fix(otel): provide Tokio runtime for OTLP HTTP exporter batch thread (#9541)

Signed-off-by: Varun Nuthalapati <nuthalapativarun@gmail.com>
Signed-off-by: Douwe M Osinga <douwe@sidewalklabs.com>
Co-authored-by: Douwe M Osinga <douwe@sidewalklabs.com>

crates/goose/src/otel/otlp.rs

@@ -8,7 +8,7 @@ use opentelemetry_sdk::resource::{EnvResourceDetector, TelemetryResourceDetector
 use opentelemetry_sdk::trace::SdkTracerProvider;
 use opentelemetry_sdk::Resource;
 use std::env;
-use std::sync::Mutex;
+use std::sync::{Arc, Mutex};
 use tracing::{Level, Metadata};
 use tracing_opentelemetry::{MetricsLayer, OpenTelemetryLayer};
 use tracing_subscriber::filter::FilterFn;
@@ -45,6 +45,133 @@ fn warn_grpc_protocol_skipped_once() {
     });
 }
 
+/// Dedicated single-thread Tokio runtime for OTLP HTTP export.
+///
+/// `BatchSpanProcessor`, `BatchLogProcessor`, and `PeriodicReader` all
+/// spawn raw `std::thread`s with no Tokio reactor. The async reqwest HTTP
+/// client calls `tokio::time::sleep` during export, which panics without a
+/// reactor. We drive every OTLP export call through this runtime so the
+/// reqwest futures always have a live Tokio handle.
+static OTEL_RT: Mutex<Option<Arc<tokio::runtime::Runtime>>> = Mutex::new(None);
+
+fn get_or_create_otel_rt() -> OtlpResult<Arc<tokio::runtime::Runtime>> {
+    let mut guard = OTEL_RT.lock().unwrap_or_else(|e| e.into_inner());
+    if let Some(rt) = guard.as_ref() {
+        return Ok(Arc::clone(rt));
+    }
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()?;
+    let rt = Arc::new(rt);
+    *guard = Some(Arc::clone(&rt));
+    Ok(rt)
+}
+
+/// Wraps an OTLP `SpanExporter` so that each `export` call is driven inside
+/// a dedicated Tokio runtime. Required because `BatchSpanProcessor` runs in
+/// a raw `std::thread` with no Tokio reactor.
+#[derive(Debug)]
+struct TokioSpanExporter {
+    inner: opentelemetry_otlp::SpanExporter,
+    rt: Arc<tokio::runtime::Runtime>,
+}
+
+impl opentelemetry_sdk::trace::SpanExporter for TokioSpanExporter {
+    async fn export(
+        &self,
+        batch: Vec<opentelemetry_sdk::trace::SpanData>,
+    ) -> opentelemetry_sdk::error::OTelSdkResult {
+        if tokio::runtime::Handle::try_current().is_ok() {
+            self.inner.export(batch).await
+        } else {
+            self.rt.block_on(self.inner.export(batch))
+        }
+    }
+
+    fn shutdown_with_timeout(
+        &self,
+        timeout: std::time::Duration,
+    ) -> opentelemetry_sdk::error::OTelSdkResult {
+        self.inner.shutdown_with_timeout(timeout)
+    }
+
+    fn force_flush(&self) -> opentelemetry_sdk::error::OTelSdkResult {
+        self.inner.force_flush()
+    }
+
+    fn set_resource(&mut self, resource: &Resource) {
+        self.inner.set_resource(resource);
+    }
+}
+
+/// Wraps an OTLP `LogExporter` so that each export runs inside a dedicated
+/// Tokio runtime for the same reason as `TokioSpanExporter`.
+#[derive(Debug)]
+struct TokioLogExporter {
+    inner: opentelemetry_otlp::LogExporter,
+    rt: Arc<tokio::runtime::Runtime>,
+}
+
+impl opentelemetry_sdk::logs::LogExporter for TokioLogExporter {
+    async fn export(
+        &self,
+        batch: opentelemetry_sdk::logs::LogBatch<'_>,
+    ) -> opentelemetry_sdk::error::OTelSdkResult {
+        if tokio::runtime::Handle::try_current().is_ok() {
+            self.inner.export(batch).await
+        } else {
+            self.rt.block_on(self.inner.export(batch))
+        }
+    }
+
+    fn shutdown_with_timeout(
+        &self,
+        timeout: std::time::Duration,
+    ) -> opentelemetry_sdk::error::OTelSdkResult {
+        self.inner.shutdown_with_timeout(timeout)
+    }
+
+    fn set_resource(&mut self, resource: &Resource) {
+        self.inner.set_resource(resource);
+    }
+}
+
+/// Wraps an OTLP `MetricExporter` so that periodic export runs inside a
+/// dedicated Tokio runtime for the same reason as `TokioSpanExporter`.
+#[derive(Debug)]
+struct TokioMetricExporter {
+    inner: opentelemetry_otlp::MetricExporter,
+    rt: Arc<tokio::runtime::Runtime>,
+}
+
+impl opentelemetry_sdk::metrics::exporter::PushMetricExporter for TokioMetricExporter {
+    async fn export(
+        &self,
+        metrics: &opentelemetry_sdk::metrics::data::ResourceMetrics,
+    ) -> opentelemetry_sdk::error::OTelSdkResult {
+        if tokio::runtime::Handle::try_current().is_ok() {
+            self.inner.export(metrics).await
+        } else {
+            self.rt.block_on(self.inner.export(metrics))
+        }
+    }
+
+    fn force_flush(&self) -> opentelemetry_sdk::error::OTelSdkResult {
+        self.inner.force_flush()
+    }
+
+    fn shutdown_with_timeout(
+        &self,
+        timeout: std::time::Duration,
+    ) -> opentelemetry_sdk::error::OTelSdkResult {
+        self.inner.shutdown_with_timeout(timeout)
+    }
+
+    fn temporality(&self) -> Temporality {
+        self.inner.temporality()
+    }
+}
+
 #[derive(Debug, Clone, PartialEq)]
 pub enum ExporterType {
     Otlp,
@@ -206,9 +333,13 @@ fn create_otlp_tracing_layer() -> OtlpResult<OtlpTracingLayer> {
                 warn_grpc_protocol_skipped_once();
                 return Err("OTLP traces protocol is grpc but goose was built without grpc-tonic; skipping traces exporter".into());
             }
-            let exporter = opentelemetry_otlp::SpanExporter::builder()
-                .with_http()
-                .build()?;
+            let rt = get_or_create_otel_rt()?;
+            let exporter = TokioSpanExporter {
+                inner: opentelemetry_otlp::SpanExporter::builder()
+                    .with_http()
+                    .build()?,
+                rt,
+            };
             SdkTracerProvider::builder()
                 .with_batch_exporter(exporter)
                 .with_resource(resource)
@@ -254,10 +385,14 @@ fn create_otlp_metrics_layer() -> OtlpResult<OtlpMetricsLayer> {
                 warn_grpc_protocol_skipped_once();
                 return Err("OTLP metrics protocol is grpc but goose was built without grpc-tonic; skipping metrics exporter".into());
             }
-            let exporter = opentelemetry_otlp::MetricExporter::builder()
-                .with_http()
-                .with_temporality(temporality_preference())
-                .build()?;
+            let rt = get_or_create_otel_rt()?;
+            let exporter = TokioMetricExporter {
+                inner: opentelemetry_otlp::MetricExporter::builder()
+                    .with_http()
+                    .with_temporality(temporality_preference())
+                    .build()?,
+                rt,
+            };
             SdkMeterProvider::builder()
                 .with_resource(resource)
                 .with_periodic_exporter(exporter)
@@ -289,9 +424,13 @@ fn create_otlp_logs_layer() -> OtlpResult<OtlpLogsLayer> {
                 warn_grpc_protocol_skipped_once();
                 return Err("OTLP logs protocol is grpc but goose was built without grpc-tonic; skipping logs exporter".into());
             }
-            let exporter = opentelemetry_otlp::LogExporter::builder()
-                .with_http()
-                .build()?;
+            let rt = get_or_create_otel_rt()?;
+            let exporter = TokioLogExporter {
+                inner: opentelemetry_otlp::LogExporter::builder()
+                    .with_http()
+                    .build()?,
+                rt,
+            };
             SdkLoggerProvider::builder()
                 .with_batch_exporter(exporter)
                 .with_resource(resource)
@@ -464,11 +603,26 @@ pub fn shutdown_otlp() {
             tracing::warn!("OTLP logger provider shutdown error: {e}");
         }
     }
+
+    if let Some(rt) = OTEL_RT.lock().unwrap_or_else(|e| e.into_inner()).take() {
+        // Dropping a `Runtime` inside an async context panics with "Cannot drop
+        // a runtime in a context where blocking is not allowed". Move the drop
+        // off-runtime via `shutdown_background` (which takes ownership of
+        // `Runtime` and shuts down without blocking the caller).  If we still
+        // have other `Arc` clones alive, fall back to dropping on a plain thread.
+        match Arc::try_unwrap(rt) {
+            Ok(runtime) => runtime.shutdown_background(),
+            Err(arc) => {
+                std::thread::spawn(move || drop(arc));
+            }
+        }
+    }
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::session_context::{session_host, session_user};
     use goose_test_support::otel::clear_otel_env;
     use opentelemetry_sdk::metrics::Temporality;
     use test_case::test_case;
@@ -682,6 +836,66 @@ mod tests {
         assert!(get("user.name").is_some());
     }
 
+    /// Verify that OTLP layers initialize without panicking even when no
+    /// Tokio runtime is active on the calling thread. This is the scenario
+    /// that triggered the "no reactor running" panic: `BatchSpanProcessor`
+    /// spawns a raw `std::thread`, and the async reqwest client inside the
+    /// exporter calls `tokio::time::sleep`, which requires a reactor.
+    #[test]
+    fn test_otlp_layers_ok_without_tokio_context() {
+        let _env = clear_otel_env(&[
+            ("OTEL_TRACES_EXPORTER", "otlp"),
+            ("OTEL_METRICS_EXPORTER", "otlp"),
+            ("OTEL_LOGS_EXPORTER", "otlp"),
+            ("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4318"),
+        ]);
+        assert!(create_otlp_tracing_layer().is_ok());
+        assert!(create_otlp_metrics_layer().is_ok());
+        assert!(create_otlp_logs_layer().is_ok());
+        shutdown_otlp();
+    }
+
+    #[test_case(
+        &[],
+        Resource::builder_empty()
+            .with_attributes([KeyValue::new("service.name", "goose"), KeyValue::new("service.version", env!("CARGO_PKG_VERSION")), KeyValue::new("service.namespace", "goose"), KeyValue::new("host.name", session_host()), KeyValue::new("user.name", session_user())])
+            .with_detector(Box::new(TelemetryResourceDetector))
+            .build();
+        "no env vars uses goose defaults"
+    )]
+    #[test_case(
+        &[("OTEL_SERVICE_NAME", "custom")],
+        Resource::builder_empty()
+            .with_attributes([KeyValue::new("service.name", "goose"), KeyValue::new("service.version", env!("CARGO_PKG_VERSION")), KeyValue::new("service.namespace", "goose"), KeyValue::new("host.name", session_host()), KeyValue::new("user.name", session_user())])
+            .with_detector(Box::new(TelemetryResourceDetector))
+            .with_service_name("custom")
+            .build();
+        "OTEL_SERVICE_NAME overrides service.name"
+    )]
+    #[test_case(
+        &[("OTEL_RESOURCE_ATTRIBUTES", "deployment.environment=prod")],
+        Resource::builder_empty()
+            .with_attributes([KeyValue::new("service.name", "goose"), KeyValue::new("service.version", env!("CARGO_PKG_VERSION")), KeyValue::new("service.namespace", "goose"), KeyValue::new("host.name", session_host()), KeyValue::new("user.name", session_user())])
+            .with_detector(Box::new(TelemetryResourceDetector))
+            .with_attribute(KeyValue::new("deployment.environment", "prod"))
+            .build();
+        "OTEL_RESOURCE_ATTRIBUTES adds custom attributes"
+    )]
+    #[test_case(
+        &[("OTEL_SERVICE_NAME", "custom"), ("OTEL_RESOURCE_ATTRIBUTES", "deployment.environment=prod")],
+        Resource::builder_empty()
+            .with_attributes([KeyValue::new("service.name", "goose"), KeyValue::new("service.version", env!("CARGO_PKG_VERSION")), KeyValue::new("service.namespace", "goose"), KeyValue::new("host.name", session_host()), KeyValue::new("user.name", session_user())])
+            .with_detector(Box::new(TelemetryResourceDetector))
+            .with_service_name("custom")
+            .with_attribute(KeyValue::new("deployment.environment", "prod"))
+            .build();
+        "OTEL_SERVICE_NAME and OTEL_RESOURCE_ATTRIBUTES combine"
+    )]
+    fn test_create_resource(env: &[(&'static str, &'static str)], expected: Resource) {
+        let _guard = clear_otel_env(env);
+        assert_eq!(create_resource(), expected);
+    }
+
     #[test_case(&[("RUST_LOG", "")], Level::INFO; "default is info")]
     #[test_case(&[("RUST_LOG", "debug")], Level::DEBUG; "RUST_LOG takes precedence")]
     #[test_case(&[("RUST_LOG", ""), ("OTEL_LOG_LEVEL", "error")], Level::ERROR; "OTEL_LOG_LEVEL fallback")]