Merge branch 'main' into cijothomas/metric-doc1

lalitb · web-flow · commit 13b2689876c8 · 2025-04-30T13:13:05.000-07:00
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -102,7 +102,7 @@ jobs:
         run: |
           cargo install cargo-check-external-types@0.1.13
           cd ${{ matrix.example }}
-          cargo check-external-types --config allowed-external-types.toml
+          cargo check-external-types --all-features --config allowed-external-types.toml
   msrv:
     strategy:
       matrix:
diff --git a/opentelemetry-otlp/CHANGELOG.md b/opentelemetry-otlp/CHANGELOG.md
@@ -3,6 +3,8 @@
 ## vNext
 
 - Update `tonic` dependency version to 0.13
+- Re-export `tonic` types under `tonic_types`
+  [2898](https://github.com/open-telemetry/opentelemetry-rust/pull/2898)
 
 ## 0.29.0
 
diff --git a/opentelemetry-otlp/allowed-external-types.toml b/opentelemetry-otlp/allowed-external-types.toml
@@ -3,20 +3,16 @@
 # This is used with cargo-check-external-types to reduce the surface area of downstream crates from
 # the public API. Ideally this can have a few exceptions as possible.
 allowed_external_types = [
-    "opentelemetry::*",
     "opentelemetry_http::*",
     "opentelemetry_sdk::*",
-    # http is a pre 1.0 crate
-    "http::uri::InvalidUri",
-    "http::header::name::InvalidHeaderName",
-    "http::header::value::InvalidHeaderValue",
-    # prost is a pre 1.0 crate
-    "prost::error::EncodeError",
+    # serde
+    "serde::de::Deserialize",
+    "serde::ser::Serialize",
     # tonic is a pre 1.0 crate
-    "tonic::status::Code",
-    "tonic::status::Status",
     "tonic::metadata::map::MetadataMap",
+    "tonic::transport::channel::tls::ClientTlsConfig",
+    "tonic::transport::tls::Certificate",
+    "tonic::transport::tls::Identity",
     "tonic::transport::channel::Channel",
-    "tonic::transport::error::Error",
     "tonic::service::interceptor::Interceptor",
 ]
diff --git a/opentelemetry-otlp/src/exporter/http/metrics.rs b/opentelemetry-otlp/src/exporter/http/metrics.rs
@@ -9,7 +9,7 @@ use opentelemetry_sdk::metrics::data::ResourceMetrics;
 use super::OtlpHttpClient;
 
 impl MetricsClient for OtlpHttpClient {
-    async fn export(&self, metrics: &mut ResourceMetrics) -> OTelSdkResult {
+    async fn export(&self, metrics: &ResourceMetrics) -> OTelSdkResult {
         let client = self
             .client
             .lock()
diff --git a/opentelemetry-otlp/src/exporter/http/mod.rs b/opentelemetry-otlp/src/exporter/http/mod.rs
@@ -326,11 +326,11 @@ impl OtlpHttpClient {
     #[cfg(feature = "metrics")]
     fn build_metrics_export_body(
         &self,
-        metrics: &mut ResourceMetrics,
+        metrics: &ResourceMetrics,
     ) -> Option<(Vec<u8>, &'static str)> {
         use opentelemetry_proto::tonic::collector::metrics::v1::ExportMetricsServiceRequest;
 
-        let req: ExportMetricsServiceRequest = (&*metrics).into();
+        let req: ExportMetricsServiceRequest = metrics.into();
 
         match self.protocol {
             #[cfg(feature = "http-json")]
diff --git a/opentelemetry-otlp/src/exporter/tonic/metrics.rs b/opentelemetry-otlp/src/exporter/tonic/metrics.rs
@@ -52,7 +52,7 @@ impl TonicMetricsClient {
 }
 
 impl MetricsClient for TonicMetricsClient {
-    async fn export(&self, metrics: &mut ResourceMetrics) -> OTelSdkResult {
+    async fn export(&self, metrics: &ResourceMetrics) -> OTelSdkResult {
         let (mut client, metadata, extensions) = self
             .inner
             .lock()
@@ -81,7 +81,7 @@ impl MetricsClient for TonicMetricsClient {
             .export(Request::from_parts(
                 metadata,
                 extensions,
-                ExportMetricsServiceRequest::from(&*metrics),
+                ExportMetricsServiceRequest::from(metrics),
             ))
             .await
             .map_err(|e| OTelSdkError::InternalFailure(format!("{e:?}")))?;
diff --git a/opentelemetry-otlp/src/lib.rs b/opentelemetry-otlp/src/lib.rs
@@ -450,3 +450,20 @@ pub enum Protocol {
 #[doc(hidden)]
 /// Placeholder type when no exporter pipeline has been configured in telemetry pipeline.
 pub struct NoExporterConfig(());
+
+/// Re-exported types from the `tonic` crate.
+#[cfg(feature = "grpc-tonic")]
+pub mod tonic_types {
+    /// Re-exported types from `tonic::metadata`.
+    pub mod metadata {
+        #[doc(no_inline)]
+        pub use tonic::metadata::MetadataMap;
+    }
+
+    /// Re-exported types from `tonic::transport`.
+    #[cfg(feature = "tls")]
+    pub mod transport {
+        #[doc(no_inline)]
+        pub use tonic::transport::{Certificate, ClientTlsConfig, Identity};
+    }
+}
diff --git a/opentelemetry-otlp/src/metric.rs b/opentelemetry-otlp/src/metric.rs
@@ -123,7 +123,7 @@ impl HasHttpConfig for MetricExporterBuilder<HttpExporterBuilderSet> {
 pub(crate) trait MetricsClient: fmt::Debug + Send + Sync + 'static {
     fn export(
         &self,
-        metrics: &mut ResourceMetrics,
+        metrics: &ResourceMetrics,
     ) -> impl std::future::Future<Output = OTelSdkResult> + Send;
     fn shutdown(&self) -> OTelSdkResult;
 }
@@ -149,7 +149,7 @@ impl Debug for MetricExporter {
 }
 
 impl PushMetricExporter for MetricExporter {
-    async fn export(&self, metrics: &mut ResourceMetrics) -> OTelSdkResult {
+    async fn export(&self, metrics: &ResourceMetrics) -> OTelSdkResult {
         match &self.client {
             #[cfg(feature = "grpc-tonic")]
             SupportedTransportClient::Tonic(client) => client.export(metrics).await,
diff --git a/opentelemetry-sdk/CHANGELOG.md b/opentelemetry-sdk/CHANGELOG.md
@@ -43,6 +43,9 @@ also modified to suppress telemetry before invoking exporters.
 - *Breaking* `Aggregation` enum moved behind feature flag
   "spec_unstable_metrics_views". This was only required when using Views.
   [#2928](https://github.com/open-telemetry/opentelemetry-rust/pull/2928)
+- *Breaking* change, affecting custom `PushMetricExporter` authors:
+  - The `export` method on `PushMetricExporter` now accepts `&ResourceMetrics`
+    instead of `&mut ResourceMetrics`.
 
 ## 0.29.0
 
diff --git a/opentelemetry-sdk/src/metrics/exporter.rs b/opentelemetry-sdk/src/metrics/exporter.rs
@@ -18,7 +18,7 @@ pub trait PushMetricExporter: Send + Sync + 'static {
     /// considered unrecoverable and will be logged.
     fn export(
         &self,
-        metrics: &mut ResourceMetrics,
+        metrics: &ResourceMetrics,
     ) -> impl std::future::Future<Output = OTelSdkResult> + Send;
 
     /// Flushes any metric data held by an exporter.
diff --git a/opentelemetry-sdk/src/metrics/in_memory_exporter.rs b/opentelemetry-sdk/src/metrics/in_memory_exporter.rs
@@ -237,7 +237,7 @@ impl InMemoryMetricExporter {
 }
 
 impl PushMetricExporter for InMemoryMetricExporter {
-    async fn export(&self, metrics: &mut ResourceMetrics) -> OTelSdkResult {
+    async fn export(&self, metrics: &ResourceMetrics) -> OTelSdkResult {
         self.metrics
             .lock()
             .map(|mut metrics_guard| {
diff --git a/opentelemetry-sdk/src/metrics/internal/histogram.rs b/opentelemetry-sdk/src/metrics/internal/histogram.rs
@@ -27,7 +27,10 @@ where
 
         buckets.total += value;
         buckets.count += 1;
-        buckets.counts[index] += 1;
+        if !buckets.counts.is_empty() {
+            buckets.counts[index] += 1;
+        }
+
         if value < buckets.min {
             buckets.min = value;
         }
@@ -96,7 +99,12 @@ impl<T: Number> Histogram<T> {
             bounds.sort_by(|a, b| a.partial_cmp(b).expect("NaNs filtered out"));
         }
 
-        let buckets_count = bounds.len() + 1;
+        let buckets_count = if bounds.is_empty() {
+            0
+        } else {
+            bounds.len() + 1
+        };
+
         Histogram {
             value_map: ValueMap::new(buckets_count, cardinality_limit),
             init_time: AggregateTimeInitiator::default(),
diff --git a/opentelemetry-sdk/src/metrics/mod.rs b/opentelemetry-sdk/src/metrics/mod.rs
@@ -440,6 +440,14 @@ mod tests {
         histogram_aggregation_with_custom_bounds_helper(Temporality::Cumulative);
     }
 
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn histogram_aggregation_with_empty_bounds() {
+        // Run this test with stdout enabled to see output.
+        // cargo test histogram_aggregation_with_empty_bounds --features=testing -- --nocapture
+        histogram_aggregation_with_empty_bounds_helper(Temporality::Delta);
+        histogram_aggregation_with_empty_bounds_helper(Temporality::Cumulative);
+    }
+
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     async fn updown_counter_aggregation_cumulative() {
         // Run this test with stdout enabled to see output.
@@ -2161,6 +2169,55 @@ mod tests {
         assert_eq!(vec![1.0, 2.5, 5.5], data_point.bounds);
         assert_eq!(vec![1, 1, 3, 0], data_point.bucket_counts);
     }
+
+    fn histogram_aggregation_with_empty_bounds_helper(temporality: Temporality) {
+        let mut test_context = TestContext::new(temporality);
+        let histogram = test_context
+            .meter()
+            .u64_histogram("test_histogram")
+            .with_boundaries(vec![])
+            .build();
+        histogram.record(1, &[KeyValue::new("key1", "value1")]);
+        histogram.record(2, &[KeyValue::new("key1", "value1")]);
+        histogram.record(3, &[KeyValue::new("key1", "value1")]);
+        histogram.record(4, &[KeyValue::new("key1", "value1")]);
+        histogram.record(5, &[KeyValue::new("key1", "value1")]);
+
+        test_context.flush_metrics();
+
+        // Assert
+        let MetricData::Histogram(histogram_data) =
+            test_context.get_aggregation::<u64>("test_histogram", None)
+        else {
+            unreachable!()
+        };
+        // Expecting 1 time-series.
+        assert_eq!(histogram_data.data_points.len(), 1);
+        if let Temporality::Cumulative = temporality {
+            assert_eq!(
+                histogram_data.temporality,
+                Temporality::Cumulative,
+                "Should produce cumulative"
+            );
+        } else {
+            assert_eq!(
+                histogram_data.temporality,
+                Temporality::Delta,
+                "Should produce delta"
+            );
+        }
+
+        // find and validate key1=value1 datapoint
+        let data_point =
+            find_histogram_datapoint_with_key_value(&histogram_data.data_points, "key1", "value1")
+                .expect("datapoint with key1=value1 expected");
+
+        assert_eq!(data_point.count, 5);
+        assert_eq!(data_point.sum, 15);
+        assert!(data_point.bounds.is_empty());
+        assert!(data_point.bucket_counts.is_empty());
+    }
+
     fn gauge_aggregation_helper(temporality: Temporality) {
         // Arrange
         let mut test_context = TestContext::new(temporality);
diff --git a/opentelemetry-sdk/src/metrics/periodic_reader.rs b/opentelemetry-sdk/src/metrics/periodic_reader.rs
@@ -411,7 +411,7 @@ impl<E: PushMetricExporter> PeriodicReaderInner<E> {
 
         // Relying on futures executor to execute async call.
         // TODO: Pass timeout to exporter
-        futures_executor::block_on(self.exporter.export(&mut rm))
+        futures_executor::block_on(self.exporter.export(&rm))
     }
 
     fn force_flush(&self) -> OTelSdkResult {
@@ -553,7 +553,7 @@ mod tests {
     }
 
     impl PushMetricExporter for MetricExporterThatFailsOnlyOnFirst {
-        async fn export(&self, _metrics: &mut ResourceMetrics) -> OTelSdkResult {
+        async fn export(&self, _metrics: &ResourceMetrics) -> OTelSdkResult {
             if self.count.fetch_add(1, Ordering::Relaxed) == 0 {
                 Err(OTelSdkError::InternalFailure("export failed".into()))
             } else {
@@ -584,7 +584,7 @@ mod tests {
     }
 
     impl PushMetricExporter for MockMetricExporter {
-        async fn export(&self, _metrics: &mut ResourceMetrics) -> OTelSdkResult {
+        async fn export(&self, _metrics: &ResourceMetrics) -> OTelSdkResult {
             Ok(())
         }
 
diff --git a/opentelemetry-sdk/src/metrics/periodic_reader_with_async_runtime.rs b/opentelemetry-sdk/src/metrics/periodic_reader_with_async_runtime.rs
@@ -259,7 +259,7 @@ impl<E: PushMetricExporter, RT: Runtime> PeriodicReaderWorker<E, RT> {
             message = "Calling exporter's export method with collected metrics.",
             count = self.rm.scope_metrics.len(),
         );
-        let export = self.reader.exporter.export(&mut self.rm);
+        let export = self.reader.exporter.export(&self.rm);
         let timeout = self.runtime.delay(self.timeout);
         pin_mut!(export);
         pin_mut!(timeout);
diff --git a/opentelemetry-stdout/src/metrics/exporter.rs b/opentelemetry-stdout/src/metrics/exporter.rs
@@ -42,7 +42,7 @@ impl fmt::Debug for MetricExporter {
 
 impl PushMetricExporter for MetricExporter {
     /// Write Metrics to stdout
-    async fn export(&self, metrics: &mut ResourceMetrics) -> OTelSdkResult {
+    async fn export(&self, metrics: &ResourceMetrics) -> OTelSdkResult {
         if self.is_shutdown.load(atomic::Ordering::SeqCst) {
             Err(opentelemetry_sdk::error::OTelSdkError::AlreadyShutdown)
         } else {
@@ -235,19 +235,21 @@ fn print_hist_data_points<T: Debug>(data_points: &[HistogramDataPoint<T>]) {
             println!("\t\t\t\t ->  {}: {}", kv.key, kv.value.as_str());
         }
 
-        println!("\t\t\tBuckets");
-        let mut lower_bound = f64::NEG_INFINITY;
-        for (i, &upper_bound) in data_point.bounds.iter().enumerate() {
-            let count = data_point.bucket_counts.get(i).unwrap_or(&0);
-            println!("\t\t\t\t {} to {} : {}", lower_bound, upper_bound, count);
-            lower_bound = upper_bound;
-        }
+        if !data_point.bucket_counts.is_empty() {
+            println!("\t\t\tBuckets");
+            let mut lower_bound = f64::NEG_INFINITY;
+            for (i, &upper_bound) in data_point.bounds.iter().enumerate() {
+                let count = data_point.bucket_counts.get(i).unwrap_or(&0);
+                println!("\t\t\t\t {} to {} : {}", lower_bound, upper_bound, count);
+                lower_bound = upper_bound;
+            }
 
-        let last_count = data_point
-            .bucket_counts
-            .get(data_point.bounds.len())
-            .unwrap_or(&0);
-        println!("\t\t\t\t{} to +Infinity : {}", lower_bound, last_count);
+            let last_count = data_point
+                .bucket_counts
+                .get(data_point.bounds.len())
+                .unwrap_or(&0);
+            println!("\t\t\t\t{} to +Infinity : {}", lower_bound, last_count);
+        }
     }
 }
 
