Skip to content

Commit

Permalink
[chore][pkg/stanza] refactor: remove function juggling (#36108)
Browse files Browse the repository at this point in the history
Separates header processing functions from content processing functions
to improve code clarity.
This is a follow-up from
#35870 (comment).
  • Loading branch information
andrzej-stencel authored Oct 31, 2024
1 parent c031332 commit 568cbb6
Show file tree
Hide file tree
Showing 2 changed files with 8 additions and 18 deletions.
11 changes: 3 additions & 8 deletions pkg/stanza/fileconsumer/internal/reader/factory.go
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,6 @@ func (f *Factory) NewReaderFromMetadata(file *os.File, m *Metadata) (r *Reader,
initialBufferSize: f.InitialBufferSize,
maxLogSize: f.MaxLogSize,
decoder: decode.New(f.Encoding),
lineSplitFunc: f.SplitFunc,
deleteAtEOF: f.DeleteAtEOF,
includeFileRecordNum: f.IncludeFileRecordNumber,
compression: f.Compression,
Expand Down Expand Up @@ -103,18 +102,14 @@ func (f *Factory) NewReaderFromMetadata(file *os.File, m *Metadata) (r *Reader,
}

flushFunc := m.FlushState.Func(f.SplitFunc, f.FlushTimeout)
r.lineSplitFunc = trim.WithFunc(trim.ToLength(flushFunc, f.MaxLogSize), f.TrimFunc)
r.contentSplitFunc = trim.WithFunc(trim.ToLength(flushFunc, f.MaxLogSize), f.TrimFunc)
r.emitFunc = f.EmitFunc
if f.HeaderConfig == nil || m.HeaderFinalized {
r.splitFunc = r.lineSplitFunc
r.processFunc = r.emitFunc
} else {
if f.HeaderConfig != nil && !m.HeaderFinalized {
r.headerSplitFunc = f.HeaderConfig.SplitFunc
r.headerReader, err = header.NewReader(f.TelemetrySettings, *f.HeaderConfig)
if err != nil {
return nil, err
}
r.splitFunc = f.HeaderConfig.SplitFunc
r.processFunc = r.headerReader.Process
}

attributes, err := f.Attributes.Resolve(file)
Expand Down
15 changes: 5 additions & 10 deletions pkg/stanza/fileconsumer/internal/reader/reader.go
Original file line number Diff line number Diff line change
Expand Up @@ -42,11 +42,10 @@ type Reader struct {
fingerprintSize int
initialBufferSize int
maxLogSize int
lineSplitFunc bufio.SplitFunc
splitFunc bufio.SplitFunc
headerSplitFunc bufio.SplitFunc
contentSplitFunc bufio.SplitFunc
decoder *decode.Decoder
headerReader *header.Reader
processFunc emit.Callback
emitFunc emit.Callback
deleteAtEOF bool
needsUpdateFingerprint bool
Expand Down Expand Up @@ -116,7 +115,7 @@ func (r *Reader) ReadToEnd(ctx context.Context) {
}

func (r *Reader) readHeader(ctx context.Context) (doneReadingFile bool) {
s := scanner.New(r, r.maxLogSize, r.initialBufferSize, r.Offset, r.splitFunc)
s := scanner.New(r, r.maxLogSize, r.initialBufferSize, r.Offset, r.headerSplitFunc)

// Read the tokens from the file until no more header tokens are found or the end of file is reached.
for {
Expand Down Expand Up @@ -167,10 +166,6 @@ func (r *Reader) readHeader(ctx context.Context) (doneReadingFile bool) {
r.HeaderFinalized = true
r.initialBufferSize = scanner.DefaultBufferSize

// Switch to the normal split and process functions.
r.splitFunc = r.lineSplitFunc
r.processFunc = r.emitFunc

// 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))
Expand All @@ -182,7 +177,7 @@ func (r *Reader) readHeader(ctx context.Context) (doneReadingFile bool) {

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)
s := scanner.New(r, r.maxLogSize, r.initialBufferSize, r.Offset, r.contentSplitFunc)

// Iterate over the contents of the file.
for {
Expand Down Expand Up @@ -214,7 +209,7 @@ func (r *Reader) readContents(ctx context.Context) {
r.FileAttributes[attrs.LogFileRecordNumber] = r.RecordNum
}

err = r.processFunc(ctx, token, r.FileAttributes)
err = r.emitFunc(ctx, token, r.FileAttributes)
if err != nil {
r.set.Logger.Error("failed to process token", zap.Error(err))
}
Expand Down

0 comments on commit 568cbb6

Please sign in to comment.