review feedback

Author: moh-osman3

exporter/otelarrowexporter/factory.go

@@ -72,14 +72,15 @@ func createDefaultConfig() component.Config {
 	}
 }
 
-func (e *baseExporter) helperOptions() []exporterhelper.Option {
+func helperOptions(e exp) []exporterhelper.Option {
+	cfg := e.getConfig().(*Config)
 	return []exporterhelper.Option{
 		exporterhelper.WithCapabilities(consumer.Capabilities{MutatesData: false}),
-		exporterhelper.WithTimeout(e.config.TimeoutSettings),
-		exporterhelper.WithRetry(e.config.RetryConfig),
-		exporterhelper.WithQueue(e.config.QueueSettings),
+		exporterhelper.WithTimeout(cfg.TimeoutSettings),
+		exporterhelper.WithRetry(cfg.RetryConfig),
+		exporterhelper.WithQueue(cfg.QueueSettings),
 		exporterhelper.WithStart(e.start),
-		exporterhelper.WithBatcher(e.config.BatcherConfig),
+		exporterhelper.WithBatcher(cfg.BatcherConfig),
 		exporterhelper.WithShutdown(e.shutdown),
 	}
 }
@@ -109,7 +110,7 @@ func createTracesExporter(
 	}
 	return exporterhelper.NewTracesExporter(ctx, exp.getSettings(), exp.getConfig(),
 		exp.pushTraces,
-		exp.helperOptions()...,
+		helperOptions(exp)...,
 	)
 }
 
@@ -128,7 +129,7 @@ func createMetricsExporter(
 	}
 	return exporterhelper.NewMetricsExporter(ctx, exp.getSettings(), exp.getConfig(),
 		exp.pushMetrics,
-		exp.helperOptions()...,
+		helperOptions(exp)...,
 	)
 }
 
@@ -147,6 +148,6 @@ func createLogsExporter(
 	}
 	return exporterhelper.NewLogsExporter(ctx, exp.getSettings(), exp.getConfig(),
 		exp.pushLogs,
-		exp.helperOptions()...,
+		helperOptions(exp)...,
 	)
 }

exporter/otelarrowexporter/metadata.go

@@ -7,22 +7,25 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"runtime"
 	"sort"
 	"strings"
 	"sync"
 
+	arrowPkg "github.com/apache/arrow/go/v16/arrow"
 	"go.opentelemetry.io/collector/client"
 	"go.opentelemetry.io/collector/component"
-	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/consumer/consumererror"
 	"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"
 	"go.opentelemetry.io/otel/attribute"
 	"go.uber.org/multierr"
 	"google.golang.org/grpc/metadata"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow/compression/zstd"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow/netstats"
 )
 
 var (
@@ -38,6 +41,10 @@ type metadataExporter struct {
 
 	metadataKeys []string
 	exporters    sync.Map
+	netReporter  *netstats.NetworkReporter
+
+	// Default user-agent header.
+	userAgent string
 
 	// Guards the size and the storing logic to ensure no more than limit items are stored.
 	// If we are willing to allow "some" extra items than the limit this can be removed and size can be made atomic.
@@ -49,20 +56,40 @@ var _ exp = (*metadataExporter)(nil)
 
 func newMetadataExporter(cfg component.Config, set exporter.Settings, streamClientFactory streamClientFactory) (exp, error) {
 	oCfg := cfg.(*Config)
+	if oCfg.Endpoint == "" {
+		return nil, errors.New("OTLP exporter config requires an Endpoint")
+	}
+
+	netReporter, err := netstats.NewExporterNetworkReporter(set)
+	if err != nil {
+		return nil, err
+	}
+	userAgent := fmt.Sprintf("%s/%s (%s/%s)",
+		set.BuildInfo.Description, set.BuildInfo.Version, runtime.GOOS, runtime.GOARCH)
+
+	if !oCfg.Arrow.Disabled {
+		// Ignoring an error because Validate() was called.
+		_ = zstd.SetEncoderConfig(oCfg.Arrow.Zstd)
+
+		userAgent += fmt.Sprintf(" ApacheArrow/%s (NumStreams/%d)", arrowPkg.PkgVersion, oCfg.Arrow.NumStreams)
+	}
+
 	// use lower-case, to be consistent with http/2 headers.
 	mks := make([]string, len(oCfg.MetadataKeys))
 	for i, k := range oCfg.MetadataKeys {
 		mks[i] = strings.ToLower(k)
 	}
 	sort.Strings(mks)
 	if len(mks) == 0 {
-		return newExporter(cfg, set, streamClientFactory)
+		return newExporter(cfg, set, streamClientFactory, userAgent, netReporter)
 	}
 	return &metadataExporter{
 		config:       oCfg,
 		settings:     set,
 		scf:          streamClientFactory,
 		metadataKeys: mks,
+		userAgent: userAgent,
+		netReporter: netReporter,
 	}, nil
 }
 
@@ -74,17 +101,6 @@ func (e *metadataExporter) getConfig() component.Config {
 	return e.config
 }
 
-func (e *metadataExporter) helperOptions() []exporterhelper.Option {
-	return []exporterhelper.Option{
-		exporterhelper.WithCapabilities(consumer.Capabilities{MutatesData: false}),
-		exporterhelper.WithTimeout(e.config.TimeoutSettings),
-		exporterhelper.WithRetry(e.config.RetryConfig),
-		exporterhelper.WithQueue(e.config.QueueSettings),
-		exporterhelper.WithStart(e.start),
-		exporterhelper.WithShutdown(e.shutdown),
-	}
-}
-
 func (e *metadataExporter) start(_ context.Context, host component.Host) (err error) {
 	e.host = host
 	return nil
@@ -145,7 +161,7 @@ func (e *metadataExporter) getOrCreateExporter(ctx context.Context, s attribute.
 		return nil, errTooManyExporters
 	}
 
-	newExp, err := newExporter(e.config, e.settings, e.scf)
+	newExp, err := newExporter(e.config, e.settings, e.scf, e.userAgent, e.netReporter)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create exporter: %w", err)
 	}

exporter/otelarrowexporter/otelarrow.go

@@ -5,12 +5,8 @@ package otelarrowexporter // import "github.com/open-telemetry/opentelemetry-col
 
 import (
 	"context"
-	"errors"
-	"fmt"
-	"runtime"
 	"time"
 
-	arrowPkg "github.com/apache/arrow/go/v16/arrow"
 	arrowRecord "github.com/open-telemetry/otel-arrow/pkg/otel/arrow_record"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configcompression"
@@ -38,7 +34,6 @@ import (
 )
 
 type exp interface {
-	helperOptions() []exporterhelper.Option
 	getSettings() exporter.Settings
 	getConfig() component.Config
 
@@ -79,27 +74,9 @@ type streamClientFactory func(conn *grpc.ClientConn) arrow.StreamClientFunc
 
 // Crete new exporter and start it. The exporter will begin connecting but
 // this function may return before the connection is established.
-func newExporter(cfg component.Config, set exporter.Settings, streamClientFactory streamClientFactory) (exp, error) {
+func newExporter(cfg component.Config, set exporter.Settings, streamClientFactory streamClientFactory, userAgent string, netReporter *netstats.NetworkReporter) (exp, error) {
 	oCfg := cfg.(*Config)
 
-	if oCfg.Endpoint == "" {
-		return nil, errors.New("OTLP exporter config requires an Endpoint")
-	}
-
-	netReporter, err := netstats.NewExporterNetworkReporter(set)
-	if err != nil {
-		return nil, err
-	}
-	userAgent := fmt.Sprintf("%s/%s (%s/%s)",
-		set.BuildInfo.Description, set.BuildInfo.Version, runtime.GOOS, runtime.GOARCH)
-
-	if !oCfg.Arrow.Disabled {
-		// Ignoring an error because Validate() was called.
-		_ = zstd.SetEncoderConfig(oCfg.Arrow.Zstd)
-
-		userAgent += fmt.Sprintf(" ApacheArrow/%s (NumStreams/%d)", arrowPkg.PkgVersion, oCfg.Arrow.NumStreams)
-	}
-
 	return &baseExporter{
 		config:              oCfg,
 		settings:            set,