Merge branch 'main' into cijothomas/028migration

Author: cijothomas

opentelemetry-appender-tracing/benches/logs.rs

@@ -16,7 +16,7 @@
 use criterion::{criterion_group, criterion_main, Criterion};
 use opentelemetry::InstrumentationScope;
 use opentelemetry_appender_tracing::layer as tracing_layer;
-use opentelemetry_sdk::logs::LogResult;
+use opentelemetry_sdk::error::OTelSdkResult;
 use opentelemetry_sdk::logs::{LogBatch, LogExporter};
 use opentelemetry_sdk::logs::{LogProcessor, SdkLogRecord, SdkLoggerProvider};
 use opentelemetry_sdk::Resource;
@@ -36,8 +36,8 @@ impl LogExporter for NoopExporter {
     fn export(
         &self,
         _batch: LogBatch<'_>,
-    ) -> impl std::future::Future<Output = LogResult<()>> + Send {
-        async { LogResult::Ok(()) }
+    ) -> impl std::future::Future<Output = OTelSdkResult> + Send {
+        async { OTelSdkResult::Ok(()) }
     }
 
     fn event_enabled(&self, _: opentelemetry::logs::Severity, _: &str, _: &str) -> bool {
@@ -61,11 +61,11 @@ impl<E: LogExporter> LogProcessor for NoopProcessor<E> {
         // no-op
     }
 
-    fn force_flush(&self) -> LogResult<()> {
+    fn force_flush(&self) -> OTelSdkResult {
         Ok(())
     }
 
-    fn shutdown(&self) -> LogResult<()> {
+    fn shutdown(&self) -> OTelSdkResult {
         Ok(())
     }
 

opentelemetry-appender-tracing/src/layer.rs

@@ -233,9 +233,10 @@ mod tests {
     use opentelemetry::trace::TracerProvider;
     use opentelemetry::trace::{TraceContextExt, TraceFlags, Tracer};
     use opentelemetry::{logs::AnyValue, Key};
+    use opentelemetry_sdk::error::OTelSdkResult;
     use opentelemetry_sdk::logs::InMemoryLogExporter;
     use opentelemetry_sdk::logs::{LogBatch, LogExporter};
-    use opentelemetry_sdk::logs::{LogResult, SdkLogRecord, SdkLoggerProvider};
+    use opentelemetry_sdk::logs::{SdkLogRecord, SdkLoggerProvider};
     use opentelemetry_sdk::trace::{Sampler, SdkTracerProvider};
     use tracing::{error, warn};
     use tracing_subscriber::prelude::__tracing_subscriber_SubscriberExt;
@@ -269,7 +270,7 @@ mod tests {
         fn export(
             &self,
             _batch: LogBatch<'_>,
-        ) -> impl std::future::Future<Output = LogResult<()>> + Send {
+        ) -> impl std::future::Future<Output = OTelSdkResult> + Send {
             async {
                 // This will cause a deadlock as the export itself creates a log
                 // while still within the lock of the SimpleLogProcessor.
@@ -345,7 +346,7 @@ mod tests {
 
         // Act
         error!(name: "my-event-name", target: "my-system", event_id = 20, user_name = "otel", user_email = "[email protected]");
-        logger_provider.force_flush();
+        assert!(logger_provider.force_flush().is_ok());
 
         // Assert TODO: move to helper methods
         let exported_logs = exporter
@@ -439,7 +440,7 @@ mod tests {
             (trace_id, span_id)
         });
 
-        logger_provider.force_flush();
+        assert!(logger_provider.force_flush().is_ok());
 
         // Assert TODO: move to helper methods
         let exported_logs = exporter
@@ -553,7 +554,7 @@ mod tests {
             });
         });
 
-        logger_provider.force_flush();
+        assert!(logger_provider.force_flush().is_ok());
 
         let logs = exporter.get_emitted_logs().expect("No emitted logs");
         assert_eq!(logs.len(), 2);
@@ -593,7 +594,7 @@ mod tests {
 
         // Act
         log::error!(target: "my-system", "log from log crate");
-        logger_provider.force_flush();
+        assert!(logger_provider.force_flush().is_ok());
 
         // Assert TODO: move to helper methods
         let exported_logs = exporter
@@ -672,7 +673,7 @@ mod tests {
             (trace_id, span_id)
         });
 
-        logger_provider.force_flush();
+        assert!(logger_provider.force_flush().is_ok());
 
         // Assert TODO: move to helper methods
         let exported_logs = exporter

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

@@ -1,60 +1,67 @@
-use std::sync::Arc;
-
+use super::OtlpHttpClient;
 use http::{header::CONTENT_TYPE, Method};
 use opentelemetry::otel_debug;
+use opentelemetry_sdk::error::{OTelSdkError, OTelSdkResult};
 use opentelemetry_sdk::logs::{LogBatch, LogExporter};
-use opentelemetry_sdk::logs::{LogError, LogResult};
-
-use super::OtlpHttpClient;
 
 impl LogExporter for OtlpHttpClient {
     #[allow(clippy::manual_async_fn)]
     fn export(
         &self,
         batch: LogBatch<'_>,
-    ) -> impl std::future::Future<Output = LogResult<()>> + Send {
+    ) -> impl std::future::Future<Output = OTelSdkResult> + Send {
         async move {
             let client = self
                 .client
                 .lock()
-                .map_err(|e| LogError::Other(e.to_string().into()))
-                .and_then(|g| match &*g {
-                    Some(client) => Ok(Arc::clone(client)),
-                    _ => Err(LogError::Other("exporter is already shut down".into())),
-                })?;
+                .map_err(|e| OTelSdkError::InternalFailure(format!("Mutex lock failed: {}", e)))?
+                .clone()
+                .ok_or(OTelSdkError::AlreadyShutdown)?;
+
+            let (body, content_type) = self
+                .build_logs_export_body(batch)
+                .map_err(|e| OTelSdkError::InternalFailure(e.to_string()))?;
 
-            let (body, content_type) = { self.build_logs_export_body(batch)? };
             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(e.to_string()))?;
 
             for (k, v) in &self.headers {
                 request.headers_mut().insert(k.clone(), v.clone());
             }
 
             let request_uri = request.uri().to_string();
             otel_debug!(name: "HttpLogsClient.CallingExport");
-            let response = client.send_bytes(request).await?;
-
+            let response = client
+                .send_bytes(request)
+                .await
+                .map_err(|e| OTelSdkError::InternalFailure(format!("{e:?}")))?;
             if !response.status().is_success() {
                 let error = format!(
                     "OpenTelemetry logs export failed. Url: {}, Status Code: {}, Response: {:?}",
-                    response.status().as_u16(),
                     request_uri,
+                    response.status().as_u16(),
                     response.body()
                 );
-                return Err(LogError::Other(error.into()));
+                return Err(OTelSdkError::InternalFailure(error));
             }
-
             Ok(())
         }
     }
 
-    fn shutdown(&mut self) {
-        let _ = self.client.lock().map(|mut c| c.take());
+    fn shutdown(&mut self) -> OTelSdkResult {
+        let mut client_guard = self.client.lock().map_err(|e| {
+            OTelSdkError::InternalFailure(format!("Failed to acquire client lock: {}", e))
+        })?;
+
+        if client_guard.take().is_none() {
+            return Err(OTelSdkError::AlreadyShutdown);
+        }
+
+        Ok(())
     }
 
     fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) {

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

@@ -3,8 +3,8 @@ use opentelemetry::otel_debug;
 use opentelemetry_proto::tonic::collector::logs::v1::{
     logs_service_client::LogsServiceClient, ExportLogsServiceRequest,
 };
+use opentelemetry_sdk::error::{OTelSdkError, OTelSdkResult};
 use opentelemetry_sdk::logs::{LogBatch, LogExporter};
-use opentelemetry_sdk::logs::{LogError, LogResult};
 use tonic::{codegen::CompressionEncoding, service::Interceptor, transport::Channel, Request};
 
 use opentelemetry_proto::transform::logs::tonic::group_logs_by_resource_and_scope;
@@ -60,7 +60,7 @@ impl LogExporter for TonicLogsClient {
     fn export(
         &self,
         batch: LogBatch<'_>,
-    ) -> impl std::future::Future<Output = LogResult<()>> + Send {
+    ) -> impl std::future::Future<Output = OTelSdkResult> + Send {
         async move {
             let (mut client, metadata, extensions) = match &self.inner {
                 Some(inner) => {
@@ -69,11 +69,11 @@ impl LogExporter for TonicLogsClient {
                         .lock()
                         .await // tokio::sync::Mutex doesn't return a poisoned error, so we can safely use the interceptor here
                         .call(Request::new(()))
-                        .map_err(|e| LogError::Other(Box::new(e)))?
+                        .map_err(|e| OTelSdkError::InternalFailure(format!("error: {:?}", e)))?
                         .into_parts();
                     (inner.client.clone(), m, e)
                 }
-                None => return Err(LogError::Other("exporter is already shut down".into())),
+                None => return Err(OTelSdkError::AlreadyShutdown),
             };
 
             let resource_logs = group_logs_by_resource_and_scope(batch, &self.resource);
@@ -87,13 +87,16 @@ impl LogExporter for TonicLogsClient {
                     ExportLogsServiceRequest { resource_logs },
                 ))
                 .await
-                .map_err(crate::Error::from)?;
+                .map_err(|e| OTelSdkError::InternalFailure(format!("export error: {:?}", e)))?;
             Ok(())
         }
     }
 
-    fn shutdown(&mut self) {
-        let _ = self.inner.take();
+    fn shutdown(&mut self) -> OTelSdkResult {
+        match self.inner.take() {
+            Some(_) => Ok(()), // Successfully took `inner`, indicating a successful shutdown.
+            None => Err(OTelSdkError::AlreadyShutdown), // `inner` was already `None`, meaning it's already shut down.
+        }
     }
 
     fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) {

opentelemetry-otlp/src/logs.rs

@@ -6,9 +6,7 @@
 use opentelemetry::otel_debug;
 use std::fmt::Debug;
 
-use opentelemetry_sdk::logs::LogResult;
-
-use opentelemetry_sdk::logs::LogBatch;
+use opentelemetry_sdk::{error::OTelSdkResult, logs::LogBatch};
 
 use crate::{HasExportConfig, NoExporterBuilderSet};
 
@@ -145,7 +143,7 @@ impl opentelemetry_sdk::logs::LogExporter for LogExporter {
     fn export(
         &self,
         batch: LogBatch<'_>,
-    ) -> impl std::future::Future<Output = LogResult<()>> + Send {
+    ) -> impl std::future::Future<Output = OTelSdkResult> + Send {
         async move {
             match &self.client {
                 #[cfg(feature = "grpc-tonic")]

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

@@ -86,7 +86,7 @@ async fn logs_tokio_helper(
             .await
             .unwrap();
     } else {
-        let _ = logger_provider.shutdown();
+        assert!(logger_provider.shutdown().is_ok());
     }
     tokio::time::sleep(Duration::from_secs(5)).await;
     assert_logs_results_contains(test_utils::LOGS_FILE, expected_uuid.as_str())?;
@@ -126,7 +126,7 @@ fn logs_non_tokio_helper(is_simple: bool, init_logs_inside_rt: bool) -> Result<(
         );
     }
 
-    let _ = logger_provider.shutdown();
+    assert!(logger_provider.shutdown().is_ok());
     std::thread::sleep(Duration::from_secs(5));
     assert_logs_results_contains(test_utils::LOGS_FILE, expected_uuid.as_str())?;
     Ok(())

opentelemetry-proto/src/transform/logs.rs

@@ -225,8 +225,9 @@ mod tests {
     use opentelemetry::logs::LoggerProvider;
     use opentelemetry::time::now;
     use opentelemetry::InstrumentationScope;
+    use opentelemetry_sdk::error::OTelSdkResult;
     use opentelemetry_sdk::logs::LogProcessor;
-    use opentelemetry_sdk::logs::{LogResult, SdkLoggerProvider};
+    use opentelemetry_sdk::logs::SdkLoggerProvider;
     use opentelemetry_sdk::{logs::LogBatch, logs::SdkLogRecord, Resource};
 
     #[derive(Debug)]
@@ -235,11 +236,11 @@ mod tests {
     impl LogProcessor for MockProcessor {
         fn emit(&self, _record: &mut SdkLogRecord, _instrumentation: &InstrumentationScope) {}
 
-        fn force_flush(&self) -> LogResult<()> {
+        fn force_flush(&self) -> OTelSdkResult {
             Ok(())
         }
 
-        fn shutdown(&self) -> LogResult<()> {
+        fn shutdown(&self) -> OTelSdkResult {
             Ok(())
         }
     }

opentelemetry-sdk/CHANGELOG.md

@@ -386,7 +386,7 @@ let processor = BatchSpanProcessor::builder(exporter)
   - The `opentelemetry_otlp` API, via `.with_tonic().with_timeout()` or `.with_http().with_timeout()`.
 
 - **Breaking**
- - The public API changes in the Tracing:
+ - The public API changes in the Trace SDK:
    - Before:
       ```rust
         fn SpanExporter::export(&mut self, batch: Vec<SpanData>) -> BoxFuture<'static, ExportResult>;
@@ -400,7 +400,7 @@ let processor = BatchSpanProcessor::builder(exporter)
         fn SpanExporter::export(&mut self, batch: Vec<SpanData>) -> BoxFuture<'static, OTelSdkResult>;
         fn SpanExporter::shutdown(&mut self) -> OTelSdkResult;
         fn SpanExporter::force_flush(&mut self) -> BoxFuture<'static, OTelSdkResult>
-        fn TraerProvider::shutdown(&self) -> OTelSdkResult;
+        fn TracerProvider::shutdown(&self) -> OTelSdkResult;
         fn TracerProvider::force_flush(&self) -> OTelSdkResult;
       ```
 - **Breaking** Renamed `LoggerProvider`, `Logger` and `LogRecord' to
@@ -413,6 +413,23 @@ let processor = BatchSpanProcessor::builder(exporter)
   type-aliased to `SdkTracer` to keep back-compat with tracing-opentelemetry.
   [#2614](https://github.com/open-telemetry/opentelemetry-rust/pull/2614)
 
+- **Breaking**
+ - The public API changes in the Logs SDK:
+   - Before:
+      ```rust
+        fn LogExporter::export(&self, _batch: LogBatch<'_>,) -> impl std::future::Future<Output = LogResult<()>> + Send
+        fn LogExporter::shutdown(&mut self);
+        fn LoggerProvider::shutdown(&self) -> LogResult<()>
+        fn LoggerProvider::force_flush(&self) -> Vec<LogResult<()>>
+      ```
+    - After:
+      ```rust
+        fn LogExporter::export(&self, _batch: LogBatch<'_>,) -> impl std::future::Future<Output = OTelSdkResult<()>> + Send
+        fn LogExporter::shutdown(&mut self) -> OTelSdkResult;
+        fn LoggerProvider::shutdown(&self) -> OTelSdkResult;
+        fn LoggerProvider::force_flush(&self) -> OTelSdkResult;
+      ```
+
 ## 0.27.1
 
 Released 2024-Nov-27

opentelemetry-sdk/benches/log.rs

@@ -24,9 +24,8 @@ use opentelemetry::logs::{AnyValue, LogRecord as _, Logger, LoggerProvider, Seve
 use opentelemetry::trace::Tracer;
 use opentelemetry::trace::TracerProvider;
 use opentelemetry::{InstrumentationScope, Key};
-use opentelemetry_sdk::logs::{
-    LogProcessor, LogResult, SdkLogRecord, SdkLogger, SdkLoggerProvider,
-};
+use opentelemetry_sdk::error::OTelSdkResult;
+use opentelemetry_sdk::logs::{LogProcessor, SdkLogRecord, SdkLogger, SdkLoggerProvider};
 use opentelemetry_sdk::trace::{Sampler, SdkTracerProvider};
 
 #[derive(Debug)]
@@ -35,11 +34,11 @@ struct NoopProcessor;
 impl LogProcessor for NoopProcessor {
     fn emit(&self, _data: &mut SdkLogRecord, _scope: &InstrumentationScope) {}
 
-    fn force_flush(&self) -> LogResult<()> {
+    fn force_flush(&self) -> OTelSdkResult {
         Ok(())
     }
 
-    fn shutdown(&self) -> LogResult<()> {
+    fn shutdown(&self) -> OTelSdkResult {
         Ok(())
     }
 }