use buffered tailer in fileTailer_test

Author: fstab

grok_exporter.go

@@ -282,5 +282,5 @@ func startTailer(cfg *v2.Config) (tailer.Tailer, error) {
 	default:
 		return nil, fmt.Errorf("Config error: Input type '%v' unknown.", cfg.Input.Type)
 	}
-	return exporter.BufferedTailerWithMetrics(tail), nil
+	return tailer.BufferedTailerWithMetrics(tail), nil
 }

exporter/bufferedTailer.go tailer/bufferedTailer.goexporter/bufferedTailer.go renamed to tailer/bufferedTailer.go

@@ -12,28 +12,27 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package exporter
+package tailer
 
 import (
 	"container/list"
-	"github.com/fstab/grok_exporter/tailer"
 	"github.com/prometheus/client_golang/prometheus"
 	"log"
 	"sync"
 	"time"
 )
 
-// implements tailer.Tailer
+// implements Tailer
 type bufferedTailerWithMetrics struct {
 	out  chan string
-	orig tailer.Tailer
+	orig Tailer
 }
 
 func (b *bufferedTailerWithMetrics) Lines() chan string {
 	return b.out
 }
 
-func (b *bufferedTailerWithMetrics) Errors() chan tailer.Error {
+func (b *bufferedTailerWithMetrics) Errors() chan Error {
 	return b.orig.Errors()
 }
 
@@ -46,7 +45,61 @@ func (b *bufferedTailerWithMetrics) Close() {
 // and does not need to wait until the lines are processed.
 // The number of buffered lines are exposed as a Prometheus metric, if lines are constantly
 // produced faster than they are consumed, we will eventually run out of memory.
-func BufferedTailerWithMetrics(orig tailer.Tailer) tailer.Tailer {
+//
+// ---
+// The buffered tailer prevents the following error (this can be reproduced on Windows,
+// where we don't keep the logfile open):
+//
+// Example test actions
+// --------------------
+//
+// Sequence of actions simulated in fileTailer_test:
+//
+// 1) write line a
+// 2) write line b
+// 3) move the old logfile away and create a new logfile
+// 4) write line c
+//
+// Good case event processing
+// --------------------------
+//
+// How Events.Process() should process the file system events triggered by the actions above:
+//
+// 1) MODIFIED : Process() reads line a
+// 2) MODIFIED : Process() reads line b
+// 3) MOVED_FROM, CREATED : Process() resets the line reader and seeks the file to position 0
+// 4) MODIFIED : Process() reads line c
+//
+// Bad case event processing
+// -------------------------
+//
+// When Events.Process() receives a MODIFIED event, it does not know how many lines have been written.
+// Therefore, it reads all new lines until EOF is reached.
+// If line processing is slow (writing to the lines channel blocks until all grok patterns are processed),
+// we might read 'line b' while we are still processing the first MODIFIED event:
+//
+// 1) MODIFIED : Process() reads 'line a' and 'line b'
+//
+// Meanwhile, the test continues with steps 3 and 4, moving the logfile away, creating a new logfile,
+// and writing 'line c'. When the tailer receives the second MODIFIED event, it learns that the file
+// has been truncated, seeks to position 0, and reads 'line c'.
+//
+// 2) MODIFIED : Process() detects the truncated file, seeks to position 0, reads 'line c'
+//
+// The tailer now receives MOVED_FROM, which makes it close the logfile, CREATED, which makes
+// it open the logfile and start reading from position 0:
+//
+// 3) MOVED_FROM, CREATED : seek to position 0, read line c again !!!
+//
+// When the last MODIFIED event is processed, there are no more changes in the file:
+//
+// 4) MODIFIED : no changes in file
+//
+// As a result, we read 'line c' two times.
+//
+// To minimize the risk, use the buffered tailer to make sure file system events are handled
+// as quickly as possible without waiting for the grok patterns to be processed.
+func BufferedTailerWithMetrics(orig Tailer) Tailer {
 	buffer := list.New()
 	bufferSync := sync.NewCond(&sync.Mutex{}) // coordinate producer and consumer
 	out := make(chan string)

exporter/bufferedTailer_test.go tailer/bufferedTailer_test.goexporter/bufferedTailer_test.go renamed to tailer/bufferedTailer_test.go

@@ -12,11 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package exporter
+package tailer
 
 import (
 	"fmt"
-	"github.com/fstab/grok_exporter/tailer"
 	"math/rand"
 	"sync"
 	"testing"
@@ -31,7 +30,7 @@ func (tail *sourceTailer) Lines() chan string {
 	return tail.lines
 }
 
-func (tail *sourceTailer) Errors() chan tailer.Error {
+func (tail *sourceTailer) Errors() chan Error {
 	return nil
 }
 

tailer/fileTailer.go

@@ -60,36 +60,7 @@ func runFileTailer(path string, readall bool, failOnMissingFile bool, logger sim
 		logger = &nilLogger{}
 	}
 
-	// Why does the lines channel have a size of 10?
-	// ---
-	// In fileTailer_test, we read a line from the channel, then we run the next action
-	// on the logfile (like writing the next line, simulating logrotate, etc.).
-	//
-	// Example:
-	// 1) write line a
-	// 2) write line b
-	// 3) move the old logfile away and create a new logfile
-	// 4) write line c
-	//
-	// With this example, we would expect 4 calls to Events.Process() with the following events:
-	// 1) MODIFIED : Process() reads line a
-	// 2) MODIFIED : Process() reads line b
-	// 3) MOVED_FROM, CREATED : Process() resets the line reader and seeks the file to position 0
-	// 4) MODIFIED : Process() reads line c
-	//
-	// However, there is a race condition: When the line reader is slow, it does not hit EOF
-	// before line b is written. The reader keeps reading upon the 1st MODIFIED event.
-	// 1) MODIFIED : Process() reads line a and line b
-	// 2) MODIFIED : Process() detects the truncated file, seeks to position 0, reads line c
-	// 3) MOVED_FROM, CREATED : seek to position 0, read line c again !!!
-	// 4) MODIFIED : no changes in file
-	//
-	// As a result, we read 'line c' two times.
-	// This problem occurs especially on Windows where we don't keep the file open.
-	//
-	// To minimize the risk, we give the lines channel size 10, so that the line reader can
-	// continue reading the next few lines and does not need to wait until the lines are processed.
-	lines := make(chan string, 10)
+	lines := make(chan string)
 	done := make(chan struct{})
 	errors := make(chan Error)
 

tailer/fileTailer_test.go

@@ -245,6 +245,7 @@ func testLogrotate(t *testing.T, log simpleLogger, watcherOpt watcherType, logro
 	case polling:
 		tail = RunPollingFileTailer(logfile, true, true, 10*time.Millisecond, log)
 	}
+	tail = BufferedTailerWithMetrics(tail)
 	defer tail.Close()
 
 	// We don't expect errors. However, start a go-routine listening on