From a0d9e217f656b602d05f49cdfd9592931936c198 Mon Sep 17 00:00:00 2001 From: Will Killian Date: Mon, 3 Aug 2026 12:59:02 -0400 Subject: [PATCH 1/3] fix: reject colliding typed OTLP destinations Signed-off-by: Will Killian --- .../src/observability/plugin_component.rs | 61 +++++++++++++++++++ .../observability/plugin_component_tests.rs | 55 ++++++++++++++++- .../observability/configuration.mdx | 9 ++- .../observability/opentelemetry.mdx | 5 +- 4 files changed, 124 insertions(+), 6 deletions(-) diff --git a/crates/core/src/observability/plugin_component.rs b/crates/core/src/observability/plugin_component.rs index 0dc46285e..d5c28941b 100644 --- a/crates/core/src/observability/plugin_component.rs +++ b/crates/core/src/observability/plugin_component.rs @@ -974,6 +974,7 @@ fn register_opentelemetry( "enabled OpenTelemetry section requires at least one endpoint".to_string(), )); } + validate_distinct_opentelemetry_destinations(§ion.endpoints)?; let subscribers = build_opentelemetry_subscribers(section.endpoints)?; for (index, _) in subscribers.iter().enumerate() { log::info!( @@ -2233,9 +2234,69 @@ fn validate_opentelemetry_section( } validate_opentelemetry_headers(diagnostics, policy, index, endpoint); } + for error in opentelemetry_destination_collision_errors(§ion.endpoints) { + diagnostics.push(ConfigDiagnostic { + level: DiagnosticLevel::Error, + code: "observability.unsafe_otel_destination_collision".to_string(), + component: Some("opentelemetry".to_string()), + field: Some(format!("endpoints[{}].endpoint", error.index)), + message: error.message, + }); + } validate_opentelemetry_feature_support(diagnostics, policy, section); } +struct OpenTelemetryDestinationCollision { + index: usize, + message: String, +} + +fn validate_distinct_opentelemetry_destinations( + endpoints: &[OpenTelemetryEndpointConfig], +) -> PluginResult<()> { + if let Some(error) = opentelemetry_destination_collision_errors(endpoints) + .into_iter() + .next() + { + return Err(PluginError::InvalidConfig(error.message)); + } + Ok(()) +} + +fn opentelemetry_destination_collision_errors( + endpoints: &[OpenTelemetryEndpointConfig], +) -> Vec { + let mut errors = Vec::new(); + for (index, endpoint) in endpoints.iter().enumerate() { + for (other_index, other) in endpoints[..index].iter().enumerate() { + if endpoint.transport == other.transport + && endpoint.endpoint.trim() == other.endpoint.trim() + && endpoint.otel_type != other.otel_type + { + errors.push(OpenTelemetryDestinationCollision { + index, + message: format!( + "OpenTelemetry endpoints[{other_index}] ({}) and endpoints[{index}] ({}) use the same {} destination {:?}; different projection types must use independent destinations", + opentelemetry_type_name(other.otel_type), + opentelemetry_type_name(endpoint.otel_type), + endpoint.transport, + endpoint.endpoint.trim(), + ), + }); + } + } + } + errors +} + +const fn opentelemetry_type_name(otel_type: OpenTelemetryType) -> &'static str { + match otel_type { + OpenTelemetryType::Full => "full", + OpenTelemetryType::GenAi => "gen_ai", + OpenTelemetryType::OpenInference => "openinference", + } +} + fn validate_opentelemetry_headers( diagnostics: &mut Vec, policy: &ConfigPolicy, diff --git a/crates/core/tests/unit/observability/plugin_component_tests.rs b/crates/core/tests/unit/observability/plugin_component_tests.rs index cadc388c0..9a9b5b8dc 100644 --- a/crates/core/tests/unit/observability/plugin_component_tests.rs +++ b/crates/core/tests/unit/observability/plugin_component_tests.rs @@ -2552,12 +2552,12 @@ fn otlp_sections_register_inferred_subscribers_with_full_config() { }, { "type": "openinference", - "endpoint": "http://127.0.0.1:4318/v1/traces", + "endpoint": "http://127.0.0.1:4319/v1/traces", "service_name": "oi-service" }, { "type": "gen_ai", - "endpoint": "http://127.0.0.1:4318/v1/traces" + "endpoint": "http://127.0.0.1:4320/v1/traces" } ] } @@ -2616,6 +2616,57 @@ fn opentelemetry_endpoints_fan_out_to_heterogeneous_and_repeated_types() { } } +#[test] +fn opentelemetry_rejects_different_projection_types_at_the_same_destination() { + let _guard = crate::observability::test_mutex().lock().unwrap(); + reset_runtime(); + let config = plugin_config(json!({ + "policy": {"unsupported_value": "ignore"}, + "opentelemetry": { + "enabled": true, + "endpoints": [ + {"type": "full", "endpoint": " http://127.0.0.1:4318/v1/traces "}, + {"type": "gen_ai", "endpoint": "http://127.0.0.1:4318/v1/traces"} + ] + } + })); + + let report = validate_plugin_config(&config); + assert!(report.has_errors()); + assert!(report.diagnostics.iter().any(|diagnostic| { + diagnostic.code == "observability.unsafe_otel_destination_collision" + && diagnostic.field.as_deref() == Some("endpoints[1].endpoint") + && diagnostic.message.contains("endpoints[0] (full)") + && diagnostic.message.contains("endpoints[1] (gen_ai)") + && diagnostic + .message + .contains("http://127.0.0.1:4318/v1/traces") + })); + assert!(futures::executor::block_on(initialize_plugins_exact(config)).is_err()); + assert!( + !global_context() + .read() + .unwrap() + .event_subscribers + .contains_key("__nemo_relay_plugin__observability__opentelemetry") + ); +} + +#[test] +fn opentelemetry_allows_repeated_projection_types_at_the_same_destination() { + let config = plugin_config(json!({ + "opentelemetry": { + "enabled": true, + "endpoints": [ + {"type": "full", "endpoint": "http://127.0.0.1:4318/v1/traces"}, + {"type": "full", "endpoint": "http://127.0.0.1:4318/v1/traces"} + ] + } + })); + + assert!(!validate_plugin_config(&config).has_errors()); +} + #[test] fn opentelemetry_endpoint_delivery_failure_does_not_block_other_endpoints() { let _guard = crate::observability::test_mutex().lock().unwrap(); diff --git a/docs/configure-plugins/observability/configuration.mdx b/docs/configure-plugins/observability/configuration.mdx index 8552b2538..06eccada3 100644 --- a/docs/configure-plugins/observability/configuration.mdx +++ b/docs/configure-plugins/observability/configuration.mdx @@ -55,7 +55,7 @@ service_name = "agent-service" [[components.config.opentelemetry.endpoints]] type = "gen_ai" -endpoint = "http://localhost:4318/v1/traces" +endpoint = "http://localhost:4319/v1/traces" service_name = "agent-service" [[components.config.opentelemetry.endpoints]] @@ -65,8 +65,11 @@ service_name = "agent-service" ``` When OpenTelemetry is enabled, `endpoints` must contain at least one entry. -Endpoint types can be combined or repeated with independent destinations. All -endpoints are constructed before the plugin registers its fan-out subscriber. +Endpoint types can be combined or repeated with independent destinations. Two +different endpoint types must not use the same endpoint and transport: Relay +rejects that configuration because their deterministic trace and span IDs would +collide at the receiver. All endpoints are constructed before the plugin +registers its fan-out subscriber. ## Multi-Endpoint Lifecycle diff --git a/docs/configure-plugins/observability/opentelemetry.mdx b/docs/configure-plugins/observability/opentelemetry.mdx index 5fbe79548..be8ad346c 100644 --- a/docs/configure-plugins/observability/opentelemetry.mdx +++ b/docs/configure-plugins/observability/opentelemetry.mdx @@ -28,7 +28,10 @@ with collector or backend schemas. NeMo Relay uses OpenTelemetry Rust `0.32`. It deterministically derives compliant trace and span IDs from Relay lifecycle UUIDs, so endpoints that -receive the same event stream use the same identifiers and parentage. +receive the same event stream use the same identifiers and parentage. Different +endpoint types must therefore use independent OTLP destinations; configuring +them with the same endpoint and transport is rejected to prevent identifier +collisions at the receiver. Rooted Relay propagation continues the Relay-derived trace across the import boundary. Rootless propagation retains Relay event parentage but starts a new OpenTelemetry trace from the first local event. Carry W3C `traceparent` and From e09d49c6c601ef9217e86a1004e740450e286ebc Mon Sep 17 00:00:00 2001 From: Will Killian <2007799+willkill07@users.noreply.github.com> Date: Mon, 3 Aug 2026 14:49:37 -0400 Subject: [PATCH 2/3] Update docs/configure-plugins/observability/configuration.mdx Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com> Signed-off-by: Will Killian <2007799+willkill07@users.noreply.github.com> --- docs/configure-plugins/observability/configuration.mdx | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/docs/configure-plugins/observability/configuration.mdx b/docs/configure-plugins/observability/configuration.mdx index 06eccada3..92cc90014 100644 --- a/docs/configure-plugins/observability/configuration.mdx +++ b/docs/configure-plugins/observability/configuration.mdx @@ -65,8 +65,9 @@ service_name = "agent-service" ``` When OpenTelemetry is enabled, `endpoints` must contain at least one entry. -Endpoint types can be combined or repeated with independent destinations. Two -different endpoint types must not use the same endpoint and transport: Relay +Endpoint types can be combined. Repeated endpoint types can use the same +endpoint and transport. Two different endpoint types must not use the same +endpoint and transport: Relay rejects that configuration because their deterministic trace and span IDs would collide at the receiver. All endpoints are constructed before the plugin registers its fan-out subscriber. From 0ff422d5e2c32296fb019d91933766d3c325f57e Mon Sep 17 00:00:00 2001 From: Will Killian Date: Mon, 3 Aug 2026 15:02:26 -0400 Subject: [PATCH 3/3] fix: normalize OTLP collision destinations Signed-off-by: Will Killian --- .../core/src/observability/plugin_component.rs | 17 +++++++++++++++-- .../observability/plugin_component_tests.rs | 4 ++-- .../observability/configuration.mdx | 4 +++- 3 files changed, 20 insertions(+), 5 deletions(-) diff --git a/crates/core/src/observability/plugin_component.rs b/crates/core/src/observability/plugin_component.rs index ab45a6ecf..116460fa4 100644 --- a/crates/core/src/observability/plugin_component.rs +++ b/crates/core/src/observability/plugin_component.rs @@ -18,6 +18,7 @@ //! metadata; their declared scope type is preserved in the exported event //! stream. +use std::borrow::Cow; use std::collections::{HashMap, HashSet}; use std::future::Future; use std::panic::{AssertUnwindSafe, catch_unwind}; @@ -52,6 +53,7 @@ use crate::observability::atof::{ }; use crate::observability::otel::{ OpenTelemetryConfig as CoreOpenTelemetryConfig, OpenTelemetrySubscriber, OtlpTransport, + resolve_http_trace_endpoint, }; use crate::observability::{ MarkProjection, OpenTelemetryType, OtlpAttributeMapping, default_mark_exclude_names, @@ -2286,8 +2288,10 @@ fn opentelemetry_destination_collision_errors( let mut errors = Vec::new(); for (index, endpoint) in endpoints.iter().enumerate() { for (other_index, other) in endpoints[..index].iter().enumerate() { + let endpoint_destination = opentelemetry_destination(endpoint); + let other_destination = opentelemetry_destination(other); if endpoint.transport == other.transport - && endpoint.endpoint.trim() == other.endpoint.trim() + && endpoint_destination == other_destination && endpoint.otel_type != other.otel_type { errors.push(OpenTelemetryDestinationCollision { @@ -2297,7 +2301,7 @@ fn opentelemetry_destination_collision_errors( opentelemetry_type_name(other.otel_type), opentelemetry_type_name(endpoint.otel_type), endpoint.transport, - endpoint.endpoint.trim(), + endpoint_destination, ), }); } @@ -2306,6 +2310,15 @@ fn opentelemetry_destination_collision_errors( errors } +fn opentelemetry_destination(endpoint: &OpenTelemetryEndpointConfig) -> Cow<'_, str> { + let configured_endpoint = endpoint.endpoint.trim(); + if endpoint.transport == "http_binary" { + resolve_http_trace_endpoint(configured_endpoint) + } else { + Cow::Borrowed(configured_endpoint) + } +} + const fn opentelemetry_type_name(otel_type: OpenTelemetryType) -> &'static str { match otel_type { OpenTelemetryType::Full => "full", diff --git a/crates/core/tests/unit/observability/plugin_component_tests.rs b/crates/core/tests/unit/observability/plugin_component_tests.rs index 9a9b5b8dc..745be8c04 100644 --- a/crates/core/tests/unit/observability/plugin_component_tests.rs +++ b/crates/core/tests/unit/observability/plugin_component_tests.rs @@ -2617,7 +2617,7 @@ fn opentelemetry_endpoints_fan_out_to_heterogeneous_and_repeated_types() { } #[test] -fn opentelemetry_rejects_different_projection_types_at_the_same_destination() { +fn opentelemetry_rejects_different_projection_types_at_the_same_effective_destination() { let _guard = crate::observability::test_mutex().lock().unwrap(); reset_runtime(); let config = plugin_config(json!({ @@ -2625,7 +2625,7 @@ fn opentelemetry_rejects_different_projection_types_at_the_same_destination() { "opentelemetry": { "enabled": true, "endpoints": [ - {"type": "full", "endpoint": " http://127.0.0.1:4318/v1/traces "}, + {"type": "full", "endpoint": " http://127.0.0.1:4318 "}, {"type": "gen_ai", "endpoint": "http://127.0.0.1:4318/v1/traces"} ] } diff --git a/docs/configure-plugins/observability/configuration.mdx b/docs/configure-plugins/observability/configuration.mdx index 92cc90014..a89534893 100644 --- a/docs/configure-plugins/observability/configuration.mdx +++ b/docs/configure-plugins/observability/configuration.mdx @@ -69,7 +69,9 @@ Endpoint types can be combined. Repeated endpoint types can use the same endpoint and transport. Two different endpoint types must not use the same endpoint and transport: Relay rejects that configuration because their deterministic trace and span IDs would -collide at the receiver. All endpoints are constructed before the plugin +collide at the receiver. For `http_binary`, this comparison uses the effective +trace destination, so a bare URL and the same URL with `/v1/traces` also +collide. All endpoints are constructed before the plugin registers its fan-out subscriber. ## Multi-Endpoint Lifecycle