Use native Rust support for async traits in LogExporter::export() method (11% improvement)

opentelemetry-appender-tracing/benches/logs.rs

@@ -34,8 +34,12 @@ struct NoopExporter {
 
 #[async_trait]
 impl LogExporter for NoopExporter {
-    async fn export(&self, _: LogBatch<'_>) -> LogResult<()> {
-        LogResult::Ok(())
+    #[allow(clippy::manual_async_fn)]
+    fn export<'a>(
+        &'a self,
+        _batch: &'a LogBatch<'a>,
+    ) -> impl std::future::Future<Output = LogResult<()>> + Send + 'a {
+        async { LogResult::Ok(()) }
     }
 
     fn event_enabled(&self, _: opentelemetry::logs::Severity, _: &str, _: &str) -> bool {
@@ -44,17 +48,17 @@ impl LogExporter for NoopExporter {
 }
 
 #[derive(Debug)]
-struct NoopProcessor {
-    exporter: Box<dyn LogExporter>,
+struct NoopProcessor<E: LogExporter> {
+    exporter: E,
 }
 
-impl NoopProcessor {
-    fn new(exporter: Box<dyn LogExporter>) -> Self {
+impl<E: LogExporter> NoopProcessor<E> {
+    fn new(exporter: E) -> Self {
         Self { exporter }
     }
 }
 
-impl LogProcessor for NoopProcessor {
+impl<E: LogExporter> LogProcessor for NoopProcessor<E> {
     fn emit(&self, _: &mut LogRecord, _: &InstrumentationScope) {
         // no-op
     }
@@ -124,7 +128,7 @@ fn benchmark_no_subscriber(c: &mut Criterion) {
 
 fn benchmark_with_ot_layer(c: &mut Criterion, enabled: bool, bench_name: &str) {
     let exporter = NoopExporter { enabled };
-    let processor = NoopProcessor::new(Box::new(exporter));
+    let processor = NoopProcessor::new(exporter);
     let provider = LoggerProvider::builder()
         .with_resource(
             Resource::builder_empty()

opentelemetry-appender-tracing/src/layer.rs

@@ -247,11 +247,17 @@
 
     #[async_trait]
     impl LogExporter for ReentrantLogExporter {
-        async fn export(&self, _batch: LogBatch<'_>) -> LogResult<()> {
-            // This will cause a deadlock as the export itself creates a log
-            // while still within the lock of the SimpleLogProcessor.
-            warn!(name: "my-event-name", target: "reentrant", event_id = 20, user_name = "otel", user_email = "[email protected]");
-            Ok(())
+        #[allow(clippy::manual_async_fn)]
+        fn export<'a>(
+            &'a self,
+            _batch: &'a LogBatch<'a>,
+        ) -> impl std::future::Future<Output = LogResult<()>> + Send + 'a {
+            async {
+                // This will cause a deadlock as the export itself creates a log
+                // while still within the lock of the SimpleLogProcessor.
+                warn!(name: "my-event-name", target: "reentrant", event_id = 20, user_name = "otel", user_email = "[email protected]");
+                Ok(())
+            }
         }
     }
 

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

@@ -9,42 +9,48 @@
 
 #[async_trait]
 impl LogExporter for OtlpHttpClient {
-    async fn export(&self, batch: LogBatch<'_>) -> LogResult<()> {
-        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())),
-            })?;
-
-        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)
-            .map_err(|e| crate::Error::RequestFailed(Box::new(e)))?;
-
-        for (k, v) in &self.headers {
-            request.headers_mut().insert(k.clone(), v.clone());
+    #[allow(clippy::manual_async_fn)]
+    fn export<'a>(
+        &'a self,
+        batch: &'a LogBatch<'a>,
+    ) -> impl std::future::Future<Output = LogResult<()>> + Send + 'a {
+        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())),
+                })?;
+
+            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)
+                .map_err(|e| crate::Error::RequestFailed(Box::new(e)))?;
+
+            for (k, v) in &self.headers {
+                request.headers_mut().insert(k.clone(), v.clone());
+            }
+
+            let request_uri = request.uri().to_string();
+            let response = client.send(request).await?;
+
+            if !response.status().is_success() {
+                let error = format!(
+                    "OpenTelemetry logs export failed. Url: {}, Status Code: {}, Response: {:?}",
+                    response.status().as_u16(),
+                    request_uri,
+                    response.body()
+                );
+                return Err(LogError::Other(error.into()));
+            }
+
+            Ok(())
         }
-
-        let request_uri = request.uri().to_string();
-        let response = client.send(request).await?;
-
-        if !response.status().is_success() {
-            let error = format!(
-                "OpenTelemetry logs export failed. Url: {}, Status Code: {}, Response: {:?}",
-                response.status().as_u16(),
-                request_uri,
-                response.body()
-            );
-            return Err(LogError::Other(error.into()));
-        }
-
-        Ok(())
     }
 
     fn shutdown(&mut self) {

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

@@ -28,7 +28,7 @@
 mod metrics;
 
 #[cfg(feature = "logs")]
-mod logs;
+pub(crate) mod logs;
 
 #[cfg(feature = "trace")]
 mod trace;
@@ -236,7 +236,7 @@
             OTEL_EXPORTER_OTLP_LOGS_HEADERS,
         )?;
 
-        Ok(crate::LogExporter::new(client))
+        Ok(crate::LogExporter::from_http(client))
     }
 
     /// Create a metrics exporter with the current configuration
@@ -262,7 +262,7 @@
 }
 
 #[derive(Debug)]
-struct OtlpHttpClient {
+pub(crate) struct OtlpHttpClient {
     client: Mutex<Option<Arc<dyn HttpClient>>>,
     collector_endpoint: Uri,
     headers: HashMap<HeaderName, HeaderValue>,
@@ -314,7 +314,7 @@
     #[cfg(feature = "logs")]
     fn build_logs_export_body(
         &self,
-        logs: LogBatch<'_>,
+        logs: &LogBatch<'_>,
     ) -> opentelemetry_sdk::logs::LogResult<(Vec<u8>, &'static str)> {
         use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceRequest;
         let resource_logs = group_logs_by_resource_and_scope(logs, &self.resource);

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

@@ -58,35 +58,40 @@
 
 #[async_trait]
 impl LogExporter for TonicLogsClient {
-    async fn export(&self, batch: LogBatch<'_>) -> LogResult<()> {
-        let (mut client, metadata, extensions) = match &self.inner {
-            Some(inner) => {
-                let (m, e, _) = inner
-                    .interceptor
-                    .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)))?
-                    .into_parts();
-                (inner.client.clone(), m, e)
-            }
-            None => return Err(LogError::Other("exporter is already shut down".into())),
-        };
+    #[allow(clippy::manual_async_fn)]
+    fn export<'a>(
+        &'a self,
+        batch: &'a LogBatch<'a>,
+    ) -> impl std::future::Future<Output = LogResult<()>> + Send + 'a {
+        async move {
+            let (mut client, metadata, extensions) = match &self.inner {
+                Some(inner) => {
+                    let (m, e, _) = inner
+                        .interceptor
+                        .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)))?
+                        .into_parts();
+                    (inner.client.clone(), m, e)
+                }
+                None => return Err(LogError::Other("exporter is already shut down".into())),
+            };
 
-        let resource_logs = group_logs_by_resource_and_scope(batch, &self.resource);
+            let resource_logs = group_logs_by_resource_and_scope(batch, &self.resource);
 
-        otel_debug!(name: "TonicsLogsClient.CallingExport");
+            otel_debug!(name: "TonicsLogsClient.CallingExport");
 
-        client
-            .export(Request::from_parts(
-                metadata,
-                extensions,
-                ExportLogsServiceRequest { resource_logs },
-            ))
-            .await
-            .map_err(crate::Error::from)?;
-
-        Ok(())
+            client
+                .export(Request::from_parts(
+                    metadata,
+                    extensions,
+                    ExportLogsServiceRequest { resource_logs },
+                ))
+                .await
+                .map_err(crate::Error::from)?;
+            Ok(())
+        }
     }
 
     fn shutdown(&mut self) {

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

@@ -20,7 +20,7 @@
 };
 
 #[cfg(feature = "logs")]
-mod logs;
+pub(crate) mod logs;
 
 #[cfg(feature = "metrics")]
 mod metrics;
@@ -273,7 +273,7 @@
 
         let client = TonicLogsClient::new(channel, interceptor, compression);
 
-        Ok(crate::logs::LogExporter::new(client))
+        Ok(crate::logs::LogExporter::from_tonic(client))
     }
 
     /// Build a new tonic metrics exporter

opentelemetry-otlp/src/logs.rs

@@ -108,7 +108,16 @@
 /// OTLP exporter that sends log data
 #[derive(Debug)]
 pub struct LogExporter {
-    client: Box<dyn opentelemetry_sdk::export::logs::LogExporter>,
+    //client: Box<dyn opentelemetry_sdk::export::logs::LogExporter>,
+    client: LogExporterInner,
+}
+
+#[derive(Debug)]
+enum LogExporterInner {
+    #[cfg(feature = "grpc-tonic")]
+    Tonic(crate::exporter::tonic::logs::TonicLogsClient),
+    #[cfg(any(feature = "http-proto", feature = "http-json"))]
+    Http(crate::exporter::http::OtlpHttpClient),
 }
 
 impl LogExporter {
@@ -117,21 +126,44 @@
         LogExporterBuilder::default()
     }
 
-    /// Create a new log exporter
-    pub fn new(client: impl opentelemetry_sdk::export::logs::LogExporter + 'static) -> Self {
+    #[cfg(any(feature = "http-proto", feature = "http-json"))]
+    pub(crate) fn from_http(client: crate::exporter::http::OtlpHttpClient) -> Self {
         LogExporter {
-            client: Box::new(client),
+            client: LogExporterInner::Http(client),
+        }
+    }
+
+    #[cfg(feature = "grpc-tonic")]
+    pub(crate) fn from_tonic(client: crate::exporter::tonic::logs::TonicLogsClient) -> Self {
+        LogExporter {
+            client: LogExporterInner::Tonic(client),
         }
     }
 }
 
 #[async_trait]
 impl opentelemetry_sdk::export::logs::LogExporter for LogExporter {
-    async fn export(&self, batch: LogBatch<'_>) -> LogResult<()> {
-        self.client.export(batch).await
+    #[allow(clippy::manual_async_fn)]
+    fn export<'a>(
+        &'a self,
+        batch: &'a LogBatch<'a>,
+    ) -> impl std::future::Future<Output = LogResult<()>> + Send + 'a {
+        async move {
+            match &self.client {
+                #[cfg(feature = "grpc-tonic")]
+                LogExporterInner::Tonic(client) => client.export(batch).await,
+                #[cfg(any(feature = "http-proto", feature = "http-json"))]
+                LogExporterInner::Http(client) => client.export(batch).await,
+            }
+        }
     }
 
     fn set_resource(&mut self, resource: &opentelemetry_sdk::Resource) {
-        self.client.set_resource(resource);
+        match &mut self.client {
+            #[cfg(feature = "grpc-tonic")]
+            LogExporterInner::Tonic(client) => client.set_resource(resource),
+            #[cfg(any(feature = "http-proto", feature = "http-json"))]
+            LogExporterInner::Http(client) => client.set_resource(resource),
+        }
     }
 }

opentelemetry-proto/src/transform/logs.rs

@@ -178,7 +178,7 @@ pub mod tonic {
     }
 
     pub fn group_logs_by_resource_and_scope(
-        logs: LogBatch<'_>,
+        logs: &LogBatch<'_>,
         resource: &ResourceAttributesWithSchema,
     ) -> Vec<ResourceLogs> {
         // Group logs by target or instrumentation name
@@ -261,7 +261,7 @@ mod tests {
         let resource: ResourceAttributesWithSchema = (&resource).into(); // Convert Resource to ResourceAttributesWithSchema
 
         let grouped_logs =
-            crate::transform::logs::tonic::group_logs_by_resource_and_scope(log_batch, &resource);
+            crate::transform::logs::tonic::group_logs_by_resource_and_scope(&log_batch, &resource);
 
         assert_eq!(grouped_logs.len(), 1);
         let resource_logs = &grouped_logs[0];
@@ -281,7 +281,7 @@ mod tests {
         let log_batch = LogBatch::new(&logs);
         let resource: ResourceAttributesWithSchema = (&resource).into(); // Convert Resource to ResourceAttributesWithSchema
         let grouped_logs =
-            crate::transform::logs::tonic::group_logs_by_resource_and_scope(log_batch, &resource);
+            crate::transform::logs::tonic::group_logs_by_resource_and_scope(&log_batch, &resource);
 
         assert_eq!(grouped_logs.len(), 1);
         let resource_logs = &grouped_logs[0];

opentelemetry-sdk/Cargo.toml

@@ -45,7 +45,7 @@ pprof = { version = "0.13", features = ["flamegraph", "criterion"] }
 default = ["trace", "metrics", "logs", "internal-logs"]
 trace = ["opentelemetry/trace", "rand", "percent-encoding"]
 jaeger_remote_sampler = ["trace", "opentelemetry-http", "http", "serde", "serde_json", "url"]
-logs = ["opentelemetry/logs", "async-trait", "serde_json"]
+logs = ["opentelemetry/logs", "serde_json"]
 spec_unstable_logs_enabled = ["logs", "opentelemetry/spec_unstable_logs_enabled"]
 metrics = ["opentelemetry/metrics", "glob", "async-trait"]
 testing = ["opentelemetry/testing", "trace", "metrics", "logs", "rt-async-std", "rt-tokio", "rt-tokio-current-thread", "tokio/macros", "tokio/rt-multi-thread"]