From 89417d9411b8461fdc4b7ee8fb2bbcca04704b80 Mon Sep 17 00:00:00 2001 From: mnajafian-nv Date: Mon, 3 Aug 2026 15:29:55 -0700 Subject: [PATCH 1/4] fix: add offline mode to doctor probes Signed-off-by: mnajafian-nv --- crates/cli/src/commands/diagnostics.rs | 3 + crates/cli/src/commands/mod.rs | 9 +- crates/cli/src/commands/root.rs | 2 +- crates/cli/src/diagnostics/mod.rs | 134 ++++++++++-- crates/cli/src/diagnostics/probes.rs | 47 ++-- .../cli/tests/coverage/commands/main_tests.rs | 9 + .../cli/tests/coverage/shared/doctor_tests.rs | 200 +++++++++++++++--- .../cli/tests/coverage/shared/probes_tests.rs | 15 +- 8 files changed, 356 insertions(+), 63 deletions(-) diff --git a/crates/cli/src/commands/diagnostics.rs b/crates/cli/src/commands/diagnostics.rs index 35034c671..dd6ad46a6 100644 --- a/crates/cli/src/commands/diagnostics.rs +++ b/crates/cli/src/commands/diagnostics.rs @@ -21,6 +21,8 @@ pub(crate) struct DoctorCommand { pub(crate) install_dir: Option, #[arg(long)] pub(crate) json: bool, + #[arg(long)] + pub(crate) offline: bool, } #[derive(Debug, Clone, Args)] @@ -41,6 +43,7 @@ pub(super) async fn execute( crate::diagnostics::run_doctor( command.agent.map(Into::into), command.json, + crate::diagnostics::DoctorProbeMode::from_offline_flag(command.offline), &gateway_overrides, logging_fallback_error, ) diff --git a/crates/cli/src/commands/mod.rs b/crates/cli/src/commands/mod.rs index 79f9f19d4..9771ba937 100644 --- a/crates/cli/src/commands/mod.rs +++ b/crates/cli/src/commands/mod.rs @@ -233,7 +233,14 @@ async fn run_default( .await?; Ok(ExitCode::SUCCESS) } else if runtime_configuration::any_config_file_exists() { - runtime_diagnostics::run_doctor(None, false, &runtime_args, None).await + runtime_diagnostics::run_doctor( + None, + false, + runtime_diagnostics::DoctorProbeMode::Live, + &runtime_args, + None, + ) + .await } else { configure::run(None, None).await?; Ok(ExitCode::SUCCESS) diff --git a/crates/cli/src/commands/root.rs b/crates/cli/src/commands/root.rs index 9cce1033a..6b8c4f48f 100644 --- a/crates/cli/src/commands/root.rs +++ b/crates/cli/src/commands/root.rs @@ -112,7 +112,7 @@ pub(crate) enum Command { Uninstall(UninstallCommand), /// Validate and configure model pricing catalogs. ModelPricing(PricingCommand), - /// Diagnose env, agents, config, observability (optionally scoped to one agent) + /// Diagnose env, agents, config, observability (use --offline to skip live network probes) Doctor(DoctorCommand), /// List supported and locally-detected agents (use `--json` for machine output) Agents(AgentsCommand), diff --git a/crates/cli/src/diagnostics/mod.rs b/crates/cli/src/diagnostics/mod.rs index 1a7dc94fc..10c1f4625 100644 --- a/crates/cli/src/diagnostics/mod.rs +++ b/crates/cli/src/diagnostics/mod.rs @@ -46,6 +46,22 @@ use crate::server::{GatewayOverrides, register_and_validate_plugin_components}; const NETWORK_TIMEOUT: Duration = Duration::from_secs(2); const PRICING_PLUGIN_KIND: &str = "pricing"; +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum DoctorProbeMode { + Live, + Offline, +} + +impl DoctorProbeMode { + pub(crate) fn from_offline_flag(offline: bool) -> Self { + if offline { Self::Offline } else { Self::Live } + } + + fn is_offline(self) -> bool { + matches!(self, Self::Offline) + } +} + struct PluginConfigurationDiagnostics { sources: Vec, error: Option, @@ -57,6 +73,7 @@ struct PluginConfigurationDiagnostics { /// the first missing directory. pub(crate) async fn collect_report( target_agent: Option, + probe_mode: DoctorProbeMode, gateway_overrides: &GatewayOverrides, ) -> Result { let (resolved, resolution) = match resolve_server_config(gateway_overrides) { @@ -131,7 +148,7 @@ pub(crate) async fn collect_report( ), agents: collect_agents(target_agent, &resolved).await, host_plugins: crate::agents::collect_default_integration_readiness(), - observability: collect_observability(&resolved.gateway).await, + observability: collect_observability(&resolved.gateway, probe_mode).await, completions: collect_completions(home.as_deref()), }) } @@ -569,7 +586,7 @@ async fn probe_version(argv: &[String]) -> Option { } } -async fn collect_observability(gateway: &GatewayConfig) -> Vec { +async fn collect_observability(gateway: &GatewayConfig, probe_mode: DoctorProbeMode) -> Vec { let mut checks = Vec::new(); let Some(plugin_value) = &gateway.plugin_config else { @@ -630,7 +647,7 @@ async fn collect_observability(gateway: &GatewayConfig) -> Vec { } if let Some(config) = observability_component_config(plugin_value) { - collect_observability_component_checks(&mut checks, config).await; + collect_observability_component_checks(&mut checks, config, probe_mode).await; } else { checks.push(Check { name: "Observability plugin", @@ -639,8 +656,13 @@ async fn collect_observability(gateway: &GatewayConfig) -> Vec { }); } collect_pricing_component_checks(&mut checks, &plugin_config); - collect_response_cache_component_checks(&mut checks, &plugin_config, response_cache_invalid) - .await; + collect_response_cache_component_checks( + &mut checks, + &plugin_config, + response_cache_invalid, + probe_mode, + ) + .await; checks } @@ -649,6 +671,7 @@ async fn collect_response_cache_component_checks( checks: &mut Vec, plugin_config: &PluginConfig, config_invalid: bool, + probe_mode: DoctorProbeMode, ) { // The response cache is a section of the adaptive component, not its own // plugin kind: find the adaptive component and look for `response_cache`. @@ -705,6 +728,17 @@ async fn collect_response_cache_component_checks( return; } }; + if probe_mode.is_offline() && config.backend.kind != "in_memory" { + checks.push(Check { + name: "Response cache", + status: Status::Info, + details: format!( + "configured; live {} backend probe skipped (--offline)", + config.backend.kind + ), + }); + return; + } checks.push(response_cache_backend_check(response_cache::check_backend_health(&config)).await); } @@ -733,15 +767,19 @@ async fn response_cache_backend_check( } } -async fn collect_observability_component_checks(checks: &mut Vec, config: &Value) { +async fn collect_observability_component_checks( + checks: &mut Vec, + config: &Value, + probe_mode: DoctorProbeMode, +) { checks.extend(observability_atof_file_checks(config)); if let Some(check) = observability_file_exporter_check(config, "atif") { checks.push(check); } - checks.extend(observability_http_exporter_checks(config).await); + checks.extend(observability_http_exporter_checks(config, probe_mode).await); if section_enabled(config, "atof") && !atof_stream_sinks(config).is_empty() { if atof_streaming_supported() { - checks.extend(observability_atof_stream_checks(config).await); + checks.extend(observability_atof_stream_checks(config, probe_mode).await); } else { checks.push(Check { name: "ATOF stream sink", @@ -810,7 +848,10 @@ fn observability_file_exporter_check(config: &Value, section: &str) -> Option Vec { +async fn observability_http_exporter_checks( + config: &Value, + probe_mode: DoctorProbeMode, +) -> Vec { if !section_enabled(config, "opentelemetry") { return Vec::new(); } @@ -838,13 +879,53 @@ async fn observability_http_exporter_checks(config: &Value) -> Vec { match endpoint.get("endpoint").and_then(Value::as_str) { Some(url) => { let mut check = if transport == "grpc" { - probe_tcp_named(label, url).await + if probe_mode.is_offline() { + match validate_grpc_endpoint(url) { + Ok(_) => Check { + name: label, + status: Status::Info, + details: format!( + "endpoints[{index}] ({endpoint_type}): live network probe skipped (--offline)" + ), + }, + Err(details) => Check { + name: label, + status: Status::Fail, + details: format!( + "endpoints[{index}] ({endpoint_type}): {details}" + ), + }, + } + } else { + probe_tcp_named(label, url).await + } } else { let effective_url = resolve_http_trace_endpoint(url); - probe_otlp_http_named(label, effective_url.as_ref()).await + if probe_mode.is_offline() { + match validate_otlp_http_endpoint(effective_url.as_ref()) { + Ok(()) => Check { + name: label, + status: Status::Info, + details: format!( + "endpoints[{index}] ({endpoint_type}): live network probe skipped (--offline)" + ), + }, + Err(details) => Check { + name: label, + status: Status::Fail, + details: format!( + "endpoints[{index}] ({endpoint_type}): {details}" + ), + }, + } + } else { + probe_otlp_http_named(label, effective_url.as_ref()).await + } }; - check.details = - format!("endpoints[{index}] ({endpoint_type}): {}", check.details); + if !probe_mode.is_offline() { + check.details = + format!("endpoints[{index}] ({endpoint_type}): {}", check.details); + } check } None => Check { @@ -993,16 +1074,23 @@ fn atof_streaming_supported() -> bool { cfg!(feature = "atof-streaming") } -async fn observability_atof_stream_checks(config: &Value) -> Vec { +async fn observability_atof_stream_checks( + config: &Value, + probe_mode: DoctorProbeMode, +) -> Vec { let streams = atof_stream_sinks(config); let mut checks = Vec::with_capacity(streams.len()); for (index, sink) in streams { - checks.push(probe_atof_stream_sink(index, sink).await); + checks.push(probe_atof_stream_sink(index, sink, probe_mode).await); } checks } -async fn probe_atof_stream_sink(index: usize, endpoint: &Value) -> Check { +async fn probe_atof_stream_sink( + index: usize, + endpoint: &Value, + probe_mode: DoctorProbeMode, +) -> Check { let name = "ATOF stream sink"; let Some(url) = endpoint.get("url").and_then(Value::as_str) else { return Check { @@ -1036,6 +1124,15 @@ async fn probe_atof_stream_sink(index: usize, endpoint: &Value) -> Check { }; } }; + if probe_mode.is_offline() { + return Check { + name, + status: Status::Info, + details: format!( + "sinks[{index}] {transport} {url}: live network probe skipped (--offline)" + ), + }; + } let payload = match doctor_atof_probe_payload() { Ok(payload) => payload, Err(err) => { @@ -1061,7 +1158,7 @@ async fn probe_atof_stream_sink(index: usize, endpoint: &Value) -> Check { #[cfg(test)] async fn probe_atof_endpoint(index: usize, endpoint: &Value) -> Check { - probe_atof_stream_sink(index, endpoint).await + probe_atof_stream_sink(index, endpoint, DoctorProbeMode::Live).await } fn endpoint_headers(endpoint: &Value) -> Result, String> { @@ -1374,10 +1471,11 @@ pub(crate) fn format_agents_json(agents: &[AgentInfo]) -> Result, json: bool, + probe_mode: DoctorProbeMode, gateway_overrides: &GatewayOverrides, logging_fallback_error: Option<&CliError>, ) -> Result { - let mut report = collect_report(target_agent, gateway_overrides).await?; + let mut report = collect_report(target_agent, probe_mode, gateway_overrides).await?; if let Some(error) = logging_fallback_error { let logging_details = format!( "could not resolve logging configuration: {error}; repair or recreate the named logging configuration file" diff --git a/crates/cli/src/diagnostics/probes.rs b/crates/cli/src/diagnostics/probes.rs index 2625e0c56..e70e35d2b 100644 --- a/crates/cli/src/diagnostics/probes.rs +++ b/crates/cli/src/diagnostics/probes.rs @@ -66,34 +66,31 @@ pub(super) async fn probe_otlp_http_named(name: &'static str, url: &str) -> Chec } else { Status::Warn }, - details: format!("{} (HTTP {})", url, response.status().as_u16()), + details: format!( + "{} (live HTTP reachability probe returned HTTP {})", + url, + response.status().as_u16() + ), }, Err(error) => Check { name, status: Status::Fail, - details: format!("{url}: {error}"), + details: format!("{url}: live HTTP reachability probe failed: {error}"), }, } } pub(super) async fn probe_tcp_named(name: &'static str, endpoint: &str) -> Check { - let parsed = match reqwest::Url::parse(endpoint) { - Ok(parsed) => parsed, - Err(error) => { + let (parsed, host) = match validate_grpc_endpoint(endpoint) { + Ok((parsed, host)) => (parsed, host), + Err(details) => { return Check { name, status: Status::Fail, - details: format!("{endpoint}: invalid gRPC endpoint: {error}"), + details, }; } }; - let Some(host) = parsed.host_str() else { - return Check { - name, - status: Status::Fail, - details: format!("{endpoint}: gRPC endpoint has no host"), - }; - }; let port = grpc_endpoint_port(&parsed); match tokio::time::timeout( NETWORK_TIMEOUT, @@ -104,21 +101,39 @@ pub(super) async fn probe_tcp_named(name: &'static str, endpoint: &str) -> Check Ok(Ok(_)) => Check { name, status: Status::Pass, - details: format!("{endpoint} (gRPC TCP connection succeeded)"), + details: format!( + "{endpoint} (live gRPC reachability probe connected to the TCP port; OTLP handshake not verified)" + ), }, Ok(Err(error)) => Check { name, status: Status::Fail, - details: format!("{endpoint}: gRPC TCP connection failed: {error}"), + details: format!("{endpoint}: live gRPC reachability probe failed: {error}"), }, Err(_) => Check { name, status: Status::Fail, - details: format!("{endpoint}: gRPC TCP connection timed out"), + details: format!("{endpoint}: live gRPC reachability probe timed out"), }, } } +pub(super) fn validate_grpc_endpoint(endpoint: &str) -> Result<(reqwest::Url, String), String> { + let parsed = reqwest::Url::parse(endpoint) + .map_err(|error| format!("{endpoint}: invalid gRPC endpoint: {error}"))?; + let host = parsed + .host_str() + .map(str::to_owned) + .ok_or_else(|| format!("{endpoint}: gRPC endpoint has no host"))?; + Ok((parsed, host)) +} + +pub(super) fn validate_otlp_http_endpoint(endpoint: &str) -> Result<(), String> { + reqwest::Url::parse(endpoint) + .map(|_| ()) + .map_err(|error| format!("{endpoint}: invalid OTLP HTTP endpoint: {error}")) +} + fn grpc_endpoint_port(endpoint: &reqwest::Url) -> u16 { endpoint.port().unwrap_or_else(|| { if endpoint.scheme() == "https" { diff --git a/crates/cli/tests/coverage/commands/main_tests.rs b/crates/cli/tests/coverage/commands/main_tests.rs index 2dfec86c5..2967a50ad 100644 --- a/crates/cli/tests/coverage/commands/main_tests.rs +++ b/crates/cli/tests/coverage/commands/main_tests.rs @@ -353,6 +353,15 @@ fn doctor_rejects_conflicting_agent_and_plugin_targets() { assert!(error.to_string().contains("cannot be used with")); } +#[test] +fn doctor_accepts_offline_flag() { + let cli = Cli::try_parse_from(["nemo-relay", "doctor", "--offline"]).unwrap(); + match cli.command { + Some(Command::Doctor(command)) => assert!(command.offline), + other => panic!("expected doctor command, got {other:?}"), + } +} + #[test] fn multi_agent_operations_attempt_every_target_before_reporting_errors() { let visited = std::cell::RefCell::new(Vec::new()); diff --git a/crates/cli/tests/coverage/shared/doctor_tests.rs b/crates/cli/tests/coverage/shared/doctor_tests.rs index 2e3d9d054..5dd84d11c 100644 --- a/crates/cli/tests/coverage/shared/doctor_tests.rs +++ b/crates/cli/tests/coverage/shared/doctor_tests.rs @@ -101,6 +101,27 @@ fn empty_report() -> DoctorReport { } } +async fn collect_live_observability(gateway: &GatewayConfig) -> Vec { + collect_observability(gateway, DoctorProbeMode::Live).await +} + +async fn collect_offline_observability(gateway: &GatewayConfig) -> Vec { + collect_observability(gateway, DoctorProbeMode::Offline).await +} + +async fn live_observability_http_exporter_checks(config: &serde_json::Value) -> Vec { + observability_http_exporter_checks(config, DoctorProbeMode::Live).await +} + +async fn offline_observability_http_exporter_checks(config: &serde_json::Value) -> Vec { + observability_http_exporter_checks(config, DoctorProbeMode::Offline).await +} + +#[cfg(test)] +async fn offline_atof_endpoint(index: usize, endpoint: &serde_json::Value) -> Check { + probe_atof_stream_sink(index, endpoint, DoctorProbeMode::Offline).await +} + #[test] fn exit_code_passes_when_no_failures() { let report = empty_report(); @@ -1066,11 +1087,16 @@ async fn opentelemetry_doctor_uses_tcp_probe_for_grpc_endpoints() { } }); - let checks = observability_http_exporter_checks(&config).await; + let checks = live_observability_http_exporter_checks(&config).await; assert_eq!(checks.len(), 1); assert_eq!(checks[0].status, Status::Pass); - assert!(checks[0].details.contains("gRPC TCP connection succeeded")); + assert!( + checks[0] + .details + .contains("live gRPC reachability probe connected to the TCP port") + ); + assert!(checks[0].details.contains("OTLP handshake not verified")); assert!(checks[0].details.contains("endpoints[0] (gen_ai)")); accept.join().unwrap(); } @@ -1078,14 +1104,14 @@ async fn opentelemetry_doctor_uses_tcp_probe_for_grpc_endpoints() { #[tokio::test] async fn opentelemetry_doctor_resolves_bare_http_endpoints_and_warns_on_missing_routes() { assert!( - observability_http_exporter_checks(&serde_json::json!({ + live_observability_http_exporter_checks(&serde_json::json!({ "opentelemetry": {"enabled": true, "endpoints": "not-a-list"} })) .await .is_empty() ); - let missing = observability_http_exporter_checks(&serde_json::json!({ + let missing = live_observability_http_exporter_checks(&serde_json::json!({ "opentelemetry": { "enabled": true, "endpoints": [{"type": "openinference"}] @@ -1105,7 +1131,7 @@ async fn opentelemetry_doctor_resolves_bare_http_endpoints_and_warns_on_missing_ .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!({ + let checks = live_observability_http_exporter_checks(&serde_json::json!({ "opentelemetry": { "enabled": true, "endpoints": [{"type": "full", "endpoint": endpoint}] @@ -1114,7 +1140,11 @@ async fn opentelemetry_doctor_resolves_bare_http_endpoints_and_warns_on_missing_ .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)")); + assert!( + checks[0] + .details + .contains("/v1/traces (live HTTP reachability probe returned HTTP 405)") + ); accept.join().unwrap(); let listener = TcpListener::bind("127.0.0.1:0").unwrap(); @@ -1127,7 +1157,7 @@ async fn opentelemetry_doctor_resolves_bare_http_endpoints_and_warns_on_missing_ .unwrap(); request }); - let checks = observability_http_exporter_checks(&serde_json::json!({ + let checks = live_observability_http_exporter_checks(&serde_json::json!({ "opentelemetry": { "enabled": true, "endpoints": [{"type": "full", "endpoint": endpoint}] @@ -1135,7 +1165,11 @@ async fn opentelemetry_doctor_resolves_bare_http_endpoints_and_warns_on_missing_ })) .await; assert_eq!(checks[0].status, Status::Pass); - assert!(checks[0].details.contains("/ (HTTP 405)")); + assert!( + checks[0] + .details + .contains("/ (live HTTP reachability probe returned HTTP 405)") + ); let request = accept.join().unwrap(); assert!(request.starts_with("GET / HTTP/1.1")); @@ -1148,7 +1182,7 @@ async fn opentelemetry_doctor_resolves_bare_http_endpoints_and_warns_on_missing_ .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!({ + let checks = live_observability_http_exporter_checks(&serde_json::json!({ "opentelemetry": { "enabled": true, "endpoints": [{"type": "full", "endpoint": endpoint}] @@ -1156,10 +1190,69 @@ async fn opentelemetry_doctor_resolves_bare_http_endpoints_and_warns_on_missing_ })) .await; assert_eq!(checks[0].status, Status::Warn); - assert!(checks[0].details.contains("/wrong (HTTP 404)")); + assert!( + checks[0] + .details + .contains("/wrong (live HTTP reachability probe returned HTTP 404)") + ); accept.join().unwrap(); } +#[tokio::test] +async fn opentelemetry_doctor_skips_live_network_probes_offline() { + let checks = offline_observability_http_exporter_checks(&serde_json::json!({ + "opentelemetry": { + "enabled": true, + "endpoints": [ + { + "type": "gen_ai", + "transport": "grpc", + "endpoint": "http://127.0.0.1:4317" + }, + { + "type": "openinference", + "endpoint": "http://127.0.0.1:4318" + } + ] + } + })) + .await; + + assert_eq!(checks.len(), 2); + assert!(checks.iter().all(|check| check.status == Status::Info)); + assert!(checks.iter().all(|check| { + check + .details + .contains("live network probe skipped (--offline)") + })); +} + +#[tokio::test] +async fn opentelemetry_doctor_offline_still_rejects_malformed_endpoints() { + let checks = offline_observability_http_exporter_checks(&serde_json::json!({ + "opentelemetry": { + "enabled": true, + "endpoints": [ + { + "type": "gen_ai", + "transport": "grpc", + "endpoint": "http://:4317" + }, + { + "type": "full", + "endpoint": "http://:4318" + } + ] + } + })) + .await; + + assert_eq!(checks.len(), 2); + assert!(checks[0].details.contains("invalid gRPC endpoint")); + assert!(checks[1].details.contains("invalid OTLP HTTP endpoint")); + assert!(checks.iter().all(|check| check.status == Status::Fail)); +} + #[test] fn atof_file_checks_preserve_configured_sink_indices() { let config = serde_json::json!({ @@ -1214,7 +1307,7 @@ async fn collect_observability_warns_for_missing_atif_dir_without_creating_it() ..GatewayConfig::default() }; - let checks = collect_observability(&gateway).await; + let checks = collect_live_observability(&gateway).await; let atif_check = checks .iter() @@ -1251,7 +1344,7 @@ async fn collect_observability_registers_adaptive_before_validation() { ..GatewayConfig::default() }; - let checks = collect_observability(&gateway).await; + let checks = collect_live_observability(&gateway).await; assert!( !checks.iter().any(|check| check @@ -1285,7 +1378,7 @@ async fn collect_observability_reports_response_cache_on_when_configured() { ..GatewayConfig::default() }; - let checks = collect_observability(&gateway).await; + let checks = collect_live_observability(&gateway).await; let cache = checks .iter() @@ -1299,6 +1392,47 @@ async fn collect_observability_reports_response_cache_on_when_configured() { ); } +#[tokio::test] +async fn collect_observability_skips_redis_response_cache_probe_offline() { + let gateway = GatewayConfig { + plugin_config: Some(serde_json::json!({ + "version": 1, + "components": [ + { + "kind": "adaptive", + "enabled": true, + "config": { + "response_cache": { + "ttl_seconds": 3600, + "namespace": "doctor-test", + "backend": { + "kind": "redis", + "config": { + "url": "redis://127.0.0.1:6379", + "key_prefix": "doctor-test:" + } + } + } + } + } + ] + })), + ..GatewayConfig::default() + }; + + let checks = collect_offline_observability(&gateway).await; + + let cache = checks + .iter() + .find(|check| check.name == "Response cache") + .expect("a Response cache check should be present"); + assert_eq!(cache.status, Status::Info, "checks: {checks:?}"); + assert_eq!( + cache.details, + "configured; live redis backend probe skipped (--offline)" + ); +} + #[tokio::test(start_paused = true)] async fn response_cache_backend_check_reports_timeout() { let check = response_cache_backend_check(std::future::pending()).await; @@ -1333,7 +1467,7 @@ async fn collect_observability_reports_response_cache_not_configured_without_sec ..GatewayConfig::default() }; - let checks = collect_observability(&gateway).await; + let checks = collect_live_observability(&gateway).await; let cache = checks .iter() @@ -1365,7 +1499,7 @@ async fn collect_observability_reports_response_cache_fail_when_config_invalid() ..GatewayConfig::default() }; - let checks = collect_observability(&gateway).await; + let checks = collect_live_observability(&gateway).await; let cache = checks .iter() @@ -1415,7 +1549,7 @@ async fn collect_observability_registers_pii_redaction_before_validation() { ..GatewayConfig::default() }; - let checks = collect_observability(&gateway).await; + let checks = collect_live_observability(&gateway).await; assert!( !checks.iter().any(|check| check @@ -1447,7 +1581,7 @@ async fn collect_observability_reports_invalid_pii_redaction_config() { ..GatewayConfig::default() }; - let checks = collect_observability(&gateway).await; + let checks = collect_live_observability(&gateway).await; let diagnostic = checks .iter() @@ -1483,7 +1617,7 @@ async fn collect_observability_probes_atof_streaming_endpoint() { ..GatewayConfig::default() }; - let checks = collect_observability(&gateway).await; + let checks = collect_live_observability(&gateway).await; let body = tokio::time::timeout(std::time::Duration::from_secs(2), async { loop { let captured = body.lock().unwrap().clone(); @@ -1510,11 +1644,11 @@ async fn collect_observability_probes_atof_streaming_endpoint() { #[tokio::test] async fn collect_observability_covers_absent_invalid_and_componentless_configs() { - let absent = collect_observability(&GatewayConfig::default()).await; + let absent = collect_live_observability(&GatewayConfig::default()).await; assert_eq!(absent[0].status, Status::Info); assert!(absent[0].details.contains("not configured")); - let invalid = collect_observability(&GatewayConfig { + let invalid = collect_live_observability(&GatewayConfig { plugin_config: Some(serde_json::json!({"version": "bad"})), ..GatewayConfig::default() }) @@ -1522,7 +1656,7 @@ async fn collect_observability_covers_absent_invalid_and_componentless_configs() assert_eq!(invalid[0].status, Status::Fail); assert!(invalid[0].details.contains("invalid plugin config")); - let no_observability = collect_observability(&GatewayConfig { + let no_observability = collect_live_observability(&GatewayConfig { plugin_config: Some(serde_json::json!({ "version": 1, "components": [] @@ -1567,7 +1701,7 @@ async fn collect_observability_rejects_websocket_endpoint_http_scheme() { ..GatewayConfig::default() }; - let checks = collect_observability(&gateway).await; + let checks = collect_live_observability(&gateway).await; let endpoint = checks .iter() @@ -1649,6 +1783,24 @@ async fn atof_endpoint_validation_rejects_missing_url_headers_timeout_and_transp assert!(unsupported.details.contains("unsupported transport")); } +#[tokio::test] +async fn atof_endpoint_offline_skips_live_network_probe_after_validation() { + let skipped = offline_atof_endpoint( + 0, + &serde_json::json!({ + "url": "http://127.0.0.1:1/events", + "transport": "http_post" + }), + ) + .await; + + assert_eq!(skipped.status, Status::Info); + assert_eq!( + skipped.details, + "sinks[0] http_post http://127.0.0.1:1/events: live network probe skipped (--offline)" + ); +} + #[tokio::test] async fn atof_http_and_websocket_probes_report_failure_branches() { let listener = TcpListener::bind("127.0.0.1:0").unwrap(); @@ -1869,7 +2021,7 @@ async fn collect_observability_validates_pricing_file_source() { ..GatewayConfig::default() }; - let checks = collect_observability(&gateway).await; + let checks = collect_live_observability(&gateway).await; let pricing = checks .iter() @@ -1901,7 +2053,7 @@ async fn collect_observability_fails_for_missing_pricing_file_source() { ..GatewayConfig::default() }; - let checks = collect_observability(&gateway).await; + let checks = collect_live_observability(&gateway).await; let pricing = checks .iter() diff --git a/crates/cli/tests/coverage/shared/probes_tests.rs b/crates/cli/tests/coverage/shared/probes_tests.rs index 17335fa38..b6c47abe5 100644 --- a/crates/cli/tests/coverage/shared/probes_tests.rs +++ b/crates/cli/tests/coverage/shared/probes_tests.rs @@ -9,7 +9,12 @@ async fn grpc_probe_uses_tcp_connectivity() { let endpoint = format!("http://{}", listener.local_addr().unwrap()); let check = probe_tcp_named("OpenTelemetry endpoint", &endpoint).await; assert_eq!(check.status, Status::Pass); - assert!(check.details.contains("gRPC TCP connection succeeded")); + assert!( + check + .details + .contains("live gRPC reachability probe connected to the TCP port") + ); + assert!(check.details.contains("OTLP handshake not verified")); } #[tokio::test] @@ -28,8 +33,12 @@ async fn grpc_probe_reports_invalid_hostless_and_refused_endpoints() { let refused = probe_tcp_named("OpenTelemetry endpoint", &endpoint).await; assert_eq!(refused.status, Status::Fail); assert!( - refused.details.contains("connection failed") - || refused.details.contains("connection timed out"), + refused + .details + .contains("live gRPC reachability probe failed") + || refused + .details + .contains("live gRPC reachability probe timed out"), "{}", refused.details ); From 200b5fb0f75ec0875bbb9cad6dbffd6642111a78 Mon Sep 17 00:00:00 2001 From: mnajafian-nv Date: Mon, 3 Aug 2026 15:55:51 -0700 Subject: [PATCH 2/4] fix: validate offline doctor endpoints Signed-off-by: mnajafian-nv --- crates/cli/src/diagnostics/mod.rs | 40 +++++++++++ crates/cli/src/diagnostics/probes.rs | 19 +++++- .../cli/tests/coverage/shared/doctor_tests.rs | 68 +++++++++++++++++++ .../cli/tests/coverage/shared/probes_tests.rs | 6 +- 4 files changed, 129 insertions(+), 4 deletions(-) diff --git a/crates/cli/src/diagnostics/mod.rs b/crates/cli/src/diagnostics/mod.rs index 10c1f4625..f6b64d459 100644 --- a/crates/cli/src/diagnostics/mod.rs +++ b/crates/cli/src/diagnostics/mod.rs @@ -1124,6 +1124,13 @@ async fn probe_atof_stream_sink( }; } }; + if let Err(details) = validate_atof_stream_probe_target(index, transport, url) { + return Check { + name, + status: Status::Fail, + details, + }; + } if probe_mode.is_offline() { return Check { name, @@ -1161,6 +1168,39 @@ async fn probe_atof_endpoint(index: usize, endpoint: &Value) -> Check { probe_atof_stream_sink(index, endpoint, DoctorProbeMode::Live).await } +fn validate_atof_stream_probe_target( + index: usize, + transport: &str, + url: &str, +) -> Result<(), String> { + let parsed = reqwest::Url::parse(url) + .map_err(|error| format!("sinks[{index}] {transport} {url}: {error}"))?; + if parsed.host_str().is_none() { + return Err(format!("sinks[{index}] {transport} {url}: missing host")); + } + let valid_scheme = match transport { + "http_post" | "ndjson" => matches!(parsed.scheme(), "http" | "https"), + "websocket" => matches!(parsed.scheme(), "ws" | "wss"), + _ => { + return Err(format!( + "sinks[{index}] {transport} {url}: unsupported transport" + )); + } + }; + if valid_scheme { + Ok(()) + } else { + let expected = match transport { + "http_post" | "ndjson" => "http or https", + "websocket" => "ws or wss", + _ => unreachable!("unsupported transports return earlier"), + }; + Err(format!( + "sinks[{index}] {transport} {url}: invalid scheme (must be {expected})" + )) + } +} + fn endpoint_headers(endpoint: &Value) -> Result, String> { let mut out = Vec::new(); let mut names = std::collections::HashSet::new(); diff --git a/crates/cli/src/diagnostics/probes.rs b/crates/cli/src/diagnostics/probes.rs index e70e35d2b..4ba55df76 100644 --- a/crates/cli/src/diagnostics/probes.rs +++ b/crates/cli/src/diagnostics/probes.rs @@ -121,6 +121,11 @@ pub(super) async fn probe_tcp_named(name: &'static str, endpoint: &str) -> Check pub(super) fn validate_grpc_endpoint(endpoint: &str) -> Result<(reqwest::Url, String), String> { let parsed = reqwest::Url::parse(endpoint) .map_err(|error| format!("{endpoint}: invalid gRPC endpoint: {error}"))?; + if !matches!(parsed.scheme(), "http" | "https") { + return Err(format!( + "{endpoint}: gRPC endpoint must use http:// or https://" + )); + } let host = parsed .host_str() .map(str::to_owned) @@ -129,9 +134,17 @@ pub(super) fn validate_grpc_endpoint(endpoint: &str) -> Result<(reqwest::Url, St } pub(super) fn validate_otlp_http_endpoint(endpoint: &str) -> Result<(), String> { - reqwest::Url::parse(endpoint) - .map(|_| ()) - .map_err(|error| format!("{endpoint}: invalid OTLP HTTP endpoint: {error}")) + let parsed = reqwest::Url::parse(endpoint) + .map_err(|error| format!("{endpoint}: invalid OTLP HTTP endpoint: {error}"))?; + if !matches!(parsed.scheme(), "http" | "https") { + return Err(format!( + "{endpoint}: OTLP HTTP endpoint must use http:// or https://" + )); + } + if parsed.host_str().is_none() { + return Err(format!("{endpoint}: OTLP HTTP endpoint has no host")); + } + Ok(()) } fn grpc_endpoint_port(endpoint: &reqwest::Url) -> u16 { diff --git a/crates/cli/tests/coverage/shared/doctor_tests.rs b/crates/cli/tests/coverage/shared/doctor_tests.rs index 5dd84d11c..b88c4a40c 100644 --- a/crates/cli/tests/coverage/shared/doctor_tests.rs +++ b/crates/cli/tests/coverage/shared/doctor_tests.rs @@ -1253,6 +1253,41 @@ async fn opentelemetry_doctor_offline_still_rejects_malformed_endpoints() { assert!(checks.iter().all(|check| check.status == Status::Fail)); } +#[tokio::test] +async fn opentelemetry_doctor_offline_rejects_unsupported_endpoint_schemes() { + let checks = offline_observability_http_exporter_checks(&serde_json::json!({ + "opentelemetry": { + "enabled": true, + "endpoints": [ + { + "type": "gen_ai", + "transport": "grpc", + "endpoint": "ftp://collector.example:4317" + }, + { + "type": "full", + "endpoint": "file:///tmp/collector" + } + ] + } + })) + .await; + + assert_eq!(checks.len(), 2); + assert_eq!(checks[0].status, Status::Fail); + assert_eq!(checks[1].status, Status::Fail); + assert!( + checks[0] + .details + .contains("gRPC endpoint must use http:// or https://") + ); + assert!( + checks[1] + .details + .contains("OTLP HTTP endpoint must use http:// or https://") + ); +} + #[test] fn atof_file_checks_preserve_configured_sink_indices() { let config = serde_json::json!({ @@ -1801,6 +1836,39 @@ async fn atof_endpoint_offline_skips_live_network_probe_after_validation() { ); } +#[tokio::test] +async fn atof_endpoint_offline_still_rejects_invalid_transport_and_scheme() { + let invalid_websocket = offline_atof_endpoint( + 0, + &serde_json::json!({ + "url": "http://127.0.0.1:1/events", + "transport": "websocket" + }), + ) + .await; + assert_eq!(invalid_websocket.status, Status::Fail); + assert!( + invalid_websocket + .details + .contains("invalid scheme (must be ws or wss)") + ); + + let unsupported_transport = offline_atof_endpoint( + 1, + &serde_json::json!({ + "url": "http://127.0.0.1:1/events", + "transport": "udp" + }), + ) + .await; + assert_eq!(unsupported_transport.status, Status::Fail); + assert!( + unsupported_transport + .details + .contains("unsupported transport") + ); +} + #[tokio::test] async fn atof_http_and_websocket_probes_report_failure_branches() { let listener = TcpListener::bind("127.0.0.1:0").unwrap(); diff --git a/crates/cli/tests/coverage/shared/probes_tests.rs b/crates/cli/tests/coverage/shared/probes_tests.rs index b6c47abe5..bb2958a25 100644 --- a/crates/cli/tests/coverage/shared/probes_tests.rs +++ b/crates/cli/tests/coverage/shared/probes_tests.rs @@ -25,7 +25,11 @@ async fn grpc_probe_reports_invalid_hostless_and_refused_endpoints() { let hostless = probe_tcp_named("OpenTelemetry endpoint", "file:///tmp/collector").await; assert_eq!(hostless.status, Status::Fail); - assert!(hostless.details.contains("has no host")); + assert!( + hostless + .details + .contains("gRPC endpoint must use http:// or https://") + ); let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); let endpoint = format!("http://{}", listener.local_addr().unwrap()); From 6bfcd39807e2767ce7525881590ed0e6f00cb4c4 Mon Sep 17 00:00:00 2001 From: mnajafian-nv Date: Mon, 3 Aug 2026 17:39:39 -0700 Subject: [PATCH 3/4] fix: validate doctor offline probe inputs Signed-off-by: mnajafian-nv --- crates/adaptive/src/response_cache/store.rs | 53 ++++++-- crates/cli/src/commands/diagnostics.rs | 5 +- crates/cli/src/diagnostics/mod.rs | 21 ++- .../cli/tests/coverage/shared/doctor_tests.rs | 121 +++++++++++++++++- 4 files changed, 186 insertions(+), 14 deletions(-) diff --git a/crates/adaptive/src/response_cache/store.rs b/crates/adaptive/src/response_cache/store.rs index 001074005..281a581fc 100644 --- a/crates/adaptive/src/response_cache/store.rs +++ b/crates/adaptive/src/response_cache/store.rs @@ -445,6 +445,18 @@ pub async fn check_backend_health( Ok(store.backend_kind().to_string()) } +/// Validates the configured backend target without attempting network access. +/// +/// Intended for offline `nemo-relay doctor` checks: it rejects malformed +/// backend targets while still skipping live reachability probes. +pub fn validate_backend_target(config: &ResponseCacheConfig) -> std::result::Result<(), String> { + match config.backend.kind.as_str() { + "in_memory" => Ok(()), + "redis" => validate_redis_backend_target(config).map_err(|err| err.to_string()), + other => Err(format!("response_cache: unknown backend kind '{other}'")), + } +} + /// Builds the response-cache backend from config. /// /// Returns the boxed [`CacheStore`] used by the intercept and the `doctor` @@ -465,16 +477,7 @@ pub(crate) async fn build_store(config: &ResponseCacheConfig) -> Result Result> { use crate::response_cache::store::RedisCacheStore; - let url = config - .backend - .config - .get("url") - .and_then(Json::as_str) - .ok_or_else(|| { - AdaptiveError::InvalidConfig( - "response_cache: redis backend requires backend.config.url".to_string(), - ) - })?; + let url = redis_backend_url(config)?; let key_prefix = config .backend .config @@ -496,6 +499,36 @@ async fn build_redis_store(_config: &ResponseCacheConfig) -> Result Result<&str> { + config + .backend + .config + .get("url") + .and_then(Json::as_str) + .ok_or_else(|| { + AdaptiveError::InvalidConfig( + "response_cache: redis backend requires backend.config.url".to_string(), + ) + }) +} + +#[cfg(feature = "redis-backend")] +fn validate_redis_backend_target(config: &ResponseCacheConfig) -> Result<()> { + let url = redis_backend_url(config)?; + redis::Client::open(url) + .map(|_| ()) + .map_err(|err| AdaptiveError::Storage(format!("response_cache: redis client: {err}"))) +} + +#[cfg(not(feature = "redis-backend"))] +fn validate_redis_backend_target(_config: &ResponseCacheConfig) -> Result<()> { + Err(AdaptiveError::InvalidConfig( + "response_cache: backend.kind = \"redis\" requires building with the 'redis-backend' \ + feature" + .to_string(), + )) +} + #[cfg(test)] #[path = "../../tests/unit/response_cache/store_tests.rs"] mod tests; diff --git a/crates/cli/src/commands/diagnostics.rs b/crates/cli/src/commands/diagnostics.rs index dd6ad46a6..ccf6cbe08 100644 --- a/crates/cli/src/commands/diagnostics.rs +++ b/crates/cli/src/commands/diagnostics.rs @@ -21,7 +21,10 @@ pub(crate) struct DoctorCommand { pub(crate) install_dir: Option, #[arg(long)] pub(crate) json: bool, - #[arg(long)] + #[arg( + long, + help = "Validate configuration and endpoint syntax without running live network probes" + )] pub(crate) offline: bool, } diff --git a/crates/cli/src/diagnostics/mod.rs b/crates/cli/src/diagnostics/mod.rs index f6b64d459..6a325ca0d 100644 --- a/crates/cli/src/diagnostics/mod.rs +++ b/crates/cli/src/diagnostics/mod.rs @@ -728,6 +728,16 @@ async fn collect_response_cache_component_checks( return; } }; + if probe_mode.is_offline() + && let Err(error) = response_cache::store::validate_backend_target(&config) + { + checks.push(Check { + name: "Response cache", + status: Status::Fail, + details: format!("invalid backend target: {error}"), + }); + return; + } if probe_mode.is_offline() && config.backend.kind != "in_memory" { checks.push(Check { name: "Response cache", @@ -1214,7 +1224,14 @@ fn endpoint_headers(endpoint: &Value) -> Result, String> { let Some(value) = value.as_str() else { return Err(format!("headers.{key} must be a string")); }; - names.insert(name); + if value.trim().is_empty() { + return Err(format!("headers.{key} must not be blank")); + } + reqwest::header::HeaderValue::from_bytes(value.as_bytes()) + .map_err(|error| format!("headers.{key} invalid: {error}"))?; + if !names.insert(name) { + return Err(format!("header {key:?} appears more than once")); + } out.push((key.clone(), value.to_string())); } } @@ -1238,6 +1255,8 @@ fn endpoint_headers(endpoint: &Value) -> Result, String> { if value.trim().is_empty() { return Err(format!("environment variable {variable:?} is blank")); } + reqwest::header::HeaderValue::from_bytes(value.as_bytes()) + .map_err(|error| format!("header_env.{key} invalid: {error}"))?; names.insert(name); out.push((key.clone(), value)); } diff --git a/crates/cli/tests/coverage/shared/doctor_tests.rs b/crates/cli/tests/coverage/shared/doctor_tests.rs index b88c4a40c..b82aa0407 100644 --- a/crates/cli/tests/coverage/shared/doctor_tests.rs +++ b/crates/cli/tests/coverage/shared/doctor_tests.rs @@ -1468,6 +1468,57 @@ async fn collect_observability_skips_redis_response_cache_probe_offline() { ); } +#[tokio::test] +async fn collect_observability_fails_invalid_redis_response_cache_target_offline() { + let gateway = GatewayConfig { + plugin_config: Some(serde_json::json!({ + "version": 1, + "components": [ + { + "kind": "adaptive", + "enabled": true, + "config": { + "response_cache": { + "ttl_seconds": 3600, + "namespace": "doctor-test", + "backend": { + "kind": "redis", + "config": { + "url": "not-a-redis-url", + "key_prefix": "doctor-test:" + } + } + } + } + } + ] + })), + ..GatewayConfig::default() + }; + + let checks = collect_offline_observability(&gateway).await; + + let cache = checks + .iter() + .find(|check| check.name == "Response cache") + .expect("a Response cache check should be present"); + assert_eq!(cache.status, Status::Fail, "checks: {checks:?}"); + assert!( + cache.details.contains("invalid backend target"), + "details: {}", + cache.details + ); + assert!( + cache.details.contains("redis client"), + "details: {}", + cache.details + ); + assert!( + !cache.details.contains("probe skipped"), + "must not report skipped for an invalid backend target" + ); +} + #[tokio::test(start_paused = true)] async fn response_cache_backend_check_reports_timeout() { let check = response_cache_backend_check(std::future::pending()).await; @@ -1790,8 +1841,23 @@ async fn atof_endpoint_validation_rejects_missing_url_headers_timeout_and_transp .contains("headers.x-test must be a string") ); - let mixed_case_duplicate = probe_atof_endpoint( + let blank_static_header = probe_atof_endpoint( 4, + &serde_json::json!({ + "url": "http://127.0.0.1:1/events", + "headers": {"x-test": " "} + }), + ) + .await; + assert_eq!(blank_static_header.status, Status::Fail); + assert!( + blank_static_header + .details + .contains("headers.x-test must not be blank") + ); + + let mixed_case_duplicate = probe_atof_endpoint( + 5, &serde_json::json!({ "url": "http://127.0.0.1:1/events", "headers": {"Authorization": "Bearer literal"}, @@ -1806,8 +1872,26 @@ async fn atof_endpoint_validation_rejects_missing_url_headers_timeout_and_transp .contains("cannot appear in both headers and header_env") ); + let duplicate_static_header = probe_atof_endpoint( + 6, + &serde_json::json!({ + "url": "http://127.0.0.1:1/events", + "headers": { + "Authorization": "Bearer a", + "authorization": "Bearer b" + } + }), + ) + .await; + assert_eq!(duplicate_static_header.status, Status::Fail); + assert!( + duplicate_static_header + .details + .contains("appears more than once") + ); + let unsupported = probe_atof_endpoint( - 5, + 7, &serde_json::json!({ "url": "http://127.0.0.1:1/events", "transport": "grpc" @@ -1867,6 +1951,39 @@ async fn atof_endpoint_offline_still_rejects_invalid_transport_and_scheme() { .details .contains("unsupported transport") ); + + let blank_static_header = offline_atof_endpoint( + 2, + &serde_json::json!({ + "url": "http://127.0.0.1:1/events", + "headers": {"x-test": " "} + }), + ) + .await; + assert_eq!(blank_static_header.status, Status::Fail); + assert!( + blank_static_header + .details + .contains("headers.x-test must not be blank") + ); + + let duplicate_static_header = offline_atof_endpoint( + 3, + &serde_json::json!({ + "url": "http://127.0.0.1:1/events", + "headers": { + "Authorization": "Bearer a", + "authorization": "Bearer b" + } + }), + ) + .await; + assert_eq!(duplicate_static_header.status, Status::Fail); + assert!( + duplicate_static_header + .details + .contains("appears more than once") + ); } #[tokio::test] From ca7adf8edf0fde476f58ecfb152d42a11923bfdc Mon Sep 17 00:00:00 2001 From: mnajafian-nv Date: Mon, 3 Aug 2026 18:08:26 -0700 Subject: [PATCH 4/4] test: cover invalid ATOF header_env values Signed-off-by: mnajafian-nv --- .../cli/tests/coverage/shared/doctor_tests.rs | 25 ++++++++++++++++--- 1 file changed, 22 insertions(+), 3 deletions(-) diff --git a/crates/cli/tests/coverage/shared/doctor_tests.rs b/crates/cli/tests/coverage/shared/doctor_tests.rs index b82aa0407..3c3da1391 100644 --- a/crates/cli/tests/coverage/shared/doctor_tests.rs +++ b/crates/cli/tests/coverage/shared/doctor_tests.rs @@ -1856,8 +1856,27 @@ async fn atof_endpoint_validation_rejects_missing_url_headers_timeout_and_transp .contains("headers.x-test must not be blank") ); - let mixed_case_duplicate = probe_atof_endpoint( + let _env = EnvScope::set(&[( + "NEMO_RELAY_INVALID_ATOF_HEADER", + Some(std::ffi::OsStr::new("bad\nvalue")), + )]); + let invalid_header_env = probe_atof_endpoint( 5, + &serde_json::json!({ + "url": "http://127.0.0.1:1/events", + "header_env": {"x-test": "NEMO_RELAY_INVALID_ATOF_HEADER"} + }), + ) + .await; + assert_eq!(invalid_header_env.status, Status::Fail); + assert!( + invalid_header_env + .details + .contains("header_env.x-test invalid") + ); + + let mixed_case_duplicate = probe_atof_endpoint( + 6, &serde_json::json!({ "url": "http://127.0.0.1:1/events", "headers": {"Authorization": "Bearer literal"}, @@ -1873,7 +1892,7 @@ async fn atof_endpoint_validation_rejects_missing_url_headers_timeout_and_transp ); let duplicate_static_header = probe_atof_endpoint( - 6, + 7, &serde_json::json!({ "url": "http://127.0.0.1:1/events", "headers": { @@ -1891,7 +1910,7 @@ async fn atof_endpoint_validation_rejects_missing_url_headers_timeout_and_transp ); let unsupported = probe_atof_endpoint( - 7, + 8, &serde_json::json!({ "url": "http://127.0.0.1:1/events", "transport": "grpc"