Merge main branch, keeping local changes

Author: cijothomas

opentelemetry-otlp/CHANGELOG.md

@@ -5,8 +5,10 @@
 - Update `tonic` dependency version to 0.13
 - Re-export `tonic` types under `tonic_types`
   [2898](https://github.com/open-telemetry/opentelemetry-rust/pull/2898)
-- It is now possible to add links to a `Span` via the `SpanRef` that you get from
-  a `Context`. [2959](https://github.com/open-telemetry/opentelemetry-rust/pull/2959)
+- Publicly re-exported `MetricExporterBuilder`, `SpanExporterBuilder`, and
+  `LogExporterBuilder` types, enabling users to directly reference and use these
+  builder types for metrics, traces, and logs exporters.
+  [2966](https://github.com/open-telemetry/opentelemetry-rust/pull/2966)
 
 ## 0.29.0
 

opentelemetry-sdk/CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## vNext
 
+- It is now possible to add links to a `Span` via the `SpanRef` that you get from
+  a `Context`. [2959](https://github.com/open-telemetry/opentelemetry-rust/pull/2959)
 - **Feature**: Added context based telemetry suppression. [#2868](https://github.com/open-telemetry/opentelemetry-rust/pull/2868)
   - `SdkLogger`, `SdkTracer` modified to respect telemetry suppression based on
 `Context`. In other words, if the current context has telemetry suppression

opentelemetry-sdk/src/metrics/instrument.rs

@@ -259,6 +259,146 @@ fn validate_bucket_boundaries(boundaries: &[f64]) -> Result<(), String> {
     Ok(())
 }
 
+/// A builder for creating Stream objects.
+///
+/// # Example
+///
+/// ```
+/// use opentelemetry_sdk::metrics::{Aggregation, Stream};
+/// use opentelemetry::Key;
+///
+/// let stream = Stream::builder()
+///     .with_name("my_stream")
+///     .with_aggregation(Aggregation::Sum)
+///     .with_cardinality_limit(100)
+///     .build()
+///     .unwrap();
+/// ```
+#[derive(Default, Debug)]
+#[non_exhaustive]
+pub struct StreamBuilder {
+    name: Option<Cow<'static, str>>,
+    description: Option<Cow<'static, str>>,
+    unit: Option<Cow<'static, str>>,
+    aggregation: Option<Aggregation>,
+    allowed_attribute_keys: Option<Arc<HashSet<Key>>>,
+    cardinality_limit: Option<usize>,
+}
+
+impl StreamBuilder {
+    /// Create a new stream builder with default values.
+    pub(crate) fn new() -> Self {
+        StreamBuilder::default()
+    }
+
+    /// Set the stream name. If this is not set, name provide while creating the instrument will be used.
+    pub fn with_name(mut self, name: impl Into<Cow<'static, str>>) -> Self {
+        self.name = Some(name.into());
+        self
+    }
+
+    /// Set the stream description. If this is not set, description provided while creating the instrument will be used.
+    pub fn with_description(mut self, description: impl Into<Cow<'static, str>>) -> Self {
+        self.description = Some(description.into());
+        self
+    }
+
+    /// Set the stream unit. If this is not set, unit provided while creating the instrument will be used.
+    pub fn with_unit(mut self, unit: impl Into<Cow<'static, str>>) -> Self {
+        self.unit = Some(unit.into());
+        self
+    }
+
+    #[cfg(feature = "spec_unstable_metrics_views")]
+    /// Set the stream aggregation. This is used to customize the aggregation.
+    /// If not set, the default aggregation based on the instrument kind will be used.
+    pub fn with_aggregation(mut self, aggregation: Aggregation) -> Self {
+        self.aggregation = Some(aggregation);
+        self
+    }
+
+    #[cfg(feature = "spec_unstable_metrics_views")]
+    /// Set the stream allowed attribute keys.
+    ///
+    /// Any attribute recorded for the stream with a key not in this set will be
+    /// dropped. If the set is empty, all attributes will be dropped, if `None` all
+    /// attributes will be kept.
+    pub fn with_allowed_attribute_keys(
+        mut self,
+        attribute_keys: impl IntoIterator<Item = Key>,
+    ) -> Self {
+        self.allowed_attribute_keys = Some(Arc::new(attribute_keys.into_iter().collect()));
+        self
+    }
+
+    /// Set the stream cardinality limit. If this is not set, the default limit of 2000 will be used.
+    pub fn with_cardinality_limit(mut self, limit: usize) -> Self {
+        self.cardinality_limit = Some(limit);
+        self
+    }
+
+    /// Build a new Stream instance using the configuration in this builder.
+    ///
+    /// # Returns
+    ///
+    /// A Result containing the new Stream instance or an error if the build failed.
+    pub fn build(self) -> Result<Stream, Box<dyn Error>> {
+        // TODO: Add same validation as already done while
+        // creating instruments. It is better to move validation logic
+        // to a common helper and call it from both places.
+        // The current implementations does a basic validation
+        // only to close the overall API design.
+
+        // if name is provided, it must not be empty
+        if let Some(name) = &self.name {
+            if name.is_empty() {
+                return Err("Stream name must not be empty".into());
+            }
+        }
+
+        // if cardinality limit is provided, it must be greater than 0
+        if let Some(limit) = self.cardinality_limit {
+            if limit == 0 {
+                return Err("Cardinality limit must be greater than 0".into());
+            }
+        }
+
+        // If the aggregation is set to ExplicitBucketHistogram, validate the bucket boundaries.
+        if let Some(Aggregation::ExplicitBucketHistogram { boundaries, .. }) = &self.aggregation {
+            validate_bucket_boundaries(boundaries)?;
+        }
+
+        Ok(Stream {
+            name: self.name,
+            description: self.description,
+            unit: self.unit,
+            aggregation: self.aggregation,
+            allowed_attribute_keys: self.allowed_attribute_keys,
+            cardinality_limit: self.cardinality_limit,
+        })
+    }
+}
+
+fn validate_bucket_boundaries(boundaries: &[f64]) -> Result<(), String> {
+    // Validate boundaries do not contain f64::NAN, f64::INFINITY, or f64::NEG_INFINITY
+    for boundary in boundaries {
+        if boundary.is_nan() || boundary.is_infinite() {
+            return Err(
+                "Bucket boundaries must not contain NaN, Infinity, or -Infinity".to_string(),
+            );
+        }
+    }
+
+    // validate that buckets are sorted and non-duplicate
+    for i in 1..boundaries.len() {
+        if boundaries[i] <= boundaries[i - 1] {
+            return Err("Bucket boundaries must be sorted and non-duplicate".to_string());
+        }
+    }
+
+    Ok(())
+}
+
 /// Describes the stream of data an instrument produces.
 ///
 /// # Example