metrics: Implement release for handles and observers

This commit implements a solution for releasing instrument handles and observers.

For the handles it is based on a ref count that is increased each time the
handled is acquired, when the ref count reaches 0 the handle is removed on
collection time.  The direct call convention is updated to release the handle
after it has been updated.

The observer instrument is only updated on collection time, so it can be removed
as soon as the user request to do so.

Author: mauriciovasquezbernal

examples/metrics/record.py

@@ -65,8 +65,14 @@
 # a labelset. A handle is essentially metric data that corresponds to a specific
 # set of labels. Therefore, getting a handle using the same set of labels will
 # yield the same metric handle.
+# Get a handle when you have to perform multiple operations using the same
+# labelset
 counter_handle = counter.get_handle(label_set)
-counter_handle.add(100)
+for i in range(1000):
+    counter_handle.add(i)
+
+# You can release the handle we you are done
+counter_handle.release()
 
 # Direct metric usage
 # You can record metrics directly using the metric instrument. You pass in a
@@ -78,4 +84,5 @@
 # (metric, value) pairs. The value would be recorded for each metric using the
 # specified labelset for each.
 meter.record_batch(label_set, [(counter, 50), (counter2, 70)])
-time.sleep(100)
+
+time.sleep(10)

opentelemetry-api/src/opentelemetry/metrics/__init__.py

@@ -57,6 +57,9 @@ def record(self, value: ValueT) -> None:
             value: The value to record to the handle.
         """
 
+    def release(self) -> None:
+        """No-op implementation of release."""
+
 
 class CounterHandle:
     def add(self, value: ValueT) -> None:
@@ -346,6 +349,14 @@ def register_observer(
         Returns: A new ``Observer`` metric instrument.
         """
 
+    @abc.abstractmethod
+    def unregister_observer(self, observer: "Observer") -> None:
+        """Unregisters an ``Observer`` metric instrument.
+
+        Args:
+            observer: The observer to unregister.
+        """
+
     @abc.abstractmethod
     def get_label_set(self, labels: Dict[str, str]) -> "LabelSet":
         """Gets a `LabelSet` with the given labels.
@@ -392,6 +403,9 @@ def register_observer(
     ) -> "Observer":
         return DefaultObserver()
 
+    def unregister_observer(self, observer: "Observer") -> None:
+        pass
+
     def get_label_set(self, labels: Dict[str, str]) -> "LabelSet":
         # pylint: disable=no-self-use
         return DefaultLabelSet()

opentelemetry-api/tests/metrics/test_metrics.py

@@ -48,7 +48,8 @@ def test_measure_record(self):
         measure.record(1, label_set)
 
     def test_default_handle(self):
-        metrics.DefaultMetricHandle()
+        handle = metrics.DefaultMetricHandle()
+        handle.release()
 
     def test_counter_handle(self):
         handle = metrics.CounterHandle()

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

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 import logging
+import threading
 from collections import OrderedDict
 from typing import Dict, Sequence, Tuple, Type
 
@@ -70,6 +71,8 @@ def __init__(
         self.enabled = enabled
         self.aggregator = aggregator
         self.last_update_timestamp = time_ns()
+        self._ref_count = 0
+        self._ref_count_lock = threading.Lock()
 
     def _validate_update(self, value: metrics_api.ValueT) -> bool:
         if not self.enabled:
@@ -85,6 +88,21 @@ def update(self, value: metrics_api.ValueT):
         self.last_update_timestamp = time_ns()
         self.aggregator.update(value)
 
+    def release(self):
+        self.decrease_ref_count()
+
+    def decrease_ref_count(self):
+        with self._ref_count_lock:
+            self._ref_count -= 1
+
+    def increase_ref_count(self):
+        with self._ref_count_lock:
+            self._ref_count += 1
+
+    def ref_count(self):
+        with self._ref_count_lock:
+            return self._ref_count
+
     def __repr__(self):
         return '{}(data="{}", last_update_timestamp={})'.format(
             type(self).__name__,
@@ -136,18 +154,21 @@ def __init__(
         self.label_keys = label_keys
         self.enabled = enabled
         self.handles = {}
+        self.handles_lock = threading.Lock()
 
     def get_handle(self, label_set: LabelSet) -> BaseHandle:
         """See `opentelemetry.metrics.Metric.get_handle`."""
-        handle = self.handles.get(label_set)
-        if not handle:
-            handle = self.HANDLE_TYPE(
-                self.value_type,
-                self.enabled,
-                # Aggregator will be created based off type of metric
-                self.meter.batcher.aggregator_for(self.__class__),
-            )
-            self.handles[label_set] = handle
+        with self.handles_lock:
+            handle = self.handles.get(label_set)
+            if handle is None:
+                handle = self.HANDLE_TYPE(
+                    self.value_type,
+                    self.enabled,
+                    # Aggregator will be created based off type of metric
+                    self.meter.batcher.aggregator_for(self.__class__),
+                )
+                self.handles[label_set] = handle
+        handle.increase_ref_count()
         return handle
 
     def __repr__(self):
@@ -186,7 +207,9 @@ def __init__(
 
     def add(self, value: metrics_api.ValueT, label_set: LabelSet) -> None:
         """See `opentelemetry.metrics.Counter.add`."""
-        self.get_handle(label_set).add(value)
+        handle = self.get_handle(label_set)
+        handle.add(value)
+        handle.release()
 
     UPDATE_FUNCTION = add
 
@@ -198,7 +221,9 @@ class Measure(Metric, metrics_api.Measure):
 
     def record(self, value: metrics_api.ValueT, label_set: LabelSet) -> None:
         """See `opentelemetry.metrics.Measure.record`."""
-        self.get_handle(label_set).record(value)
+        handle = self.get_handle(label_set)
+        handle.record(value)
+        handle.release()
 
     UPDATE_FUNCTION = record
 
@@ -295,6 +320,7 @@ def __init__(
         self.metrics = set()
         self.observers = set()
         self.batcher = UngroupedBatcher(stateful)
+        self.observers_lock = threading.Lock()
 
     def collect(self) -> None:
         """Collects all the metrics created with this `Meter` for export.
@@ -309,26 +335,39 @@ def collect(self) -> None:
 
     def _collect_metrics(self) -> None:
         for metric in self.metrics:
-            if metric.enabled:
+            if not metric.enabled:
+                continue
+
+            to_remove = []
+
+            with metric.handles_lock:
                 for label_set, handle in metric.handles.items():
                     # TODO: Consider storing records in memory?
                     record = Record(metric, label_set, handle.aggregator)
                     # Checkpoints the current aggregators
                     # Applies different batching logic based on type of batcher
                     self.batcher.process(record)
 
+                    if handle.ref_count() == 0:
+                        to_remove.append(label_set)
+
+                # Remove handles that were released
+                for label_set in to_remove:
+                    del metric.handles[label_set]
+
     def _collect_observers(self) -> None:
-        for observer in self.observers:
-            if not observer.enabled:
-                continue
+        with self.observers_lock:
+            for observer in self.observers:
+                if not observer.enabled:
+                    continue
 
-            # TODO: capture timestamp?
-            if not observer.run():
-                continue
+                # TODO: capture timestamp?
+                if not observer.run():
+                    continue
 
-            for label_set, aggregator in observer.aggregators.items():
-                record = Record(observer, label_set, aggregator)
-                self.batcher.process(record)
+                for label_set, aggregator in observer.aggregators.items():
+                    record = Record(observer, label_set, aggregator)
+                    self.batcher.process(record)
 
     def record_batch(
         self,
@@ -383,9 +422,14 @@ def register_observer(
             label_keys,
             enabled,
         )
-        self.observers.add(ob)
+        with self.observers_lock:
+            self.observers.add(ob)
         return ob
 
+    def unregister_observer(self, observer: "Observer") -> None:
+        with self.observers_lock:
+            self.observers.remove(observer)
+
     def get_label_set(self, labels: Dict[str, str]):
         """See `opentelemetry.metrics.Meter.create_metric`.
 

opentelemetry-sdk/tests/metrics/test_metrics.py

@@ -35,7 +35,7 @@ def test_collect(self):
         )
         kvp = {"key1": "value1"}
         label_set = meter.get_label_set(kvp)
-        counter.add(label_set, 1.0)
+        counter.add(1.0, label_set)
         meter.metrics.add(counter)
         meter.collect()
         self.assertTrue(batcher_mock.process.called)
@@ -163,6 +163,18 @@ def test_register_observer(self):
         self.assertEqual(observer.label_keys, ())
         self.assertTrue(observer.enabled)
 
+    def test_unregister_observer(self):
+        meter = metrics.MeterProvider().get_meter(__name__)
+
+        callback = mock.Mock()
+
+        observer = meter.register_observer(
+            callback, "name", "desc", "unit", int, (), True
+        )
+
+        meter.unregister_observer(observer)
+        self.assertEqual(len(meter.observers), 0)
+
     def test_get_label_set(self):
         meter = metrics.MeterProvider().get_meter(__name__)
         kvp = {"environment": "staging", "a": "z"}
@@ -177,6 +189,64 @@ def test_get_label_set_empty(self):
         label_set = meter.get_label_set(kvp)
         self.assertEqual(label_set, metrics.EMPTY_LABEL_SET)
 
+    def test_direct_call_release_handle(self):
+        meter = metrics.MeterProvider().get_meter(__name__)
+        label_keys = ("key1",)
+        kvp = {"key1": "value1"}
+        label_set = meter.get_label_set(kvp)
+
+        counter = metrics.Counter(
+            "name", "desc", "unit", float, meter, label_keys
+        )
+        meter.metrics.add(counter)
+        counter.add(4.0, label_set)
+
+        measure = metrics.Measure(
+            "name", "desc", "unit", float, meter, label_keys
+        )
+        meter.metrics.add(measure)
+        measure.record(42.0, label_set)
+
+        self.assertEqual(len(counter.handles), 1)
+        self.assertEqual(len(measure.handles), 1)
+
+        meter.collect()
+
+        self.assertEqual(len(counter.handles), 0)
+        self.assertEqual(len(measure.handles), 0)
+
+    def test_release_handle(self):
+        meter = metrics.MeterProvider().get_meter(__name__)
+        label_keys = ("key1",)
+        kvp = {"key1": "value1"}
+        label_set = meter.get_label_set(kvp)
+
+        counter = metrics.Counter(
+            "name", "desc", "unit", float, meter, label_keys
+        )
+        meter.metrics.add(counter)
+        counter_handle = counter.get_handle(label_set)
+        counter_handle.add(4.0)
+
+        measure = metrics.Measure(
+            "name", "desc", "unit", float, meter, label_keys
+        )
+        meter.metrics.add(measure)
+        measure_handle = measure.get_handle(label_set)
+        measure_handle.record(42)
+
+        counter_handle.release()
+        measure_handle.release()
+
+        # be sure that handles are only released after collection
+        self.assertEqual(len(counter.handles), 1)
+        self.assertEqual(len(measure.handles), 1)
+
+        meter.collect()
+
+        self.assertEqual(len(counter.handles), 0)
+        self.assertEqual(len(measure.handles), 0)
+
 
 class TestMetric(unittest.TestCase):
     def test_get_handle(self):