1#[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")]
57pub 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#[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#[derive(Debug, Copy, Clone, Default, PartialEq, Eq, Hash)]
89#[non_exhaustive]
90pub enum Temporality {
91 #[default]
96 Cumulative,
97
98 Delta,
103
104 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 #[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 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 test_context.check_no_metrics();
205 }
206
207 let invalid_bucket_boundaries = vec![
208 vec![1.0, 1.0], vec![1.0, 2.0, 3.0, 2.0], vec![1.0, 2.0, 3.0, 4.0, 2.5], vec![1.0, 2.0, 3.0, f64::INFINITY, 4.0], vec![1.0, 2.0, 3.0, f64::NAN], vec![f64::NEG_INFINITY, 2.0, 3.0], ];
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 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 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 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 let invalid_bucket_boundaries = vec![
299 vec![1.0, 1.0], vec![1.0, 2.0, 3.0, 2.0], vec![1.0, 2.0, 3.0, 4.0, 2.5], vec![1.0, 2.0, 3.0, f64::INFINITY, 4.0], vec![1.0, 2.0, 3.0, f64::NAN], vec![f64::NEG_INFINITY, 2.0, 3.0], ];
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 test_context.check_no_metrics();
319 }
320 }
321
322 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
323 async fn counter_aggregation_delta() {
324 counter_aggregation_helper(Temporality::Delta);
327 }
328
329 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
330 async fn counter_aggregation_cumulative() {
331 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 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 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 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 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 histogram_aggregation_helper(Temporality::Cumulative);
429 }
430
431 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
432 async fn histogram_aggregation_delta() {
433 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 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 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 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 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 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 updown_counter_aggregation_helper(Temporality::Delta);
482 }
483
484 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
485 async fn gauge_aggregation() {
486 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 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 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 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 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 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 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 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 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 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 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 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 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 assert_eq!(data_point.value, *v);
625 } else {
626 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 let mut test_context = TestContext::new(Temporality::Delta);
659
660 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 {
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 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 {
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 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 {
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 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 let counter = meter.u64_counter("my_counter").build();
775
776 counter.add(10, &[]);
777 provider.force_flush().unwrap();
778
779 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 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 let exporter = InMemoryMetricExporter::default();
809 let meter_provider = SdkMeterProvider::builder()
810 .with_periodic_exporter(exporter.clone())
811 .build();
812
813 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 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 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 let exporter = InMemoryMetricExporter::default();
861 let meter_provider = SdkMeterProvider::builder()
862 .with_periodic_exporter(exporter.clone())
863 .build();
864
865 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 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 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 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 let exporter = InMemoryMetricExporter::default();
952 let meter_provider = SdkMeterProvider::builder()
953 .with_periodic_exporter(exporter.clone())
954 .build();
955
956 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 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 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 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 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], 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 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 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 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 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 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 assert_eq!(
1123 metric.name, "my_counter",
1124 "View rename should be ignored due to incompatible aggregation."
1125 );
1126 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 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 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 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 assert_eq!(
1175 metric.name, "my_gauge",
1176 "View rename should be ignored due to incompatible aggregation."
1177 );
1178 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 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 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 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 assert_eq!(
1227 metric.name, "my_updown_counter",
1228 "View rename should be ignored due to incompatible aggregation."
1229 );
1230 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 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 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 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 assert_eq!(
1279 metric.name, "my_histogram",
1280 "View rename should be ignored due to incompatible aggregation."
1281 );
1282 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 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 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 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 assert_eq!(
1335 metric.name, "my_observable_gauge",
1336 "View rename should be ignored due to incompatible aggregation."
1337 );
1338 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 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 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 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 assert_eq!(
1391 metric.name, "my_observable_counter",
1392 "View rename should be ignored due to incompatible aggregation."
1393 );
1394 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 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 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 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 assert_eq!(
1447 metric.name, "my_observable_updowncounter",
1448 "View rename should be ignored due to incompatible aggregation."
1449 );
1450 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 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 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 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 assert_eq!(
1500 metric.name, "my_histogram_renamed",
1501 "View rename should be applied for compatible aggregation."
1502 );
1503 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); }
1512
1513 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
1514 async fn gauge_with_histogram_aggregation_is_valid() {
1515 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 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 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 assert_eq!(
1557 metric.name, "my_gauge_renamed",
1558 "View rename should be applied for compatible aggregation."
1559 );
1560 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); assert_eq!(dp.min, Some(3.0));
1571 assert_eq!(dp.max, Some(30.0));
1572 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 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 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 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 assert_eq!(
1621 metric.name, "my_counter_renamed",
1622 "View rename should be applied for compatible aggregation."
1623 );
1624 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); assert_eq!(dp.min, Some(3));
1635 assert_eq!(dp.max, Some(30));
1636 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 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 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 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 assert_eq!(
1684 metric.name, "my_updowncounter_renamed",
1685 "View rename should be applied for compatible aggregation."
1686 );
1687 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 }
1698
1699 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
1700 async fn counter_with_exponential_histogram_aggregation_is_valid() {
1701 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 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 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 assert_eq!(
1743 metric.name, "my_counter_renamed",
1744 "View rename should be applied for compatible aggregation."
1745 );
1746 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); 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 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 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 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 assert_eq!(
1805 metric.name, "my_gauge_renamed",
1806 "View rename should be applied for compatible aggregation."
1807 );
1808 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); 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 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 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 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 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 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 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 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 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 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 let exporter = InMemoryMetricExporter::default();
1967 let view = |i: &Instrument| {
1968 if i.name == "my_counter" {
1969 Stream::builder()
1970 .with_name("dropped_counter") .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 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 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 let exporter = InMemoryMetricExporter::default();
2010 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 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 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 assert_eq!(sum.data_points.len(), 1);
2077
2078 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 let exporter = InMemoryMetricExporter::default();
2090 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 let meter = meter_provider.meter("test");
2110 let counter = meter.u64_counter("my_counter").build();
2111
2112 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 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 assert_eq!(sum.data_points.len(), 1);
2161 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 let mut test_context = TestContext::new(Temporality::Delta);
2309 let counter = test_context.u64_counter("test", "my_counter", None);
2310
2311 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 assert_eq!(sum.data_points.len(), 2);
2329
2330 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 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 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 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 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 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 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 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 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 assert_correct_export(&mut test_context, instrument_name);
2437
2438 test_context.reset_metrics();
2440
2441 test_context.flush_metrics();
2442
2443 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 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 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 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 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 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 #[test]
2757 fn test_view_single_instrument_multiple_stream() {
2758 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 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 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 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 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 let exporter = InMemoryMetricExporter::default();
2826 let meter_provider = SdkMeterProvider::builder()
2827 .with_periodic_exporter(exporter.clone())
2828 .with_view(view)
2829 .build();
2830
2831 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 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 }
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 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 assert_correct_export(&mut test_context, instrument_name);
2909
2910 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 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 if i % 2 == 0 {
2993 test_context.flush_metrics();
2994 thread::sleep(Duration::from_millis(i)); }
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 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); 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); }
3038
3039 fn counter_f64_multithreaded_aggregation_helper(temporality: Temporality) {
3040 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 if i % 2 == 0 {
3055 test_context.flush_metrics();
3056 thread::sleep(Duration::from_millis(i)); }
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 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); 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); }
3100
3101 fn histogram_multithreaded_aggregation_helper(temporality: Temporality) {
3102 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 if i % 2 == 0 {
3118 test_context.flush_metrics();
3119 thread::sleep(Duration::from_millis(i)); }
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 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]; let mut bucket_counts_key1_value1 = vec![0; 16];
3155
3156 histograms.iter().for_each(|histogram| {
3157 assert_eq!(histogram.data_points.len(), 2); 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 assert_eq!(count_zero_attributes, 20); assert_eq!(sum_zero_attributes, 50); 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), _ => assert_eq!(*count, 0),
3225 }
3226 }
3227
3228 assert_eq!(count_key1_value1, 50); assert_eq!(sum_key1_value1, 1000); 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), 2 => assert_eq!(*count, 10), 3 => assert_eq!(*count, 10), 4 => assert_eq!(*count, 20), _ => assert_eq!(*count, 0),
3240 }
3241 }
3242 }
3243
3244 fn histogram_f64_multithreaded_aggregation_helper(temporality: Temporality) {
3245 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 if i % 2 == 0 {
3261 test_context.flush_metrics();
3262 thread::sleep(Duration::from_millis(i)); }
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 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]; let mut bucket_counts_key1_value1 = vec![0; 16];
3298
3299 histograms.iter().for_each(|histogram| {
3300 assert_eq!(histogram.data_points.len(), 2); 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 assert_eq!(count_zero_attributes, 20); assert!(f64::abs(61.0 - sum_zero_attributes) < 0.0001); 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), _ => assert_eq!(*count, 0),
3368 }
3369 }
3370
3371 assert_eq!(count_key1_value1, 50); assert!(f64::abs(1006.0 - sum_key1_value1) < 0.0001); 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), 2 => assert_eq!(*count, 10), 3 => assert_eq!(*count, 10), 4 => assert_eq!(*count, 20), _ => assert_eq!(*count, 0),
3383 }
3384 }
3385 }
3386
3387 fn histogram_aggregation_helper(temporality: Temporality) {
3388 let mut test_context = TestContext::new(temporality);
3390 let histogram = test_context.meter().u64_histogram("my_histogram").build();
3391
3392 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 let MetricData::Histogram(histogram_data) =
3412 test_context.get_aggregation::<u64>("my_histogram", None)
3413 else {
3414 unreachable!()
3415 };
3416 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 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 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 let MetricData::Histogram(histogram_data) =
3515 test_context.get_aggregation::<u64>("test_histogram", None)
3516 else {
3517 unreachable!()
3518 };
3519 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 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 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 let MetricData::Histogram(histogram_data) =
3570 test_context.get_aggregation::<u64>("test_histogram", None)
3571 else {
3572 unreachable!()
3573 };
3574 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 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 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 if specify_boundaries_in_view {
3660 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 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 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 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 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 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 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 assert_eq!(
3745 data_point.zero_count(),
3746 0,
3747 "zero_count should be 0 for positive values"
3748 );
3749
3750 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 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 let attrs: Vec<_> = data_point.attributes().collect();
3770 assert_eq!(attrs.len(), 1);
3771 assert_eq!(attrs[0].key.as_str(), "key1");
3772
3773 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 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 assert_eq!(data_point.count(), 8);
3797 assert_eq!(data_point.sum(), 75.0);
3798 assert_eq!(data_point.min(), Some(1.0)); assert_eq!(data_point.max(), Some(30.0)); } else {
3801 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 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 let mut test_context = TestContext::new(temporality);
3822 let gauge = test_context.meter().i64_gauge("my_gauge").build();
3823
3824 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 let MetricData::Gauge(gauge_data_point) =
3839 test_context.get_aggregation::<i64>("my_gauge", None)
3840 else {
3841 unreachable!()
3842 };
3843 assert_eq!(gauge_data_point.data_points.len(), 2);
3845
3846 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 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 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 let MetricData::Gauge(gauge) =
3903 test_context.get_aggregation::<i64>("test_observable_gauge", None)
3904 else {
3905 unreachable!()
3906 };
3907 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 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 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 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 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 let mut test_context = TestContext::new(temporality);
3960 let counter = test_context.u64_counter("test", "my_counter", None);
3961
3962 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 let MetricData::Sum(sum) = test_context.get_aggregation::<u64>("my_counter", None) else {
3977 unreachable!()
3978 };
3979 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 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 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 let mut test_context = TestContext::new(temporality);
4039 let counter = test_context.u64_counter("test", "my_counter", None);
4040
4041 for v in 0..2000 {
4044 counter.add(100, &[KeyValue::new("A", v.to_string())]);
4045 }
4046
4047 counter.add(3, &[]);
4049 counter.add(3, &[]);
4050
4051 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 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 = 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 test_context.reset_metrics();
4085 if temporality == Temporality::Delta {
4086 test_context.flush_metrics();
4087 test_context.reset_metrics();
4088 }
4089 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 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 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 for v in 0..cardinality_limit {
4156 counter.add(100, &[KeyValue::new("A", v.to_string())]);
4157 }
4158
4159 counter.add(3, &[]);
4161 counter.add(3, &[]);
4162
4163 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 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 = 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 test_context.reset_metrics();
4197 if temporality == Temporality::Delta {
4198 test_context.flush_metrics();
4199 test_context.reset_metrics();
4200 }
4201 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 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 let mut test_context = TestContext::new(temporality);
4249 let counter = test_context.u64_counter("test", "my_counter", None);
4250
4251 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 assert_eq!(sum.data_points.len(), 1);
4323
4324 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 let mut test_context = TestContext::new(temporality);
4332 let counter = test_context.i64_up_down_counter("test", "my_updown_counter", None);
4333
4334 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 let MetricData::Sum(sum) = test_context.get_aggregation::<i64>("my_updown_counter", None)
4349 else {
4350 unreachable!()
4351 };
4352 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 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 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 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"); 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 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 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 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 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 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 for v in 0..cardinality_limit {
4907 counter.add(1, &[KeyValue::new("A", v.to_string())]);
4908 }
4909
4910 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 assert_eq!(
4922 sum.data_points.len(),
4923 cardinality_limit + 1,
4924 "Expected {} unique + 1 overflow data points",
4925 cardinality_limit
4926 );
4927
4928 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 for v in 0..cardinality_limit {
4957 counter.add(1, &[KeyValue::new("A", v.to_string())]);
4958 }
4959
4960 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 test_context.reset_metrics();
4970 test_context.flush_metrics(); 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 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 counter.add(1, &[KeyValue::new("A", "0")]);
5020 counter.add(1, &[KeyValue::new("A", "1")]);
5021
5022 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 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 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 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 test_context.reset_metrics();
5103 test_context.flush_metrics();
5104
5105 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 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 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 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 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 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 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 for v in 0..cardinality_limit {
5281 histogram.record(1.0, &[KeyValue::new("A", v.to_string())]);
5282 }
5283
5284 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 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 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 test_context.reset_metrics();
5355 test_context.flush_metrics();
5356
5357 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 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 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 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 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 test_context.reset_metrics();
5510 test_context.flush_metrics();
5511
5512 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 }
5550
5551 test_context.reset_metrics();
5553 test_context.flush_metrics();
5554
5555 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 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(bound1);
5602 test_context.reset_metrics();
5603 test_context.flush_metrics(); 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 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 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 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 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 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 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 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 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}