fix race condition with line reader

Author: fstab

tailer/fileTailer.go

@@ -60,7 +60,36 @@ func runFileTailer(path string, readall bool, failOnMissingFile bool, logger sim
 		logger = &nilLogger{}
 	}
 
-	lines := make(chan string)
+	// 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)
 	done := make(chan struct{})
 	errors := make(chan Error)
 

tailer/fileTailer_test.go

@@ -305,7 +305,11 @@ func (l *closeFileAfterEachLineLogFileWriter) writeLine(t *testing.T, log simple
 	if err != nil {
 		t.Fatalf("%v: Failed to open file for writing: %v", l.path, err.Error())
 	}
-	_, err = f.WriteString(fmt.Sprintf("%v\n", line))
+	newline := "\n"
+	if runtime.GOOS == "windows" {
+		newline = "\r\n"
+	}
+	_, err = f.WriteString(fmt.Sprintf("%v%v", line, newline))
 	if err != nil {
 		t.Fatalf("%v: Failed to write to file: %v", l.path, err.Error())
 	}

tailer/fileTailer_windows.go

@@ -114,19 +114,19 @@ func (event *event) Process(fileBefore *File, reader *lineReader, abspath string
 	)
 	logger.Debug("File system watcher received %v.\n", event.String())
 
-	// WRITE or TRUNCATE
-	if file != nil && norm(event.Name) == norm(abspath) && event.Mask&winfsnotify.FS_MODIFY == winfsnotify.FS_MODIFY {
-		truncated, err = file.CheckTruncated()
+	// MOVED_FROM or DELETE
+	if file != nil && norm(event.Name) == norm(abspath) && (event.Mask&winfsnotify.FS_MOVED_FROM == winfsnotify.FS_MOVED_FROM || event.Mask&winfsnotify.FS_DELETE == winfsnotify.FS_DELETE) {
+		file = nil
+		reader.Clear()
+	}
+
+	// MOVED_TO or CREATE
+	if file == nil && norm(event.Name) == norm(abspath) && (event.Mask&winfsnotify.FS_MOVED_TO == winfsnotify.FS_MOVED_TO || event.Mask&winfsnotify.FS_CREATE == winfsnotify.FS_CREATE) {
+		file, err = open(abspath)
 		if err != nil {
 			return
 		}
-		if truncated {
-			_, err = file.Seek(0, io.SeekStart)
-			if err != nil {
-				return
-			}
-			reader.Clear()
-		}
+		reader.Clear()
 		for {
 			line, eof, err = reader.ReadLine(file)
 			if err != nil {
@@ -139,19 +139,19 @@ func (event *event) Process(fileBefore *File, reader *lineReader, abspath string
 		}
 	}
 
-	// MOVED_FROM or DELETE
-	if file != nil && norm(event.Name) == norm(abspath) && (event.Mask&winfsnotify.FS_MOVED_FROM == winfsnotify.FS_MOVED_FROM || event.Mask&winfsnotify.FS_DELETE == winfsnotify.FS_DELETE) {
-		file = nil
-		reader.Clear()
-	}
-
-	// MOVED_TO or CREATE
-	if file == nil && norm(event.Name) == norm(abspath) && (event.Mask&winfsnotify.FS_MOVED_TO == winfsnotify.FS_MOVED_TO || event.Mask&winfsnotify.FS_CREATE == winfsnotify.FS_CREATE) {
-		file, err = open(abspath)
+	// WRITE or TRUNCATE
+	if file != nil && norm(event.Name) == norm(abspath) && event.Mask&winfsnotify.FS_MODIFY == winfsnotify.FS_MODIFY {
+		truncated, err = file.CheckTruncated()
 		if err != nil {
 			return
 		}
-		reader.Clear()
+		if truncated {
+			_, err = file.Seek(0, io.SeekStart)
+			if err != nil {
+				return
+			}
+			reader.Clear()
+		}
 		for {
 			line, eof, err = reader.ReadLine(file)
 			if err != nil {
@@ -163,6 +163,7 @@ func (event *event) Process(fileBefore *File, reader *lineReader, abspath string
 			lines = append(lines, line)
 		}
 	}
+
 	return
 }
 

tailer/file_windows.go

@@ -48,6 +48,10 @@ func (f *File) Seek(offset int64, whence int) (int64, error) {
 		return 0, err
 	}
 	defer file.Close()
+	_, err = file.Seek(f.currentPos, io.SeekStart)
+	if err != nil {
+		return 0, err
+	}
 	result, resultErr := file.Seek(offset, whence)
 	f.currentPos, err = file.Seek(0, io.SeekCurrent)
 	if err != nil {