move netreporter and fix race?

Author: moh-osman3

exporter/otelarrowexporter/metadata.go

@@ -25,6 +25,7 @@ import (
 	"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 (
@@ -40,6 +41,7 @@ type metadataExporter struct {
 
 	metadataKeys []string
 	exporters    sync.Map
+	netReporter    *netstats.NetworkReporter
 
 	userAgent string
 
@@ -53,6 +55,10 @@ var _ exp = (*metadataExporter)(nil)
 
 func newMetadataExporter(cfg component.Config, set exporter.Settings, streamClientFactory streamClientFactory) (exp, error) {
 	oCfg := cfg.(*Config)
+	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)
 
@@ -69,13 +75,15 @@ func newMetadataExporter(cfg component.Config, set exporter.Settings, streamClie
 	}
 	sort.Strings(mks)
 	if len(mks) == 0 {
-		return newExporter(cfg, set, streamClientFactory, userAgent)
+		return newExporter(cfg, set, streamClientFactory, userAgent, netReporter)
 	}
 	return &metadataExporter{
 		config:       oCfg,
 		settings:     set,
 		scf:          streamClientFactory,
 		metadataKeys: mks,
+		userAgent: userAgent,
+		netReporter: netReporter,
 	}, nil
 }
 
@@ -135,19 +143,19 @@ func (e *metadataExporter) pushLogs(ctx context.Context, ld plog.Logs) error {
 }
 
 func (e *metadataExporter) getOrCreateExporter(ctx context.Context, s attribute.Set, md metadata.MD) (exp, error) {
-	v, ok := e.exporters.Load(s)
-	if ok {
-		return v.(exp), nil
-	}
-
 	e.lock.Lock()
 	defer e.lock.Unlock()
 
 	if e.config.MetadataCardinalityLimit != 0 && e.size >= int(e.config.MetadataCardinalityLimit) {
 		return nil, errTooManyExporters
 	}
 
-	newExp, err := newExporter(e.config, e.settings, e.scf, e.userAgent)
+	v, ok := e.exporters.Load(s)
+	if ok {
+		return v.(exp), nil
+	}
+
+	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)
 	}
@@ -164,6 +172,7 @@ func (e *metadataExporter) getOrCreateExporter(ctx context.Context, s attribute.
 			e.exporters.Delete(s)
 			return nil, fmt.Errorf("failed to start exporter: %w", err)
 		}
+
 		e.size++
 	}
 

exporter/otelarrowexporter/metadata_test.go

@@ -48,6 +48,8 @@ func TestSendTracesWithMetadata(t *testing.T) {
 	cfg.MetadataCardinalityLimit = 10
 	cfg.MetadataKeys = []string{"key1", "key2"}
 	set := exportertest.NewNopSettings()
+	set.BuildInfo.Description = "Collector"
+	set.BuildInfo.Version = "1.2.3test"
 	bg := context.Background()
 	exp, err := factory.CreateTracesExporter(bg, set, cfg)
 	require.NoError(t, err)
@@ -57,8 +59,8 @@ func TestSendTracesWithMetadata(t *testing.T) {
 	}()
 
 	host := componenttest.NewNopHost()
+
 	assert.NoError(t, exp.Start(context.Background(), host))
-	time.Sleep(1 * time.Second)
 
 	// Ensure that initially there is no data in the receiver.
 	assert.EqualValues(t, 0, rcv.requestCount.Load())
@@ -164,8 +166,8 @@ func TestMetadataExporterCardinalityLimit(t *testing.T) {
 	}()
 
 	host := componenttest.NewNopHost()
+
 	assert.NoError(t, exp.Start(context.Background(), host))
-	time.Sleep(1 * time.Second)
 
 	// Ensure that initially there is no data in the receiver.
 	assert.EqualValues(t, 0, rcv.requestCount.Load())

exporter/otelarrowexporter/otelarrow.go

@@ -75,18 +75,13 @@ 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, userAgent string) (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
-	}
-
 	return &baseExporter{
 		config:              oCfg,
 		settings:            set,