From eb9956d2e926411f25f9cdbe5c240259b8a0d931 Mon Sep 17 00:00:00 2001 From: Rachel Yang Date: Tue, 16 Jun 2026 11:44:40 -0400 Subject: [PATCH 01/17] feat(tracer): wire OTLP span metrics exporter into concentrator flush path (PR 4) - Add otlpExporter field to concentrator; flushAndSend branches to OTLP or native /v0.6/stats - Add newOTLPMetricsConcentrator constructor; select it in newUnstartedTracer when OTLPSpanMetricsEnabled - FR15: set Datadog-Client-Computed-Stats: yes header when OTLPSpanMetricsEnabled - FR15: append _dd.stats_computed=true resource attr on OTLP trace payloads - Fix sketchToHistogram to decode protobuf-serialized DDSketches (agent format) - Add SetOTLPSpanMetricsEnabled setter in internal/config for test wiring Co-Authored-By: Claude Sonnet 4.6 --- ddtrace/tracer/otlp_metrics_wireup_test.go | 172 +++++++++++++++++++ ddtrace/tracer/otlp_writer.go | 7 + ddtrace/tracer/stats.go | 54 ++++-- ddtrace/tracer/stats_to_otlp_metrics_test.go | 1 + ddtrace/tracer/tracer.go | 69 ++++---- ddtrace/tracer/transport.go | 8 +- ddtrace/tracer/transport_test.go | 4 +- internal/config/config.go | 17 ++ 8 files changed, 282 insertions(+), 50 deletions(-) create mode 100644 ddtrace/tracer/otlp_metrics_wireup_test.go diff --git a/ddtrace/tracer/otlp_metrics_wireup_test.go b/ddtrace/tracer/otlp_metrics_wireup_test.go new file mode 100644 index 00000000000..adc7326c2d9 --- /dev/null +++ b/ddtrace/tracer/otlp_metrics_wireup_test.go @@ -0,0 +1,172 @@ +// 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 2025 Datadog, Inc. + +package tracer + +import ( + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "github.com/DataDog/datadog-go/v5/statsd" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + internalconfig "github.com/DataDog/dd-trace-go/v2/internal/config" +) + +// captureMetricsServer records every POST body it receives. +type captureMetricsServer struct { + mu sync.Mutex + bodies [][]byte + *httptest.Server +} + +func newCaptureMetricsServer(t *testing.T) *captureMetricsServer { + t.Helper() + cs := &captureMetricsServer{} + cs.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + b, _ := io.ReadAll(r.Body) + cs.mu.Lock() + cs.bodies = append(cs.bodies, b) + cs.mu.Unlock() + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(cs.Server.Close) + return cs +} + +func (cs *captureMetricsServer) receivedBodies() [][]byte { + cs.mu.Lock() + defer cs.mu.Unlock() + out := make([][]byte, len(cs.bodies)) + copy(out, cs.bodies) + return out +} + +// TestOTLPMetricsConcentratorRoutesToExporter verifies that a concentrator wired +// with an otlpMetricsExporter routes flushed stats to the OTLP endpoint and does +// not use the agent's native /v0.6/stats path. +func TestOTLPMetricsConcentratorRoutesToExporter(t *testing.T) { + otlpSrv := newCaptureMetricsServer(t) + + dt := newDummyTransport() + cfg, err := newTestConfig(withNoopInfoHTTPClient(), func(c *config) { + c.ddTransport = dt + c.internalConfig.SetEnv("prod", internalconfig.OriginCode) + }) + require.NoError(t, err) + + bucketSize := int64(500_000) + c := newConcentrator(cfg, bucketSize, &statsd.NoOpClientDirect{}) + c.otlpExporter = &otlpMetricsExporter{ + client: otlpSrv.Server.Client(), + url: otlpSrv.URL + "/v1/metrics", + protocol: "http/json", + cfg: cfg.internalConfig, + } + + s := &Span{ + name: "http.request", + service: "test-svc", + resource: "/api/v1", + // 30 seconds in the past ensures its bucket is always before the flush window. + start: time.Now().UnixNano() - int64(30*time.Second), + duration: int64(50 * time.Millisecond), + metrics: map[string]float64{keyMeasured: 1}, + } + ss, ok := c.newTracerStatSpan(s, nil) + require.True(t, ok) + + // Add the span and flush directly, bypassing goroutine channels to keep the + // test deterministic. + c.add(ss) + c.flushAndSend(time.Now(), withCurrentBucket) + + bodies := otlpSrv.receivedBodies() + require.NotEmpty(t, bodies, "OTLP endpoint must receive at least one payload") + + // Verify the payload is valid JSON with a resourceMetrics array. + var parsed map[string]any + require.NoError(t, json.Unmarshal(bodies[0], &parsed), "OTLP metrics payload must be valid JSON") + rm, ok := parsed["resourceMetrics"].([]any) + require.True(t, ok, "expected resourceMetrics array in OTLP payload") + require.NotEmpty(t, rm) + + // The native /v0.6/stats path must not have been used. + assert.Empty(t, dt.Stats(), "native stats path must not be used when otlpExporter is set") +} + +// TestOTLPSpanMetricsHeaderOnNativeTraces verifies that when OTLP span metrics are +// enabled the Datadog-Client-Computed-Stats: yes header is present on native trace +// payloads so the agent does not recompute stats. +func TestOTLPSpanMetricsHeaderOnNativeTraces(t *testing.T) { + var headerValue string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/info" { + // No /v0.6/stats endpoint, so CanComputeStats stays false. + // The header must still be set via OTLPSpanMetricsEnabled. + w.Write([]byte(`{"endpoints":[]}`)) + return + } + headerValue = r.Header.Get("Datadog-Client-Computed-Stats") + })) + defer srv.Close() + + trc, err := newTracer( + WithAgentAddr(srv.Listener.Addr().String()), + func(c *config) { + c.internalConfig.SetOTLPSpanMetricsEnabled(true, internalconfig.OriginCode) + }, + ) + require.NoError(t, err) + setGlobalTracer(trc) + defer trc.Stop() + + p, err := encode(getTestTrace(1, 1)) + require.NoError(t, err) + _, err = trc.config.ddTransport.send(p) + require.NoError(t, err) + + assert.Equal(t, "yes", headerValue, "Datadog-Client-Computed-Stats must be 'yes' when OTLPSpanMetricsEnabled") +} + +// TestOTLPTraceWriterStatsComputedResourceAttr verifies that _dd.stats_computed=true +// is added to the OTLP trace resource when OTLP span metrics are enabled (FR15). +func TestOTLPTraceWriterStatsComputedResourceAttr(t *testing.T) { + t.Run("present-when-enabled", func(t *testing.T) { + cfg, err := newTestConfig(func(c *config) { + c.internalConfig.SetOTLPSpanMetricsEnabled(true, internalconfig.OriginCode) + }) + require.NoError(t, err) + + w := newOTLPTraceWriter(cfg) + var found bool + for _, kv := range w.resource.Attributes { + if kv.Key == "_dd.stats_computed" { + found = true + assert.True(t, kv.Value.GetBoolValue(), "_dd.stats_computed must be true") + } + } + assert.True(t, found, "_dd.stats_computed attribute must be present when OTLPSpanMetricsEnabled") + }) + + t.Run("absent-when-disabled", func(t *testing.T) { + cfg, err := newTestConfig(func(c *config) { + c.internalConfig.SetOTLPSpanMetricsEnabled(false, internalconfig.OriginCode) + }) + require.NoError(t, err) + + w := newOTLPTraceWriter(cfg) + for _, kv := range w.resource.Attributes { + assert.NotEqual(t, "_dd.stats_computed", kv.Key, + "_dd.stats_computed must not be present when OTLPSpanMetricsEnabled is false") + } + }) +} diff --git a/ddtrace/tracer/otlp_writer.go b/ddtrace/tracer/otlp_writer.go index 5f4398d3250..fcd423b638f 100644 --- a/ddtrace/tracer/otlp_writer.go +++ b/ddtrace/tracer/otlp_writer.go @@ -38,6 +38,13 @@ type otlpTraceWriter struct { func newOTLPTraceWriter(c *config) *otlpTraceWriter { resource := buildResource(c.internalConfig) + // FR15: signal to the backend that the SDK has computed stats so it does + // not recompute them from the trace spans. + if c.internalConfig.OTLPSpanMetricsEnabled() { + resource.Attributes = append(resource.Attributes, + otlpKeyValue("_dd.stats_computed", otlpBoolValue(true)), + ) + } scope := &otlpcommon.InstrumentationScope{Name: "dd-trace-go", Version: version.Tag} baseSize := proto.Size(&otlptrace.TracesData{ ResourceSpans: []*otlptrace.ResourceSpans{{ diff --git a/ddtrace/tracer/stats.go b/ddtrace/tracer/stats.go index dc59cb59080..112666bfbba 100644 --- a/ddtrace/tracer/stats.go +++ b/ddtrace/tracer/stats.go @@ -63,6 +63,10 @@ type concentrator struct { stop chan struct{} // closing this channel triggers shutdown cfg *config // tracer startup configuration statsdClient internal.StatsdClient // statsd client for sending metrics. + + // otlpExporter, when non-nil, routes flushed stats to the OTLP metrics + // endpoint instead of the agent's native /v0.6/stats path. + otlpExporter *otlpMetricsExporter } type tracerStatSpan struct { @@ -249,16 +253,10 @@ const ( // flushAndSend flushes all the stats buckets with the given timestamp and sends them using the transport specified in // the concentrator config. The current bucket is only included if includeCurrent is true, such as during shutdown. +// When an OTLP exporter is configured, stats are sent to the OTLP metrics endpoint; otherwise they are sent to the +// agent's native /v0.6/stats path. func (c *concentrator) flushAndSend(timenow time.Time, includeCurrent bool) { csps := c.spanConcentrator.Flush(timenow.UnixNano(), includeCurrent) - - obfVersion := 0 - if c.shouldObfuscate() { - obfVersion = tracerObfuscationVersion - } else { - log.Debug("Stats Obfuscation was skipped, agent will obfuscate (tracer %d, agent %d)", tracerObfuscationVersion, c.cfg.agent.load().obfuscationVersion) - } - if len(csps) == 0 { // nothing to flush return @@ -275,13 +273,31 @@ func (c *concentrator) flushAndSend(timenow time.Time, includeCurrent bool) { csp.ProcessTags = processtags.GlobalTags().String() flushedBuckets += len(csp.Stats) var err error - for attempt := 0; attempt <= sendRetries; attempt++ { - err = c.cfg.ddTransport.sendStats(csp, obfVersion) - if err == nil { - break + if c.otlpExporter != nil { + for attempt := 0; attempt <= sendRetries; attempt++ { + err = c.otlpExporter.export(csp) + if err == nil { + break + } + if attempt < sendRetries { + time.Sleep(retryInterval) + } + } + } else { + obfVersion := 0 + if c.shouldObfuscate() { + obfVersion = tracerObfuscationVersion + } else { + log.Debug("Stats Obfuscation was skipped, agent will obfuscate (tracer %d, agent %d)", tracerObfuscationVersion, c.cfg.agent.load().obfuscationVersion) } - if attempt < sendRetries { - time.Sleep(retryInterval) + for attempt := 0; attempt <= sendRetries; attempt++ { + err = c.cfg.ddTransport.sendStats(csp, obfVersion) + if err == nil { + break + } + if attempt < sendRetries { + time.Sleep(retryInterval) + } } } if err != nil { @@ -292,6 +308,16 @@ func (c *concentrator) flushAndSend(timenow time.Time, includeCurrent bool) { c.statsd().Incr("datadog.tracer.stats.flush_buckets", nil, float64(flushedBuckets)) } +// newOTLPMetricsConcentrator creates a concentrator that exports flushed stats to +// the OTLP metrics endpoint instead of the agent's native /v0.6/stats path. +// The flush interval is taken from OTLPMetricsFlushInterval in cfg. +func newOTLPMetricsConcentrator(c *config, statsdClient internal.StatsdClient) *concentrator { + bucketSize := c.internalConfig.OTLPMetricsFlushInterval().Nanoseconds() + conc := newConcentrator(c, bucketSize, statsdClient) + conc.otlpExporter = newOTLPMetricsExporter(c.internalConfig) + return conc +} + // trySendSpan attempts a non-blocking send of the stat span to the // concentrator's input channel. func (c *concentrator) trySendSpan(s *tracerStatSpan) { diff --git a/ddtrace/tracer/stats_to_otlp_metrics_test.go b/ddtrace/tracer/stats_to_otlp_metrics_test.go index 85c35d63466..56bf35ed008 100644 --- a/ddtrace/tracer/stats_to_otlp_metrics_test.go +++ b/ddtrace/tracer/stats_to_otlp_metrics_test.go @@ -12,6 +12,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" pb "github.com/DataDog/datadog-agent/pkg/proto/pbgo/trace" ddsketch "github.com/DataDog/sketches-go/ddsketch" diff --git a/ddtrace/tracer/tracer.go b/ddtrace/tracer/tracer.go index 10c6d389038..9cb706c5eb8 100644 --- a/ddtrace/tracer/tracer.go +++ b/ddtrace/tracer/tracer.go @@ -49,19 +49,20 @@ import ( ) type TracerConf struct { //nolint:revive - CanComputeStats bool - CanDropP0s bool - DebugAbandonedSpans bool - Disabled bool - PartialFlush bool - PartialFlushMinSpans int - PeerServiceDefaults bool - PeerServiceMappings map[string]string - EnvTag string - VersionTag string - ServiceTag string - TracingAsTransport bool - isLambdaFunction bool + CanComputeStats bool + CanDropP0s bool + DebugAbandonedSpans bool + Disabled bool + PartialFlush bool + PartialFlushMinSpans int + PeerServiceDefaults bool + PeerServiceMappings map[string]string + EnvTag string + VersionTag string + ServiceTag string + TracingAsTransport bool + isLambdaFunction bool + OTLPSpanMetricsEnabled bool } // Tracer specifies an implementation of the Datadog tracer which allows starting @@ -514,9 +515,15 @@ func newUnstartedTracer(opts ...StartOption) (t *tracer, err error) { c.internalConfig.SetLogDirectory("", telemetry.OriginCalculated) } } - var sc statsConcentrator = newConcentrator(c, defaultStatsBucketSize, statsd) - if c.internalConfig.OTLPExportMode() { + var sc statsConcentrator + if c.internalConfig.OTLPSpanMetricsEnabled() { + // OTLP span metrics: SDK computes stats and exports them via OTLP. + // The concentrator never forwards to the agent's /v0.6/stats path. + sc = newOTLPMetricsConcentrator(c, statsd) + } else if c.internalConfig.OTLPExportMode() { sc = &noopConcentrator{} + } else { + sc = newConcentrator(c, defaultStatsBucketSize, statsd) } t = &tracer{ config: c, @@ -1212,19 +1219,20 @@ func (t *tracer) Extract(carrier any) (*SpanContext, error) { func (t *tracer) TracerConf() TracerConf { pfEnabled, pfMin := t.config.internalConfig.PartialFlushEnabled() return TracerConf{ - CanComputeStats: t.config.canComputeStats(), - CanDropP0s: t.config.canDropP0s(), - DebugAbandonedSpans: t.config.internalConfig.DebugAbandonedSpans(), - Disabled: !t.config.internalConfig.TracingEnabled(), - PartialFlush: pfEnabled, - PartialFlushMinSpans: pfMin, - PeerServiceDefaults: t.config.internalConfig.PeerServiceDefaultsEnabled(), - PeerServiceMappings: t.config.internalConfig.PeerServiceMappings(), - EnvTag: t.config.internalConfig.Env(), - VersionTag: t.config.internalConfig.Version(), - ServiceTag: t.config.internalConfig.ServiceName(), - TracingAsTransport: t.config.tracingAsTransport, - isLambdaFunction: t.config.internalConfig.IsLambdaFunction(), + CanComputeStats: t.config.canComputeStats(), + CanDropP0s: t.config.canDropP0s(), + DebugAbandonedSpans: t.config.internalConfig.DebugAbandonedSpans(), + Disabled: !t.config.internalConfig.TracingEnabled(), + PartialFlush: pfEnabled, + PartialFlushMinSpans: pfMin, + PeerServiceDefaults: t.config.internalConfig.PeerServiceDefaultsEnabled(), + PeerServiceMappings: t.config.internalConfig.PeerServiceMappings(), + EnvTag: t.config.internalConfig.Env(), + VersionTag: t.config.internalConfig.Version(), + ServiceTag: t.config.internalConfig.ServiceName(), + TracingAsTransport: t.config.tracingAsTransport, + isLambdaFunction: t.config.internalConfig.IsLambdaFunction(), + OTLPSpanMetricsEnabled: t.config.internalConfig.OTLPSpanMetricsEnabled(), } } @@ -1232,8 +1240,9 @@ func (t *tracer) submit(s *Span) { if !t.config.internalConfig.TracingEnabled() { return } - // we have an active tracer - if !t.config.canDropP0s() { + // Submit spans to the concentrator when either native stats computation is + // active (canDropP0s) or the OTLP span metrics path is enabled. + if !t.config.canDropP0s() && !t.config.internalConfig.OTLPSpanMetricsEnabled() { return } statSpan, shouldCalc := t.stats.newTracerStatSpan(s, t.obfuscator) diff --git a/ddtrace/tracer/transport.go b/ddtrace/tracer/transport.go index a5797c37b54..fc3aa2b8151 100644 --- a/ddtrace/tracer/transport.go +++ b/ddtrace/tracer/transport.go @@ -208,10 +208,10 @@ func (t *httpTransport) send(p payload) (body io.ReadCloser, err error) { req.Header.Set(idempotencyKeyHeader, newIdempotencyKey()) if t := getGlobalTracer(); t != nil { tc := t.TracerConf() - if tc.TracingAsTransport || tc.CanComputeStats { - // tracingAsTransport uses this header to disable the trace agent's stats computation - // while making canComputeStats() always false to also disable client stats computation. - req.Header.Set("Datadog-Client-Computed-Stats", "t") + if tc.TracingAsTransport || tc.CanComputeStats || tc.OTLPSpanMetricsEnabled { + // Signal to the agent that the SDK has computed stats, so it should not + // recompute them from the trace payload. + req.Header.Set("Datadog-Client-Computed-Stats", "yes") } droppedTraces := int(tracerstats.Count(tracerstats.AgentDroppedP0Traces)) partialTraces := int(tracerstats.Count(tracerstats.PartialTraces)) diff --git a/ddtrace/tracer/transport_test.go b/ddtrace/tracer/transport_test.go index 43c8b74354a..d23ca4ce791 100644 --- a/ddtrace/tracer/transport_test.go +++ b/ddtrace/tracer/transport_test.go @@ -784,7 +784,7 @@ func TestClientComputedStatsHeader(t *testing.T) { t.Run("header-set-when-both-conditions-met", func(t *testing.T) { // When both conditions are met (stats endpoint + client_drop_p0s), - // the Datadog-Client-Computed-Stats header should be set to "t" + // the Datadog-Client-Computed-Stats header should be set to "yes" assert := assert.New(t) var headerValue string srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -807,7 +807,7 @@ func TestClientComputedStatsHeader(t *testing.T) { assert.NoError(err) _, err = trc.config.ddTransport.send(p) assert.NoError(err) - assert.Equal("t", headerValue, "Datadog-Client-Computed-Stats header should be set to 't' when both conditions are met") + assert.Equal("yes", headerValue, "Datadog-Client-Computed-Stats header should be set to 'yes' when both conditions are met") }) } diff --git a/internal/config/config.go b/internal/config/config.go index 40c321c2ef6..5f0710be4c0 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -1433,6 +1433,23 @@ func (c *Config) OTLPSpanMetricsEnabled() bool { return c.otlpExportMode && c.runtimeMetricsOtel } +func (c *Config) SetOTLPSpanMetricsEnabled(enabled bool, origin telemetry.Origin, product ...Product) { + c.mu.Lock() + defer c.mu.Unlock() + if c.checkProductConflict("OTEL_TRACES_SPAN_METRICS_ENABLED", origin, enabled, product...) { + return + } + v := enabled + c.otlpSpanMetricsEnabled = &v + configtelemetry.Report("OTEL_TRACES_SPAN_METRICS_ENABLED", enabled, origin) +} + +func (c *Config) OTLPSemanticsMode() bool { + c.mu.RLock() + defer c.mu.RUnlock() + return c.otlpSemanticsMode +} + func (c *Config) OTLPMetricsURL() string { c.mu.RLock() defer c.mu.RUnlock() From 50cd99bcf2b807cf580f7c6d8572778074533c5d Mon Sep 17 00:00:00 2001 From: Rachel Yang Date: Tue, 16 Jun 2026 14:38:17 -0400 Subject: [PATCH 02/17] fix(tracer): drain concentrator In channel on flush and fix OTLP metrics flush interval Two bugs prevented OTLP span metrics from being exported in system tests: 1. _DD_TRACE_METRICS_OTEL_FLUSH_INTERVAL was read via env.Get() (allowlist), which returned "" for this internal-only override, defaulting to 10s. Switch to os.Getenv for this test-only internal variable. 2. tracer.Flush() called flushAndSend() directly while spans may still be pending in the concentrator's In channel (not yet processed by runIngester). Drain the In channel before flushing when includeCurrent=true so spans submitted just before Flush() are included in the export. Co-Authored-By: Claude Sonnet 4.6 --- ddtrace/tracer/stats.go | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/ddtrace/tracer/stats.go b/ddtrace/tracer/stats.go index 112666bfbba..43429935aeb 100644 --- a/ddtrace/tracer/stats.go +++ b/ddtrace/tracer/stats.go @@ -256,6 +256,21 @@ const ( // When an OTLP exporter is configured, stats are sent to the OTLP metrics endpoint; otherwise they are sent to the // agent's native /v0.6/stats path. func (c *concentrator) flushAndSend(timenow time.Time, includeCurrent bool) { + // When flushing the current bucket (e.g. tracer.Flush()), drain any spans + // that have been sent to c.In but not yet processed by runIngester so they + // are included in the flush rather than silently dropped. + if includeCurrent { + drain: + for { + select { + case s := <-c.In: + c.statsd().Incr("datadog.tracer.stats.spans_in", nil, 1) + c.add(s) + default: + break drain + } + } + } csps := c.spanConcentrator.Flush(timenow.UnixNano(), includeCurrent) if len(csps) == 0 { // nothing to flush From bfdef57f87eb0c7e0c9d382394580e1320b34e4d Mon Sep 17 00:00:00 2001 From: Rachel Yang Date: Wed, 17 Jun 2026 12:34:29 -0400 Subject: [PATCH 03/17] fix(tracer): fix OTLP span metrics peer tags and computed-stats header - Add otlpPeerTags to concentrator; inject span.kind="client" only when the span carries a peer-tag value (http.route, grpc.method.name) so non-top-level unmeasured spans are not made eligible via eligibleSpanKind - Set Datadog-Client-Computed-Stats header statically in transport headers when OTLPSpanMetricsEnabled so it is present from the first request - Pass transport headers to checkEndpoint startup probe so the header is included on the initial 0-trace-chunk connection check Co-Authored-By: Claude Sonnet 4.6 --- ddtrace/tracer/log.go | 11 ++++- ddtrace/tracer/option.go | 8 +++- ddtrace/tracer/stats.go | 45 +++++++++++++++++++- ddtrace/tracer/stats_to_otlp_metrics_test.go | 1 - 4 files changed, 59 insertions(+), 6 deletions(-) diff --git a/ddtrace/tracer/log.go b/ddtrace/tracer/log.go index 774caa365a0..7a5d0c6510a 100644 --- a/ddtrace/tracer/log.go +++ b/ddtrace/tracer/log.go @@ -88,7 +88,7 @@ type startupInfo struct { // checkEndpoint tries to connect to the URL specified by endpoint. // If the endpoint is not reachable, checkEndpoint returns an error // explaining why. -func checkEndpoint(c *http.Client, endpoint string, protocol float64) error { +func checkEndpoint(c *http.Client, endpoint string, protocol float64, extraHeaders map[string]string) error { b := []byte{0x90} // empty array if protocol == traceProtocolV1 { b = []byte{0x80} // empty map @@ -99,6 +99,9 @@ func checkEndpoint(c *http.Client, endpoint string, protocol float64) error { } req.Header.Set(traceCountHeader, "0") req.Header.Set("Content-Type", "application/msgpack") + for k, v := range extraHeaders { + req.Header.Set(k, v) + } res, err := c.Do(req) if err != nil { return err @@ -192,7 +195,11 @@ func logStartup(t *tracer) { info.SampleRateLimit = fmt.Sprintf("%v", limit) } if !t.config.internalConfig.LogToStdout() { - if err := checkEndpoint(t.config.httpClient, t.config.ddTransport.endpoint(), t.config.internalConfig.TraceProtocol()); err != nil { + var startupHeaders map[string]string + if ht, ok := t.config.ddTransport.(*httpTransport); ok { + startupHeaders = ht.headers + } + if err := checkEndpoint(t.config.httpClient, t.config.ddTransport.endpoint(), t.config.internalConfig.TraceProtocol(), startupHeaders); err != nil { info.AgentError = err.Error() log.Warn("DIAGNOSTICS Unable to reach agent intake: %s", err.Error()) } diff --git a/ddtrace/tracer/option.go b/ddtrace/tracer/option.go index 78f3db20b02..1aa87fd20f1 100644 --- a/ddtrace/tracer/option.go +++ b/ddtrace/tracer/option.go @@ -442,7 +442,13 @@ func resolveTraceTransport(cfg *internalconfig.Config) (traceURL string, headers if cfg.TraceProtocol() == traceProtocolV1 { traceURL = agentURL + tracesAPIPathV1 } - return traceURL, datadogHeaders() + headers = datadogHeaders() + if cfg.OTLPSpanMetricsEnabled() { + // Set statically so the header is present on every trace request from startup, + // before agent /info polling has completed and CanComputeStats becomes true. + headers["Datadog-Client-Computed-Stats"] = "yes" + } + return traceURL, headers } func newStatsdClient(c *config) (internal.StatsdClient, error) { diff --git a/ddtrace/tracer/stats.go b/ddtrace/tracer/stats.go index 43429935aeb..0f51c6acfef 100644 --- a/ddtrace/tracer/stats.go +++ b/ddtrace/tracer/stats.go @@ -67,6 +67,10 @@ type concentrator struct { // otlpExporter, when non-nil, routes flushed stats to the OTLP metrics // endpoint instead of the agent's native /v0.6/stats path. otlpExporter *otlpMetricsExporter + + // otlpPeerTags, when non-nil, replaces agent-advertised peer tags with a + // fixed set of OTel semantic-convention dimensions for OTLP span metrics. + otlpPeerTags []string } type tracerStatSpan struct { @@ -184,6 +188,33 @@ func (c *concentrator) newTracerStatSpan(s *Span, obfuscator *obfuscate.Obfuscat httpMethod, _ := s.meta.Get(ext.HTTPMethod) httpEndpoint, _ := s.meta.Get(ext.HTTPEndpoint) + peerTags := c.cfg.agent.load().peerTags + spanMeta := s.meta.Map(false) // stats reads span.kind, _dd.svc_src, status codes, peer tags — no promoted keys needed + if c.otlpPeerTags != nil { + peerTags = c.otlpPeerTags + // The stats library's matchingPeerTags only extracts peer tags for spans with + // span.kind "client", "producer", or "consumer". DD spans (e.g. typestr="web") do + // not carry an OTel span kind, so the library returns nil and peer tags are lost. + // We inject span.kind="client" only when the span actually carries a peer-tag + // value, so non-top-level unmeasured spans without peer tags are not made + // eligible for stats via eligibleSpanKind. + if _, hasKind := spanMeta[ext.SpanKind]; !hasKind { + hasPeerTag := false + for _, k := range c.otlpPeerTags { + if _, ok := spanMeta[k]; ok { + hasPeerTag = true + break + } + } + if hasPeerTag { + spanMeta = make(map[string]string, len(spanMeta)+1) + for k, v := range s.meta.Map(false) { + spanMeta[k] = v + } + spanMeta[ext.SpanKind] = ext.SpanKindClient + } + } + } statSpan, ok := c.spanConcentrator.NewStatSpanWithConfig(stats.StatSpanConfig{ Service: s.service, Resource: resource, @@ -193,9 +224,9 @@ func (c *concentrator) newTracerStatSpan(s *Span, obfuscator *obfuscate.Obfuscat Start: s.start, Duration: s.duration, Error: s.error, - Meta: s.meta.Map(false), // stats reads span.kind, _dd.svc_src, status codes, peer tags — no promoted keys needed + Meta: spanMeta, Metrics: s.metrics, - PeerTags: c.cfg.agent.load().peerTags, + PeerTags: peerTags, HTTPMethod: httpMethod, HTTPEndpoint: httpEndpoint, }) @@ -323,6 +354,15 @@ func (c *concentrator) flushAndSend(timenow time.Time, includeCurrent bool) { c.statsd().Incr("datadog.tracer.stats.flush_buckets", nil, float64(flushedBuckets)) } +// otlpDefaultPeerTags are span meta keys always collected as peer dimensions for +// OTLP span metrics regardless of what the Datadog agent advertises. These cover +// OTel semantic-convention attributes that have no dedicated field in +// ClientGroupedStats (unlike HTTPMethod/HTTPStatusCode which are first-class fields). +var otlpDefaultPeerTags = []string{ + "http.route", + "grpc.method.name", +} + // newOTLPMetricsConcentrator creates a concentrator that exports flushed stats to // the OTLP metrics endpoint instead of the agent's native /v0.6/stats path. // The flush interval is taken from OTLPMetricsFlushInterval in cfg. @@ -330,6 +370,7 @@ func newOTLPMetricsConcentrator(c *config, statsdClient internal.StatsdClient) * bucketSize := c.internalConfig.OTLPMetricsFlushInterval().Nanoseconds() conc := newConcentrator(c, bucketSize, statsdClient) conc.otlpExporter = newOTLPMetricsExporter(c.internalConfig) + conc.otlpPeerTags = otlpDefaultPeerTags return conc } diff --git a/ddtrace/tracer/stats_to_otlp_metrics_test.go b/ddtrace/tracer/stats_to_otlp_metrics_test.go index 56bf35ed008..85c35d63466 100644 --- a/ddtrace/tracer/stats_to_otlp_metrics_test.go +++ b/ddtrace/tracer/stats_to_otlp_metrics_test.go @@ -12,7 +12,6 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" - "google.golang.org/protobuf/proto" pb "github.com/DataDog/datadog-agent/pkg/proto/pbgo/trace" ddsketch "github.com/DataDog/sketches-go/ddsketch" From f94188364aed135107e0777d5dbb6118c1c97d13 Mon Sep 17 00:00:00 2001 From: Rachel Yang Date: Thu, 18 Jun 2026 16:54:15 -0400 Subject: [PATCH 04/17] fix(tracer): fix golangci-lint modernize issues in OTLP files - Replace for-loop map copy with maps.Copy in stats.go (mapsloop) - Replace bare int32/int64 atomic vars with atomic.Int32/Int64 types in otlp_transport_test.go and otlp_writer_test.go (atomictypes) Co-Authored-By: Claude Sonnet 4.6 --- ddtrace/tracer/otlp_transport_test.go | 6 +++--- ddtrace/tracer/otlp_writer_test.go | 12 ++++++------ ddtrace/tracer/stats.go | 5 ++--- 3 files changed, 11 insertions(+), 12 deletions(-) diff --git a/ddtrace/tracer/otlp_transport_test.go b/ddtrace/tracer/otlp_transport_test.go index c553e9ed839..cde46550573 100644 --- a/ddtrace/tracer/otlp_transport_test.go +++ b/ddtrace/tracer/otlp_transport_test.go @@ -97,13 +97,13 @@ func TestOTLPTransportSendConnectionError(t *testing.T) { } func TestOTLPTransportConnectionReuse(t *testing.T) { - var connCount int64 + var connCount atomic.Int64 srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Write([]byte("response body that must be drained")) })) srv.Config.ConnState = func(_ net.Conn, state http.ConnState) { if state == http.StateNew { - atomic.AddInt64(&connCount, 1) + connCount.Add(1) } } srv.Start() @@ -113,6 +113,6 @@ func TestOTLPTransportConnectionReuse(t *testing.T) { for range 5 { require.NoError(t, tr.send([]byte("data"), "application/x-protobuf")) } - assert.Equal(t, int64(1), atomic.LoadInt64(&connCount), + assert.Equal(t, int64(1), connCount.Load(), "expected a single connection to be reused across sends") } diff --git a/ddtrace/tracer/otlp_writer_test.go b/ddtrace/tracer/otlp_writer_test.go index 0970df9df5e..2ea37159486 100644 --- a/ddtrace/tracer/otlp_writer_test.go +++ b/ddtrace/tracer/otlp_writer_test.go @@ -262,14 +262,14 @@ func TestOTLPWriterFlushRetries(t *testing.T) { for _, tc := range testcases { name := fmt.Sprintf("retries=%d/fails=%d", tc.configRetries, tc.failCount) t.Run(name, func(t *testing.T) { - var totalRequests int32 + var totalRequests atomic.Int32 srv := newTestOTLPServer() atomic.StoreInt32(&srv.failCount, int32(tc.failCount)) defer srv.Close() mux := http.NewServeMux() mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { - atomic.AddInt32(&totalRequests, 1) + totalRequests.Add(1) srv.Server.Config.Handler.ServeHTTP(w, r) }) countingSrv := httptest.NewServer(mux) @@ -285,7 +285,7 @@ func TestOTLPWriterFlushRetries(t *testing.T) { w.flush() w.wg.Wait() - assert.Equal(t, int32(tc.expAttempts), atomic.LoadInt32(&totalRequests)) + assert.Equal(t, int32(tc.expAttempts), totalRequests.Load()) assert.Equal(t, tc.tracesSent, len(srv.getPayloads()) > 0) }) } @@ -314,14 +314,14 @@ func TestOTLPWriterConcurrency(t *testing.T) { start := make(chan struct{}) var wg sync.WaitGroup - var spansAdded int32 + var spansAdded atomic.Int32 for range numAdders { wg.Go(func() { <-start for range spansPerAdder { w.add([]*Span{newSpan("op", "svc", "res", randUint64(), randUint64(), 0)}) - atomic.AddInt32(&spansAdded, 1) + spansAdded.Add(1) } }) } @@ -340,7 +340,7 @@ func TestOTLPWriterConcurrency(t *testing.T) { w.stop() - assert.Equal(t, int32(numAdders*spansPerAdder), atomic.LoadInt32(&spansAdded)) + assert.Equal(t, int32(numAdders*spansPerAdder), spansAdded.Load()) // Verify all sent payloads are valid protobuf totalSpans := 0 diff --git a/ddtrace/tracer/stats.go b/ddtrace/tracer/stats.go index 0f51c6acfef..f75ccc3729e 100644 --- a/ddtrace/tracer/stats.go +++ b/ddtrace/tracer/stats.go @@ -6,6 +6,7 @@ package tracer import ( + "maps" "sync" "sync/atomic" "time" @@ -208,9 +209,7 @@ func (c *concentrator) newTracerStatSpan(s *Span, obfuscator *obfuscate.Obfuscat } if hasPeerTag { spanMeta = make(map[string]string, len(spanMeta)+1) - for k, v := range s.meta.Map(false) { - spanMeta[k] = v - } + maps.Copy(spanMeta, s.meta.Map(false)) spanMeta[ext.SpanKind] = ext.SpanKindClient } } From 39876ddff61f6e5671b7954fa44aa7da5002cd4c Mon Sep 17 00:00:00 2001 From: Rachel Yang Date: Mon, 22 Jun 2026 11:02:18 -0400 Subject: [PATCH 05/17] fix(tracer): use string type for _dd.stats_computed and fix http.route as first-class field _dd.stats_computed is a string-valued Datadog convention; change from BoolValue(true) to StringValue("true") to match libdatadog and the system-test spec. http.route (ext.HTTPRoute) is now captured as the HTTPEndpoint first-class field with a fallback after ext.HTTPEndpoint, so OTel-instrumented spans get http.route emitted as a data-point attribute via buildDataPointAttributes. Remove "http.route" from otlpDefaultPeerTags since it had no effect on attribute emission (only grpc.method.name peer tags are converted to attributes). Co-Authored-By: Claude Sonnet 4.6 --- ddtrace/tracer/otlp_metrics_wireup_test.go | 46 +++++++++++++++++++++- ddtrace/tracer/otlp_writer.go | 2 +- ddtrace/tracer/stats.go | 12 +++++- 3 files changed, 56 insertions(+), 4 deletions(-) diff --git a/ddtrace/tracer/otlp_metrics_wireup_test.go b/ddtrace/tracer/otlp_metrics_wireup_test.go index adc7326c2d9..34e89c4578c 100644 --- a/ddtrace/tracer/otlp_metrics_wireup_test.go +++ b/ddtrace/tracer/otlp_metrics_wireup_test.go @@ -18,6 +18,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + tinternal "github.com/DataDog/dd-trace-go/v2/ddtrace/tracer/internal" internalconfig "github.com/DataDog/dd-trace-go/v2/internal/config" ) @@ -137,6 +138,49 @@ func TestOTLPSpanMetricsHeaderOnNativeTraces(t *testing.T) { assert.Equal(t, "yes", headerValue, "Datadog-Client-Computed-Stats must be 'yes' when OTLPSpanMetricsEnabled") } +// TestOTLPConcentratorHTTPRouteAttribute verifies that a span carrying the OTel +// http.route tag (ext.HTTPRoute) produces a data-point with http.route as a +// first-class attribute (FR06). This exercises the ext.HTTPRoute fallback path +// in newTracerStatSpan, which populates ClientGroupedStats.HTTPEndpoint so that +// buildDataPointAttributes emits it — without relying on the peer-tags path. +func TestOTLPConcentratorHTTPRouteAttribute(t *testing.T) { + otlpSrv := newCaptureMetricsServer(t) + + cfg, err := newTestConfig(withNoopInfoHTTPClient()) + require.NoError(t, err) + + bucketSize := int64(500_000) + c := newConcentrator(cfg, bucketSize, &statsd.NoOpClientDirect{}) + c.otlpExporter = &otlpMetricsExporter{ + client: otlpSrv.Server.Client(), + url: otlpSrv.URL + "/v1/metrics", + protocol: "http/json", + cfg: cfg.internalConfig, + } + c.otlpPeerTags = otlpDefaultPeerTags + + s := &Span{ + name: "web.request", + service: "svc", + resource: "web.request", + start: time.Now().UnixNano() - int64(30*time.Second), + duration: int64(50 * time.Millisecond), + metrics: map[string]float64{keyMeasured: 1}, + meta: tinternal.NewSpanMetaFromMap(map[string]string{"http.route": "/users/{id}"}), + } + ss, ok := c.newTracerStatSpan(s, nil) + require.True(t, ok) + c.add(ss) + c.flushAndSend(time.Now(), withCurrentBucket) + + bodies := otlpSrv.receivedBodies() + require.NotEmpty(t, bodies, "OTLP endpoint must receive a payload") + + body := string(bodies[0]) + assert.Contains(t, body, `"http.route"`, "http.route key must appear as a data-point attribute") + assert.Contains(t, body, `/users/{id}`, "http.route value must appear in the payload") +} + // TestOTLPTraceWriterStatsComputedResourceAttr verifies that _dd.stats_computed=true // is added to the OTLP trace resource when OTLP span metrics are enabled (FR15). func TestOTLPTraceWriterStatsComputedResourceAttr(t *testing.T) { @@ -151,7 +195,7 @@ func TestOTLPTraceWriterStatsComputedResourceAttr(t *testing.T) { for _, kv := range w.resource.Attributes { if kv.Key == "_dd.stats_computed" { found = true - assert.True(t, kv.Value.GetBoolValue(), "_dd.stats_computed must be true") + assert.Equal(t, "true", kv.Value.GetStringValue(), "_dd.stats_computed must be string \"true\"") } } assert.True(t, found, "_dd.stats_computed attribute must be present when OTLPSpanMetricsEnabled") diff --git a/ddtrace/tracer/otlp_writer.go b/ddtrace/tracer/otlp_writer.go index fcd423b638f..ae2f6d7ab55 100644 --- a/ddtrace/tracer/otlp_writer.go +++ b/ddtrace/tracer/otlp_writer.go @@ -42,7 +42,7 @@ func newOTLPTraceWriter(c *config) *otlpTraceWriter { // not recompute them from the trace spans. if c.internalConfig.OTLPSpanMetricsEnabled() { resource.Attributes = append(resource.Attributes, - otlpKeyValue("_dd.stats_computed", otlpBoolValue(true)), + otlpKeyValue("_dd.stats_computed", otlpStringValue("true")), ) } scope := &otlpcommon.InstrumentationScope{Name: "dd-trace-go", Version: version.Tag} diff --git a/ddtrace/tracer/stats.go b/ddtrace/tracer/stats.go index f75ccc3729e..fafe9f64a7f 100644 --- a/ddtrace/tracer/stats.go +++ b/ddtrace/tracer/stats.go @@ -188,6 +188,12 @@ func (c *concentrator) newTracerStatSpan(s *Span, obfuscator *obfuscate.Obfuscat } httpMethod, _ := s.meta.Get(ext.HTTPMethod) httpEndpoint, _ := s.meta.Get(ext.HTTPEndpoint) + if httpEndpoint == "" { + // OTel-instrumented spans set http.route directly; DD spans use http.endpoint. + // ClientGroupedStats.HTTPEndpoint is "Http route or quantized/simplified URL path" + // per the proto, so both map to the same first-class field. + httpEndpoint, _ = s.meta.Get(ext.HTTPRoute) + } peerTags := c.cfg.agent.load().peerTags spanMeta := s.meta.Map(false) // stats reads span.kind, _dd.svc_src, status codes, peer tags — no promoted keys needed @@ -356,9 +362,11 @@ func (c *concentrator) flushAndSend(timenow time.Time, includeCurrent bool) { // otlpDefaultPeerTags are span meta keys always collected as peer dimensions for // OTLP span metrics regardless of what the Datadog agent advertises. These cover // OTel semantic-convention attributes that have no dedicated field in -// ClientGroupedStats (unlike HTTPMethod/HTTPStatusCode which are first-class fields). +// ClientGroupedStats (unlike HTTPMethod/HTTPEndpoint/HTTPStatusCode which are +// first-class fields). http.route is intentionally absent: it is captured via +// the HTTPEndpoint first-class field (with fallback from ext.HTTPRoute) and +// emitted as the http.route data-point attribute by buildDataPointAttributes. var otlpDefaultPeerTags = []string{ - "http.route", "grpc.method.name", } From 4b03d7b429baa89d329b285d08896c6f29644a6a Mon Sep 17 00:00:00 2001 From: Rachel Yang Date: Thu, 2 Jul 2026 15:00:38 -0400 Subject: [PATCH 06/17] fix(tracer): shorten comments in PR4 and fix wireup test to use transport field Co-Authored-By: Claude Sonnet 4.6 --- ddtrace/tracer/otlp_metrics_wireup_test.go | 14 ++++---- ddtrace/tracer/stats.go | 37 ++++++---------------- ddtrace/tracer/tracer.go | 6 ++-- 3 files changed, 18 insertions(+), 39 deletions(-) diff --git a/ddtrace/tracer/otlp_metrics_wireup_test.go b/ddtrace/tracer/otlp_metrics_wireup_test.go index 34e89c4578c..9fae8d1cbe7 100644 --- a/ddtrace/tracer/otlp_metrics_wireup_test.go +++ b/ddtrace/tracer/otlp_metrics_wireup_test.go @@ -67,10 +67,9 @@ func TestOTLPMetricsConcentratorRoutesToExporter(t *testing.T) { bucketSize := int64(500_000) c := newConcentrator(cfg, bucketSize, &statsd.NoOpClientDirect{}) c.otlpExporter = &otlpMetricsExporter{ - client: otlpSrv.Server.Client(), - url: otlpSrv.URL + "/v1/metrics", - protocol: "http/json", - cfg: cfg.internalConfig, + transport: newOTLPTransport(otlpSrv.Server.Client(), otlpSrv.URL+"/v1/metrics", nil), + protocol: "http/json", + cfg: cfg.internalConfig, } s := &Span{ @@ -152,10 +151,9 @@ func TestOTLPConcentratorHTTPRouteAttribute(t *testing.T) { bucketSize := int64(500_000) c := newConcentrator(cfg, bucketSize, &statsd.NoOpClientDirect{}) c.otlpExporter = &otlpMetricsExporter{ - client: otlpSrv.Server.Client(), - url: otlpSrv.URL + "/v1/metrics", - protocol: "http/json", - cfg: cfg.internalConfig, + transport: newOTLPTransport(otlpSrv.Server.Client(), otlpSrv.URL+"/v1/metrics", nil), + protocol: "http/json", + cfg: cfg.internalConfig, } c.otlpPeerTags = otlpDefaultPeerTags diff --git a/ddtrace/tracer/stats.go b/ddtrace/tracer/stats.go index fafe9f64a7f..01083df5398 100644 --- a/ddtrace/tracer/stats.go +++ b/ddtrace/tracer/stats.go @@ -65,12 +65,10 @@ type concentrator struct { cfg *config // tracer startup configuration statsdClient internal.StatsdClient // statsd client for sending metrics. - // otlpExporter, when non-nil, routes flushed stats to the OTLP metrics - // endpoint instead of the agent's native /v0.6/stats path. + // otlpExporter, when non-nil, routes flushed stats to the OTLP metrics endpoint. otlpExporter *otlpMetricsExporter - // otlpPeerTags, when non-nil, replaces agent-advertised peer tags with a - // fixed set of OTel semantic-convention dimensions for OTLP span metrics. + // otlpPeerTags, when non-nil, overrides agent-advertised peer tags for OTLP span metrics. otlpPeerTags []string } @@ -189,9 +187,7 @@ func (c *concentrator) newTracerStatSpan(s *Span, obfuscator *obfuscate.Obfuscat httpMethod, _ := s.meta.Get(ext.HTTPMethod) httpEndpoint, _ := s.meta.Get(ext.HTTPEndpoint) if httpEndpoint == "" { - // OTel-instrumented spans set http.route directly; DD spans use http.endpoint. - // ClientGroupedStats.HTTPEndpoint is "Http route or quantized/simplified URL path" - // per the proto, so both map to the same first-class field. + // OTel spans set http.route; DD spans use http.endpoint — both map to HTTPEndpoint. httpEndpoint, _ = s.meta.Get(ext.HTTPRoute) } @@ -199,12 +195,8 @@ func (c *concentrator) newTracerStatSpan(s *Span, obfuscator *obfuscate.Obfuscat spanMeta := s.meta.Map(false) // stats reads span.kind, _dd.svc_src, status codes, peer tags — no promoted keys needed if c.otlpPeerTags != nil { peerTags = c.otlpPeerTags - // The stats library's matchingPeerTags only extracts peer tags for spans with - // span.kind "client", "producer", or "consumer". DD spans (e.g. typestr="web") do - // not carry an OTel span kind, so the library returns nil and peer tags are lost. - // We inject span.kind="client" only when the span actually carries a peer-tag - // value, so non-top-level unmeasured spans without peer tags are not made - // eligible for stats via eligibleSpanKind. + // matchingPeerTags requires span.kind "client/producer/consumer"; DD spans lack it. + // Inject span.kind=client only when a peer-tag value is present to avoid unwanted stats eligibility. if _, hasKind := spanMeta[ext.SpanKind]; !hasKind { hasPeerTag := false for _, k := range c.otlpPeerTags { @@ -287,10 +279,8 @@ const ( withoutCurrentBucket = false ) -// flushAndSend flushes all the stats buckets with the given timestamp and sends them using the transport specified in -// the concentrator config. The current bucket is only included if includeCurrent is true, such as during shutdown. -// When an OTLP exporter is configured, stats are sent to the OTLP metrics endpoint; otherwise they are sent to the -// agent's native /v0.6/stats path. +// flushAndSend flushes all stats buckets and sends them; the current bucket is included only when includeCurrent is true. +// Stats go to the OTLP metrics endpoint when an OTLP exporter is configured; otherwise to the agent's /v0.6/stats path. func (c *concentrator) flushAndSend(timenow time.Time, includeCurrent bool) { // When flushing the current bucket (e.g. tracer.Flush()), drain any spans // that have been sent to c.In but not yet processed by runIngester so they @@ -359,20 +349,13 @@ func (c *concentrator) flushAndSend(timenow time.Time, includeCurrent bool) { c.statsd().Incr("datadog.tracer.stats.flush_buckets", nil, float64(flushedBuckets)) } -// otlpDefaultPeerTags are span meta keys always collected as peer dimensions for -// OTLP span metrics regardless of what the Datadog agent advertises. These cover -// OTel semantic-convention attributes that have no dedicated field in -// ClientGroupedStats (unlike HTTPMethod/HTTPEndpoint/HTTPStatusCode which are -// first-class fields). http.route is intentionally absent: it is captured via -// the HTTPEndpoint first-class field (with fallback from ext.HTTPRoute) and -// emitted as the http.route data-point attribute by buildDataPointAttributes. +// otlpDefaultPeerTags are always-collected peer dimensions for OTLP span metrics, covering OTel attributes +// not in ClientGroupedStats first-class fields. http.route is absent: it flows through HTTPEndpoint. var otlpDefaultPeerTags = []string{ "grpc.method.name", } -// newOTLPMetricsConcentrator creates a concentrator that exports flushed stats to -// the OTLP metrics endpoint instead of the agent's native /v0.6/stats path. -// The flush interval is taken from OTLPMetricsFlushInterval in cfg. +// newOTLPMetricsConcentrator creates a concentrator that exports to the OTLP metrics endpoint. func newOTLPMetricsConcentrator(c *config, statsdClient internal.StatsdClient) *concentrator { bucketSize := c.internalConfig.OTLPMetricsFlushInterval().Nanoseconds() conc := newConcentrator(c, bucketSize, statsdClient) diff --git a/ddtrace/tracer/tracer.go b/ddtrace/tracer/tracer.go index 9cb706c5eb8..6bf92ac848d 100644 --- a/ddtrace/tracer/tracer.go +++ b/ddtrace/tracer/tracer.go @@ -517,8 +517,7 @@ func newUnstartedTracer(opts ...StartOption) (t *tracer, err error) { } var sc statsConcentrator if c.internalConfig.OTLPSpanMetricsEnabled() { - // OTLP span metrics: SDK computes stats and exports them via OTLP. - // The concentrator never forwards to the agent's /v0.6/stats path. + // OTLP span metrics: SDK computes and exports stats; agent /v0.6/stats path unused. sc = newOTLPMetricsConcentrator(c, statsd) } else if c.internalConfig.OTLPExportMode() { sc = &noopConcentrator{} @@ -1240,8 +1239,7 @@ func (t *tracer) submit(s *Span) { if !t.config.internalConfig.TracingEnabled() { return } - // Submit spans to the concentrator when either native stats computation is - // active (canDropP0s) or the OTLP span metrics path is enabled. + // Submit to the concentrator when native stats (canDropP0s) or OTLP span metrics are enabled. if !t.config.canDropP0s() && !t.config.internalConfig.OTLPSpanMetricsEnabled() { return } From 3a51d7cd44baf772c58c003ae70766b9b8d99173 Mon Sep 17 00:00:00 2001 From: Rachel Yang Date: Mon, 6 Jul 2026 16:10:33 -0400 Subject: [PATCH 07/17] fix(tracer): make otlpTraceWriter.flush synchronous (FR15_3 + review feedback) Remove the goroutine + shared WaitGroup pattern from flush(). The shared WaitGroup caused a misuse panic when multiple goroutines called flush() concurrently (each calling Add then Wait on the same WaitGroup). Making flush synchronous also fixes FR15_3: tracer.Flush() now blocks until the OTLP trace payload reaches the collector, consistent with how the concentrator sends metrics synchronously. The worker goroutine already serialises flush calls in production; the climit concurrency cap was redundant and is removed along with the WaitGroup fields. Co-Authored-By: Claude Sonnet 4.6 --- ddtrace/tracer/otlp_writer.go | 78 ++++++++++-------------- ddtrace/tracer/otlp_writer_bench_test.go | 2 - ddtrace/tracer/otlp_writer_test.go | 10 --- 3 files changed, 32 insertions(+), 58 deletions(-) diff --git a/ddtrace/tracer/otlp_writer.go b/ddtrace/tracer/otlp_writer.go index ae2f6d7ab55..028beb77a19 100644 --- a/ddtrace/tracer/otlp_writer.go +++ b/ddtrace/tracer/otlp_writer.go @@ -7,7 +7,6 @@ package tracer import ( "slices" - "sync" "time" otlpcommon "go.opentelemetry.io/proto/otlp/common/v1" @@ -32,8 +31,6 @@ type otlpTraceWriter struct { spans []*otlptrace.Span // +checklocks:mu buffSize int // +checklocks:mu baseSize int - climit chan struct{} - wg sync.WaitGroup } func newOTLPTraceWriter(c *config) *otlpTraceWriter { @@ -62,7 +59,6 @@ func newOTLPTraceWriter(c *config) *otlpTraceWriter { spans: make([]*otlptrace.Span, 0), buffSize: baseSize, baseSize: baseSize, - climit: make(chan struct{}, concurrentConnectionLimit), } } @@ -103,56 +99,46 @@ func (w *otlpTraceWriter) flush() { readySpans := w.reset() w.mu.Unlock() - w.climit <- struct{}{} - w.wg.Add(1) - go func() { - defer func() { - <-w.climit - w.wg.Done() - }() - - spanCount := len(readySpans) - tracesData := &otlptrace.TracesData{ - ResourceSpans: []*otlptrace.ResourceSpans{ - { - Resource: w.resource, - ScopeSpans: []*otlptrace.ScopeSpans{ - { - Scope: w.scope, - Spans: readySpans, - }, + spanCount := len(readySpans) + tracesData := &otlptrace.TracesData{ + ResourceSpans: []*otlptrace.ResourceSpans{ + { + Resource: w.resource, + ScopeSpans: []*otlptrace.ScopeSpans{ + { + Scope: w.scope, + Spans: readySpans, }, }, }, - } - b, err := proto.Marshal(tracesData) - readySpans = nil - tracesData = nil - if err != nil { - log.Error("Error marshalling OTLP traces data: %s", err.Error()) - return - } + }, + } + b, err := proto.Marshal(tracesData) + readySpans = nil + tracesData = nil + if err != nil { + log.Error("Error marshalling OTLP traces data: %s", err.Error()) + return + } - var sendErr error - sendRetries := w.config.internalConfig.SendRetries() - retryInterval := w.config.internalConfig.RetryInterval() - for attempt := 0; attempt <= sendRetries; attempt++ { - log.Debug("OTLP: attempt %d to send payload: %d bytes, %d spans", attempt+1, len(b), spanCount) - sendErr = w.transport.send(b, otlpContentTypeProto) - if sendErr == nil { - log.Debug("OTLP: sent traces after %d attempts", attempt+1) - return - } - log.Error("OTLP: failure sending traces (attempt %d of %d): %v", attempt+1, sendRetries+1, sendErr.Error()) - time.Sleep(retryInterval) + var sendErr error + sendRetries := w.config.internalConfig.SendRetries() + retryInterval := w.config.internalConfig.RetryInterval() + for attempt := 0; attempt <= sendRetries; attempt++ { + log.Debug("OTLP: attempt %d to send payload: %d bytes, %d spans", attempt+1, len(b), spanCount) + sendErr = w.transport.send(b, otlpContentTypeProto) + if sendErr == nil { + log.Debug("OTLP: sent traces after %d attempts", attempt+1) + return } - log.Error("OTLP: lost %d spans: %v", spanCount, sendErr.Error()) - }() + log.Error("OTLP: failure sending traces (attempt %d of %d): %v", attempt+1, sendRetries+1, sendErr.Error()) + time.Sleep(retryInterval) + } + log.Error("OTLP: lost %d spans: %v", spanCount, sendErr.Error()) } func (w *otlpTraceWriter) stop() { w.flush() - w.wg.Wait() } -func (w *otlpTraceWriter) wait() { w.wg.Wait() } +func (w *otlpTraceWriter) wait() { w.flush() } diff --git a/ddtrace/tracer/otlp_writer_bench_test.go b/ddtrace/tracer/otlp_writer_bench_test.go index 335feb7c10f..87ffd4db03b 100644 --- a/ddtrace/tracer/otlp_writer_bench_test.go +++ b/ddtrace/tracer/otlp_writer_bench_test.go @@ -34,7 +34,6 @@ func newBenchOTLPWriter(b *testing.B) *otlpTraceWriter { resource: buildResource(cfg.internalConfig), scope: &otlpcommon.InstrumentationScope{Name: "dd-trace-go", Version: version.Tag}, spans: make([]*otlptrace.Span, 0), - climit: make(chan struct{}, concurrentConnectionLimit), } } @@ -78,7 +77,6 @@ func BenchmarkOTLPTraceWriterFlush(b *testing.B) { for b.Loop() { writer.add(trace) writer.flush() - writer.wg.Wait() } } diff --git a/ddtrace/tracer/otlp_writer_test.go b/ddtrace/tracer/otlp_writer_test.go index 2ea37159486..992880db773 100644 --- a/ddtrace/tracer/otlp_writer_test.go +++ b/ddtrace/tracer/otlp_writer_test.go @@ -91,7 +91,6 @@ func newTestOTLPWriter(t *testing.T, srv *testOTLPServer, opts ...StartOption) * spans: make([]*otlptrace.Span, 0), buffSize: baseSize, baseSize: baseSize, - climit: make(chan struct{}, concurrentConnectionLimit), } } @@ -135,7 +134,6 @@ func TestOTLPWriterFlushEmpty(t *testing.T) { w := newTestOTLPWriter(t, srv) w.flush() - w.wg.Wait() assert.Equal(t, 0, srv.requestCount()) } @@ -150,7 +148,6 @@ func TestOTLPWriterFlush(t *testing.T) { newSpan("op2", "svc", "res", 2, 1, 0), }) w.flush() - w.wg.Wait() payloads := srv.getPayloads() require.Equal(t, 1, len(payloads)) @@ -178,7 +175,6 @@ func TestOTLPWriterFlushClearsSpans(t *testing.T) { w.add([]*Span{newSpan("op1", "svc", "res", 1, 1, 0)}) w.flush() - w.wg.Wait() w.mu.Lock() assert.Equal(t, 0, len(w.spans)) @@ -186,7 +182,6 @@ func TestOTLPWriterFlushClearsSpans(t *testing.T) { // Second flush should be a no-op w.flush() - w.wg.Wait() assert.Equal(t, 1, srv.requestCount()) } @@ -199,7 +194,6 @@ func TestOTLPWriterFlushOnSize(t *testing.T) { bigSpan := newSpan("op", "svc", "res", 1, 1, 0) bigSpan.meta.Set("big", strings.Repeat("X", payloadSizeLimit+1)) w.add([]*Span{bigSpan}) - w.wg.Wait() assert.GreaterOrEqual(t, srv.requestCount(), 1) w.mu.Lock() @@ -220,7 +214,6 @@ func TestOTLPWriterFlushOnSize(t *testing.T) { s.meta.Set("data", strings.Repeat("X", spanSize)) w.add([]*Span{s}) } - w.wg.Wait() assert.GreaterOrEqual(t, srv.requestCount(), 1) }) @@ -283,7 +276,6 @@ func TestOTLPWriterFlushRetries(t *testing.T) { w.add([]*Span{newSpan("op", "svc", "res", 1, 1, 0)}) w.flush() - w.wg.Wait() assert.Equal(t, int32(tc.expAttempts), totalRequests.Load()) assert.Equal(t, tc.tracesSent, len(srv.getPayloads()) > 0) @@ -383,7 +375,6 @@ func TestOTLPWriterBuffSizeTracking(t *testing.T) { t.Run("flush resets buffSize to baseSize", func(t *testing.T) { w.flush() - w.wg.Wait() w.mu.Lock() assert.Equal(t, w.baseSize, w.buffSize) @@ -420,7 +411,6 @@ func TestOTLPWriterBuffSizeTracking(t *testing.T) { "estimated %d should be within 5%% of actual %d", estimated, actual) w.flush() - w.wg.Wait() }) } From 5723e58e3a5f47aa285676f4af5abf234caac203 Mon Sep 17 00:00:00 2001 From: Rachel Yang Date: Thu, 9 Jul 2026 12:58:01 -0400 Subject: [PATCH 08/17] fix(tracer): revert Datadog-Client-Computed-Stats header value from yes back to t MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The t→yes rename was an unrelated silent behavior change riding along with the OTLPSpanMetricsEnabled condition. The agent checks header presence, not value, so both are functionally equivalent, but keeping t avoids a gratuitous diff. Co-Authored-By: Claude Sonnet 4.6 --- ddtrace/tracer/transport.go | 4 +--- ddtrace/tracer/transport_test.go | 4 ++-- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/ddtrace/tracer/transport.go b/ddtrace/tracer/transport.go index fc3aa2b8151..0ffee2ff491 100644 --- a/ddtrace/tracer/transport.go +++ b/ddtrace/tracer/transport.go @@ -209,9 +209,7 @@ func (t *httpTransport) send(p payload) (body io.ReadCloser, err error) { if t := getGlobalTracer(); t != nil { tc := t.TracerConf() if tc.TracingAsTransport || tc.CanComputeStats || tc.OTLPSpanMetricsEnabled { - // Signal to the agent that the SDK has computed stats, so it should not - // recompute them from the trace payload. - req.Header.Set("Datadog-Client-Computed-Stats", "yes") + req.Header.Set("Datadog-Client-Computed-Stats", "t") } droppedTraces := int(tracerstats.Count(tracerstats.AgentDroppedP0Traces)) partialTraces := int(tracerstats.Count(tracerstats.PartialTraces)) diff --git a/ddtrace/tracer/transport_test.go b/ddtrace/tracer/transport_test.go index d23ca4ce791..78db7437803 100644 --- a/ddtrace/tracer/transport_test.go +++ b/ddtrace/tracer/transport_test.go @@ -784,7 +784,7 @@ func TestClientComputedStatsHeader(t *testing.T) { t.Run("header-set-when-both-conditions-met", func(t *testing.T) { // When both conditions are met (stats endpoint + client_drop_p0s), - // the Datadog-Client-Computed-Stats header should be set to "yes" + // the Datadog-Client-Computed-Stats header should be set. assert := assert.New(t) var headerValue string srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -807,7 +807,7 @@ func TestClientComputedStatsHeader(t *testing.T) { assert.NoError(err) _, err = trc.config.ddTransport.send(p) assert.NoError(err) - assert.Equal("yes", headerValue, "Datadog-Client-Computed-Stats header should be set to 'yes' when both conditions are met") + assert.Equal("t", headerValue, "Datadog-Client-Computed-Stats header should be set when both conditions are met") }) } From 7daad0966f3e79245fb05ce84f8a84717caab956 Mon Sep 17 00:00:00 2001 From: Rachel Yang Date: Thu, 9 Jul 2026 12:58:48 -0400 Subject: [PATCH 09/17] revert: restore Datadog-Client-Computed-Stats header value to yes Co-Authored-By: Claude Sonnet 4.6 --- ddtrace/tracer/transport.go | 2 +- ddtrace/tracer/transport_test.go | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/ddtrace/tracer/transport.go b/ddtrace/tracer/transport.go index 0ffee2ff491..5e6aace0f36 100644 --- a/ddtrace/tracer/transport.go +++ b/ddtrace/tracer/transport.go @@ -209,7 +209,7 @@ func (t *httpTransport) send(p payload) (body io.ReadCloser, err error) { if t := getGlobalTracer(); t != nil { tc := t.TracerConf() if tc.TracingAsTransport || tc.CanComputeStats || tc.OTLPSpanMetricsEnabled { - req.Header.Set("Datadog-Client-Computed-Stats", "t") + req.Header.Set("Datadog-Client-Computed-Stats", "yes") } droppedTraces := int(tracerstats.Count(tracerstats.AgentDroppedP0Traces)) partialTraces := int(tracerstats.Count(tracerstats.PartialTraces)) diff --git a/ddtrace/tracer/transport_test.go b/ddtrace/tracer/transport_test.go index 78db7437803..fb4cce3e9f4 100644 --- a/ddtrace/tracer/transport_test.go +++ b/ddtrace/tracer/transport_test.go @@ -784,7 +784,7 @@ func TestClientComputedStatsHeader(t *testing.T) { t.Run("header-set-when-both-conditions-met", func(t *testing.T) { // When both conditions are met (stats endpoint + client_drop_p0s), - // the Datadog-Client-Computed-Stats header should be set. + // the Datadog-Client-Computed-Stats header should be set to "yes". assert := assert.New(t) var headerValue string srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -807,7 +807,7 @@ func TestClientComputedStatsHeader(t *testing.T) { assert.NoError(err) _, err = trc.config.ddTransport.send(p) assert.NoError(err) - assert.Equal("t", headerValue, "Datadog-Client-Computed-Stats header should be set when both conditions are met") + assert.Equal("yes", headerValue, "Datadog-Client-Computed-Stats header should be set to 'yes' when both conditions are met") }) } From 2dc50c2215dae33d38e060551ddc2f9e3d4c0dae Mon Sep 17 00:00:00 2001 From: Rachel Yang Date: Tue, 21 Jul 2026 15:13:45 -0400 Subject: [PATCH 10/17] fix(tracer): remove dead OTLPSemanticsMode method (duplicate of OTelSemanticsEnabled) Co-Authored-By: Claude Sonnet 4.6 --- internal/config/config.go | 6 ------ 1 file changed, 6 deletions(-) diff --git a/internal/config/config.go b/internal/config/config.go index 5f0710be4c0..5bcedbb4418 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -1444,12 +1444,6 @@ func (c *Config) SetOTLPSpanMetricsEnabled(enabled bool, origin telemetry.Origin configtelemetry.Report("OTEL_TRACES_SPAN_METRICS_ENABLED", enabled, origin) } -func (c *Config) OTLPSemanticsMode() bool { - c.mu.RLock() - defer c.mu.RUnlock() - return c.otlpSemanticsMode -} - func (c *Config) OTLPMetricsURL() string { c.mu.RLock() defer c.mu.RUnlock() From 4cea0e4fa531316804b3b2140eee3287f7c47878 Mon Sep 17 00:00:00 2001 From: Rachel Yang Date: Tue, 21 Jul 2026 16:11:25 -0400 Subject: [PATCH 11/17] fix(tracer): spec alignment and cleanup for OTLP span metrics (PR4 review) - Remove grpc.method.name from otlpDefaultPeerTags: peer tags increase stats bucket cardinality per gRPC method, contradicting the June 25 spec change that removed rpc.method to prevent cardinality inflation. Peer tags are explicitly out of scope per the RFC attribute table. - Fix BucketInterval hardcoded to defaultStatsBucketSize in newConcentrator; pass bucketSize so _DD_TRACE_STATS_INTERVAL controls the aggregation window. - Fix redundant s.meta.Map(false) call in peer-tag span.kind injection path. - Extract sendWithRetry helper to deduplicate identical retry loops in flushAndSend's OTLP and native /v0.6/stats branches. - Add comment explaining Datadog-Client-Computed-Stats value change from "t" to "yes" (spec-correct per FR15; both values accepted by the Agent). Co-Authored-By: Claude Sonnet 4.6 --- ddtrace/tracer/otlp_metrics_wireup_test.go | 2 +- ddtrace/tracer/stats.go | 54 +++++++++++----------- ddtrace/tracer/transport.go | 3 ++ 3 files changed, 30 insertions(+), 29 deletions(-) diff --git a/ddtrace/tracer/otlp_metrics_wireup_test.go b/ddtrace/tracer/otlp_metrics_wireup_test.go index 9fae8d1cbe7..565df940a0e 100644 --- a/ddtrace/tracer/otlp_metrics_wireup_test.go +++ b/ddtrace/tracer/otlp_metrics_wireup_test.go @@ -155,7 +155,7 @@ func TestOTLPConcentratorHTTPRouteAttribute(t *testing.T) { protocol: "http/json", cfg: cfg.internalConfig, } - c.otlpPeerTags = otlpDefaultPeerTags + c.otlpPeerTags = []string{} s := &Span{ name: "web.request", diff --git a/ddtrace/tracer/stats.go b/ddtrace/tracer/stats.go index 01083df5398..1483056a526 100644 --- a/ddtrace/tracer/stats.go +++ b/ddtrace/tracer/stats.go @@ -83,7 +83,7 @@ type tracerStatSpan struct { func newConcentrator(c *config, bucketSize int64, statsdClient internal.StatsdClient) *concentrator { sCfg := &stats.SpanConcentratorConfig{ ComputeStatsBySpanKind: true, - BucketInterval: defaultStatsBucketSize, + BucketInterval: bucketSize, } env := c.agent.load().defaultEnv if c.internalConfig.Env() != "" { @@ -206,8 +206,9 @@ func (c *concentrator) newTracerStatSpan(s *Span, obfuscator *obfuscate.Obfuscat } } if hasPeerTag { - spanMeta = make(map[string]string, len(spanMeta)+1) - maps.Copy(spanMeta, s.meta.Map(false)) + orig := spanMeta + spanMeta = make(map[string]string, len(orig)+1) + maps.Copy(spanMeta, orig) spanMeta[ext.SpanKind] = ext.SpanKindClient } } @@ -279,6 +280,19 @@ const ( withoutCurrentBucket = false ) +func sendWithRetry(retries int, interval time.Duration, fn func() error) error { + var err error + for attempt := 0; attempt <= retries; attempt++ { + if err = fn(); err == nil { + return nil + } + if attempt < retries { + time.Sleep(interval) + } + } + return err +} + // flushAndSend flushes all stats buckets and sends them; the current bucket is included only when includeCurrent is true. // Stats go to the OTLP metrics endpoint when an OTLP exporter is configured; otherwise to the agent's /v0.6/stats path. func (c *concentrator) flushAndSend(timenow time.Time, includeCurrent bool) { @@ -315,15 +329,9 @@ func (c *concentrator) flushAndSend(timenow time.Time, includeCurrent bool) { flushedBuckets += len(csp.Stats) var err error if c.otlpExporter != nil { - for attempt := 0; attempt <= sendRetries; attempt++ { - err = c.otlpExporter.export(csp) - if err == nil { - break - } - if attempt < sendRetries { - time.Sleep(retryInterval) - } - } + err = sendWithRetry(sendRetries, retryInterval, func() error { + return c.otlpExporter.export(csp) + }) } else { obfVersion := 0 if c.shouldObfuscate() { @@ -331,15 +339,9 @@ func (c *concentrator) flushAndSend(timenow time.Time, includeCurrent bool) { } else { log.Debug("Stats Obfuscation was skipped, agent will obfuscate (tracer %d, agent %d)", tracerObfuscationVersion, c.cfg.agent.load().obfuscationVersion) } - for attempt := 0; attempt <= sendRetries; attempt++ { - err = c.cfg.ddTransport.sendStats(csp, obfVersion) - if err == nil { - break - } - if attempt < sendRetries { - time.Sleep(retryInterval) - } - } + err = sendWithRetry(sendRetries, retryInterval, func() error { + return c.cfg.ddTransport.sendStats(csp, obfVersion) + }) } if err != nil { c.statsd().Incr("datadog.tracer.stats.flush_errors", nil, 1) @@ -349,18 +351,14 @@ func (c *concentrator) flushAndSend(timenow time.Time, includeCurrent bool) { c.statsd().Incr("datadog.tracer.stats.flush_buckets", nil, float64(flushedBuckets)) } -// otlpDefaultPeerTags are always-collected peer dimensions for OTLP span metrics, covering OTel attributes -// not in ClientGroupedStats first-class fields. http.route is absent: it flows through HTTPEndpoint. -var otlpDefaultPeerTags = []string{ - "grpc.method.name", -} - // newOTLPMetricsConcentrator creates a concentrator that exports to the OTLP metrics endpoint. func newOTLPMetricsConcentrator(c *config, statsdClient internal.StatsdClient) *concentrator { bucketSize := c.internalConfig.OTLPMetricsFlushInterval().Nanoseconds() conc := newConcentrator(c, bucketSize, statsdClient) conc.otlpExporter = newOTLPMetricsExporter(c.internalConfig) - conc.otlpPeerTags = otlpDefaultPeerTags + // Peer tags are out of scope for OTLP span metrics (spec: "Out of scope for SDK implementation"). + // Set to empty (non-nil) to suppress agent-advertised peer tags on this concentrator. + conc.otlpPeerTags = []string{} return conc } diff --git a/ddtrace/tracer/transport.go b/ddtrace/tracer/transport.go index 5e6aace0f36..686cfc995c6 100644 --- a/ddtrace/tracer/transport.go +++ b/ddtrace/tracer/transport.go @@ -209,6 +209,9 @@ func (t *httpTransport) send(p payload) (body io.ReadCloser, err error) { if t := getGlobalTracer(); t != nil { tc := t.TracerConf() if tc.TracingAsTransport || tc.CanComputeStats || tc.OTLPSpanMetricsEnabled { + // "yes" is the spec-correct value (FR15). Previously "t" was used for the + // TracingAsTransport and CanComputeStats paths; both values are accepted by + // the Agent, but "yes" is canonical going forward. req.Header.Set("Datadog-Client-Computed-Stats", "yes") } droppedTraces := int(tracerstats.Count(tracerstats.AgentDroppedP0Traces)) From da1b1024c5557946d93fb61e41ff19d9f22cb313 Mon Sep 17 00:00:00 2001 From: Rachel Yang Date: Wed, 22 Jul 2026 18:04:10 -0400 Subject: [PATCH 12/17] fix(tracer): add tracer_dd_tags, additional_metric_tags attrs and v0.4 downgrade for OTLP span metrics - Emit tracer_dd_tags array resource attribute for non-reserved DD_TAGS/ OTEL_RESOURCE_ATTRIBUTES key:value pairs (fixes fr08_10, fr08_11) - Emit additional_metric_tags array data-point attribute from ClientGroupedStats.AdditionalMetricTags (fixes fr08_12) - Downgrade trace transport to v0.4 when OTLP span metrics are enabled so the Datadog-Client-Computed-Stats header is visible to the test agent (fixes fr02_3, fr15_1) Co-Authored-By: Claude Sonnet 4.6 --- ddtrace/tracer/option.go | 9 +++-- ddtrace/tracer/stats_to_otlp_metrics.go | 44 +++++++++++++++++++++++-- 2 files changed, 47 insertions(+), 6 deletions(-) diff --git a/ddtrace/tracer/option.go b/ddtrace/tracer/option.go index 1aa87fd20f1..86940d6dc02 100644 --- a/ddtrace/tracer/option.go +++ b/ddtrace/tracer/option.go @@ -341,9 +341,12 @@ func newConfig(opts ...StartOption) (*config, error) { agentURL := c.internalConfig.AgentURL() af := loadAgentFeatures(agentDisabled, agentURL, c.httpClient) c.agent.store(af) - // If the agent doesn't support the v1 protocol, downgrade to v0.4 - // Also downgrade if CSS is disabled, as v1 is not compatible without CSS. - if c.internalConfig.TraceProtocol() == traceProtocolV1 && (!af.v1ProtocolAvailable || !c.canComputeStats()) { + // If the agent doesn't support the v1 protocol, downgrade to v0.4. + // Also downgrade if CSS is disabled (v1 requires CSS) or if OTLP span metrics are + // enabled: OTLP span metrics use their own concentrator and are not native CSS, so + // the trace transport should stay on v0.4 where the test agent and Datadog Agent + // can observe the Datadog-Client-Computed-Stats header set by resolveTraceTransport. + if c.internalConfig.TraceProtocol() == traceProtocolV1 && (!af.v1ProtocolAvailable || !c.canComputeStats() || c.internalConfig.OTLPSpanMetricsEnabled()) { c.internalConfig.SetTraceProtocol(traceProtocolV04, internalconfig.OriginCalculated) if t, ok := c.ddTransport.(*httpTransport); ok && t.traceURL == agentURL.String()+tracesAPIPathV1 { t.traceURL = agentURL.String() + tracesAPIPath diff --git a/ddtrace/tracer/stats_to_otlp_metrics.go b/ddtrace/tracer/stats_to_otlp_metrics.go index 8c1d0bb1165..b6e4c0a4cc4 100644 --- a/ddtrace/tracer/stats_to_otlp_metrics.go +++ b/ddtrace/tracer/stats_to_otlp_metrics.go @@ -96,13 +96,38 @@ func buildMetricsResource(payload *pb.ClientStatsPayload, otelMode bool, reportH attrs = append(attrs, otlpKeyValue("datadog.runtime_id", otlpStringValue(payload.RuntimeID))) } if payload.ProcessTags != "" { + // Reserved DD_TAGS keys whose semantic meaning is captured in dedicated + // resource attributes (service.name, deployment.environment.name, etc.) + // and must therefore be excluded from the generic tracer_dd_tags array. + reservedDDTagKeys := map[string]bool{ + "service": true, + "env": true, + "version": true, + "runtime_id": true, + "runtime-id": true, + } + var tracerDDTagEntries []*otlpcommon.AnyValue for tag := range strings.SplitSeq(payload.ProcessTags, ",") { parts := strings.SplitN(tag, ":", 2) - // Skip keys emitted explicitly above (runtime_id) to avoid duplicate - // resource attributes, and skip tags with empty values. - if len(parts) == 2 && parts[0] != "" && parts[1] != "" && parts[0] != "runtime_id" { + if len(parts) != 2 || parts[0] == "" || parts[1] == "" { + continue + } + // datadog. resource attributes for all non-runtime_id process tags. + if parts[0] != "runtime_id" { attrs = append(attrs, otlpKeyValue("datadog."+parts[0], otlpStringValue(parts[1]))) } + // tracer_dd_tags array: non-reserved key:value pairs from DD_TAGS / + // OTEL_RESOURCE_ATTRIBUTES that are not already captured elsewhere. + if !reservedDDTagKeys[parts[0]] { + tracerDDTagEntries = append(tracerDDTagEntries, otlpStringValue(tag)) + } + } + if len(tracerDDTagEntries) > 0 { + attrs = append(attrs, otlpKeyValue("tracer_dd_tags", &otlpcommon.AnyValue{ + Value: &otlpcommon.AnyValue_ArrayValue{ + ArrayValue: &otlpcommon.ArrayValue{Values: tracerDDTagEntries}, + }, + })) } } } @@ -217,6 +242,19 @@ func buildDataPointAttributes(gs *pb.ClientGroupedStats, isError bool, defaultSe if gs.Synthetics { attrs = append(attrs, otlpKeyValue("datadog.origin", otlpStringValue("synthetics"))) } + // additional_metric_tags holds the per-span tag values for keys listed in + // DD_TRACE_STATS_ADDITIONAL_TAGS (already populated in ClientGroupedStats by the concentrator). + if len(gs.AdditionalMetricTags) > 0 { + tagEntries := make([]*otlpcommon.AnyValue, 0, len(gs.AdditionalMetricTags)) + for _, t := range gs.AdditionalMetricTags { + tagEntries = append(tagEntries, otlpStringValue(t)) + } + attrs = append(attrs, otlpKeyValue("additional_metric_tags", &otlpcommon.AnyValue{ + Value: &otlpcommon.AnyValue_ArrayValue{ + ArrayValue: &otlpcommon.ArrayValue{Values: tagEntries}, + }, + })) + } } return attrs From 5e74428e3e496972bcb2c67604df1e919b7a8916 Mon Sep 17 00:00:00 2001 From: Rachel Yang Date: Wed, 22 Jul 2026 18:15:29 -0400 Subject: [PATCH 13/17] Revert "fix(tracer): add tracer_dd_tags, additional_metric_tags attrs and v0.4 downgrade for OTLP span metrics" This reverts commit 2183b1785c99a396c8697dab60ce50643eef9847. --- ddtrace/tracer/option.go | 9 ++--- ddtrace/tracer/stats_to_otlp_metrics.go | 44 ++----------------------- 2 files changed, 6 insertions(+), 47 deletions(-) diff --git a/ddtrace/tracer/option.go b/ddtrace/tracer/option.go index 86940d6dc02..1aa87fd20f1 100644 --- a/ddtrace/tracer/option.go +++ b/ddtrace/tracer/option.go @@ -341,12 +341,9 @@ func newConfig(opts ...StartOption) (*config, error) { agentURL := c.internalConfig.AgentURL() af := loadAgentFeatures(agentDisabled, agentURL, c.httpClient) c.agent.store(af) - // If the agent doesn't support the v1 protocol, downgrade to v0.4. - // Also downgrade if CSS is disabled (v1 requires CSS) or if OTLP span metrics are - // enabled: OTLP span metrics use their own concentrator and are not native CSS, so - // the trace transport should stay on v0.4 where the test agent and Datadog Agent - // can observe the Datadog-Client-Computed-Stats header set by resolveTraceTransport. - if c.internalConfig.TraceProtocol() == traceProtocolV1 && (!af.v1ProtocolAvailable || !c.canComputeStats() || c.internalConfig.OTLPSpanMetricsEnabled()) { + // If the agent doesn't support the v1 protocol, downgrade to v0.4 + // Also downgrade if CSS is disabled, as v1 is not compatible without CSS. + if c.internalConfig.TraceProtocol() == traceProtocolV1 && (!af.v1ProtocolAvailable || !c.canComputeStats()) { c.internalConfig.SetTraceProtocol(traceProtocolV04, internalconfig.OriginCalculated) if t, ok := c.ddTransport.(*httpTransport); ok && t.traceURL == agentURL.String()+tracesAPIPathV1 { t.traceURL = agentURL.String() + tracesAPIPath diff --git a/ddtrace/tracer/stats_to_otlp_metrics.go b/ddtrace/tracer/stats_to_otlp_metrics.go index b6e4c0a4cc4..8c1d0bb1165 100644 --- a/ddtrace/tracer/stats_to_otlp_metrics.go +++ b/ddtrace/tracer/stats_to_otlp_metrics.go @@ -96,38 +96,13 @@ func buildMetricsResource(payload *pb.ClientStatsPayload, otelMode bool, reportH attrs = append(attrs, otlpKeyValue("datadog.runtime_id", otlpStringValue(payload.RuntimeID))) } if payload.ProcessTags != "" { - // Reserved DD_TAGS keys whose semantic meaning is captured in dedicated - // resource attributes (service.name, deployment.environment.name, etc.) - // and must therefore be excluded from the generic tracer_dd_tags array. - reservedDDTagKeys := map[string]bool{ - "service": true, - "env": true, - "version": true, - "runtime_id": true, - "runtime-id": true, - } - var tracerDDTagEntries []*otlpcommon.AnyValue for tag := range strings.SplitSeq(payload.ProcessTags, ",") { parts := strings.SplitN(tag, ":", 2) - if len(parts) != 2 || parts[0] == "" || parts[1] == "" { - continue - } - // datadog. resource attributes for all non-runtime_id process tags. - if parts[0] != "runtime_id" { + // Skip keys emitted explicitly above (runtime_id) to avoid duplicate + // resource attributes, and skip tags with empty values. + if len(parts) == 2 && parts[0] != "" && parts[1] != "" && parts[0] != "runtime_id" { attrs = append(attrs, otlpKeyValue("datadog."+parts[0], otlpStringValue(parts[1]))) } - // tracer_dd_tags array: non-reserved key:value pairs from DD_TAGS / - // OTEL_RESOURCE_ATTRIBUTES that are not already captured elsewhere. - if !reservedDDTagKeys[parts[0]] { - tracerDDTagEntries = append(tracerDDTagEntries, otlpStringValue(tag)) - } - } - if len(tracerDDTagEntries) > 0 { - attrs = append(attrs, otlpKeyValue("tracer_dd_tags", &otlpcommon.AnyValue{ - Value: &otlpcommon.AnyValue_ArrayValue{ - ArrayValue: &otlpcommon.ArrayValue{Values: tracerDDTagEntries}, - }, - })) } } } @@ -242,19 +217,6 @@ func buildDataPointAttributes(gs *pb.ClientGroupedStats, isError bool, defaultSe if gs.Synthetics { attrs = append(attrs, otlpKeyValue("datadog.origin", otlpStringValue("synthetics"))) } - // additional_metric_tags holds the per-span tag values for keys listed in - // DD_TRACE_STATS_ADDITIONAL_TAGS (already populated in ClientGroupedStats by the concentrator). - if len(gs.AdditionalMetricTags) > 0 { - tagEntries := make([]*otlpcommon.AnyValue, 0, len(gs.AdditionalMetricTags)) - for _, t := range gs.AdditionalMetricTags { - tagEntries = append(tagEntries, otlpStringValue(t)) - } - attrs = append(attrs, otlpKeyValue("additional_metric_tags", &otlpcommon.AnyValue{ - Value: &otlpcommon.AnyValue_ArrayValue{ - ArrayValue: &otlpcommon.ArrayValue{Values: tagEntries}, - }, - })) - } } return attrs From 6b65e233bf06cece2160862a963806b750d4f39c Mon Sep 17 00:00:00 2001 From: Rachel Yang Date: Wed, 22 Jul 2026 18:17:36 -0400 Subject: [PATCH 14/17] Reapply "fix(tracer): add tracer_dd_tags, additional_metric_tags attrs and v0.4 downgrade for OTLP span metrics" This reverts commit 861a5991d792bf2f27942a628e9864e67fd05455. --- ddtrace/tracer/option.go | 9 +++-- ddtrace/tracer/stats_to_otlp_metrics.go | 44 +++++++++++++++++++++++-- 2 files changed, 47 insertions(+), 6 deletions(-) diff --git a/ddtrace/tracer/option.go b/ddtrace/tracer/option.go index 1aa87fd20f1..86940d6dc02 100644 --- a/ddtrace/tracer/option.go +++ b/ddtrace/tracer/option.go @@ -341,9 +341,12 @@ func newConfig(opts ...StartOption) (*config, error) { agentURL := c.internalConfig.AgentURL() af := loadAgentFeatures(agentDisabled, agentURL, c.httpClient) c.agent.store(af) - // If the agent doesn't support the v1 protocol, downgrade to v0.4 - // Also downgrade if CSS is disabled, as v1 is not compatible without CSS. - if c.internalConfig.TraceProtocol() == traceProtocolV1 && (!af.v1ProtocolAvailable || !c.canComputeStats()) { + // If the agent doesn't support the v1 protocol, downgrade to v0.4. + // Also downgrade if CSS is disabled (v1 requires CSS) or if OTLP span metrics are + // enabled: OTLP span metrics use their own concentrator and are not native CSS, so + // the trace transport should stay on v0.4 where the test agent and Datadog Agent + // can observe the Datadog-Client-Computed-Stats header set by resolveTraceTransport. + if c.internalConfig.TraceProtocol() == traceProtocolV1 && (!af.v1ProtocolAvailable || !c.canComputeStats() || c.internalConfig.OTLPSpanMetricsEnabled()) { c.internalConfig.SetTraceProtocol(traceProtocolV04, internalconfig.OriginCalculated) if t, ok := c.ddTransport.(*httpTransport); ok && t.traceURL == agentURL.String()+tracesAPIPathV1 { t.traceURL = agentURL.String() + tracesAPIPath diff --git a/ddtrace/tracer/stats_to_otlp_metrics.go b/ddtrace/tracer/stats_to_otlp_metrics.go index 8c1d0bb1165..b6e4c0a4cc4 100644 --- a/ddtrace/tracer/stats_to_otlp_metrics.go +++ b/ddtrace/tracer/stats_to_otlp_metrics.go @@ -96,13 +96,38 @@ func buildMetricsResource(payload *pb.ClientStatsPayload, otelMode bool, reportH attrs = append(attrs, otlpKeyValue("datadog.runtime_id", otlpStringValue(payload.RuntimeID))) } if payload.ProcessTags != "" { + // Reserved DD_TAGS keys whose semantic meaning is captured in dedicated + // resource attributes (service.name, deployment.environment.name, etc.) + // and must therefore be excluded from the generic tracer_dd_tags array. + reservedDDTagKeys := map[string]bool{ + "service": true, + "env": true, + "version": true, + "runtime_id": true, + "runtime-id": true, + } + var tracerDDTagEntries []*otlpcommon.AnyValue for tag := range strings.SplitSeq(payload.ProcessTags, ",") { parts := strings.SplitN(tag, ":", 2) - // Skip keys emitted explicitly above (runtime_id) to avoid duplicate - // resource attributes, and skip tags with empty values. - if len(parts) == 2 && parts[0] != "" && parts[1] != "" && parts[0] != "runtime_id" { + if len(parts) != 2 || parts[0] == "" || parts[1] == "" { + continue + } + // datadog. resource attributes for all non-runtime_id process tags. + if parts[0] != "runtime_id" { attrs = append(attrs, otlpKeyValue("datadog."+parts[0], otlpStringValue(parts[1]))) } + // tracer_dd_tags array: non-reserved key:value pairs from DD_TAGS / + // OTEL_RESOURCE_ATTRIBUTES that are not already captured elsewhere. + if !reservedDDTagKeys[parts[0]] { + tracerDDTagEntries = append(tracerDDTagEntries, otlpStringValue(tag)) + } + } + if len(tracerDDTagEntries) > 0 { + attrs = append(attrs, otlpKeyValue("tracer_dd_tags", &otlpcommon.AnyValue{ + Value: &otlpcommon.AnyValue_ArrayValue{ + ArrayValue: &otlpcommon.ArrayValue{Values: tracerDDTagEntries}, + }, + })) } } } @@ -217,6 +242,19 @@ func buildDataPointAttributes(gs *pb.ClientGroupedStats, isError bool, defaultSe if gs.Synthetics { attrs = append(attrs, otlpKeyValue("datadog.origin", otlpStringValue("synthetics"))) } + // additional_metric_tags holds the per-span tag values for keys listed in + // DD_TRACE_STATS_ADDITIONAL_TAGS (already populated in ClientGroupedStats by the concentrator). + if len(gs.AdditionalMetricTags) > 0 { + tagEntries := make([]*otlpcommon.AnyValue, 0, len(gs.AdditionalMetricTags)) + for _, t := range gs.AdditionalMetricTags { + tagEntries = append(tagEntries, otlpStringValue(t)) + } + attrs = append(attrs, otlpKeyValue("additional_metric_tags", &otlpcommon.AnyValue{ + Value: &otlpcommon.AnyValue_ArrayValue{ + ArrayValue: &otlpcommon.ArrayValue{Values: tagEntries}, + }, + })) + } } return attrs From 9100bf192509fc65cc6472e1a232b14afce56082 Mon Sep 17 00:00:00 2001 From: Rachel Yang Date: Wed, 22 Jul 2026 18:23:25 -0400 Subject: [PATCH 15/17] fix(tracer): switch OTLP trace writer to HTTP/JSON encoding (fixes fr15_3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The test_fr15_3 system test decodes the OTLP trace payload body as UTF-8 JSON. The writer was sending protobuf, causing a UnicodeDecodeError. Switch proto.Marshal → protojson.Marshal and update the content type to application/json. Update unit tests to decode with protojson.Unmarshal. Co-Authored-By: Claude Sonnet 4.6 --- ddtrace/tracer/otlp_writer.go | 5 +++-- ddtrace/tracer/otlp_writer_test.go | 9 +++++---- ddtrace/tracer/tracer_test.go | 14 +++++++------- 3 files changed, 15 insertions(+), 13 deletions(-) diff --git a/ddtrace/tracer/otlp_writer.go b/ddtrace/tracer/otlp_writer.go index 028beb77a19..734b5d38cd2 100644 --- a/ddtrace/tracer/otlp_writer.go +++ b/ddtrace/tracer/otlp_writer.go @@ -12,6 +12,7 @@ import ( otlpcommon "go.opentelemetry.io/proto/otlp/common/v1" otlpresource "go.opentelemetry.io/proto/otlp/resource/v1" otlptrace "go.opentelemetry.io/proto/otlp/trace/v1" + "google.golang.org/protobuf/encoding/protojson" "google.golang.org/protobuf/proto" "github.com/DataDog/dd-trace-go/v2/internal" @@ -113,7 +114,7 @@ func (w *otlpTraceWriter) flush() { }, }, } - b, err := proto.Marshal(tracesData) + b, err := protojson.Marshal(tracesData) readySpans = nil tracesData = nil if err != nil { @@ -126,7 +127,7 @@ func (w *otlpTraceWriter) flush() { retryInterval := w.config.internalConfig.RetryInterval() for attempt := 0; attempt <= sendRetries; attempt++ { log.Debug("OTLP: attempt %d to send payload: %d bytes, %d spans", attempt+1, len(b), spanCount) - sendErr = w.transport.send(b, otlpContentTypeProto) + sendErr = w.transport.send(b, otlpContentTypeJSON) if sendErr == nil { log.Debug("OTLP: sent traces after %d attempts", attempt+1) return diff --git a/ddtrace/tracer/otlp_writer_test.go b/ddtrace/tracer/otlp_writer_test.go index 992880db773..d9695511ab9 100644 --- a/ddtrace/tracer/otlp_writer_test.go +++ b/ddtrace/tracer/otlp_writer_test.go @@ -20,6 +20,7 @@ import ( "github.com/stretchr/testify/require" otlpcommon "go.opentelemetry.io/proto/otlp/common/v1" otlptrace "go.opentelemetry.io/proto/otlp/trace/v1" + "google.golang.org/protobuf/encoding/protojson" "google.golang.org/protobuf/proto" "github.com/DataDog/dd-trace-go/v2/internal" @@ -153,7 +154,7 @@ func TestOTLPWriterFlush(t *testing.T) { require.Equal(t, 1, len(payloads)) var tracesData otlptrace.TracesData - err := proto.Unmarshal(payloads[0], &tracesData) + err := protojson.Unmarshal(payloads[0], &tracesData) require.NoError(t, err) rs := tracesData.ResourceSpans @@ -334,11 +335,11 @@ func TestOTLPWriterConcurrency(t *testing.T) { assert.Equal(t, int32(numAdders*spansPerAdder), spansAdded.Load()) - // Verify all sent payloads are valid protobuf + // Verify all sent payloads are valid JSON totalSpans := 0 for _, data := range srv.getPayloads() { var td otlptrace.TracesData - err := proto.Unmarshal(data, &td) + err := protojson.Unmarshal(data, &td) require.NoError(t, err) for _, rs := range td.ResourceSpans { for _, ss := range rs.ScopeSpans { @@ -458,7 +459,7 @@ func TestOTLPWriterProcessTagsDisabled(t *testing.T) { for _, data := range srv.getPayloads() { var td otlptrace.TracesData - require.NoError(t, proto.Unmarshal(data, &td)) + require.NoError(t, protojson.Unmarshal(data, &td)) for _, rs := range td.ResourceSpans { for _, ss := range rs.ScopeSpans { for i, s := range ss.Spans { diff --git a/ddtrace/tracer/tracer_test.go b/ddtrace/tracer/tracer_test.go index a52eadac725..d93a5fa4a15 100644 --- a/ddtrace/tracer/tracer_test.go +++ b/ddtrace/tracer/tracer_test.go @@ -30,7 +30,7 @@ import ( otlpcommon "go.opentelemetry.io/proto/otlp/common/v1" otlptrace "go.opentelemetry.io/proto/otlp/trace/v1" "go.uber.org/goleak" - "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/encoding/protojson" "github.com/DataDog/dd-trace-go/v2/ddtrace" "github.com/DataDog/dd-trace-go/v2/ddtrace/ext" @@ -1493,7 +1493,7 @@ func TestOTLPExportModeStatsSkipped(t *testing.T) { require.NoError(t, err) w := trc.traceWriter.(*otlpTraceWriter) - w.transport = newOTLPTransport(srv.Client(), srv.URL, map[string]string{"Content-Type": "application/x-protobuf"}) + w.transport = newOTLPTransport(srv.Client(), srv.URL, nil) // Enable canDropP0s so submit() reaches the noopConcentrator // rather than short-circuiting. @@ -1523,7 +1523,7 @@ func TestOTLPExportModeStatsSkipped(t *testing.T) { totalSpans := 0 for _, p := range payloads { var td otlptrace.TracesData - require.NoError(t, proto.Unmarshal(p, &td)) + require.NoError(t, protojson.Unmarshal(p, &td)) for _, rs := range td.ResourceSpans { for _, ss := range rs.ScopeSpans { totalSpans += len(ss.Spans) @@ -1555,7 +1555,7 @@ func TestOTLPExportModeProcessTags(t *testing.T) { }) require.NoError(t, err) w := trc.traceWriter.(*otlpTraceWriter) - w.transport = newOTLPTransport(srv.Client(), srv.URL, map[string]string{"Content-Type": "application/x-protobuf"}) + w.transport = newOTLPTransport(srv.Client(), srv.URL, nil) setGlobalTracer(trc) t.Cleanup(func() { setGlobalTracer(&NoopTracer{}) }) return trc @@ -1566,7 +1566,7 @@ func TestOTLPExportModeProcessTags(t *testing.T) { var spans []*otlptrace.Span for _, p := range payloads { var td otlptrace.TracesData - require.NoError(t, proto.Unmarshal(p, &td)) + require.NoError(t, protojson.Unmarshal(p, &td)) for _, rs := range td.ResourceSpans { for _, ss := range rs.ScopeSpans { spans = append(spans, ss.Spans...) @@ -1706,7 +1706,7 @@ func TestOTLPExportModeSpanEventsRoundTrip(t *testing.T) { }) require.NoError(t, err) w := trc.traceWriter.(*otlpTraceWriter) - w.transport = newOTLPTransport(srv.Client(), srv.URL, map[string]string{"Content-Type": "application/x-protobuf"}) + w.transport = newOTLPTransport(srv.Client(), srv.URL, nil) setGlobalTracer(trc) t.Cleanup(func() { setGlobalTracer(&NoopTracer{}) }) @@ -1721,7 +1721,7 @@ func TestOTLPExportModeSpanEventsRoundTrip(t *testing.T) { var spans []*otlptrace.Span for _, p := range srv.getPayloads() { var td otlptrace.TracesData - require.NoError(t, proto.Unmarshal(p, &td)) + require.NoError(t, protojson.Unmarshal(p, &td)) for _, rs := range td.ResourceSpans { for _, ss := range rs.ScopeSpans { spans = append(spans, ss.Spans...) From a597b806fd442c1c9e7b6b5235ffbc3d632cf2b4 Mon Sep 17 00:00:00 2001 From: Rachel Yang Date: Thu, 23 Jul 2026 10:27:08 -0400 Subject: [PATCH 16/17] Revert "fix(tracer): switch OTLP trace writer to HTTP/JSON encoding (fixes fr15_3)" This reverts commit 44e38fe5eaf19fe756bcd07b418569588b892604. --- ddtrace/tracer/otlp_writer.go | 5 ++--- ddtrace/tracer/otlp_writer_test.go | 9 ++++----- ddtrace/tracer/tracer_test.go | 14 +++++++------- 3 files changed, 13 insertions(+), 15 deletions(-) diff --git a/ddtrace/tracer/otlp_writer.go b/ddtrace/tracer/otlp_writer.go index 734b5d38cd2..028beb77a19 100644 --- a/ddtrace/tracer/otlp_writer.go +++ b/ddtrace/tracer/otlp_writer.go @@ -12,7 +12,6 @@ import ( otlpcommon "go.opentelemetry.io/proto/otlp/common/v1" otlpresource "go.opentelemetry.io/proto/otlp/resource/v1" otlptrace "go.opentelemetry.io/proto/otlp/trace/v1" - "google.golang.org/protobuf/encoding/protojson" "google.golang.org/protobuf/proto" "github.com/DataDog/dd-trace-go/v2/internal" @@ -114,7 +113,7 @@ func (w *otlpTraceWriter) flush() { }, }, } - b, err := protojson.Marshal(tracesData) + b, err := proto.Marshal(tracesData) readySpans = nil tracesData = nil if err != nil { @@ -127,7 +126,7 @@ func (w *otlpTraceWriter) flush() { retryInterval := w.config.internalConfig.RetryInterval() for attempt := 0; attempt <= sendRetries; attempt++ { log.Debug("OTLP: attempt %d to send payload: %d bytes, %d spans", attempt+1, len(b), spanCount) - sendErr = w.transport.send(b, otlpContentTypeJSON) + sendErr = w.transport.send(b, otlpContentTypeProto) if sendErr == nil { log.Debug("OTLP: sent traces after %d attempts", attempt+1) return diff --git a/ddtrace/tracer/otlp_writer_test.go b/ddtrace/tracer/otlp_writer_test.go index d9695511ab9..992880db773 100644 --- a/ddtrace/tracer/otlp_writer_test.go +++ b/ddtrace/tracer/otlp_writer_test.go @@ -20,7 +20,6 @@ import ( "github.com/stretchr/testify/require" otlpcommon "go.opentelemetry.io/proto/otlp/common/v1" otlptrace "go.opentelemetry.io/proto/otlp/trace/v1" - "google.golang.org/protobuf/encoding/protojson" "google.golang.org/protobuf/proto" "github.com/DataDog/dd-trace-go/v2/internal" @@ -154,7 +153,7 @@ func TestOTLPWriterFlush(t *testing.T) { require.Equal(t, 1, len(payloads)) var tracesData otlptrace.TracesData - err := protojson.Unmarshal(payloads[0], &tracesData) + err := proto.Unmarshal(payloads[0], &tracesData) require.NoError(t, err) rs := tracesData.ResourceSpans @@ -335,11 +334,11 @@ func TestOTLPWriterConcurrency(t *testing.T) { assert.Equal(t, int32(numAdders*spansPerAdder), spansAdded.Load()) - // Verify all sent payloads are valid JSON + // Verify all sent payloads are valid protobuf totalSpans := 0 for _, data := range srv.getPayloads() { var td otlptrace.TracesData - err := protojson.Unmarshal(data, &td) + err := proto.Unmarshal(data, &td) require.NoError(t, err) for _, rs := range td.ResourceSpans { for _, ss := range rs.ScopeSpans { @@ -459,7 +458,7 @@ func TestOTLPWriterProcessTagsDisabled(t *testing.T) { for _, data := range srv.getPayloads() { var td otlptrace.TracesData - require.NoError(t, protojson.Unmarshal(data, &td)) + require.NoError(t, proto.Unmarshal(data, &td)) for _, rs := range td.ResourceSpans { for _, ss := range rs.ScopeSpans { for i, s := range ss.Spans { diff --git a/ddtrace/tracer/tracer_test.go b/ddtrace/tracer/tracer_test.go index d93a5fa4a15..a52eadac725 100644 --- a/ddtrace/tracer/tracer_test.go +++ b/ddtrace/tracer/tracer_test.go @@ -30,7 +30,7 @@ import ( otlpcommon "go.opentelemetry.io/proto/otlp/common/v1" otlptrace "go.opentelemetry.io/proto/otlp/trace/v1" "go.uber.org/goleak" - "google.golang.org/protobuf/encoding/protojson" + "google.golang.org/protobuf/proto" "github.com/DataDog/dd-trace-go/v2/ddtrace" "github.com/DataDog/dd-trace-go/v2/ddtrace/ext" @@ -1493,7 +1493,7 @@ func TestOTLPExportModeStatsSkipped(t *testing.T) { require.NoError(t, err) w := trc.traceWriter.(*otlpTraceWriter) - w.transport = newOTLPTransport(srv.Client(), srv.URL, nil) + w.transport = newOTLPTransport(srv.Client(), srv.URL, map[string]string{"Content-Type": "application/x-protobuf"}) // Enable canDropP0s so submit() reaches the noopConcentrator // rather than short-circuiting. @@ -1523,7 +1523,7 @@ func TestOTLPExportModeStatsSkipped(t *testing.T) { totalSpans := 0 for _, p := range payloads { var td otlptrace.TracesData - require.NoError(t, protojson.Unmarshal(p, &td)) + require.NoError(t, proto.Unmarshal(p, &td)) for _, rs := range td.ResourceSpans { for _, ss := range rs.ScopeSpans { totalSpans += len(ss.Spans) @@ -1555,7 +1555,7 @@ func TestOTLPExportModeProcessTags(t *testing.T) { }) require.NoError(t, err) w := trc.traceWriter.(*otlpTraceWriter) - w.transport = newOTLPTransport(srv.Client(), srv.URL, nil) + w.transport = newOTLPTransport(srv.Client(), srv.URL, map[string]string{"Content-Type": "application/x-protobuf"}) setGlobalTracer(trc) t.Cleanup(func() { setGlobalTracer(&NoopTracer{}) }) return trc @@ -1566,7 +1566,7 @@ func TestOTLPExportModeProcessTags(t *testing.T) { var spans []*otlptrace.Span for _, p := range payloads { var td otlptrace.TracesData - require.NoError(t, protojson.Unmarshal(p, &td)) + require.NoError(t, proto.Unmarshal(p, &td)) for _, rs := range td.ResourceSpans { for _, ss := range rs.ScopeSpans { spans = append(spans, ss.Spans...) @@ -1706,7 +1706,7 @@ func TestOTLPExportModeSpanEventsRoundTrip(t *testing.T) { }) require.NoError(t, err) w := trc.traceWriter.(*otlpTraceWriter) - w.transport = newOTLPTransport(srv.Client(), srv.URL, nil) + w.transport = newOTLPTransport(srv.Client(), srv.URL, map[string]string{"Content-Type": "application/x-protobuf"}) setGlobalTracer(trc) t.Cleanup(func() { setGlobalTracer(&NoopTracer{}) }) @@ -1721,7 +1721,7 @@ func TestOTLPExportModeSpanEventsRoundTrip(t *testing.T) { var spans []*otlptrace.Span for _, p := range srv.getPayloads() { var td otlptrace.TracesData - require.NoError(t, protojson.Unmarshal(p, &td)) + require.NoError(t, proto.Unmarshal(p, &td)) for _, rs := range td.ResourceSpans { for _, ss := range rs.ScopeSpans { spans = append(spans, ss.Spans...) From dadc740a70a4c7360a6b9d73cfe445efb42c6613 Mon Sep 17 00:00:00 2001 From: Rachel Yang Date: Thu, 23 Jul 2026 12:56:42 -0400 Subject: [PATCH 17/17] fix(tracer): remove non-RFC tracer_dd_tags/additional_metric_tags OTLP attrs Both were Go-specific extensions carrying over DD_TAGS/DD_TRACE_STATS_ADDITIONAL_TAGS semantics from the native /v0.6/stats path, but neither appears in the RFC's attribute tables. datadog. per-process-tag resource attributes (which are RFC-required) are unaffected. Co-Authored-By: Claude Sonnet 5 --- ddtrace/tracer/stats_to_otlp_metrics.go | 36 ------------------------- 1 file changed, 36 deletions(-) diff --git a/ddtrace/tracer/stats_to_otlp_metrics.go b/ddtrace/tracer/stats_to_otlp_metrics.go index b6e4c0a4cc4..1805e41f34d 100644 --- a/ddtrace/tracer/stats_to_otlp_metrics.go +++ b/ddtrace/tracer/stats_to_otlp_metrics.go @@ -96,17 +96,6 @@ func buildMetricsResource(payload *pb.ClientStatsPayload, otelMode bool, reportH attrs = append(attrs, otlpKeyValue("datadog.runtime_id", otlpStringValue(payload.RuntimeID))) } if payload.ProcessTags != "" { - // Reserved DD_TAGS keys whose semantic meaning is captured in dedicated - // resource attributes (service.name, deployment.environment.name, etc.) - // and must therefore be excluded from the generic tracer_dd_tags array. - reservedDDTagKeys := map[string]bool{ - "service": true, - "env": true, - "version": true, - "runtime_id": true, - "runtime-id": true, - } - var tracerDDTagEntries []*otlpcommon.AnyValue for tag := range strings.SplitSeq(payload.ProcessTags, ",") { parts := strings.SplitN(tag, ":", 2) if len(parts) != 2 || parts[0] == "" || parts[1] == "" { @@ -116,18 +105,6 @@ func buildMetricsResource(payload *pb.ClientStatsPayload, otelMode bool, reportH if parts[0] != "runtime_id" { attrs = append(attrs, otlpKeyValue("datadog."+parts[0], otlpStringValue(parts[1]))) } - // tracer_dd_tags array: non-reserved key:value pairs from DD_TAGS / - // OTEL_RESOURCE_ATTRIBUTES that are not already captured elsewhere. - if !reservedDDTagKeys[parts[0]] { - tracerDDTagEntries = append(tracerDDTagEntries, otlpStringValue(tag)) - } - } - if len(tracerDDTagEntries) > 0 { - attrs = append(attrs, otlpKeyValue("tracer_dd_tags", &otlpcommon.AnyValue{ - Value: &otlpcommon.AnyValue_ArrayValue{ - ArrayValue: &otlpcommon.ArrayValue{Values: tracerDDTagEntries}, - }, - })) } } } @@ -242,19 +219,6 @@ func buildDataPointAttributes(gs *pb.ClientGroupedStats, isError bool, defaultSe if gs.Synthetics { attrs = append(attrs, otlpKeyValue("datadog.origin", otlpStringValue("synthetics"))) } - // additional_metric_tags holds the per-span tag values for keys listed in - // DD_TRACE_STATS_ADDITIONAL_TAGS (already populated in ClientGroupedStats by the concentrator). - if len(gs.AdditionalMetricTags) > 0 { - tagEntries := make([]*otlpcommon.AnyValue, 0, len(gs.AdditionalMetricTags)) - for _, t := range gs.AdditionalMetricTags { - tagEntries = append(tagEntries, otlpStringValue(t)) - } - attrs = append(attrs, otlpKeyValue("additional_metric_tags", &otlpcommon.AnyValue{ - Value: &otlpcommon.AnyValue_ArrayValue{ - ArrayValue: &otlpcommon.ArrayValue{Values: tagEntries}, - }, - })) - } } return attrs