add force_flush_with_timeout to processors and exporters

Author: kovaszab

examples/tracing-http-propagator/src/server.rs

@@ -132,7 +132,7 @@ impl LogProcessor for EnrichWithBaggageLogProcessor {
 #[derive(Debug)]
 struct EnrichWithBaggageSpanProcessor;
 impl SpanProcessor for EnrichWithBaggageSpanProcessor {
-    fn force_flush(&self) -> OTelSdkResult {
+    fn force_flush_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
         Ok(())
     }
 

opentelemetry-sdk/src/logs/batch_log_processor.rs

@@ -129,7 +129,6 @@ pub struct BatchLogProcessor {
     logs_sender: SyncSender<LogsData>, // Data channel to store log records and instrumentation scopes
     message_sender: SyncSender<BatchMessage>, // Control channel to store control messages for the worker thread
     handle: Mutex<Option<thread::JoinHandle<()>>>,
-    forceflush_timeout: Duration,
     export_log_message_sent: Arc<AtomicBool>,
     current_batch_size: Arc<AtomicUsize>,
     max_export_batch_size: usize,
@@ -221,21 +220,19 @@ impl LogProcessor for BatchLogProcessor {
         }
     }
 
-    fn force_flush(&self) -> OTelSdkResult {
+    fn force_flush_with_timeout(&self, timeout: Duration) -> OTelSdkResult {
         let (sender, receiver) = mpsc::sync_channel(1);
         match self
             .message_sender
             .try_send(BatchMessage::ForceFlush(sender))
         {
-            Ok(_) => receiver
-                .recv_timeout(self.forceflush_timeout)
-                .map_err(|err| {
-                    if err == RecvTimeoutError::Timeout {
-                        OTelSdkError::Timeout(self.forceflush_timeout)
-                    } else {
-                        OTelSdkError::InternalFailure(format!("{err}"))
-                    }
-                })?,
+            Ok(_) => receiver.recv_timeout(timeout).map_err(|err| {
+                if err == RecvTimeoutError::Timeout {
+                    OTelSdkError::Timeout(timeout)
+                } else {
+                    OTelSdkError::InternalFailure(format!("{err}"))
+                }
+            })?,
             Err(mpsc::TrySendError::Full(_)) => {
                 // If the control message could not be sent, emit a warning.
                 otel_debug!(
@@ -489,7 +486,6 @@ impl BatchLogProcessor {
             logs_sender,
             message_sender,
             handle: Mutex::new(Some(handle)),
-            forceflush_timeout: Duration::from_secs(5), // TODO: make this configurable
             dropped_logs_count: AtomicUsize::new(0),
             max_queue_size,
             export_log_message_sent: Arc::new(AtomicBool::new(false)),

opentelemetry-sdk/src/logs/log_processor.rs

@@ -53,7 +53,13 @@ pub trait LogProcessor: Send + Sync + Debug {
     /// - `instrumentation`: The instrumentation scope associated with the log record.
     fn emit(&self, data: &mut SdkLogRecord, instrumentation: &InstrumentationScope);
     /// Force the logs lying in the cache to be exported.
-    fn force_flush(&self) -> OTelSdkResult;
+    fn force_flush_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
+        Ok(())
+    }
+    /// Force the logs lying in the cache to be exported with default timeout.
+    fn force_flush(&self) -> OTelSdkResult {
+        self.force_flush_with_timeout(Duration::from_secs(5))
+    }
     /// Shuts down the processor.
     /// After shutdown returns the log processor should stop processing any logs.
     /// It's up to the implementation on when to drop the LogProcessor.

opentelemetry-sdk/src/logs/logger_provider.rs

@@ -83,11 +83,11 @@ impl SdkLoggerProvider {
     }
 
     /// Force flush all remaining logs in log processors and return results.
-    pub fn force_flush(&self) -> OTelSdkResult {
+    pub fn force_flush_with_timeout(&self, timeout: Duration) -> OTelSdkResult {
         let result: Vec<_> = self
             .log_processors()
             .iter()
-            .map(|processor| processor.force_flush())
+            .map(|processor| processor.force_flush_with_timeout(timeout))
             .collect();
         if result.iter().all(|r| r.is_ok()) {
             Ok(())
@@ -96,6 +96,11 @@ impl SdkLoggerProvider {
         }
     }
 
+    /// Force flush all remaining logs with default timeout.
+    pub fn force_flush(&self) -> OTelSdkResult {
+        self.force_flush_with_timeout(Duration::from_secs(5))
+    }
+
     /// Shuts down this `LoggerProvider`
     pub fn shutdown_with_timeout(&self, timeout: Duration) -> OTelSdkResult {
         otel_debug!(

opentelemetry-sdk/src/trace/export.rs

@@ -66,10 +66,15 @@ pub trait SpanExporter: Send + Sync + Debug {
     /// implemented as a blocking API or an asynchronous API which notifies the caller via
     /// a callback or an event. OpenTelemetry client authors can decide if they want to
     /// make the flush timeout configurable.
-    fn force_flush(&mut self) -> OTelSdkResult {
+    fn force_flush_with_timeout(&mut self, _timeout: Duration) -> OTelSdkResult {
         Ok(())
     }
 
+    /// Force flush the exporter with default timeout.
+    fn force_flush(&mut self) -> OTelSdkResult {
+        self.force_flush_with_timeout(Duration::from_secs(5))
+    }
+
     /// Set the resource for the exporter.
     fn set_resource(&mut self, _resource: &Resource) {}
 }

opentelemetry-sdk/src/trace/mod.rs

@@ -143,7 +143,7 @@ mod tests {
             // let _c = Context::current();
         }
 
-        fn force_flush(&self) -> crate::error::OTelSdkResult {
+        fn force_flush_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
             Ok(())
         }
 

opentelemetry-sdk/src/trace/provider.rs

@@ -194,7 +194,7 @@ impl SdkTracerProvider {
         self.inner.is_shutdown.load(Ordering::Relaxed)
     }
 
-    /// Force flush all remaining spans in span processors and return results.
+    /// Force flush all remaining spans in span processors with a default timeout and return results.
     ///
     /// # Examples
     ///
@@ -228,10 +228,15 @@ impl SdkTracerProvider {
     /// }
     /// ```
     pub fn force_flush(&self) -> OTelSdkResult {
+        self.force_flush_with_timeout(Duration::from_secs(5))
+    }
+
+    /// force flush processors with a specified timeout
+    pub fn force_flush_with_timeout(&self, timeout: Duration) -> OTelSdkResult {
         let result: Vec<_> = self
             .span_processors()
             .iter()
-            .map(|processor| processor.force_flush())
+            .map(|processor| processor.force_flush_with_timeout(timeout))
             .collect();
         if result.iter().all(|r| r.is_ok()) {
             Ok(())
@@ -530,7 +535,7 @@ mod tests {
             // ignore
         }
 
-        fn force_flush(&self) -> OTelSdkResult {
+        fn force_flush_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
             if self.success {
                 Ok(())
             } else {
@@ -793,7 +798,7 @@ mod tests {
             // No operation needed for this processor
         }
 
-        fn force_flush(&self) -> OTelSdkResult {
+        fn force_flush_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
             Ok(())
         }
 

opentelemetry-sdk/src/trace/span_processor.rs

@@ -85,7 +85,11 @@ pub trait SpanProcessor: Send + Sync + std::fmt::Debug {
     /// TODO - This method should take reference to `SpanData`
     fn on_end(&self, span: SpanData);
     /// Force the spans lying in the cache to be exported.
-    fn force_flush(&self) -> OTelSdkResult;
+    fn force_flush_with_timeout(&self, _timeout: Duration) -> OTelSdkResult;
+    /// Force flush the spans with a default timeout.
+    fn force_flush(&self) -> OTelSdkResult {
+        self.force_flush_with_timeout(Duration::from_secs(5))
+    }
     /// Shuts down the processor. Called when SDK is shut down. This is an
     /// opportunity for processors to do any cleanup required.
     ///
@@ -153,7 +157,7 @@ impl<T: SpanExporter> SpanProcessor for SimpleSpanProcessor<T> {
         }
     }
 
-    fn force_flush(&self) -> OTelSdkResult {
+    fn force_flush_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
         // Nothing to flush for simple span processor.
         Ok(())
     }
@@ -286,7 +290,6 @@ pub struct BatchSpanProcessor {
     span_sender: SyncSender<SpanData>, // Data channel to store spans
     message_sender: SyncSender<BatchMessage>, // Control channel to store control messages.
     handle: Mutex<Option<thread::JoinHandle<()>>>,
-    forceflush_timeout: Duration,
     export_span_message_sent: Arc<AtomicBool>,
     current_batch_size: Arc<AtomicUsize>,
     max_export_batch_size: usize,
@@ -424,7 +427,6 @@ impl BatchSpanProcessor {
             span_sender,
             message_sender,
             handle: Mutex::new(Some(handle)),
-            forceflush_timeout: Duration::from_secs(5), // TODO: make this configurable
             dropped_spans_count: AtomicUsize::new(0),
             max_queue_size,
             export_span_message_sent: Arc::new(AtomicBool::new(false)),
@@ -593,21 +595,19 @@ impl SpanProcessor for BatchSpanProcessor {
     }
 
     /// Flushes all pending spans.
-    fn force_flush(&self) -> OTelSdkResult {
+    fn force_flush_with_timeout(&self, timeout: Duration) -> OTelSdkResult {
         let (sender, receiver) = std::sync::mpsc::sync_channel(1);
         match self
             .message_sender
             .try_send(BatchMessage::ForceFlush(sender))
         {
-            Ok(_) => receiver
-                .recv_timeout(self.forceflush_timeout)
-                .map_err(|err| {
-                    if err == std::sync::mpsc::RecvTimeoutError::Timeout {
-                        OTelSdkError::Timeout(self.forceflush_timeout)
-                    } else {
-                        OTelSdkError::InternalFailure(format!("{err}"))
-                    }
-                })?,
+            Ok(_) => receiver.recv_timeout(timeout).map_err(|err| {
+                if err == std::sync::mpsc::RecvTimeoutError::Timeout {
+                    OTelSdkError::Timeout(timeout)
+                } else {
+                    OTelSdkError::InternalFailure(format!("{err}"))
+                }
+            })?,
             Err(std::sync::mpsc::TrySendError::Full(_)) => {
                 // If the control message could not be sent, emit a warning.
                 otel_debug!(

opentelemetry-sdk/src/trace/span_processor_with_async_runtime.rs

@@ -123,7 +123,7 @@ impl<R: RuntimeChannel> SpanProcessor for BatchSpanProcessor<R> {
         }
     }
 
-    fn force_flush(&self) -> OTelSdkResult {
+    fn force_flush_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
         let (res_sender, res_receiver) = oneshot::channel();
         self.message_sender
             .try_send(BatchMessage::Flush(Some(res_sender)))

stress/src/traces.rs

@@ -41,7 +41,7 @@ impl SpanProcessor for NoOpSpanProcessor {
         // No-op
     }
 
-    fn force_flush(&self) -> OTelSdkResult {
+    fn force_flush_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
         Ok(())
     }