diff --git a/ddtrace/tracer/log.go b/ddtrace/tracer/log.go index 774caa365a..7a5d0c6510 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 4dd70fdcdf..ba70f694e2 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 @@ -442,7 +445,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/otlp_metrics_wireup_test.go b/ddtrace/tracer/otlp_metrics_wireup_test.go new file mode 100644 index 0000000000..565df940a0 --- /dev/null +++ b/ddtrace/tracer/otlp_metrics_wireup_test.go @@ -0,0 +1,214 @@ +// 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" + + tinternal "github.com/DataDog/dd-trace-go/v2/ddtrace/tracer/internal" + 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{ + transport: newOTLPTransport(otlpSrv.Server.Client(), otlpSrv.URL+"/v1/metrics", nil), + 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") +} + +// 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{ + transport: newOTLPTransport(otlpSrv.Server.Client(), otlpSrv.URL+"/v1/metrics", nil), + protocol: "http/json", + cfg: cfg.internalConfig, + } + c.otlpPeerTags = []string{} + + 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) { + 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.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") + }) + + 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_transport_test.go b/ddtrace/tracer/otlp_transport_test.go index c553e9ed83..cde4655057 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.go b/ddtrace/tracer/otlp_writer.go index 5f4398d325..028beb77a1 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,12 +31,17 @@ 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 { 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", otlpStringValue("true")), + ) + } scope := &otlpcommon.InstrumentationScope{Name: "dd-trace-go", Version: version.Tag} baseSize := proto.Size(&otlptrace.TracesData{ ResourceSpans: []*otlptrace.ResourceSpans{{ @@ -55,7 +59,6 @@ func newOTLPTraceWriter(c *config) *otlpTraceWriter { spans: make([]*otlptrace.Span, 0), buffSize: baseSize, baseSize: baseSize, - climit: make(chan struct{}, concurrentConnectionLimit), } } @@ -96,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 335feb7c10..87ffd4db03 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 0970df9df5..992880db77 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) }) @@ -262,14 +255,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) @@ -283,9 +276,8 @@ 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), atomic.LoadInt32(&totalRequests)) + assert.Equal(t, int32(tc.expAttempts), totalRequests.Load()) assert.Equal(t, tc.tracesSent, len(srv.getPayloads()) > 0) }) } @@ -314,14 +306,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 +332,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 @@ -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() }) } diff --git a/ddtrace/tracer/stats.go b/ddtrace/tracer/stats.go index 473ef154c9..308a9de550 100644 --- a/ddtrace/tracer/stats.go +++ b/ddtrace/tracer/stats.go @@ -6,6 +6,7 @@ package tracer import ( + "maps" "sync" "sync/atomic" "time" @@ -64,6 +65,12 @@ 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. + otlpExporter *otlpMetricsExporter + + // otlpPeerTags, when non-nil, overrides agent-advertised peer tags for OTLP span metrics. + otlpPeerTags []string } type tracerStatSpan struct { @@ -192,7 +199,33 @@ func (c *concentrator) newTracerStatSpan(s *Span, obfuscator *obfuscate.Obfuscat } httpMethod, _ := s.meta.Get(ext.HTTPMethod) httpEndpoint, _ := s.meta.Get(ext.HTTPEndpoint) + if httpEndpoint == "" { + // OTel spans set http.route; DD spans use http.endpoint — both map to HTTPEndpoint. + 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 + if c.otlpPeerTags != nil { + peerTags = c.otlpPeerTags + // 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 { + if _, ok := spanMeta[k]; ok { + hasPeerTag = true + break + } + } + if hasPeerTag { + orig := spanMeta + spanMeta = make(map[string]string, len(orig)+1) + maps.Copy(spanMeta, orig) + spanMeta[ext.SpanKind] = ext.SpanKindClient + } + } + } statSpan, ok := c.spanConcentrator.NewStatSpanWithConfig(stats.StatSpanConfig{ Service: s.service, Resource: resource, @@ -202,9 +235,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: agentInfo.peerTags, + PeerTags: peerTags, AdditionalMetricTagKeys: c.cfg.internalConfig.StatsAdditionalTags(), HTTPMethod: httpMethod, HTTPEndpoint: httpEndpoint, @@ -261,20 +294,40 @@ 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. +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) { + // 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) bc := c.spanConcentrator.DrainBlockCounts() c.emitCollapseMetrics(bc) - - 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 @@ -291,14 +344,20 @@ 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 attempt < sendRetries { - time.Sleep(retryInterval) + if c.otlpExporter != nil { + err = sendWithRetry(sendRetries, retryInterval, func() error { + return c.otlpExporter.export(csp) + }) + } 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) } + 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) @@ -308,6 +367,17 @@ 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 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) + // 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 +} + // emitCollapseMetrics sends health and instrumentation telemetry for cardinality collapse events. // Per the Cardinality Limits RFC: // - Health metric: datadog.tracer.stats.collapsed_spans (statsd, public) diff --git a/ddtrace/tracer/stats_to_otlp_metrics.go b/ddtrace/tracer/stats_to_otlp_metrics.go index 8c1d0bb116..1805e41f34 100644 --- a/ddtrace/tracer/stats_to_otlp_metrics.go +++ b/ddtrace/tracer/stats_to_otlp_metrics.go @@ -98,9 +98,11 @@ func buildMetricsResource(payload *pb.ClientStatsPayload, otelMode bool, reportH if payload.ProcessTags != "" { 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]))) } } diff --git a/ddtrace/tracer/tracer.go b/ddtrace/tracer/tracer.go index 10c6d38903..6bf92ac848 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,14 @@ 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 and exports stats; agent /v0.6/stats path unused. + sc = newOTLPMetricsConcentrator(c, statsd) + } else if c.internalConfig.OTLPExportMode() { sc = &noopConcentrator{} + } else { + sc = newConcentrator(c, defaultStatsBucketSize, statsd) } t = &tracer{ config: c, @@ -1212,19 +1218,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 +1239,8 @@ func (t *tracer) submit(s *Span) { if !t.config.internalConfig.TracingEnabled() { return } - // we have an active tracer - if !t.config.canDropP0s() { + // Submit to the concentrator when native stats (canDropP0s) or OTLP span metrics are 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 a5797c37b5..686cfc995c 100644 --- a/ddtrace/tracer/transport.go +++ b/ddtrace/tracer/transport.go @@ -208,10 +208,11 @@ 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 { + // "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)) partialTraces := int(tracerstats.Count(tracerstats.PartialTraces)) diff --git a/ddtrace/tracer/transport_test.go b/ddtrace/tracer/transport_test.go index 43c8b74354..fb4cce3e9f 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 9a7f0cb53a..43f48d4211 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -1629,6 +1629,17 @@ 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) OTLPMetricsURL() string { c.mu.RLock() defer c.mu.RUnlock()