Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[chore][pkg/stanza] refactor: remove function juggling #36108

Merged
merged 2 commits into from
Oct 31, 2024
Merged

[chore][pkg/stanza] refactor: remove function juggling #36108

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0ec9afb
refactor: remove function juggling
andrzej-stencel Oct 31, 2024
b480ac7
Merge branch 'main' into remove-function-juggling
andrzej-stencel Oct 31, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 3 additions & 8 deletions pkg/stanza/fileconsumer/internal/reader/factory.go
View file
Open in desktop
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,
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I have removed this line, as lineSplitFunc is being reassigned below in line 106.

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
View file
Open in desktop
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
Loading