From 8f95faea3c0a795f45767036f9ac519a13921bf1 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Sun, 28 Jun 2026 14:59:05 +0000
Subject: [PATCH] feat(metrics): add distribution metric support

Add a `distribution` method to the metrics backend interface, alongside
the existing `increment`, `gauge`, and `timing` methods. This allows
backends to record distribution metrics.

- Add abstract `distribution` to the `Metrics` protocol
- Implement no-op `distribution` in `DummyMetricsBackend`
- Add `Distribution` call type and implementation in the testing backend
- Document `distribution` in the metrics docs
- Add a test for the testing backend's `distribution`

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019ASHzHqVirSvv3vBRYaQPS
---
 CHANGELOG.md                |  6 ++++++
 arroyo/utils/metrics.py     | 14 ++++++++++++++
 docs/source/metrics.rst     |  6 ++++++
 tests/metrics.py            | 15 +++++++++++++--
 tests/utils/test_metrics.py | 16 ++++++++++++++++
 5 files changed, 55 insertions(+), 2 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 18f15631..9bfced07 100644
--- a/CHANGELOG.md
+++ b/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 ✨
diff --git a/arroyo/utils/metrics.py b/arroyo/utils/metrics.py
index 8a68fa78..18ce4d51 100644
--- a/arroyo/utils/metrics.py
+++ b/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__(
diff --git a/docs/source/metrics.rst b/docs/source/metrics.rst
index c005c740..9642b424 100644
--- a/docs/source/metrics.rst
+++ b/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)
diff --git a/tests/metrics.py b/tests/metrics.py
index 5c730f1f..989e1881 100644
--- a/tests/metrics.py
+++ b/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()
diff --git a/tests/utils/test_metrics.py b/tests/utils/test_metrics.py
index 6be5518a..7332568f 100644
--- a/tests/utils/test_metrics.py
+++ b/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