[chore][exporter/datadogexporter] fix data race in tests again (open-telemetry#39802)

#### Description
Having both the otlp grpc exporter & otlp grpc recevier in the tests
introduces data race, so move all tests to use HTTP instead. Also make
sure to properly close all go routines before tests complete.

#### Link to tracking issue
Fixes
open-telemetry#39782
Fixes
open-telemetry#39780
Fixes
open-telemetry#39519
Fixes
http://github.com/open-telemetry/opentelemetry-collector-contrib/issues/39804

Author: songy23

exporter/datadogexporter/integrationtest/go.mod

@@ -29,8 +29,8 @@ require (
 	go.opentelemetry.io/collector/receiver v1.31.1-0.20250509190408-4ca0f1829e0a
 	go.opentelemetry.io/collector/receiver/otlpreceiver v0.125.1-0.20250509190408-4ca0f1829e0a
 	go.opentelemetry.io/otel v1.35.0
-	go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc v0.11.0
-	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.35.0
+	go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp v0.11.0
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.35.0
 	go.opentelemetry.io/otel/sdk v1.35.0
 	go.opentelemetry.io/otel/sdk/log v0.11.0
 	go.opentelemetry.io/otel/trace v1.35.0
@@ -338,11 +338,11 @@ require (
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0 // indirect
 	go.opentelemetry.io/contrib/otelconf v0.15.0 // indirect
 	go.opentelemetry.io/contrib/propagators/b3 v1.35.0 // indirect
-	go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp v0.11.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc v0.11.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.35.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.35.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.35.0 // indirect
-	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.35.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.35.0 // indirect
 	go.opentelemetry.io/otel/exporters/prometheus v0.57.0 // indirect
 	go.opentelemetry.io/otel/exporters/stdout/stdoutlog v0.11.0 // indirect
 	go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.35.0 // indirect

exporter/datadogexporter/integrationtest/integration_test.go

@@ -39,8 +39,8 @@ import (
 	"go.opentelemetry.io/collector/receiver/otlpreceiver"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc"
-	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
+	"go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp"
+	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
 	"go.opentelemetry.io/otel/sdk/log"
 	"go.opentelemetry.io/otel/sdk/resource"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
@@ -73,46 +73,35 @@ type point struct {
 	Value     float64
 }
 
-func TestIntegration_NativeOTelAPMStatsIngest(t *testing.T) {
-	previousVal := datadogconnector.NativeIngestFeatureGate.IsEnabled()
-	err := featuregate.GlobalRegistry().Set(datadogconnector.NativeIngestFeatureGate.ID(), true)
-	require.NoError(t, err)
-	defer func() {
-		err = featuregate.GlobalRegistry().Set(datadogconnector.NativeIngestFeatureGate.ID(), previousVal)
-		require.NoError(t, err)
-	}()
-
-	testIntegration(t)
-}
-
-func TestIntegration_LegacyOTelAPMStatsIngest(t *testing.T) {
-	testIntegration(t)
-}
-
-func testIntegration(t *testing.T) {
+func TestIntegration(t *testing.T) {
 	// 1. Set up mock Datadog server
 	// See also https://github.com/DataDog/datadog-agent/blob/49c16e0d4deab396626238fa1d572b684475a53f/cmd/trace-agent/test/backend.go
 	apmstatsRec := &testutil.HTTPRequestRecorderWithChan{Pattern: testutil.APMStatsEndpoint, ReqChan: make(chan []byte)}
 	tracesRec := &testutil.HTTPRequestRecorderWithChan{Pattern: testutil.TraceEndpoint, ReqChan: make(chan []byte)}
 	server := testutil.DatadogServerMock(apmstatsRec.HandlerFunc, tracesRec.HandlerFunc)
 	defer server.Close()
 	t.Setenv("SERVER_URL", server.URL)
-	t.Setenv("OTLP_HTTP_SERVER", commonTestutil.GetAvailableLocalAddress(t))
-	otlpGRPCEndpoint := commonTestutil.GetAvailableLocalAddress(t)
-	t.Setenv("OTLP_GRPC_SERVER", otlpGRPCEndpoint)
+	otlpEndpoint := commonTestutil.GetAvailableLocalAddress(t)
+	t.Setenv("OTLP_HTTP_SERVER", otlpEndpoint)
 
 	// 2. Start in-process collector
 	factories := getIntegrationTestComponents(t)
 	app := getIntegrationTestCollector(t, "integration_test_config.yaml", factories)
+	var wg sync.WaitGroup
+	wg.Add(1)
 	go func() {
-		assert.NoError(t, app.Run(context.Background()))
+		_ = app.Run(context.Background()) // ignore shutdown error, core collector has race in shutdown: https://github.com/open-telemetry/opentelemetry-collector/issues/12944
+		wg.Done()
+	}()
+	defer func() {
+		app.Shutdown()
+		wg.Wait()
 	}()
-	defer app.Shutdown()
 
 	waitForReadiness(app)
 
 	// 3. Generate and send traces
-	sendTraces(t, otlpGRPCEndpoint)
+	sendTraces(t, otlpEndpoint)
 
 	// 4. Validate traces and APM stats from the mock server
 	var spans []*pb.Span
@@ -233,7 +222,7 @@ func sendTraces(t *testing.T, endpoint string) {
 	ctx := context.Background()
 
 	// Set up OTel-Go SDK and exporter
-	traceExporter, err := otlptracegrpc.New(ctx, otlptracegrpc.WithInsecure(), otlptracegrpc.WithEndpoint(endpoint))
+	traceExporter, err := otlptracehttp.New(ctx, otlptracehttp.WithInsecure(), otlptracehttp.WithEndpoint(endpoint))
 	require.NoError(t, err)
 	bsp := sdktrace.NewBatchSpanProcessor(traceExporter)
 	r1, _ := resource.New(ctx, resource.WithAttributes(attribute.String("k8s.node.name", "aaaa")))
@@ -290,22 +279,27 @@ func TestIntegrationComputeTopLevelBySpanKind(t *testing.T) {
 	server := testutil.DatadogServerMock(apmstatsRec.HandlerFunc, tracesRec.HandlerFunc)
 	defer server.Close()
 	t.Setenv("SERVER_URL", server.URL)
-	t.Setenv("OTLP_HTTP_SERVER", commonTestutil.GetAvailableLocalAddress(t))
-	otlpGRPCEndpoint := commonTestutil.GetAvailableLocalAddress(t)
-	t.Setenv("OTLP_GRPC_SERVER", otlpGRPCEndpoint)
+	otlpEndpoint := commonTestutil.GetAvailableLocalAddress(t)
+	t.Setenv("OTLP_HTTP_SERVER", otlpEndpoint)
 
 	// 2. Start in-process collector
 	factories := getIntegrationTestComponents(t)
 	app := getIntegrationTestCollector(t, "integration_test_toplevel_config.yaml", factories)
+	var wg sync.WaitGroup
+	wg.Add(1)
 	go func() {
-		assert.NoError(t, app.Run(context.Background()))
+		_ = app.Run(context.Background()) // ignore shutdown error, core collector has race in shutdown: https://github.com/open-telemetry/opentelemetry-collector/issues/12944
+		wg.Done()
+	}()
+	defer func() {
+		app.Shutdown()
+		wg.Wait()
 	}()
-	defer app.Shutdown()
 
 	waitForReadiness(app)
 
 	// 3. Generate and send traces
-	sendTracesComputeTopLevelBySpanKind(t, otlpGRPCEndpoint)
+	sendTracesComputeTopLevelBySpanKind(t, otlpEndpoint)
 
 	// 4. Validate traces and APM stats from the mock server
 	var spans []*pb.Span
@@ -397,7 +391,7 @@ func sendTracesComputeTopLevelBySpanKind(t *testing.T, endpoint string) {
 	ctx := context.Background()
 
 	// Set up OTel-Go SDK and exporter
-	traceExporter, err := otlptracegrpc.New(ctx, otlptracegrpc.WithInsecure(), otlptracegrpc.WithEndpoint(endpoint))
+	traceExporter, err := otlptracehttp.New(ctx, otlptracehttp.WithInsecure(), otlptracehttp.WithEndpoint(endpoint))
 	require.NoError(t, err)
 	bsp := sdktrace.NewBatchSpanProcessor(traceExporter)
 	r1, _ := resource.New(ctx, resource.WithAttributes(attribute.String("k8s.node.name", "aaaa")))
@@ -478,17 +472,16 @@ func TestIntegrationLogs(t *testing.T) {
 	})
 	defer server.Close()
 	t.Setenv("SERVER_URL", server.URL)
-	t.Setenv("OTLP_HTTP_SERVER", commonTestutil.GetAvailableLocalAddress(t))
-	otlpGRPCEndpoint := commonTestutil.GetAvailableLocalAddress(t)
-	t.Setenv("OTLP_GRPC_SERVER", otlpGRPCEndpoint)
+	otlpEndpoint := commonTestutil.GetAvailableLocalAddress(t)
+	t.Setenv("OTLP_HTTP_SERVER", otlpEndpoint)
 
 	// 2. Start in-process collector
 	factories := getIntegrationTestComponents(t)
 	app := getIntegrationTestCollector(t, "integration_test_logs_config.yaml", factories)
 	var wg sync.WaitGroup
 	wg.Add(1)
 	go func() {
-		_ = app.Run(context.Background()) // ignore shutdown error
+		_ = app.Run(context.Background()) // ignore shutdown error, core collector has race in shutdown: https://github.com/open-telemetry/opentelemetry-collector/issues/12944
 		wg.Done()
 	}()
 	defer func() {
@@ -499,7 +492,7 @@ func TestIntegrationLogs(t *testing.T) {
 	waitForReadiness(app)
 
 	// 3. Generate and send logs
-	sendLogs(t, 5, otlpGRPCEndpoint)
+	sendLogs(t, 5, otlpEndpoint)
 
 	// 4. Validate logs and metrics from the mock server
 	// Wait until `doneChannel` is closed and internal metrics are received.
@@ -545,7 +538,7 @@ func TestIntegrationLogs(t *testing.T) {
 
 func sendLogs(t *testing.T, numLogs int, endpoint string) {
 	ctx := context.Background()
-	logExporter, err := otlploggrpc.New(ctx, otlploggrpc.WithInsecure(), otlploggrpc.WithEndpoint(endpoint))
+	logExporter, err := otlploghttp.New(ctx, otlploghttp.WithInsecure(), otlploghttp.WithEndpoint(endpoint))
 	assert.NoError(t, err)
 	lr := make([]log.Record, numLogs)
 	assert.NoError(t, logExporter.Export(ctx, lr))
@@ -642,10 +635,16 @@ func testIntegrationHostMetrics(t *testing.T, expectedMetrics map[string]struct{
 	// 2. Start in-process collector
 	factories := getIntegrationTestComponents(t)
 	app := getIntegrationTestCollector(t, "integration_test_host_metrics_config.yaml", factories)
+	var wg sync.WaitGroup
+	wg.Add(1)
 	go func() {
-		assert.NoError(t, app.Run(context.Background()))
+		_ = app.Run(context.Background()) // ignore shutdown error, core collector has race in shutdown: https://github.com/open-telemetry/opentelemetry-collector/issues/12944
+		wg.Done()
+	}()
+	defer func() {
+		app.Shutdown()
+		wg.Wait()
 	}()
-	defer app.Shutdown()
 
 	waitForReadiness(app)
 
@@ -730,3 +729,95 @@ func seriesFromAPIClient(t *testing.T, metricsBytes []byte, expectedMetrics map[
 	}
 	return metricMap, nil
 }
+
+func TestIntegrationInternalMetrics(t *testing.T) {
+	require.NoError(t, featuregate.GlobalRegistry().Set("exporter.datadogexporter.metricexportserializerclient", false))
+	defer func() {
+		require.NoError(t, featuregate.GlobalRegistry().Set("exporter.datadogexporter.metricexportserializerclient", true))
+	}()
+	expectedMetrics := map[string]struct{}{
+		// Datadog internal metrics on trace and stats writers
+		"datadog.otlp_translator.resources.missing_source": {},
+		"datadog.trace_agent.stats_writer.bytes":           {},
+		"datadog.trace_agent.stats_writer.retries":         {},
+		"datadog.trace_agent.stats_writer.stats_buckets":   {},
+		"datadog.trace_agent.stats_writer.stats_entries":   {},
+		"datadog.trace_agent.stats_writer.payloads":        {},
+		"datadog.trace_agent.stats_writer.client_payloads": {},
+		"datadog.trace_agent.stats_writer.errors":          {},
+		"datadog.trace_agent.stats_writer.splits":          {},
+		"datadog.trace_agent.trace_writer.bytes":           {},
+		"datadog.trace_agent.trace_writer.retries":         {},
+		"datadog.trace_agent.trace_writer.spans":           {},
+		"datadog.trace_agent.trace_writer.traces":          {},
+		"datadog.trace_agent.trace_writer.payloads":        {},
+		"datadog.trace_agent.trace_writer.errors":          {},
+		"datadog.trace_agent.trace_writer.events":          {},
+
+		// OTel collector internal metrics
+		"otelcol_process_memory_rss":                     {},
+		"otelcol_process_runtime_total_sys_memory_bytes": {},
+		"otelcol_process_uptime":                         {},
+		"otelcol_process_cpu_seconds":                    {},
+		"otelcol_process_runtime_heap_alloc_bytes":       {},
+		"otelcol_process_runtime_total_alloc_bytes":      {},
+		"otelcol_receiver_accepted_metric_points":        {},
+		"otelcol_receiver_accepted_spans":                {},
+		"otelcol_exporter_queue_capacity":                {},
+		"otelcol_exporter_queue_size":                    {},
+		"otelcol_exporter_sent_spans":                    {},
+		"otelcol_exporter_sent_metric_points":            {},
+	}
+	testIntegrationInternalMetrics(t, expectedMetrics)
+}
+
+func testIntegrationInternalMetrics(t *testing.T, expectedMetrics map[string]struct{}) {
+	// 1. Set up mock Datadog server
+	seriesRec := &testutil.HTTPRequestRecorderWithChan{Pattern: testutil.MetricV2Endpoint, ReqChan: make(chan []byte, 100)}
+	tracesRec := &testutil.HTTPRequestRecorderWithChan{Pattern: testutil.TraceEndpoint, ReqChan: make(chan []byte, 100)}
+	server := testutil.DatadogServerMock(seriesRec.HandlerFunc, tracesRec.HandlerFunc)
+	defer server.Close()
+	t.Setenv("SERVER_URL", server.URL)
+	otlpEndpoint := commonTestutil.GetAvailableLocalAddress(t)
+	t.Setenv("OTLP_HTTP_SERVER", otlpEndpoint)
+
+	// 2. Start in-process collector
+	factories := getIntegrationTestComponents(t)
+	app := getIntegrationTestCollector(t, "integration_test_internal_metrics_config.yaml", factories)
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		_ = app.Run(context.Background()) // ignore shutdown error, core collector has race in shutdown: https://github.com/open-telemetry/opentelemetry-collector/issues/12944
+		wg.Done()
+	}()
+	defer func() {
+		app.Shutdown()
+		wg.Wait()
+	}()
+
+	waitForReadiness(app)
+
+	// 3. Generate and send traces
+	sendTraces(t, otlpEndpoint)
+
+	// 4. Validate Datadog trace agent & OTel internal metrics are sent to the mock server
+	metricMap := make(map[string]series)
+	for len(metricMap) < len(expectedMetrics) {
+		select {
+		case <-tracesRec.ReqChan:
+			// Drain the channel, no need to look into the traces
+		case metricsBytes := <-seriesRec.ReqChan:
+			var metrics seriesSlice
+			gz := getGzipReader(t, metricsBytes)
+			dec := json.NewDecoder(gz)
+			assert.NoError(t, dec.Decode(&metrics))
+			for _, s := range metrics.Series {
+				if _, ok := expectedMetrics[s.Metric]; ok {
+					metricMap[s.Metric] = s
+				}
+			}
+		case <-time.After(60 * time.Second):
+			t.Fail()
+		}
+	}
+}

exporter/datadogexporter/integrationtest/integration_test_config.yaml

@@ -4,8 +4,6 @@ receivers:
     protocols:
       http:
         endpoint: ${env:OTLP_HTTP_SERVER}
-      grpc:
-        endpoint: ${env:OTLP_GRPC_SERVER}
 
 processors:
   tail_sampling:
@@ -25,7 +23,6 @@ connectors:
 
 exporters:
   debug:
-    verbosity: detailed
   datadog:
     api:
       key: "aaa"

exporter/datadogexporter/integrationtest/integration_test_internal_metrics_config.yaml

@@ -4,8 +4,6 @@ receivers:
     protocols:
       http:
         endpoint: ${env:OTLP_HTTP_SERVER}
-      grpc:
-        endpoint: ${env:OTLP_GRPC_SERVER}
 
 exporters:
   datadog:

exporter/datadogexporter/integrationtest/integration_test_logs_config.yaml

@@ -4,8 +4,6 @@ receivers:
     protocols:
       http:
         endpoint: ${env:OTLP_HTTP_SERVER}
-      grpc:
-        endpoint: ${env:OTLP_GRPC_SERVER}
 
 exporters:
   datadog:

exporter/datadogexporter/integrationtest/integration_test_toplevel_config.yaml

@@ -4,8 +4,6 @@ receivers:
     protocols:
       http:
         endpoint: ${env:OTLP_HTTP_SERVER}
-      grpc:
-        endpoint: ${env:OTLP_GRPC_SERVER}
 
 connectors:
   datadog/connector:
@@ -14,7 +12,6 @@ connectors:
 
 exporters:
   debug:
-    verbosity: detailed
   datadog:
     api:
       key: "aaa"