Consolidate BatchConfig creation and validation via BatchConfigBuilder (#1480)

CosminLazar · web-flow · commit 26600476b3c2 · 2024-01-25T12:17:47.000-05:00
diff --git a/opentelemetry-jaeger/src/exporter/config/agent.rs b/opentelemetry-jaeger/src/exporter/config/agent.rs
@@ -237,11 +237,13 @@ impl AgentPipeline {
     /// # Examples
     /// Set max queue size.
     /// ```rust
-    /// use opentelemetry_sdk::trace::BatchConfig;
+    /// use opentelemetry_sdk::trace::BatchConfigBuilder;
     ///
     /// let pipeline = opentelemetry_jaeger::new_agent_pipeline()
     ///                 .with_batch_processor_config(
-    ///                       BatchConfig::default().with_max_queue_size(200)
+    ///                       BatchConfigBuilder::default()
+    ///                         .with_max_queue_size(200)
+    ///                         .build()
     ///                 );
     ///
     /// ```
diff --git a/opentelemetry-jaeger/src/exporter/config/collector/mod.rs b/opentelemetry-jaeger/src/exporter/config/collector/mod.rs
@@ -380,11 +380,13 @@ impl CollectorPipeline {
     /// # Examples
     /// Set max queue size.
     /// ```rust
-    /// use opentelemetry_sdk::trace::BatchConfig;
+    /// use opentelemetry_sdk::trace::BatchConfigBuilder;
     ///
     /// let pipeline = opentelemetry_jaeger::new_collector_pipeline()
     ///                 .with_batch_processor_config(
-    ///                       BatchConfig::default().with_max_queue_size(200)
+    ///                       BatchConfigBuilder::default()
+    ///                         .with_max_queue_size(200)
+    ///                         .build()
     ///                 );
     ///
     /// ```
diff --git a/opentelemetry-sdk/CHANGELOG.md b/opentelemetry-sdk/CHANGELOG.md
@@ -29,6 +29,8 @@
 - **Breaking** Remove `TextMapCompositePropagator` [#1373](https://github.com/open-telemetry/opentelemetry-rust/pull/1373). Use `TextMapCompositePropagator` in opentelemetry API.
 
 - [#1375](https://github.com/open-telemetry/opentelemetry-rust/pull/1375/) Fix metric collections during PeriodicReader shutdown
+- **Breaking** [#1480](https://github.com/open-telemetry/opentelemetry-rust/pull/1480) Remove fine grained `BatchConfig` configurations from `BatchLogProcessorBuilder` and `BatchSpanProcessorBuilder`. Use `BatchConfigBuilder` to construct a `BatchConfig` instance and pass it using `BatchLogProcessorBuilder::with_batch_config` or `BatchSpanProcessorBuilder::with_batch_config`. 
+- **Breaking** [#1480](https://github.com/open-telemetry/opentelemetry-rust/pull/1480) Remove mutating functions from `BatchConfig`, use `BatchConfigBuilder` to construct a `BatchConfig` instance.
 
 ## v0.21.2
 
diff --git a/opentelemetry-sdk/benches/batch_span_processor.rs b/opentelemetry-sdk/benches/batch_span_processor.rs
@@ -5,7 +5,9 @@ use opentelemetry::trace::{
 use opentelemetry_sdk::export::trace::SpanData;
 use opentelemetry_sdk::runtime::Tokio;
 use opentelemetry_sdk::testing::trace::NoopSpanExporter;
-use opentelemetry_sdk::trace::{BatchSpanProcessor, SpanEvents, SpanLinks, SpanProcessor};
+use opentelemetry_sdk::trace::{
+    BatchConfigBuilder, BatchSpanProcessor, SpanEvents, SpanLinks, SpanProcessor,
+};
 use opentelemetry_sdk::Resource;
 use std::borrow::Cow;
 use std::sync::Arc;
@@ -52,7 +54,11 @@ fn criterion_benchmark(c: &mut Criterion) {
                     rt.block_on(async move {
                         let span_processor =
                             BatchSpanProcessor::builder(NoopSpanExporter::new(), Tokio)
-                                .with_max_queue_size(10_000)
+                                .with_batch_config(
+                                    BatchConfigBuilder::default()
+                                        .with_max_queue_size(10_000)
+                                        .build(),
+                                )
                                 .build();
                         let mut shared_span_processor = Arc::new(span_processor);
                         let mut handles = Vec::with_capacity(10);
diff --git a/opentelemetry-sdk/src/logs/log_processor.rs b/opentelemetry-sdk/src/logs/log_processor.rs
@@ -13,7 +13,7 @@ use opentelemetry::{
     global,
     logs::{LogError, LogResult},
 };
-use std::{env, sync::Mutex};
+use std::{cmp::min, env, sync::Mutex};
 use std::{
     fmt::{self, Debug, Formatter},
     str::FromStr,
@@ -246,7 +246,7 @@ impl<R: RuntimeChannel> BatchLogProcessor<R> {
     {
         BatchLogProcessorBuilder {
             exporter,
-            config: BatchConfig::default(),
+            config: Default::default(),
             runtime,
         }
     }
@@ -276,7 +276,8 @@ where
     }
 }
 
-/// Batch log processor configuration
+/// Batch log processor configuration.
+/// Use [`BatchConfigBuilder`] to configure your own instance of [`BatchConfig`].
 #[derive(Debug)]
 pub struct BatchConfig {
     /// The maximum queue size to buffer logs for delayed processing. If the
@@ -299,55 +300,36 @@ pub struct BatchConfig {
 
 impl Default for BatchConfig {
     fn default() -> Self {
-        let mut config = BatchConfig {
+        BatchConfigBuilder::default().build()
+    }
+}
+
+/// A builder for creating [`BatchConfig`] instances.
+#[derive(Debug)]
+pub struct BatchConfigBuilder {
+    max_queue_size: usize,
+    scheduled_delay: Duration,
+    max_export_batch_size: usize,
+    max_export_timeout: Duration,
+}
+
+impl Default for BatchConfigBuilder {
+    /// Create a new [`BatchConfigBuilder`] initialized with default batch config values as per the specs.
+    /// The values are overriden by environment variables if set.
+    /// For a list of supported environment variables see [Batch LogRecord Processor](https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/configuration/sdk-environment-variables.md#batch-logrecord-processor).
+    fn default() -> Self {
+        BatchConfigBuilder {
             max_queue_size: OTEL_BLRP_MAX_QUEUE_SIZE_DEFAULT,
             scheduled_delay: Duration::from_millis(OTEL_BLRP_SCHEDULE_DELAY_DEFAULT),
             max_export_batch_size: OTEL_BLRP_MAX_EXPORT_BATCH_SIZE_DEFAULT,
             max_export_timeout: Duration::from_millis(OTEL_BLRP_EXPORT_TIMEOUT_DEFAULT),
-        };
-
-        if let Some(max_queue_size) = env::var(OTEL_BLRP_MAX_QUEUE_SIZE)
-            .ok()
-            .and_then(|queue_size| usize::from_str(&queue_size).ok())
-        {
-            config.max_queue_size = max_queue_size;
-        }
-
-        if let Some(max_export_batch_size) = env::var(OTEL_BLRP_MAX_EXPORT_BATCH_SIZE)
-            .ok()
-            .and_then(|batch_size| usize::from_str(&batch_size).ok())
-        {
-            config.max_export_batch_size = max_export_batch_size;
         }
-
-        // max export batch size must be less or equal to max queue size.
-        // we set max export batch size to max queue size if it's larger than max queue size.
-        if config.max_export_batch_size > config.max_queue_size {
-            config.max_export_batch_size = config.max_queue_size;
-        }
-
-        if let Some(scheduled_delay) = env::var(OTEL_BLRP_SCHEDULE_DELAY)
-            .ok()
-            .or_else(|| env::var("OTEL_BLRP_SCHEDULE_DELAY_MILLIS").ok())
-            .and_then(|delay| u64::from_str(&delay).ok())
-        {
-            config.scheduled_delay = Duration::from_millis(scheduled_delay);
-        }
-
-        if let Some(max_export_timeout) = env::var(OTEL_BLRP_EXPORT_TIMEOUT)
-            .ok()
-            .or_else(|| env::var("OTEL_BLRP_EXPORT_TIMEOUT_MILLIS").ok())
-            .and_then(|s| u64::from_str(&s).ok())
-        {
-            config.max_export_timeout = Duration::from_millis(max_export_timeout);
-        }
-
-        config
+        .init_from_env_vars()
     }
 }
 
-impl BatchConfig {
-    /// Set max_queue_size for [`BatchConfig`].
+impl BatchConfigBuilder {
+    /// Set max_queue_size for [`BatchConfigBuilder`].
     /// It's the maximum queue size to buffer logs for delayed processing.
     /// If the queue gets full it will drop the logs.
     /// The default value of is 2048.
@@ -356,23 +338,23 @@ impl BatchConfig {
         self
     }
 
-    /// Set scheduled_delay for [`BatchConfig`].
+    /// Set scheduled_delay for [`BatchConfigBuilder`].
     /// It's the delay interval in milliseconds between two consecutive processing of batches.
     /// The default value is 1000 milliseconds.
     pub fn with_scheduled_delay(mut self, scheduled_delay: Duration) -> Self {
         self.scheduled_delay = scheduled_delay;
         self
     }
 
-    /// Set max_export_timeout for [`BatchConfig`].
+    /// Set max_export_timeout for [`BatchConfigBuilder`].
     /// It's the maximum duration to export a batch of data.
     /// The default value is 30000 milliseconds.
     pub fn with_max_export_timeout(mut self, max_export_timeout: Duration) -> Self {
         self.max_export_timeout = max_export_timeout;
         self
     }
 
-    /// Set max_export_batch_size for [`BatchConfig`].
+    /// Set max_export_batch_size for [`BatchConfigBuilder`].
     /// It's the maximum number of logs to process in a single batch. If there are
     /// more than one batch worth of logs then it processes multiple batches
     /// of logs one batch after the other without any delay.
@@ -381,6 +363,55 @@ impl BatchConfig {
         self.max_export_batch_size = max_export_batch_size;
         self
     }
+
+    /// Builds a `BatchConfig` enforcing the following invariants:
+    /// * `max_export_batch_size` must be less than or equal to `max_queue_size`.
+    pub fn build(self) -> BatchConfig {
+        // max export batch size must be less or equal to max queue size.
+        // we set max export batch size to max queue size if it's larger than max queue size.
+        let max_export_batch_size = min(self.max_export_batch_size, self.max_queue_size);
+
+        BatchConfig {
+            max_queue_size: self.max_queue_size,
+            scheduled_delay: self.scheduled_delay,
+            max_export_timeout: self.max_export_timeout,
+            max_export_batch_size,
+        }
+    }
+
+    fn init_from_env_vars(mut self) -> Self {
+        if let Some(max_queue_size) = env::var(OTEL_BLRP_MAX_QUEUE_SIZE)
+            .ok()
+            .and_then(|queue_size| usize::from_str(&queue_size).ok())
+        {
+            self.max_queue_size = max_queue_size;
+        }
+
+        if let Some(max_export_batch_size) = env::var(OTEL_BLRP_MAX_EXPORT_BATCH_SIZE)
+            .ok()
+            .and_then(|batch_size| usize::from_str(&batch_size).ok())
+        {
+            self.max_export_batch_size = max_export_batch_size;
+        }
+
+        if let Some(scheduled_delay) = env::var(OTEL_BLRP_SCHEDULE_DELAY)
+            .ok()
+            .or_else(|| env::var("OTEL_BLRP_SCHEDULE_DELAY_MILLIS").ok())
+            .and_then(|delay| u64::from_str(&delay).ok())
+        {
+            self.scheduled_delay = Duration::from_millis(scheduled_delay);
+        }
+
+        if let Some(max_export_timeout) = env::var(OTEL_BLRP_EXPORT_TIMEOUT)
+            .ok()
+            .or_else(|| env::var("OTEL_BLRP_EXPORT_TIMEOUT_MILLIS").ok())
+            .and_then(|s| u64::from_str(&s).ok())
+        {
+            self.max_export_timeout = Duration::from_millis(max_export_timeout);
+        }
+
+        self
+    }
 }
 
 /// A builder for creating [`BatchLogProcessor`] instances.
@@ -397,44 +428,6 @@ where
     E: LogExporter + 'static,
     R: RuntimeChannel,
 {
-    /// Set max queue size for batches
-    pub fn with_max_queue_size(self, size: usize) -> Self {
-        let mut config = self.config;
-        config.max_queue_size = size;
-
-        BatchLogProcessorBuilder { config, ..self }
-    }
-
-    /// Set scheduled delay for batches
-    pub fn with_scheduled_delay(self, delay: Duration) -> Self {
-        let mut config = self.config;
-        config.scheduled_delay = delay;
-
-        BatchLogProcessorBuilder { config, ..self }
-    }
-
-    /// Set max timeout for exporting.
-    pub fn with_max_timeout(self, timeout: Duration) -> Self {
-        let mut config = self.config;
-        config.max_export_timeout = timeout;
-
-        BatchLogProcessorBuilder { config, ..self }
-    }
-
-    /// Set max export size for batches, should always less than or equals to max queue size.
-    ///
-    /// If input is larger than max queue size, will lower it to be equal to max queue size
-    pub fn with_max_export_batch_size(self, size: usize) -> Self {
-        let mut config = self.config;
-        if size > config.max_queue_size {
-            config.max_export_batch_size = config.max_queue_size;
-        } else {
-            config.max_export_batch_size = size;
-        }
-
-        BatchLogProcessorBuilder { config, ..self }
-    }
-
     /// Set the BatchConfig for [`BatchLogProcessorBuilder`]
     pub fn with_batch_config(self, config: BatchConfig) -> Self {
         BatchLogProcessorBuilder { config, ..self }
@@ -471,7 +464,7 @@ mod tests {
                 OTEL_BLRP_EXPORT_TIMEOUT_DEFAULT, OTEL_BLRP_MAX_EXPORT_BATCH_SIZE_DEFAULT,
                 OTEL_BLRP_MAX_QUEUE_SIZE_DEFAULT, OTEL_BLRP_SCHEDULE_DELAY_DEFAULT,
             },
-            BatchConfig,
+            BatchConfig, BatchConfigBuilder,
         },
         runtime,
         testing::logs::InMemoryLogsExporter,
@@ -590,11 +583,12 @@ mod tests {
 
     #[test]
     fn test_batch_config_with_fields() {
-        let batch = BatchConfig::default()
+        let batch = BatchConfigBuilder::default()
             .with_max_export_batch_size(1)
             .with_scheduled_delay(Duration::from_millis(2))
             .with_max_export_timeout(Duration::from_millis(3))
-            .with_max_queue_size(4);
+            .with_max_queue_size(4)
+            .build();
 
         assert_eq!(batch.max_export_batch_size, 1);
         assert_eq!(batch.scheduled_delay, Duration::from_millis(2));
@@ -640,11 +634,12 @@ mod tests {
 
     #[test]
     fn test_build_batch_log_processor_builder_with_custom_config() {
-        let expected = BatchConfig::default()
+        let expected = BatchConfigBuilder::default()
             .with_max_export_batch_size(1)
             .with_scheduled_delay(Duration::from_millis(2))
             .with_max_export_timeout(Duration::from_millis(3))
-            .with_max_queue_size(4);
+            .with_max_queue_size(4)
+            .build();
 
         let builder = BatchLogProcessor::builder(InMemoryLogsExporter::default(), runtime::Tokio)
             .with_batch_config(expected);
diff --git a/opentelemetry-sdk/src/logs/mod.rs b/opentelemetry-sdk/src/logs/mod.rs
@@ -7,7 +7,8 @@ mod log_processor;
 pub use config::{config, Config};
 pub use log_emitter::{Builder, Logger, LoggerProvider};
 pub use log_processor::{
-    BatchConfig, BatchLogProcessor, BatchLogProcessorBuilder, LogProcessor, SimpleLogProcessor,
+    BatchConfig, BatchConfigBuilder, BatchLogProcessor, BatchLogProcessorBuilder, LogProcessor,
+    SimpleLogProcessor,
 };
 
 #[cfg(all(test, feature = "testing"))]
diff --git a/opentelemetry-sdk/src/trace/mod.rs b/opentelemetry-sdk/src/trace/mod.rs
@@ -26,7 +26,8 @@ pub use sampler::{Sampler, ShouldSample};
 pub use span::Span;
 pub use span_limit::SpanLimits;
 pub use span_processor::{
-    BatchConfig, BatchSpanProcessor, BatchSpanProcessorBuilder, SimpleSpanProcessor, SpanProcessor,
+    BatchConfig, BatchConfigBuilder, BatchSpanProcessor, BatchSpanProcessorBuilder,
+    SimpleSpanProcessor, SpanProcessor,
 };
 pub use tracer::Tracer;
 
diff --git a/opentelemetry-sdk/src/trace/span_processor.rs b/opentelemetry-sdk/src/trace/span_processor.rs
