Merge branch 'main' into fix-metrics-u64-serialization

Author: lalitb

README.md

@@ -65,9 +65,10 @@ Project versioning information and stability guarantees can be found
 ## Getting Started
 
 ```rust
-use opentelemetry::{
-    global,
-    trace::{Tracer, TracerProvider as _},
+use opentelemetry::trace::{
+    TraceContextExt,
+    Tracer,
+    TracerProvider as _,
 };
 use opentelemetry_sdk::trace::TracerProvider;
 
@@ -92,9 +93,9 @@ The example above requires the following packages:
 ```toml
 # Cargo.toml
 [dependencies]
-opentelemetry = "0.22"
-opentelemetry_sdk = "0.22"
-opentelemetry-stdout = { version = "0.3", features = ["trace"] }
+opentelemetry = "0.27"
+opentelemetry_sdk = "0.27"
+opentelemetry-stdout = { version = "0.27", features = ["trace"] }
 ```
 
 See the [examples](./examples) directory for different integration patterns.

examples/metrics-advanced/src/main.rs

@@ -49,14 +49,12 @@ fn init_meter_provider() -> opentelemetry_sdk::metrics::SdkMeterProvider {
         .with_temporality(Temporality::Delta)
         .build();
 
-    let reader = PeriodicReader::builder(exporter).build();
-
     let resource = Resource::builder()
         .with_service_name("metrics-advanced-example")
         .build();
 
     let provider = SdkMeterProvider::builder()
-        .with_reader(reader)
+        .with_periodic_exporter(exporter)
         .with_resource(resource)
         .with_view(my_view_rename_and_unit)
         .with_view(my_view_drop_attributes)

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;
@@ -9,9 +10,8 @@ fn init_meter_provider() -> opentelemetry_sdk::metrics::SdkMeterProvider {
         // Build exporter using Delta Temporality (Defaults to Temporality::Cumulative)
         // .with_temporality(opentelemetry_sdk::metrics::Temporality::Delta)
         .build();
-    let reader = PeriodicReader::builder(exporter).build();
     let provider = SdkMeterProvider::builder()
-        .with_reader(reader)
+        .with_periodic_exporter(exporter)
         .with_resource(
             Resource::builder()
                 .with_service_name("metrics-basic-example")
@@ -23,7 +23,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 +137,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-otlp/CHANGELOG.md

@@ -13,6 +13,15 @@
 - Remove unnecessarily public trait `opentelemetry_otlp::metrics::MetricsClient`
   and `MetricExporter::new(..)` method. Use
   `MetricExporter::builder()...build()` to obtain `MetricExporter`.
+- The HTTP clients (reqwest, reqwest-blocking, hyper) now support the
+timeout internal configured in below order
+  - Signal specific env variable `OTEL_EXPORTER_OTLP_TRACES_TIMEOUT`, 
+  `OTEL_EXPORTER_OTLP_LOGS_TIMEOUT` or `OTEL_EXPORTER_OTLP_TIMEOUT`.
+  - `OTEL_EXPORTER_OTLP_TIMEOUT` env variable.
+  - `with_http().with_timeout()` API method of
+`LogExporterBuilder` and `SpanExporterBuilder` and `MetricsExporterBuilder`.
+  - The default interval of 10sec is used if none is configured.
+
 
 ## 0.27.0
 

opentelemetry-otlp/examples/basic-otlp-http/src/main.rs

@@ -61,10 +61,8 @@ fn init_metrics() -> Result<opentelemetry_sdk::metrics::SdkMeterProvider, Metric
         .with_endpoint("http://localhost:4318/v1/metrics")
         .build()?;
 
-    let reader = opentelemetry_sdk::metrics::PeriodicReader::builder(exporter).build();
-
     Ok(SdkMeterProvider::builder()
-        .with_reader(reader)
+        .with_periodic_exporter(exporter)
         .with_resource(RESOURCE.clone())
         .build())
 }

opentelemetry-otlp/examples/basic-otlp/src/main.rs

@@ -33,10 +33,9 @@ fn init_traces() -> Result<sdktrace::TracerProvider, TraceError> {
 
 fn init_metrics() -> Result<opentelemetry_sdk::metrics::SdkMeterProvider, MetricError> {
     let exporter = MetricExporter::builder().with_tonic().build()?;
-    let reader = PeriodicReader::builder(exporter).build();
 
     Ok(SdkMeterProvider::builder()
-        .with_reader(reader)
+        .with_periodic_exporter(exporter)
         .with_resource(RESOURCE.clone())
         .build())
 }

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};
 
@@ -43,8 +44,11 @@ impl MetricsClient for OtlpHttpClient {
         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/mod.rs

@@ -44,15 +44,7 @@ mod trace;
 use opentelemetry_http::hyper::HyperClient;
 
 /// Configuration of the http transport
-#[derive(Debug)]
-#[cfg_attr(
-    all(
-        not(feature = "reqwest-client"),
-        not(feature = "reqwest-blocking-client"),
-        not(feature = "hyper-client")
-    ),
-    derive(Default)
-)]
+#[derive(Debug, Default)]
 pub struct HttpConfig {
     /// Select the HTTP client
     client: Option<Arc<dyn HttpClient>>,
@@ -61,44 +53,6 @@ pub struct HttpConfig {
     headers: Option<HashMap<String, String>>,
 }
 
-#[cfg(any(
-    feature = "reqwest-blocking-client",
-    feature = "reqwest-client",
-    feature = "hyper-client"
-))]
-impl Default for HttpConfig {
-    fn default() -> Self {
-        #[cfg(feature = "reqwest-blocking-client")]
-        let default_client = std::thread::spawn(|| {
-            Some(Arc::new(reqwest::blocking::Client::new()) as Arc<dyn HttpClient>)
-        })
-        .join()
-        .expect("creating reqwest::blocking::Client on a new thread not to fail");
-        #[cfg(all(not(feature = "reqwest-blocking-client"), feature = "reqwest-client"))]
-        let default_client = Some(Arc::new(reqwest::Client::new()) as Arc<dyn HttpClient>);
-        #[cfg(all(
-            not(feature = "reqwest-client"),
-            not(feature = "reqwest-blocking-client"),
-            feature = "hyper-client"
-        ))]
-        // TODO - support configuring custom connector and executor
-        let default_client = Some(Arc::new(HyperClient::with_default_connector(
-            Duration::from_secs(10),
-            None,
-        )) as Arc<dyn HttpClient>);
-        #[cfg(all(
-            not(feature = "reqwest-client"),
-            not(feature = "reqwest-blocking-client"),
-            not(feature = "hyper-client")
-        ))]
-        let default_client = None;
-        HttpConfig {
-            client: default_client,
-            headers: None,
-        }
-    }
-}
-
 /// Configuration for the OTLP HTTP exporter.
 ///
 /// ## Examples
@@ -171,11 +125,56 @@ impl HttpExporterBuilder {
             },
             None => self.exporter_config.timeout,
         };
-        let http_client = self
-            .http_config
-            .client
-            .take()
-            .ok_or(crate::Error::NoHttpClient)?;
+
+        #[allow(unused_mut)] // TODO - clippy thinks mut is not needed, but it is
+        let mut http_client = self.http_config.client.take();
+
+        if http_client.is_none() {
+            #[cfg(all(
+                not(feature = "reqwest-client"),
+                not(feature = "reqwest-blocking-client"),
+                feature = "hyper-client"
+            ))]
+            {
+                // TODO - support configuring custom connector and executor
+                http_client = Some(Arc::new(HyperClient::with_default_connector(timeout, None))
+                    as Arc<dyn HttpClient>);
+            }
+            #[cfg(all(
+                not(feature = "hyper-client"),
+                not(feature = "reqwest-blocking-client"),
+                feature = "reqwest-client"
+            ))]
+            {
+                http_client = Some(Arc::new(
+                    reqwest::Client::builder()
+                        .timeout(timeout)
+                        .build()
+                        .unwrap_or_default(),
+                ) as Arc<dyn HttpClient>);
+            }
+            #[cfg(all(
+                not(feature = "hyper-client"),
+                not(feature = "reqwest-client"),
+                feature = "reqwest-blocking-client"
+            ))]
+            {
+                let timeout_clone = timeout;
+                http_client = Some(Arc::new(
+                    std::thread::spawn(move || {
+                        reqwest::blocking::Client::builder()
+                            .timeout(timeout_clone)
+                            .build()
+                            .unwrap_or_else(|_| reqwest::blocking::Client::new())
+                    })
+                    .join()
+                    .unwrap(), // Unwrap thread result
+                ) as Arc<dyn HttpClient>);
+            }
+        }
+
+        let http_client = http_client.ok_or(crate::Error::NoHttpClient)?;
+
         #[allow(clippy::mutable_key_type)] // http headers are not mutated
         let mut headers: HashMap<HeaderName, HeaderValue> = self
             .http_config

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::{
@@ -123,7 +124,7 @@ impl HasHttpConfig for MetricExporterBuilder<HttpExporterBuilderSet> {
 #[async_trait]
 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()
     }
 

opentelemetry-otlp/tests/integration_test/src/lib.rs

@@ -1,4 +1,4 @@
 pub mod logs_asserter;
-pub mod metrics_asserter;
+pub mod metric_helpers;
 pub mod test_utils;
 pub mod trace_asserter;