Merge branch 'main' into anujnegi/disable-name-check

Author: anujnegi270

examples/metrics-basic/src/main.rs

@@ -1,4 +1,5 @@
 use opentelemetry::{global, KeyValue};
+use opentelemetry_sdk::error::ShutdownError;
 use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider};
 use opentelemetry_sdk::Resource;
 use std::error::Error;
@@ -23,7 +24,7 @@ fn init_meter_provider() -> opentelemetry_sdk::metrics::SdkMeterProvider {
 }
 
 #[tokio::main]
-async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
+async fn main() -> Result<(), Box<dyn Error>> {
     // Initialize the MeterProvider with the stdout Exporter.
     let meter_provider = init_meter_provider();
 
@@ -137,9 +138,41 @@ async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
         })
         .build();
 
-    // Metrics are exported by default every 30 seconds when using stdout exporter,
-    // however shutting down the MeterProvider here instantly flushes
-    // the metrics, instead of waiting for the 30 sec interval.
+    // Metrics are exported by default every 30 seconds when using stdout
+    // exporter, however shutting down the MeterProvider here instantly flushes
+    // the metrics, instead of waiting for the 30 sec interval. Shutdown returns
+    // a result, which is bubbled up to the caller The commented code below
+    // demonstrates handling the shutdown result, instead of bubbling up the
+    // error.
     meter_provider.shutdown()?;
+
+    // let shutdown_result = meter_provider.shutdown();
+
+    // Handle the shutdown result.
+    // match shutdown_result {
+    //     Ok(_) => println!("MeterProvider shutdown successfully"),
+    //     Err(e) => {
+    //         match e {
+    //             opentelemetry_sdk::error::ShutdownError::InternalFailure(message) => {
+    //                 // This indicates some internal failure during shutdown. The
+    //                 // error message is intended for logging purposes only and
+    //                 // should not be used to make programmatic decisions.
+    //                 println!("MeterProvider shutdown failed: {}", message)
+    //             }
+    //             opentelemetry_sdk::error::ShutdownError::AlreadyShutdown => {
+    //                 // This indicates some user code tried to shutdown
+    //                 // elsewhere. user need to review their code to ensure
+    //                 // shutdown is called only once.
+    //                 println!("MeterProvider already shutdown")
+    //             }
+    //             opentelemetry_sdk::error::ShutdownError::Timeout(e) => {
+    //                 // This indicates the shutdown timed out, and a good hint to
+    //                 // user to increase the timeout. (Shutdown method does not
+    //                 // allow custom timeout today, but that is temporary)
+    //                 println!("MeterProvider shutdown timed out after {:?}", e)
+    //             }
+    //         }
+    //     }
+    // }
     Ok(())
 }

opentelemetry-http/CHANGELOG.md

@@ -4,6 +4,7 @@
 
 - Bump msrv to 1.75.0.
 - Add "internal-logs" feature flag (enabled by default), and emit internal logs.
+- Add `HttpClient::send_bytes` with `bytes::Bytes` request payload and deprecate old `HttpClient::send` function.
 
 ## 0.27.0
 

opentelemetry-http/src/lib.rs

@@ -55,13 +55,24 @@ pub type HttpError = Box<dyn std::error::Error + Send + Sync + 'static>;
 /// users to bring their choice of HTTP client.
 #[async_trait]
 pub trait HttpClient: Debug + Send + Sync {
-    /// Send the specified HTTP request
+    /// Send the specified HTTP request with `Vec<u8>` payload
     ///
     /// Returns the HTTP response including the status code and body.
     ///
     /// Returns an error if it can't connect to the server or the request could not be completed,
     /// e.g. because of a timeout, infinite redirects, or a loss of connection.
-    async fn send(&self, request: Request<Vec<u8>>) -> Result<Response<Bytes>, HttpError>;
+    #[deprecated(note = "Use `send_bytes` with `Bytes` payload instead.")]
+    async fn send(&self, request: Request<Vec<u8>>) -> Result<Response<Bytes>, HttpError> {
+        self.send_bytes(request.map(Into::into)).await
+    }
+
+    /// Send the specified HTTP request with `Bytes` payload.
+    ///
+    /// Returns the HTTP response including the status code and body.
+    ///
+    /// Returns an error if it can't connect to the server or the request could not be completed,
+    /// e.g. because of a timeout, infinite redirects, or a loss of connection.
+    async fn send_bytes(&self, request: Request<Bytes>) -> Result<Response<Bytes>, HttpError>;
 }
 
 #[cfg(feature = "reqwest")]
@@ -72,7 +83,7 @@ mod reqwest {
 
     #[async_trait]
     impl HttpClient for reqwest::Client {
-        async fn send(&self, request: Request<Vec<u8>>) -> Result<Response<Bytes>, HttpError> {
+        async fn send_bytes(&self, request: Request<Bytes>) -> Result<Response<Bytes>, HttpError> {
             otel_debug!(name: "ReqwestClient.Send");
             let request = request.try_into()?;
             let mut response = self.execute(request).await?.error_for_status()?;
@@ -89,7 +100,7 @@ mod reqwest {
     #[cfg(not(target_arch = "wasm32"))]
     #[async_trait]
     impl HttpClient for reqwest::blocking::Client {
-        async fn send(&self, request: Request<Vec<u8>>) -> Result<Response<Bytes>, HttpError> {
+        async fn send_bytes(&self, request: Request<Bytes>) -> Result<Response<Bytes>, HttpError> {
             otel_debug!(name: "ReqwestBlockingClient.Send");
             let request = request.try_into()?;
             let mut response = self.execute(request)?.error_for_status()?;
@@ -159,7 +170,7 @@ pub mod hyper {
 
     #[async_trait]
     impl HttpClient for HyperClient {
-        async fn send(&self, request: Request<Vec<u8>>) -> Result<Response<Bytes>, HttpError> {
+        async fn send_bytes(&self, request: Request<Bytes>) -> Result<Response<Bytes>, HttpError> {
             otel_debug!(name: "HyperClient.Send");
             let (parts, body) = request.into_parts();
             let mut request = Request::from_parts(parts, Body(Full::from(body)));

opentelemetry-otlp/CHANGELOG.md

@@ -10,6 +10,9 @@
   "reqwest-blocking-client" features as default, to align with the
   specification.
   [2516](https://github.com/open-telemetry/opentelemetry-rust/pull/2516)
+- Remove unnecessarily public trait `opentelemetry_otlp::metrics::MetricsClient`
+  and `MetricExporter::new(..)` method. Use
+  `MetricExporter::builder()...build()` to obtain `MetricExporter`.
 
 ## 0.27.0
 

opentelemetry-otlp/src/exporter/http/logs.rs

@@ -28,7 +28,7 @@ impl LogExporter for OtlpHttpClient {
                 .method(Method::POST)
                 .uri(&self.collector_endpoint)
                 .header(CONTENT_TYPE, content_type)
-                .body(body)
+                .body(body.into())
                 .map_err(|e| crate::Error::RequestFailed(Box::new(e)))?;
 
             for (k, v) in &self.headers {
@@ -37,7 +37,7 @@ impl LogExporter for OtlpHttpClient {
 
             let request_uri = request.uri().to_string();
             otel_debug!(name: "HttpLogsClient.CallingExport");
-            let response = client.send(request).await?;
+            let response = client.send_bytes(request).await?;
 
             if !response.status().is_success() {
                 let error = format!(

opentelemetry-otlp/src/exporter/http/metrics.rs

@@ -3,6 +3,7 @@ use std::sync::Arc;
 use async_trait::async_trait;
 use http::{header::CONTENT_TYPE, Method};
 use opentelemetry::otel_debug;
+use opentelemetry_sdk::error::{ShutdownError, ShutdownResult};
 use opentelemetry_sdk::metrics::data::ResourceMetrics;
 use opentelemetry_sdk::metrics::{MetricError, MetricResult};
 
@@ -27,7 +28,7 @@ impl MetricsClient for OtlpHttpClient {
             .method(Method::POST)
             .uri(&self.collector_endpoint)
             .header(CONTENT_TYPE, content_type)
-            .body(body)
+            .body(body.into())
             .map_err(|e| crate::Error::RequestFailed(Box::new(e)))?;
 
         for (k, v) in &self.headers {
@@ -36,15 +37,18 @@ impl MetricsClient for OtlpHttpClient {
 
         otel_debug!(name: "HttpMetricsClient.CallingExport");
         client
-            .send(request)
+            .send_bytes(request)
             .await
             .map_err(|e| MetricError::ExportErr(Box::new(Error::RequestFailed(e))))?;
 
         Ok(())
     }
 
-    fn shutdown(&self) -> MetricResult<()> {
-        let _ = self.client.lock()?.take();
+    fn shutdown(&self) -> ShutdownResult {
+        self.client
+            .lock()
+            .map_err(|e| ShutdownError::InternalFailure(format!("Failed to acquire lock: {}", e)))?
+            .take();
 
         Ok(())
     }

opentelemetry-otlp/src/exporter/http/trace.rs

@@ -30,7 +30,7 @@ impl SpanExporter for OtlpHttpClient {
             .method(Method::POST)
             .uri(&self.collector_endpoint)
             .header(CONTENT_TYPE, content_type)
-            .body(body)
+            .body(body.into())
         {
             Ok(req) => req,
             Err(e) => {
@@ -48,7 +48,7 @@ impl SpanExporter for OtlpHttpClient {
         Box::pin(async move {
             let request_uri = request.uri().to_string();
             otel_debug!(name: "HttpTracesClient.CallingExport");
-            let response = client.send(request).await?;
+            let response = client.send_bytes(request).await?;
 
             if !response.status().is_success() {
                 let error = format!(

opentelemetry-otlp/src/exporter/tonic/metrics.rs

@@ -6,6 +6,7 @@ use opentelemetry::otel_debug;
 use opentelemetry_proto::tonic::collector::metrics::v1::{
     metrics_service_client::MetricsServiceClient, ExportMetricsServiceRequest,
 };
+use opentelemetry_sdk::error::{ShutdownError, ShutdownResult};
 use opentelemetry_sdk::metrics::data::ResourceMetrics;
 use opentelemetry_sdk::metrics::{MetricError, MetricResult};
 use tonic::{codegen::CompressionEncoding, service::Interceptor, transport::Channel, Request};
@@ -89,8 +90,11 @@ impl MetricsClient for TonicMetricsClient {
         Ok(())
     }
 
-    fn shutdown(&self) -> MetricResult<()> {
-        let _ = self.inner.lock()?.take();
+    fn shutdown(&self) -> ShutdownResult {
+        self.inner
+            .lock()
+            .map_err(|e| ShutdownError::InternalFailure(format!("Failed to acquire lock: {}", e)))?
+            .take();
 
         Ok(())
     }

opentelemetry-otlp/src/metric.rs

@@ -16,6 +16,7 @@ use crate::NoExporterBuilderSet;
 
 use async_trait::async_trait;
 use core::fmt;
+use opentelemetry_sdk::error::ShutdownResult;
 use opentelemetry_sdk::metrics::MetricResult;
 
 use opentelemetry_sdk::metrics::{
@@ -121,9 +122,9 @@ impl HasHttpConfig for MetricExporterBuilder<HttpExporterBuilderSet> {
 
 /// An interface for OTLP metrics clients
 #[async_trait]
-pub trait MetricsClient: fmt::Debug + Send + Sync + 'static {
+pub(crate) trait MetricsClient: fmt::Debug + Send + Sync + 'static {
     async fn export(&self, metrics: &mut ResourceMetrics) -> MetricResult<()>;
-    fn shutdown(&self) -> MetricResult<()>;
+    fn shutdown(&self) -> ShutdownResult;
 }
 
 /// Export metrics in OTEL format.
@@ -149,7 +150,7 @@ impl PushMetricExporter for MetricExporter {
         Ok(())
     }
 
-    fn shutdown(&self) -> MetricResult<()> {
+    fn shutdown(&self) -> ShutdownResult {
         self.client.shutdown()
     }
 
@@ -165,7 +166,7 @@ impl MetricExporter {
     }
 
     /// Create a new metrics exporter
-    pub fn new(client: impl MetricsClient, temporality: Temporality) -> MetricExporter {
+    pub(crate) fn new(client: impl MetricsClient, temporality: Temporality) -> MetricExporter {
         MetricExporter {
             client: Box::new(client),
             temporality,

opentelemetry-sdk/benches/metric.rs

@@ -7,6 +7,7 @@ use opentelemetry::{
     Key, KeyValue,
 };
 use opentelemetry_sdk::{
+    error::ShutdownResult,
     metrics::{
         data::ResourceMetrics, new_view, reader::MetricReader, Aggregation, Instrument,
         InstrumentKind, ManualReader, MetricResult, Pipeline, SdkMeterProvider, Stream,
@@ -31,7 +32,7 @@ impl MetricReader for SharedReader {
         self.0.force_flush()
     }
 
-    fn shutdown(&self) -> MetricResult<()> {
+    fn shutdown(&self) -> ShutdownResult {
         self.0.shutdown()
     }