Flush fix for SimpleLogProcessor

opentelemetry-sdk/Cargo.toml

@@ -40,6 +40,10 @@ indexmap = "2.0"
 criterion = { version = "0.5", features = ["html_reports"] }
 [target.'cfg(not(target_os = "windows"))'.dev-dependencies]
 pprof = { version = "0.13", features = ["flamegraph", "criterion"] }
+opentelemetry-appender-tracing = { path = "../opentelemetry-appender-tracing"}
+tracing = {version = "0.1.37", default-features = false, features = ["std"]}
+tracing-core = "0.1.31"
+tracing-subscriber = { version = "0.3.0", default-features = false, features = ["registry", "std"] }
 
 [features]
 default = ["trace"]

opentelemetry-sdk/src/logs/log_processor.rs

@@ -13,7 +13,7 @@
     global,
     logs::{LogError, LogResult},
 };
-use std::thread;
+use std::sync::Mutex;
 use std::{
     fmt::{self, Debug, Formatter},
     time::Duration,
@@ -42,62 +42,32 @@
 /// emitted. If you find this limiting, consider the batch processor instead.
 #[derive(Debug)]
 pub struct SimpleLogProcessor {
-    sender: crossbeam_channel::Sender<Option<LogData>>,
-    shutdown: crossbeam_channel::Receiver<()>,
+    exporter: Mutex<Box<dyn LogExporter>>,
 }
 
 impl SimpleLogProcessor {
-    pub(crate) fn new(mut exporter: Box<dyn LogExporter>) -> Self {
-        let (log_tx, log_rx) = crossbeam_channel::unbounded();
-        let (shutdown_tx, shutdown_rx) = crossbeam_channel::bounded(0);
-
-        let _ = thread::Builder::new()
-            .name("opentelemetry-log-exporter".to_string())
-            .spawn(move || {
-                while let Ok(Some(log)) = log_rx.recv() {
-                    if let Err(err) = futures_executor::block_on(exporter.export(vec![log])) {
-                        global::handle_error(err);
-                    }
-                }
-
-                exporter.shutdown();
-
-                if let Err(err) = shutdown_tx.send(()) {
-                    global::handle_error(LogError::from(format!(
-                        "could not send shutdown: {:?}",
-                        err
-                    )));
-                }
-            });
-
+    pub(crate) fn new(exporter: Box<dyn LogExporter>) -> Self {
         SimpleLogProcessor {
-            sender: log_tx,
-            shutdown: shutdown_rx,
+            exporter: Mutex::new(exporter),
         }
     }
 }
 
 impl LogProcessor for SimpleLogProcessor {
     fn emit(&self, data: LogData) {
-        if let Err(err) = self.sender.send(Some(data)) {
-            global::handle_error(LogError::from(format!("error processing log {:?}", err)));
+        if let Err(err) =
+            futures_executor::block_on(self.exporter.lock().unwrap().export(vec![data]))
+        {
+            global::handle_error(err);
         }
     }
 
     fn force_flush(&self) -> LogResult<()> {
-        // Ignored since all logs in Simple Processor will be exported as they ended.
         Ok(())
     }
 
     fn shutdown(&mut self) -> LogResult<()> {
-        if self.sender.send(None).is_ok() {
-            if let Err(err) = self.shutdown.recv() {
-                global::handle_error(LogError::from(format!(
-                    "error shutting down log processor: {:?}",
-                    err
-                )))
-            }
-        }
+        self.exporter.lock().unwrap().shutdown();
         Ok(())
     }
 
@@ -108,7 +78,7 @@
 }
 
 /// A [`LogProcessor`] that asynchronously buffers log records and reports
-/// them at a preconfigured interval.
+/// them at a pre-configured interval.
 pub struct BatchLogProcessor<R: RuntimeChannel<BatchMessage>> {
     message_sender: R::Sender,
 }

opentelemetry-sdk/src/logs/mod.rs

@@ -10,3 +10,49 @@ pub use log_processor::{
     BatchConfig, BatchLogProcessor, BatchLogProcessorBuilder, BatchMessage, LogProcessor,
     SimpleLogProcessor,
 };
+
+#[cfg(all(test, feature = "testing"))]
+mod tests {
+    // use std::thread::sleep;
+
+    use super::*;
+    use crate::testing::logs::InMemoryLogsExporter;
+    use opentelemetry::{logs::AnyValue, Key};
+    use opentelemetry_appender_tracing::layer;
+    use tracing::error;
+    use tracing_subscriber::prelude::*;
+
+    #[test]
+    fn logging_sdk_test() {
+        // Arrange
+        let exporter: InMemoryLogsExporter = InMemoryLogsExporter::default();
+        let logger_provider = LoggerProvider::builder()
+            .with_log_processor(SimpleLogProcessor::new(Box::new(exporter.clone())))
+            .build();
+
+        // Act
+        let layer = layer::OpenTelemetryTracingBridge::new(&logger_provider);
+        tracing_subscriber::registry().with(layer).init();
+        error!(target: "my-system", event_id = 20, event_name = "my-event_name");
+
+        logger_provider.force_flush();
+        // TODO: To remove this comment.
+        // The test will fail without the sleep prior to the flush fix.
+        // sleep(std::time::Duration::from_millis(10));
+
+        // Assert
+        let exported_logs = exporter
+            .get_emitted_logs()
+            .expect("Logs are expected to be exported.");
+        assert_eq!(exported_logs.len(), 1);
+        let log = exported_logs
+            .get(0)
+            .expect("Atleast one log is expected to be present.");
+        let attributes: Vec<(Key, AnyValue)> = log
+            .record
+            .attributes
+            .clone()
+            .expect("Attributes are expected");
+        assert_eq!(attributes.len(), 2);
+    }
+}