opentelemetry_sdk/logs/
simple_log_processor.rs

1//! # OpenTelemetry Simple Log Processor
2//! The `SimpleLogProcessor` is one implementation of the `LogProcessor` interface.
3//!
4//! It forwards log records to the exporter immediately after they are emitted
5//! (or one exporter after another if applicable). This processor is
6//! **synchronous** and is designed for debugging or testing purposes. It is
7//! **not suitable for production** environments due to its lack of batching,
8//! performance optimizations, or support for high-throughput scenarios.
9//!
10//! ## Diagram
11//!
12//! ```ascii
13//!   +-----+---------------+   +-----------------------+   +-------------------+
14//!   |     |               |   |                       |   |                   |
15//!   | SDK | Logger.emit() +---> (Simple)LogProcessor  +--->  LogExporter      |
16//!   +-----+---------------+   +-----------------------+   +-------------------+
17//! ```
18
19use crate::error::{OTelSdkError, OTelSdkResult};
20use crate::logs::log_processor::LogProcessor;
21use crate::{
22    logs::{LogBatch, LogExporter, SdkLogRecord},
23    Resource,
24};
25
26use opentelemetry::{otel_debug, otel_error, otel_warn, Context, InstrumentationScope};
27
28use std::fmt::Debug;
29use std::sync::atomic::AtomicBool;
30use std::sync::Mutex;
31use std::time::Duration;
32
33/// A [`LogProcessor`] designed for testing and debugging purpose, that immediately
34/// exports log records as they are emitted. Log records are exported synchronously
35/// in the same thread that emits the log record.
36/// When using this processor with the OTLP Exporter, the following exporter
37/// features are supported:
38/// - `grpc-tonic`: This requires LoggerProvider to be created within a tokio
39///   runtime. Logs can be emitted from any thread, including tokio runtime
40///   threads.
41/// - `reqwest-blocking-client`: LoggerProvider may be created anywhere, but
42///   logs must be emitted from a non-tokio runtime thread.
43/// - `reqwest-client`: LoggerProvider may be created anywhere, but logs must be
44///   emitted from a tokio runtime thread.
45///
46/// ## Example
47///
48/// ### Using a SimpleLogProcessor
49///
50/// ```rust
51/// # #[cfg(feature = "testing")]
52/// # {
53/// use opentelemetry_sdk::logs::{SimpleLogProcessor, SdkLoggerProvider, LogExporter};
54/// use opentelemetry::global;
55/// use opentelemetry_sdk::logs::InMemoryLogExporter;
56///
57/// let exporter = InMemoryLogExporter::default(); // Replace with an actual exporter
58/// let provider = SdkLoggerProvider::builder()
59///     .with_simple_exporter(exporter)
60///     .build();
61/// # }
62/// ```
63///
64#[derive(Debug)]
65pub struct SimpleLogProcessor<T: LogExporter> {
66    exporter: Mutex<T>,
67    is_shutdown: AtomicBool,
68}
69
70impl<T: LogExporter> SimpleLogProcessor<T> {
71    /// Creates a new instance of `SimpleLogProcessor`.
72    pub fn new(exporter: T) -> Self {
73        SimpleLogProcessor {
74            exporter: Mutex::new(exporter),
75            is_shutdown: AtomicBool::new(false),
76        }
77    }
78}
79
80impl<T: LogExporter> LogProcessor for SimpleLogProcessor<T> {
81    fn emit(&self, record: &mut SdkLogRecord, instrumentation: &InstrumentationScope) {
82        let _suppress_guard = Context::enter_telemetry_suppressed_scope();
83        // noop after shutdown
84        if self.is_shutdown.load(std::sync::atomic::Ordering::Relaxed) {
85            // this is a warning, as the user is trying to log after the processor has been shutdown
86            otel_warn!(
87                name: "SimpleLogProcessor.Emit.ProcessorShutdown",
88            );
89            return;
90        }
91
92        let result = self
93            .exporter
94            .lock()
95            .map_err(|_| OTelSdkError::InternalFailure("SimpleLogProcessor mutex poison".into()))
96            .and_then(|exporter| {
97                let log_tuple = &[(record as &SdkLogRecord, instrumentation)];
98                futures_executor::block_on(exporter.export(LogBatch::new(log_tuple)))
99            });
100        // Handle errors with specific static names
101        match result {
102            Err(OTelSdkError::InternalFailure(_)) => {
103                // logging as debug as this is not a user error
104                otel_debug!(
105                    name: "SimpleLogProcessor.Emit.MutexPoisoning",
106                );
107            }
108            Err(err) => {
109                otel_error!(
110                    name: "SimpleLogProcessor.Emit.ExportError",
111                    error = format!("{}",err)
112                );
113            }
114            _ => {}
115        }
116    }
117
118    fn force_flush(&self) -> OTelSdkResult {
119        Ok(())
120    }
121
122    fn shutdown_with_timeout(&self, timeout: Duration) -> OTelSdkResult {
123        self.is_shutdown
124            .store(true, std::sync::atomic::Ordering::Relaxed);
125        if let Ok(exporter) = self.exporter.lock() {
126            exporter.shutdown_with_timeout(timeout)
127        } else {
128            Err(OTelSdkError::InternalFailure(
129                "SimpleLogProcessor mutex poison at shutdown".into(),
130            ))
131        }
132    }
133
134    fn set_resource(&mut self, resource: &Resource) {
135        if let Ok(mut exporter) = self.exporter.lock() {
136            exporter.set_resource(resource);
137        }
138    }
139
140    #[inline]
141    fn event_enabled(
142        &self,
143        level: opentelemetry::logs::Severity,
144        target: &str,
145        name: Option<&str>,
146    ) -> bool {
147        if let Ok(exporter) = self.exporter.lock() {
148            exporter.event_enabled(level, target, name)
149        } else {
150            true
151        }
152    }
153}
154
155#[cfg(all(test, feature = "testing", feature = "logs"))]
156mod tests {
157    use crate::logs::log_processor::tests::MockLogExporter;
158    use crate::logs::{LogBatch, LogExporter, SdkLogRecord, SdkLogger};
159    use crate::{
160        error::OTelSdkResult,
161        logs::{InMemoryLogExporterBuilder, LogProcessor, SdkLoggerProvider, SimpleLogProcessor},
162        Resource,
163    };
164    use opentelemetry::logs::{LogRecord, Logger, LoggerProvider};
165    use opentelemetry::InstrumentationScope;
166    use opentelemetry::KeyValue;
167    use std::sync::atomic::{AtomicUsize, Ordering};
168    use std::sync::{Arc, Mutex};
169    use std::time;
170    use std::time::Duration;
171
172    #[derive(Debug, Clone)]
173    struct LogExporterThatRequiresTokio {
174        export_count: Arc<AtomicUsize>,
175    }
176
177    impl LogExporterThatRequiresTokio {
178        /// Creates a new instance of `LogExporterThatRequiresTokio`.
179        fn new() -> Self {
180            LogExporterThatRequiresTokio {
181                export_count: Arc::new(AtomicUsize::new(0)),
182            }
183        }
184
185        /// Returns the number of logs stored in the exporter.
186        fn len(&self) -> usize {
187            self.export_count.load(Ordering::Acquire)
188        }
189    }
190
191    impl LogExporter for LogExporterThatRequiresTokio {
192        async fn export(&self, batch: LogBatch<'_>) -> OTelSdkResult {
193            // Simulate minimal dependency on tokio by sleeping asynchronously for a short duration
194            tokio::time::sleep(Duration::from_millis(50)).await;
195
196            for _ in batch.iter() {
197                self.export_count.fetch_add(1, Ordering::Acquire);
198            }
199            Ok(())
200        }
201        fn shutdown_with_timeout(&self, _timeout: time::Duration) -> OTelSdkResult {
202            Ok(())
203        }
204    }
205
206    #[test]
207    fn test_set_resource_simple_processor() {
208        let exporter = MockLogExporter {
209            resource: Arc::new(Mutex::new(None)),
210        };
211        let processor = SimpleLogProcessor::new(exporter.clone());
212        let _ = SdkLoggerProvider::builder()
213            .with_log_processor(processor)
214            .with_resource(
215                Resource::builder_empty()
216                    .with_attributes([
217                        KeyValue::new("k1", "v1"),
218                        KeyValue::new("k2", "v3"),
219                        KeyValue::new("k3", "v3"),
220                        KeyValue::new("k4", "v4"),
221                        KeyValue::new("k5", "v5"),
222                    ])
223                    .build(),
224            )
225            .build();
226        assert_eq!(exporter.get_resource().unwrap().into_iter().count(), 5);
227    }
228
229    #[test]
230    fn test_simple_shutdown() {
231        let exporter = InMemoryLogExporterBuilder::default()
232            .keep_records_on_shutdown()
233            .build();
234        let processor = SimpleLogProcessor::new(exporter.clone());
235
236        let mut record: SdkLogRecord = SdkLogRecord::new();
237        let instrumentation: InstrumentationScope = Default::default();
238
239        processor.emit(&mut record, &instrumentation);
240
241        processor.shutdown().unwrap();
242
243        let is_shutdown = processor
244            .is_shutdown
245            .load(std::sync::atomic::Ordering::Relaxed);
246        assert!(is_shutdown);
247
248        processor.emit(&mut record, &instrumentation);
249
250        assert_eq!(1, exporter.get_emitted_logs().unwrap().len());
251        assert!(exporter.is_shutdown_called());
252    }
253
254    #[test]
255    fn test_simple_processor_sync_exporter_without_runtime() {
256        let exporter = InMemoryLogExporterBuilder::default().build();
257        let processor = SimpleLogProcessor::new(exporter.clone());
258
259        let mut record: SdkLogRecord = SdkLogRecord::new();
260        let instrumentation: InstrumentationScope = Default::default();
261
262        processor.emit(&mut record, &instrumentation);
263
264        assert_eq!(exporter.get_emitted_logs().unwrap().len(), 1);
265    }
266
267    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
268    async fn test_simple_processor_sync_exporter_with_runtime() {
269        let exporter = InMemoryLogExporterBuilder::default().build();
270        let processor = SimpleLogProcessor::new(exporter.clone());
271
272        let mut record: SdkLogRecord = SdkLogRecord::new();
273        let instrumentation: InstrumentationScope = Default::default();
274
275        processor.emit(&mut record, &instrumentation);
276
277        assert_eq!(exporter.get_emitted_logs().unwrap().len(), 1);
278    }
279
280    #[tokio::test(flavor = "multi_thread")]
281    async fn test_simple_processor_sync_exporter_with_multi_thread_runtime() {
282        let exporter = InMemoryLogExporterBuilder::default().build();
283        let processor = Arc::new(SimpleLogProcessor::new(exporter.clone()));
284
285        let mut handles = vec![];
286        for _ in 0..10 {
287            let processor_clone = Arc::clone(&processor);
288            let handle = tokio::spawn(async move {
289                let mut record: SdkLogRecord = SdkLogRecord::new();
290                let instrumentation: InstrumentationScope = Default::default();
291                processor_clone.emit(&mut record, &instrumentation);
292            });
293            handles.push(handle);
294        }
295
296        for handle in handles {
297            handle.await.unwrap();
298        }
299
300        assert_eq!(exporter.get_emitted_logs().unwrap().len(), 10);
301    }
302
303    #[tokio::test(flavor = "current_thread")]
304    async fn test_simple_processor_sync_exporter_with_current_thread_runtime() {
305        let exporter = InMemoryLogExporterBuilder::default().build();
306        let processor = SimpleLogProcessor::new(exporter.clone());
307
308        let mut record: SdkLogRecord = SdkLogRecord::new();
309        let instrumentation: InstrumentationScope = Default::default();
310
311        processor.emit(&mut record, &instrumentation);
312
313        assert_eq!(exporter.get_emitted_logs().unwrap().len(), 1);
314    }
315
316    #[test]
317    fn test_simple_processor_async_exporter_without_runtime() {
318        // Use `catch_unwind` to catch the panic caused by missing Tokio runtime
319        let result = std::panic::catch_unwind(|| {
320            let exporter = LogExporterThatRequiresTokio::new();
321            let processor = SimpleLogProcessor::new(exporter.clone());
322
323            let mut record: SdkLogRecord = SdkLogRecord::new();
324            let instrumentation: InstrumentationScope = Default::default();
325
326            // This will panic because an tokio async operation within exporter without a runtime.
327            processor.emit(&mut record, &instrumentation);
328        });
329
330        // Verify that the panic occurred and check the panic message for the absence of a Tokio runtime
331        assert!(
332            result.is_err(),
333            "The test should fail due to missing Tokio runtime, but it did not."
334        );
335        let panic_payload = result.unwrap_err();
336        let panic_message = panic_payload
337            .downcast_ref::<String>()
338            .map(|s| s.as_str())
339            .or_else(|| panic_payload.downcast_ref::<&str>().copied())
340            .unwrap_or("No panic message");
341
342        assert!(
343            panic_message.contains("no reactor running")
344                || panic_message.contains("must be called from the context of a Tokio 1.x runtime"),
345            "Expected panic message about missing Tokio runtime, but got: {panic_message}"
346        );
347    }
348
349    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
350    #[ignore]
351    // This test demonstrates a potential deadlock scenario in a multi-threaded Tokio runtime.
352    // It spawns Tokio tasks equal to the number of runtime worker threads (4) to emit log events.
353    // Each task attempts to acquire a mutex on the exporter in `SimpleLogProcessor::emit`.
354    // Only one task obtains the lock, while the others are blocked, waiting for its release.
355    //
356    // The task holding the lock invokes the LogExporterThatRequiresTokio, which performs an
357    // asynchronous operation (e.g., network I/O simulated by `tokio::sleep`). This operation
358    // requires yielding control back to the Tokio runtime to make progress.
359    //
360    // However, all worker threads are occupied:
361    // - One thread is executing the async exporter operation
362    // - Three threads are blocked waiting for the mutex
363    //
364    // This leads to a deadlock as there are no available threads to drive the async operation
365    // to completion, preventing the mutex from being released. Consequently, neither the blocked
366    // tasks nor the exporter can proceed.
367    async fn test_simple_processor_async_exporter_with_all_runtime_worker_threads_blocked() {
368        let exporter = LogExporterThatRequiresTokio::new();
369        let processor = Arc::new(SimpleLogProcessor::new(exporter.clone()));
370
371        let concurrent_emit = 4; // number of worker threads
372
373        let mut handles = vec![];
374        // try send `concurrent_emit` events concurrently
375        for _ in 0..concurrent_emit {
376            let processor_clone = Arc::clone(&processor);
377            let handle = tokio::spawn(async move {
378                let mut record: SdkLogRecord = SdkLogRecord::new();
379                let instrumentation: InstrumentationScope = Default::default();
380                processor_clone.emit(&mut record, &instrumentation);
381            });
382            handles.push(handle);
383        }
384
385        // below code won't get executed
386        for handle in handles {
387            handle.await.unwrap();
388        }
389        assert_eq!(exporter.len(), concurrent_emit);
390    }
391
392    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
393    // This test uses a multi-threaded runtime setup with a single worker thread. Note that even
394    // though only one worker thread is created, it is distinct from the main thread. The processor
395    // emits a log event, and the exporter performs an async operation that requires the runtime.
396    // The single worker thread handles this operation without deadlocking, as long as no other
397    // tasks occupy the runtime.
398    async fn test_simple_processor_async_exporter_with_runtime() {
399        let exporter = LogExporterThatRequiresTokio::new();
400        let processor = SimpleLogProcessor::new(exporter.clone());
401
402        let mut record: SdkLogRecord = SdkLogRecord::new();
403        let instrumentation: InstrumentationScope = Default::default();
404
405        processor.emit(&mut record, &instrumentation);
406
407        assert_eq!(exporter.len(), 1);
408    }
409
410    #[tokio::test(flavor = "multi_thread")]
411    // This test uses a multi-threaded runtime setup with the default number of worker threads.
412    // The processor emits a log event, and the exporter, which requires the runtime for its async
413    // operations, can access one of the available worker threads to complete its task. As there
414    // are multiple threads, the exporter can proceed without blocking other tasks, ensuring the
415    // test completes successfully.
416    async fn test_simple_processor_async_exporter_with_multi_thread_runtime() {
417        let exporter = LogExporterThatRequiresTokio::new();
418
419        let processor = SimpleLogProcessor::new(exporter.clone());
420
421        let mut record: SdkLogRecord = SdkLogRecord::new();
422        let instrumentation: InstrumentationScope = Default::default();
423
424        processor.emit(&mut record, &instrumentation);
425
426        assert_eq!(exporter.len(), 1);
427    }
428
429    #[tokio::test(flavor = "current_thread")]
430    #[ignore]
431    // This test uses a current-thread runtime, where all operations run on the main thread.
432    // The processor emits a log event while the runtime is blocked using `futures::block_on`
433    // to complete the export operation. The exporter, which performs an async operation and
434    // requires the runtime, cannot progress because the main thread is already blocked.
435    // This results in a deadlock, as the runtime cannot move forward.
436    async fn test_simple_processor_async_exporter_with_current_thread_runtime() {
437        let exporter = LogExporterThatRequiresTokio::new();
438
439        let processor = SimpleLogProcessor::new(exporter.clone());
440
441        let mut record: SdkLogRecord = SdkLogRecord::new();
442        let instrumentation: InstrumentationScope = Default::default();
443
444        processor.emit(&mut record, &instrumentation);
445
446        assert_eq!(exporter.len(), 1);
447    }
448
449    #[derive(Debug, Clone)]
450    struct ReentrantLogExporter {
451        logger: Arc<Mutex<Option<SdkLogger>>>,
452    }
453
454    impl ReentrantLogExporter {
455        fn new() -> Self {
456            Self {
457                logger: Arc::new(Mutex::new(None)),
458            }
459        }
460
461        fn set_logger(&self, logger: SdkLogger) {
462            let mut guard = self.logger.lock().unwrap();
463            *guard = Some(logger);
464        }
465    }
466
467    impl LogExporter for ReentrantLogExporter {
468        async fn export(&self, _batch: LogBatch<'_>) -> OTelSdkResult {
469            let logger = self.logger.lock().unwrap();
470            if let Some(logger) = logger.as_ref() {
471                let mut log_record = logger.create_log_record();
472                log_record.set_severity_number(opentelemetry::logs::Severity::Error);
473                logger.emit(log_record);
474            }
475
476            Ok(())
477        }
478    }
479
480    #[test]
481    fn exporter_internal_log_does_not_deadlock_with_simple_processor() {
482        // This tests that even when exporter produces logs while
483        // exporting, it does not deadlock, as SimpleLogProcessor
484        // activates SuppressGuard before calling the exporter.
485        let exporter: ReentrantLogExporter = ReentrantLogExporter::new();
486        let logger_provider = SdkLoggerProvider::builder()
487            .with_simple_exporter(exporter.clone())
488            .build();
489        exporter.set_logger(logger_provider.logger("processor-logger"));
490
491        let logger = logger_provider.logger("test-logger");
492        let mut log_record = logger.create_log_record();
493        log_record.set_severity_number(opentelemetry::logs::Severity::Error);
494        logger.emit(log_record);
495    }
496}
opentelemetry_sdk/logs/simple_log_processor.rs

opentelemetry_sdk/logs/
simple_log_processor.rs
