Merge main

Author: scottgerring

opentelemetry-appender-tracing/Cargo.toml

@@ -24,7 +24,6 @@ opentelemetry-stdout = { path = "../opentelemetry-stdout", features = ["logs"] }
 opentelemetry_sdk = { path = "../opentelemetry-sdk", features = ["logs", "testing"]  }
 tracing-subscriber = { workspace = true, features = ["registry", "std", "env-filter"] }
 tracing-log = "0.2"
-async-trait = { workspace = true }
 criterion = { workspace = true }
 tokio = { workspace = true, features = ["full"]}
 

opentelemetry-appender-tracing/benches/logs.rs

@@ -13,7 +13,6 @@
     | ot_layer_enabled            | 196 ns      |
 */
 
-use async_trait::async_trait;
 use criterion::{criterion_group, criterion_main, Criterion};
 use opentelemetry::InstrumentationScope;
 use opentelemetry_appender_tracing::layer as tracing_layer;
@@ -32,10 +31,13 @@ struct NoopExporter {
     enabled: bool,
 }
 
-#[async_trait]
 impl LogExporter for NoopExporter {
-    async fn export(&self, _: LogBatch<'_>) -> LogResult<()> {
-        LogResult::Ok(())
+    #[allow(clippy::manual_async_fn)]
+    fn export(
+        &self,
+        _batch: LogBatch<'_>,
+    ) -> impl std::future::Future<Output = LogResult<()>> + Send {
+        async { LogResult::Ok(()) }
     }
 
     fn event_enabled(&self, _: opentelemetry::logs::Severity, _: &str, _: &str) -> bool {
@@ -44,17 +46,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 +126,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

@@ -209,7 +209,6 @@ const fn severity_of_level(level: &Level) -> Severity {
 #[cfg(test)]
 mod tests {
     use crate::layer;
-    use async_trait::async_trait;
     use opentelemetry::logs::Severity;
     use opentelemetry::trace::TracerProvider as _;
     use opentelemetry::trace::{TraceContextExt, TraceFlags, Tracer};
@@ -245,13 +244,18 @@ mod tests {
     #[derive(Clone, Debug, Default)]
     struct ReentrantLogExporter;
 
-    #[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(
+            &self,
+            _batch: LogBatch<'_>,
+        ) -> impl std::future::Future<Output = LogResult<()>> + Send {
+            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/examples/basic-otlp-http/Cargo.toml

@@ -6,11 +6,9 @@ license = "Apache-2.0"
 publish = false
 
 [features]
-default = ["reqwest", "experimental_metrics_periodicreader_with_async_runtime"]
-reqwest = ["opentelemetry-otlp/reqwest-client"]
+default = ["reqwest-blocking"]
+reqwest-blocking = ["opentelemetry-otlp/reqwest-blocking-client"]
 hyper = ["opentelemetry-otlp/hyper-client"]
-experimental_metrics_periodicreader_with_async_runtime = ["opentelemetry_sdk/experimental_metrics_periodicreader_with_async_runtime"]
-
 
 [dependencies]
 once_cell = { workspace = true }

opentelemetry-otlp/examples/basic-otlp-http/README.md

@@ -13,11 +13,9 @@ applications, these filters should be adjusted appropriately.
 
 The example employs a `BatchExporter` for logs and traces, which is the
 recommended approach when using OTLP exporters. While it can be modified to use
-a `SimpleExporter`, this requires enabling feature flag `reqwest-blocking-client` and
-making the `main()` a normal main and *not* `tokio::main`
+a `SimpleExporter`, this requires making the main function a regular main and
+*not* tokio main.
 
-// TODO: Metrics does not work with non tokio main when using `reqwest-blocking-client` today, fix that when switching
-// default to use own thread.
 // TODO: Document `hyper` feature flag when using SimpleProcessor.
 
 ## Usage

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

@@ -11,7 +11,6 @@ use opentelemetry_otlp::{LogExporter, MetricExporter, Protocol, SpanExporter};
 use opentelemetry_sdk::{
     logs::LoggerProvider,
     metrics::{MetricError, SdkMeterProvider},
-    runtime,
     trace::{self as sdktrace, TracerProvider},
 };
 use opentelemetry_sdk::{
@@ -50,7 +49,9 @@ fn init_traces() -> Result<sdktrace::TracerProvider, TraceError> {
         .build()?;
 
     Ok(TracerProvider::builder()
-        .with_batch_exporter(exporter, runtime::Tokio)
+        // TODO: Enable BatchExporter after
+        // https://github.com/open-telemetry/opentelemetry-rust/pull/2456
+        .with_simple_exporter(exporter)
         .with_resource(RESOURCE.clone())
         .build())
 }
@@ -62,15 +63,6 @@ fn init_metrics() -> Result<opentelemetry_sdk::metrics::SdkMeterProvider, Metric
         .with_endpoint("http://localhost:4318/v1/metrics")
         .build()?;
 
-    #[cfg(feature = "experimental_metrics_periodicreader_with_async_runtime")]
-    let reader =
-        opentelemetry_sdk::metrics::periodic_reader_with_async_runtime::PeriodicReader::builder(
-            exporter,
-            runtime::Tokio,
-        )
-        .build();
-    // TODO: This does not work today. See https://github.com/open-telemetry/opentelemetry-rust/issues/2400
-    #[cfg(not(feature = "experimental_metrics_periodicreader_with_async_runtime"))]
     let reader = opentelemetry_sdk::metrics::PeriodicReader::builder(exporter).build();
 
     Ok(SdkMeterProvider::builder()
@@ -79,8 +71,10 @@ fn init_metrics() -> Result<opentelemetry_sdk::metrics::SdkMeterProvider, Metric
         .build())
 }
 
-#[tokio::main]
-async fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
+// #[tokio::main]
+// TODO: Re-enable tokio::main, if needed, after
+// https://github.com/open-telemetry/opentelemetry-rust/pull/2456
+fn main() -> Result<(), Box<dyn Error + Send + Sync + 'static>> {
     let logger_provider = init_logs()?;
 
     // Create a new OpenTelemetryTracingBridge using the above LoggerProvider.

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

@@ -1,52 +1,56 @@
 use std::sync::Arc;
 
-use async_trait::async_trait;
 use http::{header::CONTENT_TYPE, Method};
 use opentelemetry::otel_debug;
 use opentelemetry_sdk::export::logs::{LogBatch, LogExporter, ShutdownResult};
 use opentelemetry_sdk::logs::{LogError, LogResult};
 
 use super::OtlpHttpClient;
 
-#[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(
+        &self,
+        batch: LogBatch<'_>,
+    ) -> impl std::future::Future<Output = LogResult<()>> + 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())),
+                })?;
 
-        let request_uri = request.uri().to_string();
-        otel_debug!(name: "HttpLogsClient.CallingExport");
-        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()));
-        }
+            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)))?;
 
-        Ok(())
+            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(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) -> ShutdownResult {

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

@@ -28,7 +28,7 @@ use std::time::Duration;
 mod metrics;
 
 #[cfg(feature = "logs")]
-mod logs;
+pub(crate) mod logs;
 
 #[cfg(feature = "trace")]
 mod trace;
@@ -239,7 +239,7 @@ impl HttpExporterBuilder {
             OTEL_EXPORTER_OTLP_LOGS_HEADERS,
         )?;
 
-        Ok(crate::LogExporter::new(client))
+        Ok(crate::LogExporter::from_http(client))
     }
 
     /// Create a metrics exporter with the current configuration
@@ -265,7 +265,7 @@ impl HttpExporterBuilder {
 }
 
 #[derive(Debug)]
-struct OtlpHttpClient {
+pub(crate) struct OtlpHttpClient {
     client: Mutex<Option<Arc<dyn HttpClient>>>,
     collector_endpoint: Uri,
     headers: HashMap<HeaderName, HeaderValue>,

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

@@ -1,4 +1,3 @@
-use async_trait::async_trait;
 use core::fmt;
 use opentelemetry::otel_debug;
 use opentelemetry_proto::tonic::collector::logs::v1::{
@@ -56,37 +55,41 @@ impl TonicLogsClient {
     }
 }
 
-#[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(
+        &self,
+        batch: LogBatch<'_>,
+    ) -> impl std::future::Future<Output = LogResult<()>> + Send {
+        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) -> ShutdownResult {

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

@@ -20,7 +20,7 @@ use crate::{
 };
 
 #[cfg(feature = "logs")]
-mod logs;
+pub(crate) mod logs;
 
 #[cfg(feature = "metrics")]
 mod metrics;
@@ -273,7 +273,7 @@ impl TonicExporterBuilder {
 
         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