fix: fix record counting with header

This refactors the `Reader::ReadToEnd` method
by separating reading the file's header from reading the file's contents.
This results in very similar code in `readHeader` and `readContents` methods,
which was previosly deduplicated at the cost of slightly higher complexity.
The bug could be fixed without separating header reading from contents reading,
but I hope this separation will make it easier to implement content batching
in the Reader (open-telemetry#35455).
Content batching was my original motivation for these code changes.
I only discovered the problem with record counting when reading the code.

Author: andrzej-stencel

pkg/stanza/fileconsumer/file_test.go

@@ -1643,10 +1643,8 @@ func TestIncludeFileRecordNumberWithHeaderConfigured(t *testing.T) {
 	}()
 
 	sink.ExpectCall(t, []byte("testlog1"), map[string]any{
-		attrs.LogFileName: filepath.Base(temp.Name()),
-		// The record number should be 1, but it's actually 4 = 1 + 2 header lines + 1
-		// https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/35869
-		attrs.LogFileRecordNumber: int64(4),
+		attrs.LogFileName:         filepath.Base(temp.Name()),
+		attrs.LogFileRecordNumber: int64(1),
 		"header_attr":             "xyz",
 	})
 }
@@ -1671,9 +1669,7 @@ func TestIncludeFileRecordNumberWithHeaderConfiguredButMissing(t *testing.T) {
 	}()
 
 	sink.ExpectCall(t, []byte("testlog1"), map[string]any{
-		attrs.LogFileName: filepath.Base(temp.Name()),
-		// The record number should be 1, but it's actually 2 = 1 + 1
-		// https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/35869
-		attrs.LogFileRecordNumber: int64(2),
+		attrs.LogFileName:         filepath.Base(temp.Name()),
+		attrs.LogFileRecordNumber: int64(1),
 	})
 }

pkg/stanza/fileconsumer/internal/reader/reader.go

@@ -70,7 +70,7 @@ func (r *Reader) ReadToEnd(ctx context.Context) {
 		// SectionReader can only read a fixed window (from previous offset to EOF).
 		info, err := r.file.Stat()
 		if err != nil {
-			r.set.Logger.Error("Failed to stat", zap.Error(err))
+			r.set.Logger.Error("failed to stat", zap.Error(err))
 			return
 		}
 		currentEOF := info.Size()
@@ -80,7 +80,7 @@ func (r *Reader) ReadToEnd(ctx context.Context) {
 		gzipReader, err := gzip.NewReader(io.NewSectionReader(r.file, r.Offset, currentEOF))
 		if err != nil {
 			if !errors.Is(err, io.EOF) {
-				r.set.Logger.Error("Failed to create gzip reader", zap.Error(err))
+				r.set.Logger.Error("failed to create gzip reader", zap.Error(err))
 			}
 			return
 		} else {
@@ -96,7 +96,7 @@ func (r *Reader) ReadToEnd(ctx context.Context) {
 	}
 
 	if _, err := r.file.Seek(r.Offset, 0); err != nil {
-		r.set.Logger.Error("Failed to seek", zap.Error(err))
+		r.set.Logger.Error("failed to seek", zap.Error(err))
 		return
 	}
 
@@ -106,9 +106,90 @@ func (r *Reader) ReadToEnd(ctx context.Context) {
 		}
 	}()
 
+	doneReadingFile := r.readHeader(ctx)
+	if doneReadingFile {
+		return
+	}
+
+	// Reset position in file to r.Offest after the header scanner might have moved it past a content token.
+	if _, err := r.file.Seek(r.Offset, 0); err != nil {
+		r.set.Logger.Error("failed to seek post-header", zap.Error(err))
+		return
+	}
+
+	r.readContents(ctx)
+}
+
+func (r *Reader) readHeader(ctx context.Context) (doneReadingFile bool) {
+	if r.headerReader == nil {
+		r.set.Logger.Debug("no need to read header", zap.Bool("header_finalized", r.HeaderFinalized))
+		return false
+	}
+
 	s := scanner.New(r, r.maxLogSize, r.initialBufferSize, r.Offset, r.splitFunc)
 
-	// Iterate over the tokenized file, emitting entries as we go
+	// Read the tokens from the file until no more header tokens are found or the end of file is reached.
+	for {
+		select {
+		case <-ctx.Done():
+			return true
+		default:
+		}
+
+		ok := s.Scan()
+		if !ok {
+			if err := s.Error(); err != nil {
+				r.set.Logger.Error("failed during header scan", zap.Error(err))
+			} else {
+				r.set.Logger.Debug("end of file reached", zap.Bool("delete_at_eof", r.deleteAtEOF))
+				if r.deleteAtEOF {
+					r.delete()
+				}
+			}
+			// Either end of file was reached, or file cannot be scanned.
+			return true
+		}
+
+		token, err := r.decoder.Decode(s.Bytes())
+		if err != nil {
+			r.set.Logger.Error("decode header: %w", zap.Error(err))
+			r.Offset = s.Pos() // move past the bad token or we may be stuck
+			continue
+		}
+
+		err = r.headerReader.Process(ctx, token, r.FileAttributes)
+		if err != nil {
+			if errors.Is(err, header.ErrEndOfHeader) {
+				// End of header reached.
+				break
+			} else {
+				r.set.Logger.Error("process header: %w", zap.Error(err))
+			}
+		}
+
+		r.Offset = s.Pos()
+	}
+
+	// Clean up the header machinery
+	if err := r.headerReader.Stop(); err != nil {
+		r.set.Logger.Error("failed to stop header pipeline during finalization", zap.Error(err))
+	}
+	r.headerReader = nil
+	r.HeaderFinalized = true
+	r.initialBufferSize = scanner.DefaultBufferSize
+
+	// Switch to the normal split and process functions.
+	r.splitFunc = r.lineSplitFunc
+	r.processFunc = r.emitFunc
+
+	return false
+}
+
+func (r *Reader) readContents(ctx context.Context) {
+	// Create the scanner to read the contents of the file.
+	s := scanner.New(r, r.maxLogSize, r.initialBufferSize, r.Offset, r.splitFunc)
+
+	// Iterate over the contents of the file.
 	for {
 		select {
 		case <-ctx.Done():
@@ -119,7 +200,7 @@ func (r *Reader) ReadToEnd(ctx context.Context) {
 		ok := s.Scan()
 		if !ok {
 			if err := s.Error(); err != nil {
-				r.set.Logger.Error("Failed during scan", zap.Error(err))
+				r.set.Logger.Error("failed during scan", zap.Error(err))
 			} else if r.deleteAtEOF {
 				r.delete()
 			}
@@ -139,36 +220,11 @@ func (r *Reader) ReadToEnd(ctx context.Context) {
 		}
 
 		err = r.processFunc(ctx, token, r.FileAttributes)
-		if err == nil {
-			r.Offset = s.Pos() // successful emit, update offset
-			continue
-		}
-
-		if !errors.Is(err, header.ErrEndOfHeader) {
+		if err != nil {
 			r.set.Logger.Error("process: %w", zap.Error(err))
-			r.Offset = s.Pos() // move past the bad token or we may be stuck
-			continue
 		}
 
-		// Clean up the header machinery
-		if err = r.headerReader.Stop(); err != nil {
-			r.set.Logger.Error("Failed to stop header pipeline during finalization", zap.Error(err))
-		}
-		r.headerReader = nil
-		r.HeaderFinalized = true
-
-		// Switch to the normal split and process functions.
-		r.splitFunc = r.lineSplitFunc
-		r.processFunc = r.emitFunc
-
-		// Recreate the scanner with the normal split func.
-		// Do not use the updated offset from the old scanner, as the most recent token
-		// could be split differently with the new splitter.
-		if _, err = r.file.Seek(r.Offset, 0); err != nil {
-			r.set.Logger.Error("Failed to seek post-header", zap.Error(err))
-			return
-		}
-		s = scanner.New(r, r.maxLogSize, scanner.DefaultBufferSize, r.Offset, r.splitFunc)
+		r.Offset = s.Pos()
 	}
 }