restructured fileexporter to prepare for group_by (#31068)

**Description:** 

In fileexporter: restructured code by splitting out two functionality
from `fileExporter`: marshalling and compression logic was moved to
`marshaller`, and logic related to writing to file and buffering writes
was moved to, `fileWriter`.

This pr introduces no changes to the behavior. The restructure was made
in preparation for adding the new group_by functionality (see linked
ticket for more detail).

**Link to tracking Issue:** #24654

**Testing:** Tests have been updated for the new structure. No tests
were added or modified beyond the structural changes.

**Documentation:** This pr introduces no user-facing changes.

exporter/fileexporter/error_component.go

@@ -0,0 +1,26 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package fileexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/fileexporter"
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/component"
+)
+
+// errorComponent is used to return error from a factory method. SharedComponent does
+// not handle errors, so wrapping the error into a component is necessary.
+type errorComponent struct {
+	err error
+}
+
+// Start will return the cached error.
+func (e *errorComponent) Start(context.Context, component.Host) error {
+	return e.err
+}
+
+// Shutdown will return the cached error.
+func (e *errorComponent) Shutdown(context.Context) error {
+	return e.err
+}

exporter/fileexporter/factory.go

@@ -7,11 +7,15 @@ import (
 	"context"
 	"io"
 	"os"
+	"time"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/exporter"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/pdata/ptrace"
 	"gopkg.in/natefinch/lumberjack.v2"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/fileexporter/internal/metadata"
@@ -30,6 +34,13 @@ const (
 	compressionZSTD = "zstd"
 )
 
+type FileExporter interface {
+	component.Component
+	consumeTraces(_ context.Context, td ptrace.Traces) error
+	consumeMetrics(_ context.Context, md pmetric.Metrics) error
+	consumeLogs(_ context.Context, ld plog.Logs) error
+}
+
 // NewFactory creates a factory for OTLP exporter.
 func NewFactory() exporter.Factory {
 	return exporter.NewFactory(
@@ -52,19 +63,15 @@ func createTracesExporter(
 	set exporter.CreateSettings,
 	cfg component.Config,
 ) (exporter.Traces, error) {
-	conf := cfg.(*Config)
-	writer, err := buildFileWriter(conf)
+	fe, err := getOrCreateFileExporter(cfg)
 	if err != nil {
 		return nil, err
 	}
-	fe := exporters.GetOrAdd(cfg, func() component.Component {
-		return newFileExporter(conf, writer)
-	})
 	return exporterhelper.NewTracesExporter(
 		ctx,
 		set,
 		cfg,
-		fe.Unwrap().(*fileExporter).consumeTraces,
+		fe.consumeTraces,
 		exporterhelper.WithStart(fe.Start),
 		exporterhelper.WithShutdown(fe.Shutdown),
 		exporterhelper.WithCapabilities(consumer.Capabilities{MutatesData: false}),
@@ -76,19 +83,15 @@ func createMetricsExporter(
 	set exporter.CreateSettings,
 	cfg component.Config,
 ) (exporter.Metrics, error) {
-	conf := cfg.(*Config)
-	writer, err := buildFileWriter(conf)
+	fe, err := getOrCreateFileExporter(cfg)
 	if err != nil {
 		return nil, err
 	}
-	fe := exporters.GetOrAdd(cfg, func() component.Component {
-		return newFileExporter(conf, writer)
-	})
 	return exporterhelper.NewMetricsExporter(
 		ctx,
 		set,
 		cfg,
-		fe.Unwrap().(*fileExporter).consumeMetrics,
+		fe.consumeMetrics,
 		exporterhelper.WithStart(fe.Start),
 		exporterhelper.WithShutdown(fe.Shutdown),
 		exporterhelper.WithCapabilities(consumer.Capabilities{MutatesData: false}),
@@ -100,59 +103,94 @@ func createLogsExporter(
 	set exporter.CreateSettings,
 	cfg component.Config,
 ) (exporter.Logs, error) {
-	conf := cfg.(*Config)
-	writer, err := buildFileWriter(conf)
+	fe, err := getOrCreateFileExporter(cfg)
 	if err != nil {
 		return nil, err
 	}
-	fe := exporters.GetOrAdd(cfg, func() component.Component {
-		return newFileExporter(conf, writer)
-	})
 	return exporterhelper.NewLogsExporter(
 		ctx,
 		set,
 		cfg,
-		fe.Unwrap().(*fileExporter).consumeLogs,
+		fe.consumeLogs,
 		exporterhelper.WithStart(fe.Start),
 		exporterhelper.WithShutdown(fe.Shutdown),
 		exporterhelper.WithCapabilities(consumer.Capabilities{MutatesData: false}),
 	)
 }
 
-func newFileExporter(conf *Config, writer io.WriteCloser) *fileExporter {
-	return &fileExporter{
-		path:             conf.Path,
+// getOrCreateFileExporter creates a FileExporter and caches it for a particular configuration,
+// or returns the already cached one. Caching is required because the factory is asked trace and
+// metric receivers separately when it gets CreateTracesReceiver() and CreateMetricsReceiver()
+// but they must not create separate objects, they must use one Exporter object per configuration.
+func getOrCreateFileExporter(cfg component.Config) (FileExporter, error) {
+	conf := cfg.(*Config)
+	fe := exporters.GetOrAdd(cfg, func() component.Component {
+		e, err := newFileExporter(conf)
+		if err != nil {
+			return &errorComponent{err: err}
+		}
+
+		return e
+	})
+
+	component := fe.Unwrap()
+	if errComponent, ok := component.(*errorComponent); ok {
+		return nil, errComponent.err
+	}
+
+	return component.(FileExporter), nil
+}
+
+func newFileExporter(conf *Config) (FileExporter, error) {
+	marshaller := &marshaller{
 		formatType:       conf.FormatType,
-		file:             writer,
 		tracesMarshaler:  tracesMarshalers[conf.FormatType],
 		metricsMarshaler: metricsMarshalers[conf.FormatType],
 		logsMarshaler:    logsMarshalers[conf.FormatType],
-		exporter:         buildExportFunc(conf),
 		compression:      conf.Compression,
 		compressor:       buildCompressor(conf.Compression),
-		flushInterval:    conf.FlushInterval,
 	}
+	export := buildExportFunc(conf)
+
+	writer, err := newFileWriter(conf.Path, conf.Rotation, conf.FlushInterval, export)
+	if err != nil {
+		return nil, err
+	}
+
+	return &fileExporter{
+		marshaller: marshaller,
+		writer:     writer,
+	}, nil
 }
 
-func buildFileWriter(cfg *Config) (io.WriteCloser, error) {
-	if cfg.Rotation == nil {
-		f, err := os.OpenFile(cfg.Path, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0600)
+func newFileWriter(path string, rotation *Rotation, flushInterval time.Duration, export exportFunc) (*fileWriter, error) {
+	var wc io.WriteCloser
+	if rotation == nil {
+		f, err := os.OpenFile(path, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0644)
 		if err != nil {
 			return nil, err
 		}
-		return newBufferedWriteCloser(f), nil
+		wc = newBufferedWriteCloser(f)
+	} else {
+		wc = &lumberjack.Logger{
+			Filename:   path,
+			MaxSize:    rotation.MaxMegabytes,
+			MaxAge:     rotation.MaxDays,
+			MaxBackups: rotation.MaxBackups,
+			LocalTime:  rotation.LocalTime,
+		}
 	}
-	return &lumberjack.Logger{
-		Filename:   cfg.Path,
-		MaxSize:    cfg.Rotation.MaxMegabytes,
-		MaxAge:     cfg.Rotation.MaxDays,
-		MaxBackups: cfg.Rotation.MaxBackups,
-		LocalTime:  cfg.Rotation.LocalTime,
+
+	return &fileWriter{
+		path:          path,
+		file:          wc,
+		exporter:      export,
+		flushInterval: flushInterval,
 	}, nil
 }
 
 // This is the map of already created File exporters for particular configurations.
 // We maintain this map because the Factory is asked trace and metric receivers separately
 // when it gets CreateTracesReceiver() and CreateMetricsReceiver() but they must not
-// create separate objects, they must use one Receiver object per configuration.
+// create separate objects, they must use one Exporter object per configuration.
 var exporters = sharedcomponent.NewSharedComponents()

exporter/fileexporter/factory_test.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"io"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -96,25 +97,27 @@ func TestCreateLogsExporterError(t *testing.T) {
 	assert.Error(t, err)
 }
 
-func TestBuildFileWriter(t *testing.T) {
+func TestNewFileWriter(t *testing.T) {
 	type args struct {
 		cfg *Config
 	}
 	tests := []struct {
 		name     string
 		args     args
 		want     io.WriteCloser
-		validate func(*testing.T, io.WriteCloser)
+		validate func(*testing.T, *fileWriter)
 	}{
 		{
 			name: "single file",
 			args: args{
 				cfg: &Config{
-					Path: tempFileName(t),
+					Path:          tempFileName(t),
+					FlushInterval: 5 * time.Second,
 				},
 			},
-			validate: func(t *testing.T, closer io.WriteCloser) {
-				_, ok := closer.(*bufferedWriteCloser)
+			validate: func(t *testing.T, writer *fileWriter) {
+				assert.Equal(t, 5*time.Second, writer.flushInterval)
+				_, ok := writer.file.(*bufferedWriteCloser)
 				assert.Equal(t, true, ok)
 			},
 		},
@@ -128,10 +131,10 @@ func TestBuildFileWriter(t *testing.T) {
 					},
 				},
 			},
-			validate: func(t *testing.T, closer io.WriteCloser) {
-				writer, ok := closer.(*lumberjack.Logger)
+			validate: func(t *testing.T, writer *fileWriter) {
+				logger, ok := writer.file.(*lumberjack.Logger)
 				assert.Equal(t, true, ok)
-				assert.Equal(t, defaultMaxBackups, writer.MaxBackups)
+				assert.Equal(t, defaultMaxBackups, logger.MaxBackups)
 			},
 		},
 		{
@@ -147,21 +150,21 @@ func TestBuildFileWriter(t *testing.T) {
 					},
 				},
 			},
-			validate: func(t *testing.T, closer io.WriteCloser) {
-				writer, ok := closer.(*lumberjack.Logger)
+			validate: func(t *testing.T, writer *fileWriter) {
+				logger, ok := writer.file.(*lumberjack.Logger)
 				assert.Equal(t, true, ok)
-				assert.Equal(t, 3, writer.MaxBackups)
-				assert.Equal(t, 30, writer.MaxSize)
-				assert.Equal(t, 100, writer.MaxAge)
-				assert.Equal(t, true, writer.LocalTime)
+				assert.Equal(t, 3, logger.MaxBackups)
+				assert.Equal(t, 30, logger.MaxSize)
+				assert.Equal(t, 100, logger.MaxAge)
+				assert.Equal(t, true, logger.LocalTime)
 			},
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := buildFileWriter(tt.args.cfg)
+			got, err := newFileWriter(tt.args.cfg.Path, tt.args.cfg.Rotation, tt.args.cfg.FlushInterval, nil)
 			defer func() {
-				assert.NoError(t, got.Close())
+				assert.NoError(t, got.file.Close())
 			}()
 			assert.NoError(t, err)
 			tt.validate(t, got)