Updating hpcloud tail library to forked version modified for our purp…

…oses.

Changes to the library move the handling of symlinks and re-opening to the tail library.
fsnotify events are only used now to start new tailers for files not previously seen.

Gopkg.toml

@@ -65,3 +65,8 @@
 [prune]
   go-tests = true
   unused-packages = true
+
+[[constraint]]
+  name = "github.com/hpcloud/tail"
+  source = "github.com/slim-bean/tail"
+  branch = "master"

pkg/promtail/promtail_test.go

@@ -180,9 +180,6 @@ func fileRoll(t *testing.T, filename string, prefix string) int {
 		time.Sleep(1 * time.Millisecond)
 	}
 
-	//FIXME this is a hack to make sure the hpcloud tail polling implementation reads all lines before we roll the file
-	time.Sleep(300 * time.Millisecond)
-
 	if err = os.Rename(filename, filename+".1"); err != nil {
 		t.Fatal("Failed to rename file for test: ", err)
 	}
@@ -199,9 +196,6 @@ func fileRoll(t *testing.T, filename string, prefix string) int {
 		time.Sleep(1 * time.Millisecond)
 	}
 
-	//FIXME this is a hack to make sure the hpcloud tail polling implementation reads all lines before we roll the file
-	time.Sleep(300 * time.Millisecond)
-
 	return 200
 }
 
@@ -231,9 +225,6 @@ func symlinkRoll(t *testing.T, testDir string, filename string, prefix string) i
 		time.Sleep(1 * time.Millisecond)
 	}
 
-	//FIXME this is a hack to make sure the hpcloud tail polling implementation reads all lines before we roll the file
-	time.Sleep(300 * time.Millisecond)
-
 	// Remove the link, make a new file, link to the new file.
 	if err := os.Remove(filename); err != nil {
 		t.Fatal(err)
@@ -255,9 +246,6 @@ func symlinkRoll(t *testing.T, testDir string, filename string, prefix string) i
 		time.Sleep(1 * time.Millisecond)
 	}
 
-	//FIXME this is a hack to make sure the hpcloud tail polling implementation reads all lines before we roll the file
-	time.Sleep(300 * time.Millisecond)
-
 	return 200
 
 }

pkg/promtail/targets/filetarget.go

@@ -135,13 +135,6 @@ func (t *FileTarget) run() {
 		case event := <-t.watcher.Events:
 			switch event.Op {
 			case fsnotify.Create:
-				// If the file was a symlink we don't get a Remove notification if the symlink resolves to a non watched directory.
-				// Close and re-open the tailer to make sure we tail the new file.
-				if tailer, ok := t.tails[event.Name]; ok {
-					level.Info(t.logger).Log("msg", "create for file being tailed. Will close and re-open", "filename", event.Name)
-					helpers.LogError("stopping tailer", tailer.stop)
-					delete(t.tails, event.Name)
-				}
 				matched, err := filepath.Match(t.path, event.Name)
 				if err != nil {
 					level.Error(t.logger).Log("msg", "failed to match file", "error", err, "filename", event.Name)
@@ -152,11 +145,6 @@ func (t *FileTarget) run() {
 					continue
 				}
 				t.startTailing([]string{event.Name})
-			case fsnotify.Remove:
-				t.stopTailing([]string{event.Name})
-			case fsnotify.Rename:
-				// Rename is only issued on the original file path; the new name receives a Create event
-				t.stopTailing([]string{event.Name})
 			default:
 				level.Debug(t.logger).Log("msg", "got unknown event", "event", event)
 			}
@@ -188,7 +176,7 @@ func (t *FileTarget) sync() error {
 	}
 
 	// Record the size of all the files matched by the Glob pattern.
-	t.updateTotalBytesMetric(matches)
+	matches = t.reportSizeAndRemoveMissing(matches)
 
 	// Get the current unique set of dirs to watch.
 	dirs := map[string]struct{}{}
@@ -298,15 +286,18 @@ func toStopTailing(nt []string, et map[string]*tailer) []string {
 	return ta
 }
 
-func (t *FileTarget) updateTotalBytesMetric(ms []string) {
+func (t *FileTarget) reportSizeAndRemoveMissing(ms []string) []string {
+	mso := ms[:0]
 	for _, m := range ms {
 		fi, err := os.Stat(m)
 		if err != nil {
-			level.Error(t.logger).Log("msg", "failed to stat matched file, cannot report size", m, "error", err)
+			//If we can't stat the file, skip it
 			continue
 		}
+		mso = append(mso, m)
 		totalBytes.WithLabelValues(m).Set(float64(fi.Size()))
 	}
+	return mso
 }
 
 // Returns the elements from set b which are missing from set a

pkg/promtail/targets/tailer.go

@@ -18,50 +18,30 @@ type tailer struct {
 	handler   api.EntryHandler
 	positions *positions.Positions
 
-	path     string
-	filename string
-	tail     *tail.Tail
+	path string
+	tail *tail.Tail
 
 	quit chan struct{}
 	done chan struct{}
 }
 
 func newTailer(logger log.Logger, handler api.EntryHandler, positions *positions.Positions, path string) (*tailer, error) {
-	filename := path
-	var reOpen bool
-
-	// Check if the path requested is a symbolic link
-	fi, err := os.Lstat(path)
-	if err != nil {
-		return nil, err
-	}
-	if fi.Mode()&os.ModeSymlink == os.ModeSymlink {
-		filename, err = os.Readlink(path)
-		if err != nil {
-			return nil, err
-		}
-
-		// if we are tailing a symbolic link then we need to automatically re-open
-		// as we wont get a Create event when a file is rotated.
-		reOpen = true
-	}
-
 	// Simple check to make sure the file we are tailing doesn't
 	// have a position already saved which is past the end of the file.
-	fi, err = os.Stat(filename)
+	fi, err := os.Stat(path)
 	if err != nil {
 		return nil, err
 	}
-	if fi.Size() < positions.Get(filename) {
-		positions.Remove(filename)
+	if fi.Size() < positions.Get(path) {
+		positions.Remove(path)
 	}
 
-	tail, err := tail.TailFile(filename, tail.Config{
+	tail, err := tail.TailFile(path, tail.Config{
 		Follow: true,
 		Poll:   true,
-		ReOpen: reOpen,
+		ReOpen: true,
 		Location: &tail.SeekInfo{
-			Offset: positions.Get(filename),
+			Offset: positions.Get(path),
 			Whence: 0,
 		},
 	})
@@ -74,11 +54,10 @@ func newTailer(logger log.Logger, handler api.EntryHandler, positions *positions
 		handler:   api.AddLabelsMiddleware(model.LabelSet{filenameLabel: model.LabelValue(path)}).Wrap(handler),
 		positions: positions,
 
-		path:     path,
-		filename: filename,
-		tail:     tail,
-		quit:     make(chan struct{}),
-		done:     make(chan struct{}),
+		path: path,
+		tail: tail,
+		quit: make(chan struct{}),
+		done: make(chan struct{}),
 	}
 	go tailer.run()
 	filesActive.Add(1.)
@@ -131,8 +110,8 @@ func (t *tailer) markPosition() error {
 	}
 
 	readBytes.WithLabelValues(t.path).Set(float64(pos))
-	level.Debug(t.logger).Log("path", t.path, "filename", t.filename, "current_position", pos)
-	t.positions.Put(t.filename, pos)
+	level.Debug(t.logger).Log("path", t.path, "current_position", pos)
+	t.positions.Put(t.path, pos)
 	return nil
 }
 
@@ -151,5 +130,5 @@ func (t *tailer) stop() error {
 }
 
 func (t *tailer) cleanup() {
-	t.positions.Remove(t.filename)
+	t.positions.Remove(t.path)
 }