diff --git a/faust/sensors/prometheus.py b/faust/sensors/prometheus.py index f3c497697..08f2510a0 100644 --- a/faust/sensors/prometheus.py +++ b/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, diff --git a/tests/unit/sensors/test_prometheus.py b/tests/unit/sensors/test_prometheus.py index 0df2f41b9..e3f5a745b 100644 --- a/tests/unit/sensors/test_prometheus.py +++ b/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