test(sdk): add regression tests for tokio runtime deadlock fix

Add tests with TokioSpawn*Exporter mocks that call tokio::spawn()
inside export(), simulating tonic/gRPC exporters. These prove that
BlockingStrategy correctly provides tokio runtime context on the
processor's dedicated OS thread, preventing deadlocks on constrained
multi_thread(1) runtimes (open-telemetry#2802, open-telemetry#3356).

Author: bryantbiggs

opentelemetry-sdk/src/logs/batch_log_processor.rs

@@ -1036,6 +1036,66 @@ mod tests {
         processor.shutdown().unwrap();
     }
 
+    // Mock exporter that uses tokio::spawn internally, simulating tonic/gRPC
+    // exporters where the export future depends on tokio tasks. Without
+    // BlockingStrategy, this deadlocks on constrained runtimes because
+    // futures_executor::block_on() cannot drive tokio::spawn-ed tasks.
+    #[derive(Debug, Clone)]
+    struct TokioSpawnLogExporter {
+        exported_count: Arc<AtomicUsize>,
+    }
+
+    impl TokioSpawnLogExporter {
+        fn new() -> Self {
+            Self {
+                exported_count: Arc::new(AtomicUsize::new(0)),
+            }
+        }
+    }
+
+    impl LogExporter for TokioSpawnLogExporter {
+        async fn export(&self, batch: LogBatch<'_>) -> OTelSdkResult {
+            let count = batch.len();
+            // Simulate tonic/gRPC: the export future depends on a tokio::spawn-ed task.
+            let result = tokio::spawn(async move { count }).await.unwrap();
+            assert_eq!(result, count);
+            self.exported_count.fetch_add(count, Ordering::Relaxed);
+            Ok(())
+        }
+
+        fn shutdown(&self) -> OTelSdkResult {
+            Ok(())
+        }
+    }
+
+    // Regression test for deadlock on constrained tokio runtimes (#2802, #3356).
+    // Uses TokioSpawnLogExporter which internally calls tokio::spawn(),
+    // simulating tonic/gRPC exporters where the export future depends on
+    // tokio-spawned tasks. Without BlockingStrategy, this deadlocks because
+    // futures_executor::block_on() cannot drive tokio::spawn-ed tasks
+    // without the runtime context.
+    //
+    // Note: current_thread runtime is not tested here because it has a
+    // fundamental limitation — the single runtime thread is blocked by
+    // force_flush()'s recv(), so no thread is available to drive spawned
+    // tasks. The multi_thread(1) scenario (1-vCPU k8s pods) is the primary
+    // target of this fix.
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn test_batch_log_processor_multi_thread_1_worker_with_tokio_spawn_exporter() {
+        let exporter = TokioSpawnLogExporter::new();
+        let exported_count = exporter.exported_count.clone();
+        let processor = BatchLogProcessor::new(exporter, BatchConfig::default());
+
+        let mut record = SdkLogRecord::new();
+        let instrumentation = InstrumentationScope::default();
+        processor.emit(&mut record, &instrumentation);
+
+        processor.force_flush().unwrap();
+
+        assert_eq!(exported_count.load(Ordering::Relaxed), 1);
+    }
+
     #[tokio::test(flavor = "multi_thread")]
     async fn test_batch_log_processor_shutdown_with_async_runtime_multi_flavor_current_thread() {
         let exporter = InMemoryLogExporterBuilder::default().build();

opentelemetry-sdk/src/metrics/periodic_reader.rs

@@ -606,6 +606,75 @@ mod tests {
         }
     }
 
+    // Mock exporter that uses tokio::spawn internally, simulating tonic/gRPC
+    // exporters where the export future depends on tokio tasks. Without
+    // BlockingStrategy, this deadlocks on constrained runtimes because
+    // futures_executor::block_on() cannot drive tokio::spawn-ed tasks.
+    #[derive(Debug, Clone, Default)]
+    struct TokioSpawnMetricExporter {
+        exported_count: Arc<AtomicUsize>,
+        is_shutdown: Arc<AtomicBool>,
+    }
+
+    impl PushMetricExporter for TokioSpawnMetricExporter {
+        async fn export(&self, _metrics: &ResourceMetrics) -> OTelSdkResult {
+            // Simulate tonic/gRPC: the export future depends on a tokio::spawn-ed task.
+            let result = tokio::spawn(async { 42 }).await.unwrap();
+            assert_eq!(result, 42);
+            self.exported_count.fetch_add(1, Ordering::Relaxed);
+            Ok(())
+        }
+
+        fn force_flush(&self) -> OTelSdkResult {
+            Ok(())
+        }
+
+        fn shutdown(&self) -> OTelSdkResult {
+            self.is_shutdown.store(true, Ordering::Relaxed);
+            Ok(())
+        }
+
+        fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
+            self.shutdown()
+        }
+
+        fn temporality(&self) -> Temporality {
+            Temporality::Cumulative
+        }
+    }
+
+    // Regression test for deadlock on constrained tokio runtimes (#2802, #3356).
+    // Uses TokioSpawnMetricExporter which internally calls tokio::spawn(),
+    // simulating tonic/gRPC exporters where the export future depends on
+    // tokio-spawned tasks. Without BlockingStrategy, this deadlocks because
+    // futures_executor::block_on() cannot drive tokio::spawn-ed tasks
+    // without the runtime context.
+    //
+    // Note: current_thread runtime is not tested here because it has a
+    // fundamental limitation — the single runtime thread is blocked by
+    // force_flush()'s recv(), so no thread is available to drive spawned
+    // tasks. The multi_thread(1) scenario (1-vCPU k8s pods) is the primary
+    // target of this fix.
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn test_periodic_reader_multi_thread_1_worker_with_tokio_spawn_exporter() {
+        let exporter = TokioSpawnMetricExporter::default();
+        let exported_count = exporter.exported_count.clone();
+
+        let reader = PeriodicReader::builder(exporter)
+            .with_interval(Duration::from_secs(120))
+            .build();
+
+        let meter_provider = SdkMeterProvider::builder().with_reader(reader).build();
+        let meter = meter_provider.meter("test");
+        let counter = meter.u64_counter("test.counter").build();
+        counter.add(1, &[]);
+
+        meter_provider.force_flush().unwrap();
+
+        assert!(exported_count.load(Ordering::Relaxed) >= 1);
+    }
+
     #[test]
     fn collection_triggered_by_interval_multiple() {
         // Arrange

opentelemetry-sdk/src/trace/span_processor.rs

@@ -988,6 +988,7 @@ mod tests {
         OTEL_BSP_SCHEDULE_DELAY, OTEL_BSP_SCHEDULE_DELAY_DEFAULT,
     };
     use crate::error::OTelSdkResult;
+    use crate::util::BlockingStrategy;
     use crate::testing::trace::new_test_export_span_data;
     use crate::trace::span_processor::{
         OTEL_BSP_EXPORT_TIMEOUT_DEFAULT, OTEL_BSP_MAX_CONCURRENT_EXPORTS,
@@ -1302,13 +1303,15 @@ mod tests {
         sender.send(create_test_span("counted")).unwrap();
         sender.send(create_test_span("unaccounted")).unwrap();
 
+        let blocking_strategy = BlockingStrategy::new();
         let result = BatchSpanProcessor::get_spans_and_export(
             &receiver,
             &exporter,
             &mut spans,
             &mut last_export_time,
             &current_batch_size,
             &config,
+            &blocking_strategy,
         );
 
         assert!(result.is_ok(), "export should succeed");
@@ -1546,6 +1549,39 @@ mod tests {
         );
     }
 
+    // Mock exporter that uses tokio::spawn internally, simulating tonic/gRPC
+    // exporters where the export future depends on tokio tasks. Without
+    // BlockingStrategy, this deadlocks on constrained runtimes because
+    // futures_executor::block_on() cannot drive tokio::spawn-ed tasks.
+    #[derive(Debug, Clone)]
+    struct TokioSpawnSpanExporter {
+        exported_spans: Arc<Mutex<Vec<SpanData>>>,
+    }
+
+    impl TokioSpawnSpanExporter {
+        fn new() -> Self {
+            Self {
+                exported_spans: Arc::new(Mutex::new(Vec::new())),
+            }
+        }
+    }
+
+    impl SpanExporter for TokioSpawnSpanExporter {
+        async fn export(&self, batch: Vec<SpanData>) -> OTelSdkResult {
+            // Simulate tonic/gRPC: the export future depends on a tokio::spawn-ed task.
+            // Without tokio runtime context, this panics or deadlocks.
+            let count = batch.len();
+            let result = tokio::spawn(async move { count }).await.unwrap();
+            assert_eq!(result, batch.len());
+            self.exported_spans.lock().unwrap().extend(batch);
+            Ok(())
+        }
+
+        fn shutdown(&self) -> OTelSdkResult {
+            Ok(())
+        }
+    }
+
     #[tokio::test(flavor = "current_thread")]
     async fn test_batch_processor_current_thread_runtime() {
         let exporter = MockSpanExporter::new();
@@ -1592,6 +1628,42 @@ mod tests {
         assert_eq!(exported_spans.len(), 4);
     }
 
+    // Regression test for deadlock on constrained tokio runtimes (#2802, #3356).
+    // Uses TokioSpawnSpanExporter which internally calls tokio::spawn(),
+    // simulating tonic/gRPC exporters where the export future depends on
+    // tokio-spawned tasks. Without BlockingStrategy, this deadlocks because
+    // futures_executor::block_on() cannot drive tokio::spawn-ed tasks
+    // without the runtime context.
+    //
+    // Note: current_thread runtime is not tested here because it has a
+    // fundamental limitation — the single runtime thread is blocked by
+    // force_flush()'s recv(), so no thread is available to drive spawned
+    // tasks. The multi_thread(1) scenario (1-vCPU k8s pods) is the primary
+    // target of this fix.
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn test_batch_processor_multi_thread_1_worker_with_tokio_spawn_exporter() {
+        let exporter = TokioSpawnSpanExporter::new();
+        let exporter_shared = exporter.exported_spans.clone();
+
+        let config = BatchConfigBuilder::default()
+            .with_max_queue_size(5)
+            .with_max_export_batch_size(3)
+            .build();
+
+        let processor = BatchSpanProcessor::new(exporter, config);
+
+        for _ in 0..4 {
+            let span = new_test_export_span_data();
+            processor.on_end(span);
+        }
+
+        processor.force_flush().unwrap();
+
+        let exported_spans = exporter_shared.lock().unwrap();
+        assert_eq!(exported_spans.len(), 4);
+    }
+
     #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
     async fn test_batch_processor_multi_thread() {
         let exporter = MockSpanExporter::new();