Create new aggregator per ViewData, config dict

Author: cnnradams

docs/examples/basic_meter/view.py

@@ -60,31 +60,32 @@
 # dropped from the aggregation
 counter_view1 = View(
     requests_counter,
-    SumAggregator(),
+    SumAggregator,
     label_keys=["environment"],
-    config=ViewConfig.LABEL_KEYS,
+    view_config=ViewConfig.LABEL_KEYS,
 )
 counter_view2 = View(
     requests_counter,
-    MinMaxSumCountAggregator(),
+    MinMaxSumCountAggregator,
     label_keys=["os_type"],
-    config=ViewConfig.LABEL_KEYS,
+    view_config=ViewConfig.LABEL_KEYS,
 )
 # This view has ViewConfig set to UNGROUPED, meaning all recorded metrics take
 # the labels directly without and consideration for label_keys
 counter_view3 = View(
     requests_counter,
-    LastValueAggregator(),
+    LastValueAggregator,
     label_keys=["environment"],  # is not used due to ViewConfig.UNGROUPED
-    config=ViewConfig.UNGROUPED,
+    view_config=ViewConfig.UNGROUPED,
 )
 # This view uses the HistogramAggregator which accepts an option config
 # parameter to specify the bucket ranges
 size_view = View(
     requests_size,
-    HistogramAggregator(config={"bounds": [20, 40, 60, 80, 100]}),
+    HistogramAggregator,
     label_keys=["environment"],  # is not used due to ViewConfig.UNGROUPED
-    config=ViewConfig.UNGROUPED,
+    aggregator_config={"bounds": [20, 40, 60, 80, 100]},
+    view_config=ViewConfig.UNGROUPED,
 )
 
 # Register the views to the view manager to use the views. Views MUST be

opentelemetry-sdk/src/opentelemetry/sdk/metrics/export/aggregate.py

@@ -166,7 +166,7 @@ def __init__(self, config=None):
         super().__init__(config=config)
         self._lock = threading.Lock()
         self.last_update_timestamp = None
-        boundaries = self.config.get("bounds", None)
+        boundaries = self.config.get("bounds")
         if boundaries and self._validate_boundaries(boundaries):
             self._boundaries = boundaries
         else:

opentelemetry-sdk/src/opentelemetry/sdk/metrics/view.py

@@ -74,29 +74,40 @@ class View:
     def __init__(
         self,
         metric: InstrumentT,
-        aggregator: Aggregator,
+        aggregator: type,
+        aggregator_config: dict = None,
         label_keys: Sequence[str] = None,
-        config: ViewConfig = ViewConfig.UNGROUPED,
+        view_config: ViewConfig = ViewConfig.UNGROUPED,
     ):
         self.metric = metric
         self.aggregator = aggregator
+        if aggregator_config is None:
+            aggregator_config = {}
+        self.aggregator_config = aggregator_config
         if label_keys is None:
             label_keys = []
         self.label_keys = sorted(label_keys)
-        self.config = config
+        self.view_config = view_config
 
     # Uniqueness is based on metric, aggregator type, ordered label keys and ViewConfig
     def __hash__(self):
         return hash(
-            (self.metric, self.aggregator, tuple(self.label_keys), self.config)
+            (
+                self.metric,
+                self.aggregator,
+                tuple(self.label_keys),
+                tuple(self.aggregator_config),
+                self.view_config,
+            )
         )
 
     def __eq__(self, other):
         return (
             self.metric == other.metric
             and self.aggregator.__class__ == other.aggregator.__class__
             and self.label_keys == other.label_keys
-            and self.config == other.config
+            and self.aggregator_config == other.aggregator_config
+            and self.view_config == other.view_config
         )
 
 
@@ -132,7 +143,7 @@ def generate_view_datas(self, metric, labels):
             for view in views:
                 updated_labels = []
                 dropped_labels = []
-                if view.config == ViewConfig.LABEL_KEYS:
+                if view.view_config == ViewConfig.LABEL_KEYS:
                     label_key_set = set(view.label_keys)
                     for label in labels:
                         # Only keep labels that are in configured label_keys
@@ -142,7 +153,7 @@ def generate_view_datas(self, metric, labels):
                             dropped_labels.append(label)
                     updated_labels = tuple(updated_labels)
                     dropped_labels = tuple(dropped_labels)
-                elif view.config == ViewConfig.UNGROUPED:
+                elif view.view_config == ViewConfig.UNGROUPED:
                     updated_labels = labels
                 else:
                     dropped_labels = labels
@@ -151,7 +162,7 @@ def generate_view_datas(self, metric, labels):
                 view_datas.add(
                     ViewData(
                         tuple(updated_labels),
-                        view.aggregator,
+                        view.aggregator(view.aggregator_config),
                         dropped_labels=dropped_labels,
                     )
                 )

opentelemetry-sdk/tests/metrics/export/test_exemplars.py

@@ -489,15 +489,14 @@ def test_histogram(self):
 
         size_view = View(
             requests_size,
-            HistogramAggregator(
-                config={
-                    "bounds": [20, 40, 60, 80, 100],
-                    "num_exemplars": 1,
-                    "statistical_exemplars": True,
-                }
-            ),
+            HistogramAggregator,
+            aggregator_config={
+                "bounds": [20, 40, 60, 80, 100],
+                "num_exemplars": 1,
+                "statistical_exemplars": True,
+            },
             label_keys=["environment"],
-            config=ViewConfig.LABEL_KEYS,
+            view_config=ViewConfig.LABEL_KEYS,
         )
 
         meter.register_view(size_view)

opentelemetry-sdk/tests/metrics/test_metrics.py

@@ -80,7 +80,7 @@ def test_collect_metrics(self):
             "name", "desc", "unit", float, metrics.Counter
         )
         labels = {"key1": "value1"}
-        meter.register_view(View(counter, SumAggregator()))
+        meter.register_view(View(counter, SumAggregator))
         counter.add(1.0, labels)
         meter.collect()
         self.assertTrue(batcher_mock.process.called)
@@ -108,7 +108,7 @@ def test_collect_disabled_metric(self):
         meter.batcher = batcher_mock
         counter = metrics.Counter("name", "desc", "unit", float, meter, False)
         labels = {"key1": "value1"}
-        meter.register_view(View(counter, SumAggregator()))
+        meter.register_view(View(counter, SumAggregator))
         counter.add(1.0, labels)
         meter.collect()
         self.assertFalse(batcher_mock.process.called)
@@ -137,8 +137,8 @@ def test_record_batch(self):
         valuerecorder = metrics.ValueRecorder(
             "name", "desc", "unit", float, meter
         )
-        counter_v = View(counter, SumAggregator())
-        measure_v = View(valuerecorder, MinMaxSumCountAggregator())
+        counter_v = View(counter, SumAggregator)
+        measure_v = View(valuerecorder, MinMaxSumCountAggregator)
         meter.register_view(counter_v)
         meter.register_view(measure_v)
         record_tuples = [(counter, 1.0), (valuerecorder, 3.0)]
@@ -241,7 +241,7 @@ def test_add(self):
         meter = metrics.MeterProvider().get_meter(__name__)
         metric = metrics.Counter("name", "desc", "unit", int, meter)
         labels = {"key": "value"}
-        counter_v = View(metric, SumAggregator())
+        counter_v = View(metric, SumAggregator)
         meter.register_view(counter_v)
         bound_mock = metric.bind(labels)
         metric.add(3, labels)
@@ -253,7 +253,7 @@ def test_add_non_decreasing_int_error(self, logger_mock):
         meter = metrics.MeterProvider().get_meter(__name__)
         metric = metrics.Counter("name", "desc", "unit", int, meter)
         labels = {"key": "value"}
-        counter_v = View(metric, SumAggregator())
+        counter_v = View(metric, SumAggregator)
         meter.register_view(counter_v)
         bound_counter = metric.bind(labels)
         metric.add(3, labels)
@@ -267,7 +267,7 @@ def test_add_non_decreasing_float_error(self, logger_mock):
         meter = metrics.MeterProvider().get_meter(__name__)
         metric = metrics.Counter("name", "desc", "unit", float, meter,)
         labels = {"key": "value"}
-        counter_v = View(metric, SumAggregator())
+        counter_v = View(metric, SumAggregator)
         meter.register_view(counter_v)
         bound_counter = metric.bind(labels)
         metric.add(3.3, labels)
@@ -286,7 +286,7 @@ def test_add(self):
         metric = metrics.UpDownCounter("name", "desc", "unit", int, meter,)
         labels = {"key": "value"}
         bound_counter = metric.bind(labels)
-        counter_v = View(metric, SumAggregator())
+        counter_v = View(metric, SumAggregator)
         meter.register_view(counter_v)
         metric.add(3, labels)
         metric.add(2, labels)
@@ -298,7 +298,7 @@ def test_record(self):
         meter = metrics.MeterProvider().get_meter(__name__)
         metric = metrics.ValueRecorder("name", "desc", "unit", int, meter,)
         labels = {"key": "value"}
-        measure_v = View(metric, MinMaxSumCountAggregator())
+        measure_v = View(metric, MinMaxSumCountAggregator)
         bound_valuerecorder = metric.bind(labels)
         meter.register_view(measure_v)
         values = (37, 42, 7)

opentelemetry-sdk/tests/metrics/test_view.py

@@ -18,6 +18,13 @@
 from opentelemetry.sdk import metrics
 from opentelemetry.sdk.metrics import view
 from opentelemetry.sdk.metrics.export import aggregate
+from opentelemetry.sdk.metrics import Counter
+from opentelemetry.sdk.metrics.export.aggregate import SumAggregator
+from opentelemetry.sdk.metrics.export.controller import PushController
+from opentelemetry.sdk.metrics.export.in_memory_metrics_exporter import (
+    InMemoryMetricsExporter,
+)
+from opentelemetry.sdk.metrics.view import View, ViewConfig
 
 
 class TestUtil(unittest.TestCase):
@@ -74,6 +81,44 @@ def test_default_aggregator(self, logger_mock):
         self.assertEqual(logger_mock.warning.call_count, 1)
 
 
+class TestPipeline(unittest.TestCase):
+    def setUp(self):
+        self.meter = metrics.MeterProvider(stateful=False).get_meter(__name__)
+        self.exporter = InMemoryMetricsExporter()
+        self.controller = PushController(self.meter, self.exporter, 30)
+
+    def tearDown(self):
+        self.controller.shutdown()
+
+    def test_stateless_label_keys(self):
+        test_counter = self.meter.create_metric(
+            name="test_counter",
+            description="description",
+            unit="By",
+            value_type=int,
+            metric_type=Counter,
+        )
+        counter_view = View(
+            test_counter,
+            SumAggregator,
+            label_keys=["environment"],
+            view_config=ViewConfig.LABEL_KEYS,
+        )
+
+        self.meter.register_view(counter_view)
+        test_counter.add(6, {"environment": "production", "customer_id": 123})
+        test_counter.add(5, {"environment": "production", "customer_id": 247})
+
+        self.controller.tick()
+
+        metric_data = self.exporter.get_exported_metrics()
+        self.assertEqual(len(metric_data), 1)
+        self.assertEqual(
+            metric_data[0].labels, (("environment", "production"),)
+        )
+        self.assertEqual(metric_data[0].aggregator.checkpoint, 11)
+
+
 class DummyMetric(metrics.Metric):
     # pylint: disable=W0231
     def __init__(self):