From 0c33e81944142a3bbfb844807bf7c320ebfd8f50 Mon Sep 17 00:00:00 2001 From: Will Killian Date: Mon, 13 Jul 2026 20:54:33 -0400 Subject: [PATCH 1/8] feat(observability)!: project typed OTLP attributes Signed-off-by: Will Killian --- crates/cli/tests/coverage/plugins_tests.rs | 7 + crates/core/src/observability/mod.rs | 233 +++++++++++++++ .../core/src/observability/openinference.rs | 194 +++++++++---- crates/core/src/observability/otel.rs | 182 ++++++++---- .../src/observability/plugin_component.rs | 50 +++- .../tests/integration/middleware_tests.rs | 12 +- .../observability/exporter_parity_tests.rs | 61 ++-- .../unit/observability/openinference_tests.rs | 267 ++++++++++++------ .../tests/unit/observability/otel_tests.rs | 165 +++++++++-- .../observability/plugin_component_tests.rs | 39 +++ crates/ffi/nemo_relay.h | 40 +++ crates/ffi/src/api/observability.rs | 104 +++++++ crates/ffi/tests/unit/api/plugin_tests.rs | 81 ++++++ crates/node/observability.d.ts | 7 + crates/node/observability.js | 1 + crates/node/src/api/mod.rs | 31 ++ .../node/tests/observability_plugin_tests.mjs | 1 + crates/node/tests/openinference_tests.mjs | 8 + crates/node/tests/otel_tests.mjs | 8 + crates/python/src/py_types/observability.rs | 55 ++++ go/nemo_relay/nemo_relay.go | 36 ++- go/nemo_relay/observability_plugin.go | 1 + go/nemo_relay/openinference_test.go | 4 + go/nemo_relay/otel_test.go | 4 + python/nemo_relay/_native.pyi | 16 ++ python/nemo_relay/observability.py | 2 + python/nemo_relay/observability.pyi | 1 + python/tests/test_observability_plugin.py | 1 + python/tests/test_types.py | 10 + 29 files changed, 1369 insertions(+), 252 deletions(-) diff --git a/crates/cli/tests/coverage/plugins_tests.rs b/crates/cli/tests/coverage/plugins_tests.rs index 493c487b6..cfeb2c83f 100644 --- a/crates/cli/tests/coverage/plugins_tests.rs +++ b/crates/cli/tests/coverage/plugins_tests.rs @@ -200,6 +200,13 @@ fn typed_editor_model_contains_observability_sections() { .iter() .any(|field| field.name == "endpoint") ); + let attribute_mappings = openinference.field("attribute_mappings").unwrap(); + assert_eq!(attribute_mappings.kind, EditorFieldKind::List); + let mapping = attribute_mappings.list_item.unwrap(); + assert_eq!(mapping.kind, EditorFieldKind::Section); + let mapping_schema = mapping.schema.unwrap()(); + assert_eq!(mapping_schema.fields[0].name, "key"); + assert_eq!(mapping_schema.fields[1].name, "alias"); } #[test] diff --git a/crates/core/src/observability/mod.rs b/crates/core/src/observability/mod.rs index 6884cbd96..a518bff35 100644 --- a/crates/core/src/observability/mod.rs +++ b/crates/core/src/observability/mod.rs @@ -6,6 +6,29 @@ use crate::api::event::EventNormalizationExt; use serde::{Deserialize, Serialize}; +/// Copies a projected OTLP attribute to a second attribute name. +/// +/// `key` names the fully-qualified projected attribute and `alias` names the +/// additional attribute to emit with the same typed value. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "schema", derive(schemars::JsonSchema))] +pub struct OtlpAttributeMapping { + /// Fully-qualified projected attribute to copy. + pub key: String, + /// Additional attribute name receiving the copied value. + pub alias: String, +} + +impl OtlpAttributeMapping { + /// Creates an attribute mapping. + pub fn new(key: impl Into, alias: impl Into) -> Self { + Self { + key: key.into(), + alias: alias.into(), + } + } +} + #[cfg(test)] use std::sync::Mutex; @@ -49,6 +72,137 @@ pub(crate) fn default_mark_exclude_names() -> Vec { vec!["llm.chunk".to_string()] } +/// Validates OTLP attribute mappings shared by exporter configuration surfaces. +pub fn validate_attribute_mappings( + mappings: &[OtlpAttributeMapping], +) -> std::result::Result<(), String> { + let mut aliases = std::collections::HashSet::new(); + for mapping in mappings { + if mapping.key.trim().is_empty() { + return Err("attribute mapping key must not be blank".to_string()); + } + if mapping.alias.trim().is_empty() { + return Err("attribute mapping alias must not be blank".to_string()); + } + if !aliases.insert(mapping.alias.as_str()) { + return Err(format!( + "attribute mapping alias {:?} is duplicated", + mapping.alias + )); + } + } + Ok(()) +} + +#[cfg(any(feature = "otel", feature = "openinference"))] +/// Projects only top-level JSON fields as OTLP attributes. +/// +/// Nested objects and arrays remain JSON strings so arbitrary payloads do not +/// create ambiguous dotted attribute paths or unbounded attribute sets. +pub(crate) fn push_top_level_json_attributes( + attributes: &mut Vec, + prefix: &str, + value: Option<&crate::json::Json>, +) { + let Some(value) = value else { + return; + }; + match value { + crate::json::Json::Object(values) => { + for (field, value) in values { + push_top_level_json_value(attributes, &format!("{prefix}.{field}"), value); + } + } + value => push_top_level_json_value(attributes, prefix, value), + } +} + +#[cfg(any(feature = "otel", feature = "openinference"))] +/// Serializes a value and projects its top-level JSON fields as OTLP attributes. +pub(crate) fn push_serialized_top_level_attributes( + attributes: &mut Vec, + prefix: &str, + value: Option<&T>, +) { + let Some(value) = value else { + return; + }; + if let Ok(value) = serde_json::to_value(value) { + push_top_level_json_attributes(attributes, prefix, Some(&value)); + } +} + +#[cfg(any(feature = "otel", feature = "openinference"))] +fn push_top_level_json_value( + attributes: &mut Vec, + key: &str, + value: &crate::json::Json, +) { + use opentelemetry::KeyValue; + + match value { + crate::json::Json::Null => {} + crate::json::Json::Bool(value) => attributes.push(KeyValue::new(key.to_string(), *value)), + crate::json::Json::String(value) => { + attributes.push(KeyValue::new(key.to_string(), value.clone())) + } + crate::json::Json::Number(value) => { + if let Some(value) = value.as_i64() { + attributes.push(KeyValue::new(key.to_string(), value)); + } else if let Some(value) = value.as_u64() { + if let Ok(value) = i64::try_from(value) { + attributes.push(KeyValue::new(key.to_string(), value)); + } else { + attributes.push(KeyValue::new(key.to_string(), value.to_string())); + } + } else if let Some(value) = value.as_f64() { + attributes.push(KeyValue::new(key.to_string(), value)); + } + } + crate::json::Json::Array(_) | crate::json::Json::Object(_) => { + if let Ok(value) = serde_json::to_string(value) { + attributes.push(KeyValue::new(key.to_string(), value)); + } + } + } +} + +#[cfg(any(feature = "otel", feature = "openinference"))] +pub(crate) fn apply_attribute_mappings( + attributes: &mut Vec, + mappings: &[OtlpAttributeMapping], +) { + attributes.extend(attribute_mapping_aliases(attributes, mappings)); +} + +/// Resolves typed aliases from a complete set of projected attributes. +/// +/// Callers that project a span across multiple lifecycle events must pass every +/// real span attribute so projected fields always take precedence over aliases. +#[cfg(any(feature = "otel", feature = "openinference"))] +pub(crate) fn attribute_mapping_aliases( + projected_attributes: &[opentelemetry::KeyValue], + mappings: &[OtlpAttributeMapping], +) -> Vec { + let existing = projected_attributes + .iter() + .map(|attribute| attribute.key.as_str().to_string()) + .collect::>(); + mappings + .iter() + .filter(|mapping| !existing.contains(mapping.alias.as_str())) + .filter_map(|mapping| { + projected_attributes + .iter() + .rev() + .find(|attribute| attribute.key.as_str() == mapping.key) + .map(|attribute| { + opentelemetry::KeyValue::new(mapping.alias.clone(), attribute.value.clone()) + }) + }) + .collect() +} + /// Returns whether a mark matches a configured projection exclusion. /// /// Agent hook adapters may preserve the canonical event name in metadata while @@ -200,3 +354,82 @@ where }; span.set_status(status); } + +#[cfg(all(test, any(feature = "otel", feature = "openinference")))] +mod attribute_projection_tests { + use super::{OtlpAttributeMapping, apply_attribute_mappings, push_top_level_json_attributes}; + + #[test] + fn projects_typed_json_and_copies_configured_aliases() { + let mut attributes = Vec::new(); + push_top_level_json_attributes( + &mut attributes, + "nemo_relay.start.metadata", + Some(&serde_json::json!({ + "tenant": "acme", + "attempt": 2, + "tags": ["a", "b"], + "context": {"region": "us-east-1"}, + "request": {"id": "nested-id"}, + "request.id": "flat-id", + "event_id": 18446744073709551615u64 + })), + ); + apply_attribute_mappings( + &mut attributes, + &[OtlpAttributeMapping::new( + "nemo_relay.start.metadata.tenant", + "tenant.id", + )], + ); + + let values = attributes + .iter() + .map(|attribute| (attribute.key.as_str(), attribute.value.to_string())) + .collect::>(); + assert_eq!( + values.get("nemo_relay.start.metadata.tenant"), + Some(&"acme".to_string()) + ); + assert_eq!( + values.get("nemo_relay.start.metadata.attempt"), + Some(&"2".to_string()) + ); + assert_eq!( + values.get("nemo_relay.start.metadata.tags"), + Some(&"[\"a\",\"b\"]".to_string()) + ); + assert_eq!( + values.get("nemo_relay.start.metadata.context"), + Some(&"{\"region\":\"us-east-1\"}".to_string()) + ); + assert_eq!( + values.get("nemo_relay.start.metadata.request"), + Some(&"{\"id\":\"nested-id\"}".to_string()) + ); + assert_eq!( + values.get("nemo_relay.start.metadata.request.id"), + Some(&"flat-id".to_string()) + ); + assert_eq!( + values.get("nemo_relay.start.metadata.event_id"), + Some(&"18446744073709551615".to_string()) + ); + assert_eq!(values.get("tenant.id"), Some(&"acme".to_string())); + assert!(!values.contains_key("nemo_relay.start.metadata_json")); + } + + #[test] + fn rejects_invalid_attribute_mappings() { + assert!( + super::validate_attribute_mappings(&[OtlpAttributeMapping::new("", "alias")]).is_err() + ); + assert!( + super::validate_attribute_mappings(&[ + OtlpAttributeMapping::new("one", "duplicate"), + OtlpAttributeMapping::new("two", "duplicate"), + ]) + .is_err() + ); + } +} diff --git a/crates/core/src/observability/openinference.rs b/crates/core/src/observability/openinference.rs index 37aa8ea16..28faede98 100644 --- a/crates/core/src/observability/openinference.rs +++ b/crates/core/src/observability/openinference.rs @@ -21,9 +21,11 @@ use std::sync::{Arc, Mutex}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use super::{ - MarkProjection, default_mark_exclude_names, effective_mark_projection, - estimate_cost_for_response_or_model, estimate_cost_for_response_or_requested_model, manual, - merge_usage, model_name_for_llm_event, + MarkProjection, OtlpAttributeMapping, apply_attribute_mappings, attribute_mapping_aliases, + default_mark_exclude_names, effective_mark_projection, estimate_cost_for_response_or_model, + estimate_cost_for_response_or_requested_model, manual, merge_usage, model_name_for_llm_event, + push_serialized_top_level_attributes, push_top_level_json_attributes, + validate_attribute_mappings, }; use crate::api::event::{Event, EventNormalizationExt, ScopeCategory}; use crate::api::runtime::EventSubscriberFn; @@ -77,6 +79,9 @@ pub enum OpenInferenceError { /// The underlying tracer provider returned an error. #[error("OpenInference tracer provider error: {0}")] Provider(String), + /// Attribute mapping configuration was invalid. + #[error("invalid attribute mappings: {0}")] + InvalidAttributeMappings(String), /// Registration errors from the core runtime. #[error(transparent)] Core(#[from] FlowError), @@ -104,6 +109,7 @@ pub struct OpenInferenceConfig { instrumentation_scope: String, mark_projection: MarkProjection, mark_exclude_names: Vec, + attribute_mappings: Vec, timeout: Duration, transport: OtlpTransport, } @@ -120,6 +126,7 @@ impl Default for OpenInferenceConfig { instrumentation_scope: "nemo-relay-openinference".to_string(), mark_projection: MarkProjection::default(), mark_exclude_names: default_mark_exclude_names(), + attribute_mappings: Vec::new(), timeout: Duration::from_secs(3), transport: OtlpTransport::HttpBinary, } @@ -207,6 +214,26 @@ impl OpenInferenceConfig { self.mark_exclude_names = names.into_iter().map(Into::into).collect(); self } + + /// Adds a typed attribute copy after event payload projection. + pub fn with_attribute_mapping( + mut self, + key: impl Into, + alias: impl Into, + ) -> Self { + self.attribute_mappings + .push(OtlpAttributeMapping::new(key, alias)); + self + } + + /// Replaces the configured typed attribute copies. + pub fn with_attribute_mappings(mut self, mappings: I) -> Self + where + I: IntoIterator, + { + self.attribute_mappings = mappings.into_iter().collect(); + self + } } /// OpenInference-backed NeMo Relay subscriber. @@ -227,6 +254,8 @@ impl OpenInferenceSubscriber { { return Err(OpenInferenceError::MissingTokioRuntime); } + validate_attribute_mappings(&config.attribute_mappings) + .map_err(OpenInferenceError::InvalidAttributeMappings)?; let provider = build_tracer_provider(&config)?; Ok(Self::from_tracer_provider_with_scope( @@ -234,6 +263,7 @@ impl OpenInferenceSubscriber { config.instrumentation_scope, config.mark_projection, config.mark_exclude_names, + config.attribute_mappings, )) } @@ -247,6 +277,7 @@ impl OpenInferenceSubscriber { instrumentation_scope.into(), MarkProjection::default(), default_mark_exclude_names(), + Vec::new(), ) } @@ -261,6 +292,7 @@ impl OpenInferenceSubscriber { instrumentation_scope.into(), mark_projection, default_mark_exclude_names(), + Vec::new(), ) } @@ -280,21 +312,45 @@ impl OpenInferenceSubscriber { instrumentation_scope.into(), mark_projection, mark_exclude_names.into_iter().map(Into::into).collect(), + Vec::new(), ) } + /// Builds a subscriber from a tracer provider with typed attribute copies. + pub fn from_tracer_provider_with_attribute_mappings( + provider: SdkTracerProvider, + instrumentation_scope: impl Into, + attribute_mappings: I, + ) -> Result + where + I: IntoIterator, + { + let attribute_mappings = attribute_mappings.into_iter().collect::>(); + validate_attribute_mappings(&attribute_mappings) + .map_err(OpenInferenceError::InvalidAttributeMappings)?; + Ok(Self::from_tracer_provider_with_scope( + provider, + instrumentation_scope.into(), + MarkProjection::default(), + default_mark_exclude_names(), + attribute_mappings, + )) + } + fn from_tracer_provider_with_scope( provider: SdkTracerProvider, instrumentation_scope: String, mark_projection: MarkProjection, mark_exclude_names: Vec, + attribute_mappings: Vec, ) -> Self { let processor = Arc::new(Mutex::new( - OpenInferenceEventProcessor::new_with_mark_projection_and_exclusions( + OpenInferenceEventProcessor::new_with_mark_projection_and_exclusions_and_mappings( provider, instrumentation_scope, mark_projection, mark_exclude_names, + attribute_mappings, ), )); let processor_for_callback = Arc::clone(&processor); @@ -442,6 +498,7 @@ fn build_grpc_metadata(headers: &HashMap) -> Result struct ActiveSpan { span: Span, span_context: SpanContext, + projected_attributes: Vec, } struct OpenInferenceEventProcessor { @@ -452,6 +509,7 @@ struct OpenInferenceEventProcessor { tracer: SdkTracer, mark_projection: MarkProjection, mark_exclude_names: Vec, + attribute_mappings: Vec, } impl OpenInferenceEventProcessor { @@ -474,11 +532,28 @@ impl OpenInferenceEventProcessor { ) } + #[cfg(test)] fn new_with_mark_projection_and_exclusions( provider: SdkTracerProvider, instrumentation_scope: String, mark_projection: MarkProjection, mark_exclude_names: Vec, + ) -> Self { + Self::new_with_mark_projection_and_exclusions_and_mappings( + provider, + instrumentation_scope, + mark_projection, + mark_exclude_names, + Vec::new(), + ) + } + + fn new_with_mark_projection_and_exclusions_and_mappings( + provider: SdkTracerProvider, + instrumentation_scope: String, + mark_projection: MarkProjection, + mark_exclude_names: Vec, + attribute_mappings: Vec, ) -> Self { let tracer = provider.tracer(instrumentation_scope); Self { @@ -489,6 +564,7 @@ impl OpenInferenceEventProcessor { tracer, mark_projection, mark_exclude_names, + attribute_mappings, } } @@ -520,10 +596,17 @@ impl OpenInferenceEventProcessor { .with_kind(span_kind(event)) .with_start_time(to_system_time(*event.timestamp())) .start_with_context(&self.tracer, &self.parent_context(event)); - span.set_attributes(start_attributes(event)); + let attributes = start_attributes(event); + span.set_attributes(attributes.clone()); let span_context = local_parent_span_context(span.span_context()); - self.active_spans - .insert(event.uuid(), ActiveSpan { span, span_context }); + self.active_spans.insert( + event.uuid(), + ActiveSpan { + span, + span_context, + projected_attributes: attributes, + }, + ); } fn process_end(&mut self, event: &Event) { @@ -532,7 +615,14 @@ impl OpenInferenceEventProcessor { }; self.record_completed_span_context(event.uuid(), active_span.span_context.clone()); super::set_span_status_from_event_metadata(&mut active_span.span, event); - active_span.span.set_attributes(end_attributes(event)); + let mut attributes = end_attributes(event); + let mut projected_attributes = active_span.projected_attributes; + projected_attributes.extend(attributes.iter().cloned()); + attributes.extend(attribute_mapping_aliases( + &projected_attributes, + &self.attribute_mappings, + )); + active_span.span.set_attributes(attributes); active_span .span .end_with_timestamp(to_system_time(*event.timestamp())); @@ -547,9 +637,13 @@ impl OpenInferenceEventProcessor { } let mark_name = event.name().to_string(); let timestamp = to_system_time(*event.timestamp()); - let attributes = mark_attributes(event); + let mut attributes = mark_attributes(event); - if let Some(parent_span) = self.find_parent_span_mut(event) { + if self.find_parent_span(event).is_some() { + apply_attribute_mappings(&mut attributes, &self.attribute_mappings); + let parent_span = self + .find_parent_span_mut(event) + .expect("parent span was present during mark projection"); parent_span .span .add_event_with_timestamp(mark_name, timestamp, attributes); @@ -562,13 +656,13 @@ impl OpenInferenceEventProcessor { .with_kind(SpanKind::Internal) .with_start_time(timestamp) .start_with_context(&self.tracer, &self.parent_context(event)); - let mut span_attributes = attributes; - span_attributes.push(KeyValue::new( + attributes.push(KeyValue::new( oi::OPENINFERENCE_SPAN_KIND, OpenInferenceSpanKind::Chain, )); - span_attributes.push(KeyValue::new("nemo_relay.mark.orphan", true)); - span.set_attributes(span_attributes); + attributes.push(KeyValue::new("nemo_relay.mark.orphan", true)); + apply_attribute_mappings(&mut attributes, &self.attribute_mappings); + span.set_attributes(attributes); span.end_with_timestamp(timestamp); } @@ -584,6 +678,7 @@ impl OpenInferenceEventProcessor { if orphan { attributes.push(KeyValue::new("nemo_relay.mark.orphan", true)); } + apply_attribute_mappings(&mut attributes, &self.attribute_mappings); let mut span = self .tracer @@ -682,29 +777,24 @@ fn start_attributes(event: &Event) -> Vec { let mut attributes = common_attributes(event); let is_llm = event .category() - .is_some_and(|category| category.as_str() == "llm"); + .is_some_and(|category| category.as_str() == "llm") + || semantic_scope_type(event) == Some(ScopeType::Llm); if is_llm { // Final span metadata should reflect the completed event, especially for mixed-fidelity // Hermes flows where the request can be exact but the terminal error is lossy. - attributes.retain(|attribute| attribute.key.as_str() != oi::METADATA.as_str()); - } - let handle_attributes = event.attributes(); - if handle_attributes.is_some_and(|attributes| !attributes.is_empty()) { - push_serialized( - &mut attributes, - "nemo_relay.handle_attributes_json", - handle_attributes, - ); + attributes.retain(|attribute| { + attribute.key.as_str() != oi::METADATA.as_str() + && !attribute.key.as_str().starts_with("openinference.metadata") + }); } - if event - .category() - .is_none_or(|category| category.as_str() != "llm") - { - push_serialized( + if !is_llm { + push_serialized_top_level_attributes( &mut attributes, - "nemo_relay.start.input_json", - event.input(), + "nemo_relay.handle_attributes", + event.attributes(), ); + push_top_level_json_attributes(&mut attributes, "nemo_relay.start.data", event.data()); + push_top_level_json_attributes(&mut attributes, "nemo_relay.start.input", event.input()); } if event .category() @@ -739,17 +829,15 @@ fn end_attributes(event: &Event) -> Vec { let mut attributes = Vec::new(); let is_llm = event .category() - .is_some_and(|category| category.as_str() == "llm"); + .is_some_and(|category| category.as_str() == "llm") + || semantic_scope_type(event) == Some(ScopeType::Llm); + push_top_level_json_attributes(&mut attributes, "nemo_relay.end.data", event.data()); if let Some(metadata) = event.metadata().and_then(to_json_string) { attributes.push(KeyValue::new(oi::METADATA, metadata)); } - - push_serialized( - &mut attributes, - "nemo_relay.end.output_json", - event.output(), - ); + push_top_level_json_attributes(&mut attributes, "openinference.metadata", event.metadata()); + push_top_level_json_attributes(&mut attributes, "nemo_relay.end.output", event.output()); if let Some((output, mime_type)) = openinference_output_value(event) { attributes.push(KeyValue::new(oi::output::VALUE, output)); attributes.push(KeyValue::new(oi::output::MIME_TYPE, mime_type)); @@ -1357,7 +1445,6 @@ fn cost_total_from_llm_event( } fn mark_attributes(event: &Event) -> Vec { - let handle_attributes = event.attributes(); let mut attributes = vec![ KeyValue::new("nemo_relay.mark.uuid", event.uuid().to_string()), KeyValue::new( @@ -1368,15 +1455,15 @@ fn mark_attributes(event: &Event) -> Vec { .unwrap_or_default(), ), ]; - push_serialized( + push_serialized_top_level_attributes( &mut attributes, - "nemo_relay.mark.attributes_json", - handle_attributes, + "nemo_relay.mark.attributes", + event.attributes(), ); - push_serialized(&mut attributes, "nemo_relay.mark.data_json", event.data()); - push_serialized( + push_top_level_json_attributes(&mut attributes, "nemo_relay.mark.data", event.data()); + push_top_level_json_attributes( &mut attributes, - "nemo_relay.mark.metadata_json", + "nemo_relay.mark.metadata", event.metadata(), ); if let Some(category) = event.category() { @@ -1385,9 +1472,9 @@ fn mark_attributes(event: &Event) -> Vec { category.as_str().to_string(), )); } - push_serialized( + push_serialized_top_level_attributes( &mut attributes, - "nemo_relay.mark.category_profile_json", + "nemo_relay.mark.category_profile", event.category_profile(), ); attributes @@ -1422,6 +1509,7 @@ fn common_attributes(event: &Event) -> Vec { if let Some(metadata) = event.metadata().and_then(to_json_string) { attributes.push(KeyValue::new(oi::METADATA, metadata)); } + push_top_level_json_attributes(&mut attributes, "openinference.metadata", event.metadata()); attributes } @@ -1442,18 +1530,6 @@ fn openinference_span_kind(scope_type: Option) -> OpenInferenceSpanKi } } -fn push_serialized( - attributes: &mut Vec, - key: &'static str, - value: Option<&T>, -) { - if let Some(value) = value - && let Ok(json) = serde_json::to_string(value) - { - attributes.push(KeyValue::new(key, json)); - } -} - fn openinference_input_value(event: &Event) -> Option<(String, &'static str)> { let input = event.input()?; diff --git a/crates/core/src/observability/otel.rs b/crates/core/src/observability/otel.rs index 9624b0527..1a99b4e23 100644 --- a/crates/core/src/observability/otel.rs +++ b/crates/core/src/observability/otel.rs @@ -21,9 +21,11 @@ use std::sync::{Arc, Mutex}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use super::{ - MarkProjection, 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, + MarkProjection, OtlpAttributeMapping, apply_attribute_mappings, attribute_mapping_aliases, + 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_top_level_json_attributes, + validate_attribute_mappings, }; use crate::api::event::{Event, EventNormalizationExt, ScopeCategory}; use crate::api::runtime::EventSubscriberFn; @@ -39,7 +41,6 @@ use opentelemetry::{Context, KeyValue}; use opentelemetry_otlp::{Protocol, SpanExporter, WithExportConfig, WithHttpConfig}; use opentelemetry_sdk::Resource; use opentelemetry_sdk::trace::{SdkTracer, SdkTracerProvider, Span}; -use serde::Serialize; use uuid::Uuid; const COMPLETED_SPAN_CONTEXT_LIMIT: usize = 4096; @@ -71,6 +72,9 @@ pub enum OpenTelemetryError { /// The underlying tracer provider returned an error. #[error("OpenTelemetry tracer provider error: {0}")] Provider(String), + /// Attribute mapping configuration was invalid. + #[error("invalid attribute mappings: {0}")] + InvalidAttributeMappings(String), /// Registration errors from the core runtime. #[error(transparent)] Core(#[from] FlowError), @@ -98,6 +102,7 @@ pub struct OpenTelemetryConfig { instrumentation_scope: String, mark_projection: MarkProjection, mark_exclude_names: Vec, + attribute_mappings: Vec, timeout: Duration, transport: OtlpTransport, } @@ -114,6 +119,7 @@ impl Default for OpenTelemetryConfig { instrumentation_scope: "nemo-relay-otel".to_string(), mark_projection: MarkProjection::default(), mark_exclude_names: default_mark_exclude_names(), + attribute_mappings: Vec::new(), timeout: Duration::from_secs(3), transport: OtlpTransport::HttpBinary, } @@ -202,6 +208,26 @@ impl OpenTelemetryConfig { self.mark_exclude_names = names.into_iter().map(Into::into).collect(); self } + + /// Adds a typed attribute copy after event payload projection. + pub fn with_attribute_mapping( + mut self, + key: impl Into, + alias: impl Into, + ) -> Self { + self.attribute_mappings + .push(OtlpAttributeMapping::new(key, alias)); + self + } + + /// Replaces the configured typed attribute copies. + pub fn with_attribute_mappings(mut self, mappings: I) -> Self + where + I: IntoIterator, + { + self.attribute_mappings = mappings.into_iter().collect(); + self + } } /// OpenTelemetry-backed NeMo Relay subscriber. @@ -222,6 +248,8 @@ impl OpenTelemetrySubscriber { { return Err(OpenTelemetryError::MissingTokioRuntime); } + validate_attribute_mappings(&config.attribute_mappings) + .map_err(OpenTelemetryError::InvalidAttributeMappings)?; let provider = build_tracer_provider(&config)?; Ok(Self::from_tracer_provider_with_scope( @@ -229,6 +257,7 @@ impl OpenTelemetrySubscriber { config.instrumentation_scope, config.mark_projection, config.mark_exclude_names, + config.attribute_mappings, )) } @@ -242,6 +271,7 @@ impl OpenTelemetrySubscriber { instrumentation_scope.into(), MarkProjection::default(), default_mark_exclude_names(), + Vec::new(), ) } @@ -256,6 +286,7 @@ impl OpenTelemetrySubscriber { instrumentation_scope.into(), mark_projection, default_mark_exclude_names(), + Vec::new(), ) } @@ -275,21 +306,45 @@ impl OpenTelemetrySubscriber { instrumentation_scope.into(), mark_projection, mark_exclude_names.into_iter().map(Into::into).collect(), + Vec::new(), ) } + /// Builds a subscriber from a tracer provider with typed attribute copies. + pub fn from_tracer_provider_with_attribute_mappings( + provider: SdkTracerProvider, + instrumentation_scope: impl Into, + attribute_mappings: I, + ) -> Result + where + I: IntoIterator, + { + let attribute_mappings = attribute_mappings.into_iter().collect::>(); + validate_attribute_mappings(&attribute_mappings) + .map_err(OpenTelemetryError::InvalidAttributeMappings)?; + Ok(Self::from_tracer_provider_with_scope( + provider, + instrumentation_scope.into(), + MarkProjection::default(), + default_mark_exclude_names(), + attribute_mappings, + )) + } + fn from_tracer_provider_with_scope( provider: SdkTracerProvider, instrumentation_scope: String, mark_projection: MarkProjection, mark_exclude_names: Vec, + attribute_mappings: Vec, ) -> Self { let processor = Arc::new(Mutex::new( - OtelEventProcessor::new_with_mark_projection_and_exclusions( + OtelEventProcessor::new_with_mark_projection_and_exclusions_and_mappings( provider, instrumentation_scope, mark_projection, mark_exclude_names, + attribute_mappings, ), )); let processor_for_callback = Arc::clone(&processor); @@ -436,6 +491,7 @@ fn build_grpc_metadata(headers: &HashMap) -> Result struct ActiveSpan { span: Span, span_context: SpanContext, + projected_attributes: Vec, } struct OtelEventProcessor { @@ -446,6 +502,7 @@ struct OtelEventProcessor { tracer: SdkTracer, mark_projection: MarkProjection, mark_exclude_names: Vec, + attribute_mappings: Vec, } impl OtelEventProcessor { @@ -468,11 +525,28 @@ impl OtelEventProcessor { ) } + #[cfg(test)] fn new_with_mark_projection_and_exclusions( provider: SdkTracerProvider, instrumentation_scope: String, mark_projection: MarkProjection, mark_exclude_names: Vec, + ) -> Self { + Self::new_with_mark_projection_and_exclusions_and_mappings( + provider, + instrumentation_scope, + mark_projection, + mark_exclude_names, + Vec::new(), + ) + } + + fn new_with_mark_projection_and_exclusions_and_mappings( + provider: SdkTracerProvider, + instrumentation_scope: String, + mark_projection: MarkProjection, + mark_exclude_names: Vec, + attribute_mappings: Vec, ) -> Self { let tracer = provider.tracer(instrumentation_scope); Self { @@ -483,6 +557,7 @@ impl OtelEventProcessor { tracer, mark_projection, mark_exclude_names, + attribute_mappings, } } @@ -514,10 +589,17 @@ impl OtelEventProcessor { .with_kind(span_kind(event)) .with_start_time(to_system_time(*event.timestamp())) .start_with_context(&self.tracer, &self.parent_context(event)); - span.set_attributes(start_attributes(event)); + let attributes = start_attributes(event); + span.set_attributes(attributes.clone()); let span_context = local_parent_span_context(span.span_context()); - self.active_spans - .insert(event.uuid(), ActiveSpan { span, span_context }); + self.active_spans.insert( + event.uuid(), + ActiveSpan { + span, + span_context, + projected_attributes: attributes, + }, + ); } fn process_end(&mut self, event: &Event) { @@ -527,7 +609,14 @@ impl OtelEventProcessor { self.record_completed_span_context(event.uuid(), active_span.span_context.clone()); super::set_span_status_from_event_metadata(&mut active_span.span, event); - active_span.span.set_attributes(end_attributes(event)); + let mut attributes = end_attributes(event); + let mut projected_attributes = active_span.projected_attributes; + projected_attributes.extend(attributes.iter().cloned()); + attributes.extend(attribute_mapping_aliases( + &projected_attributes, + &self.attribute_mappings, + )); + active_span.span.set_attributes(attributes); active_span .span .end_with_timestamp(to_system_time(*event.timestamp())); @@ -542,9 +631,13 @@ impl OtelEventProcessor { } let mark_name = event.name().to_string(); let timestamp = to_system_time(*event.timestamp()); - let attributes = mark_attributes(event); + let mut attributes = mark_attributes(event); - if let Some(parent_span) = self.find_parent_span_mut(event) { + if self.find_parent_span(event).is_some() { + apply_attribute_mappings(&mut attributes, &self.attribute_mappings); + let parent_span = self + .find_parent_span_mut(event) + .expect("parent span was present during mark projection"); parent_span .span .add_event_with_timestamp(mark_name, timestamp, attributes); @@ -557,9 +650,9 @@ impl OtelEventProcessor { .with_kind(SpanKind::Internal) .with_start_time(timestamp) .start_with_context(&self.tracer, &self.parent_context(event)); - let mut span_attributes = attributes; - span_attributes.push(KeyValue::new("nemo_relay.mark.orphan", true)); - span.set_attributes(span_attributes); + attributes.push(KeyValue::new("nemo_relay.mark.orphan", true)); + apply_attribute_mappings(&mut attributes, &self.attribute_mappings); + span.set_attributes(attributes); span.end_with_timestamp(timestamp); } @@ -572,6 +665,7 @@ impl OtelEventProcessor { if orphan { attributes.push(KeyValue::new("nemo_relay.mark.orphan", true)); } + apply_attribute_mappings(&mut attributes, &self.attribute_mappings); let mut span = self .tracer @@ -668,37 +762,26 @@ fn scope_type_name(scope_type: Option) -> &'static str { fn start_attributes(event: &Event) -> Vec { let mut attributes = common_attributes(event); - let handle_attributes = event.attributes(); - push_serialized( + push_serialized_top_level_attributes( &mut attributes, - "nemo_relay.handle_attributes_json", - handle_attributes, + "nemo_relay.handle_attributes", + event.attributes(), ); - push_serialized(&mut attributes, "nemo_relay.start.data_json", event.data()); - push_serialized( + push_top_level_json_attributes(&mut attributes, "nemo_relay.start.data", event.data()); + push_top_level_json_attributes( &mut attributes, - "nemo_relay.start.metadata_json", + "nemo_relay.start.metadata", event.metadata(), ); - push_serialized( - &mut attributes, - "nemo_relay.start.input_json", - event.input(), - ); + push_top_level_json_attributes(&mut attributes, "nemo_relay.start.input", event.input()); attributes } fn end_attributes(event: &Event) -> Vec { let mut attributes = Vec::new(); - push_serialized(&mut attributes, "nemo_relay.end.data_json", event.data()); - - let metadata = event.metadata(); - push_serialized(&mut attributes, "nemo_relay.end.metadata_json", metadata); - push_serialized( - &mut attributes, - "nemo_relay.end.output_json", - event.output(), - ); + push_top_level_json_attributes(&mut attributes, "nemo_relay.end.data", event.data()); + push_top_level_json_attributes(&mut attributes, "nemo_relay.end.metadata", event.metadata()); + push_top_level_json_attributes(&mut attributes, "nemo_relay.end.output", event.output()); if event .category() .is_some_and(|category| category.as_str() == "llm") @@ -894,7 +977,6 @@ fn cost_total_and_currency(cost: &CostEstimate) -> Option<(f64, String)> { } fn mark_attributes(event: &Event) -> Vec { - let handle_attributes = event.attributes(); let mut attributes = vec![ KeyValue::new("nemo_relay.mark.uuid", event.uuid().to_string()), KeyValue::new( @@ -905,15 +987,15 @@ fn mark_attributes(event: &Event) -> Vec { .unwrap_or_default(), ), ]; - push_serialized( + push_serialized_top_level_attributes( &mut attributes, - "nemo_relay.mark.attributes_json", - handle_attributes, + "nemo_relay.mark.attributes", + event.attributes(), ); - push_serialized(&mut attributes, "nemo_relay.mark.data_json", event.data()); - push_serialized( + push_top_level_json_attributes(&mut attributes, "nemo_relay.mark.data", event.data()); + push_top_level_json_attributes( &mut attributes, - "nemo_relay.mark.metadata_json", + "nemo_relay.mark.metadata", event.metadata(), ); if let Some(category) = event.category() { @@ -922,9 +1004,9 @@ fn mark_attributes(event: &Event) -> Vec { category.as_str().to_string(), )); } - push_serialized( + push_serialized_top_level_attributes( &mut attributes, - "nemo_relay.mark.category_profile_json", + "nemo_relay.mark.category_profile", event.category_profile(), ); attributes @@ -959,18 +1041,6 @@ fn common_attributes(event: &Event) -> Vec { attributes } -fn push_serialized( - attributes: &mut Vec, - key: &'static str, - value: Option<&T>, -) { - if let Some(value) = value - && let Ok(json) = serde_json::to_string(value) - { - attributes.push(KeyValue::new(key, json)); - } -} - fn local_parent_span_context(span_context: &SpanContext) -> SpanContext { SpanContext::new( span_context.trace_id(), diff --git a/crates/core/src/observability/plugin_component.rs b/crates/core/src/observability/plugin_component.rs index f90de36f5..f9765bb48 100644 --- a/crates/core/src/observability/plugin_component.rs +++ b/crates/core/src/observability/plugin_component.rs @@ -51,7 +51,9 @@ use crate::observability::openinference::{ use crate::observability::otel::{ OpenTelemetryConfig as CoreOpenTelemetryConfig, OpenTelemetrySubscriber, }; -use crate::observability::{MarkProjection, default_mark_exclude_names}; +use crate::observability::{ + MarkProjection, OtlpAttributeMapping, default_mark_exclude_names, validate_attribute_mappings, +}; use crate::plugin::{ ConfigDiagnostic, ConfigPolicy, DiagnosticLevel, Plugin, PluginComponentSpec, PluginError, PluginRegistration, PluginRegistrationContext, Result as PluginResult, UnsupportedBehavior, @@ -385,6 +387,9 @@ pub struct OtlpSectionConfig { /// Mark names excluded from tool projection. Defaults to `llm.chunk`. #[serde(default = "default_mark_exclude_names")] pub mark_exclude_names: Vec, + /// Typed projected attributes copied to aliases. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub attribute_mappings: Vec, /// OTLP transport: `http_binary` or `grpc`. #[serde(default = "default_otlp_transport")] #[cfg_attr(feature = "schema", schemars(schema_with = "otlp_transport_schema"))] @@ -421,6 +426,7 @@ impl Default for OtlpSectionConfig { enabled: false, mark_projection: MarkProjection::default(), mark_exclude_names: default_mark_exclude_names(), + attribute_mappings: Vec::new(), transport: default_otlp_transport(), endpoint: None, headers: HashMap::new(), @@ -497,11 +503,37 @@ crate::editor_config! { } } +crate::editor_config! { + impl OtlpAttributeMapping { + key => { label: "key", kind: String }, + alias => { label: "alias", kind: String }, + } +} + +fn otlp_attribute_mapping_editor_schema() -> &'static crate::config_editor::EditorSchema { + ::editor_schema() +} + +fn default_otlp_attribute_mapping() -> Json { + serde_json::to_value(OtlpAttributeMapping::new("", "")) + .expect("attribute mapping should serialize") +} + +static OTLP_ATTRIBUTE_MAPPING_LIST_ITEM: crate::config_editor::EditorListItemSpec = + crate::config_editor::EditorListItemSpec { + kind: crate::config_editor::EditorFieldKind::Section, + schema: Some(otlp_attribute_mapping_editor_schema), + default: Some(default_otlp_attribute_mapping), + tagged_union: None, + list_item: None, + }; + crate::editor_config! { impl OtlpSectionConfig { enabled => { label: "enabled", kind: Boolean }, mark_projection => { label: "mark_projection", kind: Enum, values: ["inherit", "event", "tool"] }, mark_exclude_names => { label: "mark_exclude_names", kind: Json }, + attribute_mappings => { label: "attribute_mappings", kind: List, list: &OTLP_ATTRIBUTE_MAPPING_LIST_ITEM }, transport => { label: "transport", kind: Enum, values: ["http_binary", "grpc"] }, endpoint => { label: "endpoint", kind: String, optional: true }, headers => { label: "headers", kind: StringMap }, @@ -1368,7 +1400,8 @@ fn build_otel_config(section: OtlpSectionConfig) -> PluginResult PluginResult( - attributes.get("nemo_relay.end.output_json").unwrap(), - ) - .unwrap(), - json!({"status": "done", "metrics": {"tokens": 42}}) + attributes.get("nemo_relay.end.output.status"), + Some(&"done".to_string()) + ); + assert_eq!( + attributes.get("nemo_relay.end.output.metrics"), + Some(&"{\"tokens\":42}".to_string()) ); } @@ -2857,6 +2957,10 @@ fn helper_functions_cover_additional_openinference_branches() { raw_model_attributes.get(oi::llm::MODEL_NAME.as_str()), Some(&"raw-model".to_string()) ); + assert_eq!( + llm_attributes.get("openinference.metadata.phase"), + Some(&"done".to_string()) + ); assert_eq!( llm_attributes.get(oi::METADATA.as_str()), Some(&"{\"phase\":\"done\"}".to_string()) @@ -2928,12 +3032,12 @@ fn helper_functions_cover_additional_openinference_branches() { )); let mark_attributes = attr_map(&mark_attributes(&mark)); assert_eq!( - mark_attributes.get("nemo_relay.mark.data_json"), - Some(&"{\"kind\":\"aux\"}".to_string()) + mark_attributes.get("nemo_relay.mark.data.kind"), + Some(&"aux".to_string()) ); assert_eq!( - mark_attributes.get("nemo_relay.mark.metadata_json"), - Some(&"{\"source\":\"unit\"}".to_string()) + mark_attributes.get("nemo_relay.mark.metadata.source"), + Some(&"unit".to_string()) ); let llm_with_scalar_input = make_start_event( @@ -3840,11 +3944,13 @@ fn hermes_exact_api_payloads_emit_openinference_text_usage_and_metadata() { attributes.get("llm.cost.total"), Some(&"0.0042".to_string()) ); - assert_attr_contains(&attributes, "metadata", "\"provider_payload_exact\":true"); - assert_attr_contains( - &attributes, - "metadata", - "\"fidelity_source\":\"hermes_api_hooks_sanitized\"", + assert_eq!( + attributes.get("openinference.metadata.provider_payload_exact"), + Some(&"true".to_string()) + ); + assert_eq!( + attributes.get("openinference.metadata.fidelity_source"), + Some(&"hermes_api_hooks_sanitized".to_string()) ); } @@ -3934,26 +4040,25 @@ fn hermes_api_request_error_emits_openinference_json_output_and_metadata() { Some(&"application/json".to_string()) ); assert_eq!( - serde_json::from_str::( - attributes.get("nemo_relay.end.output_json").unwrap(), - ) - .unwrap(), - json!({ - "status_code": 502, - "retry_count": 1, - "max_retries": 2, - "retryable": true, - "reason": "upstream", - "error": { - "type": "BadGateway", - "message": "gateway upstream error" - } - }) + attributes.get("nemo_relay.end.output.status_code"), + Some(&"502".to_string()) + ); + assert_eq!( + attributes.get("openinference.metadata.provider_payload_exact"), + Some(&"false".to_string()) + ); + assert_eq!( + attributes.get("openinference.metadata.fidelity_source"), + Some(&"hermes_api_hooks".to_string()) + ); + assert_attr_contains( + &attributes, + oi::METADATA.as_str(), + "\"provider_payload_exact\":false", ); - assert_attr_contains(&attributes, "metadata", "\"provider_payload_exact\":false"); assert_attr_contains( &attributes, - "metadata", + oi::METADATA.as_str(), "\"fidelity_source\":\"hermes_api_hooks\"", ); } diff --git a/crates/core/tests/unit/observability/otel_tests.rs b/crates/core/tests/unit/observability/otel_tests.rs index b764c29a6..3471266b6 100644 --- a/crates/core/tests/unit/observability/otel_tests.rs +++ b/crates/core/tests/unit/observability/otel_tests.rs @@ -584,6 +584,111 @@ fn subscriber_registration_and_provider_lifecycle_methods_work() { subscriber.shutdown().unwrap(); } +#[test] +fn mapped_aliases_are_typed_and_cannot_replace_projected_span_fields() { + let (provider, exporter) = make_provider(); + let subscriber = OpenTelemetrySubscriber::from_tracer_provider_with_attribute_mappings( + provider, + "mapping-scope", + [ + crate::observability::OtlpAttributeMapping::new( + "nemo_relay.start.data.tenant", + "tenant.id", + ), + crate::observability::OtlpAttributeMapping::new( + "nemo_relay.end.data.tenant", + "nemo_relay.start.data.tenant", + ), + crate::observability::OtlpAttributeMapping::new("missing.source", "ignored.alias"), + ], + ) + .unwrap(); + let callback = subscriber.subscriber(); + let uuid = Uuid::now_v7(); + callback(&make_start_event( + uuid, + None, + "mapped-scope", + ScopeType::Agent, + Some(json!({"tenant": 7})), + )); + callback(&make_end_event( + uuid, + None, + "mapped-scope", + ScopeType::Agent, + Some(json!({"tenant": 8})), + )); + subscriber.force_flush().unwrap(); + + let spans = exporter.get_finished_spans().unwrap(); + let span = spans + .iter() + .find(|span| span.name.as_ref() == "mapped-scope") + .unwrap(); + assert_eq!( + span.attributes + .iter() + .filter(|attribute| attribute.key.as_str() == "nemo_relay.start.data.tenant") + .count(), + 1 + ); + assert_eq!( + span.attributes + .iter() + .find(|attribute| attribute.key.as_str() == "tenant.id") + .map(|attribute| &attribute.value), + Some(&opentelemetry::Value::I64(7)) + ); + assert!( + !span + .attributes + .iter() + .any(|attribute| attribute.key.as_str() == "ignored.alias") + ); +} + +#[test] +fn mapped_orphan_mark_alias_cannot_replace_intrinsic_mark_fields() { + let (provider, exporter) = make_provider(); + let subscriber = OpenTelemetrySubscriber::from_tracer_provider_with_attribute_mappings( + provider, + "mapping-scope", + [crate::observability::OtlpAttributeMapping::new( + "nemo_relay.mark.data.value", + "nemo_relay.mark.orphan", + )], + ) + .unwrap(); + let callback = subscriber.subscriber(); + callback(&make_mark_event( + None, + "mapped-orphan-mark", + Some(json!({"value": "not-an-orphan-flag"})), + )); + subscriber.force_flush().unwrap(); + + let spans = exporter.get_finished_spans().unwrap(); + let span = spans + .iter() + .find(|span| span.name.as_ref() == "mark:mapped-orphan-mark") + .unwrap(); + assert_eq!( + span.attributes + .iter() + .filter(|attribute| attribute.key.as_str() == "nemo_relay.mark.orphan") + .count(), + 1 + ); + assert_eq!( + span.attributes + .iter() + .find(|attribute| attribute.key.as_str() == "nemo_relay.mark.orphan") + .map(|attribute| &attribute.value), + Some(&opentelemetry::Value::Bool(true)) + ); +} + #[test] fn registered_subscriber_emits_spans_for_scope_push_pop_and_marks() { let _guard = crate::observability::test_mutex().lock().unwrap(); @@ -634,22 +739,22 @@ fn registered_subscriber_emits_spans_for_scope_push_pop_and_marks() { let attributes = attr_map(&span.attributes); assert_eq!( - attributes.get("nemo_relay.start.data_json"), - Some(&"{\"task\":\"scope-start\"}".to_string()) + attributes.get("nemo_relay.start.data.task"), + Some(&"scope-start".to_string()) ); assert_eq!( - attributes.get("nemo_relay.start.metadata_json"), - Some(&"{\"phase\":\"start\"}".to_string()) + attributes.get("nemo_relay.start.metadata.phase"), + Some(&"start".to_string()) ); let event_attributes = attr_map(&span.events.events[0].attributes); assert_eq!( - event_attributes.get("nemo_relay.mark.data_json"), - Some(&"{\"step\":1}".to_string()) + event_attributes.get("nemo_relay.mark.data.step"), + Some(&"1".to_string()) ); assert_eq!( - event_attributes.get("nemo_relay.mark.metadata_json"), - Some(&"{\"source\":\"rust-test\"}".to_string()) + event_attributes.get("nemo_relay.mark.metadata.source"), + Some(&"rust-test".to_string()) ); } @@ -747,12 +852,12 @@ fn records_span_start_mark_and_end() { Some(&root_uuid.to_string()) ); assert_eq!( - attributes.get("nemo_relay.start.input_json"), - Some(&"{\"query\":\"hello\"}".to_string()) + attributes.get("nemo_relay.start.input.query"), + Some(&"hello".to_string()) ); assert_eq!( - attributes.get("nemo_relay.end.output_json"), - Some(&"{\"result\":\"ok\"}".to_string()) + attributes.get("nemo_relay.end.output.result"), + Some(&"ok".to_string()) ); } @@ -977,16 +1082,16 @@ fn tool_projection_emits_generic_mark_as_parented_zero_duration_span() { Some(&"tool".to_string()) ); assert_eq!( - attributes.get("nemo_relay.mark.data_json"), - Some(&"{\"count\":3}".to_string()) + attributes.get("nemo_relay.mark.data.count"), + Some(&"3".to_string()) ); assert_eq!( attributes.get("nemo_relay.mark.category"), Some(&"custom".to_string()) ); assert_eq!( - attributes.get("nemo_relay.mark.category_profile_json"), - Some(&"{\"subtype\":\"example.compaction\"}".to_string()) + attributes.get("nemo_relay.mark.category_profile.subtype"), + Some(&"example.compaction".to_string()) ); assert!(!attributes.contains_key("nemo_relay.mark.orphan")); } @@ -1240,6 +1345,12 @@ fn semantic_scope_type_and_span_kind_follow_event_variants() { ); assert_eq!(semantic_scope_type(&remote_tool), Some(ScopeType::Tool)); assert_eq!(span_kind(&remote_tool), SpanKind::Client); + let remote_tool_attributes = attr_map(&start_attributes(&remote_tool)); + assert_eq!( + remote_tool_attributes.get("nemo_relay.handle_attributes"), + Some(&"[\"remote\"]".to_string()) + ); + assert!(!remote_tool_attributes.contains_key("nemo_relay.handle_attributes_json")); let llm_event = make_end_event( Uuid::now_v7(), @@ -1328,7 +1439,7 @@ fn llm_end_emits_cost_only_no_token_or_gen_ai_attributes() { assert!(keys.iter().any(|k| k == "nemo_relay.llm.cost.currency")); assert!( keys.iter() - .all(|k| !k.to_ascii_lowercase().contains("token") && !k.starts_with("gen_ai")), + .all(|k| !k.starts_with("llm.token") && !k.starts_with("gen_ai")), "no token attributes expected on the LLM span: {keys:?}" ); } @@ -1453,12 +1564,12 @@ fn helper_functions_cover_additional_otel_branches() { let start_attributes = attr_map(&start_attributes(&tool_event)); assert_eq!( - start_attributes.get("nemo_relay.start.input_json"), - Some(&"{\"query\":\"hello\"}".to_string()) + start_attributes.get("nemo_relay.start.input.query"), + Some(&"hello".to_string()) ); assert_eq!( - start_attributes.get("nemo_relay.start.metadata_json"), - Some(&"{\"meta\":true}".to_string()) + start_attributes.get("nemo_relay.start.metadata.meta"), + Some(&"true".to_string()) ); let tool_end_attributes = attr_map(&end_attributes(&Event::Scope(ScopeEvent::new( @@ -1473,8 +1584,8 @@ fn helper_functions_cover_additional_otel_branches() { Some(CategoryProfile::builder().tool_call_id("call-456").build()), )))); assert_eq!( - tool_end_attributes.get("nemo_relay.end.output_json"), - Some(&"{\"result\":true}".to_string()) + tool_end_attributes.get("nemo_relay.end.data.result"), + Some(&"true".to_string()) ); { @@ -1767,12 +1878,12 @@ fn helper_functions_cover_additional_otel_branches() { )); let mark_attributes = attr_map(&mark_attributes(&mark)); assert_eq!( - mark_attributes.get("nemo_relay.mark.data_json"), - Some(&"{\"kind\":\"aux\"}".to_string()) + mark_attributes.get("nemo_relay.mark.data.kind"), + Some(&"aux".to_string()) ); assert_eq!( - mark_attributes.get("nemo_relay.mark.metadata_json"), - Some(&"{\"source\":\"unit\"}".to_string()) + mark_attributes.get("nemo_relay.mark.metadata.source"), + Some(&"unit".to_string()) ); let mut processor = OtelEventProcessor::new(make_provider().0, "test".into()); diff --git a/crates/core/tests/unit/observability/plugin_component_tests.rs b/crates/core/tests/unit/observability/plugin_component_tests.rs index 318b1b417..78aec9c21 100644 --- a/crates/core/tests/unit/observability/plugin_component_tests.rs +++ b/crates/core/tests/unit/observability/plugin_component_tests.rs @@ -81,6 +81,18 @@ fn editor_schema_tracks_observability_config_types() { .expect("openinference editor schema"); let headers = otlp.field("headers").expect("headers field"); assert_eq!(headers.kind, EditorFieldKind::StringMap); + + let attribute_mappings = otlp + .field("attribute_mappings") + .expect("attribute mappings field"); + assert_eq!(attribute_mappings.kind, EditorFieldKind::List); + let mapping = attribute_mappings + .list_item + .expect("attribute mapping list item"); + assert_eq!(mapping.kind, EditorFieldKind::Section); + let mapping_schema = mapping.schema.expect("attribute mapping schema")(); + assert_eq!(mapping_schema.fields[0].name, "key"); + assert_eq!(mapping_schema.fields[1].name, "alias"); } fn push_agent(name: &str) -> crate::api::scope::ScopeHandle { @@ -222,6 +234,7 @@ fn default_config_and_component_conversion_cover_public_shape() { assert!(!otlp.enabled); assert_eq!(otlp.mark_projection, MarkProjection::Inherit); assert_eq!(otlp.mark_exclude_names, vec!["llm.chunk"]); + assert!(otlp.attribute_mappings.is_empty()); assert_eq!(otlp.transport, "http_binary"); assert_eq!(otlp.service_name, "nemo-relay"); assert_eq!(otlp.timeout_millis, 3_000); @@ -242,6 +255,15 @@ fn default_config_and_component_conversion_cover_public_shape() { #[test] fn mark_projection_parses_for_otlp_and_rejects_unknown_values() { + let mappings: OtlpSectionConfig = serde_json::from_value(json!({ + "attribute_mappings": [{ + "key": "nemo_relay.start.metadata.tenant", + "alias": "tenant.id" + }] + })) + .unwrap(); + assert_eq!(mappings.attribute_mappings.len(), 1); + let otlp: OtlpSectionConfig = serde_json::from_value(json!({ "mark_projection": "tool" })) @@ -287,6 +309,22 @@ fn mark_projection_parses_for_otlp_and_rejects_unknown_values() { diagnostic.code == "observability.unknown_field" && diagnostic.field.as_deref() == Some("mark_projection") })); + + let report = validate_plugin_config(&plugin_config(json!({ + "openinference": { + "attribute_mappings": [ + {"key": "", "alias": "tenant.id"}, + {"key": "openinference.metadata.tenant", "alias": "tenant.id"} + ] + } + }))); + assert!(report.diagnostics.iter().any(|diagnostic| { + diagnostic.code == "observability.unsupported_value" + && diagnostic.field.as_deref() == Some("attribute_mappings") + && diagnostic + .message + .contains("attribute mapping key must not be blank") + })); } #[cfg(feature = "schema")] @@ -310,6 +348,7 @@ fn schema_contains_every_supported_observability_option() { "model_name", "mark_projection", "mark_exclude_names", + "attribute_mappings", "tool_definitions", "extra", "filename_template", diff --git a/crates/ffi/nemo_relay.h b/crates/ffi/nemo_relay.h index 6355d579c..e346db863 100644 --- a/crates/ffi/nemo_relay.h +++ b/crates/ffi/nemo_relay.h @@ -1321,6 +1321,26 @@ NemoRelayStatus nemo_relay_otel_subscriber_create(const char *transport, uint64_t timeout_millis, struct FfiOpenTelemetrySubscriber **out); +/** + * Creates a new OpenTelemetry subscriber with typed attribute mappings. + * + * `attribute_mappings_json` is a JSON array of `{ "key": string, "alias": string }` objects. + * + * # Safety + * Any non-null C strings must be valid and `out` must be non-null. + */ +NemoRelayStatus nemo_relay_otel_subscriber_create_with_attribute_mappings(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 *attribute_mappings_json, + struct FfiOpenTelemetrySubscriber **out); + /** * Registers the OpenTelemetry subscriber as an event subscriber. * @@ -1375,6 +1395,26 @@ NemoRelayStatus nemo_relay_openinference_subscriber_create(const char *transport uint64_t timeout_millis, struct FfiOpenInferenceSubscriber **out); +/** + * Creates a new OpenInference subscriber with typed attribute mappings. + * + * `attribute_mappings_json` is a JSON array of `{ "key": string, "alias": string }` objects. + * + * # Safety + * Any non-null C strings must be valid and `out` must be non-null. + */ +NemoRelayStatus nemo_relay_openinference_subscriber_create_with_attribute_mappings(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 *attribute_mappings_json, + struct FfiOpenInferenceSubscriber **out); + /** * Registers the OpenInference subscriber as an event subscriber. * diff --git a/crates/ffi/src/api/observability.rs b/crates/ffi/src/api/observability.rs index 2713c394e..b3476407b 100644 --- a/crates/ffi/src/api/observability.rs +++ b/crates/ffi/src/api/observability.rs @@ -16,6 +16,7 @@ type OpenTelemetryConfig = nemo_relay::observability::otel::OpenTelemetryConfig; type OpenTelemetrySubscriber = nemo_relay::observability::otel::OpenTelemetrySubscriber; type OpenInferenceConfig = nemo_relay::observability::openinference::OpenInferenceConfig; type OpenInferenceSubscriber = nemo_relay::observability::openinference::OpenInferenceSubscriber; +type OtlpAttributeMapping = nemo_relay::observability::OtlpAttributeMapping; type ObservabilityComponentSpec = nemo_relay::observability::plugin_component::ComponentSpec; type ObservabilityConfig = nemo_relay::observability::plugin_component::ObservabilityConfig; @@ -529,6 +530,25 @@ fn parse_string_map_json( Ok(out) } +fn parse_attribute_mappings_json( + json_ptr: *const c_char, +) -> Result, NemoRelayStatus> { + if json_ptr.is_null() { + return Ok(Vec::new()); + } + let json_string = c_str_to_string(json_ptr)?; + let mappings: Vec = + serde_json::from_str(&json_string).map_err(|error| { + set_last_error(&format!("invalid attribute_mappings JSON: {error}")); + NemoRelayStatus::InvalidJson + })?; + nemo_relay::observability::validate_attribute_mappings(&mappings).map_err(|error| { + set_last_error(&format!("invalid attribute_mappings: {error}")); + NemoRelayStatus::InvalidArg + })?; + Ok(mappings) +} + fn required_out_ptr(out: *mut *mut T) -> Result<(), NemoRelayStatus> { if out.is_null() { set_last_error("out pointer is null"); @@ -666,6 +686,43 @@ pub unsafe extern "C" fn nemo_relay_otel_subscriber_create( instrumentation_scope: *const c_char, timeout_millis: u64, out: *mut *mut FfiOpenTelemetrySubscriber, +) -> NemoRelayStatus { + unsafe { + nemo_relay_otel_subscriber_create_with_attribute_mappings( + transport, + endpoint, + headers_json, + resource_attributes_json, + service_name, + service_namespace, + service_version, + instrumentation_scope, + timeout_millis, + std::ptr::null(), + out, + ) + } +} + +/// Creates a new OpenTelemetry subscriber with typed attribute mappings. +/// +/// `attribute_mappings_json` is a JSON array of `{ "key": string, "alias": string }` objects. +/// +/// # Safety +/// Any non-null C strings must be valid and `out` must be non-null. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn nemo_relay_otel_subscriber_create_with_attribute_mappings( + 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, + attribute_mappings_json: *const c_char, + out: *mut *mut FfiOpenTelemetrySubscriber, ) -> NemoRelayStatus { clear_last_error(); if let Err(status) = required_out_ptr(out) { @@ -723,6 +780,11 @@ pub unsafe extern "C" fn nemo_relay_otel_subscriber_create( Ok(config) => config, Err(status) => return status, }; + let attribute_mappings = match parse_attribute_mappings_json(attribute_mappings_json) { + Ok(mappings) => mappings, + Err(status) => return status, + }; + config = config.with_attribute_mappings(attribute_mappings); config = match apply_string_map( config, resource_attributes_json, @@ -855,6 +917,43 @@ pub unsafe extern "C" fn nemo_relay_openinference_subscriber_create( instrumentation_scope: *const c_char, timeout_millis: u64, out: *mut *mut FfiOpenInferenceSubscriber, +) -> NemoRelayStatus { + unsafe { + nemo_relay_openinference_subscriber_create_with_attribute_mappings( + transport, + endpoint, + headers_json, + resource_attributes_json, + service_name, + service_namespace, + service_version, + instrumentation_scope, + timeout_millis, + std::ptr::null(), + out, + ) + } +} + +/// Creates a new OpenInference subscriber with typed attribute mappings. +/// +/// `attribute_mappings_json` is a JSON array of `{ "key": string, "alias": string }` objects. +/// +/// # Safety +/// Any non-null C strings must be valid and `out` must be non-null. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn nemo_relay_openinference_subscriber_create_with_attribute_mappings( + 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, + attribute_mappings_json: *const c_char, + out: *mut *mut FfiOpenInferenceSubscriber, ) -> NemoRelayStatus { clear_last_error(); if let Err(status) = required_out_ptr(out) { @@ -912,6 +1011,11 @@ pub unsafe extern "C" fn nemo_relay_openinference_subscriber_create( Ok(config) => config, Err(status) => return status, }; + let attribute_mappings = match parse_attribute_mappings_json(attribute_mappings_json) { + Ok(mappings) => mappings, + Err(status) => return status, + }; + config = config.with_attribute_mappings(attribute_mappings); config = match apply_string_map( config, resource_attributes_json, diff --git a/crates/ffi/tests/unit/api/plugin_tests.rs b/crates/ffi/tests/unit/api/plugin_tests.rs index 2d37db847..e58154f37 100644 --- a/crates/ffi/tests/unit/api/plugin_tests.rs +++ b/crates/ffi/tests/unit/api/plugin_tests.rs @@ -1299,6 +1299,87 @@ fn test_ffi_specialized_subscriber_and_exporter_default_and_invalid_name_paths() } } +#[test] +fn test_ffi_typed_attribute_mapping_constructors_validate_and_accept_mappings() { + let _lock = TEST_MUTEX.lock().unwrap_or_else(|e| e.into_inner()); + reset_globals(); + + unsafe { + let valid = cstring(r#"[{"key":"nemo_relay.start.data.tenant","alias":"tenant.id"}]"#); + let invalid = cstring(r#"[{"key":"","alias":"tenant.id"}]"#); + + let mut otel = ptr::null_mut(); + assert_eq!( + nemo_relay_otel_subscriber_create_with_attribute_mappings( + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + 0, + valid.as_ptr(), + &mut otel, + ), + NemoRelayStatus::Ok + ); + nemo_relay_otel_subscriber_free(otel); + assert_eq!( + nemo_relay_otel_subscriber_create_with_attribute_mappings( + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + 0, + invalid.as_ptr(), + &mut otel, + ), + NemoRelayStatus::InvalidArg + ); + + let mut openinference = ptr::null_mut(); + assert_eq!( + nemo_relay_openinference_subscriber_create_with_attribute_mappings( + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + 0, + valid.as_ptr(), + &mut openinference, + ), + NemoRelayStatus::Ok + ); + nemo_relay_openinference_subscriber_free(openinference); + assert_eq!( + nemo_relay_openinference_subscriber_create_with_attribute_mappings( + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + 0, + invalid.as_ptr(), + &mut openinference, + ), + NemoRelayStatus::InvalidArg + ); + } +} + #[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 eb15be134..3689f8f4b 100644 --- a/crates/node/observability.d.ts +++ b/crates/node/observability.d.ts @@ -58,6 +58,7 @@ export interface OtlpConfig { enabled?: boolean; mark_projection?: 'inherit' | 'event' | 'tool'; mark_exclude_names?: string[]; + attribute_mappings?: OtlpAttributeMapping[]; transport?: 'http_binary' | 'grpc' | string; endpoint?: string; headers?: Record; @@ -69,6 +70,12 @@ export interface OtlpConfig { timeout_millis?: number; } +/** Copy a projected OTLP attribute to an additional attribute name. */ +export interface OtlpAttributeMapping { + key: string; + alias: string; +} + export interface Config { version?: number; atof?: AtofConfig; diff --git a/crates/node/observability.js b/crates/node/observability.js index 97e5c8415..943c4913c 100644 --- a/crates/node/observability.js +++ b/crates/node/observability.js @@ -59,6 +59,7 @@ function otlpConfig(config = {}) { enabled: false, mark_projection: 'inherit', mark_exclude_names: ['llm.chunk'], + attribute_mappings: [], transport: 'http_binary', headers: {}, resource_attributes: {}, diff --git a/crates/node/src/api/mod.rs b/crates/node/src/api/mod.rs index 9f06b649f..a9df4c3f6 100644 --- a/crates/node/src/api/mod.rs +++ b/crates/node/src/api/mod.rs @@ -109,6 +109,22 @@ fn parse_string_map( Ok(out) } +fn parse_attribute_mappings( + value: Option>, +) -> napi::Result> { + let mappings = value + .unwrap_or_default() + .into_iter() + .map(|mapping| nemo_relay::observability::OtlpAttributeMapping { + key: mapping.key, + alias: mapping.alias, + }) + .collect::>(); + nemo_relay::observability::validate_attribute_mappings(&mappings) + .map_err(napi::Error::from_reason)?; + Ok(mappings) +} + fn otel_status_metadata(status_code: &'static str, status_message: Option) -> Json { let mut metadata = serde_json::Map::new(); metadata.insert( @@ -168,6 +184,7 @@ fn build_otel_config( for (key, value) in parse_string_map(options.resource_attributes, "resourceAttributes")? { config = config.with_resource_attribute(key, value); } + config = config.with_attribute_mappings(parse_attribute_mappings(options.attribute_mappings)?); Ok(config) } @@ -277,6 +294,7 @@ fn build_openinference_config( for (key, value) in parse_string_map(options.resource_attributes, "resourceAttributes")? { config = config.with_resource_attribute(key, value); } + config = config.with_attribute_mappings(parse_attribute_mappings(options.attribute_mappings)?); Ok(config) } @@ -3449,6 +3467,17 @@ pub struct OpenTelemetryConfig { pub instrumentation_scope: Option, /// Export timeout in milliseconds. Defaults to `3000`. pub timeout_millis: Option, + /// Typed projected attributes copied to aliases. + pub attribute_mappings: Option>, +} + +/// Typed projected attribute copy configuration. +#[napi(object)] +pub struct OtlpAttributeMapping { + /// Fully-qualified projected attribute to copy. + pub key: String, + /// Additional attribute name receiving the copied value. + pub alias: String, } /// Mutable configuration object for `OpenInferenceSubscriber`. @@ -3473,6 +3502,8 @@ pub struct OpenInferenceConfig { pub instrumentation_scope: Option, /// Export timeout in milliseconds. Defaults to `3000`. pub timeout_millis: Option, + /// Typed projected attributes copied to aliases. + pub attribute_mappings: Option>, } /// OpenTelemetry-backed event subscriber. diff --git a/crates/node/tests/observability_plugin_tests.mjs b/crates/node/tests/observability_plugin_tests.mjs index 94b825031..77e5bdd19 100644 --- a/crates/node/tests/observability_plugin_tests.mjs +++ b/crates/node/tests/observability_plugin_tests.mjs @@ -31,6 +31,7 @@ describe('observability plugin helpers', () => { enabled: false, mark_projection: 'inherit', mark_exclude_names: ['llm.chunk'], + attribute_mappings: [], transport: 'http_binary', headers: {}, resource_attributes: {}, diff --git a/crates/node/tests/openinference_tests.mjs b/crates/node/tests/openinference_tests.mjs index 8ab01a977..323f73c19 100644 --- a/crates/node/tests/openinference_tests.mjs +++ b/crates/node/tests/openinference_tests.mjs @@ -32,6 +32,7 @@ describe('OpenInferenceSubscriber', () => { resourceAttributes: { 'deployment.environment': 'test', }, + attributeMappings: [{ key: 'openinference.metadata.tenant', alias: 'tenant.id' }], }); const name = unique('node_openinference'); @@ -68,6 +69,13 @@ describe('OpenInferenceSubscriber', () => { }), /resourceAttributes must be an object of string values/i, ); + assert.throws( + () => + new OpenInferenceSubscriber({ + attributeMappings: [{ key: '', alias: 'tenant.id' }], + }), + /attribute mapping key must not be blank/i, + ); }); it('exports scope push/pop and mark events end to end', async () => { diff --git a/crates/node/tests/otel_tests.mjs b/crates/node/tests/otel_tests.mjs index f9228d843..3853ffaf4 100644 --- a/crates/node/tests/otel_tests.mjs +++ b/crates/node/tests/otel_tests.mjs @@ -28,6 +28,7 @@ describe('OpenTelemetrySubscriber', () => { resourceAttributes: { 'deployment.environment': 'test', }, + attributeMappings: [{ key: 'nemo_relay.start.data.tenant', alias: 'tenant.id' }], }); const name = uniqueId('node_otel'); @@ -64,6 +65,13 @@ describe('OpenTelemetrySubscriber', () => { }), /resourceAttributes must be an object of string values/i, ); + assert.throws( + () => + new OpenTelemetrySubscriber({ + attributeMappings: [{ key: '', alias: 'tenant.id' }], + }), + /attribute mapping key must not be blank/i, + ); }); it('exports scope push/pop and mark events end to end', async () => { diff --git a/crates/python/src/py_types/observability.rs b/crates/python/src/py_types/observability.rs index 1f7e33b0a..772800429 100644 --- a/crates/python/src/py_types/observability.rs +++ b/crates/python/src/py_types/observability.rs @@ -14,6 +14,21 @@ use super::{ FORCE_ATIF_EXPORT_JSON_SERIALIZATION_ERROR, FORCE_ATIF_EXPORT_VALUE_SERIALIZATION_ERROR, }; +fn py_attribute_mappings( + value: &Bound<'_, PyAny>, +) -> PyResult> { + let value = py_to_json(value)?; + let mappings: Vec = + serde_json::from_value(value).map_err(|error| { + pyo3::exceptions::PyValueError::new_err(format!( + "attribute_mappings must be a list of {{key: str, alias: str}} objects: {error}" + )) + })?; + nemo_relay::observability::validate_attribute_mappings(&mappings) + .map_err(pyo3::exceptions::PyValueError::new_err)?; + Ok(mappings) +} + // --------------------------------------------------------------------------- // AtifExporter // --------------------------------------------------------------------------- @@ -380,6 +395,7 @@ pub struct PyOpenTelemetryConfig { pub(crate) timeout_millis: u64, pub(crate) headers: HashMap, pub(crate) resource_attributes: HashMap, + pub(crate) attribute_mappings: Vec, } impl PyOpenTelemetryConfig { @@ -417,6 +433,7 @@ impl PyOpenTelemetryConfig { for (key, value) in &self.resource_attributes { config = config.with_resource_attribute(key.clone(), value.clone()); } + config = config.with_attribute_mappings(self.attribute_mappings.clone()); Ok(config) } } @@ -435,6 +452,7 @@ impl PyOpenTelemetryConfig { timeout_millis: 3_000, headers: HashMap::new(), resource_attributes: HashMap::new(), + attribute_mappings: Vec::new(), } } @@ -466,6 +484,23 @@ impl PyOpenTelemetryConfig { Ok(()) } + #[getter] + pub(crate) fn attribute_mappings(&self, py: Python<'_>) -> PyResult> { + json_to_py( + py, + &serde_json::to_value(&self.attribute_mappings).unwrap_or_default(), + ) + } + + #[setter] + pub(crate) fn set_attribute_mappings( + &mut self, + attribute_mappings: &Bound<'_, PyAny>, + ) -> PyResult<()> { + self.attribute_mappings = py_attribute_mappings(attribute_mappings)?; + Ok(()) + } + pub(crate) fn set_header(&mut self, key: String, value: String) { self.headers.insert(key, value); } @@ -559,6 +594,7 @@ pub struct PyOpenInferenceConfig { pub(crate) timeout_millis: u64, pub(crate) headers: HashMap, pub(crate) resource_attributes: HashMap, + pub(crate) attribute_mappings: Vec, } impl PyOpenInferenceConfig { @@ -596,6 +632,7 @@ impl PyOpenInferenceConfig { for (key, value) in &self.resource_attributes { config = config.with_resource_attribute(key.clone(), value.clone()); } + config = config.with_attribute_mappings(self.attribute_mappings.clone()); Ok(config) } } @@ -614,6 +651,7 @@ impl PyOpenInferenceConfig { timeout_millis: 3_000, headers: HashMap::new(), resource_attributes: HashMap::new(), + attribute_mappings: Vec::new(), } } @@ -645,6 +683,23 @@ impl PyOpenInferenceConfig { Ok(()) } + #[getter] + pub(crate) fn attribute_mappings(&self, py: Python<'_>) -> PyResult> { + json_to_py( + py, + &serde_json::to_value(&self.attribute_mappings).unwrap_or_default(), + ) + } + + #[setter] + pub(crate) fn set_attribute_mappings( + &mut self, + attribute_mappings: &Bound<'_, PyAny>, + ) -> PyResult<()> { + self.attribute_mappings = py_attribute_mappings(attribute_mappings)?; + Ok(()) + } + pub(crate) fn set_header(&mut self, key: String, value: String) { self.headers.insert(key, value); } diff --git a/go/nemo_relay/nemo_relay.go b/go/nemo_relay/nemo_relay.go index b09718ff5..7e138e3da 100644 --- a/go/nemo_relay/nemo_relay.go +++ b/go/nemo_relay/nemo_relay.go @@ -251,6 +251,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*, uint64_t, void**); +extern int32_t nemo_relay_otel_subscriber_create_with_attribute_mappings(const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, uint64_t, 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*); @@ -259,6 +260,7 @@ extern void nemo_relay_otel_subscriber_free(void*); // OpenInference subscriber extern int32_t nemo_relay_openinference_subscriber_create(const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, uint64_t, void**); +extern int32_t nemo_relay_openinference_subscriber_create_with_attribute_mappings(const char*, const char*, const char*, const char*, const char*, const char*, const char*, const char*, uint64_t, const char*, void**); extern int32_t nemo_relay_openinference_subscriber_register(const void*, const char*); extern int32_t nemo_relay_openinference_subscriber_deregister(const char*); extern int32_t nemo_relay_openinference_subscriber_force_flush(const void*); @@ -1835,6 +1837,13 @@ type OpenTelemetryConfig struct { ServiceVersion string InstrumentationScope string Timeout time.Duration + AttributeMappings []OtlpAttributeMapping +} + +// OtlpAttributeMapping copies a projected OTLP attribute to an alias. +type OtlpAttributeMapping struct { + Key string `json:"key"` + Alias string `json:"alias"` } // NewOpenTelemetryConfig returns a config initialized with sensible defaults. @@ -1898,6 +1907,16 @@ func NewOpenTelemetrySubscriber(config OpenTelemetryConfig) (*OpenTelemetrySubsc cResourceAttrsJSON := C.CString(string(resourceAttrsJSON)) defer C.free(unsafe.Pointer(cResourceAttrsJSON)) + var cAttributeMappingsJSON *C.char + if config.AttributeMappings != nil { + attributeMappingsJSON, err := jsonMarshal(config.AttributeMappings) + if err != nil { + return nil, err + } + cAttributeMappingsJSON = C.CString(string(attributeMappingsJSON)) + defer C.free(unsafe.Pointer(cAttributeMappingsJSON)) + } + cServiceName := C.CString(config.ServiceName) defer C.free(unsafe.Pointer(cServiceName)) @@ -1917,7 +1936,7 @@ func NewOpenTelemetrySubscriber(config OpenTelemetryConfig) (*OpenTelemetrySubsc defer C.free(unsafe.Pointer(cInstrumentationScope)) var ptr unsafe.Pointer - status := C.nemo_relay_otel_subscriber_create( + status := C.nemo_relay_otel_subscriber_create_with_attribute_mappings( cTransport, cEndpoint, cHeadersJSON, @@ -1927,6 +1946,7 @@ func NewOpenTelemetrySubscriber(config OpenTelemetryConfig) (*OpenTelemetrySubsc cServiceVersion, cInstrumentationScope, C.uint64_t(config.Timeout/time.Millisecond), + cAttributeMappingsJSON, &ptr, ) if err := checkStatus(status); err != nil { @@ -1999,6 +2019,7 @@ type OpenInferenceConfig struct { ServiceVersion string InstrumentationScope string Timeout time.Duration + AttributeMappings []OtlpAttributeMapping } // NewOpenInferenceConfig returns a config initialized with sensible defaults. @@ -2062,6 +2083,16 @@ func NewOpenInferenceSubscriber(config OpenInferenceConfig) (*OpenInferenceSubsc cResourceAttrsJSON := C.CString(string(resourceAttrsJSON)) defer C.free(unsafe.Pointer(cResourceAttrsJSON)) + var cAttributeMappingsJSON *C.char + if config.AttributeMappings != nil { + attributeMappingsJSON, err := jsonMarshal(config.AttributeMappings) + if err != nil { + return nil, err + } + cAttributeMappingsJSON = C.CString(string(attributeMappingsJSON)) + defer C.free(unsafe.Pointer(cAttributeMappingsJSON)) + } + cServiceName := C.CString(config.ServiceName) defer C.free(unsafe.Pointer(cServiceName)) @@ -2081,7 +2112,7 @@ func NewOpenInferenceSubscriber(config OpenInferenceConfig) (*OpenInferenceSubsc defer C.free(unsafe.Pointer(cInstrumentationScope)) var ptr unsafe.Pointer - status := C.nemo_relay_openinference_subscriber_create( + status := C.nemo_relay_openinference_subscriber_create_with_attribute_mappings( cTransport, cEndpoint, cHeadersJSON, @@ -2091,6 +2122,7 @@ func NewOpenInferenceSubscriber(config OpenInferenceConfig) (*OpenInferenceSubsc cServiceVersion, cInstrumentationScope, C.uint64_t(config.Timeout/time.Millisecond), + cAttributeMappingsJSON, &ptr, ) if err := checkStatus(status); err != nil { diff --git a/go/nemo_relay/observability_plugin.go b/go/nemo_relay/observability_plugin.go index c79488919..7ccb8f5be 100644 --- a/go/nemo_relay/observability_plugin.go +++ b/go/nemo_relay/observability_plugin.go @@ -143,6 +143,7 @@ type ObservabilityOtlpConfig struct { Enabled bool `json:"enabled,omitempty"` MarkProjection ObservabilityMarkProjection `json:"mark_projection,omitempty"` MarkExcludeNames []string `json:"mark_exclude_names,omitempty"` + AttributeMappings []OtlpAttributeMapping `json:"attribute_mappings,omitempty"` Transport string `json:"transport,omitempty"` Endpoint string `json:"endpoint,omitempty"` Headers map[string]string `json:"headers,omitempty"` diff --git a/go/nemo_relay/openinference_test.go b/go/nemo_relay/openinference_test.go index 75f5f8fb9..43556b6e0 100644 --- a/go/nemo_relay/openinference_test.go +++ b/go/nemo_relay/openinference_test.go @@ -52,6 +52,10 @@ func TestOpenInferenceSubscriberLifecycle(t *testing.T) { config.Timeout = 1250 * time.Millisecond config.Headers["authorization"] = "Bearer token" config.ResourceAttributes["deployment.environment"] = "test" + config.AttributeMappings = []OtlpAttributeMapping{{ + Key: "openinference.metadata.tenant", + Alias: "tenant.id", + }} subscriber, err := NewOpenInferenceSubscriber(config) if err != nil { diff --git a/go/nemo_relay/otel_test.go b/go/nemo_relay/otel_test.go index 4ba6d7870..a84bb998b 100644 --- a/go/nemo_relay/otel_test.go +++ b/go/nemo_relay/otel_test.go @@ -45,6 +45,10 @@ func TestOpenTelemetrySubscriberLifecycle(t *testing.T) { config.Timeout = 1250 * time.Millisecond config.Headers["authorization"] = "Bearer token" config.ResourceAttributes["deployment.environment"] = "test" + config.AttributeMappings = []OtlpAttributeMapping{{ + Key: "nemo_relay.start.data.tenant", + Alias: "tenant.id", + }} subscriber, err := NewOpenTelemetrySubscriber(config) if err != nil { diff --git a/python/nemo_relay/_native.pyi b/python/nemo_relay/_native.pyi index 400307416..86e9a7778 100644 --- a/python/nemo_relay/_native.pyi +++ b/python/nemo_relay/_native.pyi @@ -964,6 +964,14 @@ class OpenTelemetryConfig: def resource_attributes(self, value: dict[str, str]) -> None: """Replace additional OpenTelemetry resource attributes.""" ... + @property + def attribute_mappings(self) -> list[dict[str, str]]: + """Return typed projected-attribute aliases.""" + ... + @attribute_mappings.setter + def attribute_mappings(self, value: list[dict[str, str]]) -> None: + """Replace typed projected-attribute aliases.""" + ... def set_header(self, key: str, value: str) -> None: """Set one exporter header key/value pair.""" ... @@ -1036,6 +1044,14 @@ class OpenInferenceConfig: def resource_attributes(self, value: dict[str, str]) -> None: """Replace additional OpenInference resource attributes.""" ... + @property + def attribute_mappings(self) -> list[dict[str, str]]: + """Return typed projected-attribute aliases.""" + ... + @attribute_mappings.setter + def attribute_mappings(self, value: list[dict[str, str]]) -> None: + """Replace typed projected-attribute aliases.""" + ... def set_header(self, key: str, value: str) -> None: """Set one exporter header key/value pair.""" ... diff --git a/python/nemo_relay/observability.py b/python/nemo_relay/observability.py index ad837a0a1..f1c502446 100644 --- a/python/nemo_relay/observability.py +++ b/python/nemo_relay/observability.py @@ -198,6 +198,7 @@ class OtlpConfig: enabled: bool = False mark_projection: MarkProjection = "inherit" mark_exclude_names: list[str] = field(default_factory=lambda: ["llm.chunk"]) + attribute_mappings: list[dict[str, str]] = field(default_factory=list) transport: Literal["http_binary", "grpc"] = "http_binary" endpoint: str | None = None headers: dict[str, str] = field(default_factory=dict) @@ -215,6 +216,7 @@ def to_dict(self) -> JsonObject: "enabled": self.enabled, "mark_projection": self.mark_projection, "mark_exclude_names": self.mark_exclude_names, + "attribute_mappings": self.attribute_mappings, "transport": self.transport, "endpoint": self.endpoint, "headers": self.headers, diff --git a/python/nemo_relay/observability.pyi b/python/nemo_relay/observability.pyi index f562a10a1..d6f640bac 100644 --- a/python/nemo_relay/observability.pyi +++ b/python/nemo_relay/observability.pyi @@ -75,6 +75,7 @@ class OtlpConfig: enabled: bool = ... mark_projection: MarkProjection = ... mark_exclude_names: list[str] = ... + attribute_mappings: list[dict[str, str]] = field(default_factory=list) transport: Literal["http_binary", "grpc"] = ... endpoint: str | None = ... headers: dict[str, str] = field(default_factory=dict) diff --git a/python/tests/test_observability_plugin.py b/python/tests/test_observability_plugin.py index e536fcaf8..c75fe938d 100644 --- a/python/tests/test_observability_plugin.py +++ b/python/tests/test_observability_plugin.py @@ -40,6 +40,7 @@ def test_defaults_and_component_wrapper(self): "enabled": False, "mark_projection": "inherit", "mark_exclude_names": ["llm.chunk"], + "attribute_mappings": [], "transport": "http_binary", "headers": {}, "resource_attributes": {}, diff --git a/python/tests/test_types.py b/python/tests/test_types.py index 7cf13d3c5..3b7a61b65 100644 --- a/python/tests/test_types.py +++ b/python/tests/test_types.py @@ -569,6 +569,8 @@ def test_config_defaults_mutation_and_repr(self): assert config.headers == {"authorization": "Bearer token"} assert config.resource_attributes == {"deployment.environment": "test"} + config.attribute_mappings = [{"key": "nemo_relay.start.metadata.tenant", "alias": "tenant.id"}] + assert config.attribute_mappings == [{"key": "nemo_relay.start.metadata.tenant", "alias": "tenant.id"}] assert "OpenTelemetryConfig" in repr(config) def test_config_rejects_invalid_map_values(self): @@ -580,6 +582,9 @@ def test_config_rejects_invalid_map_values(self): with pytest.raises(ValueError, match="dict\\[str, str\\]"): config.resource_attributes = cast(dict[str, str], {"env": 1}) + with pytest.raises(ValueError, match="attribute mapping key must not be blank"): + config.attribute_mappings = [{"key": "", "alias": "tenant.id"}] + def test_subscriber_lifecycle_and_invalid_transport(self): config = OpenTelemetryConfig() config.endpoint = "http://localhost:4318/v1/traces" @@ -658,6 +663,8 @@ def test_config_defaults_mutation_and_repr(self): assert config.headers == {"authorization": "Bearer token"} assert config.resource_attributes == {"deployment.environment": "test"} + config.attribute_mappings = [{"key": "openinference.metadata.tenant", "alias": "tenant.id"}] + assert config.attribute_mappings == [{"key": "openinference.metadata.tenant", "alias": "tenant.id"}] assert "OpenInferenceConfig" in repr(config) def test_config_rejects_invalid_map_values(self): @@ -669,6 +676,9 @@ def test_config_rejects_invalid_map_values(self): with pytest.raises(ValueError, match="dict\\[str, str\\]"): config.resource_attributes = cast(dict[str, str], {"env": 1}) + with pytest.raises(ValueError, match="attribute mapping key must not be blank"): + config.attribute_mappings = [{"key": "", "alias": "tenant.id"}] + def test_subscriber_lifecycle_and_invalid_transport(self): config = OpenInferenceConfig() config.endpoint = "http://localhost:4318/v1/traces" From d14422cdb3cb21bd63d55d2d9635e87d4528f1cb Mon Sep 17 00:00:00 2001 From: Will Killian Date: Mon, 13 Jul 2026 21:28:57 -0400 Subject: [PATCH 2/8] fix(observability): address attribute mapping review feedback Signed-off-by: Will Killian --- crates/ffi/src/api/observability.rs | 13 +++--- crates/ffi/tests/unit/api/plugin_tests.rs | 36 ++++++++++++++++ crates/node/tests/openinference_tests.mjs | 16 +++----- crates/node/tests/otel_tests.mjs | 18 ++++---- go/nemo_relay/openinference_test.go | 50 +++++++++++++++++++++-- go/nemo_relay/otel_test.go | 17 ++++++++ 6 files changed, 121 insertions(+), 29 deletions(-) diff --git a/crates/ffi/src/api/observability.rs b/crates/ffi/src/api/observability.rs index b3476407b..407579bd0 100644 --- a/crates/ffi/src/api/observability.rs +++ b/crates/ffi/src/api/observability.rs @@ -537,11 +537,14 @@ fn parse_attribute_mappings_json( return Ok(Vec::new()); } let json_string = c_str_to_string(json_ptr)?; - let mappings: Vec = - serde_json::from_str(&json_string).map_err(|error| { - set_last_error(&format!("invalid attribute_mappings JSON: {error}")); - NemoRelayStatus::InvalidJson - })?; + let value: serde_json::Value = serde_json::from_str(&json_string).map_err(|error| { + set_last_error(&format!("invalid attribute_mappings JSON: {error}")); + NemoRelayStatus::InvalidJson + })?; + let mappings: Vec = serde_json::from_value(value).map_err(|error| { + set_last_error(&format!("invalid attribute_mappings: {error}")); + NemoRelayStatus::InvalidArg + })?; nemo_relay::observability::validate_attribute_mappings(&mappings).map_err(|error| { set_last_error(&format!("invalid attribute_mappings: {error}")); NemoRelayStatus::InvalidArg diff --git a/crates/ffi/tests/unit/api/plugin_tests.rs b/crates/ffi/tests/unit/api/plugin_tests.rs index e58154f37..d520e427d 100644 --- a/crates/ffi/tests/unit/api/plugin_tests.rs +++ b/crates/ffi/tests/unit/api/plugin_tests.rs @@ -1377,6 +1377,42 @@ fn test_ffi_typed_attribute_mapping_constructors_validate_and_accept_mappings() ), NemoRelayStatus::InvalidArg ); + + for invalid_shape in ["{}", "null", r#"[{"key":1,"alias":"tenant.id"}]"#] { + let invalid_shape = cstring(invalid_shape); + assert_eq!( + nemo_relay_otel_subscriber_create_with_attribute_mappings( + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + 0, + invalid_shape.as_ptr(), + &mut otel, + ), + NemoRelayStatus::InvalidArg + ); + assert_eq!( + nemo_relay_openinference_subscriber_create_with_attribute_mappings( + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + ptr::null(), + 0, + invalid_shape.as_ptr(), + &mut openinference, + ), + NemoRelayStatus::InvalidArg + ); + } } } diff --git a/crates/node/tests/openinference_tests.mjs b/crates/node/tests/openinference_tests.mjs index 323f73c19..223ec51e7 100644 --- a/crates/node/tests/openinference_tests.mjs +++ b/crates/node/tests/openinference_tests.mjs @@ -83,21 +83,13 @@ describe('OpenInferenceSubscriber', () => { const subscriber = new OpenInferenceSubscriber({ endpoint: collector.endpoint, serviceName: 'node-agent', + attributeMappings: [{ key: 'openinference.metadata.tenant', alias: 'tenant.id' }], }); const name = unique('node_openinference_e2e'); subscriber.register(name); try { - const scope = pushScope( - 'openinference_scope', - ScopeType.Agent, - null, - null, - { - scope: true, - }, - null, - ); + const scope = pushScope('openinference_scope', ScopeType.Agent, null, null, null, null); event( 'openinference_mark', scope, @@ -108,7 +100,7 @@ describe('OpenInferenceSubscriber', () => { source: 'node', }, ); - popScope(scope); + popScope(scope, null, null, { tenant: 'node' }); subscriber.forceFlush(); const request = await collector.nextRequest(); @@ -119,6 +111,8 @@ describe('OpenInferenceSubscriber', () => { assertBodyContains(request.body, 'AGENT'); assertBodyContains(request.body, 'metadata'); assertBodyContains(request.body, 'openinference_mark'); + assertBodyContains(request.body, 'tenant.id'); + assertBodyContains(request.body, 'node'); } finally { subscriber.deregister(name); subscriber.shutdown(); diff --git a/crates/node/tests/otel_tests.mjs b/crates/node/tests/otel_tests.mjs index 3853ffaf4..5d46a0634 100644 --- a/crates/node/tests/otel_tests.mjs +++ b/crates/node/tests/otel_tests.mjs @@ -13,6 +13,10 @@ function uniqueId(prefix) { return `${prefix}_${Date.now()}_${Math.random().toString(16).slice(2)}`; } +function assertBodyContains(body, text) { + assert.equal(body.includes(Buffer.from(text, 'utf8')), true, `expected OTLP payload to contain ${text}`); +} + describe('OpenTelemetrySubscriber', () => { it('constructs from a mutable config object and supports lifecycle methods', () => { const subscriber = new OpenTelemetrySubscriber({ @@ -79,21 +83,13 @@ describe('OpenTelemetrySubscriber', () => { const subscriber = new OpenTelemetrySubscriber({ endpoint: collector.endpoint, serviceName: 'node-agent', + attributeMappings: [{ key: 'nemo_relay.mark.metadata.source', alias: 'tenant.id' }], }); const name = uniqueId('node_otel_e2e'); subscriber.register(name); try { - const scope = pushScope( - 'otel_scope', - ScopeType.Agent, - null, - null, - { - scope: true, - }, - null, - ); + const scope = pushScope('otel_scope', ScopeType.Agent, null, null, null, null); event( 'otel_mark', scope, @@ -111,6 +107,8 @@ describe('OpenTelemetrySubscriber', () => { assert.equal(request.url, '/v1/traces'); assert.equal(request.headers['content-type'], 'application/x-protobuf'); assert.ok(request.body.length > 0); + assertBodyContains(request.body, 'tenant.id'); + assertBodyContains(request.body, 'node'); } finally { subscriber.deregister(name); subscriber.shutdown(); diff --git a/go/nemo_relay/openinference_test.go b/go/nemo_relay/openinference_test.go index 43556b6e0..5aac34aef 100644 --- a/go/nemo_relay/openinference_test.go +++ b/go/nemo_relay/openinference_test.go @@ -91,14 +91,55 @@ func TestOpenInferenceSubscriberRejectsInvalidTransport(t *testing.T) { } } -func TestOpenInferenceSubscriberExportsScopeLifecycleAndMarks(t *testing.T) { +func TestOpenInferenceSubscriberRejectsInvalidAttributeMapping(t *testing.T) { + config := NewOpenInferenceConfig() + config.AttributeMappings = []OtlpAttributeMapping{{Key: "", Alias: "tenant.id"}} + + if _, err := NewOpenInferenceSubscriber(config); err == nil { + t.Fatal("expected invalid attribute mapping error") + } +} + +func TestOpenInferenceSubscriberExportsScopeLifecycleAndMappedAttributes(t *testing.T) { requests := make(chan otelRequest, 4) server := NewOtelTestServer(t, requests) defer server.Close() - subscriber := NewRegisteredOpenInferenceSubscriber(t, server.URL+"/v1/traces") + config := NewOpenInferenceConfig() + config.Endpoint = server.URL + "/v1/traces" + config.ServiceName = "go-agent" + config.AttributeMappings = []OtlpAttributeMapping{{ + Key: "openinference.metadata.tenant", + Alias: "tenant.id", + }} + subscriber, err := NewOpenInferenceSubscriber(config) + if err != nil { + t.Fatalf("NewOpenInferenceSubscriber failed: %v", err) + } defer subscriber.Close() - EmitOpenInferenceScopeLifecycle(t) + name := "go_openinference_e2e_" + time.Now().Format("150405.000000") + if err := subscriber.Register(name); err != nil { + t.Fatalf("Register failed: %v", err) + } + defer func() { _ = subscriber.Deregister(name) }() + + runWithTestScopeStack(t, func() { + handle, err := PushScope("openinference_scope", ScopeTypeAgent) + if err != nil { + t.Fatalf("PushScope failed: %v", err) + } + requireNoError(t, EmitEvent( + "openinference_mark", + WithEventParent(handle), + WithEventData(json.RawMessage(`{"step":1}`)), + WithEventMetadata(json.RawMessage(`{"source":"go"}`)), + ), "EmitEvent failed") + requireNoError( + t, + PopScope(handle, WithScopeEndMetadata(json.RawMessage(`{"tenant":"go"}`))), + "PopScope failed", + ) + }) if err := subscriber.ForceFlush(); err != nil { t.Fatalf("ForceFlush failed: %v", err) } @@ -106,6 +147,9 @@ func TestOpenInferenceSubscriberExportsScopeLifecycleAndMarks(t *testing.T) { select { case request := <-requests: AssertOpenInferenceRequest(t, request) + if !bytes.Contains(request.Body, []byte("tenant.id")) || !bytes.Contains(request.Body, []byte("go")) { + t.Fatal("expected OTLP request body to contain mapped tenant alias") + } case <-time.After(5 * time.Second): t.Fatal("timed out waiting for OTLP request") } diff --git a/go/nemo_relay/otel_test.go b/go/nemo_relay/otel_test.go index a84bb998b..e13986644 100644 --- a/go/nemo_relay/otel_test.go +++ b/go/nemo_relay/otel_test.go @@ -4,6 +4,7 @@ package nemo_relay import ( + "bytes" "encoding/json" "io" "net/http" @@ -84,6 +85,15 @@ func TestOpenTelemetrySubscriberRejectsInvalidTransport(t *testing.T) { } } +func TestOpenTelemetrySubscriberRejectsInvalidAttributeMapping(t *testing.T) { + config := NewOpenTelemetryConfig() + config.AttributeMappings = []OtlpAttributeMapping{{Key: "", Alias: "tenant.id"}} + + if _, err := NewOpenTelemetrySubscriber(config); err == nil { + t.Fatal("expected invalid attribute mapping error") + } +} + func TestOpenTelemetrySubscriberExportsScopeLifecycleAndMarks(t *testing.T) { type otelRequest struct { Path string @@ -109,6 +119,10 @@ func TestOpenTelemetrySubscriberExportsScopeLifecycleAndMarks(t *testing.T) { config := NewOpenTelemetryConfig() config.Endpoint = server.URL + "/v1/traces" config.ServiceName = "go-agent" + config.AttributeMappings = []OtlpAttributeMapping{{ + Key: "nemo_relay.mark.metadata.source", + Alias: "tenant.id", + }} subscriber, err := NewOpenTelemetrySubscriber(config) if err != nil { @@ -154,6 +168,9 @@ func TestOpenTelemetrySubscriberExportsScopeLifecycleAndMarks(t *testing.T) { if len(request.Body) == 0 { t.Fatal("expected non-empty OTLP request body") } + if !bytes.Contains(request.Body, []byte("tenant.id")) || !bytes.Contains(request.Body, []byte("go")) { + t.Fatal("expected OTLP request body to contain mapped tenant alias") + } case <-time.After(5 * time.Second): t.Fatal("timed out waiting for OTLP request") } From a9f926a7625ba07d1cd78ed9168f58d34d438d05 Mon Sep 17 00:00:00 2001 From: Will Killian Date: Mon, 13 Jul 2026 22:51:16 -0400 Subject: [PATCH 3/8] fix(observability): harden attribute mapping validation Signed-off-by: Will Killian --- crates/core/src/observability/mod.rs | 12 ++++++- .../tests/unit/observability/otel_tests.rs | 4 +-- crates/node/tests/openinference_tests.mjs | 13 ++++++-- crates/node/tests/otel_tests.mjs | 13 ++++++-- go/nemo_relay/openinference_test.go | 4 +-- go/nemo_relay/otel_test.go | 14 ++++++-- python/nemo_relay/observability.py | 2 +- python/nemo_relay/observability.pyi | 2 +- python/tests/test_types.py | 33 +++++++++++++++++++ 9 files changed, 82 insertions(+), 15 deletions(-) diff --git a/crates/core/src/observability/mod.rs b/crates/core/src/observability/mod.rs index a518bff35..d166e42ee 100644 --- a/crates/core/src/observability/mod.rs +++ b/crates/core/src/observability/mod.rs @@ -84,7 +84,7 @@ pub fn validate_attribute_mappings( if mapping.alias.trim().is_empty() { return Err("attribute mapping alias must not be blank".to_string()); } - if !aliases.insert(mapping.alias.as_str()) { + if !aliases.insert(mapping.alias.trim()) { return Err(format!( "attribute mapping alias {:?} is duplicated", mapping.alias @@ -184,6 +184,9 @@ pub(crate) fn attribute_mapping_aliases( projected_attributes: &[opentelemetry::KeyValue], mappings: &[OtlpAttributeMapping], ) -> Vec { + if mappings.is_empty() { + return Vec::new(); + } let existing = projected_attributes .iter() .map(|attribute| attribute.key.as_str().to_string()) @@ -431,5 +434,12 @@ mod attribute_projection_tests { ]) .is_err() ); + assert!( + super::validate_attribute_mappings(&[ + OtlpAttributeMapping::new("one", "duplicate"), + OtlpAttributeMapping::new("two", " duplicate "), + ]) + .is_err() + ); } } diff --git a/crates/core/tests/unit/observability/otel_tests.rs b/crates/core/tests/unit/observability/otel_tests.rs index 3471266b6..5afc55f3d 100644 --- a/crates/core/tests/unit/observability/otel_tests.rs +++ b/crates/core/tests/unit/observability/otel_tests.rs @@ -739,7 +739,7 @@ fn registered_subscriber_emits_spans_for_scope_push_pop_and_marks() { let attributes = attr_map(&span.attributes); assert_eq!( - attributes.get("nemo_relay.start.data.task"), + attributes.get("nemo_relay.start.input.task"), Some(&"scope-start".to_string()) ); assert_eq!( @@ -1564,7 +1564,7 @@ fn helper_functions_cover_additional_otel_branches() { let start_attributes = attr_map(&start_attributes(&tool_event)); assert_eq!( - start_attributes.get("nemo_relay.start.input.query"), + start_attributes.get("nemo_relay.start.data.query"), Some(&"hello".to_string()) ); assert_eq!( diff --git a/crates/node/tests/openinference_tests.mjs b/crates/node/tests/openinference_tests.mjs index 223ec51e7..6ed81a6c2 100644 --- a/crates/node/tests/openinference_tests.mjs +++ b/crates/node/tests/openinference_tests.mjs @@ -17,6 +17,16 @@ function assertBodyContains(body, text) { assert.equal(body.includes(Buffer.from(text, 'utf8')), true, `expected OTLP payload to contain ${text}`); } +function assertStringAttribute(body, key, value) { + const encoded = Buffer.concat([ + Buffer.from([0x0a, Buffer.byteLength(key)]), + Buffer.from(key), + Buffer.from([0x12, Buffer.byteLength(value) + 2, 0x0a, Buffer.byteLength(value)]), + Buffer.from(value), + ]); + assert.equal(body.includes(encoded), true, `expected OTLP string attribute ${key}=${value}`); +} + describe('OpenInferenceSubscriber', () => { it('constructs from a mutable config object and supports lifecycle methods', () => { const subscriber = new OpenInferenceSubscriber({ @@ -111,8 +121,7 @@ describe('OpenInferenceSubscriber', () => { assertBodyContains(request.body, 'AGENT'); assertBodyContains(request.body, 'metadata'); assertBodyContains(request.body, 'openinference_mark'); - assertBodyContains(request.body, 'tenant.id'); - assertBodyContains(request.body, 'node'); + assertStringAttribute(request.body, 'tenant.id', 'node'); } finally { subscriber.deregister(name); subscriber.shutdown(); diff --git a/crates/node/tests/otel_tests.mjs b/crates/node/tests/otel_tests.mjs index 5d46a0634..e6429168f 100644 --- a/crates/node/tests/otel_tests.mjs +++ b/crates/node/tests/otel_tests.mjs @@ -17,6 +17,16 @@ function assertBodyContains(body, text) { assert.equal(body.includes(Buffer.from(text, 'utf8')), true, `expected OTLP payload to contain ${text}`); } +function assertStringAttribute(body, key, value) { + const encoded = Buffer.concat([ + Buffer.from([0x0a, Buffer.byteLength(key)]), + Buffer.from(key), + Buffer.from([0x12, Buffer.byteLength(value) + 2, 0x0a, Buffer.byteLength(value)]), + Buffer.from(value), + ]); + assert.equal(body.includes(encoded), true, `expected OTLP string attribute ${key}=${value}`); +} + describe('OpenTelemetrySubscriber', () => { it('constructs from a mutable config object and supports lifecycle methods', () => { const subscriber = new OpenTelemetrySubscriber({ @@ -107,8 +117,7 @@ describe('OpenTelemetrySubscriber', () => { assert.equal(request.url, '/v1/traces'); assert.equal(request.headers['content-type'], 'application/x-protobuf'); assert.ok(request.body.length > 0); - assertBodyContains(request.body, 'tenant.id'); - assertBodyContains(request.body, 'node'); + assertStringAttribute(request.body, 'tenant.id', 'node'); } finally { subscriber.deregister(name); subscriber.shutdown(); diff --git a/go/nemo_relay/openinference_test.go b/go/nemo_relay/openinference_test.go index 5aac34aef..0f3c8c36f 100644 --- a/go/nemo_relay/openinference_test.go +++ b/go/nemo_relay/openinference_test.go @@ -147,9 +147,7 @@ func TestOpenInferenceSubscriberExportsScopeLifecycleAndMappedAttributes(t *test select { case request := <-requests: AssertOpenInferenceRequest(t, request) - if !bytes.Contains(request.Body, []byte("tenant.id")) || !bytes.Contains(request.Body, []byte("go")) { - t.Fatal("expected OTLP request body to contain mapped tenant alias") - } + assertOtlpStringAttribute(t, request.Body, "tenant.id", "go") case <-time.After(5 * time.Second): t.Fatal("timed out waiting for OTLP request") } diff --git a/go/nemo_relay/otel_test.go b/go/nemo_relay/otel_test.go index e13986644..d88822250 100644 --- a/go/nemo_relay/otel_test.go +++ b/go/nemo_relay/otel_test.go @@ -13,6 +13,16 @@ import ( "time" ) +func assertOtlpStringAttribute(t *testing.T, body []byte, key string, value string) { + t.Helper() + encoded := append([]byte{0x0a, byte(len(key))}, key...) + encoded = append(encoded, 0x12, byte(len(value)+2), 0x0a, byte(len(value))) + encoded = append(encoded, value...) + if !bytes.Contains(body, encoded) { + t.Fatalf("expected OTLP string attribute %s=%s", key, value) + } +} + func TestNewOpenTelemetryConfigDefaults(t *testing.T) { config := NewOpenTelemetryConfig() @@ -168,9 +178,7 @@ func TestOpenTelemetrySubscriberExportsScopeLifecycleAndMarks(t *testing.T) { if len(request.Body) == 0 { t.Fatal("expected non-empty OTLP request body") } - if !bytes.Contains(request.Body, []byte("tenant.id")) || !bytes.Contains(request.Body, []byte("go")) { - t.Fatal("expected OTLP request body to contain mapped tenant alias") - } + assertOtlpStringAttribute(t, request.Body, "tenant.id", "go") case <-time.After(5 * time.Second): t.Fatal("timed out waiting for OTLP request") } diff --git a/python/nemo_relay/observability.py b/python/nemo_relay/observability.py index f1c502446..b1b77d39c 100644 --- a/python/nemo_relay/observability.py +++ b/python/nemo_relay/observability.py @@ -198,7 +198,6 @@ class OtlpConfig: enabled: bool = False mark_projection: MarkProjection = "inherit" mark_exclude_names: list[str] = field(default_factory=lambda: ["llm.chunk"]) - attribute_mappings: list[dict[str, str]] = field(default_factory=list) transport: Literal["http_binary", "grpc"] = "http_binary" endpoint: str | None = None headers: dict[str, str] = field(default_factory=dict) @@ -208,6 +207,7 @@ class OtlpConfig: service_version: str | None = None instrumentation_scope: str | None = None timeout_millis: int = 3000 + attribute_mappings: list[dict[str, str]] = field(default_factory=list) def to_dict(self) -> JsonObject: """Serialize this OTLP config to the canonical JSON object shape.""" diff --git a/python/nemo_relay/observability.pyi b/python/nemo_relay/observability.pyi index d6f640bac..7f44229f6 100644 --- a/python/nemo_relay/observability.pyi +++ b/python/nemo_relay/observability.pyi @@ -75,7 +75,6 @@ class OtlpConfig: enabled: bool = ... mark_projection: MarkProjection = ... mark_exclude_names: list[str] = ... - attribute_mappings: list[dict[str, str]] = field(default_factory=list) transport: Literal["http_binary", "grpc"] = ... endpoint: str | None = ... headers: dict[str, str] = field(default_factory=dict) @@ -85,6 +84,7 @@ class OtlpConfig: service_version: str | None = ... instrumentation_scope: str | None = ... timeout_millis: int = ... + attribute_mappings: list[dict[str, str]] = field(default_factory=list) def to_dict(self) -> JsonObject: ... @dataclass(slots=True) diff --git a/python/tests/test_types.py b/python/tests/test_types.py index 3b7a61b65..f0b06fca1 100644 --- a/python/tests/test_types.py +++ b/python/tests/test_types.py @@ -92,6 +92,17 @@ class _OtelCollectorServer(http.server.ThreadingHTTPServer): request_event: threading.Event +def _otlp_string_attribute(key: str, value: str) -> bytes: + return ( + b"\x0a" + + bytes([len(key)]) + + key.encode() + + b"\x12" + + bytes([len(value) + 2, 0x0A, len(value)]) + + value.encode() + ) + + def _scope_event(events, name: str, category: str, scope_category: str) -> ScopeEvent: return next( event @@ -585,6 +596,15 @@ 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": "tenant.id"}] + with pytest.raises(ValueError, match="attribute mapping alias must not be blank"): + config.attribute_mappings = [{"key": "nemo_relay.mark.metadata.source", "alias": ""}] + + with pytest.raises(ValueError, match="attribute mapping alias .* duplicated"): + config.attribute_mappings = [ + {"key": "one", "alias": "tenant.id"}, + {"key": "two", "alias": "tenant.id"}, + ] + def test_subscriber_lifecycle_and_invalid_transport(self): config = OpenTelemetryConfig() config.endpoint = "http://localhost:4318/v1/traces" @@ -613,6 +633,7 @@ def test_subscriber_exports_scope_and_mark_events_end_to_end(self): config = OpenTelemetryConfig() config.endpoint = collector.endpoint config.service_name = "py-agent" + config.attribute_mappings = [{"key": "nemo_relay.mark.metadata.source", "alias": "tenant.id"}] subscriber = OpenTelemetrySubscriber(config) subscriber_name = f"py_otel_e2e_{uuid4().hex}" @@ -635,6 +656,7 @@ def test_subscriber_exports_scope_and_mark_events_end_to_end(self): assert request["path"] == "/v1/traces" assert request["headers"]["content-type"] == "application/x-protobuf" assert request["body"] + assert _otlp_string_attribute("tenant.id", "python") in request["body"] finally: subscriber.deregister(subscriber_name) subscriber.shutdown() @@ -679,6 +701,15 @@ 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": "tenant.id"}] + with pytest.raises(ValueError, match="attribute mapping alias must not be blank"): + config.attribute_mappings = [{"key": "openinference.metadata.tenant", "alias": ""}] + + with pytest.raises(ValueError, match="attribute mapping alias .* duplicated"): + config.attribute_mappings = [ + {"key": "one", "alias": "tenant.id"}, + {"key": "two", "alias": "tenant.id"}, + ] + def test_subscriber_lifecycle_and_invalid_transport(self): config = OpenInferenceConfig() config.endpoint = "http://localhost:4318/v1/traces" @@ -714,6 +745,7 @@ def test_subscriber_exports_scope_and_mark_events_end_to_end(self): config = OpenInferenceConfig() config.endpoint = collector.endpoint config.service_name = "py-agent" + config.attribute_mappings = [{"key": "nemo_relay.mark.metadata.source", "alias": "tenant.id"}] subscriber = OpenInferenceSubscriber(config) subscriber_name = f"py_openinference_e2e_{uuid4().hex}" @@ -740,6 +772,7 @@ def test_subscriber_exports_scope_and_mark_events_end_to_end(self): assert b"AGENT" in request["body"] assert b"metadata" in request["body"] assert b"openinference_mark" in request["body"] + assert _otlp_string_attribute("tenant.id", "python") in request["body"] finally: subscriber.deregister(subscriber_name) subscriber.shutdown() From a5e7b33227ceceb7d4baf4311687d433d20aa262 Mon Sep 17 00:00:00 2001 From: Will Killian Date: Mon, 13 Jul 2026 23:03:17 -0400 Subject: [PATCH 4/8] fix(observability): avoid unnecessary mapping copies Signed-off-by: Will Killian --- .../core/src/observability/openinference.rs | 23 +++++++----- crates/core/src/observability/otel.rs | 23 +++++++----- crates/node/tests/openinference_tests.mjs | 14 ++------ crates/node/tests/otel_tests.mjs | 14 ++------ go/nemo_relay/otel_test.go | 11 ++++-- python/tests/test_types.py | 4 +-- scripts/test-support/otel_test_utils.mjs | 35 +++++++++++++++++-- 7 files changed, 77 insertions(+), 47 deletions(-) diff --git a/crates/core/src/observability/openinference.rs b/crates/core/src/observability/openinference.rs index 28faede98..06997921e 100644 --- a/crates/core/src/observability/openinference.rs +++ b/crates/core/src/observability/openinference.rs @@ -597,14 +597,19 @@ impl OpenInferenceEventProcessor { .with_start_time(to_system_time(*event.timestamp())) .start_with_context(&self.tracer, &self.parent_context(event)); let attributes = start_attributes(event); - span.set_attributes(attributes.clone()); + let projected_attributes = if self.attribute_mappings.is_empty() { + Vec::new() + } else { + attributes.clone() + }; + span.set_attributes(attributes); let span_context = local_parent_span_context(span.span_context()); self.active_spans.insert( event.uuid(), ActiveSpan { span, span_context, - projected_attributes: attributes, + projected_attributes, }, ); } @@ -616,12 +621,14 @@ impl OpenInferenceEventProcessor { self.record_completed_span_context(event.uuid(), active_span.span_context.clone()); super::set_span_status_from_event_metadata(&mut active_span.span, event); let mut attributes = end_attributes(event); - let mut projected_attributes = active_span.projected_attributes; - projected_attributes.extend(attributes.iter().cloned()); - attributes.extend(attribute_mapping_aliases( - &projected_attributes, - &self.attribute_mappings, - )); + if !self.attribute_mappings.is_empty() { + let mut projected_attributes = active_span.projected_attributes; + projected_attributes.extend(attributes.iter().cloned()); + attributes.extend(attribute_mapping_aliases( + &projected_attributes, + &self.attribute_mappings, + )); + } active_span.span.set_attributes(attributes); active_span .span diff --git a/crates/core/src/observability/otel.rs b/crates/core/src/observability/otel.rs index 1a99b4e23..c083c40fe 100644 --- a/crates/core/src/observability/otel.rs +++ b/crates/core/src/observability/otel.rs @@ -590,14 +590,19 @@ impl OtelEventProcessor { .with_start_time(to_system_time(*event.timestamp())) .start_with_context(&self.tracer, &self.parent_context(event)); let attributes = start_attributes(event); - span.set_attributes(attributes.clone()); + let projected_attributes = if self.attribute_mappings.is_empty() { + Vec::new() + } else { + attributes.clone() + }; + span.set_attributes(attributes); let span_context = local_parent_span_context(span.span_context()); self.active_spans.insert( event.uuid(), ActiveSpan { span, span_context, - projected_attributes: attributes, + projected_attributes, }, ); } @@ -610,12 +615,14 @@ impl OtelEventProcessor { super::set_span_status_from_event_metadata(&mut active_span.span, event); let mut attributes = end_attributes(event); - let mut projected_attributes = active_span.projected_attributes; - projected_attributes.extend(attributes.iter().cloned()); - attributes.extend(attribute_mapping_aliases( - &projected_attributes, - &self.attribute_mappings, - )); + if !self.attribute_mappings.is_empty() { + let mut projected_attributes = active_span.projected_attributes; + projected_attributes.extend(attributes.iter().cloned()); + attributes.extend(attribute_mapping_aliases( + &projected_attributes, + &self.attribute_mappings, + )); + } active_span.span.set_attributes(attributes); active_span .span diff --git a/crates/node/tests/openinference_tests.mjs b/crates/node/tests/openinference_tests.mjs index 6ed81a6c2..a5f3d7f11 100644 --- a/crates/node/tests/openinference_tests.mjs +++ b/crates/node/tests/openinference_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 { OpenInferenceSubscriber, ScopeType, pushScope, popScope, event } = require('../index.js'); @@ -17,16 +17,6 @@ function assertBodyContains(body, text) { assert.equal(body.includes(Buffer.from(text, 'utf8')), true, `expected OTLP payload to contain ${text}`); } -function assertStringAttribute(body, key, value) { - const encoded = Buffer.concat([ - Buffer.from([0x0a, Buffer.byteLength(key)]), - Buffer.from(key), - Buffer.from([0x12, Buffer.byteLength(value) + 2, 0x0a, Buffer.byteLength(value)]), - Buffer.from(value), - ]); - assert.equal(body.includes(encoded), true, `expected OTLP string attribute ${key}=${value}`); -} - describe('OpenInferenceSubscriber', () => { it('constructs from a mutable config object and supports lifecycle methods', () => { const subscriber = new OpenInferenceSubscriber({ @@ -121,7 +111,7 @@ describe('OpenInferenceSubscriber', () => { assertBodyContains(request.body, 'AGENT'); assertBodyContains(request.body, 'metadata'); assertBodyContains(request.body, 'openinference_mark'); - assertStringAttribute(request.body, 'tenant.id', 'node'); + assertOtlpStringAttribute(request.body, 'tenant.id', 'node'); } finally { subscriber.deregister(name); subscriber.shutdown(); diff --git a/crates/node/tests/otel_tests.mjs b/crates/node/tests/otel_tests.mjs index e6429168f..9144178b6 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 { OpenTelemetrySubscriber, ScopeType, pushScope, popScope, event } = require('../index.js'); @@ -17,16 +17,6 @@ function assertBodyContains(body, text) { assert.equal(body.includes(Buffer.from(text, 'utf8')), true, `expected OTLP payload to contain ${text}`); } -function assertStringAttribute(body, key, value) { - const encoded = Buffer.concat([ - Buffer.from([0x0a, Buffer.byteLength(key)]), - Buffer.from(key), - Buffer.from([0x12, Buffer.byteLength(value) + 2, 0x0a, Buffer.byteLength(value)]), - Buffer.from(value), - ]); - assert.equal(body.includes(encoded), true, `expected OTLP string attribute ${key}=${value}`); -} - describe('OpenTelemetrySubscriber', () => { it('constructs from a mutable config object and supports lifecycle methods', () => { const subscriber = new OpenTelemetrySubscriber({ @@ -117,7 +107,7 @@ describe('OpenTelemetrySubscriber', () => { assert.equal(request.url, '/v1/traces'); assert.equal(request.headers['content-type'], 'application/x-protobuf'); assert.ok(request.body.length > 0); - assertStringAttribute(request.body, 'tenant.id', 'node'); + assertOtlpStringAttribute(request.body, 'tenant.id', 'node'); } finally { subscriber.deregister(name); subscriber.shutdown(); diff --git a/go/nemo_relay/otel_test.go b/go/nemo_relay/otel_test.go index d88822250..73850d591 100644 --- a/go/nemo_relay/otel_test.go +++ b/go/nemo_relay/otel_test.go @@ -5,6 +5,7 @@ package nemo_relay import ( "bytes" + "encoding/binary" "encoding/json" "io" "net/http" @@ -15,9 +16,13 @@ import ( func assertOtlpStringAttribute(t *testing.T, body []byte, key string, value string) { t.Helper() - encoded := append([]byte{0x0a, byte(len(key))}, key...) - encoded = append(encoded, 0x12, byte(len(value)+2), 0x0a, byte(len(value))) - encoded = append(encoded, value...) + encoded := append([]byte{0x0a}, binary.AppendUvarint(nil, uint64(len(key)))...) + encoded = append(encoded, key...) + attributeValue := append([]byte{0x0a}, binary.AppendUvarint(nil, uint64(len(value)))...) + attributeValue = append(attributeValue, value...) + encoded = append(encoded, 0x12) + encoded = binary.AppendUvarint(encoded, uint64(len(attributeValue))) + encoded = append(encoded, attributeValue...) if !bytes.Contains(body, encoded) { t.Fatalf("expected OTLP string attribute %s=%s", key, value) } diff --git a/python/tests/test_types.py b/python/tests/test_types.py index f0b06fca1..02b57c374 100644 --- a/python/tests/test_types.py +++ b/python/tests/test_types.py @@ -599,7 +599,7 @@ def test_config_rejects_invalid_map_values(self): with pytest.raises(ValueError, match="attribute mapping alias must not be blank"): config.attribute_mappings = [{"key": "nemo_relay.mark.metadata.source", "alias": ""}] - with pytest.raises(ValueError, match="attribute mapping alias .* duplicated"): + with pytest.raises(ValueError, match=r"attribute mapping alias .* duplicated"): config.attribute_mappings = [ {"key": "one", "alias": "tenant.id"}, {"key": "two", "alias": "tenant.id"}, @@ -704,7 +704,7 @@ def test_config_rejects_invalid_map_values(self): with pytest.raises(ValueError, match="attribute mapping alias must not be blank"): config.attribute_mappings = [{"key": "openinference.metadata.tenant", "alias": ""}] - with pytest.raises(ValueError, match="attribute mapping alias .* duplicated"): + with pytest.raises(ValueError, match=r"attribute mapping alias .* duplicated"): config.attribute_mappings = [ {"key": "one", "alias": "tenant.id"}, {"key": "two", "alias": "tenant.id"}, diff --git a/scripts/test-support/otel_test_utils.mjs b/scripts/test-support/otel_test_utils.mjs index e7fd87117..8e7727321 100644 --- a/scripts/test-support/otel_test_utils.mjs +++ b/scripts/test-support/otel_test_utils.mjs @@ -5,6 +5,33 @@ import { spawn } from 'node:child_process'; import { once } from 'node:events'; import { fileURLToPath } from 'node:url'; +function encodeVarint(value) { + const bytes = []; + do { + const byte = value % 128; + value = Math.floor(value / 128); + bytes.push(value === 0 ? byte : byte | 0x80); + } while (value !== 0); + return Buffer.from(bytes); +} + +export function assertOtlpStringAttribute(body, key, value) { + const keyBuffer = Buffer.from(key, 'utf8'); + const valueBuffer = Buffer.from(value, 'utf8'); + const attributeValue = Buffer.concat([Buffer.from([0x0a]), encodeVarint(valueBuffer.length), valueBuffer]); + const encoded = Buffer.concat([ + Buffer.from([0x0a]), + encodeVarint(keyBuffer.length), + keyBuffer, + Buffer.from([0x12]), + encodeVarint(attributeValue.length), + attributeValue, + ]); + if (!body.includes(encoded)) { + throw new Error(`expected OTLP string attribute ${key}=${value}`); + } +} + export async function startCollector() { const requests = []; let nextRequestIndex = 0; @@ -68,7 +95,9 @@ export async function startCollector() { }); const endpoint = await Promise.race([ readyPromise, - new Promise((_, reject) => setTimeout(() => reject(new Error('timed out waiting for OTLP collector startup')), 5000)), + new Promise((_, reject) => + setTimeout(() => reject(new Error('timed out waiting for OTLP collector startup')), 5000), + ), ]); return { @@ -86,7 +115,9 @@ export async function startCollector() { }); return await Promise.race([ requestPromise, - new Promise((_, reject) => setTimeout(() => reject(new Error('timed out waiting for OTLP request')), timeoutMs)), + new Promise((_, reject) => + setTimeout(() => reject(new Error('timed out waiting for OTLP request')), timeoutMs), + ), ]); }, async close() { From 967732b6b28afa6c81d0e324aa5c0b0be66b7a48 Mon Sep 17 00:00:00 2001 From: Will Killian Date: Tue, 14 Jul 2026 10:40:53 -0400 Subject: [PATCH 5/8] fix(observability): address typed OTLP projection feedback Signed-off-by: Will Killian --- crates/core/src/observability/mod.rs | 51 +++++++++++++++- .../core/src/observability/openinference.rs | 59 +++++++++++++++---- crates/core/src/observability/otel.rs | 57 ++++++++++++++---- .../unit/observability/openinference_tests.rs | 48 +++++++++++---- .../tests/unit/observability/otel_tests.rs | 48 +++++++++++---- python/tests/test_types.py | 31 ++++++---- 6 files changed, 230 insertions(+), 64 deletions(-) diff --git a/crates/core/src/observability/mod.rs b/crates/core/src/observability/mod.rs index d166e42ee..765cf9beb 100644 --- a/crates/core/src/observability/mod.rs +++ b/crates/core/src/observability/mod.rs @@ -175,6 +175,28 @@ pub(crate) fn apply_attribute_mappings( attributes.extend(attribute_mapping_aliases(attributes, mappings)); } +/// Keeps the start attributes needed to resolve mappings at the end of a span. +/// +/// The final span attributes must still take precedence over mapped aliases, so +/// retain both mapped source keys and aliases that were already present at +/// start. The span itself owns all other start attributes and does not need a +/// second copy in the active-span state. +#[cfg(any(feature = "otel", feature = "openinference"))] +pub(crate) fn attribute_mapping_inputs( + attributes: &[opentelemetry::KeyValue], + mappings: &[OtlpAttributeMapping], +) -> Vec { + attributes + .iter() + .filter(|attribute| { + mappings.iter().any(|mapping| { + attribute.key.as_str() == mapping.key || attribute.key.as_str() == mapping.alias + }) + }) + .cloned() + .collect() +} + /// Resolves typed aliases from a complete set of projected attributes. /// /// Callers that project a span across multiple lifecycle events must pass every @@ -360,7 +382,34 @@ where #[cfg(all(test, any(feature = "otel", feature = "openinference")))] mod attribute_projection_tests { - use super::{OtlpAttributeMapping, apply_attribute_mappings, push_top_level_json_attributes}; + use super::{ + OtlpAttributeMapping, apply_attribute_mappings, attribute_mapping_inputs, + push_top_level_json_attributes, + }; + + #[test] + fn retains_only_mapping_sources_and_existing_aliases_between_span_events() { + let attributes = vec![ + opentelemetry::KeyValue::new("source", "value"), + opentelemetry::KeyValue::new("alias", "existing"), + opentelemetry::KeyValue::new("large.request", "payload"), + ]; + + let retained = + attribute_mapping_inputs(&attributes, &[OtlpAttributeMapping::new("source", "alias")]); + + assert_eq!(retained.len(), 2); + assert!( + retained + .iter() + .any(|attribute| attribute.key.as_str() == "source") + ); + assert!( + retained + .iter() + .any(|attribute| attribute.key.as_str() == "alias") + ); + } #[test] fn projects_typed_json_and_copies_configured_aliases() { diff --git a/crates/core/src/observability/openinference.rs b/crates/core/src/observability/openinference.rs index 06997921e..4e3d9c1a0 100644 --- a/crates/core/src/observability/openinference.rs +++ b/crates/core/src/observability/openinference.rs @@ -22,10 +22,10 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH}; use super::{ MarkProjection, OtlpAttributeMapping, apply_attribute_mappings, attribute_mapping_aliases, - default_mark_exclude_names, effective_mark_projection, estimate_cost_for_response_or_model, - estimate_cost_for_response_or_requested_model, manual, merge_usage, model_name_for_llm_event, - push_serialized_top_level_attributes, push_top_level_json_attributes, - validate_attribute_mappings, + attribute_mapping_inputs, default_mark_exclude_names, effective_mark_projection, + estimate_cost_for_response_or_model, estimate_cost_for_response_or_requested_model, manual, + merge_usage, model_name_for_llm_event, push_serialized_top_level_attributes, + push_top_level_json_attributes, validate_attribute_mappings, }; use crate::api::event::{Event, EventNormalizationExt, ScopeCategory}; use crate::api::runtime::EventSubscriberFn; @@ -242,6 +242,27 @@ pub struct OpenInferenceSubscriber { inner: Arc, } +/// Options for constructing an OpenInference subscriber from an existing tracer provider. +#[derive(Debug, Clone)] +pub struct OpenInferenceSubscriberOptions { + /// How mark events are projected into the trace. + pub mark_projection: MarkProjection, + /// Mark names excluded from tool projection. + pub mark_exclude_names: Vec, + /// Typed OTLP attributes copied to alias keys. + pub attribute_mappings: Vec, +} + +impl Default for OpenInferenceSubscriberOptions { + fn default() -> Self { + Self { + mark_projection: MarkProjection::default(), + mark_exclude_names: default_mark_exclude_names(), + attribute_mappings: Vec::new(), + } + } +} + struct Inner { processor: Arc>, subscriber: EventSubscriberFn, @@ -326,14 +347,30 @@ impl OpenInferenceSubscriber { I: IntoIterator, { let attribute_mappings = attribute_mappings.into_iter().collect::>(); - validate_attribute_mappings(&attribute_mappings) + Self::from_tracer_provider_with_options( + provider, + instrumentation_scope, + OpenInferenceSubscriberOptions { + attribute_mappings, + ..Default::default() + }, + ) + } + + /// Builds a subscriber from a tracer provider with composable projection options. + pub fn from_tracer_provider_with_options( + provider: SdkTracerProvider, + instrumentation_scope: impl Into, + options: OpenInferenceSubscriberOptions, + ) -> Result { + validate_attribute_mappings(&options.attribute_mappings) .map_err(OpenInferenceError::InvalidAttributeMappings)?; Ok(Self::from_tracer_provider_with_scope( provider, instrumentation_scope.into(), - MarkProjection::default(), - default_mark_exclude_names(), - attribute_mappings, + options.mark_projection, + options.mark_exclude_names, + options.attribute_mappings, )) } @@ -597,11 +634,7 @@ impl OpenInferenceEventProcessor { .with_start_time(to_system_time(*event.timestamp())) .start_with_context(&self.tracer, &self.parent_context(event)); let attributes = start_attributes(event); - let projected_attributes = if self.attribute_mappings.is_empty() { - Vec::new() - } else { - attributes.clone() - }; + let projected_attributes = attribute_mapping_inputs(&attributes, &self.attribute_mappings); span.set_attributes(attributes); let span_context = local_parent_span_context(span.span_context()); self.active_spans.insert( diff --git a/crates/core/src/observability/otel.rs b/crates/core/src/observability/otel.rs index c083c40fe..671367036 100644 --- a/crates/core/src/observability/otel.rs +++ b/crates/core/src/observability/otel.rs @@ -22,9 +22,9 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH}; use super::{ MarkProjection, OtlpAttributeMapping, apply_attribute_mappings, attribute_mapping_aliases, - 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_top_level_json_attributes, + 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_top_level_json_attributes, validate_attribute_mappings, }; use crate::api::event::{Event, EventNormalizationExt, ScopeCategory}; @@ -236,6 +236,27 @@ pub struct OpenTelemetrySubscriber { inner: Arc, } +/// Options for constructing an OpenTelemetry subscriber from an existing tracer provider. +#[derive(Debug, Clone)] +pub struct OpenTelemetrySubscriberOptions { + /// How mark events are projected into the trace. + pub mark_projection: MarkProjection, + /// Mark names excluded from tool projection. + pub mark_exclude_names: Vec, + /// Typed OTLP attributes copied to alias keys. + pub attribute_mappings: Vec, +} + +impl Default for OpenTelemetrySubscriberOptions { + fn default() -> Self { + Self { + mark_projection: MarkProjection::default(), + mark_exclude_names: default_mark_exclude_names(), + attribute_mappings: Vec::new(), + } + } +} + struct Inner { processor: Arc>, subscriber: EventSubscriberFn, @@ -320,14 +341,30 @@ impl OpenTelemetrySubscriber { I: IntoIterator, { let attribute_mappings = attribute_mappings.into_iter().collect::>(); - validate_attribute_mappings(&attribute_mappings) + Self::from_tracer_provider_with_options( + provider, + instrumentation_scope, + OpenTelemetrySubscriberOptions { + attribute_mappings, + ..Default::default() + }, + ) + } + + /// Builds a subscriber from a tracer provider with composable projection options. + pub fn from_tracer_provider_with_options( + provider: SdkTracerProvider, + instrumentation_scope: impl Into, + options: OpenTelemetrySubscriberOptions, + ) -> Result { + validate_attribute_mappings(&options.attribute_mappings) .map_err(OpenTelemetryError::InvalidAttributeMappings)?; Ok(Self::from_tracer_provider_with_scope( provider, instrumentation_scope.into(), - MarkProjection::default(), - default_mark_exclude_names(), - attribute_mappings, + options.mark_projection, + options.mark_exclude_names, + options.attribute_mappings, )) } @@ -590,11 +627,7 @@ impl OtelEventProcessor { .with_start_time(to_system_time(*event.timestamp())) .start_with_context(&self.tracer, &self.parent_context(event)); let attributes = start_attributes(event); - let projected_attributes = if self.attribute_mappings.is_empty() { - Vec::new() - } else { - attributes.clone() - }; + let projected_attributes = attribute_mapping_inputs(&attributes, &self.attribute_mappings); span.set_attributes(attributes); let span_context = local_parent_span_context(span.span_context()); self.active_spans.insert( diff --git a/crates/core/tests/unit/observability/openinference_tests.rs b/crates/core/tests/unit/observability/openinference_tests.rs index aaf1a783f..7f5a03c12 100644 --- a/crates/core/tests/unit/observability/openinference_tests.rs +++ b/crates/core/tests/unit/observability/openinference_tests.rs @@ -690,20 +690,28 @@ fn subscriber_registration_and_provider_lifecycle_methods_work() { #[test] fn mapped_aliases_are_typed_and_cannot_replace_projected_span_fields() { let (provider, exporter) = make_provider(); - let subscriber = OpenInferenceSubscriber::from_tracer_provider_with_attribute_mappings( + let subscriber = OpenInferenceSubscriber::from_tracer_provider_with_options( provider, "mapping-scope", - [ - crate::observability::OtlpAttributeMapping::new( - "nemo_relay.start.data.tenant", - "tenant.id", - ), - crate::observability::OtlpAttributeMapping::new( - "nemo_relay.end.data.tenant", - "nemo_relay.start.data.tenant", - ), - crate::observability::OtlpAttributeMapping::new("missing.source", "ignored.alias"), - ], + OpenInferenceSubscriberOptions { + mark_projection: MarkProjection::Tool, + mark_exclude_names: vec!["custom.mark".to_string()], + attribute_mappings: vec![ + crate::observability::OtlpAttributeMapping::new( + "nemo_relay.start.data.tenant", + "tenant.id", + ), + crate::observability::OtlpAttributeMapping::new( + "nemo_relay.end.data.tenant", + "nemo_relay.start.data.tenant", + ), + crate::observability::OtlpAttributeMapping::new( + "nemo_relay.start.data.tenant", + "nemo_relay.start.data.existing", + ), + crate::observability::OtlpAttributeMapping::new("missing.source", "ignored.alias"), + ], + }, ) .unwrap(); let callback = subscriber.subscriber(); @@ -713,7 +721,7 @@ fn mapped_aliases_are_typed_and_cannot_replace_projected_span_fields() { None, "mapped-scope", ScopeType::Agent, - Some(json!({"tenant": 7})), + Some(json!({"tenant": 7, "existing": 9})), )); callback(&make_end_event( uuid, @@ -743,6 +751,20 @@ fn mapped_aliases_are_typed_and_cannot_replace_projected_span_fields() { .map(|attribute| &attribute.value), Some(&opentelemetry::Value::I64(7)) ); + assert_eq!( + span.attributes + .iter() + .filter(|attribute| attribute.key.as_str() == "nemo_relay.start.data.existing") + .count(), + 1 + ); + assert_eq!( + span.attributes + .iter() + .find(|attribute| attribute.key.as_str() == "nemo_relay.start.data.existing") + .map(|attribute| &attribute.value), + Some(&opentelemetry::Value::I64(9)) + ); assert!( !span .attributes diff --git a/crates/core/tests/unit/observability/otel_tests.rs b/crates/core/tests/unit/observability/otel_tests.rs index 5afc55f3d..a5b790ccb 100644 --- a/crates/core/tests/unit/observability/otel_tests.rs +++ b/crates/core/tests/unit/observability/otel_tests.rs @@ -587,20 +587,28 @@ fn subscriber_registration_and_provider_lifecycle_methods_work() { #[test] fn mapped_aliases_are_typed_and_cannot_replace_projected_span_fields() { let (provider, exporter) = make_provider(); - let subscriber = OpenTelemetrySubscriber::from_tracer_provider_with_attribute_mappings( + let subscriber = OpenTelemetrySubscriber::from_tracer_provider_with_options( provider, "mapping-scope", - [ - crate::observability::OtlpAttributeMapping::new( - "nemo_relay.start.data.tenant", - "tenant.id", - ), - crate::observability::OtlpAttributeMapping::new( - "nemo_relay.end.data.tenant", - "nemo_relay.start.data.tenant", - ), - crate::observability::OtlpAttributeMapping::new("missing.source", "ignored.alias"), - ], + OpenTelemetrySubscriberOptions { + mark_projection: MarkProjection::Tool, + mark_exclude_names: vec!["custom.mark".to_string()], + attribute_mappings: vec![ + crate::observability::OtlpAttributeMapping::new( + "nemo_relay.start.data.tenant", + "tenant.id", + ), + crate::observability::OtlpAttributeMapping::new( + "nemo_relay.end.data.tenant", + "nemo_relay.start.data.tenant", + ), + crate::observability::OtlpAttributeMapping::new( + "nemo_relay.start.data.tenant", + "nemo_relay.start.data.existing", + ), + crate::observability::OtlpAttributeMapping::new("missing.source", "ignored.alias"), + ], + }, ) .unwrap(); let callback = subscriber.subscriber(); @@ -610,7 +618,7 @@ fn mapped_aliases_are_typed_and_cannot_replace_projected_span_fields() { None, "mapped-scope", ScopeType::Agent, - Some(json!({"tenant": 7})), + Some(json!({"tenant": 7, "existing": 9})), )); callback(&make_end_event( uuid, @@ -640,6 +648,20 @@ fn mapped_aliases_are_typed_and_cannot_replace_projected_span_fields() { .map(|attribute| &attribute.value), Some(&opentelemetry::Value::I64(7)) ); + assert_eq!( + span.attributes + .iter() + .filter(|attribute| attribute.key.as_str() == "nemo_relay.start.data.existing") + .count(), + 1 + ); + assert_eq!( + span.attributes + .iter() + .find(|attribute| attribute.key.as_str() == "nemo_relay.start.data.existing") + .map(|attribute| &attribute.value), + Some(&opentelemetry::Value::I64(9)) + ); assert!( !span .attributes diff --git a/python/tests/test_types.py b/python/tests/test_types.py index 02b57c374..a33766bb4 100644 --- a/python/tests/test_types.py +++ b/python/tests/test_types.py @@ -92,15 +92,20 @@ class _OtelCollectorServer(http.server.ThreadingHTTPServer): request_event: threading.Event +def _encode_varint(value: int) -> bytes: + result = bytearray() + while value >= 0x80: + result.append((value & 0x7F) | 0x80) + value >>= 7 + result.append(value) + return bytes(result) + + def _otlp_string_attribute(key: str, value: str) -> bytes: - return ( - b"\x0a" - + bytes([len(key)]) - + key.encode() - + b"\x12" - + bytes([len(value) + 2, 0x0A, len(value)]) - + value.encode() - ) + key_bytes = key.encode() + value_bytes = value.encode() + any_value = b"\x0a" + _encode_varint(len(value_bytes)) + value_bytes + return b"\x0a" + _encode_varint(len(key_bytes)) + key_bytes + b"\x12" + _encode_varint(len(any_value)) + any_value def _scope_event(events, name: str, category: str, scope_category: str) -> ScopeEvent: @@ -630,6 +635,7 @@ def test_subscriber_lifecycle_and_invalid_transport(self): def test_subscriber_exports_scope_and_mark_events_end_to_end(self): with _OtelCollector() as collector: + source = "python-é" * 20 config = OpenTelemetryConfig() config.endpoint = collector.endpoint config.service_name = "py-agent" @@ -646,7 +652,7 @@ def test_subscriber_exports_scope_and_mark_events_end_to_end(self): "otel_mark", handle=handle, data={"step": 1}, - metadata={"source": "python"}, + metadata={"source": source}, ) finally: scope.pop(handle) @@ -656,7 +662,7 @@ def test_subscriber_exports_scope_and_mark_events_end_to_end(self): assert request["path"] == "/v1/traces" assert request["headers"]["content-type"] == "application/x-protobuf" assert request["body"] - assert _otlp_string_attribute("tenant.id", "python") in request["body"] + assert _otlp_string_attribute("tenant.id", source) in request["body"] finally: subscriber.deregister(subscriber_name) subscriber.shutdown() @@ -742,6 +748,7 @@ def test_subscriber_lifecycle_and_invalid_transport(self): def test_subscriber_exports_scope_and_mark_events_end_to_end(self): with _OtelCollector() as collector: + source = "python-é" * 20 config = OpenInferenceConfig() config.endpoint = collector.endpoint config.service_name = "py-agent" @@ -758,7 +765,7 @@ def test_subscriber_exports_scope_and_mark_events_end_to_end(self): "openinference_mark", handle=handle, data={"step": 1}, - metadata={"source": "python"}, + metadata={"source": source}, ) finally: scope.pop(handle) @@ -772,7 +779,7 @@ def test_subscriber_exports_scope_and_mark_events_end_to_end(self): assert b"AGENT" in request["body"] assert b"metadata" in request["body"] assert b"openinference_mark" in request["body"] - assert _otlp_string_attribute("tenant.id", "python") in request["body"] + assert _otlp_string_attribute("tenant.id", source) in request["body"] finally: subscriber.deregister(subscriber_name) subscriber.shutdown() From b70841b2564d1b0406fe075adc481f1127de00f3 Mon Sep 17 00:00:00 2001 From: Will Killian Date: Tue, 14 Jul 2026 12:37:08 -0400 Subject: [PATCH 6/8] fix(observability): accept typed attribute plugin mappings Signed-off-by: Will Killian --- crates/core/src/observability/mod.rs | 113 +----------------- .../src/observability/plugin_component.rs | 2 + .../attribute_projection_tests.rs | 112 +++++++++++++++++ .../observability/plugin_component_tests.rs | 21 ++++ 4 files changed, 137 insertions(+), 111 deletions(-) create mode 100644 crates/core/tests/unit/observability/attribute_projection_tests.rs diff --git a/crates/core/src/observability/mod.rs b/crates/core/src/observability/mod.rs index 765cf9beb..c3c6bf74c 100644 --- a/crates/core/src/observability/mod.rs +++ b/crates/core/src/observability/mod.rs @@ -381,114 +381,5 @@ where } #[cfg(all(test, any(feature = "otel", feature = "openinference")))] -mod attribute_projection_tests { - use super::{ - OtlpAttributeMapping, apply_attribute_mappings, attribute_mapping_inputs, - push_top_level_json_attributes, - }; - - #[test] - fn retains_only_mapping_sources_and_existing_aliases_between_span_events() { - let attributes = vec![ - opentelemetry::KeyValue::new("source", "value"), - opentelemetry::KeyValue::new("alias", "existing"), - opentelemetry::KeyValue::new("large.request", "payload"), - ]; - - let retained = - attribute_mapping_inputs(&attributes, &[OtlpAttributeMapping::new("source", "alias")]); - - assert_eq!(retained.len(), 2); - assert!( - retained - .iter() - .any(|attribute| attribute.key.as_str() == "source") - ); - assert!( - retained - .iter() - .any(|attribute| attribute.key.as_str() == "alias") - ); - } - - #[test] - fn projects_typed_json_and_copies_configured_aliases() { - let mut attributes = Vec::new(); - push_top_level_json_attributes( - &mut attributes, - "nemo_relay.start.metadata", - Some(&serde_json::json!({ - "tenant": "acme", - "attempt": 2, - "tags": ["a", "b"], - "context": {"region": "us-east-1"}, - "request": {"id": "nested-id"}, - "request.id": "flat-id", - "event_id": 18446744073709551615u64 - })), - ); - apply_attribute_mappings( - &mut attributes, - &[OtlpAttributeMapping::new( - "nemo_relay.start.metadata.tenant", - "tenant.id", - )], - ); - - let values = attributes - .iter() - .map(|attribute| (attribute.key.as_str(), attribute.value.to_string())) - .collect::>(); - assert_eq!( - values.get("nemo_relay.start.metadata.tenant"), - Some(&"acme".to_string()) - ); - assert_eq!( - values.get("nemo_relay.start.metadata.attempt"), - Some(&"2".to_string()) - ); - assert_eq!( - values.get("nemo_relay.start.metadata.tags"), - Some(&"[\"a\",\"b\"]".to_string()) - ); - assert_eq!( - values.get("nemo_relay.start.metadata.context"), - Some(&"{\"region\":\"us-east-1\"}".to_string()) - ); - assert_eq!( - values.get("nemo_relay.start.metadata.request"), - Some(&"{\"id\":\"nested-id\"}".to_string()) - ); - assert_eq!( - values.get("nemo_relay.start.metadata.request.id"), - Some(&"flat-id".to_string()) - ); - assert_eq!( - values.get("nemo_relay.start.metadata.event_id"), - Some(&"18446744073709551615".to_string()) - ); - assert_eq!(values.get("tenant.id"), Some(&"acme".to_string())); - assert!(!values.contains_key("nemo_relay.start.metadata_json")); - } - - #[test] - fn rejects_invalid_attribute_mappings() { - assert!( - super::validate_attribute_mappings(&[OtlpAttributeMapping::new("", "alias")]).is_err() - ); - assert!( - super::validate_attribute_mappings(&[ - OtlpAttributeMapping::new("one", "duplicate"), - OtlpAttributeMapping::new("two", "duplicate"), - ]) - .is_err() - ); - assert!( - super::validate_attribute_mappings(&[ - OtlpAttributeMapping::new("one", "duplicate"), - OtlpAttributeMapping::new("two", " duplicate "), - ]) - .is_err() - ); - } -} +#[path = "../../tests/unit/observability/attribute_projection_tests.rs"] +mod attribute_projection_tests; diff --git a/crates/core/src/observability/plugin_component.rs b/crates/core/src/observability/plugin_component.rs index f9765bb48..dad7d8be1 100644 --- a/crates/core/src/observability/plugin_component.rs +++ b/crates/core/src/observability/plugin_component.rs @@ -1563,6 +1563,7 @@ fn validate_observability_section_fields( "enabled", "mark_projection", "mark_exclude_names", + "attribute_mappings", "transport", "endpoint", "headers", @@ -1583,6 +1584,7 @@ fn validate_observability_section_fields( "enabled", "mark_projection", "mark_exclude_names", + "attribute_mappings", "transport", "endpoint", "headers", diff --git a/crates/core/tests/unit/observability/attribute_projection_tests.rs b/crates/core/tests/unit/observability/attribute_projection_tests.rs new file mode 100644 index 000000000..0a5f27bf4 --- /dev/null +++ b/crates/core/tests/unit/observability/attribute_projection_tests.rs @@ -0,0 +1,112 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Unit tests for shared OTLP attribute projection. + +use super::{ + OtlpAttributeMapping, apply_attribute_mappings, attribute_mapping_inputs, + push_top_level_json_attributes, +}; + +#[test] +fn retains_only_mapping_sources_and_existing_aliases_between_span_events() { + let attributes = vec![ + opentelemetry::KeyValue::new("source", "value"), + opentelemetry::KeyValue::new("alias", "existing"), + opentelemetry::KeyValue::new("large.request", "payload"), + ]; + + let retained = + attribute_mapping_inputs(&attributes, &[OtlpAttributeMapping::new("source", "alias")]); + + assert_eq!(retained.len(), 2); + assert!( + retained + .iter() + .any(|attribute| attribute.key.as_str() == "source") + ); + assert!( + retained + .iter() + .any(|attribute| attribute.key.as_str() == "alias") + ); +} + +#[test] +fn projects_typed_json_and_copies_configured_aliases() { + let mut attributes = Vec::new(); + push_top_level_json_attributes( + &mut attributes, + "nemo_relay.start.metadata", + Some(&serde_json::json!({ + "tenant": "acme", + "attempt": 2, + "tags": ["a", "b"], + "context": {"region": "us-east-1"}, + "request": {"id": "nested-id"}, + "request.id": "flat-id", + "event_id": 18446744073709551615u64 + })), + ); + apply_attribute_mappings( + &mut attributes, + &[OtlpAttributeMapping::new( + "nemo_relay.start.metadata.tenant", + "tenant.id", + )], + ); + + let values = attributes + .iter() + .map(|attribute| (attribute.key.as_str(), attribute.value.to_string())) + .collect::>(); + assert_eq!( + values.get("nemo_relay.start.metadata.tenant"), + Some(&"acme".to_string()) + ); + assert_eq!( + values.get("nemo_relay.start.metadata.attempt"), + Some(&"2".to_string()) + ); + assert_eq!( + values.get("nemo_relay.start.metadata.tags"), + Some(&"[\"a\",\"b\"]".to_string()) + ); + assert_eq!( + values.get("nemo_relay.start.metadata.context"), + Some(&"{\"region\":\"us-east-1\"}".to_string()) + ); + assert_eq!( + values.get("nemo_relay.start.metadata.request"), + Some(&"{\"id\":\"nested-id\"}".to_string()) + ); + assert_eq!( + values.get("nemo_relay.start.metadata.request.id"), + Some(&"flat-id".to_string()) + ); + assert_eq!( + values.get("nemo_relay.start.metadata.event_id"), + Some(&"18446744073709551615".to_string()) + ); + assert_eq!(values.get("tenant.id"), Some(&"acme".to_string())); + assert!(!values.contains_key("nemo_relay.start.metadata_json")); +} + +#[test] +fn rejects_invalid_attribute_mappings() { + assert!(super::validate_attribute_mappings(&[OtlpAttributeMapping::new("", "alias")]).is_err()); + assert!( + super::validate_attribute_mappings(&[ + OtlpAttributeMapping::new("one", "duplicate"), + OtlpAttributeMapping::new("two", "duplicate"), + ]) + .is_err() + ); + assert!( + super::validate_attribute_mappings(&[ + OtlpAttributeMapping::new("one", "duplicate"), + OtlpAttributeMapping::new("two", " duplicate "), + ]) + .is_err() + ); +} diff --git a/crates/core/tests/unit/observability/plugin_component_tests.rs b/crates/core/tests/unit/observability/plugin_component_tests.rs index 78aec9c21..8aaabf59e 100644 --- a/crates/core/tests/unit/observability/plugin_component_tests.rs +++ b/crates/core/tests/unit/observability/plugin_component_tests.rs @@ -325,6 +325,27 @@ fn mark_projection_parses_for_otlp_and_rejects_unknown_values() { .message .contains("attribute mapping key must not be blank") })); + + let report = validate_plugin_config(&plugin_config(json!({ + "policy": {"unknown_field": "error"}, + "opentelemetry": { + "attribute_mappings": [{ + "key": "nemo_relay.start.metadata.tenant", + "alias": "tenant.id" + }] + }, + "openinference": { + "attribute_mappings": [{ + "key": "openinference.metadata.tenant", + "alias": "tenant.id" + }] + } + }))); + assert!( + !report.has_errors(), + "valid attribute mappings must not be reported as unknown: {:?}", + report.diagnostics + ); } #[cfg(feature = "schema")] From dcea96c04382000c53d24d4f235f83009e09c7d6 Mon Sep 17 00:00:00 2001 From: Will Killian Date: Tue, 14 Jul 2026 12:45:58 -0400 Subject: [PATCH 7/8] test(observability): cover typed projection edge cases Signed-off-by: Will Killian --- .../unit/observability/attribute_projection_tests.rs | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/crates/core/tests/unit/observability/attribute_projection_tests.rs b/crates/core/tests/unit/observability/attribute_projection_tests.rs index 0a5f27bf4..407b41f87 100644 --- a/crates/core/tests/unit/observability/attribute_projection_tests.rs +++ b/crates/core/tests/unit/observability/attribute_projection_tests.rs @@ -41,6 +41,8 @@ fn projects_typed_json_and_copies_configured_aliases() { Some(&serde_json::json!({ "tenant": "acme", "attempt": 2, + "enabled": true, + "unset": null, "tags": ["a", "b"], "context": {"region": "us-east-1"}, "request": {"id": "nested-id"}, @@ -68,6 +70,11 @@ fn projects_typed_json_and_copies_configured_aliases() { values.get("nemo_relay.start.metadata.attempt"), Some(&"2".to_string()) ); + assert_eq!( + values.get("nemo_relay.start.metadata.enabled"), + Some(&"true".to_string()) + ); + assert!(!values.contains_key("nemo_relay.start.metadata.unset")); assert_eq!( values.get("nemo_relay.start.metadata.tags"), Some(&"[\"a\",\"b\"]".to_string()) @@ -109,4 +116,7 @@ fn rejects_invalid_attribute_mappings() { ]) .is_err() ); + assert!( + super::validate_attribute_mappings(&[OtlpAttributeMapping::new("key", " ")]).is_err() + ); } From 15e1656aa515c6db070cb69fcc89ffca03005e5a Mon Sep 17 00:00:00 2001 From: Will Killian Date: Tue, 14 Jul 2026 12:54:59 -0400 Subject: [PATCH 8/8] test(observability): assert boolean projection type Signed-off-by: Will Killian --- .../unit/observability/attribute_projection_tests.rs | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/crates/core/tests/unit/observability/attribute_projection_tests.rs b/crates/core/tests/unit/observability/attribute_projection_tests.rs index 407b41f87..3b6dbc0d3 100644 --- a/crates/core/tests/unit/observability/attribute_projection_tests.rs +++ b/crates/core/tests/unit/observability/attribute_projection_tests.rs @@ -58,6 +58,14 @@ fn projects_typed_json_and_copies_configured_aliases() { )], ); + assert_eq!( + attributes + .iter() + .find(|attribute| attribute.key.as_str() == "nemo_relay.start.metadata.enabled") + .map(|attribute| &attribute.value), + Some(&opentelemetry::Value::Bool(true)) + ); + let values = attributes .iter() .map(|attribute| (attribute.key.as_str(), attribute.value.to_string())) @@ -70,10 +78,6 @@ fn projects_typed_json_and_copies_configured_aliases() { values.get("nemo_relay.start.metadata.attempt"), Some(&"2".to_string()) ); - assert_eq!( - values.get("nemo_relay.start.metadata.enabled"), - Some(&"true".to_string()) - ); assert!(!values.contains_key("nemo_relay.start.metadata.unset")); assert_eq!( values.get("nemo_relay.start.metadata.tags"),