feat(tracer): add OTLP metrics HTTP exporter for span metrics #4899
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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) | ||
|
Contributor
There was a problem hiding this comment. We're implicitly defaulting to protobuf unless "http/json" was specified. Does internalconfig already inform the user that anything other than json or protobuf will be dropped/default to protobuf?
Contributor
Author
There was a problem hiding this comment. Yes, internalconfig already handles this. Also worth noting: the default was changed to "http/json" in this PR to align with the spec's transport section ("HTTP/JSON as the fallback protocol"), so the implicit behavior now favors JSON rather than protobuf.
Contributor
There was a problem hiding this comment. Right. it feels like this block is duplicating some of that. |
||
| 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}) | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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()) | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Does newOTLPMetricsExporter need all of
cfg? Can't it accept the parameters it needs explicitly/only?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
cfg is used across both construction and the export() call path. there are 7 fields in total: AgentTimeout, OTLPMetricsURL, OTLPMetricsHeaders, OTLPMetricsProtocol at construction, plus OTelSemanticsEnabled, ReportHostname, Hostname inside buildOTLPMetricsRequest. Passing those explicitly would produce a 7-parameter constructor, which is harder to read and maintain than the single cfg argument. Storing cfg on the struct also avoids threading those values through separately on every export() call.
This matches how
newOTLPTraceWriteris structured. it also takes the full config rather than individual fields.