#5: replace polling file tailer with multi file version

Author: fstab

grok_exporter.go

@@ -22,6 +22,8 @@ import (
 	"github.com/fstab/grok_exporter/exporter"
 	"github.com/fstab/grok_exporter/oniguruma"
 	"github.com/fstab/grok_exporter/tailer"
+	"github.com/fstab/grok_exporter/tailer/fswatcher"
+	"github.com/fstab/grok_exporter/tailer/glob"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/sirupsen/logrus"
 	"net/http"
@@ -87,21 +89,21 @@ func main() {
 			matched := false
 			for _, metric := range metrics {
 				start := time.Now()
-				match, err := metric.ProcessMatch(line)
+				match, err := metric.ProcessMatch(line.Line)
 				if err != nil {
 					fmt.Fprintf(os.Stderr, "WARNING: skipping log line: %v\n", err.Error())
-					fmt.Fprintf(os.Stderr, "%v\n", line)
+					fmt.Fprintf(os.Stderr, "%v\n", line.Line)
 					nErrorsByMetric.WithLabelValues(metric.Name()).Inc()
 				}
 				if match != nil {
 					nMatchesByMetric.WithLabelValues(metric.Name()).Inc()
 					procTimeMicrosecondsByMetric.WithLabelValues(metric.Name()).Add(float64(time.Since(start).Nanoseconds() / int64(1000)))
 					matched = true
 				}
-				_, err = metric.ProcessDeleteMatch(line)
+				_, err = metric.ProcessDeleteMatch(line.Line)
 				if err != nil {
 					fmt.Fprintf(os.Stderr, "WARNING: skipping log line: %v\n", err.Error())
-					fmt.Fprintf(os.Stderr, "%v\n", line)
+					fmt.Fprintf(os.Stderr, "%v\n", line.Line)
 					nErrorsByMetric.WithLabelValues(metric.Name()).Inc()
 				}
 				// TODO: create metric to monitor number of matching delete_patterns
@@ -269,16 +271,20 @@ func startServer(cfg v2.ServerConfig, handler http.Handler) chan error {
 	return serverErrors
 }
 
-func startTailer(cfg *v2.Config) (tailer.Tailer, error) {
+func startTailer(cfg *v2.Config) (fswatcher.FileTailer, error) {
 	logger := logrus.New()
 	logger.Level = logrus.WarnLevel
-	var tail tailer.Tailer
+	var tail fswatcher.FileTailer
+	g, err := glob.FromPath(cfg.Input.Path)
+	if err != nil {
+		return nil, err
+	}
 	switch {
 	case cfg.Input.Type == "file":
 		if cfg.Input.PollInterval == 0 {
-			tail = tailer.RunFseventFileTailer(cfg.Input.Path, cfg.Input.Readall, cfg.Input.FailOnMissingLogfile, logger)
+			tail, err = fswatcher.RunFileTailer([]glob.Glob{g}, cfg.Input.Readall, cfg.Input.FailOnMissingLogfile, logger)
 		} else {
-			tail = tailer.RunPollingFileTailer(cfg.Input.Path, cfg.Input.Readall, cfg.Input.FailOnMissingLogfile, cfg.Input.PollInterval, logger)
+			tail, err = fswatcher.RunPollingFileTailer([]glob.Glob{g}, cfg.Input.Readall, cfg.Input.FailOnMissingLogfile, cfg.Input.PollInterval, logger)
 		}
 	case cfg.Input.Type == "stdin":
 		tail = tailer.RunStdinTailer()

tailer/bufferedTailer.go

@@ -16,30 +16,31 @@ package tailer
 
 import (
 	"container/list"
+	"github.com/fstab/grok_exporter/tailer/fswatcher"
 	"log"
 	"sync"
 	"time"
 )
 
-// implements Tailer
+// implements fswatcher.FileTailer
 type bufferedTailer struct {
-	out  chan string
-	orig Tailer
+	out  chan fswatcher.Line
+	orig fswatcher.FileTailer
 }
 
-func (b *bufferedTailer) Lines() chan string {
+func (b *bufferedTailer) Lines() chan fswatcher.Line {
 	return b.out
 }
 
-func (b *bufferedTailer) Errors() chan Error {
+func (b *bufferedTailer) Errors() chan fswatcher.Error {
 	return b.orig.Errors()
 }
 
 func (b *bufferedTailer) Close() {
 	b.orig.Close()
 }
 
-func BufferedTailer(orig Tailer) Tailer {
+func BufferedTailer(orig fswatcher.FileTailer) fswatcher.FileTailer {
 	return BufferedTailerWithMetrics(orig, &noopMetric{})
 }
 
@@ -102,10 +103,10 @@ func BufferedTailer(orig Tailer) Tailer {
 //
 // 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, bufferLoadMetric BufferLoadMetric) Tailer {
+func BufferedTailerWithMetrics(orig fswatcher.FileTailer, bufferLoadMetric BufferLoadMetric) fswatcher.FileTailer {
 	buffer := list.New()
 	bufferSync := sync.NewCond(&sync.Mutex{}) // coordinate producer and consumer
-	out := make(chan string)
+	out := make(chan fswatcher.Line)
 
 	// producer
 	go func() {
@@ -155,7 +156,7 @@ func BufferedTailerWithMetrics(orig Tailer, bufferLoadMetric BufferLoadMetric) T
 			buffer.Remove(first)
 			bufferSync.L.Unlock()
 			switch line := first.Value.(type) {
-			case string:
+			case fswatcher.Line:
 				out <- line
 			default:
 				// this cannot happen

tailer/bufferedTailer_test.go

@@ -16,21 +16,22 @@ package tailer
 
 import (
 	"fmt"
+	"github.com/fstab/grok_exporter/tailer/fswatcher"
 	"math/rand"
 	"sync"
 	"testing"
 	"time"
 )
 
 type sourceTailer struct {
-	lines chan string
+	lines chan fswatcher.Line
 }
 
-func (tail *sourceTailer) Lines() chan string {
+func (tail *sourceTailer) Lines() chan fswatcher.Line {
 	return tail.lines
 }
 
-func (tail *sourceTailer) Errors() chan Error {
+func (tail *sourceTailer) Errors() chan fswatcher.Error {
 	return nil
 }
 
@@ -40,15 +41,15 @@ func (tail *sourceTailer) Close() {
 
 // First produce 10,000 lines, then consume 10,000 lines.
 func TestLineBufferSequential(t *testing.T) {
-	src := &sourceTailer{lines: make(chan string)}
+	src := &sourceTailer{lines: make(chan fswatcher.Line)}
 	metric := &peakLoadMetric{}
 	buffered := BufferedTailerWithMetrics(src, metric)
 	for i := 1; i <= 10000; i++ {
-		src.lines <- fmt.Sprintf("This is line number %v.", i)
+		src.lines <- fswatcher.Line{Line: fmt.Sprintf("This is line number %v.", i)}
 	}
 	for i := 1; i <= 10000; i++ {
 		line := <-buffered.Lines()
-		if line != fmt.Sprintf("This is line number %v.", i) {
+		if line.Line != fmt.Sprintf("This is line number %v.", i) {
 			t.Errorf("Expected 'This is line number %v', but got '%v'.", i, line)
 		}
 	}
@@ -76,14 +77,14 @@ func TestLineBufferSequential(t *testing.T) {
 
 // Produce and consume in parallel.
 func TestLineBufferParallel(t *testing.T) {
-	src := &sourceTailer{lines: make(chan string)}
+	src := &sourceTailer{lines: make(chan fswatcher.Line)}
 	metric := &peakLoadMetric{}
 	buffered := BufferedTailerWithMetrics(src, metric)
 	var wg sync.WaitGroup
 	go func() {
 		start := time.Now()
 		for i := 1; i <= 10000; i++ {
-			src.lines <- fmt.Sprintf("This is line number %v.", i)
+			src.lines <- fswatcher.Line{Line: fmt.Sprintf("This is line number %v.", i)}
 			if rand.Int()%64 == 0 { // Sleep from time to time
 				time.Sleep(10 * time.Millisecond)
 			}
@@ -95,7 +96,7 @@ func TestLineBufferParallel(t *testing.T) {
 		start := time.Now()
 		for i := 1; i <= 10000; i++ {
 			line := <-buffered.Lines()
-			if line != fmt.Sprintf("This is line number %v.", i) {
+			if line.Line != fmt.Sprintf("This is line number %v.", i) {
 				t.Errorf("Expected 'This is line number %v', but got '%v'.", i, line)
 			}
 			if rand.Int()%64 == 0 { // Sleep from time to time

tailer/fileTailer_test.go

@@ -16,6 +16,8 @@ package tailer
 
 import (
 	"fmt"
+	"github.com/fstab/grok_exporter/tailer/fswatcher"
+	"github.com/fstab/grok_exporter/tailer/glob"
 	"github.com/sirupsen/logrus"
 	"io/ioutil"
 	"os"
@@ -166,7 +168,14 @@ func TestVisibleInOSXFinder(t *testing.T) {
 	logFileWriter := newLogFileWriter(t, logfile, closeFileAfterEachLine)
 	defer logFileWriter.close(t)
 	logFileWriter.writeLine(t, log, "test line 1")
-	tail := RunFseventFileTailer(logfile, false, true, log)
+	g, err := glob.FromPath(logfile)
+	if err != nil {
+		t.Fatal(err)
+	}
+	tail, err := fswatcher.RunFileTailer([]glob.Glob{g}, false, true, log)
+	if err != nil {
+		t.Fatalf("failed to start file tailer: %v", err)
+	}
 	defer func() {
 		tail.Close()
 		// wait until closed
@@ -204,7 +213,14 @@ func TestFileMissingOnStartup(t *testing.T) {
 	tmpDir := mkTmpDirOrFail(t)
 	defer cleanUp(t, tmpDir)
 	var logfile = fmt.Sprintf("%s%c%s", tmpDir, os.PathSeparator, logfileName)
-	tail := RunFseventFileTailer(logfile, true, false, log)
+	g, err := glob.FromPath(logfile)
+	if err != nil {
+		t.Fatal(err)
+	}
+	tail, err := fswatcher.RunFileTailer([]glob.Glob{g}, true, false, log)
+	if err != nil {
+		t.Fatalf("failed to start file system tailer: %v", err)
+	}
 	defer func() {
 		tail.Close()
 		// wait until closed
@@ -226,7 +242,7 @@ func TestFileMissingOnStartup(t *testing.T) {
 	}()
 
 	// Double check that file does not exist yet
-	_, err := os.Stat(logfile)
+	_, err = os.Stat(logfile)
 	if err == nil || !os.IsNotExist(err) {
 		t.Fatalf("%v should not exist yet", logfile)
 	}
@@ -265,12 +281,19 @@ func testLogrotate(t *testing.T, log logrus.FieldLogger, watcherOpt watcherType,
 	logFileWriter.writeLine(t, log, "test line 1")
 	logFileWriter.writeLine(t, log, "test line 2")
 
-	var tail Tailer
+	var tail fswatcher.FileTailer
+	g, err := glob.FromPath(logfile)
+	if err != nil {
+		t.Fatal(err)
+	}
 	switch watcherOpt {
 	case fsevent:
-		tail = RunFseventFileTailer(logfile, true, true, log)
+		tail, err = fswatcher.RunFileTailer([]glob.Glob{g}, true, true, log)
 	case polling:
-		tail = RunPollingFileTailer(logfile, true, true, 10*time.Millisecond, log)
+		tail, err = fswatcher.RunPollingFileTailer([]glob.Glob{g}, true, true, 10*time.Millisecond, log)
+	}
+	if err != nil {
+		t.Fatalf("failed to start file system watcher: %v", err)
 	}
 	tail = BufferedTailer(tail)
 	defer func() {
@@ -608,15 +631,15 @@ func truncateOrFail(t *testing.T, logfile string) {
 	}
 }
 
-func expect(t *testing.T, log logrus.FieldLogger, c chan string, line string, timeout time.Duration) {
+func expect(t *testing.T, log logrus.FieldLogger, lines chan fswatcher.Line, line string, timeout time.Duration) {
 	timeoutChan := make(chan bool, 1)
 	go func() {
 		time.Sleep(timeout)
 		close(timeoutChan)
 	}()
 	select {
-	case result := <-c:
-		if result != line {
+	case result := <-lines:
+		if result.Line != line {
 			t.Fatalf("Expected '%v', but got '%v'.", line, result)
 		} else {
 			log.Debugf("Read expected line '%v'\n", line)

tailer/fswatcher/fswatcher_windows.go

@@ -150,16 +150,8 @@ func (w *watcher) processEvent(t *fileTailer, fsevent fsevent, log logrus.FieldL
 	return nil
 }
 
-func isTruncated(file *os.File) (bool, Error) {
-	currentPos, err := file.Seek(0, io.SeekCurrent)
-	if err != nil {
-		return false, NewError(NotSpecified, os.NewSyscallError("seek", err), file.Name())
-	}
-	fileInfo, err := file.Stat()
-	if err != nil {
-		return false, NewError(NotSpecified, os.NewSyscallError("stat", err), file.Name())
-	}
-	return currentPos > fileInfo.Size(), nil
+func isTruncated(file *File) (bool, Error) {
+	return file.CheckTruncated()
 }
 
 func findSameFile(t *fileTailer, newFileInfo *fileInfo, path string) (*fileWithReader, Error) {

tailer/fswatcher/pollingwatcher.go

@@ -2,6 +2,7 @@ package fswatcher
 
 import (
 	"github.com/sirupsen/logrus"
+	"io"
 	"time"
 )
 
@@ -26,6 +27,23 @@ func (w *pollingWatcher) processEvent(t *fileTailer, fsevent fsevent, log logrus
 			return err
 		}
 	}
+	for _, file := range t.watchedFiles {
+		truncated, err := isTruncated(file.file)
+		if err != nil {
+			return NewErrorf(NotSpecified, err, "%v: seek() or stat() failed", file.file.Name())
+		}
+		if truncated {
+			_, err = file.file.Seek(0, io.SeekStart)
+			if err != nil {
+				return NewErrorf(NotSpecified, err, "%v: seek() failed", file.file.Name())
+			}
+			file.reader.Clear()
+		}
+		readErr := t.readNewLines(file, log)
+		if readErr != nil {
+			return readErr
+		}
+	}
 	return nil
 }
 

tailer/glob/glob.go

@@ -51,6 +51,17 @@ func (g Glob) Match(path string) bool {
 	return matched
 }
 
+// The file tailer implementation switched from watching single paths to globs,
+// but the rest of grok_exporter just supports single files.
+// FromPath creates a Glob from a file path, so that we can use the new file
+// tailers but be sure only a single file is watched.
+func FromPath(path string) (Glob, error) {
+	if containsWildcards(path) {
+		return "", fmt.Errorf("%v: illegal file name", path)
+	}
+	return Parse(path)
+}
+
 func containsWildcards(pattern string) bool {
 	p := []rune(pattern)
 	escaped := false // p[i] is escaped by '\\'