Merge pull request #22 from snonux/develop

Add context awareness to dgrep

29 files changed, 604 insertions, 199 deletions

diff --git a/.gitignore b/.gitignore
index c9e8333..bac8dd1 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,14 +1,10 @@
-*_proprietary.go
-<<<<<<< HEAD
-cache/
-log/
-tags
-=======
-/cache/
-/log/
+/cache
+/log
+/tags
 /dtail
 /dgrep
 /dcat
 /dmap
 /dserver
->>>>>>> 7ee0121afed3e7cab6457142f70e411020ab2b21
+/bin
+serverlist.txt
diff --git a/cmd/dgrep/main.go b/cmd/dgrep/main.go
index 4da1bb3..123d061 100644
--- a/cmd/dgrep/main.go
+++ b/cmd/dgrep/main.go
@@ -36,15 +36,24 @@ func main() {
 	flag.IntVar(&sshPort, "port", 2222, "SSH server port")
 	flag.StringVar(&args.Discovery, "discovery", "", "Server discovery method")
 	flag.StringVar(&args.PrivateKeyPathFile, "key", "", "Path to private key")
-	flag.StringVar(&args.RegexStr, "regex", ".", "Regular expression")
 	flag.StringVar(&args.ServersStr, "servers", "", "Remote servers to connect")
 	flag.StringVar(&args.UserName, "user", userName, "Your system user name")
 	flag.StringVar(&args.What, "files", "", "File(s) to read")
 	flag.StringVar(&cfgFile, "cfg", "", "Config file path")
+
+	// Line context awareness.
+	flag.StringVar(&args.RegexStr, "regex", ".", "Regular expression")
 	flag.StringVar(&grep, "grep", "", "Alias for -regex")
+	flag.IntVar(&args.LContext.BeforeContext, "before", 0, "Print lines of leading context before matching lines")
+	flag.IntVar(&args.LContext.AfterContext, "after", 0, "Print lines of trailing context after matching lines")
+	flag.IntVar(&args.LContext.MaxCount, "max", 0, "Stop reading file after NUM matching lines")
 
 	flag.Parse()
 
+	if grep != "" {
+		args.RegexStr = grep
+	}
+
 	config.Read(cfgFile, sshPort)
 	color.Colored = !noColor
 
@@ -58,10 +67,6 @@ func main() {
 		Quiet: args.Quiet,
 	})
 
-	if grep != "" {
-		args.RegexStr = grep
-	}
-
 	client, err := clients.NewGrepClient(args)
 	if err != nil {
 		panic(err)
diff --git a/cmd/dtail/main.go b/cmd/dtail/main.go
index f2a039f..2639b4b 100644
--- a/cmd/dtail/main.go
+++ b/cmd/dtail/main.go
@@ -50,14 +50,19 @@ func main() {
 	flag.IntVar(&sshPort, "port", 2222, "SSH server port")
 	flag.StringVar(&args.Discovery, "discovery", "", "Server discovery method")
 	flag.StringVar(&args.PrivateKeyPathFile, "key", "", "Path to private key")
-	flag.StringVar(&args.RegexStr, "regex", ".", "Regular expression")
 	flag.StringVar(&args.ServersStr, "servers", "", "Remote servers to connect")
 	flag.StringVar(&args.UserName, "user", userName, "Your system user name")
 	flag.StringVar(&args.What, "files", "", "File(s) to read")
 	flag.StringVar(&cfgFile, "cfg", "", "Config file path")
-	flag.StringVar(&grep, "grep", "", "Alias for -regex")
 	flag.StringVar(&queryStr, "query", "", "Map reduce query")
 
+	// Line context awareness.
+	flag.StringVar(&args.RegexStr, "regex", ".", "Regular expression")
+	flag.StringVar(&grep, "grep", "", "Alias for -regex")
+	flag.IntVar(&args.LContext.BeforeContext, "before", 0, "Print lines of leading context before matching lines")
+	flag.IntVar(&args.LContext.AfterContext, "after", 0, "Print lines of trailing context after matching lines")
+	flag.IntVar(&args.LContext.MaxCount, "max", 0, "Stop reading file after NUM matching lines")
+
 	flag.Parse()
 
 	if grep != "" {
diff --git a/docker/Dockerfile b/docker/Dockerfile
index 8632832..61a1f7d 100644
--- a/docker/Dockerfile
+++ b/docker/Dockerfile
@@ -14,5 +14,6 @@ USER dserver
 
 WORKDIR /var/run/dserver
 EXPOSE 2222/tcp
+EXPOSE 8080/tcp
 
-CMD ["/usr/local/bin/dserver", "-relaxedAuth", "-cfg", "/etc/dserver/dtail.json"]
+CMD ["/usr/local/bin/dserver", "-relaxedAuth", "-cfg", "/etc/dserver/dtail.json", "-pprof", "8080"]
diff --git a/docker/Makefile b/docker/Makefile
index 029adf6..f09d9e0 100644
--- a/docker/Makefile
+++ b/docker/Makefile
@@ -3,7 +3,8 @@ all:
 	docker build . -t dserver:develop
 	rm ./dserver
 run:
-	docker run -p 2222:2222 dserver:develop
+	# http://localhost:8080/debug/pprof/goroutines?debug=1
+	docker run -p 2222:2222 -p 8080:8080 dserver:develop
 spinup:
 	./spinup.sh 10
 spindown:
diff --git a/docker/spindown.sh b/docker/spindown.sh
index 73ed059..b3d3a8c 100755
--- a/docker/spindown.sh
+++ b/docker/spindown.sh
@@ -6,4 +6,5 @@ declare -i BASE_PORT=2222
 for (( i=0; i < $NUM_INSTANCES; i++ )); do
     port=$[ BASE_PORT + i + 1 ]
 	docker stop dserver-serv$i 
+	docker rm dserver-serv$i 
 done
diff --git a/docker/spinup.sh b/docker/spinup.sh
index 3890ce6..0625967 100755
--- a/docker/spinup.sh
+++ b/docker/spinup.sh
@@ -3,7 +3,10 @@
 declare -i NUM_INSTANCES=$1
 declare -i BASE_PORT=2222
 
+echo > serverlist.txt
+
 for (( i=0; i < $NUM_INSTANCES; i++ )); do
     port=$[ BASE_PORT + i + 1 ]
 	docker run -d --name dserver-serv$i --hostname serv$i -p $port:2222 dserver:develop
+	echo localhost:$port >> serverlist.txt
 done
diff --git a/internal/clients/args.go b/internal/clients/args.go
index 7f782f1..684dadd 100644
--- a/internal/clients/args.go
+++ b/internal/clients/args.go
@@ -1,6 +1,7 @@
 package clients
 
 import (
+	"github.com/mimecast/dtail/internal/lcontext"
 	"github.com/mimecast/dtail/internal/omode"
 
 	gossh "golang.org/x/crypto/ssh"
@@ -8,12 +9,13 @@ import (
 
 // Args is a helper struct to summarize common client arguments.
 type Args struct {
+	lcontext.LContext
+	RegexStr           string
 	Mode               omode.Mode
 	ServersStr         string
 	UserName           string
 	What               string
 	Arguments          []string
-	RegexStr           string
 	RegexInvert        bool
 	TrustAllHosts      bool
 	Discovery          string
@@ -22,5 +24,5 @@ type Args struct {
 	SSHAuthMethods     []gossh.AuthMethod
 	SSHHostKeyCallback gossh.HostKeyCallback
 	PrivateKeyPathFile string
-	Quiet            bool
+	Quiet              bool
 }
diff --git a/internal/clients/baseclient.go b/internal/clients/baseclient.go
index f20156f..f83fcfd 100644
--- a/internal/clients/baseclient.go
+++ b/internal/clients/baseclient.go
@@ -2,6 +2,8 @@ package clients
 
 import (
 	"context"
+	"fmt"
+	"strings"
 	"sync"
 	"time"
 
@@ -118,10 +120,44 @@ func (c *baseClient) start(ctx context.Context, active chan struct{}, i int, con
 	}
 }
 
+func (c *baseClient) makeCommandOptions() map[string]string {
+	options := make(map[string]string)
+
+	if c.Args.Quiet {
+		options["quiet"] = fmt.Sprintf("%v", c.Args.Quiet)
+	}
+	if c.Args.LContext.MaxCount != 0 {
+		options["max"] = fmt.Sprintf("%d", c.Args.LContext.MaxCount)
+	}
+	if c.Args.LContext.BeforeContext != 0 {
+		options["before"] = fmt.Sprintf("%d", c.Args.LContext.BeforeContext)
+	}
+	if c.Args.LContext.AfterContext != 0 {
+		options["after"] = fmt.Sprintf("%d", c.Args.LContext.AfterContext)
+	}
+
+	return options
+}
+
+func (c *baseClient) commandOptionsToString(options map[string]string) string {
+	var sb strings.Builder
+
+	count := 0
+	for k, v := range options {
+		if count > 0 {
+			sb.WriteString(":")
+		}
+		sb.WriteString(fmt.Sprintf("%s=%s", k, v))
+		count++
+	}
+
+	return sb.String()
+}
+
 func (c *baseClient) makeConnection(server string, sshAuthMethods []gossh.AuthMethod, hostKeyCallback client.HostKeyCallback) *remote.Connection {
 	conn := remote.NewConnection(server, c.UserName, sshAuthMethods, hostKeyCallback)
 	conn.Handler = c.maker.makeHandler(server)
-	conn.Commands = c.maker.makeCommands()
+	conn.Commands = c.maker.makeCommands(c.makeCommandOptions())
 
 	return conn
 }
diff --git a/internal/clients/catclient.go b/internal/clients/catclient.go
index b7b6131..db892f1 100644
--- a/internal/clients/catclient.go
+++ b/internal/clients/catclient.go
@@ -41,10 +41,10 @@ func (c CatClient) makeHandler(server string) handlers.Handler {
 	return handlers.NewClientHandler(server)
 }
 
-func (c CatClient) makeCommands() (commands []string) {
-	options := fmt.Sprintf("quiet=%v", c.Args.Quiet)
+func (c CatClient) makeCommands(options map[string]string) (commands []string) {
+	optionsStr := c.commandOptionsToString(options)
 	for _, file := range strings.Split(c.What, ",") {
-		commands = append(commands, fmt.Sprintf("%s:%s %s %s", c.Mode.String(), options, file, c.Regex.Serialize()))
+		commands = append(commands, fmt.Sprintf("%s:%s %s %s", c.Mode.String(), optionsStr, file, c.Regex.Serialize()))
 	}
 	return
 }
diff --git a/internal/clients/grepclient.go b/internal/clients/grepclient.go
index 652c31b..567193a 100644
--- a/internal/clients/grepclient.go
+++ b/internal/clients/grepclient.go
@@ -40,10 +40,10 @@ func (c GrepClient) makeHandler(server string) handlers.Handler {
 	return handlers.NewClientHandler(server)
 }
 
-func (c GrepClient) makeCommands() (commands []string) {
-	options := fmt.Sprintf("quiet=%v", c.Args.Quiet)
+func (c GrepClient) makeCommands(options map[string]string) (commands []string) {
+	optionsStr := c.commandOptionsToString(options)
 	for _, file := range strings.Split(c.What, ",") {
-		commands = append(commands, fmt.Sprintf("%s:%s %s %s", c.Mode.String(), options, file, c.Regex.Serialize()))
+		commands = append(commands, fmt.Sprintf("%s:%s %s %s", c.Mode.String(), optionsStr, file, c.Regex.Serialize()))
 	}
 
 	return
diff --git a/internal/clients/handlers/basehandler.go b/internal/clients/handlers/basehandler.go
index f07fd90..602a7ac 100644
--- a/internal/clients/handlers/basehandler.go
+++ b/internal/clients/handlers/basehandler.go
@@ -101,15 +101,7 @@ func (h *baseHandler) handleMessageType(message string) {
 // Handle messages received from server which are not meant to be displayed
 // to the end user.
 func (h *baseHandler) handleHiddenMessage(message string) {
-	switch {
-	case strings.HasPrefix(message, ".syn close connection"):
+	if strings.HasPrefix(message, ".syn close connection") {
 		h.SendMessage(".ack close connection")
-		select {
-		case <-time.After(time.Second * 5):
-			logger.Debug("Shutting down client after timeout and sending ack to server")
-			h.Shutdown()
-		case <-h.Done():
-			return
-		}
 	}
 }
diff --git a/internal/clients/maker.go b/internal/clients/maker.go
index d5ffd8b..a1d6864 100644
--- a/internal/clients/maker.go
+++ b/internal/clients/maker.go
@@ -9,5 +9,5 @@ import (
 // and send different commands to the DTail server.
 type maker interface {
 	makeHandler(server string) handlers.Handler
-	makeCommands() (commands []string)
+	makeCommands(options map[string]string) (commands []string)
 }
diff --git a/internal/clients/maprclient.go b/internal/clients/maprclient.go
index 1c0c2cc..feb7e47 100644
--- a/internal/clients/maprclient.go
+++ b/internal/clients/maprclient.go
@@ -110,27 +110,31 @@ func (c MaprClient) makeHandler(server string) handlers.Handler {
 	return handlers.NewMaprHandler(server, c.query, c.globalGroup)
 }
 
-func (c MaprClient) makeCommands() (commands []string) {
+func (c MaprClient) makeCommands(options map[string]string) (commands []string) {
 	commands = append(commands, fmt.Sprintf("map %s", c.query.RawQuery))
-	options := fmt.Sprintf("quiet=%v", c.Args.Quiet)
 
 	modeStr := "cat"
 	if c.Mode == omode.TailClient {
 		modeStr = "tail"
 	}
+	optionsStr := c.commandOptionsToString(options)
 
 	for _, file := range strings.Split(c.What, ",") {
 		if c.Timeout > 0 {
 			commands = append(commands, fmt.Sprintf("timeout %d %s %s %s", c.Timeout, modeStr, file, c.Regex.Serialize()))
 			continue
 		}
-		commands = append(commands, fmt.Sprintf("%s:%s %s %s", modeStr, options, file, c.Regex.Serialize()))
+		commands = append(commands, fmt.Sprintf("%s:%s %s %s", modeStr, optionsStr, file, c.Regex.Serialize()))
 	}
 
 	return
 }
 
 func (c *MaprClient) periodicReportResults(ctx context.Context) {
+	rampUpSleep := c.query.Interval / 2
+	logger.Debug("Ramp up sleeping before processing mapreduce results", rampUpSleep)
+	time.Sleep(rampUpSleep)
+
 	for {
 		select {
 		case <-time.After(c.query.Interval):
diff --git a/internal/clients/tailclient.go b/internal/clients/tailclient.go
index cefbaa7..853ef1d 100644
--- a/internal/clients/tailclient.go
+++ b/internal/clients/tailclient.go
@@ -37,10 +37,10 @@ func (c TailClient) makeHandler(server string) handlers.Handler {
 	return handlers.NewClientHandler(server)
 }
 
-func (c TailClient) makeCommands() (commands []string) {
-	options := fmt.Sprintf("quiet=%v", c.Args.Quiet)
+func (c TailClient) makeCommands(options map[string]string) (commands []string) {
+	optionsStr := c.commandOptionsToString(options)
 	for _, file := range strings.Split(c.What, ",") {
-		commands = append(commands, fmt.Sprintf("%s:%s %s %s", c.Mode.String(), options, file, c.Regex.Serialize()))
+		commands = append(commands, fmt.Sprintf("%s:%s %s %s", c.Mode.String(), optionsStr, file, c.Regex.Serialize()))
 	}
 	logger.Debug(commands)
 
diff --git a/internal/datas/rbuffer.go b/internal/datas/rbuffer.go
new file mode 100644
index 0000000..df8f622
--- /dev/null
+++ b/internal/datas/rbuffer.go
@@ -0,0 +1,49 @@
+package datas
+
+import "fmt"
+
+// RBuffer is a simple circular string ring buffer data structure.
+type RBuffer struct {
+	Capacity int
+	size     int
+	readPos  int
+	writePos int
+	data     []string
+}
+
+// NewRBuffer creates a new string ring buffer.
+func NewRBuffer(capacity int) (*RBuffer, error) {
+	if capacity < 1 {
+		return nil, fmt.Errorf("RBuffer capacity must not be less than 1")
+	}
+
+	r := RBuffer{
+		Capacity: capacity,
+		size:     capacity + 1,
+		data:     make([]string, capacity+1),
+	}
+
+	return &r, nil
+}
+
+// Add a value.
+func (r *RBuffer) Add(value string) {
+	r.data[r.writePos] = value
+	r.writePos = (r.writePos + 1) % r.size
+
+	if r.writePos == r.readPos {
+		r.readPos = (r.readPos + 1) % r.size
+	}
+}
+
+// Get a value.
+func (r *RBuffer) Get() (string, bool) {
+	if r.readPos == r.writePos {
+		// RBuffer is empty.
+		return "", false
+	}
+
+	value := r.data[r.readPos]
+	r.readPos = (r.readPos + 1) % r.size
+	return value, true
+}
diff --git a/internal/datas/rbuffer_test.go b/internal/datas/rbuffer_test.go
new file mode 100644
index 0000000..456511a
--- /dev/null
+++ b/internal/datas/rbuffer_test.go
@@ -0,0 +1,106 @@
+package datas
+
+import (
+	"fmt"
+	"math/rand"
+	"testing"
+	"time"
+)
+
+func TestRBufferOneElement(t *testing.T) {
+	r, err := NewRBuffer(1)
+	if err != nil {
+		t.Errorf("Expected error creating ring buffer with capacity 1")
+	}
+
+	testRBufferValues(t, r, []string{"Hello world"})
+	testRBufferValues(t, r, []string{"Hello world", "Hello universe"})
+}
+
+func TestRBuffer(t *testing.T) {
+	if _, err := NewRBuffer(0); err == nil {
+		t.Errorf("Expected error creating ring buffer with capacity 0")
+	}
+
+	r, err := NewRBuffer(10)
+	if err != nil {
+		t.Errorf("Error creating ring buffer with capacity 10: %v", err)
+	}
+
+	fiveValues := []string{
+		"42 is the answer!",
+		"Scroption: Get over here!",
+		"Have you swiped your nectar card?",
+		"Please mind the gap between the train and the platform!",
+		"Visit DTail at https://dtail.dev",
+	}
+	testRBufferValues(t, r, fiveValues)
+
+	moreFiveValues := []string{
+		"I love Golang",
+		"As a contrast, I also love Perl",
+		"Mimecast: Stop Bad Things From Happening to Good Organizations",
+		"We are the Buetow Brothers",
+		"London is calling",
+	}
+	tenValues := append(fiveValues, moreFiveValues...)
+	testRBufferValues(t, r, tenValues)
+}
+
+func TestRandomRBuffer(t *testing.T) {
+	for i := 0; i < 100; i++ {
+		testRandomRBuffer(t)
+	}
+}
+
+func testRandomRBuffer(t *testing.T) {
+	rand.Seed(time.Now().UnixNano())
+
+	maxCapacity := 1000
+	minCapacity := 1
+	capacity := rand.Intn(maxCapacity-minCapacity) + minCapacity
+	r, err := NewRBuffer(capacity)
+	if err != nil {
+		t.Errorf("Error creating ring buffer with capacity %d: %v", capacity, err)
+	}
+
+	numValues := rand.Intn(capacity * 2)
+	values := make([]string, numValues)
+	for i := 0; i < numValues; i++ {
+		values = append(values, fmt.Sprintf("%d.%d", i, rand.Int()))
+	}
+
+	testRBufferValues(t, r, values)
+}
+
+func testRBufferValues(t *testing.T, r *RBuffer, values []string) {
+	value, ok := r.Get()
+	if ok {
+		t.Errorf("Expected not ok reading from empty ring buffer but got ok and  value '%s'", value)
+	}
+
+	for _, value := range values {
+		r.Add(value)
+	}
+
+	expectedValues := values
+	overCapacity := len(values) - r.Capacity
+	if overCapacity > 0 {
+		expectedValues = values[overCapacity:]
+	}
+
+	for _, expected := range expectedValues {
+		value, ok := r.Get()
+		if !ok {
+			t.Errorf("Expected value '%s' but got nothing", expected)
+		}
+		if value != expected {
+			t.Errorf("Expected value '%s' but got value '%v'", expected, value)
+		}
+	}
+
+	value, ok = r.Get()
+	if ok {
+		t.Errorf("Expected not ok reading from empty ring buffer but got ok and  value '%s'", value)
+	}
+}
diff --git a/internal/io/fs/filereader.go b/internal/io/fs/filereader.go
index 0774837..efd410e 100644
--- a/internal/io/fs/filereader.go
+++ b/internal/io/fs/filereader.go
@@ -4,12 +4,13 @@ import (
 	"context"
 
 	"github.com/mimecast/dtail/internal/io/line"
+	"github.com/mimecast/dtail/internal/lcontext"
 	"github.com/mimecast/dtail/internal/regex"
 )
 
 // FileReader is the interface used on the dtail server to read/cat/grep/mapr... a file.
 type FileReader interface {
-	Start(ctx context.Context, lines chan<- line.Line, re regex.Regex) error
+	Start(ctx context.Context, lContext lcontext.LContext, lines chan<- line.Line, re regex.Regex) error
 	FilePath() string
 	Retry() bool
 }
diff --git a/internal/io/fs/filter.go b/internal/io/fs/filter.go
new file mode 100644
index 0000000..c4f605e
--- /dev/null
+++ b/internal/io/fs/filter.go
@@ -0,0 +1,167 @@
+package fs
+
+import (
+	"context"
+
+	"github.com/mimecast/dtail/internal/io/line"
+	"github.com/mimecast/dtail/internal/lcontext"
+	"github.com/mimecast/dtail/internal/regex"
+)
+
+func (f readFile) filter(ctx context.Context, rawLines <-chan []byte, lines chan<- line.Line, re regex.Regex, lContext lcontext.LContext) {
+	// Do we have any kind of local context settings? If so then run the more complex
+	// filterWithLContext method.
+	if lContext.Has() {
+		// We can not skip transmitting any lines to the client with a local
+		// grep context specified.
+		f.canSkipLines = false
+		f.filterWithLContext(ctx, rawLines, lines, re, lContext)
+		return
+	}
+
+	f.filterWithoutLContext(ctx, rawLines, lines, re)
+}
+
+// Filter log lines matching a given regular expression, however with local grep context.
+func (f readFile) filterWithLContext(ctx context.Context, rawLines <-chan []byte, lines chan<- line.Line, re regex.Regex, lContext lcontext.LContext) {
+	// Scenario 1: Finish once maxCount hits found
+	maxCount := lContext.MaxCount
+	processMaxCount := maxCount > 0
+	maxReached := false
+
+	// Scenario 2: Print prev. N lines when current line matches.
+	before := lContext.BeforeContext
+	processBefore := before > 0
+	var beforeBuf chan []byte
+	if processBefore {
+		beforeBuf = make(chan []byte, before)
+	}
+
+	// Screnario 3: Print next N lines when current line matches.
+	after := 0
+	processAfter := lContext.AfterContext > 0
+
+	for rawLine := range rawLines {
+		// logger.Debug("rawLine", string(rawLine))
+		f.updatePosition()
+
+		if !re.Match(rawLine) {
+			f.updateLineNotMatched()
+
+			if processAfter && after > 0 {
+				after--
+				myLine := line.Line{Content: rawLine, SourceID: f.globID, Count: f.totalLineCount(), TransmittedPerc: 100}
+				select {
+				case lines <- myLine:
+				case <-ctx.Done():
+					return
+				}
+
+			} else if processBefore {
+				// Keep last num BeforeContext raw messages.
+				select {
+				case beforeBuf <- rawLine:
+				default:
+					<-beforeBuf
+					beforeBuf <- rawLine
+				}
+			}
+			continue
+		}
+
+		f.updateLineMatched()
+
+		if processAfter {
+			if maxReached {
+				return
+			}
+			after = lContext.AfterContext
+		}
+
+		if processBefore {
+			i := uint64(len(beforeBuf))
+			for {
+				select {
+				case myRawLine := <-beforeBuf:
+					myLine := line.Line{Content: myRawLine, SourceID: f.globID, Count: f.totalLineCount() - i, TransmittedPerc: 100}
+					i--
+					select {
+					case lines <- myLine:
+					case <-ctx.Done():
+						return
+					}
+				default:
+					// beforeBuf is now empty.
+				}
+				if len(beforeBuf) == 0 {
+					break
+				}
+			}
+		}
+
+		line := line.Line{Content: rawLine, SourceID: f.globID, Count: f.totalLineCount(), TransmittedPerc: 100}
+
+		select {
+		case lines <- line:
+			if processMaxCount {
+				maxCount--
+				if maxCount == 0 {
+					if !processAfter || after == 0 {
+						return
+					}
+					// Unfortunatley we have to continue filter, as there might be more lines to print
+					maxReached = true
+				}
+			}
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+// Filter log lines matching a given regular expression, there is no local grep context specified.
+func (f readFile) filterWithoutLContext(ctx context.Context, rawLines <-chan []byte, lines chan<- line.Line, re regex.Regex) {
+	for {
+		select {
+		case rawLine, ok := <-rawLines:
+			f.updatePosition()
+			if !ok {
+				return
+			}
+
+			if f.lineUntransmittable(rawLine, len(lines), cap(lines), re) {
+				continue
+			}
+
+			line := line.Line{Content: rawLine, SourceID: f.globID, Count: f.totalLineCount(), TransmittedPerc: f.transmittedPerc()}
+
+			select {
+			case lines <- line:
+				continue
+			case <-ctx.Done():
+				return
+			}
+		}
+	}
+}
+
+func (f readFile) lineUntransmittable(rawLine []byte, length, capacity int, re regex.Regex) bool {
+	if !re.Match(rawLine) {
+		f.updateLineNotMatched()
+		f.updateLineNotTransmitted()
+		// Regex dosn't match, so not interested in it.
+		return true
+	}
+	f.updateLineMatched()
+
+	// Can we actually send more messages, channel capacity reached?
+	if f.canSkipLines && length >= capacity {
+		f.updateLineNotTransmitted()
+		// Matching, not transmittable
+		return true
+	}
+	f.updateLineTransmitted()
+
+	// Matching, transmittable
+	return false
+}
diff --git a/internal/io/fs/readfile.go b/internal/io/fs/readfile.go
index 6757bd6..161e3f0 100644
--- a/internal/io/fs/readfile.go
+++ b/internal/io/fs/readfile.go
@@ -4,16 +4,15 @@ import (
 	"bufio"
 	"compress/gzip"
 	"context"
-	"errors"
 	"fmt"
 	"io"
 	"os"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/mimecast/dtail/internal/io/line"
 	"github.com/mimecast/dtail/internal/io/logger"
+	"github.com/mimecast/dtail/internal/lcontext"
 	"github.com/mimecast/dtail/internal/regex"
 
 	"github.com/DataDog/zstd"
@@ -38,6 +37,28 @@ type readFile struct {
 	limiter chan struct{}
 }
 
+func (f readFile) makeReader(fd *os.File) (reader *bufio.Reader, err error) {
+	switch {
+	case strings.HasSuffix(f.FilePath(), ".gz"):
+		fallthrough
+	case strings.HasSuffix(f.FilePath(), ".gzip"):
+		logger.Info(f.FilePath(), "Detected gzip compression format")
+		var gzipReader *gzip.Reader
+		gzipReader, err = gzip.NewReader(fd)
+		if err != nil {
+			return
+		}
+		reader = bufio.NewReader(gzipReader)
+	case strings.HasSuffix(f.FilePath(), ".zst"):
+		logger.Info(f.FilePath(), "Detected zstd compression format")
+		reader = bufio.NewReader(zstd.NewReader(fd))
+	default:
+		reader = bufio.NewReader(fd)
+	}
+
+	return
+}
+
 // String returns the string representation of the readFile
 func (f readFile) String() string {
 	return fmt.Sprintf("readFile(filePath:%s,globID:%s,retry:%v,canSkipLines:%v,seekEOF:%v)",
@@ -59,7 +80,7 @@ func (f readFile) Retry() bool {
 }
 
 // Start tailing a log file.
-func (f readFile) Start(ctx context.Context, lines chan<- line.Line, re regex.Regex) error {
+func (f readFile) Start(ctx context.Context, lContext lcontext.LContext, lines chan<- line.Line, re regex.Regex) error {
 	logger.Debug("readFile", f)
 	defer func() {
 		select {
@@ -90,58 +111,28 @@ func (f readFile) Start(ctx context.Context, lines chan<- line.Line, re regex.Re
 	}
 
 	rawLines := make(chan []byte, 100)
-	truncate := make(chan struct{})
-
-	var wg sync.WaitGroup
-	wg.Add(1)
-
-	go f.periodicTruncateCheck(ctx, truncate)
-	go f.filter(ctx, &wg, rawLines, lines, re)
+	readCtx, readCancel := context.WithCancel(ctx)
+
+	filterDone := make(chan struct{})
+	go func() {
+		f.filter(ctx, rawLines, lines, re, lContext)
+		close(filterDone)
+		// If the filter stopped, make the reader stop too, no need to read
+		// more data if there is nothing more the filter wants to filter for!
+		// E.g. it could be that we only want to filter N matches but not more.
+		readCancel()
+	}()
 
-	err = f.read(ctx, fd, rawLines, truncate)
+	err = f.read(readCtx, fd, rawLines)
 	close(rawLines)
-	wg.Wait()
-
-	return err
-}
 
-func (f readFile) periodicTruncateCheck(ctx context.Context, truncate chan struct{}) {
-	for {
-		select {
-		case <-time.After(time.Second * 3):
-			select {
-			case truncate <- struct{}{}:
-			case <-ctx.Done():
-			}
-		case <-ctx.Done():
-			return
-		}
-	}
-}
-
-func (f readFile) makeReader(fd *os.File) (reader *bufio.Reader, err error) {
-	switch {
-	case strings.HasSuffix(f.FilePath(), ".gz"):
-		fallthrough
-	case strings.HasSuffix(f.FilePath(), ".gzip"):
-		logger.Info(f.FilePath(), "Detected gzip compression format")
-		var gzipReader *gzip.Reader
-		gzipReader, err = gzip.NewReader(fd)
-		if err != nil {
-			return
-		}
-		reader = bufio.NewReader(gzipReader)
-	case strings.HasSuffix(f.FilePath(), ".zst"):
-		logger.Info(f.FilePath(), "Detected zstd compression format")
-		reader = bufio.NewReader(zstd.NewReader(fd))
-	default:
-		reader = bufio.NewReader(fd)
-	}
+	// Filter may flushes some data still. So wait until it is done here.
+	<-filterDone
 
-	return
+	return err
 }
 
-func (f readFile) read(ctx context.Context, fd *os.File, rawLines chan []byte, truncate <-chan struct{}) error {
+func (f readFile) read(ctx context.Context, fd *os.File, rawLines chan []byte) error {
 	var offset uint64
 
 	reader, err := f.makeReader(fd)
@@ -153,6 +144,8 @@ func (f readFile) read(ctx context.Context, fd *os.File, rawLines chan []byte, t
 	lineLengthThreshold := 1024 * 1024 // 1mb
 	longLineWarning := false
 
+	checkTruncate := f.truncateTimer(ctx)
+
 	for {
 		select {
 		case <-ctx.Done():
@@ -161,7 +154,7 @@ func (f readFile) read(ctx context.Context, fd *os.File, rawLines chan []byte, t
 		}
 
 		select {
-		case <-truncate:
+		case <-checkTruncate:
 			if isTruncated, err := f.truncated(fd); isTruncated {
 				return err
 			}
@@ -225,82 +218,3 @@ func (f readFile) read(ctx context.Context, fd *os.File, rawLin