opentelemetry_sdk/logs/
batch_log_processor.rs

1//! # OpenTelemetry Batch Log Processor
2//! The `BatchLogProcessor` is one implementation of the `LogProcessor` interface.
3//!
4//! It buffers log records and sends them to the exporter
5//! in batches. This processor is designed for **production use** in high-throughput
6//! applications and reduces the overhead of frequent exports by using a background
7//! thread for batch processing.
8//!
9//! ## Diagram
10//!
11//! ```ascii
12//!   +-----+---------------+   +-----------------------+   +-------------------+
13//!   |     |               |   |                       |   |                   |
14//!   | SDK | Logger.emit() +---> (Batch)LogProcessor   +--->  (OTLPExporter)   |
15//!   +-----+---------------+   +-----------------------+   +-------------------+
16//! ```
17
18use crate::error::{OTelSdkError, OTelSdkResult};
19use crate::logs::log_processor::LogProcessor;
20use crate::{
21    logs::{LogBatch, LogExporter, SdkLogRecord},
22    Resource,
23};
24use std::sync::mpsc::{self, RecvTimeoutError, SyncSender};
25
26use opentelemetry::{otel_debug, otel_error, otel_warn, Context, InstrumentationScope};
27
28use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
29use std::{cmp::min, env, sync::Mutex};
30use std::{
31    fmt::{self, Debug, Formatter},
32    str::FromStr,
33    sync::Arc,
34    thread,
35    time::Duration,
36    time::Instant,
37};
38
39/// Delay interval between two consecutive exports.
40pub(crate) const OTEL_BLRP_SCHEDULE_DELAY: &str = "OTEL_BLRP_SCHEDULE_DELAY";
41/// Default delay interval between two consecutive exports.
42pub(crate) const OTEL_BLRP_SCHEDULE_DELAY_DEFAULT: Duration = Duration::from_millis(1_000);
43/// Maximum allowed time to export data.
44#[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
45pub(crate) const OTEL_BLRP_EXPORT_TIMEOUT: &str = "OTEL_BLRP_EXPORT_TIMEOUT";
46/// Default maximum allowed time to export data.
47#[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
48pub(crate) const OTEL_BLRP_EXPORT_TIMEOUT_DEFAULT: Duration = Duration::from_millis(30_000);
49/// Maximum queue size.
50pub(crate) const OTEL_BLRP_MAX_QUEUE_SIZE: &str = "OTEL_BLRP_MAX_QUEUE_SIZE";
51/// Default maximum queue size.
52pub(crate) const OTEL_BLRP_MAX_QUEUE_SIZE_DEFAULT: usize = 2_048;
53/// Maximum batch size, must be less than or equal to OTEL_BLRP_MAX_QUEUE_SIZE.
54pub(crate) const OTEL_BLRP_MAX_EXPORT_BATCH_SIZE: &str = "OTEL_BLRP_MAX_EXPORT_BATCH_SIZE";
55/// Default maximum batch size.
56pub(crate) const OTEL_BLRP_MAX_EXPORT_BATCH_SIZE_DEFAULT: usize = 512;
57
58/// Messages sent between application thread and batch log processor's work thread.
59#[allow(clippy::large_enum_variant)]
60#[derive(Debug)]
61enum BatchMessage {
62    /// This is ONLY sent when the number of logs records in the data channel has reached `max_export_batch_size`.
63    ExportLog(Arc<AtomicBool>),
64    /// ForceFlush flushes the current buffer to the exporter.
65    ForceFlush(mpsc::SyncSender<OTelSdkResult>),
66    /// Shut down the worker thread, push all logs in buffer to the exporter.
67    Shutdown(mpsc::SyncSender<OTelSdkResult>),
68    /// Set the resource for the exporter.
69    SetResource(Arc<Resource>),
70}
71
72type LogsData = Box<(SdkLogRecord, InstrumentationScope)>;
73
74/// The `BatchLogProcessor` collects finished logs in a buffer and exports them
75/// in batches to the configured `LogExporter`. This processor is ideal for
76/// high-throughput environments, as it minimizes the overhead of exporting logs
77/// individually. It uses a **dedicated background thread** to manage and export logs
78/// asynchronously, ensuring that the application's main execution flow is not blocked.
79///
80/// This processor supports the following configurations:
81/// - **Queue size**: Maximum number of log records that can be buffered.
82/// - **Batch size**: Maximum number of log records to include in a single export.
83/// - **Scheduled delay**: Frequency at which the batch is exported.
84///
85/// When using this processor with the OTLP Exporter, the following exporter
86/// features are supported:
87/// - `grpc-tonic`: Requires `LoggerProvider` to be created within a tokio runtime.
88/// - `reqwest-blocking-client`: Works with a regular `main` or `tokio::main`.
89///
90/// In other words, async HTTP clients like `reqwest-client` and `hyper-client`
91/// are not supported by this default processor. The OTLP HTTP exporter chooses
92/// its default HTTP client from enabled crate features and cannot tell which
93/// processor will drive it. If your dependency graph enables async HTTP client
94/// features, either pass an explicit blocking client for this processor or use
95/// the experimental async-runtime batch log processor.
96///
97/// `BatchLogProcessor` buffers logs in memory and exports them in batches. An
98/// export is triggered when `max_export_batch_size` is reached or every
99/// `scheduled_delay` milliseconds. Users can explicitly trigger an export using
100/// the `force_flush` method. Shutdown also triggers an export of all buffered
101/// logs and is recommended to be called before the application exits to ensure
102/// all buffered logs are exported.
103///
104/// **Warning**: When using tokio's current-thread runtime, `shutdown()`, which
105/// is a blocking call ,should not be called from your main thread. This can
106/// cause deadlock. Instead, call `shutdown()` from a separate thread or use
107/// tokio's `spawn_blocking`.
108///
109///
110/// ### Using a BatchLogProcessor:
111///
112/// ```rust
113/// # #[cfg(feature = "testing")]
114/// # {
115/// use opentelemetry_sdk::logs::{BatchLogProcessor, BatchConfigBuilder, SdkLoggerProvider};
116/// use opentelemetry::global;
117/// use std::time::Duration;
118/// use opentelemetry_sdk::logs::InMemoryLogExporter;
119///
120/// let exporter = InMemoryLogExporter::default(); // Replace with an actual exporter
121/// let processor = BatchLogProcessor::builder(exporter)
122///     .with_batch_config(
123///         BatchConfigBuilder::default()
124///             .with_max_queue_size(2048)
125///             .with_max_export_batch_size(512)
126///             .with_scheduled_delay(Duration::from_secs(5))
127///             .build(),
128///     )
129///     .build();
130///
131/// let provider = SdkLoggerProvider::builder()
132///     .with_log_processor(processor)
133///     .build();
134/// # }
135///
136pub struct BatchLogProcessor {
137    logs_sender: SyncSender<LogsData>, // Data channel to store log records and instrumentation scopes
138    message_sender: SyncSender<BatchMessage>, // Control channel to store control messages for the worker thread
139    handle: Mutex<Option<thread::JoinHandle<()>>>,
140    forceflush_timeout: Duration,
141    export_log_message_sent: Arc<AtomicBool>,
142    current_batch_size: Arc<AtomicUsize>,
143    max_export_batch_size: usize,
144
145    // Track dropped logs - we'll log this at shutdown
146    dropped_logs_count: AtomicUsize,
147
148    // Track the maximum queue size that was configured for this processor
149    max_queue_size: usize,
150}
151
152impl Debug for BatchLogProcessor {
153    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
154        f.debug_struct("BatchLogProcessor")
155            .field("message_sender", &self.message_sender)
156            .finish()
157    }
158}
159
160impl LogProcessor for BatchLogProcessor {
161    fn emit(&self, record: &mut SdkLogRecord, instrumentation: &InstrumentationScope) {
162        let result = self
163            .logs_sender
164            .try_send(Box::new((record.clone(), instrumentation.clone())));
165
166        // match for result and handle each separately
167        match result {
168            Ok(_) => {
169                // Successfully sent the log record to the data channel.
170                // Increment the current batch size and check if it has reached
171                // the max export batch size.
172                if self.current_batch_size.fetch_add(1, Ordering::AcqRel) + 1
173                    >= self.max_export_batch_size
174                {
175                    // Check if the a control message for exporting logs is
176                    // already sent to the worker thread. If not, send a control
177                    // message to export logs. `export_log_message_sent` is set
178                    // to false ONLY when the worker thread has processed the
179                    // control message.
180
181                    if !self.export_log_message_sent.load(Ordering::Relaxed) {
182                        // This is a cost-efficient check as atomic load
183                        // operations do not require exclusive access to cache
184                        // line. Perform atomic swap to
185                        // `export_log_message_sent` ONLY when the atomic load
186                        // operation above returns false. Atomic
187                        // swap/compare_exchange operations require exclusive
188                        // access to cache line on most processor architectures.
189                        // We could have used compare_exchange as well here, but
190                        // it's more verbose than swap.
191                        if !self.export_log_message_sent.swap(true, Ordering::Relaxed) {
192                            match self.message_sender.try_send(BatchMessage::ExportLog(
193                                self.export_log_message_sent.clone(),
194                            )) {
195                                Ok(_) => {
196                                    // Control message sent successfully.
197                                }
198                                Err(_err) => {
199                                    // TODO: Log error If the control message
200                                    // could not be sent, reset the
201                                    // `export_log_message_sent` flag.
202                                    self.export_log_message_sent.store(false, Ordering::Relaxed);
203                                }
204                            }
205                        }
206                    }
207                }
208            }
209            Err(mpsc::TrySendError::Full(_)) => {
210                // Increment dropped logs count. The first time we have to drop
211                // a log, emit a warning.
212                if self.dropped_logs_count.fetch_add(1, Ordering::Relaxed) == 0 {
213                    otel_warn!(name: "BatchLogProcessor.LogDroppingStarted",
214                        message = "BatchLogProcessor dropped a LogRecord due to queue full. No further log will be emitted for further drops until Shutdown. During Shutdown time, a log will be emitted with exact count of total logs dropped.");
215                }
216            }
217            Err(mpsc::TrySendError::Disconnected(_)) => {
218                // The following `otel_warn!` may cause an infinite feedback loop of
219                // 'telemetry-induced-telemetry', potentially causing a stack overflow
220                let _guard = Context::enter_telemetry_suppressed_scope();
221
222                // Given background thread is the only receiver, and it's
223                // disconnected, it indicates the thread is shutdown
224                otel_warn!(
225                    name: "BatchLogProcessor.Emit.AfterShutdown",
226                    message = "Logs are being emitted even after Shutdown. This indicates incorrect lifecycle management of OTelLoggerProvider in application. Logs will not be exported."
227                );
228            }
229        }
230    }
231
232    fn force_flush(&self) -> OTelSdkResult {
233        let (sender, receiver) = mpsc::sync_channel(1);
234        match self
235            .message_sender
236            .try_send(BatchMessage::ForceFlush(sender))
237        {
238            Ok(_) => receiver
239                .recv_timeout(self.forceflush_timeout)
240                .map_err(|err| {
241                    if err == RecvTimeoutError::Timeout {
242                        OTelSdkError::Timeout(self.forceflush_timeout)
243                    } else {
244                        OTelSdkError::InternalFailure(format!("{err}"))
245                    }
246                })?,
247            Err(mpsc::TrySendError::Full(_)) => {
248                // If the control message could not be sent, emit a warning.
249                otel_debug!(
250                    name: "BatchLogProcessor.ForceFlush.ControlChannelFull",
251                    message = "Control message to flush the worker thread could not be sent as the control channel is full. This can occur if user repeatedily calls force_flush/shutdown without finishing the previous call."
252                );
253                Err(OTelSdkError::InternalFailure("ForceFlush cannot be performed as Control channel is full. This can occur if user repeatedily calls force_flush/shutdown without finishing the previous call.".into()))
254            }
255            Err(mpsc::TrySendError::Disconnected(_)) => {
256                // Given background thread is the only receiver, and it's
257                // disconnected, it indicates the thread is shutdown
258                otel_debug!(
259                    name: "BatchLogProcessor.ForceFlush.AlreadyShutdown",
260                    message = "ForceFlush invoked after Shutdown. This will not perform Flush and indicates a incorrect lifecycle management in Application."
261                );
262
263                Err(OTelSdkError::AlreadyShutdown)
264            }
265        }
266    }
267
268    fn shutdown_with_timeout(&self, timeout: Duration) -> OTelSdkResult {
269        let dropped_logs = self.dropped_logs_count.load(Ordering::Relaxed);
270        let max_queue_size = self.max_queue_size;
271        if dropped_logs > 0 {
272            otel_warn!(
273                name: "BatchLogProcessor.LogsDropped",
274                dropped_logs_count = dropped_logs,
275                max_queue_size = max_queue_size,
276                message = "Logs were dropped due to a queue being full. The count represents the total count of log records dropped in the lifetime of this BatchLogProcessor. Consider increasing the queue size and/or decrease delay between intervals."
277            );
278        }
279
280        let (sender, receiver) = mpsc::sync_channel(1);
281        match self.message_sender.try_send(BatchMessage::Shutdown(sender)) {
282            Ok(_) => {
283                receiver
284                    .recv_timeout(timeout)
285                    .map(|_| {
286                        // join the background thread after receiving back the
287                        // shutdown signal
288                        if let Some(handle) = self.handle.lock().unwrap().take() {
289                            handle.join().unwrap();
290                        }
291                        OTelSdkResult::Ok(())
292                    })
293                    .map_err(|err| match err {
294                        RecvTimeoutError::Timeout => {
295                            otel_error!(
296                                name: "BatchLogProcessor.Shutdown.Timeout",
297                                message = "BatchLogProcessor shutdown timing out."
298                            );
299                            OTelSdkError::Timeout(timeout)
300                        }
301                        _ => {
302                            otel_error!(
303                                name: "BatchLogProcessor.Shutdown.Error",
304                                error = format!("{}", err)
305                            );
306                            OTelSdkError::InternalFailure(format!("{err}"))
307                        }
308                    })?
309            }
310            Err(mpsc::TrySendError::Full(_)) => {
311                // If the control message could not be sent, emit a warning.
312                otel_debug!(
313                    name: "BatchLogProcessor.Shutdown.ControlChannelFull",
314                    message = "Control message to shutdown the worker thread could not be sent as the control channel is full. This can occur if user repeatedily calls force_flush/shutdown without finishing the previous call."
315                );
316                Err(OTelSdkError::InternalFailure("Shutdown cannot be performed as Control channel is full. This can occur if user repeatedily calls force_flush/shutdown without finishing the previous call.".into()))
317            }
318            Err(mpsc::TrySendError::Disconnected(_)) => {
319                // Given background thread is the only receiver, and it's
320                // disconnected, it indicates the thread is shutdown
321                otel_debug!(
322                    name: "BatchLogProcessor.Shutdown.AlreadyShutdown",
323                    message = "Shutdown is being invoked more than once. This is noop, but indicates a potential issue in the application's lifecycle management."
324                );
325
326                Err(OTelSdkError::AlreadyShutdown)
327            }
328        }
329    }
330
331    fn set_resource(&mut self, resource: &Resource) {
332        let resource = Arc::new(resource.clone());
333        let _ = self
334            .message_sender
335            .try_send(BatchMessage::SetResource(resource));
336    }
337}
338
339impl BatchLogProcessor {
340    pub(crate) fn new<E>(mut exporter: E, config: BatchConfig) -> Self
341    where
342        E: LogExporter + Send + Sync + 'static,
343    {
344        let (logs_sender, logs_receiver) = mpsc::sync_channel::<LogsData>(config.max_queue_size);
345        let (message_sender, message_receiver) = mpsc::sync_channel::<BatchMessage>(64); // Is this a reasonable bound?
346        let max_queue_size = config.max_queue_size;
347        let max_export_batch_size = config.max_export_batch_size;
348        let current_batch_size = Arc::new(AtomicUsize::new(0));
349        let current_batch_size_for_thread = current_batch_size.clone();
350
351        let handle = thread::Builder::new()
352            .name("OpenTelemetry.Logs.BatchProcessor".to_string())
353            .spawn(move || {
354                let _suppress_guard = Context::enter_telemetry_suppressed_scope();
355                otel_debug!(
356                    name: "BatchLogProcessor.ThreadStarted",
357                    interval_in_millisecs = config.scheduled_delay.as_millis(),
358                    max_export_batch_size = config.max_export_batch_size,
359                    max_queue_size = max_queue_size,
360                );
361                let mut last_export_time = Instant::now();
362                let mut logs = Vec::with_capacity(config.max_export_batch_size);
363                let current_batch_size = current_batch_size_for_thread;
364
365                // This method gets up to `max_export_batch_size` amount of logs from the channel and exports them.
366                // It returns the result of the export operation.
367                // It expects the logs vec to be empty when it's called.
368                #[inline]
369                fn get_logs_and_export<E>(
370                    logs_receiver: &mpsc::Receiver<LogsData>,
371                    exporter: &E,
372                    logs: &mut Vec<LogsData>,
373                    last_export_time: &mut Instant,
374                    current_batch_size: &AtomicUsize,
375                    max_export_size: usize,
376                ) -> OTelSdkResult
377                where
378                    E: LogExporter + Send + Sync + 'static,
379                {
380                    let target = current_batch_size.load(Ordering::Acquire); // `target` is used to determine the stopping criteria for exporting logs.
381                    let mut result = OTelSdkResult::Ok(());
382                    let mut total_exported_logs: usize = 0;
383
384                    while target > 0 && total_exported_logs < target {
385                        let batch_limit = max_export_size.min(target - total_exported_logs);
386
387                        // Get up to the remaining target batch size from the channel and push them to the logs vec
388                        while let Ok(log) = logs_receiver.try_recv() {
389                            logs.push(log);
390                            if logs.len() == batch_limit {
391                                break;
392                            }
393                        }
394
395                        let count_of_logs = logs.len(); // Count of logs that will be exported
396                        if count_of_logs == 0 {
397                            break;
398                        }
399                        total_exported_logs += count_of_logs;
400
401                        result = export_batch_sync(exporter, logs, last_export_time); // This method clears the logs vec after exporting
402
403                        current_batch_size.fetch_sub(count_of_logs, Ordering::AcqRel);
404                    }
405                    result
406                }
407
408                loop {
409                    let remaining_time = config
410                        .scheduled_delay
411                        .checked_sub(last_export_time.elapsed())
412                        .unwrap_or(config.scheduled_delay);
413
414                    match message_receiver.recv_timeout(remaining_time) {
415                        Ok(BatchMessage::ExportLog(export_log_message_sent)) => {
416                            // Reset the export log message sent flag now it has has been processed.
417                            export_log_message_sent.store(false, Ordering::Relaxed);
418
419                            otel_debug!(
420                                name: "BatchLogProcessor.ExportingDueToBatchSize",
421                            );
422
423                            let _ = get_logs_and_export(
424                                &logs_receiver,
425                                &exporter,
426                                &mut logs,
427                                &mut last_export_time,
428                                &current_batch_size,
429                                max_export_batch_size,
430                            );
431                        }
432                        Ok(BatchMessage::ForceFlush(sender)) => {
433                            otel_debug!(name: "BatchLogProcessor.ExportingDueToForceFlush");
434                            let result = get_logs_and_export(
435                                &logs_receiver,
436                                &exporter,
437                                &mut logs,
438                                &mut last_export_time,
439                                &current_batch_size,
440                                max_export_batch_size,
441                            );
442                            let _ = sender.send(result);
443                        }
444                        Ok(BatchMessage::Shutdown(sender)) => {
445                            otel_debug!(name: "BatchLogProcessor.ExportingDueToShutdown");
446                            let result = get_logs_and_export(
447                                &logs_receiver,
448                                &exporter,
449                                &mut logs,
450                                &mut last_export_time,
451                                &current_batch_size,
452                                max_export_batch_size,
453                            );
454                            let _ = exporter.shutdown();
455                            let _ = sender.send(result);
456
457                            otel_debug!(
458                                name: "BatchLogProcessor.ThreadExiting",
459                                reason = "ShutdownRequested"
460                            );
461                            //
462                            // break out the loop and return from the current background thread.
463                            //
464                            break;
465                        }
466                        Ok(BatchMessage::SetResource(resource)) => {
467                            exporter.set_resource(&resource);
468                        }
469                        Err(RecvTimeoutError::Timeout) => {
470                            otel_debug!(
471                                name: "BatchLogProcessor.ExportingDueToTimer",
472                            );
473
474                            let _ = get_logs_and_export(
475                                &logs_receiver,
476                                &exporter,
477                                &mut logs,
478                                &mut last_export_time,
479                                &current_batch_size,
480                                max_export_batch_size,
481                            );
482                        }
483                        Err(RecvTimeoutError::Disconnected) => {
484                            // Channel disconnected, only thing to do is break
485                            // out (i.e exit the thread)
486                            otel_debug!(
487                                name: "BatchLogProcessor.ThreadExiting",
488                                reason = "MessageSenderDisconnected"
489                            );
490                            break;
491                        }
492                    }
493                }
494                otel_debug!(
495                    name: "BatchLogProcessor.ThreadStopped"
496                );
497            })
498            .expect("Thread spawn failed."); //TODO: Handle thread spawn failure
499
500        // Return batch processor with link to worker
501        BatchLogProcessor {
502            logs_sender,
503            message_sender,
504            handle: Mutex::new(Some(handle)),
505            forceflush_timeout: Duration::from_secs(5), // TODO: make this configurable
506            dropped_logs_count: AtomicUsize::new(0),
507            max_queue_size,
508            export_log_message_sent: Arc::new(AtomicBool::new(false)),
509            current_batch_size,
510            max_export_batch_size,
511        }
512    }
513
514    /// Create a new batch processor builder
515    pub fn builder<E>(exporter: E) -> BatchLogProcessorBuilder<E>
516    where
517        E: LogExporter,
518    {
519        BatchLogProcessorBuilder {
520            exporter,
521            config: Default::default(),
522        }
523    }
524}
525
526#[allow(clippy::vec_box)]
527fn export_batch_sync<E>(
528    exporter: &E,
529    batch: &mut Vec<Box<(SdkLogRecord, InstrumentationScope)>>,
530    last_export_time: &mut Instant,
531) -> OTelSdkResult
532where
533    E: LogExporter + ?Sized,
534{
535    *last_export_time = Instant::now();
536
537    if batch.is_empty() {
538        return OTelSdkResult::Ok(());
539    }
540
541    let export = exporter.export(LogBatch::new_with_owned_data(batch.as_slice()));
542    let export_result = futures_executor::block_on(export);
543
544    // Clear the batch vec after exporting
545    batch.clear();
546
547    match export_result {
548        Ok(_) => OTelSdkResult::Ok(()),
549        Err(err) => {
550            otel_error!(
551                name: "BatchLogProcessor.ExportError",
552                error = format!("{}", err)
553            );
554            OTelSdkResult::Err(err)
555        }
556    }
557}
558
559///
560/// A builder for creating [`BatchLogProcessor`] instances.
561///
562#[derive(Debug)]
563pub struct BatchLogProcessorBuilder<E> {
564    exporter: E,
565    config: BatchConfig,
566}
567
568impl<E> BatchLogProcessorBuilder<E>
569where
570    E: LogExporter + 'static,
571{
572    /// Set the BatchConfig for [`BatchLogProcessorBuilder`]
573    pub fn with_batch_config(self, config: BatchConfig) -> Self {
574        BatchLogProcessorBuilder { config, ..self }
575    }
576
577    /// Build a batch processor
578    pub fn build(self) -> BatchLogProcessor {
579        BatchLogProcessor::new(self.exporter, self.config)
580    }
581}
582
583/// Batch log processor configuration.
584/// Use [`BatchConfigBuilder`] to configure your own instance of [`BatchConfig`].
585#[derive(Debug)]
586#[allow(dead_code)]
587pub struct BatchConfig {
588    /// The maximum queue size to buffer logs for delayed processing. If the
589    /// queue gets full it drops the logs. The default value of is 2048.
590    pub(crate) max_queue_size: usize,
591
592    /// The delay interval in milliseconds between two consecutive processing
593    /// of batches. The default value is 1 second.
594    pub(crate) scheduled_delay: Duration,
595
596    /// The maximum number of logs to process in a single batch. If there are
597    /// more than one batch worth of logs then it processes multiple batches
598    /// of logs one batch after the other without any delay. The default value
599    /// is 512.
600    pub(crate) max_export_batch_size: usize,
601
602    /// The maximum duration to export a batch of data.
603    #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
604    pub(crate) max_export_timeout: Duration,
605}
606
607impl Default for BatchConfig {
608    fn default() -> Self {
609        BatchConfigBuilder::default().build()
610    }
611}
612
613/// A builder for creating [`BatchConfig`] instances.
614#[derive(Debug)]
615pub struct BatchConfigBuilder {
616    max_queue_size: usize,
617    scheduled_delay: Duration,
618    max_export_batch_size: usize,
619    #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
620    max_export_timeout: Duration,
621}
622
623impl Default for BatchConfigBuilder {
624    /// Create a new [`BatchConfigBuilder`] initialized with default batch config values as per the specs.
625    /// The values are overridden by environment variables if set.
626    /// The supported environment variables are:
627    /// * `OTEL_BLRP_MAX_QUEUE_SIZE`
628    /// * `OTEL_BLRP_SCHEDULE_DELAY`
629    /// * `OTEL_BLRP_MAX_EXPORT_BATCH_SIZE`
630    /// * `OTEL_BLRP_EXPORT_TIMEOUT`
631    ///
632    /// Note: Programmatic configuration overrides any value set via the environment variable.
633    fn default() -> Self {
634        BatchConfigBuilder {
635            max_queue_size: OTEL_BLRP_MAX_QUEUE_SIZE_DEFAULT,
636            scheduled_delay: OTEL_BLRP_SCHEDULE_DELAY_DEFAULT,
637            max_export_batch_size: OTEL_BLRP_MAX_EXPORT_BATCH_SIZE_DEFAULT,
638            #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
639            max_export_timeout: OTEL_BLRP_EXPORT_TIMEOUT_DEFAULT,
640        }
641        .init_from_env_vars()
642    }
643}
644
645impl BatchConfigBuilder {
646    /// Set max_queue_size for [`BatchConfigBuilder`].
647    /// It's the maximum queue size to buffer logs for delayed processing.
648    /// If the queue gets full it will drop the logs.
649    /// The default value is 2048.
650    ///
651    /// Corresponding environment variable: `OTEL_BLRP_MAX_QUEUE_SIZE`.
652    ///
653    /// Note: Programmatically setting this will override any value set via the environment variable.
654    pub fn with_max_queue_size(mut self, max_queue_size: usize) -> Self {
655        self.max_queue_size = max_queue_size;
656        self
657    }
658
659    /// Set scheduled_delay for [`BatchConfigBuilder`].
660    /// It's the delay interval in milliseconds between two consecutive processing of batches.
661    /// The default value is 1000 milliseconds.
662    ///
663    /// Corresponding environment variable: `OTEL_BLRP_SCHEDULE_DELAY`.
664    ///
665    /// Note: Programmatically setting this will override any value set via the environment variable.
666    pub fn with_scheduled_delay(mut self, scheduled_delay: Duration) -> Self {
667        self.scheduled_delay = scheduled_delay;
668        self
669    }
670
671    /// Set max_export_timeout for [`BatchConfigBuilder`].
672    /// It's the maximum duration to export a batch of data.
673    /// The default value is 30000 milliseconds.
674    ///
675    /// Corresponding environment variable: `OTEL_BLRP_EXPORT_TIMEOUT`.
676    ///
677    /// Note: Programmatically setting this will override any value set via the environment variable.
678    #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
679    pub fn with_max_export_timeout(mut self, max_export_timeout: Duration) -> Self {
680        self.max_export_timeout = max_export_timeout;
681        self
682    }
683
684    /// Set max_export_batch_size for [`BatchConfigBuilder`].
685    /// It's the maximum number of logs to process in a single batch. If there are
686    /// more than one batch worth of logs then it processes multiple batches
687    /// of logs one batch after the other without any delay.
688    /// The default value is 512.
689    ///
690    /// Corresponding environment variable: `OTEL_BLRP_MAX_EXPORT_BATCH_SIZE`.
691    ///
692    /// Note: Programmatically setting this will override any value set via the environment variable.
693    pub fn with_max_export_batch_size(mut self, max_export_batch_size: usize) -> Self {
694        self.max_export_batch_size = max_export_batch_size;
695        self
696    }
697
698    /// Builds a `BatchConfig` enforcing the following invariants:
699    /// * `max_export_batch_size` must be less than or equal to `max_queue_size`.
700    pub fn build(self) -> BatchConfig {
701        // max export batch size must be less or equal to max queue size.
702        // we set max export batch size to max queue size if it's larger than max queue size.
703        let max_export_batch_size = min(self.max_export_batch_size, self.max_queue_size);
704
705        BatchConfig {
706            max_queue_size: self.max_queue_size,
707            scheduled_delay: self.scheduled_delay,
708            #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
709            max_export_timeout: self.max_export_timeout,
710            max_export_batch_size,
711        }
712    }
713
714    fn init_from_env_vars(mut self) -> Self {
715        if let Some(max_queue_size) = env::var(OTEL_BLRP_MAX_QUEUE_SIZE)
716            .ok()
717            .and_then(|queue_size| usize::from_str(&queue_size).ok())
718        {
719            self.max_queue_size = max_queue_size;
720        }
721
722        if let Some(max_export_batch_size) = env::var(OTEL_BLRP_MAX_EXPORT_BATCH_SIZE)
723            .ok()
724            .and_then(|batch_size| usize::from_str(&batch_size).ok())
725        {
726            self.max_export_batch_size = max_export_batch_size;
727        }
728
729        if let Some(scheduled_delay) = env::var(OTEL_BLRP_SCHEDULE_DELAY)
730            .ok()
731            .and_then(|delay| u64::from_str(&delay).ok())
732        {
733            self.scheduled_delay = Duration::from_millis(scheduled_delay);
734        }
735
736        #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
737        if let Some(max_export_timeout) = env::var(OTEL_BLRP_EXPORT_TIMEOUT)
738            .ok()
739            .and_then(|s| u64::from_str(&s).ok())
740        {
741            self.max_export_timeout = Duration::from_millis(max_export_timeout);
742        }
743
744        self
745    }
746}
747
748#[cfg(all(test, feature = "testing", feature = "logs"))]
749mod tests {
750    use super::{
751        BatchConfig, BatchConfigBuilder, BatchLogProcessor, OTEL_BLRP_MAX_EXPORT_BATCH_SIZE,
752        OTEL_BLRP_MAX_EXPORT_BATCH_SIZE_DEFAULT, OTEL_BLRP_MAX_QUEUE_SIZE,
753        OTEL_BLRP_MAX_QUEUE_SIZE_DEFAULT, OTEL_BLRP_SCHEDULE_DELAY,
754        OTEL_BLRP_SCHEDULE_DELAY_DEFAULT,
755    };
756    #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
757    use super::{OTEL_BLRP_EXPORT_TIMEOUT, OTEL_BLRP_EXPORT_TIMEOUT_DEFAULT};
758    use crate::error::OTelSdkResult;
759    use crate::logs::log_processor::tests::MockLogExporter;
760    use crate::logs::SdkLogRecord;
761    use crate::logs::{LogBatch, LogExporter};
762    use crate::{
763        logs::{InMemoryLogExporter, InMemoryLogExporterBuilder, LogProcessor, SdkLoggerProvider},
764        Resource,
765    };
766    use opentelemetry::logs::LogRecord;
767    use opentelemetry::InstrumentationScope;
768    use opentelemetry::KeyValue;
769    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
770    use std::sync::{Arc, Mutex};
771    use std::time::Duration;
772
773    #[test]
774    fn test_default_const_values() {
775        assert_eq!(OTEL_BLRP_SCHEDULE_DELAY, "OTEL_BLRP_SCHEDULE_DELAY");
776        assert_eq!(OTEL_BLRP_SCHEDULE_DELAY_DEFAULT.as_millis(), 1_000);
777        #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
778        assert_eq!(OTEL_BLRP_EXPORT_TIMEOUT, "OTEL_BLRP_EXPORT_TIMEOUT");
779        #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
780        assert_eq!(OTEL_BLRP_EXPORT_TIMEOUT_DEFAULT.as_millis(), 30_000);
781        assert_eq!(OTEL_BLRP_MAX_QUEUE_SIZE, "OTEL_BLRP_MAX_QUEUE_SIZE");
782        assert_eq!(OTEL_BLRP_MAX_QUEUE_SIZE_DEFAULT, 2_048);
783        assert_eq!(
784            OTEL_BLRP_MAX_EXPORT_BATCH_SIZE,
785            "OTEL_BLRP_MAX_EXPORT_BATCH_SIZE"
786        );
787        assert_eq!(OTEL_BLRP_MAX_EXPORT_BATCH_SIZE_DEFAULT, 512);
788    }
789
790    #[test]
791    fn test_default_batch_config_adheres_to_specification() {
792        // The following environment variables are expected to be unset so that their default values are used.
793        let env_vars = vec![
794            OTEL_BLRP_SCHEDULE_DELAY,
795            #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
796            OTEL_BLRP_EXPORT_TIMEOUT,
797            OTEL_BLRP_MAX_QUEUE_SIZE,
798            OTEL_BLRP_MAX_EXPORT_BATCH_SIZE,
799        ];
800
801        let config = temp_env::with_vars_unset(env_vars, BatchConfig::default);
802
803        assert_eq!(config.scheduled_delay, OTEL_BLRP_SCHEDULE_DELAY_DEFAULT);
804        #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
805        assert_eq!(config.max_export_timeout, OTEL_BLRP_EXPORT_TIMEOUT_DEFAULT);
806        assert_eq!(config.max_queue_size, OTEL_BLRP_MAX_QUEUE_SIZE_DEFAULT);
807        assert_eq!(
808            config.max_export_batch_size,
809            OTEL_BLRP_MAX_EXPORT_BATCH_SIZE_DEFAULT
810        );
811    }
812
813    #[test]
814    fn test_code_based_config_overrides_env_vars() {
815        let env_vars = vec![
816            (OTEL_BLRP_SCHEDULE_DELAY, Some("2000")),
817            (OTEL_BLRP_MAX_QUEUE_SIZE, Some("4096")),
818            (OTEL_BLRP_MAX_EXPORT_BATCH_SIZE, Some("1024")),
819        ];
820
821        temp_env::with_vars(env_vars, || {
822            let config = BatchConfigBuilder::default()
823                .with_max_queue_size(2048)
824                .with_scheduled_delay(Duration::from_millis(1000))
825                .with_max_export_batch_size(512)
826                .build();
827
828            assert_eq!(config.scheduled_delay, Duration::from_millis(1000));
829            assert_eq!(config.max_queue_size, 2048);
830            assert_eq!(config.max_export_batch_size, 512);
831        });
832    }
833
834    #[test]
835    fn test_batch_config_configurable_by_env_vars() {
836        let env_vars = vec![
837            (OTEL_BLRP_SCHEDULE_DELAY, Some("2000")),
838            #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
839            (OTEL_BLRP_EXPORT_TIMEOUT, Some("60000")),
840            (OTEL_BLRP_MAX_QUEUE_SIZE, Some("4096")),
841            (OTEL_BLRP_MAX_EXPORT_BATCH_SIZE, Some("1024")),
842        ];
843
844        let config = temp_env::with_vars(env_vars, BatchConfig::default);
845
846        assert_eq!(config.scheduled_delay, Duration::from_millis(2000));
847        #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
848        assert_eq!(config.max_export_timeout, Duration::from_millis(60000));
849        assert_eq!(config.max_queue_size, 4096);
850        assert_eq!(config.max_export_batch_size, 1024);
851    }
852    #[test]
853    fn test_force_flush_being_called() {
854        #[derive(Debug, Clone)]
855        struct MockExporter {
856            export_called: Arc<AtomicBool>,
857        }
858        impl LogExporter for MockExporter {
859            async fn export(&self, _batch: LogBatch<'_>) -> OTelSdkResult {
860                self.export_called.store(true, Ordering::SeqCst);
861                Ok(())
862            }
863        }
864        let exporter = MockExporter {
865            export_called: Arc::new(AtomicBool::new(false)),
866        };
867        let processor = BatchLogProcessor::new(exporter.clone(), BatchConfig::default());
868        let scope = opentelemetry::InstrumentationScope::builder("my-crate")
869            .with_schema_url("https://opentelemetry.io/schemas/1.17.0")
870            .build();
871        processor.emit(&mut SdkLogRecord::new(), &scope);
872        processor.force_flush().unwrap();
873        assert!(exporter.export_called.load(Ordering::SeqCst));
874    }
875
876    #[test]
877    fn test_batch_config_max_export_batch_size_validation() {
878        let env_vars = vec![
879            (OTEL_BLRP_MAX_QUEUE_SIZE, Some("256")),
880            (OTEL_BLRP_MAX_EXPORT_BATCH_SIZE, Some("1024")),
881        ];
882
883        let config = temp_env::with_vars(env_vars, BatchConfig::default);
884
885        assert_eq!(config.max_queue_size, 256);
886        assert_eq!(config.max_export_batch_size, 256);
887        assert_eq!(config.scheduled_delay, OTEL_BLRP_SCHEDULE_DELAY_DEFAULT);
888        #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
889        assert_eq!(config.max_export_timeout, OTEL_BLRP_EXPORT_TIMEOUT_DEFAULT);
890    }
891
892    #[test]
893    fn test_batch_config_with_fields() {
894        let batch_builder = BatchConfigBuilder::default()
895            .with_max_export_batch_size(1)
896            .with_scheduled_delay(Duration::from_millis(2))
897            .with_max_queue_size(4);
898
899        #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
900        let batch_builder = batch_builder.with_max_export_timeout(Duration::from_millis(3));
901        let batch = batch_builder.build();
902
903        assert_eq!(batch.max_export_batch_size, 1);
904        assert_eq!(batch.scheduled_delay, Duration::from_millis(2));
905        #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
906        assert_eq!(batch.max_export_timeout, Duration::from_millis(3));
907        assert_eq!(batch.max_queue_size, 4);
908    }
909
910    #[test]
911    fn test_build_batch_log_processor_builder() {
912        let mut env_vars = vec![
913            (OTEL_BLRP_MAX_EXPORT_BATCH_SIZE, Some("500")),
914            (OTEL_BLRP_SCHEDULE_DELAY, Some("I am not number")),
915            #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
916            (OTEL_BLRP_EXPORT_TIMEOUT, Some("2046")),
917        ];
918        temp_env::with_vars(env_vars.clone(), || {
919            let builder = BatchLogProcessor::builder(InMemoryLogExporter::default());
920
921            assert_eq!(builder.config.max_export_batch_size, 500);
922            assert_eq!(
923                builder.config.scheduled_delay,
924                OTEL_BLRP_SCHEDULE_DELAY_DEFAULT
925            );
926            assert_eq!(
927                builder.config.max_queue_size,
928                OTEL_BLRP_MAX_QUEUE_SIZE_DEFAULT
929            );
930
931            #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
932            assert_eq!(
933                builder.config.max_export_timeout,
934                Duration::from_millis(2046)
935            );
936        });
937
938        env_vars.push((OTEL_BLRP_MAX_QUEUE_SIZE, Some("120")));
939
940        temp_env::with_vars(env_vars, || {
941            let builder = BatchLogProcessor::builder(InMemoryLogExporter::default());
942            assert_eq!(builder.config.max_export_batch_size, 120);
943            assert_eq!(builder.config.max_queue_size, 120);
944        });
945    }
946
947    #[test]
948    fn test_build_batch_log_processor_builder_with_custom_config() {
949        let expected = BatchConfigBuilder::default()
950            .with_max_export_batch_size(1)
951            .with_scheduled_delay(Duration::from_millis(2))
952            .with_max_queue_size(4)
953            .build();
954
955        let builder =
956            BatchLogProcessor::builder(InMemoryLogExporter::default()).with_batch_config(expected);
957
958        let actual = &builder.config;
959        assert_eq!(actual.max_export_batch_size, 1);
960        assert_eq!(actual.scheduled_delay, Duration::from_millis(2));
961        assert_eq!(actual.max_queue_size, 4);
962    }
963
964    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
965    async fn test_set_resource_batch_processor() {
966        let exporter = MockLogExporter {
967            resource: Arc::new(Mutex::new(None)),
968        };
969        let processor = BatchLogProcessor::new(exporter.clone(), BatchConfig::default());
970        let provider = SdkLoggerProvider::builder()
971            .with_log_processor(processor)
972            .with_resource(
973                Resource::builder_empty()
974                    .with_attributes([
975                        KeyValue::new("k1", "v1"),
976                        KeyValue::new("k2", "v3"),
977                        KeyValue::new("k3", "v3"),
978                        KeyValue::new("k4", "v4"),
979                        KeyValue::new("k5", "v5"),
980                    ])
981                    .build(),
982            )
983            .build();
984
985        provider.force_flush().unwrap();
986
987        assert_eq!(exporter.get_resource().unwrap().into_iter().count(), 5);
988        let _ = provider.shutdown();
989    }
990
991    #[tokio::test(flavor = "multi_thread")]
992    async fn test_batch_shutdown() {
993        // assert we will receive an error
994        // setup
995        let exporter = InMemoryLogExporterBuilder::default()
996            .keep_records_on_shutdown()
997            .build();
998        let processor = BatchLogProcessor::new(exporter.clone(), BatchConfig::default());
999
1000        let mut record = SdkLogRecord::new();
1001        let instrumentation = InstrumentationScope::default();
1002
1003        processor.emit(&mut record, &instrumentation);
1004        processor.force_flush().unwrap();
1005        processor.shutdown().unwrap();
1006        // todo: expect to see errors here. How should we assert this?
1007        processor.emit(&mut record, &instrumentation);
1008        assert_eq!(1, exporter.get_emitted_logs().unwrap().len());
1009        assert!(exporter.is_shutdown_called());
1010    }
1011
1012    #[tokio::test(flavor = "current_thread")]
1013    async fn test_batch_log_processor_shutdown_under_async_runtime_current_flavor_multi_thread() {
1014        let exporter = InMemoryLogExporterBuilder::default().build();
1015        let processor = BatchLogProcessor::new(exporter.clone(), BatchConfig::default());
1016
1017        processor.shutdown().unwrap();
1018    }
1019
1020    #[tokio::test(flavor = "current_thread")]
1021    async fn test_batch_log_processor_shutdown_with_async_runtime_current_flavor_current_thread() {
1022        let exporter = InMemoryLogExporterBuilder::default().build();
1023        let processor = BatchLogProcessor::new(exporter.clone(), BatchConfig::default());
1024        processor.shutdown().unwrap();
1025    }
1026
1027    #[tokio::test(flavor = "multi_thread")]
1028    async fn test_batch_log_processor_shutdown_with_async_runtime_multi_flavor_multi_thread() {
1029        let exporter = InMemoryLogExporterBuilder::default().build();
1030        let processor = BatchLogProcessor::new(exporter.clone(), BatchConfig::default());
1031        processor.shutdown().unwrap();
1032    }
1033
1034    #[tokio::test(flavor = "multi_thread")]
1035    async fn test_batch_log_processor_shutdown_with_async_runtime_multi_flavor_current_thread() {
1036        let exporter = InMemoryLogExporterBuilder::default().build();
1037        let processor = BatchLogProcessor::new(exporter.clone(), BatchConfig::default());
1038        processor.shutdown().unwrap();
1039    }
1040
1041    /// A slow exporter that counts the number of logs received.
1042    /// Used for stress testing the BatchLogProcessor.
1043    #[derive(Debug, Clone)]
1044    struct CountingExporter {
1045        count: Arc<AtomicUsize>,
1046    }
1047
1048    impl CountingExporter {
1049        fn new() -> Self {
1050            CountingExporter {
1051                count: Arc::new(AtomicUsize::new(0)),
1052            }
1053        }
1054    }
1055
1056    impl LogExporter for CountingExporter {
1057        async fn export(&self, batch: LogBatch<'_>) -> OTelSdkResult {
1058            self.count.fetch_add(batch.len(), Ordering::SeqCst);
1059            // Simulate slow export to cause queue buildup and drops
1060            std::thread::sleep(std::time::Duration::from_millis(20));
1061            Ok(())
1062        }
1063    }
1064
1065    /// Stress test that verifies all logs are accounted for.
1066    /// With multiple threads pushing logs faster than the exporter can handle,
1067    /// some logs will inevitably be dropped. This test validates:
1068    /// total_logs_sent == logs_received_by_exporter + logs_dropped
1069    #[test]
1070    fn test_batch_log_processor_all_logs_accounted_for() {
1071        let exporter = CountingExporter::new();
1072        let exporter_count = exporter.count.clone();
1073
1074        // Configure with small queue to force drops under pressure
1075        let config = BatchConfigBuilder::default()
1076            .with_max_queue_size(2048)
1077            .with_max_export_batch_size(512)
1078            .with_scheduled_delay(Duration::from_millis(5))
1079            .build();
1080
1081        let processor = BatchLogProcessor::new(exporter, config);
1082
1083        let total_logs_per_thread = 100_000;
1084        let num_threads = 4;
1085        let total_logs_to_emit = total_logs_per_thread * num_threads;
1086
1087        // Use scoped threads to safely share the processor reference
1088        std::thread::scope(|s| {
1089            for _ in 0..num_threads {
1090                s.spawn(|| {
1091                    for _ in 0..total_logs_per_thread {
1092                        let mut record = SdkLogRecord::new();
1093                        record.set_body("stress test log".into());
1094                        let instrumentation = InstrumentationScope::default();
1095                        processor.emit(&mut record, &instrumentation);
1096                    }
1097                });
1098            }
1099        });
1100
1101        // Shutdown the processor to ensure all buffered logs are flushed
1102        processor.shutdown().unwrap();
1103
1104        let logs_received = exporter_count.load(Ordering::SeqCst);
1105        let logs_dropped = processor.dropped_logs_count.load(Ordering::SeqCst);
1106
1107        // The invariant: every log is either received or dropped
1108        assert_eq!(
1109            logs_received + logs_dropped,
1110            total_logs_to_emit,
1111            "Logs unaccounted for! Received: {}, Dropped: {}, Total emitted: {}",
1112            logs_received,
1113            logs_dropped,
1114            total_logs_to_emit
1115        );
1116
1117        // Also verify that some logs were actually dropped (stress test validation)
1118        // If no logs were dropped, the test parameters may need adjustment
1119        assert!(
1120            logs_dropped > 0,
1121            "Expected some logs to be dropped under stress, but none were. \
1122             Consider reducing queue size or increasing thread count/log volume."
1123        );
1124    }
1125}
opentelemetry_sdk/logs/batch_log_processor.rs

opentelemetry_sdk/logs/
batch_log_processor.rs
