add unit and integration tests

Author: czhang771

opentelemetry-sdk/tests/metrics/integration_test/test_histogram_export.py

@@ -17,6 +17,11 @@
 from opentelemetry.sdk.metrics import MeterProvider
 from opentelemetry.sdk.metrics.export import InMemoryMetricReader
 from opentelemetry.sdk.resources import SERVICE_NAME, Resource
+from opentelemetry.sdk.metrics._internal.exemplar import (
+    AlwaysOffExemplarFilter,
+    AlwaysOnExemplarFilter,
+    TraceBasedExemplarFilter,
+)
 
 
 class TestHistogramExport(TestCase):
@@ -88,3 +93,63 @@ def test_histogram_counter_collection(self):
             ),
             1,
         )
+    
+    def test_histogram_with_exemplars(self):
+
+        in_memory_metric_reader = InMemoryMetricReader()
+
+        provider = MeterProvider(
+            resource=Resource.create({SERVICE_NAME: "otel-test"}),
+            metric_readers=[in_memory_metric_reader],
+            exemplar_filter=AlwaysOnExemplarFilter()
+        )
+        meter = provider.get_meter("my-meter")
+        histogram = meter.create_histogram("my_histogram")
+        
+        histogram.record(2, {"attribute": "value1"})   # Should go in the first bucket
+        histogram.record(7, {"attribute": "value2"})   # Should go in the second bucket
+        histogram.record(9, {"attribute": "value2"})   # Should also go in the second bucket
+        histogram.record(15, {"attribute": "value3"})  # Should go in the third bucket
+
+        metric_data = in_memory_metric_reader.get_metrics_data()
+
+        self.assertEqual(len(metric_data.resource_metrics[0].scope_metrics[0].metrics), 1)
+        histogram_metric = metric_data.resource_metrics[0].scope_metrics[0].metrics[0]
+
+        self.assertEqual(len(histogram_metric.data.data_points), 3)
+
+        self.assertEqual(len(histogram_metric.data.data_points[0].exemplars), 1)
+        self.assertEqual(len(histogram_metric.data.data_points[1].exemplars), 1)
+        self.assertEqual(len(histogram_metric.data.data_points[2].exemplars), 1)
+
+        self.assertEqual(histogram_metric.data.data_points[0].sum, 2)
+        self.assertEqual(histogram_metric.data.data_points[1].sum, 16)
+        self.assertEqual(histogram_metric.data.data_points[2].sum, 15)
+
+        self.assertEqual(histogram_metric.data.data_points[0].exemplars[0].value, 2.0)  
+        self.assertEqual(histogram_metric.data.data_points[1].exemplars[0].value, 9.0)  
+        self.assertEqual(histogram_metric.data.data_points[2].exemplars[0].value, 15.0) 
+
+    def test_filter_with_exemplars(self):
+        in_memory_metric_reader = InMemoryMetricReader()
+
+        provider = MeterProvider(
+            resource=Resource.create({SERVICE_NAME: "otel-test"}),
+            metric_readers=[in_memory_metric_reader],
+            exemplar_filter=AlwaysOffExemplarFilter()
+        )
+        meter = provider.get_meter("my-meter")
+        histogram = meter.create_histogram("my_histogram")
+        
+        histogram.record(2, {"attribute": "value1"})   # Should go in the first bucket
+        histogram.record(7, {"attribute": "value2"})   # Should go in the second bucket
+        
+        metric_data = in_memory_metric_reader.get_metrics_data()
+
+        self.assertEqual(len(metric_data.resource_metrics[0].scope_metrics[0].metrics), 1)
+        histogram_metric = metric_data.resource_metrics[0].scope_metrics[0].metrics[0]
+
+        self.assertEqual(len(histogram_metric.data.data_points), 2)
+
+        self.assertEqual(len(histogram_metric.data.data_points[0].exemplars), 0)
+        self.assertEqual(len(histogram_metric.data.data_points[1].exemplars), 0)

opentelemetry-sdk/tests/metrics/integration_test/test_sum_aggregation.py

@@ -27,7 +27,12 @@
     InMemoryMetricReader,
 )
 from opentelemetry.sdk.metrics.view import SumAggregation
-
+from opentelemetry.sdk.metrics._internal.exemplar import (
+    AlwaysOffExemplarFilter,
+    AlwaysOnExemplarFilter,
+    TraceBasedExemplarFilter,
+)
+from opentelemetry.context import Context
 
 class TestSumAggregation(TestCase):
     @mark.skipif(
@@ -474,3 +479,34 @@ def test_synchronous_cumulative_temporality(self):
                 start_time_unix_nano, metric_data.start_time_unix_nano
             )
             self.assertEqual(metric_data.value, 80)
+    
+    def test_sum_aggregation_with_exemplars(self):
+
+        in_memory_metric_reader = InMemoryMetricReader()
+
+        provider = MeterProvider(
+            metric_readers=[in_memory_metric_reader],
+            exemplar_filter=AlwaysOnExemplarFilter(),
+        )
+
+        meter = provider.get_meter("my-meter")
+        counter = meter.create_counter("my_counter")
+
+        counter.add(2, {"attribute": "value1"}, context=Context())
+        counter.add(5, {"attribute": "value2"}, context=Context())
+        counter.add(3, {"attribute": "value3"}, context=Context())
+
+        metric_data = in_memory_metric_reader.get_metrics_data()
+
+        self.assertEqual(len(metric_data.resource_metrics[0].scope_metrics[0].metrics), 1)
+
+        sum_metric = metric_data.resource_metrics[0].scope_metrics[0].metrics[0]
+
+        data_points = sum_metric.data.data_points
+        self.assertEqual(len(data_points), 3)
+        
+        self.assertEqual(data_points[0].exemplars[0].value, 2.0)
+        self.assertEqual(data_points[1].exemplars[0].value, 5.0)
+        self.assertEqual(data_points[2].exemplars[0].value, 3.0)
+
+        provider.shutdown()

opentelemetry-sdk/tests/metrics/test_aggregation.py

@@ -48,7 +48,11 @@
     SumAggregation,
 )
 from opentelemetry.util.types import Attributes
-
+from opentelemetry.sdk.metrics._internal.exemplar import (
+    AlignedHistogramBucketExemplarReservoir,
+    ExemplarReservoirFactory,
+    SimpleFixedSizeExemplarReservoir,
+)
 
 def measurement(
     value: Union[int, float], attributes: Attributes = None
@@ -640,3 +644,76 @@ def test_observable_gauge(self):
             0,
         )
         self.assertIsInstance(aggregation, _LastValueAggregation)
+
+class TestExemplarsFromAggregations(TestCase):
+    
+    def test_collection_simple_fixed_size_reservoir(self):
+        exemplar_reservoir_factory = lambda: SimpleFixedSizeExemplarReservoir(size=5)
+        synchronous_sum_aggregation = _SumAggregation(
+            Mock(),
+            True,
+            AggregationTemporality.DELTA,
+            0,
+            exemplar_reservoir_factory,
+        )
+
+        synchronous_sum_aggregation.aggregate(measurement(1))
+        synchronous_sum_aggregation.aggregate(measurement(2))
+        synchronous_sum_aggregation.aggregate(measurement(3))
+
+        self.assertEqual(synchronous_sum_aggregation._value, 6)
+        datapoint = synchronous_sum_aggregation.collect(AggregationTemporality.CUMULATIVE, 0)
+        self.assertEqual(len(datapoint.exemplars), 3)    
+    
+    def test_collection_simple_fixed_size_reservoir_with_default_reservoir(self):
+        
+        synchronous_sum_aggregation = _SumAggregation(
+            Mock(),
+            True,
+            AggregationTemporality.DELTA,
+            0,
+            default_reservoir_factory(_SumAggregation),
+        )
+        
+        synchronous_sum_aggregation.aggregate(measurement(1))
+        synchronous_sum_aggregation.aggregate(measurement(2))
+        synchronous_sum_aggregation.aggregate(measurement(3))
+
+        self.assertEqual(synchronous_sum_aggregation._value, 6)
+        datapoint = synchronous_sum_aggregation.collect(AggregationTemporality.CUMULATIVE, 0)
+        self.assertEqual(len(datapoint.exemplars), 1)    
+    
+    def test_collection_aligned_histogram_bucket_reservoir(self):
+        boundaries = [5.0, 10.0, 20.0]
+        exemplar_reservoir_factory = lambda: AlignedHistogramBucketExemplarReservoir(boundaries)
+        synchronous_sum_aggregation = _SumAggregation(
+            Mock(),
+            True,
+            AggregationTemporality.DELTA,
+            0,
+            exemplar_reservoir_factory,
+        )
+
+        synchronous_sum_aggregation.aggregate(measurement(2.0)) 
+        synchronous_sum_aggregation.aggregate(measurement(4.0))  
+        synchronous_sum_aggregation.aggregate(measurement(6.0)) 
+        synchronous_sum_aggregation.aggregate(measurement(15.0))
+        synchronous_sum_aggregation.aggregate(measurement(25.0)) 
+
+        datapoint = synchronous_sum_aggregation.collect(AggregationTemporality.CUMULATIVE, 0)
+        self.assertEqual(len(datapoint.exemplars), 4)  
+
+        # Verify that exemplars are associated with the correct boundaries
+        expected_buckets = [
+            (4.0, boundaries[0]),  # First bucket, should hold the last value <= 5.0
+            (6.0, boundaries[1]),  # Second bucket, should hold the last value <= 10.0
+            (15.0, boundaries[2]), # Third bucket, should hold the last value <= 20.0
+            (25.0, None),          # Last bucket, should hold the value > 20.0
+        ]
+
+        for exemplar, (value, boundary) in zip(datapoint.exemplars, expected_buckets):
+            self.assertEqual(exemplar.value, value)
+            if boundary is not None:
+                self.assertLessEqual(exemplar.value, boundary)
+            else:
+                self.assertGreater(exemplar.value, boundaries[-1])

opentelemetry-sdk/tests/metrics/test_exemplarreservoir.py

@@ -1,6 +1,6 @@
 from unittest import TestCase
 
-#from opentelemetry.context import Context
+from opentelemetry.context import Context
 from opentelemetry.trace import INVALID_SPAN, SpanContext, TraceFlags
 from opentelemetry import trace
 from time import time_ns
@@ -58,6 +58,26 @@ def test_filter_attributes(self):
         exemplars = reservoir.collect({"key2": "value2", "key3": "value3"})
         self.assertEqual(len(exemplars), 1)
         self.assertNotEqual("key1", exemplars[0].filtered_attributes)
+    
+    def test_reset_after_collection(self):
+        reservoir = SimpleFixedSizeExemplarReservoir(4)
+
+        reservoir.offer(1.0, time_ns(), {"attribute": "value1"}, Context())
+        reservoir.offer(2.0, time_ns(), {"attribute": "value2"}, Context())
+        reservoir.offer(3.0, time_ns(), {"attribute": "value3"}, Context())
+
+        exemplars = reservoir.collect({})
+        self.assertEqual(len(exemplars), 3)
+
+        # Offer new measurements after reset
+        reservoir.offer(4.0, time_ns(), {"attribute": "value4"}, Context())
+        reservoir.offer(5.0, time_ns(), {"attribute": "value5"}, Context())
+
+        # Collect again and check the number of exemplars
+        new_exemplars = reservoir.collect({})
+        self.assertEqual(len(new_exemplars), 2)
+        self.assertEqual(new_exemplars[0].value, 4.0)
+        self.assertEqual(new_exemplars[1].value, 5.0)
 
 class TestAlignedHistogramBucketExemplarReservoir(TestCase):
 
@@ -78,14 +98,40 @@ def test_measurement_in_buckets(self):
         reservoir.offer(52, time_ns(), {"bucket": "5"}, ctx)
         reservoir.offer(7, time_ns(), {"bucket": "3"}, ctx)
         reservoir.offer(6, time_ns(), {"bucket": "3"}, ctx)
+        
         exemplars = reservoir.collect({"bucket": "3"})
+        
         self.assertEqual(len(exemplars), 2)
         self.assertEqual(exemplars[0].value, 6)            
         self.assertEqual(exemplars[1].value, 52)
         self.assertEqual(len(exemplars[0].filtered_attributes), 0)
         self.assertNotEqual(exemplars[1].filtered_attributes, {"bucket": "5"})
+    def test_last_measurement_in_bucket(self):
+        reservoir = AlignedHistogramBucketExemplarReservoir([0, 5, 10, 25])
+        span_context = SpanContext(
+            trace_id=self.TRACE_ID,
+            span_id=self.SPAN_ID,
+            is_remote=False,
+            trace_flags=TraceFlags(TraceFlags.SAMPLED),
+            trace_state={}
+        )
+        span = trace.NonRecordingSpan(span_context)
+        ctx = trace.set_span_in_context(span)
+        
+        # Offer values to the reservoir
+        reservoir.offer(2, time_ns(), {"bucket": "1"}, ctx)   # Bucket 1
+        reservoir.offer(7, time_ns(), {"bucket": "2"}, ctx)   # Bucket 2
+        reservoir.offer(8, time_ns(), {"bucket": "2"}, ctx)   # Bucket 2 - should replace the 7
+        reservoir.offer(15, time_ns(), {"bucket": "3"}, ctx)  # Bucket 3
 
-
+        exemplars = reservoir.collect({})
+        
+        # Check that each bucket has the correct value
+        self.assertEqual(len(exemplars), 3)  
+        self.assertEqual(exemplars[0].value, 2)
+        self.assertEqual(exemplars[1].value, 8)
+        self.assertEqual(exemplars[2].value, 15)
+        
 class TestExemplarReservoirFactory(TestCase):
     def test_sum_aggregation(self):
         exemplar_reservoir = default_reservoir_factory(_SumAggregation)