Add limit to number of undelivered lines to read ahead in tail (#7210)

plugins/inputs/tail/tail.go

@@ -3,6 +3,8 @@
 package tail
 
 import (
+	"context"
+	"errors"
 	"strings"
 	"sync"
 
@@ -15,7 +17,8 @@ import (
 )
 
 const (
-	defaultWatchMethod = "inotify"
+	defaultWatchMethod         = "inotify"
+	defaultMaxUndeliveredLines = 1000
 )
 
 var (
@@ -23,21 +26,25 @@ var (
 	offsetsMutex = new(sync.Mutex)
 )
 
+type empty struct{}
+type semaphore chan empty
+
 type Tail struct {
-	Files         []string
-	FromBeginning bool
-	Pipe          bool
-	WatchMethod   string
-
-	Log telegraf.Logger
+	Files               []string `toml:"files"`
+	FromBeginning       bool     `toml:"from_beginning"`
+	Pipe                bool     `toml:"pipe"`
+	WatchMethod         string   `toml:"watch_method"`
+	MaxUndeliveredLines int      `toml:"max_undelivered_lines"`
 
+	Log        telegraf.Logger `toml:"-"`
 	tailers    map[string]*tail.Tail
 	offsets    map[string]int64
 	parserFunc parsers.ParserFunc
 	wg         sync.WaitGroup
-	acc        telegraf.Accumulator
-
-	sync.Mutex
+	ctx        context.Context
+	cancel     context.CancelFunc
+	acc        telegraf.TrackingAccumulator
+	sem        semaphore
 }
 
 func NewTail() *Tail {
@@ -49,13 +56,14 @@ func NewTail() *Tail {
 	offsetsMutex.Unlock()
 
 	return &Tail{
-		FromBeginning: false,
-		offsets:       offsetsCopy,
+		FromBeginning:       false,
+		MaxUndeliveredLines: 1000,
+		offsets:             offsetsCopy,
 	}
 }
 
 const sampleConfig = `
-  ## files to tail.
+  ## File names or a pattern to tail.
   ## These accept standard unix glob matching rules, but with the addition of
   ## ** as a "super asterisk". ie:
   ##   "/var/log/**.log"  -> recursively find all .log files in /var/log
@@ -65,14 +73,21 @@ const sampleConfig = `
   ## See https://github.com/gobwas/glob for more examples
   ##
   files = ["/var/mymetrics.out"]
+
   ## Read file from beginning.
-  from_beginning = false
+  # from_beginning = false
+
   ## Whether file is a named pipe
-  pipe = false
+  # pipe = false
 
   ## Method used to watch for file updates.  Can be either "inotify" or "poll".
   # watch_method = "inotify"
 
+  ## Maximum lines of the file to process that have not yet be written by the
+  ## output.  For best throughput set based on the number of metrics on each
+  ## line and the size of the output's metric_batch_size.
+  # max_undelivered_lines = 1000
+
   ## Data format to consume.
   ## Each data format has its own unique set of configuration options, read
   ## more about them here:
@@ -88,18 +103,36 @@ func (t *Tail) Description() string {
 	return "Stream a log file, like the tail -f command"
 }
 
-func (t *Tail) Gather(acc telegraf.Accumulator) error {
-	t.Lock()
-	defer t.Unlock()
+func (t *Tail) Init() error {
+	if t.MaxUndeliveredLines == 0 {
+		return errors.New("max_undelivered_lines must be positive")
+	}
+	t.sem = make(semaphore, t.MaxUndeliveredLines)
+	return nil
+}
 
+func (t *Tail) Gather(acc telegraf.Accumulator) error {
 	return t.tailNewFiles(true)
 }
 
 func (t *Tail) Start(acc telegraf.Accumulator) error {
-	t.Lock()
-	defer t.Unlock()
+	t.acc = acc.WithTracking(t.MaxUndeliveredLines)
+
+	t.ctx, t.cancel = context.WithCancel(context.Background())
+
+	t.wg.Add(1)
+	go func() {
+		defer t.wg.Done()
+		for {
+			select {
+			case <-t.ctx.Done():
+				return
+			case <-t.acc.Delivered():
+				<-t.sem
+			}
+		}
+	}()
 
-	t.acc = acc
 	t.tailers = make(map[string]*tail.Tail)
 
 	err := t.tailNewFiles(t.FromBeginning)
@@ -175,6 +208,12 @@ func (t *Tail) tailNewFiles(fromBeginning bool) error {
 			go func() {
 				defer t.wg.Done()
 				t.receiver(parser, tailer)
+
+				t.Log.Debugf("Tail removed for %q", tailer.Filename)
+
+				if err := tailer.Err(); err != nil {
+					t.Log.Errorf("Tailing %q: %s", tailer.Filename, err.Error())
+				}
 			}()
 			t.tailers[tailer.Filename] = tailer
 		}
@@ -229,21 +268,19 @@ func (t *Tail) receiver(parser parsers.Parser, tailer *tail.Tail) {
 
 		for _, metric := range metrics {
 			metric.AddTag("path", tailer.Filename)
-			t.acc.AddMetric(metric)
 		}
-	}
 
-	t.Log.Debugf("Tail removed for %q", tailer.Filename)
-
-	if err := tailer.Err(); err != nil {
-		t.Log.Errorf("Tailing %q: %s", tailer.Filename, err.Error())
+		// Block until plugin is stopping or room is available to add metrics.
+		select {
+		case <-t.ctx.Done():
+			return
+		case t.sem <- empty{}:
+			t.acc.AddTrackingMetricGroup(metrics)
+		}
 	}
 }
 
 func (t *Tail) Stop() {
-	t.Lock()
-	defer t.Unlock()
-
 	for _, tailer := range t.tailers {
 		if !t.Pipe && !t.FromBeginning {
 			// store offset for resume
@@ -260,6 +297,7 @@ func (t *Tail) Stop() {
 		}
 	}
 
+	t.cancel()
 	t.wg.Wait()
 
 	// persist offsets