diff --git a/Cargo.lock b/Cargo.lock index 31e2104987..8df977097c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4203,6 +4203,10 @@ dependencies = [ "openshell-server-macros", "openshell-supervisor-middleware", "openshell-supervisor-middleware-builtins", + "opentelemetry", + "opentelemetry-otlp", + "opentelemetry-proto", + "opentelemetry_sdk", "petname", "pin-project-lite", "prost", @@ -4229,6 +4233,7 @@ dependencies = [ "tower 0.5.3", "tower-http 0.6.8", "tracing", + "tracing-opentelemetry", "tracing-subscriber", "url", "uuid", @@ -4405,6 +4410,68 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" +[[package]] +name = "opentelemetry" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0142c63252a9e054e68a4c61a5778f7b14f576274d593f8ce883d191a099682" +dependencies = [ + "futures-core", + "futures-sink", + "js-sys", + "pin-project-lite", + "thiserror 2.0.18", + "tracing", +] + +[[package]] +name = "opentelemetry-otlp" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9966929966d17620d7c316c643ba62631826e10021409357772d5eea84f62c35" +dependencies = [ + "http 1.4.0", + "opentelemetry", + "opentelemetry-proto", + "opentelemetry_sdk", + "prost", + "thiserror 2.0.18", + "tokio", + "tonic", + "tonic-types", +] + +[[package]] +name = "opentelemetry-proto" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "56d658ba1faf63f7b9c492cfbe6e0ec365440a16132d3270c1065f7b33f1b638" +dependencies = [ + "opentelemetry", + "opentelemetry_sdk", + "prost", + "tonic", + "tonic-prost", +] + +[[package]] +name = "opentelemetry_sdk" +version = "0.32.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b59f80e1ac4d5ff7a2db8fb6c80badb7f0f3f858211fba08dd9aaec750894f9" +dependencies = [ + "futures-channel", + "futures-executor", + "futures-util", + "opentelemetry", + "percent-encoding", + "portable-atomic", + "rand 0.9.4", + "thiserror 2.0.18", + "tokio", + "tokio-stream", +] + [[package]] name = "ordered-float" version = "2.10.1" @@ -6994,6 +7061,17 @@ dependencies = [ "tonic-build", ] +[[package]] +name = "tonic-types" +version = "0.14.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "73ab1b02061f83d519bba3caa167f88f261ef05720ab8ebc954ade70de3348e8" +dependencies = [ + "prost", + "prost-types", + "tonic", +] + [[package]] name = "tower" version = "0.4.13" @@ -7150,6 +7228,21 @@ dependencies = [ "tracing-core", ] +[[package]] +name = "tracing-opentelemetry" +version = "0.33.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "adbc64cba7137545b8044cb1fe9814f7aacf3c6b5f9b45be8bb5db538befdb26" +dependencies = [ + "js-sys", + "opentelemetry", + "tracing", + "tracing-core", + "tracing-log", + "tracing-subscriber", + "web-time", +] + [[package]] name = "tracing-serde" version = "0.2.0" diff --git a/Cargo.toml b/Cargo.toml index 4ec6a0d44f..4078c0feb5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -60,6 +60,13 @@ tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] } tracing-appender = "0.2" +# OpenTelemetry — OTLP/gRPC export. Kept in lockstep with the workspace's +# tonic 0.14 / prost 0.14 via opentelemetry-proto's `grpc-tonic` feature. +opentelemetry = "0.32" +opentelemetry_sdk = { version = "0.32", features = ["rt-tokio"] } +opentelemetry-otlp = { version = "0.32", default-features = false, features = ["grpc-tonic", "trace"] } +tracing-opentelemetry = { version = "0.33", default-features = false, features = ["tracing-log"] } + # Metrics metrics = "0.24" metrics-exporter-prometheus = { version = "0.18", default-features = false, features = ["http-listener"] } diff --git a/architecture/gateway.md b/architecture/gateway.md index c8b323ea13..cac4326ea0 100644 --- a/architecture/gateway.md +++ b/architecture/gateway.md @@ -596,6 +596,36 @@ Driver-specific values that are not part of the inheritance allowlist (e.g. Podman `socket_path`, VM `vcpus`) only come from the driver's own table. +### OTLP export + +The gateway already uses Rust's `tracing` framework for structured log events +and request-span context consumed by stdout and the sandbox log bus. OTLP export +adds an OpenTelemetry layer to the same subscriber. That layer turns selected +`tracing` spans into distributed traces; it does not export log events or +replace the existing logging paths. + +`[openshell.gateway.otlp]` is the only enablement path for OpenTelemetry +export: the table's presence is the on-switch, and `OTEL_EXPORTER_OTLP_ENDPOINT` +is ignored so enablement has a single source. TOML decides whether and where +to export; the SDK's `OTEL_*` variables tune how. Transport is OTLP over gRPC +only. + +Span emission requires no per-handler instrumentation. The `tower_http` +`TraceLayer` in `multiplex.rs` opens a span per inbound request, and that span +continues incoming W3C trace context when present or starts a new trace +otherwise. It is named for the RPC and carries the request ID that also appears +in the gateway's logs — the identifier that lets an operator pivot between a +trace and its log lines. Store and compute-driver spans become children of the +request span. Reconciliation, provider refresh, and driver-watch loops create +their own operation spans because they have no inbound request to provide a +parent. gRPC status is recorded when response trailers arrive. + +Two invariants shape the failure behavior. Telemetry is diagnostic, so no OTLP +failure stops the gateway from serving: a malformed endpoint is logged at +startup and disables export. Export is best-effort — the SDK logs runtime +failures, and a failed batch is dropped rather than retried. Buffered spans +flush after the server loop exits so `SIGTERM` does not drop in-flight traces. + ### Package-managed gateway registry The CLI reads its active-gateway and per-gateway metadata from diff --git a/crates/openshell-server/Cargo.toml b/crates/openshell-server/Cargo.toml index 4c4f289ed6..43f1097383 100644 --- a/crates/openshell-server/Cargo.toml +++ b/crates/openshell-server/Cargo.toml @@ -69,6 +69,12 @@ anyhow = { workspace = true } tracing = { workspace = true } tracing-subscriber = { workspace = true } +# OpenTelemetry (OTLP trace export, opt-in via [openshell.gateway.otlp]) +opentelemetry = { workspace = true } +opentelemetry_sdk = { workspace = true } +opentelemetry-otlp = { workspace = true } +tracing-opentelemetry = { workspace = true } + # Metrics metrics = { workspace = true } metrics-exporter-prometheus = { workspace = true } @@ -118,6 +124,11 @@ rcgen = { version = "0.13", features = ["crypto", "pem"] } tokio-tungstenite = { workspace = true } futures-util = "0.3" wiremock = "0.6" +# `testing` provides InMemorySpanExporter, so span assertions do not need a +# collector, a network hop, or a flush barrier. +opentelemetry_sdk = { workspace = true, features = ["testing"] } +# Supports asserting that the gateway sends spans over OTLP. +opentelemetry-proto = { version = "0.32", default-features = false, features = ["gen-tonic", "trace"] } [lints] workspace = true diff --git a/crates/openshell-server/src/cli.rs b/crates/openshell-server/src/cli.rs index 269b9f40e9..45fc3fc2db 100644 --- a/crates/openshell-server/src/cli.rs +++ b/crates/openshell-server/src/cli.rs @@ -10,7 +10,7 @@ use openshell_core::ComputeDriverKind; use openshell_core::config::DEFAULT_SERVER_PORT; use std::net::{IpAddr, SocketAddr}; use std::path::PathBuf; -use tracing::{info, warn}; +use tracing::{error, info, warn}; use tracing_subscriber::EnvFilter; use crate::certgen; @@ -440,9 +440,15 @@ async fn run_from_args(mut args: RunArgs, matches: ArgMatches) -> Result<()> { let prepared = prepare_server_config(&mut args, &matches)?; let tracing_log_bus = TracingLogBus::new(); - tracing_log_bus.install_subscriber( + let otlp_config = prepared + .config_file + .as_ref() + .and_then(|f| f.openshell.gateway.otlp.as_ref()); + let (tracing_handle, setup_error) = crate::tracing_setup::install( EnvFilter::try_from_default_env() .unwrap_or_else(|_| EnvFilter::new(&prepared.config.log_level)), + &tracing_log_bus, + otlp_config, ); let has_client_ca = prepared @@ -468,6 +474,18 @@ async fn run_from_args(mut args: RunArgs, matches: ArgMatches) -> Result<()> { if has_oidc { info!("OIDC authentication enabled"); } + if let Some(err) = &setup_error { + error!( + error = %err, + "OTLP exporting is configured but could not be started; continuing without it" + ); + } else if let Some(otlp) = prepared + .config_file + .as_ref() + .and_then(|f| f.openshell.gateway.otlp.as_ref()) + { + info!(endpoint = %otlp.endpoint, "OTLP exporting enabled"); + } if prepared.config.auth.allow_unauthenticated_users { warn!( "Unauthenticated user access enabled — only use this for trusted local development or a fully trusted fronting proxy" @@ -487,9 +505,11 @@ async fn run_from_args(mut args: RunArgs, matches: ArgMatches) -> Result<()> { info!(bind = %prepared.config.bind_address, "Starting OpenShell server"); - Box::pin(run_server(prepared, tracing_log_bus)) - .await - .into_diagnostic() + let result = Box::pin(run_server(prepared, tracing_log_bus)).await; + + tracing_handle.shutdown(); + + result.into_diagnostic() } fn parse_compute_driver(value: &str) -> std::result::Result { diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index b87f1b5fa6..91bd8bc272 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -57,12 +57,76 @@ use tokio::sync::{Mutex, watch}; use tonic::transport::{Channel, Endpoint}; use tonic::{Code, Request, Status}; use tower::service_fn; -use tracing::{debug, info, warn}; +use tracing::{Instrument as _, debug, info, warn}; type DriverWatchStream = Pin> + Send>>; type SharedComputeDriver = Arc + Send + Sync>; +use traced_driver::TracedDriver; + +/// Instrumenting wrapper around the compute driver. +mod traced_driver { + use std::future::Future; + + use tonic::Status; + use tracing::Instrument as _; + + use super::SharedComputeDriver; + + #[derive(Clone)] + pub(super) struct TracedDriver { + inner: SharedComputeDriver, + name: String, + } + + impl TracedDriver { + pub(super) fn new(inner: SharedComputeDriver, name: String) -> Self { + Self { inner, name } + } + + /// Run one call across the driver boundary inside its span. + /// + /// Takes a closure rather than a future so the call cannot be built + /// without going through here. + pub(super) async fn call( + &self, + operation: &'static str, + sandbox_id: Option<&str>, + call: impl FnOnce(SharedComputeDriver) -> Fut, + ) -> Result + where + Fut: Future>, + { + let span = tracing::info_span!( + "driver", + otel.name = operation, + otel.kind = "client", + otel.status_code = tracing::field::Empty, + driver.name = %self.name, + sandbox.id = tracing::field::Empty, + grpc.code = tracing::field::Empty, + ); + if let Some(sandbox_id) = sandbox_id { + span.record("sandbox.id", sandbox_id); + } + + let future = call(self.inner.clone()); + async { + let result = future.await; + if let Err(status) = &result { + let current = tracing::Span::current(); + crate::otel_tracing::mark_error(¤t); + current.record("grpc.code", status.code() as i32); + } + result + } + .instrument(span) + .await + } + } +} + const DELETE_PHASE_CAS_RETRY_LIMIT: usize = 3; /// Serializes request-side deletes for the same stable sandbox ID. @@ -357,7 +421,7 @@ impl ComputeDriver for RemoteComputeDriver { #[derive(Clone)] pub struct ComputeRuntime { - driver: SharedComputeDriver, + driver: TracedDriver, driver_info: ComputeDriverInfoSnapshot, shutdown_cleanup: Option>, startup_resume: Option>, @@ -408,13 +472,13 @@ impl ComputeRuntime { "Compute driver connected" ); let driver_info = ComputeDriverInfoSnapshot { - name: driver_name, + name: driver_name.clone(), driver_name: capabilities.driver_name, driver_version: capabilities.driver_version, }; let default_image = capabilities.default_image; Ok(Self { - driver, + driver: TracedDriver::new(driver, driver_name), driver_info, shutdown_cleanup, startup_resume, @@ -596,9 +660,17 @@ impl ComputeRuntime { let driver_sandbox = driver_sandbox_from_public(sandbox, &self.driver_info.name) .map_err(|status| *status)?; self.driver - .validate_sandbox_create(Request::new(ValidateSandboxCreateRequest { - sandbox: Some(driver_sandbox), - })) + .call( + "driver.validate_sandbox_create", + Some(sandbox.object_id()), + |driver| async move { + driver + .validate_sandbox_create(Request::new(ValidateSandboxCreateRequest { + sandbox: Some(driver_sandbox), + })) + .await + }, + ) .await .map(|_| ()) } @@ -657,9 +729,17 @@ impl ComputeRuntime { } match self .driver - .create_sandbox(Request::new(CreateSandboxRequest { - sandbox: Some(driver_sandbox), - })) + .call( + "driver.create_sandbox", + Some(sandbox.object_id()), + |driver| async move { + driver + .create_sandbox(Request::new(CreateSandboxRequest { + sandbox: Some(driver_sandbox), + })) + .await + }, + ) .await { Ok(_) => { @@ -724,11 +804,17 @@ impl ComputeRuntime { // the worker. From this commitment point onward, request cancellation // cannot stop the delete after it starts mutating durable state. let runtime = self.clone(); - tokio::spawn(async move { - runtime - .delete_sandbox_inner(target, delete_guard, global_guard) - .await - }) + // `tokio::spawn` detaches from the current span, which would orphan + // the driver span from the request trace. Carry the span across. + let request_span = tracing::Span::current(); + tokio::spawn( + async move { + runtime + .delete_sandbox_inner(target, delete_guard, global_guard) + .await + } + .instrument(request_span), + ) .await .map_err(|err| { Status::internal(format!( @@ -786,10 +872,22 @@ impl ComputeRuntime { let result = self .driver - .delete_sandbox(Request::new(DeleteSandboxRequest { - sandbox_id: transition.deleting.object_id().to_string(), - sandbox_name: transition.deleting.object_name().to_string(), - })) + .call( + "driver.delete_sandbox", + Some(transition.deleting.object_id()), + |driver| { + let sandbox_id = transition.deleting.object_id().to_string(); + let sandbox_name = transition.deleting.object_name().to_string(); + async move { + driver + .delete_sandbox(Request::new(DeleteSandboxRequest { + sandbox_id, + sandbox_name, + })) + .await + } + }, + ) .await; match result { @@ -1514,9 +1612,15 @@ impl ComputeRuntime { async fn watch_loop(self: Arc, mut cancel: watch::Receiver) { loop { + // Spans the stream open, not its lifetime: the future resolves + // once the driver accepts the watch. let mut stream = match self .driver - .watch_sandboxes(Request::new(WatchSandboxesRequest {})) + .call("driver.watch_sandboxes", None, |driver| async move { + driver + .watch_sandboxes(Request::new(WatchSandboxesRequest {})) + .await + }) .await { Ok(response) => response.into_inner(), @@ -1574,30 +1678,49 @@ impl ComputeRuntime { } } + #[tracing::instrument( + name = "reconcile", + skip_all, + fields( + otel.name = "reconcile.sandboxes", + driver.name = %self.driver_info.name, + backend_count = tracing::field::Empty, + store_count = tracing::field::Empty, + ) + )] async fn reconcile_store_with_backend(&self, grace_period: Duration) -> Result<(), String> { let sweep_started_at_ms = openshell_core::time::now_ms(); let backend_sandboxes = self .driver - .list_sandboxes(Request::new(ListSandboxesRequest {})) + .call("driver.list_sandboxes", None, |driver| async move { + driver + .list_sandboxes(Request::new(ListSandboxesRequest {})) + .await + }) .await - .map_err(|e| e.to_string())? + .map_err(|e| e.to_string()) + .inspect_err(|_| crate::otel_tracing::mark_error(&tracing::Span::current()))? .into_inner() .sandboxes; let backend_ids = backend_sandboxes .iter() .map(|sandbox| sandbox.id.clone()) .collect::>(); + tracing::Span::current().record("backend_count", backend_sandboxes.len()); for sandbox in backend_sandboxes { self.reconcile_snapshot_sandbox(sandbox, sweep_started_at_ms) - .await?; + .await + .inspect_err(|_| crate::otel_tracing::mark_error(&tracing::Span::current()))?; } let records = self .store .list_by_type(Sandbox::object_type(), 500, 0) .await - .map_err(|e| e.to_string())?; + .map_err(|e| e.to_string()) + .inspect_err(|_| crate::otel_tracing::mark_error(&tracing::Span::current()))?; + tracing::Span::current().record("store_count", records.len()); let grace_ms = grace_period.as_millis().try_into().unwrap_or(i64::MAX); @@ -1615,13 +1738,50 @@ impl ComputeRuntime { } self.prune_missing_sandbox(record, sweep_started_at_ms, grace_ms) - .await?; + .await + .inspect_err(|_| crate::otel_tracing::mark_error(&tracing::Span::current()))?; } Ok(()) } async fn apply_watch_event(&self, event: WatchSandboxesEvent) -> Result<(), String> { + let (operation, sandbox_id) = match &event.payload { + Some(watch_sandboxes_event::Payload::Sandbox(update)) => ( + "driver_watch.sandbox_updated", + update + .sandbox + .as_ref() + .map(|sandbox| sandbox.id.as_str()) + .unwrap_or_default(), + ), + Some(watch_sandboxes_event::Payload::Deleted(deleted)) => { + ("driver_watch.sandbox_deleted", deleted.sandbox_id.as_str()) + } + Some(watch_sandboxes_event::Payload::PlatformEvent(platform_event)) => ( + "driver_watch.platform_event", + platform_event.sandbox_id.as_str(), + ), + None => return Ok(()), + }; + let span = tracing::info_span!( + "driver_watch", + otel.name = operation, + otel.status_code = tracing::field::Empty, + sandbox.id = %sandbox_id, + ); + async { + let result = self.apply_watch_event_inner(event).await; + if result.is_err() { + crate::otel_tracing::mark_error(&tracing::Span::current()); + } + result + } + .instrument(span) + .await + } + + async fn apply_watch_event_inner(&self, event: WatchSandboxesEvent) -> Result<(), String> { match event.payload { Some(watch_sandboxes_event::Payload::Sandbox(sandbox)) => { if let Some(sandbox) = sandbox.sandbox { @@ -2068,10 +2228,18 @@ impl ComputeRuntime { ) -> Result, String> { match self .driver - .get_sandbox(Request::new(GetSandboxRequest { - sandbox_id: sandbox_id.to_string(), - sandbox_name: sandbox_name.to_string(), - })) + .call("driver.get_sandbox", Some(sandbox_id), |driver| { + let sandbox_id = sandbox_id.to_string(); + let sandbox_name = sandbox_name.to_string(); + async move { + driver + .get_sandbox(Request::new(GetSandboxRequest { + sandbox_id, + sandbox_name, + })) + .await + } + }) .await { Ok(response) => { @@ -2830,7 +2998,7 @@ pub async fn new_test_runtime(store: Arc) -> ComputeRuntime { #[cfg(test)] pub async fn new_test_runtime_for_driver(store: Arc, driver_name: &str) -> ComputeRuntime { ComputeRuntime { - driver: Arc::new(NoopTestDriver), + driver: TracedDriver::new(Arc::new(NoopTestDriver), "test".to_string()), driver_info: ComputeDriverInfoSnapshot { name: driver_name.to_string(), driver_name: driver_name.to_string(), @@ -3299,7 +3467,7 @@ mod tests { ) -> ComputeRuntime { let store = Arc::new(Store::connect("sqlite::memory:").await.unwrap()); ComputeRuntime { - driver, + driver: TracedDriver::new(driver, "test-driver".to_string()), driver_info: ComputeDriverInfoSnapshot { name: "test-driver".to_string(), driver_name: "test-driver".to_string(), @@ -3916,6 +4084,174 @@ mod tests { )); } + /// Driver calls are a remote boundary even in-process: they reach the + /// Docker daemon, the Kubernetes API, or a Podman socket. + #[tokio::test] + async fn driver_calls_export_child_spans_under_the_request_span() { + use tracing::Instrument as _; + + use crate::otel_tracing::test_collector; + + let runtime = test_runtime(Arc::new(TestDriver::default())).await; + let sandbox = sandbox_record("sb-trace", "sandbox-trace", SandboxPhase::Provisioning); + + let traced = test_collector::install_traced(); + async { + runtime + .create_sandbox(sandbox, None) + .await + .expect("create succeeds"); + } + .instrument(tracing::info_span!("request")) + .await; + + let driver_span = traced.span_with("driver.create_sandbox", "sandbox.id", "sb-trace"); + let spans = traced.finished_spans(); + let root = spans + .iter() + .find(|s| s.span_context.span_id() == driver_span.parent_span_id) + .expect("the driver span has a recorded parent"); + + assert_eq!( + root.name, "request", + "the driver span is a child of the request span" + ); + assert_eq!( + driver_span.span_context.trace_id(), + root.span_context.trace_id(), + "both share one trace" + ); + assert_eq!( + test_collector::attribute(&driver_span, "driver.name").as_deref(), + Some("test-driver"), + "the span names which driver was called" + ); + assert_eq!( + test_collector::attribute(&driver_span, "sandbox.id").as_deref(), + Some("sb-trace"), + ); + assert_eq!( + driver_span.span_kind, + opentelemetry::trace::SpanKind::Client, + "the gateway is the caller at this boundary" + ); + assert!( + !matches!( + driver_span.status, + opentelemetry::trace::Status::Error { .. } + ), + "a successful driver call is not marked an error, got {:?}", + driver_span.status + ); + } + + /// A failing driver call must be visible as a failure in the trace, not + /// just as a span that happens to be followed by nothing. + #[tokio::test] + async fn failed_driver_calls_are_marked_on_the_span() { + use tracing::Instrument as _; + + use crate::otel_tracing::test_collector; + + /// A driver that behaves normally except that creates fail, so the + /// test exercises only the failure attribute. + #[derive(Debug, Default)] + struct FailingDriver(TestDriver); + + #[tonic::async_trait] + impl ComputeDriver for FailingDriver { + type WatchSandboxesStream = DriverWatchStream; + + async fn create_sandbox( + &self, + _request: Request, + ) -> Result, Status> { + Err(Status::unavailable("driver is down")) + } + + async fn get_capabilities( + &self, + request: Request, + ) -> Result, Status> { + self.0.get_capabilities(request).await + } + + async fn validate_sandbox_create( + &self, + request: Request, + ) -> Result, Status> { + self.0.validate_sandbox_create(request).await + } + + async fn get_sandbox( + &self, + request: Request, + ) -> Result, Status> { + self.0.get_sandbox(request).await + } + + async fn list_sandboxes( + &self, + request: Request, + ) -> Result< + tonic::Response, + Status, + > { + self.0.list_sandboxes(request).await + } + + async fn stop_sandbox( + &self, + request: Request, + ) -> Result, Status> { + self.0.stop_sandbox(request).await + } + + async fn delete_sandbox( + &self, + request: Request, + ) -> Result, Status> { + self.0.delete_sandbox(request).await + } + + async fn watch_sandboxes( + &self, + request: Request, + ) -> Result, Status> { + self.0.watch_sandboxes(request).await + } + } + + let runtime = test_runtime(Arc::new(FailingDriver::default())).await; + let sandbox = sandbox_record("sb-fail", "sandbox-fail", SandboxPhase::Provisioning); + + let traced = test_collector::install_traced(); + async { + runtime + .create_sandbox(sandbox, None) + .await + .expect_err("driver refuses the create"); + } + .instrument(tracing::info_span!("request")) + .await; + + let driver_span = traced.span_with("driver.create_sandbox", "sandbox.id", "sb-fail"); + + assert!( + matches!( + driver_span.status, + opentelemetry::trace::Status::Error { .. } + ), + "the span carries error status so trace UIs flag it, got {:?}", + driver_span.status + ); + assert_eq!( + test_collector::attribute(&driver_span, "grpc.code").as_deref(), + Some("14"), + "the gRPC code names the cause without reading the message" + ); + } + #[tokio::test] async fn begin_sandbox_delete_retries_after_stale_snapshot_conflict() { let runtime = test_runtime(Arc::new(TestDriver::default())).await; @@ -5669,6 +6005,105 @@ mod tests { })); } + /// Driver watch events arrive on a background stream, so the store writes + /// they trigger land outside the request that caused them. + #[tokio::test] + async fn driver_watch_events_export_one_trace_rooted_at_the_event() { + use crate::otel_tracing::test_collector; + + let runtime = test_runtime(Arc::new(TestDriver::default())).await; + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Ready); + runtime.store.put_message(&sandbox).await.unwrap(); + runtime.sandbox_index.update_from_sandbox(&sandbox); + + let traced = test_collector::install_traced(); + runtime + .apply_watch_event(deleted_watch_event("sb-1")) + .await + .unwrap(); + + let spans = traced.finished_spans(); + let root = spans + .iter() + .find(|s| s.name == "driver_watch.sandbox_deleted") + .unwrap_or_else(|| { + panic!( + "the event records a span of its own, got {:?}", + spans.iter().map(|s| &s.name).collect::>() + ) + }); + + assert_eq!( + root.parent_span_id, + opentelemetry::trace::SpanId::INVALID, + "a driver-pushed event has no caller, so it roots its own trace" + ); + assert_eq!( + test_collector::attribute(root, "sandbox.id").as_deref(), + Some("sb-1"), + "the span names which sandbox the driver reported on" + ); + + assert!( + spans.iter().any(|s| { + s.name.starts_with("store.") + && s.span_context.trace_id() == root.span_context.trace_id() + }), + "the event's store writes join its trace rather than orphaning, got {:?}", + spans.iter().map(|s| &s.name).collect::>() + ); + } + + /// The reconciler runs on a timer with no inbound request, so without a + /// span of its own each store call becomes its own anonymous trace. + #[tokio::test] + async fn reconcile_sweeps_export_one_trace_rooted_at_the_sweep() { + use crate::otel_tracing::test_collector; + + let runtime = test_runtime(Arc::new(TestDriver::default())).await; + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Provisioning); + runtime.store.put_message(&sandbox).await.unwrap(); + runtime.sandbox_index.update_from_sandbox(&sandbox); + + let traced = test_collector::install_traced(); + runtime + .reconcile_store_with_backend(Duration::ZERO) + .await + .unwrap(); + + // Other tests drive their own reconcile loops into the shared + // exporter, so match on the shape of a sweep rather than assuming + // there is exactly one. + let spans = traced.finished_spans(); + let roots = traced.spans_named("reconcile.sandboxes"); + assert!( + !roots.is_empty(), + "the sweep records a span of its own, got {:?}", + spans.iter().map(|s| &s.name).collect::>() + ); + for root in &roots { + assert_eq!( + root.parent_span_id, + opentelemetry::trace::SpanId::INVALID, + "a timer-driven sweep has no caller, so it roots its own trace" + ); + } + + let complete = roots.iter().find(|root| { + let children: Vec<_> = spans + .iter() + .filter(|s| s.parent_span_id == root.span_context.span_id()) + .collect(); + children.iter().any(|s| s.name == "driver.list_sandboxes") + && children.iter().any(|s| s.name.starts_with("store.")) + }); + assert!( + complete.is_some(), + "the sweep's driver call and store reads are children of it, got {:?}", + spans.iter().map(|s| &s.name).collect::>() + ); + } + #[tokio::test] async fn reconcile_store_with_backend_does_not_recreate_missing_record_from_snapshot() { let runtime = test_runtime(Arc::new(TestDriver { diff --git a/crates/openshell-server/src/config_file.rs b/crates/openshell-server/src/config_file.rs index c4e0cbc959..c29fd21657 100644 --- a/crates/openshell-server/src/config_file.rs +++ b/crates/openshell-server/src/config_file.rs @@ -161,6 +161,8 @@ pub struct GatewayFileSection { pub mtls_auth: Option, #[serde(default)] pub gateway_jwt: Option, + #[serde(default)] + pub otlp: Option, // ── Disallowed-in-file fields ──────────────────────────────────────── // @@ -171,6 +173,23 @@ pub struct GatewayFileSection { pub database_url: Option, } +/// `[openshell.gateway.otlp]` section. +/// +/// Presence of this table enables OTLP export; there is no `enabled` flag. +/// SDK tuning knobs are deliberately absent — see [`crate::otel_tracing`] for what +/// this table owns and what the `OTEL_*` environment variables own. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct OtlpConfig { + /// OTLP/gRPC collector endpoint, e.g. + /// `http://otel-collector.observability.svc:4317`. + pub endpoint: String, + + /// `service.name` resource attribute. Defaults to `openshell-gateway`. + #[serde(default)] + pub service_name: Option, +} + /// `[openshell.supervisor]` section. #[derive(Debug, Clone, Default, Serialize, Deserialize)] #[serde(deny_unknown_fields)] @@ -436,6 +455,64 @@ grpc_endpoint = "https://openshell-gateway.agents.svc:8080" assert!(file.openshell.drivers.contains_key("kubernetes")); } + #[test] + fn parses_gateway_otlp_config() { + let toml = r#" +[openshell.gateway.otlp] +endpoint = "http://otel-collector.observability.svc:4317" +service_name = "openshell-gateway-dev" +"#; + let tmp = write_tmp(toml); + let file = load(tmp.path()).expect("valid otlp config parses"); + let otlp = file.openshell.gateway.otlp.expect("otlp config"); + assert_eq!( + otlp.endpoint, + "http://otel-collector.observability.svc:4317" + ); + assert_eq!(otlp.service_name.as_deref(), Some("openshell-gateway-dev")); + } + + #[test] + fn otlp_config_requires_only_endpoint() { + let toml = r#" +[openshell.gateway.otlp] +endpoint = "http://127.0.0.1:4317" +"#; + let tmp = write_tmp(toml); + let file = load(tmp.path()).expect("minimal otlp config parses"); + let otlp = file.openshell.gateway.otlp.expect("otlp config"); + assert_eq!(otlp.endpoint, "http://127.0.0.1:4317"); + assert!(otlp.service_name.is_none()); + } + + #[test] + fn otlp_config_rejects_unknown_fields() { + let toml = r#" +[openshell.gateway.otlp] +endpoint = "http://127.0.0.1:4317" +protocol = "http" +"#; + let tmp = write_tmp(toml); + assert!(load(tmp.path()).is_err(), "unknown otlp field is rejected"); + } + + #[test] + fn otlp_config_rejects_sdk_tuning_keys() { + // Sampling, batching, and limits are the SDK's env-var surface. A + // `deny_unknown_fields` rejection is the signal that they do not + // belong in the config file. + let toml = r#" +[openshell.gateway.otlp] +endpoint = "http://127.0.0.1:4317" +sampler = "traceidratio" +"#; + let tmp = write_tmp(toml); + assert!( + load(tmp.path()).is_err(), + "sampler is configured via OTEL_TRACES_SAMPLER, not TOML" + ); + } + #[test] fn parses_gateway_auth_config() { let toml = r" diff --git a/crates/openshell-server/src/grpc/sandbox.rs b/crates/openshell-server/src/grpc/sandbox.rs index 0d6a4e277f..58554a3c60 100644 --- a/crates/openshell-server/src/grpc/sandbox.rs +++ b/crates/openshell-server/src/grpc/sandbox.rs @@ -672,199 +672,205 @@ pub(super) async fn handle_watch_sandbox( let (tx, rx) = mpsc::channel::>(256); let state = state.clone(); - // Spawn producer task. - tokio::spawn(async move { - // Validate that the sandbox exists BEFORE subscribing to any buses. - match state.store.get_message::(&sandbox_id).await { - Ok(Some(_)) => {} - Ok(None) => { - let _ = tx.send(Err(Status::not_found("sandbox not found"))).await; - return; - } - Err(e) => { - let _ = tx - .send(Err(Status::internal(format!("fetch sandbox failed: {e}")))) - .await; - return; + // Spawn producer task. `tokio::spawn` detaches from the current span, so + // carry it across to keep the producer's store reads in the request trace. + let request_span = tracing::Span::current(); + tokio::spawn(tracing::Instrument::instrument( + async move { + // Validate that the sandbox exists BEFORE subscribing to any buses. + match state.store.get_message::(&sandbox_id).await { + Ok(Some(_)) => {} + Ok(None) => { + let _ = tx.send(Err(Status::not_found("sandbox not found"))).await; + return; + } + Err(e) => { + let _ = tx + .send(Err(Status::internal(format!("fetch sandbox failed: {e}")))) + .await; + return; + } } - } - // Subscribe to all buses BEFORE reading the snapshot. - let mut status_rx = if follow_status { - Some(state.sandbox_watch_bus.subscribe(&sandbox_id)) - } else { - None - }; - let mut log_rx = if follow_logs { - Some(state.tracing_log_bus.subscribe(&sandbox_id)) - } else { - None - }; - let mut platform_rx = if follow_events { - Some( - state - .tracing_log_bus - .platform_event_bus - .subscribe(&sandbox_id), - ) - } else { - None - }; + // Subscribe to all buses BEFORE reading the snapshot. + let mut status_rx = if follow_status { + Some(state.sandbox_watch_bus.subscribe(&sandbox_id)) + } else { + None + }; + let mut log_rx = if follow_logs { + Some(state.tracing_log_bus.subscribe(&sandbox_id)) + } else { + None + }; + let mut platform_rx = if follow_events { + Some( + state + .tracing_log_bus + .platform_event_bus + .subscribe(&sandbox_id), + ) + } else { + None + }; - // Re-read the snapshot now that we have subscriptions active. - match state.store.get_message::(&sandbox_id).await { - Ok(Some(sandbox)) => { - state.sandbox_index.update_from_sandbox(&sandbox); - let _ = tx - .send(Ok(SandboxStreamEvent { - payload: Some( - openshell_core::proto::sandbox_stream_event::Payload::Sandbox( - sandbox.clone(), + // Re-read the snapshot now that we have subscriptions active. + match state.store.get_message::(&sandbox_id).await { + Ok(Some(sandbox)) => { + state.sandbox_index.update_from_sandbox(&sandbox); + let _ = tx + .send(Ok(SandboxStreamEvent { + payload: Some( + openshell_core::proto::sandbox_stream_event::Payload::Sandbox( + sandbox.clone(), + ), ), - ), - })) - .await; + })) + .await; - if stop_on_terminal { - let phase = - SandboxPhase::try_from(sandbox.phase()).unwrap_or(SandboxPhase::Unknown); - if phase == SandboxPhase::Ready { - return; + if stop_on_terminal { + let phase = SandboxPhase::try_from(sandbox.phase()) + .unwrap_or(SandboxPhase::Unknown); + if phase == SandboxPhase::Ready { + return; + } } } + Ok(None) => { + let _ = tx.send(Err(Status::not_found("sandbox not found"))).await; + return; + } + Err(e) => { + let _ = tx + .send(Err(Status::internal(format!("fetch sandbox failed: {e}")))) + .await; + return; + } } - Ok(None) => { - let _ = tx.send(Err(Status::not_found("sandbox not found"))).await; - return; - } - Err(e) => { - let _ = tx - .send(Err(Status::internal(format!("fetch sandbox failed: {e}")))) - .await; - return; - } - } - // Replay tail logs (best-effort), filtered by log_since_ms and log_sources. - if follow_logs { - for evt in state.tracing_log_bus.tail(&sandbox_id, log_tail as usize) { - if let Some(openshell_core::proto::sandbox_stream_event::Payload::Log(ref log)) = - evt.payload - { - if log_since_ms > 0 && log.timestamp_ms < log_since_ms { - continue; - } - if !log_sources.is_empty() && !source_matches(&log.source, &log_sources) { - continue; + // Replay tail logs (best-effort), filtered by log_since_ms and log_sources. + if follow_logs { + for evt in state.tracing_log_bus.tail(&sandbox_id, log_tail as usize) { + if let Some(openshell_core::proto::sandbox_stream_event::Payload::Log( + ref log, + )) = evt.payload + { + if log_since_ms > 0 && log.timestamp_ms < log_since_ms { + continue; + } + if !log_sources.is_empty() && !source_matches(&log.source, &log_sources) { + continue; + } + if !level_matches(&log.level, &log_min_level) { + continue; + } } - if !level_matches(&log.level, &log_min_level) { - continue; + if tx.send(Ok(evt)).await.is_err() { + return; } } - if tx.send(Ok(evt)).await.is_err() { - return; - } } - } - // Replay buffered platform events. - if follow_events { - for evt in state - .tracing_log_bus - .platform_event_bus - .tail(&sandbox_id, event_tail as usize) - { - if tx.send(Ok(evt)).await.is_err() { - return; + // Replay buffered platform events. + if follow_events { + for evt in state + .tracing_log_bus + .platform_event_bus + .tail(&sandbox_id, event_tail as usize) + { + if tx.send(Ok(evt)).await.is_err() { + return; + } } } - } - loop { - tokio::select! { - res = async { - match status_rx.as_mut() { - Some(rx) => rx.recv().await, - None => future::pending().await, - } - } => { - match res { - Ok(()) => { - match state.store.get_message::(&sandbox_id).await { - Ok(Some(sandbox)) => { - state.sandbox_index.update_from_sandbox(&sandbox); - if tx.send(Ok(SandboxStreamEvent { payload: Some(openshell_core::proto::sandbox_stream_event::Payload::Sandbox(sandbox.clone()))})).await.is_err() { - return; - } - if stop_on_terminal { - let phase = SandboxPhase::try_from(sandbox.phase()).unwrap_or(SandboxPhase::Unknown); - if phase == SandboxPhase::Ready { + loop { + tokio::select! { + res = async { + match status_rx.as_mut() { + Some(rx) => rx.recv().await, + None => future::pending().await, + } + } => { + match res { + Ok(()) => { + match state.store.get_message::(&sandbox_id).await { + Ok(Some(sandbox)) => { + state.sandbox_index.update_from_sandbox(&sandbox); + if tx.send(Ok(SandboxStreamEvent { payload: Some(openshell_core::proto::sandbox_stream_event::Payload::Sandbox(sandbox.clone()))})).await.is_err() { return; } + if stop_on_terminal { + let phase = SandboxPhase::try_from(sandbox.phase()).unwrap_or(SandboxPhase::Unknown); + if phase == SandboxPhase::Ready { + return; + } + } + } + Ok(None) => { + return; + } + Err(e) => { + let _ = tx.send(Err(Status::internal(format!("fetch sandbox failed: {e}")))).await; + return; } - } - Ok(None) => { - return; - } - Err(e) => { - let _ = tx.send(Err(Status::internal(format!("fetch sandbox failed: {e}")))).await; - return; } } - } - Err(err) => { - let _ = tx.send(Err(crate::sandbox_watch::broadcast_to_status(err))).await; - return; + Err(err) => { + let _ = tx.send(Err(crate::sandbox_watch::broadcast_to_status(err))).await; + return; + } } } - } - res = async { - match log_rx.as_mut() { - Some(rx) => rx.recv().await, - None => future::pending().await, - } - } => { - match res { - Ok(evt) => { - if let Some(openshell_core::proto::sandbox_stream_event::Payload::Log(ref log)) = evt.payload { - if !log_sources.is_empty() && !source_matches(&log.source, &log_sources) { - continue; + res = async { + match log_rx.as_mut() { + Some(rx) => rx.recv().await, + None => future::pending().await, + } + } => { + match res { + Ok(evt) => { + if let Some(openshell_core::proto::sandbox_stream_event::Payload::Log(ref log)) = evt.payload { + if !log_sources.is_empty() && !source_matches(&log.source, &log_sources) { + continue; + } + if !level_matches(&log.level, &log_min_level) { + continue; + } } - if !level_matches(&log.level, &log_min_level) { - continue; + if tx.send(Ok(evt)).await.is_err() { + return; } } - if tx.send(Ok(evt)).await.is_err() { + Err(err) => { + let _ = tx.send(Err(crate::sandbox_watch::broadcast_to_status(err))).await; return; } } - Err(err) => { - let _ = tx.send(Err(crate::sandbox_watch::broadcast_to_status(err))).await; - return; - } - } - } - res = async { - match platform_rx.as_mut() { - Some(rx) => rx.recv().await, - None => future::pending().await, } - } => { - match res { - Ok(evt) => { - if tx.send(Ok(evt)).await.is_err() { + res = async { + match platform_rx.as_mut() { + Some(rx) => rx.recv().await, + None => future::pending().await, + } + } => { + match res { + Ok(evt) => { + if tx.send(Ok(evt)).await.is_err() { + return; + } + } + Err(err) => { + let _ = tx.send(Err(crate::sandbox_watch::broadcast_to_status(err))).await; return; } } - Err(err) => { - let _ = tx.send(Err(crate::sandbox_watch::broadcast_to_status(err))).await; - return; - } } } } - } - }); + }, + request_span, + )); Ok(Response::new(ReceiverStream::new(rx))) } diff --git a/crates/openshell-server/src/lib.rs b/crates/openshell-server/src/lib.rs index f2967a833d..d052e8d0d4 100644 --- a/crates/openshell-server/src/lib.rs +++ b/crates/openshell-server/src/lib.rs @@ -34,6 +34,7 @@ mod http; mod inference; mod middleware; mod multiplex; +mod otel_tracing; mod persistence; pub(crate) mod policy_store; mod provider_profile_sources; @@ -51,6 +52,7 @@ mod tls; #[cfg(test)] pub(crate) mod tls_test_utils; pub mod tracing_bus; +mod tracing_setup; mod ws_tunnel; use metrics_exporter_prometheus::PrometheusBuilder; @@ -70,6 +72,10 @@ use tracing::{debug, error, info, warn}; #[cfg(test)] pub(crate) static TEST_ENV_LOCK: LazyLock> = LazyLock::new(|| Mutex::new(())); +/// Serializes tests that assert on captured spans, which share one exporter. +#[cfg(test)] +pub(crate) static TEST_TRACING_LOCK: LazyLock> = LazyLock::new(|| Mutex::new(())); + use compute::ComputeRuntime; pub use grpc::OpenShellService; pub use http::{health_router, http_router, metrics_router, service_http_router}; diff --git a/crates/openshell-server/src/multiplex.rs b/crates/openshell-server/src/multiplex.rs index a58f66c916..d49379128e 100644 --- a/crates/openshell-server/src/multiplex.rs +++ b/crates/openshell-server/src/multiplex.rs @@ -22,6 +22,9 @@ use openshell_core::proto::{ inference_server::InferenceServer, open_shell_server::OpenShellServer, }; use openshell_gateway_interceptors::{EvaluationContext, GatewayInterceptorRuntime}; +use opentelemetry::propagation::{Extractor, TextMapPropagator}; +use opentelemetry::trace::TraceContextExt as _; +use opentelemetry_sdk::propagation::TraceContextPropagator; use std::collections::BTreeMap; use std::convert::Infallible; use std::future::Future; @@ -33,6 +36,7 @@ use tokio::io::{AsyncRead, AsyncWrite}; use tower::ServiceExt; use tower_http::request_id::{MakeRequestId, RequestId}; use tracing::{Span, warn}; +use tracing_opentelemetry::OpenTelemetrySpanExt as _; use crate::{ OpenShellService, ServerState, @@ -67,32 +71,112 @@ fn make_request_span(req: &Request) -> Span { .and_then(|v| v.to_str().ok()) .unwrap_or("-"); - if matches!(path, "/health" | "/healthz" | "/readyz") { + // `otel.name` and `otel.kind` are consumed by `tracing-opentelemetry` to + // set the exported span's name and kind; they are not emitted as + // attributes. See [`otel_span_name`] for why the name cannot simply be + // the callsite name. + let otel_name = otel_span_name(req.method(), path); + + let span = if matches!(path, "/health" | "/healthz" | "/readyz") { tracing::debug_span!( "request", method = %req.method(), path, request_id, + otel.name = %otel_name, + otel.kind = "server", + otel.status_code = tracing::field::Empty, + http.response.status_code = tracing::field::Empty, ) } else { - tracing::info_span!( + let span = tracing::info_span!( "request", method = %req.method(), path, request_id, - ) + otel.name = %otel_name, + otel.kind = "server", + otel.status_code = tracing::field::Empty, + http.response.status_code = tracing::field::Empty, + rpc.system = tracing::field::Empty, + rpc.service = tracing::field::Empty, + rpc.method = tracing::field::Empty, + rpc.grpc.status_code = tracing::field::Empty, + ); + // RPC-aware backends build service maps from these; without them a + // gRPC call is just an HTTP span. + if let Some((service, method)) = grpc_service_method(path) { + span.record("rpc.system", "grpc"); + span.record("rpc.service", service); + span.record("rpc.method", method); + } + span + }; + + let propagator = TraceContextPropagator::new(); + let parent = propagator.extract_with_context( + &opentelemetry::Context::new(), + &HeaderExtractor(req.headers()), + ); + if parent.span().span_context().is_valid() { + let _ = span.set_parent(parent); } + + span } -/// Log response status and latency within the request span. -fn log_response(res: &Response, latency: Duration, _span: &Span) { +/// Log response status and latency, record protocol status, and mark failures. +fn log_response(res: &Response, latency: Duration, span: &Span) { + let status = res.status(); + span.record("http.response.status_code", status.as_u16()); + record_grpc_status(res.headers(), span); + if status.is_server_error() { + crate::otel_tracing::mark_error(span); + } tracing::info!( - status = res.status().as_u16(), + status = status.as_u16(), latency_ms = latency.as_millis(), "response" ); } +fn record_response_trailers( + trailers: Option<&http::HeaderMap>, + _stream_duration: Duration, + span: &Span, +) { + if let Some(trailers) = trailers { + record_grpc_status(trailers, span); + } +} + +fn record_grpc_status(headers: &http::HeaderMap, span: &Span) { + let Some(code) = headers + .get("grpc-status") + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.parse::().ok()) + else { + return; + }; + + span.record("rpc.grpc.status_code", code); + if code != 0 { + crate::otel_tracing::mark_error(span); + } +} + +struct HeaderExtractor<'a>(&'a http::HeaderMap); + +impl Extractor for HeaderExtractor<'_> { + fn get(&self, key: &str) -> Option<&str> { + self.0.get(key).and_then(|value| value.to_str().ok()) + } + + fn keys(&self) -> Vec<&str> { + self.0.keys().map(http::HeaderName::as_str).collect() + } +} + /// Wrap a service with the standard request-ID middleware stack. /// /// Layer order: `SetRequestId` → `TraceLayer` → `PropagateRequestId`. @@ -108,7 +192,8 @@ macro_rules! request_id_middleware { ::tower_http::trace::TraceLayer::new_for_http() .make_span_with(make_request_span) .on_request(()) - .on_response(log_response), + .on_response(log_response) + .on_eos(record_response_trailers), ) .layer(::tower_http::request_id::PropagateRequestIdLayer::new( x_request_id, @@ -992,6 +1077,33 @@ fn grpc_method_from_path(path: &str) -> String { path.rsplit('/').next().unwrap_or(path).to_string() } +/// Name for the exported `OpenTelemetry` span, per the `OTel` semantic +/// conventions: `$service/$method` for RPCs, `{method} {path}` for plain HTTP. +/// +/// The `tracing` callsite name is the constant `"request"` because `tracing` +/// requires `'static` span names, so the per-request name is carried in the +/// `otel.name` field instead. +fn otel_span_name(method: &http::Method, path: &str) -> String { + grpc_service_method(path).map_or_else( + || format!("{method} {path}"), + |(service, rpc_method)| format!("{service}/{rpc_method}"), + ) +} + +/// Split a gRPC path into its service and method. +/// +/// A gRPC path is exactly "/package.Service/Method". Anything else — a health +/// check, /metrics, a sandbox service URL — is plain HTTP. +fn grpc_service_method(path: &str) -> Option<(&str, &str)> { + let mut segments = path.strip_prefix('/')?.split('/'); + let service = segments.next()?; + let method = segments.next()?; + if segments.next().is_some() || !service.contains('.') || method.is_empty() { + return None; + } + Some((service, method)) +} + fn grpc_status_from_response(res: &Response) -> String { res.headers() .get("grpc-status") @@ -1713,7 +1825,6 @@ mod tests { #[test] fn request_id_appears_in_trace_span() { use tracing_subscriber::fmt::format::FmtSpan; - use tracing_subscriber::layer::SubscriberExt; let log_buf: Arc>> = Arc::new(Mutex::new(Vec::new())); let writer = TraceBuf(log_buf.clone()); @@ -1723,12 +1834,12 @@ mod tests { .with_ansi(false) .with_span_events(FmtSpan::CLOSE); - let subscriber = tracing_subscriber::registry().with(fmt_layer); - tracing::subscriber::with_default(subscriber, || { - // Other parallel tests may register this callsite while no subscriber - // is active. Refresh the process-wide cache after installing this - // thread-local subscriber so the span cannot remain disabled. - tracing::callsite::rebuild_interest_cache(); + let subscriber = { + use tracing_subscriber::layer::SubscriberExt as _; + tracing_subscriber::registry().with(fmt_layer) + }; + { + let _traced = crate::otel_tracing::test_collector::install_scoped(subscriber); let req = Request::builder() .uri("/test-path") @@ -1738,7 +1849,7 @@ mod tests { let span = make_request_span(&req); drop(span.enter()); drop(span); - }); + } let output = String::from_utf8(log_buf.lock().unwrap().clone()).unwrap(); assert!( @@ -1747,6 +1858,271 @@ mod tests { ); } + /// The `TraceLayer` creates the server span, so no gRPC handler needs + /// `#[instrument]`. The request ID carries into it so a trace can be + /// correlated with the gateway's logs. + #[tokio::test] + async fn request_span_exports_over_otlp_with_request_id() { + use crate::otel_tracing::test_collector; + + let traced = test_collector::install_traced(); + let req = Request::builder() + .uri("/openshell.v1.OpenShell/CreateSandbox") + .header("x-request-id", "otlp-req-id-9876") + .body(Empty::::new()) + .unwrap(); + let span = make_request_span(&req); + drop(span.enter()); + drop(span); + + let spans = traced.finished_spans(); + let span = spans + .iter() + .find(|s| s.name == "openshell.v1.OpenShell/CreateSandbox") + .unwrap_or_else(|| { + panic!( + "the per-request span is recorded under its RPC name, got {:?}", + spans.iter().map(|s| &s.name).collect::>() + ) + }); + assert_eq!( + test_collector::attribute(span, "request_id").as_deref(), + Some("otlp-req-id-9876"), + ); + assert_eq!( + test_collector::attribute(span, "path").as_deref(), + Some("/openshell.v1.OpenShell/CreateSandbox"), + ); + assert_eq!( + span.span_kind, + opentelemetry::trace::SpanKind::Server, + "trace UIs lay this out as a served call, not an internal operation" + ); + assert_eq!( + span.parent_span_id, + opentelemetry::trace::SpanId::INVALID, + "the request span is a trace root" + ); + assert_eq!( + test_collector::attribute(span, "rpc.system").as_deref(), + Some("grpc"), + ); + assert_eq!( + test_collector::attribute(span, "rpc.service").as_deref(), + Some("openshell.v1.OpenShell"), + ); + assert_eq!( + test_collector::attribute(span, "rpc.method").as_deref(), + Some("CreateSandbox"), + ); + } + + #[tokio::test] + async fn request_span_continues_the_incoming_trace() { + use crate::otel_tracing::test_collector; + + let traced = test_collector::install_traced(); + let req = Request::builder() + .uri("/openshell.v1.OpenShell/CreateSandbox") + .header( + "traceparent", + "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01", + ) + .body(Empty::::new()) + .unwrap(); + let span = make_request_span(&req); + drop(span.enter()); + drop(span); + + let span = traced.span_with( + "openshell.v1.OpenShell/CreateSandbox", + "rpc.method", + "CreateSandbox", + ); + assert_eq!( + span.span_context.trace_id().to_string(), + "4bf92f3577b34da6a3ce929d0e0e4736" + ); + assert_eq!( + span.parent_span_id.to_string(), + "00f067aa0ba902b7", + "the server span is a child of the caller's span" + ); + } + + /// A failed request must be distinguishable from a successful one in a + /// trace UI, which keys off span status rather than a logged field. + #[tokio::test] + async fn request_spans_record_the_response_outcome() { + use crate::otel_tracing::test_collector; + + let traced = test_collector::install_traced(); + for (path, status) in [ + ("/openshell.v1.OpenShell/CreateSandbox", 500), + ("/openshell.v1.OpenShell/ListSandboxes", 200), + ] { + let req = Request::builder() + .uri(path) + .body(Empty::::new()) + .unwrap(); + let span = make_request_span(&req); + let res = Response::builder() + .status(status) + .body(Empty::::new()) + .unwrap(); + let entered = span.enter(); + log_response(&res, Duration::from_millis(3), &span); + drop(entered); + drop(span); + } + + let spans = traced.finished_spans(); + let failed = spans + .iter() + .find(|s| s.name == "openshell.v1.OpenShell/CreateSandbox") + .expect("failed request span recorded"); + let succeeded = spans + .iter() + .find(|s| s.name == "openshell.v1.OpenShell/ListSandboxes") + .expect("successful request span recorded"); + + assert_eq!( + test_collector::attribute(failed, "http.response.status_code").as_deref(), + Some("500"), + "the response status is an attribute, not only a log field" + ); + assert!( + matches!(failed.status, opentelemetry::trace::Status::Error { .. }), + "the span carries error status so trace UIs flag it, got {:?}", + failed.status + ); + assert!( + !matches!(succeeded.status, opentelemetry::trace::Status::Error { .. }), + "got {:?}", + succeeded.status + ); + } + + #[tokio::test] + async fn request_span_records_grpc_status_from_trailers() { + use crate::otel_tracing::test_collector; + + let traced = test_collector::install_traced(); + let req = Request::builder() + .uri("/openshell.v1.OpenShell/CreateSandbox") + .body(Empty::::new()) + .unwrap(); + let span = make_request_span(&req); + let mut trailers = http::HeaderMap::new(); + trailers.insert("grpc-status", HeaderValue::from_static("13")); + record_response_trailers(Some(&trailers), Duration::from_millis(3), &span); + drop(span); + + let span = traced.span_with( + "openshell.v1.OpenShell/CreateSandbox", + "rpc.method", + "CreateSandbox", + ); + assert_eq!( + test_collector::attribute(&span, "rpc.grpc.status_code").as_deref(), + Some("13") + ); + assert!( + matches!(span.status, opentelemetry::trace::Status::Error { .. }), + "a non-OK gRPC trailer marks the span as failed" + ); + } + + /// Without upstream trace context, each inbound entrypoint roots a trace + /// named for itself. + #[tokio::test] + async fn each_entrypoint_gets_its_own_root_span() { + use crate::otel_tracing::test_collector; + + let paths = [ + "/openshell.v1.OpenShell/CreateSandbox", + "/openshell.v1.OpenShell/ListSandboxes", + "/openshell.v1.OpenShell/DeleteSandbox", + "/openshell.inference.v1.Inference/GetInferenceBundle", + "/metrics", + ]; + + let traced = test_collector::install_traced(); + for path in paths { + let req = Request::builder() + .uri(path) + .body(Empty::::new()) + .unwrap(); + let span = make_request_span(&req); + drop(span.enter()); + drop(span); + } + + let names: std::collections::BTreeSet = traced + .finished_spans() + .iter() + .map(|s| s.name.to_string()) + .collect(); + + let expected = [ + "GET /metrics", + "openshell.inference.v1.Inference/GetInferenceBundle", + "openshell.v1.OpenShell/CreateSandbox", + "openshell.v1.OpenShell/DeleteSandbox", + "openshell.v1.OpenShell/ListSandboxes", + ] + .into_iter() + .map(String::from) + .collect::>(); + + assert!( + expected.is_subset(&names), + "each entrypoint exports under its own name, got {names:?}" + ); + assert!( + !names.contains("request"), + "no entrypoint falls back to the generic callsite name, got {names:?}" + ); + } + + /// gRPC spans are named for the RPC, per the OpenTelemetry RPC semantic + /// conventions (`$service/$method`). + #[test] + fn grpc_request_spans_are_named_for_the_rpc() { + assert_eq!( + otel_span_name(&http::Method::POST, "/openshell.v1.OpenShell/CreateSandbox"), + "openshell.v1.OpenShell/CreateSandbox" + ); + assert_eq!( + otel_span_name( + &http::Method::POST, + "/openshell.inference.v1.Inference/GetInferenceBundle" + ), + "openshell.inference.v1.Inference/GetInferenceBundle" + ); + } + + /// Non-RPC paths follow the HTTP conventions instead: `{method} {path}`. + #[test] + fn http_request_spans_are_named_method_and_path() { + assert_eq!( + otel_span_name(&http::Method::GET, "/healthz"), + "GET /healthz" + ); + assert_eq!( + otel_span_name(&http::Method::GET, "/metrics"), + "GET /metrics" + ); + } + + /// A path with no service segment must not produce a span named after a + /// stray slash or an empty string. + #[test] + fn bare_paths_fall_back_to_the_http_shape() { + assert_eq!(otel_span_name(&http::Method::GET, "/"), "GET /"); + assert_eq!(otel_span_name(&http::Method::POST, "/Foo"), "POST /Foo"); + } + #[test] fn grpc_method_extracts_last_segment() { assert_eq!( diff --git a/crates/openshell-server/src/otel_tracing.rs b/crates/openshell-server/src/otel_tracing.rs new file mode 100644 index 0000000000..a56a956a3c --- /dev/null +++ b/crates/openshell-server/src/otel_tracing.rs @@ -0,0 +1,658 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! OpenTelemetry tracing integration for the gateway. +//! +//! Converts selected Rust `tracing` spans into OpenTelemetry traces and +//! exports them over OTLP/gRPC when configured. +//! +//! # Configuration split +//! +//! `[openshell.gateway.otlp]` decides **whether and where** to export: the +//! table's presence is the on-switch, its `endpoint` the destination. +//! `OTEL_EXPORTER_OTLP_ENDPOINT` is deliberately not read, so enablement has +//! one source. +//! +//! **How** to export — sampling, batching, span limits, transport headers — +//! is the SDK's `OTEL_*` environment surface, read as the provider is built +//! and mirrored nowhere here. `docs/reference/gateway-config.mdx` documents +//! the variables operators are likely to want. +//! +//! Only traces are exported. Logs and metrics have their own surfaces (OCSF +//! JSONL and the Prometheus `/metrics` endpoint). + +use opentelemetry::KeyValue; +use opentelemetry::trace::TracerProvider as _; +use opentelemetry_otlp::{SpanExporter, WithExportConfig}; +use opentelemetry_sdk::Resource; +use opentelemetry_sdk::trace::{SdkTracer, SdkTracerProvider}; +use tracing::Subscriber; +use tracing_opentelemetry::OpenTelemetryLayer; +use tracing_subscriber::Layer as _; +use tracing_subscriber::registry::LookupSpan; + +use crate::config_file::OtlpConfig; + +/// `service.name` reported when the config file does not override it. +const DEFAULT_SERVICE_NAME: &str = "openshell-gateway"; + +/// Instrumentation scope recorded on spans this gateway emits. +const INSTRUMENTATION_SCOPE: &str = "openshell-gateway"; + +/// Prefix of the placeholder `service.name` the SDK's own detector supplies +/// when nothing else set one. The full value is `unknown_service:`. +/// Used to tell "operator configured nothing" apart from "operator set +/// `OTEL_SERVICE_NAME`". +const SDK_UNKNOWN_SERVICE_PREFIX: &str = "unknown_service"; + +/// Failure to construct the OTLP export pipeline from gateway config. +/// +/// Every variant is a configuration error, surfaced at startup. Collector +/// *reachability* is deliberately not an error here — the batch exporter +/// connects lazily so a down collector never blocks the gateway from serving. +#[derive(Debug, thiserror::Error)] +pub enum SetupError { + #[error("otlp endpoint is empty; set [openshell.gateway.otlp].endpoint")] + EmptyEndpoint, + + #[error("invalid otlp endpoint {endpoint:?}: {source}")] + InvalidEndpoint { + endpoint: String, + source: http::uri::InvalidUri, + }, + + #[error("failed to build the otlp span exporter: {0}")] + Exporter(#[from] opentelemetry_otlp::ExporterBuildError), +} + +/// Build the `OTel` resource describing this gateway process. +/// +/// `Resource::builder` also runs the SDK's env detector, so +/// `OTEL_RESOURCE_ATTRIBUTES` and `OTEL_SERVICE_NAME` are merged in on top of +/// these defaults. +fn build_resource(cfg: &OtlpConfig) -> Resource { + let version = KeyValue::new("service.version", openshell_core::VERSION); + + if let Some(name) = cfg + .service_name + .as_deref() + .map(str::trim) + .filter(|s| !s.is_empty()) + { + return Resource::builder() + .with_service_name(name.to_string()) + .with_attribute(version) + .build(); + } + + // No name configured: prefer whatever the env detector found, and fall + // back to our own default only if it found nothing. Setting the default + // unconditionally would mask `OTEL_SERVICE_NAME`. + let detected = Resource::builder().with_attribute(version.clone()).build(); + if detected + .get(&opentelemetry::Key::from_static_str("service.name")) + .is_some_and(|v| !v.to_string().starts_with(SDK_UNKNOWN_SERVICE_PREFIX)) + { + return detected; + } + + Resource::builder() + .with_service_name(DEFAULT_SERVICE_NAME) + .with_attribute(version) + .build() +} + +/// Build a tracer provider exporting over OTLP/gRPC to the configured endpoint. +/// +/// Must be called from within a Tokio runtime — the tonic exporter binds to +/// the current reactor as it is constructed. It does not connect: an +/// unreachable collector produces export failures, never a startup failure. +/// +/// The sampler and span limits are left at the SDK's defaults, which are +/// themselves resolved from `OTEL_*` env vars (see the module docs). +fn build_provider(cfg: &OtlpConfig) -> Result { + let endpoint = cfg.endpoint.trim(); + if endpoint.is_empty() { + return Err(SetupError::EmptyEndpoint); + } + // Validate up front: the exporter defers connection, so without this a + // typo'd endpoint would surface only as export failures at runtime. + endpoint + .parse::() + .map_err(|source| SetupError::InvalidEndpoint { + endpoint: endpoint.to_string(), + source, + })?; + + let exporter = SpanExporter::builder() + .with_tonic() + .with_endpoint(endpoint) + .build()?; + + Ok(SdkTracerProvider::builder() + .with_batch_exporter(exporter) + .with_resource(build_resource(cfg)) + .build()) +} + +/// Resolve the tracer provider for a gateway config file's optional +/// `[openshell.gateway.otlp]` table. +/// +/// `None` means export is off — not configured, or configured and unusable. +/// Telemetry is diagnostic, so a broken exporter never stops the gateway. +/// +/// The error is returned rather than logged because the provider is built +/// before the subscriber it attaches to, so logging here would go nowhere. +pub fn provider_for(cfg: Option<&OtlpConfig>) -> (Option, Option) { + match cfg.map(build_provider) { + None => (None, None), + Some(Ok(provider)) => (Some(provider), None), + Some(Err(err)) => (None, Some(err)), + } +} + +/// Build the `tracing` layer that forwards spans to `provider`. +/// +/// Events stay on the gateway's logging layers. Spans emitted by the +/// OpenTelemetry crates are excluded to prevent recursive export traffic. +pub fn layer( + provider: &SdkTracerProvider, +) -> tracing_subscriber::filter::Filtered< + OpenTelemetryLayer, + tracing_subscriber::filter::FilterFn, + S, +> +where + S: Subscriber + for<'span> LookupSpan<'span>, +{ + tracing_opentelemetry::layer() + .with_tracer(provider.tracer(INSTRUMENTATION_SCOPE)) + .with_filter(tracing_subscriber::filter::filter_fn(|meta| { + meta.is_span() && !is_opentelemetry_target(meta.target()) + })) +} + +/// Mark `span` as failed. +/// +/// The field must be declared on the span at creation — `tracing` drops +/// records for fields a span does not have. +pub fn mark_error(span: &tracing::Span) { + span.record("otel.status_code", "ERROR"); +} + +/// Whether `target` belongs to the OpenTelemetry crates themselves. +/// +/// OpenTelemetry crates use their crate name as the target (`opentelemetry`, +/// `opentelemetry_sdk`, `opentelemetry-otlp`), so a prefix match covers them. +fn is_opentelemetry_target(target: &str) -> bool { + target.starts_with("opentelemetry") +} + +/// In-process OTLP/gRPC trace collector, for tests that need to assert what +/// the gateway actually put on the wire. +#[cfg(test)] +pub mod test_collector { + use std::net::SocketAddr; + use std::sync::{Arc, Mutex}; + + use opentelemetry_proto::tonic::collector::trace::v1::{ + ExportTraceServiceRequest, ExportTraceServiceResponse, + trace_service_server::{TraceService, TraceServiceServer}, + }; + use opentelemetry_proto::tonic::trace::v1::Span; + + /// Spans received by a running [`start`] collector. + pub type Received = Arc>>; + + /// The one exporter every traced test reads, installed for the whole test + /// binary and never swapped. + /// + /// A per-test subscriber cannot work here: `tracing` caches callsite + /// interest process-wide, so a callsite first hit by an untraced test + /// stays dark for every later one. + static EXPORTER: std::sync::LazyLock = + std::sync::LazyLock::new(|| { + use tracing_subscriber::layer::SubscriberExt as _; + + let exporter = opentelemetry_sdk::trace::InMemorySpanExporterBuilder::new().build(); + let provider = opentelemetry_sdk::trace::SdkTracerProvider::builder() + .with_simple_exporter(exporter.clone()) + .build(); + let subscriber = tracing_subscriber::registry().with(super::layer(&provider)); + tracing::subscriber::set_global_default(subscriber) + .expect("test subscriber installs once"); + // Leaked so the provider outlives every span the binary records; + // dropping it would shut the exporter down mid-suite. + std::mem::forget(provider); + exporter + }); + + /// Captures spans until the returned guard is dropped. + /// + /// Serialized on [`crate::TEST_TRACING_LOCK`] because the exporter it + /// clears is shared by the whole binary. + #[must_use] + pub fn install_traced() -> TracingTestGuard { + let lock = crate::TEST_TRACING_LOCK + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + EXPORTER.reset(); + TracingTestGuard { + _lock: lock, + exporter: &EXPORTER, + } + } + + impl TracingTestGuard { + /// Every span recorded since [`install_traced`], including any written + /// concurrently by other callers. + /// + /// Prefer [`Self::spans_named`] or [`Self::span_with`] to select spans. + pub fn finished_spans(&self) -> Vec { + self.exporter.get_finished_spans().expect("in-memory spans") + } + + /// Spans named `name`. + pub fn spans_named(&self, name: &str) -> Vec { + self.finished_spans() + .into_iter() + .filter(|span| span.name == name) + .collect() + } + + /// The span named `name` carrying `key` = `value`. + pub fn span_with( + &self, + name: &str, + key: &str, + value: &str, + ) -> opentelemetry_sdk::trace::SpanData { + let spans = self.finished_spans(); + spans + .iter() + .find(|span| span.name == name && attribute(span, key).as_deref() == Some(value)) + .cloned() + .unwrap_or_else(|| { + panic!( + "no span {name:?} with {key}={value:?}, got {:?}", + spans.iter().map(|s| &s.name).collect::>() + ) + }) + } + } + + /// Installs `subscriber` for the current thread until dropped, for tests + /// asserting on log output rather than exported spans. + /// + /// Forces the global subscriber up first so callsite interest is decided + /// by a registry that records, not by the no-op default. + #[must_use] + pub fn install_scoped(subscriber: impl Into) -> ScopedTracingTestGuard { + let lock = crate::TEST_TRACING_LOCK + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + std::sync::LazyLock::force(&EXPORTER); + ScopedTracingTestGuard { + _default: tracing::dispatcher::set_default(&subscriber.into()), + _lock: lock, + } + } + + /// Uninstalls the scoped subscriber before releasing the lock. + pub struct ScopedTracingTestGuard { + _default: tracing::dispatcher::DefaultGuard, + _lock: std::sync::MutexGuard<'static, ()>, + } + + pub struct TracingTestGuard { + _lock: std::sync::MutexGuard<'static, ()>, + exporter: &'static opentelemetry_sdk::trace::InMemorySpanExporter, + } + + #[derive(Clone)] + struct Collector { + spans: Received, + } + + #[tonic::async_trait] + impl TraceService for Collector { + async fn export( + &self, + request: tonic::Request, + ) -> Result, tonic::Status> { + let mut spans = self.spans.lock().expect("collector lock"); + for resource_span in request.into_inner().resource_spans { + for scope_span in resource_span.scope_spans { + spans.extend(scope_span.spans); + } + } + Ok(tonic::Response::new(ExportTraceServiceResponse::default())) + } + } + + /// Start a collector on an ephemeral port. Returns its address and a + /// handle to the spans it receives. + pub async fn start() -> (SocketAddr, Received) { + let spans: Received = Arc::new(Mutex::new(Vec::new())); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind collector"); + let addr = listener.local_addr().expect("collector addr"); + + let service = Collector { + spans: Arc::clone(&spans), + }; + tokio::spawn(async move { + tonic::transport::Server::builder() + .add_service(TraceServiceServer::new(service)) + .serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(listener)) + .await + .ok(); + }); + + (addr, spans) + } + + /// Wait for spans named `expected` to arrive, returning everything + /// received. + /// + /// `SdkTracerProvider::force_flush` is not a delivery barrier — it reports + /// `Ok` even when the collector is unreachable — so asserting on the + /// collector's contents immediately after it is a race. + pub async fn wait_for_spans(received: &Received, expected: &[&str]) -> Vec { + for _ in 0..200 { + let spans = received.lock().expect("collector lock").clone(); + if expected + .iter() + .all(|name| spans.iter().any(|s| s.name == *name)) + { + return spans; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + received.lock().expect("collector lock").clone() + } + + /// Value of `key` on an in-memory span, if present. + pub fn attribute(span: &opentelemetry_sdk::trace::SpanData, key: &str) -> Option { + span.attributes + .iter() + .find(|kv| kv.key.as_str() == key) + .map(|kv| kv.value.to_string()) + } + + /// Value of `key` on `span`, if present as a string attribute. + pub fn string_attribute(span: &Span, key: &str) -> Option { + use opentelemetry_proto::tonic::common::v1::any_value::Value; + + span.attributes + .iter() + .find(|kv| kv.key == key) + .and_then(|kv| kv.value.as_ref()) + .and_then(|v| v.value.as_ref()) + .map(|v| match v { + Value::StringValue(s) => s.clone(), + other => format!("{other:?}"), + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn config() -> OtlpConfig { + OtlpConfig { + endpoint: "http://127.0.0.1:4317".into(), + service_name: None, + } + } + + #[test] + fn resource_defaults_the_service_name() { + let _lock = crate::TEST_ENV_LOCK + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let _env = EnvVarGuard::remove("OTEL_SERVICE_NAME"); + + assert_eq!( + service_name_of(&build_resource(&config())), + Some(DEFAULT_SERVICE_NAME.to_string()) + ); + } + + #[test] + fn resource_honors_configured_service_name_and_carries_version() { + let mut cfg = config(); + cfg.service_name = Some("gateway-staging".into()); + let resource = build_resource(&cfg); + + assert_eq!( + resource + .get(&opentelemetry::Key::from_static_str("service.name")) + .map(|v| v.to_string()), + Some("gateway-staging".to_string()) + ); + assert_eq!( + resource + .get(&opentelemetry::Key::from_static_str("service.version")) + .map(|v| v.to_string()), + Some(openshell_core::VERSION.to_string()) + ); + } + + struct EnvVarGuard { + key: &'static str, + original: Option, + } + + impl EnvVarGuard { + #[allow(unsafe_code)] + fn remove(key: &'static str) -> Self { + let original = std::env::var(key).ok(); + // SAFETY: tests serialize environment mutation with TEST_ENV_LOCK. + unsafe { std::env::remove_var(key) }; + Self { key, original } + } + + #[allow(unsafe_code)] + fn set(key: &'static str, value: &str) -> Self { + let original = std::env::var(key).ok(); + // SAFETY: tests serialize environment mutation with TEST_ENV_LOCK. + unsafe { std::env::set_var(key, value) }; + Self { key, original } + } + } + + impl Drop for EnvVarGuard { + #[allow(unsafe_code)] + fn drop(&mut self) { + // SAFETY: tests serialize environment mutation with TEST_ENV_LOCK. + match self.original.as_deref() { + Some(value) => unsafe { std::env::set_var(self.key, value) }, + None => unsafe { std::env::remove_var(self.key) }, + } + } + } + + fn service_name_of(resource: &Resource) -> Option { + resource + .get(&opentelemetry::Key::from_static_str("service.name")) + .map(|v| v.to_string()) + } + + /// Documented in `docs/reference/gateway-config.mdx`: the config file wins + /// over `OTEL_SERVICE_NAME`, because the gateway owns its own identity + /// when an operator has stated it explicitly. + #[test] + fn configured_service_name_wins_over_the_env_var() { + let _lock = crate::TEST_ENV_LOCK + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let _env = EnvVarGuard::set("OTEL_SERVICE_NAME", "from-env"); + + let mut cfg = config(); + cfg.service_name = Some("from-config".into()); + + assert_eq!( + service_name_of(&build_resource(&cfg)), + Some("from-config".to_string()) + ); + } + + /// With no `service_name` in the config file, the SDK's env detector is + /// the fallback rather than the built-in default. + #[test] + fn env_service_name_applies_when_config_omits_it() { + let _lock = crate::TEST_ENV_LOCK + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let _env = EnvVarGuard::set("OTEL_SERVICE_NAME", "from-env"); + + assert_eq!( + service_name_of(&build_resource(&config())), + Some("from-env".to_string()) + ); + } + + #[test] + fn blank_service_name_falls_back_to_the_default() { + let _lock = crate::TEST_ENV_LOCK + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let _env = EnvVarGuard::remove("OTEL_SERVICE_NAME"); + + let mut cfg = config(); + cfg.service_name = Some(" ".into()); + assert_eq!( + service_name_of(&build_resource(&cfg)), + Some(DEFAULT_SERVICE_NAME.to_string()) + ); + } + + #[test] + fn provider_rejects_a_malformed_endpoint() { + let mut cfg = config(); + cfg.endpoint = "definitely not a url".into(); + let err = build_provider(&cfg).expect_err("malformed endpoint"); + assert!( + err.to_string().contains("definitely not a url"), + "error names the offending endpoint: {err}" + ); + } + + #[test] + fn provider_rejects_an_empty_endpoint() { + let mut cfg = config(); + cfg.endpoint = " ".into(); + assert!(build_provider(&cfg).is_err(), "empty endpoint is rejected"); + } + + #[tokio::test] + async fn provider_builds_without_a_reachable_collector() { + // The OTLP batch exporter connects lazily, so a valid endpoint must + // build even when nothing is listening — the gateway must not fail to + // start because its collector is down. + let provider = build_provider(&config()).expect("provider builds"); + provider.shutdown().ok(); + } + + /// Not configuring export is not a failure, so it produces nothing to + /// report. This is distinct from a *broken* configuration, which yields an + /// error for the caller to log — see the misconfigured-endpoint test. + #[tokio::test] + async fn absent_otlp_table_disables_export() { + let (provider, err) = provider_for(None); + assert!(provider.is_none(), "export is off"); + assert!( + err.is_none(), + "an absent table is a choice, not an error to report" + ); + } + + #[tokio::test] + async fn present_otlp_table_enables_export() { + let (provider, err) = provider_for(Some(&config())); + assert!(err.is_none()); + provider.expect("provider is present").shutdown().ok(); + } + + /// Telemetry must never be able to take the gateway down. A bad endpoint + /// disables export and surfaces an error to report; it does not stop the + /// gateway from starting. + #[tokio::test] + async fn misconfigured_endpoint_disables_export_without_failing_startup() { + let mut cfg = config(); + cfg.endpoint = "definitely not a url".into(); + + let (provider, err) = provider_for(Some(&cfg)); + assert!( + provider.is_none(), + "a bad endpoint degrades to no export rather than failing startup" + ); + assert!(err.is_some(), "the failure is reportable, not swallowed"); + } + + #[tokio::test] + async fn tracing_events_are_not_exported() { + let traced = test_collector::install_traced(); + let span = tracing::info_span!("outer"); + let entered = span.enter(); + tracing::warn!(target: "opentelemetry-otlp", "export failed"); + tracing::warn!(target: "openshell_server", "gateway warning"); + drop(entered); + drop(span); + + let spans = traced.finished_spans(); + let outer = spans + .iter() + .find(|s| s.name == "outer") + .expect("outer span recorded"); + + assert!( + outer.events.is_empty(), + "structured log events stay on the logging paths" + ); + } + + #[tokio::test(flavor = "multi_thread")] + async fn tracing_spans_reach_the_collector_over_otlp() { + let (addr, received) = test_collector::start().await; + + let cfg = OtlpConfig { + endpoint: format!("http://{addr}"), + service_name: None, + }; + let provider = build_provider(&cfg).expect("provider builds"); + + { + // Its own subscriber, not the shared in-memory one: this test + // asserts on what reaches a real collector over the wire. + use tracing_subscriber::layer::SubscriberExt as _; + let subscriber = tracing_subscriber::registry().with(layer(&provider)); + let _traced = test_collector::install_scoped(subscriber); + let span = tracing::info_span!("create_sandbox", sandbox_id = "sb-otlp"); + let _entered = span.enter(); + } + + provider.force_flush().expect("flush spans"); + + let spans = test_collector::wait_for_spans(&received, &["create_sandbox"]).await; + let span = spans + .iter() + .find(|s| s.name == "create_sandbox") + .unwrap_or_else(|| { + panic!( + "collector received the exported span, got {:?}", + spans.iter().map(|s| &s.name).collect::>() + ) + }); + assert_eq!( + test_collector::string_attribute(span, "sandbox_id").as_deref(), + Some("sb-otlp"), + "tracing fields are exported as span attributes" + ); + + provider.shutdown().ok(); + } +} diff --git a/crates/openshell-server/src/persistence/mod.rs b/crates/openshell-server/src/persistence/mod.rs index 291e2eafd7..ad71658086 100644 --- a/crates/openshell-server/src/persistence/mod.rs +++ b/crates/openshell-server/src/persistence/mod.rs @@ -52,6 +52,15 @@ pub enum PersistenceError { } impl PersistenceError { + /// Whether this error is a signal the caller acts on rather than a failure. + /// + /// Both variants are how the store reports contention: `MustCreate` losing + /// a race is how [`crate::compute::lease`] learns the lease is held, and a + /// version conflict is what drives an optimistic-concurrency retry. + pub fn is_expected(&self) -> bool { + matches!(self, Self::UniqueViolation { .. } | Self::Conflict { .. }) + } + pub fn unique_violation(constraint: Option, detail: Option) -> Self { let constraint_msg = constraint .as_ref() @@ -171,6 +180,20 @@ macro_rules! store_dispatch { }; } +/// [`store_dispatch`] for methods carrying a span, marking that span failed +/// unless the error is one the caller is expected to act on. +macro_rules! store_dispatch_traced { + ($self:ident . $method:ident ( $($arg:expr),* )) => {{ + let result = store_dispatch!($self.$method($($arg),*)); + if let Err(err) = &result + && !err.is_expected() + { + crate::otel_tracing::mark_error(&tracing::Span::current()); + } + result + }}; +} + impl Store { /// Returns `true` for single-replica backends (`SQLite`) where no lease /// coordination is needed, `false` for multi-replica backends (`Postgres`). @@ -237,6 +260,11 @@ impl Store { /// * `Err(Conflict)` - Resource version mismatch (for `MatchResourceVersion`) /// * `Err(UniqueViolation)` - Object already exists (for `MustCreate`) or name conflict #[allow(clippy::too_many_arguments)] + #[tracing::instrument( + name = "store", + skip_all, + fields(otel.name = "store.put_if", otel.status_code = tracing::field::Empty, object_type = %object_type, object.id = %id, object.name = %name, workspace = %workspace) + )] pub async fn put_if( &self, object_type: &str, @@ -247,7 +275,15 @@ impl Store { labels: Option<&str>, condition: WriteCondition, ) -> PersistenceResult { - store_dispatch!(self.put_if(object_type, id, name, workspace, payload, labels, condition)) + store_dispatch_traced!(self.put_if( + object_type, + id, + name, + workspace, + payload, + labels, + condition + )) } /// Delete an object by id with compare-and-swap support. @@ -261,17 +297,27 @@ impl Store { /// * `Ok(true)` - Object was deleted /// * `Ok(false)` - Object not found /// * `Err(Conflict)` - Resource version mismatch + #[tracing::instrument( + name = "store", + skip_all, + fields(otel.name = "store.delete_if", otel.status_code = tracing::field::Empty, object_type = %object_type, object.id = %id) + )] pub async fn delete_if( &self, object_type: &str, id: &str, expected_resource_version: u64, ) -> PersistenceResult { - store_dispatch!(self.delete_if(object_type, id, expected_resource_version)) + store_dispatch_traced!(self.delete_if(object_type, id, expected_resource_version)) } /// Insert or update a generic named object with an application-owned scope. #[allow(clippy::too_many_arguments)] + #[tracing::instrument( + name = "store", + skip_all, + fields(otel.name = "store.put_scoped", otel.status_code = tracing::field::Empty, object_type = %object_type, object.id = %id, object.name = %name, workspace = %workspace, scope = %scope) + )] pub async fn put_scoped( &self, object_type: &str, @@ -282,67 +328,120 @@ impl Store { payload: &[u8], labels: Option<&str>, ) -> PersistenceResult<()> { - store_dispatch!(self.put_scoped(object_type, id, name, workspace, scope, payload, labels)) + store_dispatch_traced!(self.put_scoped( + object_type, + id, + name, + workspace, + scope, + payload, + labels + )) } /// Fetch an object by id. + #[tracing::instrument( + name = "store", + skip_all, + fields(otel.name = "store.get", otel.status_code = tracing::field::Empty, object_type = %object_type, object.id = %id) + )] pub async fn get( &self, object_type: &str, id: &str, ) -> PersistenceResult> { - store_dispatch!(self.get(object_type, id)) + store_dispatch_traced!(self.get(object_type, id)) } /// Fetch an object by name within an object type and workspace. + #[tracing::instrument( + name = "store", + skip_all, + fields( + otel.name = "store.get_by_name", otel.status_code = tracing::field::Empty, + object_type = %object_type, + workspace = %workspace, + object.name = %name + ) + )] pub async fn get_by_name( &self, object_type: &str, workspace: &str, name: &str, ) -> PersistenceResult> { - store_dispatch!(self.get_by_name(object_type, workspace, name)) + store_dispatch_traced!(self.get_by_name(object_type, workspace, name)) } /// Delete an object by id. + #[tracing::instrument( + name = "store", + skip_all, + fields(otel.name = "store.delete", otel.status_code = tracing::field::Empty, object_type = %object_type, object.id = %id) + )] pub async fn delete(&self, object_type: &str, id: &str) -> PersistenceResult { - store_dispatch!(self.delete(object_type, id)) + store_dispatch_traced!(self.delete(object_type, id)) } /// Count objects of a given type within a workspace. + #[tracing::instrument( + name = "store", + skip_all, + fields(otel.name = "store.count_in_workspace", otel.status_code = tracing::field::Empty, object_type = %object_type, workspace = %workspace) + )] pub async fn count_in_workspace( &self, object_type: &str, workspace: &str, ) -> PersistenceResult { - store_dispatch!(self.count_in_workspace(object_type, workspace)) + store_dispatch_traced!(self.count_in_workspace(object_type, workspace)) } /// Delete all objects of a given type within a workspace. + #[tracing::instrument( + name = "store", + skip_all, + fields(otel.name = "store.delete_all_in_workspace", otel.status_code = tracing::field::Empty, object_type = %object_type, workspace = %workspace) + )] pub async fn delete_all_in_workspace( &self, object_type: &str, workspace: &str, ) -> PersistenceResult { - store_dispatch!(self.delete_all_in_workspace(object_type, workspace)) + store_dispatch_traced!(self.delete_all_in_workspace(object_type, workspace)) } /// Delete all objects of a given type with a matching scope. + #[tracing::instrument( + name = "store", + skip_all, + fields(otel.name = "store.delete_by_scope", otel.status_code = tracing::field::Empty, object_type = %object_type, scope = %scope) + )] pub async fn delete_by_scope(&self, object_type: &str, scope: &str) -> PersistenceResult { - store_dispatch!(self.delete_by_scope(object_type, scope)) + store_dispatch_traced!(self.delete_by_scope(object_type, scope)) } /// Delete an object by name within an object type and workspace. + #[tracing::instrument( + name = "store", + skip_all, + fields(otel.name = "store.delete_by_name", otel.status_code = tracing::field::Empty, object_type = %object_type, workspace = %workspace, object.name = %name) + )] pub async fn delete_by_name( &self, object_type: &str, workspace: &str, name: &str, ) -> PersistenceResult { - store_dispatch!(self.delete_by_name(object_type, workspace, name)) + store_dispatch_traced!(self.delete_by_name(object_type, workspace, name)) } /// List objects by type and workspace. + #[tracing::instrument( + name = "store", + skip_all, + fields(otel.name = "store.list", otel.status_code = tracing::field::Empty, object_type = %object_type, workspace = %workspace) + )] pub async fn list( &self, object_type: &str, @@ -350,23 +449,33 @@ impl Store { limit: u32, offset: u32, ) -> PersistenceResult> { - store_dispatch!(self.list(object_type, workspace, limit, offset)) + store_dispatch_traced!(self.list(object_type, workspace, limit, offset)) } /// List objects by type across all workspaces. + #[tracing::instrument( + name = "store", + skip_all, + fields(otel.name = "store.list_by_type", otel.status_code = tracing::field::Empty, object_type = %object_type) + )] pub async fn list_by_type( &self, object_type: &str, limit: u32, offset: u32, ) -> PersistenceResult> { - store_dispatch!(self.list_by_type(object_type, limit, offset)) + store_dispatch_traced!(self.list_by_type(object_type, limit, offset)) } /// List objects by type and application-owned scope. /// /// Workspace filtering is intentionally omitted: scope values are sandbox /// UUIDs which are globally unique. Revisit if non-UUID scopes are introduced. + #[tracing::instrument( + name = "store", + skip_all, + fields(otel.name = "store.list_by_scope", otel.status_code = tracing::field::Empty, object_type = %object_type, scope = %scope) + )] pub async fn list_by_scope( &self, object_type: &str, @@ -374,11 +483,21 @@ impl Store { limit: u32, offset: u32, ) -> PersistenceResult> { - store_dispatch!(self.list_by_scope(object_type, scope, limit, offset)) + store_dispatch_traced!(self.list_by_scope(object_type, scope, limit, offset)) } /// List objects by type and workspace with label selector filtering. /// Label selector format: "key1=value1,key2=value2" (comma-separated equality matches). + #[tracing::instrument( + name = "store", + skip_all, + fields( + otel.name = "store.list_with_selector", otel.status_code = tracing::field::Empty, + object_type = %object_type, + workspace = %workspace, + label_selector = %label_selector + ) + )] pub async fn list_with_selector( &self, object_type: &str, @@ -387,7 +506,7 @@ impl Store { limit: u32, offset: u32, ) -> PersistenceResult> { - store_dispatch!(self.list_with_selector( + store_dispatch_traced!(self.list_with_selector( object_type, workspace, label_selector, @@ -397,6 +516,11 @@ impl Store { } /// List objects by type across all workspaces with label selector filtering. + #[tracing::instrument( + name = "store", + skip_all, + fields(otel.name = "store.list_all_with_selector", otel.status_code = tracing::field::Empty, object_type = %object_type, label_selector = %label_selector) + )] pub async fn list_all_with_selector( &self, object_type: &str, @@ -404,7 +528,12 @@ impl Store { limit: u32, offset: u32, ) -> PersistenceResult> { - store_dispatch!(self.list_all_with_selector(object_type, label_selector, limit, offset)) + store_dispatch_traced!(self.list_all_with_selector( + object_type, + label_selector, + limit, + offset + )) } // ----------------------------------------------------------------------- @@ -728,6 +857,11 @@ pub fn parse_label_selector(selector: &str) -> PersistenceResult, ) -> PersistenceResult<()> { - store_dispatch!(self.put(object_type, id, name, workspace, payload, labels)) + store_dispatch_traced!(self.put(object_type, id, name, workspace, payload, labels)) } pub async fn put_message< @@ -772,6 +906,17 @@ impl Store { } } +#[cfg(test)] +impl Store { + /// Closes the backing connection pool. + pub(crate) async fn close_for_test(&self) { + match self { + Self::Sqlite(store) => store.close_for_test().await, + Self::Postgres(_) => unreachable!("tests use SQLite"), + } + } +} + #[cfg(test)] pub async fn test_store() -> Store { Store::connect("sqlite::memory:?cache=shared") diff --git a/crates/openshell-server/src/persistence/sqlite.rs b/crates/openshell-server/src/persistence/sqlite.rs index 86f79a69e6..7e3616df62 100644 --- a/crates/openshell-server/src/persistence/sqlite.rs +++ b/crates/openshell-server/src/persistence/sqlite.rs @@ -29,6 +29,12 @@ pub struct SqliteStore { } impl SqliteStore { + /// Closes the connection pool. + #[cfg(test)] + pub(crate) async fn close_for_test(&self) { + self.pool.close().await; + } + pub async fn connect(url: &str) -> PersistenceResult { let is_in_memory = url.contains(":memory:") || url.contains("mode=memory"); let max_connections = if is_in_memory { 1 } else { 5 }; diff --git a/crates/openshell-server/src/persistence/tests.rs b/crates/openshell-server/src/persistence/tests.rs index 9539eab49d..948de71f1f 100644 --- a/crates/openshell-server/src/persistence/tests.rs +++ b/crates/openshell-server/src/persistence/tests.rs @@ -8,6 +8,98 @@ use openshell_core::proto::{ObjectForTest, Sandbox, SandboxPolicy, SandboxSpec}; use prost::Message; use std::collections::HashMap as StdHashMap; +/// A failed store call must be visible as a failure in the trace, not as a +/// span that merely happened to return nothing. +#[tokio::test] +async fn failed_store_calls_are_marked_on_the_span() { + use crate::otel_tracing::test_collector; + + let store = test_store().await; + let traced = test_collector::install_traced(); + store.close_for_test().await; + store + .get("sandbox", "failed-store-call") + .await + .expect_err("a closed pool fails the query"); + + let span = traced.span_with("store.get", "object.id", "failed-store-call"); + + assert!( + matches!(span.status, opentelemetry::trace::Status::Error { .. }), + "the span carries error status so trace UIs flag it, got {:?}", + span.status + ); +} + +/// Losing a `MustCreate` race is how callers learn a record already exists, so +/// the span must stay clean — otherwise every lease a replica does not win, and +/// every gateway restart, exports as a failure. +#[tokio::test] +async fn expected_conflicts_leave_the_span_unmarked() { + use crate::otel_tracing::test_collector; + + let store = test_store().await; + let traced = test_collector::install_traced(); + let put = async |id: &str| { + store + .put_if( + "workspace", + id, + "expected-conflict", + "", + b"payload", + None, + super::WriteCondition::MustCreate, + ) + .await + }; + + put("expected-conflict-first").await.expect("first write"); + put("expected-conflict-second") + .await + .expect_err("the name is already taken"); + + let span = traced.span_with("store.put_if", "object.id", "expected-conflict-second"); + + assert_eq!( + span.status, + opentelemetry::trace::Status::Unset, + "a unique violation is a return value the caller acts on, got {:?}", + span.status + ); +} + +/// Span names stay low-cardinality so they group across object types; what +/// each call touched is carried as attributes. +#[tokio::test] +async fn store_spans_record_what_they_touched_as_attributes() { + use crate::otel_tracing::test_collector; + + let store = test_store().await; + store + .put("sandbox", "abc", "my-sandbox", "default", b"payload", None) + .await + .unwrap(); + + let traced = test_collector::install_traced(); + store.get("sandbox", "abc").await.unwrap(); + store + .get_by_name("sandbox", "default", "my-sandbox") + .await + .unwrap(); + store.list("sandbox", "default", 10, 0).await.unwrap(); + + let by_name = traced.span_with("store.get_by_name", "object.name", "my-sandbox"); + assert_eq!( + test_collector::attribute(&by_name, "object_type").as_deref(), + Some("sandbox"), + "the span records which type it queried" + ); + + traced.span_with("store.get", "object.id", "abc"); + traced.span_with("store.list", "object_type", "sandbox"); +} + #[tokio::test] async fn sqlite_put_get_round_trip() { let store = test_store().await; @@ -1770,3 +1862,56 @@ async fn list_by_scope_returns_resource_version() { "list_by_scope must return the actual resource_version, not a default" ); } + +/// Store operations open a child span under whatever request span is active, +/// so a trace decomposes an RPC into the storage work it did rather than +/// bottoming out at the request boundary. +#[tokio::test] +async fn store_operations_export_child_spans_under_the_request_span() { + use tracing::Instrument as _; + + use crate::otel_tracing::test_collector; + + let store = test_store().await; + + let traced = test_collector::install_traced(); + async { + store + .list("sandbox", "default", 10, 0) + .await + .expect("list succeeds"); + } + .instrument(tracing::info_span!("request")) + .await; + + let spans = traced.finished_spans(); + let root = spans + .iter() + .find(|s| s.name == "request") + .expect("root span recorded"); + let child = spans + .iter() + .find(|s| s.name == "store.list") + .unwrap_or_else(|| { + panic!( + "a store span is recorded, got {:?}", + spans.iter().map(|s| &s.name).collect::>() + ) + }); + + assert_eq!( + child.parent_span_id, + root.span_context.span_id(), + "the store span is a child of the request span" + ); + assert_eq!( + child.span_context.trace_id(), + root.span_context.trace_id(), + "both share one trace" + ); + assert_eq!( + test_collector::attribute(child, "object_type").as_deref(), + Some("sandbox"), + "the store span records what it queried" + ); +} diff --git a/crates/openshell-server/src/provider_refresh.rs b/crates/openshell-server/src/provider_refresh.rs index a51ec53373..89427430f8 100644 --- a/crates/openshell-server/src/provider_refresh.rs +++ b/crates/openshell-server/src/provider_refresh.rs @@ -1000,9 +1000,20 @@ pub fn spawn_refresh_worker(state: std::sync::Arc, interval: }); } +#[tracing::instrument( + name = "refresh", + skip_all, + fields( + otel.name = "refresh.provider_credentials", + watched_count = tracing::field::Empty, + due_count = tracing::field::Empty, + ) +)] async fn run_refresh_worker_tick(store: &Store) -> Result<(), Status> { let now_ms = current_time_ms(); - let states = list_all_refresh_states(store).await?; + let states = list_all_refresh_states(store).await.inspect_err(|_| { + crate::otel_tracing::mark_error(&tracing::Span::current()); + })?; let watched_count = states.len(); let due_count = states .iter() @@ -1012,6 +1023,9 @@ async fn run_refresh_worker_tick(store: &Store) -> Result<(), Status> { .iter() .filter(|state| state.status == "rotation_requested") .count(); + let span = tracing::Span::current(); + span.record("watched_count", watched_count); + span.record("due_count", due_count); info!( watched_count, due_count, rotation_requested_count, "provider credential refresh worker sweep" @@ -1509,6 +1523,43 @@ mod tests { ); } + /// The worker ticks on a timer with no inbound request, so without a span + /// of its own its store reads export as anonymous single-span traces. + #[tokio::test] + async fn refresh_worker_ticks_export_one_trace_rooted_at_the_tick() { + use crate::otel_tracing::test_collector; + + let store = test_store().await; + + let traced = test_collector::install_traced(); + run_refresh_worker_tick(&store).await.unwrap(); + + let spans = traced.finished_spans(); + let root = spans + .iter() + .find(|s| s.name == "refresh.provider_credentials") + .unwrap_or_else(|| { + panic!( + "the tick records a span of its own, got {:?}", + spans.iter().map(|s| &s.name).collect::>() + ) + }); + + assert_eq!( + root.parent_span_id, + opentelemetry::trace::SpanId::INVALID, + "a timer-driven tick has no caller, so it roots its own trace" + ); + + assert!( + spans.iter().any(|s| { + s.name.starts_with("store.") + && s.span_context.trace_id() == root.span_context.trace_id() + }), + "the tick's store reads join its trace rather than orphaning" + ); + } + #[test] fn refresh_strategy_name_includes_aws_sts() { assert_eq!( diff --git a/crates/openshell-server/src/tracing_bus.rs b/crates/openshell-server/src/tracing_bus.rs index cc7b64ad32..a91a5fd877 100644 --- a/crates/openshell-server/src/tracing_bus.rs +++ b/crates/openshell-server/src/tracing_bus.rs @@ -10,9 +10,8 @@ use openshell_core::proto::{SandboxLogLine, SandboxStreamEvent}; use openshell_ocsf::OCSF_TARGET; use tokio::sync::broadcast; use tracing::{Event, Subscriber}; +use tracing_subscriber::Layer; use tracing_subscriber::layer::Context; -use tracing_subscriber::prelude::*; -use tracing_subscriber::{EnvFilter, Layer}; /// Bus that publishes server log lines keyed by sandbox id. #[derive(Debug, Clone)] @@ -45,18 +44,11 @@ impl TracingLogBus { } } - /// Install a tracing subscriber that logs to stdout and publishes events into this bus. - pub fn install_subscriber(&self, env_filter: EnvFilter) { - let layer = SandboxLogLayer { + pub(crate) fn layer(&self) -> impl Layer { + SandboxLogLayer { bus: self.clone(), default_tail: Self::DEFAULT_TAIL, - }; - - tracing_subscriber::registry() - .with(env_filter) - .with(tracing_subscriber::fmt::layer()) - .with(layer) - .init(); + } } fn sender_for(&self, sandbox_id: &str) -> broadcast::Sender { diff --git a/crates/openshell-server/src/tracing_setup.rs b/crates/openshell-server/src/tracing_setup.rs new file mode 100644 index 0000000000..321edefafe --- /dev/null +++ b/crates/openshell-server/src/tracing_setup.rs @@ -0,0 +1,47 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Process-wide tracing subscriber setup for the gateway. +//! +//! This module routes gateway logs and spans to configured diagnostic outputs. +//! `OpenShell` product telemetry collected for maintainers is handled by +//! [`crate::telemetry`]. + +use opentelemetry_sdk::trace::SdkTracerProvider; +use tracing_subscriber::EnvFilter; +use tracing_subscriber::prelude::*; + +use crate::config_file::OtlpConfig; +use crate::otel_tracing::SetupError; +use crate::tracing_bus::TracingLogBus; + +pub struct TracingHandle { + tracer_provider: Option, +} + +impl TracingHandle { + pub fn shutdown(&self) { + if let Some(provider) = &self.tracer_provider + && let Err(err) = provider.shutdown() + { + tracing::warn!(error = %err, "OTLP tracer provider shutdown failed"); + } + } +} + +pub fn install( + env_filter: EnvFilter, + tracing_log_bus: &TracingLogBus, + otlp_config: Option<&OtlpConfig>, +) -> (TracingHandle, Option) { + let (tracer_provider, setup_error) = crate::otel_tracing::provider_for(otlp_config); + + tracing_subscriber::registry() + .with(env_filter) + .with(tracing_subscriber::fmt::layer()) + .with(tracing_log_bus.layer()) + .with(tracer_provider.as_ref().map(crate::otel_tracing::layer)) + .init(); + + (TracingHandle { tracer_provider }, setup_error) +} diff --git a/docs/reference/gateway-config.mdx b/docs/reference/gateway-config.mdx index 27a1d3d45a..12f912e45f 100644 --- a/docs/reference/gateway-config.mdx +++ b/docs/reference/gateway-config.mdx @@ -140,6 +140,11 @@ allow_unauthenticated_users = false [openshell.gateway.mtls_auth] enabled = false +# OTLP export. Omit this table entirely to disable it. +[openshell.gateway.otlp] +endpoint = "http://otel-collector.observability.svc:4317" +service_name = "openshell-gateway" + [openshell.gateway.oidc] issuer = "https://idp.example.com/realms/openshell" audience = "openshell-cli" @@ -175,6 +180,51 @@ Local Docker, Podman, and VM gateways can also set `[openshell.gateway.mtls_auth `[openshell.gateway.auth] allow_unauthenticated_users = true` is an unsafe local-development and trusted-proxy escape hatch. It accepts user-facing CLI/API calls without OIDC or mTLS credentials while sandbox supervisors still authenticate with gateway-minted sandbox JWTs. Leave it false for shared and production gateways. +## OTLP Export + +`[openshell.gateway.otlp]` enables OpenTelemetry export over OTLP/gRPC. Omit the table to disable export; there is no separate `enabled` flag. + +The gateway already uses the Rust `tracing` framework for structured logs sent to stdout and the sandbox log stream. Enabling this section adds an OpenTelemetry layer to the same tracing subscriber. It exports span trees to an OTLP collector without exporting, replacing, or redirecting the existing log events. + +```toml +[openshell.gateway.otlp] +endpoint = "http://otel-collector.observability.svc:4317" +service_name = "openshell-gateway" +``` + +`endpoint` is required and must be a valid URI. If it is malformed, the gateway logs the configuration error and continues with export disabled. It does not connect at startup: an unreachable collector produces export failures, never a failure to serve. + +The transport is **OTLP over gRPC only**. HTTP/protobuf and HTTP/JSON are not supported, and `OTEL_EXPORTER_OTLP_PROTOCOL` has no effect. Point `endpoint` at a collector's gRPC receiver, conventionally port `4317`, not the HTTP receiver on `4318`. A URI alone cannot distinguish the two, so an HTTP endpoint is accepted at startup and then fails on export. + +The OpenTelemetry SDK logs export failures after startup. Spans in a failed batch are dropped rather than retried. + +`service_name` sets the `service.name` resource attribute and defaults to `openshell-gateway`. The gateway also reports `service.version`. + +Only OpenTelemetry traces are exported. Inbound gRPC and HTTP requests produce server spans named for the RPC or HTTP method and path. Store and compute-driver operations appear as child spans. Internal reconciliation, credential-refresh, and driver-watch loops create operation roots for their store work because no inbound request supplies a parent. The gateway continues valid W3C `traceparent` context and starts a new trace when none is supplied. Request spans carry `method`, `path`, and the `request_id` that also appears in gateway logs. Health endpoint spans use DEBUG level and are not exported by the default INFO filter. + +### Tuning + +This table decides whether and where to export. How the SDK exports is controlled by the standard OpenTelemetry environment variables, which the gateway reads through the SDK rather than mirroring as TOML keys: + +| Variable | Effect | +|---|---| +| `OTEL_TRACES_SAMPLER`, `OTEL_TRACES_SAMPLER_ARG` | Sampling strategy and ratio. Defaults to `parentbased_always_on`. | +| `OTEL_BSP_SCHEDULE_DELAY`, `OTEL_BSP_MAX_QUEUE_SIZE`, `OTEL_BSP_MAX_EXPORT_BATCH_SIZE`, `OTEL_BSP_EXPORT_TIMEOUT` | Batch span processor tuning. | +| `OTEL_RESOURCE_ATTRIBUTES` | Additional resource attributes, such as `deployment.environment=prod`. | +| `OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT`, `OTEL_SPAN_EVENT_COUNT_LIMIT`, `OTEL_SPAN_LINK_COUNT_LIMIT` | Per-span limits. | +| `OTEL_EXPORTER_OTLP_HEADERS`, `OTEL_EXPORTER_OTLP_COMPRESSION`, `OTEL_EXPORTER_OTLP_TIMEOUT` | Exporter transport tuning. | +| `OTEL_EXPORTER_OTLP_PROTOCOL` | No effect. The gateway is built with the gRPC exporter only. | + +To sample 10% of traces: + +```shell +OTEL_TRACES_SAMPLER=parentbased_traceidratio OTEL_TRACES_SAMPLER_ARG=0.1 +``` + +`OTEL_EXPORTER_OTLP_ENDPOINT` and `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` are deliberately ignored. Enablement has one source, so an environment variable cannot silently turn export on or redirect it. `OTEL_RESOURCE_ATTRIBUTES` does apply and adds attributes, but a `service_name` set here wins over `OTEL_SERVICE_NAME`. + +The gateway flushes buffered spans during shutdown, so spans from in-flight requests survive a `SIGTERM`. + ## Supervisor Middleware Services Register operator-run supervisor middleware services with one or more `[[openshell.supervisor.middleware]]` entries. Registration is static and operator-owned; changing it requires restarting the gateway.