Merge branch 'main' into log-attributes-update

Author: lalitb

.github/ISSUE_TEMPLATE/BUG-REPORT.yml

@@ -28,7 +28,7 @@ body:
   - type: textarea
     id: sdk-version
     attributes:
-      label: label: OpenTelemetry SDK Version (i.e version of `opentelemetry_sdk` crate)
+      label: OpenTelemetry SDK Version (i.e version of `opentelemetry_sdk` crate)
       description: What version of the `opentelemetry_sdk` crate are you using?
       placeholder: 0.x, 1.x, etc.
     validations:

opentelemetry-otlp/tests/integration_test/Cargo.toml

@@ -29,6 +29,7 @@ reqwest-client = ["opentelemetry-otlp/reqwest-client", "opentelemetry-otlp/http-
 reqwest-blocking-client = ["opentelemetry-otlp/reqwest-blocking-client", "opentelemetry-otlp/http-proto", "opentelemetry-otlp/trace","opentelemetry-otlp/logs", "opentelemetry-otlp/metrics", "internal-logs"]
 tonic-client = ["opentelemetry-otlp/grpc-tonic", "opentelemetry-otlp/trace", "opentelemetry-otlp/logs", "opentelemetry-otlp/metrics", "internal-logs"]
 internal-logs = ["opentelemetry-otlp/internal-logs"]
+experimental_metadata_attributes = ["opentelemetry-appender-tracing/experimental_metadata_attributes"]
 
 # Keep tonic as the default client
 default = ["tonic-client", "internal-logs"]

opentelemetry-otlp/tests/integration_test/src/logs_asserter.rs

@@ -1,5 +1,8 @@
 use anyhow::Result;
-use opentelemetry_proto::tonic::logs::v1::{LogRecord, LogsData, ResourceLogs};
+use opentelemetry_proto::tonic::{
+    common::v1::KeyValue,
+    logs::v1::{LogRecord, LogsData, ResourceLogs},
+};
 use std::fs::File;
 
 // Given two ResourceLogs, assert that they are equal except for the timestamps
@@ -75,8 +78,22 @@ impl PartialEq for LogRecordWrapper {
             a.severity_text, b.severity_text,
             "severity_text does not match"
         );
+        let a_attrs = a.attributes.clone();
+        #[cfg(feature = "experimental_metadata_attributes")]
+        let a_attrs: Vec<_> = a_attrs
+            .into_iter()
+            .filter(|KeyValue { key, .. }| !key.as_str().starts_with("code."))
+            .collect();
+
+        let b_attrs = b.attributes.clone();
+        #[cfg(feature = "experimental_metadata_attributes")]
+        let b_attrs: Vec<_> = b_attrs
+            .into_iter()
+            .filter(|KeyValue { key, .. }| !key.as_str().starts_with("code."))
+            .collect();
+
         assert_eq!(a.body, b.body, "body does not match");
-        assert_eq!(a.attributes, b.attributes, "attributes do not match");
+        assert_eq!(a_attrs, b_attrs, "attributes do not match");
         assert_eq!(
             a.dropped_attributes_count, b.dropped_attributes_count,
             "dropped_attributes_count does not match"

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

@@ -255,6 +255,7 @@ mod logtests {
     // current thread
     #[test]
     #[cfg(feature = "reqwest-blocking-client")]
+    #[cfg(not(feature = "tonic-client"))]
     pub fn logs_batch_non_tokio_main_with_init_logs_outside_rt() -> Result<()> {
         logs_non_tokio_helper(false, false)
     }
@@ -295,7 +296,8 @@ mod logtests {
     // Client - Reqwest-blocking
     #[test]
     #[cfg(feature = "reqwest-blocking-client")]
-    pub fn logs_simple_non_tokio_main_with_init_logs_outsie_rt_blocking() -> Result<()> {
+    #[cfg(not(feature = "tonic-client"))]
+    pub fn logs_simple_non_tokio_main_with_init_logs_outside_rt_blocking() -> Result<()> {
         logs_non_tokio_helper(true, false)
     }
 
@@ -309,7 +311,7 @@ mod logtests {
         feature = "tonic-client",
         feature = "reqwest-client"
     ))]
-    pub fn logs_simple_non_tokio_main_with_init_logs_outsie_rt() -> Result<()> {
+    pub fn logs_simple_non_tokio_main_with_init_logs_outside_rt() -> Result<()> {
         logs_non_tokio_helper(true, false)
     }
 

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

@@ -37,7 +37,7 @@ pub async fn test_logs() -> Result<()> {
         info!(target: "my-target", "hello from {}. My price is {}.", "banana", 2.99);
     }
 
-    let _ = logger_provider.shutdown();
+    logger_provider.shutdown().unwrap();
     tokio::time::sleep(std::time::Duration::from_secs(5)).await;
     assert_logs_results(test_utils::LOGS_FILE, "expected/logs.json")?;
     Ok(())

opentelemetry-sdk/Cargo.toml

@@ -56,6 +56,7 @@ internal-logs = ["tracing"]
 experimental_metrics_periodicreader_with_async_runtime = ["metrics"]
 spec_unstable_metrics_views = ["metrics"]
 experimental_logs_batch_log_processor_with_async_runtime = ["logs"]
+experimental_logs_concurrent_log_processor = ["logs"]
 experimental_trace_batch_span_processor_with_async_runtime = ["trace"]
 experimental_metrics_disable_name_validation = ["metrics"]
 
@@ -88,6 +89,11 @@ harness = false
 name = "log_processor"
 harness = false
 
+[[bench]]
+name = "log_enabled"
+harness = false
+required-features = ["spec_unstable_logs_enabled", "experimental_logs_concurrent_log_processor"]
+
 [[bench]]
 name = "tracer_creation"
 harness = false

opentelemetry-sdk/benches/log_enabled.rs

@@ -0,0 +1,81 @@
+/*
+    The benchmark results:
+    criterion = "0.5.1"
+    Hardware: Apple M4 Pro
+    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     |
+*/
+
+use criterion::{criterion_group, criterion_main, Criterion};
+use opentelemetry::logs::{Logger, LoggerProvider};
+use opentelemetry_sdk::error::OTelSdkResult;
+use opentelemetry_sdk::logs::concurrent_log_processor::SimpleConcurrentLogProcessor;
+use opentelemetry_sdk::logs::{
+    LogBatch, LogExporter, LogProcessor, SdkLoggerProvider, SimpleLogProcessor,
+};
+use opentelemetry_sdk::Resource;
+#[cfg(not(target_os = "windows"))]
+use pprof::criterion::{Output, PProfProfiler};
+
+#[derive(Debug)]
+struct NoopExporter;
+impl LogExporter for NoopExporter {
+    async fn export(&self, _: LogBatch<'_>) -> OTelSdkResult {
+        Ok(())
+    }
+
+    fn shutdown(&self) -> OTelSdkResult {
+        Ok(())
+    }
+
+    fn event_enabled(
+        &self,
+        _level: opentelemetry::logs::Severity,
+        _target: &str,
+        _name: Option<&str>,
+    ) -> bool {
+        false
+    }
+
+    fn set_resource(&mut self, _: &Resource) {}
+}
+
+fn benchmark_exporter_enabled_false<T>(c: &mut Criterion, name: &str, processor: T)
+where
+    T: LogProcessor + Send + Sync + 'static,
+{
+    let provider = SdkLoggerProvider::builder()
+        .with_log_processor(processor)
+        .build();
+    let logger = provider.logger("test_logger");
+
+    c.bench_function(name, |b| {
+        b.iter(|| {
+            logger.event_enabled(opentelemetry::logs::Severity::Debug, "target", Some("name"));
+        });
+    });
+}
+
+fn criterion_benchmark(c: &mut Criterion) {
+    let processor = SimpleConcurrentLogProcessor::new(NoopExporter);
+    benchmark_exporter_enabled_false(c, "exporter_disabled_concurrent_processor", processor);
+    let simple = SimpleLogProcessor::new(NoopExporter);
+    benchmark_exporter_enabled_false(c, "exporter_disabled_simple_processor", simple);
+}
+
+#[cfg(not(target_os = "windows"))]
+criterion_group! {
+    name = benches;
+    config = Criterion::default().with_profiler(PProfProfiler::new(100, Output::Flamegraph(None)));
+    targets = criterion_benchmark
+}
+#[cfg(target_os = "windows")]
+criterion_group! {
+    name = benches;
+    config = Criterion::default();
+    targets = criterion_benchmark
+}
+criterion_main!(benches);

opentelemetry-sdk/src/logs/batch_log_processor.rs

@@ -614,6 +614,8 @@ impl Default for BatchConfigBuilder {
     /// * `OTEL_BLRP_SCHEDULE_DELAY`
     /// * `OTEL_BLRP_MAX_EXPORT_BATCH_SIZE`
     /// * `OTEL_BLRP_EXPORT_TIMEOUT`
+    ///
+    /// Note: Programmatic configuration overrides any value set via the environment variable.
     fn default() -> Self {
         BatchConfigBuilder {
             max_queue_size: OTEL_BLRP_MAX_QUEUE_SIZE_DEFAULT,
@@ -630,7 +632,11 @@ impl BatchConfigBuilder {
     /// Set max_queue_size for [`BatchConfigBuilder`].
     /// It's the maximum queue size to buffer logs for delayed processing.
     /// If the queue gets full it will drop the logs.
-    /// The default value of is 2048.
+    /// The default value is 2048.
+    ///
+    /// Corresponding environment variable: `OTEL_BLRP_MAX_QUEUE_SIZE`.
+    ///
+    /// Note: Programmatically setting this will override any value set via the environment variable.
     pub fn with_max_queue_size(mut self, max_queue_size: usize) -> Self {
         self.max_queue_size = max_queue_size;
         self
@@ -639,6 +645,10 @@ impl BatchConfigBuilder {
     /// Set scheduled_delay for [`BatchConfigBuilder`].
     /// It's the delay interval in milliseconds between two consecutive processing of batches.
     /// The default value is 1000 milliseconds.
+    ///
+    /// Corresponding environment variable: `OTEL_BLRP_SCHEDULE_DELAY`.
+    ///
+    /// Note: Programmatically setting this will override any value set via the environment variable.
     pub fn with_scheduled_delay(mut self, scheduled_delay: Duration) -> Self {
         self.scheduled_delay = scheduled_delay;
         self
@@ -647,6 +657,10 @@ impl BatchConfigBuilder {
     /// Set max_export_timeout for [`BatchConfigBuilder`].
     /// It's the maximum duration to export a batch of data.
     /// The default value is 30000 milliseconds.
+    ///
+    /// Corresponding environment variable: `OTEL_BLRP_EXPORT_TIMEOUT`.
+    ///
+    /// Note: Programmatically setting this will override any value set via the environment variable.
     #[cfg(feature = "experimental_logs_batch_log_processor_with_async_runtime")]
     pub fn with_max_export_timeout(mut self, max_export_timeout: Duration) -> Self {
         self.max_export_timeout = max_export_timeout;
@@ -658,6 +672,10 @@ impl BatchConfigBuilder {
     /// more than one batch worth of logs then it processes multiple batches
     /// of logs one batch after the other without any delay.
     /// The default value is 512.
+    ///
+    /// Corresponding environment variable: `OTEL_BLRP_MAX_EXPORT_BATCH_SIZE`.
+    ///
+    /// Note: Programmatically setting this will override any value set via the environment variable.
     pub fn with_max_export_batch_size(mut self, max_export_batch_size: usize) -> Self {
         self.max_export_batch_size = max_export_batch_size;
         self
@@ -774,6 +792,27 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_code_based_config_overrides_env_vars() {
+        let env_vars = vec![
+            (OTEL_BLRP_SCHEDULE_DELAY, Some("2000")),
+            (OTEL_BLRP_MAX_QUEUE_SIZE, Some("4096")),
+            (OTEL_BLRP_MAX_EXPORT_BATCH_SIZE, Some("1024")),
+        ];
+
+        temp_env::with_vars(env_vars, || {
+            let config = BatchConfigBuilder::default()
+                .with_max_queue_size(2048)
+                .with_scheduled_delay(Duration::from_millis(1000))
+                .with_max_export_batch_size(512)
+                .build();
+
+            assert_eq!(config.scheduled_delay, Duration::from_millis(1000));
+            assert_eq!(config.max_queue_size, 2048);
+            assert_eq!(config.max_export_batch_size, 512);
+        });
+    }
+
     #[test]
     fn test_batch_config_configurable_by_env_vars() {
         let env_vars = vec![

opentelemetry-sdk/src/logs/concurrent_log_processor.rs

@@ -0,0 +1,59 @@
+use opentelemetry::{otel_info, InstrumentationScope};
+
+use crate::error::OTelSdkResult;
+
+use super::{LogBatch, LogExporter, LogProcessor, SdkLogRecord};
+
+/// A concurrent log processor calls exporter's export method on each emit. This
+/// processor does not buffer logs. Note: This invokes exporter's export method
+/// on the current thread without synchronization. i.e multiple export() calls
+/// can happen simultaneously from different threads. This is not a problem if
+/// the exporter is designed to handle that. As of now, exporters in the
+/// opentelemetry-rust project (stdout/otlp) are not thread-safe.
+/// This is intended to be used when exporting to operating system
+/// tracing facilities like Windows ETW, Linux TracePoints etc.
+#[derive(Debug)]
+pub struct SimpleConcurrentLogProcessor<T: LogExporter> {
+    exporter: T,
+}
+
+impl<T: LogExporter> SimpleConcurrentLogProcessor<T> {
+    /// Creates a new `ConcurrentExportProcessor` with the given exporter.
+    pub fn new(exporter: T) -> Self {
+        Self { exporter }
+    }
+}
+
+impl<T: LogExporter> LogProcessor for SimpleConcurrentLogProcessor<T> {
+    fn emit(&self, record: &mut SdkLogRecord, instrumentation: &InstrumentationScope) {
+        let log_tuple = &[(record as &SdkLogRecord, instrumentation)];
+        let result = futures_executor::block_on(self.exporter.export(LogBatch::new(log_tuple)));
+        if let Err(err) = result {
+            otel_info!(
+                name: "SimpleConcurrentLogProcessor.Emit.ExportError",
+                error = format!("{}",err)
+            );
+        }
+    }
+
+    fn force_flush(&self) -> OTelSdkResult {
+        // TODO: invoke flush on exporter
+        // once https://github.com/open-telemetry/opentelemetry-rust/issues/2261
+        // is resolved
+        Ok(())
+    }
+
+    fn shutdown(&self) -> OTelSdkResult {
+        self.exporter.shutdown()
+    }
+
+    #[cfg(feature = "spec_unstable_logs_enabled")]
+    fn event_enabled(
+        &self,
+        level: opentelemetry::logs::Severity,
+        target: &str,
+        name: Option<&str>,
+    ) -> bool {
+        self.exporter.event_enabled(level, target, name)
+    }
+}

opentelemetry-sdk/src/logs/logger_provider.rs

@@ -200,15 +200,22 @@ impl LoggerProviderBuilder {
         LoggerProviderBuilder { processors, ..self }
     }
 
-    /// Adds a [BatchLogProcessor] with the configured exporter to the pipeline.
+    /// Adds a [BatchLogProcessor] with the configured exporter to the pipeline,
+    /// using the default [super::BatchConfig].
+    ///
+    /// The following environment variables can be used to configure the batching configuration:
+    ///
+    /// * `OTEL_BLRP_SCHEDULE_DELAY` - Corresponds to `with_scheduled_delay`.
+    /// * `OTEL_BLRP_MAX_QUEUE_SIZE` - Corresponds to `with_max_queue_size`.
+    /// * `OTEL_BLRP_MAX_EXPORT_BATCH_SIZE` - Corresponds to `with_max_export_batch_size`.
     ///
     /// # Arguments
     ///
-    /// * `exporter` - The exporter to be used by the BatchLogProcessor.
+    /// * `exporter` - The exporter to be used by the `BatchLogProcessor`.
     ///
     /// # Returns
     ///
-    /// A new `Builder` instance with the BatchLogProcessor added to the pipeline.
+    /// A new `LoggerProviderBuilder` instance with the `BatchLogProcessor` added to the pipeline.
     ///
     /// Processors are invoked in the order they are added.
     pub fn with_batch_exporter<T: LogExporter + 'static>(self, exporter: T) -> Self {