fix(tracer): skip stats concentrator in OTLP export mode

ddtrace/tracer/civisibility_transport.go

@@ -33,8 +33,8 @@ const (
 	EvpProxyPath       = "evp_proxy/v2"       // Path for EVP proxy.
 )
 
-// Ensure that civisibilityTransport implements the transport interface.
-var _ transport = (*ciVisibilityTransport)(nil)
+// Ensure that ciVisibilityTransport implements the ddTransport interface.
+var _ ddTransport = (*ciVisibilityTransport)(nil)
 
 // ciVisibilityTransport is a structure that handles sending CI Visibility payloads
 // to the Datadog endpoint, either in agentless mode or through the EVP proxy.

ddtrace/tracer/civisibility_writer.go

@@ -108,7 +108,7 @@ func (w *ciVisibilityTraceWriter) flush() {
 		var err error
 
 		requestCompressedType := telemetry.UncompressedRequestCompressedType
-		if ciTransport, ok := w.config.transport.(*ciVisibilityTransport); ok && ciTransport.agentless {
+		if ciTransport, ok := w.config.ddTransport.(*ciVisibilityTransport); ok && ciTransport.agentless {
 			requestCompressedType = telemetry.CompressedRequestCompressedType
 		}
 		telemetry.EndpointPayloadRequests(telemetry.TestCycleEndpointType, requestCompressedType)
@@ -117,7 +117,7 @@ func (w *ciVisibilityTraceWriter) flush() {
 			stats := p.stats()
 			size, count = stats.size, stats.itemCount
 			log.Debug("ciVisibilityTraceWriter: sending payload: size: %d events: %d\n", size, count)
-			_, err = w.config.transport.send(p.payload)
+			_, err = w.config.ddTransport.send(p.payload)
 			if err == nil {
 				log.Debug("ciVisibilityTraceWriter: sent events after %d attempts", attempt+1)
 				return

ddtrace/tracer/civisibility_writer_test.go

@@ -91,7 +91,7 @@ func TestCiVisibilityTraceWriterFlushRetries(t *testing.T) {
 				assert:    assert,
 			}
 			c, err := newTestConfig(func(c *config) {
-				c.transport = p
+				c.ddTransport = p
 				c.sendRetries = test.configRetries
 				c.internalConfig.SetRetryInterval(test.retryInterval, internalconfig.OriginCode)
 			})

ddtrace/tracer/log.go

@@ -122,7 +122,7 @@ func logStartup(t *tracer) {
 	if srcURL := t.config.internalConfig.RawAgentURL(); srcURL != nil && srcURL.Scheme == "unix" {
 		agentURL = srcURL.String()
 	} else {
-		agentURL = t.config.transport.endpoint()
+		agentURL = t.config.ddTransport.endpoint()
 	}
 	info := startupInfo{
 		Date:                        time.Now().Format(time.RFC3339),
@@ -170,7 +170,7 @@ func logStartup(t *tracer) {
 		info.SampleRateLimit = fmt.Sprintf("%v", limit)
 	}
 	if !t.config.internalConfig.LogToStdout() {
-		if err := checkEndpoint(t.config.httpClient, t.config.transport.endpoint(), t.config.internalConfig.TraceProtocol()); err != nil {
+		if err := checkEndpoint(t.config.httpClient, t.config.ddTransport.endpoint(), t.config.internalConfig.TraceProtocol()); err != nil {
 			info.AgentError = err.Error()
 			log.Warn("DIAGNOSTICS Unable to reach agent intake: %s", err.Error())
 		}

ddtrace/tracer/option.go

@@ -174,8 +174,8 @@ type config struct {
 	// all spans.
 	globalTags dynamicConfig[map[string]any]
 
-	// transport specifies the Transport interface which will be used to send data to the agent.
-	transport transport
+	// ddTransport specifies the Datadog transport used to send msgpack traces and stats to the agent.
+	ddTransport ddTransport
 
 	// httpClientTimeout specifies the timeout for the HTTP client.
 	httpClientTimeout time.Duration
@@ -417,10 +417,10 @@ func newConfig(opts ...StartOption) (*config, error) {
 	} else {
 		globalconfig.SetServiceName(c.internalConfig.ServiceName())
 	}
-	if c.transport == nil {
+	if c.ddTransport == nil {
 		agentURL := c.internalConfig.AgentURL().String()
 		traceURL, headers := resolveTraceTransport(c.internalConfig)
-		c.transport = newHTTPTransport(traceURL, agentURL+statsAPIPath, c.httpClient, headers)
+		c.ddTransport = newHTTPTransport(traceURL, agentURL+statsAPIPath, c.httpClient, headers)
 	}
 	if c.propagator == nil {
 		envKey := "DD_TRACE_X_DATADOG_TAGS_MAX_LENGTH"
@@ -448,7 +448,7 @@ func newConfig(opts ...StartOption) (*config, error) {
 		c.httpClientTimeout = time.Second * 45                                 // Increase timeout up to 45 seconds (same as other tracers in CIVis mode)
 		c.internalConfig.SetLogStartup(false, internalconfig.OriginCalculated) // If we are in CI Visibility mode we don't want to log the startup to stdout to avoid polluting the output
 		ciTransport := newCiVisibilityTransport(c)                             // Create a default CI Visibility Transport
-		c.transport = ciTransport                                              // Replace the default transport with the CI Visibility transport
+		c.ddTransport = ciTransport                                            // Replace the default transport with the CI Visibility transport
 		c.ciVisibilityAgentless = ciTransport.agentless
 		c.ciVisibilityNoopTracer = internal.BoolEnv(constants.CIVisibilityUseNoopTracer, false)
 	}
@@ -461,7 +461,7 @@ func newConfig(opts ...StartOption) (*config, error) {
 	// If the agent doesn't support the v1 protocol, downgrade to v0.4
 	if c.internalConfig.TraceProtocol() == traceProtocolV1 && !af.v1ProtocolAvailable {
 		c.internalConfig.SetTraceProtocol(traceProtocolV04, internalconfig.OriginCalculated)
-		if t, ok := c.transport.(*httpTransport); ok {
+		if t, ok := c.ddTransport.(*httpTransport); ok {
 			t.traceURL = agentURL.String() + tracesAPIPath
 		}
 	}
@@ -529,12 +529,11 @@ func apmTracingDisabled(c *config) {
 	c.internalConfig.SetRuntimeMetricsV2Enabled(false, internalconfig.OriginCalculated)
 }
 
-// resolveTraceTransport returns the trace URL and headers for the transport
-// based on whether OTLP export mode is active.
+// resolveTraceTransport returns the trace URL and headers for the Datadog
+// agent transport. In OTLP export mode the ddTransport is not used for trace
+// sending (otlpTransport handles that), but it may still be used for stats
+// and agent discovery, so it always points at the DD agent.
 func resolveTraceTransport(cfg *internalconfig.Config) (traceURL string, headers map[string]string) {
-	if cfg.OTLPExportMode() {
-		return cfg.OTLPTraceURL(), cfg.OTLPHeaders()
-	}
 	agentURL := cfg.AgentURL().String()
 	traceURL = agentURL + tracesAPIPath
 	if cfg.TraceProtocol() == traceProtocolV1 {
@@ -1455,7 +1454,7 @@ func WithTestDefaults(statsdClient any) StartOption {
 			statsdClient = &statsd.NoOpClientDirect{}
 		}
 		c.statsdClient = statsdClient.(internal.StatsdClient)
-		c.transport = newDummyTransport()
+		c.ddTransport = newDummyTransport()
 	}
 }
 

ddtrace/tracer/option_test.go

@@ -38,9 +38,9 @@ import (
 	"github.com/DataDog/dd-trace-go/v2/internal/version"
 )
 
-func withTransport(t transport) StartOption {
+func withTransport(t ddTransport) StartOption {
 	return func(c *config) {
-		c.transport = t
+		c.ddTransport = t
 	}
 }
 

ddtrace/tracer/otlp_transport.go

@@ -0,0 +1,48 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016 Datadog, Inc.
+
+package tracer
+
+import (
+	"bytes"
+	"fmt"
+	"net/http"
+)
+
+// otlpTransport sends protobuf-encoded OTLP payloads over HTTP.
+// It is the OTLP counterpart to httpTransport (which handles Datadog-protocol traffic).
+type otlpTransport struct {
+	client   *http.Client
+	endpoint string
+	headers  map[string]string
+}
+
+func newOTLPTransport(client *http.Client, endpoint string, headers map[string]string) *otlpTransport {
+	return &otlpTransport{
+		client:   client,
+		endpoint: endpoint,
+		headers:  headers,
+	}
+}
+
+// send posts a protobuf-encoded payload to the configured OTLP endpoint.
+func (t *otlpTransport) send(data []byte) error {
+	req, err := http.NewRequest("POST", t.endpoint, bytes.NewReader(data))
+	if err != nil {
+		return fmt.Errorf("cannot create http request: %s", err)
+	}
+	for header, value := range t.headers {
+		req.Header.Set(header, value)
+	}
+	resp, err := t.client.Do(req)
+	if err != nil {
+		return err
+	}
+	resp.Body.Close()
+	if code := resp.StatusCode; code >= 400 {
+		return fmt.Errorf("HTTP %d: %s", code, http.StatusText(code))
+	}
+	return nil
+}

ddtrace/tracer/otlp_transport_test.go

@@ -0,0 +1,97 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2016 Datadog, Inc.
+
+package tracer
+
+import (
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestOTLPTransportSendSuccess(t *testing.T) {
+	var received []byte
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		received, _ = io.ReadAll(r.Body)
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer srv.Close()
+
+	tr := newOTLPTransport(srv.Client(), srv.URL, map[string]string{"Content-Type": "application/x-protobuf"})
+	err := tr.send([]byte("hello"))
+	require.NoError(t, err)
+	assert.Equal(t, []byte("hello"), received)
+}
+
+func TestOTLPTransportSendHeaders(t *testing.T) {
+	var gotHeaders http.Header
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		gotHeaders = r.Header.Clone()
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer srv.Close()
+
+	headers := map[string]string{
+		"Content-Type": "application/x-protobuf",
+		"Api-Key":      "secret",
+		"X-Custom":     "value",
+	}
+	tr := newOTLPTransport(srv.Client(), srv.URL, headers)
+	err := tr.send([]byte("data"))
+	require.NoError(t, err)
+
+	assert.Equal(t, "application/x-protobuf", gotHeaders.Get("Content-Type"))
+	assert.Equal(t, "secret", gotHeaders.Get("Api-Key"))
+	assert.Equal(t, "value", gotHeaders.Get("X-Custom"))
+}
+
+func TestOTLPTransportSendHTTPMethod(t *testing.T) {
+	var gotMethod string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		gotMethod = r.Method
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer srv.Close()
+
+	tr := newOTLPTransport(srv.Client(), srv.URL, nil)
+	err := tr.send([]byte("data"))
+	require.NoError(t, err)
+	assert.Equal(t, "POST", gotMethod)
+}
+
+func TestOTLPTransportSendErrorStatus(t *testing.T) {
+	tests := []struct {
+		code int
+		text string
+	}{
+		{http.StatusBadRequest, "Bad Request"},
+		{http.StatusUnauthorized, "Unauthorized"},
+		{http.StatusInternalServerError, "Internal Server Error"},
+		{http.StatusServiceUnavailable, "Service Unavailable"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.text, func(t *testing.T) {
+			srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				w.WriteHeader(tt.code)
+			}))
+			defer srv.Close()
+
+			tr := newOTLPTransport(srv.Client(), srv.URL, nil)
+			err := tr.send([]byte("data"))
+			require.Error(t, err)
+			assert.Contains(t, err.Error(), tt.text)
+		})
+	}
+}
+
+func TestOTLPTransportSendConnectionError(t *testing.T) {
+	tr := newOTLPTransport(http.DefaultClient, "http://127.0.0.1:0/nonexistent", nil)
+	err := tr.send([]byte("data"))
+	require.Error(t, err)
+}