opentelemetry_sdk/logs/
log_processor.rs

1//! # OpenTelemetry Log Processor Interface
2//!
3//! The `LogProcessor` interface provides hooks for log record processing and
4//! exporting. Log processors receive `LogRecord`s emitted by the SDK's
5//! `Logger` and determine how these records are handled.
6//!
7//! Built-in log processors are responsible for converting logs to exportable
8//! representations and passing them to configured exporters. They can be
9//! registered directly with a `LoggerProvider`.
10//!
11//! ## Types of Log Processors
12//!
13//! There are currently two types of log processors available in the SDK:
14//! - **SimpleLogProcessor**: Forwards log records to the exporter immediately.
15//! - **BatchLogProcessor**: Buffers log records and sends them to the exporter in batches.
16//!
17//! For more information, see simple_log_processor.rs and batch_log_processor.rs.
18//!
19//! ## Diagram
20//!
21//! ```ascii
22//!   +-----+---------------+   +-----------------------+   +-------------------+
23//!   |     |               |   |                       |   |                   |
24//!   | SDK | Logger.emit() +---> (Simple)LogProcessor  +--->  LogExporter      |
25//!   |     |               |   | (Batch)LogProcessor   +--->  (OTLPExporter)   |
26//!   +-----+---------------+   +-----------------------+   +-------------------+
27//! ```
28
29use crate::error::OTelSdkResult;
30use crate::{logs::SdkLogRecord, Resource};
31
32use opentelemetry::logs::Severity;
33use opentelemetry::{otel_warn, InstrumentationScope};
34
35use std::fmt::Debug;
36use std::time::Duration;
37
38/// The interface for plugging into a [`SdkLogger`].
39///
40/// [`SdkLogger`]: crate::logs::SdkLogger
41pub trait LogProcessor: Send + Sync + Debug {
42    /// Called when a log record is ready to processed and exported.
43    ///
44    /// This method receives a mutable reference to `LogRecord`. If the processor
45    /// needs to handle the export asynchronously, it should clone the data to
46    /// ensure it can be safely processed without lifetime issues. Any changes
47    /// made to the log data in this method will be reflected in the next log
48    /// processor in the chain.
49    ///
50    /// # Parameters
51    /// - `record`: A mutable reference to `LogRecord` representing the log record.
52    /// - `instrumentation`: The instrumentation scope associated with the log record.
53    fn emit(&self, data: &mut SdkLogRecord, instrumentation: &InstrumentationScope);
54    /// Force the logs lying in the cache to be exported.
55    fn force_flush(&self) -> OTelSdkResult;
56    /// Shuts down the processor.
57    /// After shutdown returns the log processor should stop processing any logs.
58    /// It's up to the implementation on when to drop the LogProcessor.
59    ///
60    /// Implementors that manage resources (background threads, network connections,
61    /// file handles, etc.) should override this method to properly clean up.
62    /// Processors that wrap other processors should forward the shutdown call to
63    /// the wrapped processor(s).
64    /// Simple processors that only transform log data can use the default implementation.
65    fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
66        // It would have been better to make this method required, but that ship
67        // sailed when the logs API was declared stable.
68        otel_warn!(
69            name: "LogProcessor.DefaultShutdownWithTimeout",
70            message = "LogProcessor is using default shutdown implementation. If this processor manages background threads, network connections, file handles, or other resources that need cleanup, implement `shutdown_with_timeout()` to properly release them. Simple processors that only transform log data can safely use this default."
71        );
72        Ok(())
73    }
74    /// Shuts down the processor with default timeout.
75    ///
76    /// Implementors typically do not need to change this method, and can just
77    /// implement `shutdown_with_timeout`.
78    fn shutdown(&self) -> OTelSdkResult {
79        self.shutdown_with_timeout(Duration::from_secs(5))
80    }
81
82    /// Check if logging is enabled
83    fn event_enabled(&self, _level: Severity, _target: &str, _name: Option<&str>) -> bool {
84        // By default, all logs are enabled
85        true
86    }
87
88    /// Set the resource for the log processor.
89    fn set_resource(&mut self, _resource: &Resource) {}
90}
91
92#[cfg(all(test, feature = "testing", feature = "logs"))]
93pub(crate) mod tests {
94    use crate::logs::{LogBatch, LogExporter, SdkLogRecord};
95    use crate::Resource;
96    use crate::{
97        error::OTelSdkResult,
98        logs::{LogProcessor, SdkLoggerProvider},
99    };
100    use opentelemetry::logs::AnyValue;
101    use opentelemetry::logs::LogRecord as _;
102    use opentelemetry::logs::{Logger, LoggerProvider};
103    use opentelemetry::{InstrumentationScope, Key};
104    use std::sync::{Arc, Mutex};
105
106    #[derive(Debug, Clone)]
107    pub(crate) struct MockLogExporter {
108        pub resource: Arc<Mutex<Option<Resource>>>,
109    }
110
111    impl LogExporter for MockLogExporter {
112        async fn export(&self, _batch: LogBatch<'_>) -> OTelSdkResult {
113            Ok(())
114        }
115
116        fn set_resource(&mut self, resource: &Resource) {
117            self.resource
118                .lock()
119                .map(|mut res_opt| {
120                    res_opt.replace(resource.clone());
121                })
122                .expect("mock log exporter shouldn't error when setting resource");
123        }
124    }
125
126    // Implementation specific to the MockLogExporter, not part of the LogExporter trait
127    impl MockLogExporter {
128        pub(crate) fn get_resource(&self) -> Option<Resource> {
129            (*self.resource).lock().unwrap().clone()
130        }
131    }
132
133    #[derive(Debug)]
134    struct FirstProcessor {
135        pub(crate) logs: Arc<Mutex<Vec<(SdkLogRecord, InstrumentationScope)>>>,
136    }
137
138    impl LogProcessor for FirstProcessor {
139        fn emit(&self, record: &mut SdkLogRecord, instrumentation: &InstrumentationScope) {
140            // add attribute
141            record.add_attribute(
142                Key::from_static_str("processed_by"),
143                AnyValue::String("FirstProcessor".into()),
144            );
145            // update body
146            record.body = Some("Updated by FirstProcessor".into());
147
148            self.logs
149                .lock()
150                .unwrap()
151                .push((record.clone(), instrumentation.clone())); //clone as the LogProcessor is storing the data.
152        }
153
154        fn force_flush(&self) -> OTelSdkResult {
155            Ok(())
156        }
157
158        fn shutdown_with_timeout(&self, _timeout: std::time::Duration) -> OTelSdkResult {
159            Ok(())
160        }
161    }
162
163    #[derive(Debug)]
164    struct SecondProcessor {
165        pub(crate) logs: Arc<Mutex<Vec<(SdkLogRecord, InstrumentationScope)>>>,
166    }
167
168    impl LogProcessor for SecondProcessor {
169        fn emit(&self, record: &mut SdkLogRecord, instrumentation: &InstrumentationScope) {
170            assert!(record.attributes_contains(
171                &Key::from_static_str("processed_by"),
172                &AnyValue::String("FirstProcessor".into())
173            ));
174            assert!(
175                record.body.clone().unwrap()
176                    == AnyValue::String("Updated by FirstProcessor".into())
177            );
178            self.logs
179                .lock()
180                .unwrap()
181                .push((record.clone(), instrumentation.clone()));
182        }
183
184        fn force_flush(&self) -> OTelSdkResult {
185            Ok(())
186        }
187
188        fn shutdown_with_timeout(&self, _timeout: std::time::Duration) -> OTelSdkResult {
189            Ok(())
190        }
191    }
192
193    #[test]
194    fn test_log_data_modification_by_multiple_processors() {
195        let first_processor_logs = Arc::new(Mutex::new(Vec::new()));
196        let second_processor_logs = Arc::new(Mutex::new(Vec::new()));
197
198        let first_processor = FirstProcessor {
199            logs: Arc::clone(&first_processor_logs),
200        };
201        let second_processor = SecondProcessor {
202            logs: Arc::clone(&second_processor_logs),
203        };
204
205        let logger_provider = SdkLoggerProvider::builder()
206            .with_log_processor(first_processor)
207            .with_log_processor(second_processor)
208            .build();
209
210        let logger = logger_provider.logger("test-logger");
211        let mut log_record = logger.create_log_record();
212        log_record.body = Some(AnyValue::String("Test log".into()));
213
214        logger.emit(log_record);
215
216        assert_eq!(first_processor_logs.lock().unwrap().len(), 1);
217        assert_eq!(second_processor_logs.lock().unwrap().len(), 1);
218
219        let first_log = &first_processor_logs.lock().unwrap()[0];
220        let second_log = &second_processor_logs.lock().unwrap()[0];
221
222        assert!(first_log.0.attributes_contains(
223            &Key::from_static_str("processed_by"),
224            &AnyValue::String("FirstProcessor".into())
225        ));
226        assert!(second_log.0.attributes_contains(
227            &Key::from_static_str("processed_by"),
228            &AnyValue::String("FirstProcessor".into())
229        ));
230
231        assert!(
232            first_log.0.body.clone().unwrap()
233                == AnyValue::String("Updated by FirstProcessor".into())
234        );
235        assert!(
236            second_log.0.body.clone().unwrap()
237                == AnyValue::String("Updated by FirstProcessor".into())
238        );
239    }
240}
opentelemetry_sdk/logs/log_processor.rs

opentelemetry_sdk/logs/
log_processor.rs
