review comments, move test to sdk itself

Author: cijothomas

opentelemetry-appender-tracing/src/layer.rs

@@ -327,56 +327,6 @@ mod tests {
         }
     }
 
-    #[test]
-    #[ignore = "While this test runs fine, this uses global subscriber and does not play well with other tests and hence ignored in CI."]
-    fn simple_processor_deadlock() {
-        // TODO: This test maybe better suited in the opentelemetry-sdk crate tests
-        let exporter: ReentrantLogExporter = ReentrantLogExporter;
-        let logger_provider = SdkLoggerProvider::builder()
-            .with_simple_exporter(exporter)
-            .build();
-
-        let layer = layer::OpenTelemetryTracingBridge::new(&logger_provider);
-        tracing_subscriber::registry().with(layer).init();
-
-        warn!(name: "my-event-name", target: "my-system", event_id = 20, user_name = "otel", user_email = "[email protected]");
-    }
-
-    #[test]
-    #[ignore = "While this test runs fine, this uses global subscriber and does not play well with other tests."]
-    fn simple_processor_no_deadlock() {
-        let exporter: ReentrantLogExporter = ReentrantLogExporter;
-        let logger_provider = SdkLoggerProvider::builder()
-            .with_simple_exporter(exporter.clone())
-            .build();
-
-        let layer = layer::OpenTelemetryTracingBridge::new(&logger_provider);
-
-        // This filter will prevent the deadlock as the reentrant log will be
-        // ignored.
-        let filter = EnvFilter::new("debug").add_directive("reentrant=error".parse().unwrap());
-        // Setting subscriber as global as that is the only way to test this scenario.
-        tracing_subscriber::registry()
-            .with(filter)
-            .with(layer)
-            .init();
-        warn!(name: "my-event-name", target: "my-system", event_id = 20, user_name = "otel", user_email = "[email protected]");
-    }
-
-    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
-    #[ignore = "While this test runs fine, this uses global subscriber and does not play well with other tests."]
-    async fn batch_processor_no_deadlock() {
-        let exporter: ReentrantLogExporter = ReentrantLogExporter;
-        let logger_provider = SdkLoggerProvider::builder()
-            .with_batch_exporter(exporter.clone())
-            .build();
-
-        let layer = layer::OpenTelemetryTracingBridge::new(&logger_provider);
-
-        tracing_subscriber::registry().with(layer).init();
-        warn!(name: "my-event-name", target: "my-system", event_id = 20, user_name = "otel", user_email = "[email protected]");
-    }
-
     #[test]
     fn tracing_appender_standalone() {
         // Arrange

opentelemetry-sdk/CHANGELOG.md

@@ -2,15 +2,16 @@
 
 ## vNext
 
-`SdkLogger` modified to respect telemetry suppression based on `Context`. In
-other words, if the current context has telemetry suppression enabled, then logs
-will be ignored. The flag is typically set by OTel components to prevent
-telemetry from itself being fed back into OTel. `BatchLogProcessor`,
-`BatchSpanProcessor`, and `PeriodicReader` modified to set the suppression flag
-in their dedicated thread, so that telemetry generated from those threads will
-not be fed back into OTel. Similarly, `SimpleSpanProcessor` and
-`SimpleLogProcessor` also modified to suppress telemetry before invoking
-exporters.
+`SdkLogger` modified to respect telemetry suppression based on `Context`. A
+similar change is done for Tracing signal too, by returning `Sampling::Drop`
+when telemetry is suppressed. In other words, if the current context has
+telemetry suppression enabled, then logs/spans will be ignored. The flag is
+typically set by OTel components to prevent telemetry from itself being fed back
+into OTel. `BatchLogProcessor`, `BatchSpanProcessor`, and `PeriodicReader`
+modified to set the suppression flag in their dedicated thread, so that
+telemetry generated from those threads will not be fed back into OTel.
+Similarly, `SimpleLogProcessor` also modified to suppress telemetry before
+invoking exporters.
 
 ## 0.29.0
 

opentelemetry-sdk/src/logs/mod.rs

@@ -36,6 +36,7 @@ pub mod log_processor_with_async_runtime;
 #[cfg(all(test, feature = "testing"))]
 mod tests {
     use super::*;
+    use crate::error::OTelSdkResult;
     use crate::Resource;
     use opentelemetry::baggage::BaggageExt;
     use opentelemetry::logs::LogRecord;
@@ -44,6 +45,7 @@ mod tests {
     use opentelemetry::{Context, InstrumentationScope};
     use std::borrow::Borrow;
     use std::collections::HashMap;
+    use std::sync::{Arc, Mutex};
 
     #[test]
     fn logging_sdk_test() {
@@ -239,4 +241,59 @@ mod tests {
             "There should be a no logs as log emission is done inside a suppressed context"
         );
     }
+
+    #[derive(Debug, Clone)]
+    struct ReentrantLogProcessor {
+        logger: Arc<Mutex<Option<SdkLogger>>>,
+    }
+
+    impl ReentrantLogProcessor {
+        fn new() -> Self {
+            Self {
+                logger: Arc::new(Mutex::new(None)),
+            }
+        }
+
+        fn set_logger(&self, logger: SdkLogger) {
+            let mut guard = self.logger.lock().unwrap();
+            *guard = Some(logger);
+        }
+    }
+
+    impl LogProcessor for ReentrantLogProcessor {
+        fn emit(&self, _data: &mut SdkLogRecord, _instrumentation: &InstrumentationScope) {
+            let _suppress = Context::enter_telemetry_suppressed_scope();
+            // Without the suppression above, the logger.emit(log_record) below will cause a deadlock,
+            // as it emits another log, which will attempt to acquire the same lock that is
+            // already held by itself!
+            let logger = self.logger.lock().unwrap();
+            if let Some(logger) = logger.as_ref() {
+                let mut log_record = logger.create_log_record();
+                log_record.set_severity_number(Severity::Error);
+                logger.emit(log_record);
+            }
+        }
+
+        fn force_flush(&self) -> OTelSdkResult {
+            Ok(())
+        }
+
+        fn shutdown(&self) -> OTelSdkResult {
+            Ok(())
+        }
+    }
+
+    #[test]
+    fn log_does_not_deadlock_when_suppression_enabled() {
+        let processor: ReentrantLogProcessor = ReentrantLogProcessor::new();
+        let logger_provider = SdkLoggerProvider::builder()
+            .with_log_processor(processor.clone())
+            .build();
+        processor.set_logger(logger_provider.logger("processor-logger"));
+
+        let logger = logger_provider.logger("test-logger");
+        let mut log_record = logger.create_log_record();
+        log_record.set_severity_number(Severity::Error);
+        logger.emit(log_record);
+    }
 }

opentelemetry-sdk/src/logs/simple_log_processor.rs

@@ -153,12 +153,13 @@ impl<T: LogExporter> LogProcessor for SimpleLogProcessor<T> {
 #[cfg(all(test, feature = "testing", feature = "logs"))]
 mod tests {
     use crate::logs::log_processor::tests::MockLogExporter;
-    use crate::logs::{LogBatch, LogExporter, SdkLogRecord};
+    use crate::logs::{LogBatch, LogExporter, SdkLogRecord, SdkLogger};
     use crate::{
         error::OTelSdkResult,
         logs::{InMemoryLogExporterBuilder, LogProcessor, SdkLoggerProvider, SimpleLogProcessor},
         Resource,
     };
+    use opentelemetry::logs::{LogRecord, Logger, LoggerProvider};
     use opentelemetry::InstrumentationScope;
     use opentelemetry::KeyValue;
     use std::sync::atomic::{AtomicUsize, Ordering};
@@ -439,4 +440,56 @@ mod tests {
 
         assert_eq!(exporter.len(), 1);
     }
+
+    #[derive(Debug, Clone)]
+    struct ReentrantLogExporter {
+        logger: Arc<Mutex<Option<SdkLogger>>>,
+    }
+
+    impl ReentrantLogExporter {
+        fn new() -> Self {
+            Self {
+                logger: Arc::new(Mutex::new(None)),
+            }
+        }
+
+        fn set_logger(&self, logger: SdkLogger) {
+            let mut guard = self.logger.lock().unwrap();
+            *guard = Some(logger);
+        }
+    }
+
+    impl LogExporter for ReentrantLogExporter {
+        fn shutdown(&self) -> OTelSdkResult {
+            Ok(())
+        }
+
+        async fn export(&self, _batch: LogBatch<'_>) -> OTelSdkResult {
+            let logger = self.logger.lock().unwrap();
+            if let Some(logger) = logger.as_ref() {
+                let mut log_record = logger.create_log_record();
+                log_record.set_severity_number(opentelemetry::logs::Severity::Error);
+                logger.emit(log_record);
+            }
+
+            Ok(())
+        }
+    }
+
+    #[test]
+    fn does_not_deadlock_when_exporter_is_rentrant_simple_log_processor() {
+        // This tests that even when exporter produces logs while
+        // exporting, it does not deadlock, as SimpleLogProcessor
+        // activates SuppressGuard before calling the exporter.
+        let exporter: ReentrantLogExporter = ReentrantLogExporter::new();
+        let logger_provider = SdkLoggerProvider::builder()
+            .with_simple_exporter(exporter.clone())
+            .build();
+        exporter.set_logger(logger_provider.logger("processor-logger"));
+
+        let logger = logger_provider.logger("test-logger");
+        let mut log_record = logger.create_log_record();
+        log_record.set_severity_number(opentelemetry::logs::Severity::Error);
+        logger.emit(log_record);
+    }
 }

opentelemetry-sdk/src/trace/mod.rs

@@ -547,8 +547,7 @@ mod tests {
         let tracer = tracer_provider.tracer("test");
         {
             let _suppressed_context = Context::enter_telemetry_suppressed_scope();
-            let span = tracer.span_builder("span_name").start(&tracer);
-            drop(span);
+            let _span = tracer.span_builder("span_name").start(&tracer);
         }
 
         // Assert