From 7db7591edf059dd5ed682ea44b9f8a22383023f8 Mon Sep 17 00:00:00 2001
From: Nick White <nwhite@palantir.com>
Date: Thu, 8 Dec 2016 16:32:08 +0000
Subject: [PATCH] Make Logparser Plugin Check For New Files

Check in the Gather metric to see if any new files matching the glob
have appeared. If so, start tailing them from the beginning.
---
 plugins/inputs/logparser/logparser.go      | 36 ++++++++++++++----
 plugins/inputs/logparser/logparser_test.go | 43 ++++++++++++++++++++++
 2 files changed, 72 insertions(+), 7 deletions(-)

diff --git a/plugins/inputs/logparser/logparser.go b/plugins/inputs/logparser/logparser.go
index 0778a8a6d7ab2..8ec32835829a1 100644
--- a/plugins/inputs/logparser/logparser.go
+++ b/plugins/inputs/logparser/logparser.go
@@ -26,7 +26,7 @@ type LogParserPlugin struct {
 	Files         []string
 	FromBeginning bool
 
-	tailers []*tail.Tail
+	tailers map[string]*tail.Tail
 	lines   chan string
 	done    chan struct{}
 	wg      sync.WaitGroup
@@ -46,7 +46,9 @@ const sampleConfig = `
   ##   /var/log/*/*.log    -> find all .log files with a parent dir in /var/log
   ##   /var/log/apache.log -> only tail the apache log file
   files = ["/var/log/apache/access.log"]
-  ## Read file from beginning.
+  ## Read files that currently exist from the beginning. Files that are created
+  ## while telegraf is running (and that match the "files" globs) will always
+  ## be read from the beginning.
   from_beginning = false
 
   ## Parse logstash-style "grok" patterns:
@@ -77,7 +79,11 @@ func (l *LogParserPlugin) Description() string {
 }
 
 func (l *LogParserPlugin) Gather(acc telegraf.Accumulator) error {
-	return nil
+	l.Lock()
+	defer l.Unlock()
+
+	// always start from the beginning of files that appear while we're running
+	return l.tailNewfiles(true)
 }
 
 func (l *LogParserPlugin) Start(acc telegraf.Accumulator) error {
@@ -87,6 +93,7 @@ func (l *LogParserPlugin) Start(acc telegraf.Accumulator) error {
 	l.acc = acc
 	l.lines = make(chan string, 1000)
 	l.done = make(chan struct{})
+	l.tailers = make(map[string]*tail.Tail)
 
 	// Looks for fields which implement LogParser interface
 	l.parsers = []LogParser{}
@@ -121,14 +128,22 @@ func (l *LogParserPlugin) Start(acc telegraf.Accumulator) error {
 		return err
 	}
 
+	l.wg.Add(1)
+	go l.parser()
+
+	return l.tailNewfiles(l.FromBeginning)
+}
+
+// check the globs against files on disk, and start tailing any new files.
+// Assumes l's lock is held!
+func (l *LogParserPlugin) tailNewfiles(fromBeginning bool) error {
 	var seek tail.SeekInfo
-	if !l.FromBeginning {
+	if !fromBeginning {
 		seek.Whence = 2
 		seek.Offset = 0
 	}
 
-	l.wg.Add(1)
-	go l.parser()
+	errChan := errchan.New(len(l.Files))
 
 	// Create a "tailer" for each file
 	for _, filepath := range l.Files {
@@ -139,7 +154,13 @@ func (l *LogParserPlugin) Start(acc telegraf.Accumulator) error {
 		}
 		files := g.Match()
 		errChan = errchan.New(len(files))
+
 		for file, _ := range files {
+			if _, ok := l.tailers[file]; ok {
+				// we're already tailing this file
+				continue
+			}
+
 			tailer, err := tail.TailFile(file,
 				tail.Config{
 					ReOpen:    true,
@@ -152,7 +173,7 @@ func (l *LogParserPlugin) Start(acc telegraf.Accumulator) error {
 			// create a goroutine for each "tailer"
 			l.wg.Add(1)
 			go l.receiver(tailer)
-			l.tailers = append(l.tailers, tailer)
+			l.tailers[file] = tailer
 		}
 	}
 
@@ -166,6 +187,7 @@ func (l *LogParserPlugin) receiver(tailer *tail.Tail) {
 
 	var line *tail.Line
 	for line = range tailer.Lines {
+
 		if line.Err != nil {
 			log.Printf("E! Error tailing file %s, Error: %s\n",
 				tailer.Filename, line.Err)
diff --git a/plugins/inputs/logparser/logparser_test.go b/plugins/inputs/logparser/logparser_test.go
index 97f33067e0b22..059bfd266ac92 100644
--- a/plugins/inputs/logparser/logparser_test.go
+++ b/plugins/inputs/logparser/logparser_test.go
@@ -1,6 +1,8 @@
 package logparser
 
 import (
+	"io/ioutil"
+	"os"
 	"runtime"
 	"strings"
 	"testing"
@@ -80,6 +82,47 @@ func TestGrokParseLogFiles(t *testing.T) {
 		map[string]string{})
 }
 
+func TestGrokParseLogFilesAppearLater(t *testing.T) {
+	emptydir, err := ioutil.TempDir("", "TestGrokParseLogFilesAppearLater")
+	defer os.RemoveAll(emptydir)
+	assert.NoError(t, err)
+
+	thisdir := getCurrentDir()
+	p := &grok.Parser{
+		Patterns:           []string{"%{TEST_LOG_A}", "%{TEST_LOG_B}"},
+		CustomPatternFiles: []string{thisdir + "grok/testdata/test-patterns"},
+	}
+
+	logparser := &LogParserPlugin{
+		FromBeginning: true,
+		Files:         []string{emptydir + "/*.log"},
+		GrokParser:    p,
+	}
+
+	acc := testutil.Accumulator{}
+	assert.NoError(t, logparser.Start(&acc))
+
+	time.Sleep(time.Millisecond * 500)
+	assert.Equal(t, acc.NFields(), 0)
+
+	os.Symlink(
+		thisdir+"grok/testdata/test_a.log",
+		emptydir+"/test_a.log")
+	assert.NoError(t, logparser.Gather(&acc))
+	time.Sleep(time.Millisecond * 500)
+
+	logparser.Stop()
+
+	acc.AssertContainsTaggedFields(t, "logparser_grok",
+		map[string]interface{}{
+			"clientip":      "192.168.1.1",
+			"myfloat":       float64(1.25),
+			"response_time": int64(5432),
+			"myint":         int64(101),
+		},
+		map[string]string{"response_code": "200"})
+}
+
 // Test that test_a.log line gets parsed even though we don't have the correct
 // pattern available for test_b.log
 func TestGrokParseLogFilesOneBad(t *testing.T) {