[fileconsumer] Generlize emit function (#24036)

Author: djaglowski

.chloggen/fileconsumer-emitfunc-generalize.yaml

@@ -0,0 +1,20 @@
+# Use this changelog template to create an entry for release notes.
+# If your change doesn't affect end users, such as a test fix or a tooling change,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: deprecation
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: pkg/stanza
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Deprecate fileconsumer.EmitFunc in favor of fileconsumer.emit.Callback
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [24036]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:

pkg/stanza/fileconsumer/attributes.go

@@ -4,14 +4,10 @@
 package fileconsumer // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer"
 
 import (
-	"path/filepath"
-	"runtime"
-
-	"go.uber.org/multierr"
-
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/util"
 )
 
+// Deprecated: [v0.82.0] This will be removed in a future release, tentatively v0.84.0.
 type FileAttributes struct {
 	Name             string `json:"-"`
 	Path             string `json:"-"`
@@ -21,28 +17,8 @@ type FileAttributes struct {
 }
 
 // HeaderAttributesCopy gives a copy of the HeaderAttributes, in order to restrict mutation of the HeaderAttributes.
+//
+// Deprecated: [v0.82.0] This will be removed in a future release, tentatively v0.84.0.
 func (f *FileAttributes) HeaderAttributesCopy() map[string]any {
 	return util.MapCopy(f.HeaderAttributes)
 }
-
-// resolveFileAttributes resolves file attributes
-// and sets it to empty string in case of error
-func resolveFileAttributes(path string) (*FileAttributes, error) {
-	resolved := ""
-	var symErr error
-	// Dirty solution, waiting for this permanent fix https://github.com/golang/go/issues/39786
-	// EvalSymlinks on windows is partially working depending on the way you use Symlinks and Junctions
-	if runtime.GOOS != "windows" {
-		resolved, symErr = filepath.EvalSymlinks(path)
-	} else {
-		resolved = path
-	}
-	abs, absErr := filepath.Abs(resolved)
-
-	return &FileAttributes{
-		Path:         path,
-		Name:         filepath.Base(path),
-		PathResolved: abs,
-		NameResolved: filepath.Base(abs),
-	}, multierr.Combine(symErr, absErr)
-}

pkg/stanza/fileconsumer/config.go

@@ -13,6 +13,7 @@ import (
 	"go.opentelemetry.io/collector/featuregate"
 	"go.uber.org/zap"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/emit"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/internal/fingerprint"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper"
 )
@@ -72,7 +73,7 @@ type Config struct {
 }
 
 // Build will build a file input operator from the supplied configuration
-func (c Config) Build(logger *zap.SugaredLogger, emit EmitFunc) (*Manager, error) {
+func (c Config) Build(logger *zap.SugaredLogger, emit emit.Callback) (*Manager, error) {
 	if err := c.validate(); err != nil {
 		return nil, err
 	}
@@ -87,7 +88,7 @@ func (c Config) Build(logger *zap.SugaredLogger, emit EmitFunc) (*Manager, error
 }
 
 // BuildWithSplitFunc will build a file input operator with customized splitFunc function
-func (c Config) BuildWithSplitFunc(logger *zap.SugaredLogger, emit EmitFunc, splitFunc bufio.SplitFunc) (*Manager, error) {
+func (c Config) BuildWithSplitFunc(logger *zap.SugaredLogger, emit emit.Callback, splitFunc bufio.SplitFunc) (*Manager, error) {
 	if err := c.validate(); err != nil {
 		return nil, err
 	}
@@ -105,7 +106,7 @@ func (c Config) BuildWithSplitFunc(logger *zap.SugaredLogger, emit EmitFunc, spl
 	return c.buildManager(logger, emit, factory)
 }
 
-func (c Config) buildManager(logger *zap.SugaredLogger, emit EmitFunc, factory splitterFactory) (*Manager, error) {
+func (c Config) buildManager(logger *zap.SugaredLogger, emit emit.Callback, factory splitterFactory) (*Manager, error) {
 	if emit == nil {
 		return nil, fmt.Errorf("must provide emit function")
 	}
@@ -138,9 +139,13 @@ func (c Config) buildManager(logger *zap.SugaredLogger, emit EmitFunc, factory s
 		readerFactory: readerFactory{
 			SugaredLogger: logger.With("component", "fileconsumer"),
 			readerConfig: &readerConfig{
-				fingerprintSize: int(c.FingerprintSize),
-				maxLogSize:      int(c.MaxLogSize),
-				emit:            emit,
+				fingerprintSize:         int(c.FingerprintSize),
+				maxLogSize:              int(c.MaxLogSize),
+				emit:                    emit,
+				includeFileName:         c.IncludeFileName,
+				includeFilePath:         c.IncludeFilePath,
+				includeFileNameResolved: c.IncludeFileNameResolved,
+				includeFilePathResolved: c.IncludeFilePathResolved,
 			},
 			fromBeginning:   startAtBeginning,
 			splitterFactory: factory,

pkg/stanza/fileconsumer/config_test.go

@@ -4,7 +4,6 @@
 package fileconsumer
 
 import (
-	"context"
 	"path/filepath"
 	"testing"
 	"time"
@@ -636,9 +635,7 @@ func TestBuild(t *testing.T) {
 			cfg := basicConfig()
 			tc.modifyBaseConfig(cfg)
 
-			nopEmit := func(_ context.Context, _ *FileAttributes, _ []byte) {}
-
-			input, err := cfg.Build(testutil.Logger(t), nopEmit)
+			input, err := cfg.Build(testutil.Logger(t), nopEmitFunc)
 			tc.errorRequirement(t, err)
 			if err != nil {
 				return
@@ -708,7 +705,6 @@ func TestBuildWithSplitFunc(t *testing.T) {
 			cfg := basicConfig()
 			tc.modifyBaseConfig(cfg)
 
-			nopEmit := func(_ context.Context, _ *FileAttributes, _ []byte) {}
 			splitNone := func(data []byte, atEOF bool) (advance int, token []byte, err error) {
 				if !atEOF {
 					return 0, nil, nil
@@ -719,7 +715,7 @@ func TestBuildWithSplitFunc(t *testing.T) {
 				return len(data), data, nil
 			}
 
-			input, err := cfg.BuildWithSplitFunc(testutil.Logger(t), nopEmit, splitNone)
+			input, err := cfg.BuildWithSplitFunc(testutil.Logger(t), nopEmitFunc, splitNone)
 			tc.errorRequirement(t, err)
 			if err != nil {
 				return
@@ -809,9 +805,7 @@ func TestBuildWithHeader(t *testing.T) {
 			cfg := basicConfig()
 			tc.modifyBaseConfig(cfg)
 
-			nopEmit := func(_ context.Context, _ *FileAttributes, _ []byte) {}
-
-			input, err := cfg.Build(testutil.Logger(t), nopEmit)
+			input, err := cfg.Build(testutil.Logger(t), nopEmitFunc)
 			tc.errorRequirement(t, err)
 			if err != nil {
 				return

pkg/stanza/fileconsumer/emit/emit.go

@@ -0,0 +1,10 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package emit // import "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/fileconsumer/emit"
+
+import (
+	"context"
+)
+
+type Callback func(ctx context.Context, token []byte, attrs map[string]any) error

pkg/stanza/fileconsumer/file.go

@@ -18,6 +18,14 @@ import (
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator"
 )
 
+const (
+	logFileName         = "log.file.name"
+	logFilePath         = "log.file.path"
+	logFileNameResolved = "log.file.name_resolved"
+	logFilePathResolved = "log.file.path_resolved"
+)
+
+// Deprecated: [v0.82.0] Use emit.Callback instead. This will be removed in a future release, tentatively v0.84.0.
 type EmitFunc func(ctx context.Context, attrs *FileAttributes, token []byte)
 
 type Manager struct {
@@ -369,6 +377,21 @@ func (m *Manager) loadLastPollFiles(ctx context.Context) error {
 		if err = dec.Decode(unsafeReader); err != nil {
 			return err
 		}
+
+		// Migrate readers that used FileAttributes.HeaderAttributes
+		// This block can be removed in a future release, tentatively v0.90.0
+		if ha, ok := unsafeReader.FileAttributes["HeaderAttributes"]; ok {
+			switch hat := ha.(type) {
+			case map[string]any:
+				for k, v := range hat {
+					unsafeReader.FileAttributes[k] = v
+				}
+				delete(unsafeReader.FileAttributes, "HeaderAttributes")
+			default:
+				m.Errorw("migrate header attributes: unexpected format")
+			}
+		}
+
 		m.knownFiles = append(m.knownFiles, unsafeReader)
 	}
 

pkg/stanza/fileconsumer/file_test.go

@@ -54,6 +54,8 @@ func TestAddFileFields(t *testing.T) {
 	cfg.StartAt = "beginning"
 	cfg.IncludeFileName = true
 	cfg.IncludeFilePath = true
+	cfg.IncludeFileNameResolved = false
+	cfg.IncludeFilePathResolved = false
 	operator, emitCalls := buildTestManager(t, cfg)
 
 	// Create a file, then start
@@ -66,8 +68,10 @@ func TestAddFileFields(t *testing.T) {
 	}()
 
 	emitCall := waitForEmit(t, emitCalls)
-	require.Equal(t, filepath.Base(temp.Name()), emitCall.attrs.Name)
-	require.Equal(t, temp.Name(), emitCall.attrs.Path)
+	require.Equal(t, filepath.Base(temp.Name()), emitCall.attrs[logFileName])
+	require.Equal(t, temp.Name(), emitCall.attrs[logFilePath])
+	require.Nil(t, emitCall.attrs[logFileNameResolved])
+	require.Nil(t, emitCall.attrs[logFilePathResolved])
 }
 
 // AddFileResolvedFields tests that the `log.file.name_resolved` and `log.file.path_resolved` fields are included
@@ -116,10 +120,10 @@ func TestAddFileResolvedFields(t *testing.T) {
 	}()
 
 	emitCall := waitForEmit(t, emitCalls)
-	require.Equal(t, filepath.Base(symLinkPath), emitCall.attrs.Name)
-	require.Equal(t, symLinkPath, emitCall.attrs.Path)
-	require.Equal(t, filepath.Base(resolved), emitCall.attrs.NameResolved)
-	require.Equal(t, resolved, emitCall.attrs.PathResolved)
+	require.Equal(t, filepath.Base(symLinkPath), emitCall.attrs[logFileName])
+	require.Equal(t, symLinkPath, emitCall.attrs[logFilePath])
+	require.Equal(t, filepath.Base(resolved), emitCall.attrs[logFileNameResolved])
+	require.Equal(t, resolved, emitCall.attrs[logFilePathResolved])
 }
 
 // AddFileResolvedFields tests that the `log.file.name_resolved` and `log.file.path_resolved` fields are included
@@ -186,10 +190,10 @@ func TestAddFileResolvedFieldsWithChangeOfSymlinkTarget(t *testing.T) {
 	}()
 
 	emitCall := waitForEmit(t, emitCalls)
-	require.Equal(t, filepath.Base(symLinkPath), emitCall.attrs.Name)
-	require.Equal(t, symLinkPath, emitCall.attrs.Path)
-	require.Equal(t, filepath.Base(resolved1), emitCall.attrs.NameResolved)
-	require.Equal(t, resolved1, emitCall.attrs.PathResolved)
+	require.Equal(t, filepath.Base(symLinkPath), emitCall.attrs[logFileName])
+	require.Equal(t, symLinkPath, emitCall.attrs[logFilePath])
+	require.Equal(t, filepath.Base(resolved1), emitCall.attrs[logFileNameResolved])
+	require.Equal(t, resolved1, emitCall.attrs[logFilePathResolved])
 
 	// Change middleSymLink to point to file2
 	err = os.Remove(middleSymLinkPath)
@@ -201,10 +205,10 @@ func TestAddFileResolvedFieldsWithChangeOfSymlinkTarget(t *testing.T) {
 	writeString(t, file2, "testlog2\n")
 
 	emitCall = waitForEmit(t, emitCalls)
-	require.Equal(t, filepath.Base(symLinkPath), emitCall.attrs.Name)
-	require.Equal(t, symLinkPath, emitCall.attrs.Path)
-	require.Equal(t, filepath.Base(resolved2), emitCall.attrs.NameResolved)
-	require.Equal(t, resolved2, emitCall.attrs.PathResolved)
+	require.Equal(t, filepath.Base(symLinkPath), emitCall.attrs[logFileName])
+	require.Equal(t, symLinkPath, emitCall.attrs[logFilePath])
+	require.Equal(t, filepath.Base(resolved2), emitCall.attrs[logFileNameResolved])
+	require.Equal(t, resolved2, emitCall.attrs[logFilePathResolved])
 }
 
 func TestFileFieldsUpdatedAfterRestart(t *testing.T) {
@@ -227,8 +231,10 @@ func TestFileFieldsUpdatedAfterRestart(t *testing.T) {
 
 	emitCall1 := waitForEmit(t, emitCalls1)
 	assert.Equal(t, []byte("testlog1"), emitCall1.token)
-	assert.Equal(t, filepath.Base(temp.Name()), emitCall1.attrs.Name)
-	assert.Equal(t, temp.Name(), emitCall1.attrs.Path)
+	assert.Equal(t, filepath.Base(temp.Name()), emitCall1.attrs[logFileName])
+	assert.Equal(t, temp.Name(), emitCall1.attrs[logFilePath])
+	assert.Nil(t, emitCall1.attrs[logFileNameResolved])
+	assert.Nil(t, emitCall1.attrs[logFilePathResolved])
 
 	require.NoError(t, op1.Stop())
 	temp.Close() // On windows, we must close the file before renaming it
@@ -245,8 +251,10 @@ func TestFileFieldsUpdatedAfterRestart(t *testing.T) {
 
 	emitCall2 := waitForEmit(t, emitCalls2)
 	assert.Equal(t, []byte("testlog2"), emitCall2.token)
-	assert.Equal(t, filepath.Base(newPath), emitCall2.attrs.Name)
-	assert.Equal(t, newPath, emitCall2.attrs.Path)
+	assert.Equal(t, filepath.Base(newPath), emitCall2.attrs[logFileName])
+	assert.Equal(t, newPath, emitCall2.attrs[logFilePath])
+	assert.Nil(t, emitCall2.attrs[logFileNameResolved])
+	assert.Nil(t, emitCall2.attrs[logFilePathResolved])
 
 	require.NoError(t, op2.Stop())
 }
@@ -1457,9 +1465,10 @@ func TestReadExistingLogsWithHeader(t *testing.T) {
 		require.NoError(t, operator.Stop())
 	}()
 
-	waitForTokenHeaderAttributes(t, emitCalls, []byte("testlog"), map[string]any{
+	waitForTokenWithAttributes(t, emitCalls, []byte("testlog"), map[string]any{
 		"header_key":   "headerField",
 		"header_value": "headerValue",
+		logFileName:    filepath.Base(temp.Name()),
 	})
 }
 
@@ -1555,9 +1564,10 @@ func TestHeaderPersistance(t *testing.T) {
 	persister := testutil.NewUnscopedMockPersister()
 	require.NoError(t, op1.Start(persister))
 
-	waitForTokenHeaderAttributes(t, emitCalls1, []byte("log line"), map[string]any{
+	waitForTokenWithAttributes(t, emitCalls1, []byte("log line"), map[string]any{
 		"header_key":   "headerField",
 		"header_value": "headerValue",
+		logFileName:    filepath.Base(temp.Name()),
 	})
 
 	require.NoError(t, op1.Stop())
@@ -1568,13 +1578,13 @@ func TestHeaderPersistance(t *testing.T) {
 
 	require.NoError(t, op2.Start(persister))
 
-	waitForTokenHeaderAttributes(t, emitCalls2, []byte("log line 2"), map[string]any{
+	waitForTokenWithAttributes(t, emitCalls2, []byte("log line 2"), map[string]any{
 		"header_key":   "headerField",
 		"header_value": "headerValue",
+		logFileName:    filepath.Base(temp.Name()),
 	})
 
 	require.NoError(t, op2.Stop())
-
 }
 
 func TestHeaderPersistanceInHeader(t *testing.T) {
@@ -1613,9 +1623,10 @@ func TestHeaderPersistanceInHeader(t *testing.T) {
 
 	require.NoError(t, op2.Start(persister))
 
-	waitForTokenHeaderAttributes(t, emitCalls, []byte("log line"), map[string]any{
+	waitForTokenWithAttributes(t, emitCalls, []byte("log line"), map[string]any{
 		"header_value_1": "headerValue1",
 		"header_value_2": "headerValue2",
+		logFileName:      filepath.Base(temp.Name()),
 	})
 
 	require.NoError(t, op2.Stop())