views: properly hash config dict, don't copy aggregator when stateful (open-telemetry#967)

cnnradams · web-flow · commit e8f7b6ff6843 · 2020-08-13T19:13:36.000-07:00
diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/metrics/export/batcher.py b/opentelemetry-sdk/src/opentelemetry/sdk/metrics/export/batcher.py
@@ -61,6 +61,7 @@ def process(self, record) -> None:
         """Stores record information to be ready for exporting."""
         # Checkpoints the current aggregator value to be collected for export
         aggregator = record.aggregator
+        aggregator.take_checkpoint()
 
         # The uniqueness of a batch record is defined by a specific metric
         # using an aggregator type with a specific set of labels.
@@ -72,19 +73,20 @@ def process(self, record) -> None:
             get_dict_as_key(aggregator.config),
             record.labels,
         )
+
         batch_value = self._batch_map.get(key)
+
         if batch_value:
-            if batch_value != aggregator:
-                aggregator.take_checkpoint()
-                batch_value.merge(aggregator)
-        else:
-            aggregator.take_checkpoint()
-
-        if self.stateful:
-            # if stateful batcher, create a copy of the aggregator and update
-            # it with the current checkpointed value for long-term storage
-            aggregator = record.aggregator.__class__(
-                config=record.aggregator.config
-            )
-            aggregator.merge(record.aggregator)
+            # Update the stored checkpointed value if exists. The call to merge
+            # here combines only identical records (same key).
+            batch_value.merge(aggregator)
+            return
+
+        # create a copy of the aggregator and update
+        # it with the current checkpointed value for long-term storage
+        aggregator = record.aggregator.__class__(
+            config=record.aggregator.config
+        )
+        aggregator.merge(record.aggregator)
+
         self._batch_map[key] = aggregator
diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/util/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/util/__init__.py
@@ -33,7 +33,16 @@ def ns_to_iso_str(nanoseconds):
 
 def get_dict_as_key(labels):
     """Converts a dict to be used as a unique key"""
-    return tuple(sorted(labels.items()))
+    return tuple(
+        sorted(
+            map(
+                lambda kv: (kv[0], tuple(kv[1]))
+                if isinstance(kv[1], list)
+                else kv,
+                labels.items(),
+            )
+        )
+    )
 
 
 class BoundedList(Sequence):
diff --git a/opentelemetry-sdk/tests/metrics/test_view.py b/opentelemetry-sdk/tests/metrics/test_view.py
@@ -16,9 +16,10 @@
 from unittest import mock
 
 from opentelemetry.sdk import metrics
-from opentelemetry.sdk.metrics import Counter, view
+from opentelemetry.sdk.metrics import Counter, ValueRecorder, view
 from opentelemetry.sdk.metrics.export import aggregate
 from opentelemetry.sdk.metrics.export.aggregate import (
+    HistogramAggregator,
     MinMaxSumCountAggregator,
     SumAggregator,
 )
@@ -185,6 +186,117 @@ def test_multiple_views(self):
         self.assertTrue((label2, 5) in sum_set)
 
 
+class TestHistogramView(unittest.TestCase):
+    def test_histogram_stateful(self):
+        meter = metrics.MeterProvider(stateful=True).get_meter(__name__)
+        exporter = InMemoryMetricsExporter()
+        controller = PushController(meter, exporter, 30)
+
+        requests_size = meter.create_metric(
+            name="requests_size",
+            description="size of requests",
+            unit="1",
+            value_type=int,
+            metric_type=ValueRecorder,
+        )
+
+        size_view = View(
+            requests_size,
+            HistogramAggregator,
+            aggregator_config={"bounds": [20, 40, 60, 80, 100]},
+            label_keys=["environment"],
+            view_config=ViewConfig.LABEL_KEYS,
+        )
+
+        meter.register_view(size_view)
+
+        # Since this is using the HistogramAggregator, the bucket counts will be reflected
+        # with each record
+        requests_size.record(25, {"environment": "staging", "test": "value"})
+        requests_size.record(1, {"environment": "staging", "test": "value2"})
+        requests_size.record(200, {"environment": "staging", "test": "value3"})
+
+        controller.tick()
+
+        metrics_list = exporter.get_exported_metrics()
+        self.assertEqual(len(metrics_list), 1)
+        checkpoint = metrics_list[0].aggregator.checkpoint
+        self.assertEqual(
+            tuple(checkpoint.items()),
+            ((20, 1), (40, 1), (60, 0), (80, 0), (100, 0), (">", 1)),
+        )
+        exporter.clear()
+
+        requests_size.record(25, {"environment": "staging", "test": "value"})
+        requests_size.record(1, {"environment": "staging", "test": "value2"})
+        requests_size.record(200, {"environment": "staging", "test": "value3"})
+
+        controller.tick()
+
+        metrics_list = exporter.get_exported_metrics()
+        self.assertEqual(len(metrics_list), 1)
+        checkpoint = metrics_list[0].aggregator.checkpoint
+        self.assertEqual(
+            tuple(checkpoint.items()),
+            ((20, 2), (40, 2), (60, 0), (80, 0), (100, 0), (">", 2)),
+        )
+
+    def test_histogram_stateless(self):
+        # Use the meter type provided by the SDK package
+        meter = metrics.MeterProvider(stateful=False).get_meter(__name__)
+        exporter = InMemoryMetricsExporter()
+        controller = PushController(meter, exporter, 30)
+
+        requests_size = meter.create_metric(
+            name="requests_size",
+            description="size of requests",
+            unit="1",
+            value_type=int,
+            metric_type=ValueRecorder,
+        )
+
+        size_view = View(
+            requests_size,
+            HistogramAggregator,
+            aggregator_config={"bounds": [20, 40, 60, 80, 100]},
+            label_keys=["environment"],
+            view_config=ViewConfig.LABEL_KEYS,
+        )
+
+        meter.register_view(size_view)
+
+        # Since this is using the HistogramAggregator, the bucket counts will be reflected
+        # with each record
+        requests_size.record(25, {"environment": "staging", "test": "value"})
+        requests_size.record(1, {"environment": "staging", "test": "value2"})
+        requests_size.record(200, {"environment": "staging", "test": "value3"})
+
+        controller.tick()
+
+        metrics_list = exporter.get_exported_metrics()
+        self.assertEqual(len(metrics_list), 1)
+        checkpoint = metrics_list[0].aggregator.checkpoint
+        self.assertEqual(
+            tuple(checkpoint.items()),
+            ((20, 1), (40, 1), (60, 0), (80, 0), (100, 0), (">", 1)),
+        )
+        exporter.clear()
+
+        requests_size.record(25, {"environment": "staging", "test": "value"})
+        requests_size.record(1, {"environment": "staging", "test": "value2"})
+        requests_size.record(200, {"environment": "staging", "test": "value3"})
+
+        controller.tick()
+
+        metrics_list = exporter.get_exported_metrics()
+        self.assertEqual(len(metrics_list), 1)
+        checkpoint = metrics_list[0].aggregator.checkpoint
+        self.assertEqual(
+            tuple(checkpoint.items()),
+            ((20, 1), (40, 1), (60, 0), (80, 0), (100, 0), (">", 1)),
+        )
+
+
 class DummyMetric(metrics.Metric):
     # pylint: disable=W0231
     def __init__(self):
