From a04c0e55c10ef3a0cc1b7592098db2674d14d5c5 Mon Sep 17 00:00:00 2001 From: Will Killian Date: Tue, 4 Aug 2026 12:07:49 -0400 Subject: [PATCH] fix: canonicalize OpenTelemetry destinations Signed-off-by: Will Killian --- .../src/observability/plugin_component.rs | 107 ++++++++++++- .../observability/plugin_component_tests.rs | 145 +++++++++++++++++- .../observability/configuration.mdx | 8 +- .../observability/opentelemetry.mdx | 6 +- 4 files changed, 253 insertions(+), 13 deletions(-) diff --git a/crates/core/src/observability/plugin_component.rs b/crates/core/src/observability/plugin_component.rs index 4374e7da7..36976378d 100644 --- a/crates/core/src/observability/plugin_component.rs +++ b/crates/core/src/observability/plugin_component.rs @@ -21,6 +21,7 @@ use std::borrow::Cow; use std::collections::{HashMap, HashSet}; use std::future::Future; +use std::net::IpAddr; use std::panic::{AssertUnwindSafe, catch_unwind}; use std::path::{Component, Path, PathBuf}; use std::pin::Pin; @@ -2364,6 +2365,23 @@ struct OpenTelemetryDestinationCollision { message: String, } +#[derive(Debug, PartialEq, Eq)] +enum OpenTelemetryDestinationKey { + Url { + scheme: String, + host: String, + port: Option, + path: String, + query: Option, + }, + Raw(String), +} + +struct OpenTelemetryDestination { + key: OpenTelemetryDestinationKey, + display: String, +} + fn validate_distinct_opentelemetry_destinations( endpoints: &[OpenTelemetryEndpointConfig], ) -> PluginResult<()> { @@ -2385,7 +2403,7 @@ fn opentelemetry_destination_collision_errors( let endpoint_destination = opentelemetry_destination(endpoint); let other_destination = opentelemetry_destination(other); if endpoint.transport == other.transport - && endpoint_destination == other_destination + && endpoint_destination.key == other_destination.key && endpoint.otel_type != other.otel_type { errors.push(OpenTelemetryDestinationCollision { @@ -2395,7 +2413,7 @@ fn opentelemetry_destination_collision_errors( opentelemetry_type_name(other.otel_type), opentelemetry_type_name(endpoint.otel_type), endpoint.transport, - endpoint_destination, + endpoint_destination.display, ), }); } @@ -2404,13 +2422,94 @@ fn opentelemetry_destination_collision_errors( errors } -fn opentelemetry_destination(endpoint: &OpenTelemetryEndpointConfig) -> Cow<'_, str> { +fn opentelemetry_destination(endpoint: &OpenTelemetryEndpointConfig) -> OpenTelemetryDestination { let configured_endpoint = endpoint.endpoint.trim(); - if endpoint.transport == "http_binary" { + let effective_endpoint = if endpoint.transport == "http_binary" { resolve_http_trace_endpoint(configured_endpoint) } else { Cow::Borrowed(configured_endpoint) + }; + canonicalize_opentelemetry_destination(&effective_endpoint) +} + +fn canonicalize_opentelemetry_destination(endpoint: &str) -> OpenTelemetryDestination { + let Ok(url) = reqwest::Url::parse(endpoint) else { + return raw_opentelemetry_destination(endpoint); + }; + if !matches!(url.scheme(), "http" | "https") { + return raw_opentelemetry_destination(endpoint); + } + let Some(url_host) = url.host_str() else { + return raw_opentelemetry_destination(endpoint); + }; + + let scheme = url.scheme().to_string(); + let host = canonical_opentelemetry_host(url_host); + let port = url.port_or_known_default(); + let path = normalize_opentelemetry_path(url.path()); + let query = url.query().map(str::to_string); + let display = format!( + "{scheme}://{host}{}{path}{}", + port.map(|port| format!(":{port}")).unwrap_or_default(), + query + .as_deref() + .map(|query| format!("?{query}")) + .unwrap_or_default(), + ); + OpenTelemetryDestination { + key: OpenTelemetryDestinationKey::Url { + scheme, + host, + port, + path, + query, + }, + display, + } +} + +fn raw_opentelemetry_destination(endpoint: &str) -> OpenTelemetryDestination { + OpenTelemetryDestination { + key: OpenTelemetryDestinationKey::Raw(endpoint.to_string()), + display: endpoint.to_string(), + } +} + +fn canonical_opentelemetry_host(host: &str) -> String { + let domain = host.strip_suffix('.').unwrap_or(host); + let unbracketed = host + .strip_prefix('[') + .and_then(|host| host.strip_suffix(']')) + .unwrap_or(host); + let is_loopback_domain = domain == "localhost" || domain.ends_with(".localhost"); + let is_loopback_address = unbracketed + .parse::() + .is_ok_and(|address| address.is_loopback()); + if is_loopback_domain || is_loopback_address { + "".to_string() + } else { + host.to_string() + } +} + +fn normalize_opentelemetry_path(path: &str) -> String { + let mut normalized = String::with_capacity(path.len()); + let mut previous_was_slash = false; + for character in path.chars() { + if character == '/' { + if !previous_was_slash { + normalized.push(character); + } + previous_was_slash = true; + } else { + normalized.push(character); + previous_was_slash = false; + } + } + while normalized.len() > 1 && normalized.ends_with('/') { + normalized.pop(); } + normalized } const fn opentelemetry_type_name(otel_type: OpenTelemetryType) -> &'static str { diff --git a/crates/core/tests/unit/observability/plugin_component_tests.rs b/crates/core/tests/unit/observability/plugin_component_tests.rs index 494b7b731..eac9211e3 100644 --- a/crates/core/tests/unit/observability/plugin_component_tests.rs +++ b/crates/core/tests/unit/observability/plugin_component_tests.rs @@ -2707,7 +2707,140 @@ fn opentelemetry_endpoints_fan_out_to_heterogeneous_and_repeated_types() { } #[test] -fn opentelemetry_rejects_different_projection_types_at_the_same_effective_destination() { +fn opentelemetry_rejects_canonical_equivalent_destinations() { + for (first, second) in [ + ( + "http://collector.example/v1/traces", + "http://collector.example:80/v1/traces", + ), + ( + "https://collector.example/v1/traces", + "https://collector.example:443/v1/traces", + ), + ( + "HTTP://COLLECTOR.EXAMPLE/v1/traces", + "http://collector.example/v1/traces", + ), + ( + "http://collector.example//v1///traces", + "http://collector.example/v1/traces/", + ), + ("http://localhost/v1/traces", "http://LOCALHOST/v1/traces"), + ("http://localhost/v1/traces", "http://localhost./v1/traces"), + ( + "http://localhost/v1/traces", + "http://agent.localhost/v1/traces", + ), + ("http://localhost/v1/traces", "http://127.0.0.2/v1/traces"), + ("http://localhost/v1/traces", "http://127.1/v1/traces"), + ("http://localhost/v1/traces", "http://[::1]/v1/traces"), + ] { + let config = plugin_config(json!({ + "opentelemetry": { + "enabled": true, + "endpoints": [ + {"type": "full", "endpoint": first}, + {"type": "gen_ai", "endpoint": second} + ] + } + })); + + let report = validate_plugin_config(&config); + assert!( + report.diagnostics.iter().any(|diagnostic| { + diagnostic.code == "observability.unsafe_otel_destination_collision" + }), + "expected equivalent destinations {first:?} and {second:?} to collide" + ); + } + + let grpc = plugin_config(json!({ + "opentelemetry": { + "enabled": true, + "endpoints": [ + { + "type": "full", + "transport": "grpc", + "endpoint": "https://collector.example" + }, + { + "type": "gen_ai", + "transport": "grpc", + "endpoint": "https://collector.example:443/" + } + ] + } + })); + assert!(validate_plugin_config(&grpc).has_errors()); +} + +#[test] +fn opentelemetry_allows_distinct_canonical_destinations() { + for (first, second) in [ + ( + "http://collector.example:4318/v1/traces", + "http://collector.example:4319/v1/traces", + ), + ( + "http://collector.example:443/v1/traces", + "https://collector.example/v1/traces", + ), + ( + "http://collector.example/v1/traces", + "http://collector.example/custom/traces", + ), + ( + "http://collector.example/v1/traces", + "http://collector.example/v1%2Ftraces", + ), + ( + "http://collector.example/v1/traces?tenant=one", + "http://collector.example/v1/traces?tenant=two", + ), + ( + "http://localhost.example/v1/traces", + "http://localhost/v1/traces", + ), + ("http://[::2]/v1/traces", "http://localhost/v1/traces"), + ] { + let config = plugin_config(json!({ + "opentelemetry": { + "enabled": true, + "endpoints": [ + {"type": "full", "endpoint": first}, + {"type": "gen_ai", "endpoint": second} + ] + } + })); + + assert!( + !validate_plugin_config(&config).has_errors(), + "expected distinct destinations {first:?} and {second:?} to remain valid" + ); + } + + let different_transports = plugin_config(json!({ + "opentelemetry": { + "enabled": true, + "endpoints": [ + { + "type": "full", + "transport": "http_binary", + "endpoint": "http://collector.example/v1/traces" + }, + { + "type": "gen_ai", + "transport": "grpc", + "endpoint": "http://collector.example/v1/traces" + } + ] + } + })); + assert!(!validate_plugin_config(&different_transports).has_errors()); +} + +#[test] +fn opentelemetry_rejects_canonical_collision_during_validation_and_activation() { let _guard = crate::observability::test_mutex().lock().unwrap(); reset_runtime(); let config = plugin_config(json!({ @@ -2715,8 +2848,8 @@ fn opentelemetry_rejects_different_projection_types_at_the_same_effective_destin "opentelemetry": { "enabled": true, "endpoints": [ - {"type": "full", "endpoint": " http://127.0.0.1:4318 "}, - {"type": "gen_ai", "endpoint": "http://127.0.0.1:4318/v1/traces"} + {"type": "full", "endpoint": " http://LOCALHOST:80//v1///traces/ "}, + {"type": "gen_ai", "endpoint": "http://127.1/v1/traces"} ] } })); @@ -2730,7 +2863,7 @@ fn opentelemetry_rejects_different_projection_types_at_the_same_effective_destin && diagnostic.message.contains("endpoints[1] (gen_ai)") && diagnostic .message - .contains("http://127.0.0.1:4318/v1/traces") + .contains("http://:80/v1/traces") })); assert!(futures::executor::block_on(initialize_plugins_exact(config)).is_err()); assert!( @@ -2748,8 +2881,8 @@ fn opentelemetry_allows_repeated_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/v1/traces"} + {"type": "full", "endpoint": "http://LOCALHOST:80//v1///traces/"}, + {"type": "full", "endpoint": "http://127.1/v1/traces"} ] } })); diff --git a/docs/configure-plugins/observability/configuration.mdx b/docs/configure-plugins/observability/configuration.mdx index 2cf596424..88b8b61ca 100644 --- a/docs/configure-plugins/observability/configuration.mdx +++ b/docs/configure-plugins/observability/configuration.mdx @@ -71,8 +71,12 @@ endpoint and transport: Relay rejects that configuration because their deterministic trace and span IDs would 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. +collide. The comparison realizes HTTP port `80` and HTTPS port `443`, collapses +repeated path slashes, ignores a non-root trailing slash, and treats standardized +loopback forms (`localhost`, names under `.localhost`, `127.0.0.0/8`, and `::1`) +as the same host without resolving DNS. Query strings remain part of the +destination. 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 25187fa80..c57880b35 100644 --- a/docs/configure-plugins/observability/opentelemetry.mdx +++ b/docs/configure-plugins/observability/opentelemetry.mdx @@ -31,7 +31,11 @@ compliant trace and span IDs from Relay lifecycle UUIDs, so endpoints that 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. +collisions at the receiver. Duplicate detection compares canonical destinations: +HTTP and HTTPS default ports are realized, repeated and trailing path slashes +are normalized, and standardized loopback hosts such as `localhost`, names +under `.localhost`, `127.0.0.0/8`, and `::1` are equivalent. Relay does not use +DNS resolution for this comparison, and query strings remain significant. 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