Fix Prometheus latency histograms mis-scaled by 1000x

faust/sensors/prometheus.py

@@ -39,6 +39,29 @@
 
 __all__ = ["setup_prometheus_sensors"]
 
+#: Faust's latency metrics are all observed in milliseconds (see
+#: ``Monitor.ms_since``), but ``prometheus_client.Histogram`` defaults to
+#: second-scale buckets (5ms-10s). Without explicit millisecond-scale
+#: buckets, real latency observations mostly land in the ``+Inf`` overflow
+#: bucket. This mirrors the shape of ``Histogram.DEFAULT_BUCKETS`` scaled
+#: x1000 for milliseconds.
+MS_LATENCY_BUCKETS = (
+    5,
+    10,
+    25,
+    50,
+    75,
+    100,
+    250,
+    500,
+    750,
+    1000,
+    2500,
+    5000,
+    7500,
+    10000,
+)
+
 
 def setup_prometheus_sensors(
     app: AppT,
@@ -144,7 +167,10 @@ def create(cls, registry: CollectorRegistry, app_name: str) -> "FaustMetrics":
             registry=registry,
         )
         events_runtime_latency = Histogram(
-            f"{app_name}_events_runtime_ms", "Events runtime in ms", registry=registry
+            f"{app_name}_events_runtime_ms",
+            "Events runtime in ms",
+            registry=registry,
+            buckets=MS_LATENCY_BUCKETS,
         )
         total_events = Counter(
             f"{app_name}_total_events", "Total events received", registry=registry
@@ -177,6 +203,7 @@ def create(cls, registry: CollectorRegistry, app_name: str) -> "FaustMetrics":
             f"{app_name}_producer_send_latency",
             "Producer send latency in ms",
             registry=registry,
+            buckets=MS_LATENCY_BUCKETS,
         )
         total_error_messages_sent = Counter(
             f"{app_name}_total_error_messages_sent",
@@ -187,6 +214,7 @@ def create(cls, registry: CollectorRegistry, app_name: str) -> "FaustMetrics":
             f"{app_name}_producer_error_send_latency",
             "Producer error send latency in ms",
             registry=registry,
+            buckets=MS_LATENCY_BUCKETS,
         )
         assignment_operations = Counter(
             f"{app_name}_assignment_operations",
@@ -195,7 +223,10 @@ def create(cls, registry: CollectorRegistry, app_name: str) -> "FaustMetrics":
             registry=registry,
         )
         assign_latency = Histogram(
-            f"{app_name}_assign_latency", "Assignment latency in ms", registry=registry
+            f"{app_name}_assign_latency",
+            "Assignment latency in ms",
+            registry=registry,
+            buckets=MS_LATENCY_BUCKETS,
         )
         total_rebalances = Gauge(
             f"{app_name}_total_rebalances", "Total rebalances", registry=registry
@@ -209,11 +240,13 @@ def create(cls, registry: CollectorRegistry, app_name: str) -> "FaustMetrics":
             f"{app_name}_rebalance_done_consumer_latency",
             "Consumer replying that rebalance is done to broker in ms",
             registry=registry,
+            buckets=MS_LATENCY_BUCKETS,
         )
         rebalance_done_latency = Histogram(
             f"{app_name}_rebalance_done_latency",
             "Rebalance finished latency in ms",
             registry=registry,
+            buckets=MS_LATENCY_BUCKETS,
         )
         count_metrics_by_name = Gauge(
             f"{app_name}_metrics_by_name",
@@ -228,7 +261,10 @@ def create(cls, registry: CollectorRegistry, app_name: str) -> "FaustMetrics":
             registry=registry,
         )
         http_latency = Histogram(
-            f"{app_name}_http_latency", "Http response latency in ms", registry=registry
+            f"{app_name}_http_latency",
+            "Http response latency in ms",
+            registry=registry,
+            buckets=MS_LATENCY_BUCKETS,
         )
         topic_partition_end_offset = Gauge(
             f"{app_name}_topic_partition_end_offset",
@@ -246,6 +282,7 @@ def create(cls, registry: CollectorRegistry, app_name: str) -> "FaustMetrics":
             f"{app_name}_consumer_commit_latency",
             "Consumer commit latency in ms",
             registry=registry,
+            buckets=MS_LATENCY_BUCKETS,
         )
         return cls(
             messages_received=messages_received,

tests/unit/sensors/test_prometheus.py

@@ -523,3 +523,49 @@ def _handle_event(
         monitor.on_message_in(topic_partition, offset, event.message)
         monitor.on_stream_event_in(topic_partition, offset, stream, event)
         monitor.on_tp_commit({topic_partition: offset})
+
+
+# Every latency histogram in FaustMetrics observes milliseconds (see
+# Monitor.ms_since), so its bucket boundaries must be millisecond-scale too
+# (see faust-streaming/faust#260) -- otherwise real observations mostly land
+# in the +Inf overflow bucket instead of a meaningful bucket.
+LATENCY_HISTOGRAM_FIELDS = [
+    "events_runtime_latency",
+    "producer_send_latency",
+    "producer_error_send_latency",
+    "assign_latency",
+    "rebalance_done_consumer_latency",
+    "rebalance_done_latency",
+    "http_latency",
+    "consumer_commit_latency",
+]
+
+
+@pytest.mark.parametrize("field", LATENCY_HISTOGRAM_FIELDS)
+def test_latency_histograms_use_millisecond_scale_buckets(field: str) -> None:
+    from faust.sensors.prometheus import MS_LATENCY_BUCKETS
+
+    registry = CollectorRegistry()
+    metrics = FaustMetrics.create(registry, "test")
+    histogram = getattr(metrics, field)
+
+    # _upper_bounds always ends with +Inf; everything before it must match
+    # the millisecond-scale bucket boundaries exactly.
+    assert histogram._upper_bounds[:-1] == [float(b) for b in MS_LATENCY_BUCKETS]
+
+
+def test_millisecond_latency_observation_lands_in_a_real_bucket() -> None:
+    # A typical ~250ms latency must not be swallowed entirely by the +Inf
+    # overflow bucket -- it should land at/under the 250ms boundary.
+    registry = CollectorRegistry()
+    metrics = FaustMetrics.create(registry, "test")
+
+    metrics.events_runtime_latency.observe(250)
+
+    buckets = dict(
+        zip(
+            metrics.events_runtime_latency._upper_bounds,
+            metrics.events_runtime_latency._buckets,
+        )
+    )
+    assert buckets[250.0].get() == 1