[connector/datadog] - Use waitgroup instead of closing a channel and waiting on it (#40849)

VihasMakwana · web-flow · commit 20c2fd504588 · 2025-06-23T15:17:50.000-04:00
#### Description The `run()` goroutine closes the channel in defer. There are two sceanrios: 1. Best case scenario: no error encountered while calling `ConsumeMetrics`. 2. Worst case scenario: we encounter an error while calling `ConsumeMetrics`, causing the goroutine to exit early and as a result, close the `c.exit` chan. We close the channel, again, in `Shutdown` and this results in a "send on closed channel" panic in worst case scenario. In best case scenario, no panic is encountered. This PR adds waitgroup to avoid the panic.  #### Link to tracking issue Fixes #40845  #### Testing Added a test case
diff --git a/.chloggen/datadogconnector-panic.yaml b/.chloggen/datadogconnector-panic.yaml
@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: datadogconnector
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Use proper syncronization to fix a collector panic when an error occurs
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [40845]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []
diff --git a/connector/datadogconnector/connector_native.go b/connector/datadogconnector/connector_native.go
@@ -6,6 +6,7 @@ package datadogconnector // import "github.com/open-telemetry/opentelemetry-coll
 import (
 	"context"
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/DataDog/datadog-agent/comp/otelcol/otlp/components/statsprocessor"
@@ -29,6 +30,8 @@ type traceToMetricConnectorNative struct {
 	metricsConsumer consumer.Metrics // the next component in the pipeline to ingest metrics after connector
 	logger          *zap.Logger
 
+	wg sync.WaitGroup
+
 	// concentrator ingests spans and produces APM stats
 	concentrator *stats.Concentrator
 
@@ -100,6 +103,7 @@ func newTraceToMetricConnectorNative(set component.TelemetrySettings, cfg compon
 func (c *traceToMetricConnectorNative) Start(_ context.Context, _ component.Host) error {
 	c.logger.Info("Starting datadogconnector")
 	c.concentrator.Start()
+	c.wg.Add(1)
 	go c.run()
 	c.isStarted = true
 	return nil
@@ -117,8 +121,8 @@ func (c *traceToMetricConnectorNative) Shutdown(context.Context) error {
 	// stop the obfuscator and concentrator and wait for the run loop to exit
 	c.obfuscator.Stop()
 	c.concentrator.Stop()
-	c.exit <- struct{}{} // signal exit
-	<-c.exit             // wait for close
+	close(c.exit)
+	c.wg.Wait()
 	return nil
 }
 
@@ -139,7 +143,7 @@ func (c *traceToMetricConnectorNative) ConsumeTraces(_ context.Context, traces p
 // run awaits incoming stats resulting from the agent's ingestion, converts them
 // to metrics and flushes them using the configured metrics exporter.
 func (c *traceToMetricConnectorNative) run() {
-	defer close(c.exit)
+	defer c.wg.Done()
 	for {
 		select {
 		case stats := <-c.statsout:
diff --git a/connector/datadogconnector/connector_native_test.go b/connector/datadogconnector/connector_native_test.go
@@ -5,6 +5,7 @@ package datadogconnector
 
 import (
 	"context"
+	"errors"
 	"sort"
 	"testing"
 	"time"
@@ -357,3 +358,18 @@ func TestObfuscate(t *testing.T) {
 		t.Errorf("Diff between APM stats -want +got:\n%v", diff)
 	}
 }
+
+func TestNoPanic(t *testing.T) {
+	c, _ := creteConnectorNative(t)
+	c.metricsConsumer = consumertest.NewErr(errors.New("error"))
+	require.NoError(t, c.Start(context.Background(), componenttest.NewNopHost()))
+	trace1 := generateTrace()
+
+	err := c.ConsumeTraces(context.Background(), trace1)
+	assert.NoError(t, err)
+
+	time.Sleep(2 * time.Second)
+
+	err = c.Shutdown(context.Background())
+	require.NoError(t, err)
+}
