Add Unit Tests for Sync and Async Log Exporters - with and without runtime - for SimpleLogProcessor (#2218)

Author: lalitb

opentelemetry-sdk/src/logs/log_processor.rs

@@ -553,6 +553,7 @@ mod tests {
     use opentelemetry::InstrumentationLibrary;
     use opentelemetry::Key;
     use opentelemetry::{logs::LogResult, KeyValue};
+    use std::sync::atomic::{AtomicUsize, Ordering};
     use std::sync::{Arc, Mutex};
     use std::time::Duration;
 
@@ -831,9 +832,7 @@ mod tests {
     #[tokio::test(flavor = "current_thread")]
     #[ignore = "See issue https://github.com/open-telemetry/opentelemetry-rust/issues/1968"]
     async fn test_batch_log_processor_shutdown_with_async_runtime_current_flavor_multi_thread() {
-        let exporter = InMemoryLogsExporterBuilder::default()
-            .keep_records_on_shutdown()
-            .build();
+        let exporter = InMemoryLogsExporterBuilder::default().build();
         let processor = BatchLogProcessor::new(
             Box::new(exporter.clone()),
             BatchConfig::default(),
@@ -848,9 +847,7 @@ mod tests {
 
     #[tokio::test(flavor = "current_thread")]
     async fn test_batch_log_processor_shutdown_with_async_runtime_current_flavor_current_thread() {
-        let exporter = InMemoryLogsExporterBuilder::default()
-            .keep_records_on_shutdown()
-            .build();
+        let exporter = InMemoryLogsExporterBuilder::default().build();
         let processor = BatchLogProcessor::new(
             Box::new(exporter.clone()),
             BatchConfig::default(),
@@ -862,9 +859,7 @@ mod tests {
 
     #[tokio::test(flavor = "multi_thread")]
     async fn test_batch_log_processor_shutdown_with_async_runtime_multi_flavor_multi_thread() {
-        let exporter = InMemoryLogsExporterBuilder::default()
-            .keep_records_on_shutdown()
-            .build();
+        let exporter = InMemoryLogsExporterBuilder::default().build();
         let processor = BatchLogProcessor::new(
             Box::new(exporter.clone()),
             BatchConfig::default(),
@@ -876,9 +871,7 @@ mod tests {
 
     #[tokio::test(flavor = "multi_thread")]
     async fn test_batch_log_processor_shutdown_with_async_runtime_multi_flavor_current_thread() {
-        let exporter = InMemoryLogsExporterBuilder::default()
-            .keep_records_on_shutdown()
-            .build();
+        let exporter = InMemoryLogsExporterBuilder::default().build();
         let processor = BatchLogProcessor::new(
             Box::new(exporter.clone()),
             BatchConfig::default(),
@@ -994,4 +987,232 @@ mod tests {
                 == AnyValue::String("Updated by FirstProcessor".into())
         );
     }
+
+    #[test]
+    fn test_simple_processor_sync_exporter_without_runtime() {
+        let exporter = InMemoryLogsExporterBuilder::default().build();
+        let processor = SimpleLogProcessor::new(Box::new(exporter.clone()));
+
+        let mut record: LogRecord = Default::default();
+        let instrumentation: InstrumentationLibrary = Default::default();
+
+        processor.emit(&mut record, &instrumentation);
+
+        assert_eq!(exporter.get_emitted_logs().unwrap().len(), 1);
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn test_simple_processor_sync_exporter_with_runtime() {
+        let exporter = InMemoryLogsExporterBuilder::default().build();
+        let processor = SimpleLogProcessor::new(Box::new(exporter.clone()));
+
+        let mut record: LogRecord = Default::default();
+        let instrumentation: InstrumentationLibrary = Default::default();
+
+        processor.emit(&mut record, &instrumentation);
+
+        assert_eq!(exporter.get_emitted_logs().unwrap().len(), 1);
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_simple_processor_sync_exporter_with_multi_thread_runtime() {
+        let exporter = InMemoryLogsExporterBuilder::default().build();
+        let processor = Arc::new(SimpleLogProcessor::new(Box::new(exporter.clone())));
+
+        let mut handles = vec![];
+        for _ in 0..10 {
+            let processor_clone = Arc::clone(&processor);
+            let handle = tokio::spawn(async move {
+                let mut record: LogRecord = Default::default();
+                let instrumentation: InstrumentationLibrary = Default::default();
+                processor_clone.emit(&mut record, &instrumentation);
+            });
+            handles.push(handle);
+        }
+
+        for handle in handles {
+            handle.await.unwrap();
+        }
+
+        assert_eq!(exporter.get_emitted_logs().unwrap().len(), 10);
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn test_simple_processor_sync_exporter_with_current_thread_runtime() {
+        let exporter = InMemoryLogsExporterBuilder::default().build();
+        let processor = SimpleLogProcessor::new(Box::new(exporter.clone()));
+
+        let mut record: LogRecord = Default::default();
+        let instrumentation: InstrumentationLibrary = Default::default();
+
+        processor.emit(&mut record, &instrumentation);
+
+        assert_eq!(exporter.get_emitted_logs().unwrap().len(), 1);
+    }
+
+    #[derive(Debug, Clone)]
+    struct LogExporterThatRequiresTokio {
+        export_count: Arc<AtomicUsize>,
+    }
+
+    impl LogExporterThatRequiresTokio {
+        /// Creates a new instance of `LogExporterThatRequiresTokio`.
+        fn new() -> Self {
+            LogExporterThatRequiresTokio {
+                export_count: Arc::new(AtomicUsize::new(0)),
+            }
+        }
+
+        /// Returns the number of logs stored in the exporter.
+        fn len(&self) -> usize {
+            self.export_count.load(Ordering::Acquire)
+        }
+    }
+
+    #[async_trait::async_trait]
+    impl LogExporter for LogExporterThatRequiresTokio {
+        async fn export(&mut self, batch: LogBatch<'_>) -> LogResult<()> {
+            // Simulate minimal dependency on tokio by sleeping asynchronously for a short duration
+            tokio::time::sleep(Duration::from_millis(50)).await;
+
+            for _ in batch.iter() {
+                self.export_count.fetch_add(1, Ordering::Acquire);
+            }
+            Ok(())
+        }
+    }
+
+    #[test]
+    fn test_simple_processor_async_exporter_without_runtime() {
+        // Use `catch_unwind` to catch the panic caused by missing Tokio runtime
+        let result = std::panic::catch_unwind(|| {
+            let exporter = LogExporterThatRequiresTokio::new();
+            let processor = SimpleLogProcessor::new(Box::new(exporter.clone()));
+
+            let mut record: LogRecord = Default::default();
+            let instrumentation: InstrumentationLibrary = Default::default();
+
+            // This will panic because an tokio async operation within exporter without a runtime.
+            processor.emit(&mut record, &instrumentation);
+        });
+
+        // Verify that the panic occurred and check the panic message for the absence of a Tokio runtime
+        assert!(
+            result.is_err(),
+            "The test should fail due to missing Tokio runtime, but it did not."
+        );
+        let panic_payload = result.unwrap_err();
+        let panic_message = panic_payload
+            .downcast_ref::<String>()
+            .map(|s| s.as_str())
+            .or_else(|| panic_payload.downcast_ref::<&str>().copied())
+            .unwrap_or("No panic message");
+
+        assert!(
+            panic_message.contains("no reactor running")
+                || panic_message.contains("must be called from the context of a Tokio 1.x runtime"),
+            "Expected panic message about missing Tokio runtime, but got: {}",
+            panic_message
+        );
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    #[ignore]
+    // This test demonstrates a potential deadlock scenario in a multi-threaded Tokio runtime.
+    // It spawns Tokio tasks equal to the number of runtime worker threads (4) to emit log events.
+    // Each task attempts to acquire a mutex on the exporter in `SimpleLogProcessor::emit`.
+    // Only one task obtains the lock, while the others are blocked, waiting for its release.
+    //
+    // The task holding the lock invokes the LogExporterThatRequiresTokio, which performs an
+    // asynchronous operation (e.g., network I/O simulated by `tokio::sleep`). This operation
+    // requires yielding control back to the Tokio runtime to make progress.
+    //
+    // However, all worker threads are occupied:
+    // - One thread is executing the async exporter operation
+    // - Three threads are blocked waiting for the mutex
+    //
+    // This leads to a deadlock as there are no available threads to drive the async operation
+    // to completion, preventing the mutex from being released. Consequently, neither the blocked
+    // tasks nor the exporter can proceed.
+    async fn test_simple_processor_async_exporter_with_all_runtime_worker_threads_blocked() {
+        let exporter = LogExporterThatRequiresTokio::new();
+        let processor = Arc::new(SimpleLogProcessor::new(Box::new(exporter.clone())));
+
+        let concurrent_emit = 4; // number of worker threads
+
+        let mut handles = vec![];
+        // try send `concurrent_emit` events concurrently
+        for _ in 0..concurrent_emit {
+            let processor_clone = Arc::clone(&processor);
+            let handle = tokio::spawn(async move {
+                let mut record: LogRecord = Default::default();
+                let instrumentation: InstrumentationLibrary = Default::default();
+                processor_clone.emit(&mut record, &instrumentation);
+            });
+            handles.push(handle);
+        }
+
+        // below code won't get executed
+        for handle in handles {
+            handle.await.unwrap();
+        }
+        assert_eq!(exporter.len(), concurrent_emit);
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    // This test uses a multi-threaded runtime setup with a single worker thread. Note that even
+    // though only one worker thread is created, it is distinct from the main thread. The processor
+    // emits a log event, and the exporter performs an async operation that requires the runtime.
+    // The single worker thread handles this operation without deadlocking, as long as no other
+    // tasks occupy the runtime.
+    async fn test_simple_processor_async_exporter_with_runtime() {
+        let exporter = LogExporterThatRequiresTokio::new();
+        let processor = SimpleLogProcessor::new(Box::new(exporter.clone()));
+
+        let mut record: LogRecord = Default::default();
+        let instrumentation: InstrumentationLibrary = Default::default();
+
+        processor.emit(&mut record, &instrumentation);
+
+        assert_eq!(exporter.len(), 1);
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    // This test uses a multi-threaded runtime setup with the default number of worker threads.
+    // The processor emits a log event, and the exporter, which requires the runtime for its async
+    // operations, can access one of the available worker threads to complete its task. As there
+    // are multiple threads, the exporter can proceed without blocking other tasks, ensuring the
+    // test completes successfully.
+    async fn test_simple_processor_async_exporter_with_multi_thread_runtime() {
+        let exporter = LogExporterThatRequiresTokio::new();
+
+        let processor = SimpleLogProcessor::new(Box::new(exporter.clone()));
+
+        let mut record: LogRecord = Default::default();
+        let instrumentation: InstrumentationLibrary = Default::default();
+
+        processor.emit(&mut record, &instrumentation);
+
+        assert_eq!(exporter.len(), 1);
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    #[ignore]
+    // This test uses a current-thread runtime, where all operations run on the main thread.
+    // The processor emits a log event while the runtime is blocked using `futures::block_on`
+    // to complete the export operation. The exporter, which performs an async operation and
+    // requires the runtime, cannot progress because the main thread is already blocked.
+    // This results in a deadlock, as the runtime cannot move forward.
+    async fn test_simple_processor_async_exporter_with_current_thread_runtime() {
+        let exporter = LogExporterThatRequiresTokio::new();
+
+        let processor = SimpleLogProcessor::new(Box::new(exporter.clone()));
+
+        let mut record: LogRecord = Default::default();
+        let instrumentation: InstrumentationLibrary = Default::default();
+
+        processor.emit(&mut record, &instrumentation);
+
+        assert_eq!(exporter.len(), 1);
+    }
 }