Merge branch 'main' into otlp-exporter-env

Author: lalitb

CONTRIBUTING.md

@@ -172,7 +172,7 @@ It's important to regularly review and remove the `otel_unstable` flag from the
 
 The potential features include:
 
-- Stable and non-experimental features that compliant to specification, and have a feature flag to minimize compilation size. Example: feature flags for signals (like `logs`, `traces`, `metrics`) and runtimes (`rt-tokio`, `rt-tokio-current-thread`, `rt-async-std`).
+- Stable and non-experimental features that are compliant with the specification and have a feature flag to minimize compilation size. Example: feature flags for signals (like `logs`, `traces`, `metrics`) and runtimes (`rt-tokio`, `rt-tokio-current-thread`).
 - Stable and non-experimental features, although not part of the specification, are crucial for enhancing the tracing/log crate's functionality or boosting performance. These features are also subject to discussion and approval by the OpenTelemetry Rust Maintainers.
 
 All such features should adhere to naming convention  `<signal>_<feature_name>`

docs/design/logs.md

@@ -195,6 +195,13 @@ They get called in the order of registration. Log records are passed to the
 records, enrich them, filter them, and export to destinations by leveraging
 LogRecord Exporters.
 
+Similar to [LoggerProvider](#sdkloggerprovider), methods on the `LogProcessor`
+trait also takes a immutable self (`&self`) only, forcing the need to use
+interior mutability, if any mutation is required. The exception to this is
+`set_resource`, which takes a `&mut self`. This is acceptable as `set_resource`
+is called by the `SdkLoggerProvider` during build() method only, and is not
+required after that.
+
 Following built-in Log processors are provided in the Log SDK:
 
 ##### SimpleLogProcessor
@@ -232,8 +239,13 @@ other words, that many logs can be lost if the app crashes in the middle.
 
 ## LogExporters
 
-LogExporters are responsible for exporting logs to a destination. Some of them
-include:
+LogExporters are responsible for exporting logs to a destination.
+`SdkLoggerProvider` does not have a direct knowledge of the `LogExporter`, as it
+only deals with `LogProcessors`. It is the `LogProcessor`s that invokes
+`LogExporter` methods. Most methods on `LogExporter` trait also only takes
+`&self`, following the same reasoning as [LogProcessors](#logrecord-processors)
+
+Some of the exporters are:
 
 1. **InMemoryExporter** - exports to an in-memory list, primarily for
    unit-testing. This is used extensively in the repo itself, and external users

opentelemetry-otlp/src/lib.rs

@@ -141,7 +141,6 @@
 //! ```
 //!
 //! [`tokio`]: https://tokio.rs
-//! [`async-std`]: https://async.rs
 //!
 //! # Feature Flags
 //! The following feature flags can enable exporters for different telemetry signals:

opentelemetry-sdk/CHANGELOG.md

@@ -4,6 +4,7 @@
 
 - **Breaking**: The `Runtime` trait has been simplified and refined. See the [#2641](https://github.com/open-telemetry/opentelemetry-rust/pull/2641)
   for the changes.
+- Removed `async-std` support for `Runtime`, as [`async-std` crate is deprecated](https://crates.io/crates/async-std).
 - Calls to `MeterProviderBuilder::with_resource`, `TracerProviderBuilder::with_resource`,
   `LoggerProviderBuilder::with_resource` are now additive ([#2677](https://github.com/open-telemetry/opentelemetry-rust/pull/2677)).
 - Moved `ExportError` trait from `opentelemetry::trace::ExportError` to `opentelemetry_sdk::export::ExportError`

opentelemetry-sdk/Cargo.toml

@@ -12,7 +12,6 @@ rust-version = "1.75.0"
 [dependencies]
 opentelemetry = { version = "0.28", path = "../opentelemetry/" }
 opentelemetry-http = { version = "0.28", path = "../opentelemetry-http", optional = true }
-async-std = { workspace = true, features = ["unstable"], optional = true }
 futures-channel = { workspace = true }
 futures-executor = { workspace = true }
 futures-util = { workspace = true, features = ["std", "sink", "async-await-macro"] }
@@ -47,11 +46,10 @@ jaeger_remote_sampler = ["trace", "opentelemetry-http", "http", "serde", "serde_
 logs = ["opentelemetry/logs", "serde_json"]
 spec_unstable_logs_enabled = ["logs", "opentelemetry/spec_unstable_logs_enabled"]
 metrics = ["opentelemetry/metrics", "glob"]
-testing = ["opentelemetry/testing", "trace", "metrics", "logs", "rt-async-std", "rt-tokio", "rt-tokio-current-thread", "tokio/macros", "tokio/rt-multi-thread"]
+testing = ["opentelemetry/testing", "trace", "metrics", "logs", "rt-tokio", "rt-tokio-current-thread", "tokio/macros", "tokio/rt-multi-thread"]
 experimental_async_runtime = []
 rt-tokio = ["tokio", "tokio-stream", "experimental_async_runtime"]
 rt-tokio-current-thread = ["tokio", "tokio-stream", "experimental_async_runtime"]
-rt-async-std = ["async-std", "experimental_async_runtime"]
 internal-logs = ["tracing"]
 experimental_metrics_periodicreader_with_async_runtime = ["metrics"]
 spec_unstable_metrics_views = ["metrics"]

opentelemetry-sdk/benches/log_enabled.rs

@@ -5,10 +5,12 @@
     Total Number of Cores:   14 (10 performance and 4 efficiency)
     | Test                                         | Average time|
     |---------------------------------------------|-------------|
-    | exporter_disabled_concurrent_processor      |  1.9 ns     |
-    | exporter_disabled_simple_processor          |  5.0 ns     |
+    | exporter_disabled_concurrent_processor      |  1.0 ns     |
+    | exporter_disabled_simple_processor          |  4.5 ns     |
 */
 
+// cargo bench --bench log_enabled --features="spec_unstable_logs_enabled,experimental_logs_concurrent_log_processor"
+
 use criterion::{criterion_group, criterion_main, Criterion};
 use opentelemetry::logs::{Logger, LoggerProvider};
 use opentelemetry_sdk::error::OTelSdkResult;

opentelemetry-sdk/src/lib.rs

@@ -94,10 +94,8 @@
 //! * `experimental_async_runtime`: Enables the experimental `Runtime` trait and related functionality.
 //! * `rt-tokio`: Spawn telemetry tasks using [tokio]'s multi-thread runtime.
 //! * `rt-tokio-current-thread`: Spawn telemetry tasks on a separate runtime so that the main runtime won't be blocked.
-//! * `rt-async-std`: Spawn telemetry tasks using [async-std]'s runtime.
 //!
 //! [tokio]: https://crates.io/crates/tokio
-//! [async-std]: https://crates.io/crates/async-std
 #![warn(
     future_incompatible,
     missing_debug_implementations,

opentelemetry-sdk/src/logs/logger.rs

@@ -50,6 +50,7 @@ impl opentelemetry::logs::Logger for SdkLogger {
     }
 
     #[cfg(feature = "spec_unstable_logs_enabled")]
+    #[inline]
     fn event_enabled(&self, level: Severity, target: &str, name: Option<&str>) -> bool {
         self.provider
             .log_processors()

opentelemetry-sdk/src/runtime.rs

@@ -1,20 +1,18 @@
 //! Provides an abstraction of several async runtimes
 //!
-//! This  allows OpenTelemetry to work with any current or future runtime. There are currently
-//! builtin implementations for [Tokio] and [async-std].
+//! This  allows OpenTelemetry to work with any current or future runtime. There is currently
+//! built-in implementation for [Tokio].
 //!
 //! [Tokio]: https://crates.io/crates/tokio
-//! [async-std]: https://crates.io/crates/async-std
 
 use futures_util::stream::{unfold, Stream};
 use std::{fmt::Debug, future::Future, time::Duration};
 use thiserror::Error;
 
-/// A runtime is an abstraction of an async runtime like [Tokio] or [async-std]. It allows
+/// A runtime is an abstraction of an async runtime like [Tokio]. It allows
 /// OpenTelemetry to work with any current and hopefully future runtime implementations.
 ///
 /// [Tokio]: https://crates.io/crates/tokio
-/// [async-std]: https://crates.io/crates/async-std
 ///
 /// # Note
 ///
@@ -139,34 +137,6 @@ impl Runtime for TokioCurrentThread {
     }
 }
 
-/// Runtime implementation, which works with async-std.
-#[cfg(all(feature = "experimental_async_runtime", feature = "rt-async-std"))]
-#[cfg_attr(
-    docsrs,
-    doc(cfg(all(feature = "experimental_async_runtime", feature = "rt-async-std")))
-)]
-#[derive(Debug, Clone)]
-pub struct AsyncStd;
-
-#[cfg(all(feature = "experimental_async_runtime", feature = "rt-async-std"))]
-#[cfg_attr(
-    docsrs,
-    doc(cfg(all(feature = "experimental_async_runtime", feature = "rt-async-std")))
-)]
-impl Runtime for AsyncStd {
-    fn spawn<F>(&self, future: F)
-    where
-        F: Future<Output = ()> + Send + 'static,
-    {
-        #[allow(clippy::let_underscore_future)]
-        let _ = async_std::task::spawn(future);
-    }
-
-    fn delay(&self, duration: Duration) -> impl Future<Output = ()> + Send + 'static {
-        async_std::task::sleep(duration)
-    }
-}
-
 /// `RuntimeChannel` is an extension to [`Runtime`]. Currently, it provides a
 /// channel that is used by the [log] and [span] batch processors.
 ///
@@ -275,32 +245,3 @@ impl RuntimeChannel for TokioCurrentThread {
         )
     }
 }
-
-#[cfg(all(feature = "experimental_async_runtime", feature = "rt-async-std"))]
-impl<T: Send> TrySend for async_std::channel::Sender<T> {
-    type Message = T;
-
-    fn try_send(&self, item: Self::Message) -> Result<(), TrySendError> {
-        self.try_send(item).map_err(|err| match err {
-            async_std::channel::TrySendError::Full(_) => TrySendError::ChannelFull,
-            async_std::channel::TrySendError::Closed(_) => TrySendError::ChannelClosed,
-        })
-    }
-}
-
-#[cfg(all(feature = "experimental_async_runtime", feature = "rt-async-std"))]
-#[cfg_attr(
-    docsrs,
-    doc(cfg(all(feature = "experimental_async_runtime", feature = "rt-async-std")))
-)]
-impl RuntimeChannel for AsyncStd {
-    type Receiver<T: Debug + Send> = async_std::channel::Receiver<T>;
-    type Sender<T: Debug + Send> = async_std::channel::Sender<T>;
-
-    fn batch_message_channel<T: Debug + Send>(
-        &self,
-        capacity: usize,
-    ) -> (Self::Sender<T>, Self::Receiver<T>) {
-        async_std::channel::bounded(capacity)
-    }
-}

opentelemetry-sdk/src/trace/span_processor_with_async_runtime.rs

@@ -44,7 +44,7 @@ use std::sync::Arc;
 ///
 /// This processor can be configured with an [`executor`] of your choice to
 /// batch and upload spans asynchronously when they end. If you have added a
-/// library like [`tokio`] or [`async-std`], you can pass in their respective
+/// library like [`tokio`], you can pass in their respective
 /// `spawn` and `interval` functions to have batching performed in those
 /// contexts.
 ///
@@ -79,7 +79,6 @@ use std::sync::Arc;
 ///
 /// [`executor`]: https://docs.rs/futures/0.3/futures/executor/index.html
 /// [`tokio`]: https://tokio.rs
-/// [`async-std`]: https://async.rs
 pub struct BatchSpanProcessor<R: RuntimeChannel> {
     message_sender: R::Sender<BatchMessage>,
 
@@ -577,41 +576,4 @@ mod tests {
             .unwrap();
         runtime.block_on(timeout_test_tokio(false));
     }
-
-    #[test]
-    #[cfg(feature = "rt-async-std")]
-    fn test_timeout_async_std_timeout() {
-        async_std::task::block_on(timeout_test_std_async(true));
-    }
-
-    #[test]
-    #[cfg(feature = "rt-async-std")]
-    fn test_timeout_async_std_not_timeout() {
-        async_std::task::block_on(timeout_test_std_async(false));
-    }
-
-    // If the time_out is true, then the result suppose to ended with timeout.
-    // otherwise the exporter should be able to export within time out duration.
-    #[cfg(feature = "rt-async-std")]
-    async fn timeout_test_std_async(time_out: bool) {
-        let config = BatchConfig {
-            max_export_timeout: Duration::from_millis(if time_out { 5 } else { 60 }),
-            scheduled_delay: Duration::from_secs(60 * 60 * 24), // set the tick to 24 hours so we know the span must be exported via force_flush
-            ..Default::default()
-        };
-        let exporter = BlockingExporter {
-            delay_for: Duration::from_millis(if !time_out { 5 } else { 60 }),
-            delay_fn: async_std::task::sleep,
-        };
-        let processor = BatchSpanProcessor::new(exporter, config, runtime::AsyncStd);
-        processor.on_end(new_test_export_span_data());
-        let flush_res = processor.force_flush();
-        if time_out {
-            assert!(flush_res.is_err());
-        } else {
-            assert!(flush_res.is_ok());
-        }
-        let shutdown_res = processor.shutdown();
-        assert!(shutdown_res.is_ok());
-    }
 }