# Metrics (/integrations/rayserve/how-to/metrics) `get_metrics_sync()` returns a single dict with eight entries that describe the health and throughput of an adapter: how much work is queued, how much is in flight, how much has failed, and how long handlers take. Three of the eight are live gauges read from KubeMQ on each call; the rest are tracked in-memory by a thread-safe collector. This guide covers how to collect them and what to chart. For the exact schema and types, see the [metrics section of the API reference](/integrations/rayserve/reference/api#metrics). ## Prerequisites [#prerequisites] * `kubemq-rayserve` installed, with an initialized `KubeMQTaskProcessorAdapter` (see [Getting Started with Ray Serve](/integrations/rayserve/tutorials/getting-started)) * A running KubeMQ broker reachable from the adapter, since three of the eight metrics are read live from the broker ## Reading the metrics dict [#reading-the-metrics-dict] `get_metrics()` (the alias for `get_metrics_sync()`) returns the full dict at any time after `initialize()`. The packaged `examples/metrics/basic_metrics.py` enqueues a handful of tasks, waits for them to finish, and prints every key. ```python title="basic_metrics.py" from __future__ import annotations import os import time import uuid from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter BROKER = os.environ.get("KUBEMQ_ADDRESS", "localhost:50000") def compute(x: int) -> int: time.sleep(0.05 * (x % 3 + 1)) return x * x def main(): channel = f"example-metrics-{uuid.uuid4().hex[:8]}" adapter = KubeMQTaskProcessorAdapter(KubeMQAdapterConfig(address=BROKER)) class _Cfg: queue_name = channel max_retries = 0 failed_task_queue_name = f"{channel}.dlq" unprocessable_task_queue_name = "" adapter.initialize(consumer_concurrency=1, task_processor_config=_Cfg()) adapter.register_task_handler(compute, name="compute") adapter.start_consumer() try: task_ids = [adapter.enqueue_task_sync("compute", args=[i]).id for i in range(5)] for task_id in task_ids: for _ in range(30): if adapter.get_task_status_sync(task_id).status in ("SUCCESS", "FAILURE"): break time.sleep(0.5) metrics = adapter.get_metrics() print(f"queue_depth: {metrics['queue_depth']}") print(f"in_flight: {metrics['in_flight']}") print(f"dlq_depth: {metrics['dlq_depth']}") print(f"tasks_enqueued_total: {metrics['tasks_enqueued_total']}") print(f"tasks_completed_total: {metrics['tasks_completed_total']}") print(f"duration: {metrics['task_processing_duration_seconds']}") finally: adapter.stop_consumer() if __name__ == "__main__": main() ``` The three gauges — `queue_depth`, `in_flight`, and `dlq_depth` — are queried live from the broker on each `get_metrics()` call. The counters and the duration histogram are accumulated in memory per adapter process, so they reset when the process restarts. ## Monitoring queue depth over time [#monitoring-queue-depth-over-time] `queue_depth` is the most important gauge: it is the same waiting-message count the [autoscaling policy](/integrations/rayserve/how-to/autoscaling) scales on. Sampling it on an interval shows the queue filling as tasks arrive and draining as the consumer works. The `queue_depth_monitoring.py` example enqueues a batch with the consumer stopped (depth climbs), then starts the consumer (depth falls). ```python title="queue_depth_monitoring.py" depth_history: list[int] = [] # Phase 1: enqueue without a consumer — depth increases for i in range(8): adapter.enqueue_task_sync("slow_task", args=[i]) depth_history.append(adapter.get_metrics()["queue_depth"]) # Phase 2: start the consumer — depth decreases adapter.start_consumer() for _ in range(12): time.sleep(0.5) depth = adapter.get_metrics()["queue_depth"] depth_history.append(depth) if depth == 0: break # queue drained print(f"Peak depth: {max(depth_history)} Final: {depth_history[-1]}") ``` A rising `queue_depth` with a flat `in_flight` means consumers are saturated — they cannot pull work faster than it arrives. That is the signal to add replicas, which the [queue-depth autoscaling policy](/integrations/rayserve/how-to/autoscaling) does automatically. ## Processing-duration statistics [#processing-duration-statistics] `task_processing_duration_seconds` is a histogram tracked with O(1) running statistics — `min`, `max`, `avg`, and `count` — rather than storing every sample, so it stays cheap under sustained load. The `processing_duration_stats.py` example runs a mix of fast, medium, and slow handlers and reads the aggregate. ```python title="processing_duration_stats.py" # After running a mix of ~0.05s, ~0.20s, and ~0.50s handlers: duration = adapter.get_metrics()["task_processing_duration_seconds"] print(f"count: {duration['count']}") # total tasks processed print(f"min: {duration['min']:.4f}s") print(f"max: {duration['max']:.4f}s") print(f"avg: {duration['avg']:.4f}s") ``` Before any task completes, the histogram reports all-zero stats. The `count` field doubles as a processed-task total, which pairs with `tasks_enqueued_total` to spot a backlog (enqueued growing faster than count). ## Periodic collection for dashboards [#periodic-collection-for-dashboards] For a dashboard or scraper, poll `get_metrics()` on a fixed interval and forward the dict to your metrics sink. The `metrics_dashboard.py` example formats a periodic snapshot; the same loop feeds a Prometheus pushgateway, a StatsD client, or a log line. ```python title="metrics_dashboard.py (pattern)" import time while running: m = adapter.get_metrics() publish_gauge("rayserve_queue_depth", m["queue_depth"]) publish_gauge("rayserve_in_flight", m["in_flight"]) publish_gauge("rayserve_dlq_depth", m["dlq_depth"]) publish_counter("rayserve_tasks_enqueued_total", m["tasks_enqueued_total"]) time.sleep(5) ``` The counters and histogram are per-process and in-memory — they do **not** survive a restart and are **not** aggregated across replicas. To chart a fleet, scrape each replica and aggregate in your metrics backend (sum the counters, take the max of `queue_depth`). The gauges, being live broker reads, report the same shared queue from every replica. ## What to alert on [#what-to-alert-on] | Signal | Condition | Why | | ------------------------------- | ----------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------- | | `dlq_depth` | Any sustained growth | Tasks are exhausting retries — a model or input is broken. Pair with [`on_dlq`](/integrations/rayserve/how-to/dlq-and-retries) for per-task alerts. | | `queue_depth` | High and flat while `in_flight` is capped | Consumers are saturated at `maxReplicaCount`; the backlog is growing. | | `result_storage_retries_total` | Increasing | Result writes are failing — often an oversized result exceeding `max_send_size`. | | `consumer_poll_latency_seconds` | Spiking | The broker is slow or unreachable; the consumer loop is stalling. | ## Related [#related]