From 53a86e00f6cbca509be59376b19e4a981937cfa7 Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Tue, 4 Aug 2026 15:42:48 -0400 Subject: [PATCH 01/15] fix(data-pipeline): correct OTLP trace-metrics attribute shapes Fixes several OTLP Span Metrics Connector attribute conventions on the `traces.span.sdk.metrics.duration` histogram export: - process_tags/peer_tags: emit as a single `datadog.{process,peer}_tags` arrayValue instead of fanning out per-key attributes. - is_trace_root: rename `_datadog.is_trace_root` to `datadog.is_trace_root` and always emit true/false rather than only when true. - span.kind: canonicalize to the OTel SMC uppercase convention (e.g. `SPAN_KIND_SERVER`) instead of the raw Datadog tag value. - status.code: emit the SMC string convention (`STATUS_CODE_OK`/ `STATUS_CODE_ERROR`) unconditionally on every data point, replacing the int enum that was only set on error. Companion system-tests spec: https://github.com/DataDog/system-tests/pull/7363 --- libdd-data-pipeline/src/otlp/metrics.rs | 133 +++++++++++++------ libdd-trace-utils/src/otlp_encoder/mapper.rs | 12 ++ 2 files changed, 106 insertions(+), 39 deletions(-) diff --git a/libdd-data-pipeline/src/otlp/metrics.rs b/libdd-data-pipeline/src/otlp/metrics.rs index 3d95c29306..e175234bc6 100644 --- a/libdd-data-pipeline/src/otlp/metrics.rs +++ b/libdd-data-pipeline/src/otlp/metrics.rs @@ -13,7 +13,7 @@ use libdd_ddsketch::DDSketch; use libdd_shared_runtime::Worker; use libdd_trace_protobuf::pb; use libdd_trace_stats::span_concentrator::{OtlpStatsBucket, SpanConcentrator}; -use libdd_trace_utils::otlp_encoder::mapper::status_code; +use libdd_trace_utils::otlp_encoder::mapper::tag_to_otlp_kind_str_name; use libdd_trace_utils::otlp_encoder::OtlpResourceInfo; use serde_json::{json, Value}; use std::sync::{Arc, Mutex}; @@ -49,6 +49,9 @@ const GRPC_STATUS_NAMES: [&str; 17] = [ fn grpc_status_code_to_name(code: &str) -> Option<&'static str> { GRPC_STATUS_NAMES.get(code.parse::().ok()?).copied() } + +const STATUS_CODE_OK: &str = "STATUS_CODE_OK"; +const STATUS_CODE_ERROR: &str = "STATUS_CODE_ERROR"; /// Fixed bucket boundaries (seconds) mirroring the OTel spanmetrics-connector defaults. const EXPLICIT_BOUNDS_SECONDS: [f64; 16] = [ 0.002, 0.004, 0.006, 0.008, 0.01, 0.05, 0.1, 0.2, 0.4, 0.8, 1.0, 1.4, 2.0, 5.0, 10.0, 15.0, @@ -60,6 +63,13 @@ fn kv_str(key: &str, value: &str) -> Value { fn kv_int(key: &str, value: i64) -> Value { json!({ "key": key, "value": { "intValue": value.to_string() } }) } +fn kv_str_array<'a>(key: &str, values: impl IntoIterator) -> Value { + let values: Vec = values + .into_iter() + .map(|v| json!({ "stringValue": v })) + .collect(); + json!({ "key": key, "value": { "arrayValue": { "values": values } } }) +} /// Build an OTLP metrics export request (`ExportMetricsServiceRequest`) as a JSON value. /// @@ -152,12 +162,13 @@ fn build_attributes( resource_info: &OtlpResourceInfo, otel_trace_semantics_enabled: bool, ) -> Vec { - let mut attrs = Vec::new(); - let mut push = |k: &str, v: &str| { + fn push(attrs: &mut Vec, k: &str, v: &str) { if !v.is_empty() { attrs.push(kv_str(k, v)); } - }; + } + + let mut attrs = Vec::new(); // Service identity is on the resource; emit on the data point only when overridden. let group_service = if group.service.is_empty() { @@ -166,30 +177,32 @@ fn build_attributes( group.service.as_str() }; if group_service != resource_info.service { - push("service.name", group_service); + push(&mut attrs, "service.name", group_service); } - push("span.name", &group.resource); - push("span.kind", &group.span_kind); - push("http.request.method", &group.http_method); - push("http.route", &group.http_endpoint); + push(&mut attrs, "span.name", &group.resource); + if !group.span_kind.is_empty() { + push( + &mut attrs, + "span.kind", + tag_to_otlp_kind_str_name(&group.span_kind), + ); + } + push(&mut attrs, "http.request.method", &group.http_method); + push(&mut attrs, "http.route", &group.http_endpoint); // group.grpc_status_code is the numeric code as a string; emit the canonical OTel status name. if let Some(name) = grpc_status_code_to_name(&group.grpc_status_code) { - push("rpc.response.status_code", name); - } - for tag in &group.peer_tags { - if let Some((k, v)) = tag.split_once(':') { - push(k, v); - } + push(&mut attrs, "rpc.response.status_code", name); } + // additional_metric_tags support is still evolving/TBD across most SDKs. for tag in &group.additional_metric_tags { if let Some((k, v)) = tag.split_once(':') { - push(k, v); + push(&mut attrs, k, v); } } if !otel_trace_semantics_enabled { - push("datadog.operation.name", &group.name); - push("datadog.span.type", &group.r#type); + push(&mut attrs, "datadog.operation.name", &group.name); + push(&mut attrs, "datadog.span.type", &group.r#type); } if group.http_status_code != 0 { attrs.push(kv_int( @@ -203,17 +216,32 @@ fn build_attributes( if group.synthetics { attrs.push(kv_str("datadog.origin", "synthetics")); } - if group.is_trace_root == pb::Trilean::True as i32 { - attrs.push(json!({ "key": "_datadog.is_trace_root", "value": { "boolValue": true } })); - } + let is_trace_root = group.is_trace_root == pb::Trilean::True as i32; + attrs.push(json!({ + "key": "datadog.is_trace_root", "value": { "boolValue": is_trace_root } + })); let top_level = group.hits > 0 && group.top_level_hits == group.hits; attrs.push(json!({ "key": "datadog.span.top_level", "value": { "boolValue": top_level } })); + // Unlike process_tags (a resource attribute), peer_tags is per-span and belongs on the data + // point. + if !group.peer_tags.is_empty() { + attrs.push(kv_str_array( + "datadog.peer_tags", + group.peer_tags.iter().map(String::as_str), + )); + } } - if is_error { - attrs.push(kv_int("status.code", status_code::ERROR as i64)); - } + push( + &mut attrs, + "status.code", + if is_error { + STATUS_CODE_ERROR + } else { + STATUS_CODE_OK + }, + ); attrs } @@ -239,11 +267,17 @@ fn build_resource_attributes( if !info.runtime_id.is_empty() { attrs.push(kv_str("datadog.runtime_id", &info.runtime_id)); } - attrs.extend(info.process_tags.split(',').filter_map(|p| { - let (k, v) = p.split_once(':')?; - let (k, v) = (k.trim(), v.trim()); - (!k.is_empty() && !v.is_empty()).then(|| kv_str(&format!("datadog.{k}"), v)) - })); + // Mirrors the v0.6/legacy stats export's process_tags string; keep both in sync if that + // format changes. + let process_tags: Vec<&str> = info + .process_tags + .split(',') + .map(str::trim) + .filter(|p| !p.is_empty()) + .collect(); + if !process_tags.is_empty() { + attrs.push(kv_str_array("datadog.process_tags", process_tags)); + } } attrs } @@ -430,17 +464,19 @@ mod tests { .and_then(|kv| kv["value"]["stringValue"].as_str()) } - fn err_code() -> String { - (status_code::ERROR as i64).to_string() + fn str_array_at<'a>(attrs: &'a [Value], key: &str) -> Option> { + attrs.iter().find(|kv| kv["key"] == key).map(|kv| { + kv["value"]["arrayValue"]["values"] + .as_array() + .unwrap() + .iter() + .map(|v| v["stringValue"].as_str().unwrap()) + .collect() + }) } fn is_error_point(p: &Value) -> bool { - let ec = err_code(); - p["attributes"] - .as_array() - .unwrap() - .iter() - .any(|kv| kv["key"] == "status.code" && kv["value"]["intValue"].as_str() == Some(&ec)) + str_at(p["attributes"].as_array().unwrap(), "status.code") == Some(STATUS_CODE_ERROR) } #[test] @@ -466,7 +502,10 @@ mod tests { assert_eq!(str_at(a, "telemetry.sdk.name"), Some("datadog")); let dd = !otel; assert_eq!(str_at(a, "datadog.runtime_id").is_some(), dd); - assert_eq!(str_at(a, "datadog.entrypoint.name").is_some(), dd); + assert_eq!( + str_array_at(a, "datadog.process_tags"), + dd.then(|| vec!["entrypoint.name:server"]) + ); } } @@ -477,6 +516,9 @@ mod tests { g.http_method = "POST".into(); g.http_endpoint = "/users/:id".into(); g.synthetics = true; + g.span_kind = "server".into(); + g.is_trace_root = pb::Trilean::True as i32; + g.peer_tags = vec!["db.hostname:prod-db-1".into(), "db.name:orders".into()]; }); let custom_pair = group_with_exact(&[1_000_000_000], &[], |g| { g.service = "svc-other".into(); @@ -492,18 +534,30 @@ mod tests { .as_array() .unwrap(); assert_eq!(str_at(a, "span.name"), Some("GET /foo")); + assert_eq!(str_at(a, "span.kind"), Some("SPAN_KIND_SERVER")); assert_eq!(str_at(a, "http.request.method"), Some("POST")); assert_eq!(str_at(a, "http.route"), Some("/users/:id")); assert!(a.iter().any(|kv| kv["key"] == "http.response.status_code")); assert_eq!(str_at(a, "datadog.operation.name"), Some("test.op")); assert_eq!(str_at(a, "datadog.span.type"), Some("web")); assert_eq!(str_at(a, "datadog.origin"), Some("synthetics")); + assert_eq!( + a.iter() + .find(|kv| kv["key"] == "datadog.is_trace_root") + .and_then(|kv| kv["value"]["boolValue"].as_bool()), + Some(true) + ); assert!(a.iter().any(|kv| kv["key"] == "datadog.span.top_level")); + assert_eq!( + str_array_at(a, "datadog.peer_tags"), + Some(vec!["db.hostname:prod-db-1", "db.name:orders"]) + ); + assert_eq!(str_at(a, "status.code"), Some(STATUS_CODE_OK)); assert!(pts.iter().any( |p| str_at(p["attributes"].as_array().unwrap(), "service.name") == Some("svc-other") )); - // OTel mode strips datadog.*/_datadog.* attributes. + // OTel mode strips datadog.*/_datadog.* attributes but keeps status.code always present. let req = map_stats_to_otlp_metrics(&buckets(vec![g_pair]), &resource(), true).unwrap(); let a = points(&req)[0]["attributes"].as_array().unwrap(); assert!(!a.iter().any(|kv| { @@ -511,6 +565,7 @@ mod tests { k.starts_with("datadog.") || k.starts_with("_datadog.") })); assert_eq!(str_at(a, "http.request.method"), Some("POST")); + assert_eq!(str_at(a, "status.code"), Some(STATUS_CODE_OK)); } #[test] diff --git a/libdd-trace-utils/src/otlp_encoder/mapper.rs b/libdd-trace-utils/src/otlp_encoder/mapper.rs index bc3a94c65c..bf21f3b143 100644 --- a/libdd-trace-utils/src/otlp_encoder/mapper.rs +++ b/libdd-trace-utils/src/otlp_encoder/mapper.rs @@ -110,6 +110,18 @@ fn tag_to_otlp_kind(t: &str) -> i32 { } } +/// Maps a Datadog `span.kind` tag value to the OTel Span Metrics Connector's string convention. +pub fn tag_to_otlp_kind_str_name(t: &str) -> &'static str { + match tag_to_otlp_kind(t) { + span_kind::SERVER => "SPAN_KIND_SERVER", + span_kind::CLIENT => "SPAN_KIND_CLIENT", + span_kind::PRODUCER => "SPAN_KIND_PRODUCER", + span_kind::CONSUMER => "SPAN_KIND_CONSUMER", + span_kind::INTERNAL => "SPAN_KIND_INTERNAL", + _ => "SPAN_KIND_UNSPECIFIED", + } +} + /// Maps the Datadog span type field (set by DD-instrumented tracers) to an OTLP SpanKind. fn dd_type_to_otlp_kind(t: &str) -> i32 { // Case-insensitive match without allocating (see `tag_to_otlp_kind`). From 8ee6f7e6f474601e9428c7946187f71ffd530f81 Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Wed, 5 Aug 2026 19:25:37 -0400 Subject: [PATCH 02/15] fix(data-pipeline): finalize OTLP trace metrics semantics --- libdd-data-pipeline/src/otlp/metrics.rs | 168 ++++++++++++++++-------- 1 file changed, 110 insertions(+), 58 deletions(-) diff --git a/libdd-data-pipeline/src/otlp/metrics.rs b/libdd-data-pipeline/src/otlp/metrics.rs index e175234bc6..be6f8d5bde 100644 --- a/libdd-data-pipeline/src/otlp/metrics.rs +++ b/libdd-data-pipeline/src/otlp/metrics.rs @@ -170,24 +170,37 @@ fn build_attributes( let mut attrs = Vec::new(); - // Service identity is on the resource; emit on the data point only when overridden. let group_service = if group.service.is_empty() { resource_info.service.as_str() } else { group.service.as_str() }; - if group_service != resource_info.service { - push(&mut attrs, "service.name", group_service); - } + attrs.push(kv_str("service.name", group_service)); push(&mut attrs, "span.name", &group.resource); - if !group.span_kind.is_empty() { - push( - &mut attrs, - "span.kind", - tag_to_otlp_kind_str_name(&group.span_kind), - ); + push( + &mut attrs, + "span.kind", + if group.span_kind.is_empty() { + "SPAN_KIND_INTERNAL" + } else { + tag_to_otlp_kind_str_name(&group.span_kind) + }, + ); + push( + &mut attrs, + "status.code", + if is_error { + STATUS_CODE_ERROR + } else { + STATUS_CODE_OK + }, + ); + + if otel_trace_semantics_enabled { + return attrs; } + push(&mut attrs, "http.request.method", &group.http_method); push(&mut attrs, "http.route", &group.http_endpoint); // group.grpc_status_code is the numeric code as a string; emit the canonical OTel status name. @@ -200,48 +213,41 @@ fn build_attributes( push(&mut attrs, k, v); } } - if !otel_trace_semantics_enabled { - push(&mut attrs, "datadog.operation.name", &group.name); - push(&mut attrs, "datadog.span.type", &group.r#type); - } + push(&mut attrs, "datadog.operation.name", &group.name); + push(&mut attrs, "datadog.span.type", &group.r#type); if group.http_status_code != 0 { attrs.push(kv_int( "http.response.status_code", group.http_status_code as i64, )); } - if !otel_trace_semantics_enabled { - // Only `synthetics` is surfaced as `datadog.origin`: the aggregation key carries just a - // boolean, not the full origin string, so other origins are lost upstream. - if group.synthetics { - attrs.push(kv_str("datadog.origin", "synthetics")); - } - let is_trace_root = group.is_trace_root == pb::Trilean::True as i32; + // Only `synthetics` is surfaced as `datadog.origin`: the aggregation key carries just a + // boolean, not the full origin string, so other origins are lost upstream. + if group.synthetics { + attrs.push(kv_str("datadog.origin", "synthetics")); + } + let is_trace_root = match pb::Trilean::try_from(group.is_trace_root) { + Ok(pb::Trilean::True) => Some(true), + Ok(pb::Trilean::False) => Some(false), + _ => None, + }; + if let Some(is_trace_root) = is_trace_root { attrs.push(json!({ "key": "datadog.is_trace_root", "value": { "boolValue": is_trace_root } })); - let top_level = group.hits > 0 && group.top_level_hits == group.hits; - attrs.push(json!({ - "key": "datadog.span.top_level", "value": { "boolValue": top_level } - })); - // Unlike process_tags (a resource attribute), peer_tags is per-span and belongs on the data - // point. - if !group.peer_tags.is_empty() { - attrs.push(kv_str_array( - "datadog.peer_tags", - group.peer_tags.iter().map(String::as_str), - )); - } } - push( - &mut attrs, - "status.code", - if is_error { - STATUS_CODE_ERROR - } else { - STATUS_CODE_OK - }, - ); + let top_level = group.hits > 0 && group.top_level_hits == group.hits; + attrs.push(json!({ + "key": "datadog.span.top_level", "value": { "boolValue": top_level } + })); + // Unlike process_tags (a resource attribute), peer_tags is per-span and belongs on the data + // point. + if !group.peer_tags.is_empty() { + attrs.push(kv_str_array( + "datadog.peer_tags", + group.peer_tags.iter().map(String::as_str), + )); + } attrs } @@ -475,6 +481,13 @@ mod tests { }) } + fn bool_at(attrs: &[Value], key: &str) -> Option { + attrs + .iter() + .find(|kv| kv["key"] == key) + .and_then(|kv| kv["value"]["boolValue"].as_bool()) + } + fn is_error_point(p: &Value) -> bool { str_at(p["attributes"].as_array().unwrap(), "status.code") == Some(STATUS_CODE_ERROR) } @@ -517,8 +530,10 @@ mod tests { g.http_endpoint = "/users/:id".into(); g.synthetics = true; g.span_kind = "server".into(); + g.grpc_status_code = "5".into(); g.is_trace_root = pb::Trilean::True as i32; g.peer_tags = vec!["db.hostname:prod-db-1".into(), "db.name:orders".into()]; + g.additional_metric_tags = vec!["custom.primary:a".into()]; }); let custom_pair = group_with_exact(&[1_000_000_000], &[], |g| { g.service = "svc-other".into(); @@ -529,10 +544,12 @@ mod tests { let pts = points(&req); let a = pts .iter() - .find(|p| str_at(p["attributes"].as_array().unwrap(), "service.name").is_none()) + .find(|p| str_at(p["attributes"].as_array().unwrap(), "service.name") == Some("svc")) .unwrap()["attributes"] .as_array() .unwrap(); + // service.name is on the data point even though it matches the resource default. + assert_eq!(str_at(a, "service.name"), Some("svc")); assert_eq!(str_at(a, "span.name"), Some("GET /foo")); assert_eq!(str_at(a, "span.kind"), Some("SPAN_KIND_SERVER")); assert_eq!(str_at(a, "http.request.method"), Some("POST")); @@ -541,12 +558,7 @@ mod tests { assert_eq!(str_at(a, "datadog.operation.name"), Some("test.op")); assert_eq!(str_at(a, "datadog.span.type"), Some("web")); assert_eq!(str_at(a, "datadog.origin"), Some("synthetics")); - assert_eq!( - a.iter() - .find(|kv| kv["key"] == "datadog.is_trace_root") - .and_then(|kv| kv["value"]["boolValue"].as_bool()), - Some(true) - ); + assert_eq!(bool_at(a, "datadog.is_trace_root"), Some(true)); assert!(a.iter().any(|kv| kv["key"] == "datadog.span.top_level")); assert_eq!( str_array_at(a, "datadog.peer_tags"), @@ -557,15 +569,56 @@ mod tests { |p| str_at(p["attributes"].as_array().unwrap(), "service.name") == Some("svc-other") )); - // OTel mode strips datadog.*/_datadog.* attributes but keeps status.code always present. + // OTel mode emits exactly the Span Metrics Connector's default dimensions. let req = map_stats_to_otlp_metrics(&buckets(vec![g_pair]), &resource(), true).unwrap(); let a = points(&req)[0]["attributes"].as_array().unwrap(); - assert!(!a.iter().any(|kv| { - let k = kv["key"].as_str().unwrap_or(""); - k.starts_with("datadog.") || k.starts_with("_datadog.") - })); - assert_eq!(str_at(a, "http.request.method"), Some("POST")); + let mut keys: Vec<&str> = a.iter().map(|kv| kv["key"].as_str().unwrap()).collect(); + keys.sort_unstable(); + assert_eq!( + keys, + ["service.name", "span.kind", "span.name", "status.code"] + ); + for prefix in ["http.", "rpc.", "datadog.", "_datadog."] { + assert!(!keys.iter().any(|key| key.starts_with(prefix))); + } + assert!(!keys.contains(&"additional_metric_tags")); + assert!(!keys.contains(&"custom.primary")); assert_eq!(str_at(a, "status.code"), Some(STATUS_CODE_OK)); + assert_eq!(str_at(a, "service.name"), Some("svc")); + } + + #[test] + fn is_trace_root_preserves_trilean_semantics() { + for (value, expected) in [ + (pb::Trilean::NotSet, None), + (pb::Trilean::True, Some(true)), + (pb::Trilean::False, Some(false)), + ] { + let g = group_with_exact(&[1_000_000_000], &[], |g| { + g.is_trace_root = value as i32; + }); + let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), false).unwrap(); + assert_eq!( + bool_at( + points(&req)[0]["attributes"].as_array().unwrap(), + "datadog.is_trace_root" + ), + expected + ); + } + } + + #[test] + fn empty_span_kind_defaults_to_internal() { + let req = + map_stats_to_otlp_metrics(&buckets(vec![one_ok_group()]), &resource(), true).unwrap(); + assert_eq!( + str_at( + points(&req)[0]["attributes"].as_array().unwrap(), + "span.kind" + ), + Some("SPAN_KIND_INTERNAL") + ); } #[test] @@ -663,11 +716,10 @@ mod tests { assert_eq!(str_at(a, "region"), Some("us-east")); assert_eq!(str_at(a, "endpoint"), Some("https://host:8080")); - // Additional metric tags are user/tracer-defined (not Datadog-internal), so unlike - // `datadog.*` attributes they still pass through in OTel-semantics mode. + // OTel-semantics mode is limited to the Span Metrics Connector defaults. let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), true).unwrap(); let a = points(&req)[0]["attributes"].as_array().unwrap(); - assert_eq!(str_at(a, "custom.primary"), Some("a")); + assert_eq!(str_at(a, "custom.primary"), None); // Malformed (no `:`) or empty-value entries are skipped rather than emitted verbatim. let g = group_with_exact(&[1_000_000_000], &[], |g| { From 130a4decd4f8e15f46dce6a634dbcf56c546f2f9 Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Wed, 5 Aug 2026 21:52:58 -0400 Subject: [PATCH 03/15] fix(data-pipeline): preserve OTel trace metric attributes --- libdd-data-pipeline/src/otlp/metrics.rs | 68 +++++++++++++++++-------- 1 file changed, 46 insertions(+), 22 deletions(-) diff --git a/libdd-data-pipeline/src/otlp/metrics.rs b/libdd-data-pipeline/src/otlp/metrics.rs index be6f8d5bde..279318eab2 100644 --- a/libdd-data-pipeline/src/otlp/metrics.rs +++ b/libdd-data-pipeline/src/otlp/metrics.rs @@ -197,30 +197,36 @@ fn build_attributes( }, ); - if otel_trace_semantics_enabled { - return attrs; - } - push(&mut attrs, "http.request.method", &group.http_method); push(&mut attrs, "http.route", &group.http_endpoint); // group.grpc_status_code is the numeric code as a string; emit the canonical OTel status name. if let Some(name) = grpc_status_code_to_name(&group.grpc_status_code) { push(&mut attrs, "rpc.response.status_code", name); } + if group.http_status_code != 0 { + attrs.push(kv_int( + "http.response.status_code", + group.http_status_code as i64, + )); + } // additional_metric_tags support is still evolving/TBD across most SDKs. for tag in &group.additional_metric_tags { if let Some((k, v)) = tag.split_once(':') { + if otel_trace_semantics_enabled + && (k.starts_with("datadog.") || k.starts_with("_datadog.")) + { + continue; + } push(&mut attrs, k, v); } } + + if otel_trace_semantics_enabled { + return attrs; + } + push(&mut attrs, "datadog.operation.name", &group.name); push(&mut attrs, "datadog.span.type", &group.r#type); - if group.http_status_code != 0 { - attrs.push(kv_int( - "http.response.status_code", - group.http_status_code as i64, - )); - } // Only `synthetics` is surfaced as `datadog.origin`: the aggregation key carries just a // boolean, not the full origin string, so other origins are lost upstream. if group.synthetics { @@ -559,7 +565,7 @@ mod tests { assert_eq!(str_at(a, "datadog.span.type"), Some("web")); assert_eq!(str_at(a, "datadog.origin"), Some("synthetics")); assert_eq!(bool_at(a, "datadog.is_trace_root"), Some(true)); - assert!(a.iter().any(|kv| kv["key"] == "datadog.span.top_level")); + assert_eq!(bool_at(a, "datadog.span.top_level"), Some(true)); assert_eq!( str_array_at(a, "datadog.peer_tags"), Some(vec!["db.hostname:prod-db-1", "db.name:orders"]) @@ -569,20 +575,29 @@ mod tests { |p| str_at(p["attributes"].as_array().unwrap(), "service.name") == Some("svc-other") )); - // OTel mode emits exactly the Span Metrics Connector's default dimensions. + // OTel mode retains semantic and custom attributes while suppressing Datadog attributes. let req = map_stats_to_otlp_metrics(&buckets(vec![g_pair]), &resource(), true).unwrap(); let a = points(&req)[0]["attributes"].as_array().unwrap(); let mut keys: Vec<&str> = a.iter().map(|kv| kv["key"].as_str().unwrap()).collect(); keys.sort_unstable(); assert_eq!( keys, - ["service.name", "span.kind", "span.name", "status.code"] + [ + "custom.primary", + "http.request.method", + "http.response.status_code", + "http.route", + "rpc.response.status_code", + "service.name", + "span.kind", + "span.name", + "status.code", + ] ); - for prefix in ["http.", "rpc.", "datadog.", "_datadog."] { + for prefix in ["datadog.", "_datadog."] { assert!(!keys.iter().any(|key| key.starts_with(prefix))); } assert!(!keys.contains(&"additional_metric_tags")); - assert!(!keys.contains(&"custom.primary")); assert_eq!(str_at(a, "status.code"), Some(STATUS_CODE_OK)); assert_eq!(str_at(a, "service.name"), Some("svc")); } @@ -596,15 +611,16 @@ mod tests { ] { let g = group_with_exact(&[1_000_000_000], &[], |g| { g.is_trace_root = value as i32; + g.top_level_hits = 0; }); let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), false).unwrap(); + let attrs = points(&req)[0]["attributes"].as_array().unwrap(); + let is_trace_root = attrs.iter().find(|kv| kv["key"] == "datadog.is_trace_root"); assert_eq!( - bool_at( - points(&req)[0]["attributes"].as_array().unwrap(), - "datadog.is_trace_root" - ), - expected + is_trace_root.map(|kv| kv["value"].clone()), + expected.map(|value| json!({ "boolValue": value })) ); + assert_eq!(bool_at(attrs, "datadog.span.top_level"), Some(false)); } } @@ -708,6 +724,8 @@ mod tests { "region:us-east".into(), // Only the first `:` is a delimiter; the value keeps any embedded `:`. "endpoint:https://host:8080".into(), + "datadog.custom:dd".into(), + "_datadog.custom:internal".into(), ]; }); let req = map_stats_to_otlp_metrics(&buckets(vec![g.clone()]), &resource(), false).unwrap(); @@ -715,11 +733,17 @@ mod tests { assert_eq!(str_at(a, "custom.primary"), Some("a")); assert_eq!(str_at(a, "region"), Some("us-east")); assert_eq!(str_at(a, "endpoint"), Some("https://host:8080")); + assert_eq!(str_at(a, "datadog.custom"), Some("dd")); + assert_eq!(str_at(a, "_datadog.custom"), Some("internal")); - // OTel-semantics mode is limited to the Span Metrics Connector defaults. + // OTel-semantics mode keeps custom tags and filters only Datadog-prefixed tags. let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), true).unwrap(); let a = points(&req)[0]["attributes"].as_array().unwrap(); - assert_eq!(str_at(a, "custom.primary"), None); + assert_eq!(str_at(a, "custom.primary"), Some("a")); + assert_eq!(str_at(a, "region"), Some("us-east")); + assert_eq!(str_at(a, "endpoint"), Some("https://host:8080")); + assert_eq!(str_at(a, "datadog.custom"), None); + assert_eq!(str_at(a, "_datadog.custom"), None); // Malformed (no `:`) or empty-value entries are skipped rather than emitted verbatim. let g = group_with_exact(&[1_000_000_000], &[], |g| { From 2e2e3cbf0ceea1685f58bd14afa29b44658b417f Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Thu, 6 Aug 2026 09:22:45 -0400 Subject: [PATCH 04/15] fix(data-pipeline): reserve only datadog OTLP attributes --- libdd-data-pipeline/src/otlp/metrics.rs | 11 ++--------- 1 file changed, 2 insertions(+), 9 deletions(-) diff --git a/libdd-data-pipeline/src/otlp/metrics.rs b/libdd-data-pipeline/src/otlp/metrics.rs index 279318eab2..b4c161717b 100644 --- a/libdd-data-pipeline/src/otlp/metrics.rs +++ b/libdd-data-pipeline/src/otlp/metrics.rs @@ -212,9 +212,7 @@ fn build_attributes( // additional_metric_tags support is still evolving/TBD across most SDKs. for tag in &group.additional_metric_tags { if let Some((k, v)) = tag.split_once(':') { - if otel_trace_semantics_enabled - && (k.starts_with("datadog.") || k.starts_with("_datadog.")) - { + if otel_trace_semantics_enabled && k.starts_with("datadog.") { continue; } push(&mut attrs, k, v); @@ -594,9 +592,7 @@ mod tests { "status.code", ] ); - for prefix in ["datadog.", "_datadog."] { - assert!(!keys.iter().any(|key| key.starts_with(prefix))); - } + assert!(!keys.iter().any(|key| key.starts_with("datadog."))); assert!(!keys.contains(&"additional_metric_tags")); assert_eq!(str_at(a, "status.code"), Some(STATUS_CODE_OK)); assert_eq!(str_at(a, "service.name"), Some("svc")); @@ -725,7 +721,6 @@ mod tests { // Only the first `:` is a delimiter; the value keeps any embedded `:`. "endpoint:https://host:8080".into(), "datadog.custom:dd".into(), - "_datadog.custom:internal".into(), ]; }); let req = map_stats_to_otlp_metrics(&buckets(vec![g.clone()]), &resource(), false).unwrap(); @@ -734,7 +729,6 @@ mod tests { assert_eq!(str_at(a, "region"), Some("us-east")); assert_eq!(str_at(a, "endpoint"), Some("https://host:8080")); assert_eq!(str_at(a, "datadog.custom"), Some("dd")); - assert_eq!(str_at(a, "_datadog.custom"), Some("internal")); // OTel-semantics mode keeps custom tags and filters only Datadog-prefixed tags. let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), true).unwrap(); @@ -743,7 +737,6 @@ mod tests { assert_eq!(str_at(a, "region"), Some("us-east")); assert_eq!(str_at(a, "endpoint"), Some("https://host:8080")); assert_eq!(str_at(a, "datadog.custom"), None); - assert_eq!(str_at(a, "_datadog.custom"), None); // Malformed (no `:`) or empty-value entries are skipped rather than emitted verbatim. let g = group_with_exact(&[1_000_000_000], &[], |g| { From d7108bf733f403eb5d6160073fe2333696ec4052 Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Thu, 6 Aug 2026 09:39:58 -0400 Subject: [PATCH 05/15] test(data-pipeline): cover unknown trace-root values --- libdd-data-pipeline/src/otlp/metrics.rs | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/libdd-data-pipeline/src/otlp/metrics.rs b/libdd-data-pipeline/src/otlp/metrics.rs index b4c161717b..910c9ff646 100644 --- a/libdd-data-pipeline/src/otlp/metrics.rs +++ b/libdd-data-pipeline/src/otlp/metrics.rs @@ -601,12 +601,13 @@ mod tests { #[test] fn is_trace_root_preserves_trilean_semantics() { for (value, expected) in [ - (pb::Trilean::NotSet, None), - (pb::Trilean::True, Some(true)), - (pb::Trilean::False, Some(false)), + (pb::Trilean::NotSet as i32, None), + (pb::Trilean::True as i32, Some(true)), + (pb::Trilean::False as i32, Some(false)), + (i32::MAX, None), ] { let g = group_with_exact(&[1_000_000_000], &[], |g| { - g.is_trace_root = value as i32; + g.is_trace_root = value; g.top_level_hits = 0; }); let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), false).unwrap(); From 5de747c840172e2d341ad8082bef353c95759514 Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Thu, 6 Aug 2026 14:34:42 -0400 Subject: [PATCH 06/15] fix(data-pipeline)!: remove trace-metrics semantics mode Always emit available Datadog attributes for OTLP trace metrics and map non-empty service_source values to datadog.svc_src. BREAKING CHANGE: OtlpMetricsConfig and map_stats_to_otlp_metrics no longer accept the OTel trace-semantics setting. --- libdd-data-pipeline/src/otlp/config.rs | 3 - libdd-data-pipeline/src/otlp/metrics.rs | 176 ++++++++---------- .../src/trace_exporter/builder.rs | 1 - 3 files changed, 74 insertions(+), 106 deletions(-) diff --git a/libdd-data-pipeline/src/otlp/config.rs b/libdd-data-pipeline/src/otlp/config.rs index 349be5791e..4f509d8358 100644 --- a/libdd-data-pipeline/src/otlp/config.rs +++ b/libdd-data-pipeline/src/otlp/config.rs @@ -134,7 +134,4 @@ pub struct OtlpMetricsConfig { /// Protocol (for future use; currently only HttpJson is supported). #[allow(dead_code)] pub(crate) protocol: OtlpProtocol, - /// When `true`, emit only OTel attributes; omit `dd.*`/`_dd.*` ones - /// (`DD_TRACE_OTEL_SEMANTICS_ENABLED`). - pub otel_trace_semantics_enabled: bool, } diff --git a/libdd-data-pipeline/src/otlp/metrics.rs b/libdd-data-pipeline/src/otlp/metrics.rs index 910c9ff646..5cf20ab9fe 100644 --- a/libdd-data-pipeline/src/otlp/metrics.rs +++ b/libdd-data-pipeline/src/otlp/metrics.rs @@ -79,7 +79,6 @@ fn kv_str_array<'a>(key: &str, values: impl IntoIterator) -> Val pub fn map_stats_to_otlp_metrics( buckets: &[OtlpStatsBucket], resource_info: &OtlpResourceInfo, - otel_trace_semantics_enabled: bool, ) -> Option { let mut data_points = Vec::new(); for b in buckets { @@ -96,7 +95,7 @@ pub fn map_stats_to_otlp_metrics( continue; }; data_points.push(json!({ - "attributes": build_attributes(group, is_error, resource_info, otel_trace_semantics_enabled), + "attributes": build_attributes(group, is_error, resource_info), "startTimeUnixNano": b.bucket.start.to_string(), "timeUnixNano": end.to_string(), "count": cell.count.to_string(), @@ -114,7 +113,7 @@ pub fn map_stats_to_otlp_metrics( } Some(json!({ "resourceMetrics": [{ - "resource": { "attributes": build_resource_attributes(resource_info, otel_trace_semantics_enabled) }, + "resource": { "attributes": build_resource_attributes(resource_info) }, "scopeMetrics": [{ "metrics": [{ "name": METRIC_NAME, @@ -160,7 +159,6 @@ fn build_attributes( group: &pb::ClientGroupedStats, is_error: bool, resource_info: &OtlpResourceInfo, - otel_trace_semantics_enabled: bool, ) -> Vec { fn push(attrs: &mut Vec, k: &str, v: &str) { if !v.is_empty() { @@ -212,19 +210,13 @@ fn build_attributes( // additional_metric_tags support is still evolving/TBD across most SDKs. for tag in &group.additional_metric_tags { if let Some((k, v)) = tag.split_once(':') { - if otel_trace_semantics_enabled && k.starts_with("datadog.") { - continue; - } push(&mut attrs, k, v); } } - if otel_trace_semantics_enabled { - return attrs; - } - push(&mut attrs, "datadog.operation.name", &group.name); push(&mut attrs, "datadog.span.type", &group.r#type); + push(&mut attrs, "datadog.svc_src", &group.service_source); // Only `synthetics` is surfaced as `datadog.origin`: the aggregation key carries just a // boolean, not the full origin string, so other origins are lost upstream. if group.synthetics { @@ -255,10 +247,7 @@ fn build_attributes( attrs } -fn build_resource_attributes( - info: &OtlpResourceInfo, - otel_trace_semantics_enabled: bool, -) -> Vec { +fn build_resource_attributes(info: &OtlpResourceInfo) -> Vec { let mut attrs: Vec = [ ("service.name", info.service.as_str()), ("service.version", info.app_version.as_str()), @@ -273,21 +262,19 @@ fn build_resource_attributes( .map(|(k, v)| kv_str(k, v)) .collect(); - if !otel_trace_semantics_enabled { - if !info.runtime_id.is_empty() { - attrs.push(kv_str("datadog.runtime_id", &info.runtime_id)); - } - // Mirrors the v0.6/legacy stats export's process_tags string; keep both in sync if that - // format changes. - let process_tags: Vec<&str> = info - .process_tags - .split(',') - .map(str::trim) - .filter(|p| !p.is_empty()) - .collect(); - if !process_tags.is_empty() { - attrs.push(kv_str_array("datadog.process_tags", process_tags)); - } + if !info.runtime_id.is_empty() { + attrs.push(kv_str("datadog.runtime_id", &info.runtime_id)); + } + // Mirrors the v0.6/legacy stats export's process_tags string; keep both in sync if that + // format changes. + let process_tags: Vec<&str> = info + .process_tags + .split(',') + .map(str::trim) + .filter(|p| !p.is_empty()) + .collect(); + if !process_tags.is_empty() { + attrs.push(kv_str_array("datadog.process_tags", process_tags)); } attrs } @@ -313,11 +300,7 @@ impl OtlpStatsExporter { if buckets.is_empty() { return Ok(false); } - let Some(request) = map_stats_to_otlp_metrics( - &buckets, - &self.resource, - self.config.otel_trace_semantics_enabled, - ) else { + let Some(request) = map_stats_to_otlp_metrics(&buckets, &self.resource) else { return Ok(false); }; send_otlp_http( @@ -498,36 +481,33 @@ mod tests { #[test] fn metric_shape_and_resource_attributes() { - assert!(map_stats_to_otlp_metrics(&[], &resource(), false).is_none()); + assert!(map_stats_to_otlp_metrics(&[], &resource()).is_none()); let mut r = resource(); r.app_version = "1.2.3".to_string(); r.hostname = "my-host".to_string(); r.runtime_id = "abc-123".to_string(); r.process_tags = "entrypoint.name:server".to_string(); - for otel in [false, true] { - let req = map_stats_to_otlp_metrics(&buckets(vec![one_ok_group()]), &r, otel).unwrap(); - let m = metric(&req); - assert_eq!(m["name"], "traces.span.sdk.metrics.duration"); - assert_eq!(m["unit"], "s"); - assert_eq!(m["histogram"]["aggregationTemporality"], 1); - assert!(req["resourceMetrics"][0]["scopeMetrics"][0]["scope"].is_null()); - let a = resource_attrs(&req); - assert_eq!(str_at(a, "service.name"), Some("svc")); - assert_eq!(str_at(a, "service.version"), Some("1.2.3")); - assert_eq!(str_at(a, "deployment.environment.name"), Some("test")); - assert_eq!(str_at(a, "host.name"), Some("my-host")); - assert_eq!(str_at(a, "telemetry.sdk.name"), Some("datadog")); - let dd = !otel; - assert_eq!(str_at(a, "datadog.runtime_id").is_some(), dd); - assert_eq!( - str_array_at(a, "datadog.process_tags"), - dd.then(|| vec!["entrypoint.name:server"]) - ); - } + let req = map_stats_to_otlp_metrics(&buckets(vec![one_ok_group()]), &r).unwrap(); + let m = metric(&req); + assert_eq!(m["name"], "traces.span.sdk.metrics.duration"); + assert_eq!(m["unit"], "s"); + assert_eq!(m["histogram"]["aggregationTemporality"], 1); + assert!(req["resourceMetrics"][0]["scopeMetrics"][0]["scope"].is_null()); + let a = resource_attrs(&req); + assert_eq!(str_at(a, "service.name"), Some("svc")); + assert_eq!(str_at(a, "service.version"), Some("1.2.3")); + assert_eq!(str_at(a, "deployment.environment.name"), Some("test")); + assert_eq!(str_at(a, "host.name"), Some("my-host")); + assert_eq!(str_at(a, "telemetry.sdk.name"), Some("datadog")); + assert_eq!(str_at(a, "datadog.runtime_id"), Some("abc-123")); + assert_eq!( + str_array_at(a, "datadog.process_tags"), + Some(vec!["entrypoint.name:server"]) + ); } #[test] - fn data_point_attributes_and_otel_strip() { + fn data_point_attributes() { let g_pair = group_with_exact(&[1_000_000_000], &[], |g| { g.http_status_code = 404; g.http_method = "POST".into(); @@ -542,9 +522,9 @@ mod tests { let custom_pair = group_with_exact(&[1_000_000_000], &[], |g| { g.service = "svc-other".into(); }); - let bs = buckets(vec![g_pair.clone(), custom_pair]); + let bs = buckets(vec![g_pair, custom_pair]); - let req = map_stats_to_otlp_metrics(&bs, &resource(), false).unwrap(); + let req = map_stats_to_otlp_metrics(&bs, &resource()).unwrap(); let pts = points(&req); let a = pts .iter() @@ -572,30 +552,6 @@ mod tests { assert!(pts.iter().any( |p| str_at(p["attributes"].as_array().unwrap(), "service.name") == Some("svc-other") )); - - // OTel mode retains semantic and custom attributes while suppressing Datadog attributes. - let req = map_stats_to_otlp_metrics(&buckets(vec![g_pair]), &resource(), true).unwrap(); - let a = points(&req)[0]["attributes"].as_array().unwrap(); - let mut keys: Vec<&str> = a.iter().map(|kv| kv["key"].as_str().unwrap()).collect(); - keys.sort_unstable(); - assert_eq!( - keys, - [ - "custom.primary", - "http.request.method", - "http.response.status_code", - "http.route", - "rpc.response.status_code", - "service.name", - "span.kind", - "span.name", - "status.code", - ] - ); - assert!(!keys.iter().any(|key| key.starts_with("datadog."))); - assert!(!keys.contains(&"additional_metric_tags")); - assert_eq!(str_at(a, "status.code"), Some(STATUS_CODE_OK)); - assert_eq!(str_at(a, "service.name"), Some("svc")); } #[test] @@ -610,7 +566,7 @@ mod tests { g.is_trace_root = value; g.top_level_hits = 0; }); - let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), false).unwrap(); + let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource()).unwrap(); let attrs = points(&req)[0]["attributes"].as_array().unwrap(); let is_trace_root = attrs.iter().find(|kv| kv["key"] == "datadog.is_trace_root"); assert_eq!( @@ -621,10 +577,35 @@ mod tests { } } + #[test] + fn service_source_is_emitted_as_string() { + let g = group_with_exact(&[1_000_000_000], &[], |g| { + g.service_source = "lambda".into(); + }); + let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource()).unwrap(); + let attrs = points(&req)[0]["attributes"].as_array().unwrap(); + let service_source = attrs.iter().find(|kv| kv["key"] == "datadog.svc_src"); + + assert_eq!( + service_source, + Some(&json!({ + "key": "datadog.svc_src", + "value": { "stringValue": "lambda" } + })) + ); + } + + #[test] + fn empty_service_source_is_omitted() { + let req = map_stats_to_otlp_metrics(&buckets(vec![one_ok_group()]), &resource()).unwrap(); + let attrs = points(&req)[0]["attributes"].as_array().unwrap(); + + assert!(!attrs.iter().any(|kv| kv["key"] == "datadog.svc_src")); + } + #[test] fn empty_span_kind_defaults_to_internal() { - let req = - map_stats_to_otlp_metrics(&buckets(vec![one_ok_group()]), &resource(), true).unwrap(); + let req = map_stats_to_otlp_metrics(&buckets(vec![one_ok_group()]), &resource()).unwrap(); assert_eq!( str_at( points(&req)[0]["attributes"].as_array().unwrap(), @@ -639,7 +620,7 @@ mod tests { let g = group_with_exact(&[1_000_000_000], &[], |g| { g.grpc_status_code = "5".into(); }); - let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), false).unwrap(); + let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource()).unwrap(); let a = points(&req)[0]["attributes"].as_array().unwrap(); assert_eq!(str_at(a, "rpc.response.status_code"), Some("NOT_FOUND")); @@ -648,7 +629,7 @@ mod tests { let g = group_with_exact(&[1_000_000_000], &[], |g| { g.grpc_status_code = code.into(); }); - let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), false).unwrap(); + let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource()).unwrap(); let a = points(&req)[0]["attributes"].as_array().unwrap(); assert!(!a.iter().any(|kv| kv["key"] == "rpc.response.status_code")); } @@ -657,8 +638,7 @@ mod tests { #[test] fn histogram_values_are_exact_and_distribution_uses_sketch() { // Single 1s ok span: count/sum/min/max all exact, distribution shaped by the sketch. - let req = - map_stats_to_otlp_metrics(&buckets(vec![one_ok_group()]), &resource(), false).unwrap(); + let req = map_stats_to_otlp_metrics(&buckets(vec![one_ok_group()]), &resource()).unwrap(); let p = &points(&req)[0]; assert_eq!(p["count"], "1"); assert_eq!(p["sum"].as_f64().unwrap(), 1.0); @@ -674,7 +654,7 @@ mod tests { // 3ms, 300ms, 3s land in three distinct buckets; exact sum = 3.303s. let g = group_with_exact(&[3_000_000, 300_000_000, 3_000_000_000], &[], |_| {}); - let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), false).unwrap(); + let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource()).unwrap(); let p = &points(&req)[0]; assert_eq!(p["count"], "3"); assert_eq!(p["sum"].as_f64().unwrap(), ns_to_s(3_303_000_000)); @@ -691,7 +671,7 @@ mod tests { let err = [700_000_000_u64]; let combined_ns = ok.iter().sum::() + err.iter().sum::(); let g = group_with_exact(&ok, &err, |_| {}); - let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), false).unwrap(); + let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource()).unwrap(); let pts = points(&req); assert_eq!(pts.len(), 2); let ok_pt = pts.iter().find(|p| !is_error_point(p)).unwrap(); @@ -724,26 +704,18 @@ mod tests { "datadog.custom:dd".into(), ]; }); - let req = map_stats_to_otlp_metrics(&buckets(vec![g.clone()]), &resource(), false).unwrap(); + let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource()).unwrap(); let a = points(&req)[0]["attributes"].as_array().unwrap(); assert_eq!(str_at(a, "custom.primary"), Some("a")); assert_eq!(str_at(a, "region"), Some("us-east")); assert_eq!(str_at(a, "endpoint"), Some("https://host:8080")); assert_eq!(str_at(a, "datadog.custom"), Some("dd")); - // OTel-semantics mode keeps custom tags and filters only Datadog-prefixed tags. - let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), true).unwrap(); - let a = points(&req)[0]["attributes"].as_array().unwrap(); - assert_eq!(str_at(a, "custom.primary"), Some("a")); - assert_eq!(str_at(a, "region"), Some("us-east")); - assert_eq!(str_at(a, "endpoint"), Some("https://host:8080")); - assert_eq!(str_at(a, "datadog.custom"), None); - // Malformed (no `:`) or empty-value entries are skipped rather than emitted verbatim. let g = group_with_exact(&[1_000_000_000], &[], |g| { g.additional_metric_tags = vec!["malformed".into(), "empty:".into()]; }); - let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), false).unwrap(); + let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource()).unwrap(); let a = points(&req)[0]["attributes"].as_array().unwrap(); assert!(!a.iter().any(|kv| kv["key"] == "malformed")); assert!(!a.iter().any(|kv| kv["key"] == "empty")); diff --git a/libdd-data-pipeline/src/trace_exporter/builder.rs b/libdd-data-pipeline/src/trace_exporter/builder.rs index a672028237..ef5184be51 100644 --- a/libdd-data-pipeline/src/trace_exporter/builder.rs +++ b/libdd-data-pipeline/src/trace_exporter/builder.rs @@ -754,7 +754,6 @@ impl TraceExporterBuilder { headers: build_otlp_header_map(self.otlp_metrics_headers), timeout: otlp_timeout, protocol: OtlpProtocol::HttpJson, - otel_trace_semantics_enabled: self.otel_trace_semantics_enabled, }); let runtime_id = self From db885913550dabaf7160e8f8decb738773685691 Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Thu, 6 Aug 2026 14:53:29 -0400 Subject: [PATCH 07/15] fix(data-pipeline): preserve metrics config compatibility Retain the OTel semantics field and mapper argument for downstream callers while keeping OTLP trace-metrics output unconditional. --- libdd-data-pipeline/src/otlp/config.rs | 3 ++ libdd-data-pipeline/src/otlp/metrics.rs | 38 +++++++++++-------- .../src/trace_exporter/builder.rs | 1 + 3 files changed, 27 insertions(+), 15 deletions(-) diff --git a/libdd-data-pipeline/src/otlp/config.rs b/libdd-data-pipeline/src/otlp/config.rs index 4f509d8358..1abc8c8fb6 100644 --- a/libdd-data-pipeline/src/otlp/config.rs +++ b/libdd-data-pipeline/src/otlp/config.rs @@ -134,4 +134,7 @@ pub struct OtlpMetricsConfig { /// Protocol (for future use; currently only HttpJson is supported). #[allow(dead_code)] pub(crate) protocol: OtlpProtocol, + /// Whether OTel trace semantics are enabled. Retained for downstream compatibility; OTLP + /// trace-metrics export does not branch on this value. + pub otel_trace_semantics_enabled: bool, } diff --git a/libdd-data-pipeline/src/otlp/metrics.rs b/libdd-data-pipeline/src/otlp/metrics.rs index 5cf20ab9fe..f363be69af 100644 --- a/libdd-data-pipeline/src/otlp/metrics.rs +++ b/libdd-data-pipeline/src/otlp/metrics.rs @@ -79,6 +79,7 @@ fn kv_str_array<'a>(key: &str, values: impl IntoIterator) -> Val pub fn map_stats_to_otlp_metrics( buckets: &[OtlpStatsBucket], resource_info: &OtlpResourceInfo, + _otel_trace_semantics_enabled: bool, ) -> Option { let mut data_points = Vec::new(); for b in buckets { @@ -300,7 +301,11 @@ impl OtlpStatsExporter { if buckets.is_empty() { return Ok(false); } - let Some(request) = map_stats_to_otlp_metrics(&buckets, &self.resource) else { + let Some(request) = map_stats_to_otlp_metrics( + &buckets, + &self.resource, + self.config.otel_trace_semantics_enabled, + ) else { return Ok(false); }; send_otlp_http( @@ -481,13 +486,13 @@ mod tests { #[test] fn metric_shape_and_resource_attributes() { - assert!(map_stats_to_otlp_metrics(&[], &resource()).is_none()); + assert!(map_stats_to_otlp_metrics(&[], &resource(), false).is_none()); let mut r = resource(); r.app_version = "1.2.3".to_string(); r.hostname = "my-host".to_string(); r.runtime_id = "abc-123".to_string(); r.process_tags = "entrypoint.name:server".to_string(); - let req = map_stats_to_otlp_metrics(&buckets(vec![one_ok_group()]), &r).unwrap(); + let req = map_stats_to_otlp_metrics(&buckets(vec![one_ok_group()]), &r, false).unwrap(); let m = metric(&req); assert_eq!(m["name"], "traces.span.sdk.metrics.duration"); assert_eq!(m["unit"], "s"); @@ -524,7 +529,7 @@ mod tests { }); let bs = buckets(vec![g_pair, custom_pair]); - let req = map_stats_to_otlp_metrics(&bs, &resource()).unwrap(); + let req = map_stats_to_otlp_metrics(&bs, &resource(), false).unwrap(); let pts = points(&req); let a = pts .iter() @@ -566,7 +571,7 @@ mod tests { g.is_trace_root = value; g.top_level_hits = 0; }); - let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource()).unwrap(); + let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), false).unwrap(); let attrs = points(&req)[0]["attributes"].as_array().unwrap(); let is_trace_root = attrs.iter().find(|kv| kv["key"] == "datadog.is_trace_root"); assert_eq!( @@ -582,7 +587,7 @@ mod tests { let g = group_with_exact(&[1_000_000_000], &[], |g| { g.service_source = "lambda".into(); }); - let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource()).unwrap(); + let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), false).unwrap(); let attrs = points(&req)[0]["attributes"].as_array().unwrap(); let service_source = attrs.iter().find(|kv| kv["key"] == "datadog.svc_src"); @@ -597,7 +602,8 @@ mod tests { #[test] fn empty_service_source_is_omitted() { - let req = map_stats_to_otlp_metrics(&buckets(vec![one_ok_group()]), &resource()).unwrap(); + let req = + map_stats_to_otlp_metrics(&buckets(vec![one_ok_group()]), &resource(), false).unwrap(); let attrs = points(&req)[0]["attributes"].as_array().unwrap(); assert!(!attrs.iter().any(|kv| kv["key"] == "datadog.svc_src")); @@ -605,7 +611,8 @@ mod tests { #[test] fn empty_span_kind_defaults_to_internal() { - let req = map_stats_to_otlp_metrics(&buckets(vec![one_ok_group()]), &resource()).unwrap(); + let req = + map_stats_to_otlp_metrics(&buckets(vec![one_ok_group()]), &resource(), false).unwrap(); assert_eq!( str_at( points(&req)[0]["attributes"].as_array().unwrap(), @@ -620,7 +627,7 @@ mod tests { let g = group_with_exact(&[1_000_000_000], &[], |g| { g.grpc_status_code = "5".into(); }); - let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource()).unwrap(); + let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), false).unwrap(); let a = points(&req)[0]["attributes"].as_array().unwrap(); assert_eq!(str_at(a, "rpc.response.status_code"), Some("NOT_FOUND")); @@ -629,7 +636,7 @@ mod tests { let g = group_with_exact(&[1_000_000_000], &[], |g| { g.grpc_status_code = code.into(); }); - let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource()).unwrap(); + let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), false).unwrap(); let a = points(&req)[0]["attributes"].as_array().unwrap(); assert!(!a.iter().any(|kv| kv["key"] == "rpc.response.status_code")); } @@ -638,7 +645,8 @@ mod tests { #[test] fn histogram_values_are_exact_and_distribution_uses_sketch() { // Single 1s ok span: count/sum/min/max all exact, distribution shaped by the sketch. - let req = map_stats_to_otlp_metrics(&buckets(vec![one_ok_group()]), &resource()).unwrap(); + let req = + map_stats_to_otlp_metrics(&buckets(vec![one_ok_group()]), &resource(), false).unwrap(); let p = &points(&req)[0]; assert_eq!(p["count"], "1"); assert_eq!(p["sum"].as_f64().unwrap(), 1.0); @@ -654,7 +662,7 @@ mod tests { // 3ms, 300ms, 3s land in three distinct buckets; exact sum = 3.303s. let g = group_with_exact(&[3_000_000, 300_000_000, 3_000_000_000], &[], |_| {}); - let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource()).unwrap(); + let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), false).unwrap(); let p = &points(&req)[0]; assert_eq!(p["count"], "3"); assert_eq!(p["sum"].as_f64().unwrap(), ns_to_s(3_303_000_000)); @@ -671,7 +679,7 @@ mod tests { let err = [700_000_000_u64]; let combined_ns = ok.iter().sum::() + err.iter().sum::(); let g = group_with_exact(&ok, &err, |_| {}); - let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource()).unwrap(); + let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), false).unwrap(); let pts = points(&req); assert_eq!(pts.len(), 2); let ok_pt = pts.iter().find(|p| !is_error_point(p)).unwrap(); @@ -704,7 +712,7 @@ mod tests { "datadog.custom:dd".into(), ]; }); - let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource()).unwrap(); + let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), false).unwrap(); let a = points(&req)[0]["attributes"].as_array().unwrap(); assert_eq!(str_at(a, "custom.primary"), Some("a")); assert_eq!(str_at(a, "region"), Some("us-east")); @@ -715,7 +723,7 @@ mod tests { let g = group_with_exact(&[1_000_000_000], &[], |g| { g.additional_metric_tags = vec!["malformed".into(), "empty:".into()]; }); - let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource()).unwrap(); + let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), false).unwrap(); let a = points(&req)[0]["attributes"].as_array().unwrap(); assert!(!a.iter().any(|kv| kv["key"] == "malformed")); assert!(!a.iter().any(|kv| kv["key"] == "empty")); diff --git a/libdd-data-pipeline/src/trace_exporter/builder.rs b/libdd-data-pipeline/src/trace_exporter/builder.rs index ef5184be51..a672028237 100644 --- a/libdd-data-pipeline/src/trace_exporter/builder.rs +++ b/libdd-data-pipeline/src/trace_exporter/builder.rs @@ -754,6 +754,7 @@ impl TraceExporterBuilder { headers: build_otlp_header_map(self.otlp_metrics_headers), timeout: otlp_timeout, protocol: OtlpProtocol::HttpJson, + otel_trace_semantics_enabled: self.otel_trace_semantics_enabled, }); let runtime_id = self From f0ac2a203ee10ffb5a8c864f8448f2e29ebba782 Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Thu, 6 Aug 2026 15:00:23 -0400 Subject: [PATCH 08/15] fix(data-pipeline): default unknown metric span kinds --- libdd-data-pipeline/src/otlp/config.rs | 3 +- libdd-data-pipeline/src/otlp/metrics.rs | 37 +++++++++----------- libdd-trace-utils/src/otlp_encoder/mapper.rs | 2 +- 3 files changed, 18 insertions(+), 24 deletions(-) diff --git a/libdd-data-pipeline/src/otlp/config.rs b/libdd-data-pipeline/src/otlp/config.rs index 1abc8c8fb6..ddd2987ff9 100644 --- a/libdd-data-pipeline/src/otlp/config.rs +++ b/libdd-data-pipeline/src/otlp/config.rs @@ -134,7 +134,6 @@ pub struct OtlpMetricsConfig { /// Protocol (for future use; currently only HttpJson is supported). #[allow(dead_code)] pub(crate) protocol: OtlpProtocol, - /// Whether OTel trace semantics are enabled. Retained for downstream compatibility; OTLP - /// trace-metrics export does not branch on this value. + /// Retained for downstream compatibility; OTLP trace metrics ignore this value. pub otel_trace_semantics_enabled: bool, } diff --git a/libdd-data-pipeline/src/otlp/metrics.rs b/libdd-data-pipeline/src/otlp/metrics.rs index f363be69af..71fef8023e 100644 --- a/libdd-data-pipeline/src/otlp/metrics.rs +++ b/libdd-data-pipeline/src/otlp/metrics.rs @@ -180,11 +180,7 @@ fn build_attributes( push( &mut attrs, "span.kind", - if group.span_kind.is_empty() { - "SPAN_KIND_INTERNAL" - } else { - tag_to_otlp_kind_str_name(&group.span_kind) - }, + tag_to_otlp_kind_str_name(&group.span_kind), ); push( &mut attrs, @@ -208,7 +204,6 @@ fn build_attributes( group.http_status_code as i64, )); } - // additional_metric_tags support is still evolving/TBD across most SDKs. for tag in &group.additional_metric_tags { if let Some((k, v)) = tag.split_once(':') { push(&mut attrs, k, v); @@ -237,8 +232,6 @@ fn build_attributes( attrs.push(json!({ "key": "datadog.span.top_level", "value": { "boolValue": top_level } })); - // Unlike process_tags (a resource attribute), peer_tags is per-span and belongs on the data - // point. if !group.peer_tags.is_empty() { attrs.push(kv_str_array( "datadog.peer_tags", @@ -266,8 +259,7 @@ fn build_resource_attributes(info: &OtlpResourceInfo) -> Vec { if !info.runtime_id.is_empty() { attrs.push(kv_str("datadog.runtime_id", &info.runtime_id)); } - // Mirrors the v0.6/legacy stats export's process_tags string; keep both in sync if that - // format changes. + // Keep this representation aligned with v0.6 stats export. let process_tags: Vec<&str> = info .process_tags .split(',') @@ -537,7 +529,6 @@ mod tests { .unwrap()["attributes"] .as_array() .unwrap(); - // service.name is on the data point even though it matches the resource default. assert_eq!(str_at(a, "service.name"), Some("svc")); assert_eq!(str_at(a, "span.name"), Some("GET /foo")); assert_eq!(str_at(a, "span.kind"), Some("SPAN_KIND_SERVER")); @@ -610,16 +601,20 @@ mod tests { } #[test] - fn empty_span_kind_defaults_to_internal() { - let req = - map_stats_to_otlp_metrics(&buckets(vec![one_ok_group()]), &resource(), false).unwrap(); - assert_eq!( - str_at( - points(&req)[0]["attributes"].as_array().unwrap(), - "span.kind" - ), - Some("SPAN_KIND_INTERNAL") - ); + fn absent_or_unknown_span_kind_defaults_to_internal() { + for span_kind in ["", "unknown"] { + let group = group_with_exact(&[1_000_000_000], &[], |g| { + g.span_kind = span_kind.into(); + }); + let req = map_stats_to_otlp_metrics(&buckets(vec![group]), &resource(), false).unwrap(); + assert_eq!( + str_at( + points(&req)[0]["attributes"].as_array().unwrap(), + "span.kind" + ), + Some("SPAN_KIND_INTERNAL") + ); + } } #[test] diff --git a/libdd-trace-utils/src/otlp_encoder/mapper.rs b/libdd-trace-utils/src/otlp_encoder/mapper.rs index bf21f3b143..2bfe5122c7 100644 --- a/libdd-trace-utils/src/otlp_encoder/mapper.rs +++ b/libdd-trace-utils/src/otlp_encoder/mapper.rs @@ -118,7 +118,7 @@ pub fn tag_to_otlp_kind_str_name(t: &str) -> &'static str { span_kind::PRODUCER => "SPAN_KIND_PRODUCER", span_kind::CONSUMER => "SPAN_KIND_CONSUMER", span_kind::INTERNAL => "SPAN_KIND_INTERNAL", - _ => "SPAN_KIND_UNSPECIFIED", + _ => "SPAN_KIND_INTERNAL", } } From b3945eda9bccc19ef3d1bbaf26fc96bb0989131d Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Thu, 6 Aug 2026 15:06:35 -0400 Subject: [PATCH 09/15] fix(data-pipeline): decouple metrics exporter config --- libdd-data-pipeline/src/otlp/config.rs | 1 + libdd-data-pipeline/src/otlp/metrics.rs | 6 +----- 2 files changed, 2 insertions(+), 5 deletions(-) diff --git a/libdd-data-pipeline/src/otlp/config.rs b/libdd-data-pipeline/src/otlp/config.rs index ddd2987ff9..e8edcdf610 100644 --- a/libdd-data-pipeline/src/otlp/config.rs +++ b/libdd-data-pipeline/src/otlp/config.rs @@ -135,5 +135,6 @@ pub struct OtlpMetricsConfig { #[allow(dead_code)] pub(crate) protocol: OtlpProtocol, /// Retained for downstream compatibility; OTLP trace metrics ignore this value. + #[allow(dead_code)] pub otel_trace_semantics_enabled: bool, } diff --git a/libdd-data-pipeline/src/otlp/metrics.rs b/libdd-data-pipeline/src/otlp/metrics.rs index 71fef8023e..7cbaef7c18 100644 --- a/libdd-data-pipeline/src/otlp/metrics.rs +++ b/libdd-data-pipeline/src/otlp/metrics.rs @@ -293,11 +293,7 @@ impl OtlpStatsExporter { if buckets.is_empty() { return Ok(false); } - let Some(request) = map_stats_to_otlp_metrics( - &buckets, - &self.resource, - self.config.otel_trace_semantics_enabled, - ) else { + let Some(request) = map_stats_to_otlp_metrics(&buckets, &self.resource, false) else { return Ok(false); }; send_otlp_http( From e5998987f4fd53af3ae3c3004de0fb40aeaf1b7a Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Fri, 7 Aug 2026 10:45:52 -0400 Subject: [PATCH 10/15] refactor(data-pipeline): simplify OTLP metric mapping --- libdd-data-pipeline/src/otlp/metrics.rs | 91 +++++++------------- libdd-trace-utils/src/otlp_encoder/mapper.rs | 12 --- 2 files changed, 31 insertions(+), 72 deletions(-) diff --git a/libdd-data-pipeline/src/otlp/metrics.rs b/libdd-data-pipeline/src/otlp/metrics.rs index 7cbaef7c18..d3d63b3448 100644 --- a/libdd-data-pipeline/src/otlp/metrics.rs +++ b/libdd-data-pipeline/src/otlp/metrics.rs @@ -13,7 +13,6 @@ use libdd_ddsketch::DDSketch; use libdd_shared_runtime::Worker; use libdd_trace_protobuf::pb; use libdd_trace_stats::span_concentrator::{OtlpStatsBucket, SpanConcentrator}; -use libdd_trace_utils::otlp_encoder::mapper::tag_to_otlp_kind_str_name; use libdd_trace_utils::otlp_encoder::OtlpResourceInfo; use serde_json::{json, Value}; use std::sync::{Arc, Mutex}; @@ -50,6 +49,20 @@ fn grpc_status_code_to_name(code: &str) -> Option<&'static str> { GRPC_STATUS_NAMES.get(code.parse::().ok()?).copied() } +fn span_kind_name(kind: &str) -> &'static str { + if kind.eq_ignore_ascii_case("server") { + "SPAN_KIND_SERVER" + } else if kind.eq_ignore_ascii_case("client") { + "SPAN_KIND_CLIENT" + } else if kind.eq_ignore_ascii_case("producer") { + "SPAN_KIND_PRODUCER" + } else if kind.eq_ignore_ascii_case("consumer") { + "SPAN_KIND_CONSUMER" + } else { + "SPAN_KIND_INTERNAL" + } +} + const STATUS_CODE_OK: &str = "STATUS_CODE_OK"; const STATUS_CODE_ERROR: &str = "STATUS_CODE_ERROR"; /// Fixed bucket boundaries (seconds) mirroring the OTel spanmetrics-connector defaults. @@ -177,11 +190,7 @@ fn build_attributes( attrs.push(kv_str("service.name", group_service)); push(&mut attrs, "span.name", &group.resource); - push( - &mut attrs, - "span.kind", - tag_to_otlp_kind_str_name(&group.span_kind), - ); + push(&mut attrs, "span.kind", span_kind_name(&group.span_kind)); push( &mut attrs, "status.code", @@ -204,12 +213,6 @@ fn build_attributes( group.http_status_code as i64, )); } - for tag in &group.additional_metric_tags { - if let Some((k, v)) = tag.split_once(':') { - push(&mut attrs, k, v); - } - } - push(&mut attrs, "datadog.operation.name", &group.name); push(&mut attrs, "datadog.span.type", &group.r#type); push(&mut attrs, "datadog.svc_src", &group.service_source); @@ -238,6 +241,11 @@ fn build_attributes( group.peer_tags.iter().map(String::as_str), )); } + for tag in &group.additional_metric_tags { + if let Some((k, v)) = tag.split_once(':') { + push(&mut attrs, k, v); + } + } attrs } @@ -509,6 +517,7 @@ mod tests { g.span_kind = "server".into(); g.grpc_status_code = "5".into(); g.is_trace_root = pb::Trilean::True as i32; + g.service_source = "lambda".into(); g.peer_tags = vec!["db.hostname:prod-db-1".into(), "db.name:orders".into()]; g.additional_metric_tags = vec!["custom.primary:a".into()]; }); @@ -533,6 +542,7 @@ mod tests { assert!(a.iter().any(|kv| kv["key"] == "http.response.status_code")); assert_eq!(str_at(a, "datadog.operation.name"), Some("test.op")); assert_eq!(str_at(a, "datadog.span.type"), Some("web")); + assert_eq!(str_at(a, "datadog.svc_src"), Some("lambda")); assert_eq!(str_at(a, "datadog.origin"), Some("synthetics")); assert_eq!(bool_at(a, "datadog.is_trace_root"), Some(true)); assert_eq!(bool_at(a, "datadog.span.top_level"), Some(true)); @@ -541,9 +551,15 @@ mod tests { Some(vec!["db.hostname:prod-db-1", "db.name:orders"]) ); assert_eq!(str_at(a, "status.code"), Some(STATUS_CODE_OK)); - assert!(pts.iter().any( - |p| str_at(p["attributes"].as_array().unwrap(), "service.name") == Some("svc-other") - )); + let custom = pts + .iter() + .find(|p| { + str_at(p["attributes"].as_array().unwrap(), "service.name") == Some("svc-other") + }) + .unwrap()["attributes"] + .as_array() + .unwrap(); + assert_eq!(str_at(custom, "span.kind"), Some("SPAN_KIND_INTERNAL")); } #[test] @@ -552,7 +568,6 @@ mod tests { (pb::Trilean::NotSet as i32, None), (pb::Trilean::True as i32, Some(true)), (pb::Trilean::False as i32, Some(false)), - (i32::MAX, None), ] { let g = group_with_exact(&[1_000_000_000], &[], |g| { g.is_trace_root = value; @@ -569,50 +584,6 @@ mod tests { } } - #[test] - fn service_source_is_emitted_as_string() { - let g = group_with_exact(&[1_000_000_000], &[], |g| { - g.service_source = "lambda".into(); - }); - let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), false).unwrap(); - let attrs = points(&req)[0]["attributes"].as_array().unwrap(); - let service_source = attrs.iter().find(|kv| kv["key"] == "datadog.svc_src"); - - assert_eq!( - service_source, - Some(&json!({ - "key": "datadog.svc_src", - "value": { "stringValue": "lambda" } - })) - ); - } - - #[test] - fn empty_service_source_is_omitted() { - let req = - map_stats_to_otlp_metrics(&buckets(vec![one_ok_group()]), &resource(), false).unwrap(); - let attrs = points(&req)[0]["attributes"].as_array().unwrap(); - - assert!(!attrs.iter().any(|kv| kv["key"] == "datadog.svc_src")); - } - - #[test] - fn absent_or_unknown_span_kind_defaults_to_internal() { - for span_kind in ["", "unknown"] { - let group = group_with_exact(&[1_000_000_000], &[], |g| { - g.span_kind = span_kind.into(); - }); - let req = map_stats_to_otlp_metrics(&buckets(vec![group]), &resource(), false).unwrap(); - assert_eq!( - str_at( - points(&req)[0]["attributes"].as_array().unwrap(), - "span.kind" - ), - Some("SPAN_KIND_INTERNAL") - ); - } - } - #[test] fn emits_canonical_grpc_status_name_for_rpc_response_status_code() { let g = group_with_exact(&[1_000_000_000], &[], |g| { diff --git a/libdd-trace-utils/src/otlp_encoder/mapper.rs b/libdd-trace-utils/src/otlp_encoder/mapper.rs index 2bfe5122c7..bc3a94c65c 100644 --- a/libdd-trace-utils/src/otlp_encoder/mapper.rs +++ b/libdd-trace-utils/src/otlp_encoder/mapper.rs @@ -110,18 +110,6 @@ fn tag_to_otlp_kind(t: &str) -> i32 { } } -/// Maps a Datadog `span.kind` tag value to the OTel Span Metrics Connector's string convention. -pub fn tag_to_otlp_kind_str_name(t: &str) -> &'static str { - match tag_to_otlp_kind(t) { - span_kind::SERVER => "SPAN_KIND_SERVER", - span_kind::CLIENT => "SPAN_KIND_CLIENT", - span_kind::PRODUCER => "SPAN_KIND_PRODUCER", - span_kind::CONSUMER => "SPAN_KIND_CONSUMER", - span_kind::INTERNAL => "SPAN_KIND_INTERNAL", - _ => "SPAN_KIND_INTERNAL", - } -} - /// Maps the Datadog span type field (set by DD-instrumented tracers) to an OTLP SpanKind. fn dd_type_to_otlp_kind(t: &str) -> i32 { // Case-insensitive match without allocating (see `tag_to_otlp_kind`). From bfef125870b40f6e2e4e4ccbae62f77ebb5df8b5 Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Fri, 7 Aug 2026 10:53:02 -0400 Subject: [PATCH 11/15] refactor(data-pipeline): align OTLP metric attribute order --- libdd-data-pipeline/src/otlp/metrics.rs | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/libdd-data-pipeline/src/otlp/metrics.rs b/libdd-data-pipeline/src/otlp/metrics.rs index d3d63b3448..8049f5555e 100644 --- a/libdd-data-pipeline/src/otlp/metrics.rs +++ b/libdd-data-pipeline/src/otlp/metrics.rs @@ -189,17 +189,16 @@ fn build_attributes( }; attrs.push(kv_str("service.name", group_service)); - push(&mut attrs, "span.name", &group.resource); - push(&mut attrs, "span.kind", span_kind_name(&group.span_kind)); - push( - &mut attrs, + attrs.push(kv_str( "status.code", if is_error { STATUS_CODE_ERROR } else { STATUS_CODE_OK }, - ); + )); + push(&mut attrs, "span.kind", span_kind_name(&group.span_kind)); + push(&mut attrs, "span.name", &group.resource); push(&mut attrs, "http.request.method", &group.http_method); push(&mut attrs, "http.route", &group.http_endpoint); @@ -213,7 +212,7 @@ fn build_attributes( group.http_status_code as i64, )); } - push(&mut attrs, "datadog.operation.name", &group.name); + attrs.push(kv_str("datadog.operation.name", &group.name)); push(&mut attrs, "datadog.span.type", &group.r#type); push(&mut attrs, "datadog.svc_src", &group.service_source); // Only `synthetics` is surfaced as `datadog.origin`: the aggregation key carries just a From 0034e1bdcac1680dea5e0b0cc929e1ed59734232 Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Fri, 7 Aug 2026 11:18:32 -0400 Subject: [PATCH 12/15] refactor(data-pipeline): remove unused OTLP mapper option --- libdd-data-pipeline/src/otlp/metrics.rs | 26 ++++++++++++------------- 1 file changed, 12 insertions(+), 14 deletions(-) diff --git a/libdd-data-pipeline/src/otlp/metrics.rs b/libdd-data-pipeline/src/otlp/metrics.rs index 8049f5555e..d2d124df55 100644 --- a/libdd-data-pipeline/src/otlp/metrics.rs +++ b/libdd-data-pipeline/src/otlp/metrics.rs @@ -92,7 +92,6 @@ fn kv_str_array<'a>(key: &str, values: impl IntoIterator) -> Val pub fn map_stats_to_otlp_metrics( buckets: &[OtlpStatsBucket], resource_info: &OtlpResourceInfo, - _otel_trace_semantics_enabled: bool, ) -> Option { let mut data_points = Vec::new(); for b in buckets { @@ -300,7 +299,7 @@ impl OtlpStatsExporter { if buckets.is_empty() { return Ok(false); } - let Some(request) = map_stats_to_otlp_metrics(&buckets, &self.resource, false) else { + let Some(request) = map_stats_to_otlp_metrics(&buckets, &self.resource) else { return Ok(false); }; send_otlp_http( @@ -481,13 +480,13 @@ mod tests { #[test] fn metric_shape_and_resource_attributes() { - assert!(map_stats_to_otlp_metrics(&[], &resource(), false).is_none()); + assert!(map_stats_to_otlp_metrics(&[], &resource()).is_none()); let mut r = resource(); r.app_version = "1.2.3".to_string(); r.hostname = "my-host".to_string(); r.runtime_id = "abc-123".to_string(); r.process_tags = "entrypoint.name:server".to_string(); - let req = map_stats_to_otlp_metrics(&buckets(vec![one_ok_group()]), &r, false).unwrap(); + let req = map_stats_to_otlp_metrics(&buckets(vec![one_ok_group()]), &r).unwrap(); let m = metric(&req); assert_eq!(m["name"], "traces.span.sdk.metrics.duration"); assert_eq!(m["unit"], "s"); @@ -525,7 +524,7 @@ mod tests { }); let bs = buckets(vec![g_pair, custom_pair]); - let req = map_stats_to_otlp_metrics(&bs, &resource(), false).unwrap(); + let req = map_stats_to_otlp_metrics(&bs, &resource()).unwrap(); let pts = points(&req); let a = pts .iter() @@ -572,7 +571,7 @@ mod tests { g.is_trace_root = value; g.top_level_hits = 0; }); - let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), false).unwrap(); + let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource()).unwrap(); let attrs = points(&req)[0]["attributes"].as_array().unwrap(); let is_trace_root = attrs.iter().find(|kv| kv["key"] == "datadog.is_trace_root"); assert_eq!( @@ -588,7 +587,7 @@ mod tests { let g = group_with_exact(&[1_000_000_000], &[], |g| { g.grpc_status_code = "5".into(); }); - let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), false).unwrap(); + let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource()).unwrap(); let a = points(&req)[0]["attributes"].as_array().unwrap(); assert_eq!(str_at(a, "rpc.response.status_code"), Some("NOT_FOUND")); @@ -597,7 +596,7 @@ mod tests { let g = group_with_exact(&[1_000_000_000], &[], |g| { g.grpc_status_code = code.into(); }); - let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), false).unwrap(); + let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource()).unwrap(); let a = points(&req)[0]["attributes"].as_array().unwrap(); assert!(!a.iter().any(|kv| kv["key"] == "rpc.response.status_code")); } @@ -606,8 +605,7 @@ mod tests { #[test] fn histogram_values_are_exact_and_distribution_uses_sketch() { // Single 1s ok span: count/sum/min/max all exact, distribution shaped by the sketch. - let req = - map_stats_to_otlp_metrics(&buckets(vec![one_ok_group()]), &resource(), false).unwrap(); + let req = map_stats_to_otlp_metrics(&buckets(vec![one_ok_group()]), &resource()).unwrap(); let p = &points(&req)[0]; assert_eq!(p["count"], "1"); assert_eq!(p["sum"].as_f64().unwrap(), 1.0); @@ -623,7 +621,7 @@ mod tests { // 3ms, 300ms, 3s land in three distinct buckets; exact sum = 3.303s. let g = group_with_exact(&[3_000_000, 300_000_000, 3_000_000_000], &[], |_| {}); - let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), false).unwrap(); + let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource()).unwrap(); let p = &points(&req)[0]; assert_eq!(p["count"], "3"); assert_eq!(p["sum"].as_f64().unwrap(), ns_to_s(3_303_000_000)); @@ -640,7 +638,7 @@ mod tests { let err = [700_000_000_u64]; let combined_ns = ok.iter().sum::() + err.iter().sum::(); let g = group_with_exact(&ok, &err, |_| {}); - let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), false).unwrap(); + let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource()).unwrap(); let pts = points(&req); assert_eq!(pts.len(), 2); let ok_pt = pts.iter().find(|p| !is_error_point(p)).unwrap(); @@ -673,7 +671,7 @@ mod tests { "datadog.custom:dd".into(), ]; }); - let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), false).unwrap(); + let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource()).unwrap(); let a = points(&req)[0]["attributes"].as_array().unwrap(); assert_eq!(str_at(a, "custom.primary"), Some("a")); assert_eq!(str_at(a, "region"), Some("us-east")); @@ -684,7 +682,7 @@ mod tests { let g = group_with_exact(&[1_000_000_000], &[], |g| { g.additional_metric_tags = vec!["malformed".into(), "empty:".into()]; }); - let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource(), false).unwrap(); + let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource()).unwrap(); let a = points(&req)[0]["attributes"].as_array().unwrap(); assert!(!a.iter().any(|kv| kv["key"] == "malformed")); assert!(!a.iter().any(|kv| kv["key"] == "empty")); From 78ce6ff5a2ae69c9068109a79e902371f4aa49af Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Mon, 10 Aug 2026 14:31:30 -0400 Subject: [PATCH 13/15] feat(data-pipeline): support OTLP trace metrics for ddtrace-py --- libdd-data-pipeline/src/otlp/metrics.rs | 31 +++++++++++++++++++ .../src/trace_exporter/builder.rs | 18 ++++++++++- libdd-data-pipeline/src/trace_exporter/mod.rs | 1 + .../src/trace_exporter/stats.rs | 4 ++- libdd-trace-utils/src/otlp_encoder/mod.rs | 1 + 5 files changed, 53 insertions(+), 2 deletions(-) diff --git a/libdd-data-pipeline/src/otlp/metrics.rs b/libdd-data-pipeline/src/otlp/metrics.rs index d2d124df55..1b0cb7dc52 100644 --- a/libdd-data-pipeline/src/otlp/metrics.rs +++ b/libdd-data-pipeline/src/otlp/metrics.rs @@ -275,6 +275,12 @@ fn build_resource_attributes(info: &OtlpResourceInfo) -> Vec { if !process_tags.is_empty() { attrs.push(kv_str_array("datadog.process_tags", process_tags)); } + if !info.tracer_tags.is_empty() { + attrs.push(kv_str_array( + "datadog.tracer_tags", + info.tracer_tags.iter().map(String::as_str), + )); + } attrs } @@ -486,6 +492,7 @@ mod tests { r.hostname = "my-host".to_string(); r.runtime_id = "abc-123".to_string(); r.process_tags = "entrypoint.name:server".to_string(); + r.tracer_tags = vec!["team:apm".to_string(), "tier:backend".to_string()]; let req = map_stats_to_otlp_metrics(&buckets(vec![one_ok_group()]), &r).unwrap(); let m = metric(&req); assert_eq!(m["name"], "traces.span.sdk.metrics.duration"); @@ -503,6 +510,10 @@ mod tests { str_array_at(a, "datadog.process_tags"), Some(vec!["entrypoint.name:server"]) ); + assert_eq!( + str_array_at(a, "datadog.tracer_tags"), + Some(vec!["team:apm", "tier:backend"]) + ); } #[test] @@ -560,6 +571,26 @@ mod tests { assert_eq!(str_at(custom, "span.kind"), Some("SPAN_KIND_INTERNAL")); } + #[test] + fn empty_optional_data_point_attributes_are_omitted() { + let g = group_with_exact(&[1_000_000_000], &[], |g| { + g.http_method.clear(); + g.http_endpoint.clear(); + g.r#type.clear(); + g.service_source.clear(); + }); + let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource()).unwrap(); + let attrs = points(&req)[0]["attributes"].as_array().unwrap(); + for key in [ + "http.request.method", + "http.route", + "datadog.span.type", + "datadog.svc_src", + ] { + assert!(!attrs.iter().any(|kv| kv["key"] == key)); + } + } + #[test] fn is_trace_root_preserves_trilean_semantics() { for (value, expected) in [ diff --git a/libdd-data-pipeline/src/trace_exporter/builder.rs b/libdd-data-pipeline/src/trace_exporter/builder.rs index 4a01daf78a..79f057ca4d 100644 --- a/libdd-data-pipeline/src/trace_exporter/builder.rs +++ b/libdd-data-pipeline/src/trace_exporter/builder.rs @@ -66,6 +66,7 @@ pub struct TraceExporterBuilder { instrumentation_scope_version: String, git_commit_sha: String, process_tags: String, + tracer_tags: Vec, container_id: String, input_format: TraceExporterInputFormat, output_format: TraceExporterOutputFormat, @@ -76,6 +77,7 @@ pub struct TraceExporterBuilder { /// A Some value enables stats-computation, None if it is disabled stats_bucket_size: Option, peer_tags: Vec, + additional_metric_tag_keys: Vec, stats_cardinality_limits: Option, #[cfg(feature = "stats-obfuscation")] client_side_stats_obfuscation_enabled: bool, @@ -141,6 +143,7 @@ impl TraceExporterBuilder { instrumentation_scope_version: String::new(), git_commit_sha: String::new(), process_tags: String::new(), + tracer_tags: Vec::new(), container_id: String::new(), input_format: TraceExporterInputFormat::default(), output_format: TraceExporterOutputFormat::default(), @@ -149,6 +152,7 @@ impl TraceExporterBuilder { client_computed_top_level: false, stats_bucket_size: None, peer_tags: Vec::new(), + additional_metric_tag_keys: Vec::new(), stats_cardinality_limits: None, #[cfg(feature = "stats-obfuscation")] client_side_stats_obfuscation_enabled: false, @@ -242,6 +246,11 @@ impl TraceExporterBuilder { self } + pub fn set_tracer_tags(&mut self, tracer_tags: Vec) -> &mut Self { + self.tracer_tags = tracer_tags; + self + } + /// Set the `Datadog-Container-Id` header pub fn set_container_id(&mut self, container_id: &str) -> &mut Self { container_id.clone_into(&mut self.container_id); @@ -339,6 +348,11 @@ impl TraceExporterBuilder { self } + pub fn set_additional_metric_tag_keys(&mut self, tag_keys: Vec) -> &mut Self { + self.additional_metric_tag_keys = tag_keys; + self + } + /// Sets the cardinality limit for client-side stats computation. /// /// When the number of distinct stats groups exceeds `limit`, additional groups are @@ -783,7 +797,7 @@ impl TraceExporterBuilder { span_kinds, self.peer_tags.clone(), self.stats_cardinality_limits, - vec![], + self.additional_metric_tag_keys.clone(), #[cfg(feature = "stats-obfuscation")] None, ))); @@ -796,6 +810,7 @@ impl TraceExporterBuilder { resource.runtime_id = runtime_id.clone(); resource.hostname = self.hostname.clone(); resource.process_tags = self.process_tags.clone(); + resource.tracer_tags = self.tracer_tags.clone(); let worker = OtlpStatsExporter { flush_interval: bucket_size, concentrator: concentrator.clone(), @@ -860,6 +875,7 @@ impl TraceExporterBuilder { client_side_stats: StatsComputationConfig { status: ArcSwap::new(stats.into()), stats_cardinality_limits: self.stats_cardinality_limits, + additional_metric_tag_keys: self.additional_metric_tag_keys, #[cfg(feature = "stats-obfuscation")] obfuscation_config: Arc::new(ArcSwap::from_pointee( StatsComputationObfuscationConfig::default(), diff --git a/libdd-data-pipeline/src/trace_exporter/mod.rs b/libdd-data-pipeline/src/trace_exporter/mod.rs index a29ffc6a55..24dc646e0e 100644 --- a/libdd-data-pipeline/src/trace_exporter/mod.rs +++ b/libdd-data-pipeline/src/trace_exporter/mod.rs @@ -481,6 +481,7 @@ impl< endpoint_url: &self.endpoint.url, shared_runtime: &*self.shared_runtime, stats_cardinality_limits: self.client_side_stats.stats_cardinality_limits, + additional_metric_tag_keys: &self.client_side_stats.additional_metric_tag_keys, restart_after_fork: self.restart_after_fork, dogstatsd: if self.health_metrics_enabled { self.dogstatsd.clone() diff --git a/libdd-data-pipeline/src/trace_exporter/stats.rs b/libdd-data-pipeline/src/trace_exporter/stats.rs index 8fd9ef3356..5da3db50ad 100644 --- a/libdd-data-pipeline/src/trace_exporter/stats.rs +++ b/libdd-data-pipeline/src/trace_exporter/stats.rs @@ -50,6 +50,7 @@ pub(crate) struct StatsContext< pub endpoint_url: &'a http::Uri, pub shared_runtime: &'a R, pub stats_cardinality_limits: Option, + pub additional_metric_tag_keys: &'a [String], /// Configuration option to pass to [SharedRuntime::spawn_worker] pub restart_after_fork: bool, /// Optional DogStatsD client forwarded to the [`StatsExporter`]. @@ -80,6 +81,7 @@ pub(crate) enum StatsComputationStatus { pub(crate) struct StatsComputationConfig { pub(crate) status: ArcSwap, pub(crate) stats_cardinality_limits: Option, + pub(crate) additional_metric_tag_keys: Vec, #[cfg(feature = "stats-obfuscation")] pub(crate) obfuscation_config: SharedStatsComputationObfuscationConfig, /// Builder-level opt-in. When false, stats obfuscation stays off @@ -142,7 +144,7 @@ pub(crate) fn start_stats_computation< span_kinds, peer_tags, ctx.stats_cardinality_limits, - vec![], + ctx.additional_metric_tag_keys.to_vec(), #[cfg(feature = "stats-obfuscation")] Some(client_side_stats.obfuscation_config.clone()), ))); diff --git a/libdd-trace-utils/src/otlp_encoder/mod.rs b/libdd-trace-utils/src/otlp_encoder/mod.rs index 753f27adc6..70ef928a1f 100644 --- a/libdd-trace-utils/src/otlp_encoder/mod.rs +++ b/libdd-trace-utils/src/otlp_encoder/mod.rs @@ -38,6 +38,7 @@ pub struct OtlpResourceInfo { pub runtime_id: String, pub hostname: String, pub process_tags: String, + pub tracer_tags: Vec, pub instrumentation_scope_name: String, pub instrumentation_scope_version: String, /// When true, emits `_dd.stats_computed: "true"` on the OTLP resource to prevent From 5961433109564dca68309d5bd3ab4e4e8d343c7a Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Mon, 10 Aug 2026 16:53:23 -0400 Subject: [PATCH 14/15] fix(data-pipeline): preserve OTLP metric attribute contracts --- libdd-data-pipeline/src/otlp/metrics.rs | 13 +++++++++---- libdd-data-pipeline/src/trace_exporter/builder.rs | 8 +++----- 2 files changed, 12 insertions(+), 9 deletions(-) diff --git a/libdd-data-pipeline/src/otlp/metrics.rs b/libdd-data-pipeline/src/otlp/metrics.rs index 1b0cb7dc52..c6c1e92850 100644 --- a/libdd-data-pipeline/src/otlp/metrics.rs +++ b/libdd-data-pipeline/src/otlp/metrics.rs @@ -241,7 +241,9 @@ fn build_attributes( } for tag in &group.additional_metric_tags { if let Some((k, v)) = tag.split_once(':') { - push(&mut attrs, k, v); + if attrs.iter().all(|attr| attr["key"].as_str() != Some(k)) { + attrs.push(kv_str(k, v)); + } } } attrs @@ -700,6 +702,7 @@ mod tests { // Only the first `:` is a delimiter; the value keeps any embedded `:`. "endpoint:https://host:8080".into(), "datadog.custom:dd".into(), + "status.code:custom".into(), ]; }); let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource()).unwrap(); @@ -708,15 +711,17 @@ mod tests { assert_eq!(str_at(a, "region"), Some("us-east")); assert_eq!(str_at(a, "endpoint"), Some("https://host:8080")); assert_eq!(str_at(a, "datadog.custom"), Some("dd")); + assert_eq!(str_at(a, "status.code"), Some(STATUS_CODE_OK)); + assert_eq!(a.iter().filter(|kv| kv["key"] == "status.code").count(), 1); - // Malformed (no `:`) or empty-value entries are skipped rather than emitted verbatim. + // Preserve the cardinality-overflow marker even though its value is empty. let g = group_with_exact(&[1_000_000_000], &[], |g| { - g.additional_metric_tags = vec!["malformed".into(), "empty:".into()]; + g.additional_metric_tags = vec!["malformed".into(), "tracer_blocked_value:".into()]; }); let req = map_stats_to_otlp_metrics(&buckets(vec![g]), &resource()).unwrap(); let a = points(&req)[0]["attributes"].as_array().unwrap(); assert!(!a.iter().any(|kv| kv["key"] == "malformed")); - assert!(!a.iter().any(|kv| kv["key"] == "empty")); + assert_eq!(str_at(a, "tracer_blocked_value"), Some("")); } #[test] diff --git a/libdd-data-pipeline/src/trace_exporter/builder.rs b/libdd-data-pipeline/src/trace_exporter/builder.rs index 79f057ca4d..e190308461 100644 --- a/libdd-data-pipeline/src/trace_exporter/builder.rs +++ b/libdd-data-pipeline/src/trace_exporter/builder.rs @@ -516,13 +516,11 @@ impl TraceExporterBuilder { self } - /// Enables OTel trace semantics, which does not add DD-specific per-span attributes + /// Enables OTel trace semantics for trace export, which does not add DD-specific per-span + /// attributes /// (`service.name`, `operation.name`, `resource.name`, `span.type`, `error.msg`, /// `error.message`, `span.kind`) to the OTLP payload. - /// Also strips Datadog-specific `dd.*`/`_dd.*` data-point attributes from the exported - /// histogram. This is useful when exporting to a native OTel backend that does not expect - /// Datadog semantics. The host language tracer is expected to observe this behavior by - /// setting the `DD_TRACE_OTEL_SEMANTICS_ENABLED` environment variable to `true`. + /// OTLP trace metrics are unaffected and always include available Datadog attributes. pub fn enable_otel_trace_semantics(&mut self) -> &mut Self { self.otel_trace_semantics_enabled = true; self From 9890ff2c5f5ab0209afc59bc48f8c9e99448913e Mon Sep 17 00:00:00 2001 From: Munir Abdinur Date: Wed, 12 Aug 2026 10:08:13 -0400 Subject: [PATCH 15/15] refactor(data-pipeline): clarify optional OTLP attributes --- libdd-data-pipeline/src/otlp/metrics.rs | 27 +++++++++++++------------ 1 file changed, 14 insertions(+), 13 deletions(-) diff --git a/libdd-data-pipeline/src/otlp/metrics.rs b/libdd-data-pipeline/src/otlp/metrics.rs index c6c1e92850..7f56c0cb35 100644 --- a/libdd-data-pipeline/src/otlp/metrics.rs +++ b/libdd-data-pipeline/src/otlp/metrics.rs @@ -73,6 +73,13 @@ const EXPLICIT_BOUNDS_SECONDS: [f64; 16] = [ fn kv_str(key: &str, value: &str) -> Value { json!({ "key": key, "value": { "stringValue": value } }) } + +fn push_if_non_empty(attrs: &mut Vec, key: &str, value: &str) { + if !value.is_empty() { + attrs.push(kv_str(key, value)); + } +} + fn kv_int(key: &str, value: i64) -> Value { json!({ "key": key, "value": { "intValue": value.to_string() } }) } @@ -173,12 +180,6 @@ fn build_attributes( is_error: bool, resource_info: &OtlpResourceInfo, ) -> Vec { - fn push(attrs: &mut Vec, k: &str, v: &str) { - if !v.is_empty() { - attrs.push(kv_str(k, v)); - } - } - let mut attrs = Vec::new(); let group_service = if group.service.is_empty() { @@ -196,14 +197,14 @@ fn build_attributes( STATUS_CODE_OK }, )); - push(&mut attrs, "span.kind", span_kind_name(&group.span_kind)); - push(&mut attrs, "span.name", &group.resource); + push_if_non_empty(&mut attrs, "span.kind", span_kind_name(&group.span_kind)); + push_if_non_empty(&mut attrs, "span.name", &group.resource); - push(&mut attrs, "http.request.method", &group.http_method); - push(&mut attrs, "http.route", &group.http_endpoint); + push_if_non_empty(&mut attrs, "http.request.method", &group.http_method); + push_if_non_empty(&mut attrs, "http.route", &group.http_endpoint); // group.grpc_status_code is the numeric code as a string; emit the canonical OTel status name. if let Some(name) = grpc_status_code_to_name(&group.grpc_status_code) { - push(&mut attrs, "rpc.response.status_code", name); + push_if_non_empty(&mut attrs, "rpc.response.status_code", name); } if group.http_status_code != 0 { attrs.push(kv_int( @@ -212,8 +213,8 @@ fn build_attributes( )); } attrs.push(kv_str("datadog.operation.name", &group.name)); - push(&mut attrs, "datadog.span.type", &group.r#type); - push(&mut attrs, "datadog.svc_src", &group.service_source); + push_if_non_empty(&mut attrs, "datadog.span.type", &group.r#type); + push_if_non_empty(&mut attrs, "datadog.svc_src", &group.service_source); // Only `synthetics` is surfaced as `datadog.origin`: the aggregation key carries just a // boolean, not the full origin string, so other origins are lost upstream. if group.synthetics {