Merge branch 'main' into fix-json-deser-bytes

Author: adriangb

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,16 +141,16 @@ 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
     }
 
-    async fn force_flush(&self) -> MetricResult<()> {
+    async fn force_flush(&self) -> OTelSdkResult {
         // this component is stateless
         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-otlp/tests/integration_test/tests/metrics.rs

@@ -69,10 +69,11 @@ mod metrictests {
         );
 
         // In tokio::current_thread flavor, shutdown must be done in a separate thread
-        let _res = Handle::current()
+        let shutdown_resut = Handle::current()
             .spawn_blocking(move || meter_provider.shutdown())
             .await
             .unwrap();
+        assert!(shutdown_resut.is_ok());
         // We still need to sleep, to give otel-collector a chance to flush to disk
         std::thread::sleep(SLEEP_DURATION);
 
@@ -101,7 +102,8 @@ mod metrictests {
             ],
         );
 
-        meter_provider.shutdown()?;
+        let shutdown_resut = meter_provider.shutdown();
+        assert!(shutdown_resut.is_ok());
         // We still need to sleep, to give otel-collector a chance to flush to disk
         std::thread::sleep(SLEEP_DURATION);
 
@@ -127,7 +129,8 @@ mod metrictests {
             ],
         );
 
-        meter_provider.shutdown()?;
+        let shutdown_resut = meter_provider.shutdown();
+        assert!(shutdown_resut.is_ok());
         // We still need to sleep, to give otel-collector a chance to flush to disk
         std::thread::sleep(SLEEP_DURATION);
 

opentelemetry-proto/src/transform/logs.rs

@@ -223,11 +223,11 @@ mod tests {
     use opentelemetry::logs::LogRecord as _;
     use opentelemetry::logs::Logger as _;
     use opentelemetry::logs::LoggerProvider as _;
+    use opentelemetry::time::now;
     use opentelemetry::InstrumentationScope;
     use opentelemetry_sdk::logs::LogProcessor;
     use opentelemetry_sdk::logs::{LogResult, LoggerProvider};
     use opentelemetry_sdk::{logs::LogBatch, logs::LogRecord, Resource};
-    use std::time::SystemTime;
 
     #[derive(Debug)]
     struct MockProcessor;
@@ -254,8 +254,8 @@ mod tests {
             .build()
             .logger("test");
         let mut logrecord = logger.create_log_record();
-        logrecord.set_timestamp(SystemTime::now());
-        logrecord.set_observed_timestamp(SystemTime::now());
+        logrecord.set_timestamp(now());
+        logrecord.set_observed_timestamp(now());
         let instrumentation =
             InstrumentationScope::builder(instrumentation_name.to_string()).build();
         (logrecord, instrumentation)

opentelemetry-proto/src/transform/trace.rs

@@ -193,6 +193,7 @@ pub mod tonic {
 mod tests {
     use crate::tonic::common::v1::any_value::Value;
     use crate::transform::common::tonic::ResourceAttributesWithSchema;
+    use opentelemetry::time::now;
     use opentelemetry::trace::{
         SpanContext, SpanId, SpanKind, Status, TraceFlags, TraceId, TraceState,
     };
@@ -202,7 +203,7 @@ mod tests {
     use opentelemetry_sdk::trace::SpanData;
     use opentelemetry_sdk::trace::{SpanEvents, SpanLinks};
     use std::borrow::Cow;
-    use std::time::{Duration, SystemTime};
+    use std::time::Duration;
 
     fn create_test_span_data(instrumentation_name: &'static str) -> SpanData {
         let span_context = SpanContext::new(
@@ -218,8 +219,8 @@ mod tests {
             parent_span_id: SpanId::from_u64(0),
             span_kind: SpanKind::Internal,
             name: Cow::Borrowed("test_span"),
-            start_time: SystemTime::now(),
-            end_time: SystemTime::now() + Duration::from_secs(1),
+            start_time: now(),
+            end_time: now() + Duration::from_secs(1),
             attributes: vec![KeyValue::new("key", "value")],
             dropped_attributes_count: 0,
             events: SpanEvents::default(),