Nit improvements to batchprocessors (#2466)

open-telemetry · Dec 24, 2024 · 8d5f222 · 8d5f222
1 parent ef49833
commit 8d5f222
Show file tree

Hide file tree

Showing 3 changed files with 30 additions and 16 deletions.
diff --git a/opentelemetry-sdk/src/logs/log_processor.rs b/opentelemetry-sdk/src/logs/log_processor.rs
@@ -337,35 +337,42 @@ impl BatchLogProcessor {
                     match message_receiver.recv_timeout(remaining_time) {
                         Ok(BatchMessage::ExportLog(log)) => {
                             logs.push(log);
-                            if logs.len() == config.max_export_batch_size
-                                || last_export_time.elapsed() >= config.scheduled_delay
-                            {
+                            if logs.len() == config.max_export_batch_size {
+                                otel_debug!(
+                                    name: "BatchLogProcessor.ExportingDueToBatchSize",
+                                );
                                 let _ = export_with_timeout_sync(
-                                    remaining_time,
+                                    config.max_export_timeout,
                                     &mut exporter,
                                     logs.split_off(0),
                                     &mut last_export_time,
                                 );
                             }
                         }
                         Ok(BatchMessage::ForceFlush(sender)) => {
+                            otel_debug!(name: "BatchLogProcessor.ExportingDueToForceFlush");
                             let result = export_with_timeout_sync(
-                                remaining_time,
+                                config.max_export_timeout,
                                 &mut exporter,
                                 logs.split_off(0),
                                 &mut last_export_time,
                             );
                             let _ = sender.send(result);
                         }
                         Ok(BatchMessage::Shutdown(sender)) => {
+                            otel_debug!(name: "BatchLogProcessor.ExportingDueToShutdown");
                             let result = export_with_timeout_sync(
-                                remaining_time,
+                                config.max_export_timeout,
                                 &mut exporter,
                                 logs.split_off(0),
                                 &mut last_export_time,
                             );
                             let _ = sender.send(result);
 
+                            otel_debug!(
+                                name: "BatchLogProcessor.ThreadExiting",
+                                reason = "ShutdownRequested"
+                            );
                             //
                             // break out the loop and return from the current background thread.
                             //
@@ -375,19 +382,24 @@ impl BatchLogProcessor {
                             exporter.set_resource(&resource);
                         }
                         Err(RecvTimeoutError::Timeout) => {
+                            otel_debug!(
+                                name: "BatchLogProcessor.ExportingDueToTimer",
+                            );
                             let _ = export_with_timeout_sync(
-                                remaining_time,
+                                config.max_export_timeout,
                                 &mut exporter,
                                 logs.split_off(0),
                                 &mut last_export_time,
                             );
                         }
-                        Err(err) => {
-                            // TODO: this should not happen! Log the error and continue for now.
-                            otel_error!(
-                                name: "BatchLogProcessor.InternalError",
-                                error = format!("{}", err)
+                        Err(RecvTimeoutError::Disconnected) => {
+                            // Channel disconnected, only thing to do is break
+                            // out (i.e exit the thread)
+                            otel_debug!(
+                                name: "BatchLogProcessor.ThreadExiting",
+                                reason = "MessageSenderDisconnected"
                             );
+                            break;
                         }
                     }
                 }

diff --git a/opentelemetry-sdk/src/metrics/periodic_reader.rs b/opentelemetry-sdk/src/metrics/periodic_reader.rs
@@ -260,7 +260,7 @@ impl PeriodicReader {
                             // out (i.e exit the thread)
                             otel_debug!(
                                 name: "PeriodReaderThreadExiting",
-                                reason = "MessageReceiverDisconnected"
+                                reason = "MessageSenderDisconnected"
                             );
                             break;
                         }

diff --git a/opentelemetry-sdk/src/trace/span_processor.rs b/opentelemetry-sdk/src/trace/span_processor.rs
@@ -319,9 +319,11 @@ impl BatchSpanProcessor {
                             }
                         }
                         Err(RecvTimeoutError::Disconnected) => {
-                            otel_error!(
-                                name: "BatchSpanProcessor.InternalError.ChannelDisconnected",
-                                message = "Channel disconnected, shutting down processor thread."
+                            // Channel disconnected, only thing to do is break
+                            // out (i.e exit the thread)
+                            otel_debug!(
+                                name: "BatchSpanProcessor.ThreadExiting",
+                                reason = "MessageSenderDisconnected"
                             );
                             break;
                         }