Change fileexporter to use the new SharedComponents because of the file (open-telemetry#3201)

Signed-off-by: Bogdan Drutu <[email protected]>

Author: bogdandrutu

exporter/fileexporter/config.go

@@ -15,6 +15,8 @@
 package fileexporter
 
 import (
+	"errors"
+
 	"go.opentelemetry.io/collector/config"
 )
 
@@ -30,5 +32,9 @@ var _ config.Exporter = (*Config)(nil)
 
 // Validate checks if the exporter configuration is valid
 func (cfg *Config) Validate() error {
+	if cfg.Path == "" {
+		return errors.New("path must be non-empty")
+	}
+
 	return nil
 }

exporter/fileexporter/config_test.go

@@ -33,8 +33,7 @@ func TestLoadConfig(t *testing.T) {
 	factory := NewFactory()
 	factories.Exporters[typeStr] = factory
 	cfg, err := configtest.LoadConfigFile(t, path.Join(".", "testdata", "config.yaml"), factories)
-
-	require.NoError(t, err)
+	require.EqualError(t, err, "exporter \"file\" has invalid configuration: path must be non-empty")
 	require.NotNil(t, cfg)
 
 	e0 := cfg.Exporters[config.NewID(typeStr)]

exporter/fileexporter/factory.go

@@ -16,11 +16,11 @@ package fileexporter
 
 import (
 	"context"
-	"os"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
+	"go.opentelemetry.io/collector/internal/sharedcomponent"
 )
 
 const (
@@ -46,52 +46,39 @@ func createDefaultConfig() config.Exporter {
 
 func createTracesExporter(
 	_ context.Context,
-	_ component.ExporterCreateParams,
+	params component.ExporterCreateParams,
 	cfg config.Exporter,
 ) (component.TracesExporter, error) {
-	return createExporter(cfg)
+	fe := exporters.GetOrAdd(cfg, func() component.Component {
+		return &fileExporter{path: cfg.(*Config).Path}
+	})
+	return exporterhelper.NewTracesExporter(cfg, params.Logger, fe.Unwrap().(*fileExporter).ConsumeTraces)
 }
 
 func createMetricsExporter(
 	_ context.Context,
-	_ component.ExporterCreateParams,
+	params component.ExporterCreateParams,
 	cfg config.Exporter,
 ) (component.MetricsExporter, error) {
-	return createExporter(cfg)
+	fe := exporters.GetOrAdd(cfg, func() component.Component {
+		return &fileExporter{path: cfg.(*Config).Path}
+	})
+	return exporterhelper.NewMetricsExporter(cfg, params.Logger, fe.Unwrap().(*fileExporter).ConsumeMetrics)
 }
 
 func createLogsExporter(
 	_ context.Context,
-	_ component.ExporterCreateParams,
+	params component.ExporterCreateParams,
 	cfg config.Exporter,
 ) (component.LogsExporter, error) {
-	return createExporter(cfg)
-}
-
-func createExporter(config config.Exporter) (*fileExporter, error) {
-	cfg := config.(*Config)
-
-	// There must be one exporter for metrics, traces, and logs. We maintain a
-	// map of exporters per config.
-
-	// Check to see if there is already a exporter for this config.
-	exporter, ok := exporters[cfg]
-
-	if !ok {
-		file, err := os.OpenFile(cfg.Path, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0600)
-		if err != nil {
-			return nil, err
-		}
-		exporter = &fileExporter{file: file}
-
-		// Remember the receiver in the map
-		exporters[cfg] = exporter
-	}
-	return exporter, nil
+	fe := exporters.GetOrAdd(cfg, func() component.Component {
+		return &fileExporter{path: cfg.(*Config).Path}
+	})
+	return exporterhelper.NewLogsExporter(cfg, params.Logger, fe.Unwrap().(*fileExporter).ConsumeLogs)
 }
 
 // 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.
-var exporters = map[*Config]*fileExporter{}
+var exporters = sharedcomponent.NewSharedComponents()

exporter/fileexporter/factory_test.go

@@ -38,8 +38,8 @@ func TestCreateMetricsExporter(t *testing.T) {
 		context.Background(),
 		component.ExporterCreateParams{Logger: zap.NewNop()},
 		cfg)
-	assert.Error(t, err)
-	require.Nil(t, exp)
+	assert.NoError(t, err)
+	require.NotNil(t, exp)
 }
 
 func TestCreateTracesExporter(t *testing.T) {
@@ -48,17 +48,16 @@ func TestCreateTracesExporter(t *testing.T) {
 		context.Background(),
 		component.ExporterCreateParams{Logger: zap.NewNop()},
 		cfg)
-	assert.Error(t, err)
-	require.Nil(t, exp)
+	assert.NoError(t, err)
+	require.NotNil(t, exp)
 }
 
 func TestCreateLogsExporter(t *testing.T) {
 	cfg := createDefaultConfig()
-
 	exp, err := createLogsExporter(
 		context.Background(),
 		component.ExporterCreateParams{Logger: zap.NewNop()},
 		cfg)
-	assert.Error(t, err)
-	require.Nil(t, exp)
+	assert.NoError(t, err)
+	require.NotNil(t, exp)
 }

exporter/fileexporter/file_exporter.go

@@ -17,6 +17,7 @@ package fileexporter
 import (
 	"context"
 	"io"
+	"os"
 	"sync"
 
 	"github.com/gogo/protobuf/jsonpb"
@@ -34,6 +35,7 @@ var marshaler = &jsonpb.Marshaler{}
 // fileExporter is the implementation of file exporter that writes telemetry data to a file
 // in Protobuf-JSON format.
 type fileExporter struct {
+	path  string
 	file  io.WriteCloser
 	mutex sync.Mutex
 }
@@ -68,7 +70,9 @@ func exportMessageAsLine(e *fileExporter, message proto.Message) error {
 }
 
 func (e *fileExporter) Start(context.Context, component.Host) error {
-	return nil
+	var err error
+	e.file, err = os.OpenFile(e.path, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0600)
+	return err
 }
 
 // Shutdown stops the exporter and is invoked during shutdown.

exporter/fileexporter/file_exporter_test.go

@@ -14,7 +14,10 @@
 package fileexporter
 
 import (
+	"bytes"
 	"context"
+	"io/ioutil"
+	"os"
 	"testing"
 
 	"github.com/gogo/protobuf/jsonpb"
@@ -31,8 +34,7 @@ import (
 )
 
 func TestFileTracesExporter(t *testing.T) {
-	mf := &testutil.LimitedWriter{}
-	fe := &fileExporter{file: mf}
+	fe := &fileExporter{path: tempFileName(t)}
 	require.NotNil(t, fe)
 
 	td := testdata.GenerateTracesTwoSpansSameResource()
@@ -42,7 +44,9 @@ func TestFileTracesExporter(t *testing.T) {
 
 	var unmarshaler = &jsonpb.Unmarshaler{}
 	got := &collectortrace.ExportTraceServiceRequest{}
-	assert.NoError(t, unmarshaler.Unmarshal(mf, got))
+	buf, err := ioutil.ReadFile(fe.path)
+	assert.NoError(t, err)
+	assert.NoError(t, unmarshaler.Unmarshal(bytes.NewReader(buf), got))
 	assert.EqualValues(t, internal.TracesToOtlp(td.InternalRep()), got)
 }
 
@@ -54,14 +58,13 @@ func TestFileTracesExporterError(t *testing.T) {
 	require.NotNil(t, fe)
 
 	td := testdata.GenerateTracesTwoSpansSameResource()
-	assert.NoError(t, fe.Start(context.Background(), componenttest.NewNopHost()))
+	// Cannot call Start since we inject directly the WriterCloser.
 	assert.Error(t, fe.ConsumeTraces(context.Background(), td))
 	assert.NoError(t, fe.Shutdown(context.Background()))
 }
 
 func TestFileMetricsExporter(t *testing.T) {
-	mf := &testutil.LimitedWriter{}
-	fe := &fileExporter{file: mf}
+	fe := &fileExporter{path: tempFileName(t)}
 	require.NotNil(t, fe)
 
 	md := testdata.GenerateMetricsTwoMetrics()
@@ -71,7 +74,9 @@ func TestFileMetricsExporter(t *testing.T) {
 
 	var unmarshaler = &jsonpb.Unmarshaler{}
 	got := &collectormetrics.ExportMetricsServiceRequest{}
-	assert.NoError(t, unmarshaler.Unmarshal(mf, got))
+	buf, err := ioutil.ReadFile(fe.path)
+	assert.NoError(t, err)
+	assert.NoError(t, unmarshaler.Unmarshal(bytes.NewReader(buf), got))
 	assert.EqualValues(t, internal.MetricsToOtlp(md.InternalRep()), got)
 }
 
@@ -83,14 +88,13 @@ func TestFileMetricsExporterError(t *testing.T) {
 	require.NotNil(t, fe)
 
 	md := testdata.GenerateMetricsTwoMetrics()
-	assert.NoError(t, fe.Start(context.Background(), componenttest.NewNopHost()))
+	// Cannot call Start since we inject directly the WriterCloser.
 	assert.Error(t, fe.ConsumeMetrics(context.Background(), md))
 	assert.NoError(t, fe.Shutdown(context.Background()))
 }
 
 func TestFileLogsExporter(t *testing.T) {
-	mf := &testutil.LimitedWriter{}
-	fe := &fileExporter{file: mf}
+	fe := &fileExporter{path: tempFileName(t)}
 	require.NotNil(t, fe)
 
 	otlp := testdata.GenerateLogsTwoLogRecordsSameResource()
@@ -100,7 +104,9 @@ func TestFileLogsExporter(t *testing.T) {
 
 	var unmarshaler = &jsonpb.Unmarshaler{}
 	got := &collectorlogs.ExportLogsServiceRequest{}
-	assert.NoError(t, unmarshaler.Unmarshal(mf, got))
+	buf, err := ioutil.ReadFile(fe.path)
+	assert.NoError(t, err)
+	assert.NoError(t, unmarshaler.Unmarshal(bytes.NewReader(buf), got))
 	assert.EqualValues(t, internal.LogsToOtlp(otlp.InternalRep()), got)
 }
 
@@ -112,7 +118,17 @@ func TestFileLogsExporterErrors(t *testing.T) {
 	require.NotNil(t, fe)
 
 	otlp := testdata.GenerateLogsTwoLogRecordsSameResource()
-	assert.NoError(t, fe.Start(context.Background(), componenttest.NewNopHost()))
+	// Cannot call Start since we inject directly the WriterCloser.
 	assert.Error(t, fe.ConsumeLogs(context.Background(), otlp))
 	assert.NoError(t, fe.Shutdown(context.Background()))
 }
+
+// tempFileName provides a temporary file name for testing.
+func tempFileName(t *testing.T) string {
+	tmpfile, err := ioutil.TempFile("", "*.json")
+	require.NoError(t, err)
+	require.NoError(t, tmpfile.Close())
+	socket := tmpfile.Name()
+	require.NoError(t, os.Remove(socket))
+	return socket
+}

service/defaultcomponents/default_exporters_test.go

@@ -168,8 +168,9 @@ func verifyExporterLifecycle(t *testing.T, factory component.ExporterFactory, ge
 		BuildInfo: component.DefaultBuildInfo(),
 	}
 
-	if getConfigFn == nil {
-		getConfigFn = factory.CreateDefaultConfig
+	cfg := factory.CreateDefaultConfig()
+	if getConfigFn != nil {
+		cfg = getConfigFn()
 	}
 
 	createFns := []createExporterFn{
@@ -178,19 +179,20 @@ func verifyExporterLifecycle(t *testing.T, factory component.ExporterFactory, ge
 		wrapCreateMetricsExp(factory),
 	}
 
-	for _, createFn := range createFns {
-		firstExp, err := createFn(ctx, expCreateParams, getConfigFn())
-		if errors.Is(err, componenterror.ErrDataTypeIsNotSupported) {
-			continue
+	for i := 0; i < 2; i++ {
+		var exps []component.Exporter
+		for _, createFn := range createFns {
+			exp, err := createFn(ctx, expCreateParams, cfg)
+			if errors.Is(err, componenterror.ErrDataTypeIsNotSupported) {
+				continue
+			}
+			require.NoError(t, err)
+			require.NoError(t, exp.Start(ctx, host))
+			exps = append(exps, exp)
+		}
+		for _, exp := range exps {
+			assert.NoError(t, exp.Shutdown(ctx))
 		}
-		require.NoError(t, err)
-		require.NoError(t, firstExp.Start(ctx, host))
-		require.NoError(t, firstExp.Shutdown(ctx))
-
-		secondExp, err := createFn(ctx, expCreateParams, getConfigFn())
-		require.NoError(t, err)
-		require.NoError(t, secondExp.Start(ctx, host))
-		require.NoError(t, secondExp.Shutdown(ctx))
 	}
 }