include file record num in reader config

sfc-gh-jikim · sfc-gh-jikim · commit b6c56e443851 · 2024-06-14T09:40:54.000-07:00
diff --git a/pkg/stanza/fileconsumer/attrs/attrs.go b/pkg/stanza/fileconsumer/attrs/attrs.go
@@ -27,7 +27,6 @@ type Resolver struct {
 	IncludeFilePathResolved   bool `mapstructure:"include_file_path_resolved,omitempty"`
 	IncludeFileOwnerName      bool `mapstructure:"include_file_owner_name,omitempty"`
 	IncludeFileOwnerGroupName bool `mapstructure:"include_file_owner_group_name,omitempty"`
-	IncludeFileRecordNumber   bool `mapstructure:"include_file_record_number,omitempty"`
 }
 
 func (r *Resolver) Resolve(file *os.File) (attributes map[string]any, err error) {
@@ -46,10 +45,6 @@ func (r *Resolver) Resolve(file *os.File) (attributes map[string]any, err error)
 			return nil, err
 		}
 	}
-	if r.IncludeFileRecordNumber {
-		// non-zero value to flag for setting
-		attributes[LogFileRecordNumber] = 1
-	}
 	if !r.IncludeFileNameResolved && !r.IncludeFilePathResolved {
 		return attributes, nil
 	}
diff --git a/pkg/stanza/fileconsumer/attrs/attrs_test.go b/pkg/stanza/fileconsumer/attrs/attrs_test.go
@@ -17,10 +17,10 @@ import (
 func TestResolver(t *testing.T) {
 	t.Parallel()
 
-	for i := 0; i < 128; i++ {
+	for i := 0; i < 64; i++ {
 
-		// Create a 7 bit string where each bit represents the value of a config option
-		bitString := fmt.Sprintf("%07b", i)
+		// Create a 6 bit string where each bit represents the value of a config option
+		bitString := fmt.Sprintf("%06b", i)
 
 		// Create a resolver with a config that matches the bit pattern of i
 		r := Resolver{
@@ -30,7 +30,6 @@ func TestResolver(t *testing.T) {
 			IncludeFilePathResolved:   bitString[3] == '1',
 			IncludeFileOwnerName:      bitString[4] == '1' && runtime.GOOS != "windows",
 			IncludeFileOwnerGroupName: bitString[5] == '1' && runtime.GOOS != "windows",
-			IncludeFileRecordNumber:   bitString[6] == '1',
 		}
 
 		t.Run(bitString, func(t *testing.T) {
@@ -54,12 +53,6 @@ func TestResolver(t *testing.T) {
 			} else {
 				assert.Empty(t, attributes[LogFilePath])
 			}
-			if r.IncludeFileRecordNumber {
-				expectLen++
-				assert.Equal(t, 1, attributes[LogFileRecordNumber])
-			} else {
-				assert.Empty(t, attributes[LogFileRecordNumber])
-			}
 
 			// We don't have an independent way to resolve the path, so the only meaningful validate
 			// is to ensure that the resolver returns nothing vs something based on the config.
diff --git a/pkg/stanza/fileconsumer/config.go b/pkg/stanza/fileconsumer/config.go
@@ -71,20 +71,21 @@ func NewConfig() *Config {
 
 // Config is the configuration of a file input operator
 type Config struct {
-	matcher.Criteria   `mapstructure:",squash"`
-	attrs.Resolver     `mapstructure:",squash"`
-	PollInterval       time.Duration   `mapstructure:"poll_interval,omitempty"`
-	MaxConcurrentFiles int             `mapstructure:"max_concurrent_files,omitempty"`
-	MaxBatches         int             `mapstructure:"max_batches,omitempty"`
-	StartAt            string          `mapstructure:"start_at,omitempty"`
-	FingerprintSize    helper.ByteSize `mapstructure:"fingerprint_size,omitempty"`
-	MaxLogSize         helper.ByteSize `mapstructure:"max_log_size,omitempty"`
-	Encoding           string          `mapstructure:"encoding,omitempty"`
-	SplitConfig        split.Config    `mapstructure:"multiline,omitempty"`
-	TrimConfig         trim.Config     `mapstructure:",squash,omitempty"`
-	FlushPeriod        time.Duration   `mapstructure:"force_flush_period,omitempty"`
-	Header             *HeaderConfig   `mapstructure:"header,omitempty"`
-	DeleteAfterRead    bool            `mapstructure:"delete_after_read,omitempty"`
+	matcher.Criteria        `mapstructure:",squash"`
+	attrs.Resolver          `mapstructure:",squash"`
+	PollInterval            time.Duration   `mapstructure:"poll_interval,omitempty"`
+	MaxConcurrentFiles      int             `mapstructure:"max_concurrent_files,omitempty"`
+	MaxBatches              int             `mapstructure:"max_batches,omitempty"`
+	StartAt                 string          `mapstructure:"start_at,omitempty"`
+	FingerprintSize         helper.ByteSize `mapstructure:"fingerprint_size,omitempty"`
+	MaxLogSize              helper.ByteSize `mapstructure:"max_log_size,omitempty"`
+	Encoding                string          `mapstructure:"encoding,omitempty"`
+	SplitConfig             split.Config    `mapstructure:"multiline,omitempty"`
+	TrimConfig              trim.Config     `mapstructure:",squash,omitempty"`
+	FlushPeriod             time.Duration   `mapstructure:"force_flush_period,omitempty"`
+	Header                  *HeaderConfig   `mapstructure:"header,omitempty"`
+	DeleteAfterRead         bool            `mapstructure:"delete_after_read,omitempty"`
+	IncludeFileRecordNumber bool            `mapstructure:"include_file_record_number,omitempty"`
 }
 
 type HeaderConfig struct {
@@ -153,19 +154,20 @@ func (c Config) Build(set component.TelemetrySettings, emit emit.Callback, opts
 
 	set.Logger = set.Logger.With(zap.String("component", "fileconsumer"))
 	readerFactory := reader.Factory{
-		TelemetrySettings: set,
-		FromBeginning:     startAtBeginning,
-		FingerprintSize:   int(c.FingerprintSize),
-		InitialBufferSize: scanner.DefaultBufferSize,
-		MaxLogSize:        int(c.MaxLogSize),
-		Encoding:          enc,
-		SplitFunc:         splitFunc,
-		TrimFunc:          trimFunc,
-		FlushTimeout:      c.FlushPeriod,
-		EmitFunc:          emit,
-		Attributes:        c.Resolver,
-		HeaderConfig:      hCfg,
-		DeleteAtEOF:       c.DeleteAfterRead,
+		TelemetrySettings:       set,
+		FromBeginning:           startAtBeginning,
+		FingerprintSize:         int(c.FingerprintSize),
+		InitialBufferSize:       scanner.DefaultBufferSize,
+		MaxLogSize:              int(c.MaxLogSize),
+		Encoding:                enc,
+		SplitFunc:               splitFunc,
+		TrimFunc:                trimFunc,
+		FlushTimeout:            c.FlushPeriod,
+		EmitFunc:                emit,
+		Attributes:              c.Resolver,
+		HeaderConfig:            hCfg,
+		DeleteAtEOF:             c.DeleteAfterRead,
+		IncludeFileRecordNumber: c.IncludeFileRecordNumber,
 	}
 
 	var t tracker.Tracker
diff --git a/pkg/stanza/fileconsumer/internal/reader/factory.go b/pkg/stanza/fileconsumer/internal/reader/factory.go
@@ -30,18 +30,19 @@ const (
 
 type Factory struct {
 	component.TelemetrySettings
-	HeaderConfig      *header.Config
-	FromBeginning     bool
-	FingerprintSize   int
-	InitialBufferSize int
-	MaxLogSize        int
-	Encoding          encoding.Encoding
-	SplitFunc         bufio.SplitFunc
-	TrimFunc          trim.Func
-	FlushTimeout      time.Duration
-	EmitFunc          emit.Callback
-	Attributes        attrs.Resolver
-	DeleteAtEOF       bool
+	HeaderConfig            *header.Config
+	FromBeginning           bool
+	FingerprintSize         int
+	InitialBufferSize       int
+	MaxLogSize              int
+	Encoding                encoding.Encoding
+	SplitFunc               bufio.SplitFunc
+	TrimFunc                trim.Func
+	FlushTimeout            time.Duration
+	EmitFunc                emit.Callback
+	Attributes              attrs.Resolver
+	DeleteAtEOF             bool
+	IncludeFileRecordNumber bool
 }
 
 func (f *Factory) NewFingerprint(file *os.File) (*fingerprint.Fingerprint, error) {
@@ -63,16 +64,17 @@ func (f *Factory) NewReader(file *os.File, fp *fingerprint.Fingerprint) (*Reader
 func (f *Factory) NewReaderFromMetadata(file *os.File, m *Metadata) (r *Reader, err error) {
 
 	r = &Reader{
-		Metadata:          m,
-		set:               f.TelemetrySettings,
-		file:              file,
-		fileName:          file.Name(),
-		fingerprintSize:   f.FingerprintSize,
-		initialBufferSize: f.InitialBufferSize,
-		maxLogSize:        f.MaxLogSize,
-		decoder:           decode.New(f.Encoding),
-		lineSplitFunc:     f.SplitFunc,
-		deleteAtEOF:       f.DeleteAtEOF,
+		Metadata:             m,
+		set:                  f.TelemetrySettings,
+		file:                 file,
+		fileName:             file.Name(),
+		fingerprintSize:      f.FingerprintSize,
+		initialBufferSize:    f.InitialBufferSize,
+		maxLogSize:           f.MaxLogSize,
+		decoder:              decode.New(f.Encoding),
+		lineSplitFunc:        f.SplitFunc,
+		deleteAtEOF:          f.DeleteAtEOF,
+		includeFileRecordNum: f.IncludeFileRecordNumber,
 	}
 	r.set.Logger = r.set.Logger.With(zap.String("path", r.fileName))
 
diff --git a/pkg/stanza/fileconsumer/internal/reader/reader.go b/pkg/stanza/fileconsumer/internal/reader/reader.go
@@ -48,6 +48,7 @@ type Reader struct {
 	emitFunc               emit.Callback
 	deleteAtEOF            bool
 	needsUpdateFingerprint bool
+	includeFileRecordNum   bool
 }
 
 // ReadToEnd will read until the end of the file
@@ -90,8 +91,7 @@ func (r *Reader) ReadToEnd(ctx context.Context) {
 			continue
 		}
 
-		_, fileRecordNumEnabled := r.FileAttributes[attrs.LogFileRecordNumber]
-		if fileRecordNumEnabled {
+		if r.includeFileRecordNum {
 			r.RecordNum++
 			r.FileAttributes[attrs.LogFileRecordNumber] = r.RecordNum
 		}

Original file line number	Diff line number	Diff line change
`@@ -27,7 +27,6 @@ type Resolver struct {`
`27`	`27`	IncludeFilePathResolved bool `mapstructure:"include_file_path_resolved,omitempty"`
`28`	`28`	IncludeFileOwnerName bool `mapstructure:"include_file_owner_name,omitempty"`
`29`	`29`	IncludeFileOwnerGroupName bool `mapstructure:"include_file_owner_group_name,omitempty"`
`30`		- IncludeFileRecordNumber bool `mapstructure:"include_file_record_number,omitempty"`
`31`	`30`	`}`
`32`	`31`
`33`	`32`	`func (r Resolver) Resolve(file os.File) (attributes map[string]any, err error) {`
`@@ -46,10 +45,6 @@ func (r Resolver) Resolve(file os.File) (attributes map[string]any, err error)`
`46`	`45`	`return nil, err`
`47`	`46`	`}`
`48`	`47`	`}`
`49`		`- if r.IncludeFileRecordNumber {`
`50`		`- // non-zero value to flag for setting`
`51`		`- attributes[LogFileRecordNumber] = 1`
`52`		`- }`
`53`	`48`	`if !r.IncludeFileNameResolved && !r.IncludeFilePathResolved {`
`54`	`49`	`return attributes, nil`
`55`	`50`	`}`
Original file line number	Diff line number	Diff line change
`@@ -48,6 +48,7 @@ type Reader struct {`
`48`	`48`	`emitFunc emit.Callback`
`49`	`49`	`deleteAtEOF bool`
`50`	`50`	`needsUpdateFingerprint bool`
	`51`	`+ includeFileRecordNum bool`
`51`	`52`	`}`
`52`	`53`
`53`	`54`	`// ReadToEnd will read until the end of the file`
`@@ -90,8 +91,7 @@ func (r *Reader) ReadToEnd(ctx context.Context) {`
`90`	`91`	`continue`
`91`	`92`	`}`
`92`	`93`
`93`		`- _, fileRecordNumEnabled := r.FileAttributes[attrs.LogFileRecordNumber]`
`94`		`- if fileRecordNumEnabled {`
	`94`	`+ if r.includeFileRecordNum {`
`95`	`95`	`r.RecordNum++`
`96`	`96`	`r.FileAttributes[attrs.LogFileRecordNumber] = r.RecordNum`
`97`	`97`	`}`