diff --git a/libdd-data-pipeline/src/otlp/config.rs b/libdd-data-pipeline/src/otlp/config.rs index 349be5791e..e8edcdf610 100644 --- a/libdd-data-pipeline/src/otlp/config.rs +++ b/libdd-data-pipeline/src/otlp/config.rs @@ -134,7 +134,7 @@ 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`). + /// 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 3d95c29306..7f56c0cb35 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::status_code; use libdd_trace_utils::otlp_encoder::OtlpResourceInfo; use serde_json::{json, Value}; use std::sync::{Arc, Mutex}; @@ -49,6 +48,23 @@ 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() } + +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. 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, @@ -57,9 +73,23 @@ 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() } }) } +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. /// @@ -69,7 +99,6 @@ fn kv_int(key: &str, value: i64) -> Value { 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 { @@ -86,7 +115,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(), @@ -104,7 +133,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, @@ -150,46 +179,32 @@ fn build_attributes( group: &pb::ClientGroupedStats, is_error: bool, resource_info: &OtlpResourceInfo, - otel_trace_semantics_enabled: bool, ) -> Vec { let mut attrs = Vec::new(); - let mut push = |k: &str, v: &str| { - if !v.is_empty() { - attrs.push(kv_str(k, v)); - } - }; - // 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("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); + attrs.push(kv_str("service.name", group_service)); + + attrs.push(kv_str( + "status.code", + if is_error { + STATUS_CODE_ERROR + } else { + STATUS_CODE_OK + }, + )); + 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_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("rpc.response.status_code", name); - } - for tag in &group.peer_tags { - if let Some((k, v)) = tag.split_once(':') { - push(k, v); - } - } - for tag in &group.additional_metric_tags { - if let Some((k, v)) = tag.split_once(':') { - push(k, v); - } - } - if !otel_trace_semantics_enabled { - push("datadog.operation.name", &group.name); - push("datadog.span.type", &group.r#type); + push_if_non_empty(&mut attrs, "rpc.response.status_code", name); } if group.http_status_code != 0 { attrs.push(kv_int( @@ -197,30 +212,45 @@ fn build_attributes( 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")); - } - if group.is_trace_root == pb::Trilean::True as i32 { - attrs.push(json!({ "key": "_datadog.is_trace_root", "value": { "boolValue": true } })); - } - let top_level = group.hits > 0 && group.top_level_hits == group.hits; + attrs.push(kv_str("datadog.operation.name", &group.name)); + 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 { + 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.span.top_level", "value": { "boolValue": top_level } + "key": "datadog.is_trace_root", "value": { "boolValue": is_trace_root } })); } - if is_error { - attrs.push(kv_int("status.code", status_code::ERROR as i64)); + let top_level = group.hits > 0 && group.top_level_hits == group.hits; + attrs.push(json!({ + "key": "datadog.span.top_level", "value": { "boolValue": top_level } + })); + if !group.peer_tags.is_empty() { + attrs.push(kv_str_array( + "datadog.peer_tags", + group.peer_tags.iter().map(String::as_str), + )); + } + for tag in &group.additional_metric_tags { + if let Some((k, v)) = tag.split_once(':') { + if attrs.iter().all(|attr| attr["key"].as_str() != Some(k)) { + attrs.push(kv_str(k, v)); + } + } } 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()), @@ -235,15 +265,24 @@ 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)); - } - 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)) - })); + if !info.runtime_id.is_empty() { + attrs.push(kv_str("datadog.runtime_id", &info.runtime_id)); + } + // Keep this representation aligned with v0.6 stats export. + 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.tracer_tags.is_empty() { + attrs.push(kv_str_array( + "datadog.tracer_tags", + info.tracer_tags.iter().map(String::as_str), + )); } attrs } @@ -269,11 +308,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( @@ -430,87 +465,155 @@ 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() + fn bool_at(attrs: &[Value], key: &str) -> Option { + attrs .iter() - .any(|kv| kv["key"] == "status.code" && kv["value"]["intValue"].as_str() == Some(&ec)) + .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) } #[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_at(a, "datadog.entrypoint.name").is_some(), dd); - } + 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"); + 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"]) + ); + assert_eq!( + str_array_at(a, "datadog.tracer_tags"), + Some(vec!["team:apm", "tier:backend"]) + ); } #[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(); 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.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()]; }); 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() - .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(); + 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")); 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.svc_src"), Some("lambda")); assert_eq!(str_at(a, "datadog.origin"), Some("synthetics")); - assert!(a.iter().any(|kv| kv["key"] == "datadog.span.top_level")); - assert!(pts.iter().any( - |p| str_at(p["attributes"].as_array().unwrap(), "service.name") == Some("svc-other") - )); + assert_eq!(bool_at(a, "datadog.is_trace_root"), Some(true)); + 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"]) + ); + assert_eq!(str_at(a, "status.code"), Some(STATUS_CODE_OK)); + 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")); + } - // OTel mode strips datadog.*/_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(); - 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")); + #[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 [ + (pb::Trilean::NotSet as i32, None), + (pb::Trilean::True as i32, Some(true)), + (pb::Trilean::False as i32, Some(false)), + ] { + let g = group_with_exact(&[1_000_000_000], &[], |g| { + g.is_trace_root = value; + g.top_level_hits = 0; + }); + 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!( + 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)); + } } #[test] @@ -518,7 +621,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")); @@ -527,7 +630,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")); } @@ -536,8 +639,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); @@ -553,7 +655,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)); @@ -570,7 +672,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(); @@ -600,28 +702,27 @@ 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(), + "status.code:custom".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")); + assert_eq!(str_at(a, "status.code"), Some(STATUS_CODE_OK)); + assert_eq!(a.iter().filter(|kv| kv["key"] == "status.code").count(), 1); - // Additional metric tags are user/tracer-defined (not Datadog-internal), so unlike - // `datadog.*` attributes they still pass through in OTel-semantics mode. - 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")); - - // 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(), 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")); + 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 4a01daf78a..e190308461 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 @@ -502,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 @@ -783,7 +795,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 +808,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 +873,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