api/sdk: Implement UpDownCounter instrument (open-telemetry#796)

Author: aabmass

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

@@ -66,7 +66,17 @@ def add(self, value: ValueT) -> None:
         """Increases the value of the bound counter by ``value``.
 
         Args:
-            value: The value to add to the bound counter.
+            value: The value to add to the bound counter. Must be positive.
+        """
+
+
+class BoundUpDownCounter:
+    def add(self, value: ValueT) -> None:
+        """Increases the value of the bound counter by ``value``.
+
+        Args:
+            value: The value to add to the bound counter. Can be positive or
+                negative.
         """
 
 
@@ -137,7 +147,28 @@ def add(self, value: ValueT, labels: Dict[str, str]) -> None:
         """Increases the value of the counter by ``value``.
 
         Args:
-            value: The value to add to the counter metric.
+            value: The value to add to the counter metric. Should be positive
+                or zero. For a Counter that can decrease, use
+                `UpDownCounter`.
+            labels: Labels to associate with the bound instrument.
+        """
+
+
+class UpDownCounter(Metric):
+    """A counter type metric that expresses the computation of a sum,
+    allowing negative increments."""
+
+    def bind(self, labels: Dict[str, str]) -> "BoundUpDownCounter":
+        """Gets a `BoundUpDownCounter`."""
+        return BoundUpDownCounter()
+
+    def add(self, value: ValueT, labels: Dict[str, str]) -> None:
+        """Increases the value of the counter by ``value``.
+
+        Args:
+            value: The value to add to the counter metric. Can be positive or
+                negative. For a Counter that is never decreasing, use
+                `Counter`.
             labels: Labels to associate with the bound instrument.
         """
 
@@ -268,7 +299,9 @@ def get_meter(
 
 
 MetricT = TypeVar("MetricT", Counter, ValueRecorder)
-InstrumentT = TypeVar("InstrumentT", Counter, Observer, ValueRecorder)
+InstrumentT = TypeVar(
+    "InstrumentT", Counter, UpDownCounter, Observer, ValueRecorder
+)
 ObserverT = TypeVar("ObserverT", bound=Observer)
 ObserverCallbackT = Callable[[Observer], None]
 

opentelemetry-api/tests/metrics/test_metrics.py

@@ -35,6 +35,16 @@ def test_counter_add(self):
         counter = metrics.Counter()
         counter.add(1, {})
 
+    def test_updowncounter(self):
+        counter = metrics.UpDownCounter()
+        bound_counter = counter.bind({})
+        self.assertIsInstance(bound_counter, metrics.BoundUpDownCounter)
+
+    def test_updowncounter_add(self):
+        counter = metrics.Counter()
+        counter.add(1, {})
+        counter.add(-1, {})
+
     def test_valuerecorder(self):
         valuerecorder = metrics.ValueRecorder()
         bound_valuerecorder = valuerecorder.bind({})

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

@@ -102,6 +102,25 @@ def add(self, value: metrics_api.ValueT) -> None:
         if self._validate_update(value):
             self.update(value)
 
+    def _validate_update(self, value: metrics_api.ValueT) -> bool:
+        if not super()._validate_update(value):
+            return False
+        if value < 0:
+            logger.warning(
+                "Invalid value %s passed to Counter, value must be non-negative. "
+                "For a Counter that can decrease, use UpDownCounter.",
+                value,
+            )
+            return False
+        return True
+
+
+class BoundUpDownCounter(metrics_api.BoundUpDownCounter, BaseBoundInstrument):
+    def add(self, value: metrics_api.ValueT) -> None:
+        """See `opentelemetry.metrics.BoundUpDownCounter.add`."""
+        if self._validate_update(value):
+            self.update(value)
+
 
 class BoundValueRecorder(metrics_api.BoundValueRecorder, BaseBoundInstrument):
     def record(self, value: metrics_api.ValueT) -> None:
@@ -184,6 +203,21 @@ def add(self, value: metrics_api.ValueT, labels: Dict[str, str]) -> None:
     UPDATE_FUNCTION = add
 
 
+class UpDownCounter(Metric, metrics_api.UpDownCounter):
+    """See `opentelemetry.metrics.UpDownCounter`.
+    """
+
+    BOUND_INSTR_TYPE = BoundUpDownCounter
+
+    def add(self, value: metrics_api.ValueT, labels: Dict[str, str]) -> None:
+        """See `opentelemetry.metrics.UpDownCounter.add`."""
+        bound_intrument = self.bind(labels)
+        bound_intrument.add(value)
+        bound_intrument.release()
+
+    UPDATE_FUNCTION = add
+
+
 class ValueRecorder(Metric, metrics_api.ValueRecorder):
     """See `opentelemetry.metrics.ValueRecorder`."""
 

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

@@ -19,6 +19,7 @@
     Counter,
     InstrumentT,
     SumObserver,
+    UpDownCounter,
     UpDownSumObserver,
     ValueObserver,
     ValueRecorder,
@@ -55,7 +56,7 @@ def aggregator_for(self, instrument_type: Type[InstrumentT]) -> Aggregator:
         Aggregators keep track of and updates values when metrics get updated.
         """
         # pylint:disable=R0201
-        if issubclass(instrument_type, Counter):
+        if issubclass(instrument_type, (Counter, UpDownCounter)):
             return CounterAggregator()
         if issubclass(instrument_type, (SumObserver, UpDownSumObserver)):
             return LastValueAggregator()

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

@@ -69,6 +69,15 @@ def test_aggregator_for_counter(self):
             )
         )
 
+    def test_aggregator_for_updowncounter(self):
+        batcher = UngroupedBatcher(True)
+        self.assertTrue(
+            isinstance(
+                batcher.aggregator_for(metrics.UpDownCounter),
+                CounterAggregator,
+            )
+        )
+
     # TODO: Add other aggregator tests
 
     def test_checkpoint_set(self):

opentelemetry-sdk/tests/metrics/test_metrics.py

@@ -174,6 +174,15 @@ def test_create_metric(self):
         self.assertEqual(counter.name, "name")
         self.assertIs(counter.meter.resource, resource)
 
+    def test_create_updowncounter(self):
+        meter = metrics.MeterProvider().get_meter(__name__)
+        updowncounter = meter.create_metric(
+            "name", "desc", "unit", float, metrics.UpDownCounter, ()
+        )
+        self.assertIsInstance(updowncounter, metrics.UpDownCounter)
+        self.assertEqual(updowncounter.value_type, float)
+        self.assertEqual(updowncounter.name, "name")
+
     def test_create_valuerecorder(self):
         meter = metrics.MeterProvider().get_meter(__name__)
         valuerecorder = meter.create_metric(
@@ -296,6 +305,53 @@ def test_add(self):
         metric.add(2, labels)
         self.assertEqual(bound_counter.aggregator.current, 5)
 
+    @mock.patch("opentelemetry.sdk.metrics.logger")
+    def test_add_non_decreasing_int_error(self, logger_mock):
+        meter = metrics.MeterProvider().get_meter(__name__)
+        metric = metrics.Counter("name", "desc", "unit", int, meter, ("key",))
+        labels = {"key": "value"}
+        bound_counter = metric.bind(labels)
+        metric.add(3, labels)
+        metric.add(0, labels)
+        metric.add(-1, labels)
+        self.assertEqual(bound_counter.aggregator.current, 3)
+        self.assertEqual(logger_mock.warning.call_count, 1)
+
+    @mock.patch("opentelemetry.sdk.metrics.logger")
+    def test_add_non_decreasing_float_error(self, logger_mock):
+        meter = metrics.MeterProvider().get_meter(__name__)
+        metric = metrics.Counter(
+            "name", "desc", "unit", float, meter, ("key",)
+        )
+        labels = {"key": "value"}
+        bound_counter = metric.bind(labels)
+        metric.add(3.3, labels)
+        metric.add(0.0, labels)
+        metric.add(0.1, labels)
+        metric.add(-0.1, labels)
+        self.assertEqual(bound_counter.aggregator.current, 3.4)
+        self.assertEqual(logger_mock.warning.call_count, 1)
+
+
+class TestUpDownCounter(unittest.TestCase):
+    @mock.patch("opentelemetry.sdk.metrics.logger")
+    def test_add(self, logger_mock):
+        meter = metrics.MeterProvider().get_meter(__name__)
+        metric = metrics.UpDownCounter(
+            "name", "desc", "unit", int, meter, ("key",)
+        )
+        labels = {"key": "value"}
+        bound_counter = metric.bind(labels)
+        metric.add(3, labels)
+        metric.add(2, labels)
+        self.assertEqual(bound_counter.aggregator.current, 5)
+
+        metric.add(0, labels)
+        metric.add(-3, labels)
+        metric.add(-1, labels)
+        self.assertEqual(bound_counter.aggregator.current, 1)
+        self.assertEqual(logger_mock.warning.call_count, 0)
+
 
 class TestValueRecorder(unittest.TestCase):
     def test_record(self):