opentelemetry_sdk/trace/
span_processor.rs

1//! # OpenTelemetry Span Processor Interface
2//!
3//! Span processor is an interface which allows hooks for span start and end method
4//! invocations. The span processors are invoked only when
5//! [`is_recording`] is true.
6//!
7//! Built-in span processors are responsible for batching and conversion of spans to
8//! exportable representation and passing batches to exporters.
9//!
10//! Span processors can be registered directly on SDK [`TracerProvider`] and they are
11//! invoked in the same order as they were registered.
12//!
13//! All `Tracer` instances created by a `TracerProvider` share the same span processors.
14//! Changes to this collection reflect in all `Tracer` instances.
15//!
16//! The following diagram shows `SpanProcessor`'s relationship to other components
17//! in the SDK:
18//!
19//! ```ascii
20//!   +-----+--------------+   +-----------------------+   +-------------------+
21//!   |     |              |   |                       |   |                   |
22//!   |     |              |   | (Batch)SpanProcessor  |   |    SpanExporter   |
23//!   |     |              +---> (Simple)SpanProcessor +--->  (OTLPExporter)   |
24//!   |     |              |   |                       |   |                   |
25//!   | SDK | Tracer.span()|   +-----------------------+   +-------------------+
26//!   |     | Span.end()   |
27//!   |     |              |
28//!   |     |              |
29//!   |     |              |
30//!   |     |              |
31//!   +-----+--------------+
32//! ```
33//!
34//! [`is_recording`]: opentelemetry::trace::Span::is_recording()
35//! [`TracerProvider`]: opentelemetry::trace::TracerProvider
36
37use crate::error::{OTelSdkError, OTelSdkResult};
38use crate::resource::Resource;
39use crate::trace::Span;
40use crate::trace::{SpanData, SpanExporter};
41use opentelemetry::Context;
42use opentelemetry::{otel_debug, otel_error, otel_warn};
43use std::cmp::min;
44use std::sync::atomic::{AtomicUsize, Ordering};
45use std::sync::{Arc, Mutex};
46use std::{env, str::FromStr, time::Duration};
47
48use std::sync::atomic::AtomicBool;
49use std::thread;
50use std::time::Instant;
51
52/// Delay interval between two consecutive exports.
53pub(crate) const OTEL_BSP_SCHEDULE_DELAY: &str = "OTEL_BSP_SCHEDULE_DELAY";
54/// Default delay interval between two consecutive exports.
55pub(crate) const OTEL_BSP_SCHEDULE_DELAY_DEFAULT: Duration = Duration::from_millis(5_000);
56/// Maximum queue size
57pub(crate) const OTEL_BSP_MAX_QUEUE_SIZE: &str = "OTEL_BSP_MAX_QUEUE_SIZE";
58/// Default maximum queue size
59pub(crate) const OTEL_BSP_MAX_QUEUE_SIZE_DEFAULT: usize = 2_048;
60/// Maximum batch size, must be less than or equal to OTEL_BSP_MAX_QUEUE_SIZE
61pub(crate) const OTEL_BSP_MAX_EXPORT_BATCH_SIZE: &str = "OTEL_BSP_MAX_EXPORT_BATCH_SIZE";
62/// Default maximum batch size
63pub(crate) const OTEL_BSP_MAX_EXPORT_BATCH_SIZE_DEFAULT: usize = 512;
64/// Maximum allowed time to export data.
65pub(crate) const OTEL_BSP_EXPORT_TIMEOUT: &str = "OTEL_BSP_EXPORT_TIMEOUT";
66/// Default maximum allowed time to export data.
67pub(crate) const OTEL_BSP_EXPORT_TIMEOUT_DEFAULT: Duration = Duration::from_millis(30_000);
68pub(crate) const OTEL_BSP_MAX_CONCURRENT_EXPORTS: &str = "OTEL_BSP_MAX_CONCURRENT_EXPORTS";
69/// Default max concurrent exports for BSP
70pub(crate) const OTEL_BSP_MAX_CONCURRENT_EXPORTS_DEFAULT: usize = 1;
71
72/// `SpanProcessor` is an interface which allows hooks for span start and end
73/// method invocations. The span processors are invoked only when is_recording
74/// is true.
75pub trait SpanProcessor: Send + Sync + std::fmt::Debug {
76    /// `on_start` is called when a `Span` is started.  This method is called
77    /// synchronously on the thread that started the span, therefore it should
78    /// not block or throw exceptions.
79    fn on_start(&self, span: &mut Span, cx: &Context);
80
81    /// `on_end` is called after a `Span` is ended (i.e., the end timestamp is
82    /// already set). This method is called synchronously within the `Span::end`
83    /// API, therefore it should not block or throw an exception.
84    ///
85    /// # Accessing Context
86    ///
87    /// **Important**: Do not rely on [`Context::current()`] in `on_end`. When `on_end`
88    /// is called during span cleanup, `Context::current()` returns whatever context
89    /// happens to be active at that moment, which is typically unrelated to the span
90    /// being ended. Contexts can be activated in any order and are not necessarily
91    /// hierarchical.
92    ///
93    /// **Best Practice**: Extract any needed context information in [`on_start`]
94    /// and store it as span attributes. This ensures the information is available
95    /// in the [`SpanData`] passed to `on_end`.
96    ///
97    /// # Example
98    ///
99    /// ```rust,ignore
100    /// impl SpanProcessor for MyProcessor {
101    ///     fn on_start(&self, span: &mut Span, cx: &Context) {
102    ///         // Extract baggage and store as span attribute
103    ///         if let Some(value) = cx.baggage().get("my-key") {
104    ///             span.set_attribute(KeyValue::new("my-key", value.to_string()));
105    ///         }
106    ///     }
107    ///
108    ///     fn on_end(&self, span: SpanData) {
109    ///         // Access the attribute stored in on_start
110    ///         let my_value = span.attributes.iter()
111    ///             .find(|kv| kv.key.as_str() == "my-key");
112    ///     }
113    /// }
114    /// ```
115    ///
116    /// [`on_start`]: SpanProcessor::on_start
117    /// [`Context::current()`]: opentelemetry::Context::current
118    ///
119    /// TODO - This method should take reference to `SpanData`
120    fn on_end(&self, span: SpanData);
121    /// Force the spans lying in the cache to be exported.
122    fn force_flush(&self) -> OTelSdkResult;
123    /// Shuts down the processor. Called when SDK is shut down. This is an
124    /// opportunity for processors to do any cleanup required.
125    ///
126    /// Implementation should make sure shutdown can be called multiple times.
127    fn shutdown_with_timeout(&self, timeout: Duration) -> OTelSdkResult;
128    /// shutdown the processor with a default timeout.
129    fn shutdown(&self) -> OTelSdkResult {
130        self.shutdown_with_timeout(Duration::from_secs(5))
131    }
132    /// Set the resource for the span processor.
133    fn set_resource(&mut self, _resource: &Resource) {}
134}
135
136/// A [SpanProcessor] that passes finished spans to the configured
137/// `SpanExporter`, as soon as they are finished, without any batching. This is
138/// typically useful for debugging and testing. For scenarios requiring higher
139/// performance/throughput, consider using [BatchSpanProcessor].
140/// Spans are exported synchronously
141/// in the same thread that emits the log record.
142/// When using this processor with the OTLP Exporter, the following exporter
143/// features are supported:
144/// - `grpc-tonic`: This requires TracerProvider to be created within a tokio
145///   runtime. Spans can be emitted from any thread, including tokio runtime
146///   threads.
147/// - `reqwest-blocking-client`: TracerProvider may be created anywhere, but
148///   spans must be emitted from a non-tokio runtime thread.
149/// - `reqwest-client`: TracerProvider may be created anywhere, but spans must be
150///   emitted from a tokio runtime thread.
151///
152/// The OTLP HTTP exporter chooses its default HTTP client from enabled crate
153/// features. That choice is not processor-aware. If you enable async HTTP
154/// clients such as `reqwest-client` or `hyper-client`, ensure this processor is
155/// only used from a thread where those clients can run.
156#[derive(Debug)]
157pub struct SimpleSpanProcessor<T: SpanExporter> {
158    exporter: Mutex<T>,
159}
160
161impl<T: SpanExporter> SimpleSpanProcessor<T> {
162    /// Create a new [SimpleSpanProcessor] using the provided exporter.
163    pub fn new(exporter: T) -> Self {
164        Self {
165            exporter: Mutex::new(exporter),
166        }
167    }
168}
169
170impl<T: SpanExporter> SpanProcessor for SimpleSpanProcessor<T> {
171    fn on_start(&self, _span: &mut Span, _cx: &Context) {
172        // Ignored
173    }
174
175    fn on_end(&self, span: SpanData) {
176        if !span.span_context.is_sampled() {
177            return;
178        }
179
180        let result = self
181            .exporter
182            .lock()
183            .map_err(|_| OTelSdkError::InternalFailure("SimpleSpanProcessor mutex poison".into()))
184            .and_then(|exporter| futures_executor::block_on(exporter.export(vec![span])));
185
186        if let Err(err) = result {
187            // TODO: check error type, and log `error` only if the error is user-actionable, else log `debug`
188            otel_debug!(
189                name: "SimpleProcessor.OnEnd.Error",
190                reason = format!("{:?}", err)
191            );
192        }
193    }
194
195    fn force_flush(&self) -> OTelSdkResult {
196        // Nothing to flush for simple span processor.
197        Ok(())
198    }
199
200    fn shutdown_with_timeout(&self, timeout: Duration) -> OTelSdkResult {
201        if let Ok(exporter) = self.exporter.lock() {
202            exporter.shutdown_with_timeout(timeout)
203        } else {
204            Err(OTelSdkError::InternalFailure(
205                "SimpleSpanProcessor mutex poison at shutdown".into(),
206            ))
207        }
208    }
209
210    fn set_resource(&mut self, resource: &Resource) {
211        if let Ok(mut exporter) = self.exporter.lock() {
212            exporter.set_resource(resource);
213        }
214    }
215}
216
217/// The `BatchSpanProcessor` collects finished spans in a buffer and exports them
218/// in batches to the configured `SpanExporter`. This processor is ideal for
219/// high-throughput environments, as it minimizes the overhead of exporting spans
220/// individually. It uses a **dedicated background thread** to manage and export spans
221/// asynchronously, ensuring that the application's main execution flow is not blocked.
222///
223/// When using this processor with the OTLP Exporter, the following exporter
224/// features are supported:
225/// - `grpc-tonic`: This requires `TracerProvider` to be created within a tokio
226///   runtime.
227/// - `reqwest-blocking-client`: Works with a regular `main` or `tokio::main`.
228///
229/// In other words, async HTTP clients like `reqwest-client` and `hyper-client`
230/// are not supported by this default processor. The OTLP HTTP exporter chooses
231/// its default HTTP client from enabled crate features and cannot tell which
232/// processor will drive it. If your dependency graph enables async HTTP client
233/// features, either pass an explicit blocking client for this processor or use
234/// the experimental async-runtime batch span processor.
235///
236/// # Example
237///
238/// This example demonstrates how to configure and use the `BatchSpanProcessor`
239/// with a custom configuration. Note that a dedicated thread is used internally
240/// to manage the export process.
241///
242/// ```rust
243/// # #[cfg(feature = "testing")]
244/// # {
245/// use opentelemetry::global;
246/// use opentelemetry_sdk::trace::{
247///     BatchSpanProcessor, BatchConfigBuilder, SdkTracerProvider, InMemorySpanExporter,
248/// };
249/// use opentelemetry::trace::Tracer as _;
250/// use opentelemetry::trace::Span;
251/// use std::time::Duration;
252///
253/// // Step 1: Create an exporter (e.g., an In-Memory Exporter for demonstration).
254/// let exporter = InMemorySpanExporter::default();
255///
256/// // Step 2: Configure the BatchSpanProcessor.
257/// let batch_processor = BatchSpanProcessor::builder(exporter)
258///     .with_batch_config(
259///         BatchConfigBuilder::default()
260///             .with_max_queue_size(1024) // Buffer up to 1024 spans.
261///             .with_max_export_batch_size(256) // Export in batches of up to 256 spans.
262///             .with_scheduled_delay(Duration::from_secs(5)) // Export every 5 seconds.
263///             .build(),
264///     )
265///     .build();
266///
267/// // Step 3: Set up a TracerProvider with the configured processor.
268/// let provider = SdkTracerProvider::builder()
269///     .with_span_processor(batch_processor)
270///     .build();
271/// global::set_tracer_provider(provider.clone());
272///
273/// // Step 4: Create spans and record operations.
274/// let tracer = global::tracer("example-tracer");
275/// let mut span = tracer.start("example-span");
276/// span.end(); // Mark the span as completed.
277///
278/// // Step 5: Ensure all spans are flushed before exiting.
279/// provider.shutdown();
280/// # }
281/// ```
282use std::sync::mpsc::sync_channel;
283use std::sync::mpsc::Receiver;
284use std::sync::mpsc::RecvTimeoutError;
285use std::sync::mpsc::SyncSender;
286
287/// Messages exchanged between the main thread and the background thread.
288#[allow(clippy::large_enum_variant)]
289#[derive(Debug)]
290enum BatchMessage {
291    //ExportSpan(SpanData),
292    ExportSpan(Arc<AtomicBool>),
293    ForceFlush(SyncSender<OTelSdkResult>),
294    Shutdown(SyncSender<OTelSdkResult>),
295    SetResource(Arc<Resource>),
296}
297
298/// The `BatchSpanProcessor` collects finished spans in a buffer and exports them
299/// in batches to the configured `SpanExporter`. This processor is ideal for
300/// high-throughput environments, as it minimizes the overhead of exporting spans
301/// individually. It uses a **dedicated background thread** to manage and export spans
302/// asynchronously, ensuring that the application's main execution flow is not blocked.
303///
304/// This processor supports the following configurations:
305/// - **Queue size**: Maximum number of spans that can be buffered.
306/// - **Batch size**: Maximum number of spans to include in a single export.
307/// - **Scheduled delay**: Frequency at which the batch is exported.
308///
309/// When using this processor with the OTLP Exporter, the following exporter
310/// features are supported:
311/// - `grpc-tonic`: Requires `TracerProvider` to be created within a tokio runtime.
312/// - `reqwest-blocking-client`: Works with a regular `main` or `tokio::main`.
313///
314/// In other words, async HTTP clients like `reqwest-client` and `hyper-client`
315/// are not supported by this default processor. The OTLP HTTP exporter chooses
316/// its default HTTP client from enabled crate features and cannot tell which
317/// processor will drive it. If your dependency graph enables async HTTP client
318/// features, either pass an explicit blocking client for this processor or use
319/// the experimental async-runtime batch span processor.
320///
321/// `BatchSpanProcessor` buffers spans in memory and exports them in batches. An
322/// export is triggered when `max_export_batch_size` is reached or every
323/// `scheduled_delay` milliseconds. Users can explicitly trigger an export using
324/// the `force_flush` method. Shutdown also triggers an export of all buffered
325/// spans and is recommended to be called before the application exits to ensure
326/// all buffered spans are exported.
327///
328/// **Warning**: When using tokio's current-thread runtime, `shutdown()`, which
329/// is a blocking call ,should not be called from your main thread. This can
330/// cause deadlock. Instead, call `shutdown()` from a separate thread or use
331/// tokio's `spawn_blocking`.
332///
333#[derive(Debug)]
334pub struct BatchSpanProcessor {
335    span_sender: SyncSender<SpanData>, // Data channel to store spans
336    message_sender: SyncSender<BatchMessage>, // Control channel to store control messages.
337    handle: Mutex<Option<thread::JoinHandle<()>>>,
338    forceflush_timeout: Duration,
339    export_span_message_sent: Arc<AtomicBool>,
340    current_batch_size: Arc<AtomicUsize>,
341    max_export_batch_size: usize,
342    dropped_spans_count: AtomicUsize,
343    max_queue_size: usize,
344}
345
346impl BatchSpanProcessor {
347    /// Creates a new instance of `BatchSpanProcessor`.
348    pub fn new<E>(
349        mut exporter: E,
350        config: BatchConfig,
351        //max_queue_size: usize,
352        //scheduled_delay: Duration,
353        //shutdown_timeout: Duration,
354    ) -> Self
355    where
356        E: SpanExporter + Send + 'static,
357    {
358        let (span_sender, span_receiver) = sync_channel::<SpanData>(config.max_queue_size);
359        let (message_sender, message_receiver) = sync_channel::<BatchMessage>(64); // Is this a reasonable bound?
360        let max_queue_size = config.max_queue_size;
361        let max_export_batch_size = config.max_export_batch_size;
362        let current_batch_size = Arc::new(AtomicUsize::new(0));
363        let current_batch_size_for_thread = current_batch_size.clone();
364
365        let handle = thread::Builder::new()
366            .name("OpenTelemetry.Traces.BatchProcessor".to_string())
367            .spawn(move || {
368                let _suppress_guard = Context::enter_telemetry_suppressed_scope();
369                otel_debug!(
370                    name: "BatchSpanProcessor.ThreadStarted",
371                    interval_in_millisecs = config.scheduled_delay.as_millis(),
372                    max_export_batch_size = config.max_export_batch_size,
373                    max_queue_size = config.max_queue_size,
374                );
375                let mut spans = Vec::with_capacity(config.max_export_batch_size);
376                let mut last_export_time = Instant::now();
377                let current_batch_size = current_batch_size_for_thread;
378                loop {
379                    let remaining_time_option = config
380                        .scheduled_delay
381                        .checked_sub(last_export_time.elapsed());
382                    let remaining_time = match remaining_time_option {
383                        Some(remaining_time) => remaining_time,
384                        None => config.scheduled_delay,
385                    };
386                    match message_receiver.recv_timeout(remaining_time) {
387                        Ok(message) => match message {
388                            BatchMessage::ExportSpan(export_span_message_sent) => {
389                                // Reset the export span message sent flag now it has has been processed.
390                                export_span_message_sent.store(false, Ordering::Relaxed);
391                                otel_debug!(
392                                    name: "BatchSpanProcessor.ExportingDueToBatchSize",
393                                );
394                                let _ = Self::get_spans_and_export(
395                                    &span_receiver,
396                                    &exporter,
397                                    &mut spans,
398                                    &mut last_export_time,
399                                    &current_batch_size,
400                                    &config,
401                                );
402                            }
403                            BatchMessage::ForceFlush(sender) => {
404                                otel_debug!(name: "BatchSpanProcessor.ExportingDueToForceFlush");
405                                let result = Self::get_spans_and_export(
406                                    &span_receiver,
407                                    &exporter,
408                                    &mut spans,
409                                    &mut last_export_time,
410                                    &current_batch_size,
411                                    &config,
412                                );
413                                let _ = sender.send(result);
414                            }
415                            BatchMessage::Shutdown(sender) => {
416                                otel_debug!(name: "BatchSpanProcessor.ExportingDueToShutdown");
417                                let result = Self::get_spans_and_export(
418                                    &span_receiver,
419                                    &exporter,
420                                    &mut spans,
421                                    &mut last_export_time,
422                                    &current_batch_size,
423                                    &config,
424                                );
425                                let _ = exporter.shutdown();
426                                let _ = sender.send(result);
427
428                                otel_debug!(
429                                    name: "BatchSpanProcessor.ThreadExiting",
430                                    reason = "ShutdownRequested"
431                                );
432                                //
433                                // break out the loop and return from the current background thread.
434                                //
435                                break;
436                            }
437                            BatchMessage::SetResource(resource) => {
438                                exporter.set_resource(&resource);
439                            }
440                        },
441                        Err(RecvTimeoutError::Timeout) => {
442                            otel_debug!(
443                                name: "BatchSpanProcessor.ExportingDueToTimer",
444                            );
445
446                            let _ = Self::get_spans_and_export(
447                                &span_receiver,
448                                &exporter,
449                                &mut spans,
450                                &mut last_export_time,
451                                &current_batch_size,
452                                &config,
453                            );
454                        }
455                        Err(RecvTimeoutError::Disconnected) => {
456                            // Channel disconnected, only thing to do is break
457                            // out (i.e exit the thread)
458                            otel_debug!(
459                                name: "BatchSpanProcessor.ThreadExiting",
460                                reason = "MessageSenderDisconnected"
461                            );
462                            break;
463                        }
464                    }
465                }
466                otel_debug!(
467                    name: "BatchSpanProcessor.ThreadStopped"
468                );
469            })
470            .expect("Failed to spawn thread"); //TODO: Handle thread spawn failure
471
472        Self {
473            span_sender,
474            message_sender,
475            handle: Mutex::new(Some(handle)),
476            forceflush_timeout: Duration::from_secs(5), // TODO: make this configurable
477            dropped_spans_count: AtomicUsize::new(0),
478            max_queue_size,
479            export_span_message_sent: Arc::new(AtomicBool::new(false)),
480            current_batch_size,
481            max_export_batch_size,
482        }
483    }
484
485    /// builder
486    pub fn builder<E>(exporter: E) -> BatchSpanProcessorBuilder<E>
487    where
488        E: SpanExporter + Send + 'static,
489    {
490        BatchSpanProcessorBuilder {
491            exporter,
492            config: BatchConfig::default(),
493        }
494    }
495
496    // This method gets up to `max_export_batch_size` amount of spans from the channel and exports them.
497    // It returns the result of the export operation.
498    // It expects the spans vec to be empty when it's called.
499    #[inline]
500    fn get_spans_and_export<E>(
501        spans_receiver: &Receiver<SpanData>,
502        exporter: &E,
503        spans: &mut Vec<SpanData>,
504        last_export_time: &mut Instant,
505        current_batch_size: &AtomicUsize,
506        config: &BatchConfig,
507    ) -> OTelSdkResult
508    where
509        E: SpanExporter + Send + Sync + 'static,
510    {
511        let target = current_batch_size.load(Ordering::Acquire); // `target` is used to determine the stopping criteria for exporting spans.
512        let mut result = OTelSdkResult::Ok(());
513        let mut total_exported_spans: usize = 0;
514
515        while target > 0 && total_exported_spans < target {
516            let batch_limit = config
517                .max_export_batch_size
518                .min(target - total_exported_spans);
519
520            // Get up to the remaining target batch size from the channel and push them to the spans vec
521            while let Ok(span) = spans_receiver.try_recv() {
522                spans.push(span);
523                if spans.len() == batch_limit {
524                    break;
525                }
526            }
527
528            let count_of_spans = spans.len(); // Count of spans that will be exported
529            if count_of_spans == 0 {
530                break;
531            }
532            total_exported_spans += count_of_spans;
533
534            result = Self::export_batch_sync(exporter, spans, last_export_time); // This method clears the spans vec after exporting
535
536            current_batch_size.fetch_sub(count_of_spans, Ordering::AcqRel);
537        }
538        result
539    }
540
541    #[allow(clippy::vec_box)]
542    fn export_batch_sync<E>(
543        exporter: &E,
544        batch: &mut Vec<SpanData>,
545        last_export_time: &mut Instant,
546    ) -> OTelSdkResult
547    where
548        E: SpanExporter + ?Sized,
549    {
550        *last_export_time = Instant::now();
551
552        if batch.is_empty() {
553            return OTelSdkResult::Ok(());
554        }
555
556        // Splitting off batch clears the existing batch capacity, and is ready
557        // for re-use in the next export. The newly returned vec! from split_off
558        // is passed to the exporter.
559        // TODO: Compared to Logs, this requires new allocation for vec for
560        // every export. See if this can be optimized by
561        // *not* requiring ownership in the exporter.
562        let export = exporter.export(batch.split_off(0));
563        let export_result = futures_executor::block_on(export);
564
565        match export_result {
566            Ok(_) => OTelSdkResult::Ok(()),
567            Err(err) => {
568                otel_error!(
569                    name: "BatchSpanProcessor.ExportError",
570                    error = format!("{}", err)
571                );
572                OTelSdkResult::Err(err)
573            }
574        }
575    }
576}
577
578impl SpanProcessor for BatchSpanProcessor {
579    /// Handles span start.
580    fn on_start(&self, _span: &mut Span, _cx: &Context) {
581        // Ignored
582    }
583
584    /// Handles span end.
585    fn on_end(&self, span: SpanData) {
586        let result = self.span_sender.try_send(span);
587
588        // match for result and handle each separately
589        match result {
590            Ok(_) => {
591                // Successfully sent the span to the data channel.
592                // Increment the current batch size and check if it has reached
593                // the max export batch size.
594                if self.current_batch_size.fetch_add(1, Ordering::AcqRel) + 1
595                    >= self.max_export_batch_size
596                {
597                    // Check if the a control message for exporting spans is
598                    // already sent to the worker thread. If not, send a control
599                    // message to export spans. `export_span_message_sent` is set
600                    // to false ONLY when the worker thread has processed the
601                    // control message.
602
603                    if !self.export_span_message_sent.load(Ordering::Relaxed) {
604                        // This is a cost-efficient check as atomic load
605                        // operations do not require exclusive access to cache
606                        // line. Perform atomic swap to
607                        // `export_span_message_sent` ONLY when the atomic load
608                        // operation above returns false. Atomic
609                        // swap/compare_exchange operations require exclusive
610                        // access to cache line on most processor architectures.
611                        // We could have used compare_exchange as well here, but
612                        // it's more verbose than swap.
613                        if !self.export_span_message_sent.swap(true, Ordering::Relaxed) {
614                            match self.message_sender.try_send(BatchMessage::ExportSpan(
615                                self.export_span_message_sent.clone(),
616                            )) {
617                                Ok(_) => {
618                                    // Control message sent successfully.
619                                }
620                                Err(_err) => {
621                                    // TODO: Log error If the control message
622                                    // could not be sent, reset the
623                                    // `export_span_message_sent` flag.
624                                    self.export_span_message_sent
625                                        .store(false, Ordering::Relaxed);
626                                }
627                            }
628                        }
629                    }
630                }
631            }
632            Err(std::sync::mpsc::TrySendError::Full(_)) => {
633                // Increment dropped spans count. The first time we have to drop
634                // a span, emit a warning.
635                if self.dropped_spans_count.fetch_add(1, Ordering::Relaxed) == 0 {
636                    otel_warn!(name: "BatchSpanProcessor.SpanDroppingStarted",
637                        message = "BatchSpanProcessor dropped a Span 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 spans dropped.");
638                }
639            }
640            Err(std::sync::mpsc::TrySendError::Disconnected(_)) => {
641                // Given background thread is the only receiver, and it's
642                // disconnected, it indicates the thread is shutdown
643                otel_warn!(
644                    name: "BatchSpanProcessor.OnEnd.AfterShutdown",
645                    message = "Spans are being emitted even after Shutdown. This indicates incorrect lifecycle management of TracerProvider in application. Spans will not be exported."
646                );
647            }
648        }
649    }
650
651    /// Flushes all pending spans.
652    fn force_flush(&self) -> OTelSdkResult {
653        let (sender, receiver) = std::sync::mpsc::sync_channel(1);
654        match self
655            .message_sender
656            .try_send(BatchMessage::ForceFlush(sender))
657        {
658            Ok(_) => receiver
659                .recv_timeout(self.forceflush_timeout)
660                .map_err(|err| {
661                    if err == std::sync::mpsc::RecvTimeoutError::Timeout {
662                        OTelSdkError::Timeout(self.forceflush_timeout)
663                    } else {
664                        OTelSdkError::InternalFailure(format!("{err}"))
665                    }
666                })?,
667            Err(std::sync::mpsc::TrySendError::Full(_)) => {
668                // If the control message could not be sent, emit a warning.
669                otel_debug!(
670                    name: "BatchSpanProcessor.ForceFlush.ControlChannelFull",
671                    message = "Control message to flush the worker thread could not be sent as the control channel is full. This can occur if user repeatedly calls force_flush/shutdown without finishing the previous call."
672                );
673                Err(OTelSdkError::InternalFailure("ForceFlush cannot be performed as Control channel is full. This can occur if user repeatedly calls force_flush/shutdown without finishing the previous call.".into()))
674            }
675            Err(std::sync::mpsc::TrySendError::Disconnected(_)) => {
676                // Given background thread is the only receiver, and it's
677                // disconnected, it indicates the thread is shutdown
678                otel_debug!(
679                    name: "BatchSpanProcessor.ForceFlush.AlreadyShutdown",
680                    message = "ForceFlush invoked after Shutdown. This will not perform Flush and indicates a incorrect lifecycle management in Application."
681                );
682
683                Err(OTelSdkError::AlreadyShutdown)
684            }
685        }
686    }
687
688    /// Shuts down the processor.
689    fn shutdown_with_timeout(&self, timeout: Duration) -> OTelSdkResult {
690        let dropped_spans = self.dropped_spans_count.load(Ordering::Relaxed);
691        let max_queue_size = self.max_queue_size;
692        if dropped_spans > 0 {
693            otel_warn!(
694                name: "BatchSpanProcessor.SpansDropped",
695                dropped_span_count = dropped_spans,
696                max_queue_size = max_queue_size,
697                message = "Spans were dropped due to a queue being full. The count represents the total count of spans dropped in the lifetime of this BatchSpanProcessor. Consider increasing the queue size and/or decrease delay between intervals."
698            );
699        }
700
701        let (sender, receiver) = std::sync::mpsc::sync_channel(1);
702        match self.message_sender.try_send(BatchMessage::Shutdown(sender)) {
703            Ok(_) => {
704                receiver
705                    .recv_timeout(timeout)
706                    .map(|_| {
707                        // join the background thread after receiving back the
708                        // shutdown signal
709                        if let Some(handle) = self.handle.lock().unwrap().take() {
710                            handle.join().unwrap();
711                        }
712                        OTelSdkResult::Ok(())
713                    })
714                    .map_err(|err| match err {
715                        std::sync::mpsc::RecvTimeoutError::Timeout => {
716                            otel_error!(
717                                name: "BatchSpanProcessor.Shutdown.Timeout",
718                                message = "BatchSpanProcessor shutdown timing out."
719                            );
720                            OTelSdkError::Timeout(timeout)
721                        }
722                        _ => {
723                            otel_error!(
724                                name: "BatchSpanProcessor.Shutdown.Error",
725                                error = format!("{}", err)
726                            );
727                            OTelSdkError::InternalFailure(format!("{err}"))
728                        }
729                    })?
730            }
731            Err(std::sync::mpsc::TrySendError::Full(_)) => {
732                // If the control message could not be sent, emit a warning.
733                otel_debug!(
734                    name: "BatchSpanProcessor.Shutdown.ControlChannelFull",
735                    message = "Control message to shutdown the worker thread could not be sent as the control channel is full. This can occur if user repeatedly calls force_flush/shutdown without finishing the previous call."
736                );
737                Err(OTelSdkError::InternalFailure("Shutdown cannot be performed as Control channel is full. This can occur if user repeatedly calls force_flush/shutdown without finishing the previous call.".into()))
738            }
739            Err(std::sync::mpsc::TrySendError::Disconnected(_)) => {
740                // Given background thread is the only receiver, and it's
741                // disconnected, it indicates the thread is shutdown
742                otel_debug!(
743                    name: "BatchSpanProcessor.Shutdown.AlreadyShutdown",
744                    message = "Shutdown is being invoked more than once. This is noop, but indicates a potential issue in the application's lifecycle management."
745                );
746
747                Err(OTelSdkError::AlreadyShutdown)
748            }
749        }
750    }
751
752    /// Set the resource for the processor.
753    fn set_resource(&mut self, resource: &Resource) {
754        let resource = Arc::new(resource.clone());
755        let _ = self
756            .message_sender
757            .try_send(BatchMessage::SetResource(resource));
758    }
759}
760
761/// Builder for `BatchSpanProcessorDedicatedThread`.
762#[derive(Debug, Default)]
763pub struct BatchSpanProcessorBuilder<E>
764where
765    E: SpanExporter + Send + 'static,
766{
767    exporter: E,
768    config: BatchConfig,
769}
770
771impl<E> BatchSpanProcessorBuilder<E>
772where
773    E: SpanExporter + Send + 'static,
774{
775    /// Set the BatchConfig for [BatchSpanProcessorBuilder]
776    pub fn with_batch_config(self, config: BatchConfig) -> Self {
777        BatchSpanProcessorBuilder { config, ..self }
778    }
779
780    /// Build a new instance of `BatchSpanProcessor`.
781    pub fn build(self) -> BatchSpanProcessor {
782        BatchSpanProcessor::new(self.exporter, self.config)
783    }
784}
785
786/// Batch span processor configuration.
787/// Use [`BatchConfigBuilder`] to configure your own instance of [`BatchConfig`].
788#[derive(Debug)]
789pub struct BatchConfig {
790    /// The maximum queue size to buffer spans for delayed processing. If the
791    /// queue gets full it drops the spans. The default value of is 2048.
792    pub(crate) max_queue_size: usize,
793
794    /// The delay interval in milliseconds between two consecutive processing
795    /// of batches. The default value is 5 seconds.
796    pub(crate) scheduled_delay: Duration,
797
798    #[allow(dead_code)]
799    /// The maximum number of spans to process in a single batch. If there are
800    /// more than one batch worth of spans then it processes multiple batches
801    /// of spans one batch after the other without any delay. The default value
802    /// is 512.
803    pub(crate) max_export_batch_size: usize,
804
805    #[allow(dead_code)]
806    /// The maximum duration to export a batch of data.
807    pub(crate) max_export_timeout: Duration,
808
809    #[allow(dead_code)]
810    pub(crate) max_concurrent_exports: usize,
811}
812
813impl Default for BatchConfig {
814    fn default() -> Self {
815        BatchConfigBuilder::default().build()
816    }
817}
818
819/// A builder for creating [`BatchConfig`] instances.
820#[derive(Debug)]
821pub struct BatchConfigBuilder {
822    max_queue_size: usize,
823    scheduled_delay: Duration,
824    max_export_batch_size: usize,
825    max_export_timeout: Duration,
826    max_concurrent_exports: usize,
827}
828
829impl Default for BatchConfigBuilder {
830    /// Create a new [`BatchConfigBuilder`] initialized with default batch config values as per the specs.
831    /// The values are overriden by environment variables if set.
832    /// The supported environment variables are:
833    /// * `OTEL_BSP_MAX_QUEUE_SIZE`
834    /// * `OTEL_BSP_SCHEDULE_DELAY`
835    /// * `OTEL_BSP_MAX_EXPORT_BATCH_SIZE`
836    /// * `OTEL_BSP_EXPORT_TIMEOUT`
837    /// * `OTEL_BSP_MAX_CONCURRENT_EXPORTS`
838    ///
839    /// Note: Programmatic configuration overrides any value set via the environment variable.
840    fn default() -> Self {
841        BatchConfigBuilder {
842            max_queue_size: OTEL_BSP_MAX_QUEUE_SIZE_DEFAULT,
843            scheduled_delay: OTEL_BSP_SCHEDULE_DELAY_DEFAULT,
844            max_export_batch_size: OTEL_BSP_MAX_EXPORT_BATCH_SIZE_DEFAULT,
845            max_export_timeout: OTEL_BSP_EXPORT_TIMEOUT_DEFAULT,
846            max_concurrent_exports: OTEL_BSP_MAX_CONCURRENT_EXPORTS_DEFAULT,
847        }
848        .init_from_env_vars()
849    }
850}
851
852impl BatchConfigBuilder {
853    /// Set max_queue_size for [`BatchConfigBuilder`].
854    /// It's the maximum queue size to buffer spans for delayed processing.
855    /// If the queue gets full it will drops the spans.
856    /// The default value is 2048.
857    ///
858    /// Corresponding environment variable: `OTEL_BSP_MAX_QUEUE_SIZE`.
859    ///
860    /// Note: Programmatically setting this will override any value set via the environment variable.
861    pub fn with_max_queue_size(mut self, max_queue_size: usize) -> Self {
862        self.max_queue_size = max_queue_size;
863        self
864    }
865
866    /// Set max_export_batch_size for [`BatchConfigBuilder`].
867    /// It's the maximum number of spans to process in a single batch. If there are
868    /// more than one batch worth of spans then it processes multiple batches
869    /// of spans one batch after the other without any delay. The default value
870    /// is 512.
871    ///
872    /// Corresponding environment variable: `OTEL_BSP_MAX_EXPORT_BATCH_SIZE`.
873    ///
874    /// Note: Programmatically setting this will override any value set via the environment variable.
875    pub fn with_max_export_batch_size(mut self, max_export_batch_size: usize) -> Self {
876        self.max_export_batch_size = max_export_batch_size;
877        self
878    }
879
880    #[cfg(feature = "experimental_trace_batch_span_processor_with_async_runtime")]
881    /// Set max_concurrent_exports for [`BatchConfigBuilder`].
882    ///
883    /// This value is honored by
884    /// `span_processor_with_async_runtime::BatchSpanProcessor`, where it limits
885    /// the number of concurrent export tasks.
886    ///
887    /// The thread-based `BatchSpanProcessor` exports serially and ignores this
888    /// setting.
889    ///
890    /// Corresponding environment variable: `OTEL_BSP_MAX_CONCURRENT_EXPORTS`.
891    ///
892    /// For concurrent exports, enable
893    /// `experimental_trace_batch_span_processor_with_async_runtime` and use the
894    /// async-runtime processor.
895    ///
896    /// Note: Programmatically setting this will override any value set via the environment variable.
897    pub fn with_max_concurrent_exports(mut self, max_concurrent_exports: usize) -> Self {
898        self.max_concurrent_exports = max_concurrent_exports;
899        self
900    }
901
902    /// Set scheduled_delay_duration for [`BatchConfigBuilder`].
903    /// It's the delay interval in milliseconds between two consecutive processing of batches.
904    /// The default value is 5000 milliseconds.
905    ///
906    /// Corresponding environment variable: `OTEL_BSP_SCHEDULE_DELAY`.
907    ///
908    /// Note: Programmatically setting this will override any value set via the environment variable.
909    pub fn with_scheduled_delay(mut self, scheduled_delay: Duration) -> Self {
910        self.scheduled_delay = scheduled_delay;
911        self
912    }
913
914    /// Set max_export_timeout for [`BatchConfigBuilder`].
915    /// It's the maximum duration to export a batch of data.
916    /// The The default value is 30000 milliseconds.
917    ///
918    /// Corresponding environment variable: `OTEL_BSP_EXPORT_TIMEOUT`.
919    ///
920    /// Note: Programmatically setting this will override any value set via the environment variable.
921    #[cfg(feature = "experimental_trace_batch_span_processor_with_async_runtime")]
922    pub fn with_max_export_timeout(mut self, max_export_timeout: Duration) -> Self {
923        self.max_export_timeout = max_export_timeout;
924        self
925    }
926
927    /// Builds a `BatchConfig` enforcing the following invariants:
928    /// * `max_export_batch_size` must be less than or equal to `max_queue_size`.
929    pub fn build(self) -> BatchConfig {
930        // max export batch size must be less or equal to max queue size.
931        // we set max export batch size to max queue size if it's larger than max queue size.
932        let max_export_batch_size = min(self.max_export_batch_size, self.max_queue_size);
933
934        BatchConfig {
935            max_queue_size: self.max_queue_size,
936            scheduled_delay: self.scheduled_delay,
937            max_export_timeout: self.max_export_timeout,
938            max_concurrent_exports: self.max_concurrent_exports,
939            max_export_batch_size,
940        }
941    }
942
943    fn init_from_env_vars(mut self) -> Self {
944        if let Some(max_concurrent_exports) = env::var(OTEL_BSP_MAX_CONCURRENT_EXPORTS)
945            .ok()
946            .and_then(|max_concurrent_exports| usize::from_str(&max_concurrent_exports).ok())
947        {
948            self.max_concurrent_exports = max_concurrent_exports;
949        }
950
951        if let Some(max_queue_size) = env::var(OTEL_BSP_MAX_QUEUE_SIZE)
952            .ok()
953            .and_then(|queue_size| usize::from_str(&queue_size).ok())
954        {
955            self.max_queue_size = max_queue_size;
956        }
957
958        if let Some(scheduled_delay) = env::var(OTEL_BSP_SCHEDULE_DELAY)
959            .ok()
960            .and_then(|delay| u64::from_str(&delay).ok())
961        {
962            self.scheduled_delay = Duration::from_millis(scheduled_delay);
963        }
964
965        if let Some(max_export_batch_size) = env::var(OTEL_BSP_MAX_EXPORT_BATCH_SIZE)
966            .ok()
967            .and_then(|batch_size| usize::from_str(&batch_size).ok())
968        {
969            self.max_export_batch_size = max_export_batch_size;
970        }
971
972        // max export batch size must be less or equal to max queue size.
973        // we set max export batch size to max queue size if it's larger than max queue size.
974        if self.max_export_batch_size > self.max_queue_size {
975            self.max_export_batch_size = self.max_queue_size;
976        }
977
978        if let Some(max_export_timeout) = env::var(OTEL_BSP_EXPORT_TIMEOUT)
979            .ok()
980            .and_then(|timeout| u64::from_str(&timeout).ok())
981        {
982            self.max_export_timeout = Duration::from_millis(max_export_timeout);
983        }
984
985        self
986    }
987}
988
989#[cfg(all(test, feature = "testing", feature = "trace"))]
990mod tests {
991    // cargo test trace::span_processor::tests:: --features=testing
992    use super::{
993        BatchSpanProcessor, SimpleSpanProcessor, SpanProcessor, OTEL_BSP_EXPORT_TIMEOUT,
994        OTEL_BSP_MAX_EXPORT_BATCH_SIZE, OTEL_BSP_MAX_QUEUE_SIZE, OTEL_BSP_MAX_QUEUE_SIZE_DEFAULT,
995        OTEL_BSP_SCHEDULE_DELAY, OTEL_BSP_SCHEDULE_DELAY_DEFAULT,
996    };
997    use crate::error::OTelSdkResult;
998    use crate::testing::trace::new_test_export_span_data;
999    use crate::trace::span_processor::{
1000        OTEL_BSP_EXPORT_TIMEOUT_DEFAULT, OTEL_BSP_MAX_CONCURRENT_EXPORTS,
1001        OTEL_BSP_MAX_CONCURRENT_EXPORTS_DEFAULT, OTEL_BSP_MAX_EXPORT_BATCH_SIZE_DEFAULT,
1002    };
1003    use crate::trace::InMemorySpanExporterBuilder;
1004    use crate::trace::{BatchConfig, BatchConfigBuilder, SpanEvents, SpanLinks};
1005    use crate::trace::{SpanData, SpanExporter};
1006    use opentelemetry::trace::{SpanContext, SpanId, SpanKind, Status};
1007    use std::fmt::Debug;
1008    use std::time::Duration;
1009
1010    #[test]
1011    fn simple_span_processor_on_end_calls_export() {
1012        let exporter = InMemorySpanExporterBuilder::new().build();
1013        let processor = SimpleSpanProcessor::new(exporter.clone());
1014        let span_data = new_test_export_span_data();
1015        processor.on_end(span_data.clone());
1016        assert_eq!(exporter.get_finished_spans().unwrap()[0], span_data);
1017        let _result = processor.shutdown();
1018    }
1019
1020    #[test]
1021    fn simple_span_processor_on_end_skips_export_if_not_sampled() {
1022        let exporter = InMemorySpanExporterBuilder::new().build();
1023        let processor = SimpleSpanProcessor::new(exporter.clone());
1024        let unsampled = SpanData {
1025            span_context: SpanContext::empty_context(),
1026            parent_span_id: SpanId::INVALID,
1027            parent_span_is_remote: false,
1028            span_kind: SpanKind::Internal,
1029            name: "opentelemetry".into(),
1030            start_time: opentelemetry::time::now(),
1031            end_time: opentelemetry::time::now(),
1032            attributes: Vec::new(),
1033            dropped_attributes_count: 0,
1034            events: SpanEvents::default(),
1035            links: SpanLinks::default(),
1036            status: Status::Unset,
1037            instrumentation_scope: Default::default(),
1038        };
1039        processor.on_end(unsampled);
1040        assert!(exporter.get_finished_spans().unwrap().is_empty());
1041    }
1042
1043    #[test]
1044    fn simple_span_processor_shutdown_calls_shutdown() {
1045        let exporter = InMemorySpanExporterBuilder::new().build();
1046        let processor = SimpleSpanProcessor::new(exporter.clone());
1047        let span_data = new_test_export_span_data();
1048        processor.on_end(span_data.clone());
1049        assert!(!exporter.get_finished_spans().unwrap().is_empty());
1050        let _result = processor.shutdown();
1051        // Assume shutdown is called by ensuring spans are empty in the exporter
1052        assert!(exporter.get_finished_spans().unwrap().is_empty());
1053    }
1054
1055    #[test]
1056    fn test_default_const_values() {
1057        assert_eq!(OTEL_BSP_MAX_QUEUE_SIZE, "OTEL_BSP_MAX_QUEUE_SIZE");
1058        assert_eq!(OTEL_BSP_MAX_QUEUE_SIZE_DEFAULT, 2048);
1059        assert_eq!(OTEL_BSP_SCHEDULE_DELAY, "OTEL_BSP_SCHEDULE_DELAY");
1060        assert_eq!(OTEL_BSP_SCHEDULE_DELAY_DEFAULT.as_millis(), 5000);
1061        assert_eq!(
1062            OTEL_BSP_MAX_EXPORT_BATCH_SIZE,
1063            "OTEL_BSP_MAX_EXPORT_BATCH_SIZE"
1064        );
1065        assert_eq!(OTEL_BSP_MAX_EXPORT_BATCH_SIZE_DEFAULT, 512);
1066        assert_eq!(OTEL_BSP_EXPORT_TIMEOUT, "OTEL_BSP_EXPORT_TIMEOUT");
1067        assert_eq!(OTEL_BSP_EXPORT_TIMEOUT_DEFAULT.as_millis(), 30000);
1068    }
1069
1070    #[test]
1071    fn test_default_batch_config_adheres_to_specification() {
1072        let env_vars = vec![
1073            OTEL_BSP_SCHEDULE_DELAY,
1074            OTEL_BSP_EXPORT_TIMEOUT,
1075            OTEL_BSP_MAX_QUEUE_SIZE,
1076            OTEL_BSP_MAX_EXPORT_BATCH_SIZE,
1077            OTEL_BSP_MAX_CONCURRENT_EXPORTS,
1078        ];
1079
1080        let config = temp_env::with_vars_unset(env_vars, BatchConfig::default);
1081
1082        assert_eq!(
1083            config.max_concurrent_exports,
1084            OTEL_BSP_MAX_CONCURRENT_EXPORTS_DEFAULT
1085        );
1086        assert_eq!(config.scheduled_delay, OTEL_BSP_SCHEDULE_DELAY_DEFAULT);
1087        assert_eq!(config.max_export_timeout, OTEL_BSP_EXPORT_TIMEOUT_DEFAULT);
1088        assert_eq!(config.max_queue_size, OTEL_BSP_MAX_QUEUE_SIZE_DEFAULT);
1089        assert_eq!(
1090            config.max_export_batch_size,
1091            OTEL_BSP_MAX_EXPORT_BATCH_SIZE_DEFAULT
1092        );
1093    }
1094
1095    #[test]
1096    fn test_code_based_config_overrides_env_vars() {
1097        let env_vars = vec![
1098            (OTEL_BSP_EXPORT_TIMEOUT, Some("60000")),
1099            (OTEL_BSP_MAX_CONCURRENT_EXPORTS, Some("5")),
1100            (OTEL_BSP_MAX_EXPORT_BATCH_SIZE, Some("1024")),
1101            (OTEL_BSP_MAX_QUEUE_SIZE, Some("4096")),
1102            (OTEL_BSP_SCHEDULE_DELAY, Some("2000")),
1103        ];
1104
1105        temp_env::with_vars(env_vars, || {
1106            let config = BatchConfigBuilder::default()
1107                .with_max_export_batch_size(512)
1108                .with_max_queue_size(2048)
1109                .with_scheduled_delay(Duration::from_millis(1000));
1110            #[cfg(feature = "experimental_trace_batch_span_processor_with_async_runtime")]
1111            let config = {
1112                config
1113                    .with_max_concurrent_exports(10)
1114                    .with_max_export_timeout(Duration::from_millis(2000))
1115            };
1116            let config = config.build();
1117
1118            assert_eq!(config.max_export_batch_size, 512);
1119            assert_eq!(config.max_queue_size, 2048);
1120            assert_eq!(config.scheduled_delay, Duration::from_millis(1000));
1121            #[cfg(feature = "experimental_trace_batch_span_processor_with_async_runtime")]
1122            {
1123                assert_eq!(config.max_concurrent_exports, 10);
1124                assert_eq!(config.max_export_timeout, Duration::from_millis(2000));
1125            }
1126        });
1127    }
1128
1129    #[test]
1130    fn test_batch_config_configurable_by_env_vars() {
1131        let env_vars = vec![
1132            (OTEL_BSP_SCHEDULE_DELAY, Some("2000")),
1133            (OTEL_BSP_EXPORT_TIMEOUT, Some("60000")),
1134            (OTEL_BSP_MAX_QUEUE_SIZE, Some("4096")),
1135            (OTEL_BSP_MAX_EXPORT_BATCH_SIZE, Some("1024")),
1136        ];
1137
1138        let config = temp_env::with_vars(env_vars, BatchConfig::default);
1139
1140        assert_eq!(config.scheduled_delay, Duration::from_millis(2000));
1141        assert_eq!(config.max_export_timeout, Duration::from_millis(60000));
1142        assert_eq!(config.max_queue_size, 4096);
1143        assert_eq!(config.max_export_batch_size, 1024);
1144    }
1145
1146    #[test]
1147    fn test_batch_config_max_export_batch_size_validation() {
1148        let env_vars = vec![
1149            (OTEL_BSP_MAX_QUEUE_SIZE, Some("256")),
1150            (OTEL_BSP_MAX_EXPORT_BATCH_SIZE, Some("1024")),
1151        ];
1152
1153        let config = temp_env::with_vars(env_vars, BatchConfig::default);
1154
1155        assert_eq!(config.max_queue_size, 256);
1156        assert_eq!(config.max_export_batch_size, 256);
1157        assert_eq!(config.scheduled_delay, OTEL_BSP_SCHEDULE_DELAY_DEFAULT);
1158        assert_eq!(config.max_export_timeout, OTEL_BSP_EXPORT_TIMEOUT_DEFAULT);
1159    }
1160
1161    #[test]
1162    fn test_batch_config_with_fields() {
1163        let batch = BatchConfigBuilder::default()
1164            .with_max_export_batch_size(10)
1165            .with_scheduled_delay(Duration::from_millis(10))
1166            .with_max_queue_size(10);
1167        #[cfg(feature = "experimental_trace_batch_span_processor_with_async_runtime")]
1168        let batch = {
1169            batch
1170                .with_max_concurrent_exports(10)
1171                .with_max_export_timeout(Duration::from_millis(10))
1172        };
1173        let batch = batch.build();
1174        assert_eq!(batch.max_export_batch_size, 10);
1175        assert_eq!(batch.scheduled_delay, Duration::from_millis(10));
1176        assert_eq!(batch.max_queue_size, 10);
1177        #[cfg(feature = "experimental_trace_batch_span_processor_with_async_runtime")]
1178        {
1179            assert_eq!(batch.max_concurrent_exports, 10);
1180            assert_eq!(batch.max_export_timeout, Duration::from_millis(10));
1181        }
1182    }
1183
1184    // Helper function to create a default test span
1185    fn create_test_span(name: &str) -> SpanData {
1186        SpanData {
1187            span_context: SpanContext::empty_context(),
1188            parent_span_id: SpanId::INVALID,
1189            parent_span_is_remote: false,
1190            span_kind: SpanKind::Internal,
1191            name: name.to_string().into(),
1192            start_time: opentelemetry::time::now(),
1193            end_time: opentelemetry::time::now(),
1194            attributes: Vec::new(),
1195            dropped_attributes_count: 0,
1196            events: SpanEvents::default(),
1197            links: SpanLinks::default(),
1198            status: Status::Unset,
1199            instrumentation_scope: Default::default(),
1200        }
1201    }
1202
1203    use crate::Resource;
1204    use opentelemetry::{Key, KeyValue, Value};
1205    use std::{
1206        sync::{
1207            atomic::{AtomicUsize, Ordering},
1208            Arc, Mutex,
1209        },
1210        time::Instant,
1211    };
1212
1213    // Mock exporter to test functionality
1214    #[derive(Debug)]
1215    struct MockSpanExporter {
1216        exported_spans: Arc<Mutex<Vec<SpanData>>>,
1217        exported_resource: Arc<Mutex<Option<Resource>>>,
1218    }
1219
1220    impl MockSpanExporter {
1221        fn new() -> Self {
1222            Self {
1223                exported_spans: Arc::new(Mutex::new(Vec::new())),
1224                exported_resource: Arc::new(Mutex::new(None)),
1225            }
1226        }
1227    }
1228
1229    impl SpanExporter for MockSpanExporter {
1230        async fn export(&self, batch: Vec<SpanData>) -> OTelSdkResult {
1231            let exported_spans = self.exported_spans.clone();
1232            exported_spans.lock().unwrap().extend(batch);
1233            Ok(())
1234        }
1235
1236        fn shutdown(&self) -> OTelSdkResult {
1237            Ok(())
1238        }
1239        fn set_resource(&mut self, resource: &Resource) {
1240            let mut exported_resource = self.exported_resource.lock().unwrap();
1241            *exported_resource = Some(resource.clone());
1242        }
1243    }
1244
1245    #[test]
1246    fn batchspanprocessor_handles_on_end() {
1247        let exporter = MockSpanExporter::new();
1248        let exporter_shared = exporter.exported_spans.clone();
1249        let config = BatchConfigBuilder::default()
1250            .with_max_queue_size(10)
1251            .with_max_export_batch_size(10)
1252            .with_scheduled_delay(Duration::from_secs(5))
1253            .build();
1254        let processor = BatchSpanProcessor::new(exporter, config);
1255
1256        let test_span = create_test_span("test_span");
1257        processor.on_end(test_span.clone());
1258
1259        // Wait for flush interval to ensure the span is processed
1260        std::thread::sleep(Duration::from_secs(6));
1261
1262        let exported_spans = exporter_shared.lock().unwrap();
1263        assert_eq!(exported_spans.len(), 1);
1264        assert_eq!(exported_spans[0].name, "test_span");
1265    }
1266
1267    #[test]
1268    fn batchspanprocessor_force_flush() {
1269        let exporter = MockSpanExporter::new();
1270        let exporter_shared = exporter.exported_spans.clone(); // Shared access to verify exported spans
1271        let config = BatchConfigBuilder::default()
1272            .with_max_queue_size(10)
1273            .with_max_export_batch_size(10)
1274            .with_scheduled_delay(Duration::from_secs(5))
1275            .build();
1276        let processor = BatchSpanProcessor::new(exporter, config);
1277
1278        // Create a test span and send it to the processor
1279        let test_span = create_test_span("force_flush_span");
1280        processor.on_end(test_span.clone());
1281
1282        // Call force_flush to immediately export the spans
1283        let flush_result = processor.force_flush();
1284        assert!(flush_result.is_ok(), "Force flush failed unexpectedly");
1285
1286        // Verify the exported spans in the mock exporter
1287        let exported_spans = exporter_shared.lock().unwrap();
1288        assert_eq!(
1289            exported_spans.len(),
1290            1,
1291            "Unexpected number of exported spans"
1292        );
1293        assert_eq!(exported_spans[0].name, "force_flush_span");
1294    }
1295
1296    #[test]
1297    fn batchspanprocessor_does_not_overdrain_unaccounted_spans() {
1298        let exporter = MockSpanExporter::new();
1299        let exported_spans = exporter.exported_spans.clone();
1300        let (sender, receiver) = std::sync::mpsc::sync_channel(4);
1301        let current_batch_size = AtomicUsize::new(1);
1302        let config = BatchConfigBuilder::default()
1303            .with_max_queue_size(4)
1304            .with_max_export_batch_size(4)
1305            .build();
1306        let mut spans = Vec::with_capacity(config.max_export_batch_size);
1307        let mut last_export_time = Instant::now();
1308
1309        sender.send(create_test_span("counted")).unwrap();
1310        sender.send(create_test_span("unaccounted")).unwrap();
1311
1312        let result = BatchSpanProcessor::get_spans_and_export(
1313            &receiver,
1314            &exporter,
1315            &mut spans,
1316            &mut last_export_time,
1317            &current_batch_size,
1318            &config,
1319        );
1320
1321        assert!(result.is_ok(), "export should succeed");
1322        assert_eq!(
1323            current_batch_size.load(Ordering::Relaxed),
1324            0,
1325            "helper should only subtract the counted span"
1326        );
1327        assert_eq!(
1328            exported_spans.lock().unwrap().len(),
1329            1,
1330            "helper should export at most the target batch size snapshot"
1331        );
1332        assert!(
1333            receiver.try_recv().is_ok(),
1334            "one span should remain queued for a later export cycle"
1335        );
1336    }
1337
1338    #[test]
1339    fn batchspanprocessor_shutdown() {
1340        // Setup exporter and processor - following the same pattern as test_batch_shutdown from logs
1341        let exporter = InMemorySpanExporterBuilder::new()
1342            .keep_records_on_shutdown()
1343            .build();
1344        let processor = BatchSpanProcessor::new(exporter.clone(), BatchConfig::default());
1345
1346        let record = create_test_span("test_span");
1347
1348        processor.on_end(record);
1349        processor.force_flush().unwrap();
1350        processor.shutdown().unwrap();
1351
1352        // todo: expect to see errors here. How should we assert this?
1353        processor.on_end(create_test_span("after_shutdown_span"));
1354
1355        assert_eq!(1, exporter.get_finished_spans().unwrap().len());
1356        assert!(exporter.is_shutdown_called());
1357    }
1358
1359    #[test]
1360    fn batchspanprocessor_handles_dropped_spans() {
1361        // This test verifies that BSP drops spans when the queue is full.
1362
1363        #[derive(Debug)]
1364        struct SlowExporter {
1365            exported_count: Arc<std::sync::atomic::AtomicUsize>,
1366        }
1367
1368        impl SpanExporter for SlowExporter {
1369            async fn export(&self, batch: Vec<SpanData>) -> OTelSdkResult {
1370                // Simulate slow export
1371                std::thread::sleep(Duration::from_millis(50));
1372                self.exported_count
1373                    .fetch_add(batch.len(), Ordering::Relaxed);
1374                Ok(())
1375            }
1376
1377            fn shutdown(&self) -> OTelSdkResult {
1378                Ok(())
1379            }
1380
1381            fn set_resource(&mut self, _resource: &Resource) {}
1382        }
1383
1384        let exported_count = Arc::new(std::sync::atomic::AtomicUsize::new(0));
1385        let exporter = SlowExporter {
1386            exported_count: exported_count.clone(),
1387        };
1388
1389        let max_queue_size = 10;
1390        let config = BatchConfigBuilder::default()
1391            .with_max_queue_size(max_queue_size)
1392            .with_max_export_batch_size(5)
1393            .with_scheduled_delay(Duration::from_millis(10))
1394            .build();
1395        let processor = BatchSpanProcessor::new(exporter, config);
1396
1397        // Rapidly send many more spans than the queue can hold
1398        let total_spans_to_send = 100;
1399        for i in 0..total_spans_to_send {
1400            let span = create_test_span(&format!("span_{}", i));
1401            processor.on_end(span);
1402        }
1403
1404        // Force flush any remaining spans - this waits for export to complete
1405        let _ = processor.force_flush();
1406
1407        let dropped = processor.dropped_spans_count.load(Ordering::Relaxed);
1408        let exported = exported_count.load(Ordering::Relaxed);
1409
1410        // Verify that dropped + exported = total (every span is accounted for)
1411        assert_eq!(
1412            dropped + exported,
1413            total_spans_to_send,
1414            "dropped ({}) + exported ({}) should equal total sent ({})",
1415            dropped,
1416            exported,
1417            total_spans_to_send
1418        );
1419
1420        // With 100 spans sent rapidly and a slow exporter, we should have some drops
1421        assert!(
1422            dropped > 0,
1423            "Expected some spans to be dropped due to full queue. Exported: {}",
1424            exported
1425        );
1426    }
1427
1428    #[test]
1429    fn batchspanprocessor_sync_ignores_max_concurrent_exports() {
1430        #[derive(Debug)]
1431        struct TrackingExporter {
1432            active: Arc<AtomicUsize>,
1433            max_inflight: Arc<AtomicUsize>,
1434            export_calls: Arc<AtomicUsize>,
1435            delay: Duration,
1436        }
1437
1438        impl SpanExporter for TrackingExporter {
1439            async fn export(&self, _batch: Vec<SpanData>) -> OTelSdkResult {
1440                self.export_calls.fetch_add(1, Ordering::SeqCst);
1441                let inflight = self.active.fetch_add(1, Ordering::SeqCst) + 1;
1442                self.max_inflight.fetch_max(inflight, Ordering::SeqCst);
1443
1444                std::thread::sleep(self.delay);
1445                self.active.fetch_sub(1, Ordering::SeqCst);
1446                Ok(())
1447            }
1448        }
1449
1450        let active = Arc::new(AtomicUsize::new(0));
1451        let max_inflight = Arc::new(AtomicUsize::new(0));
1452        let export_calls = Arc::new(AtomicUsize::new(0));
1453        let exporter = TrackingExporter {
1454            active: active.clone(),
1455            max_inflight: max_inflight.clone(),
1456            export_calls: export_calls.clone(),
1457            delay: Duration::from_millis(50),
1458        };
1459
1460        let config = BatchConfig {
1461            max_export_batch_size: 1,
1462            max_queue_size: 16,
1463            scheduled_delay: Duration::from_secs(3600),
1464            max_export_timeout: Duration::from_secs(5),
1465            max_concurrent_exports: 4,
1466        };
1467
1468        let processor = BatchSpanProcessor::new(exporter, config);
1469
1470        processor.on_end(new_test_export_span_data());
1471        processor.on_end(new_test_export_span_data());
1472        processor.on_end(new_test_export_span_data());
1473
1474        processor.force_flush().expect("force flush failed");
1475        processor.shutdown().expect("shutdown failed");
1476
1477        assert_eq!(
1478            export_calls.load(Ordering::SeqCst),
1479            3,
1480            "expected three exports for three spans with max_export_batch_size=1"
1481        );
1482        assert_eq!(
1483            max_inflight.load(Ordering::SeqCst),
1484            1,
1485            "sync BatchSpanProcessor should export serially regardless of max_concurrent_exports"
1486        );
1487    }
1488
1489    #[test]
1490    fn validate_span_attributes_exported_correctly() {
1491        let exporter = MockSpanExporter::new();
1492        let exporter_shared = exporter.exported_spans.clone();
1493        let config = BatchConfigBuilder::default().build();
1494        let processor = BatchSpanProcessor::new(exporter, config);
1495
1496        // Create a span with attributes
1497        let mut span_data = create_test_span("attribute_validation");
1498        span_data.attributes = vec![
1499            KeyValue::new("key1", "value1"),
1500            KeyValue::new("key2", "value2"),
1501        ];
1502        processor.on_end(span_data.clone());
1503
1504        // Force flush to export the span
1505        let _ = processor.force_flush();
1506
1507        // Validate the exported attributes
1508        let exported_spans = exporter_shared.lock().unwrap();
1509        assert_eq!(exported_spans.len(), 1);
1510        let exported_span = &exported_spans[0];
1511        assert!(exported_span
1512            .attributes
1513            .contains(&KeyValue::new("key1", "value1")));
1514        assert!(exported_span
1515            .attributes
1516            .contains(&KeyValue::new("key2", "value2")));
1517    }
1518
1519    #[test]
1520    fn batchspanprocessor_sets_and_exports_with_resource() {
1521        let exporter = MockSpanExporter::new();
1522        let exporter_shared = exporter.exported_spans.clone();
1523        let resource_shared = exporter.exported_resource.clone();
1524        let config = BatchConfigBuilder::default().build();
1525        let mut processor = BatchSpanProcessor::new(exporter, config);
1526
1527        // Set a resource for the processor
1528        let resource = Resource::builder_empty()
1529            .with_attributes(vec![KeyValue::new("service.name", "test_service")])
1530            .build();
1531        processor.set_resource(&resource);
1532
1533        // Create a span and send it to the processor
1534        let test_span = create_test_span("resource_test");
1535        processor.on_end(test_span.clone());
1536
1537        // Force flush to ensure the span is exported
1538        let _ = processor.force_flush();
1539
1540        // Validate spans are exported
1541        let exported_spans = exporter_shared.lock().unwrap();
1542        assert_eq!(exported_spans.len(), 1);
1543
1544        // Validate the resource is correctly set in the exporter
1545        let exported_resource = resource_shared.lock().unwrap();
1546        assert!(exported_resource.is_some());
1547        assert_eq!(
1548            exported_resource
1549                .as_ref()
1550                .unwrap()
1551                .get(&Key::new("service.name")),
1552            Some(Value::from("test_service"))
1553        );
1554    }
1555
1556    #[tokio::test(flavor = "current_thread")]
1557    async fn test_batch_processor_current_thread_runtime() {
1558        let exporter = MockSpanExporter::new();
1559        let exporter_shared = exporter.exported_spans.clone();
1560
1561        let config = BatchConfigBuilder::default()
1562            .with_max_queue_size(5)
1563            .with_max_export_batch_size(3)
1564            .build();
1565
1566        let processor = BatchSpanProcessor::new(exporter, config);
1567
1568        for _ in 0..4 {
1569            let span = new_test_export_span_data();
1570            processor.on_end(span);
1571        }
1572
1573        processor.force_flush().unwrap();
1574
1575        let exported_spans = exporter_shared.lock().unwrap();
1576        assert_eq!(exported_spans.len(), 4);
1577    }
1578
1579    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
1580    async fn test_batch_processor_multi_thread_count_1_runtime() {
1581        let exporter = MockSpanExporter::new();
1582        let exporter_shared = exporter.exported_spans.clone();
1583
1584        let config = BatchConfigBuilder::default()
1585            .with_max_queue_size(5)
1586            .with_max_export_batch_size(3)
1587            .build();
1588
1589        let processor = BatchSpanProcessor::new(exporter, config);
1590
1591        for _ in 0..4 {
1592            let span = new_test_export_span_data();
1593            processor.on_end(span);
1594        }
1595
1596        processor.force_flush().unwrap();
1597
1598        let exported_spans = exporter_shared.lock().unwrap();
1599        assert_eq!(exported_spans.len(), 4);
1600    }
1601
1602    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1603    async fn test_batch_processor_multi_thread() {
1604        let exporter = MockSpanExporter::new();
1605        let exporter_shared = exporter.exported_spans.clone();
1606
1607        let config = BatchConfigBuilder::default()
1608            .with_max_queue_size(20)
1609            .with_max_export_batch_size(5)
1610            .build();
1611
1612        // Create the processor with the thread-safe exporter
1613        let processor = Arc::new(BatchSpanProcessor::new(exporter, config));
1614
1615        let mut handles = vec![];
1616        for _ in 0..10 {
1617            let processor_clone = Arc::clone(&processor);
1618            let handle = tokio::spawn(async move {
1619                let span = new_test_export_span_data();
1620                processor_clone.on_end(span);
1621            });
1622            handles.push(handle);
1623        }
1624
1625        for handle in handles {
1626            handle.await.unwrap();
1627        }
1628
1629        processor.force_flush().unwrap();
1630
1631        // Verify exported spans
1632        let exported_spans = exporter_shared.lock().unwrap();
1633        assert_eq!(exported_spans.len(), 10);
1634    }
1635}
opentelemetry_sdk/trace/span_processor.rs

opentelemetry_sdk/trace/
span_processor.rs
