opentelemetry_sdk/metrics/
periodic_reader.rs

1use std::{
2    env, fmt,
3    sync::{
4        mpsc::{self, Receiver, Sender},
5        Arc, Mutex, Weak,
6    },
7    thread,
8    time::{Duration, Instant},
9};
10
11use opentelemetry::{otel_debug, otel_error, otel_info, otel_warn, Context};
12
13use crate::{
14    error::{OTelSdkError, OTelSdkResult},
15    metrics::{exporter::PushMetricExporter, reader::SdkProducer},
16    Resource,
17};
18
19use super::{
20    data::ResourceMetrics, instrument::InstrumentKind, pipeline::Pipeline, reader::MetricReader,
21    Temporality,
22};
23
24/// Environment variable for configuring the delay interval (in milliseconds)
25/// between two consecutive exports for [PeriodicReader].
26pub const OTEL_METRIC_EXPORT_INTERVAL: &str = "OTEL_METRIC_EXPORT_INTERVAL";
27
28/// Default delay interval between two consecutive exports for [PeriodicReader].
29pub const OTEL_METRIC_EXPORT_INTERVAL_DEFAULT: Duration = Duration::from_secs(60);
30
31/// Configuration options for [PeriodicReader].
32#[derive(Debug)]
33pub struct PeriodicReaderBuilder<E> {
34    interval: Duration,
35    exporter: E,
36}
37
38impl<E> PeriodicReaderBuilder<E>
39where
40    E: PushMetricExporter,
41{
42    fn new(exporter: E) -> Self {
43        let interval = env::var(OTEL_METRIC_EXPORT_INTERVAL)
44            .ok()
45            .and_then(|v| v.parse().map(Duration::from_millis).ok())
46            .unwrap_or(OTEL_METRIC_EXPORT_INTERVAL_DEFAULT);
47
48        PeriodicReaderBuilder { interval, exporter }
49    }
50
51    /// Configures the intervening time between exports for a [PeriodicReader].
52    ///
53    /// This option overrides any value set for the `OTEL_METRIC_EXPORT_INTERVAL`
54    /// environment variable.
55    ///
56    /// If this option is not used or `interval` is equal to zero, 60 seconds is
57    /// used as the default.
58    pub fn with_interval(mut self, interval: Duration) -> Self {
59        if !interval.is_zero() {
60            self.interval = interval;
61        }
62        self
63    }
64
65    /// Create a [PeriodicReader] with the given config.
66    pub fn build(self) -> PeriodicReader<E> {
67        PeriodicReader::new(self.exporter, self.interval)
68    }
69}
70
71/// A `MetricReader` that periodically collects and exports metrics at a configurable interval.
72///
73/// By default, [`PeriodicReader`] collects and exports metrics every **60 seconds**.
74/// The time taken for export is **not** included in the interval. Use [`PeriodicReaderBuilder`]
75/// to customize the interval.
76///
77/// [`PeriodicReader`] spawns a background thread to handle metric collection and export.
78/// This thread remains active until [`shutdown()`] is called.
79///
80/// ## Collection Process
81/// "Collection" refers to gathering aggregated metrics from the SDK's internal storage.
82/// During this phase, callbacks from observable instruments are also triggered.
83///
84/// [`PeriodicReader`] does **not** enforce a timeout for collection. If an
85/// observable callback takes too long, it may delay the next collection cycle.
86/// If a callback never returns, it **will stall** all metric collection (and exports)
87/// indefinitely. Furthermore, the SDK does not isolate callbacks from panics; if a
88/// callback panics, the background worker thread will terminate, stopping all subsequent
89/// periodic metric collection and export.
90///
91/// ## Exporter Compatibility
92/// When used with the [`OTLP Exporter`](https://docs.rs/opentelemetry-otlp), the following
93/// transport options are supported:
94///
95/// - **`grpc-tonic`**: Requires [`MeterProvider`] to be initialized within a `tokio` runtime.
96/// - **`reqwest-blocking-client`**: Works with both a standard (`main`) function and `tokio::main`.
97///
98/// Async HTTP clients such as `reqwest-client` and `hyper-client` are not
99/// supported by this default reader. The OTLP HTTP exporter chooses its default
100/// HTTP client from enabled crate features and cannot tell which reader will
101/// drive it. If your dependency graph enables async HTTP client features, either
102/// pass an explicit blocking client for this reader or use the experimental
103/// async-runtime periodic reader.
104///
105/// [`PeriodicReader`] does **not** enforce a timeout for exports either. Instead,
106/// the configured exporter is responsible for enforcing timeouts. If an export operation
107/// never returns, [`PeriodicReader`] will **stop exporting new metrics**, stalling
108/// metric collection.
109///
110/// ## Manual Export & Shutdown
111/// Users can manually trigger an export via [`force_flush()`]. Calling [`shutdown()`]
112/// exports any remaining metrics and should be done before application exit to ensure
113/// all data is sent.
114///
115/// **Warning**: If using **tokio’s current-thread runtime**, calling [`shutdown()`]
116/// from the main thread may cause a deadlock. To prevent this, call [`shutdown()`]
117/// from a separate thread or use tokio's `spawn_blocking`.
118///
119/// [`PeriodicReader`]: crate::metrics::PeriodicReader
120/// [`PeriodicReaderBuilder`]: crate::metrics::PeriodicReaderBuilder
121/// [`MeterProvider`]: crate::metrics::SdkMeterProvider
122/// [`shutdown()`]: crate::metrics::SdkMeterProvider::shutdown
123/// [`force_flush()`]: crate::metrics::SdkMeterProvider::force_flush
124///
125/// # Example
126///
127/// ```no_run
128/// use opentelemetry_sdk::metrics::PeriodicReader;
129/// # fn example<E>(get_exporter: impl Fn() -> E)
130/// # where
131/// #     E: opentelemetry_sdk::metrics::exporter::PushMetricExporter,
132/// # {
133///
134/// let exporter = get_exporter(); // set up a push exporter
135///
136/// let reader = PeriodicReader::builder(exporter).build();
137/// # drop(reader);
138/// # }
139/// ```
140pub struct PeriodicReader<E: PushMetricExporter> {
141    inner: Arc<PeriodicReaderInner<E>>,
142}
143
144impl<E: PushMetricExporter> Clone for PeriodicReader<E> {
145    fn clone(&self) -> Self {
146        Self {
147            inner: Arc::clone(&self.inner),
148        }
149    }
150}
151
152impl<E: PushMetricExporter> PeriodicReader<E> {
153    /// Configuration options for a periodic reader with own thread
154    pub fn builder(exporter: E) -> PeriodicReaderBuilder<E> {
155        PeriodicReaderBuilder::new(exporter)
156    }
157
158    fn new(exporter: E, interval: Duration) -> Self {
159        let (message_sender, message_receiver): (Sender<Message>, Receiver<Message>) =
160            mpsc::channel();
161        let exporter_arc = Arc::new(exporter);
162        let reader = PeriodicReader {
163            inner: Arc::new(PeriodicReaderInner {
164                message_sender,
165                producer: Mutex::new(None),
166                exporter: exporter_arc.clone(),
167            }),
168        };
169        let cloned_reader = reader.clone();
170
171        let mut rm = ResourceMetrics {
172            resource: Resource::empty(),
173            scope_metrics: Vec::new(),
174        };
175
176        #[cfg(all(test, feature = "internal-logs"))]
177        let dispatcher = tracing::dispatcher::get_default(Clone::clone);
178
179        let result_thread_creation = thread::Builder::new()
180            .name("OpenTelemetry.Metrics.PeriodicReader".to_string())
181            .spawn(move || {
182                let _suppress_guard = Context::enter_telemetry_suppressed_scope();
183                // Logs worker-thread panics; drops before _suppress_guard so suppression is still active.
184                let _termination_guard = WorkerTerminationGuard::new(
185                    #[cfg(all(test, feature = "internal-logs"))]
186                    &dispatcher,
187                );
188                let mut interval_start = Instant::now();
189                let mut remaining_interval = interval;
190                otel_debug!(
191                    name: "PeriodReaderThreadStarted",
192                    interval_in_millisecs = interval.as_millis(),
193                );
194                loop {
195                    otel_debug!(
196                        name: "PeriodReaderThreadLoopAlive", message = "Next export will happen after interval, unless flush or shutdown is triggered.", interval_in_millisecs = remaining_interval.as_millis()
197                    );
198                    match message_receiver.recv_timeout(remaining_interval) {
199                        Ok(Message::Flush(response_sender)) => {
200                            otel_debug!(
201                                name: "PeriodReaderThreadExportingDueToFlush"
202                            );
203                            let export_result = cloned_reader.collect_and_export(&mut rm);
204                            otel_debug!(
205                                name: "PeriodReaderInvokedExport",
206                                export_result = format!("{:?}", export_result)
207                            );
208
209                            // If response_sender is disconnected, we can't send
210                            // the result back. This occurs when the thread that
211                            // initiated flush gave up due to timeout.
212                            // Gracefully handle that with internal logs. A
213                            // failed export has already been logged at Error
214                            // level by collect_and_export, so these are Debug
215                            // level and only note that the result could not be
216                            // delivered to the caller.
217
218                            if export_result.is_err() {
219                                if response_sender.send(false).is_err() {
220                                    otel_debug!(
221                                        name: "PeriodReader.Flush.ResponseSendError",
222                                        message = "PeriodicReader's flush has failed, but unable to send this info back to caller.
223                                        This occurs when the caller has timed out waiting for the response. If you see this occuring frequently, consider increasing the flush timeout."
224                                    );
225                                }
226                            } else if response_sender.send(true).is_err() {
227                                otel_debug!(
228                                    name: "PeriodReader.Flush.ResponseSendError",
229                                    message = "PeriodicReader's flush has completed successfully, but unable to send this info back to caller.
230                                    This occurs when the caller has timed out waiting for the response. If you see this occuring frequently, consider increasing the flush timeout."
231                                );
232                            }
233
234                            // Adjust the remaining interval after the flush
235                            let elapsed = interval_start.elapsed();
236                            if elapsed < interval {
237                                remaining_interval = interval - elapsed;
238                                otel_debug!(
239                                    name: "PeriodReaderThreadAdjustingRemainingIntervalAfterFlush",
240                                    remaining_interval = remaining_interval.as_secs()
241                                );
242                            } else {
243                                otel_debug!(
244                                    name: "PeriodReaderThreadAdjustingExportAfterFlush",
245                                );
246                                // Reset the interval if the flush finishes after the expected export time
247                                // effectively missing the normal export.
248                                // Should we attempt to do the missed export immediately?
249                                // Or do the next export at the next interval?
250                                // Currently this attempts the next export immediately.
251                                // i.e calling Flush can affect the regularity.
252                                interval_start = Instant::now();
253                                remaining_interval = Duration::ZERO;
254                            }
255                        }
256                        Ok(Message::Shutdown(response_sender)) => {
257                            // Perform final export and break out of loop and exit the thread
258                            otel_debug!(name: "PeriodReaderThreadExportingDueToShutdown");
259                            let export_result = cloned_reader.collect_and_export(&mut rm);
260                            otel_debug!(
261                                name: "PeriodReaderInvokedExport",
262                                export_result = format!("{:?}", export_result)
263                            );
264                            let shutdown_result = exporter_arc.shutdown();
265                            otel_debug!(
266                                name: "PeriodReaderInvokedExporterShutdown",
267                                shutdown_result = format!("{:?}", shutdown_result)
268                            );
269
270                            // If response_sender is disconnected, we can't send
271                            // the result back. This occurs when the thread that
272                            // initiated shutdown gave up due to timeout.
273                            // Gracefully handle that with internal logs and
274                            // continue with shutdown (i.e exit thread) The
275                            // internal errors are of Info level, as this is
276                            // useful for user to know whether the shutdown was
277                            // successful or not, when shutdown() itself merely
278                            // tells that it timed out.
279                            if export_result.is_err() || shutdown_result.is_err() {
280                                if response_sender.send(false).is_err() {
281                                    otel_info!(
282                                        name: "PeriodReaderThreadShutdown.ResponseSendError",
283                                        message = "PeriodicReader's shutdown has failed, but unable to send this info back to caller.
284                                        This occurs when the caller has timed out waiting for the response. If you see this occuring frequently, consider increasing the shutdown timeout."
285                                    );
286                                }
287                            } else if response_sender.send(true).is_err() {
288                                otel_debug!(
289                                    name: "PeriodReaderThreadShutdown.ResponseSendError",
290                                    message = "PeriodicReader completed its shutdown, but unable to send this info back to caller.
291                                    This occurs when the caller has timed out waiting for the response. If you see this occuring frequently, consider increasing the shutdown timeout."
292                                );
293                            }
294
295                            otel_debug!(
296                                name: "PeriodReaderThreadExiting",
297                                reason = "ShutdownRequested"
298                            );
299                            break;
300                        }
301                        Err(mpsc::RecvTimeoutError::Timeout) => {
302                            let export_start = Instant::now();
303                            otel_debug!(
304                                name: "PeriodReaderThreadExportingDueToTimer"
305                            );
306
307                            let export_result = cloned_reader.collect_and_export(&mut rm);
308                            otel_debug!(
309                                name: "PeriodReaderInvokedExport",
310                                export_result = format!("{:?}", export_result)
311                            );
312
313                            let time_taken_for_export = export_start.elapsed();
314                            if time_taken_for_export > interval {
315                                otel_debug!(
316                                    name: "PeriodReaderThreadExportTookLongerThanInterval"
317                                );
318                                // if export took longer than interval, do the
319                                // next export immediately.
320                                // Alternatively, we could skip the next export
321                                // and wait for the next interval.
322                                // Or enforce that export timeout is less than interval.
323                                // What is the desired behavior?
324                                interval_start = Instant::now();
325                                remaining_interval = Duration::ZERO;
326                            } else {
327                                remaining_interval = interval - time_taken_for_export;
328                                interval_start = Instant::now();
329                            }
330                        }
331                        Err(mpsc::RecvTimeoutError::Disconnected) => {
332                            // Channel disconnected, only thing to do is break
333                            // out (i.e exit the thread)
334                            otel_debug!(
335                                name: "PeriodReaderThreadExiting",
336                                reason = "MessageSenderDisconnected"
337                            );
338                            break;
339                        }
340                    }
341                }
342                otel_debug!(
343                    name: "PeriodReaderThreadStopped"
344                );
345            });
346
347        // TODO: Should we fail-fast here and bubble up the error to user?
348        #[allow(unused_variables)]
349        if let Err(e) = result_thread_creation {
350            otel_error!(
351                name: "PeriodReaderThreadStartError",
352                message = "Failed to start PeriodicReader thread. Metrics will not be exported.",
353                error = format!("{:?}", e)
354            );
355        }
356        reader
357    }
358
359    fn collect_and_export(&self, rm: &mut ResourceMetrics) -> OTelSdkResult {
360        self.inner.collect_and_export(rm)
361    }
362}
363
364impl<E: PushMetricExporter> fmt::Debug for PeriodicReader<E> {
365    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
366        f.debug_struct("PeriodicReader").finish()
367    }
368}
369
370struct PeriodicReaderInner<E: PushMetricExporter> {
371    exporter: Arc<E>,
372    message_sender: mpsc::Sender<Message>,
373    producer: Mutex<Option<Weak<dyn SdkProducer>>>,
374}
375
376impl<E: PushMetricExporter> PeriodicReaderInner<E> {
377    fn register_pipeline(&self, producer: Weak<dyn SdkProducer>) {
378        let mut inner = self.producer.lock().expect("lock poisoned");
379        *inner = Some(producer);
380    }
381
382    fn temporality(&self, _kind: InstrumentKind) -> Temporality {
383        self.exporter.temporality()
384    }
385
386    fn collect(&self, rm: &mut ResourceMetrics) -> OTelSdkResult {
387        let producer = self.producer.lock().expect("lock poisoned");
388        if let Some(p) = producer.as_ref() {
389            p.upgrade()
390                .ok_or(OTelSdkError::AlreadyShutdown)?
391                .produce(rm)?;
392            Ok(())
393        } else {
394            otel_warn!(
395            name: "PeriodReader.MeterProviderNotRegistered",
396            message = "PeriodicReader is not registered with MeterProvider. Metrics will not be collected. \
397                   This occurs when a periodic reader is created but not associated with a MeterProvider \
398                   by calling `.with_reader(reader)` on MeterProviderBuilder."
399            );
400            Err(OTelSdkError::InternalFailure(
401                "MeterProvider is not registered".into(),
402            ))
403        }
404    }
405
406    fn collect_and_export(&self, rm: &mut ResourceMetrics) -> OTelSdkResult {
407        let current_time = Instant::now();
408        let collect_result = self.collect(rm);
409        let time_taken_for_collect = current_time.elapsed();
410
411        #[allow(clippy::question_mark)]
412        if let Err(e) = collect_result {
413            otel_warn!(
414                name: "PeriodReaderCollectError",
415                error = format!("{:?}", e)
416            );
417            return Err(OTelSdkError::InternalFailure(e.to_string()));
418        }
419
420        if rm.scope_metrics.is_empty() {
421            otel_debug!(name: "NoMetricsCollected");
422            return Ok(());
423        }
424
425        let metrics_count = rm.scope_metrics.iter().fold(0, |count, scope_metrics| {
426            count + scope_metrics.metrics.len()
427        });
428        otel_debug!(name: "PeriodicReaderMetricsCollected", count = metrics_count, time_taken_in_millis = time_taken_for_collect.as_millis());
429
430        // Relying on futures executor to execute async call.
431        // TODO: Pass timeout to exporter
432        match futures_executor::block_on(self.exporter.export(rm)) {
433            Ok(()) => Ok(()),
434            Err(err) => {
435                otel_error!(
436                    name: "PeriodicReader.ExportError",
437                    error = format!("{}", err)
438                );
439                Err(err)
440            }
441        }
442    }
443
444    fn force_flush(&self) -> OTelSdkResult {
445        // TODO: Better message for this scenario.
446        // Flush and Shutdown called from 2 threads Flush check shutdown
447        // flag before shutdown thread sets it. Both threads attempt to send
448        // message to the same channel. Case1: Flush thread sends message first,
449        // shutdown thread sends message next. Flush would succeed, as
450        // background thread won't process shutdown message until flush
451        // triggered export is done. Case2: Shutdown thread sends message first,
452        // flush thread sends message next. Shutdown would succeed, as
453        // background thread would process shutdown message first. The
454        // background exits so it won't receive the flush message. ForceFlush
455        // returns Failure, but we could indicate specifically that shutdown has
456        // completed. TODO is to see if this message can be improved.
457
458        let (response_tx, response_rx) = mpsc::channel();
459        self.message_sender
460            .send(Message::Flush(response_tx))
461            .map_err(|e| OTelSdkError::InternalFailure(e.to_string()))?;
462
463        if let Ok(response) = response_rx.recv() {
464            // TODO: call exporter's force_flush method.
465            if response {
466                Ok(())
467            } else {
468                Err(OTelSdkError::InternalFailure("Failed to flush".into()))
469            }
470        } else {
471            Err(OTelSdkError::InternalFailure("Failed to flush".into()))
472        }
473    }
474
475    fn shutdown(&self) -> OTelSdkResult {
476        // TODO: See if this is better to be created upfront.
477        let (response_tx, response_rx) = mpsc::channel();
478        self.message_sender
479            .send(Message::Shutdown(response_tx))
480            .map_err(|e| OTelSdkError::InternalFailure(e.to_string()))?;
481
482        // TODO: Make this timeout configurable.
483        match response_rx.recv_timeout(Duration::from_secs(5)) {
484            Ok(response) => {
485                if response {
486                    Ok(())
487                } else {
488                    Err(OTelSdkError::InternalFailure("Failed to shutdown".into()))
489                }
490            }
491            Err(mpsc::RecvTimeoutError::Timeout) => {
492                Err(OTelSdkError::Timeout(Duration::from_secs(5)))
493            }
494            Err(mpsc::RecvTimeoutError::Disconnected) => {
495                Err(OTelSdkError::InternalFailure("Failed to shutdown".into()))
496            }
497        }
498    }
499}
500
501#[derive(Debug)]
502enum Message {
503    Flush(Sender<bool>),
504    Shutdown(Sender<bool>),
505}
506
507impl<E: PushMetricExporter> MetricReader for PeriodicReader<E> {
508    fn register_pipeline(&self, pipeline: Weak<Pipeline>) {
509        self.inner.register_pipeline(pipeline);
510    }
511
512    fn collect(&self, rm: &mut ResourceMetrics) -> OTelSdkResult {
513        self.inner.collect(rm)
514    }
515
516    fn force_flush(&self) -> OTelSdkResult {
517        self.inner.force_flush()
518    }
519
520    // TODO: Offer an async version of shutdown so users can await the shutdown
521    // completion, and avoid blocking the thread. The default shutdown on drop
522    // can still use blocking call. If user already explicitly called shutdown,
523    // drop won't call shutdown again.
524    fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
525        self.inner.shutdown()
526    }
527
528    /// To construct a [MetricReader][metric-reader] when setting up an SDK,
529    /// The output temporality (optional), a function of instrument kind.
530    /// This function SHOULD be obtained from the exporter.
531    ///
532    /// If not configured, the Cumulative temporality SHOULD be used.
533    ///
534    /// [metric-reader]: https://github.com/open-telemetry/opentelemetry-specification/blob/0a78571045ca1dca48621c9648ec3c832c3c541c/specification/metrics/sdk.md#metricreader
535    fn temporality(&self, kind: InstrumentKind) -> Temporality {
536        kind.temporality_preference(self.inner.temporality(kind))
537    }
538}
539
540/// Reports a worker panic during unwind. In tests, keeps the scoped subscriber
541/// active until reporting finishes, then restores the previous subscriber.
542struct WorkerTerminationGuard {
543    #[cfg(all(test, feature = "internal-logs"))]
544    _subscriber_guard: tracing::dispatcher::DefaultGuard,
545}
546
547impl WorkerTerminationGuard {
548    fn new(#[cfg(all(test, feature = "internal-logs"))] dispatcher: &tracing::Dispatch) -> Self {
549        Self {
550            #[cfg(all(test, feature = "internal-logs"))]
551            _subscriber_guard: tracing::dispatcher::set_default(dispatcher),
552        }
553    }
554}
555
556impl Drop for WorkerTerminationGuard {
557    fn drop(&mut self) {
558        if thread::panicking() {
559            otel_error!(
560                name: "PeriodicReader.WorkerThreadTerminated",
561                message = "PeriodicReader worker thread is no longer running; future periodic metric collection and export will not occur. An observable metric callback panic is one possible cause."
562            );
563        }
564    }
565}
566
567#[cfg(all(test, feature = "testing"))]
568mod tests {
569    use super::PeriodicReader;
570    use crate::{
571        error::{OTelSdkError, OTelSdkResult},
572        metrics::{
573            data::ResourceMetrics, exporter::PushMetricExporter, reader::MetricReader,
574            InMemoryMetricExporter, SdkMeterProvider, Temporality,
575        },
576        Resource,
577    };
578    use opentelemetry::metrics::MeterProvider;
579    use std::{
580        sync::{
581            atomic::{AtomicBool, AtomicUsize, Ordering},
582            mpsc, Arc,
583        },
584        time::Duration,
585    };
586
587    // use below command to run all tests
588    // cargo test metrics::periodic_reader::tests --features=testing,spec_unstable_metrics_views -- --nocapture
589
590    #[derive(Debug, Clone)]
591    struct MetricExporterThatFailsOnlyOnFirst {
592        count: Arc<AtomicUsize>,
593    }
594
595    impl Default for MetricExporterThatFailsOnlyOnFirst {
596        fn default() -> Self {
597            MetricExporterThatFailsOnlyOnFirst {
598                count: Arc::new(AtomicUsize::new(0)),
599            }
600        }
601    }
602
603    impl MetricExporterThatFailsOnlyOnFirst {
604        fn get_count(&self) -> usize {
605            self.count.load(Ordering::Relaxed)
606        }
607    }
608
609    impl PushMetricExporter for MetricExporterThatFailsOnlyOnFirst {
610        async fn export(&self, _metrics: &ResourceMetrics) -> OTelSdkResult {
611            if self.count.fetch_add(1, Ordering::Relaxed) == 0 {
612                Err(OTelSdkError::InternalFailure("export failed".into()))
613            } else {
614                Ok(())
615            }
616        }
617
618        fn force_flush(&self) -> OTelSdkResult {
619            Ok(())
620        }
621
622        fn shutdown(&self) -> OTelSdkResult {
623            Ok(())
624        }
625
626        fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
627            Ok(())
628        }
629
630        fn temporality(&self) -> Temporality {
631            Temporality::Cumulative
632        }
633    }
634
635    #[derive(Debug, Clone, Default)]
636    struct MockMetricExporter {
637        is_shutdown: Arc<AtomicBool>,
638    }
639
640    impl PushMetricExporter for MockMetricExporter {
641        async fn export(&self, _metrics: &ResourceMetrics) -> OTelSdkResult {
642            Ok(())
643        }
644
645        fn force_flush(&self) -> OTelSdkResult {
646            Ok(())
647        }
648
649        fn shutdown(&self) -> OTelSdkResult {
650            self.shutdown_with_timeout(Duration::from_secs(5))
651        }
652
653        fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult {
654            self.is_shutdown.store(true, Ordering::Relaxed);
655            Ok(())
656        }
657
658        fn temporality(&self) -> Temporality {
659            Temporality::Cumulative
660        }
661    }
662
663    #[test]
664    fn collection_triggered_by_interval_multiple() {
665        // Arrange
666        let interval = std::time::Duration::from_millis(1);
667        let exporter = InMemoryMetricExporter::default();
668        let reader = PeriodicReader::builder(exporter.clone())
669            .with_interval(interval)
670            .build();
671        let i = Arc::new(AtomicUsize::new(0));
672        let i_clone = i.clone();
673
674        // Act
675        let meter_provider = SdkMeterProvider::builder().with_reader(reader).build();
676        let meter = meter_provider.meter("test");
677        let _counter = meter
678            .u64_observable_counter("testcounter")
679            .with_callback(move |_| {
680                i_clone.fetch_add(1, Ordering::Relaxed);
681            })
682            .build();
683
684        // Sleep for a duration 5X (plus liberal buffer to account for potential
685        // CI slowness) the interval to ensure multiple collection.
686        // Not a fan of such tests, but this seems to be the only way to test
687        // if periodic reader is doing its job.
688        // TODO: Decide if this should be ignored in CI
689        std::thread::sleep(interval * 5 * 20);
690
691        // Assert
692        assert!(i.load(Ordering::Relaxed) >= 5);
693    }
694
695    #[test]
696    fn shutdown_repeat() {
697        // Arrange
698        let exporter = InMemoryMetricExporter::default();
699        let reader = PeriodicReader::builder(exporter.clone()).build();
700
701        let meter_provider = SdkMeterProvider::builder().with_reader(reader).build();
702        let result = meter_provider.shutdown();
703        assert!(result.is_ok());
704
705        // calling shutdown again should return Err
706        let result = meter_provider.shutdown();
707        assert!(result.is_err());
708        assert!(matches!(result, Err(OTelSdkError::AlreadyShutdown)));
709
710        // calling shutdown again should return Err
711        let result = meter_provider.shutdown();
712        assert!(result.is_err());
713        assert!(matches!(result, Err(OTelSdkError::AlreadyShutdown)));
714    }
715
716    #[test]
717    fn flush_after_shutdown() {
718        // Arrange
719        let exporter = InMemoryMetricExporter::default();
720        let reader = PeriodicReader::builder(exporter.clone()).build();
721
722        let meter_provider = SdkMeterProvider::builder().with_reader(reader).build();
723        let result = meter_provider.force_flush();
724        assert!(result.is_ok());
725
726        let result = meter_provider.shutdown();
727        assert!(result.is_ok());
728
729        // calling force_flush after shutdown should return Err
730        let result = meter_provider.force_flush();
731        assert!(result.is_err());
732    }
733
734    #[test]
735    fn flush_repeat() {
736        // Arrange
737        let exporter = InMemoryMetricExporter::default();
738        let reader = PeriodicReader::builder(exporter.clone()).build();
739
740        let meter_provider = SdkMeterProvider::builder().with_reader(reader).build();
741        let result = meter_provider.force_flush();
742        assert!(result.is_ok());
743
744        // calling force_flush again should return Ok
745        let result = meter_provider.force_flush();
746        assert!(result.is_ok());
747    }
748
749    #[test]
750    fn periodic_reader_without_pipeline() {
751        // Arrange
752        let exporter = InMemoryMetricExporter::default();
753        let reader = PeriodicReader::builder(exporter.clone()).build();
754
755        let rm = &mut ResourceMetrics {
756            resource: Resource::empty(),
757            scope_metrics: Vec::new(),
758        };
759        // Pipeline is not registered, so collect should return an error
760        let result = reader.collect(rm);
761        assert!(result.is_err());
762
763        // Pipeline is not registered, so flush should return an error
764        let result = reader.force_flush();
765        assert!(result.is_err());
766
767        // Adding reader to meter provider should register the pipeline
768        // TODO: This part might benefit from a different design.
769        let meter_provider = SdkMeterProvider::builder()
770            .with_reader(reader.clone())
771            .build();
772
773        // Now collect and flush should succeed
774        let result = reader.collect(rm);
775        assert!(result.is_ok());
776
777        let result = meter_provider.force_flush();
778        assert!(result.is_ok());
779    }
780
781    #[test]
782    fn exporter_failures_are_handled() {
783        // create a mock exporter that fails 1st time and succeeds 2nd time
784        // Validate using this exporter that periodic reader can handle exporter failure
785        // and continue to export metrics.
786        // Arrange
787        let interval = std::time::Duration::from_millis(10);
788        let exporter = MetricExporterThatFailsOnlyOnFirst::default();
789        let reader = PeriodicReader::builder(exporter.clone())
790            .with_interval(interval)
791            .build();
792
793        let meter_provider = SdkMeterProvider::builder().with_reader(reader).build();
794        let meter = meter_provider.meter("test");
795        let counter = meter.u64_counter("sync_counter").build();
796        counter.add(1, &[]);
797        let _obs_counter = meter
798            .u64_observable_counter("testcounter")
799            .with_callback(move |observer| {
800                observer.observe(1, &[]);
801            })
802            .build();
803
804        // Sleep for a duration much longer than the interval to trigger
805        // multiple exports, including failures.
806        // Not a fan of such tests, but this seems to be the
807        // only way to test if periodic reader is doing its job. TODO: Decide if
808        // this should be ignored in CI
809        std::thread::sleep(Duration::from_millis(500));
810
811        // Assert that atleast 2 exports are attempted given the 1st one fails.
812        assert!(exporter.get_count() >= 2);
813    }
814
815    #[test]
816    fn shutdown_passed_to_exporter() {
817        // Arrange
818        let exporter = MockMetricExporter::default();
819        let reader = PeriodicReader::builder(exporter.clone()).build();
820
821        let meter_provider = SdkMeterProvider::builder().with_reader(reader).build();
822        let meter = meter_provider.meter("test");
823        let counter = meter.u64_counter("sync_counter").build();
824        counter.add(1, &[]);
825
826        // shutdown the provider, which should call shutdown on periodic reader
827        // which in turn should call shutdown on exporter.
828        let result = meter_provider.shutdown();
829        assert!(result.is_ok());
830        assert!(exporter.is_shutdown.load(Ordering::Relaxed));
831    }
832
833    #[test]
834    fn collection() {
835        collection_triggered_by_interval_helper();
836        collection_triggered_by_flush_helper();
837        collection_triggered_by_shutdown_helper();
838        collection_triggered_by_drop_helper();
839    }
840
841    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
842    async fn collection_from_tokio_multi_with_one_worker() {
843        collection_triggered_by_interval_helper();
844        collection_triggered_by_flush_helper();
845        collection_triggered_by_shutdown_helper();
846        collection_triggered_by_drop_helper();
847    }
848
849    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
850    async fn collection_from_tokio_with_two_worker() {
851        collection_triggered_by_interval_helper();
852        collection_triggered_by_flush_helper();
853        collection_triggered_by_shutdown_helper();
854        collection_triggered_by_drop_helper();
855    }
856
857    #[tokio::test(flavor = "current_thread")]
858    async fn collection_from_tokio_current() {
859        collection_triggered_by_interval_helper();
860        collection_triggered_by_flush_helper();
861        collection_triggered_by_shutdown_helper();
862        collection_triggered_by_drop_helper();
863    }
864
865    fn collection_triggered_by_interval_helper() {
866        collection_helper(|_| {
867            // Sleep for a duration longer than the interval to ensure at least one collection
868            // Not a fan of such tests, but this seems to be the only way to test
869            // if periodic reader is doing its job.
870            // TODO: Decide if this should be ignored in CI
871            std::thread::sleep(Duration::from_millis(500));
872        });
873    }
874
875    fn collection_triggered_by_flush_helper() {
876        collection_helper(|meter_provider| {
877            meter_provider.force_flush().expect("flush should succeed");
878        });
879    }
880
881    fn collection_triggered_by_shutdown_helper() {
882        collection_helper(|meter_provider| {
883            meter_provider.shutdown().expect("shutdown should succeed");
884        });
885    }
886
887    fn collection_triggered_by_drop_helper() {
888        collection_helper(|meter_provider| {
889            drop(meter_provider);
890        });
891    }
892
893    fn collection_helper(trigger: fn(SdkMeterProvider)) {
894        // Arrange
895        let exporter = InMemoryMetricExporter::default();
896        let reader = PeriodicReader::builder(exporter.clone()).build();
897        let (sender, receiver) = mpsc::channel();
898
899        let meter_provider = SdkMeterProvider::builder().with_reader(reader).build();
900        let meter = meter_provider.meter("test");
901        let _counter = meter
902            .u64_observable_counter("testcounter")
903            .with_callback(move |observer| {
904                observer.observe(1, &[]);
905                sender.send(()).expect("channel should still be open");
906            })
907            .build();
908
909        // Act
910        trigger(meter_provider);
911
912        // Assert
913        receiver
914            .recv_timeout(Duration::ZERO)
915            .expect("message should be available in channel, indicating a collection occurred, which should trigger observable callback");
916
917        let exported_metrics = exporter
918            .get_finished_metrics()
919            .expect("this should not fail");
920        assert!(
921            !exported_metrics.is_empty(),
922            "Metrics should be available in exporter."
923        );
924    }
925
926    async fn some_async_function() -> u64 {
927        // No dependency on any particular async runtime.
928        std::thread::sleep(std::time::Duration::from_millis(1));
929        1
930    }
931
932    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
933    async fn async_inside_observable_callback_from_tokio_multi_with_one_worker() {
934        async_inside_observable_callback_helper();
935    }
936
937    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
938    async fn async_inside_observable_callback_from_tokio_multi_with_two_worker() {
939        async_inside_observable_callback_helper();
940    }
941
942    #[tokio::test(flavor = "current_thread")]
943    async fn async_inside_observable_callback_from_tokio_current_thread() {
944        async_inside_observable_callback_helper();
945    }
946
947    #[test]
948    fn async_inside_observable_callback_from_regular_main() {
949        async_inside_observable_callback_helper();
950    }
951
952    fn async_inside_observable_callback_helper() {
953        let interval = std::time::Duration::from_millis(10);
954        let exporter = InMemoryMetricExporter::default();
955        let reader = PeriodicReader::builder(exporter.clone())
956            .with_interval(interval)
957            .build();
958
959        let meter_provider = SdkMeterProvider::builder().with_reader(reader).build();
960        let meter = meter_provider.meter("test");
961        let _gauge = meter
962            .u64_observable_gauge("my_observable_gauge")
963            .with_callback(|observer| {
964                // using futures_executor::block_on intentionally and avoiding
965                // any particular async runtime.
966                let value = futures_executor::block_on(some_async_function());
967                observer.observe(value, &[]);
968            })
969            .build();
970
971        meter_provider.force_flush().expect("flush should succeed");
972        let exported_metrics = exporter
973            .get_finished_metrics()
974            .expect("this should not fail");
975        assert!(
976            !exported_metrics.is_empty(),
977            "Metrics should be available in exporter."
978        );
979    }
980
981    async fn some_tokio_async_function() -> u64 {
982        // Tokio specific async function
983        tokio::time::sleep(Duration::from_millis(1)).await;
984        1
985    }
986
987    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
988
989    async fn tokio_async_inside_observable_callback_from_tokio_multi_with_one_worker() {
990        tokio_async_inside_observable_callback_helper(true);
991    }
992
993    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
994    async fn tokio_async_inside_observable_callback_from_tokio_multi_with_two_worker() {
995        tokio_async_inside_observable_callback_helper(true);
996    }
997
998    #[tokio::test(flavor = "current_thread")]
999    #[ignore] //TODO: Investigate if this can be fixed.
1000    async fn tokio_async_inside_observable_callback_from_tokio_current_thread() {
1001        tokio_async_inside_observable_callback_helper(true);
1002    }
1003
1004    #[test]
1005    fn tokio_async_inside_observable_callback_from_regular_main() {
1006        tokio_async_inside_observable_callback_helper(false);
1007    }
1008
1009    fn tokio_async_inside_observable_callback_helper(use_current_tokio_runtime: bool) {
1010        let exporter = InMemoryMetricExporter::default();
1011        let reader = PeriodicReader::builder(exporter.clone()).build();
1012
1013        let meter_provider = SdkMeterProvider::builder().with_reader(reader).build();
1014        let meter = meter_provider.meter("test");
1015
1016        if use_current_tokio_runtime {
1017            let rt = tokio::runtime::Handle::current().clone();
1018            let _gauge = meter
1019                .u64_observable_gauge("my_observable_gauge")
1020                .with_callback(move |observer| {
1021                    // call tokio specific async function from here
1022                    let value = rt.block_on(some_tokio_async_function());
1023                    observer.observe(value, &[]);
1024                })
1025                .build();
1026            // rt here is a reference to the current tokio runtime.
1027            // Dropping it occurs when the tokio::main itself ends.
1028        } else {
1029            let rt = tokio::runtime::Runtime::new().unwrap();
1030            let _gauge = meter
1031                .u64_observable_gauge("my_observable_gauge")
1032                .with_callback(move |observer| {
1033                    // call tokio specific async function from here
1034                    let value = rt.block_on(some_tokio_async_function());
1035                    observer.observe(value, &[]);
1036                })
1037                .build();
1038            // rt is not dropped here as it is moved to the closure,
1039            // and is dropped only when MeterProvider itself is dropped.
1040            // This works when called from normal main.
1041        };
1042
1043        meter_provider.force_flush().expect("flush should succeed");
1044        let exported_metrics = exporter
1045            .get_finished_metrics()
1046            .expect("this should not fail");
1047        assert!(
1048            !exported_metrics.is_empty(),
1049            "Metrics should be available in exporter."
1050        );
1051    }
1052
1053    #[cfg(feature = "internal-logs")]
1054    struct TerminationEvents {
1055        count: Arc<AtomicUsize>,
1056        sender: mpsc::Sender<bool>,
1057    }
1058
1059    #[cfg(feature = "internal-logs")]
1060    impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for TerminationEvents {
1061        fn on_event(
1062            &self,
1063            event: &tracing::Event<'_>,
1064            _ctx: tracing_subscriber::layer::Context<'_, S>,
1065        ) {
1066            if event.metadata().name() == "PeriodicReader.WorkerThreadTerminated"
1067                && *event.metadata().level() == tracing::Level::ERROR
1068            {
1069                self.count.fetch_add(1, Ordering::Relaxed);
1070                let _ = self
1071                    .sender
1072                    .send(opentelemetry::Context::is_current_telemetry_suppressed());
1073            }
1074        }
1075    }
1076
1077    #[cfg(feature = "internal-logs")]
1078    fn termination_subscriber() -> (
1079        impl tracing::Subscriber + Send + Sync,
1080        Arc<AtomicUsize>,
1081        mpsc::Receiver<bool>,
1082    ) {
1083        use tracing_subscriber::prelude::*;
1084        let count = Arc::new(AtomicUsize::new(0));
1085        let (sender, receiver) = mpsc::channel();
1086        let subscriber = tracing_subscriber::registry().with(TerminationEvents {
1087            count: count.clone(),
1088            sender,
1089        });
1090        (subscriber, count, receiver)
1091    }
1092
1093    #[cfg(feature = "internal-logs")]
1094    #[test]
1095    fn callback_panic_during_flush_reports_termination_once() {
1096        let (subscriber, count, events) = termination_subscriber();
1097        tracing::subscriber::with_default(subscriber, || {
1098            let reader = PeriodicReader::builder(InMemoryMetricExporter::default()).build();
1099            let provider = SdkMeterProvider::builder()
1100                .with_reader(reader.clone())
1101                .build();
1102            let _counter = provider
1103                .meter("test")
1104                .u64_observable_counter("panicking_counter")
1105                .with_callback(|_| panic!("intentional callback panic"))
1106                .build();
1107
1108            assert!(matches!(
1109                reader.force_flush(),
1110                Err(OTelSdkError::InternalFailure(message)) if message == "Failed to flush"
1111            ));
1112            assert!(events.recv_timeout(Duration::from_secs(5)).unwrap());
1113            assert!(provider.force_flush().is_err());
1114            assert!(reader.shutdown().is_err());
1115            assert_eq!(count.load(Ordering::Relaxed), 1);
1116        });
1117    }
1118
1119    #[cfg(feature = "internal-logs")]
1120    #[test]
1121    fn callback_panic_during_periodic_collection_reports_termination_once() {
1122        let (subscriber, count, events) = termination_subscriber();
1123        tracing::subscriber::with_default(subscriber, || {
1124            let reader = PeriodicReader::builder(InMemoryMetricExporter::default())
1125                .with_interval(Duration::from_millis(10))
1126                .build();
1127            let provider = SdkMeterProvider::builder()
1128                .with_reader(reader.clone())
1129                .build();
1130            let _counter = provider
1131                .meter("test")
1132                .u64_observable_counter("panicking_counter")
1133                .with_callback(|_| panic!("intentional callback panic"))
1134                .build();
1135
1136            // Wait for the worker diagnostic without triggering flush or shutdown.
1137            assert!(events.recv_timeout(Duration::from_secs(5)).unwrap());
1138            assert!(provider.force_flush().is_err());
1139            assert!(provider.shutdown().is_err());
1140            assert_eq!(count.load(Ordering::Relaxed), 1);
1141        });
1142    }
1143
1144    #[cfg(feature = "internal-logs")]
1145    #[test]
1146    fn callback_panic_during_shutdown_reports_termination_once() {
1147        let (subscriber, count, events) = termination_subscriber();
1148        tracing::subscriber::with_default(subscriber, || {
1149            let provider = SdkMeterProvider::builder()
1150                .with_reader(PeriodicReader::builder(InMemoryMetricExporter::default()).build())
1151                .build();
1152            let _counter = provider
1153                .meter("test")
1154                .u64_observable_counter("panicking_counter")
1155                .with_callback(|_| panic!("intentional callback panic"))
1156                .build();
1157
1158            assert!(provider.shutdown().is_err());
1159            assert!(events.recv_timeout(Duration::from_secs(5)).unwrap());
1160            assert_eq!(count.load(Ordering::Relaxed), 1);
1161        });
1162    }
1163
1164    #[cfg(feature = "internal-logs")]
1165    #[test]
1166    fn flush_queued_after_shutdown_does_not_report_termination() {
1167        let (subscriber, count, _events) = termination_subscriber();
1168        tracing::subscriber::with_default(subscriber, || {
1169            let reader = PeriodicReader::builder(InMemoryMetricExporter::default()).build();
1170            let provider = SdkMeterProvider::builder()
1171                .with_reader(reader.clone())
1172                .build();
1173            let (entered_tx, entered_rx) = mpsc::channel();
1174            let (release_tx, release_rx) = mpsc::channel();
1175            let release_rx = std::sync::Mutex::new(release_rx);
1176            let _counter = provider
1177                .meter("test")
1178                .u64_observable_counter("blocking_counter")
1179                .with_callback(move |_| {
1180                    entered_tx.send(()).unwrap();
1181                    release_rx
1182                        .lock()
1183                        .unwrap()
1184                        .recv_timeout(Duration::from_secs(5))
1185                        .unwrap();
1186                })
1187                .build();
1188
1189            let (shutdown_tx, shutdown_rx) = mpsc::channel();
1190            reader
1191                .inner
1192                .message_sender
1193                .send(super::Message::Shutdown(shutdown_tx))
1194                .unwrap();
1195            entered_rx.recv_timeout(Duration::from_secs(5)).unwrap();
1196            let (flush_tx, flush_rx) = mpsc::channel();
1197            reader
1198                .inner
1199                .message_sender
1200                .send(super::Message::Flush(flush_tx))
1201                .unwrap();
1202            release_tx.send(()).unwrap();
1203            assert_eq!(shutdown_rx.recv_timeout(Duration::from_secs(5)), Ok(true));
1204            assert_eq!(
1205                flush_rx.recv_timeout(Duration::from_secs(5)),
1206                Err(mpsc::RecvTimeoutError::Disconnected)
1207            );
1208            assert!(reader.force_flush().is_err());
1209            assert!(reader.shutdown().is_err());
1210            assert_eq!(count.load(Ordering::Relaxed), 0);
1211        });
1212    }
1213}
opentelemetry_sdk/metrics/periodic_reader.rs

opentelemetry_sdk/metrics/
periodic_reader.rs
