Metrics
Collect the adapter's 8-metric dict — queue depth, in-flight, DLQ depth, counters, and duration stats — for dashboards and alerting.
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.
Prerequisites
kubemq-rayserveinstalled, with an initializedKubeMQTaskProcessorAdapter(see Getting Started with Ray Serve)- A running KubeMQ broker reachable from the adapter, since three of the eight metrics are read live from the broker
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.
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
queue_depth is the most important gauge: it is the same waiting-message count the
autoscaling policy 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).
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 does automatically.
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.
# 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
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.
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
| Signal | Condition | Why |
|---|---|---|
dlq_depth | Any sustained growth | Tasks are exhausting retries — a model or input is broken. Pair with on_dlq 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
Was this page helpful?