fix(sdk): resolve exporter deadlock on constrained tokio runtimes

The default thread-based processors (BatchSpanProcessor, BatchLogProcessor,
PeriodicReader) call futures_executor::block_on() on their dedicated worker
threads. When the exporter uses tonic/gRPC, the export future depends on
tokio tasks (e.g. tonic's Buffer worker) that can only be polled by tokio
worker threads. If all tokio worker threads are blocked (single-threaded
runtime, or multi-thread with 1 worker), this creates a circular wait.

Add BlockingStrategy that captures the tokio runtime handle at construction
time and enters the runtime context via Handle::enter() before calling
futures_executor::block_on(). This makes tokio types available on the
dedicated background threads without taking ownership of the reactor.
Falls back to plain futures_executor::block_on() without tokio.

Fixes: open-telemetry#2802

Author: bryantbiggs

opentelemetry-sdk/src/logs/batch_log_processor.rs

@@ -17,6 +17,7 @@
 
 use crate::error::{OTelSdkError, OTelSdkResult};
 use crate::logs::log_processor::LogProcessor;
+use crate::util::BlockingStrategy;
 use crate::{
     logs::{LogBatch, LogExporter, SdkLogRecord},
     Resource,
@@ -342,6 +343,7 @@ impl BatchLogProcessor {
         let max_export_batch_size = config.max_export_batch_size;
         let current_batch_size = Arc::new(AtomicUsize::new(0));
         let current_batch_size_for_thread = current_batch_size.clone();
+        let blocking_strategy = BlockingStrategy::new();
 
         let handle = thread::Builder::new()
             .name("OpenTelemetry.Logs.BatchProcessor".to_string())
@@ -368,6 +370,7 @@ impl BatchLogProcessor {
                     last_export_time: &mut Instant,
                     current_batch_size: &AtomicUsize,
                     max_export_size: usize,
+                    blocking_strategy: &BlockingStrategy,
                 ) -> OTelSdkResult
                 where
                     E: LogExporter + Send + Sync + 'static,
@@ -393,13 +396,15 @@ impl BatchLogProcessor {
                         }
                         total_exported_logs += count_of_logs;
 
-                        result = export_batch_sync(exporter, logs, last_export_time); // This method clears the logs vec after exporting
+                        result =
+                            export_batch_sync(exporter, logs, last_export_time, blocking_strategy); // This method clears the logs vec after exporting
 
                         current_batch_size.fetch_sub(count_of_logs, Ordering::AcqRel);
                     }
                     result
                 }
 
+                let blocking_strategy = blocking_strategy;
                 loop {
                     let remaining_time = config
                         .scheduled_delay
@@ -422,6 +427,7 @@ impl BatchLogProcessor {
                                 &mut last_export_time,
                                 &current_batch_size,
                                 max_export_batch_size,
+                                &blocking_strategy,
                             );
                         }
                         Ok(BatchMessage::ForceFlush(sender)) => {
@@ -433,6 +439,7 @@ impl BatchLogProcessor {
                                 &mut last_export_time,
                                 &current_batch_size,
                                 max_export_batch_size,
+                                &blocking_strategy,
                             );
                             let _ = sender.send(result);
                         }
@@ -445,6 +452,7 @@ impl BatchLogProcessor {
                                 &mut last_export_time,
                                 &current_batch_size,
                                 max_export_batch_size,
+                                &blocking_strategy,
                             );
                             let _ = exporter.shutdown();
                             let _ = sender.send(result);
@@ -473,6 +481,7 @@ impl BatchLogProcessor {
                                 &mut last_export_time,
                                 &current_batch_size,
                                 max_export_batch_size,
+                                &blocking_strategy,
                             );
                         }
                         Err(RecvTimeoutError::Disconnected) => {
@@ -523,6 +532,7 @@ fn export_batch_sync<E>(
     exporter: &E,
     batch: &mut Vec<Box<(SdkLogRecord, InstrumentationScope)>>,
     last_export_time: &mut Instant,
+    blocking_strategy: &BlockingStrategy,
 ) -> OTelSdkResult
 where
     E: LogExporter + ?Sized,
@@ -534,7 +544,7 @@ where
     }
 
     let export = exporter.export(LogBatch::new_with_owned_data(batch.as_slice()));
-    let export_result = futures_executor::block_on(export);
+    let export_result = blocking_strategy.block_on(export);
 
     // Clear the batch vec after exporting
     batch.clear();

opentelemetry-sdk/src/metrics/periodic_reader.rs

@@ -13,6 +13,7 @@ use opentelemetry::{otel_debug, otel_error, otel_info, otel_warn, Context};
 use crate::{
     error::{OTelSdkError, OTelSdkResult},
     metrics::{exporter::PushMetricExporter, reader::SdkProducer},
+    util::BlockingStrategy,
     Resource,
 };
 
@@ -152,6 +153,7 @@ impl<E: PushMetricExporter> PeriodicReader<E> {
                 message_sender,
                 producer: Mutex::new(None),
                 exporter: exporter_arc.clone(),
+                blocking_strategy: BlockingStrategy::new(),
             }),
         };
         let cloned_reader = reader.clone();
@@ -351,6 +353,7 @@ struct PeriodicReaderInner<E: PushMetricExporter> {
     exporter: Arc<E>,
     message_sender: mpsc::Sender<Message>,
     producer: Mutex<Option<Weak<dyn SdkProducer>>>,
+    blocking_strategy: BlockingStrategy,
 }
 
 impl<E: PushMetricExporter> PeriodicReaderInner<E> {
@@ -407,9 +410,8 @@ impl<E: PushMetricExporter> PeriodicReaderInner<E> {
         });
         otel_debug!(name: "PeriodicReaderMetricsCollected", count = metrics_count, time_taken_in_millis = time_taken_for_collect.as_millis());
 
-        // Relying on futures executor to execute async call.
         // TODO: Pass timeout to exporter
-        futures_executor::block_on(self.exporter.export(rm))
+        self.blocking_strategy.block_on(self.exporter.export(rm))
     }
 
     fn force_flush(&self) -> OTelSdkResult {

opentelemetry-sdk/src/trace/span_processor.rs

@@ -38,6 +38,7 @@ use crate::error::{OTelSdkError, OTelSdkResult};
 use crate::resource::Resource;
 use crate::trace::Span;
 use crate::trace::{SpanData, SpanExporter};
+use crate::util::BlockingStrategy;
 use opentelemetry::Context;
 use opentelemetry::{otel_debug, otel_error, otel_warn};
 use std::cmp::min;
@@ -345,6 +346,7 @@ impl BatchSpanProcessor {
         let max_export_batch_size = config.max_export_batch_size;
         let current_batch_size = Arc::new(AtomicUsize::new(0));
         let current_batch_size_for_thread = current_batch_size.clone();
+        let blocking_strategy = BlockingStrategy::new();
 
         let handle = thread::Builder::new()
             .name("OpenTelemetry.Traces.BatchProcessor".to_string())
@@ -359,6 +361,7 @@ impl BatchSpanProcessor {
                 let mut spans = Vec::with_capacity(config.max_export_batch_size);
                 let mut last_export_time = Instant::now();
                 let current_batch_size = current_batch_size_for_thread;
+                let blocking_strategy = blocking_strategy;
                 loop {
                     let remaining_time_option = config
                         .scheduled_delay
@@ -382,6 +385,7 @@ impl BatchSpanProcessor {
                                     &mut last_export_time,
                                     &current_batch_size,
                                     &config,
+                                    &blocking_strategy,
                                 );
                             }
                             BatchMessage::ForceFlush(sender) => {
@@ -393,6 +397,7 @@ impl BatchSpanProcessor {
                                     &mut last_export_time,
                                     &current_batch_size,
                                     &config,
+                                    &blocking_strategy,
                                 );
                                 let _ = sender.send(result);
                             }
@@ -405,6 +410,7 @@ impl BatchSpanProcessor {
                                     &mut last_export_time,
                                     &current_batch_size,
                                     &config,
+                                    &blocking_strategy,
                                 );
                                 let _ = exporter.shutdown();
                                 let _ = sender.send(result);
@@ -434,6 +440,7 @@ impl BatchSpanProcessor {
                                 &mut last_export_time,
                                 &current_batch_size,
                                 &config,
+                                &blocking_strategy,
                             );
                         }
                         Err(RecvTimeoutError::Disconnected) => {
@@ -488,6 +495,7 @@ impl BatchSpanProcessor {
         last_export_time: &mut Instant,
         current_batch_size: &AtomicUsize,
         config: &BatchConfig,
+        blocking_strategy: &BlockingStrategy,
     ) -> OTelSdkResult
     where
         E: SpanExporter + Send + Sync + 'static,
@@ -515,7 +523,7 @@ impl BatchSpanProcessor {
             }
             total_exported_spans += count_of_spans;
 
-            result = Self::export_batch_sync(exporter, spans, last_export_time); // This method clears the spans vec after exporting
+            result = Self::export_batch_sync(exporter, spans, last_export_time, blocking_strategy); // This method clears the spans vec after exporting
 
             current_batch_size.fetch_sub(count_of_spans, Ordering::AcqRel);
         }
@@ -527,6 +535,7 @@ impl BatchSpanProcessor {
         exporter: &E,
         batch: &mut Vec<SpanData>,
         last_export_time: &mut Instant,
+        blocking_strategy: &BlockingStrategy,
     ) -> OTelSdkResult
     where
         E: SpanExporter + ?Sized,
@@ -544,7 +553,7 @@ impl BatchSpanProcessor {
         // every export. See if this can be optimized by
         // *not* requiring ownership in the exporter.
         let export = exporter.export(batch.split_off(0));
-        let export_result = futures_executor::block_on(export);
+        let export_result = blocking_strategy.block_on(export);
 
         match export_result {
             Ok(_) => OTelSdkResult::Ok(()),

opentelemetry-sdk/src/util.rs

@@ -7,3 +7,44 @@ pub fn tokio_interval_stream(
 ) -> tokio_stream::wrappers::IntervalStream {
     tokio_stream::wrappers::IntervalStream::new(tokio::time::interval(period))
 }
+
+/// Strategy for blocking on async futures from synchronous contexts.
+///
+/// When constructed within a tokio runtime, captures the runtime handle
+/// and enters the runtime context via [`tokio::runtime::Handle::enter()`]
+/// before blocking with [`futures_executor::block_on()`]. This makes tokio
+/// types (spawn, timers, IO resources) available on dedicated background
+/// threads without taking ownership of the reactor — IO continues to be
+/// driven by the runtime's own threads.
+///
+/// Falls back to plain [`futures_executor::block_on()`] when no tokio runtime
+/// is available (e.g., non-tokio environments).
+#[cfg(any(feature = "trace", feature = "logs", feature = "metrics"))]
+#[derive(Clone, Debug)]
+pub(crate) enum BlockingStrategy {
+    #[cfg(feature = "rt-tokio")]
+    TokioHandle(tokio::runtime::Handle),
+    FuturesExecutor,
+}
+
+#[cfg(any(feature = "trace", feature = "logs", feature = "metrics"))]
+impl BlockingStrategy {
+    pub(crate) fn new() -> Self {
+        #[cfg(feature = "rt-tokio")]
+        if let Ok(handle) = tokio::runtime::Handle::try_current() {
+            return Self::TokioHandle(handle);
+        }
+        Self::FuturesExecutor
+    }
+
+    pub(crate) fn block_on<F: std::future::Future>(&self, future: F) -> F::Output {
+        match self {
+            #[cfg(feature = "rt-tokio")]
+            Self::TokioHandle(handle) => {
+                let _guard = handle.enter();
+                futures_executor::block_on(future)
+            }
+            Self::FuturesExecutor => futures_executor::block_on(future),
+        }
+    }
+}