diff --git a/Cargo.lock b/Cargo.lock index ff8a754f7..f89a80a84 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6015,7 +6015,12 @@ dependencies = [ "omnigraph-engine", "omnigraph-mcp", "omnigraph-policy", + "opentelemetry", + "opentelemetry-otlp", + "opentelemetry-proto", + "opentelemetry_sdk", "p256", + "prost", "regex", "reqwest 0.12.28", "rsa", @@ -6030,6 +6035,7 @@ dependencies = [ "tower", "tower-http", "tracing", + "tracing-opentelemetry", "tracing-subscriber", "url", "utoipa", @@ -6381,6 +6387,80 @@ dependencies = [ "vcpkg", ] +[[package]] +name = "opentelemetry" +version = "0.31.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b84bcd6ae87133e903af7ef497404dda70c60d0ea14895fc8a5e6722754fc2a0" +dependencies = [ + "futures-core", + "futures-sink", + "js-sys", + "pin-project-lite", + "thiserror", + "tracing", +] + +[[package]] +name = "opentelemetry-http" +version = "0.31.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d7a6d09a73194e6b66df7c8f1b680f156d916a1a942abf2de06823dd02b7855d" +dependencies = [ + "async-trait", + "bytes", + "http 1.4.2", + "opentelemetry", + "reqwest 0.12.28", +] + +[[package]] +name = "opentelemetry-otlp" +version = "0.31.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f69cd6acbb9af919df949cd1ec9e5e7fdc2ef15d234b6b795aaa525cc02f71f" +dependencies = [ + "http 1.4.2", + "opentelemetry", + "opentelemetry-http", + "opentelemetry-proto", + "opentelemetry_sdk", + "prost", + "reqwest 0.12.28", + "thiserror", + "tracing", +] + +[[package]] +name = "opentelemetry-proto" +version = "0.31.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7175df06de5eaee9909d4805a3d07e28bb752c34cab57fa9cff549da596b30f" +dependencies = [ + "opentelemetry", + "opentelemetry_sdk", + "prost", + "tonic", + "tonic-prost", +] + +[[package]] +name = "opentelemetry_sdk" +version = "0.31.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e14ae4f5991976fd48df6d843de219ca6d31b01daaab2dad5af2badeded372bd" +dependencies = [ + "futures-channel", + "futures-executor", + "futures-util", + "opentelemetry", + "percent-encoding", + "rand 0.9.4", + "thiserror", + "tokio", + "tokio-stream", +] + [[package]] name = "option-ext" version = "0.2.0" @@ -8836,6 +8916,38 @@ version = "1.1.2+spec-1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7d56353a2a665ad0f41a421187180aab746c8c325620617ad883a99a1cbe66d2" +[[package]] +name = "tonic" +version = "0.14.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac2a5518c70fa84342385732db33fb3f44bc4cc748936eb5833d2df34d6445ef" +dependencies = [ + "async-trait", + "base64 0.22.1", + "bytes", + "http 1.4.2", + "http-body 1.0.1", + "http-body-util", + "percent-encoding", + "pin-project", + "sync_wrapper", + "tokio-stream", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "tonic-prost" +version = "0.14.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "50849f68853be452acf590cde0b146665b8d507b3b8af17261df47e02c209ea0" +dependencies = [ + "bytes", + "prost", + "tonic", +] + [[package]] name = "tower" version = "0.5.3" @@ -8955,6 +9067,20 @@ dependencies = [ "tracing-core", ] +[[package]] +name = "tracing-opentelemetry" +version = "0.32.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ac28f2d093c6c477eaa76b23525478f38de514fa9aeb1285738d4b97a9552fc" +dependencies = [ + "js-sys", + "opentelemetry", + "tracing", + "tracing-core", + "tracing-subscriber", + "web-time", +] + [[package]] name = "tracing-serde" version = "0.2.0" diff --git a/changelog.d/server-otel-traces.added.md b/changelog.d/server-otel-traces.added.md new file mode 100644 index 000000000..e6697b500 --- /dev/null +++ b/changelog.d/server-otel-traces.added.md @@ -0,0 +1,6 @@ +- The server supports optional OTLP HTTP/protobuf tracing and structured JSON + logs with trace correlation. Request diagnostics use route templates and + exclude request payloads and credentials. Log verbosity does not disable + tracing or warning/error correlation. See [logs and traces][server-otel-guide]. + +[server-otel-guide]: ../docs/user/operations/observability.md diff --git a/crates/omnigraph-server/Cargo.toml b/crates/omnigraph-server/Cargo.toml index 804dd76c1..4619f4a73 100644 --- a/crates/omnigraph-server/Cargo.toml +++ b/crates/omnigraph-server/Cargo.toml @@ -33,8 +33,12 @@ serde_json = { workspace = true } tokio = { workspace = true } yaml_serde = { workspace = true } tracing = { workspace = true } -tracing-subscriber = { workspace = true } +tracing-subscriber = { workspace = true, features = ["json"] } tower-http = { workspace = true } +opentelemetry = { version = "0.31", default-features = false, features = ["trace"] } +opentelemetry_sdk = { version = "0.31", default-features = false, features = ["trace"] } +opentelemetry-otlp = { version = "0.31", default-features = false, features = ["trace", "http-proto", "reqwest-blocking-client", "reqwest-rustls-webpki-roots", "internal-logs"] } +tracing-opentelemetry = { version = "0.32", default-features = false } utoipa = { workspace = true } futures = { workspace = true } url = { workspace = true } @@ -55,6 +59,9 @@ aws-config = { version = "1", optional = true, default-features = false, feature aws-sdk-secretsmanager = { version = "1", optional = true, default-features = false, features = ["default-https-client", "rt-tokio"] } [dev-dependencies] +opentelemetry_sdk = { version = "0.31", default-features = false, features = ["trace", "testing"] } +opentelemetry-proto = { version = "0.31", default-features = false, features = ["gen-tonic-messages", "trace"] } +prost = "0.14" omnigraph = { package = "omnigraph-engine", path = "../omnigraph", features = ["failpoints"] } omnigraph-cluster = { path = "../omnigraph-cluster", version = "0.14.1", features = ["test-util", "failpoints"] } tempfile = { workspace = true } diff --git a/crates/omnigraph-server/src/deployment.rs b/crates/omnigraph-server/src/deployment.rs index d73138175..587a308d6 100644 --- a/crates/omnigraph-server/src/deployment.rs +++ b/crates/omnigraph-server/src/deployment.rs @@ -456,6 +456,7 @@ pub(crate) async fn apply( "a canonical deployment_id is required", )); } + crate::telemetry::deployment_id(&request.deployment_id); let gate = state .deployments .gate diff --git a/crates/omnigraph-server/src/lib.rs b/crates/omnigraph-server/src/lib.rs index a0829806b..8435934dc 100644 --- a/crates/omnigraph-server/src/lib.rs +++ b/crates/omnigraph-server/src/lib.rs @@ -15,6 +15,7 @@ mod mcp; pub mod operations; mod redacted_cause; mod settings; +pub mod telemetry; use extract::{ApiJson, ApiPath, ApiQuery}; use handlers::*; use settings::*; @@ -87,7 +88,6 @@ use sha2::{Digest, Sha256}; use std::io::{self, Write}; use subtle::ConstantTimeEq; use tokio::net::TcpListener; -use tower_http::trace::TraceLayer; use tracing::{error, info, warn}; use tracing_subscriber::EnvFilter; use utoipa::OpenApi; @@ -3434,23 +3434,17 @@ fn server_log_subscriber(filter: EnvFilter, writer: W) -> impl tracing::Subsc where W: for<'a> tracing_subscriber::fmt::MakeWriter<'a> + Send + Sync + 'static, { - use tracing_subscriber::layer::SubscriberExt as _; - - tracing_subscriber::registry() - .with(filter) - // This independent metadata filter cannot be overridden by a more - // specific RUST_LOG directive, and keeps graph values, storage URIs - // and entity IDs out of dependency logs. - .with(tracing_subscriber::filter::filter_fn(|metadata| { - let target = metadata.target(); - let restricted = RESTRICTED_LOG_TARGETS.iter().any(|restricted| { - target - .strip_prefix(restricted) - .is_some_and(|rest| rest.is_empty() || rest.starts_with("::")) - }); - !restricted || *metadata.level() <= tracing::Level::WARN - })) - .with(tracing_subscriber::fmt::layer().with_writer(writer)) + telemetry::subscriber(filter, writer, false, None) +} + +fn safe_log_metadata(metadata: &tracing::Metadata<'_>) -> bool { + let target = metadata.target(); + let restricted = RESTRICTED_LOG_TARGETS.iter().any(|restricted| { + target + .strip_prefix(restricted) + .is_some_and(|rest| rest.is_empty() || rest.starts_with("::")) + }); + !restricted || *metadata.level() <= tracing::Level::WARN } /// Install native server logging with MCP protocol payload logs and Lance's @@ -3487,6 +3481,19 @@ pub(crate) mod test_log_capture { super::server_log_subscriber(EnvFilter::new(directives), move || writer.clone()) } + pub(crate) fn json_subscriber( + &self, + directives: &str, + ) -> impl tracing::Subscriber + Send + Sync { + let writer = self.clone(); + super::telemetry::subscriber( + EnvFilter::new(directives), + move || writer.clone(), + true, + None, + ) + } + /// Everything written so far, with ANSI styling removed so a test can /// match `field="value"` spellings. pub(crate) fn output(&self) -> String { @@ -3528,16 +3535,23 @@ mod log_filter_tests { #[test] fn restricted_dependency_logs_remain_filtered_under_specific_directives() { // Each directive set, and whether it enables DEBUG on unrestricted targets. - for (directives, debug) in [ + for ((directives, debug), json) in [ ("trace", true), ("debug,rmcp::service=trace", true), ( "info,lance::dataset_events=trace,lance::file_audit=trace", false, ), - ] { + ] + .into_iter() + .flat_map(|case| [false, true].map(|json| (case, json))) + { let captured = Capture::default(); - let subscriber = captured.subscriber(directives); + let subscriber: Box = if json { + Box::new(captured.json_subscriber(directives)) + } else { + Box::new(captured.subscriber(directives)) + }; tracing::subscriber::with_default(subscriber, || { tracing::debug!(target: "rmcp::service", request = "PRIVATE_REQUEST_MARKER", "received request"); tracing::trace!(target: "rmcp::transport::streamable_http_server::tower", message = "PRIVATE_RESULT_MARKER"); @@ -3747,7 +3761,7 @@ pub fn build_app(state: AppState) -> Router { .layer(middleware::from_fn(http_contract::identify_response)) .layer(middleware::from_fn_with_state(challenge, challenge::answer)) .layer(DefaultBodyLimit::max(DEFAULT_REQUEST_BODY_LIMIT_BYTES)) - .layer(TraceLayer::new_for_http()) + .layer(middleware::from_fn(telemetry::http_request)) .with_state(state) } diff --git a/crates/omnigraph-server/src/main.rs b/crates/omnigraph-server/src/main.rs index 3f051ec59..fb5fa076f 100644 --- a/crates/omnigraph-server/src/main.rs +++ b/crates/omnigraph-server/src/main.rs @@ -3,7 +3,7 @@ use std::path::PathBuf; use clap::Parser; use color_eyre::eyre::Result; use omnigraph_server::{ - McpSettings, init_tracing, load_server_settings, load_server_settings_with_bootstrap_handoff, + McpSettings, load_server_settings, load_server_settings_with_bootstrap_handoff, load_server_settings_with_identity_trust, resolve_shutdown_grace, serve_with_data_token_trust, serve_with_mcp_settings, }; @@ -76,9 +76,14 @@ struct Cli { #[tokio::main] async fn main() -> Result<()> { color_eyre::install()?; - init_tracing(); - let cli = Cli::parse(); + let telemetry = tokio::task::spawn_blocking(omnigraph_server::telemetry::init).await??; + let result = run(cli).await; + telemetry.shutdown_in_background(); + result +} + +async fn run(cli: Cli) -> Result<()> { let mcp = McpSettings::resolve( cli.mcp_allow_adhoc, cli.mcp_allow_writes, diff --git a/crates/omnigraph-server/src/operations.rs b/crates/omnigraph-server/src/operations.rs index 7ac838dbf..99fc5299b 100644 --- a/crates/omnigraph-server/src/operations.rs +++ b/crates/omnigraph-server/src/operations.rs @@ -262,6 +262,8 @@ impl OperationRuntime { R: Send + 'static, F: Future> + Send + 'static, { + use tracing::instrument::WithSubscriber as _; + { let mut state = locked(&self.inner.state); if state.closed { @@ -275,6 +277,8 @@ impl OperationRuntime { reservation: Some(Box::new(reservation)), }; let (sender, receiver) = oneshot::channel(); + // A span retains its subscriber, but entering it does not install that + // subscriber for Span::current() or events on a detached task. tokio::spawn( async move { // The write runs as its request's submitted write: an error it @@ -296,6 +300,12 @@ impl OperationRuntime { // Mark the error as its waiter receives it, so the log below // records the status its response sends. let result = result.map_err(ApiError::after_owned_write); + // Process containment and completion uncertainty are distinct: + // scoped storage uncertainty leaves admission open, but remains + // an uncertain outcome in telemetry. + let telemetry_uncertain = + uncertain || result.as_ref().is_err_and(ApiError::completion_uncertain); + crate::telemetry::operation_finished("write", result.is_ok(), telemetry_uncertain); // A known result carries its reservation through pending delivery. // Consumption releases admission before the handler can respond; // abandonment destroys the output before releasing its ownership. @@ -324,7 +334,8 @@ impl OperationRuntime { owner, }); } - .in_current_span(), + .instrument(crate::telemetry::operation_span("write")) + .with_current_subscriber(), ); Ok(OwnedResponse(receiver)) } @@ -478,6 +489,8 @@ impl ReadObserver { R: Send + 'static, F: Future> + Send + 'static, { + use tracing::instrument::WithSubscriber as _; + let (sender, receiver) = oneshot::channel(); let response_observer = self.clone(); tokio::spawn( @@ -494,10 +507,12 @@ impl ReadObserver { result.map_err(ApiError::with_request_facts) }) .await; + crate::telemetry::operation_finished("read", result.is_ok(), false); // No result history: an absent waiter drops the result here. let _ = sender.send(result); } - .in_current_span(), + .instrument(crate::telemetry::operation_span("read")) + .with_current_subscriber(), ); OwnedReadResponse { receiver, @@ -816,6 +831,18 @@ mod tests { #[test] fn disconnected_owned_operation_keeps_its_request_span() { + fn request_parent() -> Option { + use tracing_subscriber::registry::LookupSpan; + tracing::Span::current() + .with_subscriber(|(id, dispatch)| { + let registry = dispatch.downcast_ref::()?; + let operation = registry.span(id)?; + assert_eq!(operation.name(), "native.operation"); + operation.parent().map(|parent| parent.id()) + }) + .flatten() + } + let subscriber = tracing_subscriber::registry().with(tracing_subscriber::filter::LevelFilter::DEBUG); use tracing_subscriber::prelude::*; @@ -832,7 +859,7 @@ mod tests { let _entered = span.enter(); runtime .submit((), async move { - sent.send(tracing::Span::current().id()).unwrap(); + sent.send(request_parent()).unwrap(); Ok::<_, ApiError>(()).into() }) .unwrap() @@ -843,7 +870,7 @@ mod tests { let response = { let _entered = span.enter(); runtime.try_observe().unwrap().spawn_read((), async move { - sent.send(tracing::Span::current().id()).unwrap(); + sent.send(request_parent()).unwrap(); Ok::<_, ApiError>(()) }) }; diff --git a/crates/omnigraph-server/src/telemetry.rs b/crates/omnigraph-server/src/telemetry.rs new file mode 100644 index 000000000..ba95ea154 --- /dev/null +++ b/crates/omnigraph-server/src/telemetry.rs @@ -0,0 +1,756 @@ +//! Optional vendor-neutral tracing and safe HTTP request diagnostics. +//! +//! Export failure never changes request admission, execution, or ownership. + +use std::fmt; +use std::time::{Duration, Instant}; + +use axum::extract::{MatchedPath, Request}; +use axum::http::HeaderMap; +use axum::middleware::Next; +use axum::response::Response; +use color_eyre::eyre::{Result, bail, eyre}; +use opentelemetry::KeyValue; +use opentelemetry::propagation::{Extractor, TextMapPropagator}; +use opentelemetry::trace::{Status, TracerProvider}; +use opentelemetry_otlp::{Protocol, WithExportConfig}; +use opentelemetry_sdk::Resource; +use opentelemetry_sdk::propagation::TraceContextPropagator; +use opentelemetry_sdk::trace::{ + BatchConfigBuilder, BatchSpanProcessor, SdkTracer, SdkTracerProvider, +}; +use tracing::{Instrument, Span, Subscriber}; +use tracing_opentelemetry::OpenTelemetrySpanExt; +use tracing_subscriber::filter::FilterExt; +use tracing_subscriber::fmt::format::{JsonFields, Writer}; +use tracing_subscriber::fmt::{FmtContext, FormatEvent, MakeWriter}; +use tracing_subscriber::layer::SubscriberExt; +use tracing_subscriber::registry::LookupSpan; +use tracing_subscriber::util::SubscriberInitExt; +use tracing_subscriber::{EnvFilter, Layer}; + +const TARGET: &str = "omnigraph_server::telemetry"; +const EXPORT_TIMEOUT: Duration = Duration::from_secs(3); +const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5); + +/// Retain until serving has finished. Its bounded flush is telemetry-only, +/// never native cleanup or a reason to delay the native shutdown deadline. +pub struct TelemetryGuard(Option); + +impl TelemetryGuard { + /// Start a best-effort flush without delaying process exit. An embedder + /// without a native shutdown deadline may call [`Self::shutdown`] instead. + pub fn shutdown_in_background(self) { + if self.0.is_none() { + return; + } + // If spawning fails, leave the provider for process exit rather than + // running its blocking Drop on the native shutdown path. + let guard = std::mem::ManuallyDrop::new(self); + if std::thread::Builder::new() + .name("telemetry-shutdown".into()) + .spawn(move || std::mem::ManuallyDrop::into_inner(guard).shutdown()) + .is_err() + { + tracing::warn!(target: TARGET, "trace exporter shutdown worker could not start"); + } + } + + /// Flush on a blocking thread, bounded independently of native cleanup. + pub fn shutdown(self) { + if let Some(provider) = self.0 + && provider.shutdown_with_timeout(SHUTDOWN_TIMEOUT).is_err() + { + tracing::warn!(target: TARGET, "trace exporter shutdown did not complete"); + } + } +} + +/// Install the binary's logging and optional OTLP exporter. Call from a +/// blocking thread: the exporter owns a bounded worker and blocking HTTP client. +/// An explicit OTLP endpoint opts in; no default network destination is used. +pub fn init() -> Result { + let config = Config::read(|key| std::env::var(key).ok())?; + let provider = config.endpoint.as_deref().map(build_provider).transpose()?; + let tracer = provider + .as_ref() + .map(|provider| provider.tracer("omnigraph-server")); + let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")); + if subscriber(filter, std::io::stdout, config.json, tracer) + .try_init() + .is_err() + { + if let Some(provider) = provider { + let _ = provider.shutdown_with_timeout(SHUTDOWN_TIMEOUT); + } + bail!("server telemetry subscriber is already installed"); + } + Ok(TelemetryGuard(provider)) +} + +#[derive(Debug, PartialEq, Eq)] +struct Config { + json: bool, + endpoint: Option, +} + +impl Config { + fn read(env: impl Fn(&str) -> Option) -> Result { + let json = match env("OMNIGRAPH_LOG_FORMAT").as_deref().unwrap_or("text") { + "text" => false, + "json" => true, + _ => bail!("OMNIGRAPH_LOG_FORMAT must be text or json"), + }; + let disabled = + env("OTEL_SDK_DISABLED").is_some_and(|value| value.eq_ignore_ascii_case("true")); + let specific = env("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT").filter(|value| !value.is_empty()); + let general = env("OTEL_EXPORTER_OTLP_ENDPOINT").filter(|value| !value.is_empty()); + let endpoint = if disabled { + None + } else if let Some(endpoint) = specific { + Some(endpoint) + } else { + general.map(|endpoint| format!("{}/v1/traces", endpoint.trim_end_matches('/'))) + }; + if let Some(endpoint) = &endpoint { + let protocol = env("OTEL_EXPORTER_OTLP_TRACES_PROTOCOL") + .or_else(|| env("OTEL_EXPORTER_OTLP_PROTOCOL")); + if protocol.is_some_and(|value| value != "http/protobuf") { + bail!("server trace export supports only OTLP http/protobuf"); + } + // Never put a potentially credential-bearing endpoint in errors. + let url = + url::Url::parse(endpoint).map_err(|_| eyre!("invalid OTLP trace endpoint"))?; + if !matches!(url.scheme(), "http" | "https") + || url.host_str().is_none() + || !url.username().is_empty() + || url.password().is_some() + || url.query().is_some() + || url.fragment().is_some() + { + bail!("OTLP trace endpoint must be HTTP(S) without userinfo, query or fragment"); + } + } + Ok(Self { json, endpoint }) + } +} + +fn build_provider(endpoint: &str) -> Result { + let exporter = opentelemetry_otlp::SpanExporter::builder() + .with_http() + .with_endpoint(endpoint) + .with_protocol(Protocol::HttpBinary) + .with_timeout(EXPORT_TIMEOUT) + .build() + .map_err(|_| eyre!("could not initialize OTLP trace exporter"))?; + let processor = BatchSpanProcessor::builder(exporter) + .with_batch_config( + BatchConfigBuilder::default() + .with_max_queue_size(2048) + .with_max_export_batch_size(256) + .with_scheduled_delay(Duration::from_secs(5)) + .build(), + ) + .build(); + // Resource::builder reads OTEL_SERVICE_NAME and OTEL_RESOURCE_ATTRIBUTES. + let detected = Resource::builder().build(); + let resource = Resource::builder_empty() + .with_service_name("omnigraph-server") + .with_attribute(KeyValue::new("service.version", env!("CARGO_PKG_VERSION"))) + .with_attributes( + detected + .iter() + .filter(|(key, value)| { + key.as_str() != "service.name" || value.to_string() != "unknown_service" + }) + .map(|(key, value)| KeyValue::new(key.clone(), value.clone())), + ) + .build(); + Ok(SdkTracerProvider::builder() + .with_span_processor(processor) + .with_resource(resource) + .build()) +} + +pub(crate) fn subscriber( + filter: EnvFilter, + writer: W, + json: bool, + tracer: Option, +) -> Box +where + W: for<'writer> MakeWriter<'writer> + Send + Sync + 'static, +{ + let telemetry = tracer.map(|tracer| { + tracing_opentelemetry::layer() + .with_tracer(tracer) + .with_location(false) + .with_threads(false) + .with_tracked_inactivity(false) + // Export only explicitly authored spans/events. A library's DEBUG + // request body or storage path must never become a trace payload. + .with_filter(tracing_subscriber::filter::filter_fn(|metadata| { + metadata.target() == TARGET + })) + }); + // Keep authored spans available to log formatting so warning/error events + // retain trace correlation even when their INFO parents are not logged. + // RUST_LOG controls events independently of OTLP sampling and export. + let log_filter = filter.or(tracing_subscriber::filter::filter_fn(|metadata| { + metadata.is_span() && metadata.target() == TARGET + })); + let subscriber = tracing_subscriber::registry() + .with(tracing_subscriber::filter::filter_fn( + super::safe_log_metadata, + )) + .with(telemetry); + if json { + Box::new( + subscriber.with( + tracing_subscriber::fmt::layer() + .fmt_fields(JsonFields::new()) + .event_format(CorrelatedJson) + .with_writer(writer) + .with_filter(log_filter), + ), + ) + } else { + Box::new( + subscriber.with( + tracing_subscriber::fmt::layer() + .with_writer(writer) + .with_filter(log_filter), + ), + ) + } +} + +struct CorrelatedJson; + +impl FormatEvent for CorrelatedJson +where + S: Subscriber + for<'lookup> LookupSpan<'lookup>, +{ + fn format_event( + &self, + ctx: &FmtContext<'_, S, JsonFields>, + mut writer: Writer<'_>, + event: &tracing::Event<'_>, + ) -> fmt::Result { + // Keep tracing-subscriber's field/time handling; add only standard + // correlation IDs. Never recover IDs from raw incoming header text. + let mut encoded = String::new(); + tracing_subscriber::fmt::format() + .json() + .flatten_event(true) + .format_event(ctx, Writer::new(&mut encoded), event)?; + let mut record: serde_json::Value = + serde_json::from_str(&encoded).map_err(|_| fmt::Error)?; + // Use the event's registry scope: recursive dispatcher access through + // Span::current() inside a subscriber can return a disabled span. + if let Some(scope) = ctx.event_scope() { + for span in scope { + let extensions = span.extensions(); + if let Some(data) = extensions.get::() + && let (Some(trace_id), Some(span_id)) = (data.trace_id(), data.span_id()) + { + record["trace_id"] = trace_id.to_string().into(); + record["span_id"] = span_id.to_string().into(); + break; + } + } + } + writeln!(writer, "{record}") + } +} + +struct TraceHeaders<'a>(&'a HeaderMap); + +impl Extractor for TraceHeaders<'_> { + fn get(&self, key: &str) -> Option<&str> { + // W3C context is correlation only. Never import baggage, credentials or + // arbitrary request headers into a span or log. + matches!(key, "traceparent" | "tracestate") + .then(|| { + self.0 + .get(key)? + .to_str() + .ok() + .filter(|value| value.len() <= 512) + }) + .flatten() + } + + fn keys(&self) -> Vec<&str> { + vec!["traceparent", "tracestate"] + } +} + +pub(crate) async fn http_request(request: Request, next: Next) -> Response { + let route = request + .extensions() + .get::() + .map_or("", MatchedPath::as_str); + let method = match request.method().as_str() { + "GET" | "POST" | "PUT" | "DELETE" | "PATCH" | "HEAD" | "OPTIONS" | "CONNECT" | "TRACE" => { + request.method().as_str() + } + _ => "_OTHER", + }; + let span = tracing::info_span!( + target: TARGET, + parent: None, + "http.server", + otel.name = %format!("{method} {route}"), + otel.kind = "server", + http.request.method = method, + http.route = route, + http.response.status_code = tracing::field::Empty, + deployment.id = tracing::field::Empty, + ); + let parent = TraceContextPropagator::new().extract(&TraceHeaders(request.headers())); + let _ = span.set_parent(parent); + async move { + let started = Instant::now(); + let response = next.run(request).await; + let status = response.status(); + let current = Span::current(); + current.record("http.response.status_code", status.as_u16()); + if status.is_server_error() { + current.set_status(Status::error("HTTP server error")); + } + tracing::info!(target: TARGET, status_code = status.as_u16(), response_headers_ms = started.elapsed().as_secs_f64() * 1000.0, "http request completed"); + response + } + .instrument(span) + .await +} + +pub(crate) fn operation_span(kind: &'static str) -> Span { + tracing::info_span!(target: TARGET, "native.operation", otel.kind = "internal", operation.kind = kind) +} + +pub(crate) fn operation_finished(kind: &'static str, succeeded: bool, uncertain: bool) { + let outcome = if uncertain { + "uncertain" + } else if succeeded { + "succeeded" + } else { + "failed" + }; + if !succeeded || uncertain { + Span::current().set_status(Status::error(outcome)); + } + tracing::info!(target: TARGET, operation_kind = kind, operation_outcome = outcome, "native operation completed"); +} + +pub(crate) fn deployment_id(id: &str) { + if !id.is_empty() + && id.len() <= 75 + && id + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || b"-_:".contains(&byte)) + { + Span::current().record("deployment.id", id); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use axum::body::{Body, Bytes}; + use axum::http::StatusCode; + use axum::routing::post; + use axum::{Router, middleware}; + use opentelemetry::trace::{SpanKind, Tracer}; + use opentelemetry_sdk::trace::InMemorySpanExporter; + use prost::Message; + use tower::ServiceExt; + use tracing::instrument::WithSubscriber; + + use crate::operations::{OperationRuntime, OwnedResult}; + use crate::test_log_capture::Capture; + + fn config(entries: &[(&str, &str)]) -> Result { + Config::read(|key| { + entries + .iter() + .find(|(name, _)| *name == key) + .map(|(_, value)| value.to_string()) + }) + } + + #[test] + fn exporter_requires_an_explicit_endpoint_and_valid_safe_configuration() { + assert_eq!( + config(&[]).unwrap(), + Config { + json: false, + endpoint: None + } + ); + let general = ("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector:4318/"); + let specific = ( + "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT", + "https://collector/custom-traces", + ); + assert_eq!( + config(&[general]).unwrap().endpoint.as_deref(), + Some("http://collector:4318/v1/traces") + ); + assert_eq!( + config(&[general, specific]).unwrap().endpoint.as_deref(), + Some(specific.1) + ); + assert!( + config(&[general, ("OTEL_SDK_DISABLED", "true")]) + .unwrap() + .endpoint + .is_none() + ); + assert!(config(&[("OMNIGRAPH_LOG_FORMAT", "json")]).unwrap().json); + assert!(config(&[("OMNIGRAPH_LOG_FORMAT", "invalid")]).is_err()); + assert!(config(&[general, ("OTEL_EXPORTER_OTLP_PROTOCOL", "grpc")]).is_err()); + for endpoint in [ + "not-a-url", + "file:///private", + "http://user:PRIVATE@collector", + "https://collector?token=PRIVATE", + "https://collector/#PRIVATE", + ] { + let error = config(&[("OTEL_EXPORTER_OTLP_ENDPOINT", endpoint)]) + .unwrap_err() + .to_string(); + assert!(!error.contains("PRIVATE"), "{error}"); + } + } + + #[tokio::test] + async fn http_and_owned_operations_preserve_parent_context_without_payloads() { + for (directives, json) in [ + ("info", true), + ("warn", true), + ("error", true), + ("off", true), + ("warn", false), + ] { + http_and_owned_operations_case(directives, json).await; + } + } + + async fn http_and_owned_operations_case(directives: &str, json: bool) { + let exporter = InMemorySpanExporter::default(); + let provider = SdkTracerProvider::builder() + .with_simple_exporter(exporter.clone()) + .build(); + let captured = Capture::default(); + let writer = captured.clone(); + let subscriber = subscriber( + EnvFilter::new(directives), + move || writer.clone(), + json, + Some(provider.tracer("test")), + ); + async { + let app = Router::new() + .route( + "/graphs/{graph_id}/query", + post(|| async { + let runtime = OperationRuntime::new(); + runtime + .submit((), async { + OwnedResult { + result: Ok(()), + uncertain: false, + } + }) + .unwrap() + .result() + .await + .unwrap(); + runtime + .try_observe() + .unwrap() + .spawn_read((), async { Ok(()) }) + .result() + .await + .unwrap(); + tracing::warn!(target: TARGET, "request warning marker"); + tracing::error!(target: TARGET, "request error marker"); + StatusCode::SERVICE_UNAVAILABLE + }), + ) + .layer(middleware::from_fn(http_request)); + let response = app + .oneshot( + Request::post("/graphs/PRIVATE_GRAPH/query?secret=PRIVATE_QUERY") + .header( + "traceparent", + "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01", + ) + .header("authorization", "Bearer PRIVATE_TOKEN") + .header("baggage", "secret=PRIVATE_BAGGAGE") + .body(Body::from("PRIVATE_BODY")) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); + } + .with_subscriber(subscriber) + .await; + provider.force_flush().unwrap(); + let spans = exporter.get_finished_spans().unwrap(); + assert_eq!(spans.len(), 3, "{spans:?}"); + let http = spans + .iter() + .find(|span| span.span_kind == SpanKind::Server) + .unwrap(); + assert_eq!(http.name, "POST /graphs/{graph_id}/query"); + assert_eq!( + http.span_context.trace_id().to_string(), + "0af7651916cd43dd8448eb211c80319c" + ); + assert_eq!(http.parent_span_id.to_string(), "b7ad6b7169203331"); + assert!(matches!(http.status, Status::Error { .. })); + let logs = captured.output(); + for native in spans.iter().filter(|span| span.name == "native.operation") { + assert_eq!(native.parent_span_id, http.span_context.span_id()); + assert_eq!(native.span_context.trace_id(), http.span_context.trace_id()); + } + if directives == "info" { + let native_completions: Vec = logs + .lines() + .map(|line| serde_json::from_str::(line).unwrap()) + .filter(|line| line["message"] == "native operation completed") + .collect(); + assert_eq!(native_completions.len(), 2, "{logs}"); + for native in spans.iter().filter(|span| span.name == "native.operation") { + let completion = native_completions + .iter() + .find(|line| line["span_id"] == native.span_context.span_id().to_string()) + .expect("owned task completion uses its captured subscriber and span"); + assert_eq!( + completion["trace_id"], + http.span_context.trace_id().to_string() + ); + assert_eq!(completion["operation_outcome"], "succeeded"); + } + assert!( + native_completions + .iter() + .any(|line| line["operation_kind"] == "write") + ); + assert!( + native_completions + .iter() + .any(|line| line["operation_kind"] == "read") + ); + let completion: serde_json::Value = logs + .lines() + .map(|line| serde_json::from_str::(line).unwrap()) + .find(|line| line["message"] == "http request completed") + .unwrap(); + assert_eq!( + completion["trace_id"], + http.span_context.trace_id().to_string() + ); + assert_eq!( + completion["span_id"], + http.span_context.span_id().to_string() + ); + assert_eq!(completion["status_code"], 503); + assert!(completion["response_headers_ms"].is_number()); + } else { + assert!(!logs.contains("http request completed"), "{logs}"); + assert!(!logs.contains("native operation completed"), "{logs}"); + if directives == "warn" || directives == "error" { + assert_eq!( + logs.contains("request warning marker"), + directives == "warn" + ); + assert!(logs.contains("request error marker"), "{logs}"); + if json { + for line in logs.lines() { + let record: serde_json::Value = serde_json::from_str(line).unwrap(); + assert_eq!(record["trace_id"], http.span_context.trace_id().to_string()); + assert_eq!(record["span_id"], http.span_context.span_id().to_string()); + } + } + } else { + assert!(logs.is_empty(), "{logs}"); + } + } + for private in [ + "PRIVATE_GRAPH", + "PRIVATE_QUERY", + "PRIVATE_TOKEN", + "PRIVATE_BAGGAGE", + "PRIVATE_BODY", + ] { + assert!(!logs.contains(private), "{logs}"); + assert!(!format!("{spans:?}").contains(private)); + } + provider.shutdown().unwrap(); + } + + #[tokio::test] + async fn scoped_write_uncertainty_is_reported_without_containing_the_process() { + use omnigraph::error::OmniError; + + for (error, outcome, effects, action) in [ + ( + OmniError::manifest_publish_in_doubt("PRIVATE_PUBLICATION_DETAIL"), + "uncertain", + crate::api::DispositionEffects::Unknown, + crate::api::DispositionAction::Reconcile, + ), + ( + OmniError::Policy("PRIVATE_POLICY_DETAIL".into()), + "failed", + crate::api::DispositionEffects::None, + crate::api::DispositionAction::Refresh, + ), + ] { + let exporter = InMemorySpanExporter::default(); + let provider = SdkTracerProvider::builder() + .with_simple_exporter(exporter.clone()) + .build(); + let captured = Capture::default(); + let writer = captured.clone(); + let subscriber = subscriber( + EnvFilter::new("info"), + move || writer.clone(), + true, + Some(provider.tracer("test")), + ); + async { + let runtime = OperationRuntime::new(); + let response = runtime + .submit((), async move { + OwnedResult::from(Err::<(), _>(crate::ApiError::from_omni(error))) + }) + .unwrap() + .result() + .await + .unwrap_err(); + assert_eq!(response.completion_uncertain(), outcome == "uncertain"); + assert!(!response.contains_process()); + let disposition = response.disposition(); + assert_eq!(disposition.effects, effects); + assert_eq!(disposition.action, action); + assert!(runtime.wait_logical_owners().await); + assert!(!runtime.snapshot().closed); + assert_eq!(runtime.snapshot().uncertain_writes, 0); + // Observing the failure must not change the native admission + // contract: a scoped uncertainty leaves the process available. + drop(runtime.try_observe().unwrap()); + } + .with_subscriber(subscriber) + .await; + provider.force_flush().unwrap(); + let spans = exporter.get_finished_spans().unwrap(); + assert_eq!(spans.len(), 1, "{spans:?}"); + assert_eq!(spans[0].name, "native.operation"); + assert_eq!(spans[0].status, Status::error(outcome)); + let logs = captured.output(); + let completions: Vec = logs + .lines() + .map(|line| serde_json::from_str::(line).unwrap()) + .filter(|line| line["message"] == "native operation completed") + .collect(); + assert_eq!(completions.len(), 1); + assert_eq!(completions[0]["operation_kind"], "write"); + assert_eq!(completions[0]["operation_outcome"], outcome); + assert!(!logs.contains("PRIVATE_"), "{logs}"); + assert!(!format!("{spans:?}").contains("PRIVATE_")); + provider.shutdown().unwrap(); + } + } + + #[tokio::test] + async fn malformed_context_and_unknown_routes_never_echo_request_input() { + let exporter = InMemorySpanExporter::default(); + let provider = SdkTracerProvider::builder() + .with_simple_exporter(exporter.clone()) + .build(); + let captured = Capture::default(); + let writer = captured.clone(); + let subscriber = subscriber( + EnvFilter::new("info"), + move || writer.clone(), + true, + Some(provider.tracer("test")), + ); + async { + let response = Router::new() + .layer(middleware::from_fn(http_request)) + .oneshot( + Request::get("/PRIVATE_PATH?PRIVATE_QUERY") + .header("traceparent", "PRIVATE_INVALID_CONTEXT") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::NOT_FOUND); + } + .with_subscriber(subscriber) + .await; + let spans = exporter.get_finished_spans().unwrap(); + assert_eq!(spans.len(), 1); + assert_eq!(spans[0].name, "GET "); + assert_eq!( + spans[0].parent_span_id, + opentelemetry::trace::SpanId::INVALID + ); + assert!(spans[0].span_context.is_valid()); + assert!(!captured.output().contains("PRIVATE")); + provider.shutdown().unwrap(); + } + + #[tokio::test] + async fn bounded_worker_exports_real_otlp_protobuf_and_flushes_before_shutdown() { + let (sent, received) = tokio::sync::oneshot::channel(); + let sender = std::sync::Arc::new(std::sync::Mutex::new(Some(sent))); + let app = Router::new().route( + "/v1/traces", + post(move |headers: HeaderMap, body: Bytes| { + let sender = sender.clone(); + async move { + let _ = sender.lock().unwrap().take().unwrap().send((headers, body)); + StatusCode::OK + } + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + tokio::task::spawn_blocking(move || { + let provider = build_provider(&format!("http://{address}/v1/traces")).unwrap(); + provider + .tracer("otlp-proof") + .in_span("export-proof", |_| {}); + TelemetryGuard(Some(provider)).shutdown(); + }) + .await + .unwrap(); + let (headers, body) = tokio::time::timeout(Duration::from_secs(5), received) + .await + .unwrap() + .unwrap(); + assert_eq!(headers["content-type"], "application/x-protobuf"); + let payload = + opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest::decode( + body, + ) + .unwrap(); + let span = &payload.resource_spans[0].scope_spans[0].spans[0]; + assert_eq!(span.name, "export-proof"); + assert_eq!(span.trace_id.len(), 16); + assert_eq!(span.span_id.len(), 8); + server.abort(); + } +} diff --git a/docs/dev/control-plane.md b/docs/dev/control-plane.md index 72a1959cb..81502ae40 100644 --- a/docs/dev/control-plane.md +++ b/docs/dev/control-plane.md @@ -4,6 +4,16 @@ The cluster control plane turns a local declarative bundle into a durable applie ## Authority model +Server telemetry is separate from deployment authority. The optional OTLP +subscriber exports an explicit span allowlist with W3C request context; it +does not export arbitrary dependency spans or protocol payloads. Owned reads +and writes carry child spans through their existing process-owned futures. +Exporter loss or shutdown does not alter native admission, publication, recovery +or clean-exit evidence. JSON logs use the same restricted dependency-target +filter as text logs. Configuration is in the +[observability guide](../user/operations/observability.md); telemetry and log-filter +unit tests own context propagation, redaction and the actual OTLP transport. + There are three distinct views: 1. **Desired configuration** — `cluster.yaml` and referenced files in an operator workspace. diff --git a/docs/user/index.md b/docs/user/index.md index 2cfe5ad79..d1568b071 100644 --- a/docs/user/index.md +++ b/docs/user/index.md @@ -45,6 +45,7 @@ deployment. - [`cluster.yaml` reference](clusters/config.md) - [Deployment](deployment.md) - [Storage failures and restart](operations/storage-failures.md) +- [Server logs and traces](operations/observability.md) - [Authorization and actors](operations/policy.md) - [Maintenance](operations/maintenance.md) - [Upgrade from 0.13 to 0.14](operations/upgrade-0.14.md): checklist, YAML and CLI replacements diff --git a/docs/user/operations/observability.md b/docs/user/operations/observability.md new file mode 100644 index 000000000..36ad71342 --- /dev/null +++ b/docs/user/operations/observability.md @@ -0,0 +1,54 @@ +# Server logs and traces + +The server writes text logs to stdout by default. Set +`OMNIGRAPH_LOG_FORMAT=json` for structured logs. HTTP completion records include +the route template, method, status and `response_headers_ms`: elapsed time until +response headers are available, not the duration of a streaming body. Unknown +routes use ``; request URLs, query strings, headers and bodies are not +included in these records. + +To export traces, point the server at an OpenTelemetry collector: + +```bash +OMNIGRAPH_LOG_FORMAT=json \ +OTEL_EXPORTER_OTLP_ENDPOINT=http://collector:4318 \ +OTEL_SERVICE_NAME=omnigraph-server \ +OTEL_RESOURCE_ATTRIBUTES=deployment.environment.name=production,service.version=build-123 \ + omnigraph-server --cluster /srv/omnigraph/company-brain --bind 0.0.0.0:8080 +``` + +Export is off unless an endpoint is explicitly set. The supported transport is +OTLP HTTP/protobuf; a general endpoint gets `/v1/traces` appended. +`OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` overrides it with an exact trace URL. +Endpoints must use HTTP or HTTPS without userinfo, query parameters or fragments. +Use HTTPS outside a trusted private network. `OTEL_SDK_DISABLED=true` disables +export without disabling logs. Normal authentication configuration still applies. + +Incoming W3C `traceparent` and `tracestate` connect HTTP requests to their callers. +Owned read/write spans follow the work even when the caller stops waiting; +JSON logs include their `trace_id` and `span_id`. Context is correlation only and +never supplies an actor or changes permissions. Baggage is not imported. +The SDK's `OTEL_TRACES_SAMPLER` and `OTEL_TRACES_SAMPLER_ARG` control sampling; +the default follows the parent, or samples a root request. + +Only explicitly authored request and native-operation spans are exported. +`RUST_LOG` controls stdout log events (`info` by default), independently of +trace export and sampling. Setting it to `warn`, `error` or `off` does not +disable traces; warnings and errors still retain trace correlation when INFO +logs are suppressed. The `rmcp`, `lance::dataset_events` and +`lance::file_audit` targets remain restricted to warnings/errors, even with +debug overrides. These targets can include query arguments/results, full storage +URIs and deletion predicates. `omnigraph_server::init_tracing()` also enforces +the restrictions; embedders using their own subscriber must preserve them. +A failed Blob delivery, change-route request or streamed response is logged by +its error class, never by its message; see [HTTP errors](troubleshooting.md#http-errors). +Collect stdout separately for logs; the server does not send +the same logs over OTLP. Resource attributes are operator-provided metadata; +do not place secrets or graph contents in them. + +Export uses a bounded queue of 2,048 spans, batches of up to 256, a five-second +flush interval and a three-second HTTP timeout. A full queue drops telemetry; +collector failure never blocks a graph operation. After serving stops, a +best-effort flush starts in the background with a five-second bound; it never +delays process exit or changes native shutdown deadlines. Process exit can +lose buffered spans; telemetry is not durable operation evidence. diff --git a/docs/user/operations/server.md b/docs/user/operations/server.md index 3e0a88103..20dee7194 100644 --- a/docs/user/operations/server.md +++ b/docs/user/operations/server.md @@ -44,18 +44,8 @@ claim, not a restart or recovery path; see ### Logs -The server logs to standard output at `info` unless `RUST_LOG` sets other -levels. `omnigraph_server::init_tracing()` limits three dependency targets, and -their `::` children, to warnings and errors even with `RUST_LOG=trace`: `rmcp`, -whose verbose SDK logs contain query arguments and results, and Lance's -`lance::dataset_events` and `lance::file_audit`, which at `info` name every -table's full storage URI, every file Lance creates or deletes, and each delete -predicate with its entity IDs. Other targets keep their configured levels. -Embedders using their own subscriber must enforce the same filter. - -A failed Blob delivery, change-route request or streamed response is logged by -its error class, never by its message; see -[HTTP errors](troubleshooting.md#http-errors). +Use [server logs and traces](observability.md) for JSON logging, optional OTLP +export and correlation with incoming W3C trace context. ## HTTP contract