From 413e8a257f737f7bc1d6c57ea0c8712277311211 Mon Sep 17 00:00:00 2001 From: Eric Evans <194135482+ericevans-nv@users.noreply.github.com> Date: Tue, 18 Aug 2026 15:09:07 -0500 Subject: [PATCH 1/5] feat(observability): promote selected metadata to OTel attributes Signed-off-by: Eric Evans <194135482+ericevans-nv@users.noreply.github.com> --- crates/core/src/observability/mod.rs | 187 +++++++++++++++ crates/core/src/observability/otel.rs | 123 +++++++++- .../src/observability/plugin_component.rs | 29 ++- .../attribute_projection_tests.rs | 187 ++++++++++++++- .../unit/observability/openinference_tests.rs | 1 + .../tests/unit/observability/otel_tests.rs | 220 ++++++++++++++++++ .../observability/plugin_component_tests.rs | 31 +++ crates/ffi/nemo_relay.h | 25 ++ crates/ffi/src/api/observability.rs | 77 +++++- crates/ffi/tests/unit/api/plugin_tests.rs | 61 +++++ crates/node/observability.d.ts | 1 + crates/node/observability.js | 1 + crates/node/src/api/mod.rs | 5 +- .../node/tests/observability_plugin_tests.mjs | 1 + crates/node/tests/otel_tests.mjs | 22 +- crates/python/src/py_types/observability.rs | 10 +- go/nemo_relay/nemo_relay.go | 63 +++-- go/nemo_relay/observability_plugin.go | 58 ++--- go/nemo_relay/observability_plugin_test.go | 5 + go/nemo_relay/otel_test.go | 26 ++- python/nemo_relay/_native.pyi | 1 + python/nemo_relay/observability.py | 2 + python/nemo_relay/observability.pyi | 1 + python/tests/test_observability_plugin.py | 7 + python/tests/test_types.py | 19 +- 25 files changed, 1094 insertions(+), 69 deletions(-) diff --git a/crates/core/src/observability/mod.rs b/crates/core/src/observability/mod.rs index 7c479c3d4..462bd5908 100644 --- a/crates/core/src/observability/mod.rs +++ b/crates/core/src/observability/mod.rs @@ -336,6 +336,49 @@ pub fn validate_attribute_mappings( Ok(()) } +/// Validates literal Event metadata prefixes promoted to OTLP attributes. +pub fn validate_metadata_promotion_prefixes( + prefixes: &[String], +) -> std::result::Result<(), String> { + let mut unique = std::collections::HashSet::new(); + for prefix in prefixes { + if is_blank_attribute_mapping_name(prefix) { + return Err("metadata promotion prefix must not be blank".to_string()); + } + if prefix.trim() != prefix { + return Err(format!( + "metadata promotion prefix {prefix:?} must not have surrounding whitespace" + )); + } + if prefix.contains(['*', '?', '[', ']']) { + return Err(format!( + "metadata promotion prefix {prefix:?} must be a literal prefix, not a glob" + )); + } + if !is_valid_metadata_promotion_prefix(prefix) { + return Err(format!( + "metadata promotion prefix {prefix:?} must contain letter, number, underscore, or hyphen segments separated by single dots, with an optional trailing dot" + )); + } + if !unique.insert(prefix.as_str()) { + return Err(format!( + "metadata promotion prefix {prefix:?} is duplicated" + )); + } + } + Ok(()) +} + +fn is_valid_metadata_promotion_prefix(prefix: &str) -> bool { + let key = prefix.strip_suffix('.').unwrap_or(prefix); + key.split('.').all(|segment| { + !segment.is_empty() + && segment.chars().all(|character| { + character.is_ascii_alphanumeric() || matches!(character, '_' | '-') + }) + }) +} + fn is_blank_attribute_mapping_name(value: &str) -> bool { value.chars().all(|character| { character.is_whitespace() @@ -485,6 +528,150 @@ fn push_top_level_json_value( } } +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct MetadataPromotionIssue { + pub(crate) key: String, + pub(crate) reason: &'static str, +} + +/// Copies selected top-level Event metadata entries to typed OTLP attributes. +/// +/// Existing projection-owned attributes always win. Metadata is read without +/// modification, and unsupported values are returned to the caller for +/// bounded runtime diagnostics. +pub(crate) fn promote_event_metadata_attributes( + attributes: &mut Vec, + event: &crate::api::event::Event, + prefixes: &[String], + protected_keys: &std::collections::HashSet, +) -> Vec { + if prefixes.is_empty() { + return Vec::new(); + } + let Some(metadata) = event.metadata().and_then(crate::json::Json::as_object) else { + return Vec::new(); + }; + let mut existing_keys = attributes + .iter() + .map(|attribute| attribute.key.as_str().to_string()) + .chain(protected_keys.iter().cloned()) + .collect::>(); + let mut issues = Vec::new(); + for (key, value) in metadata { + if !prefixes.iter().any(|prefix| key.starts_with(prefix)) || existing_keys.contains(key) { + continue; + } + match metadata_value_to_otel(value) { + Ok(value) => { + attributes.push(opentelemetry::KeyValue::new(key.clone(), value)); + existing_keys.insert(key.clone()); + } + Err(reason) => issues.push(MetadataPromotionIssue { + key: key.clone(), + reason, + }), + } + } + issues +} + +fn metadata_value_to_otel( + value: &crate::json::Json, +) -> std::result::Result { + use opentelemetry::Value; + + match value { + crate::json::Json::String(value) => Ok(Value::String(value.clone().into())), + crate::json::Json::Bool(value) => Ok(Value::Bool(*value)), + crate::json::Json::Number(value) => metadata_number_to_otel(value), + crate::json::Json::Array(values) => metadata_array_to_otel(values), + crate::json::Json::Null => Err("null values are not OTLP attributes"), + crate::json::Json::Object(_) => Err("object values are not supported"), + } +} + +fn metadata_number_to_otel( + value: &serde_json::Number, +) -> std::result::Result { + use opentelemetry::Value; + + if let Some(value) = value.as_i64() { + return Ok(Value::I64(value)); + } + if let Some(value) = value.as_u64() { + return i64::try_from(value) + .map(Value::I64) + .map_err(|_| "unsigned integer is larger than OTLP i64"); + } + value + .as_f64() + .map(Value::F64) + .ok_or("number is not representable as an OTLP attribute") +} + +fn metadata_array_to_otel( + values: &[crate::json::Json], +) -> std::result::Result { + use opentelemetry::{Array, Value}; + + let Some(first) = values.first() else { + return Err("empty arrays do not declare an OTLP element type"); + }; + match first { + crate::json::Json::String(_) => values + .iter() + .map(|value| value.as_str().map(|value| value.to_string().into())) + .collect::>>() + .map(|values| Value::Array(Array::String(values))) + .ok_or("array values must have one primitive type"), + crate::json::Json::Bool(_) => values + .iter() + .map(crate::json::Json::as_bool) + .collect::>>() + .map(|values| Value::Array(Array::Bool(values))) + .ok_or("array values must have one primitive type"), + crate::json::Json::Number(_) => metadata_number_array_to_otel(values), + crate::json::Json::Null => Err("arrays of null are not OTLP attributes"), + crate::json::Json::Array(_) | crate::json::Json::Object(_) => { + Err("nested arrays and objects are not supported") + } + } +} + +fn metadata_number_array_to_otel( + values: &[crate::json::Json], +) -> std::result::Result { + use opentelemetry::{Array, Value}; + + let numbers = values + .iter() + .map(crate::json::Json::as_number) + .collect::>>() + .ok_or("array values must have one primitive type")?; + let integers = numbers + .iter() + .map(|value| { + value + .as_i64() + .or_else(|| value.as_u64().and_then(|value| i64::try_from(value).ok())) + }) + .collect::>>(); + if let Some(values) = integers { + return Ok(Value::Array(Array::I64(values))); + } + if numbers + .iter() + .any(|value| value.as_u64().is_some_and(|value| value > i64::MAX as u64)) + { + return Err("array contains an unsigned integer larger than OTLP i64"); + } + numbers + .iter() + .map(|value| value.as_f64()) + .collect::>>() + .map(|values| Value::Array(Array::F64(values))) + .ok_or("number array is not representable as an OTLP attribute") +} pub(crate) fn apply_attribute_mappings( attributes: &mut Vec, mappings: &[OtlpAttributeMapping], diff --git a/crates/core/src/observability/otel.rs b/crates/core/src/observability/otel.rs index 687451475..0c2d08c11 100644 --- a/crates/core/src/observability/otel.rs +++ b/crates/core/src/observability/otel.rs @@ -33,9 +33,10 @@ use super::{ apply_attribute_mappings, attribute_mapping_aliases, attribute_mapping_inputs, default_mark_exclude_names, effective_mark_projection, estimate_cost_for_response_or_model, estimate_cost_for_response_or_requested_model, manual, model_name_for_llm_event, - push_serialized_top_level_attributes, push_session_identity_attributes, - push_tool_result_annotation_attribute, push_top_level_json_attributes, relay_span_id, - relay_trace_id, validate_attribute_mappings, + promote_event_metadata_attributes, push_serialized_top_level_attributes, + push_session_identity_attributes, push_tool_result_annotation_attribute, + push_top_level_json_attributes, relay_span_id, relay_trace_id, validate_attribute_mappings, + validate_metadata_promotion_prefixes, }; use crate::api::event::{Event, EventNormalizationExt, ScopeCategory}; use crate::api::runtime::{EventSubscriberFn, current_scope_stack}; @@ -148,6 +149,9 @@ pub enum OpenTelemetryError { /// Attribute mapping configuration was invalid. #[error("invalid attribute mappings: {0}")] InvalidAttributeMappings(String), + /// Metadata promotion prefix configuration was invalid. + #[error("invalid metadata promotion prefixes: {0}")] + InvalidMetadataPromotionPrefixes(String), /// Registration errors from the core runtime. #[error(transparent)] Core(#[from] FlowError), @@ -200,6 +204,7 @@ pub struct OpenTelemetryConfig { mark_projection: MarkProjection, mark_exclude_names: Vec, attribute_mappings: Vec, + promote_metadata_prefixes: Vec, timeout: Duration, transport: OtlpTransport, max_queue_size: Option, @@ -221,6 +226,7 @@ impl OpenTelemetryConfig { mark_projection: MarkProjection::default(), mark_exclude_names: default_mark_exclude_names(), attribute_mappings: Vec::new(), + promote_metadata_prefixes: Vec::new(), timeout: Duration::from_secs(3), transport: OtlpTransport::HttpBinary, max_queue_size: None, @@ -383,6 +389,16 @@ impl OpenTelemetryConfig { self.attribute_mappings = mappings.into_iter().collect(); self } + + /// Selects literal Event metadata prefixes copied to OTLP attributes. + pub fn with_promote_metadata_prefixes(mut self, prefixes: I) -> Self + where + I: IntoIterator, + S: Into, + { + self.promote_metadata_prefixes = prefixes.into_iter().map(Into::into).collect(); + self + } } #[cfg(test)] @@ -407,6 +423,8 @@ pub struct OpenTelemetrySubscriberOptions { pub mark_exclude_names: Vec, /// Typed OTLP attributes copied to alias keys. pub attribute_mappings: Vec, + /// Literal Event metadata prefixes copied to OTLP attributes. + pub promote_metadata_prefixes: Vec, } impl Default for OpenTelemetrySubscriberOptions { @@ -415,6 +433,7 @@ impl Default for OpenTelemetrySubscriberOptions { mark_projection: MarkProjection::default(), mark_exclude_names: default_mark_exclude_names(), attribute_mappings: Vec::new(), + promote_metadata_prefixes: Vec::new(), } } } @@ -471,6 +490,8 @@ impl OpenTelemetrySubscriber { } validate_attribute_mappings(&config.attribute_mappings) .map_err(OpenTelemetryError::InvalidAttributeMappings)?; + validate_metadata_promotion_prefixes(&config.promote_metadata_prefixes) + .map_err(OpenTelemetryError::InvalidMetadataPromotionPrefixes)?; reject_global_header_environment()?; validate_headers(&config.headers)?; let runtime_diagnostics = SignalRuntimeDiagnostics::new(diagnostic_field); @@ -483,6 +504,7 @@ impl OpenTelemetrySubscriber { config.mark_projection, config.mark_exclude_names, config.attribute_mappings, + config.promote_metadata_prefixes, Some(runtime), )) } @@ -513,6 +535,7 @@ impl OpenTelemetrySubscriber { MarkProjection::default(), default_mark_exclude_names(), Vec::new(), + Vec::new(), None, ) } @@ -545,6 +568,8 @@ impl OpenTelemetrySubscriber { ) -> Result { validate_attribute_mappings(&options.attribute_mappings) .map_err(OpenTelemetryError::InvalidAttributeMappings)?; + validate_metadata_promotion_prefixes(&options.promote_metadata_prefixes) + .map_err(OpenTelemetryError::InvalidMetadataPromotionPrefixes)?; Ok(Self::from_tracer_provider_with_scope_and_type( provider, instrumentation_scope.into(), @@ -552,6 +577,7 @@ impl OpenTelemetrySubscriber { options.mark_projection, options.mark_exclude_names, options.attribute_mappings, + options.promote_metadata_prefixes, None, )) } @@ -565,6 +591,8 @@ impl OpenTelemetrySubscriber { ) -> Result { validate_attribute_mappings(&options.attribute_mappings) .map_err(OpenTelemetryError::InvalidAttributeMappings)?; + validate_metadata_promotion_prefixes(&options.promote_metadata_prefixes) + .map_err(OpenTelemetryError::InvalidMetadataPromotionPrefixes)?; Ok(Self::from_tracer_provider_with_scope_and_type( provider, instrumentation_scope.into(), @@ -572,10 +600,12 @@ impl OpenTelemetrySubscriber { options.mark_projection, options.mark_exclude_names, options.attribute_mappings, + options.promote_metadata_prefixes, None, )) } + #[allow(clippy::too_many_arguments)] fn from_tracer_provider_with_scope_and_type( provider: SdkTracerProvider, instrumentation_scope: String, @@ -583,6 +613,7 @@ impl OpenTelemetrySubscriber { mark_projection: MarkProjection, mark_exclude_names: Vec, attribute_mappings: Vec, + promote_metadata_prefixes: Vec, runtime: Option, ) -> Self { let runtime_diagnostics = runtime @@ -597,6 +628,7 @@ impl OpenTelemetrySubscriber { mark_projection, mark_exclude_names, attribute_mappings, + promote_metadata_prefixes, runtime_diagnostics.clone(), ), )); @@ -996,6 +1028,8 @@ pub(super) struct ActiveSpan { span_context: SpanContext, start_model_name: Option, projected_attributes: Vec, + projection_attribute_keys: HashSet, + start_promoted_metadata: Vec, descendant_error_type: Option, descendant_exception_type: Option, } @@ -1010,6 +1044,7 @@ pub(super) struct OtelEventProcessor { mark_projection: MarkProjection, mark_exclude_names: Vec, attribute_mappings: Vec, + promote_metadata_prefixes: Vec, invalid_metric_count: u64, runtime_diagnostics: SignalRuntimeDiagnostics, } @@ -1115,10 +1150,12 @@ impl OtelEventProcessor { mark_projection, mark_exclude_names, attribute_mappings, + Vec::new(), SignalRuntimeDiagnostics::new(None), ) } + #[allow(clippy::too_many_arguments)] fn new_with_mark_projection_and_exclusions_and_mappings_and_runtime_diagnostics( provider: SdkTracerProvider, instrumentation_scope: String, @@ -1126,6 +1163,7 @@ impl OtelEventProcessor { mark_projection: MarkProjection, mark_exclude_names: Vec, attribute_mappings: Vec, + promote_metadata_prefixes: Vec, runtime_diagnostics: SignalRuntimeDiagnostics, ) -> Self { let tracer = provider.tracer(instrumentation_scope); @@ -1139,6 +1177,7 @@ impl OtelEventProcessor { mark_projection, mark_exclude_names, attribute_mappings, + promote_metadata_prefixes, invalid_metric_count: 0, runtime_diagnostics, } @@ -1200,11 +1239,30 @@ impl OtelEventProcessor { if self.otel_type != OpenTelemetryType::GenAi && is_trace_root { push_session_identity_attributes(&mut attributes, event); } + // Snapshot keys claimed by the projection so promoted metadata cannot + // replace them when the span completes. + let mut projection_attribute_keys = attributes + .iter() + .map(|attribute| attribute.key.as_str().to_string()) + .collect::>(); + if self.otel_type != OpenTelemetryType::GenAi { + projection_attribute_keys.extend( + self.attribute_mappings + .iter() + .map(|mapping| mapping.alias.clone()), + ); + } let projected_attributes = if self.otel_type == OpenTelemetryType::GenAi { Vec::new() } else { attribute_mapping_inputs(&attributes, &self.attribute_mappings) }; + let mut start_promoted_metadata = Vec::new(); + self.promote_metadata( + &mut start_promoted_metadata, + event, + &projection_attribute_keys, + ); span.set_attributes(attributes); let span_context = local_parent_span_context(span.span_context()); self.active_spans.insert( @@ -1214,6 +1272,8 @@ impl OtelEventProcessor { span_context, start_model_name, projected_attributes, + projection_attribute_keys, + start_promoted_metadata, descendant_error_type: None, descendant_exception_type: None, }, @@ -1282,6 +1342,25 @@ impl OtelEventProcessor { &self.attribute_mappings, )); } + // Preserve every projection-owned start/end key while promoting the + // final metadata carried by the scope-end Event. + active_span.projection_attribute_keys.extend( + attributes + .iter() + .map(|attribute| attribute.key.as_str().to_string()), + ); + let end_metadata = event.metadata().and_then(crate::json::Json::as_object); + active_span.start_promoted_metadata.retain(|attribute| { + let key = attribute.key.as_str(); + !active_span.projection_attribute_keys.contains(key) + && !end_metadata.is_some_and(|metadata| metadata.contains_key(key)) + }); + attributes.extend(active_span.start_promoted_metadata); + self.promote_metadata( + &mut attributes, + event, + &active_span.projection_attribute_keys, + ); if is_error && let Some(parent_span) = self.find_parent_span_mut(event) { if parent_span.descendant_error_type.is_none() { parent_span.descendant_error_type = error_type; @@ -1339,6 +1418,7 @@ impl OtelEventProcessor { if self.find_parent_span(event).is_some() { apply_attribute_mappings(&mut attributes, &self.attribute_mappings); + self.promote_metadata(&mut attributes, event, &HashSet::new()); let parent_span = self .find_parent_span_mut(event) .expect("parent span was present during mark projection"); @@ -1361,6 +1441,7 @@ impl OtelEventProcessor { attributes.push(KeyValue::new("nemo_relay.mark.orphan", true)); } apply_attribute_mappings(&mut attributes, &self.attribute_mappings); + self.promote_metadata(&mut attributes, event, &HashSet::new()); span.set_attributes(attributes); span.end_with_timestamp(timestamp); } @@ -1382,6 +1463,7 @@ impl OtelEventProcessor { attributes.push(KeyValue::new("nemo_relay.mark.orphan", true)); } apply_attribute_mappings(&mut attributes, &self.attribute_mappings); + self.promote_metadata(&mut attributes, event, &HashSet::new()); let mut span = with_relay_ids(event.uuid(), || { self.tracer @@ -1394,6 +1476,41 @@ impl OtelEventProcessor { span.end_with_timestamp(timestamp); } + // Report unsupported metadata by key without exposing its value or + // interrupting Event export. + fn promote_metadata( + &self, + attributes: &mut Vec, + event: &Event, + protected_keys: &HashSet, + ) { + let issues = promote_event_metadata_attributes( + attributes, + event, + &self.promote_metadata_prefixes, + protected_keys, + ); + + for issue in issues { + let diagnostic_count = self.runtime_diagnostics.record( + "otel.metadata_promotion_value_unsupported", + format!( + "OpenTelemetry metadata attribute {:?} was not promoted: {}", + issue.key, issue.reason + ), + 1, + ); + if should_relog_runtime_diagnostic(diagnostic_count) { + log::warn!( + target: "nemo_relay.observability", + event = "otel_metadata_promotion_value_unsupported", + metadata_key = issue.key.as_str(); + "OpenTelemetry metadata attribute was not promoted: {}", + issue.reason + ); + } + } + } fn mark_attributes(&self, event: &Event) -> Vec { match self.otel_type { OpenTelemetryType::Full => mark_attributes(event), diff --git a/crates/core/src/observability/plugin_component.rs b/crates/core/src/observability/plugin_component.rs index 41f14ac7a..f1496919c 100644 --- a/crates/core/src/observability/plugin_component.rs +++ b/crates/core/src/observability/plugin_component.rs @@ -70,7 +70,7 @@ use crate::observability::otel_signal::{ }; use crate::observability::{ MarkProjection, OpenTelemetryType, OtlpAttributeMapping, default_mark_exclude_names, - validate_attribute_mappings, + validate_attribute_mappings, validate_metadata_promotion_prefixes, }; use crate::plugin::{ ATIF_RUNTIME_DELIVERY_FAILURE_MARKER, ConfigDiagnostic, ConfigPolicy, DiagnosticLevel, @@ -323,6 +323,9 @@ pub struct OpenTelemetryEndpointConfig { /// Projected attributes copied to aliases. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub attribute_mappings: Vec, + /// Literal Event metadata prefixes copied to top-level OTLP attributes. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub promote_metadata_prefixes: Vec, /// OTLP transport: `http_binary` or `grpc`. #[serde(default = "default_otlp_transport")] #[cfg_attr(feature = "schema", schemars(schema_with = "otlp_transport_schema"))] @@ -731,6 +734,12 @@ impl EditorConfig for OpenTelemetryEndpointConfig { ), otel_editor_field("mark_exclude_names", EditorFieldKind::Json, &[], false), otel_editor_field("attribute_mappings", EditorFieldKind::List, &[], false), + otel_editor_field( + "promote_metadata_prefixes", + EditorFieldKind::List, + &[], + true, + ), otel_editor_field( "transport", EditorFieldKind::Enum, @@ -2954,7 +2963,8 @@ fn build_otel_config( .with_instrumentation_scope(section.instrumentation_scope) .with_mark_projection(section.mark_projection) .with_mark_exclude_names(section.mark_exclude_names) - .with_attribute_mappings(section.attribute_mappings); + .with_attribute_mappings(section.attribute_mappings) + .with_promote_metadata_prefixes(section.promote_metadata_prefixes); if let Some(max_queue_size) = section.max_queue_size { config = config.with_max_queue_size(max_queue_size); } @@ -3189,6 +3199,7 @@ fn validate_observability_section_fields( "mark_projection", "mark_exclude_names", "attribute_mappings", + "promote_metadata_prefixes", "transport", "endpoint", "headers", @@ -3208,6 +3219,7 @@ fn validate_observability_section_fields( "mark_projection", "mark_exclude_names", "attribute_mappings", + "promote_metadata_prefixes", "transport", "endpoint", "headers", @@ -3319,6 +3331,7 @@ fn validate_opentelemetry_endpoint_fields( "mark_projection", "mark_exclude_names", "attribute_mappings", + "promote_metadata_prefixes", "transport", "headers", "header_env", @@ -3539,6 +3552,18 @@ fn validate_opentelemetry_section( error, ); } + if let Err(error) = + validate_metadata_promotion_prefixes(&endpoint.promote_metadata_prefixes) + { + push_policy_diag( + diagnostics, + policy.unsupported_value, + "observability.unsupported_value", + Some("opentelemetry".to_string()), + Some(format!("endpoints[{index}].promote_metadata_prefixes")), + error, + ); + } validate_opentelemetry_batch_config(diagnostics, policy, index, endpoint); validate_opentelemetry_headers(diagnostics, policy, index, endpoint); } diff --git a/crates/core/tests/unit/observability/attribute_projection_tests.rs b/crates/core/tests/unit/observability/attribute_projection_tests.rs index 1c084826e..9564490fd 100644 --- a/crates/core/tests/unit/observability/attribute_projection_tests.rs +++ b/crates/core/tests/unit/observability/attribute_projection_tests.rs @@ -5,8 +5,11 @@ use super::{ OtlpAttributeMapping, apply_attribute_mappings, attribute_mapping_inputs, - push_top_level_json_attributes, + promote_event_metadata_attributes, push_top_level_json_attributes, + validate_metadata_promotion_prefixes, }; +use crate::api::event::{BaseEvent, Event, MarkEvent}; +use std::collections::HashSet; #[test] fn retains_only_mapping_sources_and_existing_aliases_between_span_events() { @@ -142,3 +145,185 @@ fn rejects_invalid_attribute_mappings() { ); assert!(super::validate_attribute_mappings(&[OtlpAttributeMapping::new("key", ".")]).is_ok()); } + +#[test] +fn promotes_matching_primitive_metadata_without_overwriting_owned_keys() { + let event = Event::Mark(MarkEvent::new( + BaseEvent::builder() + .name("metadata-promotion") + .metadata(serde_json::json!({ + "nv.string": "value", + "nv.bool": true, + "nv.integer": 2, + "nv.strings": ["a", "b"], + "nv.bools": [true, false], + "nv.integers": [2, 3], + "nv.floats": [1.5, 2.5], + "nv.nested": {"unsupported": true}, + "nv.owned": "attempted-overwrite", + "other.unmatched": "ignored" + })) + .build(), + None, + None, + )); + let mut unpromoted_attributes = Vec::new(); + let unpromoted_issues = + promote_event_metadata_attributes(&mut unpromoted_attributes, &event, &[], &HashSet::new()); + assert!(unpromoted_attributes.is_empty()); + assert!(unpromoted_issues.is_empty()); + + let mut attributes = vec![opentelemetry::KeyValue::new("nv.owned", "projection")]; + + let issues = promote_event_metadata_attributes( + &mut attributes, + &event, + &["nv.".to_string()], + &HashSet::new(), + ); + + let value = |key| { + attributes + .iter() + .find(|attribute| attribute.key.as_str() == key) + .map(|attribute| &attribute.value) + }; + assert_eq!( + value("nv.string"), + Some(&opentelemetry::Value::String("value".into())) + ); + assert_eq!(value("nv.bool"), Some(&opentelemetry::Value::Bool(true))); + assert_eq!(value("nv.integer"), Some(&opentelemetry::Value::I64(2))); + assert_eq!( + value("nv.strings"), + Some(&opentelemetry::Value::Array(opentelemetry::Array::String( + vec!["a".into(), "b".into()] + ))) + ); + assert_eq!( + value("nv.bools"), + Some(&opentelemetry::Value::Array(opentelemetry::Array::Bool( + vec![true, false] + ))) + ); + assert_eq!( + value("nv.integers"), + Some(&opentelemetry::Value::Array(opentelemetry::Array::I64( + vec![2, 3] + ))) + ); + assert_eq!( + value("nv.floats"), + Some(&opentelemetry::Value::Array(opentelemetry::Array::F64( + vec![1.5, 2.5] + ))) + ); + assert_eq!( + value("nv.owned"), + Some(&opentelemetry::Value::String("projection".into())) + ); + assert_eq!(value("other.unmatched"), None); + assert_eq!( + issues, + vec![super::MetadataPromotionIssue { + key: "nv.nested".to_string(), + reason: "object values are not supported", + }] + ); +} + +#[test] +fn reports_unsupported_metadata_array_shapes() { + let event = Event::Mark(MarkEvent::new( + BaseEvent::builder() + .name("metadata-array-rejections") + .metadata(serde_json::json!({ + "nv.empty": [], + "nv.mixed": [1, "two"], + "nv.nested": [[1]], + "nv.nulls": [null], + "nv.oversized": [18446744073709551615u64] + })) + .build(), + None, + None, + )); + let mut attributes = Vec::new(); + + let issues = promote_event_metadata_attributes( + &mut attributes, + &event, + &["nv.".to_string()], + &HashSet::new(), + ); + + assert!(attributes.is_empty()); + let issues = issues + .into_iter() + .map(|issue| (issue.key, issue.reason)) + .collect::>(); + assert_eq!(issues.len(), 5); + assert_eq!( + issues.get("nv.empty"), + Some(&"empty arrays do not declare an OTLP element type") + ); + assert_eq!( + issues.get("nv.mixed"), + Some(&"array values must have one primitive type") + ); + assert_eq!( + issues.get("nv.nested"), + Some(&"nested arrays and objects are not supported") + ); + assert_eq!( + issues.get("nv.nulls"), + Some(&"arrays of null are not OTLP attributes") + ); + assert_eq!( + issues.get("nv.oversized"), + Some(&"array contains an unsigned integer larger than OTLP i64") + ); +} + +#[test] +fn validates_metadata_promotion_prefixes_against_metadata_key_syntax() { + assert!(validate_metadata_promotion_prefixes(&[]).is_ok()); + + for prefix in [ + "nv", + "nv.", + "nv_", + "nv-", + "nv.client", + "nv.client.", + "nv_client", + "nv-client", + "nv2.", + "NV.", + ] { + assert!( + validate_metadata_promotion_prefixes(&[prefix.to_string()]).is_ok(), + "expected {prefix:?} to be accepted" + ); + } + + assert!( + validate_metadata_promotion_prefixes(&[ + "nv.".to_string(), + "os.".to_string(), + "host_".to_string(), + ]) + .is_ok() + ); + + for prefix in [ + "", " ", " nv.", "nv. ", ".nv", "nv..", "nv:", "nv/", "nv value", "nv.*", + ] { + assert!( + validate_metadata_promotion_prefixes(&[prefix.to_string()]).is_err(), + "expected {prefix:?} to be rejected" + ); + } + + assert!(validate_metadata_promotion_prefixes(&["nv.".to_string(), "nv.".to_string()]).is_err()); +} diff --git a/crates/core/tests/unit/observability/openinference_tests.rs b/crates/core/tests/unit/observability/openinference_tests.rs index 07f36cac6..91c87572a 100644 --- a/crates/core/tests/unit/observability/openinference_tests.rs +++ b/crates/core/tests/unit/observability/openinference_tests.rs @@ -795,6 +795,7 @@ fn mapped_aliases_are_typed_and_cannot_replace_projected_span_fields() { ), crate::observability::OtlpAttributeMapping::new("missing.source", "ignored.alias"), ], + promote_metadata_prefixes: Vec::new(), }, ) .unwrap(); diff --git a/crates/core/tests/unit/observability/otel_tests.rs b/crates/core/tests/unit/observability/otel_tests.rs index 6bbfcf33f..26e6ac2ad 100644 --- a/crates/core/tests/unit/observability/otel_tests.rs +++ b/crates/core/tests/unit/observability/otel_tests.rs @@ -492,6 +492,225 @@ fn rootless_propagation_starts_a_new_otel_trace() { assert!(!span.parent_span_is_remote); } +#[test] +fn promotes_final_scope_and_mark_metadata_without_duplicate_span_keys() { + for otel_type in [OpenTelemetryType::Full, OpenTelemetryType::OpenInference] { + let (provider, exporter) = make_provider(); + let mut processor = + OtelEventProcessor::new_with_mark_projection_and_exclusions_and_mappings_and_runtime_diagnostics( + provider, + "metadata-promotion-test".into(), + otel_type, + MarkProjection::default(), + default_mark_exclude_names(), + Vec::new(), + vec!["nv.".to_string(), "nemo_relay.".to_string()], + SignalRuntimeDiagnostics::new(None), + ); + let uuid = Uuid::now_v7(); + processor.process(&make_start_event_with_metadata( + uuid, + None, + "metadata-promotion-scope", + json!({ + "nv.source": "start", + "nemo_relay.scope_type": "attempted-overwrite" + }), + )); + processor.process(&make_mark_event_with_metadata( + Some(uuid), + json!({"nv.source": "mark"}), + )); + processor.process(&make_end_event_with_metadata( + uuid, + None, + "metadata-promotion-scope", + ScopeType::Agent, + json!({ + "nv.source": "end", + "nv.completed": true, + "nemo_relay.scope_type": "attempted-overwrite" + }), + )); + processor.force_flush().unwrap(); + + let spans = exporter.get_finished_spans().unwrap(); + assert_eq!(spans.len(), 1); + let span = &spans[0]; + assert_eq!( + span.attributes + .iter() + .filter(|attribute| attribute.key.as_str() == "nv.source") + .count(), + 1 + ); + let attributes = attr_map(&span.attributes); + assert_eq!(attributes.get("nv.source"), Some(&"end".to_string())); + assert_eq!(attributes.get("nv.completed"), Some(&"true".to_string())); + assert_eq!( + attributes.get("nemo_relay.scope_type"), + Some(&"agent".to_string()) + ); + let mark_attributes = attr_map(&span.events.events[0].attributes); + assert_eq!(mark_attributes.get("nv.source"), Some(&"mark".to_string())); + } +} + +#[test] +fn promotes_orphan_and_tool_projection_mark_metadata() { + for otel_type in [OpenTelemetryType::Full, OpenTelemetryType::OpenInference] { + let (provider, exporter) = make_provider(); + let mut processor = + OtelEventProcessor::new_with_mark_projection_and_exclusions_and_mappings_and_runtime_diagnostics( + provider, + "metadata-promotion-mark-test".into(), + otel_type, + MarkProjection::Tool, + default_mark_exclude_names(), + Vec::new(), + vec!["nv.".to_string()], + SignalRuntimeDiagnostics::new(None), + ); + let parent_uuid = Uuid::now_v7(); + processor.process(&Event::Mark(MarkEvent::new( + BaseEvent::builder() + .name("metadata.orphan") + .metadata(json!({"nv.source": "orphan"})) + .build(), + None, + None, + ))); + processor.process(&make_start_event( + parent_uuid, + None, + "metadata-promotion-parent", + ScopeType::Agent, + None, + )); + processor.process(&Event::Mark(MarkEvent::new( + BaseEvent::builder() + .parent_uuid(parent_uuid) + .name("metadata.projected") + .metadata(json!({"nv.source": "projected"})) + .build(), + None, + None, + ))); + processor.process(&make_end_event( + parent_uuid, + None, + "metadata-promotion-parent", + ScopeType::Agent, + None, + )); + processor.force_flush().unwrap(); + + let spans = exporter.get_finished_spans().unwrap(); + assert_eq!(spans.len(), 3); + let parent = finished_span_named(&spans, "metadata-promotion-parent"); + let orphan = finished_span_named(&spans, "mark:metadata.orphan"); + let projected = finished_span_named(&spans, "mark:metadata.projected"); + assert!(!attr_map(&parent.attributes).contains_key("nv.source")); + assert_eq!( + attr_map(&orphan.attributes).get("nv.source"), + Some(&"orphan".to_string()) + ); + assert_eq!( + attr_map(&projected.attributes).get("nv.source"), + Some(&"projected".to_string()) + ); + assert_eq!(projected.parent_span_id, parent.span_context.span_id()); + } +} + +#[test] +fn promotes_final_scope_metadata_across_trace_projections() { + for otel_type in [ + OpenTelemetryType::Full, + OpenTelemetryType::GenAi, + OpenTelemetryType::OpenInference, + ] { + let (provider, exporter) = make_provider(); + let mut processor = + OtelEventProcessor::new_with_mark_projection_and_exclusions_and_mappings_and_runtime_diagnostics( + provider, + "metadata-promotion-projection-test".into(), + otel_type, + MarkProjection::default(), + default_mark_exclude_names(), + Vec::new(), + vec!["nv.".to_string()], + SignalRuntimeDiagnostics::new(None), + ); + let uuid = Uuid::now_v7(); + processor.process(&make_start_event_with_metadata( + uuid, + None, + "metadata-promotion-projection-scope", + json!({"nv.source": "start"}), + )); + processor.process(&make_end_event_with_metadata( + uuid, + None, + "metadata-promotion-projection-scope", + ScopeType::Agent, + json!({"nv.source": "end"}), + )); + processor.force_flush().unwrap(); + + let spans = exporter.get_finished_spans().unwrap(); + assert_eq!(spans.len(), 1); + assert_eq!( + attr_map(&spans[0].attributes).get("nv.source"), + Some(&"end".to_string()) + ); + } +} + +#[test] +fn promotes_start_only_scope_metadata_across_trace_projections() { + for otel_type in [ + OpenTelemetryType::Full, + OpenTelemetryType::GenAi, + OpenTelemetryType::OpenInference, + ] { + let (provider, exporter) = make_provider(); + let mut processor = + OtelEventProcessor::new_with_mark_projection_and_exclusions_and_mappings_and_runtime_diagnostics( + provider, + "start-metadata-promotion-projection-test".into(), + otel_type, + MarkProjection::default(), + default_mark_exclude_names(), + Vec::new(), + vec!["nv.".to_string()], + SignalRuntimeDiagnostics::new(None), + ); + let uuid = Uuid::now_v7(); + processor.process(&make_start_event_with_metadata( + uuid, + None, + "start-metadata-promotion-projection-scope", + json!({"nv.start_only": "configured"}), + )); + processor.process(&make_end_event_with_metadata( + uuid, + None, + "start-metadata-promotion-projection-scope", + ScopeType::Agent, + json!({}), + )); + processor.force_flush().unwrap(); + + let spans = exporter.get_finished_spans().unwrap(); + assert_eq!(spans.len(), 1); + assert_eq!( + attr_map(&spans[0].attributes).get("nv.start_only"), + Some(&"configured".to_string()) + ); + } +} + fn make_start_event_with_metadata( uuid: Uuid, parent_uuid: Option, @@ -936,6 +1155,7 @@ fn mapped_aliases_are_typed_and_cannot_replace_projected_span_fields() { ), crate::observability::OtlpAttributeMapping::new("missing.source", "ignored.alias"), ], + promote_metadata_prefixes: Vec::new(), }, ) .unwrap(); diff --git a/crates/core/tests/unit/observability/plugin_component_tests.rs b/crates/core/tests/unit/observability/plugin_component_tests.rs index 8e376d8f4..0446c8ee7 100644 --- a/crates/core/tests/unit/observability/plugin_component_tests.rs +++ b/crates/core/tests/unit/observability/plugin_component_tests.rs @@ -325,6 +325,11 @@ fn assert_trace_endpoint_editor_schema(otlp: &EditorSchema) { .kind, EditorFieldKind::StringMap ); + let promote_metadata_prefixes = otlp_endpoint_schema + .field("promote_metadata_prefixes") + .expect("OTLP endpoint promote_metadata_prefixes"); + assert_eq!(promote_metadata_prefixes.kind, EditorFieldKind::List); + assert!(promote_metadata_prefixes.optional); for field in [ "max_queue_size", "max_export_batch_size", @@ -770,6 +775,7 @@ fn default_config_and_component_conversion_cover_public_shape() { mark_projection: MarkProjection::default(), mark_exclude_names: default_mark_exclude_names(), attribute_mappings: Vec::new(), + promote_metadata_prefixes: Vec::new(), }], logs: None, metrics: None, @@ -956,6 +962,7 @@ fn opentelemetry_endpoint_header_env_is_resolved_and_snapshotted() { mark_projection: MarkProjection::default(), mark_exclude_names: default_mark_exclude_names(), attribute_mappings: Vec::new(), + promote_metadata_prefixes: Vec::new(), }, ) .unwrap(); @@ -983,6 +990,7 @@ fn test_opentelemetry_endpoint() -> OpenTelemetryEndpointConfig { mark_projection: MarkProjection::default(), mark_exclude_names: default_mark_exclude_names(), attribute_mappings: Vec::new(), + promote_metadata_prefixes: Vec::new(), } } @@ -1343,6 +1351,7 @@ fn opentelemetry_endpoint_accepts_legacy_projection_controls_and_rejects_unknown "mark_projection": "tool", "mark_exclude_names": ["notification"], "attribute_mappings": [{"key": "nemo_relay.model_name", "alias": "model.alias"}], + "promote_metadata_prefixes": ["nv."], "max_queue_size": 4096, "max_export_batch_size": 256, "scheduled_delay_millis": 750, @@ -1371,6 +1380,7 @@ fn opentelemetry_endpoint_accepts_legacy_projection_controls_and_rejects_unknown Some("endpoints[0].mark_projection") | Some("endpoints[0].mark_exclude_names") | Some("endpoints[0].attribute_mappings") + | Some("endpoints[0].promote_metadata_prefixes") | Some("endpoints[0].max_queue_size") | Some("endpoints[0].max_export_batch_size") | Some("endpoints[0].scheduled_delay_millis") @@ -1450,6 +1460,27 @@ fn opentelemetry_endpoint_accepts_valid_attribute_mappings() { ); } +#[test] +fn opentelemetry_endpoint_rejects_glob_metadata_promotion_prefix() { + let report = validate_plugin_config(&plugin_config(json!({ + "opentelemetry": { + "enabled": true, + "endpoints": [{ + "type": "full", + "endpoint": "http://localhost:4318/v1/traces", + "promote_metadata_prefixes": ["nv.*"] + }] + } + }))); + + assert!(report.has_errors()); + assert!(report.diagnostics.iter().any(|diagnostic| { + diagnostic.code == "observability.unsupported_value" + && diagnostic.field.as_deref() == Some("endpoints[0].promote_metadata_prefixes") + && diagnostic.message.contains("literal prefix, not a glob") + })); +} + #[test] fn opentelemetry_endpoint_rejects_invalid_and_case_duplicate_headers() { let report = validate_plugin_config(&plugin_config(json!({ diff --git a/crates/ffi/nemo_relay.h b/crates/ffi/nemo_relay.h index 2a330e2a2..0d2bc97ef 100644 --- a/crates/ffi/nemo_relay.h +++ b/crates/ffi/nemo_relay.h @@ -1706,6 +1706,31 @@ NemoRelayStatus nemo_relay_otel_subscriber_create_with_projection_options(const const char *attribute_mappings_json, struct FfiOpenTelemetrySubscriber **out); +/** + * Creates one typed OpenTelemetry exporter subscriber with projection and metadata controls. + * + * `promote_metadata_prefixes_json` is a JSON array of literal metadata prefixes, + * such as `["nv."]`. Pass null to disable metadata promotion. + * + * # Safety + * Any non-null C strings must be valid and `out` must be non-null. + */ +NemoRelayStatus nemo_relay_otel_subscriber_create_with_projection_options_v2(const char *otel_type, + const char *transport, + const char *endpoint, + const char *headers_json, + const char *resource_attributes_json, + const char *service_name, + const char *service_namespace, + const char *service_version, + const char *instrumentation_scope, + uint64_t timeout_millis, + const char *mark_projection, + const char *mark_exclude_names_json, + const char *attribute_mappings_json, + const char *promote_metadata_prefixes_json, + struct FfiOpenTelemetrySubscriber **out); + /** * Registers the OpenTelemetry subscriber as an event subscriber. * diff --git a/crates/ffi/src/api/observability.rs b/crates/ffi/src/api/observability.rs index 970691c41..99bd60b9e 100644 --- a/crates/ffi/src/api/observability.rs +++ b/crates/ffi/src/api/observability.rs @@ -668,6 +668,28 @@ fn parse_mark_exclude_names(ptr: *const c_char) -> Result, NemoRelay }) } +fn parse_promote_metadata_prefixes(ptr: *const c_char) -> Result, NemoRelayStatus> { + if ptr.is_null() { + return Ok(Vec::new()); + } + let Some(value) = c_str_to_json(ptr) else { + return Err(NemoRelayStatus::InvalidJson); + }; + let prefixes: Vec = serde_json::from_value(value).map_err(|error| { + set_last_error(&format!( + "promote_metadata_prefixes must be an array of strings: {error}" + )); + NemoRelayStatus::InvalidArg + })?; + nemo_relay::observability::validate_metadata_promotion_prefixes(&prefixes).map_err( + |error| { + set_last_error(&error); + NemoRelayStatus::InvalidArg + }, + )?; + Ok(prefixes) +} + fn parse_attribute_mappings( ptr: *const c_char, ) -> Result, NemoRelayStatus> { @@ -870,6 +892,53 @@ pub unsafe extern "C" fn nemo_relay_otel_subscriber_create_with_projection_optio mark_exclude_names_json: *const c_char, attribute_mappings_json: *const c_char, out: *mut *mut FfiOpenTelemetrySubscriber, +) -> NemoRelayStatus { + unsafe { + nemo_relay_otel_subscriber_create_with_projection_options_v2( + otel_type, + transport, + endpoint, + headers_json, + resource_attributes_json, + service_name, + service_namespace, + service_version, + instrumentation_scope, + timeout_millis, + mark_projection, + mark_exclude_names_json, + attribute_mappings_json, + std::ptr::null(), + out, + ) + } +} + +/// Creates one typed OpenTelemetry exporter subscriber with projection and metadata controls. +/// +/// `promote_metadata_prefixes_json` is a JSON array of literal metadata prefixes, +/// such as `["nv."]`. Pass null to disable metadata promotion. +/// +/// # Safety +/// Any non-null C strings must be valid and `out` must be non-null. +#[allow(clippy::too_many_arguments)] +#[unsafe(no_mangle)] +pub unsafe extern "C" fn nemo_relay_otel_subscriber_create_with_projection_options_v2( + otel_type: *const c_char, + transport: *const c_char, + endpoint: *const c_char, + headers_json: *const c_char, + resource_attributes_json: *const c_char, + service_name: *const c_char, + service_namespace: *const c_char, + service_version: *const c_char, + instrumentation_scope: *const c_char, + timeout_millis: u64, + mark_projection: *const c_char, + mark_exclude_names_json: *const c_char, + attribute_mappings_json: *const c_char, + promote_metadata_prefixes_json: *const c_char, + out: *mut *mut FfiOpenTelemetrySubscriber, ) -> NemoRelayStatus { clear_last_error(); if let Err(status) = required_out_ptr(out) { @@ -902,7 +971,13 @@ pub unsafe extern "C" fn nemo_relay_otel_subscriber_create_with_projection_optio .with_attribute_mappings(match parse_attribute_mappings(attribute_mappings_json) { Ok(value) => value, Err(status) => return status, - }); + }) + .with_promote_metadata_prefixes( + match parse_promote_metadata_prefixes(promote_metadata_prefixes_json) { + Ok(value) => value, + Err(status) => return status, + }, + ); let subscriber = match create_otel_subscriber(config) { Ok(subscriber) => subscriber, Err(status) => return status, diff --git a/crates/ffi/tests/unit/api/plugin_tests.rs b/crates/ffi/tests/unit/api/plugin_tests.rs index f038cdf28..1dfa4850e 100644 --- a/crates/ffi/tests/unit/api/plugin_tests.rs +++ b/crates/ffi/tests/unit/api/plugin_tests.rs @@ -1525,6 +1525,67 @@ fn test_ffi_otel_projection_options_accept_and_validate_legacy_controls() { } } +#[test] +fn test_ffi_otel_projection_options_v2_accepts_and_validates_metadata_prefixes() { + let _lock = TEST_MUTEX.lock().unwrap_or_else(|error| error.into_inner()); + reset_globals(); + + unsafe { + let endpoint = c"http://localhost:4318/v1/traces"; + let valid_prefixes = c"[\"nv.\",\"host_\"]"; + let mut subscriber: *mut FfiOpenTelemetrySubscriber = ptr::null_mut(); + assert_status!( + nemo_relay_otel_subscriber_create_with_projection_options_v2( + c"full".as_ptr(), + ptr::null(), + endpoint.as_ptr(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + 0, + c"inherit".as_ptr(), + ptr::null(), + ptr::null(), + valid_prefixes.as_ptr(), + &mut subscriber, + ), + NemoRelayStatus::Ok + ); + nemo_relay_otel_subscriber_free(subscriber); + + for (prefixes, expected_status) in [ + (c"[\"nv.*\"]".as_ptr(), NemoRelayStatus::InvalidArg), + (c"{".as_ptr(), NemoRelayStatus::InvalidJson), + ] { + let mut invalid_subscriber: *mut FfiOpenTelemetrySubscriber = ptr::null_mut(); + assert_status!( + nemo_relay_otel_subscriber_create_with_projection_options_v2( + c"full".as_ptr(), + ptr::null(), + endpoint.as_ptr(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + 0, + c"inherit".as_ptr(), + ptr::null(), + ptr::null(), + prefixes, + &mut invalid_subscriber, + ), + expected_status + ); + assert!(invalid_subscriber.is_null()); + } + } +} + #[test] fn test_ffi_specialized_constructor_invalid_utf8_and_malformed_json_sweep() { let _lock = TEST_MUTEX.lock().unwrap_or_else(|e| e.into_inner()); diff --git a/crates/node/observability.d.ts b/crates/node/observability.d.ts index ae8bd2365..e80f6b484 100644 --- a/crates/node/observability.d.ts +++ b/crates/node/observability.d.ts @@ -72,6 +72,7 @@ export interface OpenTelemetryEndpointConfig { mark_projection?: 'inherit' | 'event' | 'tool'; mark_exclude_names?: string[]; attribute_mappings?: Array<{ key: string; alias: string }>; + promote_metadata_prefixes?: string[]; transport?: 'http_binary' | 'grpc'; headers?: Record; header_env?: Record; diff --git a/crates/node/observability.js b/crates/node/observability.js index d6061d5f6..c4bfc6620 100644 --- a/crates/node/observability.js +++ b/crates/node/observability.js @@ -71,6 +71,7 @@ function openTelemetryEndpoint(config) { headers: {}, header_env: {}, resource_attributes: {}, + promote_metadata_prefixes: [], ...config, }; } diff --git a/crates/node/src/api/mod.rs b/crates/node/src/api/mod.rs index 36af02086..0935a3f40 100644 --- a/crates/node/src/api/mod.rs +++ b/crates/node/src/api/mod.rs @@ -298,7 +298,8 @@ fn build_otel_config( .mark_exclude_names .unwrap_or_else(nemo_relay::observability::default_mark_exclude_names), ) - .with_attribute_mappings(parse_attribute_mappings(options.attribute_mappings)?); + .with_attribute_mappings(parse_attribute_mappings(options.attribute_mappings)?) + .with_promote_metadata_prefixes(options.promote_metadata_prefixes.unwrap_or_default()); Ok(config) } @@ -4560,6 +4561,8 @@ pub struct OpenTelemetryConfig { pub mark_exclude_names: Option>, /// Attribute aliases for full and OpenInference projections. pub attribute_mappings: Option, + /// Literal Event metadata prefixes copied to top-level OTLP attributes. + pub promote_metadata_prefixes: Option>, } /// OpenTelemetry-backed event subscriber. diff --git a/crates/node/tests/observability_plugin_tests.mjs b/crates/node/tests/observability_plugin_tests.mjs index 516500243..46079a33d 100644 --- a/crates/node/tests/observability_plugin_tests.mjs +++ b/crates/node/tests/observability_plugin_tests.mjs @@ -51,6 +51,7 @@ describe('observability plugin helpers', () => { headers: {}, header_env: { authorization: 'OTEL_AUTHORIZATION' }, resource_attributes: {}, + promote_metadata_prefixes: [], service_name: 'unknown_service', instrumentation_scope: 'opentelemetry', timeout_millis: 3000, diff --git a/crates/node/tests/otel_tests.mjs b/crates/node/tests/otel_tests.mjs index 32861416f..d1104c00e 100644 --- a/crates/node/tests/otel_tests.mjs +++ b/crates/node/tests/otel_tests.mjs @@ -4,7 +4,7 @@ import { describe, it } from 'node:test'; import assert from 'node:assert/strict'; import { createRequire } from 'node:module'; -import { startCollector } from '../../../scripts/test-support/otel_test_utils.mjs'; +import { assertOtlpStringAttribute, startCollector } from '../../../scripts/test-support/otel_test_utils.mjs'; const require = createRequire(import.meta.url); const { @@ -49,6 +49,7 @@ describe('OpenTelemetrySubscriber', () => { markProjection: 'tool', markExcludeNames: ['custom.mark'], attributeMappings: [{ key: 'nemo_relay.model_name', alias: 'model.alias' }], + promoteMetadataPrefixes: ['nv.'], }); const name = uniqueId('node_otel'); @@ -100,6 +101,15 @@ describe('OpenTelemetrySubscriber', () => { }), /attribute mapping key must not be blank/i, ); + assert.throws( + () => + new OpenTelemetrySubscriber({ + type: 'full', + endpoint: 'http://localhost:4318/v1/traces', + promoteMetadataPrefixes: ['nv.*'], + }), + /literal prefix, not a glob/i, + ); assert.throws(() => new OpenTelemetrySubscriber({ endpoint: 'http://localhost:4318' }), /missing field `type`/i); assert.throws(() => new OpenTelemetrySubscriber({ type: 'full' }), /missing field `endpoint`/i); assert.throws( @@ -122,12 +132,15 @@ describe('OpenTelemetrySubscriber', () => { type: 'full', endpoint: collector.endpoint, serviceName: 'node-agent', + promoteMetadataPrefixes: ['nv.'], }); const name = uniqueId('node_otel_e2e'); subscriber.register(name); try { - const scope = pushScope('otel_scope', ScopeType.Agent, null, null, null, null); + const scope = pushScope('otel_scope', ScopeType.Agent, null, null, null, { + 'nv.binding': 'node', + }); event( 'otel_mark', scope, @@ -138,7 +151,9 @@ describe('OpenTelemetrySubscriber', () => { source: 'node', }, ); - popScope(scope); + popScope(scope, null, null, { + 'nv.binding': 'node', + }); subscriber.forceFlush(); const request = await collector.nextRequest(); @@ -146,6 +161,7 @@ describe('OpenTelemetrySubscriber', () => { assert.equal(request.headers['content-type'], 'application/x-protobuf'); assert.ok(request.body.length > 0); assertBodyContains(request.body, 'nemo_relay.mark.metadata.source'); + assertOtlpStringAttribute(request.body, 'nv.binding', 'node'); } finally { subscriber.deregister(name); subscriber.shutdown(); diff --git a/crates/python/src/py_types/observability.rs b/crates/python/src/py_types/observability.rs index be1533416..a72e90482 100644 --- a/crates/python/src/py_types/observability.rs +++ b/crates/python/src/py_types/observability.rs @@ -490,6 +490,8 @@ pub struct PyOpenTelemetryConfig { pub(crate) headers: HashMap, pub(crate) resource_attributes: HashMap, pub(crate) attribute_mappings: Vec, + #[pyo3(get, set)] + pub(crate) promote_metadata_prefixes: Vec, } impl PyOpenTelemetryConfig { @@ -534,10 +536,15 @@ impl PyOpenTelemetryConfig { .map_err(|error| pyo3::exceptions::PyValueError::new_err(error.to_string()))?; nemo_relay::observability::validate_attribute_mappings(&self.attribute_mappings) .map_err(|error| pyo3::exceptions::PyValueError::new_err(error.to_string()))?; + nemo_relay::observability::validate_metadata_promotion_prefixes( + &self.promote_metadata_prefixes, + ) + .map_err(|error| pyo3::exceptions::PyValueError::new_err(error.to_string()))?; Ok(config .with_mark_projection(mark_projection) .with_mark_exclude_names(self.mark_exclude_names.clone()) - .with_attribute_mappings(self.attribute_mappings.clone())) + .with_attribute_mappings(self.attribute_mappings.clone()) + .with_promote_metadata_prefixes(self.promote_metadata_prefixes.clone())) } } @@ -559,6 +566,7 @@ impl PyOpenTelemetryConfig { headers: HashMap::new(), resource_attributes: HashMap::new(), attribute_mappings: Vec::new(), + promote_metadata_prefixes: Vec::new(), } } diff --git a/go/nemo_relay/nemo_relay.go b/go/nemo_relay/nemo_relay.go index d95c900ee..72da4d715 100644 --- a/go/nemo_relay/nemo_relay.go +++ b/go/nemo_relay/nemo_relay.go @@ -272,6 +272,7 @@ extern void nemo_relay_atof_exporter_free(void*); // OpenTelemetry subscriber extern int32_t nemo_relay_otel_subscriber_create(const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, uint64_t, void**); extern int32_t nemo_relay_otel_subscriber_create_with_projection_options(const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, uint64_t, const char*, const char*, const char*, void**); +extern int32_t nemo_relay_otel_subscriber_create_with_projection_options_v2(const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, uint64_t, const char*, const char*, const char*, const char*, void**); extern int32_t nemo_relay_otel_subscriber_register(const void*, const char*); extern int32_t nemo_relay_otel_subscriber_deregister(const char*); extern int32_t nemo_relay_otel_subscriber_force_flush(const void*); @@ -2242,35 +2243,37 @@ const ( // Create it with [NewOpenTelemetryConfig], then mutate fields as needed before // passing it to [NewOpenTelemetrySubscriber]. type OpenTelemetryConfig struct { - Type OpenTelemetryType - Transport OpenTelemetryTransport - Endpoint string - Headers map[string]string - ResourceAttributes map[string]string - ServiceName string - ServiceNamespace string - ServiceVersion string - InstrumentationScope string - Timeout time.Duration - MarkProjection MarkProjection - MarkExcludeNames []string - AttributeMappings []OtlpAttributeMapping + Type OpenTelemetryType + Transport OpenTelemetryTransport + Endpoint string + Headers map[string]string + ResourceAttributes map[string]string + ServiceName string + ServiceNamespace string + ServiceVersion string + InstrumentationScope string + Timeout time.Duration + MarkProjection MarkProjection + MarkExcludeNames []string + AttributeMappings []OtlpAttributeMapping + PromoteMetadataPrefixes []string } // NewOpenTelemetryConfig returns a typed config for the required endpoint. func NewOpenTelemetryConfig(otelType OpenTelemetryType, endpoint string) OpenTelemetryConfig { return OpenTelemetryConfig{ - Type: otelType, - Transport: OpenTelemetryTransportHTTPBinary, - Endpoint: endpoint, - Headers: map[string]string{}, - ResourceAttributes: map[string]string{}, - ServiceName: "unknown_service", - InstrumentationScope: "opentelemetry", - Timeout: 3 * time.Second, - MarkProjection: MarkProjectionInherit, - MarkExcludeNames: []string{"llm.chunk"}, - AttributeMappings: []OtlpAttributeMapping{}, + Type: otelType, + Transport: OpenTelemetryTransportHTTPBinary, + Endpoint: endpoint, + Headers: map[string]string{}, + ResourceAttributes: map[string]string{}, + ServiceName: "unknown_service", + InstrumentationScope: "opentelemetry", + Timeout: 3 * time.Second, + MarkProjection: MarkProjectionInherit, + MarkExcludeNames: []string{"llm.chunk"}, + AttributeMappings: []OtlpAttributeMapping{}, + PromoteMetadataPrefixes: []string{}, } } @@ -2330,6 +2333,9 @@ func normalizeOpenTelemetryConfig(config OpenTelemetryConfig) (OpenTelemetryConf if config.AttributeMappings == nil { config.AttributeMappings = []OtlpAttributeMapping{} } + if config.PromoteMetadataPrefixes == nil { + config.PromoteMetadataPrefixes = []string{} + } return config, nil } @@ -2394,9 +2400,15 @@ func NewOpenTelemetrySubscriber(config OpenTelemetryConfig) (*OpenTelemetrySubsc } cAttributeMappingsJSON := C.CString(string(attributeMappingsJSON)) defer C.free(unsafe.Pointer(cAttributeMappingsJSON)) + promoteMetadataPrefixesJSON, err := jsonMarshal(config.PromoteMetadataPrefixes) + if err != nil { + return nil, err + } + cPromoteMetadataPrefixesJSON := C.CString(string(promoteMetadataPrefixesJSON)) + defer C.free(unsafe.Pointer(cPromoteMetadataPrefixesJSON)) var ptr unsafe.Pointer - status := C.nemo_relay_otel_subscriber_create_with_projection_options( + status := C.nemo_relay_otel_subscriber_create_with_projection_options_v2( cType, cTransport, cEndpoint, @@ -2410,6 +2422,7 @@ func NewOpenTelemetrySubscriber(config OpenTelemetryConfig) (*OpenTelemetrySubsc cMarkProjection, cMarkExcludeNamesJSON, cAttributeMappingsJSON, + cPromoteMetadataPrefixesJSON, &ptr, ) if err := checkStatus(status); err != nil { diff --git a/go/nemo_relay/observability_plugin.go b/go/nemo_relay/observability_plugin.go index 29cfedc7f..987d71cfd 100644 --- a/go/nemo_relay/observability_plugin.go +++ b/go/nemo_relay/observability_plugin.go @@ -65,23 +65,24 @@ type ObservabilityOpenTelemetryMetricConfig struct { // ObservabilityOpenTelemetryEndpointConfig configures one typed OTLP destination. type ObservabilityOpenTelemetryEndpointConfig struct { - Type OpenTelemetryType `json:"type"` - Endpoint string `json:"endpoint"` - MarkProjection string `json:"mark_projection,omitempty"` - MarkExcludeNames []string `json:"mark_exclude_names,omitempty"` - AttributeMappings []OtlpAttributeMapping `json:"attribute_mappings,omitempty"` - Transport string `json:"transport,omitempty"` - Headers map[string]string `json:"headers,omitempty"` - HeaderEnv map[string]string `json:"header_env,omitempty"` - ResourceAttributes map[string]string `json:"resource_attributes,omitempty"` - ServiceName string `json:"service_name,omitempty"` - ServiceNamespace string `json:"service_namespace,omitempty"` - ServiceVersion string `json:"service_version,omitempty"` - InstrumentationScope string `json:"instrumentation_scope,omitempty"` - TimeoutMillis uint64 `json:"timeout_millis,omitempty"` - MaxQueueSize *uint64 `json:"max_queue_size,omitempty"` - MaxExportBatchSize *uint64 `json:"max_export_batch_size,omitempty"` - ScheduledDelayMillis *uint64 `json:"scheduled_delay_millis,omitempty"` + Type OpenTelemetryType `json:"type"` + Endpoint string `json:"endpoint"` + MarkProjection string `json:"mark_projection,omitempty"` + MarkExcludeNames []string `json:"mark_exclude_names,omitempty"` + AttributeMappings []OtlpAttributeMapping `json:"attribute_mappings,omitempty"` + PromoteMetadataPrefixes []string `json:"promote_metadata_prefixes,omitempty"` + Transport string `json:"transport,omitempty"` + Headers map[string]string `json:"headers,omitempty"` + HeaderEnv map[string]string `json:"header_env,omitempty"` + ResourceAttributes map[string]string `json:"resource_attributes,omitempty"` + ServiceName string `json:"service_name,omitempty"` + ServiceNamespace string `json:"service_namespace,omitempty"` + ServiceVersion string `json:"service_version,omitempty"` + InstrumentationScope string `json:"instrumentation_scope,omitempty"` + TimeoutMillis uint64 `json:"timeout_millis,omitempty"` + MaxQueueSize *uint64 `json:"max_queue_size,omitempty"` + MaxExportBatchSize *uint64 `json:"max_export_batch_size,omitempty"` + ScheduledDelayMillis *uint64 `json:"scheduled_delay_millis,omitempty"` } // ObservabilityAtofConfig configures filesystem-backed raw ATOF JSONL export. @@ -329,17 +330,18 @@ func NewObservabilityOpenTelemetryMetricConfig() ObservabilityOpenTelemetryMetri // NewObservabilityOpenTelemetryEndpointConfig returns one typed endpoint with defaults. func NewObservabilityOpenTelemetryEndpointConfig(otelType OpenTelemetryType, endpoint string) ObservabilityOpenTelemetryEndpointConfig { return ObservabilityOpenTelemetryEndpointConfig{ - Type: otelType, - Endpoint: endpoint, - Transport: "http_binary", - MarkProjection: "inherit", - MarkExcludeNames: []string{"llm.chunk"}, - Headers: map[string]string{}, - HeaderEnv: map[string]string{}, - ResourceAttributes: map[string]string{}, - ServiceName: "unknown_service", - InstrumentationScope: "opentelemetry", - TimeoutMillis: 3000, + Type: otelType, + Endpoint: endpoint, + Transport: "http_binary", + MarkProjection: "inherit", + MarkExcludeNames: []string{"llm.chunk"}, + PromoteMetadataPrefixes: []string{}, + Headers: map[string]string{}, + HeaderEnv: map[string]string{}, + ResourceAttributes: map[string]string{}, + ServiceName: "unknown_service", + InstrumentationScope: "opentelemetry", + TimeoutMillis: 3000, } } diff --git a/go/nemo_relay/observability_plugin_test.go b/go/nemo_relay/observability_plugin_test.go index 81def0b9e..90369ae69 100644 --- a/go/nemo_relay/observability_plugin_test.go +++ b/go/nemo_relay/observability_plugin_test.go @@ -75,6 +75,7 @@ func TestObservabilityConfigHelpers(t *testing.T) { NewObservabilityOpenTelemetryEndpointConfig(OpenTelemetryTypeFull, "http://localhost:4318/v1/traces"), } otel.Endpoints[0].HeaderEnv["authorization"] = "OTEL_AUTHORIZATION" + otel.Endpoints[0].PromoteMetadataPrefixes = []string{"nv."} maxQueueSize := uint64(4096) maxExportBatchSize := uint64(256) scheduledDelayMillis := uint64(750) @@ -162,6 +163,10 @@ func assertWrappedObservabilityConfig(t *testing.T, wrapped PluginComponentSpec) if otelEndpoints[0].(map[string]any)["header_env"].(map[string]any)["authorization"] != "OTEL_AUTHORIZATION" { t.Fatalf("expected OpenTelemetry header_env in serialized config: %#v", wrapped.Config) } + promotePrefixes := otelEndpoints[0].(map[string]any)["promote_metadata_prefixes"].([]any) + if len(promotePrefixes) != 1 || promotePrefixes[0] != "nv." { + t.Fatalf("expected OpenTelemetry metadata promotion prefixes in serialized config: %#v", wrapped.Config) + } if otelEndpoints[0].(map[string]any)["max_queue_size"] != float64(4096) || otelEndpoints[0].(map[string]any)["max_export_batch_size"] != float64(256) || otelEndpoints[0].(map[string]any)["scheduled_delay_millis"] != float64(750) { diff --git a/go/nemo_relay/otel_test.go b/go/nemo_relay/otel_test.go index 518a888cd..4f1e13392 100644 --- a/go/nemo_relay/otel_test.go +++ b/go/nemo_relay/otel_test.go @@ -66,6 +66,9 @@ func TestNewOpenTelemetryConfigDefaults(t *testing.T) { if config.AttributeMappings == nil || len(config.AttributeMappings) != 0 { t.Fatalf("expected empty attribute mappings, got %#v", config.AttributeMappings) } + if config.PromoteMetadataPrefixes == nil || len(config.PromoteMetadataPrefixes) != 0 { + t.Fatalf("expected empty metadata promotion prefixes, got %#v", config.PromoteMetadataPrefixes) + } } func TestOpenTelemetrySubscriberAcceptsProjectionControls(t *testing.T) { @@ -76,6 +79,7 @@ func TestOpenTelemetrySubscriberAcceptsProjectionControls(t *testing.T) { Key: "nemo_relay.model_name", Alias: "model.alias", }} + config.PromoteMetadataPrefixes = []string{"nv."} subscriber, err := NewOpenTelemetrySubscriber(config) if err != nil { @@ -84,6 +88,15 @@ func TestOpenTelemetrySubscriberAcceptsProjectionControls(t *testing.T) { defer subscriber.Close() } +func TestOpenTelemetrySubscriberRejectsInvalidMetadataPromotionPrefix(t *testing.T) { + config := NewOpenTelemetryConfig(OpenTelemetryTypeFull, otelTestEndpoint) + config.PromoteMetadataPrefixes = []string{"nv.*"} + + if _, err := NewOpenTelemetrySubscriber(config); err == nil { + t.Fatal("expected invalid metadata promotion prefix error") + } +} + func TestOpenTelemetrySubscriberRejectsInvalidAttributeMappings(t *testing.T) { config := NewOpenTelemetryConfig(OpenTelemetryTypeFull, otelTestEndpoint) config.AttributeMappings = []OtlpAttributeMapping{{Key: "", Alias: "model.alias"}} @@ -194,6 +207,7 @@ func TestOpenTelemetrySubscriberExportsScopeLifecycleAndMarks(t *testing.T) { config := NewOpenTelemetryConfig(OpenTelemetryTypeFull, server.URL+otelTestPath) config.ServiceName = "go-agent" + config.PromoteMetadataPrefixes = []string{"nv."} subscriber, err := NewOpenTelemetrySubscriber(config) if err != nil { t.Fatalf(newOpenTelemetrySubscriberFailed, err) @@ -207,7 +221,11 @@ func TestOpenTelemetrySubscriberExportsScopeLifecycleAndMarks(t *testing.T) { defer func() { _ = subscriber.Deregister(name) }() runWithTestScopeStack(t, func() { - handle, err := PushScope("otel_scope", ScopeTypeAgent) + handle, err := PushScope( + "otel_scope", + ScopeTypeAgent, + WithMetadata(json.RawMessage(`{"nv.binding":"go"}`)), + ) if err != nil { t.Fatalf("PushScope failed: %v", err) } @@ -219,7 +237,10 @@ func TestOpenTelemetrySubscriberExportsScopeLifecycleAndMarks(t *testing.T) { ); err != nil { t.Fatalf("EmitEvent failed: %v", err) } - if err := PopScope(handle); err != nil { + if err := PopScope( + handle, + WithScopeEndMetadata(json.RawMessage(`{"nv.binding":"go"}`)), + ); err != nil { t.Fatalf("PopScope failed: %v", err) } }) @@ -239,6 +260,7 @@ func TestOpenTelemetrySubscriberExportsScopeLifecycleAndMarks(t *testing.T) { t.Fatal("expected non-empty OTLP request body") } assertOtlpStringAttribute(t, request.Body, "nemo_relay.scope_type", "agent") + assertOtlpStringAttribute(t, request.Body, "nv.binding", "go") case <-time.After(5 * time.Second): t.Fatal("timed out waiting for OTLP request") } diff --git a/python/nemo_relay/_native.pyi b/python/nemo_relay/_native.pyi index 024b333fa..ee4a46ae4 100644 --- a/python/nemo_relay/_native.pyi +++ b/python/nemo_relay/_native.pyi @@ -1117,6 +1117,7 @@ class OpenTelemetryConfig: timeout_millis: int mark_projection: Literal["inherit", "event", "tool"] mark_exclude_names: list[str] + promote_metadata_prefixes: list[str] def __init__( self, diff --git a/python/nemo_relay/observability.py b/python/nemo_relay/observability.py index 6fd892621..113ef1575 100644 --- a/python/nemo_relay/observability.py +++ b/python/nemo_relay/observability.py @@ -232,6 +232,7 @@ class OpenTelemetryEndpointConfig: max_queue_size: int | None = None max_export_batch_size: int | None = None scheduled_delay_millis: int | None = None + promote_metadata_prefixes: list[str] = field(default_factory=list) def to_dict(self) -> JsonObject: """Serialize this endpoint to the canonical plugin shape.""" @@ -242,6 +243,7 @@ def to_dict(self) -> JsonObject: "mark_projection": self.mark_projection, "mark_exclude_names": self.mark_exclude_names, "attribute_mappings": self.attribute_mappings, + "promote_metadata_prefixes": self.promote_metadata_prefixes, "transport": self.transport, "service_name": self.service_name, "service_namespace": self.service_namespace, diff --git a/python/nemo_relay/observability.pyi b/python/nemo_relay/observability.pyi index 2113fc541..b8d358cc3 100644 --- a/python/nemo_relay/observability.pyi +++ b/python/nemo_relay/observability.pyi @@ -95,6 +95,7 @@ class OpenTelemetryEndpointConfig: max_queue_size: int | None = ... max_export_batch_size: int | None = ... scheduled_delay_millis: int | None = ... + promote_metadata_prefixes: list[str] = field(default_factory=list) def to_dict(self) -> JsonObject: ... @dataclass(slots=True) diff --git a/python/tests/test_observability_plugin.py b/python/tests/test_observability_plugin.py index c84dde1da..07796c097 100644 --- a/python/tests/test_observability_plugin.py +++ b/python/tests/test_observability_plugin.py @@ -137,6 +137,7 @@ def test_defaults_and_component_wrapper(self): "mark_projection": "inherit", "mark_exclude_names": ["llm.chunk"], "attribute_mappings": [], + "promote_metadata_prefixes": [], "transport": "http_binary", "service_name": "unknown_service", "instrumentation_scope": "opentelemetry", @@ -148,6 +149,12 @@ def test_defaults_and_component_wrapper(self): "header_env": {"authorization": "OTEL_AUTHORIZATION"}, "resource_attributes": {}, } + endpoint = OpenTelemetryEndpointConfig( + "gen_ai", + "http://localhost:4318/v1/traces", + promote_metadata_prefixes=["nv."], + ) + assert endpoint.to_dict()["promote_metadata_prefixes"] == ["nv."] wrapped = ComponentSpec(ObservabilityConfig(atof=AtofConfig())).to_dict() assert wrapped["kind"] == OBSERVABILITY_PLUGIN_KIND diff --git a/python/tests/test_types.py b/python/tests/test_types.py index 4ead622f4..b4eed78ca 100644 --- a/python/tests/test_types.py +++ b/python/tests/test_types.py @@ -708,6 +708,7 @@ def test_config_defaults_mutation_and_repr(self): assert config.mark_projection == "inherit" assert config.mark_exclude_names == ["llm.chunk"] assert config.attribute_mappings == [] + assert config.promote_metadata_prefixes == [] config.service_name = "py-agent" config.service_namespace = "agents" @@ -719,12 +720,14 @@ def test_config_defaults_mutation_and_repr(self): config.mark_projection = "tool" config.mark_exclude_names = ["custom.mark"] config.attribute_mappings = [{"key": "nemo_relay.model_name", "alias": "model.alias"}] + config.promote_metadata_prefixes = ["nv."] assert config.headers == {"authorization": "Bearer token"} assert config.resource_attributes == {"deployment.environment": "test"} assert config.mark_projection == "tool" assert config.mark_exclude_names == ["custom.mark"] assert config.attribute_mappings == [{"key": "nemo_relay.model_name", "alias": "model.alias"}] + assert config.promote_metadata_prefixes == ["nv."] assert "OpenTelemetryConfig" in repr(config) def test_config_rejects_invalid_map_values(self): @@ -739,6 +742,11 @@ def test_config_rejects_invalid_map_values(self): with pytest.raises(ValueError, match="attribute mapping key must not be blank"): config.attribute_mappings = [{"key": "", "alias": "x"}] + config.attribute_mappings = [] + config.promote_metadata_prefixes = ["nv.*"] + with pytest.raises(ValueError, match="literal prefix, not a glob"): + OpenTelemetrySubscriber(config) + def test_subscriber_lifecycle_and_invalid_transport(self): config = OpenTelemetryConfig("full", "http://localhost:4318/v1/traces") config.service_name = "py-agent" @@ -781,13 +789,19 @@ def test_subscriber_exports_scope_and_mark_events_end_to_end(self): source = "python-é" * 20 config = OpenTelemetryConfig("full", collector.endpoint) config.service_name = "py-agent" + config.promote_metadata_prefixes = ["nv."] subscriber = OpenTelemetrySubscriber(config) subscriber_name = f"py_otel_e2e_{uuid4().hex}" subscriber.register(subscriber_name) try: - handle = scope.push("otel_scope", ScopeType.Agent, data={"scope": True}) + handle = scope.push( + "otel_scope", + ScopeType.Agent, + data={"scope": True}, + metadata={"nv.binding": "python"}, + ) try: scope.event( "otel_mark", @@ -796,7 +810,7 @@ def test_subscriber_exports_scope_and_mark_events_end_to_end(self): metadata={"source": source}, ) finally: - scope.pop(handle) + scope.pop(handle, metadata={"nv.binding": "python"}) subscriber.force_flush() request = collector.wait_for_request() @@ -804,6 +818,7 @@ def test_subscriber_exports_scope_and_mark_events_end_to_end(self): assert request["headers"]["content-type"] == "application/x-protobuf" assert request["body"] assert b"nemo_relay.mark.metadata.source" in request["body"] + assert _otlp_string_attribute("nv.binding", "python") in request["body"] finally: subscriber.deregister(subscriber_name) subscriber.shutdown() From 6b4a42bcbd53512cd182b02e7c169cc3ba94faee Mon Sep 17 00:00:00 2001 From: Eric Evans <194135482+ericevans-nv@users.noreply.github.com> Date: Tue, 18 Aug 2026 17:28:41 -0500 Subject: [PATCH 2/5] fix(observability): protect reserved OTel attributes Signed-off-by: Eric Evans <194135482+ericevans-nv@users.noreply.github.com> --- crates/core/src/observability/mod.rs | 36 ++++++++++- .../attribute_projection_tests.rs | 59 +++++++++++++++++++ .../observability/opentelemetry.mdx | 9 +++ 3 files changed, 103 insertions(+), 1 deletion(-) diff --git a/crates/core/src/observability/mod.rs b/crates/core/src/observability/mod.rs index 462bd5908..526b32169 100644 --- a/crates/core/src/observability/mod.rs +++ b/crates/core/src/observability/mod.rs @@ -534,6 +534,30 @@ pub(crate) struct MetadataPromotionIssue { pub(crate) reason: &'static str, } +const RESERVED_OTEL_ATTRIBUTE_NAMESPACES: &[&str] = &[ + "error.", + "exception.", + "gen_ai.", + "input.", + "llm.", + "nemo_relay.", + "openinference.", + "output.", + "server.", + "service.", + "session.", + "tool.", + "tool_call.", + "user.", +]; + +fn is_reserved_otel_attribute_key(key: &str) -> bool { + key == "metadata" + || RESERVED_OTEL_ATTRIBUTE_NAMESPACES + .iter() + .any(|namespace| key.starts_with(namespace)) +} + /// Copies selected top-level Event metadata entries to typed OTLP attributes. /// /// Existing projection-owned attributes always win. Metadata is read without @@ -558,7 +582,17 @@ pub(crate) fn promote_event_metadata_attributes( .collect::>(); let mut issues = Vec::new(); for (key, value) in metadata { - if !prefixes.iter().any(|prefix| key.starts_with(prefix)) || existing_keys.contains(key) { + if !prefixes.iter().any(|prefix| key.starts_with(prefix)) { + continue; + } + if is_reserved_otel_attribute_key(key) { + issues.push(MetadataPromotionIssue { + key: key.clone(), + reason: "attribute key is reserved by Relay or an OpenTelemetry projection", + }); + continue; + } + if existing_keys.contains(key) { continue; } match metadata_value_to_otel(value) { diff --git a/crates/core/tests/unit/observability/attribute_projection_tests.rs b/crates/core/tests/unit/observability/attribute_projection_tests.rs index 9564490fd..6a12c70c7 100644 --- a/crates/core/tests/unit/observability/attribute_projection_tests.rs +++ b/crates/core/tests/unit/observability/attribute_projection_tests.rs @@ -232,6 +232,65 @@ fn promotes_matching_primitive_metadata_without_overwriting_owned_keys() { ); } +#[test] +fn rejects_metadata_keys_owned_by_relay_and_otel_projections() { + let reserved_keys = [ + "error.type", + "exception.type", + "gen_ai.request.model", + "input.value", + "llm.model_name", + "metadata", + "nemo_relay.uuid", + "openinference.span.kind", + "output.value", + "server.address", + "service.name", + "session.id", + "tool.name", + "tool_call.id", + "user.id", + ]; + let mut metadata = serde_json::Map::new(); + for key in reserved_keys { + metadata.insert(key.to_string(), serde_json::json!("blocked")); + } + metadata.insert("nv.source".to_string(), serde_json::json!("allowed")); + let event = Event::Mark(MarkEvent::new( + BaseEvent::builder() + .name("reserved-metadata-promotion") + .metadata(serde_json::Value::Object(metadata)) + .build(), + None, + None, + )); + let mut attributes = Vec::new(); + let prefixes = reserved_keys + .iter() + .copied() + .chain(std::iter::once("nv.")) + .map(str::to_string) + .collect::>(); + + let issues = + promote_event_metadata_attributes(&mut attributes, &event, &prefixes, &HashSet::new()); + + assert_eq!( + attributes, + vec![opentelemetry::KeyValue::new("nv.source", "allowed")] + ); + assert_eq!( + issues + .iter() + .map(|issue| issue.key.as_str()) + .collect::>(), + std::collections::HashSet::from(reserved_keys) + ); + assert!(issues.iter().all(|issue| { + issue.reason == "attribute key is reserved by Relay or an OpenTelemetry projection" + })); +} + #[test] fn reports_unsupported_metadata_array_shapes() { let event = Event::Mark(MarkEvent::new( diff --git a/docs/configure-plugins/observability/opentelemetry.mdx b/docs/configure-plugins/observability/opentelemetry.mdx index e9af34819..8ea56d230 100644 --- a/docs/configure-plugins/observability/opentelemetry.mdx +++ b/docs/configure-plugins/observability/opentelemetry.mdx @@ -143,6 +143,15 @@ to the other exporters. | `mark_projection` | `inherit` | Mark representation for `full` and `openinference`: `inherit`, `event`, or `tool`. | | `mark_exclude_names` | `["llm.chunk"]` | Mark names excluded from `full` and `openinference` projection. | | `attribute_mappings` | `[]` | `{ key, alias }` copies applied by `full` and `openinference` projection. | +| `promote_metadata_prefixes` | `[]` | Literal prefixes that select sanitized Event metadata to copy to top-level span attributes. | + +Metadata promotion preserves the original Event metadata and does not replace +attributes produced by a trace projection or `attribute_mappings`. Relay omits +selected keys in namespaces owned by Relay or supported semantic projections: +`nemo_relay.`, `gen_ai.`, `error.`, `exception.`, `input.`, `output.`, `llm.`, +`openinference.`, `server.`, `service.`, `session.`, `tool.`, `tool_call.`, and +`user.`. Relay also omits the bare `metadata` key. Rejected values produce a +rate-limited operational diagnostic without dropping the Event or span. ## Log and Metric Endpoint Resolution From c60b4ce71dfdaa3b20aaecee3a03b60e621c1ca8 Mon Sep 17 00:00:00 2001 From: Eric Evans <194135482+ericevans-nv@users.noreply.github.com> Date: Tue, 18 Aug 2026 18:05:26 -0500 Subject: [PATCH 3/5] test(observability): cover unsupported scope-end metadata Document Scope-end metadata as authoritative and verify that unsupported final values omit the promoted attribute instead of restoring a stale Scope-start value. Signed-off-by: Eric Evans <194135482+ericevans-nv@users.noreply.github.com> --- crates/core/src/observability/otel.rs | 5 +- .../tests/unit/observability/otel_tests.rs | 49 +++++++++++++++++++ 2 files changed, 52 insertions(+), 2 deletions(-) diff --git a/crates/core/src/observability/otel.rs b/crates/core/src/observability/otel.rs index 0c2d08c11..59b3908f4 100644 --- a/crates/core/src/observability/otel.rs +++ b/crates/core/src/observability/otel.rs @@ -1342,8 +1342,9 @@ impl OtelEventProcessor { &self.attribute_mappings, )); } - // Preserve every projection-owned start/end key while promoting the - // final metadata carried by the scope-end Event. + // Scope-end key presence is authoritative even when the final value + // cannot be represented as an OTLP attribute. Do not restore a stale + // promoted value retained from scope start. active_span.projection_attribute_keys.extend( attributes .iter() diff --git a/crates/core/tests/unit/observability/otel_tests.rs b/crates/core/tests/unit/observability/otel_tests.rs index 26e6ac2ad..a5838af20 100644 --- a/crates/core/tests/unit/observability/otel_tests.rs +++ b/crates/core/tests/unit/observability/otel_tests.rs @@ -667,6 +667,55 @@ fn promotes_final_scope_metadata_across_trace_projections() { } } +#[test] +fn omits_scope_metadata_when_final_value_is_unsupported_across_trace_projections() { + for otel_type in [ + OpenTelemetryType::Full, + OpenTelemetryType::GenAi, + OpenTelemetryType::OpenInference, + ] { + let (provider, exporter) = make_provider(); + let runtime_diagnostics = SignalRuntimeDiagnostics::new(None); + let mut processor = + OtelEventProcessor::new_with_mark_projection_and_exclusions_and_mappings_and_runtime_diagnostics( + provider, + "unsupported-final-metadata-promotion-test".into(), + otel_type, + MarkProjection::default(), + default_mark_exclude_names(), + Vec::new(), + vec!["nv.".to_string()], + runtime_diagnostics.clone(), + ); + let uuid = Uuid::now_v7(); + processor.process(&make_start_event_with_metadata( + uuid, + None, + "unsupported-final-metadata-promotion-scope", + json!({"nv.source": "start"}), + )); + processor.process(&make_end_event_with_metadata( + uuid, + None, + "unsupported-final-metadata-promotion-scope", + ScopeType::Agent, + json!({"nv.source": {"unsupported": true}}), + )); + processor.force_flush().unwrap(); + + let spans = exporter.get_finished_spans().unwrap(); + assert_eq!(spans.len(), 1); + assert!(!attr_map(&spans[0].attributes).contains_key("nv.source")); + + let diagnostics = runtime_diagnostics.snapshot(); + let diagnostic = diagnostics + .get("otel.metadata_promotion_value_unsupported") + .expect("unsupported final metadata diagnostic"); + assert_eq!(diagnostic.count, 1); + assert!(diagnostic.message.contains("nv.source")); + } +} + #[test] fn promotes_start_only_scope_metadata_across_trace_projections() { for otel_type in [ From fcb325d5de0534d45bd7f889f3b597b39a27dc5f Mon Sep 17 00:00:00 2001 From: Eric Evans <194135482+ericevans-nv@users.noreply.github.com> Date: Tue, 18 Aug 2026 18:53:26 -0500 Subject: [PATCH 4/5] test(observability): cover literal metadata prefix matching Signed-off-by: Eric Evans <194135482+ericevans-nv@users.noreply.github.com> --- .../attribute_projection_tests.rs | 35 +++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/crates/core/tests/unit/observability/attribute_projection_tests.rs b/crates/core/tests/unit/observability/attribute_projection_tests.rs index 6a12c70c7..8c86f93e4 100644 --- a/crates/core/tests/unit/observability/attribute_projection_tests.rs +++ b/crates/core/tests/unit/observability/attribute_projection_tests.rs @@ -232,6 +232,41 @@ fn promotes_matching_primitive_metadata_without_overwriting_owned_keys() { ); } +#[test] +fn treats_metadata_promotion_prefixes_as_literal_string_prefixes() { + let event = Event::Mark(MarkEvent::new( + BaseEvent::builder() + .name("literal-metadata-prefixes") + .metadata(serde_json::json!({ + "nv.dot": "dot", + "nv_underscore": "underscore", + "unrelated": "ignored", + "user_api_key": "api-key", + "username": "name" + })) + .build(), + None, + None, + )); + let mut attributes = Vec::new(); + + let issues = promote_event_metadata_attributes( + &mut attributes, + &event, + &["nv.".to_string(), "nv_".to_string(), "user".to_string()], + &HashSet::new(), + ); + + assert!(issues.is_empty()); + assert_eq!( + attributes + .iter() + .map(|attribute| attribute.key.as_str()) + .collect::>(), + HashSet::from(["nv.dot", "nv_underscore", "user_api_key", "username"]) + ); +} + #[test] fn rejects_metadata_keys_owned_by_relay_and_otel_projections() { let reserved_keys = [ From 496a8fe078b6d6b34ae89d6e470ed84d51ddc64d Mon Sep 17 00:00:00 2001 From: Eric Evans <194135482+ericevans-nv@users.noreply.github.com> Date: Tue, 18 Aug 2026 22:29:39 -0500 Subject: [PATCH 5/5] fix(observability): align metadata validation contracts Signed-off-by: Eric Evans <194135482+ericevans-nv@users.noreply.github.com> --- crates/core/src/api/event.rs | 9 ++++ crates/core/src/api/runtime/state.rs | 21 +++++----- crates/core/src/observability/mod.rs | 11 ++--- .../attribute_projection_tests.rs | 41 ++++++++++++++++--- crates/core/tests/unit/runtime_state_tests.rs | 10 +++++ 5 files changed, 67 insertions(+), 25 deletions(-) diff --git a/crates/core/src/api/event.rs b/crates/core/src/api/event.rs index 520509ced..283abb3a4 100644 --- a/crates/core/src/api/event.rs +++ b/crates/core/src/api/event.rs @@ -12,6 +12,15 @@ use nemo_relay_types::codec::response::AnnotatedLlmResponse; use crate::codec::resolve; +pub(crate) fn is_valid_event_metadata_attribute_key(key: &str) -> bool { + key.split('.').all(|segment| { + !segment.is_empty() + && segment.chars().all(|character| { + character.is_ascii_alphanumeric() || matches!(character, '_' | '-') + }) + }) +} + /// Core-only normalized LLM accessors for ATOF events. /// /// These helpers use built-in codec resolution, so they live in the runtime diff --git a/crates/core/src/api/runtime/state.rs b/crates/core/src/api/runtime/state.rs index 34b24e745..e5878e35e 100644 --- a/crates/core/src/api/runtime/state.rs +++ b/crates/core/src/api/runtime/state.rs @@ -21,8 +21,8 @@ use futures_util::{FutureExt, Stream}; use crate::api::event::{ BaseEvent, CategoryProfile, Event, EventCategory, EventSanitizeFields, MarkEvent, - ScopeCategory, ScopeEvent, llm_attributes_to_strings, scope_attributes_to_strings, - tool_attributes_to_strings, + ScopeCategory, ScopeEvent, is_valid_event_metadata_attribute_key, llm_attributes_to_strings, + scope_attributes_to_strings, tool_attributes_to_strings, }; use crate::api::llm::{CreateLlmHandleParams, EndLlmHandleParams}; use crate::api::llm::{LlmHandle, LlmRequest}; @@ -1785,27 +1785,26 @@ fn validate_event_metadata_attributes(attributes: &BTreeMap) -> Re Ok(()) } -fn is_valid_event_metadata_attribute_key(key: &str) -> bool { - key.split('.').all(|segment| { - !segment.is_empty() - && segment.chars().all(|character| { - character.is_ascii_alphanumeric() || matches!(character, '_' | '-') - }) - }) +fn is_otel_compatible_attribute_number(value: &serde_json::Number) -> bool { + if let Some(value) = value.as_u64() { + return i64::try_from(value).is_ok(); + } + value.as_i64().is_some() || value.as_f64().is_some() } fn is_otel_compatible_attribute_value(value: &Json) -> bool { fn primitive_kind(value: &Json) -> Option { match value { Json::Bool(_) => Some(0), - Json::Number(_) => Some(1), + Json::Number(value) if is_otel_compatible_attribute_number(value) => Some(1), Json::String(_) => Some(2), _ => None, } } match value { - Json::Bool(_) | Json::Number(_) | Json::String(_) => true, + Json::Bool(_) | Json::String(_) => true, + Json::Number(value) => is_otel_compatible_attribute_number(value), Json::Array(values) => match values.first().and_then(primitive_kind) { None => values.is_empty(), Some(kind) => values diff --git a/crates/core/src/observability/mod.rs b/crates/core/src/observability/mod.rs index 526b32169..8040e8f3a 100644 --- a/crates/core/src/observability/mod.rs +++ b/crates/core/src/observability/mod.rs @@ -3,7 +3,7 @@ //! Optional observability integrations for NeMo Relay Core. -use crate::api::event::EventNormalizationExt; +use crate::api::event::{EventNormalizationExt, is_valid_event_metadata_attribute_key}; use crate::codec::response::{AnnotatedLlmResponse, ApiSpecificResponse, Usage}; use serde::{Deserialize, Serialize}; @@ -371,12 +371,7 @@ pub fn validate_metadata_promotion_prefixes( fn is_valid_metadata_promotion_prefix(prefix: &str) -> bool { let key = prefix.strip_suffix('.').unwrap_or(prefix); - key.split('.').all(|segment| { - !segment.is_empty() - && segment.chars().all(|character| { - character.is_ascii_alphanumeric() || matches!(character, '_' | '-') - }) - }) + is_valid_event_metadata_attribute_key(key) } fn is_blank_attribute_mapping_name(value: &str) -> bool { @@ -649,7 +644,7 @@ fn metadata_array_to_otel( use opentelemetry::{Array, Value}; let Some(first) = values.first() else { - return Err("empty arrays do not declare an OTLP element type"); + return Ok(Value::Array(Array::String(Vec::new()))); }; match first { crate::json::Json::String(_) => values diff --git a/crates/core/tests/unit/observability/attribute_projection_tests.rs b/crates/core/tests/unit/observability/attribute_projection_tests.rs index 8c86f93e4..2d7231581 100644 --- a/crates/core/tests/unit/observability/attribute_projection_tests.rs +++ b/crates/core/tests/unit/observability/attribute_projection_tests.rs @@ -327,15 +327,44 @@ fn rejects_metadata_keys_owned_by_relay_and_otel_projections() { } #[test] -fn reports_unsupported_metadata_array_shapes() { +fn promotes_empty_metadata_arrays() { + let event = Event::Mark(MarkEvent::new( + BaseEvent::builder() + .name("empty-metadata-array") + .metadata(serde_json::json!({"nv.empty": []})) + .build(), + None, + None, + )); + let mut attributes = Vec::new(); + + let issues = promote_event_metadata_attributes( + &mut attributes, + &event, + &["nv.".to_string()], + &HashSet::new(), + ); + + assert!(issues.is_empty()); + assert_eq!( + attributes, + vec![opentelemetry::KeyValue::new( + "nv.empty", + opentelemetry::Value::Array(opentelemetry::Array::String(Vec::new())), + )] + ); +} + +#[test] +fn reports_unsupported_metadata_values() { let event = Event::Mark(MarkEvent::new( BaseEvent::builder() .name("metadata-array-rejections") .metadata(serde_json::json!({ - "nv.empty": [], "nv.mixed": [1, "two"], "nv.nested": [[1]], "nv.nulls": [null], + "nv.oversized_scalar": 18446744073709551615u64, "nv.oversized": [18446744073709551615u64] })) .build(), @@ -357,10 +386,6 @@ fn reports_unsupported_metadata_array_shapes() { .map(|issue| (issue.key, issue.reason)) .collect::>(); assert_eq!(issues.len(), 5); - assert_eq!( - issues.get("nv.empty"), - Some(&"empty arrays do not declare an OTLP element type") - ); assert_eq!( issues.get("nv.mixed"), Some(&"array values must have one primitive type") @@ -373,6 +398,10 @@ fn reports_unsupported_metadata_array_shapes() { issues.get("nv.nulls"), Some(&"arrays of null are not OTLP attributes") ); + assert_eq!( + issues.get("nv.oversized_scalar"), + Some(&"unsigned integer is larger than OTLP i64") + ); assert_eq!( issues.get("nv.oversized"), Some(&"array contains an unsigned integer larger than OTLP i64") diff --git a/crates/core/tests/unit/runtime_state_tests.rs b/crates/core/tests/unit/runtime_state_tests.rs index 6bda1aa33..a304d339e 100644 --- a/crates/core/tests/unit/runtime_state_tests.rs +++ b/crates/core/tests/unit/runtime_state_tests.rs @@ -33,6 +33,10 @@ async fn event_metadata_injection_accepts_flat_otel_values_and_empty_output() { ("experiment.variant".into(), json!("value")), ("nv.test.boolean".into(), json!(true)), ("nv.test.number".into(), json!(42)), + ( + "nv.test.max_unsigned_integer".into(), + json!(i64::MAX as u64), + ), ("nv.test.strings".into(), json!(["a", "b"])), ("nv.test.booleans".into(), json!([true, false])), ("nv.test.numbers".into(), json!([1, 2])), @@ -58,6 +62,10 @@ async fn event_metadata_injection_accepts_flat_otel_values_and_empty_output() { assert_eq!(metadata["experiment.variant"], json!("value")); assert_eq!(metadata["nv.test.boolean"], json!(true)); assert_eq!(metadata["nv.test.number"], json!(42)); + assert_eq!( + metadata["nv.test.max_unsigned_integer"], + json!(i64::MAX as u64) + ); assert_eq!(metadata["nv.test.strings"], json!(["a", "b"])); assert_eq!(metadata["nv.test.booleans"], json!([true, false])); assert_eq!(metadata["nv.test.numbers"], json!([1, 2])); @@ -84,6 +92,8 @@ async fn event_metadata_injection_rejects_invalid_output_atomically() { BTreeMap::from([("nv.test.object".into(), json!({"nested": true}))]), BTreeMap::from([("nv.test.nested_list".into(), json!([[1]]))]), BTreeMap::from([("nv.test.mixed_list".into(), json!([1, "two"]))]), + BTreeMap::from([("nv.test.oversized_number".into(), json!(u64::MAX))]), + BTreeMap::from([("nv.test.oversized_list".into(), json!([u64::MAX]))]), ]; for invalid_output in invalid_outputs {