diff --git a/ddtrace/tracer/otlp_metrics_exporter.go b/ddtrace/tracer/otlp_metrics_exporter.go new file mode 100644 index 0000000000..8c8d7fef08 --- /dev/null +++ b/ddtrace/tracer/otlp_metrics_exporter.go @@ -0,0 +1,104 @@ +// 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" + "fmt" + + pb "github.com/DataDog/datadog-agent/pkg/proto/pbgo/trace" + otlpmetrics "go.opentelemetry.io/proto/otlp/metrics/v1" + "google.golang.org/protobuf/encoding/protojson" + "google.golang.org/protobuf/encoding/protowire" + "google.golang.org/protobuf/proto" + + "github.com/DataDog/dd-trace-go/v2/internal" + internalconfig "github.com/DataDog/dd-trace-go/v2/internal/config" + "github.com/DataDog/dd-trace-go/v2/internal/log" +) + +// otlpMetricsExporter converts ClientStatsPayload to OTLP metrics and sends them over HTTP. +type otlpMetricsExporter struct { + transport *otlpTransport + protocol string // "http/json" or "http/protobuf" + cfg *internalconfig.Config +} + +func newOTLPMetricsExporter(cfg *internalconfig.Config) *otlpMetricsExporter { + return &otlpMetricsExporter{ + transport: newOTLPTransport( + internal.DefaultHTTPClient(cfg.AgentTimeout(), false), + cfg.OTLPMetricsURL(), + cfg.OTLPMetricsHeaders(), + ), + protocol: cfg.OTLPMetricsProtocol(), + cfg: cfg, + } +} + +// export converts payload to an OTLP ExportMetricsServiceRequest and sends it. +func (e *otlpMetricsExporter) export(payload *pb.ClientStatsPayload) error { + rms := buildOTLPMetricsRequest(payload, e.cfg) + if len(rms) == 0 { + return nil + } + + var body []byte + var contentType string + var err error + + if e.protocol == "http/json" { + body, err = marshalExportRequestJSON(rms) + contentType = otlpContentTypeJSON + } else { + body, err = marshalExportRequestProto(rms) + contentType = otlpContentTypeProto + } + if err != nil { + return fmt.Errorf("otlp_metrics_exporter: marshal failed: %w", err) + } + + // No retry: a failed metrics interval is dropped rather than retried. + // Span metrics are lossy by design — the next flush interval replaces the lost window. + if sendErr := e.transport.send(body, contentType); sendErr != nil { + log.Error("otlp_metrics_exporter: export to %s failed: %v", e.transport.endpoint, sendErr.Error()) + return sendErr + } + log.Debug("otlp_metrics_exporter: exported %d bytes (%s) to %s", len(body), e.protocol, e.transport.endpoint) + return nil +} + +// marshalExportRequestProto hand-encodes ExportMetricsServiceRequest (field 1: repeated ResourceMetrics) +// to avoid importing the collector package, which conflicts with some contrib modules via google.golang.org/genproto. +func marshalExportRequestProto(rms []*otlpmetrics.ResourceMetrics) ([]byte, error) { + var buf []byte + for _, rm := range rms { + b, err := proto.Marshal(rm) + if err != nil { + return nil, err + } + buf = protowire.AppendTag(buf, 1, protowire.BytesType) + buf = protowire.AppendBytes(buf, b) + } + return buf, nil +} + +// marshalExportRequestJSON encodes a ResourceMetrics slice as the JSON of +// ExportMetricsServiceRequest: {"resourceMetrics": [...]}. +func marshalExportRequestJSON(rms []*otlpmetrics.ResourceMetrics) ([]byte, error) { + type exportReq struct { + ResourceMetrics []json.RawMessage `json:"resourceMetrics"` + } + items := make([]json.RawMessage, 0, len(rms)) + for _, rm := range rms { + b, err := protojson.Marshal(rm) + if err != nil { + return nil, err + } + items = append(items, json.RawMessage(b)) + } + return json.Marshal(exportReq{ResourceMetrics: items}) +} diff --git a/ddtrace/tracer/otlp_metrics_exporter_test.go b/ddtrace/tracer/otlp_metrics_exporter_test.go new file mode 100644 index 0000000000..838415aeea --- /dev/null +++ b/ddtrace/tracer/otlp_metrics_exporter_test.go @@ -0,0 +1,183 @@ +// 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" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + pb "github.com/DataDog/datadog-agent/pkg/proto/pbgo/trace" + otlpmetrics "go.opentelemetry.io/proto/otlp/metrics/v1" + "google.golang.org/protobuf/encoding/protowire" + "google.golang.org/protobuf/proto" + + internalconfig "github.com/DataDog/dd-trace-go/v2/internal/config" +) + +// captureServer creates a test HTTP server that records the last request it received. +type captureServer struct { + *httptest.Server + lastBody []byte + lastContentType string +} + +func newCaptureServer(t *testing.T) *captureServer { + t.Helper() + cs := &captureServer{} + cs.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + cs.lastContentType = r.Header.Get("Content-Type") + cs.lastBody, _ = io.ReadAll(r.Body) + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(cs.Server.Close) + return cs +} + +func makeExporterWithServer(t *testing.T, srv *captureServer, protocol string) *otlpMetricsExporter { + t.Helper() + cfg := internalconfig.CreateNew() + return &otlpMetricsExporter{ + transport: newOTLPTransport(srv.Server.Client(), srv.URL+"/v1/metrics", nil), + protocol: protocol, + cfg: cfg, + } +} + +// ---- otlpMetricsExporter.export ---- + +func TestOTLPMetricsExporterExportEmptyPayload(t *testing.T) { + srv := newCaptureServer(t) + exp := makeExporterWithServer(t, srv, "http/json") + // A payload with no groups yields a nil request; no HTTP call is made. + err := exp.export(makePayload("svc", "", "", nil)) + require.NoError(t, err) + assert.Empty(t, srv.lastBody, "no HTTP call expected for empty payload") +} + +func TestOTLPMetricsExporterExportJSONContentType(t *testing.T) { + srv := newCaptureServer(t) + exp := makeExporterWithServer(t, srv, "http/json") + gs := &pb.ClientGroupedStats{ + Service: "svc", + Resource: "web.request", + OkSummary: encodeSketch(t, 50e6), + } + err := exp.export(makePayload("svc", "", "", []*pb.ClientGroupedStats{gs})) + require.NoError(t, err) + assert.Equal(t, otlpContentTypeJSON, srv.lastContentType) + assert.NotEmpty(t, srv.lastBody) +} + +func TestOTLPMetricsExporterExportProtoContentType(t *testing.T) { + srv := newCaptureServer(t) + exp := makeExporterWithServer(t, srv, "http/protobuf") + gs := &pb.ClientGroupedStats{ + Service: "svc", + Resource: "web.request", + OkSummary: encodeSketch(t, 50e6), + } + err := exp.export(makePayload("svc", "", "", []*pb.ClientGroupedStats{gs})) + require.NoError(t, err) + assert.Equal(t, otlpContentTypeProto, srv.lastContentType) + assert.NotEmpty(t, srv.lastBody) +} + +func TestOTLPMetricsExporterExportJSONIsValidOTLP(t *testing.T) { + srv := newCaptureServer(t) + exp := makeExporterWithServer(t, srv, "http/json") + gs := &pb.ClientGroupedStats{ + Service: "svc", + Resource: "web.request", + OkSummary: encodeSketch(t, 50e6), + } + require.NoError(t, exp.export(makePayload("svc", "prod", "1.0", []*pb.ClientGroupedStats{gs}))) + + // The body must be valid JSON with the expected metric name. + var parsed map[string]any + require.NoError(t, json.Unmarshal(srv.lastBody, &parsed)) + body := string(srv.lastBody) + assert.Contains(t, body, spanDurationMetricName) + assert.Contains(t, body, "service.name") +} + +func TestOTLPMetricsExporterExportProtobufIsDecodable(t *testing.T) { + srv := newCaptureServer(t) + exp := makeExporterWithServer(t, srv, "http/protobuf") + gs := &pb.ClientGroupedStats{ + Service: "svc", + Resource: "web.request", + OkSummary: encodeSketch(t, 50e6), + } + require.NoError(t, exp.export(makePayload("svc", "", "", []*pb.ClientGroupedStats{gs}))) + + // Decode without importing collector/metrics/v1 to avoid the genproto split + // ambiguity with confluent-kafka-go. ExportMetricsServiceRequest wire format: + // field 1 (bytes) = repeated ResourceMetrics. + var resourceMetrics []*otlpmetrics.ResourceMetrics + b := srv.lastBody + for len(b) > 0 { + num, typ, n := protowire.ConsumeTag(b) + require.Positive(t, n) + b = b[n:] + val, n := protowire.ConsumeBytes(b) + require.Positive(t, n) + b = b[n:] + if num == 1 && typ == protowire.BytesType { + var rm otlpmetrics.ResourceMetrics + require.NoError(t, proto.Unmarshal(val, &rm)) + resourceMetrics = append(resourceMetrics, &rm) + } + } + require.Len(t, resourceMetrics, 1) + require.Len(t, resourceMetrics[0].ScopeMetrics, 1) + assert.Equal(t, spanDurationMetricName, resourceMetrics[0].ScopeMetrics[0].Metrics[0].Name) +} + +func TestOTLPMetricsExporterExportHTTPError(t *testing.T) { + errSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + })) + t.Cleanup(errSrv.Close) + exp := &otlpMetricsExporter{ + transport: newOTLPTransport(errSrv.Client(), errSrv.URL, nil), + protocol: "http/json", + cfg: internalconfig.CreateNew(), + } + gs := &pb.ClientGroupedStats{Service: "svc", Resource: "op", OkSummary: encodeSketch(t, 50e6)} + err := exp.export(makePayload("svc", "", "", []*pb.ClientGroupedStats{gs})) + require.Error(t, err) + assert.Contains(t, err.Error(), "500") +} + +func TestOTLPMetricsExporterCustomHeaders(t *testing.T) { + var gotHeader string + hdrSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotHeader = r.Header.Get("X-Custom-Header") + w.WriteHeader(http.StatusOK) + })) + t.Cleanup(hdrSrv.Close) + exp := &otlpMetricsExporter{ + transport: newOTLPTransport(hdrSrv.Client(), hdrSrv.URL, map[string]string{"X-Custom-Header": "my-value"}), + protocol: "http/json", + cfg: internalconfig.CreateNew(), + } + gs := &pb.ClientGroupedStats{Service: "svc", Resource: "op", OkSummary: encodeSketch(t, 50e6)} + require.NoError(t, exp.export(makePayload("svc", "", "", []*pb.ClientGroupedStats{gs}))) + assert.Equal(t, "my-value", gotHeader) +} + +// ---- config integration ---- + +func TestOTLPMetricsProtocolDefaultIsProtobuf(t *testing.T) { + cfg := internalconfig.CreateNew() + assert.Equal(t, "http/protobuf", cfg.OTLPMetricsProtocol()) +} diff --git a/ddtrace/tracer/otlp_transport.go b/ddtrace/tracer/otlp_transport.go index b4a1acc9ef..60cd460182 100644 --- a/ddtrace/tracer/otlp_transport.go +++ b/ddtrace/tracer/otlp_transport.go @@ -12,6 +12,11 @@ import ( "net/http" ) +const ( + otlpContentTypeJSON = "application/json" + otlpContentTypeProto = "application/x-protobuf" +) + // otlpTransport sends protobuf-encoded OTLP payloads over HTTP. // It is the OTLP counterpart to httpTransport (which handles Datadog-protocol traffic). type otlpTransport struct { @@ -28,13 +33,13 @@ func newOTLPTransport(client *http.Client, endpoint string, customHeaders map[st } } -// send posts a protobuf-encoded payload to the configured OTLP endpoint. -func (t *otlpTransport) send(data []byte) error { +// send posts a payload to the configured OTLP endpoint with the given content type. +func (t *otlpTransport) send(data []byte, contentType string) error { req, err := http.NewRequest("POST", t.endpoint, bytes.NewReader(data)) if err != nil { return fmt.Errorf("cannot create http request: %w", err) } - req.Header.Set("Content-Type", "application/x-protobuf") + req.Header.Set("Content-Type", contentType) for header, value := range t.customHeaders { req.Header.Set(header, value) } diff --git a/ddtrace/tracer/otlp_transport_test.go b/ddtrace/tracer/otlp_transport_test.go index 555ca87373..c553e9ed83 100644 --- a/ddtrace/tracer/otlp_transport_test.go +++ b/ddtrace/tracer/otlp_transport_test.go @@ -26,7 +26,7 @@ func TestOTLPTransportSendSuccess(t *testing.T) { defer srv.Close() tr := newOTLPTransport(srv.Client(), srv.URL, nil) - err := tr.send([]byte("hello")) + err := tr.send([]byte("hello"), "application/x-protobuf") require.NoError(t, err) assert.Equal(t, []byte("hello"), received) } @@ -43,10 +43,10 @@ func TestOTLPTransportSendHeaders(t *testing.T) { "Api-Key": "secret", "X-Custom": "value", }) - err := tr.send([]byte("data")) + err := tr.send([]byte("data"), "application/x-protobuf") require.NoError(t, err) - assert.Equal(t, "application/x-protobuf", gotHeaders.Get("Content-Type"), "default Content-Type must be set") + assert.Equal(t, "application/x-protobuf", gotHeaders.Get("Content-Type"), "Content-Type is forwarded from caller") assert.Equal(t, "secret", gotHeaders.Get("Api-Key")) assert.Equal(t, "value", gotHeaders.Get("X-Custom")) } @@ -60,7 +60,7 @@ func TestOTLPTransportSendHTTPMethod(t *testing.T) { defer srv.Close() tr := newOTLPTransport(srv.Client(), srv.URL, nil) - err := tr.send([]byte("data")) + err := tr.send([]byte("data"), "application/x-protobuf") require.NoError(t, err) assert.Equal(t, "POST", gotMethod) } @@ -83,7 +83,7 @@ func TestOTLPTransportSendErrorStatus(t *testing.T) { defer srv.Close() tr := newOTLPTransport(srv.Client(), srv.URL, nil) - err := tr.send([]byte("data")) + err := tr.send([]byte("data"), "application/x-protobuf") require.Error(t, err) assert.Contains(t, err.Error(), tt.text) }) @@ -92,7 +92,7 @@ func TestOTLPTransportSendErrorStatus(t *testing.T) { func TestOTLPTransportSendConnectionError(t *testing.T) { tr := newOTLPTransport(http.DefaultClient, "http://127.0.0.1:0/nonexistent", nil) - err := tr.send([]byte("data")) + err := tr.send([]byte("data"), "application/x-protobuf") require.Error(t, err) } @@ -111,7 +111,7 @@ func TestOTLPTransportConnectionReuse(t *testing.T) { tr := newOTLPTransport(srv.Client(), srv.URL, nil) for range 5 { - require.NoError(t, tr.send([]byte("data"))) + require.NoError(t, tr.send([]byte("data"), "application/x-protobuf")) } assert.Equal(t, int64(1), atomic.LoadInt64(&connCount), "expected a single connection to be reused across sends") diff --git a/ddtrace/tracer/otlp_writer.go b/ddtrace/tracer/otlp_writer.go index f772b4dd16..5f4398d325 100644 --- a/ddtrace/tracer/otlp_writer.go +++ b/ddtrace/tracer/otlp_writer.go @@ -131,7 +131,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) + 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/span_to_otlp.go b/ddtrace/tracer/span_to_otlp.go index a5c7fa332b..a694186b02 100644 --- a/ddtrace/tracer/span_to_otlp.go +++ b/ddtrace/tracer/span_to_otlp.go @@ -27,25 +27,31 @@ const maxAttributesCount = 128 // Resource construction // ----------------------------------------------------------------------------- -// buildResource constructs the OTLP Resource from resolved tracer configuration. -// If cfg is nil, an empty resource is returned. -func buildResource(cfg *internalconfig.Config) *otlpresource.Resource { - if cfg == nil { - return &otlpresource.Resource{} - } +// buildBaseResourceAttrs returns the telemetry.sdk.* and service.* resource attributes +// shared by both trace and metrics OTLP exports. +func buildBaseResourceAttrs(serviceName, svcVersion, env string) []*otlpcommon.KeyValue { attrs := []*otlpcommon.KeyValue{ - otlpKeyValue("service.name", otlpStringValue(cfg.ServiceName())), + otlpKeyValue("service.name", otlpStringValue(serviceName)), otlpKeyValue("telemetry.sdk.language", otlpStringValue("go")), otlpKeyValue("telemetry.sdk.name", otlpStringValue("datadog")), otlpKeyValue("telemetry.sdk.version", otlpStringValue(version.Tag)), } - if v := cfg.Env(); v != "" { - attrs = append(attrs, otlpKeyValue("deployment.environment.name", otlpStringValue(v))) + if env != "" { + attrs = append(attrs, otlpKeyValue("deployment.environment.name", otlpStringValue(env))) } - if v := cfg.Version(); v != "" { - attrs = append(attrs, otlpKeyValue("service.version", otlpStringValue(v))) + if svcVersion != "" { + attrs = append(attrs, otlpKeyValue("service.version", otlpStringValue(svcVersion))) + } + return attrs +} + +// buildResource constructs the OTLP Resource from resolved tracer configuration. +// If cfg is nil, an empty resource is returned. +func buildResource(cfg *internalconfig.Config) *otlpresource.Resource { + if cfg == nil { + return &otlpresource.Resource{} } - return &otlpresource.Resource{Attributes: attrs} + return &otlpresource.Resource{Attributes: buildBaseResourceAttrs(cfg.ServiceName(), cfg.Version(), cfg.Env())} } // ----------------------------------------------------------------------------- diff --git a/ddtrace/tracer/stats_to_otlp_metrics.go b/ddtrace/tracer/stats_to_otlp_metrics.go index 851f829916..8c1d0bb116 100644 --- a/ddtrace/tracer/stats_to_otlp_metrics.go +++ b/ddtrace/tracer/stats_to_otlp_metrics.go @@ -22,7 +22,6 @@ import ( internalconfig "github.com/DataDog/dd-trace-go/v2/internal/config" "github.com/DataDog/dd-trace-go/v2/internal/log" - "github.com/DataDog/dd-trace-go/v2/internal/version" ) const spanDurationMetricName = "traces.span.sdk.metrics.duration" @@ -88,18 +87,7 @@ func buildOTLPMetricsRequest(payload *pb.ClientStatsPayload, cfg *internalconfig // buildMetricsResource builds the OTLP Resource; adds host.name and datadog.* attrs in default mode. func buildMetricsResource(payload *pb.ClientStatsPayload, otelMode bool, reportHostname bool, hostname string) *otlpresource.Resource { - attrs := []*otlpcommon.KeyValue{ - otlpKeyValue("telemetry.sdk.language", otlpStringValue("go")), - otlpKeyValue("telemetry.sdk.name", otlpStringValue("datadog")), - otlpKeyValue("telemetry.sdk.version", otlpStringValue(version.Tag)), - otlpKeyValue("service.name", otlpStringValue(payload.Service)), - } - if payload.Version != "" { - attrs = append(attrs, otlpKeyValue("service.version", otlpStringValue(payload.Version))) - } - if payload.Env != "" { - attrs = append(attrs, otlpKeyValue("deployment.environment.name", otlpStringValue(payload.Env))) - } + attrs := buildBaseResourceAttrs(payload.Service, payload.Version, payload.Env) if reportHostname && hostname != "" { attrs = append(attrs, otlpKeyValue("host.name", otlpStringValue(hostname))) } @@ -208,13 +196,6 @@ func buildDataPointAttributes(gs *pb.ClientGroupedStats, isError bool, defaultSe statusCode = 2 // STATUS_CODE_ERROR } attrs = append(attrs, otlpKeyValue("status.code", otlpIntValue(statusCode))) - // grpc.method.name arrives via PeerTags (no dedicated field in ClientGroupedStats) and maps to rpc.method. - for _, tag := range gs.PeerTags { - if k, v, ok := strings.Cut(tag, ":"); ok && k == "grpc.method.name" && v != "" { - attrs = append(attrs, otlpKeyValue("rpc.method", otlpStringValue(v))) - break - } - } if svc := gs.Service; svc != "" && svc != defaultService { attrs = append(attrs, otlpKeyValue("service.name", otlpStringValue(svc))) @@ -230,6 +211,9 @@ func buildDataPointAttributes(gs *pb.ClientGroupedStats, isError bool, defaultSe } // top_level is true only when all spans in the group were top-level (TopLevelHits == Hits). attrs = append(attrs, otlpKeyValue("datadog.span.top_level", otlpBoolValue(gs.Hits > 0 && gs.TopLevelHits == gs.Hits))) + // ClientGroupedStats carries only a boolean Synthetics field; finer-grained + // origin values (synthetics-browser, rum, ciapp-test, lambda) are not available + // at the stats aggregation layer and require a proto change upstream to support. if gs.Synthetics { attrs = append(attrs, otlpKeyValue("datadog.origin", otlpStringValue("synthetics"))) } diff --git a/ddtrace/tracer/stats_to_otlp_metrics_test.go b/ddtrace/tracer/stats_to_otlp_metrics_test.go index a7a7944128..85c35d6346 100644 --- a/ddtrace/tracer/stats_to_otlp_metrics_test.go +++ b/ddtrace/tracer/stats_to_otlp_metrics_test.go @@ -381,37 +381,6 @@ func TestDataPointAttributesOptionalFieldsAbsentWhenUnset(t *testing.T) { assert.NotContains(t, m, "rpc.response.status_code") } -func TestDataPointAttributesGRPCMethodName(t *testing.T) { - // grpc.method.name in PeerTags is translated to the RFC-required rpc.method attribute. - gs := &pb.ClientGroupedStats{ - Resource: "grpc.request", - PeerTags: []string{"grpc.method.name:GetUser"}, - } - m := kvAttrsToMap(buildDataPointAttributes(gs, false, "", true)) - assert.Equal(t, "GetUser", m["rpc.method"]) - assert.NotContains(t, m, "grpc.method.name") -} - -func TestDataPointAttributesGRPCMethodNameFirstOnly(t *testing.T) { - // Only the first grpc.method.name value is used; no duplicates emitted. - gs := &pb.ClientGroupedStats{ - Resource: "grpc.request", - PeerTags: []string{"grpc.method.name:GetUser", "grpc.method.name:ListUsers"}, - } - m := kvAttrsToMap(buildDataPointAttributes(gs, false, "", true)) - assert.Equal(t, "GetUser", m["rpc.method"]) -} - -func TestDataPointAttributesGRPCMethodNameEmptySkipped(t *testing.T) { - // A grpc.method.name tag with an empty value must not emit rpc.method. - gs := &pb.ClientGroupedStats{ - Resource: "grpc.request", - PeerTags: []string{"grpc.method.name:"}, - } - m := kvAttrsToMap(buildDataPointAttributes(gs, false, "", true)) - assert.NotContains(t, m, "rpc.method") -} - func TestDataPointAttributesPeerTagsNotEmitted(t *testing.T) { // peer.* tags and other non-grpc.method.name peer tags are not forwarded (out of scope per RFC). gs := &pb.ClientGroupedStats{ diff --git a/internal/config/config.go b/internal/config/config.go index 0e4c96233c..9a7f0cb53a 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -216,6 +216,8 @@ type Config struct { otlpMetricsHeaders map[string]string // otlpMetricsFlushInterval is the span metrics flush cadence (default 10s). otlpMetricsFlushInterval time.Duration + // otlpMetricsProtocol is the OTLP export protocol for metrics: "http/json" or "http/protobuf". + otlpMetricsProtocol string // traceID128BitEnabled controls if trace IDs are generated as 128-bits or 64-bits. traceID128BitEnabled bool // apiKey is the Datadog API key from DD_API_KEY (used for agentless intake, LLM Obs, etc.). @@ -410,7 +412,14 @@ func loadConfig() *Config { p.GetMap("OTEL_EXPORTER_OTLP_HEADERS", nil, internal.OtelTagsDelimeter), p.GetMap("OTEL_EXPORTER_OTLP_METRICS_HEADERS", nil, internal.OtelTagsDelimeter), ) - cfg.otlpMetricsFlushInterval = resolveOTLPMetricsFlushInterval(env.Get("_DD_TRACE_METRICS_OTEL_FLUSH_INTERVAL")) + cfg.otlpMetricsFlushInterval = resolveOTLPMetricsFlushInterval(env.Get("_DD_TRACE_STATS_INTERVAL")) + otlpProtocolFallback := p.GetString("OTEL_EXPORTER_OTLP_PROTOCOL", "http/protobuf") + if !validateOTLPProtocol(otlpProtocolFallback, "OTEL_EXPORTER_OTLP_PROTOCOL") { + otlpProtocolFallback = "http/protobuf" + } + cfg.otlpMetricsProtocol = p.GetStringWithValidator("OTEL_EXPORTER_OTLP_METRICS_PROTOCOL", otlpProtocolFallback, func(v string) bool { + return validateOTLPProtocol(v, "OTEL_EXPORTER_OTLP_METRICS_PROTOCOL") + }) cfg.traceID128BitEnabled = p.GetBool("DD_TRACE_128_BIT_TRACEID_GENERATION_ENABLED", true) cfg.httpClientTimeout = time.Duration(p.GetIntWithValidator("DD_TRACE_AGENT_TIMEOUT", 10, validateAgentTimeout)) * time.Second cfg.propagationStyleInject = p.GetString("DD_TRACE_PROPAGATION_STYLE_INJECT", "") @@ -1639,6 +1648,13 @@ func (c *Config) OTLPMetricsFlushInterval() time.Duration { return c.otlpMetricsFlushInterval } +// OTLPMetricsProtocol returns the OTLP export protocol for metrics ("http/json" or "http/protobuf"). +func (c *Config) OTLPMetricsProtocol() string { + c.mu.RLock() + defer c.mu.RUnlock() + return c.otlpMetricsProtocol +} + func (c *Config) TraceID128BitEnabled() bool { c.mu.RLock() defer c.mu.RUnlock() diff --git a/internal/config/config_helpers.go b/internal/config/config_helpers.go index a72f69f962..a1a3143929 100644 --- a/internal/config/config_helpers.go +++ b/internal/config/config_helpers.go @@ -306,9 +306,6 @@ func formatDogstatsdAddr(u *url.URL) string { return u.Host } -// resolveOTLPTraceURL resolves the OTLP trace endpoint from OTEL_EXPORTER_OTLP_TRACES_ENDPOINT if set, else agentURL host + default OTLP port 4318 + /v1/traces. -// When the user-provided endpoint is set, it is validated: it must be a parseable URL with an http or https scheme. -// If validation fails, the default endpoint is used instead. // parseAndValidateOTLPURL parses rawURL and validates that it uses http or https. // Logs a warning and returns (nil, false) on failure. func parseAndValidateOTLPURL(envVar, rawURL string) (*url.URL, bool) { @@ -324,6 +321,9 @@ func parseAndValidateOTLPURL(envVar, rawURL string) (*url.URL, bool) { return u, true } +// resolveOTLPTraceURL resolves the OTLP trace endpoint from OTEL_EXPORTER_OTLP_TRACES_ENDPOINT if set, +// else derives a default from agentURL host + port 4318 + /v1/traces. +// When the user-provided endpoint is set it is validated; if invalid the default is used instead. func resolveOTLPTraceURL(rawAgentURL *url.URL, otlpTracesEndpoint string) string { if otlpTracesEndpoint != "" { if _, ok := parseAndValidateOTLPURL("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", otlpTracesEndpoint); ok { @@ -449,7 +449,17 @@ func buildOTLPMetricsHeaders(genericHeaders, signalHeaders map[string]string) ma return merged } -// resolveOTLPMetricsFlushInterval parses _DD_TRACE_METRICS_OTEL_FLUSH_INTERVAL (milliseconds). +// validateOTLPProtocol returns true for the two supported OTLP HTTP protocol values. +// envVar is used in the warning message to identify which env var had the bad value. +func validateOTLPProtocol(v, envVar string) bool { + if v == "http/json" || v == "http/protobuf" { + return true + } + log.Warn("Unsupported %s %q; must be http/json or http/protobuf. Falling back to default.", envVar, v) + return false +} + +// resolveOTLPMetricsFlushInterval parses _DD_TRACE_STATS_INTERVAL (milliseconds). // The variable is internal and intended for tests only; in production it returns the default 10 s. func resolveOTLPMetricsFlushInterval(raw string) time.Duration { if raw == "" { @@ -457,7 +467,7 @@ func resolveOTLPMetricsFlushInterval(raw string) time.Duration { } ms, err := strconv.ParseInt(raw, 10, 64) if err != nil || ms <= 0 { - log.Warn("Invalid _DD_TRACE_METRICS_OTEL_FLUSH_INTERVAL %q; using default %s.", raw, OTLPMetricsFlushInterval) + log.Warn("Invalid _DD_TRACE_STATS_INTERVAL %q; using default %s.", raw, OTLPMetricsFlushInterval) return OTLPMetricsFlushInterval } return time.Duration(ms) * time.Millisecond diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 48b5a5ba7f..9ac7b91a02 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -1026,11 +1026,11 @@ func TestOTLPMetricsFlushInterval(t *testing.T) { assert.Equal(t, OTLPMetricsFlushInterval, cfg.OTLPMetricsFlushInterval()) }) - t.Run("_DD_TRACE_METRICS_OTEL_FLUSH_INTERVAL overrides in milliseconds", func(t *testing.T) { + t.Run("_DD_TRACE_STATS_INTERVAL overrides in milliseconds", func(t *testing.T) { resetGlobalState() defer resetGlobalState() - t.Setenv("_DD_TRACE_METRICS_OTEL_FLUSH_INTERVAL", "1000") + t.Setenv("_DD_TRACE_STATS_INTERVAL", "1000") cfg := Get() require.NotNil(t, cfg) @@ -1042,7 +1042,7 @@ func TestOTLPMetricsFlushInterval(t *testing.T) { resetGlobalState() defer resetGlobalState() - t.Setenv("_DD_TRACE_METRICS_OTEL_FLUSH_INTERVAL", "not-a-number") + t.Setenv("_DD_TRACE_STATS_INTERVAL", "not-a-number") cfg := Get() require.NotNil(t, cfg) @@ -1051,6 +1051,73 @@ func TestOTLPMetricsFlushInterval(t *testing.T) { }) } +func TestOTLPMetricsProtocol(t *testing.T) { + t.Run("defaults to http/protobuf", func(t *testing.T) { + resetGlobalState() + defer resetGlobalState() + + cfg := Get() + require.NotNil(t, cfg) + assert.Equal(t, "http/protobuf", cfg.OTLPMetricsProtocol()) + }) + + t.Run("http/json via OTEL_EXPORTER_OTLP_METRICS_PROTOCOL", func(t *testing.T) { + resetGlobalState() + defer resetGlobalState() + + t.Setenv("OTEL_EXPORTER_OTLP_METRICS_PROTOCOL", "http/json") + + cfg := Get() + require.NotNil(t, cfg) + assert.Equal(t, "http/json", cfg.OTLPMetricsProtocol()) + }) + + t.Run("falls back to OTEL_EXPORTER_OTLP_PROTOCOL when signal-specific not set", func(t *testing.T) { + resetGlobalState() + defer resetGlobalState() + + t.Setenv("OTEL_EXPORTER_OTLP_PROTOCOL", "http/json") + + cfg := Get() + require.NotNil(t, cfg) + assert.Equal(t, "http/json", cfg.OTLPMetricsProtocol()) + }) + + t.Run("signal-specific takes precedence over generic", func(t *testing.T) { + resetGlobalState() + defer resetGlobalState() + + t.Setenv("OTEL_EXPORTER_OTLP_PROTOCOL", "http/json") + t.Setenv("OTEL_EXPORTER_OTLP_METRICS_PROTOCOL", "http/protobuf") + + cfg := Get() + require.NotNil(t, cfg) + assert.Equal(t, "http/protobuf", cfg.OTLPMetricsProtocol()) + }) + + t.Run("unknown value in signal-specific falls back to http/protobuf", func(t *testing.T) { + resetGlobalState() + defer resetGlobalState() + + t.Setenv("OTEL_EXPORTER_OTLP_METRICS_PROTOCOL", "grpc") + + cfg := Get() + require.NotNil(t, cfg) + assert.Equal(t, "http/protobuf", cfg.OTLPMetricsProtocol()) + }) + + t.Run("unknown value in generic falls back to http/protobuf", func(t *testing.T) { + resetGlobalState() + defer resetGlobalState() + + t.Setenv("OTEL_EXPORTER_OTLP_PROTOCOL", "grpc") + + cfg := Get() + require.NotNil(t, cfg) + assert.Equal(t, "http/protobuf", cfg.OTLPMetricsProtocol()) + }) +} + func TestHostnameConfiguration(t *testing.T) { t.Run("default behavior - hostname empty when not configured", func(t *testing.T) { resetGlobalState()