[test] proxy: Switch proxy to JSON logging

Cargo.toml

@@ -186,6 +186,7 @@ tower-service = "0.3.2"
 tracing = "0.1"
 tracing-error = "0.2"
 tracing-opentelemetry = "0.25"
+tracing-serde = "0.1.3"
 tracing-subscriber = { version = "0.3", default-features = false, features = ["smallvec", "fmt", "tracing-log", "std", "env-filter", "json"] }
 try-lock = "0.2.5"
 twox-hash = { version = "1.6.3", default-features = false }

compute_tools/src/logger.rs

@@ -22,7 +22,10 @@ pub fn init_tracing_and_logging(default_log_level: &str) -> anyhow::Result<()> {
         .with_writer(std::io::stderr);
 
     // Initialize OpenTelemetry
-    let otlp_layer = tracing_utils::init_tracing_without_runtime("compute_ctl");
+    let otlp_layer = tracing_utils::init_tracing_without_runtime(
+        "compute_ctl",
+        None::<fn(opentelemetry::global::Error)>,
+    );
 
     // Put it all together
     tracing_subscriber::registry()

deny.toml

@@ -32,6 +32,7 @@ reason = "the marvin attack only affects private key decryption, not public key
 # https://embarkstudios.github.io/cargo-deny/checks/licenses/cfg.html
 [licenses]
 allow = [
+    "0BSD",
     "Apache-2.0",
     "Artistic-2.0",
     "BSD-2-Clause",

libs/tracing-utils/src/lib.rs

@@ -21,7 +21,7 @@
 //!         .with_writer(std::io::stderr);
 //!
 //!     // Initialize OpenTelemetry. Exports tracing spans as OpenTelemetry traces
-//!     let otlp_layer = tracing_utils::init_tracing("my_application").await;
+//!     let otlp_layer = tracing_utils::init_tracing("my_application", None::<fn(opentelemetry::global::Error)>).await;
 //!
 //!     // Put it all together
 //!     tracing_subscriber::registry()
@@ -70,19 +70,28 @@ use tracing_subscriber::Layer;
 ///
 /// This doesn't block, but is marked as 'async' to hint that this must be called in
 /// asynchronous execution context.
-pub async fn init_tracing<S>(service_name: &str) -> Option<impl Layer<S>>
+pub async fn init_tracing<S>(
+    service_name: &str,
+    error_handler: Option<impl Fn(opentelemetry::global::Error) + Send + Sync + 'static>,
+) -> Option<impl Layer<S>>
 where
     S: Subscriber + for<'span> LookupSpan<'span>,
 {
     if std::env::var("OTEL_SDK_DISABLED") == Ok("true".to_string()) {
         return None;
     };
-    Some(init_tracing_internal(service_name.to_string()))
+    Some(init_tracing_internal(
+        service_name.to_string(),
+        error_handler,
+    ))
 }
 
 /// Like `init_tracing`, but creates a separate tokio Runtime for the tracing
 /// tasks.
-pub fn init_tracing_without_runtime<S>(service_name: &str) -> Option<impl Layer<S>>
+pub fn init_tracing_without_runtime<S>(
+    service_name: &str,
+    error_handler: Option<impl Fn(opentelemetry::global::Error) + Send + Sync + 'static>,
+) -> Option<impl Layer<S>>
 where
     S: Subscriber + for<'span> LookupSpan<'span>,
 {
@@ -113,13 +122,24 @@ where
     ));
     let _guard = runtime.enter();
 
-    Some(init_tracing_internal(service_name.to_string()))
+    Some(init_tracing_internal(
+        service_name.to_string(),
+        error_handler,
+    ))
 }
 
-fn init_tracing_internal<S>(service_name: String) -> impl Layer<S>
+fn init_tracing_internal<S>(
+    service_name: String,
+    error_handler: Option<impl Fn(opentelemetry::global::Error) + Send + Sync + 'static>,
+) -> impl Layer<S>
 where
     S: Subscriber + for<'span> LookupSpan<'span>,
 {
+    if let Some(error_handler) = error_handler {
+        opentelemetry::global::set_error_handler(error_handler)
+            .expect("could not set opentelemetry error handler");
+    }
+
     // Sets up exporter from the OTEL_EXPORTER_* environment variables.
     let exporter = opentelemetry_otlp::new_exporter().http();
 

proxy/Cargo.toml

@@ -42,6 +42,7 @@ hyper0.workspace = true
 hyper = { workspace = true, features = ["server", "http1", "http2"] }
 hyper-util = { version = "0.1", features = ["server", "http1", "http2", "tokio"] }
 http-body-util = { version = "0.1" }
+gettid = "0.1.3"
 indexmap = { workspace = true, features = ["serde"] }
 ipnet.workspace = true
 itertools.workspace = true
@@ -50,6 +51,7 @@ lasso = { workspace = true, features = ["multi-threaded"] }
 measured = { workspace = true, features = ["lasso"] }
 metrics.workspace = true
 once_cell.workspace = true
+opentelemetry.workspace = true
 parking_lot.workspace = true
 parquet.workspace = true
 parquet_derive.workspace = true
@@ -69,7 +71,7 @@ rustls-pemfile.workspace = true
 rustls.workspace = true
 scopeguard.workspace = true
 serde.workspace = true
-serde_json.workspace = true
+serde_json = { workspace = true, features = ["raw_value"] }
 sha2 = { workspace = true, features = ["asm", "oid"] }
 smol_str.workspace = true
 smallvec.workspace = true
@@ -88,6 +90,7 @@ tokio = { workspace = true, features = ["signal"] }
 tracing-subscriber.workspace = true
 tracing-utils.workspace = true
 tracing.workspace = true
+tracing-serde.workspace = true
 try-lock.workspace = true
 typed-json.workspace = true
 url.workspace = true