diff --git a/crates/goose/src/otel/otlp.rs b/crates/goose/src/otel/otlp.rs index ae1c59390e..829c8407f2 100644 --- a/crates/goose/src/otel/otlp.rs +++ b/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>> = Mutex::new(None); + +fn get_or_create_otel_rt() -> OtlpResult> { + 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, +} + +impl opentelemetry_sdk::trace::SpanExporter for TokioSpanExporter { + async fn export( + &self, + batch: Vec, + ) -> 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, +} + +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, +} + +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 { 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 { 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 { 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")]