diff --git a/crates/cli/src/diagnostics/mod.rs b/crates/cli/src/diagnostics/mod.rs index 428f03d39..00e93c824 100644 --- a/crates/cli/src/diagnostics/mod.rs +++ b/crates/cli/src/diagnostics/mod.rs @@ -25,6 +25,7 @@ use std::time::Duration; use futures_util::{SinkExt, future::join_all}; use nemo_relay::api::event::{BaseEvent, Event, MarkEvent}; use nemo_relay::codec::model_pricing::{PricingCatalog, PricingConfig, PricingSourceConfig}; +use nemo_relay::observability::otel::resolve_http_trace_endpoint; use nemo_relay::observability::plugin_component::OBSERVABILITY_PLUGIN_KIND; use nemo_relay::plugin::{DiagnosticLevel, PluginConfig, validate_plugin_config}; use nemo_relay_adaptive::plugin_component::ADAPTIVE_PLUGIN_KIND; @@ -798,7 +799,8 @@ async fn observability_http_exporter_checks(config: &Value) -> Vec { let mut check = if transport == "grpc" { probe_tcp_named(label, url).await } else { - probe_http_named(label, url).await + let effective_url = resolve_http_trace_endpoint(url); + probe_otlp_http_named(label, effective_url.as_ref()).await }; check.details = format!("endpoints[{index}] ({endpoint_type}): {}", check.details); diff --git a/crates/cli/src/diagnostics/probes.rs b/crates/cli/src/diagnostics/probes.rs index 88ee1f750..2625e0c56 100644 --- a/crates/cli/src/diagnostics/probes.rs +++ b/crates/cli/src/diagnostics/probes.rs @@ -44,7 +44,7 @@ pub(super) fn check_dir_writable(directory: &Path) -> Result<(), std::io::Error> std::fs::remove_file(probe) } -pub(super) async fn probe_http_named(name: &'static str, url: &str) -> Check { +pub(super) async fn probe_otlp_http_named(name: &'static str, url: &str) -> Check { let client = match reqwest::Client::builder().timeout(NETWORK_TIMEOUT).build() { Ok(client) => client, Err(error) => { @@ -58,7 +58,10 @@ pub(super) async fn probe_http_named(name: &'static str, url: &str) -> Check { match client.get(url).send().await { Ok(response) => Check { name, - status: if response.status().is_success() || response.status().is_redirection() { + status: if response.status().is_success() + || response.status().is_redirection() + || response.status() == reqwest::StatusCode::METHOD_NOT_ALLOWED + { Status::Pass } else { Status::Warn diff --git a/crates/cli/tests/coverage/shared/doctor_tests.rs b/crates/cli/tests/coverage/shared/doctor_tests.rs index a16133830..590df44ef 100644 --- a/crates/cli/tests/coverage/shared/doctor_tests.rs +++ b/crates/cli/tests/coverage/shared/doctor_tests.rs @@ -980,7 +980,7 @@ async fn opentelemetry_doctor_uses_tcp_probe_for_grpc_endpoints() { } #[tokio::test] -async fn opentelemetry_doctor_covers_http_missing_and_malformed_endpoints() { +async fn opentelemetry_doctor_resolves_bare_http_endpoints_and_warns_on_missing_routes() { assert!( observability_http_exporter_checks(&serde_json::json!({ "opentelemetry": {"enabled": true, "endpoints": "not-a-list"} @@ -1006,7 +1006,7 @@ async fn opentelemetry_doctor_covers_http_missing_and_malformed_endpoints() { let mut stream = accept_bounded(&listener); let _ = read_headers(&mut stream); stream - .write_all(b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\n\r\n") + .write_all(b"HTTP/1.1 405 Method Not Allowed\r\nContent-Length: 0\r\n\r\n") .unwrap(); }); let checks = observability_http_exporter_checks(&serde_json::json!({ @@ -1018,6 +1018,49 @@ async fn opentelemetry_doctor_covers_http_missing_and_malformed_endpoints() { .await; assert_eq!(checks[0].status, Status::Pass); assert!(checks[0].details.contains("endpoints[0] (full)")); + assert!(checks[0].details.contains("/v1/traces (HTTP 405)")); + accept.join().unwrap(); + + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let endpoint = format!("http://{}/", listener.local_addr().unwrap()); + let accept = std::thread::spawn(move || { + let mut stream = accept_bounded(&listener); + let request = read_headers(&mut stream); + stream + .write_all(b"HTTP/1.1 405 Method Not Allowed\r\nContent-Length: 0\r\n\r\n") + .unwrap(); + request + }); + let checks = observability_http_exporter_checks(&serde_json::json!({ + "opentelemetry": { + "enabled": true, + "endpoints": [{"type": "full", "endpoint": endpoint}] + } + })) + .await; + assert_eq!(checks[0].status, Status::Pass); + assert!(checks[0].details.contains("/ (HTTP 405)")); + let request = accept.join().unwrap(); + assert!(request.starts_with("GET / HTTP/1.1")); + + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let endpoint = format!("http://{}/wrong", listener.local_addr().unwrap()); + let accept = std::thread::spawn(move || { + let mut stream = accept_bounded(&listener); + let _ = read_headers(&mut stream); + stream + .write_all(b"HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\n\r\n") + .unwrap(); + }); + let checks = observability_http_exporter_checks(&serde_json::json!({ + "opentelemetry": { + "enabled": true, + "endpoints": [{"type": "full", "endpoint": endpoint}] + } + })) + .await; + assert_eq!(checks[0].status, Status::Warn); + assert!(checks[0].details.contains("/wrong (HTTP 404)")); accept.join().unwrap(); } @@ -1620,7 +1663,7 @@ async fn atof_http_and_websocket_timeout_errors_are_reported() { } #[tokio::test] -async fn probe_http_named_warns_on_http_errors() { +async fn otlp_http_probe_warns_on_http_errors() { let listener = TcpListener::bind("127.0.0.1:0").unwrap(); let url = format!("http://{}", listener.local_addr().unwrap()); let handle = std::thread::spawn(move || { @@ -1632,14 +1675,14 @@ async fn probe_http_named_warns_on_http_errors() { .unwrap(); }); - let check = probe_http_named("OpenTelemetry endpoint", &url).await; + let check = probe_otlp_http_named("OpenTelemetry endpoint", &url).await; assert_eq!(check.status, Status::Warn); assert!(check.details.contains("HTTP 500")); handle.join().unwrap(); } #[tokio::test] -async fn http_probe_passes_success_and_ndjson_upload_success() { +async fn otlp_http_probe_passes_success_method_not_allowed_and_ndjson_upload_success() { let success_listener = TcpListener::bind("127.0.0.1:0").unwrap(); let success_url = format!("http://{}", success_listener.local_addr().unwrap()); let success_handle = std::thread::spawn(move || { @@ -1650,10 +1693,25 @@ async fn http_probe_passes_success_and_ndjson_upload_success() { .write_all(b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\n\r\n") .unwrap(); }); - let check = probe_http_named("OpenTelemetry endpoint", &success_url).await; + let check = probe_otlp_http_named("OpenTelemetry endpoint", &success_url).await; assert_eq!(check.status, Status::Pass); success_handle.join().unwrap(); + let method_listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let method_url = format!("http://{}", method_listener.local_addr().unwrap()); + let method_handle = std::thread::spawn(move || { + let mut stream = accept_bounded(&method_listener); + let mut buf = [0_u8; 1024]; + let _ = stream.read(&mut buf).unwrap(); + stream + .write_all(b"HTTP/1.1 405 Method Not Allowed\r\nContent-Length: 0\r\n\r\n") + .unwrap(); + }); + let check = probe_otlp_http_named("OpenTelemetry endpoint", &method_url).await; + assert_eq!(check.status, Status::Pass); + assert!(check.details.contains("HTTP 405")); + method_handle.join().unwrap(); + let (url, body, server_thread) = start_doctor_http_capture_server(); let check = probe_atof_ndjson( &url, diff --git a/crates/core/src/observability/otel.rs b/crates/core/src/observability/otel.rs index ccd428a39..5999442d1 100644 --- a/crates/core/src/observability/otel.rs +++ b/crates/core/src/observability/otel.rs @@ -14,6 +14,7 @@ //! - [`OpenTelemetrySubscriber`] exposes a NeMo Relay [`EventSubscriberFn`] and //! convenience `register` / `deregister` / `force_flush` / `shutdown` methods +use std::borrow::Cow; use std::cell::RefCell; use std::collections::{HashMap, HashSet, VecDeque}; use std::sync::mpsc; @@ -149,6 +150,27 @@ pub enum OtlpTransport { Grpc, } +/// Completes a bare OTLP/HTTP base URL with the standard trace signal path. +#[doc(hidden)] +pub fn resolve_http_trace_endpoint(endpoint: &str) -> Cow<'_, str> { + let Ok(mut parsed) = reqwest::Url::parse(endpoint) else { + return Cow::Borrowed(endpoint); + }; + + let has_explicit_root_path = endpoint + .split(['?', '#']) + .next() + .is_some_and(|url| url.ends_with('/')); + if !matches!(parsed.scheme(), "http" | "https") + || parsed.path() != "/" + || has_explicit_root_path + { + return Cow::Borrowed(endpoint); + } + parsed.set_path("/v1/traces"); + Cow::Owned(parsed.into()) +} + /// Configuration for the OpenTelemetry subscriber. #[derive(Debug, Clone)] pub struct OpenTelemetryConfig { @@ -681,7 +703,8 @@ fn build_tracer_provider(config: &OpenTelemetryConfig) -> Result