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

Author: anujnegi270

examples/metrics-basic/src/main.rs

@@ -1,5 +1,5 @@
 use opentelemetry::{global, KeyValue};
-use opentelemetry_sdk::error::ShutdownError;
+use opentelemetry_sdk::error::OTelSdkError;
 use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider};
 use opentelemetry_sdk::Resource;
 use std::error::Error;

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

@@ -1,35 +1,35 @@
 use std::sync::Arc;
 
+use crate::metric::MetricsClient;
 use async_trait::async_trait;
 use http::{header::CONTENT_TYPE, Method};
 use opentelemetry::otel_debug;
-use opentelemetry_sdk::error::{ShutdownError, ShutdownResult};
+use opentelemetry_sdk::error::{OTelSdkError, OTelSdkResult};
 use opentelemetry_sdk::metrics::data::ResourceMetrics;
-use opentelemetry_sdk::metrics::{MetricError, MetricResult};
-
-use crate::{metric::MetricsClient, Error};
 
 use super::OtlpHttpClient;
 
 #[async_trait]
 impl MetricsClient for OtlpHttpClient {
-    async fn export(&self, metrics: &mut ResourceMetrics) -> MetricResult<()> {
+    async fn export(&self, metrics: &mut ResourceMetrics) -> OTelSdkResult {
         let client = self
             .client
             .lock()
-            .map_err(Into::into)
+            .map_err(|e| OTelSdkError::InternalFailure(format!("Failed to acquire lock: {e:?}")))
             .and_then(|g| match &*g {
                 Some(client) => Ok(Arc::clone(client)),
-                _ => Err(MetricError::Other("exporter is already shut down".into())),
+                _ => Err(OTelSdkError::AlreadyShutdown),
             })?;
 
-        let (body, content_type) = self.build_metrics_export_body(metrics)?;
+        let (body, content_type) = self.build_metrics_export_body(metrics).map_err(|e| {
+            OTelSdkError::InternalFailure(format!("Failed to serialize metrics: {e:?}"))
+        })?;
         let mut request = http::Request::builder()
             .method(Method::POST)
             .uri(&self.collector_endpoint)
             .header(CONTENT_TYPE, content_type)
             .body(body.into())
-            .map_err(|e| crate::Error::RequestFailed(Box::new(e)))?;
+            .map_err(|e| OTelSdkError::InternalFailure(format!("{e:?}")))?;
 
         for (k, v) in &self.headers {
             request.headers_mut().insert(k.clone(), v.clone());
@@ -39,15 +39,15 @@ impl MetricsClient for OtlpHttpClient {
         client
             .send_bytes(request)
             .await
-            .map_err(|e| MetricError::ExportErr(Box::new(Error::RequestFailed(e))))?;
+            .map_err(|e| OTelSdkError::InternalFailure(format!("{e:?}")))?;
 
         Ok(())
     }
 
-    fn shutdown(&self) -> ShutdownResult {
+    fn shutdown(&self) -> OTelSdkResult {
         self.client
             .lock()
-            .map_err(|e| ShutdownError::InternalFailure(format!("Failed to acquire lock: {}", e)))?
+            .map_err(|e| OTelSdkError::InternalFailure(format!("Failed to acquire lock: {}", e)))?
             .take();
 
         Ok(())

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

@@ -6,9 +6,8 @@ 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::error::{OTelSdkError, OTelSdkResult};
 use opentelemetry_sdk::metrics::data::ResourceMetrics;
-use opentelemetry_sdk::metrics::{MetricError, MetricResult};
 use tonic::{codegen::CompressionEncoding, service::Interceptor, transport::Channel, Request};
 
 use super::BoxInterceptor;
@@ -55,26 +54,28 @@ impl TonicMetricsClient {
 
 #[async_trait]
 impl MetricsClient for TonicMetricsClient {
-    async fn export(&self, metrics: &mut ResourceMetrics) -> MetricResult<()> {
-        let (mut client, metadata, extensions) =
-            self.inner
-                .lock()
-                .map_err(Into::into)
-                .and_then(|mut inner| match &mut *inner {
-                    Some(inner) => {
-                        let (m, e, _) = inner
-                            .interceptor
-                            .call(Request::new(()))
-                            .map_err(|e| {
-                                MetricError::Other(format!(
-                                    "unexpected status while exporting {e:?}"
-                                ))
-                            })?
-                            .into_parts();
-                        Ok((inner.client.clone(), m, e))
-                    }
-                    None => Err(MetricError::Other("exporter is already shut down".into())),
-                })?;
+    async fn export(&self, metrics: &mut ResourceMetrics) -> OTelSdkResult {
+        let (mut client, metadata, extensions) = self
+            .inner
+            .lock()
+            .map_err(|e| OTelSdkError::InternalFailure(format!("Failed to acquire lock: {e:?}")))
+            .and_then(|mut inner| match &mut *inner {
+                Some(inner) => {
+                    let (m, e, _) = inner
+                        .interceptor
+                        .call(Request::new(()))
+                        .map_err(|e| {
+                            OTelSdkError::InternalFailure(format!(
+                                "unexpected status while exporting {e:?}"
+                            ))
+                        })?
+                        .into_parts();
+                    Ok((inner.client.clone(), m, e))
+                }
+                None => Err(OTelSdkError::InternalFailure(
+                    "exporter is already shut down".into(),
+                )),
+            })?;
 
         otel_debug!(name: "TonicsMetricsClient.CallingExport");
 
@@ -85,15 +86,15 @@ impl MetricsClient for TonicMetricsClient {
                 ExportMetricsServiceRequest::from(&*metrics),
             ))
             .await
-            .map_err(crate::Error::from)?;
+            .map_err(|e| OTelSdkError::InternalFailure(format!("{e:?}")))?;
 
         Ok(())
     }
 
-    fn shutdown(&self) -> ShutdownResult {
+    fn shutdown(&self) -> OTelSdkResult {
         self.inner
             .lock()
-            .map_err(|e| ShutdownError::InternalFailure(format!("Failed to acquire lock: {}", e)))?
+            .map_err(|e| OTelSdkError::InternalFailure(format!("Failed to acquire lock: {}", e)))?
             .take();
 
         Ok(())

opentelemetry-otlp/src/metric.rs

@@ -16,7 +16,7 @@ use crate::NoExporterBuilderSet;
 
 use async_trait::async_trait;
 use core::fmt;
-use opentelemetry_sdk::error::ShutdownResult;
+use opentelemetry_sdk::error::OTelSdkResult;
 use opentelemetry_sdk::metrics::MetricResult;
 
 use opentelemetry_sdk::metrics::{
@@ -123,8 +123,8 @@ impl HasHttpConfig for MetricExporterBuilder<HttpExporterBuilderSet> {
 /// An interface for OTLP metrics clients
 #[async_trait]
 pub(crate) trait MetricsClient: fmt::Debug + Send + Sync + 'static {
-    async fn export(&self, metrics: &mut ResourceMetrics) -> MetricResult<()>;
-    fn shutdown(&self) -> ShutdownResult;
+    async fn export(&self, metrics: &mut ResourceMetrics) -> OTelSdkResult;
+    fn shutdown(&self) -> OTelSdkResult;
 }
 
 /// Export metrics in OTEL format.
@@ -141,7 +141,7 @@ impl Debug for MetricExporter {
 
 #[async_trait]
 impl PushMetricExporter for MetricExporter {
-    async fn export(&self, metrics: &mut ResourceMetrics) -> MetricResult<()> {
+    async fn export(&self, metrics: &mut ResourceMetrics) -> OTelSdkResult {
         self.client.export(metrics).await
     }
 
@@ -150,7 +150,7 @@ impl PushMetricExporter for MetricExporter {
         Ok(())
     }
 
-    fn shutdown(&self) -> ShutdownResult {
+    fn shutdown(&self) -> OTelSdkResult {
         self.client.shutdown()
     }
 

opentelemetry-otlp/tests/integration_test/tests/logs.rs

@@ -46,7 +46,11 @@ fn init_logs(is_simple: bool) -> Result<sdklogs::LoggerProvider> {
     Ok(logger_provider)
 }
 
-async fn logs_tokio_helper(is_simple: bool, log_send_outside_rt: bool) -> Result<()> {
+async fn logs_tokio_helper(
+    is_simple: bool,
+    log_send_outside_rt: bool,
+    current_thread: bool,
+) -> Result<()> {
     use crate::{assert_logs_results_contains, init_logs};
     test_utils::start_collector_container().await?;
 
@@ -76,7 +80,14 @@ async fn logs_tokio_helper(is_simple: bool, log_send_outside_rt: bool) -> Result
             info!(target: "my-target",  uuid = expected_uuid.as_str(), "hello from {}. My price is {}.", "banana", 2.99);
         }
     }
-    let _ = logger_provider.shutdown();
+    if current_thread {
+        let _res = tokio::runtime::Handle::current()
+            .spawn_blocking(move || logger_provider.shutdown())
+            .await
+            .unwrap();
+    } else {
+        let _ = logger_provider.shutdown();
+    }
     tokio::time::sleep(Duration::from_secs(5)).await;
     assert_logs_results_contains(test_utils::LOGS_FILE, expected_uuid.as_str())?;
     Ok(())
@@ -175,7 +186,7 @@ mod logtests {
     #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
     #[cfg(any(feature = "tonic-client", feature = "reqwest-blocking-client"))]
     pub async fn logs_batch_tokio_multi_thread() -> Result<()> {
-        logs_tokio_helper(false, false).await
+        logs_tokio_helper(false, false, false).await
     }
 
     // logger initialization - Inside RT
@@ -185,7 +196,7 @@ mod logtests {
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     #[cfg(any(feature = "tonic-client", feature = "reqwest-blocking-client"))]
     pub async fn logs_batch_tokio_multi_with_one_worker() -> Result<()> {
-        logs_tokio_helper(false, false).await
+        logs_tokio_helper(false, false, false).await
     }
 
     // logger initialization - Inside RT
@@ -195,7 +206,7 @@ mod logtests {
     #[tokio::test(flavor = "current_thread")]
     #[cfg(any(feature = "tonic-client", feature = "reqwest-blocking-client"))]
     pub async fn logs_batch_tokio_current() -> Result<()> {
-        logs_tokio_helper(false, false).await
+        logs_tokio_helper(false, false, true).await
     }
 
     // logger initialization - Inside RT
@@ -205,7 +216,7 @@ mod logtests {
     #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
     #[cfg(any(feature = "tonic-client", feature = "reqwest-blocking-client"))]
     pub async fn logs_batch_tokio_log_outside_rt_multi_thread() -> Result<()> {
-        logs_tokio_helper(false, true).await
+        logs_tokio_helper(false, true, false).await
     }
 
     // logger initialization - Inside RT
@@ -215,7 +226,7 @@ mod logtests {
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
     #[cfg(any(feature = "tonic-client", feature = "reqwest-blocking-client"))]
     pub async fn logs_batch_tokio_log_outside_rt_multi_with_one_worker() -> Result<()> {
-        logs_tokio_helper(false, true).await
+        logs_tokio_helper(false, true, false).await
     }
 
     // logger initialization - Inside RT
@@ -225,7 +236,7 @@ mod logtests {
     #[tokio::test(flavor = "current_thread")]
     #[cfg(any(feature = "tonic-client", feature = "reqwest-blocking-client"))]
     pub async fn logs_batch_tokio_log_outside_rt_current_thread() -> Result<()> {
-        logs_tokio_helper(false, true).await
+        logs_tokio_helper(false, true, true).await
     }
 
     // logger initialization  - Inside RT
@@ -310,7 +321,7 @@ mod logtests {
     #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
     #[cfg(feature = "reqwest-blocking-client")]
     pub async fn logs_simple_tokio_multi_thread() -> Result<()> {
-        logs_tokio_helper(true, false).await
+        logs_tokio_helper(true, false, false).await
     }
 
     // logger initialization - Inside RT
@@ -324,7 +335,7 @@ mod logtests {
         feature = "hyper-client"
     ))]
     pub async fn logs_simple_tokio_multi_thread() -> Result<()> {
-        logs_tokio_helper(true, false).await
+        logs_tokio_helper(true, false, false).await
     }
 
     // logger initialization - Inside RT
@@ -338,7 +349,7 @@ mod logtests {
         feature = "hyper-client"
     ))]
     pub async fn logs_simple_tokio_multi_with_one_worker() -> Result<()> {
-        logs_tokio_helper(true, false).await
+        logs_tokio_helper(true, false, false).await
     }
 
     // logger initialization - Inside RT
@@ -353,7 +364,7 @@ mod logtests {
         feature = "hyper-client"
     ))]
     pub async fn logs_simple_tokio_current() -> Result<()> {
-        logs_tokio_helper(true, false).await
+        logs_tokio_helper(true, false, false).await
     }
 }
 ///

opentelemetry-sdk/benches/metric.rs

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

opentelemetry-sdk/src/error.rs

@@ -11,8 +11,8 @@ pub trait ExportError: std::error::Error + Send + Sync + 'static {
 }
 
 #[derive(Error, Debug)]
-/// Errors that can occur during shutdown.
-pub enum ShutdownError {
+/// Errors that can occur during SDK operations export(), force_flush() and shutdown().
+pub enum OTelSdkError {
     /// Shutdown has already been invoked.
     ///
     /// While shutdown is idempotent and calling it multiple times has no
@@ -42,4 +42,4 @@ pub enum ShutdownError {
 }
 
 /// A specialized `Result` type for Shutdown operations.
-pub type ShutdownResult = Result<(), ShutdownError>;
+pub type OTelSdkResult = Result<(), OTelSdkError>;

opentelemetry-sdk/src/metrics/exporter.rs

@@ -1,7 +1,7 @@
 //! Interfaces for exporting metrics
 use async_trait::async_trait;
 
-use crate::error::ShutdownResult;
+use crate::error::OTelSdkResult;
 use crate::metrics::MetricResult;
 
 use crate::metrics::data::ResourceMetrics;
@@ -17,9 +17,8 @@ pub trait PushMetricExporter: Send + Sync + 'static {
     ///
     /// All retry logic must be contained in this function. The SDK does not
     /// implement any retry logic. All errors returned by this function are
-    /// considered unrecoverable and will be reported to a configured error
-    /// Handler.
-    async fn export(&self, metrics: &mut ResourceMetrics) -> MetricResult<()>;
+    /// considered unrecoverable and will be logged.
+    async fn export(&self, metrics: &mut ResourceMetrics) -> OTelSdkResult;
 
     /// Flushes any metric data held by an exporter.
     async fn force_flush(&self) -> MetricResult<()>;
@@ -28,7 +27,7 @@ pub trait PushMetricExporter: Send + Sync + 'static {
     ///
     /// After Shutdown is called, calls to Export will perform no operation and
     /// instead will return an error indicating the shutdown state.
-    fn shutdown(&self) -> ShutdownResult;
+    fn shutdown(&self) -> OTelSdkResult;
 
     /// Access the [Temporality] of the MetricExporter.
     fn temporality(&self) -> Temporality;