feat(metrics): add distribution metric support

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog and versioning
 
+## Unreleased
+
+### New Features ✨
+
+- Add `distribution` method to the metrics backend interface for recording distribution metrics
+
 ## 2.40.3
 
 ### New Features ✨

arroyo/utils/metrics.py

@@ -44,6 +44,15 @@ def timing(
         """
         raise NotImplementedError
 
+    @abstractmethod
+    def distribution(
+        self, name: MetricName, value: Union[int, float], tags: Optional[Tags] = None
+    ) -> None:
+        """
+        Records a distribution metric.
+        """
+        raise NotImplementedError
+
 
 class DummyMetricsBackend(Metrics):
     """
@@ -68,6 +77,11 @@ def timing(
     ) -> None:
         pass
 
+    def distribution(
+        self, name: MetricName, value: Union[int, float], tags: Optional[Tags] = None
+    ) -> None:
+        pass
+
 
 class Gauge:
     def __init__(

docs/source/metrics.rst

@@ -35,6 +35,12 @@ This can be done like so:
             # Emit a timing metric with the given value.
             record_timing(name, value, tags)
 
+        def distribution(
+            self, name: MetricName, value: Union[int, float], tags: Optional[Tags] = None
+        ) -> None:
+            # Emit a distribution metric with the given value.
+            record_distribution(name, value, tags)
+
     metrics_backend = MyMetrics()
 
     configure_metrics(metrics_backend)

tests/metrics.py

@@ -21,7 +21,13 @@ class Timing(NamedTuple):
     tags: Optional[Tags]
 
 
-MetricCall = Union[Increment, Gauge, Timing]
+class Distribution(NamedTuple):
+    name: MetricName
+    value: Union[int, float]
+    tags: Optional[Tags]
+
+
+MetricCall = Union[Increment, Gauge, Timing, Distribution]
 
 
 class _TestingMetricsBackend(Metrics):
@@ -32,7 +38,7 @@ class _TestingMetricsBackend(Metrics):
     """
 
     def __init__(self) -> None:
-        self.calls: MutableSequence[Union[Increment, Gauge, Timing]] = []
+        self.calls: MutableSequence[Union[Increment, Gauge, Timing, Distribution]] = []
 
     def increment(
         self,
@@ -52,5 +58,10 @@ def timing(
     ) -> None:
         self.calls.append(Timing(name, value, tags))
 
+    def distribution(
+        self, name: MetricName, value: Union[int, float], tags: Optional[Tags] = None
+    ) -> None:
+        self.calls.append(Distribution(name, value, tags))
+
 
 TestingMetricsBackend = _TestingMetricsBackend()

tests/utils/test_metrics.py

@@ -1,6 +1,7 @@
 import pytest
 
 from arroyo.utils.metrics import Gauge, MetricName, configure_metrics, get_metrics
+from tests.metrics import Distribution as DistributionCall
 from tests.metrics import Gauge as GaugeCall
 from tests.metrics import TestingMetricsBackend, _TestingMetricsBackend
 
@@ -22,6 +23,21 @@ def test_gauge_simple() -> None:
     ]
 
 
+def test_distribution_simple() -> None:
+    backend = _TestingMetricsBackend()
+
+    name: MetricName = "name"  # type: ignore
+    tags = {"tag": "value"}
+
+    backend.distribution(name, 1.0, tags)
+    backend.distribution(name, 2.0, tags)
+
+    assert backend.calls == [
+        DistributionCall(name, 1.0, tags),
+        DistributionCall(name, 2.0, tags),
+    ]
+
+
 def test_configure_metrics() -> None:
     assert get_metrics() == TestingMetricsBackend