feat(input.tail): Add initial_read_offset config for controlling read behavior

plugins/inputs/tail/README.md

@@ -12,7 +12,7 @@ tail -F --lines=0 myfile.log
 that it will be compatible with log-rotated files, and that it will retry on
 inaccessible files.
 - `--lines=0` means that it will start at the end of the file (unless
-the `from_beginning` option is set).
+the `initial_read_offset` option is set).
 
 see <http://man7.org/linux/man-pages/man1/tail.1.html> for more details.
 
@@ -56,8 +56,13 @@ See the [CONFIGURATION.md][CONFIGURATION.md] for more details.
   ##
   files = ["/var/mymetrics.out"]
 
-  ## Read file from beginning.
-  # from_beginning = false
+  ## Offset to start reading at
+  ## The following methods are available:
+  ##   beginning          -- start reading from the beginning of the file ignoring any persisted offset
+  ##   end                -- start reading from the end of the file ignoring any persisted offset
+  ##   saved-or-beginning --use the persisted offset of the file or, if no offset persisted, start from the beginning  of the file
+  ##   saved-or-end       -- use the persisted offset of the file or, if no offset persisted, start from the end of the file
+  # initial_read_offset = "saved-or-end"
 
   ## Whether file is a named pipe
   # pipe = false

plugins/inputs/tail/sample.conf

@@ -12,8 +12,13 @@
   ##
   files = ["/var/mymetrics.out"]
 
-  ## Read file from beginning.
-  # from_beginning = false
+  ## Offset to start reading at
+  ## The following methods are available:
+  ##   beginning          -- start reading from the beginning of the file ignoring any persisted offset
+  ##   end                -- start reading from the end of the file ignoring any persisted offset
+  ##   saved-or-beginning --use the persisted offset of the file or, if no offset persisted, start from the beginning  of the file
+  ##   saved-or-end       -- use the persisted offset of the file or, if no offset persisted, start from the end of the file
+  # initial_read_offset = "saved-or-end"
 
   ## Whether file is a named pipe
   # pipe = false

plugins/inputs/tail/tail.go

@@ -40,7 +40,8 @@ type semaphore chan empty
 
 type Tail struct {
 	Files               []string `toml:"files"`
-	FromBeginning       bool     `toml:"from_beginning"`
+	FromBeginning       bool     `toml:"from_beginning" deprecated:"1.33.0;1.40.0;use 'initial_read_offset' with value 'beginning' instead"`
+	InitialReadOffset   string   `toml:"initial_read_offset"`
 	Pipe                bool     `toml:"pipe"`
 	WatchMethod         string   `toml:"watch_method"`
 	MaxUndeliveredLines int      `toml:"max_undelivered_lines"`
@@ -100,6 +101,9 @@ func (t *Tail) Init() error {
 	}
 	// init offsets
 	t.offsets = make(map[string]int64)
+	if t.FromBeginning && t.InitialReadOffset == "" {
+		t.InitialReadOffset = "beginning"
+	}
 
 	var err error
 	t.decoder, err = encoding.NewDecoder(t.CharacterEncoding)
@@ -122,7 +126,7 @@ func (t *Tail) SetState(state interface{}) error {
 }
 
 func (t *Tail) Gather(_ telegraf.Accumulator) error {
-	return t.tailNewFiles(true)
+	return t.tailNewFiles()
 }
 
 func (t *Tail) Start(acc telegraf.Accumulator) error {
@@ -152,7 +156,10 @@ func (t *Tail) Start(acc telegraf.Accumulator) error {
 
 	t.tailers = make(map[string]*tail.Tail)
 
-	err = t.tailNewFiles(t.FromBeginning)
+	err = t.tailNewFiles()
+	if err != nil {
+		return err
+	}
 
 	// assumption that once Start is called, all parallel plugins have already been initialized
 	offsetsMutex.Lock()
@@ -162,7 +169,32 @@ func (t *Tail) Start(acc telegraf.Accumulator) error {
 	return err
 }
 
-func (t *Tail) tailNewFiles(fromBeginning bool) error {
+func (t *Tail) getSeekInfo(file string) (*tail.SeekInfo, error) {
+	switch t.InitialReadOffset {
+	case "beginning":
+		return &tail.SeekInfo{Whence: 0, Offset: 0}, nil
+	case "end":
+		return &tail.SeekInfo{Whence: 2, Offset: 0}, nil
+	case "", "save-or-end":
+		if offset, ok := t.offsets[file]; ok {
+			t.Log.Debugf("Using offset %d for %q", offset, file)
+			return &tail.SeekInfo{Whence: 0, Offset: offset}, nil
+		} else {
+			return &tail.SeekInfo{Whence: 2, Offset: 0}, nil
+		}
+	case "save-or-beginning":
+		if offset, ok := t.offsets[file]; ok {
+			t.Log.Debugf("Using offset %d for %q", offset, file)
+			return &tail.SeekInfo{Whence: 0, Offset: offset}, nil
+		} else {
+			return &tail.SeekInfo{Whence: 0, Offset: 0}, nil
+		}
+	default:
+		return nil, errors.New("invalid 'initial_read_offset' setting")
+	}
+}
+
+func (t *Tail) tailNewFiles() error {
 	var poll bool
 	if t.WatchMethod == "poll" {
 		poll = true
@@ -180,20 +212,9 @@ func (t *Tail) tailNewFiles(fromBeginning bool) error {
 				continue
 			}
 
-			var seek *tail.SeekInfo
-			if !t.Pipe && !fromBeginning {
-				if offset, ok := t.offsets[file]; ok {
-					t.Log.Debugf("Using offset %d for %q", offset, file)
-					seek = &tail.SeekInfo{
-						Whence: 0,
-						Offset: offset,
-					}
-				} else {
-					seek = &tail.SeekInfo{
-						Whence: 2,
-						Offset: 0,
-					}
-				}
+			seek, err := t.getSeekInfo(file)
+			if err != nil {
+				return err
 			}
 
 			tailer, err := tail.TailFile(file,
@@ -379,7 +400,7 @@ func (t *Tail) receiver(parser telegraf.Parser, tailer *tail.Tail) {
 
 func (t *Tail) Stop() {
 	for _, tailer := range t.tailers {
-		if !t.Pipe && !t.FromBeginning {
+		if !t.Pipe {
 			// store offset for resume
 			offset, err := tailer.Tell()
 			if err == nil {