feat(server): add optional OTLP traces and correlated JSON logs

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

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 }

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

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<W>(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<dyn tracing::Subscriber + Send + Sync> = 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)
 }
 

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,