Skip to main content

opentelemetry_sdk/metrics/
mod.rs

1//! The crust of the OpenTelemetry metrics SDK.
2//!
3//! ## Configuration
4//!
5//! The metrics SDK configuration is stored with each [SdkMeterProvider].
6//! Configuration for [Resource]s, views, and `ManualReader` or
7//! [PeriodicReader] instances can be specified.
8//!
9//! ### Example
10//!
11//! ```
12//! use opentelemetry::global;
13//! use opentelemetry::KeyValue;
14//! use opentelemetry_sdk::{metrics::SdkMeterProvider, Resource};
15//!
16//! // Generate SDK configuration, resource, views, etc
17//! let resource = Resource::builder().build(); // default attributes about the current process
18//!
19//! // Create a meter provider with the desired config
20//! let meter_provider = SdkMeterProvider::builder().with_resource(resource).build();
21//! global::set_meter_provider(meter_provider.clone());
22//!
23//! // Use the meter provider to create meter instances
24//! let meter = global::meter("my_app");
25//!
26//! // Create instruments scoped to the meter
27//! let counter = meter
28//!     .u64_counter("power_consumption")
29//!     .with_unit("kWh")
30//!     .build();
31//!
32//! // use instruments to record measurements
33//! counter.add(10, &[KeyValue::new("rate", "standard")]);
34//!
35//! // shutdown the provider at the end of the application to ensure any metrics not yet
36//! // exported are flushed.
37//! meter_provider.shutdown().unwrap();
38//! ```
39//!
40//! [Resource]: crate::Resource
41
42#[allow(unreachable_pub)]
43#[allow(unused)]
44pub(crate) mod aggregation;
45pub mod data;
46mod error;
47pub mod exporter;
48pub(crate) mod instrument;
49pub(crate) mod internal;
50#[cfg(feature = "experimental_metrics_custom_reader")]
51pub(crate) mod manual_reader;
52pub(crate) mod meter;
53mod meter_provider;
54pub(crate) mod noop;
55pub(crate) mod periodic_reader;
56#[cfg(feature = "experimental_metrics_periodicreader_with_async_runtime")]
57/// Module for periodic reader with async runtime.
58pub mod periodic_reader_with_async_runtime;
59pub(crate) mod pipeline;
60#[cfg(feature = "experimental_metrics_custom_reader")]
61pub mod reader;
62#[cfg(not(feature = "experimental_metrics_custom_reader"))]
63pub(crate) mod reader;
64pub(crate) mod view;
65
66/// In-Memory metric exporter for testing purpose.
67#[cfg(any(feature = "testing", test))]
68#[cfg_attr(docsrs, doc(cfg(any(feature = "testing", test))))]
69pub mod in_memory_exporter;
70#[cfg(any(feature = "testing", test))]
71#[cfg_attr(docsrs, doc(cfg(any(feature = "testing", test))))]
72pub use in_memory_exporter::{InMemoryMetricExporter, InMemoryMetricExporterBuilder};
73
74pub use aggregation::*;
75#[cfg(feature = "experimental_metrics_custom_reader")]
76pub use manual_reader::*;
77pub use meter_provider::*;
78pub use periodic_reader::*;
79#[cfg(feature = "experimental_metrics_custom_reader")]
80pub use pipeline::Pipeline;
81
82pub use instrument::{Instrument, InstrumentKind, Stream, StreamBuilder};
83
84use std::hash::Hash;
85use std::str::FromStr;
86
87/// Defines the window that an aggregation was calculated over.
88#[derive(Debug, Copy, Clone, Default, PartialEq, Eq, Hash)]
89#[non_exhaustive]
90pub enum Temporality {
91    /// A measurement interval that continues to expand forward in time from a
92    /// starting point.
93    ///
94    /// New measurements are added to all previous measurements since a start time.
95    #[default]
96    Cumulative,
97
98    /// A measurement interval that resets each cycle.
99    ///
100    /// Measurements from one cycle are recorded independently, measurements from
101    /// other cycles do not affect them.
102    Delta,
103
104    /// Configures Synchronous Counter and Histogram instruments to use
105    /// Delta aggregation temporality, which allows them to shed memory
106    /// following a cardinality explosion, thus use less memory.
107    LowMemory,
108}
109
110impl FromStr for Temporality {
111    type Err = ();
112
113    fn from_str(s: &str) -> Result<Self, Self::Err> {
114        match s.to_lowercase().as_str() {
115            "cumulative" => Ok(Temporality::Cumulative),
116            "delta" => Ok(Temporality::Delta),
117            "lowmemory" => Ok(Temporality::LowMemory),
118            _ => Err(()),
119        }
120    }
121}
122
123#[cfg(all(test, feature = "testing"))]
124mod tests {
125    #[cfg(feature = "experimental_metrics_bound_instruments")]
126    use self::data::ExponentialHistogramDataPoint;
127    use self::data::{HistogramDataPoint, MetricData, ScopeMetrics, SumDataPoint};
128    use super::internal::Number;
129    use super::*;
130    use crate::metrics::data::ResourceMetrics;
131    use crate::metrics::internal::AggregatedMetricsAccess;
132    use crate::metrics::InMemoryMetricExporter;
133    use crate::metrics::InMemoryMetricExporterBuilder;
134    use data::GaugeDataPoint;
135    use opentelemetry::metrics::{Counter, Meter, UpDownCounter};
136    use opentelemetry::InstrumentationScope;
137    use opentelemetry::Value;
138    use opentelemetry::{metrics::MeterProvider as _, KeyValue};
139    use rand::{rngs, Rng, SeedableRng};
140    use std::cmp::{max, min};
141    use std::sync::atomic::{AtomicBool, Ordering};
142    use std::sync::{Arc, Mutex};
143    use std::thread;
144    use std::time::Duration;
145
146    // Run all tests in this mod
147    // cargo test metrics::tests --features=testing,spec_unstable_metrics_views
148    // Note for all tests from this point onwards in this mod:
149    // "multi_thread" tokio flavor must be used else flush won't
150    // be able to make progress!
151
152    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
153    #[cfg(not(feature = "experimental_metrics_disable_name_validation"))]
154    async fn invalid_instrument_config_noops() {
155        // Run this test with stdout enabled to see output.
156        // cargo test invalid_instrument_config_noops --features=testing,spec_unstable_metrics_views -- --nocapture
157        let invalid_instrument_names = vec![
158            "_startWithNoneAlphabet",
159            "utf8char锈",
160            "a".repeat(256).leak(),
161            "invalid name",
162        ];
163        for name in invalid_instrument_names {
164            let test_context = TestContext::new(Temporality::Cumulative);
165            let counter = test_context.meter().u64_counter(name).build();
166            counter.add(1, &[]);
167
168            let up_down_counter = test_context.meter().i64_up_down_counter(name).build();
169            up_down_counter.add(1, &[]);
170
171            let gauge = test_context.meter().f64_gauge(name).build();
172            gauge.record(1.9, &[]);
173
174            let histogram = test_context.meter().f64_histogram(name).build();
175            histogram.record(1.0, &[]);
176
177            let _observable_counter = test_context
178                .meter()
179                .u64_observable_counter(name)
180                .with_callback(move |observer| {
181                    observer.observe(1, &[]);
182                })
183                .build();
184
185            let _observable_gauge = test_context
186                .meter()
187                .f64_observable_gauge(name)
188                .with_callback(move |observer| {
189                    observer.observe(1.0, &[]);
190                })
191                .build();
192
193            let _observable_up_down_counter = test_context
194                .meter()
195                .i64_observable_up_down_counter(name)
196                .with_callback(move |observer| {
197                    observer.observe(1, &[]);
198                })
199                .build();
200
201            test_context.flush_metrics();
202
203            // As instrument name is invalid, no metrics should be exported
204            test_context.check_no_metrics();
205        }
206
207        let invalid_bucket_boundaries = vec![
208            vec![1.0, 1.0],                          // duplicate boundaries
209            vec![1.0, 2.0, 3.0, 2.0],                // duplicate non consequent boundaries
210            vec![1.0, 2.0, 3.0, 4.0, 2.5],           // unsorted boundaries
211            vec![1.0, 2.0, 3.0, f64::INFINITY, 4.0], // boundaries with positive infinity
212            vec![1.0, 2.0, 3.0, f64::NAN],           // boundaries with NaNs
213            vec![f64::NEG_INFINITY, 2.0, 3.0],       // boundaries with negative infinity
214        ];
215        for bucket_boundaries in invalid_bucket_boundaries {
216            let test_context = TestContext::new(Temporality::Cumulative);
217            let histogram = test_context
218                .meter()
219                .f64_histogram("test")
220                .with_boundaries(bucket_boundaries)
221                .build();
222            histogram.record(1.9, &[]);
223            test_context.flush_metrics();
224
225            // As bucket boundaries provided via advisory params are invalid,
226            // no metrics should be exported
227            test_context.check_no_metrics();
228        }
229    }
230
231    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
232    #[cfg(feature = "experimental_metrics_disable_name_validation")]
233    async fn valid_instrument_config_with_feature_experimental_metrics_disable_name_validation() {
234        // Run this test with stdout enabled to see output.
235        // cargo test valid_instrument_config_with_feature_experimental_metrics_disable_name_validation --all-features -- --nocapture
236        let invalid_instrument_names = vec![
237            "_startWithNoneAlphabet",
238            "utf8char锈",
239            "",
240            "a".repeat(256).leak(),
241            "\\allow\\slash /sec",
242            "\\allow\\$$slash /sec",
243            "Total $ Count",
244            "\\test\\UsagePercent(Total) > 80%",
245            "invalid name",
246        ];
247        for name in invalid_instrument_names {
248            let test_context = TestContext::new(Temporality::Cumulative);
249            let counter = test_context.meter().u64_counter(name).build();
250            counter.add(1, &[]);
251
252            let up_down_counter = test_context.meter().i64_up_down_counter(name).build();
253            up_down_counter.add(1, &[]);
254
255            let gauge = test_context.meter().f64_gauge(name).build();
256            gauge.record(1.9, &[]);
257
258            let histogram = test_context.meter().f64_histogram(name).build();
259            histogram.record(1.0, &[]);
260
261            let _observable_counter = test_context
262                .meter()
263                .u64_observable_counter(name)
264                .with_callback(move |observer| {
265                    observer.observe(1, &[]);
266                })
267                .build();
268
269            let _observable_gauge = test_context
270                .meter()
271                .f64_observable_gauge(name)
272                .with_callback(move |observer| {
273                    observer.observe(1.0, &[]);
274                })
275                .build();
276
277            let _observable_up_down_counter = test_context
278                .meter()
279                .i64_observable_up_down_counter(name)
280                .with_callback(move |observer| {
281                    observer.observe(1, &[]);
282                })
283                .build();
284
285            test_context.flush_metrics();
286
287            // As instrument name are valid because of the feature flag, metrics should be exported
288            let resource_metrics = test_context
289                .exporter
290                .get_finished_metrics()
291                .expect("metrics expected to be exported");
292
293            assert!(!resource_metrics.is_empty(), "metrics should be exported");
294        }
295
296        // Ensuring that the Histograms with invalid bucket boundaries are not exported
297        // when using the feature flag
298        let invalid_bucket_boundaries = vec![
299            vec![1.0, 1.0],                          // duplicate boundaries
300            vec![1.0, 2.0, 3.0, 2.0],                // duplicate non consequent boundaries
301            vec![1.0, 2.0, 3.0, 4.0, 2.5],           // unsorted boundaries
302            vec![1.0, 2.0, 3.0, f64::INFINITY, 4.0], // boundaries with positive infinity
303            vec![1.0, 2.0, 3.0, f64::NAN],           // boundaries with NaNs
304            vec![f64::NEG_INFINITY, 2.0, 3.0],       // boundaries with negative infinity
305        ];
306        for bucket_boundaries in invalid_bucket_boundaries {
307            let test_context = TestContext::new(Temporality::Cumulative);
308            let histogram = test_context
309                .meter()
310                .f64_histogram("test")
311                .with_boundaries(bucket_boundaries)
312                .build();
313            histogram.record(1.9, &[]);
314            test_context.flush_metrics();
315
316            // As bucket boundaries provided via advisory params are invalid,
317            // no metrics should be exported
318            test_context.check_no_metrics();
319        }
320    }
321
322    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
323    async fn counter_aggregation_delta() {
324        // Run this test with stdout enabled to see output.
325        // cargo test counter_aggregation_delta --features=testing -- --nocapture
326        counter_aggregation_helper(Temporality::Delta);
327    }
328
329    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
330    async fn counter_aggregation_cumulative() {
331        // Run this test with stdout enabled to see output.
332        // cargo test counter_aggregation_cumulative --features=testing -- --nocapture
333        counter_aggregation_helper(Temporality::Cumulative);
334    }
335
336    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
337    async fn counter_aggregation_no_attributes_cumulative() {
338        let mut test_context = TestContext::new(Temporality::Cumulative);
339        let counter = test_context.u64_counter("test", "my_counter", None);
340
341        counter.add(50, &[]);
342        test_context.flush_metrics();
343
344        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
345            unreachable!()
346        };
347
348        assert_eq!(sum.data_points.len(), 1, "Expected only one data point");
349        assert!(sum.is_monotonic, "Should produce monotonic.");
350        assert_eq!(
351            sum.temporality,
352            Temporality::Cumulative,
353            "Should produce cumulative"
354        );
355
356        let data_point = &sum.data_points[0];
357        assert!(data_point.attributes.is_empty(), "Non-empty attribute set");
358        assert_eq!(data_point.value, 50, "Unexpected data point value");
359    }
360
361    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
362    async fn counter_aggregation_no_attributes_delta() {
363        let mut test_context = TestContext::new(Temporality::Delta);
364        let counter = test_context.u64_counter("test", "my_counter", None);
365
366        counter.add(50, &[]);
367        test_context.flush_metrics();
368
369        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
370            unreachable!()
371        };
372
373        assert_eq!(sum.data_points.len(), 1, "Expected only one data point");
374        assert!(sum.is_monotonic, "Should produce monotonic.");
375        assert_eq!(sum.temporality, Temporality::Delta, "Should produce delta");
376
377        let data_point = &sum.data_points[0];
378        assert!(data_point.attributes.is_empty(), "Non-empty attribute set");
379        assert_eq!(data_point.value, 50, "Unexpected data point value");
380    }
381
382    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
383    async fn counter_aggregation_overflow_delta() {
384        counter_aggregation_overflow_helper(Temporality::Delta);
385        counter_aggregation_overflow_helper_custom_limit(Temporality::Delta);
386    }
387
388    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
389    async fn counter_aggregation_overflow_cumulative() {
390        counter_aggregation_overflow_helper(Temporality::Cumulative);
391        counter_aggregation_overflow_helper_custom_limit(Temporality::Cumulative);
392    }
393
394    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
395    async fn counter_aggregation_attribute_order_sorted_first_delta() {
396        // Run this test with stdout enabled to see output.
397        // cargo test counter_aggregation_attribute_order_sorted_first_delta --features=testing -- --nocapture
398        counter_aggregation_attribute_order_helper(Temporality::Delta, true);
399    }
400
401    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
402    async fn counter_aggregation_attribute_order_sorted_first_cumulative() {
403        // Run this test with stdout enabled to see output.
404        // cargo test counter_aggregation_attribute_order_sorted_first_cumulative --features=testing -- --nocapture
405        counter_aggregation_attribute_order_helper(Temporality::Cumulative, true);
406    }
407
408    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
409    async fn counter_aggregation_attribute_order_unsorted_first_delta() {
410        // Run this test with stdout enabled to see output.
411        // cargo test counter_aggregation_attribute_order_unsorted_first_delta --features=testing -- --nocapture
412
413        counter_aggregation_attribute_order_helper(Temporality::Delta, false);
414    }
415
416    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
417    async fn counter_aggregation_attribute_order_unsorted_first_cumulative() {
418        // Run this test with stdout enabled to see output.
419        // cargo test counter_aggregation_attribute_order_unsorted_first_cumulative --features=testing -- --nocapture
420
421        counter_aggregation_attribute_order_helper(Temporality::Cumulative, false);
422    }
423
424    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
425    async fn histogram_aggregation_cumulative() {
426        // Run this test with stdout enabled to see output.
427        // cargo test histogram_aggregation_cumulative --features=testing -- --nocapture
428        histogram_aggregation_helper(Temporality::Cumulative);
429    }
430
431    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
432    async fn histogram_aggregation_delta() {
433        // Run this test with stdout enabled to see output.
434        // cargo test histogram_aggregation_delta --features=testing -- --nocapture
435        histogram_aggregation_helper(Temporality::Delta);
436    }
437
438    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
439    async fn histogram_aggregation_with_custom_bounds() {
440        // Run this test with stdout enabled to see output.
441        // cargo test histogram_aggregation_with_custom_bounds --features=testing -- --nocapture
442        histogram_aggregation_with_custom_bounds_helper(Temporality::Delta);
443        histogram_aggregation_with_custom_bounds_helper(Temporality::Cumulative);
444    }
445
446    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
447    async fn histogram_aggregation_with_empty_bounds() {
448        // Run this test with stdout enabled to see output.
449        // cargo test histogram_aggregation_with_empty_bounds --features=testing -- --nocapture
450        histogram_aggregation_with_empty_bounds_helper(Temporality::Delta);
451        histogram_aggregation_with_empty_bounds_helper(Temporality::Cumulative);
452    }
453
454    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
455    async fn histogram_aggregation_with_custom_bounds_and_view() {
456        // Run this test with stdout enabled to see output.
457        // cargo test histogram_aggregation_with_custom_bounds_and_view --features=testing -- --nocapture
458        histogram_aggregation_with_custom_bounds_and_view_helper(Temporality::Delta);
459        histogram_aggregation_with_custom_bounds_and_view_helper(Temporality::Cumulative);
460    }
461
462    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
463    async fn exponential_histogram_aggregation_with_view() {
464        // Run this test with stdout enabled to see output.
465        // cargo test exponential_histogram_aggregation_with_view --features=testing -- --nocapture
466        exponential_histogram_aggregation_with_view_helper(Temporality::Delta);
467        exponential_histogram_aggregation_with_view_helper(Temporality::Cumulative);
468    }
469
470    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
471    async fn updown_counter_aggregation_cumulative() {
472        // Run this test with stdout enabled to see output.
473        // cargo test updown_counter_aggregation_cumulative --features=testing -- --nocapture
474        updown_counter_aggregation_helper(Temporality::Cumulative);
475    }
476
477    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
478    async fn updown_counter_aggregation_delta() {
479        // Run this test with stdout enabled to see output.
480        // cargo test updown_counter_aggregation_delta --features=testing -- --nocapture
481        updown_counter_aggregation_helper(Temporality::Delta);
482    }
483
484    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
485    async fn gauge_aggregation() {
486        // Run this test with stdout enabled to see output.
487        // cargo test gauge_aggregation --features=testing -- --nocapture
488
489        // Gauge should use last value aggregation regardless of the aggregation temporality used.
490        gauge_aggregation_helper(Temporality::Delta);
491        gauge_aggregation_helper(Temporality::Cumulative);
492    }
493
494    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
495    async fn observable_gauge_aggregation() {
496        // Run this test with stdout enabled to see output.
497        // cargo test observable_gauge_aggregation --features=testing -- --nocapture
498
499        // Gauge should use last value aggregation regardless of the aggregation temporality used.
500        observable_gauge_aggregation_helper(Temporality::Delta, false);
501        observable_gauge_aggregation_helper(Temporality::Delta, true);
502        observable_gauge_aggregation_helper(Temporality::Cumulative, false);
503        observable_gauge_aggregation_helper(Temporality::Cumulative, true);
504    }
505
506    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
507    async fn observable_counter_aggregation_cumulative_non_zero_increment() {
508        // Run this test with stdout enabled to see output.
509        // cargo test observable_counter_aggregation_cumulative_non_zero_increment --features=testing -- --nocapture
510        observable_counter_aggregation_helper(Temporality::Cumulative, 100, 10, 4, false);
511    }
512
513    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
514    async fn observable_counter_aggregation_cumulative_non_zero_increment_no_attrs() {
515        // Run this test with stdout enabled to see output.
516        // cargo test observable_counter_aggregation_cumulative_non_zero_increment_no_attrs --features=testing -- --nocapture
517        observable_counter_aggregation_helper(Temporality::Cumulative, 100, 10, 4, true);
518    }
519
520    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
521    async fn observable_counter_aggregation_delta_non_zero_increment() {
522        // Run this test with stdout enabled to see output.
523        // cargo test observable_counter_aggregation_delta_non_zero_increment --features=testing -- --nocapture
524        observable_counter_aggregation_helper(Temporality::Delta, 100, 10, 4, false);
525    }
526
527    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
528    async fn observable_counter_aggregation_delta_non_zero_increment_no_attrs() {
529        // Run this test with stdout enabled to see output.
530        // cargo test observable_counter_aggregation_delta_non_zero_increment_no_attrs --features=testing -- --nocapture
531        observable_counter_aggregation_helper(Temporality::Delta, 100, 10, 4, true);
532    }
533
534    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
535    async fn observable_counter_aggregation_cumulative_zero_increment() {
536        // Run this test with stdout enabled to see output.
537        // cargo test observable_counter_aggregation_cumulative_zero_increment --features=testing -- --nocapture
538        observable_counter_aggregation_helper(Temporality::Cumulative, 100, 0, 4, false);
539    }
540
541    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
542    async fn observable_counter_aggregation_cumulative_zero_increment_no_attrs() {
543        // Run this test with stdout enabled to see output.
544        // cargo test observable_counter_aggregation_cumulative_zero_increment_no_attrs --features=testing -- --nocapture
545        observable_counter_aggregation_helper(Temporality::Cumulative, 100, 0, 4, true);
546    }
547
548    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
549    async fn observable_counter_aggregation_delta_zero_increment() {
550        // Run this test with stdout enabled to see output.
551        // cargo test observable_counter_aggregation_delta_zero_increment --features=testing -- --nocapture
552        observable_counter_aggregation_helper(Temporality::Delta, 100, 0, 4, false);
553    }
554
555    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
556    async fn observable_counter_aggregation_delta_zero_increment_no_attrs() {
557        // Run this test with stdout enabled to see output.
558        // cargo test observable_counter_aggregation_delta_zero_increment_no_attrs --features=testing -- --nocapture
559        observable_counter_aggregation_helper(Temporality::Delta, 100, 0, 4, true);
560    }
561
562    fn observable_counter_aggregation_helper(
563        temporality: Temporality,
564        start: u64,
565        increment: u64,
566        length: u64,
567        is_empty_attributes: bool,
568    ) {
569        // Arrange
570        let mut test_context = TestContext::new(temporality);
571        let attributes = if is_empty_attributes {
572            vec![]
573        } else {
574            vec![KeyValue::new("key1", "value1")]
575        };
576        // The Observable counter reports values[0], values[1],....values[n] on each flush.
577        let values: Vec<u64> = (0..length).map(|i| start + i * increment).collect();
578        println!("Testing with observable values: {values:?}");
579        let values = Arc::new(values);
580        let values_clone = values.clone();
581        let i = Arc::new(Mutex::new(0));
582        let _observable_counter = test_context
583            .meter()
584            .u64_observable_counter("my_observable_counter")
585            .with_unit("my_unit")
586            .with_callback(move |observer| {
587                let mut index = i.lock().unwrap();
588                if *index < values.len() {
589                    observer.observe(values[*index], &attributes);
590                    *index += 1;
591                }
592            })
593            .build();
594
595        for (iter, v) in values_clone.iter().enumerate() {
596            test_context.flush_metrics();
597            let MetricData::Sum(sum) =
598                test_context.get_aggregation::<u64>("my_observable_counter", None)
599            else {
600                unreachable!()
601            };
602            assert_eq!(sum.data_points.len(), 1);
603            assert!(sum.is_monotonic, "Counter should produce monotonic.");
604            if let Temporality::Cumulative = temporality {
605                assert_eq!(
606                    sum.temporality,
607                    Temporality::Cumulative,
608                    "Should produce cumulative"
609                );
610            } else {
611                assert_eq!(sum.temporality, Temporality::Delta, "Should produce delta");
612            }
613
614            // find and validate datapoint
615            let data_point = if is_empty_attributes {
616                &sum.data_points[0]
617            } else {
618                find_sum_datapoint_with_key_value(&sum.data_points, "key1", "value1")
619                    .expect("datapoint with key1=value1 expected")
620            };
621
622            if let Temporality::Cumulative = temporality {
623                // Cumulative counter should have the value as is.
624                assert_eq!(data_point.value, *v);
625            } else {
626                // Delta counter should have the increment value.
627                // Except for the first value which should be the start value.
628                if iter == 0 {
629                    assert_eq!(data_point.value, start);
630                } else {
631                    assert_eq!(data_point.value, increment);
632                }
633            }
634
635            test_context.reset_metrics();
636        }
637    }
638
639    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
640    async fn observable_counter_delta_attribute_set_reappears_after_gap() {
641        // Run this test with stdout enabled to see output.
642        // cargo test observable_counter_delta_attribute_set_reappears_after_gap --features=testing -- --nocapture
643
644        // This test verifies the behavior when an attribute set is not reported
645        // for one collection cycle and then reappears.
646        // See: https://github.com/open-telemetry/opentelemetry-specification/issues/4861
647        //
648        // Scenario (Observable Counter with Delta temporality):
649        // | Collection | Callback Reports  | Expected Delta Export  |
650        // |------------|-------------------|------------------------|
651        // | 1          | A=100, B=50       | A=100, B=50            |
652        // | 2          | A=150 (B missing) | A=50 (B not exported)  |
653        // | 3          | A=200, B=80       | A=50, B=80             |
654        //
655        // Current implementation: When B reappears, its delta is calculated from zero
656        // (fresh start), not from the last known value. This is Option 1 from the spec issue.
657
658        let mut test_context = TestContext::new(Temporality::Delta);
659
660        // Shared state for callback: (collection_cycle, value_a, value_b_option)
661        // value_b_option is None when B should not be reported
662        let callback_state = Arc::new(Mutex::new((0u32, 0u64, Option::<u64>::None)));
663        let callback_state_clone = callback_state.clone();
664
665        let _observable_counter = test_context
666            .meter()
667            .u64_observable_counter("my_observable_counter")
668            .with_callback(move |observer| {
669                let state = callback_state_clone.lock().unwrap();
670                let (_cycle, value_a, value_b_option) = *state;
671
672                observer.observe(value_a, &[KeyValue::new("key", "A")]);
673                if let Some(value_b) = value_b_option {
674                    observer.observe(value_b, &[KeyValue::new("key", "B")]);
675                }
676            })
677            .build();
678
679        // Collection 1: A=100, B=50
680        {
681            *callback_state.lock().unwrap() = (1, 100, Some(50));
682            test_context.flush_metrics();
683
684            let MetricData::Sum(sum) =
685                test_context.get_aggregation::<u64>("my_observable_counter", None)
686            else {
687                unreachable!()
688            };
689
690            assert_eq!(sum.data_points.len(), 2);
691            assert_eq!(sum.temporality, Temporality::Delta);
692
693            let dp_a = find_sum_datapoint_with_key_value(&sum.data_points, "key", "A")
694                .expect("datapoint for A expected");
695            let dp_b = find_sum_datapoint_with_key_value(&sum.data_points, "key", "B")
696                .expect("datapoint for B expected");
697
698            // First collection: delta = value - 0
699            assert_eq!(
700                dp_a.value, 100,
701                "A's delta should be 100 (first collection)"
702            );
703            assert_eq!(dp_b.value, 50, "B's delta should be 50 (first collection)");
704
705            test_context.reset_metrics();
706        }
707
708        // Collection 2: A=150, B missing
709        {
710            *callback_state.lock().unwrap() = (2, 150, None);
711            test_context.flush_metrics();
712
713            let MetricData::Sum(sum) =
714                test_context.get_aggregation::<u64>("my_observable_counter", None)
715            else {
716                unreachable!()
717            };
718
719            // Only A should be exported, B is not observed so not exported (per spec)
720            assert_eq!(
721                sum.data_points.len(),
722                1,
723                "Only A should be exported when B is not observed"
724            );
725
726            let dp_a = find_sum_datapoint_with_key_value(&sum.data_points, "key", "A")
727                .expect("datapoint for A expected");
728            assert_eq!(dp_a.value, 50, "A's delta should be 50 (150 - 100)");
729
730            test_context.reset_metrics();
731        }
732
733        // Collection 3: A=200, B=80 (B reappears)
734        {
735            *callback_state.lock().unwrap() = (3, 200, Some(80));
736            test_context.flush_metrics();
737
738            let MetricData::Sum(sum) =
739                test_context.get_aggregation::<u64>("my_observable_counter", None)
740            else {
741                unreachable!()
742            };
743
744            assert_eq!(sum.data_points.len(), 2);
745
746            let dp_a = find_sum_datapoint_with_key_value(&sum.data_points, "key", "A")
747                .expect("datapoint for A expected");
748            let dp_b = find_sum_datapoint_with_key_value(&sum.data_points, "key", "B")
749                .expect("datapoint for B expected");
750
751            assert_eq!(dp_a.value, 50, "A's delta should be 50 (200 - 150)");
752
753            // B reappears after a gap. Current implementation uses "delta from zero" (Option 1).
754            // This means B's delta = 80 - 0 = 80, not 80 - 50 = 30.
755            // See: https://github.com/open-telemetry/opentelemetry-specification/issues/4861
756            // TODO: Watch for spec clarification on this behavior.
757            assert_eq!(
758                dp_b.value, 80,
759                "B's delta should be 80 (fresh start after gap, not 30 from last known value)"
760            );
761
762            test_context.reset_metrics();
763        }
764    }
765
766    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
767    async fn empty_meter_name_retained() {
768        async fn meter_name_retained_helper(
769            meter: Meter,
770            provider: SdkMeterProvider,
771            exporter: InMemoryMetricExporter,
772        ) {
773            // Act
774            let counter = meter.u64_counter("my_counter").build();
775
776            counter.add(10, &[]);
777            provider.force_flush().unwrap();
778
779            // Assert
780            let resource_metrics = exporter
781                .get_finished_metrics()
782                .expect("metrics are expected to be exported.");
783            assert!(
784                resource_metrics[0].scope_metrics[0].metrics.len() == 1,
785                "There should be a single metric"
786            );
787            let meter_name = resource_metrics[0].scope_metrics[0].scope.name();
788            assert_eq!(meter_name, "");
789        }
790
791        let exporter = InMemoryMetricExporter::default();
792        let meter_provider = SdkMeterProvider::builder()
793            .with_periodic_exporter(exporter.clone())
794            .build();
795
796        // Test Meter creation in 2 ways, both with empty string as meter name
797        let meter1 = meter_provider.meter("");
798        meter_name_retained_helper(meter1, meter_provider.clone(), exporter.clone()).await;
799
800        let meter_scope = InstrumentationScope::builder("").build();
801        let meter2 = meter_provider.meter_with_scope(meter_scope);
802        meter_name_retained_helper(meter2, meter_provider, exporter).await;
803    }
804
805    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
806    async fn counter_duplicate_instrument_merge() {
807        // Arrange
808        let exporter = InMemoryMetricExporter::default();
809        let meter_provider = SdkMeterProvider::builder()
810            .with_periodic_exporter(exporter.clone())
811            .build();
812
813        // Act
814        let meter = meter_provider.meter("test");
815        let counter = meter
816            .u64_counter("my_counter")
817            .with_unit("my_unit")
818            .with_description("my_description")
819            .build();
820
821        let counter_duplicated = meter
822            .u64_counter("my_counter")
823            .with_unit("my_unit")
824            .with_description("my_description")
825            .build();
826
827        let attribute = vec![KeyValue::new("key1", "value1")];
828        counter.add(10, &attribute);
829        counter_duplicated.add(5, &attribute);
830
831        meter_provider.force_flush().unwrap();
832
833        // Assert
834        let resource_metrics = exporter
835            .get_finished_metrics()
836            .expect("metrics are expected to be exported.");
837        assert!(
838            resource_metrics[0].scope_metrics[0].metrics.len() == 1,
839            "There should be single metric merging duplicate instruments"
840        );
841        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
842        assert_eq!(metric.name, "my_counter");
843        assert_eq!(metric.unit, "my_unit");
844        let MetricData::Sum(sum) = u64::extract_metrics_data_ref(&metric.data)
845            .expect("Sum aggregation expected for Counter instruments by default")
846        else {
847            unreachable!()
848        };
849
850        // Expecting 1 time-series.
851        assert_eq!(sum.data_points.len(), 1);
852
853        let datapoint = &sum.data_points[0];
854        assert_eq!(datapoint.value, 15);
855    }
856
857    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
858    async fn counter_duplicate_instrument_different_meter_no_merge() {
859        // Arrange
860        let exporter = InMemoryMetricExporter::default();
861        let meter_provider = SdkMeterProvider::builder()
862            .with_periodic_exporter(exporter.clone())
863            .build();
864
865        // Act
866        let meter1 = meter_provider.meter("test.meter1");
867        let meter2 = meter_provider.meter("test.meter2");
868        let counter1 = meter1
869            .u64_counter("my_counter")
870            .with_unit("my_unit")
871            .with_description("my_description")
872            .build();
873
874        let counter2 = meter2
875            .u64_counter("my_counter")
876            .with_unit("my_unit")
877            .with_description("my_description")
878            .build();
879
880        let attribute = vec![KeyValue::new("key1", "value1")];
881        counter1.add(10, &attribute);
882        counter2.add(5, &attribute);
883
884        meter_provider.force_flush().unwrap();
885
886        // Assert
887        let resource_metrics = exporter
888            .get_finished_metrics()
889            .expect("metrics are expected to be exported.");
890        assert!(
891            resource_metrics[0].scope_metrics.len() == 2,
892            "There should be 2 separate scope"
893        );
894        assert!(
895            resource_metrics[0].scope_metrics[0].metrics.len() == 1,
896            "There should be single metric for the scope"
897        );
898        assert!(
899            resource_metrics[0].scope_metrics[1].metrics.len() == 1,
900            "There should be single metric for the scope"
901        );
902
903        let scope1 = find_scope_metric(&resource_metrics[0].scope_metrics, "test.meter1");
904        let scope2 = find_scope_metric(&resource_metrics[0].scope_metrics, "test.meter2");
905
906        if let Some(scope1) = scope1 {
907            let metric1 = &scope1.metrics[0];
908            assert_eq!(metric1.name, "my_counter");
909            assert_eq!(metric1.unit, "my_unit");
910            assert_eq!(metric1.description, "my_description");
911            let MetricData::Sum(sum1) = u64::extract_metrics_data_ref(&metric1.data)
912                .expect("Sum aggregation expected for Counter instruments by default")
913            else {
914                unreachable!()
915            };
916
917            // Expecting 1 time-series.
918            assert_eq!(sum1.data_points.len(), 1);
919
920            let datapoint1 = &sum1.data_points[0];
921            assert_eq!(datapoint1.value, 10);
922        } else {
923            panic!("No MetricScope found for 'test.meter1'");
924        }
925
926        if let Some(scope2) = scope2 {
927            let metric2 = &scope2.metrics[0];
928            assert_eq!(metric2.name, "my_counter");
929            assert_eq!(metric2.unit, "my_unit");
930            assert_eq!(metric2.description, "my_description");
931
932            let MetricData::Sum(sum2) = u64::extract_metrics_data_ref(&metric2.data)
933                .expect("Sum aggregation expected for Counter instruments by default")
934            else {
935                unreachable!()
936            };
937
938            // Expecting 1 time-series.
939            assert_eq!(sum2.data_points.len(), 1);
940
941            let datapoint2 = &sum2.data_points[0];
942            assert_eq!(datapoint2.value, 5);
943        } else {
944            panic!("No MetricScope found for 'test.meter2'");
945        }
946    }
947
948    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
949    async fn instrumentation_scope_identity_test() {
950        // Arrange
951        let exporter = InMemoryMetricExporter::default();
952        let meter_provider = SdkMeterProvider::builder()
953            .with_periodic_exporter(exporter.clone())
954            .build();
955
956        // Act
957        // Meters are identical.
958        // Hence there should be a single metric stream output for this test.
959        let make_scope = |attributes| {
960            InstrumentationScope::builder("test.meter")
961                .with_version("v0.1.0")
962                .with_schema_url("http://example.com")
963                .with_attributes(attributes)
964                .build()
965        };
966
967        let meter1 =
968            meter_provider.meter_with_scope(make_scope(vec![KeyValue::new("key", "value1")]));
969        let meter2 =
970            meter_provider.meter_with_scope(make_scope(vec![KeyValue::new("key", "value1")]));
971
972        let counter1 = meter1
973            .u64_counter("my_counter")
974            .with_unit("my_unit")
975            .with_description("my_description")
976            .build();
977
978        let counter2 = meter2
979            .u64_counter("my_counter")
980            .with_unit("my_unit")
981            .with_description("my_description")
982            .build();
983
984        let attribute = vec![KeyValue::new("key1", "value1")];
985        counter1.add(10, &attribute);
986        counter2.add(5, &attribute);
987
988        meter_provider.force_flush().unwrap();
989
990        // Assert
991        let resource_metrics = exporter
992            .get_finished_metrics()
993            .expect("metrics are expected to be exported.");
994        println!("resource_metrics: {resource_metrics:?}");
995        assert!(
996            resource_metrics[0].scope_metrics.len() == 1,
997            "There should be a single scope as the meters are identical"
998        );
999        assert!(
1000            resource_metrics[0].scope_metrics[0].metrics.len() == 1,
1001            "There should be single metric for the scope as instruments are identical"
1002        );
1003
1004        let scope = &resource_metrics[0].scope_metrics[0].scope;
1005        assert_eq!(scope.name(), "test.meter");
1006        assert_eq!(scope.version(), Some("v0.1.0"));
1007        assert_eq!(scope.schema_url(), Some("http://example.com"));
1008
1009        // This is validating current behavior, but it is not guaranteed to be the case in the future,
1010        // as this is a user error and SDK reserves right to change this behavior.
1011        assert!(scope.attributes().eq(&[KeyValue::new("key", "value1")]));
1012
1013        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
1014        assert_eq!(metric.name, "my_counter");
1015        assert_eq!(metric.unit, "my_unit");
1016        assert_eq!(metric.description, "my_description");
1017
1018        let MetricData::Sum(sum) = u64::extract_metrics_data_ref(&metric.data)
1019            .expect("Sum aggregation expected for Counter instruments by default")
1020        else {
1021            unreachable!()
1022        };
1023
1024        // Expecting 1 time-series.
1025        assert_eq!(sum.data_points.len(), 1);
1026
1027        let datapoint = &sum.data_points[0];
1028        assert_eq!(datapoint.value, 15);
1029    }
1030
1031    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
1032    async fn histogram_aggregation_with_invalid_aggregation_should_proceed_as_if_view_not_exist() {
1033        // Run this test with stdout enabled to see output.
1034        // cargo test histogram_aggregation_with_invalid_aggregation_should_proceed_as_if_view_not_exist --features=testing -- --nocapture
1035
1036        // Arrange
1037        let exporter = InMemoryMetricExporter::default();
1038        let view = |i: &Instrument| {
1039            if i.name == "test_histogram" {
1040                Stream::builder()
1041                    .with_aggregation(aggregation::Aggregation::ExplicitBucketHistogram {
1042                        boundaries: vec![0.9, 1.9, 1.2, 1.3, 1.4, 1.5], // invalid boundaries
1043                        record_min_max: false,
1044                    })
1045                    .with_name("test_histogram_renamed")
1046                    .with_unit("test_unit_renamed")
1047                    .build()
1048                    .ok()
1049            } else {
1050                None
1051            }
1052        };
1053        let meter_provider = SdkMeterProvider::builder()
1054            .with_periodic_exporter(exporter.clone())
1055            .with_view(view)
1056            .build();
1057
1058        // Act
1059        let meter = meter_provider.meter("test");
1060        let histogram = meter
1061            .f64_histogram("test_histogram")
1062            .with_unit("test_unit")
1063            .build();
1064
1065        histogram.record(1.5, &[KeyValue::new("key1", "value1")]);
1066        meter_provider.force_flush().unwrap();
1067
1068        // Assert
1069        let resource_metrics = exporter
1070            .get_finished_metrics()
1071            .expect("metrics are expected to be exported.");
1072        assert!(!resource_metrics.is_empty());
1073        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
1074        assert_eq!(
1075            metric.name, "test_histogram",
1076            "View rename should be ignored and original name retained."
1077        );
1078        assert_eq!(
1079            metric.unit, "test_unit",
1080            "View rename of unit should be ignored and original unit retained."
1081        );
1082    }
1083
1084    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
1085    async fn counter_with_lastvalue_aggregation_uses_default() {
1086        // LastValue aggregation is only valid for Gauge instruments.
1087        // When applied to a Counter via a view, the view is ignored and
1088        // the default aggregation (Sum) is used per the spec:
1089        // "proceed as if the View did not match"
1090
1091        // Arrange
1092        let exporter = InMemoryMetricExporter::default();
1093        let view = |i: &Instrument| {
1094            if i.name == "my_counter" {
1095                Stream::builder()
1096                    .with_aggregation(aggregation::Aggregation::LastValue)
1097                    .with_name("my_counter_renamed")
1098                    .build()
1099                    .ok()
1100            } else {
1101                None
1102            }
1103        };
1104        let meter_provider = SdkMeterProvider::builder()
1105            .with_periodic_exporter(exporter.clone())
1106            .with_view(view)
1107            .build();
1108
1109        // Act
1110        let meter = meter_provider.meter("test");
1111        let counter = meter.u64_counter("my_counter").build();
1112        counter.add(10, &[KeyValue::new("key1", "value1")]);
1113        meter_provider.force_flush().unwrap();
1114
1115        // Assert - view is ignored, default aggregation is used
1116        let resource_metrics = exporter
1117            .get_finished_metrics()
1118            .expect("metrics are expected to be exported.");
1119        assert!(!resource_metrics.is_empty());
1120        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
1121        // Original name is used (view rename is ignored)
1122        assert_eq!(
1123            metric.name, "my_counter",
1124            "View rename should be ignored due to incompatible aggregation."
1125        );
1126        // Default Sum aggregation is used
1127        assert!(
1128            matches!(
1129                &metric.data,
1130                data::AggregatedMetrics::U64(data::MetricData::Sum(_))
1131            ),
1132            "Counter should use default Sum aggregation when LastValue is incompatible."
1133        );
1134    }
1135
1136    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
1137    async fn gauge_with_sum_aggregation_uses_default() {
1138        // Sum aggregation is not valid for Gauge instruments.
1139        // When applied to a Gauge via a view, the view is ignored and
1140        // the default aggregation (LastValue/Gauge) is used per the spec:
1141        // "proceed as if the View did not match"
1142
1143        // Arrange
1144        let exporter = InMemoryMetricExporter::default();
1145        let view = |i: &Instrument| {
1146            if i.name == "my_gauge" {
1147                Stream::builder()
1148                    .with_aggregation(aggregation::Aggregation::Sum)
1149                    .with_name("my_gauge_renamed")
1150                    .build()
1151                    .ok()
1152            } else {
1153                None
1154            }
1155        };
1156        let meter_provider = SdkMeterProvider::builder()
1157            .with_periodic_exporter(exporter.clone())
1158            .with_view(view)
1159            .build();
1160
1161        // Act
1162        let meter = meter_provider.meter("test");
1163        let gauge = meter.f64_gauge("my_gauge").build();
1164        gauge.record(42.0, &[KeyValue::new("key1", "value1")]);
1165        meter_provider.force_flush().unwrap();
1166
1167        // Assert - view is ignored, default aggregation is used
1168        let resource_metrics = exporter
1169            .get_finished_metrics()
1170            .expect("metrics are expected to be exported.");
1171        assert!(!resource_metrics.is_empty());
1172        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
1173        // Original name is used (view rename is ignored)
1174        assert_eq!(
1175            metric.name, "my_gauge",
1176            "View rename should be ignored due to incompatible aggregation."
1177        );
1178        // Default Gauge (LastValue) aggregation is used
1179        assert!(
1180            matches!(
1181                &metric.data,
1182                data::AggregatedMetrics::F64(data::MetricData::Gauge(_))
1183            ),
1184            "Gauge should use default LastValue aggregation when Sum is incompatible."
1185        );
1186    }
1187
1188    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
1189    async fn updowncounter_with_lastvalue_aggregation_uses_default() {
1190        // LastValue aggregation is only valid for Gauge instruments.
1191        // When applied to an UpDownCounter via a view, the view is ignored and
1192        // the default aggregation (Sum) is used per the spec:
1193        // "proceed as if the View did not match"
1194
1195        // Arrange
1196        let exporter = InMemoryMetricExporter::default();
1197        let view = |i: &Instrument| {
1198            if i.name == "my_updown_counter" {
1199                Stream::builder()
1200                    .with_aggregation(aggregation::Aggregation::LastValue)
1201                    .with_name("my_updown_counter_renamed")
1202                    .build()
1203                    .ok()
1204            } else {
1205                None
1206            }
1207        };
1208        let meter_provider = SdkMeterProvider::builder()
1209            .with_periodic_exporter(exporter.clone())
1210            .with_view(view)
1211            .build();
1212
1213        // Act
1214        let meter = meter_provider.meter("test");
1215        let counter = meter.i64_up_down_counter("my_updown_counter").build();
1216        counter.add(-5, &[KeyValue::new("key1", "value1")]);
1217        meter_provider.force_flush().unwrap();
1218
1219        // Assert - view is ignored, default aggregation is used
1220        let resource_metrics = exporter
1221            .get_finished_metrics()
1222            .expect("metrics are expected to be exported.");
1223        assert!(!resource_metrics.is_empty());
1224        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
1225        // Original name is used (view rename is ignored)
1226        assert_eq!(
1227            metric.name, "my_updown_counter",
1228            "View rename should be ignored due to incompatible aggregation."
1229        );
1230        // Default Sum aggregation is used
1231        assert!(
1232            matches!(
1233                &metric.data,
1234                data::AggregatedMetrics::I64(data::MetricData::Sum(_))
1235            ),
1236            "UpDownCounter should use default Sum aggregation when LastValue is incompatible."
1237        );
1238    }
1239
1240    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
1241    async fn histogram_with_lastvalue_aggregation_uses_default() {
1242        // LastValue aggregation is only valid for Gauge instruments.
1243        // When applied to a Histogram via a view, the view is ignored and
1244        // the default aggregation (ExplicitBucketHistogram) is used per the spec:
1245        // "proceed as if the View did not match"
1246
1247        // Arrange
1248        let exporter = InMemoryMetricExporter::default();
1249        let view = |i: &Instrument| {
1250            if i.name == "my_histogram" {
1251                Stream::builder()
1252                    .with_aggregation(aggregation::Aggregation::LastValue)
1253                    .with_name("my_histogram_renamed")
1254                    .build()
1255                    .ok()
1256            } else {
1257                None
1258            }
1259        };
1260        let meter_provider = SdkMeterProvider::builder()
1261            .with_periodic_exporter(exporter.clone())
1262            .with_view(view)
1263            .build();
1264
1265        // Act
1266        let meter = meter_provider.meter("test");
1267        let histogram = meter.f64_histogram("my_histogram").build();
1268        histogram.record(42.0, &[KeyValue::new("key1", "value1")]);
1269        meter_provider.force_flush().unwrap();
1270
1271        // Assert - view is ignored, default aggregation is used
1272        let resource_metrics = exporter
1273            .get_finished_metrics()
1274            .expect("metrics are expected to be exported.");
1275        assert!(!resource_metrics.is_empty());
1276        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
1277        // Original name is used (view rename is ignored)
1278        assert_eq!(
1279            metric.name, "my_histogram",
1280            "View rename should be ignored due to incompatible aggregation."
1281        );
1282        // Default Histogram aggregation is used
1283        assert!(
1284            matches!(
1285                &metric.data,
1286                data::AggregatedMetrics::F64(data::MetricData::Histogram(_))
1287            ),
1288            "Histogram should use default ExplicitBucketHistogram aggregation when LastValue is incompatible."
1289        );
1290    }
1291
1292    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
1293    async fn observable_gauge_with_sum_aggregation_uses_default() {
1294        // Sum aggregation is not valid for Observable Gauge instruments.
1295        // When applied to an Observable Gauge via a view, the view is ignored and
1296        // the default aggregation (LastValue/Gauge) is used per the spec:
1297        // "proceed as if the View did not match"
1298
1299        // Arrange
1300        let exporter = InMemoryMetricExporter::default();
1301        let view = |i: &Instrument| {
1302            if i.name == "my_observable_gauge" {
1303                Stream::builder()
1304                    .with_aggregation(aggregation::Aggregation::Sum)
1305                    .with_name("my_observable_gauge_renamed")
1306                    .build()
1307                    .ok()
1308            } else {
1309                None
1310            }
1311        };
1312        let meter_provider = SdkMeterProvider::builder()
1313            .with_periodic_exporter(exporter.clone())
1314            .with_view(view)
1315            .build();
1316
1317        // Act
1318        let meter = meter_provider.meter("test");
1319        let _observable_gauge = meter
1320            .f64_observable_gauge("my_observable_gauge")
1321            .with_callback(|observer| {
1322                observer.observe(42.0, &[KeyValue::new("key1", "value1")]);
1323            })
1324            .build();
1325        meter_provider.force_flush().unwrap();
1326
1327        // Assert - view is ignored, default aggregation is used
1328        let resource_metrics = exporter
1329            .get_finished_metrics()
1330            .expect("metrics are expected to be exported.");
1331        assert!(!resource_metrics.is_empty());
1332        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
1333        // Original name is used (view rename is ignored)
1334        assert_eq!(
1335            metric.name, "my_observable_gauge",
1336            "View rename should be ignored due to incompatible aggregation."
1337        );
1338        // Default Gauge (LastValue) aggregation is used
1339        assert!(
1340            matches!(
1341                &metric.data,
1342                data::AggregatedMetrics::F64(data::MetricData::Gauge(_))
1343            ),
1344            "Observable Gauge should use default LastValue aggregation when Sum is incompatible."
1345        );
1346    }
1347
1348    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
1349    async fn observable_counter_with_lastvalue_aggregation_uses_default() {
1350        // LastValue aggregation is only valid for Gauge instruments.
1351        // When applied to an Observable Counter via a view, the view is ignored and
1352        // the default aggregation (Sum) is used per the spec:
1353        // "proceed as if the View did not match"
1354
1355        // Arrange
1356        let exporter = InMemoryMetricExporter::default();
1357        let view = |i: &Instrument| {
1358            if i.name == "my_observable_counter" {
1359                Stream::builder()
1360                    .with_aggregation(aggregation::Aggregation::LastValue)
1361                    .with_name("my_observable_counter_renamed")
1362                    .build()
1363                    .ok()
1364            } else {
1365                None
1366            }
1367        };
1368        let meter_provider = SdkMeterProvider::builder()
1369            .with_periodic_exporter(exporter.clone())
1370            .with_view(view)
1371            .build();
1372
1373        // Act
1374        let meter = meter_provider.meter("test");
1375        let _observable_counter = meter
1376            .u64_observable_counter("my_observable_counter")
1377            .with_callback(|observer| {
1378                observer.observe(100, &[KeyValue::new("key1", "value1")]);
1379            })
1380            .build();
1381        meter_provider.force_flush().unwrap();
1382
1383        // Assert - view is ignored, default aggregation is used
1384        let resource_metrics = exporter
1385            .get_finished_metrics()
1386            .expect("metrics are expected to be exported.");
1387        assert!(!resource_metrics.is_empty());
1388        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
1389        // Original name is used (view rename is ignored)
1390        assert_eq!(
1391            metric.name, "my_observable_counter",
1392            "View rename should be ignored due to incompatible aggregation."
1393        );
1394        // Default Sum aggregation is used
1395        assert!(
1396            matches!(
1397                &metric.data,
1398                data::AggregatedMetrics::U64(data::MetricData::Sum(_))
1399            ),
1400            "Observable Counter should use default Sum aggregation when LastValue is incompatible."
1401        );
1402    }
1403
1404    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
1405    async fn observable_updowncounter_with_lastvalue_aggregation_uses_default() {
1406        // LastValue aggregation is only valid for Gauge instruments.
1407        // When applied to an Observable UpDownCounter via a view, the view is ignored and
1408        // the default aggregation (Sum) is used per the spec:
1409        // "proceed as if the View did not match"
1410
1411        // Arrange
1412        let exporter = InMemoryMetricExporter::default();
1413        let view = |i: &Instrument| {
1414            if i.name == "my_observable_updowncounter" {
1415                Stream::builder()
1416                    .with_aggregation(aggregation::Aggregation::LastValue)
1417                    .with_name("my_observable_updowncounter_renamed")
1418                    .build()
1419                    .ok()
1420            } else {
1421                None
1422            }
1423        };
1424        let meter_provider = SdkMeterProvider::builder()
1425            .with_periodic_exporter(exporter.clone())
1426            .with_view(view)
1427            .build();
1428
1429        // Act
1430        let meter = meter_provider.meter("test");
1431        let _observable_updowncounter = meter
1432            .i64_observable_up_down_counter("my_observable_updowncounter")
1433            .with_callback(|observer| {
1434                observer.observe(-50, &[KeyValue::new("key1", "value1")]);
1435            })
1436            .build();
1437        meter_provider.force_flush().unwrap();
1438
1439        // Assert - view is ignored, default aggregation is used
1440        let resource_metrics = exporter
1441            .get_finished_metrics()
1442            .expect("metrics are expected to be exported.");
1443        assert!(!resource_metrics.is_empty());
1444        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
1445        // Original name is used (view rename is ignored)
1446        assert_eq!(
1447            metric.name, "my_observable_updowncounter",
1448            "View rename should be ignored due to incompatible aggregation."
1449        );
1450        // Default Sum aggregation is used
1451        assert!(
1452            matches!(
1453                &metric.data,
1454                data::AggregatedMetrics::I64(data::MetricData::Sum(_))
1455            ),
1456            "Observable UpDownCounter should use default Sum aggregation when LastValue is incompatible."
1457        );
1458    }
1459
1460    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
1461    async fn histogram_with_sum_aggregation_is_valid() {
1462        // Sum aggregation is valid for Histogram instruments.
1463        // When applied via a view, the aggregation should change from
1464        // ExplicitBucketHistogram to Sum.
1465
1466        // Arrange
1467        let exporter = InMemoryMetricExporter::default();
1468        let view = |i: &Instrument| {
1469            if i.name == "my_histogram" {
1470                Stream::builder()
1471                    .with_aggregation(aggregation::Aggregation::Sum)
1472                    .with_name("my_histogram_renamed")
1473                    .build()
1474                    .ok()
1475            } else {
1476                None
1477            }
1478        };
1479        let meter_provider = SdkMeterProvider::builder()
1480            .with_periodic_exporter(exporter.clone())
1481            .with_view(view)
1482            .build();
1483
1484        // Act
1485        let meter = meter_provider.meter("test");
1486        let histogram = meter.f64_histogram("my_histogram").build();
1487        histogram.record(10.0, &[KeyValue::new("key1", "value1")]);
1488        histogram.record(20.0, &[KeyValue::new("key1", "value1")]);
1489        histogram.record(30.0, &[KeyValue::new("key1", "value1")]);
1490        meter_provider.force_flush().unwrap();
1491
1492        // Assert - view is applied, Sum aggregation is used
1493        let resource_metrics = exporter
1494            .get_finished_metrics()
1495            .expect("metrics are expected to be exported.");
1496        assert!(!resource_metrics.is_empty());
1497        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
1498        // View rename is applied
1499        assert_eq!(
1500            metric.name, "my_histogram_renamed",
1501            "View rename should be applied for compatible aggregation."
1502        );
1503        // Sum aggregation is used instead of default Histogram
1504        let MetricData::Sum(sum) = f64::extract_metrics_data_ref(&metric.data)
1505            .expect("Sum aggregation expected when view specifies Sum")
1506        else {
1507            panic!("Expected Sum aggregation for Histogram with Sum view");
1508        };
1509        assert_eq!(sum.data_points.len(), 1);
1510        assert_eq!(sum.data_points[0].value, 60.0); // 10 + 20 + 30
1511    }
1512
1513    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
1514    async fn gauge_with_histogram_aggregation_is_valid() {
1515        // Histogram aggregation is valid for Gauge instruments.
1516        // When applied via a view, the aggregation should change from
1517        // LastValue to ExplicitBucketHistogram.
1518
1519        // Arrange
1520        let exporter = InMemoryMetricExporter::default();
1521        let view = |i: &Instrument| {
1522            if i.name == "my_gauge" {
1523                Stream::builder()
1524                    .with_aggregation(aggregation::Aggregation::ExplicitBucketHistogram {
1525                        boundaries: vec![5.0, 10.0, 25.0, 50.0],
1526                        record_min_max: true,
1527                    })
1528                    .with_name("my_gauge_renamed")
1529                    .build()
1530                    .ok()
1531            } else {
1532                None
1533            }
1534        };
1535        let meter_provider = SdkMeterProvider::builder()
1536            .with_periodic_exporter(exporter.clone())
1537            .with_view(view)
1538            .build();
1539
1540        // Act
1541        let meter = meter_provider.meter("test");
1542        let gauge = meter.f64_gauge("my_gauge").build();
1543        gauge.record(3.0, &[KeyValue::new("key1", "value1")]);
1544        gauge.record(7.0, &[KeyValue::new("key1", "value1")]);
1545        gauge.record(15.0, &[KeyValue::new("key1", "value1")]);
1546        gauge.record(30.0, &[KeyValue::new("key1", "value1")]);
1547        meter_provider.force_flush().unwrap();
1548
1549        // Assert - view is applied, Histogram aggregation is used
1550        let resource_metrics = exporter
1551            .get_finished_metrics()
1552            .expect("metrics are expected to be exported.");
1553        assert!(!resource_metrics.is_empty());
1554        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
1555        // View rename is applied
1556        assert_eq!(
1557            metric.name, "my_gauge_renamed",
1558            "View rename should be applied for compatible aggregation."
1559        );
1560        // Histogram aggregation is used instead of default LastValue
1561        let MetricData::Histogram(histogram) = f64::extract_metrics_data_ref(&metric.data)
1562            .expect("Histogram aggregation expected when view specifies Histogram")
1563        else {
1564            panic!("Expected Histogram aggregation for Gauge with Histogram view");
1565        };
1566        assert_eq!(histogram.data_points.len(), 1);
1567        let dp = &histogram.data_points[0];
1568        assert_eq!(dp.count, 4);
1569        assert_eq!(dp.sum, 55.0); // 3 + 7 + 15 + 30
1570        assert_eq!(dp.min, Some(3.0));
1571        assert_eq!(dp.max, Some(30.0));
1572        // Bucket boundaries: [5.0, 10.0, 25.0, 50.0]
1573        // Values: 3.0 (bucket 0), 7.0 (bucket 1), 15.0 (bucket 2), 30.0 (bucket 3)
1574        assert_eq!(dp.bucket_counts, vec![1, 1, 1, 1, 0]);
1575    }
1576
1577    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
1578    async fn counter_with_histogram_aggregation_is_valid() {
1579        // Histogram aggregation is valid for Counter instruments.
1580        // When applied via a view, the aggregation should change from
1581        // Sum to ExplicitBucketHistogram.
1582
1583        // Arrange
1584        let exporter = InMemoryMetricExporter::default();
1585        let view = |i: &Instrument| {
1586            if i.name == "my_counter" {
1587                Stream::builder()
1588                    .with_aggregation(aggregation::Aggregation::ExplicitBucketHistogram {
1589                        boundaries: vec![5.0, 10.0, 25.0, 50.0],
1590                        record_min_max: true,
1591                    })
1592                    .with_name("my_counter_renamed")
1593                    .build()
1594                    .ok()
1595            } else {
1596                None
1597            }
1598        };
1599        let meter_provider = SdkMeterProvider::builder()
1600            .with_periodic_exporter(exporter.clone())
1601            .with_view(view)
1602            .build();
1603
1604        // Act
1605        let meter = meter_provider.meter("test");
1606        let counter = meter.u64_counter("my_counter").build();
1607        counter.add(3, &[KeyValue::new("key1", "value1")]);
1608        counter.add(7, &[KeyValue::new("key1", "value1")]);
1609        counter.add(15, &[KeyValue::new("key1", "value1")]);
1610        counter.add(30, &[KeyValue::new("key1", "value1")]);
1611        meter_provider.force_flush().unwrap();
1612
1613        // Assert - view is applied, Histogram aggregation is used
1614        let resource_metrics = exporter
1615            .get_finished_metrics()
1616            .expect("metrics are expected to be exported.");
1617        assert!(!resource_metrics.is_empty());
1618        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
1619        // View rename is applied
1620        assert_eq!(
1621            metric.name, "my_counter_renamed",
1622            "View rename should be applied for compatible aggregation."
1623        );
1624        // Histogram aggregation is used instead of default Sum
1625        let MetricData::Histogram(histogram) = u64::extract_metrics_data_ref(&metric.data)
1626            .expect("Histogram aggregation expected when view specifies Histogram")
1627        else {
1628            panic!("Expected Histogram aggregation for Counter with Histogram view");
1629        };
1630        assert_eq!(histogram.data_points.len(), 1);
1631        let dp = &histogram.data_points[0];
1632        assert_eq!(dp.count, 4);
1633        assert_eq!(dp.sum, 55); // 3 + 7 + 15 + 30
1634        assert_eq!(dp.min, Some(3));
1635        assert_eq!(dp.max, Some(30));
1636        // Bucket boundaries: [5.0, 10.0, 25.0, 50.0]
1637        // Values: 3 (bucket 0), 7 (bucket 1), 15 (bucket 2), 30 (bucket 3)
1638        assert_eq!(dp.bucket_counts, vec![1, 1, 1, 1, 0]);
1639    }
1640
1641    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
1642    async fn updowncounter_with_histogram_aggregation_is_valid() {
1643        // Histogram aggregation is valid for UpDownCounter instruments.
1644        // When applied via a view, the aggregation should change from
1645        // Sum to ExplicitBucketHistogram.
1646
1647        // Arrange
1648        let exporter = InMemoryMetricExporter::default();
1649        let view = |i: &Instrument| {
1650            if i.name == "my_updowncounter" {
1651                Stream::builder()
1652                    .with_aggregation(aggregation::Aggregation::ExplicitBucketHistogram {
1653                        boundaries: vec![0.0, 10.0, 20.0, 50.0],
1654                        record_min_max: true,
1655                    })
1656                    .with_name("my_updowncounter_renamed")
1657                    .build()
1658                    .ok()
1659            } else {
1660                None
1661            }
1662        };
1663        let meter_provider = SdkMeterProvider::builder()
1664            .with_periodic_exporter(exporter.clone())
1665            .with_view(view)
1666            .build();
1667
1668        // Act
1669        let meter = meter_provider.meter("test");
1670        let updowncounter = meter.i64_up_down_counter("my_updowncounter").build();
1671        updowncounter.add(-5, &[KeyValue::new("key1", "value1")]);
1672        updowncounter.add(15, &[KeyValue::new("key1", "value1")]);
1673        updowncounter.add(25, &[KeyValue::new("key1", "value1")]);
1674        meter_provider.force_flush().unwrap();
1675
1676        // Assert - view is applied, Histogram aggregation is used
1677        let resource_metrics = exporter
1678            .get_finished_metrics()
1679            .expect("metrics are expected to be exported.");
1680        assert!(!resource_metrics.is_empty());
1681        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
1682        // View rename is applied
1683        assert_eq!(
1684            metric.name, "my_updowncounter_renamed",
1685            "View rename should be applied for compatible aggregation."
1686        );
1687        // Histogram aggregation is used instead of default Sum
1688        let MetricData::Histogram(histogram) = i64::extract_metrics_data_ref(&metric.data)
1689            .expect("Histogram aggregation expected when view specifies Histogram")
1690        else {
1691            panic!("Expected Histogram aggregation for UpDownCounter with Histogram view");
1692        };
1693        assert_eq!(histogram.data_points.len(), 1);
1694        let dp = &histogram.data_points[0];
1695        assert_eq!(dp.count, 3);
1696        // Note: Sum is not recorded for UpDownCounter histogram per the spec
1697    }
1698
1699    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
1700    async fn counter_with_exponential_histogram_aggregation_is_valid() {
1701        // ExponentialHistogram aggregation is valid for Counter instruments.
1702        // When applied via a view, the aggregation should change from
1703        // Sum to Base2ExponentialHistogram.
1704
1705        // Arrange
1706        let exporter = InMemoryMetricExporter::default();
1707        let view = |i: &Instrument| {
1708            if i.name == "my_counter" {
1709                Stream::builder()
1710                    .with_aggregation(aggregation::Aggregation::Base2ExponentialHistogram {
1711                        max_size: 160,
1712                        max_scale: 20,
1713                        record_min_max: true,
1714                    })
1715                    .with_name("my_counter_renamed")
1716                    .build()
1717                    .ok()
1718            } else {
1719                None
1720            }
1721        };
1722        let meter_provider = SdkMeterProvider::builder()
1723            .with_periodic_exporter(exporter.clone())
1724            .with_view(view)
1725            .build();
1726
1727        // Act
1728        let meter = meter_provider.meter("test");
1729        let counter = meter.u64_counter("my_counter").build();
1730        counter.add(5, &[KeyValue::new("key1", "value1")]);
1731        counter.add(10, &[KeyValue::new("key1", "value1")]);
1732        counter.add(20, &[KeyValue::new("key1", "value1")]);
1733        meter_provider.force_flush().unwrap();
1734
1735        // Assert - view is applied, ExponentialHistogram aggregation is used
1736        let resource_metrics = exporter
1737            .get_finished_metrics()
1738            .expect("metrics are expected to be exported.");
1739        assert!(!resource_metrics.is_empty());
1740        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
1741        // View rename is applied
1742        assert_eq!(
1743            metric.name, "my_counter_renamed",
1744            "View rename should be applied for compatible aggregation."
1745        );
1746        // ExponentialHistogram aggregation is used instead of default Sum
1747        let MetricData::ExponentialHistogram(exp_hist) =
1748            u64::extract_metrics_data_ref(&metric.data)
1749                .expect("ExponentialHistogram aggregation expected when view specifies it")
1750        else {
1751            panic!("Expected ExponentialHistogram aggregation for Counter with ExponentialHistogram view");
1752        };
1753        assert_eq!(exp_hist.data_points.len(), 1);
1754        let dp = &exp_hist.data_points[0];
1755        assert_eq!(dp.count(), 3);
1756        assert_eq!(dp.sum(), 35); // 5 + 10 + 20
1757        assert_eq!(dp.min(), Some(5));
1758        assert_eq!(dp.max(), Some(20));
1759    }
1760
1761    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
1762    async fn gauge_with_exponential_histogram_aggregation_is_valid() {
1763        // ExponentialHistogram aggregation is valid for Gauge instruments.
1764        // When applied via a view, the aggregation should change from
1765        // LastValue to Base2ExponentialHistogram.
1766
1767        // Arrange
1768        let exporter = InMemoryMetricExporter::default();
1769        let view = |i: &Instrument| {
1770            if i.name == "my_gauge" {
1771                Stream::builder()
1772                    .with_aggregation(aggregation::Aggregation::Base2ExponentialHistogram {
1773                        max_size: 160,
1774                        max_scale: 20,
1775                        record_min_max: true,
1776                    })
1777                    .with_name("my_gauge_renamed")
1778                    .build()
1779                    .ok()
1780            } else {
1781                None
1782            }
1783        };
1784        let meter_provider = SdkMeterProvider::builder()
1785            .with_periodic_exporter(exporter.clone())
1786            .with_view(view)
1787            .build();
1788
1789        // Act
1790        let meter = meter_provider.meter("test");
1791        let gauge = meter.f64_gauge("my_gauge").build();
1792        gauge.record(2.5, &[KeyValue::new("key1", "value1")]);
1793        gauge.record(7.5, &[KeyValue::new("key1", "value1")]);
1794        gauge.record(15.0, &[KeyValue::new("key1", "value1")]);
1795        meter_provider.force_flush().unwrap();
1796
1797        // Assert - view is applied, ExponentialHistogram aggregation is used
1798        let resource_metrics = exporter
1799            .get_finished_metrics()
1800            .expect("metrics are expected to be exported.");
1801        assert!(!resource_metrics.is_empty());
1802        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
1803        // View rename is applied
1804        assert_eq!(
1805            metric.name, "my_gauge_renamed",
1806            "View rename should be applied for compatible aggregation."
1807        );
1808        // ExponentialHistogram aggregation is used instead of default LastValue
1809        let MetricData::ExponentialHistogram(exp_hist) =
1810            f64::extract_metrics_data_ref(&metric.data)
1811                .expect("ExponentialHistogram aggregation expected when view specifies it")
1812        else {
1813            panic!("Expected ExponentialHistogram aggregation for Gauge with ExponentialHistogram view");
1814        };
1815        assert_eq!(exp_hist.data_points.len(), 1);
1816        let dp = &exp_hist.data_points[0];
1817        assert_eq!(dp.count(), 3);
1818        assert_eq!(dp.sum(), 25.0); // 2.5 + 7.5 + 15.0
1819        assert_eq!(dp.min(), Some(2.5));
1820        assert_eq!(dp.max(), Some(15.0));
1821    }
1822
1823    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
1824    async fn counter_with_drop_aggregation_is_dropped() {
1825        // Run this test with stdout enabled to see output.
1826        // cargo test counter_with_drop_aggregation_is_dropped --features=testing -- --nocapture
1827
1828        // When a view matches an instrument and specifies Aggregation::Drop,
1829        // the instrument should be dropped and no metrics should be exported.
1830
1831        // Arrange
1832        let exporter = InMemoryMetricExporter::default();
1833        let view = |i: &Instrument| {
1834            if i.name == "my_counter_to_drop" {
1835                Stream::builder()
1836                    .with_aggregation(aggregation::Aggregation::Drop)
1837                    .build()
1838                    .ok()
1839            } else {
1840                None
1841            }
1842        };
1843        let meter_provider = SdkMeterProvider::builder()
1844            .with_periodic_exporter(exporter.clone())
1845            .with_view(view)
1846            .build();
1847
1848        // Act
1849        let meter = meter_provider.meter("test");
1850        let counter = meter.u64_counter("my_counter_to_drop").build();
1851        counter.add(10, &[KeyValue::new("key1", "value1")]);
1852        meter_provider.force_flush().unwrap();
1853
1854        // Assert - no metrics should be exported because the view drops the instrument
1855        let resource_metrics = exporter
1856            .get_finished_metrics()
1857            .expect("metrics result expected");
1858        assert!(
1859            resource_metrics.is_empty()
1860                || resource_metrics[0].scope_metrics.is_empty()
1861                || resource_metrics[0].scope_metrics[0].metrics.is_empty(),
1862            "No metrics should be exported when view uses Aggregation::Drop. Got: {:?}",
1863            resource_metrics
1864        );
1865    }
1866
1867    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
1868    async fn histogram_with_drop_aggregation_is_dropped() {
1869        // Run this test with stdout enabled to see output.
1870        // cargo test histogram_with_drop_aggregation_is_dropped --features=testing -- --nocapture
1871
1872        // When a view matches a histogram and specifies Aggregation::Drop,
1873        // the instrument should be dropped and no metrics should be exported.
1874
1875        // Arrange
1876        let exporter = InMemoryMetricExporter::default();
1877        let view = |i: &Instrument| {
1878            if i.name == "my_histogram_to_drop" {
1879                Stream::builder()
1880                    .with_aggregation(aggregation::Aggregation::Drop)
1881                    .build()
1882                    .ok()
1883            } else {
1884                None
1885            }
1886        };
1887        let meter_provider = SdkMeterProvider::builder()
1888            .with_periodic_exporter(exporter.clone())
1889            .with_view(view)
1890            .build();
1891
1892        // Act
1893        let meter = meter_provider.meter("test");
1894        let histogram = meter.f64_histogram("my_histogram_to_drop").build();
1895        histogram.record(42.0, &[KeyValue::new("key1", "value1")]);
1896        meter_provider.force_flush().unwrap();
1897
1898        // Assert - no metrics should be exported because the view drops the instrument
1899        let resource_metrics = exporter
1900            .get_finished_metrics()
1901            .expect("metrics result expected");
1902        assert!(
1903            resource_metrics.is_empty()
1904                || resource_metrics[0].scope_metrics.is_empty()
1905                || resource_metrics[0].scope_metrics[0].metrics.is_empty(),
1906            "No metrics should be exported when view uses Aggregation::Drop. Got: {:?}",
1907            resource_metrics
1908        );
1909    }
1910
1911    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
1912    async fn gauge_with_drop_aggregation_is_dropped() {
1913        // Run this test with stdout enabled to see output.
1914        // cargo test gauge_with_drop_aggregation_is_dropped --features=testing -- --nocapture
1915
1916        // When a view matches a gauge and specifies Aggregation::Drop,
1917        // the instrument should be dropped and no metrics should be exported.
1918
1919        // Arrange
1920        let exporter = InMemoryMetricExporter::default();
1921        let view = |i: &Instrument| {
1922            if i.name == "my_gauge_to_drop" {
1923                Stream::builder()
1924                    .with_aggregation(aggregation::Aggregation::Drop)
1925                    .build()
1926                    .ok()
1927            } else {
1928                None
1929            }
1930        };
1931        let meter_provider = SdkMeterProvider::builder()
1932            .with_periodic_exporter(exporter.clone())
1933            .with_view(view)
1934            .build();
1935
1936        // Act
1937        let meter = meter_provider.meter("test");
1938        let gauge = meter.f64_gauge("my_gauge_to_drop").build();
1939        gauge.record(42.0, &[KeyValue::new("key1", "value1")]);
1940        meter_provider.force_flush().unwrap();
1941
1942        // Assert - no metrics should be exported because the view drops the instrument
1943        let resource_metrics = exporter
1944            .get_finished_metrics()
1945            .expect("metrics result expected");
1946        assert!(
1947            resource_metrics.is_empty()
1948                || resource_metrics[0].scope_metrics.is_empty()
1949                || resource_metrics[0].scope_metrics[0].metrics.is_empty(),
1950            "No metrics should be exported when view uses Aggregation::Drop. Got: {:?}",
1951            resource_metrics
1952        );
1953    }
1954
1955    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
1956    async fn counter_with_drop_aggregation_and_rename_should_still_drop() {
1957        // Run this test with stdout enabled to see output.
1958        // cargo test counter_with_drop_aggregation_and_rename_should_still_drop --features=testing -- --nocapture
1959
1960        // When a view matches and specifies Aggregation::Drop, the instrument should be
1961        // dropped even if the view also specifies other customizations (like name).
1962        // The name customization is meaningless for a dropped instrument, but the Drop
1963        // intent should still be honored.
1964
1965        // Arrange
1966        let exporter = InMemoryMetricExporter::default();
1967        let view = |i: &Instrument| {
1968            if i.name == "my_counter" {
1969                Stream::builder()
1970                    .with_name("dropped_counter") // Meaningless but shouldn't break Drop
1971                    .with_aggregation(aggregation::Aggregation::Drop)
1972                    .build()
1973                    .ok()
1974            } else {
1975                None
1976            }
1977        };
1978        let meter_provider = SdkMeterProvider::builder()
1979            .with_periodic_exporter(exporter.clone())
1980            .with_view(view)
1981            .build();
1982
1983        // Act
1984        let meter = meter_provider.meter("test");
1985        let counter = meter.u64_counter("my_counter").build();
1986        counter.add(10, &[KeyValue::new("key1", "value1")]);
1987        meter_provider.force_flush().unwrap();
1988
1989        // Assert - no metrics should be exported because the view drops the instrument
1990        let resource_metrics = exporter
1991            .get_finished_metrics()
1992            .expect("metrics result expected");
1993        assert!(
1994            resource_metrics.is_empty()
1995                || resource_metrics[0].scope_metrics.is_empty()
1996                || resource_metrics[0].scope_metrics[0].metrics.is_empty(),
1997            "No metrics should be exported when view uses Aggregation::Drop, even with rename. Got: {:?}",
1998            resource_metrics
1999        );
2000    }
2001
2002    #[cfg(feature = "spec_unstable_metrics_views")]
2003    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
2004    #[ignore = "Spatial aggregation is not yet implemented."]
2005    async fn spatial_aggregation_when_view_drops_attributes_observable_counter() {
2006        // cargo test metrics::tests::spatial_aggregation_when_view_drops_attributes_observable_counter --features=testing
2007
2008        // Arrange
2009        let exporter = InMemoryMetricExporter::default();
2010        // View drops all attributes.
2011        let view = |i: &Instrument| {
2012            if i.name == "my_observable_counter" {
2013                Stream::builder()
2014                    .with_allowed_attribute_keys(vec![])
2015                    .build()
2016                    .ok()
2017            } else {
2018                None
2019            }
2020        };
2021        let meter_provider = SdkMeterProvider::builder()
2022            .with_periodic_exporter(exporter.clone())
2023            .with_view(view)
2024            .build();
2025
2026        // Act
2027        let meter = meter_provider.meter("test");
2028        let _observable_counter = meter
2029            .u64_observable_counter("my_observable_counter")
2030            .with_callback(|observer| {
2031                observer.observe(
2032                    100,
2033                    &[
2034                        KeyValue::new("statusCode", "200"),
2035                        KeyValue::new("verb", "get"),
2036                    ],
2037                );
2038
2039                observer.observe(
2040                    100,
2041                    &[
2042                        KeyValue::new("statusCode", "200"),
2043                        KeyValue::new("verb", "post"),
2044                    ],
2045                );
2046
2047                observer.observe(
2048                    100,
2049                    &[
2050                        KeyValue::new("statusCode", "500"),
2051                        KeyValue::new("verb", "get"),
2052                    ],
2053                );
2054            })
2055            .build();
2056
2057        meter_provider.force_flush().unwrap();
2058
2059        // Assert
2060        let resource_metrics = exporter
2061            .get_finished_metrics()
2062            .expect("metrics are expected to be exported.");
2063        assert!(!resource_metrics.is_empty());
2064        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
2065        assert_eq!(metric.name, "my_observable_counter",);
2066
2067        let MetricData::Sum(sum) = u64::extract_metrics_data_ref(&metric.data)
2068            .expect("Sum aggregation expected for ObservableCounter instruments by default")
2069        else {
2070            unreachable!()
2071        };
2072
2073        // Expecting 1 time-series only, as the view drops all attributes resulting
2074        // in a single time-series.
2075        // This is failing today, due to lack of support for spatial aggregation.
2076        assert_eq!(sum.data_points.len(), 1);
2077
2078        // find and validate the single datapoint
2079        let data_point = &sum.data_points[0];
2080        assert_eq!(data_point.value, 300);
2081    }
2082
2083    #[cfg(feature = "spec_unstable_metrics_views")]
2084    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
2085    async fn spatial_aggregation_when_view_drops_attributes_counter() {
2086        // cargo test spatial_aggregation_when_view_drops_attributes_counter --features=testing
2087
2088        // Arrange
2089        let exporter = InMemoryMetricExporter::default();
2090        // View drops all attributes.
2091        let view = |i: &Instrument| {
2092            if i.name == "my_counter" {
2093                Some(
2094                    Stream::builder()
2095                        .with_allowed_attribute_keys(vec![])
2096                        .build()
2097                        .unwrap(),
2098                )
2099            } else {
2100                None
2101            }
2102        };
2103        let meter_provider = SdkMeterProvider::builder()
2104            .with_periodic_exporter(exporter.clone())
2105            .with_view(view)
2106            .build();
2107
2108        // Act
2109        let meter = meter_provider.meter("test");
2110        let counter = meter.u64_counter("my_counter").build();
2111
2112        // Normally, this would generate 3 time-series, but since the view
2113        // drops all attributes, we expect only 1 time-series.
2114        counter.add(
2115            10,
2116            [
2117                KeyValue::new("statusCode", "200"),
2118                KeyValue::new("verb", "Get"),
2119            ]
2120            .as_ref(),
2121        );
2122
2123        counter.add(
2124            10,
2125            [
2126                KeyValue::new("statusCode", "500"),
2127                KeyValue::new("verb", "Get"),
2128            ]
2129            .as_ref(),
2130        );
2131
2132        counter.add(
2133            10,
2134            [
2135                KeyValue::new("statusCode", "200"),
2136                KeyValue::new("verb", "Post"),
2137            ]
2138            .as_ref(),
2139        );
2140
2141        meter_provider.force_flush().unwrap();
2142
2143        // Assert
2144        let resource_metrics = exporter
2145            .get_finished_metrics()
2146            .expect("metrics are expected to be exported.");
2147        assert!(!resource_metrics.is_empty());
2148        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
2149        assert_eq!(metric.name, "my_counter",);
2150
2151        let MetricData::Sum(sum) = u64::extract_metrics_data_ref(&metric.data)
2152            .expect("Sum aggregation expected for Counter instruments by default")
2153        else {
2154            unreachable!()
2155        };
2156
2157        // Expecting 1 time-series only, as the view drops all attributes resulting
2158        // in a single time-series.
2159        // This is failing today, due to lack of support for spatial aggregation.
2160        assert_eq!(sum.data_points.len(), 1);
2161        // find and validate the single datapoint
2162        let data_point = &sum.data_points[0];
2163        assert_eq!(data_point.value, 30);
2164    }
2165
2166    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
2167    async fn no_attr_cumulative_up_down_counter() {
2168        let mut test_context = TestContext::new(Temporality::Cumulative);
2169        let counter = test_context.i64_up_down_counter("test", "my_counter", Some("my_unit"));
2170
2171        counter.add(50, &[]);
2172        test_context.flush_metrics();
2173
2174        let MetricData::Sum(sum) =
2175            test_context.get_aggregation::<i64>("my_counter", Some("my_unit"))
2176        else {
2177            unreachable!()
2178        };
2179
2180        assert_eq!(sum.data_points.len(), 1, "Expected only one data point");
2181        assert!(!sum.is_monotonic, "Should not produce monotonic.");
2182        assert_eq!(
2183            sum.temporality,
2184            Temporality::Cumulative,
2185            "Should produce cumulative"
2186        );
2187
2188        let data_point = &sum.data_points[0];
2189        assert!(data_point.attributes.is_empty(), "Non-empty attribute set");
2190        assert_eq!(data_point.value, 50, "Unexpected data point value");
2191    }
2192
2193    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
2194    async fn no_attr_up_down_counter_always_cumulative() {
2195        let mut test_context = TestContext::new(Temporality::Delta);
2196        let counter = test_context.i64_up_down_counter("test", "my_counter", Some("my_unit"));
2197
2198        counter.add(50, &[]);
2199        test_context.flush_metrics();
2200
2201        let MetricData::Sum(sum) =
2202            test_context.get_aggregation::<i64>("my_counter", Some("my_unit"))
2203        else {
2204            unreachable!()
2205        };
2206
2207        assert_eq!(sum.data_points.len(), 1, "Expected only one data point");
2208        assert!(!sum.is_monotonic, "Should not produce monotonic.");
2209        assert_eq!(
2210            sum.temporality,
2211            Temporality::Cumulative,
2212            "Should produce Cumulative due to UpDownCounter temporality_preference"
2213        );
2214
2215        let data_point = &sum.data_points[0];
2216        assert!(data_point.attributes.is_empty(), "Non-empty attribute set");
2217        assert_eq!(data_point.value, 50, "Unexpected data point value");
2218    }
2219
2220    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
2221    async fn no_attr_cumulative_counter_value_added_after_export() {
2222        let mut test_context = TestContext::new(Temporality::Cumulative);
2223        let counter = test_context.u64_counter("test", "my_counter", None);
2224
2225        counter.add(50, &[]);
2226        test_context.flush_metrics();
2227        let _ = test_context.get_aggregation::<u64>("my_counter", None);
2228        test_context.reset_metrics();
2229
2230        counter.add(5, &[]);
2231        test_context.flush_metrics();
2232        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
2233            unreachable!()
2234        };
2235
2236        assert_eq!(sum.data_points.len(), 1, "Expected only one data point");
2237        assert!(sum.is_monotonic, "Should produce monotonic.");
2238        assert_eq!(
2239            sum.temporality,
2240            Temporality::Cumulative,
2241            "Should produce cumulative"
2242        );
2243
2244        let data_point = &sum.data_points[0];
2245        assert!(data_point.attributes.is_empty(), "Non-empty attribute set");
2246        assert_eq!(data_point.value, 55, "Unexpected data point value");
2247    }
2248
2249    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
2250    async fn no_attr_delta_counter_value_reset_after_export() {
2251        let mut test_context = TestContext::new(Temporality::Delta);
2252        let counter = test_context.u64_counter("test", "my_counter", None);
2253
2254        counter.add(50, &[]);
2255        test_context.flush_metrics();
2256        let _ = test_context.get_aggregation::<u64>("my_counter", None);
2257        test_context.reset_metrics();
2258
2259        counter.add(5, &[]);
2260        test_context.flush_metrics();
2261        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
2262            unreachable!()
2263        };
2264
2265        assert_eq!(sum.data_points.len(), 1, "Expected only one data point");
2266        assert!(sum.is_monotonic, "Should produce monotonic.");
2267        assert_eq!(
2268            sum.temporality,
2269            Temporality::Delta,
2270            "Should produce cumulative"
2271        );
2272
2273        let data_point = &sum.data_points[0];
2274        assert!(data_point.attributes.is_empty(), "Non-empty attribute set");
2275        assert_eq!(data_point.value, 5, "Unexpected data point value");
2276    }
2277
2278    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
2279    async fn second_delta_export_does_not_give_no_attr_value_if_add_not_called() {
2280        let mut test_context = TestContext::new(Temporality::Delta);
2281        let counter = test_context.u64_counter("test", "my_counter", None);
2282
2283        counter.add(50, &[]);
2284        test_context.flush_metrics();
2285        let _ = test_context.get_aggregation::<u64>("my_counter", None);
2286        test_context.reset_metrics();
2287
2288        counter.add(50, &[KeyValue::new("a", "b")]);
2289        test_context.flush_metrics();
2290        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
2291            unreachable!()
2292        };
2293
2294        let no_attr_data_point = sum.data_points.iter().find(|x| x.attributes.is_empty());
2295
2296        assert!(
2297            no_attr_data_point.is_none(),
2298            "Expected no data points with no attributes"
2299        );
2300    }
2301
2302    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
2303    async fn delta_memory_efficiency_test() {
2304        // Run this test with stdout enabled to see output.
2305        // cargo test delta_memory_efficiency_test --features=testing -- --nocapture
2306
2307        // Arrange
2308        let mut test_context = TestContext::new(Temporality::Delta);
2309        let counter = test_context.u64_counter("test", "my_counter", None);
2310
2311        // Act
2312        counter.add(1, &[KeyValue::new("key1", "value1")]);
2313        counter.add(1, &[KeyValue::new("key1", "value1")]);
2314        counter.add(1, &[KeyValue::new("key1", "value1")]);
2315        counter.add(1, &[KeyValue::new("key1", "value1")]);
2316        counter.add(1, &[KeyValue::new("key1", "value1")]);
2317
2318        counter.add(1, &[KeyValue::new("key1", "value2")]);
2319        counter.add(1, &[KeyValue::new("key1", "value2")]);
2320        counter.add(1, &[KeyValue::new("key1", "value2")]);
2321        test_context.flush_metrics();
2322
2323        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
2324            unreachable!()
2325        };
2326
2327        // Expecting 2 time-series.
2328        assert_eq!(sum.data_points.len(), 2);
2329
2330        // find and validate key1=value1 datapoint
2331        let data_point1 = find_sum_datapoint_with_key_value(&sum.data_points, "key1", "value1")
2332            .expect("datapoint with key1=value1 expected");
2333        assert_eq!(data_point1.value, 5);
2334
2335        // find and validate key1=value2 datapoint
2336        let data_point1 = find_sum_datapoint_with_key_value(&sum.data_points, "key1", "value2")
2337            .expect("datapoint with key1=value2 expected");
2338        assert_eq!(data_point1.value, 3);
2339
2340        test_context.exporter.reset();
2341        // flush again, and validate that nothing is flushed
2342        // as delta temporality.
2343        test_context.flush_metrics();
2344
2345        let resource_metrics = test_context
2346            .exporter
2347            .get_finished_metrics()
2348            .expect("metrics are expected to be exported.");
2349        println!("resource_metrics: {resource_metrics:?}");
2350        assert!(resource_metrics.is_empty(), "No metrics should be exported as no new measurements were recorded since last collect.");
2351    }
2352
2353    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
2354    async fn counter_multithreaded() {
2355        // Run this test with stdout enabled to see output.
2356        // cargo test counter_multithreaded --features=testing -- --nocapture
2357
2358        counter_multithreaded_aggregation_helper(Temporality::Delta);
2359        counter_multithreaded_aggregation_helper(Temporality::Cumulative);
2360    }
2361
2362    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
2363    async fn counter_f64_multithreaded() {
2364        // Run this test with stdout enabled to see output.
2365        // cargo test counter_f64_multithreaded --features=testing -- --nocapture
2366
2367        counter_f64_multithreaded_aggregation_helper(Temporality::Delta);
2368        counter_f64_multithreaded_aggregation_helper(Temporality::Cumulative);
2369    }
2370
2371    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
2372    async fn histogram_multithreaded() {
2373        // Run this test with stdout enabled to see output.
2374        // cargo test histogram_multithreaded --features=testing -- --nocapture
2375
2376        histogram_multithreaded_aggregation_helper(Temporality::Delta);
2377        histogram_multithreaded_aggregation_helper(Temporality::Cumulative);
2378    }
2379
2380    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
2381    async fn histogram_f64_multithreaded() {
2382        // Run this test with stdout enabled to see output.
2383        // cargo test histogram_f64_multithreaded --features=testing -- --nocapture
2384
2385        histogram_f64_multithreaded_aggregation_helper(Temporality::Delta);
2386        histogram_f64_multithreaded_aggregation_helper(Temporality::Cumulative);
2387    }
2388    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
2389    async fn synchronous_instruments_cumulative_with_gap_in_measurements() {
2390        // Run this test with stdout enabled to see output.
2391        // cargo test synchronous_instruments_cumulative_with_gap_in_measurements --features=testing -- --nocapture
2392
2393        synchronous_instruments_cumulative_with_gap_in_measurements_helper("counter");
2394        synchronous_instruments_cumulative_with_gap_in_measurements_helper("updown_counter");
2395        synchronous_instruments_cumulative_with_gap_in_measurements_helper("histogram");
2396        synchronous_instruments_cumulative_with_gap_in_measurements_helper("gauge");
2397    }
2398
2399    fn synchronous_instruments_cumulative_with_gap_in_measurements_helper(
2400        instrument_name: &'static str,
2401    ) {
2402        let mut test_context = TestContext::new(Temporality::Cumulative);
2403        let attributes = &[KeyValue::new("key1", "value1")];
2404
2405        // Create instrument and emit measurements
2406        match instrument_name {
2407            "counter" => {
2408                let counter = test_context.meter().u64_counter("test_counter").build();
2409                counter.add(5, &[]);
2410                counter.add(10, attributes);
2411            }
2412            "updown_counter" => {
2413                let updown_counter = test_context
2414                    .meter()
2415                    .i64_up_down_counter("test_updowncounter")
2416                    .build();
2417                updown_counter.add(15, &[]);
2418                updown_counter.add(20, attributes);
2419            }
2420            "histogram" => {
2421                let histogram = test_context.meter().u64_histogram("test_histogram").build();
2422                histogram.record(25, &[]);
2423                histogram.record(30, attributes);
2424            }
2425            "gauge" => {
2426                let gauge = test_context.meter().u64_gauge("test_gauge").build();
2427                gauge.record(35, &[]);
2428                gauge.record(40, attributes);
2429            }
2430            _ => panic!("Incorrect instrument kind provided"),
2431        };
2432
2433        test_context.flush_metrics();
2434
2435        // Test the first export
2436        assert_correct_export(&mut test_context, instrument_name);
2437
2438        // Reset and export again without making any measurements
2439        test_context.reset_metrics();
2440
2441        test_context.flush_metrics();
2442
2443        // Test that latest export has the same data as the previous one
2444        assert_correct_export(&mut test_context, instrument_name);
2445
2446        fn assert_correct_export(test_context: &mut TestContext, instrument_name: &'static str) {
2447            match instrument_name {
2448                "counter" => {
2449                    let MetricData::Sum(sum) =
2450                        test_context.get_aggregation::<u64>("test_counter", None)
2451                    else {
2452                        unreachable!()
2453                    };
2454                    assert_eq!(sum.data_points.len(), 2);
2455                    let zero_attribute_datapoint =
2456                        find_sum_datapoint_with_no_attributes(&sum.data_points)
2457                            .expect("datapoint with no attributes expected");
2458                    assert_eq!(zero_attribute_datapoint.value, 5);
2459                    let data_point1 =
2460                        find_sum_datapoint_with_key_value(&sum.data_points, "key1", "value1")
2461                            .expect("datapoint with key1=value1 expected");
2462                    assert_eq!(data_point1.value, 10);
2463                }
2464                "updown_counter" => {
2465                    let MetricData::Sum(sum) =
2466                        test_context.get_aggregation::<i64>("test_updowncounter", None)
2467                    else {
2468                        unreachable!()
2469                    };
2470                    assert_eq!(sum.data_points.len(), 2);
2471                    let zero_attribute_datapoint =
2472                        find_sum_datapoint_with_no_attributes(&sum.data_points)
2473                            .expect("datapoint with no attributes expected");
2474                    assert_eq!(zero_attribute_datapoint.value, 15);
2475                    let data_point1 =
2476                        find_sum_datapoint_with_key_value(&sum.data_points, "key1", "value1")
2477                            .expect("datapoint with key1=value1 expected");
2478                    assert_eq!(data_point1.value, 20);
2479                }
2480                "histogram" => {
2481                    let MetricData::Histogram(histogram_data) =
2482                        test_context.get_aggregation::<u64>("test_histogram", None)
2483                    else {
2484                        unreachable!()
2485                    };
2486                    assert_eq!(histogram_data.data_points.len(), 2);
2487                    let zero_attribute_datapoint =
2488                        find_histogram_datapoint_with_no_attributes(&histogram_data.data_points)
2489                            .expect("datapoint with no attributes expected");
2490                    assert_eq!(zero_attribute_datapoint.count, 1);
2491                    assert_eq!(zero_attribute_datapoint.sum, 25);
2492                    assert_eq!(zero_attribute_datapoint.min, Some(25));
2493                    assert_eq!(zero_attribute_datapoint.max, Some(25));
2494                    let data_point1 = find_histogram_datapoint_with_key_value(
2495                        &histogram_data.data_points,
2496                        "key1",
2497                        "value1",
2498                    )
2499                    .expect("datapoint with key1=value1 expected");
2500                    assert_eq!(data_point1.count, 1);
2501                    assert_eq!(data_point1.sum, 30);
2502                    assert_eq!(data_point1.min, Some(30));
2503                    assert_eq!(data_point1.max, Some(30));
2504                }
2505                "gauge" => {
2506                    let MetricData::Gauge(gauge_data) =
2507                        test_context.get_aggregation::<u64>("test_gauge", None)
2508                    else {
2509                        unreachable!()
2510                    };
2511                    assert_eq!(gauge_data.data_points.len(), 2);
2512                    let zero_attribute_datapoint =
2513                        find_gauge_datapoint_with_no_attributes(&gauge_data.data_points)
2514                            .expect("datapoint with no attributes expected");
2515                    assert_eq!(zero_attribute_datapoint.value, 35);
2516                    let data_point1 = find_gauge_datapoint_with_key_value(
2517                        &gauge_data.data_points,
2518                        "key1",
2519                        "value1",
2520                    )
2521                    .expect("datapoint with key1=value1 expected");
2522                    assert_eq!(data_point1.value, 40);
2523                }
2524                _ => panic!("Incorrect instrument kind provided"),
2525            }
2526        }
2527    }
2528
2529    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
2530    async fn asynchronous_instruments_cumulative_data_points_only_from_last_measurement() {
2531        // Run this test with stdout enabled to see output.
2532        // cargo test asynchronous_instruments_cumulative_data_points_only_from_last_measurement --features=testing -- --nocapture
2533
2534        asynchronous_instruments_cumulative_data_points_only_from_last_measurement_helper("gauge");
2535        asynchronous_instruments_cumulative_data_points_only_from_last_measurement_helper(
2536            "counter",
2537        );
2538        asynchronous_instruments_cumulative_data_points_only_from_last_measurement_helper(
2539            "updown_counter",
2540        );
2541    }
2542
2543    #[test]
2544    fn view_test_rename() {
2545        test_view_customization(
2546            |i| {
2547                if i.name == "my_counter" {
2548                    Some(
2549                        Stream::builder()
2550                            .with_name("my_counter_renamed")
2551                            .build()
2552                            .unwrap(),
2553                    )
2554                } else {
2555                    None
2556                }
2557            },
2558            "my_counter_renamed",
2559            "my_unit",
2560            "my_description",
2561        )
2562    }
2563
2564    #[test]
2565    fn view_test_change_unit() {
2566        test_view_customization(
2567            |i| {
2568                if i.name == "my_counter" {
2569                    Some(Stream::builder().with_unit("my_unit_new").build().unwrap())
2570                } else {
2571                    None
2572                }
2573            },
2574            "my_counter",
2575            "my_unit_new",
2576            "my_description",
2577        )
2578    }
2579
2580    #[test]
2581    fn view_test_change_description() {
2582        test_view_customization(
2583            |i| {
2584                if i.name == "my_counter" {
2585                    Some(
2586                        Stream::builder()
2587                            .with_description("my_description_new")
2588                            .build()
2589                            .unwrap(),
2590                    )
2591                } else {
2592                    None
2593                }
2594            },
2595            "my_counter",
2596            "my_unit",
2597            "my_description_new",
2598        )
2599    }
2600
2601    #[test]
2602    fn view_test_change_name_unit() {
2603        test_view_customization(
2604            |i| {
2605                if i.name == "my_counter" {
2606                    Some(
2607                        Stream::builder()
2608                            .with_name("my_counter_renamed")
2609                            .with_unit("my_unit_new")
2610                            .build()
2611                            .unwrap(),
2612                    )
2613                } else {
2614                    None
2615                }
2616            },
2617            "my_counter_renamed",
2618            "my_unit_new",
2619            "my_description",
2620        )
2621    }
2622
2623    #[test]
2624    fn view_test_change_name_unit_desc() {
2625        test_view_customization(
2626            |i| {
2627                if i.name == "my_counter" {
2628                    Some(
2629                        Stream::builder()
2630                            .with_name("my_counter_renamed")
2631                            .with_unit("my_unit_new")
2632                            .with_description("my_description_new")
2633                            .build()
2634                            .unwrap(),
2635                    )
2636                } else {
2637                    None
2638                }
2639            },
2640            "my_counter_renamed",
2641            "my_unit_new",
2642            "my_description_new",
2643        )
2644    }
2645
2646    #[test]
2647    fn view_test_match_unit() {
2648        test_view_customization(
2649            |i| {
2650                if i.unit == "my_unit" {
2651                    Some(Stream::builder().with_unit("my_unit_new").build().unwrap())
2652                } else {
2653                    None
2654                }
2655            },
2656            "my_counter",
2657            "my_unit_new",
2658            "my_description",
2659        )
2660    }
2661
2662    #[test]
2663    fn view_test_match_none() {
2664        test_view_customization(
2665            |i| {
2666                if i.name == "not_expected_to_match" {
2667                    Some(Stream::builder().build().unwrap())
2668                } else {
2669                    None
2670                }
2671            },
2672            "my_counter",
2673            "my_unit",
2674            "my_description",
2675        )
2676    }
2677
2678    #[test]
2679    fn view_test_match_multiple() {
2680        test_view_customization(
2681            |i| {
2682                if i.name == "my_counter" && i.unit == "my_unit" {
2683                    Some(
2684                        Stream::builder()
2685                            .with_name("my_counter_renamed")
2686                            .build()
2687                            .unwrap(),
2688                    )
2689                } else {
2690                    None
2691                }
2692            },
2693            "my_counter_renamed",
2694            "my_unit",
2695            "my_description",
2696        )
2697    }
2698
2699    /// Helper function to test view customizations
2700    fn test_view_customization<F>(
2701        view_function: F,
2702        expected_name: &str,
2703        expected_unit: &str,
2704        expected_description: &str,
2705    ) where
2706        F: Fn(&Instrument) -> Option<Stream> + Send + Sync + 'static,
2707    {
2708        // Run this test with stdout enabled to see output.
2709        // cargo test view_test_* --all-features -- --nocapture
2710
2711        // Arrange
2712        let exporter = InMemoryMetricExporter::default();
2713        let meter_provider = SdkMeterProvider::builder()
2714            .with_periodic_exporter(exporter.clone())
2715            .with_view(view_function)
2716            .build();
2717
2718        // Act
2719        let meter = meter_provider.meter("test");
2720        let counter = meter
2721            .f64_counter("my_counter")
2722            .with_unit("my_unit")
2723            .with_description("my_description")
2724            .build();
2725
2726        counter.add(1.5, &[KeyValue::new("key1", "value1")]);
2727        meter_provider.force_flush().unwrap();
2728
2729        // Assert
2730        let resource_metrics = exporter
2731            .get_finished_metrics()
2732            .expect("metrics are expected to be exported.");
2733        assert_eq!(resource_metrics.len(), 1);
2734        assert_eq!(resource_metrics[0].scope_metrics.len(), 1);
2735        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
2736        assert_eq!(
2737            metric.name, expected_name,
2738            "Expected name: {expected_name}."
2739        );
2740        assert_eq!(
2741            metric.unit, expected_unit,
2742            "Expected unit: {expected_unit}."
2743        );
2744        assert_eq!(
2745            metric.description, expected_description,
2746            "Expected description: {expected_description}."
2747        );
2748    }
2749
2750    // Following are just a basic set of advanced View tests - Views bring a lot
2751    // of permutations and combinations, and we need
2752    // to expand coverage for more scenarios in future.
2753    // It is best to first split this file into multiple files
2754    // based on scenarios (eg: regular aggregation, cardinality, views, view_advanced, etc)
2755    // and then add more tests for each of the scenarios.
2756    #[test]
2757    fn test_view_single_instrument_multiple_stream() {
2758        // Run this test with stdout enabled to see output.
2759        // cargo test test_view_multiple_stream --all-features
2760
2761        // Each of the views match the instrument name "my_counter" and create a
2762        // new stream with a different name. In other words, View can be used to
2763        // create multiple streams for the same instrument.
2764
2765        let view1 = |i: &Instrument| {
2766            if i.name() == "my_counter" {
2767                Some(Stream::builder().with_name("my_counter_1").build().unwrap())
2768            } else {
2769                None
2770            }
2771        };
2772
2773        let view2 = |i: &Instrument| {
2774            if i.name() == "my_counter" {
2775                Some(Stream::builder().with_name("my_counter_2").build().unwrap())
2776            } else {
2777                None
2778            }
2779        };
2780
2781        // Arrange
2782        let exporter = InMemoryMetricExporter::default();
2783        let meter_provider = SdkMeterProvider::builder()
2784            .with_periodic_exporter(exporter.clone())
2785            .with_view(view1)
2786            .with_view(view2)
2787            .build();
2788
2789        // Act
2790        let meter = meter_provider.meter("test");
2791        let counter = meter.f64_counter("my_counter").build();
2792
2793        counter.add(1.5, &[KeyValue::new("key1", "value1")]);
2794        meter_provider.force_flush().unwrap();
2795
2796        // Assert
2797        let resource_metrics = exporter
2798            .get_finished_metrics()
2799            .expect("metrics are expected to be exported.");
2800        assert_eq!(resource_metrics.len(), 1);
2801        assert_eq!(resource_metrics[0].scope_metrics.len(), 1);
2802        let metrics = &resource_metrics[0].scope_metrics[0].metrics;
2803        assert_eq!(metrics.len(), 2);
2804        assert_eq!(metrics[0].name, "my_counter_1");
2805        assert_eq!(metrics[1].name, "my_counter_2");
2806    }
2807
2808    #[test]
2809    fn test_view_multiple_instrument_single_stream() {
2810        // Run this test with stdout enabled to see output.
2811        // cargo test test_view_multiple_instrument_single_stream --all-features
2812
2813        // The view matches the instrument name "my_counter1" and "my_counter1"
2814        // and create a single new stream for both. In other words, View can be used to
2815        // "merge" multiple instruments into a single stream.
2816        let view = |i: &Instrument| {
2817            if i.name() == "my_counter1" || i.name() == "my_counter2" {
2818                Some(Stream::builder().with_name("my_counter").build().unwrap())
2819            } else {
2820                None
2821            }
2822        };
2823
2824        // Arrange
2825        let exporter = InMemoryMetricExporter::default();
2826        let meter_provider = SdkMeterProvider::builder()
2827            .with_periodic_exporter(exporter.clone())
2828            .with_view(view)
2829            .build();
2830
2831        // Act
2832        let meter = meter_provider.meter("test");
2833        let counter1 = meter.f64_counter("my_counter1").build();
2834        let counter2 = meter.f64_counter("my_counter2").build();
2835
2836        counter1.add(1.5, &[KeyValue::new("key1", "value1")]);
2837        counter2.add(1.5, &[KeyValue::new("key1", "value1")]);
2838        meter_provider.force_flush().unwrap();
2839
2840        // Assert
2841        let resource_metrics = exporter
2842            .get_finished_metrics()
2843            .expect("metrics are expected to be exported.");
2844        assert_eq!(resource_metrics.len(), 1);
2845        assert_eq!(resource_metrics[0].scope_metrics.len(), 1);
2846        let metrics = &resource_metrics[0].scope_metrics[0].metrics;
2847        assert_eq!(metrics.len(), 1);
2848        assert_eq!(metrics[0].name, "my_counter");
2849        // TODO: Assert that the data points are aggregated correctly.
2850    }
2851
2852    fn asynchronous_instruments_cumulative_data_points_only_from_last_measurement_helper(
2853        instrument_name: &'static str,
2854    ) {
2855        let mut test_context = TestContext::new(Temporality::Cumulative);
2856        let attributes = Arc::new([KeyValue::new("key1", "value1")]);
2857
2858        // Create instrument and emit measurements once
2859        match instrument_name {
2860            "counter" => {
2861                let has_run = AtomicBool::new(false);
2862                let _observable_counter = test_context
2863                    .meter()
2864                    .u64_observable_counter("test_counter")
2865                    .with_callback(move |observer| {
2866                        if !has_run.load(Ordering::SeqCst) {
2867                            observer.observe(5, &[]);
2868                            observer.observe(10, &*attributes.clone());
2869                            has_run.store(true, Ordering::SeqCst);
2870                        }
2871                    })
2872                    .build();
2873            }
2874            "updown_counter" => {
2875                let has_run = AtomicBool::new(false);
2876                let _observable_up_down_counter = test_context
2877                    .meter()
2878                    .i64_observable_up_down_counter("test_updowncounter")
2879                    .with_callback(move |observer| {
2880                        if !has_run.load(Ordering::SeqCst) {
2881                            observer.observe(15, &[]);
2882                            observer.observe(20, &*attributes.clone());
2883                            has_run.store(true, Ordering::SeqCst);
2884                        }
2885                    })
2886                    .build();
2887            }
2888            "gauge" => {
2889                let has_run = AtomicBool::new(false);
2890                let _observable_gauge = test_context
2891                    .meter()
2892                    .u64_observable_gauge("test_gauge")
2893                    .with_callback(move |observer| {
2894                        if !has_run.load(Ordering::SeqCst) {
2895                            observer.observe(25, &[]);
2896                            observer.observe(30, &*attributes.clone());
2897                            has_run.store(true, Ordering::SeqCst);
2898                        }
2899                    })
2900                    .build();
2901            }
2902            _ => panic!("Incorrect instrument kind provided"),
2903        };
2904
2905        test_context.flush_metrics();
2906
2907        // Test the first export
2908        assert_correct_export(&mut test_context, instrument_name);
2909
2910        // Reset and export again without making any measurements
2911        test_context.reset_metrics();
2912
2913        test_context.flush_metrics();
2914
2915        test_context.check_no_metrics();
2916
2917        fn assert_correct_export(test_context: &mut TestContext, instrument_name: &'static str) {
2918            match instrument_name {
2919                "counter" => {
2920                    let MetricData::Sum(sum) =
2921                        test_context.get_aggregation::<u64>("test_counter", None)
2922                    else {
2923                        unreachable!()
2924                    };
2925                    assert_eq!(sum.data_points.len(), 2);
2926                    assert!(sum.is_monotonic);
2927                    let zero_attribute_datapoint =
2928                        find_sum_datapoint_with_no_attributes(&sum.data_points)
2929                            .expect("datapoint with no attributes expected");
2930                    assert_eq!(zero_attribute_datapoint.value, 5);
2931                    let data_point1 =
2932                        find_sum_datapoint_with_key_value(&sum.data_points, "key1", "value1")
2933                            .expect("datapoint with key1=value1 expected");
2934                    assert_eq!(data_point1.value, 10);
2935                }
2936                "updown_counter" => {
2937                    let MetricData::Sum(sum) =
2938                        test_context.get_aggregation::<i64>("test_updowncounter", None)
2939                    else {
2940                        unreachable!()
2941                    };
2942                    assert_eq!(sum.data_points.len(), 2);
2943                    assert!(!sum.is_monotonic);
2944                    let zero_attribute_datapoint =
2945                        find_sum_datapoint_with_no_attributes(&sum.data_points)
2946                            .expect("datapoint with no attributes expected");
2947                    assert_eq!(zero_attribute_datapoint.value, 15);
2948                    let data_point1 =
2949                        find_sum_datapoint_with_key_value(&sum.data_points, "key1", "value1")
2950                            .expect("datapoint with key1=value1 expected");
2951                    assert_eq!(data_point1.value, 20);
2952                }
2953                "gauge" => {
2954                    let MetricData::Gauge(gauge_data) =
2955                        test_context.get_aggregation::<u64>("test_gauge", None)
2956                    else {
2957                        unreachable!()
2958                    };
2959                    assert_eq!(gauge_data.data_points.len(), 2);
2960                    let zero_attribute_datapoint =
2961                        find_gauge_datapoint_with_no_attributes(&gauge_data.data_points)
2962                            .expect("datapoint with no attributes expected");
2963                    assert_eq!(zero_attribute_datapoint.value, 25);
2964                    let data_point1 = find_gauge_datapoint_with_key_value(
2965                        &gauge_data.data_points,
2966                        "key1",
2967                        "value1",
2968                    )
2969                    .expect("datapoint with key1=value1 expected");
2970                    assert_eq!(data_point1.value, 30);
2971                }
2972                _ => panic!("Incorrect instrument kind provided"),
2973            }
2974        }
2975    }
2976
2977    fn counter_multithreaded_aggregation_helper(temporality: Temporality) {
2978        // Arrange
2979        let mut test_context = TestContext::new(temporality);
2980        let counter = Arc::new(test_context.u64_counter("test", "my_counter", None));
2981
2982        for i in 0..10 {
2983            thread::scope(|s| {
2984                s.spawn(|| {
2985                    counter.add(1, &[]);
2986
2987                    counter.add(1, &[KeyValue::new("key1", "value1")]);
2988                    counter.add(1, &[KeyValue::new("key1", "value1")]);
2989                    counter.add(1, &[KeyValue::new("key1", "value1")]);
2990
2991                    // Test concurrent collection by forcing half of the update threads to `force_flush` metrics and sleep for some time.
2992                    if i % 2 == 0 {
2993                        test_context.flush_metrics();
2994                        thread::sleep(Duration::from_millis(i)); // Make each thread sleep for some time duration for better testing
2995                    }
2996
2997                    counter.add(1, &[KeyValue::new("key1", "value1")]);
2998                    counter.add(1, &[KeyValue::new("key1", "value1")]);
2999                });
3000            });
3001        }
3002
3003        test_context.flush_metrics();
3004
3005        // Assert
3006        // We invoke `test_context.flush_metrics()` six times.
3007        let sums = test_context
3008            .get_from_multiple_aggregations::<u64>("my_counter", None, 6)
3009            .into_iter()
3010            .map(|data| {
3011                if let MetricData::Sum(sum) = data {
3012                    sum
3013                } else {
3014                    unreachable!()
3015                }
3016            })
3017            .collect::<Vec<_>>();
3018
3019        let mut sum_zero_attributes = 0;
3020        let mut sum_key1_value1 = 0;
3021        sums.iter().for_each(|sum| {
3022            assert_eq!(sum.data_points.len(), 2); // Expecting 1 time-series.
3023            assert!(sum.is_monotonic, "Counter should produce monotonic.");
3024            assert_eq!(sum.temporality, temporality);
3025
3026            if temporality == Temporality::Delta {
3027                sum_zero_attributes += sum.data_points[0].value;
3028                sum_key1_value1 += sum.data_points[1].value;
3029            } else {
3030                sum_zero_attributes = sum.data_points[0].value;
3031                sum_key1_value1 = sum.data_points[1].value;
3032            };
3033        });
3034
3035        assert_eq!(sum_zero_attributes, 10);
3036        assert_eq!(sum_key1_value1, 50); // Each of the 10 update threads record measurements summing up to 5.
3037    }
3038
3039    fn counter_f64_multithreaded_aggregation_helper(temporality: Temporality) {
3040        // Arrange
3041        let mut test_context = TestContext::new(temporality);
3042        let counter = Arc::new(test_context.meter().f64_counter("test_counter").build());
3043
3044        for i in 0..10 {
3045            thread::scope(|s| {
3046                s.spawn(|| {
3047                    counter.add(1.23, &[]);
3048
3049                    counter.add(1.23, &[KeyValue::new("key1", "value1")]);
3050                    counter.add(1.23, &[KeyValue::new("key1", "value1")]);
3051                    counter.add(1.23, &[KeyValue::new("key1", "value1")]);
3052
3053                    // Test concurrent collection by forcing half of the update threads to `force_flush` metrics and sleep for some time.
3054                    if i % 2 == 0 {
3055                        test_context.flush_metrics();
3056                        thread::sleep(Duration::from_millis(i)); // Make each thread sleep for some time duration for better testing
3057                    }
3058
3059                    counter.add(1.23, &[KeyValue::new("key1", "value1")]);
3060                    counter.add(1.23, &[KeyValue::new("key1", "value1")]);
3061                });
3062            });
3063        }
3064
3065        test_context.flush_metrics();
3066
3067        // Assert
3068        // We invoke `test_context.flush_metrics()` six times.
3069        let sums = test_context
3070            .get_from_multiple_aggregations::<f64>("test_counter", None, 6)
3071            .into_iter()
3072            .map(|data| {
3073                if let MetricData::Sum(sum) = data {
3074                    sum
3075                } else {
3076                    unreachable!()
3077                }
3078            })
3079            .collect::<Vec<_>>();
3080
3081        let mut sum_zero_attributes = 0.0;
3082        let mut sum_key1_value1 = 0.0;
3083        sums.iter().for_each(|sum| {
3084            assert_eq!(sum.data_points.len(), 2); // Expecting 1 time-series.
3085            assert!(sum.is_monotonic, "Counter should produce monotonic.");
3086            assert_eq!(sum.temporality, temporality);
3087
3088            if temporality == Temporality::Delta {
3089                sum_zero_attributes += sum.data_points[0].value;
3090                sum_key1_value1 += sum.data_points[1].value;
3091            } else {
3092                sum_zero_attributes = sum.data_points[0].value;
3093                sum_key1_value1 = sum.data_points[1].value;
3094            };
3095        });
3096
3097        assert!(f64::abs(12.3 - sum_zero_attributes) < 0.0001);
3098        assert!(f64::abs(61.5 - sum_key1_value1) < 0.0001); // Each of the 10 update threads record measurements 5 times = 10 * 5 * 1.23 = 61.5
3099    }
3100
3101    fn histogram_multithreaded_aggregation_helper(temporality: Temporality) {
3102        // Arrange
3103        let mut test_context = TestContext::new(temporality);
3104        let histogram = Arc::new(test_context.meter().u64_histogram("test_histogram").build());
3105
3106        for i in 0..10 {
3107            thread::scope(|s| {
3108                s.spawn(|| {
3109                    histogram.record(1, &[]);
3110                    histogram.record(4, &[]);
3111
3112                    histogram.record(5, &[KeyValue::new("key1", "value1")]);
3113                    histogram.record(7, &[KeyValue::new("key1", "value1")]);
3114                    histogram.record(18, &[KeyValue::new("key1", "value1")]);
3115
3116                    // Test concurrent collection by forcing half of the update threads to `force_flush` metrics and sleep for some time.
3117                    if i % 2 == 0 {
3118                        test_context.flush_metrics();
3119                        thread::sleep(Duration::from_millis(i)); // Make each thread sleep for some time duration for better testing
3120                    }
3121
3122                    histogram.record(35, &[KeyValue::new("key1", "value1")]);
3123                    histogram.record(35, &[KeyValue::new("key1", "value1")]);
3124                });
3125            });
3126        }
3127
3128        test_context.flush_metrics();
3129
3130        // Assert
3131        // We invoke `test_context.flush_metrics()` six times.
3132        let histograms = test_context
3133            .get_from_multiple_aggregations::<u64>("test_histogram", None, 6)
3134            .into_iter()
3135            .map(|data| {
3136                if let MetricData::Histogram(hist) = data {
3137                    hist
3138                } else {
3139                    unreachable!()
3140                }
3141            })
3142            .collect::<Vec<_>>();
3143
3144        let (
3145            mut sum_zero_attributes,
3146            mut count_zero_attributes,
3147            mut min_zero_attributes,
3148            mut max_zero_attributes,
3149        ) = (0, 0, u64::MAX, u64::MIN);
3150        let (mut sum_key1_value1, mut count_key1_value1, mut min_key1_value1, mut max_key1_value1) =
3151            (0, 0, u64::MAX, u64::MIN);
3152
3153        let mut bucket_counts_zero_attributes = vec![0; 16]; // There are 16 buckets for the default configuration
3154        let mut bucket_counts_key1_value1 = vec![0; 16];
3155
3156        histograms.iter().for_each(|histogram| {
3157            assert_eq!(histogram.data_points.len(), 2); // Expecting 1 time-series.
3158            assert_eq!(histogram.temporality, temporality);
3159
3160            let data_point_zero_attributes =
3161                find_histogram_datapoint_with_no_attributes(&histogram.data_points).unwrap();
3162            let data_point_key1_value1 =
3163                find_histogram_datapoint_with_key_value(&histogram.data_points, "key1", "value1")
3164                    .unwrap();
3165
3166            if temporality == Temporality::Delta {
3167                sum_zero_attributes += data_point_zero_attributes.sum;
3168                sum_key1_value1 += data_point_key1_value1.sum;
3169
3170                count_zero_attributes += data_point_zero_attributes.count;
3171                count_key1_value1 += data_point_key1_value1.count;
3172
3173                min_zero_attributes =
3174                    min(min_zero_attributes, data_point_zero_attributes.min.unwrap());
3175                min_key1_value1 = min(min_key1_value1, data_point_key1_value1.min.unwrap());
3176
3177                max_zero_attributes =
3178                    max(max_zero_attributes, data_point_zero_attributes.max.unwrap());
3179                max_key1_value1 = max(max_key1_value1, data_point_key1_value1.max.unwrap());
3180
3181                assert_eq!(data_point_zero_attributes.bucket_counts.len(), 16);
3182                assert_eq!(data_point_key1_value1.bucket_counts.len(), 16);
3183
3184                for (i, _) in data_point_zero_attributes.bucket_counts.iter().enumerate() {
3185                    bucket_counts_zero_attributes[i] += data_point_zero_attributes.bucket_counts[i];
3186                }
3187
3188                for (i, _) in data_point_key1_value1.bucket_counts.iter().enumerate() {
3189                    bucket_counts_key1_value1[i] += data_point_key1_value1.bucket_counts[i];
3190                }
3191            } else {
3192                sum_zero_attributes = data_point_zero_attributes.sum;
3193                sum_key1_value1 = data_point_key1_value1.sum;
3194
3195                count_zero_attributes = data_point_zero_attributes.count;
3196                count_key1_value1 = data_point_key1_value1.count;
3197
3198                min_zero_attributes = data_point_zero_attributes.min.unwrap();
3199                min_key1_value1 = data_point_key1_value1.min.unwrap();
3200
3201                max_zero_attributes = data_point_zero_attributes.max.unwrap();
3202                max_key1_value1 = data_point_key1_value1.max.unwrap();
3203
3204                assert_eq!(data_point_zero_attributes.bucket_counts.len(), 16);
3205                assert_eq!(data_point_key1_value1.bucket_counts.len(), 16);
3206
3207                bucket_counts_zero_attributes.clone_from(&data_point_zero_attributes.bucket_counts);
3208                bucket_counts_key1_value1.clone_from(&data_point_key1_value1.bucket_counts);
3209            };
3210        });
3211
3212        // Default buckets:
3213        // (-∞, 0], (0, 5.0], (5.0, 10.0], (10.0, 25.0], (25.0, 50.0], (50.0, 75.0], (75.0, 100.0], (100.0, 250.0], (250.0, 500.0],
3214        // (500.0, 750.0], (750.0, 1000.0], (1000.0, 2500.0], (2500.0, 5000.0], (5000.0, 7500.0], (7500.0, 10000.0], (10000.0, +∞).
3215
3216        assert_eq!(count_zero_attributes, 20); // Each of the 10 update threads record two measurements.
3217        assert_eq!(sum_zero_attributes, 50); // Each of the 10 update threads record measurements summing up to 5.
3218        assert_eq!(min_zero_attributes, 1);
3219        assert_eq!(max_zero_attributes, 4);
3220
3221        for (i, count) in bucket_counts_zero_attributes.iter().enumerate() {
3222            match i {
3223                1 => assert_eq!(*count, 20), // For each of the 10 update threads, both the recorded values 1 and 4 fall under the bucket (0, 5].
3224                _ => assert_eq!(*count, 0),
3225            }
3226        }
3227
3228        assert_eq!(count_key1_value1, 50); // Each of the 10 update threads record 5 measurements.
3229        assert_eq!(sum_key1_value1, 1000); // Each of the 10 update threads record measurements summing up to 100 (5 + 7 + 18 + 35 + 35).
3230        assert_eq!(min_key1_value1, 5);
3231        assert_eq!(max_key1_value1, 35);
3232
3233        for (i, count) in bucket_counts_key1_value1.iter().enumerate() {
3234            match i {
3235                1 => assert_eq!(*count, 10), // For each of the 10 update threads, the recorded value 5 falls under the bucket (0, 5].
3236                2 => assert_eq!(*count, 10), // For each of the 10 update threads, the recorded value 7 falls under the bucket (5, 10].
3237                3 => assert_eq!(*count, 10), // For each of the 10 update threads, the recorded value 18 falls under the bucket (10, 25].
3238                4 => assert_eq!(*count, 20), // For each of the 10 update threads, the recorded value 35 (recorded twice) falls under the bucket (25, 50].
3239                _ => assert_eq!(*count, 0),
3240            }
3241        }
3242    }
3243
3244    fn histogram_f64_multithreaded_aggregation_helper(temporality: Temporality) {
3245        // Arrange
3246        let mut test_context = TestContext::new(temporality);
3247        let histogram = Arc::new(test_context.meter().f64_histogram("test_histogram").build());
3248
3249        for i in 0..10 {
3250            thread::scope(|s| {
3251                s.spawn(|| {
3252                    histogram.record(1.5, &[]);
3253                    histogram.record(4.6, &[]);
3254
3255                    histogram.record(5.0, &[KeyValue::new("key1", "value1")]);
3256                    histogram.record(7.3, &[KeyValue::new("key1", "value1")]);
3257                    histogram.record(18.1, &[KeyValue::new("key1", "value1")]);
3258
3259                    // Test concurrent collection by forcing half of the update threads to `force_flush` metrics and sleep for some time.
3260                    if i % 2 == 0 {
3261                        test_context.flush_metrics();
3262                        thread::sleep(Duration::from_millis(i)); // Make each thread sleep for some time duration for better testing
3263                    }
3264
3265                    histogram.record(35.1, &[KeyValue::new("key1", "value1")]);
3266                    histogram.record(35.1, &[KeyValue::new("key1", "value1")]);
3267                });
3268            });
3269        }
3270
3271        test_context.flush_metrics();
3272
3273        // Assert
3274        // We invoke `test_context.flush_metrics()` six times.
3275        let histograms = test_context
3276            .get_from_multiple_aggregations::<f64>("test_histogram", None, 6)
3277            .into_iter()
3278            .map(|data| {
3279                if let MetricData::Histogram(hist) = data {
3280                    hist
3281                } else {
3282                    unreachable!()
3283                }
3284            })
3285            .collect::<Vec<_>>();
3286
3287        let (
3288            mut sum_zero_attributes,
3289            mut count_zero_attributes,
3290            mut min_zero_attributes,
3291            mut max_zero_attributes,
3292        ) = (0.0, 0, f64::MAX, f64::MIN);
3293        let (mut sum_key1_value1, mut count_key1_value1, mut min_key1_value1, mut max_key1_value1) =
3294            (0.0, 0, f64::MAX, f64::MIN);
3295
3296        let mut bucket_counts_zero_attributes = vec![0; 16]; // There are 16 buckets for the default configuration
3297        let mut bucket_counts_key1_value1 = vec![0; 16];
3298
3299        histograms.iter().for_each(|histogram| {
3300            assert_eq!(histogram.data_points.len(), 2); // Expecting 1 time-series.
3301            assert_eq!(histogram.temporality, temporality);
3302
3303            let data_point_zero_attributes =
3304                find_histogram_datapoint_with_no_attributes(&histogram.data_points).unwrap();
3305            let data_point_key1_value1 =
3306                find_histogram_datapoint_with_key_value(&histogram.data_points, "key1", "value1")
3307                    .unwrap();
3308
3309            if temporality == Temporality::Delta {
3310                sum_zero_attributes += data_point_zero_attributes.sum;
3311                sum_key1_value1 += data_point_key1_value1.sum;
3312
3313                count_zero_attributes += data_point_zero_attributes.count;
3314                count_key1_value1 += data_point_key1_value1.count;
3315
3316                min_zero_attributes =
3317                    min_zero_attributes.min(data_point_zero_attributes.min.unwrap());
3318                min_key1_value1 = min_key1_value1.min(data_point_key1_value1.min.unwrap());
3319
3320                max_zero_attributes =
3321                    max_zero_attributes.max(data_point_zero_attributes.max.unwrap());
3322                max_key1_value1 = max_key1_value1.max(data_point_key1_value1.max.unwrap());
3323
3324                assert_eq!(data_point_zero_attributes.bucket_counts.len(), 16);
3325                assert_eq!(data_point_key1_value1.bucket_counts.len(), 16);
3326
3327                for (i, _) in data_point_zero_attributes.bucket_counts.iter().enumerate() {
3328                    bucket_counts_zero_attributes[i] += data_point_zero_attributes.bucket_counts[i];
3329                }
3330
3331                for (i, _) in data_point_key1_value1.bucket_counts.iter().enumerate() {
3332                    bucket_counts_key1_value1[i] += data_point_key1_value1.bucket_counts[i];
3333                }
3334            } else {
3335                sum_zero_attributes = data_point_zero_attributes.sum;
3336                sum_key1_value1 = data_point_key1_value1.sum;
3337
3338                count_zero_attributes = data_point_zero_attributes.count;
3339                count_key1_value1 = data_point_key1_value1.count;
3340
3341                min_zero_attributes = data_point_zero_attributes.min.unwrap();
3342                min_key1_value1 = data_point_key1_value1.min.unwrap();
3343
3344                max_zero_attributes = data_point_zero_attributes.max.unwrap();
3345                max_key1_value1 = data_point_key1_value1.max.unwrap();
3346
3347                assert_eq!(data_point_zero_attributes.bucket_counts.len(), 16);
3348                assert_eq!(data_point_key1_value1.bucket_counts.len(), 16);
3349
3350                bucket_counts_zero_attributes.clone_from(&data_point_zero_attributes.bucket_counts);
3351                bucket_counts_key1_value1.clone_from(&data_point_key1_value1.bucket_counts);
3352            };
3353        });
3354
3355        // Default buckets:
3356        // (-∞, 0], (0, 5.0], (5.0, 10.0], (10.0, 25.0], (25.0, 50.0], (50.0, 75.0], (75.0, 100.0], (100.0, 250.0], (250.0, 500.0],
3357        // (500.0, 750.0], (750.0, 1000.0], (1000.0, 2500.0], (2500.0, 5000.0], (5000.0, 7500.0], (7500.0, 10000.0], (10000.0, +∞).
3358
3359        assert_eq!(count_zero_attributes, 20); // Each of the 10 update threads record two measurements.
3360        assert!(f64::abs(61.0 - sum_zero_attributes) < 0.0001); // Each of the 10 update threads record measurements summing up to 6.1 (1.5 + 4.6)
3361        assert_eq!(min_zero_attributes, 1.5);
3362        assert_eq!(max_zero_attributes, 4.6);
3363
3364        for (i, count) in bucket_counts_zero_attributes.iter().enumerate() {
3365            match i {
3366                1 => assert_eq!(*count, 20), // For each of the 10 update threads, both the recorded values 1.5 and 4.6 fall under the bucket (0, 5.0].
3367                _ => assert_eq!(*count, 0),
3368            }
3369        }
3370
3371        assert_eq!(count_key1_value1, 50); // Each of the 10 update threads record 5 measurements.
3372        assert!(f64::abs(1006.0 - sum_key1_value1) < 0.0001); // Each of the 10 update threads record measurements summing up to 100.4 (5.0 + 7.3 + 18.1 + 35.1 + 35.1).
3373        assert_eq!(min_key1_value1, 5.0);
3374        assert_eq!(max_key1_value1, 35.1);
3375
3376        for (i, count) in bucket_counts_key1_value1.iter().enumerate() {
3377            match i {
3378                1 => assert_eq!(*count, 10), // For each of the 10 update threads, the recorded value 5.0 falls under the bucket (0, 5.0].
3379                2 => assert_eq!(*count, 10), // For each of the 10 update threads, the recorded value 7.3 falls under the bucket (5.0, 10.0].
3380                3 => assert_eq!(*count, 10), // For each of the 10 update threads, the recorded value 18.1 falls under the bucket (10.0, 25.0].
3381                4 => assert_eq!(*count, 20), // For each of the 10 update threads, the recorded value 35.1 (recorded twice) falls under the bucket (25.0, 50.0].
3382                _ => assert_eq!(*count, 0),
3383            }
3384        }
3385    }
3386
3387    fn histogram_aggregation_helper(temporality: Temporality) {
3388        // Arrange
3389        let mut test_context = TestContext::new(temporality);
3390        let histogram = test_context.meter().u64_histogram("my_histogram").build();
3391
3392        // Act
3393        let mut rand = rngs::SmallRng::from_os_rng();
3394        let values_kv1 = (0..50)
3395            .map(|_| rand.random_range(0..100))
3396            .collect::<Vec<u64>>();
3397        for value in values_kv1.iter() {
3398            histogram.record(*value, &[KeyValue::new("key1", "value1")]);
3399        }
3400
3401        let values_kv2 = (0..30)
3402            .map(|_| rand.random_range(0..100))
3403            .collect::<Vec<u64>>();
3404        for value in values_kv2.iter() {
3405            histogram.record(*value, &[KeyValue::new("key1", "value2")]);
3406        }
3407
3408        test_context.flush_metrics();
3409
3410        // Assert
3411        let MetricData::Histogram(histogram_data) =
3412            test_context.get_aggregation::<u64>("my_histogram", None)
3413        else {
3414            unreachable!()
3415        };
3416        // Expecting 2 time-series.
3417        assert_eq!(histogram_data.data_points.len(), 2);
3418        if let Temporality::Cumulative = temporality {
3419            assert_eq!(
3420                histogram_data.temporality,
3421                Temporality::Cumulative,
3422                "Should produce cumulative"
3423            );
3424        } else {
3425            assert_eq!(
3426                histogram_data.temporality,
3427                Temporality::Delta,
3428                "Should produce delta"
3429            );
3430        }
3431
3432        // find and validate key1=value2 datapoint
3433        let data_point1 =
3434            find_histogram_datapoint_with_key_value(&histogram_data.data_points, "key1", "value1")
3435                .expect("datapoint with key1=value1 expected");
3436        assert_eq!(data_point1.count, values_kv1.len() as u64);
3437        assert_eq!(data_point1.sum, values_kv1.iter().sum::<u64>());
3438        assert_eq!(data_point1.min.unwrap(), *values_kv1.iter().min().unwrap());
3439        assert_eq!(data_point1.max.unwrap(), *values_kv1.iter().max().unwrap());
3440
3441        let data_point2 =
3442            find_histogram_datapoint_with_key_value(&histogram_data.data_points, "key1", "value2")
3443                .expect("datapoint with key1=value2 expected");
3444        assert_eq!(data_point2.count, values_kv2.len() as u64);
3445        assert_eq!(data_point2.sum, values_kv2.iter().sum::<u64>());
3446        assert_eq!(data_point2.min.unwrap(), *values_kv2.iter().min().unwrap());
3447        assert_eq!(data_point2.max.unwrap(), *values_kv2.iter().max().unwrap());
3448
3449        // Reset and report more measurements
3450        test_context.reset_metrics();
3451        for value in values_kv1.iter() {
3452            histogram.record(*value, &[KeyValue::new("key1", "value1")]);
3453        }
3454
3455        for value in values_kv2.iter() {
3456            histogram.record(*value, &[KeyValue::new("key1", "value2")]);
3457        }
3458
3459        test_context.flush_metrics();
3460
3461        let MetricData::Histogram(histogram_data) =
3462            test_context.get_aggregation::<u64>("my_histogram", None)
3463        else {
3464            unreachable!()
3465        };
3466        assert_eq!(histogram_data.data_points.len(), 2);
3467        let data_point1 =
3468            find_histogram_datapoint_with_key_value(&histogram_data.data_points, "key1", "value1")
3469                .expect("datapoint with key1=value1 expected");
3470        if temporality == Temporality::Cumulative {
3471            assert_eq!(data_point1.count, 2 * (values_kv1.len() as u64));
3472            assert_eq!(data_point1.sum, 2 * (values_kv1.iter().sum::<u64>()));
3473            assert_eq!(data_point1.min.unwrap(), *values_kv1.iter().min().unwrap());
3474            assert_eq!(data_point1.max.unwrap(), *values_kv1.iter().max().unwrap());
3475        } else {
3476            assert_eq!(data_point1.count, values_kv1.len() as u64);
3477            assert_eq!(data_point1.sum, values_kv1.iter().sum::<u64>());
3478            assert_eq!(data_point1.min.unwrap(), *values_kv1.iter().min().unwrap());
3479            assert_eq!(data_point1.max.unwrap(), *values_kv1.iter().max().unwrap());
3480        }
3481
3482        let data_point1 =
3483            find_histogram_datapoint_with_key_value(&histogram_data.data_points, "key1", "value2")
3484                .expect("datapoint with key1=value1 expected");
3485        if temporality == Temporality::Cumulative {
3486            assert_eq!(data_point1.count, 2 * (values_kv2.len() as u64));
3487            assert_eq!(data_point1.sum, 2 * (values_kv2.iter().sum::<u64>()));
3488            assert_eq!(data_point1.min.unwrap(), *values_kv2.iter().min().unwrap());
3489            assert_eq!(data_point1.max.unwrap(), *values_kv2.iter().max().unwrap());
3490        } else {
3491            assert_eq!(data_point1.count, values_kv2.len() as u64);
3492            assert_eq!(data_point1.sum, values_kv2.iter().sum::<u64>());
3493            assert_eq!(data_point1.min.unwrap(), *values_kv2.iter().min().unwrap());
3494            assert_eq!(data_point1.max.unwrap(), *values_kv2.iter().max().unwrap());
3495        }
3496    }
3497
3498    fn histogram_aggregation_with_custom_bounds_helper(temporality: Temporality) {
3499        let mut test_context = TestContext::new(temporality);
3500        let histogram = test_context
3501            .meter()
3502            .u64_histogram("test_histogram")
3503            .with_boundaries(vec![1.0, 2.5, 5.5])
3504            .build();
3505        histogram.record(1, &[KeyValue::new("key1", "value1")]);
3506        histogram.record(2, &[KeyValue::new("key1", "value1")]);
3507        histogram.record(3, &[KeyValue::new("key1", "value1")]);
3508        histogram.record(4, &[KeyValue::new("key1", "value1")]);
3509        histogram.record(5, &[KeyValue::new("key1", "value1")]);
3510
3511        test_context.flush_metrics();
3512
3513        // Assert
3514        let MetricData::Histogram(histogram_data) =
3515            test_context.get_aggregation::<u64>("test_histogram", None)
3516        else {
3517            unreachable!()
3518        };
3519        // Expecting 2 time-series.
3520        assert_eq!(histogram_data.data_points.len(), 1);
3521        if let Temporality::Cumulative = temporality {
3522            assert_eq!(
3523                histogram_data.temporality,
3524                Temporality::Cumulative,
3525                "Should produce cumulative"
3526            );
3527        } else {
3528            assert_eq!(
3529                histogram_data.temporality,
3530                Temporality::Delta,
3531                "Should produce delta"
3532            );
3533        }
3534
3535        // find and validate key1=value1 datapoint
3536        let data_point =
3537            find_histogram_datapoint_with_key_value(&histogram_data.data_points, "key1", "value1")
3538                .expect("datapoint with key1=value1 expected");
3539
3540        assert_eq!(data_point.count, 5);
3541        assert_eq!(data_point.sum, 15);
3542
3543        // Check the bucket counts
3544        // -∞ to 1.0: 1
3545        // 1.0 to 2.5: 1
3546        // 2.5 to 5.5: 3
3547        // 5.5 to +∞: 0
3548
3549        assert_eq!(vec![1.0, 2.5, 5.5], data_point.bounds);
3550        assert_eq!(vec![1, 1, 3, 0], data_point.bucket_counts);
3551    }
3552
3553    fn histogram_aggregation_with_empty_bounds_helper(temporality: Temporality) {
3554        let mut test_context = TestContext::new(temporality);
3555        let histogram = test_context
3556            .meter()
3557            .u64_histogram("test_histogram")
3558            .with_boundaries(vec![])
3559            .build();
3560        histogram.record(1, &[KeyValue::new("key1", "value1")]);
3561        histogram.record(2, &[KeyValue::new("key1", "value1")]);
3562        histogram.record(3, &[KeyValue::new("key1", "value1")]);
3563        histogram.record(4, &[KeyValue::new("key1", "value1")]);
3564        histogram.record(5, &[KeyValue::new("key1", "value1")]);
3565
3566        test_context.flush_metrics();
3567
3568        // Assert
3569        let MetricData::Histogram(histogram_data) =
3570            test_context.get_aggregation::<u64>("test_histogram", None)
3571        else {
3572            unreachable!()
3573        };
3574        // Expecting 1 time-series.
3575        assert_eq!(histogram_data.data_points.len(), 1);
3576        if let Temporality::Cumulative = temporality {
3577            assert_eq!(
3578                histogram_data.temporality,
3579                Temporality::Cumulative,
3580                "Should produce cumulative"
3581            );
3582        } else {
3583            assert_eq!(
3584                histogram_data.temporality,
3585                Temporality::Delta,
3586                "Should produce delta"
3587            );
3588        }
3589
3590        // find and validate key1=value1 datapoint
3591        let data_point =
3592            find_histogram_datapoint_with_key_value(&histogram_data.data_points, "key1", "value1")
3593                .expect("datapoint with key1=value1 expected");
3594
3595        assert_eq!(data_point.count, 5);
3596        assert_eq!(data_point.sum, 15);
3597        assert!(data_point.bounds.is_empty());
3598        assert!(data_point.bucket_counts.is_empty());
3599    }
3600
3601    fn histogram_aggregation_with_custom_bounds_and_view_helper(temporality: Temporality) {
3602        for specify_boundaries_in_view in [false, true] {
3603            let view = move |_: &Instrument| {
3604                let mut builder = Stream::builder();
3605                if specify_boundaries_in_view {
3606                    builder = builder.with_aggregation(Aggregation::ExplicitBucketHistogram {
3607                        boundaries: vec![1.5, 4.2, 6.7],
3608                        record_min_max: true,
3609                    });
3610                }
3611                Some(builder.build().unwrap())
3612            };
3613            let mut test_context = TestContext::new_with_view(temporality, view);
3614            let histogram = test_context
3615                .meter()
3616                .u64_histogram("test_histogram")
3617                .with_boundaries(vec![1.0, 2.5, 5.5])
3618                .build();
3619            histogram.record(1, &[KeyValue::new("key1", "value1")]);
3620            histogram.record(2, &[KeyValue::new("key1", "value1")]);
3621            histogram.record(3, &[KeyValue::new("key1", "value1")]);
3622            histogram.record(4, &[KeyValue::new("key1", "value1")]);
3623            histogram.record(5, &[KeyValue::new("key1", "value1")]);
3624
3625            test_context.flush_metrics();
3626
3627            let MetricData::Histogram(histogram_data) =
3628                test_context.get_aggregation::<u64>("test_histogram", None)
3629            else {
3630                unreachable!()
3631            };
3632            assert_eq!(histogram_data.data_points.len(), 1);
3633            if let Temporality::Cumulative = temporality {
3634                assert_eq!(
3635                    histogram_data.temporality,
3636                    Temporality::Cumulative,
3637                    "Should produce cumulative"
3638                );
3639            } else {
3640                assert_eq!(
3641                    histogram_data.temporality,
3642                    Temporality::Delta,
3643                    "Should produce delta"
3644                );
3645            }
3646
3647            // find and validate key1=value1 datapoint
3648            let data_point = find_histogram_datapoint_with_key_value(
3649                &histogram_data.data_points,
3650                "key1",
3651                "value1",
3652            )
3653            .expect("datapoint with key1=value1 expected");
3654
3655            assert_eq!(data_point.count, 5);
3656            assert_eq!(data_point.sum, 15);
3657
3658            // Check the bucket counts
3659            if specify_boundaries_in_view {
3660                // If boundaries are specified in the view, they should take precedence
3661                assert_eq!(vec![1.5, 4.2, 6.7], data_point.bounds);
3662                assert_eq!(vec![1, 3, 1, 0], data_point.bucket_counts);
3663            } else {
3664                // If boundaries are not specified in the view, the ones from the instrument
3665                // should be used
3666                assert_eq!(vec![1.0, 2.5, 5.5], data_point.bounds);
3667                assert_eq!(vec![1, 1, 3, 0], data_point.bucket_counts);
3668            }
3669        }
3670    }
3671
3672    fn exponential_histogram_aggregation_with_view_helper(temporality: Temporality) {
3673        // Arrange: Create a view that converts a regular histogram to Base2ExponentialHistogram
3674        let view = |i: &Instrument| {
3675            if i.name == "test_histogram" {
3676                Some(
3677                    Stream::builder()
3678                        .with_aggregation(Aggregation::Base2ExponentialHistogram {
3679                            max_size: 160,
3680                            max_scale: 20,
3681                            record_min_max: true,
3682                        })
3683                        .build()
3684                        .unwrap(),
3685                )
3686            } else {
3687                None
3688            }
3689        };
3690        let mut test_context = TestContext::new_with_view(temporality, view);
3691        let histogram = test_context.meter().f64_histogram("test_histogram").build();
3692
3693        // Act: Record some values
3694        histogram.record(1.0, &[KeyValue::new("key1", "value1")]);
3695        histogram.record(2.0, &[KeyValue::new("key1", "value1")]);
3696        histogram.record(3.0, &[KeyValue::new("key1", "value1")]);
3697        histogram.record(4.0, &[KeyValue::new("key1", "value1")]);
3698        histogram.record(5.0, &[KeyValue::new("key1", "value1")]);
3699
3700        test_context.flush_metrics();
3701
3702        // Assert: Verify we get an ExponentialHistogram instead of regular Histogram
3703        let exponential_histogram_data =
3704            test_context.get_aggregation::<f64>("test_histogram", None);
3705        let MetricData::ExponentialHistogram(exp_hist) = exponential_histogram_data else {
3706            panic!(
3707                "Expected ExponentialHistogram aggregation, got {:?}",
3708                exponential_histogram_data
3709            );
3710        };
3711
3712        assert_eq!(exp_hist.data_points.len(), 1);
3713        if let Temporality::Cumulative = temporality {
3714            assert_eq!(
3715                exp_hist.temporality,
3716                Temporality::Cumulative,
3717                "Should produce cumulative"
3718            );
3719        } else {
3720            assert_eq!(
3721                exp_hist.temporality,
3722                Temporality::Delta,
3723                "Should produce delta"
3724            );
3725        }
3726
3727        // Validate the data point
3728        let data_point = &exp_hist.data_points[0];
3729        assert_eq!(data_point.count(), 5);
3730        assert_eq!(data_point.sum(), 15.0);
3731        assert_eq!(data_point.min(), Some(1.0));
3732        assert_eq!(data_point.max(), Some(5.0));
3733
3734        // Validate exponential histogram specific fields
3735        // Scale should be within valid range (-10 to 20)
3736        let scale = data_point.scale();
3737        assert!(
3738            (-10..=20).contains(&scale),
3739            "Scale {} should be within valid range [-10, 20]",
3740            scale
3741        );
3742
3743        // zero_count should be 0 since we only recorded positive values > 0
3744        assert_eq!(
3745            data_point.zero_count(),
3746            0,
3747            "zero_count should be 0 for positive values"
3748        );
3749
3750        // Positive bucket should have counts (we recorded positive values)
3751        let positive_bucket = data_point.positive_bucket();
3752        let positive_counts: Vec<u64> = positive_bucket.counts().collect();
3753        let total_positive_count: u64 = positive_counts.iter().sum();
3754        assert_eq!(
3755            total_positive_count, 5,
3756            "Total count in positive buckets should equal number of recorded values"
3757        );
3758
3759        // Negative bucket should be empty (we only recorded positive values)
3760        let negative_bucket = data_point.negative_bucket();
3761        let negative_counts: Vec<u64> = negative_bucket.counts().collect();
3762        let total_negative_count: u64 = negative_counts.iter().sum();
3763        assert_eq!(
3764            total_negative_count, 0,
3765            "Negative bucket should be empty for positive-only values"
3766        );
3767
3768        // Verify the attribute is present
3769        let attrs: Vec<_> = data_point.attributes().collect();
3770        assert_eq!(attrs.len(), 1);
3771        assert_eq!(attrs[0].key.as_str(), "key1");
3772
3773        // Reset and report more measurements to verify Delta vs Cumulative behavior
3774        test_context.reset_metrics();
3775        histogram.record(10.0, &[KeyValue::new("key1", "value1")]);
3776        histogram.record(20.0, &[KeyValue::new("key1", "value1")]);
3777        histogram.record(30.0, &[KeyValue::new("key1", "value1")]);
3778
3779        test_context.flush_metrics();
3780
3781        // Assert second collect
3782        let exponential_histogram_data =
3783            test_context.get_aggregation::<f64>("test_histogram", None);
3784        let MetricData::ExponentialHistogram(exp_hist) = exponential_histogram_data else {
3785            panic!(
3786                "Expected ExponentialHistogram aggregation, got {:?}",
3787                exponential_histogram_data
3788            );
3789        };
3790
3791        assert_eq!(exp_hist.data_points.len(), 1);
3792        let data_point = &exp_hist.data_points[0];
3793
3794        if temporality == Temporality::Cumulative {
3795            // Cumulative: values accumulate (5 original + 3 new = 8 count, 15 + 60 = 75 sum)
3796            assert_eq!(data_point.count(), 8);
3797            assert_eq!(data_point.sum(), 75.0);
3798            assert_eq!(data_point.min(), Some(1.0)); // min from first batch
3799            assert_eq!(data_point.max(), Some(30.0)); // max from second batch
3800        } else {
3801            // Delta: only new values (3 count, 60 sum)
3802            assert_eq!(data_point.count(), 3);
3803            assert_eq!(data_point.sum(), 60.0);
3804            assert_eq!(data_point.min(), Some(10.0));
3805            assert_eq!(data_point.max(), Some(30.0));
3806        }
3807
3808        // Verify positive bucket counts match the count
3809        let positive_bucket = data_point.positive_bucket();
3810        let positive_counts: Vec<u64> = positive_bucket.counts().collect();
3811        let total_positive_count: u64 = positive_counts.iter().sum();
3812        assert_eq!(
3813            total_positive_count,
3814            data_point.count() as u64,
3815            "Total count in positive buckets should equal count"
3816        );
3817    }
3818
3819    fn gauge_aggregation_helper(temporality: Temporality) {
3820        // Arrange
3821        let mut test_context = TestContext::new(temporality);
3822        let gauge = test_context.meter().i64_gauge("my_gauge").build();
3823
3824        // Act
3825        gauge.record(1, &[KeyValue::new("key1", "value1")]);
3826        gauge.record(2, &[KeyValue::new("key1", "value1")]);
3827        gauge.record(1, &[KeyValue::new("key1", "value1")]);
3828        gauge.record(3, &[KeyValue::new("key1", "value1")]);
3829        gauge.record(4, &[KeyValue::new("key1", "value1")]);
3830
3831        gauge.record(11, &[KeyValue::new("key1", "value2")]);
3832        gauge.record(13, &[KeyValue::new("key1", "value2")]);
3833        gauge.record(6, &[KeyValue::new("key1", "value2")]);
3834
3835        test_context.flush_metrics();
3836
3837        // Assert
3838        let MetricData::Gauge(gauge_data_point) =
3839            test_context.get_aggregation::<i64>("my_gauge", None)
3840        else {
3841            unreachable!()
3842        };
3843        // Expecting 2 time-series.
3844        assert_eq!(gauge_data_point.data_points.len(), 2);
3845
3846        // find and validate key1=value2 datapoint
3847        let data_point1 =
3848            find_gauge_datapoint_with_key_value(&gauge_data_point.data_points, "key1", "value1")
3849                .expect("datapoint with key1=value1 expected");
3850        assert_eq!(data_point1.value, 4);
3851
3852        let data_point1 =
3853            find_gauge_datapoint_with_key_value(&gauge_data_point.data_points, "key1", "value2")
3854                .expect("datapoint with key1=value2 expected");
3855        assert_eq!(data_point1.value, 6);
3856
3857        // Reset and report more measurements
3858        test_context.reset_metrics();
3859        gauge.record(1, &[KeyValue::new("key1", "value1")]);
3860        gauge.record(2, &[KeyValue::new("key1", "value1")]);
3861        gauge.record(11, &[KeyValue::new("key1", "value1")]);
3862        gauge.record(3, &[KeyValue::new("key1", "value1")]);
3863        gauge.record(41, &[KeyValue::new("key1", "value1")]);
3864
3865        gauge.record(34, &[KeyValue::new("key1", "value2")]);
3866        gauge.record(12, &[KeyValue::new("key1", "value2")]);
3867        gauge.record(54, &[KeyValue::new("key1", "value2")]);
3868
3869        test_context.flush_metrics();
3870
3871        let MetricData::Gauge(gauge) = test_context.get_aggregation::<i64>("my_gauge", None) else {
3872            unreachable!()
3873        };
3874        assert_eq!(gauge.data_points.len(), 2);
3875        let data_point1 = find_gauge_datapoint_with_key_value(&gauge.data_points, "key1", "value1")
3876            .expect("datapoint with key1=value1 expected");
3877        assert_eq!(data_point1.value, 41);
3878
3879        let data_point1 = find_gauge_datapoint_with_key_value(&gauge.data_points, "key1", "value2")
3880            .expect("datapoint with key1=value2 expected");
3881        assert_eq!(data_point1.value, 54);
3882    }
3883
3884    fn observable_gauge_aggregation_helper(temporality: Temporality, use_empty_attributes: bool) {
3885        // Arrange
3886        let mut test_context = TestContext::new(temporality);
3887        let _observable_gauge = test_context
3888            .meter()
3889            .i64_observable_gauge("test_observable_gauge")
3890            .with_callback(move |observer| {
3891                if use_empty_attributes {
3892                    observer.observe(1, &[]);
3893                }
3894                observer.observe(4, &[KeyValue::new("key1", "value1")]);
3895                observer.observe(5, &[KeyValue::new("key2", "value2")]);
3896            })
3897            .build();
3898
3899        test_context.flush_metrics();
3900
3901        // Assert
3902        let MetricData::Gauge(gauge) =
3903            test_context.get_aggregation::<i64>("test_observable_gauge", None)
3904        else {
3905            unreachable!()
3906        };
3907        // Expecting 2 time-series.
3908        let expected_time_series_count = if use_empty_attributes { 3 } else { 2 };
3909        assert_eq!(gauge.data_points.len(), expected_time_series_count);
3910
3911        if use_empty_attributes {
3912            // find and validate zero attribute datapoint
3913            let zero_attribute_datapoint =
3914                find_gauge_datapoint_with_no_attributes(&gauge.data_points)
3915                    .expect("datapoint with no attributes expected");
3916            assert_eq!(zero_attribute_datapoint.value, 1);
3917        }
3918
3919        // find and validate key1=value1 datapoint
3920        let data_point1 = find_gauge_datapoint_with_key_value(&gauge.data_points, "key1", "value1")
3921            .expect("datapoint with key1=value1 expected");
3922        assert_eq!(data_point1.value, 4);
3923
3924        // find and validate key2=value2 datapoint
3925        let data_point2 = find_gauge_datapoint_with_key_value(&gauge.data_points, "key2", "value2")
3926            .expect("datapoint with key2=value2 expected");
3927        assert_eq!(data_point2.value, 5);
3928
3929        // Reset and report more measurements
3930        test_context.reset_metrics();
3931
3932        test_context.flush_metrics();
3933
3934        let MetricData::Gauge(gauge) =
3935            test_context.get_aggregation::<i64>("test_observable_gauge", None)
3936        else {
3937            unreachable!()
3938        };
3939        assert_eq!(gauge.data_points.len(), expected_time_series_count);
3940
3941        if use_empty_attributes {
3942            let zero_attribute_datapoint =
3943                find_gauge_datapoint_with_no_attributes(&gauge.data_points)
3944                    .expect("datapoint with no attributes expected");
3945            assert_eq!(zero_attribute_datapoint.value, 1);
3946        }
3947
3948        let data_point1 = find_gauge_datapoint_with_key_value(&gauge.data_points, "key1", "value1")
3949            .expect("datapoint with key1=value1 expected");
3950        assert_eq!(data_point1.value, 4);
3951
3952        let data_point2 = find_gauge_datapoint_with_key_value(&gauge.data_points, "key2", "value2")
3953            .expect("datapoint with key2=value2 expected");
3954        assert_eq!(data_point2.value, 5);
3955    }
3956
3957    fn counter_aggregation_helper(temporality: Temporality) {
3958        // Arrange
3959        let mut test_context = TestContext::new(temporality);
3960        let counter = test_context.u64_counter("test", "my_counter", None);
3961
3962        // Act
3963        counter.add(1, &[KeyValue::new("key1", "value1")]);
3964        counter.add(1, &[KeyValue::new("key1", "value1")]);
3965        counter.add(1, &[KeyValue::new("key1", "value1")]);
3966        counter.add(1, &[KeyValue::new("key1", "value1")]);
3967        counter.add(1, &[KeyValue::new("key1", "value1")]);
3968
3969        counter.add(1, &[KeyValue::new("key1", "value2")]);
3970        counter.add(1, &[KeyValue::new("key1", "value2")]);
3971        counter.add(1, &[KeyValue::new("key1", "value2")]);
3972
3973        test_context.flush_metrics();
3974
3975        // Assert
3976        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
3977            unreachable!()
3978        };
3979        // Expecting 2 time-series.
3980        assert_eq!(sum.data_points.len(), 2);
3981        assert!(sum.is_monotonic, "Counter should produce monotonic.");
3982        if let Temporality::Cumulative = temporality {
3983            assert_eq!(
3984                sum.temporality,
3985                Temporality::Cumulative,
3986                "Should produce cumulative"
3987            );
3988        } else {
3989            assert_eq!(sum.temporality, Temporality::Delta, "Should produce delta");
3990        }
3991
3992        // find and validate key1=value2 datapoint
3993        let data_point1 = find_sum_datapoint_with_key_value(&sum.data_points, "key1", "value1")
3994            .expect("datapoint with key1=value1 expected");
3995        assert_eq!(data_point1.value, 5);
3996
3997        let data_point1 = find_sum_datapoint_with_key_value(&sum.data_points, "key1", "value2")
3998            .expect("datapoint with key1=value2 expected");
3999        assert_eq!(data_point1.value, 3);
4000
4001        // Reset and report more measurements
4002        test_context.reset_metrics();
4003        counter.add(1, &[KeyValue::new("key1", "value1")]);
4004        counter.add(1, &[KeyValue::new("key1", "value1")]);
4005        counter.add(1, &[KeyValue::new("key1", "value1")]);
4006        counter.add(1, &[KeyValue::new("key1", "value1")]);
4007        counter.add(1, &[KeyValue::new("key1", "value1")]);
4008
4009        counter.add(1, &[KeyValue::new("key1", "value2")]);
4010        counter.add(1, &[KeyValue::new("key1", "value2")]);
4011        counter.add(1, &[KeyValue::new("key1", "value2")]);
4012
4013        test_context.flush_metrics();
4014
4015        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
4016            unreachable!()
4017        };
4018        assert_eq!(sum.data_points.len(), 2);
4019        let data_point1 = find_sum_datapoint_with_key_value(&sum.data_points, "key1", "value1")
4020            .expect("datapoint with key1=value1 expected");
4021        if temporality == Temporality::Cumulative {
4022            assert_eq!(data_point1.value, 10);
4023        } else {
4024            assert_eq!(data_point1.value, 5);
4025        }
4026
4027        let data_point1 = find_sum_datapoint_with_key_value(&sum.data_points, "key1", "value2")
4028            .expect("datapoint with key1=value2 expected");
4029        if temporality == Temporality::Cumulative {
4030            assert_eq!(data_point1.value, 6);
4031        } else {
4032            assert_eq!(data_point1.value, 3);
4033        }
4034    }
4035
4036    fn counter_aggregation_overflow_helper(temporality: Temporality) {
4037        // Arrange
4038        let mut test_context = TestContext::new(temporality);
4039        let counter = test_context.u64_counter("test", "my_counter", None);
4040
4041        // Act
4042        // Record measurements with A:0, A:1,.......A:1999, which just fits in the 2000 limit
4043        for v in 0..2000 {
4044            counter.add(100, &[KeyValue::new("A", v.to_string())]);
4045        }
4046
4047        // Empty attributes is specially treated and does not count towards the limit.
4048        counter.add(3, &[]);
4049        counter.add(3, &[]);
4050
4051        // All of the below will now go into overflow.
4052        counter.add(100, &[KeyValue::new("A", "foo")]);
4053        counter.add(100, &[KeyValue::new("A", "another")]);
4054        counter.add(100, &[KeyValue::new("A", "yet_another")]);
4055        test_context.flush_metrics();
4056
4057        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
4058            unreachable!()
4059        };
4060
4061        // Expecting 2002 metric points. (2000 + 1 overflow + Empty attributes)
4062        assert_eq!(sum.data_points.len(), 2002);
4063
4064        let data_point =
4065            find_overflow_sum_datapoint(&sum.data_points).expect("overflow point expected");
4066        assert_eq!(data_point.value, 300);
4067
4068        // let empty_attrs_data_point = &sum.data_points[0];
4069        let empty_attrs_data_point = find_sum_datapoint_with_no_attributes(&sum.data_points)
4070            .expect("Empty attributes point expected");
4071        assert!(
4072            empty_attrs_data_point.attributes.is_empty(),
4073            "Non-empty attribute set"
4074        );
4075        assert_eq!(
4076            empty_attrs_data_point.value, 6,
4077            "Empty attributes value should be 3+3=6"
4078        );
4079
4080        // Phase 2 - for delta temporality, collect_and_reset uses in-place eviction:
4081        // the first collect marks entries as not-updated, and the second collect evicts
4082        // those still-stale entries. We need an extra flush to trigger that eviction
4083        // before adding new measurements that should fit under the cardinality limit.
4084        test_context.reset_metrics();
4085        if temporality == Temporality::Delta {
4086            test_context.flush_metrics();
4087            test_context.reset_metrics();
4088        }
4089        // The following should be aggregated normally for Delta,
4090        // and should go into overflow for Cumulative.
4091        counter.add(100, &[KeyValue::new("A", "foo")]);
4092        counter.add(100, &[KeyValue::new("A", "another")]);
4093        counter.add(100, &[KeyValue::new("A", "yet_another")]);
4094        test_context.flush_metrics();
4095
4096        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
4097            unreachable!()
4098        };
4099
4100        if temporality == Temporality::Delta {
4101            assert_eq!(sum.data_points.len(), 3);
4102
4103            let data_point = find_sum_datapoint_with_key_value(&sum.data_points, "A", "foo")
4104                .expect("point expected");
4105            assert_eq!(data_point.value, 100);
4106
4107            let data_point = find_sum_datapoint_with_key_value(&sum.data_points, "A", "another")
4108                .expect("point expected");
4109            assert_eq!(data_point.value, 100);
4110
4111            let data_point =
4112                find_sum_datapoint_with_key_value(&sum.data_points, "A", "yet_another")
4113                    .expect("point expected");
4114            assert_eq!(data_point.value, 100);
4115        } else {
4116            // For cumulative, overflow should still be there, and new points should not be added.
4117            assert_eq!(sum.data_points.len(), 2002);
4118            let data_point =
4119                find_overflow_sum_datapoint(&sum.data_points).expect("overflow point expected");
4120            assert_eq!(data_point.value, 600);
4121
4122            let data_point = find_sum_datapoint_with_key_value(&sum.data_points, "A", "foo");
4123            assert!(data_point.is_none(), "point should not be present");
4124
4125            let data_point = find_sum_datapoint_with_key_value(&sum.data_points, "A", "another");
4126            assert!(data_point.is_none(), "point should not be present");
4127
4128            let data_point =
4129                find_sum_datapoint_with_key_value(&sum.data_points, "A", "yet_another");
4130            assert!(data_point.is_none(), "point should not be present");
4131        }
4132    }
4133
4134    fn counter_aggregation_overflow_helper_custom_limit(temporality: Temporality) {
4135        // Arrange
4136        let cardinality_limit = 2300;
4137        let view_change_cardinality = move |i: &Instrument| {
4138            if i.name == "my_counter" {
4139                Some(
4140                    Stream::builder()
4141                        .with_name("my_counter")
4142                        .with_cardinality_limit(cardinality_limit)
4143                        .build()
4144                        .unwrap(),
4145                )
4146            } else {
4147                None
4148            }
4149        };
4150        let mut test_context = TestContext::new_with_view(temporality, view_change_cardinality);
4151        let counter = test_context.u64_counter("test", "my_counter", None);
4152
4153        // Act
4154        // Record measurements with A:0, A:1,.......A:cardinality_limit, which just fits in the cardinality_limit
4155        for v in 0..cardinality_limit {
4156            counter.add(100, &[KeyValue::new("A", v.to_string())]);
4157        }
4158
4159        // Empty attributes is specially treated and does not count towards the limit.
4160        counter.add(3, &[]);
4161        counter.add(3, &[]);
4162
4163        // All of the below will now go into overflow.
4164        counter.add(100, &[KeyValue::new("A", "foo")]);
4165        counter.add(100, &[KeyValue::new("A", "another")]);
4166        counter.add(100, &[KeyValue::new("A", "yet_another")]);
4167        test_context.flush_metrics();
4168
4169        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
4170            unreachable!()
4171        };
4172
4173        // Expecting (cardinality_limit + 1 overflow + Empty attributes) data points.
4174        assert_eq!(sum.data_points.len(), cardinality_limit + 1 + 1);
4175
4176        let data_point =
4177            find_overflow_sum_datapoint(&sum.data_points).expect("overflow point expected");
4178        assert_eq!(data_point.value, 300);
4179
4180        // let empty_attrs_data_point = &sum.data_points[0];
4181        let empty_attrs_data_point = find_sum_datapoint_with_no_attributes(&sum.data_points)
4182            .expect("Empty attributes point expected");
4183        assert!(
4184            empty_attrs_data_point.attributes.is_empty(),
4185            "Non-empty attribute set"
4186        );
4187        assert_eq!(
4188            empty_attrs_data_point.value, 6,
4189            "Empty attributes value should be 3+3=6"
4190        );
4191
4192        // Phase 2 - for delta temporality, collect_and_reset uses in-place eviction:
4193        // the first collect marks entries as not-updated, and the second collect evicts
4194        // those still-stale entries. We need an extra flush to trigger that eviction
4195        // before adding new measurements that should fit under the cardinality limit.
4196        test_context.reset_metrics();
4197        if temporality == Temporality::Delta {
4198            test_context.flush_metrics();
4199            test_context.reset_metrics();
4200        }
4201        // The following should be aggregated normally for Delta,
4202        // and should go into overflow for Cumulative.
4203        counter.add(100, &[KeyValue::new("A", "foo")]);
4204        counter.add(100, &[KeyValue::new("A", "another")]);
4205        counter.add(100, &[KeyValue::new("A", "yet_another")]);
4206        test_context.flush_metrics();
4207
4208        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
4209            unreachable!()
4210        };
4211
4212        if temporality == Temporality::Delta {
4213            assert_eq!(sum.data_points.len(), 3);
4214
4215            let data_point = find_sum_datapoint_with_key_value(&sum.data_points, "A", "foo")
4216                .expect("point expected");
4217            assert_eq!(data_point.value, 100);
4218
4219            let data_point = find_sum_datapoint_with_key_value(&sum.data_points, "A", "another")
4220                .expect("point expected");
4221            assert_eq!(data_point.value, 100);
4222
4223            let data_point =
4224                find_sum_datapoint_with_key_value(&sum.data_points, "A", "yet_another")
4225                    .expect("point expected");
4226            assert_eq!(data_point.value, 100);
4227        } else {
4228            // For cumulative, overflow should still be there, and new points should not be added.
4229            assert_eq!(sum.data_points.len(), cardinality_limit + 1 + 1);
4230            let data_point =
4231                find_overflow_sum_datapoint(&sum.data_points).expect("overflow point expected");
4232            assert_eq!(data_point.value, 600);
4233
4234            let data_point = find_sum_datapoint_with_key_value(&sum.data_points, "A", "foo");
4235            assert!(data_point.is_none(), "point should not be present");
4236
4237            let data_point = find_sum_datapoint_with_key_value(&sum.data_points, "A", "another");
4238            assert!(data_point.is_none(), "point should not be present");
4239
4240            let data_point =
4241                find_sum_datapoint_with_key_value(&sum.data_points, "A", "yet_another");
4242            assert!(data_point.is_none(), "point should not be present");
4243        }
4244    }
4245
4246    fn counter_aggregation_attribute_order_helper(temporality: Temporality, start_sorted: bool) {
4247        // Arrange
4248        let mut test_context = TestContext::new(temporality);
4249        let counter = test_context.u64_counter("test", "my_counter", None);
4250
4251        // Act
4252        // Add the same set of attributes in different order. (they are expected
4253        // to be treated as same attributes)
4254        // start with sorted order
4255        if start_sorted {
4256            counter.add(
4257                1,
4258                &[
4259                    KeyValue::new("A", "a"),
4260                    KeyValue::new("B", "b"),
4261                    KeyValue::new("C", "c"),
4262                ],
4263            );
4264        } else {
4265            counter.add(
4266                1,
4267                &[
4268                    KeyValue::new("A", "a"),
4269                    KeyValue::new("C", "c"),
4270                    KeyValue::new("B", "b"),
4271                ],
4272            );
4273        }
4274
4275        counter.add(
4276            1,
4277            &[
4278                KeyValue::new("A", "a"),
4279                KeyValue::new("C", "c"),
4280                KeyValue::new("B", "b"),
4281            ],
4282        );
4283        counter.add(
4284            1,
4285            &[
4286                KeyValue::new("B", "b"),
4287                KeyValue::new("A", "a"),
4288                KeyValue::new("C", "c"),
4289            ],
4290        );
4291        counter.add(
4292            1,
4293            &[
4294                KeyValue::new("B", "b"),
4295                KeyValue::new("C", "c"),
4296                KeyValue::new("A", "a"),
4297            ],
4298        );
4299        counter.add(
4300            1,
4301            &[
4302                KeyValue::new("C", "c"),
4303                KeyValue::new("B", "b"),
4304                KeyValue::new("A", "a"),
4305            ],
4306        );
4307        counter.add(
4308            1,
4309            &[
4310                KeyValue::new("C", "c"),
4311                KeyValue::new("A", "a"),
4312                KeyValue::new("B", "b"),
4313            ],
4314        );
4315        test_context.flush_metrics();
4316
4317        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
4318            unreachable!()
4319        };
4320
4321        // Expecting 1 time-series.
4322        assert_eq!(sum.data_points.len(), 1);
4323
4324        // validate the sole datapoint
4325        let data_point1 = &sum.data_points[0];
4326        assert_eq!(data_point1.value, 6);
4327    }
4328
4329    fn updown_counter_aggregation_helper(temporality: Temporality) {
4330        // Arrange
4331        let mut test_context = TestContext::new(temporality);
4332        let counter = test_context.i64_up_down_counter("test", "my_updown_counter", None);
4333
4334        // Act
4335        counter.add(10, &[KeyValue::new("key1", "value1")]);
4336        counter.add(-1, &[KeyValue::new("key1", "value1")]);
4337        counter.add(-5, &[KeyValue::new("key1", "value1")]);
4338        counter.add(0, &[KeyValue::new("key1", "value1")]);
4339        counter.add(1, &[KeyValue::new("key1", "value1")]);
4340
4341        counter.add(10, &[KeyValue::new("key1", "value2")]);
4342        counter.add(0, &[KeyValue::new("key1", "value2")]);
4343        counter.add(-3, &[KeyValue::new("key1", "value2")]);
4344
4345        test_context.flush_metrics();
4346
4347        // Assert
4348        let MetricData::Sum(sum) = test_context.get_aggregation::<i64>("my_updown_counter", None)
4349        else {
4350            unreachable!()
4351        };
4352        // Expecting 2 time-series.
4353        assert_eq!(sum.data_points.len(), 2);
4354        assert!(
4355            !sum.is_monotonic,
4356            "UpDownCounter should produce non-monotonic."
4357        );
4358        assert_eq!(
4359            sum.temporality,
4360            Temporality::Cumulative,
4361            "Should produce Cumulative for UpDownCounter"
4362        );
4363
4364        // find and validate key1=value2 datapoint
4365        let data_point1 = find_sum_datapoint_with_key_value(&sum.data_points, "key1", "value1")
4366            .expect("datapoint with key1=value1 expected");
4367        assert_eq!(data_point1.value, 5);
4368
4369        let data_point1 = find_sum_datapoint_with_key_value(&sum.data_points, "key1", "value2")
4370            .expect("datapoint with key1=value2 expected");
4371        assert_eq!(data_point1.value, 7);
4372
4373        // Reset and report more measurements
4374        test_context.reset_metrics();
4375        counter.add(10, &[KeyValue::new("key1", "value1")]);
4376        counter.add(-1, &[KeyValue::new("key1", "value1")]);
4377        counter.add(-5, &[KeyValue::new("key1", "value1")]);
4378        counter.add(0, &[KeyValue::new("key1", "value1")]);
4379        counter.add(1, &[KeyValue::new("key1", "value1")]);
4380
4381        counter.add(10, &[KeyValue::new("key1", "value2")]);
4382        counter.add(0, &[KeyValue::new("key1", "value2")]);
4383        counter.add(-3, &[KeyValue::new("key1", "value2")]);
4384
4385        test_context.flush_metrics();
4386
4387        let MetricData::Sum(sum) = test_context.get_aggregation::<i64>("my_updown_counter", None)
4388        else {
4389            unreachable!()
4390        };
4391        assert_eq!(sum.data_points.len(), 2);
4392        let data_point1 = find_sum_datapoint_with_key_value(&sum.data_points, "key1", "value1")
4393            .expect("datapoint with key1=value1 expected");
4394        assert_eq!(data_point1.value, 10);
4395
4396        let data_point1 = find_sum_datapoint_with_key_value(&sum.data_points, "key1", "value2")
4397            .expect("datapoint with key1=value2 expected");
4398        assert_eq!(data_point1.value, 14);
4399    }
4400
4401    fn find_sum_datapoint_with_key_value<'a, T>(
4402        data_points: &'a [SumDataPoint<T>],
4403        key: &str,
4404        value: &str,
4405    ) -> Option<&'a SumDataPoint<T>> {
4406        data_points.iter().find(|&datapoint| {
4407            datapoint
4408                .attributes
4409                .iter()
4410                .any(|kv| kv.key.as_str() == key && kv.value.as_str() == value)
4411        })
4412    }
4413
4414    fn find_overflow_sum_datapoint<T>(data_points: &[SumDataPoint<T>]) -> Option<&SumDataPoint<T>> {
4415        data_points.iter().find(|&datapoint| {
4416            datapoint.attributes.iter().any(|kv| {
4417                kv.key.as_str() == "otel.metric.overflow" && kv.value == Value::Bool(true)
4418            })
4419        })
4420    }
4421
4422    fn find_gauge_datapoint_with_key_value<'a, T>(
4423        data_points: &'a [GaugeDataPoint<T>],
4424        key: &str,
4425        value: &str,
4426    ) -> Option<&'a GaugeDataPoint<T>> {
4427        data_points.iter().find(|&datapoint| {
4428            datapoint
4429                .attributes
4430                .iter()
4431                .any(|kv| kv.key.as_str() == key && kv.value.as_str() == value)
4432        })
4433    }
4434
4435    fn find_sum_datapoint_with_no_attributes<T>(
4436        data_points: &[SumDataPoint<T>],
4437    ) -> Option<&SumDataPoint<T>> {
4438        data_points
4439            .iter()
4440            .find(|&datapoint| datapoint.attributes.is_empty())
4441    }
4442
4443    fn find_gauge_datapoint_with_no_attributes<T>(
4444        data_points: &[GaugeDataPoint<T>],
4445    ) -> Option<&GaugeDataPoint<T>> {
4446        data_points
4447            .iter()
4448            .find(|&datapoint| datapoint.attributes.is_empty())
4449    }
4450
4451    fn find_histogram_datapoint_with_key_value<'a, T>(
4452        data_points: &'a [HistogramDataPoint<T>],
4453        key: &str,
4454        value: &str,
4455    ) -> Option<&'a HistogramDataPoint<T>> {
4456        data_points.iter().find(|&datapoint| {
4457            datapoint
4458                .attributes
4459                .iter()
4460                .any(|kv| kv.key.as_str() == key && kv.value.as_str() == value)
4461        })
4462    }
4463
4464    fn find_histogram_datapoint_with_no_attributes<T>(
4465        data_points: &[HistogramDataPoint<T>],
4466    ) -> Option<&HistogramDataPoint<T>> {
4467        data_points
4468            .iter()
4469            .find(|&datapoint| datapoint.attributes.is_empty())
4470    }
4471
4472    #[cfg(feature = "experimental_metrics_bound_instruments")]
4473    fn find_overflow_histogram_datapoint<T>(
4474        data_points: &[HistogramDataPoint<T>],
4475    ) -> Option<&HistogramDataPoint<T>> {
4476        data_points.iter().find(|&datapoint| {
4477            datapoint.attributes.iter().any(|kv| {
4478                kv.key.as_str() == "otel.metric.overflow" && kv.value == Value::Bool(true)
4479            })
4480        })
4481    }
4482
4483    #[cfg(feature = "experimental_metrics_bound_instruments")]
4484    fn find_overflow_exponential_histogram_datapoint<T>(
4485        data_points: &[ExponentialHistogramDataPoint<T>],
4486    ) -> Option<&ExponentialHistogramDataPoint<T>> {
4487        data_points.iter().find(|&datapoint| {
4488            datapoint.attributes.iter().any(|kv| {
4489                kv.key.as_str() == "otel.metric.overflow" && kv.value == Value::Bool(true)
4490            })
4491        })
4492    }
4493
4494    fn find_scope_metric<'a>(
4495        metrics: &'a [ScopeMetrics],
4496        name: &'a str,
4497    ) -> Option<&'a ScopeMetrics> {
4498        metrics
4499            .iter()
4500            .find(|&scope_metric| scope_metric.scope.name() == name)
4501    }
4502
4503    struct TestContext {
4504        exporter: InMemoryMetricExporter,
4505        meter_provider: SdkMeterProvider,
4506
4507        // Saving this on the test context for lifetime simplicity
4508        resource_metrics: Vec<ResourceMetrics>,
4509    }
4510
4511    impl TestContext {
4512        fn new(temporality: Temporality) -> Self {
4513            let exporter = InMemoryMetricExporterBuilder::new().with_temporality(temporality);
4514            let exporter = exporter.build();
4515            let meter_provider = SdkMeterProvider::builder()
4516                .with_periodic_exporter(exporter.clone())
4517                .build();
4518
4519            TestContext {
4520                exporter,
4521                meter_provider,
4522                resource_metrics: vec![],
4523            }
4524        }
4525
4526        fn new_with_view<T>(temporality: Temporality, view: T) -> Self
4527        where
4528            T: Fn(&Instrument) -> Option<Stream> + Send + Sync + 'static,
4529        {
4530            let exporter = InMemoryMetricExporterBuilder::new().with_temporality(temporality);
4531            let exporter = exporter.build();
4532            let meter_provider = SdkMeterProvider::builder()
4533                .with_periodic_exporter(exporter.clone())
4534                .with_view(view)
4535                .build();
4536
4537            TestContext {
4538                exporter,
4539                meter_provider,
4540                resource_metrics: vec![],
4541            }
4542        }
4543
4544        fn u64_counter(
4545            &self,
4546            meter_name: &'static str,
4547            counter_name: &'static str,
4548            unit: Option<&'static str>,
4549        ) -> Counter<u64> {
4550            let meter = self.meter_provider.meter(meter_name);
4551            let mut counter_builder = meter.u64_counter(counter_name);
4552            if let Some(unit_name) = unit {
4553                counter_builder = counter_builder.with_unit(unit_name);
4554            }
4555            counter_builder.build()
4556        }
4557
4558        fn i64_up_down_counter(
4559            &self,
4560            meter_name: &'static str,
4561            counter_name: &'static str,
4562            unit: Option<&'static str>,
4563        ) -> UpDownCounter<i64> {
4564            let meter = self.meter_provider.meter(meter_name);
4565            let mut updown_counter_builder = meter.i64_up_down_counter(counter_name);
4566            if let Some(unit_name) = unit {
4567                updown_counter_builder = updown_counter_builder.with_unit(unit_name);
4568            }
4569            updown_counter_builder.build()
4570        }
4571
4572        fn meter(&self) -> Meter {
4573            self.meter_provider.meter("test")
4574        }
4575
4576        fn flush_metrics(&self) {
4577            self.meter_provider.force_flush().unwrap();
4578        }
4579
4580        fn reset_metrics(&self) {
4581            self.exporter.reset();
4582        }
4583
4584        fn check_no_metrics(&self) {
4585            let resource_metrics = self
4586                .exporter
4587                .get_finished_metrics()
4588                .expect("metrics expected to be exported"); // TODO: Need to fix InMemoryMetricExporter to return None.
4589
4590            assert!(resource_metrics.is_empty(), "no metrics should be exported");
4591        }
4592
4593        fn get_aggregation<T: Number>(
4594            &mut self,
4595            counter_name: &str,
4596            unit_name: Option<&str>,
4597        ) -> &MetricData<T> {
4598            self.resource_metrics = self
4599                .exporter
4600                .get_finished_metrics()
4601                .expect("metrics expected to be exported");
4602
4603            assert!(
4604                !self.resource_metrics.is_empty(),
4605                "no metrics were exported"
4606            );
4607
4608            assert!(
4609                self.resource_metrics.len() == 1,
4610                "Expected single resource metrics."
4611            );
4612            let resource_metric = self
4613                .resource_metrics
4614                .first()
4615                .expect("This should contain exactly one resource metric, as validated above.");
4616
4617            assert!(
4618                !resource_metric.scope_metrics.is_empty(),
4619                "No scope metrics in latest export"
4620            );
4621            assert!(!resource_metric.scope_metrics[0].metrics.is_empty());
4622
4623            let metric = &resource_metric.scope_metrics[0].metrics[0];
4624            assert_eq!(metric.name, counter_name);
4625            if let Some(expected_unit) = unit_name {
4626                assert_eq!(metric.unit, expected_unit);
4627            }
4628
4629            T::extract_metrics_data_ref(&metric.data)
4630                .expect("Failed to cast aggregation to expected type")
4631        }
4632
4633        fn get_from_multiple_aggregations<T: Number>(
4634            &mut self,
4635            counter_name: &str,
4636            unit_name: Option<&str>,
4637            invocation_count: usize,
4638        ) -> Vec<&MetricData<T>> {
4639            self.resource_metrics = self
4640                .exporter
4641                .get_finished_metrics()
4642                .expect("metrics expected to be exported");
4643
4644            assert!(
4645                !self.resource_metrics.is_empty(),
4646                "no metrics were exported"
4647            );
4648
4649            assert_eq!(
4650                self.resource_metrics.len(),
4651                invocation_count,
4652                "Expected collect to be called {invocation_count} times"
4653            );
4654
4655            let result = self
4656                .resource_metrics
4657                .iter()
4658                .map(|resource_metric| {
4659                    assert!(
4660                        !resource_metric.scope_metrics.is_empty(),
4661                        "An export with no scope metrics occurred"
4662                    );
4663
4664                    assert!(!resource_metric.scope_metrics[0].metrics.is_empty());
4665
4666                    let metric = &resource_metric.scope_metrics[0].metrics[0];
4667                    assert_eq!(metric.name, counter_name);
4668
4669                    if let Some(expected_unit) = unit_name {
4670                        assert_eq!(metric.unit, expected_unit);
4671                    }
4672
4673                    let aggregation = T::extract_metrics_data_ref(&metric.data)
4674                        .expect("Failed to cast aggregation to expected type");
4675                    aggregation
4676                })
4677                .collect::<Vec<_>>();
4678
4679            result
4680        }
4681    }
4682
4683    #[test]
4684    fn parse_valid_temporality_values() {
4685        assert_eq!(
4686            "cumulative".parse::<Temporality>(),
4687            Ok(Temporality::Cumulative)
4688        );
4689        assert_eq!("delta".parse::<Temporality>(), Ok(Temporality::Delta));
4690        assert_eq!(
4691            "lowmemory".parse::<Temporality>(),
4692            Ok(Temporality::LowMemory)
4693        );
4694    }
4695
4696    #[test]
4697    fn parse_temporality_case_insensitive() {
4698        assert_eq!(
4699            "Cumulative".parse::<Temporality>(),
4700            Ok(Temporality::Cumulative)
4701        );
4702        assert_eq!("DELTA".parse::<Temporality>(), Ok(Temporality::Delta));
4703        assert_eq!(
4704            "LowMemory".parse::<Temporality>(),
4705            Ok(Temporality::LowMemory)
4706        );
4707        assert_eq!(
4708            "LOWMEMORY".parse::<Temporality>(),
4709            Ok(Temporality::LowMemory)
4710        );
4711    }
4712
4713    #[test]
4714    fn parse_invalid_temporality_returns_err() {
4715        assert!("unknown".parse::<Temporality>().is_err());
4716        assert!("".parse::<Temporality>().is_err());
4717        assert!("cumulativ".parse::<Temporality>().is_err());
4718    }
4719
4720    #[cfg(feature = "experimental_metrics_bound_instruments")]
4721    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
4722    async fn bound_counter_cumulative() {
4723        let mut test_context = TestContext::new(Temporality::Cumulative);
4724        let counter = test_context.u64_counter("test", "my_counter", None);
4725        let attrs = vec![KeyValue::new("key1", "bound_value")];
4726        let bound = counter.bind(&attrs);
4727
4728        bound.add(10);
4729        bound.add(20);
4730        bound.add(30);
4731        test_context.flush_metrics();
4732
4733        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
4734            unreachable!()
4735        };
4736
4737        assert_eq!(sum.data_points.len(), 1, "Expected one data point");
4738        assert!(sum.is_monotonic);
4739        assert_eq!(sum.temporality, Temporality::Cumulative);
4740
4741        let data_point = &sum.data_points[0];
4742        assert_eq!(data_point.value, 60);
4743        assert_eq!(
4744            data_point.attributes,
4745            vec![KeyValue::new("key1", "bound_value")]
4746        );
4747    }
4748
4749    #[cfg(feature = "experimental_metrics_bound_instruments")]
4750    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
4751    async fn bound_counter_delta() {
4752        let mut test_context = TestContext::new(Temporality::Delta);
4753        let counter = test_context.u64_counter("test", "my_counter", None);
4754        let attrs = vec![KeyValue::new("key1", "bound_value")];
4755        let bound = counter.bind(&attrs);
4756
4757        bound.add(50);
4758        test_context.flush_metrics();
4759
4760        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
4761            unreachable!()
4762        };
4763        assert_eq!(sum.temporality, Temporality::Delta);
4764        assert_eq!(sum.data_points.len(), 1);
4765        assert_eq!(sum.data_points[0].value, 50);
4766
4767        // After delta collect, add more and collect again
4768        test_context.reset_metrics();
4769        bound.add(25);
4770        test_context.flush_metrics();
4771
4772        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
4773            unreachable!()
4774        };
4775        assert_eq!(sum.data_points.len(), 1);
4776        assert_eq!(
4777            sum.data_points[0].value, 25,
4778            "Delta should reset between collections"
4779        );
4780    }
4781
4782    #[cfg(feature = "experimental_metrics_bound_instruments")]
4783    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
4784    async fn bound_histogram_cumulative() {
4785        let mut test_context = TestContext::new(Temporality::Cumulative);
4786        let histogram = test_context
4787            .meter()
4788            .f64_histogram("my_histogram")
4789            .with_boundaries(vec![5.0, 10.0, 25.0, 50.0])
4790            .build();
4791        let attrs = vec![KeyValue::new("key1", "bound_value")];
4792        let bound = histogram.bind(&attrs);
4793
4794        bound.record(1.0);
4795        bound.record(7.5);
4796        bound.record(15.0);
4797        bound.record(30.0);
4798        test_context.flush_metrics();
4799
4800        let MetricData::Histogram(histogram_data) =
4801            test_context.get_aggregation::<f64>("my_histogram", None)
4802        else {
4803            unreachable!()
4804        };
4805
4806        assert_eq!(histogram_data.data_points.len(), 1);
4807        assert_eq!(histogram_data.temporality, Temporality::Cumulative);
4808
4809        let dp = &histogram_data.data_points[0];
4810        assert_eq!(dp.count, 4);
4811        assert_eq!(dp.sum, 53.5);
4812        assert_eq!(dp.min.unwrap(), 1.0);
4813        assert_eq!(dp.max.unwrap(), 30.0);
4814        assert_eq!(dp.attributes, vec![KeyValue::new("key1", "bound_value")]);
4815    }
4816
4817    #[cfg(feature = "experimental_metrics_bound_instruments")]
4818    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
4819    async fn bound_counter_matches_unbound() {
4820        let mut test_context = TestContext::new(Temporality::Cumulative);
4821        let counter = test_context.u64_counter("test", "my_counter", None);
4822        let attrs = vec![KeyValue::new("key1", "shared")];
4823        let bound = counter.bind(&attrs);
4824
4825        // Mix bound and unbound additions to the same attribute set
4826        counter.add(10, &attrs);
4827        bound.add(20);
4828        counter.add(30, &attrs);
4829        bound.add(40);
4830        test_context.flush_metrics();
4831
4832        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
4833            unreachable!()
4834        };
4835
4836        assert_eq!(
4837            sum.data_points.len(),
4838            1,
4839            "Bound and unbound should share the same data point"
4840        );
4841        assert_eq!(sum.data_points[0].value, 100);
4842    }
4843
4844    #[cfg(feature = "experimental_metrics_bound_instruments")]
4845    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
4846    async fn bound_counter_delta_no_update_no_export() {
4847        let mut test_context = TestContext::new(Temporality::Delta);
4848        let counter = test_context.u64_counter("test", "my_counter", None);
4849        let attrs = vec![KeyValue::new("key1", "bound_value")];
4850        let bound = counter.bind(&attrs);
4851
4852        // Cycle 1: add and collect
4853        bound.add(10);
4854        test_context.flush_metrics();
4855        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
4856            unreachable!()
4857        };
4858        assert_eq!(sum.data_points.len(), 1);
4859        assert_eq!(sum.data_points[0].value, 10);
4860
4861        // Cycle 2: no add, collect — should export nothing
4862        test_context.reset_metrics();
4863        test_context.flush_metrics();
4864        let resource_metrics = test_context
4865            .exporter
4866            .get_finished_metrics()
4867            .expect("metrics export should succeed");
4868        assert!(
4869            resource_metrics.is_empty(),
4870            "Bound handle with no updates should not export"
4871        );
4872
4873        // Cycle 3: add again — handle is still alive, produces fresh delta
4874        test_context.reset_metrics();
4875        bound.add(5);
4876        test_context.flush_metrics();
4877        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
4878            unreachable!()
4879        };
4880        assert_eq!(sum.data_points.len(), 1);
4881        assert_eq!(
4882            sum.data_points[0].value, 5,
4883            "Bound handle should produce fresh delta after quiet cycle"
4884        );
4885    }
4886
4887    #[cfg(feature = "experimental_metrics_bound_instruments")]
4888    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
4889    async fn bound_counter_at_overflow_attributes_to_overflow_bucket() {
4890        let cardinality_limit = 3;
4891        let view = move |i: &Instrument| {
4892            if i.name() == "my_counter" {
4893                Stream::builder()
4894                    .with_name("my_counter")
4895                    .with_cardinality_limit(cardinality_limit)
4896                    .build()
4897                    .ok()
4898            } else {
4899                None
4900            }
4901        };
4902        let mut test_context = TestContext::new_with_view(Temporality::Delta, view);
4903        let counter = test_context.u64_counter("test", "my_counter", None);
4904
4905        // Fill to cardinality limit with unbound calls
4906        for v in 0..cardinality_limit {
4907            counter.add(1, &[KeyValue::new("A", v.to_string())]);
4908        }
4909
4910        // bind() at overflow — handle binds directly to the overflow tracker
4911        let overflow_attrs = vec![KeyValue::new("A", "overflow_bind")];
4912        let bound = counter.bind(&overflow_attrs);
4913        bound.add(42);
4914
4915        test_context.flush_metrics();
4916        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
4917            unreachable!()
4918        };
4919
4920        // Expect: cardinality_limit unique + 1 overflow = cardinality_limit + 1
4921        assert_eq!(
4922            sum.data_points.len(),
4923            cardinality_limit + 1,
4924            "Expected {} unique + 1 overflow data points",
4925            cardinality_limit
4926        );
4927
4928        // The bound handle's value should appear in the overflow bucket
4929        let overflow_dp =
4930            find_overflow_sum_datapoint(&sum.data_points).expect("overflow point expected");
4931        assert_eq!(
4932            overflow_dp.value, 42,
4933            "Bound-at-overflow data should go to overflow bucket"
4934        );
4935    }
4936
4937    #[cfg(feature = "experimental_metrics_bound_instruments")]
4938    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
4939    async fn bound_counter_overflow_recovery_after_delta_eviction() {
4940        let cardinality_limit = 3;
4941        let view = move |i: &Instrument| {
4942            if i.name() == "my_counter" {
4943                Stream::builder()
4944                    .with_name("my_counter")
4945                    .with_cardinality_limit(cardinality_limit)
4946                    .build()
4947                    .ok()
4948            } else {
4949                None
4950            }
4951        };
4952        let mut test_context = TestContext::new_with_view(Temporality::Delta, view);
4953        let counter = test_context.u64_counter("test", "my_counter", None);
4954
4955        // Fill to cardinality limit with unbound calls (these are one-shot, not bound)
4956        for v in 0..cardinality_limit {
4957            counter.add(1, &[KeyValue::new("A", v.to_string())]);
4958        }
4959
4960        // Collect cycle 1: exports the 3 unique entries, then evicts them (no new updates)
4961        test_context.flush_metrics();
4962        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
4963            unreachable!()
4964        };
4965        assert_eq!(sum.data_points.len(), cardinality_limit);
4966
4967        // Cycle 2: no unbound adds, so the stale entries get evicted.
4968        // Space is now open. A new bind() should get a dedicated tracker.
4969        test_context.reset_metrics();
4970        test_context.flush_metrics(); // triggers eviction of stale entries
4971
4972        let new_attrs = vec![KeyValue::new("A", "recovered")];
4973        let bound = counter.bind(&new_attrs);
4974        bound.add(99);
4975
4976        test_context.reset_metrics();
4977        test_context.flush_metrics();
4978        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
4979            unreachable!()
4980        };
4981
4982        // The bound handle should have a dedicated tracker, NOT overflow
4983        assert_eq!(
4984            sum.data_points.len(),
4985            1,
4986            "Only the bound entry should be exported"
4987        );
4988        let dp = find_sum_datapoint_with_key_value(&sum.data_points, "A", "recovered")
4989            .expect("should find dedicated data point for recovered attrs");
4990        assert_eq!(
4991            dp.value, 99,
4992            "Bound handle after recovery should have dedicated tracker"
4993        );
4994        assert!(
4995            find_overflow_sum_datapoint(&sum.data_points).is_none(),
4996            "Should not have overflow — bind() after eviction should get a dedicated tracker"
4997        );
4998    }
4999
5000    #[cfg(feature = "experimental_metrics_bound_instruments")]
5001    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
5002    async fn bound_counter_multiple_overflow_handles_share_overflow_bucket() {
5003        let cardinality_limit = 2;
5004        let view = move |i: &Instrument| {
5005            if i.name() == "my_counter" {
5006                Stream::builder()
5007                    .with_name("my_counter")
5008                    .with_cardinality_limit(cardinality_limit)
5009                    .build()
5010                    .ok()
5011            } else {
5012                None
5013            }
5014        };
5015        let mut test_context = TestContext::new_with_view(Temporality::Delta, view);
5016        let counter = test_context.u64_counter("test", "my_counter", None);
5017
5018        // Fill to limit
5019        counter.add(1, &[KeyValue::new("A", "0")]);
5020        counter.add(1, &[KeyValue::new("A", "1")]);
5021
5022        // Bind two distinct attribute sets at overflow
5023        let bound_a = counter.bind(&[KeyValue::new("A", "overflow_a")]);
5024        let bound_b = counter.bind(&[KeyValue::new("A", "overflow_b")]);
5025
5026        bound_a.add(10);
5027        bound_b.add(20);
5028        bound_a.add(5);
5029
5030        test_context.flush_metrics();
5031        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
5032            unreachable!()
5033        };
5034
5035        let overflow_dp =
5036            find_overflow_sum_datapoint(&sum.data_points).expect("overflow point expected");
5037        assert_eq!(
5038            overflow_dp.value, 35,
5039            "All overflow-bound measurements should accumulate in overflow bucket"
5040        );
5041
5042        // Cycle 2: bound handles still work after delta collect
5043        test_context.reset_metrics();
5044        bound_a.add(7);
5045        bound_b.add(3);
5046        test_context.flush_metrics();
5047
5048        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
5049            unreachable!()
5050        };
5051
5052        let overflow_dp =
5053            find_overflow_sum_datapoint(&sum.data_points).expect("overflow point expected");
5054        assert_eq!(
5055            overflow_dp.value, 10,
5056            "Overflow-bound handles should continue working across delta cycles"
5057        );
5058    }
5059
5060    #[cfg(feature = "experimental_metrics_bound_instruments")]
5061    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
5062    async fn bound_counter_overflow_persists_across_eviction_cycles() {
5063        // Once a bind() lands in overflow, the handle's writes must continue
5064        // landing in overflow for the lifetime of the handle — even after
5065        // delta eviction frees space. This is the predictability guarantee:
5066        // a user inspecting their data should see the bound handle's
5067        // attribution as a single, stable bucket. The recovery story is
5068        // explicit (drop and re-bind), not implicit (silent self-healing).
5069        let cardinality_limit = 3;
5070        let view = move |i: &Instrument| {
5071            if i.name() == "my_counter" {
5072                Stream::builder()
5073                    .with_name("my_counter")
5074                    .with_cardinality_limit(cardinality_limit)
5075                    .build()
5076                    .ok()
5077            } else {
5078                None
5079            }
5080        };
5081        let mut test_context = TestContext::new_with_view(Temporality::Delta, view);
5082        let counter = test_context.u64_counter("test", "my_counter", None);
5083
5084        // Cycle 1: fill cardinality with unbound calls, then bind at overflow.
5085        for v in 0..cardinality_limit {
5086            counter.add(1, &[KeyValue::new("A", v.to_string())]);
5087        }
5088        let stuck_attrs = vec![KeyValue::new("A", "stuck_in_overflow")];
5089        let bound = counter.bind(&stuck_attrs);
5090        bound.add(10);
5091
5092        test_context.flush_metrics();
5093        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
5094            unreachable!()
5095        };
5096        let overflow_dp = find_overflow_sum_datapoint(&sum.data_points)
5097            .expect("cycle 1: bound write at overflow should land in overflow bucket");
5098        assert_eq!(overflow_dp.value, 10);
5099
5100        // Cycle 2: no calls. The 3 unbound entries become stale and are evicted,
5101        // freeing all of the cardinality budget.
5102        test_context.reset_metrics();
5103        test_context.flush_metrics();
5104
5105        // Cycle 3: the SAME bound handle is used again. Even though space is
5106        // available, its writes must still land in overflow — the handle is
5107        // permanently bound to overflow, not silently re-resolved.
5108        test_context.reset_metrics();
5109        bound.add(99);
5110        test_context.flush_metrics();
5111        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
5112            unreachable!()
5113        };
5114        let overflow_dp = find_overflow_sum_datapoint(&sum.data_points)
5115            .expect("cycle 3: bound write must still land in overflow even after space frees up");
5116        assert_eq!(
5117            overflow_dp.value, 99,
5118            "Bound-at-overflow handle must keep writing to overflow even after delta eviction"
5119        );
5120        assert!(
5121            find_sum_datapoint_with_key_value(&sum.data_points, "A", "stuck_in_overflow").is_none(),
5122            "Bound-at-overflow handle must not silently self-heal to a dedicated tracker"
5123        );
5124    }
5125
5126    #[cfg(feature = "experimental_metrics_bound_instruments")]
5127    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
5128    async fn bound_histogram_delta() {
5129        let mut test_context = TestContext::new(Temporality::Delta);
5130        let histogram = test_context
5131            .meter()
5132            .f64_histogram("my_histogram")
5133            .with_boundaries(vec![5.0, 10.0, 25.0, 50.0])
5134            .build();
5135        let attrs = vec![KeyValue::new("key1", "bound_value")];
5136        let bound = histogram.bind(&attrs);
5137
5138        // Cycle 1: record and collect
5139        bound.record(3.0);
5140        bound.record(12.0);
5141        test_context.flush_metrics();
5142
5143        let MetricData::Histogram(hist) = test_context.get_aggregation::<f64>("my_histogram", None)
5144        else {
5145            unreachable!()
5146        };
5147        assert_eq!(hist.temporality, Temporality::Delta);
5148        assert_eq!(hist.data_points.len(), 1);
5149        assert_eq!(hist.data_points[0].count, 2);
5150        assert_eq!(hist.data_points[0].sum, 15.0);
5151
5152        // Cycle 2: delta resets, new values
5153        test_context.reset_metrics();
5154        bound.record(40.0);
5155        test_context.flush_metrics();
5156
5157        let MetricData::Histogram(hist) = test_context.get_aggregation::<f64>("my_histogram", None)
5158        else {
5159            unreachable!()
5160        };
5161        assert_eq!(hist.data_points.len(), 1);
5162        assert_eq!(
5163            hist.data_points[0].count, 1,
5164            "Delta should reset count between collections"
5165        );
5166        assert_eq!(
5167            hist.data_points[0].sum, 40.0,
5168            "Delta should reset sum between collections"
5169        );
5170    }
5171
5172    #[cfg(feature = "experimental_metrics_bound_instruments")]
5173    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
5174    async fn bound_histogram_matches_unbound() {
5175        let mut test_context = TestContext::new(Temporality::Cumulative);
5176        let histogram = test_context
5177            .meter()
5178            .f64_histogram("my_histogram")
5179            .with_boundaries(vec![10.0, 50.0])
5180            .build();
5181        let attrs = vec![KeyValue::new("key1", "shared")];
5182        let bound = histogram.bind(&attrs);
5183
5184        // Mix bound and unbound recordings to the same attribute set
5185        histogram.record(5.0, &attrs);
5186        bound.record(15.0);
5187        histogram.record(25.0, &attrs);
5188        bound.record(35.0);
5189        test_context.flush_metrics();
5190
5191        let MetricData::Histogram(hist) = test_context.get_aggregation::<f64>("my_histogram", None)
5192        else {
5193            unreachable!()
5194        };
5195
5196        assert_eq!(
5197            hist.data_points.len(),
5198            1,
5199            "Bound and unbound should share the same data point"
5200        );
5201        assert_eq!(hist.data_points[0].count, 4);
5202        assert_eq!(hist.data_points[0].sum, 80.0);
5203        assert_eq!(hist.data_points[0].min.unwrap(), 5.0);
5204        assert_eq!(hist.data_points[0].max.unwrap(), 35.0);
5205    }
5206
5207    #[cfg(feature = "experimental_metrics_bound_instruments")]
5208    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
5209    async fn bound_histogram_delta_no_update_no_export() {
5210        let mut test_context = TestContext::new(Temporality::Delta);
5211        let histogram = test_context
5212            .meter()
5213            .f64_histogram("my_histogram")
5214            .with_boundaries(vec![10.0])
5215            .build();
5216        let attrs = vec![KeyValue::new("key1", "bound_value")];
5217        let bound = histogram.bind(&attrs);
5218
5219        // Cycle 1: record and collect
5220        bound.record(5.0);
5221        test_context.flush_metrics();
5222        let MetricData::Histogram(hist) = test_context.get_aggregation::<f64>("my_histogram", None)
5223        else {
5224            unreachable!()
5225        };
5226        assert_eq!(hist.data_points.len(), 1);
5227        assert_eq!(hist.data_points[0].count, 1);
5228
5229        // Cycle 2: no recordings — should export nothing
5230        test_context.reset_metrics();
5231        test_context.flush_metrics();
5232        let resource_metrics = test_context
5233            .exporter
5234            .get_finished_metrics()
5235            .expect("metrics export should succeed");
5236        assert!(
5237            resource_metrics.is_empty(),
5238            "Bound histogram with no updates should not export"
5239        );
5240
5241        // Cycle 3: record again — handle still alive
5242        test_context.reset_metrics();
5243        bound.record(20.0);
5244        test_context.flush_metrics();
5245        let MetricData::Histogram(hist) = test_context.get_aggregation::<f64>("my_histogram", None)
5246        else {
5247            unreachable!()
5248        };
5249        assert_eq!(hist.data_points.len(), 1);
5250        assert_eq!(
5251            hist.data_points[0].count, 1,
5252            "Fresh delta after quiet cycle"
5253        );
5254        assert_eq!(hist.data_points[0].sum, 20.0);
5255    }
5256
5257    #[cfg(feature = "experimental_metrics_bound_instruments")]
5258    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
5259    async fn bound_histogram_at_overflow_attributes_to_overflow_bucket() {
5260        let cardinality_limit = 3;
5261        let view = move |i: &Instrument| {
5262            if i.name() == "my_histogram" {
5263                Stream::builder()
5264                    .with_name("my_histogram")
5265                    .with_cardinality_limit(cardinality_limit)
5266                    .build()
5267                    .ok()
5268            } else {
5269                None
5270            }
5271        };
5272        let mut test_context = TestContext::new_with_view(Temporality::Delta, view);
5273        let histogram = test_context
5274            .meter()
5275            .f64_histogram("my_histogram")
5276            .with_boundaries(vec![10.0, 50.0])
5277            .build();
5278
5279        // Fill to cardinality limit with unbound calls
5280        for v in 0..cardinality_limit {
5281            histogram.record(1.0, &[KeyValue::new("A", v.to_string())]);
5282        }
5283
5284        // bind() at overflow — handle binds directly to the overflow tracker
5285        let overflow_attrs = vec![KeyValue::new("A", "overflow_bind")];
5286        let bound = histogram.bind(&overflow_attrs);
5287        bound.record(42.0);
5288
5289        test_context.flush_metrics();
5290        let MetricData::Histogram(hist) = test_context.get_aggregation::<f64>("my_histogram", None)
5291        else {
5292            unreachable!()
5293        };
5294
5295        assert_eq!(
5296            hist.data_points.len(),
5297            cardinality_limit + 1,
5298            "Expected {} unique + 1 overflow data points",
5299            cardinality_limit
5300        );
5301
5302        let overflow_dp =
5303            find_overflow_histogram_datapoint(&hist.data_points).expect("overflow point expected");
5304        assert_eq!(
5305            overflow_dp.sum, 42.0,
5306            "Bound-at-overflow data should go to overflow bucket"
5307        );
5308        assert_eq!(overflow_dp.count, 1);
5309    }
5310
5311    #[cfg(feature = "experimental_metrics_bound_instruments")]
5312    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
5313    async fn bound_histogram_overflow_persists_across_eviction_cycles() {
5314        // Mirror of bound_counter_overflow_persists_across_eviction_cycles for
5315        // histograms: a bound-at-overflow handle must keep landing in overflow
5316        // even after delta eviction frees space, for the lifetime of the handle.
5317        let cardinality_limit = 3;
5318        let view = move |i: &Instrument| {
5319            if i.name() == "my_histogram" {
5320                Stream::builder()
5321                    .with_name("my_histogram")
5322                    .with_cardinality_limit(cardinality_limit)
5323                    .build()
5324                    .ok()
5325            } else {
5326                None
5327            }
5328        };
5329        let mut test_context = TestContext::new_with_view(Temporality::Delta, view);
5330        let histogram = test_context
5331            .meter()
5332            .f64_histogram("my_histogram")
5333            .with_boundaries(vec![10.0, 50.0])
5334            .build();
5335
5336        // Cycle 1: fill cardinality with unbound calls, then bind at overflow.
5337        for v in 0..cardinality_limit {
5338            histogram.record(1.0, &[KeyValue::new("A", v.to_string())]);
5339        }
5340        let stuck_attrs = vec![KeyValue::new("A", "stuck_in_overflow")];
5341        let bound = histogram.bind(&stuck_attrs);
5342        bound.record(15.0);
5343
5344        test_context.flush_metrics();
5345        let MetricData::Histogram(hist) = test_context.get_aggregation::<f64>("my_histogram", None)
5346        else {
5347            unreachable!()
5348        };
5349        let overflow_dp = find_overflow_histogram_datapoint(&hist.data_points)
5350            .expect("cycle 1: bound write at overflow should land in overflow bucket");
5351        assert_eq!(overflow_dp.sum, 15.0);
5352
5353        // Cycle 2: no calls. Stale unbound entries get evicted, freeing space.
5354        test_context.reset_metrics();
5355        test_context.flush_metrics();
5356
5357        // Cycle 3: same bound handle. Even though space is free, writes must
5358        // still land in overflow.
5359        test_context.reset_metrics();
5360        bound.record(99.0);
5361        test_context.flush_metrics();
5362        let MetricData::Histogram(hist) = test_context.get_aggregation::<f64>("my_histogram", None)
5363        else {
5364            unreachable!()
5365        };
5366        let overflow_dp = find_overflow_histogram_datapoint(&hist.data_points)
5367            .expect("cycle 3: bound write must still land in overflow even after space frees up");
5368        assert_eq!(
5369            overflow_dp.sum, 99.0,
5370            "Bound-at-overflow histogram must keep writing to overflow even after delta eviction"
5371        );
5372        assert_eq!(overflow_dp.count, 1);
5373    }
5374
5375    #[cfg(feature = "experimental_metrics_bound_instruments")]
5376    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
5377    async fn bound_exponential_histogram_delta() {
5378        // Histogram configured with Base2ExponentialHistogram aggregation goes
5379        // through ExpoHistogram internally. Verify bind() returns a handle
5380        // whose direct writes accumulate correctly.
5381        let view = |i: &Instrument| {
5382            if i.name() == "my_histogram" {
5383                Stream::builder()
5384                    .with_aggregation(Aggregation::Base2ExponentialHistogram {
5385                        max_size: 160,
5386                        max_scale: 20,
5387                        record_min_max: true,
5388                    })
5389                    .build()
5390                    .ok()
5391            } else {
5392                None
5393            }
5394        };
5395        let mut test_context = TestContext::new_with_view(Temporality::Delta, view);
5396        let histogram = test_context.meter().f64_histogram("my_histogram").build();
5397        let attrs = vec![KeyValue::new("key1", "bound_value")];
5398        let bound = histogram.bind(&attrs);
5399
5400        bound.record(2.0);
5401        bound.record(4.0);
5402        bound.record(8.0);
5403        // NaN/inf must be filtered just like the unbound path.
5404        bound.record(f64::NAN);
5405        bound.record(f64::INFINITY);
5406
5407        test_context.flush_metrics();
5408        let MetricData::ExponentialHistogram(hist) =
5409            test_context.get_aggregation::<f64>("my_histogram", None)
5410        else {
5411            panic!("expected ExponentialHistogram aggregation");
5412        };
5413        assert_eq!(hist.data_points.len(), 1);
5414        let dp = &hist.data_points[0];
5415        assert_eq!(dp.count, 3, "NaN and infinity should be dropped");
5416        assert_eq!(dp.sum, 14.0);
5417        assert_eq!(dp.min, Some(2.0));
5418        assert_eq!(dp.max, Some(8.0));
5419    }
5420
5421    #[cfg(feature = "experimental_metrics_bound_instruments")]
5422    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
5423    async fn bound_exponential_histogram_at_overflow_attributes_to_overflow_bucket() {
5424        let cardinality_limit = 3;
5425        let view = move |i: &Instrument| {
5426            if i.name() == "my_histogram" {
5427                Stream::builder()
5428                    .with_aggregation(Aggregation::Base2ExponentialHistogram {
5429                        max_size: 160,
5430                        max_scale: 20,
5431                        record_min_max: true,
5432                    })
5433                    .with_cardinality_limit(cardinality_limit)
5434                    .build()
5435                    .ok()
5436            } else {
5437                None
5438            }
5439        };
5440        let mut test_context = TestContext::new_with_view(Temporality::Delta, view);
5441        let histogram = test_context.meter().f64_histogram("my_histogram").build();
5442
5443        for v in 0..cardinality_limit {
5444            histogram.record(1.0, &[KeyValue::new("A", v.to_string())]);
5445        }
5446
5447        // bind() at overflow — handle binds directly to the overflow tracker
5448        let overflow_attrs = vec![KeyValue::new("A", "overflow_bind")];
5449        let bound = histogram.bind(&overflow_attrs);
5450        bound.record(42.0);
5451
5452        test_context.flush_metrics();
5453        let MetricData::ExponentialHistogram(hist) =
5454            test_context.get_aggregation::<f64>("my_histogram", None)
5455        else {
5456            panic!("expected ExponentialHistogram aggregation");
5457        };
5458        let overflow_dp = find_overflow_exponential_histogram_datapoint(&hist.data_points)
5459            .expect("overflow point expected");
5460        assert_eq!(
5461            overflow_dp.sum, 42.0,
5462            "Bound-at-overflow data should go to overflow bucket"
5463        );
5464        assert_eq!(overflow_dp.count, 1);
5465    }
5466
5467    #[cfg(feature = "experimental_metrics_bound_instruments")]
5468    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
5469    async fn bound_exponential_histogram_overflow_persists_across_eviction_cycles() {
5470        // Same predictability invariant the counter/histogram tests assert,
5471        // verified for the exponential histogram aggregator path.
5472        let cardinality_limit = 3;
5473        let view = move |i: &Instrument| {
5474            if i.name() == "my_histogram" {
5475                Stream::builder()
5476                    .with_aggregation(Aggregation::Base2ExponentialHistogram {
5477                        max_size: 160,
5478                        max_scale: 20,
5479                        record_min_max: true,
5480                    })
5481                    .with_cardinality_limit(cardinality_limit)
5482                    .build()
5483                    .ok()
5484            } else {
5485                None
5486            }
5487        };
5488        let mut test_context = TestContext::new_with_view(Temporality::Delta, view);
5489        let histogram = test_context.meter().f64_histogram("my_histogram").build();
5490
5491        for v in 0..cardinality_limit {
5492            histogram.record(1.0, &[KeyValue::new("A", v.to_string())]);
5493        }
5494        let stuck_attrs = vec![KeyValue::new("A", "stuck_in_overflow")];
5495        let bound = histogram.bind(&stuck_attrs);
5496        bound.record(15.0);
5497
5498        test_context.flush_metrics();
5499        let MetricData::ExponentialHistogram(hist) =
5500            test_context.get_aggregation::<f64>("my_histogram", None)
5501        else {
5502            panic!("expected ExponentialHistogram aggregation");
5503        };
5504        let overflow_dp = find_overflow_exponential_histogram_datapoint(&hist.data_points)
5505            .expect("cycle 1: bound write at overflow should land in overflow bucket");
5506        assert_eq!(overflow_dp.sum, 15.0);
5507
5508        // Cycle 2: evict stale entries, freeing space.
5509        test_context.reset_metrics();
5510        test_context.flush_metrics();
5511
5512        // Cycle 3: bound handle keeps writing to overflow.
5513        test_context.reset_metrics();
5514        bound.record(99.0);
5515        test_context.flush_metrics();
5516        let MetricData::ExponentialHistogram(hist) =
5517            test_context.get_aggregation::<f64>("my_histogram", None)
5518        else {
5519            panic!("expected ExponentialHistogram aggregation");
5520        };
5521        let overflow_dp = find_overflow_exponential_histogram_datapoint(&hist.data_points)
5522            .expect("cycle 3: bound write must still land in overflow even after space frees up");
5523        assert_eq!(
5524            overflow_dp.sum, 99.0,
5525            "Bound-at-overflow ExpoHistogram must keep writing to overflow even after delta eviction"
5526        );
5527        assert_eq!(overflow_dp.count, 1);
5528    }
5529
5530    #[cfg(feature = "experimental_metrics_bound_instruments")]
5531    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
5532    async fn bound_counter_drop_enables_eviction() {
5533        let mut test_context = TestContext::new(Temporality::Delta);
5534        let counter = test_context.u64_counter("test", "my_counter", None);
5535        let attrs = vec![KeyValue::new("key1", "ephemeral")];
5536
5537        {
5538            let bound = counter.bind(&attrs);
5539            bound.add(100);
5540            test_context.flush_metrics();
5541
5542            let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None)
5543            else {
5544                unreachable!()
5545            };
5546            assert_eq!(sum.data_points.len(), 1);
5547            assert_eq!(sum.data_points[0].value, 100);
5548            // bound drops here
5549        }
5550
5551        // Cycle 2: no updates, bound handle dropped — entry becomes stale and evictable
5552        test_context.reset_metrics();
5553        test_context.flush_metrics();
5554
5555        // Cycle 3: the stale entry should have been evicted, so a new unbound add
5556        // should be the only data point
5557        test_context.reset_metrics();
5558        counter.add(1, &[KeyValue::new("key1", "new_entry")]);
5559        test_context.flush_metrics();
5560
5561        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
5562            unreachable!()
5563        };
5564
5565        // Only the new entry should be present — the old "ephemeral" entry was evicted
5566        assert_eq!(sum.data_points.len(), 1);
5567        let dp = find_sum_datapoint_with_key_value(&sum.data_points, "key1", "new_entry")
5568            .expect("new_entry should be present");
5569        assert_eq!(dp.value, 1);
5570        assert!(
5571            find_sum_datapoint_with_key_value(&sum.data_points, "key1", "ephemeral").is_none(),
5572            "Dropped bound entry should have been evicted"
5573        );
5574    }
5575
5576    #[cfg(feature = "experimental_metrics_bound_instruments")]
5577    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
5578    async fn bound_counter_multiple_handles_same_attrs() {
5579        let mut test_context = TestContext::new(Temporality::Delta);
5580        let counter = test_context.u64_counter("test", "my_counter", None);
5581        let attrs = vec![KeyValue::new("key1", "shared")];
5582
5583        let bound1 = counter.bind(&attrs);
5584        let bound2 = counter.bind(&attrs);
5585
5586        bound1.add(10);
5587        bound2.add(20);
5588        test_context.flush_metrics();
5589
5590        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
5591            unreachable!()
5592        };
5593        assert_eq!(
5594            sum.data_points.len(),
5595            1,
5596            "Multiple handles to same attrs should share data point"
5597        );
5598        assert_eq!(sum.data_points[0].value, 30);
5599
5600        // Drop one handle — entry should NOT be evicted
5601        drop(bound1);
5602        test_context.reset_metrics();
5603        test_context.flush_metrics(); // idle cycle, but bound2 still holds it
5604
5605        // bound2 still works
5606        test_context.reset_metrics();
5607        bound2.add(5);
5608        test_context.flush_metrics();
5609
5610        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
5611            unreachable!()
5612        };
5613        assert_eq!(sum.data_points.len(), 1);
5614        assert_eq!(
5615            sum.data_points[0].value, 5,
5616            "Entry should persist while any handle is alive"
5617        );
5618    }
5619
5620    #[cfg(feature = "experimental_metrics_bound_instruments")]
5621    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
5622    async fn bound_counter_empty_attributes() {
5623        let mut test_context = TestContext::new(Temporality::Cumulative);
5624        let counter = test_context.u64_counter("test", "my_counter", None);
5625        let bound = counter.bind(&[]);
5626
5627        bound.add(10);
5628        bound.add(30);
5629        test_context.flush_metrics();
5630
5631        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
5632            unreachable!()
5633        };
5634
5635        assert_eq!(sum.data_points.len(), 1);
5636        assert_eq!(sum.data_points[0].value, 40);
5637    }
5638
5639    #[cfg(feature = "experimental_metrics_bound_instruments")]
5640    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
5641    async fn bound_counter_empty_attributes_shares_with_unbound() {
5642        let mut test_context = TestContext::new(Temporality::Cumulative);
5643        let counter = test_context.u64_counter("test", "my_counter", None);
5644        let bound = counter.bind(&[]);
5645
5646        // Mix bound and unbound calls with empty attributes — they must share
5647        // the same data point (both route to no_attribute_tracker).
5648        counter.add(10, &[]);
5649        bound.add(20);
5650        counter.add(30, &[]);
5651        bound.add(40);
5652        test_context.flush_metrics();
5653
5654        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
5655            unreachable!()
5656        };
5657
5658        assert_eq!(
5659            sum.data_points.len(),
5660            1,
5661            "Bound and unbound with empty attributes must share the same data point"
5662        );
5663        assert_eq!(sum.data_points[0].value, 100);
5664    }
5665
5666    #[cfg(feature = "experimental_metrics_bound_instruments")]
5667    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
5668    async fn bound_histogram_empty_attributes_shares_with_unbound() {
5669        let mut test_context = TestContext::new(Temporality::Cumulative);
5670        let histogram = test_context
5671            .meter()
5672            .u64_histogram("my_histogram")
5673            .with_boundaries(vec![5.0, 10.0, 25.0])
5674            .build();
5675        let bound = histogram.bind(&[]);
5676
5677        histogram.record(3, &[]);
5678        bound.record(7);
5679        histogram.record(20, &[]);
5680        test_context.flush_metrics();
5681
5682        let MetricData::Histogram(hist) = test_context.get_aggregation::<u64>("my_histogram", None)
5683        else {
5684            unreachable!()
5685        };
5686
5687        assert_eq!(
5688            hist.data_points.len(),
5689            1,
5690            "Bound and unbound with empty attributes must share the same data point"
5691        );
5692        let dp = &hist.data_points[0];
5693        assert!(dp.attributes.is_empty());
5694        assert_eq!(dp.count, 3);
5695        assert_eq!(dp.sum, 30);
5696    }
5697
5698    #[cfg(feature = "experimental_metrics_bound_instruments")]
5699    #[cfg(feature = "spec_unstable_metrics_views")]
5700    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
5701    async fn bound_counter_view_filters_attributes_at_bind_time() {
5702        use opentelemetry::Key;
5703
5704        let exporter = InMemoryMetricExporter::default();
5705        let view = |i: &Instrument| {
5706            if i.name() == "my_counter" {
5707                Stream::builder()
5708                    .with_allowed_attribute_keys(vec![Key::new("k1"), Key::new("k2")])
5709                    .build()
5710                    .ok()
5711            } else {
5712                None
5713            }
5714        };
5715        let meter_provider = SdkMeterProvider::builder()
5716            .with_periodic_exporter(exporter.clone())
5717            .with_view(view)
5718            .build();
5719        let meter = meter_provider.meter("test");
5720        let counter = meter.u64_counter("my_counter").build();
5721
5722        // bind with k3 included — view should drop it at bind time
5723        let bound = counter.bind(&[
5724            KeyValue::new("k1", "v1"),
5725            KeyValue::new("k2", "v2"),
5726            KeyValue::new("k3", "v3"),
5727        ]);
5728        bound.add(10);
5729        bound.add(20);
5730
5731        // unbound call with a *different* k3 value: after view filtering both
5732        // bound and unbound must collapse into the same data point.
5733        counter.add(
5734            7,
5735            &[
5736                KeyValue::new("k1", "v1"),
5737                KeyValue::new("k2", "v2"),
5738                KeyValue::new("k3", "different"),
5739            ],
5740        );
5741
5742        meter_provider.force_flush().unwrap();
5743        let resource_metrics = exporter
5744            .get_finished_metrics()
5745            .expect("metrics are expected to be exported.");
5746        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
5747        let data::AggregatedMetrics::U64(MetricData::Sum(sum)) = &metric.data else {
5748            unreachable!()
5749        };
5750
5751        assert_eq!(
5752            sum.data_points.len(),
5753            1,
5754            "view should filter k3, leaving bound+unbound to aggregate together"
5755        );
5756        assert_eq!(sum.data_points[0].value, 37);
5757        let attrs = &sum.data_points[0].attributes;
5758        assert_eq!(attrs.len(), 2);
5759        assert!(attrs.iter().any(|kv| kv.key.as_str() == "k1"));
5760        assert!(attrs.iter().any(|kv| kv.key.as_str() == "k2"));
5761        assert!(!attrs.iter().any(|kv| kv.key.as_str() == "k3"));
5762    }
5763
5764    #[cfg(feature = "experimental_metrics_bound_instruments")]
5765    #[cfg(feature = "spec_unstable_metrics_views")]
5766    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
5767    async fn bound_histogram_view_filters_attributes_at_bind_time() {
5768        use opentelemetry::Key;
5769
5770        let exporter = InMemoryMetricExporter::default();
5771        let view = |i: &Instrument| {
5772            if i.name() == "my_hist" {
5773                Stream::builder()
5774                    .with_allowed_attribute_keys(vec![Key::new("k1"), Key::new("k2")])
5775                    .build()
5776                    .ok()
5777            } else {
5778                None
5779            }
5780        };
5781        let meter_provider = SdkMeterProvider::builder()
5782            .with_periodic_exporter(exporter.clone())
5783            .with_view(view)
5784            .build();
5785        let meter = meter_provider.meter("test");
5786        let histogram = meter
5787            .u64_histogram("my_hist")
5788            .with_boundaries(vec![5.0, 10.0, 25.0])
5789            .build();
5790
5791        let bound = histogram.bind(&[
5792            KeyValue::new("k1", "v1"),
5793            KeyValue::new("k2", "v2"),
5794            KeyValue::new("k3", "v3"),
5795        ]);
5796        bound.record(3);
5797        bound.record(20);
5798        histogram.record(
5799            7,
5800            &[
5801                KeyValue::new("k1", "v1"),
5802                KeyValue::new("k2", "v2"),
5803                KeyValue::new("k3", "different"),
5804            ],
5805        );
5806
5807        meter_provider.force_flush().unwrap();
5808        let resource_metrics = exporter
5809            .get_finished_metrics()
5810            .expect("metrics are expected to be exported.");
5811        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
5812        let data::AggregatedMetrics::U64(MetricData::Histogram(hist)) = &metric.data else {
5813            unreachable!()
5814        };
5815
5816        assert_eq!(
5817            hist.data_points.len(),
5818            1,
5819            "view should filter k3, leaving bound+unbound to aggregate together"
5820        );
5821        let dp = &hist.data_points[0];
5822        assert_eq!(dp.count, 3);
5823        assert_eq!(dp.sum, 30);
5824        assert_eq!(dp.attributes.len(), 2);
5825        assert!(dp.attributes.iter().any(|kv| kv.key.as_str() == "k1"));
5826        assert!(dp.attributes.iter().any(|kv| kv.key.as_str() == "k2"));
5827        assert!(!dp.attributes.iter().any(|kv| kv.key.as_str() == "k3"));
5828    }
5829
5830    #[cfg(feature = "experimental_metrics_bound_instruments")]
5831    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
5832    async fn bound_counter_at_overflow_attributes_to_overflow_bucket_cumulative() {
5833        // Cumulative: cardinality only grows, never evicts. A bind() at the
5834        // limit lands in overflow and accumulates there forever. Verifies the
5835        // bound handle's cumulative writes converge in the overflow bucket
5836        // across multiple collection cycles.
5837        let cardinality_limit = 3;
5838        let view = move |i: &Instrument| {
5839            if i.name() == "my_counter" {
5840                Stream::builder()
5841                    .with_name("my_counter")
5842                    .with_cardinality_limit(cardinality_limit)
5843                    .build()
5844                    .ok()
5845            } else {
5846                None
5847            }
5848        };
5849        let mut test_context = TestContext::new_with_view(Temporality::Cumulative, view);
5850        let counter = test_context.u64_counter("test", "my_counter", None);
5851
5852        for v in 0..cardinality_limit {
5853            counter.add(1, &[KeyValue::new("A", v.to_string())]);
5854        }
5855        let bound = counter.bind(&[KeyValue::new("A", "overflow_bind")]);
5856        bound.add(10);
5857
5858        test_context.flush_metrics();
5859        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
5860            unreachable!()
5861        };
5862        let overflow_dp = find_overflow_sum_datapoint(&sum.data_points)
5863            .expect("cycle 1: overflow point expected");
5864        assert_eq!(overflow_dp.value, 10);
5865
5866        // Cycle 2: cumulative state accumulates internally; reset the exporter
5867        // so the assertion sees a single export rather than two appended ones.
5868        test_context.reset_metrics();
5869        bound.add(7);
5870        test_context.flush_metrics();
5871        let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
5872            unreachable!()
5873        };
5874        let overflow_dp = find_overflow_sum_datapoint(&sum.data_points)
5875            .expect("cycle 2: overflow point expected");
5876        assert_eq!(
5877            overflow_dp.value, 17,
5878            "cumulative overflow-bound writes must accumulate"
5879        );
5880    }
5881
5882    #[cfg(feature = "experimental_metrics_bound_instruments")]
5883    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
5884    async fn bound_histogram_at_overflow_attributes_to_overflow_bucket_cumulative() {
5885        let cardinality_limit = 3;
5886        let view = move |i: &Instrument| {
5887            if i.name() == "my_histogram" {
5888                Stream::builder()
5889                    .with_name("my_histogram")
5890                    .with_cardinality_limit(cardinality_limit)
5891                    .build()
5892                    .ok()
5893            } else {
5894                None
5895            }
5896        };
5897        let mut test_context = TestContext::new_with_view(Temporality::Cumulative, view);
5898        let histogram = test_context
5899            .meter()
5900            .f64_histogram("my_histogram")
5901            .with_boundaries(vec![10.0, 50.0])
5902            .build();
5903
5904        for v in 0..cardinality_limit {
5905            histogram.record(1.0, &[KeyValue::new("A", v.to_string())]);
5906        }
5907        let bound = histogram.bind(&[KeyValue::new("A", "overflow_bind")]);
5908        bound.record(20.0);
5909
5910        test_context.flush_metrics();
5911        let MetricData::Histogram(hist) = test_context.get_aggregation::<f64>("my_histogram", None)
5912        else {
5913            unreachable!()
5914        };
5915        let overflow_dp = find_overflow_histogram_datapoint(&hist.data_points)
5916            .expect("cycle 1: overflow point expected");
5917        assert_eq!(overflow_dp.sum, 20.0);
5918        assert_eq!(overflow_dp.count, 1);
5919
5920        // Cycle 2: cumulative accumulates internally; reset exporter to see a
5921        // single fresh export rather than two appended ones.
5922        test_context.reset_metrics();
5923        bound.record(30.0);
5924        test_context.flush_metrics();
5925        let MetricData::Histogram(hist) = test_context.get_aggregation::<f64>("my_histogram", None)
5926        else {
5927            unreachable!()
5928        };
5929        let overflow_dp = find_overflow_histogram_datapoint(&hist.data_points)
5930            .expect("cycle 2: overflow point expected");
5931        assert_eq!(
5932            overflow_dp.sum, 50.0,
5933            "cumulative overflow-bound writes must accumulate"
5934        );
5935        assert_eq!(overflow_dp.count, 2);
5936    }
5937
5938    #[cfg(feature = "experimental_metrics_bound_instruments")]
5939    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
5940    async fn bound_exponential_histogram_at_overflow_attributes_to_overflow_bucket_cumulative() {
5941        let cardinality_limit = 3;
5942        let view = move |i: &Instrument| {
5943            if i.name() == "my_histogram" {
5944                Stream::builder()
5945                    .with_aggregation(Aggregation::Base2ExponentialHistogram {
5946                        max_size: 160,
5947                        max_scale: 20,
5948                        record_min_max: true,
5949                    })
5950                    .with_cardinality_limit(cardinality_limit)
5951                    .build()
5952                    .ok()
5953            } else {
5954                None
5955            }
5956        };
5957        let mut test_context = TestContext::new_with_view(Temporality::Cumulative, view);
5958        let histogram = test_context.meter().f64_histogram("my_histogram").build();
5959
5960        for v in 0..cardinality_limit {
5961            histogram.record(1.0, &[KeyValue::new("A", v.to_string())]);
5962        }
5963        let bound = histogram.bind(&[KeyValue::new("A", "overflow_bind")]);
5964        bound.record(20.0);
5965
5966        test_context.flush_metrics();
5967        let MetricData::ExponentialHistogram(hist) =
5968            test_context.get_aggregation::<f64>("my_histogram", None)
5969        else {
5970            panic!("expected ExponentialHistogram aggregation");
5971        };
5972        let overflow_dp = find_overflow_exponential_histogram_datapoint(&hist.data_points)
5973            .expect("cycle 1: overflow point expected");
5974        assert_eq!(overflow_dp.sum, 20.0);
5975        assert_eq!(overflow_dp.count, 1);
5976
5977        // Cycle 2: cumulative accumulates internally; reset exporter to see a
5978        // single fresh export rather than two appended ones.
5979        test_context.reset_metrics();
5980        bound.record(30.0);
5981        test_context.flush_metrics();
5982        let MetricData::ExponentialHistogram(hist) =
5983            test_context.get_aggregation::<f64>("my_histogram", None)
5984        else {
5985            panic!("expected ExponentialHistogram aggregation");
5986        };
5987        let overflow_dp = find_overflow_exponential_histogram_datapoint(&hist.data_points)
5988            .expect("cycle 2: overflow point expected");
5989        assert_eq!(
5990            overflow_dp.sum, 50.0,
5991            "cumulative overflow-bound writes must accumulate"
5992        );
5993        assert_eq!(overflow_dp.count, 2);
5994    }
5995
5996    #[cfg(feature = "experimental_metrics_bound_instruments")]
5997    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
5998    async fn bound_exponential_histogram_view_filters_attributes_at_bind_time() {
5999        use opentelemetry::Key;
6000
6001        let exporter = InMemoryMetricExporter::default();
6002        let view = |i: &Instrument| {
6003            if i.name() == "my_hist" {
6004                Stream::builder()
6005                    .with_aggregation(Aggregation::Base2ExponentialHistogram {
6006                        max_size: 160,
6007                        max_scale: 20,
6008                        record_min_max: true,
6009                    })
6010                    .with_allowed_attribute_keys(vec![Key::new("k1"), Key::new("k2")])
6011                    .build()
6012                    .ok()
6013            } else {
6014                None
6015            }
6016        };
6017        let meter_provider = SdkMeterProvider::builder()
6018            .with_periodic_exporter(exporter.clone())
6019            .with_view(view)
6020            .build();
6021        let meter = meter_provider.meter("test");
6022        let histogram = meter.f64_histogram("my_hist").build();
6023
6024        let bound = histogram.bind(&[
6025            KeyValue::new("k1", "v1"),
6026            KeyValue::new("k2", "v2"),
6027            KeyValue::new("k3", "v3"),
6028        ]);
6029        bound.record(3.0);
6030        bound.record(20.0);
6031        // Unbound call with a different k3: after view filtering, bound and unbound
6032        // must collapse into the same exponential histogram data point.
6033        histogram.record(
6034            7.0,
6035            &[
6036                KeyValue::new("k1", "v1"),
6037                KeyValue::new("k2", "v2"),
6038                KeyValue::new("k3", "different"),
6039            ],
6040        );
6041
6042        meter_provider.force_flush().unwrap();
6043        let resource_metrics = exporter
6044            .get_finished_metrics()
6045            .expect("metrics are expected to be exported.");
6046        let metric = &resource_metrics[0].scope_metrics[0].metrics[0];
6047        let data::AggregatedMetrics::F64(MetricData::ExponentialHistogram(hist)) = &metric.data
6048        else {
6049            panic!("expected ExponentialHistogram aggregation");
6050        };
6051
6052        assert_eq!(
6053            hist.data_points.len(),
6054            1,
6055            "view should filter k3, leaving bound+unbound to aggregate together"
6056        );
6057        let dp = &hist.data_points[0];
6058        assert_eq!(dp.count, 3);
6059        assert_eq!(dp.sum, 30.0);
6060        assert_eq!(dp.attributes.len(), 2);
6061        assert!(dp.attributes.iter().any(|kv| kv.key.as_str() == "k1"));
6062        assert!(dp.attributes.iter().any(|kv| kv.key.as_str() == "k2"));
6063        assert!(!dp.attributes.iter().any(|kv| kv.key.as_str() == "k3"));
6064    }
6065}