feat: add vllm sidecar heartbeat and load metrics reporting.

vllm_sidecar/README.md

@@ -97,9 +97,48 @@ kill <sidecar_pid> && sleep 7
 etcdctl get --prefix XLLM:DEFAULT:              # empty
 ```
 
+## Heartbeat (load metrics)
+
+Alongside lease keepalive, the sidecar scrapes vLLM's Prometheus `/metrics` and
+POSTs a heartbeat to the master every `--heartbeat-interval` seconds:
+
+    POST <xllm-service-url>/v1/internal/heartbeat
+    X-Internal-Token: <token>            # only if configured
+    {"name","incarnation_id","load_metrics":{...},"latency_metrics":{...}}
+
+The master reuses the same `handle_instance_heartbeat` path as the brpc backends:
+the JSON is parsed into a `proto::HeartbeatRequest` and feeds the scheduler's
+in-memory load metrics (which the master then publishes to `XLLM:LOADMETRICS:`
+for non-master nodes). Mapping (pinned to vLLM 0.21):
+
+| vLLM metric | heartbeat field |
+|---|---|
+| `vllm:num_requests_waiting` | `load_metrics.waiting_requests_num` |
+| `vllm:gpu_cache_usage_perc` | `load_metrics.gpu_cache_usage_perc` |
+| `vllm:time_to_first_token_seconds` | `latency_metrics.recent_max_ttft` (ms, interval avg) |
+| `vllm:time_per_output_token_seconds` | `latency_metrics.recent_max_tbt` (ms, interval avg) |
+
+Heartbeat is **metrics-only**: liveness stays with the lease. A `409` (unknown /
+stale incarnation) makes the sidecar re-register; a transient `/metrics` or POST
+failure is logged and skipped — the lease still holds the instance up.
+
+### Heartbeat flags
+
+| flag | default | notes |
+|---|---|---|
+| `--xllm-service-url` | `http://127.0.0.1:9998` | master HTTP base for `/v1/internal/heartbeat` |
+| `--internal-token` | `""` | `X-Internal-Token`; must match master `--internal_api_token`; also `XLLM_INTERNAL_TOKEN` |
+| `--heartbeat-interval` | `3` | seconds between metrics heartbeats |
+| `--metrics-url` | `<vllm-url>/metrics` | vLLM Prometheus endpoint |
+
 ## Scope
 
-This PR provides **liveness-only** auto-registration via the etcd lease. Load
-metrics (queue depth, KV-cache usage), an HTTP `/register`+`/heartbeat` contract
-on the master, and `X-Internal-Token` auth are intentionally left to a follow-up
-PR — none are needed for single-instance routing today.
+Registration is lease-based (liveness); heartbeat adds load/latency metrics for
+load-aware routing. The KV-cache event bridge (CacheAwareRouting) and
+disaggregated-PD linking remain out of scope.
+
+
+
+## Internal heartbeat security
+
+The sidecar posts load metrics to `/v1/internal/heartbeat`. Expose this endpoint only on trusted networks. For production deployments, set `--internal_api_token` on the master and pass the same value to the sidecar with `--internal-token` or `XLLM_INTERNAL_TOKEN`.

vllm_sidecar/metrics.py

@@ -0,0 +1,112 @@
+# Copyright 2026 The xLLM Authors. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://github.com/jd-opensource/xllm-service/blob/main/LICENSE
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ==============================================================================
+"""Scrape vLLM's Prometheus `/metrics` into the xllm-service heartbeat schema.
+
+Maps vLLM gauges/histograms onto proto HeartbeatRequest fields (see
+proto/xllm_rpc_service.proto LoadMetrics / LatencyMetrics):
+
+    vllm:num_requests_waiting        -> load_metrics.waiting_requests_num
+    vllm:gpu_cache_usage_perc        -> load_metrics.gpu_cache_usage_perc
+    vllm:time_to_first_token_seconds -> latency_metrics.recent_max_ttft (ms)
+    vllm:time_per_output_token_seconds -> latency_metrics.recent_max_tbt (ms)
+
+Latency histograms only expose _sum/_count, so we report the per-interval
+average (delta_sum/delta_count, in ms) as a pragmatic proxy for the proto's
+"recent max" -- good enough for load-aware routing; a true max would need
+bucket analysis. Metric names are pinned to vLLM 0.21.
+"""
+
+import logging
+import math
+import re
+
+import requests
+
+logger = logging.getLogger("vllm_sidecar.metrics")
+
+# `name{labels} value [timestamp]` -> (base_name, value)
+_SAMPLE_RE = re.compile(
+    r"^(?P<name>[a-zA-Z_:][\w:]*)(?:\{[^}]*\})?\s+"
+    r"(?P<value>[-+]?[0-9.eE+-]+|NaN|[-+]?Inf)\s*"
+)
+
+_WAITING = "vllm:num_requests_waiting"
+_GPU_CACHE = ("vllm:gpu_cache_usage_perc", "vllm:kv_cache_usage_perc")
+_TTFT = "vllm:time_to_first_token_seconds"
+_TBT = "vllm:time_per_output_token_seconds"
+
+
+def _parse_samples(text: str) -> dict[str, float]:
+    """Sum sample values by base metric name (ignoring labels / HELP lines)."""
+    totals = {}
+    for line in text.splitlines():
+        if not line or line[0] == "#":
+            continue
+        m = _SAMPLE_RE.match(line)
+        if not m:
+            continue
+        try:
+            val = float(m.group("value"))
+        except ValueError:
+            continue
+        # Drop NaN/Inf so downstream int()/JSON on these samples can't crash.
+        if not math.isfinite(val):
+            continue
+        totals[m.group("name")] = totals.get(m.group("name"), 0.0) + val
+    return totals
+
+
+class VllmMetricsScraper:
+    def __init__(self, metrics_url: str, timeout: float = 3.0) -> None:
+        self._url = metrics_url
+        self._timeout = timeout
+        self._prev = {}  # histogram base -> (sum, count) from the last scrape
+        # Reuse one connection across periodic scrapes (HTTP keep-alive).
+        self._session = requests.Session()
+
+    def _interval_avg_ms(self, samples: dict[str, float], base: str) -> int:
+        """Average over the interval since the last scrape, in ms (0 if none)."""
+        cur = (samples.get(base + "_sum", 0.0), samples.get(base + "_count", 0.0))
+        prev = self._prev.get(base)
+        self._prev[base] = cur
+        if prev is None:
+            return 0  # first observation: establish baseline, no interval yet
+        d_count = cur[1] - prev[1]
+        if d_count <= 0:
+            return 0
+        d_sum = cur[0] - prev[0]
+        return max(0, int((d_sum / d_count) * 1000.0))
+
+    def scrape(self) -> dict | None:
+        """Return the heartbeat metrics dict, or None if /metrics unreachable."""
+        try:
+            r = self._session.get(self._url, timeout=self._timeout)
+            r.raise_for_status()
+        except requests.RequestException as e:
+            logger.debug("metrics scrape failed: %s", e)
+            return None
+
+        s = _parse_samples(r.text)
+        gpu_cache = next((s[n] for n in _GPU_CACHE if n in s), 0.0)
+        return {
+            "load_metrics": {
+                "waiting_requests_num": int(s.get(_WAITING, 0.0)),
+                "gpu_cache_usage_perc": float(gpu_cache),
+            },
+            "latency_metrics": {
+                "recent_max_ttft": self._interval_avg_ms(s, _TTFT),
+                "recent_max_tbt": self._interval_avg_ms(s, _TBT),
+            },
+        }

vllm_sidecar/sidecar.py

@@ -30,12 +30,16 @@
 import os
 import signal
 import threading
+import time
 import uuid
 from types import FrameType
 
+import requests
+
 from .etcd_registry import EtcdGatewayClient, EtcdError
 from .health import VllmHealthProbe
 from .meta import InstanceType, build_instance_key, build_instance_meta
+from .metrics import VllmMetricsScraper
 
 logger = logging.getLogger("vllm_sidecar")
 
@@ -57,6 +61,14 @@ def __init__(self, args: argparse.Namespace) -> None:
         self._stop = threading.Event()
         self._lease_id = None
         self._incarnation_id = None
+        # heartbeat (LoadMetrics/LatencyMetrics) -> master HTTP endpoint
+        self._hb_url = args.xllm_service_url.rstrip("/") + "/v1/internal/heartbeat"
+        self._metrics = VllmMetricsScraper(
+            args.metrics_url, timeout=args.health_timeout
+        )
+        self._last_hb = 0.0
+        # Reuse one connection for the periodic heartbeat POSTs (keep-alive).
+        self._hb_session = requests.Session()
 
     # --- registration primitives ------------------------------------------
 
@@ -131,6 +143,7 @@ def run(self) -> None:
             if self._health.is_healthy():
                 fail = 0
                 self._keepalive_or_reregister()
+                self._maybe_heartbeat()
             else:
                 fail += 1
                 logger.warning(
@@ -173,6 +186,51 @@ def _on_signal(self, signum: int, _frame: FrameType | None) -> None:
         logger.info("received signal %d, shutting down", signum)
         self._stop.set()
 
+    # --- heartbeat (metrics) ----------------------------------------------
+
+    def _maybe_heartbeat(self) -> None:
+        """POST LoadMetrics/LatencyMetrics on cadence while registered."""
+        if not self._registered:
+            return
+        now = time.monotonic()
+        if now - self._last_hb < self._args.heartbeat_interval:
+            return
+        self._last_hb = now
+        self._send_heartbeat()
+
+    def _send_heartbeat(self) -> None:
+        metrics = self._metrics.scrape()
+        if metrics is None:
+            return  # /metrics transiently unreachable; lease still holds liveness
+        body = {
+            "name": self._args.register_addr,
+            "incarnation_id": self._incarnation_id,
+            "load_metrics": metrics["load_metrics"],
+            "latency_metrics": metrics["latency_metrics"],
+        }
+        headers = {"Content-Type": "application/json"}
+        if self._args.internal_token:
+            headers["X-Internal-Token"] = self._args.internal_token
+        try:
+            r = self._hb_session.post(
+                self._hb_url,
+                json=body,
+                headers=headers,
+                timeout=self._args.health_timeout,
+            )
+        except requests.RequestException as e:
+            logger.debug("heartbeat POST failed: %s", e)
+            return
+        if r.status_code == 409:
+            # master doesn't know this incarnation -> re-register and resync id
+            logger.warning("heartbeat 409 (stale/unknown), re-registering")
+            self._lease_id = None
+            self._register()
+        elif r.status_code == 401:
+            logger.error("heartbeat 401: invalid --internal-token")
+        elif r.status_code != 200:
+            logger.warning("heartbeat -> HTTP %d: %s", r.status_code, r.text[:120])
+
 
 def build_parser() -> argparse.ArgumentParser:
     p = argparse.ArgumentParser(
@@ -234,6 +292,28 @@ def build_parser() -> argparse.ArgumentParser:
         default=3,
         help="consecutive failed probes before deregistering",
     )
+    # heartbeat (LoadMetrics/LatencyMetrics reporting)
+    p.add_argument(
+        "--xllm-service-url",
+        default="http://127.0.0.1:9998",
+        help="master HTTP base URL for /v1/internal/heartbeat",
+    )
+    p.add_argument(
+        "--internal-token",
+        default=os.environ.get("XLLM_INTERNAL_TOKEN", ""),
+        help="X-Internal-Token; must match master --internal_api_token",
+    )
+    p.add_argument(
+        "--heartbeat-interval",
+        type=float,
+        default=3.0,
+        help="seconds between metrics heartbeats",
+    )
+    p.add_argument(
+        "--metrics-url",
+        default=None,
+        help="vLLM Prometheus endpoint (default: <vllm-url>/metrics)",
+    )
     p.add_argument("--log-level", default="INFO")
     return p
 
@@ -252,6 +332,8 @@ def main(argv: list[str] | None = None) -> int:
     )
     if not args.register_addr:
         args.register_addr = _derive_addr(args.vllm_url)
+    if not args.metrics_url:
+        args.metrics_url = args.vllm_url.rstrip("/") + "/metrics"
     logger.info(
         "sidecar starting: register %s as backend_type=%s type=%s",
         args.register_addr,

vllm_sidecar/tests/test_metrics.py

@@ -0,0 +1,81 @@
+# Copyright 2026 The xLLM Authors. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://github.com/jd-opensource/xllm-service/blob/main/LICENSE
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ==============================================================================
+"""Unit tests for vLLM /metrics parsing (no HTTP needed)."""
+
+from vllm_sidecar.metrics import VllmMetricsScraper, _parse_samples
+
+SAMPLE = """\
+# HELP vllm:num_requests_waiting Number of requests waiting.
+# TYPE vllm:num_requests_waiting gauge
+vllm:num_requests_waiting{model_name="qwen2.5-7b"} 4.0
+# TYPE vllm:gpu_cache_usage_perc gauge
+vllm:gpu_cache_usage_perc{model_name="qwen2.5-7b"} 0.37
+# TYPE vllm:time_to_first_token_seconds histogram
+vllm:time_to_first_token_seconds_sum{model_name="qwen2.5-7b"} 2.0
+vllm:time_to_first_token_seconds_count{model_name="qwen2.5-7b"} 10.0
+vllm:time_per_output_token_seconds_sum{model_name="qwen2.5-7b"} 1.0
+vllm:time_per_output_token_seconds_count{model_name="qwen2.5-7b"} 100.0
+"""
+
+
+def test_parse_samples_strips_labels_and_comments() -> None:
+    s = _parse_samples(SAMPLE)
+    assert s["vllm:num_requests_waiting"] == 4.0
+    assert s["vllm:gpu_cache_usage_perc"] == 0.37
+    assert "vllm:num_requests_waiting" in s
+    # HELP/TYPE comment lines must not appear
+    assert all(not k.startswith("#") for k in s)
+
+
+def test_parse_samples_sums_duplicate_label_sets() -> None:
+    text = (
+        'vllm:num_requests_waiting{m="a"} 2.0\nvllm:num_requests_waiting{m="b"} 3.0\n'
+    )
+    assert _parse_samples(text)["vllm:num_requests_waiting"] == 5.0
+
+
+def test_parse_samples_handles_scientific_notation() -> None:
+    # Small latency values are emitted in scientific notation with a negative
+    # exponent; the value must parse fully, not be truncated at "1.23e" -> 0.
+    text = 'vllm:time_to_first_token_seconds_sum{m="a"} 1.23e-04\n'
+    assert _parse_samples(text)["vllm:time_to_first_token_seconds_sum"] == 1.23e-04
+
+
+def test_parse_samples_drops_nan_and_inf() -> None:
+    # vLLM may emit NaN/Inf; these must be dropped so downstream int()/JSON
+    # cannot crash the scrape loop.
+    text = (
+        "vllm:num_requests_waiting NaN\n"
+        "vllm:gpu_cache_usage_perc +Inf\n"
+        "vllm:other 2.0\n"
+    )
+    s = _parse_samples(text)
+    assert "vllm:num_requests_waiting" not in s
+    assert "vllm:gpu_cache_usage_perc" not in s
+    assert s["vllm:other"] == 2.0
+
+
+def test_interval_avg_ms_uses_delta() -> None:
+    sc = VllmMetricsScraper("http://unused")
+    # first call establishes the baseline -> 0
+    assert sc._interval_avg_ms({"h_sum": 2.0, "h_count": 10.0}, "h") == 0
+    # next: delta_sum=3.0 over delta_count=5 -> 0.6s avg -> 600 ms
+    assert sc._interval_avg_ms({"h_sum": 5.0, "h_count": 15.0}, "h") == 600
+
+
+def test_interval_avg_ms_zero_when_no_new_samples() -> None:
+    sc = VllmMetricsScraper("http://unused")
+    sc._interval_avg_ms({"h_sum": 2.0, "h_count": 10.0}, "h")
+    assert sc._interval_avg_ms({"h_sum": 2.0, "h_count": 10.0}, "h") == 0