diff --git a/vllm_sidecar/.gitignore b/vllm_sidecar/.gitignore new file mode 100644 index 0000000..7a60b85 --- /dev/null +++ b/vllm_sidecar/.gitignore @@ -0,0 +1,2 @@ +__pycache__/ +*.pyc diff --git a/vllm_sidecar/README.md b/vllm_sidecar/README.md new file mode 100644 index 0000000..7a82eef --- /dev/null +++ b/vllm_sidecar/README.md @@ -0,0 +1,105 @@ +# vLLM Sidecar — auto-register a vLLM instance into xllm-service + +A vLLM server speaks only HTTP/OpenAI. It does not write etcd, hold a lease, or +emit heartbeats — so on its own it cannot join an xllm-service cluster. This +sidecar runs next to a vLLM process and bridges that gap, **with no C++ changes**: +xllm-service already discovers instances by watching the `XLLM:DEFAULT:` etcd +prefix. + +It replaces the manual `demo/register_vllm.sh` (a one-shot `etcdctl put`) with a +long-running process that keeps registration in lockstep with vLLM's health. + +## Lifecycle + +``` + ┌─────────────┐ /health up ┌────────────────────────────┐ + start ──▶ │ wait health │ ───────────────▶│ grant lease(ttl) + put key │ + └─────────────┘ └──────────────┬─────────────┘ + ▲ │ every keepalive-interval: + │ health recovers │ probe /health + │ (re-register, new incarnation) ▼ + ┌─────┴───────┐ N consecutive fails ┌──────────────────────┐ + │ deregistered│ ◀──────────────────────│ healthy → keepalive │ + └─────────────┘ │ unhealthy → count++ │ + ▲ └──────────────────────┘ + │ SIGTERM/SIGINT → revoke lease (immediate deregister) +``` + +* **health-gate** — never registers until vLLM `/health` is up. +* **lease, not heartbeat** — the etcd lease *is* the liveness signal. Keepalive + refreshes the TTL **without rewriting the key**, so the master's watcher sees + only the initial `PUT` and the final `DELETE` (on revoke or TTL expiry); no + watch churn. +* **fast + safe deregister** — on N failed probes or a signal, the lease is + revoked so the key disappears at once; if the sidecar is killed `-9`, the key + still expires within `--lease-ttl`. +* **clean restart** — each (re)registration uses a fresh `incarnation_id` + (uuid4), which the master uses to ignore stale deletes for a replaced instance + and to clean up the previous incarnation. + +## How the master picks it up (no C++ changes) + +| step | code | +|---|---| +| watch `XLLM:DEFAULT:` etc. | `instance_mgr.cpp:133` `add_watch(prefix, ...)` | +| `PUT` → register | `instance_mgr.cpp:568` `update_instance_metainfo` → `register_instance` | +| lease expiry → `DELETE` → probe → suspect/remove | `instance_mgr.cpp:615+` | +| JSON schema | `common/types.h:248` `parse_from_json` (needs `name`,`rpc_address`,`type`) | +| key prefix / namespace | `instance_mgr.cpp:45` map, `utils.cpp:105` namespace | + +> **Constraint:** a single vLLM instance with no decode peer is routable only as +> `type=DEFAULT (0)` — keep `--instance-type=DEFAULT`. This mirrors the M2 finding. + +## Install + +```bash +pip install -r vllm_sidecar/requirements.txt # only `requests` +``` + +## Run + +```bash +python -m vllm_sidecar.sidecar \ + --etcd-endpoints 127.0.0.1:2379 \ + --vllm-url http://127.0.0.1:18000 \ + --register-addr 127.0.0.1:18000 # host:port, NO scheme +``` + +`--register-addr` defaults to the host:port derived from `--vllm-url`. The master +prepends `http://` itself (`init_brpc_channel`), so pass it bare. + +### Flags + +| flag | default | notes | +|---|---|---| +| `--etcd-endpoints` | `127.0.0.1:2379` | comma-separated; also `ETCD_ENDPOINTS` | +| `--etcd-namespace` | `""` | **must match** master's `--etcd_namespace` | +| `--etcd-username` / `--etcd-password` | `""` | also `ETCD_USERNAME`/`ETCD_PASSWORD` | +| `--vllm-url` | `http://127.0.0.1:18000` | local vLLM base URL | +| `--register-addr` | derived | host:port the master dials, no scheme | +| `--backend-type` | `vllm` | written into InstanceMetaInfo | +| `--instance-type` | `DEFAULT` | keep DEFAULT for a single instance | +| `--instance-name` | `vllm` | prefix for incarnation id / logs | +| `--lease-ttl` | `6` | seconds (≈ 2× keepalive interval) | +| `--keepalive-interval` | `2` | seconds between probe + lease refresh | +| `--health-fail-threshold` | `3` | failed probes before deregister | + +## Verify + +```bash +# instance appears (with a lease) and is reachable through xllm-service +etcdctl get --prefix XLLM:DEFAULT: +etcdctl lease list +curl http://127.0.0.1:9998/v1/models # -> backend model list + +# kill the sidecar; key disappears within lease-ttl, master logs removal +kill && sleep 7 +etcdctl get --prefix XLLM:DEFAULT: # empty +``` + +## 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. diff --git a/vllm_sidecar/__init__.py b/vllm_sidecar/__init__.py new file mode 100644 index 0000000..5ec2de4 --- /dev/null +++ b/vllm_sidecar/__init__.py @@ -0,0 +1,28 @@ +# 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. +# ============================================================================== +"""vLLM sidecar: auto-register a vLLM instance into xllm-service via etcd lease. + +A vLLM process speaks only HTTP/OpenAI -- it does not write etcd, hold a lease, +or emit heartbeats. This sidecar runs alongside vLLM and bridges that gap: + + * health-gate -- only register once vLLM `/health` is up + * register -- write an InstanceMetaInfo JSON under an etcd lease + * keep-alive -- refresh the lease while vLLM stays healthy + * deregister -- revoke the lease on vLLM failure or graceful shutdown, + so xllm-service's etcd watcher removes the instance + +This replaces the manual `demo/register_vllm.sh` with no C++ changes: the master +already discovers instances by watching the `XLLM:DEFAULT:` etcd prefix. +""" diff --git a/vllm_sidecar/etcd_registry.py b/vllm_sidecar/etcd_registry.py new file mode 100644 index 0000000..f0c6d92 --- /dev/null +++ b/vllm_sidecar/etcd_registry.py @@ -0,0 +1,141 @@ +# 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. +# ============================================================================== +"""Minimal etcd v3 client over the gRPC-gateway JSON/HTTP API. + +We deliberately avoid `python-etcd3` (grpcio/protobuf native deps are brittle +inside the shared GPU container). etcd exposes the full v3 KV/Lease API as JSON +over the same client port (default 2379), verified against etcd 3.4.30: + + POST /v3/lease/grant {"TTL": , "ID": 0} -> {"ID","TTL"} + POST /v3/kv/put {"key","value","lease"} (key/value base64) + POST /v3/lease/keepalive {"ID": } -> {"result":{"TTL"}} + POST /v3/lease/revoke {"ID": } + POST /v3/kv/range {"key": } -> {"kvs":[...]} | {} + +Lease keep-alive refreshes the TTL WITHOUT touching the key, so it generates no +watch event -- the master's watcher only sees the initial PUT and the final +DELETE (on lease expiry/revoke). Only `requests` is required. +""" + +import base64 +import logging + +import requests + +logger = logging.getLogger("vllm_sidecar.etcd") + + +class EtcdError(RuntimeError): + pass + + +def _b64(s: str) -> str: + return base64.b64encode(s.encode("utf-8")).decode("ascii") + + +class EtcdGatewayClient: + """Lease-oriented etcd v3 client. Endpoints are tried in order per call.""" + + def __init__( + self, + endpoints: str | list[str], + username: str = "", + password: str = "", + timeout: float = 3.0, + ) -> None: + # `endpoints` may be a comma-separated string or a list of host:port. + if isinstance(endpoints, str): + endpoints = [e.strip() for e in endpoints.split(",") if e.strip()] + # Keep an explicit scheme if present; only default to http:// otherwise, + # so endpoints like "https://host:2379" are not turned into + # "http://https://host:2379". + self._bases = [ + e if e.startswith(("http://", "https://")) else "http://" + e + for e in (endpoint.rstrip("/") for endpoint in endpoints) + ] + if not self._bases: + raise ValueError("at least one etcd endpoint is required") + self._timeout = timeout + self._session = requests.Session() + if username: + self._authenticate(username, password) + + def _authenticate(self, username: str, password: str) -> None: + resp = self._post( + "/v3/auth/authenticate", {"name": username, "password": password} + ) + token = resp.get("token") + if not token: + raise EtcdError("etcd authenticate returned no token") + # etcd v3 gateway expects the token in the Authorization header. + self._session.headers["Authorization"] = token + + def _post(self, path: str, body: dict) -> dict: + last_err = None + for base in self._bases: + try: + r = self._session.post(base + path, json=body, timeout=self._timeout) + if r.status_code == 200: + try: + data = r.json() + except ValueError as e: + last_err = EtcdError(f"invalid JSON response from {path}: {e}") + continue + # Every v3 gateway endpoint returns a JSON object; coerce any + # other shape (null/list) to {} so callers can rely on .get(). + return data if isinstance(data, dict) else {} + last_err = EtcdError(f"{path} -> HTTP {r.status_code}: {r.text[:200]}") + except requests.RequestException as e: # connection/timeout + last_err = e + raise EtcdError(f"all etcd endpoints failed for {path}: {last_err}") + + # --- lease lifecycle --------------------------------------------------- + + def lease_grant(self, ttl_seconds: int) -> str: + """Grant a lease; returns the lease id (int64 as a decimal string).""" + resp = self._post("/v3/lease/grant", {"TTL": ttl_seconds, "ID": 0}) + lease_id = resp.get("ID") + if not lease_id: + raise EtcdError(f"lease grant returned no ID: {resp}") + return lease_id + + def lease_keepalive(self, lease_id: str) -> int: + """Refresh a lease once; returns the remaining TTL (0 == lease gone).""" + resp = self._post("/v3/lease/keepalive", {"ID": lease_id}) + # "result" may be present but null (proto3 JSON for an empty message), + # and "TTL" itself may be missing or null; treat all of these as 0. + result = resp.get("result") or {} + return int(result.get("TTL") or 0) + + def lease_revoke(self, lease_id: str) -> None: + """Revoke a lease -> its key is deleted immediately.""" + self._post("/v3/lease/revoke", {"ID": lease_id}) + + # --- kv ---------------------------------------------------------------- + + def put(self, key: str, value: str, lease_id: str) -> None: + self._post( + "/v3/kv/put", {"key": _b64(key), "value": _b64(value), "lease": lease_id} + ) + + def get(self, key: str) -> str | None: + """Return the string value at ``key`` or None if absent.""" + resp = self._post("/v3/kv/range", {"key": _b64(key)}) + kvs = resp.get("kvs") + if not kvs: + return None + # proto3 JSON omits empty values, so "value" may be absent for an + # empty string; treat that as "". + return base64.b64decode(kvs[0].get("value", "")).decode("utf-8") diff --git a/vllm_sidecar/health.py b/vllm_sidecar/health.py new file mode 100644 index 0000000..2c62cca --- /dev/null +++ b/vllm_sidecar/health.py @@ -0,0 +1,53 @@ +# 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. +# ============================================================================== +"""vLLM health probing. + +The lease only proves the sidecar is alive; it says nothing about vLLM. We gate +registration and lease refresh on vLLM's own `/health` endpoint so that a dead +vLLM (with the sidecar still running) does not leave a phantom instance routable. +""" + +import logging + +import requests + +logger = logging.getLogger("vllm_sidecar.health") + + +class VllmHealthProbe: + def __init__(self, vllm_url: str, timeout: float = 3.0) -> None: + self._base = vllm_url.rstrip("/") + self._timeout = timeout + # Reuse one connection across the periodic probes (HTTP keep-alive) to + # avoid piling up TIME_WAIT sockets on the host. + self._session = requests.Session() + + def is_healthy(self) -> bool: + """True iff vLLM `/health` returns 2xx within the timeout.""" + try: + r = self._session.get(self._base + "/health", timeout=self._timeout) + return 200 <= r.status_code < 300 + except requests.RequestException as e: + logger.debug("vLLM health probe failed: %s", e) + return False + + def served_model(self) -> str | None: + """Best-effort first model id from `/v1/models`, for logging only.""" + try: + r = self._session.get(self._base + "/v1/models", timeout=self._timeout) + data = r.json().get("data", []) + return data[0]["id"] if data else None + except (requests.RequestException, ValueError, KeyError, IndexError): + return None diff --git a/vllm_sidecar/meta.py b/vllm_sidecar/meta.py new file mode 100644 index 0000000..4c95d25 --- /dev/null +++ b/vllm_sidecar/meta.py @@ -0,0 +1,101 @@ +# 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. +# ============================================================================== +"""InstanceMetaInfo schema + etcd key construction (Python mirror of C++). + +The single source of truth for the wire schema is the C++ struct +`InstanceMetaInfo` in `xllm_service/common/types.h`: + + * `serialize_to_json()` (types.h ~:226) -- what the master writes, + * `parse_from_json()` (types.h ~:248) -- what the watcher reads. It requires + `name`, `rpc_address`, `type` (`.at()`); everything else is optional and + defaults (e.g. `backend_type` -> "xllm"). + +We therefore emit exactly the subset the watcher needs, matching the field +names and value encodings used by C++. The etcd key layout and namespace +normalization mirror `instance_mgr.cpp` (`ETCD_KEYS_PREFIX_MAP`) and +`utils.cpp` (`normalize_etcd_namespace` / `build_etcd_key_with_namespace`). +""" + +import time +from enum import IntEnum + +# Mirror of `enum class InstanceType` in common/types.h (DEFAULT = 0, ...). +# Only DEFAULT is exercised today; a single vLLM instance with no decode peer is +# routable only as DEFAULT (see the M2 routing constraint). + + +class InstanceType(IntEnum): + DEFAULT = 0 + PREFILL = 1 + DECODE = 2 + MIX = 3 + + +# Mirror of `ETCD_KEYS_PREFIX_MAP` in scheduler/managers/instance_mgr.cpp:45. +ETCD_KEYS_PREFIX_MAP = { + InstanceType.DEFAULT: "XLLM:DEFAULT:", + InstanceType.PREFILL: "XLLM:PREFILL:", + InstanceType.DECODE: "XLLM:DECODE:", + InstanceType.MIX: "XLLM:MIX:", +} + + +def normalize_etcd_namespace(etcd_namespace: str) -> str: + """Port of utils::normalize_etcd_namespace (utils.cpp:105). + + "" -> ""; "foo" -> "/foo/"; "/a/b/" -> "/a/b/"; "///" -> "". + """ + if not etcd_namespace: + return "" + trimmed = etcd_namespace.strip("/") + if not trimmed: + return "" + return "/" + trimmed + "/" + + +def build_instance_key( + addr: str, + instance_type: InstanceType = InstanceType.DEFAULT, + etcd_namespace: str = "", +) -> str: + """Full etcd key the master watches, e.g. ``XLLM:DEFAULT:127.0.0.1:18000``. + + ``addr`` is the address the master uses to reach the backend -- host:port + with NO scheme; xllm-service selects HTTP via the channel's protocol option. + """ + logical_key = ETCD_KEYS_PREFIX_MAP[instance_type] + addr + return normalize_etcd_namespace(etcd_namespace) + logical_key + + +def build_instance_meta( + addr: str, + incarnation_id: str, + instance_type: InstanceType = InstanceType.DEFAULT, + backend_type: str = "vllm", +) -> dict: + """Build the InstanceMetaInfo JSON payload stored under the lease. + + Matches `register_vllm.sh` and the required fields of + `InstanceMetaInfo::parse_from_json`. ``register_ts_ms`` is real epoch-ms so + the master's logs/ordering are meaningful (the manual script hard-coded 1). + """ + return { + "name": addr, + "rpc_address": addr, + "type": int(instance_type), + "backend_type": backend_type, + "incarnation_id": incarnation_id, + "register_ts_ms": int(time.time() * 1000), + } diff --git a/vllm_sidecar/requirements.txt b/vllm_sidecar/requirements.txt new file mode 100644 index 0000000..78ca561 --- /dev/null +++ b/vllm_sidecar/requirements.txt @@ -0,0 +1,3 @@ +# Only stdlib + requests. We use etcd's v3 gRPC-gateway JSON/HTTP API rather +# than python-etcd3 to avoid brittle grpcio/protobuf native deps. +requests>=2.20 diff --git a/vllm_sidecar/sidecar.py b/vllm_sidecar/sidecar.py new file mode 100644 index 0000000..889de56 --- /dev/null +++ b/vllm_sidecar/sidecar.py @@ -0,0 +1,266 @@ +# 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. +# ============================================================================== +"""vLLM sidecar entrypoint: health-gated etcd lease registration loop. + +Run alongside a vLLM server: + + python -m vllm_sidecar.sidecar \ + --etcd-endpoints 127.0.0.1:2379 \ + --vllm-url http://127.0.0.1:18000 \ + --register-addr 127.0.0.1:18000 + +See README.md for the full flag list and the lifecycle contract. +""" + +import argparse +import json +import logging +import os +import signal +import threading +import uuid +from types import FrameType + +from .etcd_registry import EtcdGatewayClient, EtcdError +from .health import VllmHealthProbe +from .meta import InstanceType, build_instance_key, build_instance_meta + +logger = logging.getLogger("vllm_sidecar") + + +class Sidecar: + def __init__(self, args: argparse.Namespace) -> None: + self._args = args + self._etcd = EtcdGatewayClient( + args.etcd_endpoints, + username=args.etcd_username, + password=args.etcd_password, + timeout=args.etcd_timeout, + ) + self._health = VllmHealthProbe(args.vllm_url, timeout=args.health_timeout) + self._instance_type = InstanceType[args.instance_type] + self._key = build_instance_key( + args.register_addr, self._instance_type, args.etcd_namespace + ) + self._stop = threading.Event() + self._lease_id = None + self._incarnation_id = None + + # --- registration primitives ------------------------------------------ + + def _new_incarnation(self) -> str: + token = uuid.uuid4().hex[:12] + return ( + f"{self._args.instance_name}-{token}" if self._args.instance_name else token + ) + + def _register(self) -> bool: + """Grant a fresh lease and put the instance key. Returns success.""" + try: + incarnation = self._new_incarnation() + lease_id = self._etcd.lease_grant(self._args.lease_ttl) + meta = build_instance_meta( + self._args.register_addr, + incarnation, + self._instance_type, + self._args.backend_type, + ) + self._etcd.put(self._key, json.dumps(meta), lease_id) + self._lease_id = lease_id + self._incarnation_id = incarnation + logger.info( + "registered %s (incarnation=%s, lease=%s, ttl=%ds)", + self._key, + incarnation, + lease_id, + self._args.lease_ttl, + ) + return True + except EtcdError as e: + logger.warning("register failed, will retry: %s", e) + self._lease_id = None + return False + + def _deregister(self) -> None: + """Revoke the lease so the master removes the instance immediately.""" + if self._lease_id is None: + return + try: + self._etcd.lease_revoke(self._lease_id) + logger.info("deregistered %s (lease=%s revoked)", self._key, self._lease_id) + except EtcdError as e: + # On failure the lease still expires within ttl; not fatal. + logger.warning( + "revoke failed (lease expires in <=%ds): %s", self._args.lease_ttl, e + ) + finally: + self._lease_id = None + self._incarnation_id = None + + @property + def _registered(self) -> bool: + return self._lease_id is not None + + # --- main loop --------------------------------------------------------- + + def run(self) -> None: + signal.signal(signal.SIGTERM, self._on_signal) + signal.signal(signal.SIGINT, self._on_signal) + + self._wait_until_healthy() + if self._stop.is_set(): + return + model = self._health.served_model() + logger.info("vLLM healthy (model=%s), registering ...", model or "?") + self._register() + + fail = 0 + while not self._stop.wait(self._args.keepalive_interval): + if self._health.is_healthy(): + fail = 0 + self._keepalive_or_reregister() + else: + fail += 1 + logger.warning( + "vLLM health probe failed (%d/%d)", + fail, + self._args.health_fail_threshold, + ) + if self._registered and fail >= self._args.health_fail_threshold: + logger.error("vLLM unhealthy, deregistering") + self._deregister() + + self._deregister() + logger.info("sidecar stopped") + + def _wait_until_healthy(self) -> None: + backoff = 1.0 + while not self._stop.is_set() and not self._health.is_healthy(): + logger.info( + "waiting for vLLM at %s to become healthy ...", self._args.vllm_url + ) + self._stop.wait(backoff) + backoff = min(backoff * 2, self._args.lease_ttl) + + def _keepalive_or_reregister(self) -> None: + if not self._registered: + self._register() # recovered after a previous deregister + return + try: + ttl = self._etcd.lease_keepalive(self._lease_id) + if ttl <= 0: + logger.warning("lease %s lost (ttl=0), re-registering", self._lease_id) + self._lease_id = None + self._register() + except EtcdError as e: + logger.warning("keepalive failed, re-registering: %s", e) + self._lease_id = None + self._register() + + def _on_signal(self, signum: int, _frame: FrameType | None) -> None: + logger.info("received signal %d, shutting down", signum) + self._stop.set() + + +def build_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser( + prog="vllm_sidecar", + description="Auto-register a vLLM instance into xllm-service via etcd.", + ) + p.add_argument( + "--etcd-endpoints", + default=os.environ.get("ETCD_ENDPOINTS", "127.0.0.1:2379"), + help="comma-separated host:port list (default 127.0.0.1:2379)", + ) + p.add_argument( + "--etcd-namespace", + default=os.environ.get("ETCD_NAMESPACE", ""), + help="must match master's --etcd_namespace (default empty)", + ) + p.add_argument("--etcd-username", default=os.environ.get("ETCD_USERNAME", "")) + p.add_argument("--etcd-password", default=os.environ.get("ETCD_PASSWORD", "")) + p.add_argument("--etcd-timeout", type=float, default=3.0) + p.add_argument( + "--vllm-url", + default="http://127.0.0.1:18000", + help="base URL of the local vLLM server", + ) + p.add_argument( + "--register-addr", + default=None, + help="host:port the master uses to reach vLLM, NO scheme " + "(default: derived from --vllm-url)", + ) + p.add_argument("--backend-type", default="vllm") + p.add_argument( + "--instance-type", + default="DEFAULT", + choices=[t.name for t in InstanceType], + help="single vLLM instance must be DEFAULT to be routable", + ) + p.add_argument( + "--instance-name", + default="vllm", + help="human-readable prefix for the incarnation id / logs", + ) + p.add_argument( + "--lease-ttl", + type=int, + default=6, + help="etcd lease TTL in seconds (~2x keepalive interval)", + ) + p.add_argument( + "--keepalive-interval", + type=float, + default=2.0, + help="seconds between health probe + lease refresh", + ) + p.add_argument("--health-timeout", type=float, default=3.0) + p.add_argument( + "--health-fail-threshold", + type=int, + default=3, + help="consecutive failed probes before deregistering", + ) + p.add_argument("--log-level", default="INFO") + return p + + +def _derive_addr(vllm_url: str) -> str: + # strip scheme and any trailing path -> host:port + no_scheme = vllm_url.split("://", 1)[-1] + return no_scheme.split("/", 1)[0] + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + logging.basicConfig( + level=getattr(logging, args.log_level.upper(), logging.INFO), + format="%(asctime)s %(levelname)s %(name)s: %(message)s", + ) + if not args.register_addr: + args.register_addr = _derive_addr(args.vllm_url) + logger.info( + "sidecar starting: register %s as backend_type=%s type=%s", + args.register_addr, + args.backend_type, + args.instance_type, + ) + Sidecar(args).run() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/vllm_sidecar/tests/test_meta.py b/vllm_sidecar/tests/test_meta.py new file mode 100644 index 0000000..6cadbb8 --- /dev/null +++ b/vllm_sidecar/tests/test_meta.py @@ -0,0 +1,73 @@ +# 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 the InstanceMetaInfo / etcd-key Python mirror (no etcd needed). + +Guards against drift from the C++ source of truth: + * key layout / prefixes -- instance_mgr.cpp:45 ETCD_KEYS_PREFIX_MAP + * namespace normalization -- utils.cpp:105 normalize_etcd_namespace + * required JSON fields -- types.h:248 parse_from_json +""" + +import pytest + +from vllm_sidecar.meta import ( + InstanceType, + build_instance_key, + build_instance_meta, + normalize_etcd_namespace, +) + + +@pytest.mark.parametrize( + "raw,expected", + [ + ("", ""), + ("foo", "/foo/"), + ("/foo", "/foo/"), + ("foo/", "/foo/"), + ("/a/b/", "/a/b/"), + ("///", ""), + ], +) +def test_normalize_namespace_matches_cpp(raw, expected): + assert normalize_etcd_namespace(raw) == expected + + +def test_key_default_prefix_no_namespace(): + assert build_instance_key("127.0.0.1:18000") == "XLLM:DEFAULT:127.0.0.1:18000" + + +def test_key_with_namespace(): + key = build_instance_key("127.0.0.1:18000", InstanceType.DEFAULT, "ns") + assert key == "/ns/XLLM:DEFAULT:127.0.0.1:18000" + + +def test_meta_has_required_fields_and_values(): + meta = build_instance_meta("127.0.0.1:18000", "vllm-abc123") + # types.h parse_from_json requires these three (.at()). + for required in ("name", "rpc_address", "type"): + assert required in meta + assert meta["name"] == "127.0.0.1:18000" + assert meta["rpc_address"] == "127.0.0.1:18000" + assert meta["type"] == 0 # DEFAULT, the routable type for a lone instance + assert meta["backend_type"] == "vllm" + assert meta["incarnation_id"] == "vllm-abc123" + assert isinstance(meta["register_ts_ms"], int) and meta["register_ts_ms"] > 0 + + +def test_addr_has_no_scheme(): + # init_brpc_channel prepends http://; the registered addr must be bare. + meta = build_instance_meta("127.0.0.1:18000", "x") + assert "://" not in meta["name"] diff --git a/vllm_sidecar/tests/test_sidecar.py b/vllm_sidecar/tests/test_sidecar.py new file mode 100644 index 0000000..0a8f506 --- /dev/null +++ b/vllm_sidecar/tests/test_sidecar.py @@ -0,0 +1,299 @@ +# 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. +# ============================================================================== +"""Runtime unit tests for the vLLM sidecar without live etcd/vLLM.""" + +import argparse +import base64 +import json + +import pytest +import requests + +from vllm_sidecar import sidecar as sidecar_mod +from vllm_sidecar.etcd_registry import EtcdError, EtcdGatewayClient +from vllm_sidecar.health import VllmHealthProbe +from vllm_sidecar.meta import InstanceType + + +class _Response: + def __init__(self, status_code=200, payload=None, text=""): + self.status_code = status_code + self._payload = payload if payload is not None else {} + self.text = text + + def json(self): + return self._payload + + +class _Session: + def __init__(self, responses): + self.responses = list(responses) + self.calls = [] + self.headers = {} + + def post(self, url, json, timeout): + self.calls.append((url, json, timeout)) + item = self.responses.pop(0) + if isinstance(item, Exception): + raise item + return item + + +class _Etcd: + def __init__(self, *_args, **_kwargs): + self.puts = [] + self.keepalives = [] + self.revoked = [] + self.keepalive_ttl = 5 + self.fail_grant = False + self.fail_keepalive = False + + def lease_grant(self, ttl_seconds): + if self.fail_grant: + raise EtcdError("grant failed") + return "lease-1" + + def put(self, key, value, lease_id): + self.puts.append((key, json.loads(value), lease_id)) + + def lease_keepalive(self, lease_id): + if self.fail_keepalive: + raise EtcdError("keepalive failed") + self.keepalives.append(lease_id) + return self.keepalive_ttl + + def lease_revoke(self, lease_id): + self.revoked.append(lease_id) + + +class _Health: + def is_healthy(self): + return True + + def served_model(self): + return "demo-model" + + +def _args(**kwargs): + values = dict( + etcd_endpoints="127.0.0.1:2379", + etcd_username="", + etcd_password="", + etcd_timeout=1.0, + vllm_url="http://127.0.0.1:18000", + register_addr="127.0.0.1:18000", + backend_type="vllm", + instance_type="DEFAULT", + instance_name="vllm", + etcd_namespace="", + lease_ttl=6, + keepalive_interval=2.0, + health_timeout=1.0, + health_fail_threshold=3, + xllm_service_url="http://127.0.0.1:9998", + internal_token="", + heartbeat_interval=3.0, + metrics_url="http://127.0.0.1:18000/metrics", + log_level="INFO", + ) + values.update(kwargs) + return argparse.Namespace(**values) + + +def _install_fakes(monkeypatch): + etcd = _Etcd() + monkeypatch.setattr(sidecar_mod, "EtcdGatewayClient", lambda *a, **k: etcd) + monkeypatch.setattr(sidecar_mod, "VllmHealthProbe", lambda *a, **k: _Health()) + return etcd + + +def test_health_probe_success_failure_and_model(monkeypatch): + responses = iter( + [ + _Response(204), + _Response(503), + requests.Timeout("timeout"), + _Response(200, {"data": [{"id": "model-a"}]}), + _Response(200, {"data": []}), + ] + ) + + def fake_get(*_args, **_kwargs): + item = next(responses) + if isinstance(item, Exception): + raise item + return item + + monkeypatch.setattr(requests, "get", fake_get) + probe = VllmHealthProbe("http://vllm/") + monkeypatch.setattr(probe._session, "get", fake_get) + assert probe.is_healthy() + assert not probe.is_healthy() + assert not probe.is_healthy() + assert probe.served_model() == "model-a" + assert probe.served_model() is None + + +def test_etcd_gateway_lease_kv_auth_and_errors(monkeypatch): + value = base64.b64encode(b"payload").decode("ascii") + session = _Session( + [ + _Response(payload={"token": "tok"}), + _Response(payload={"ID": "lease-1"}), + _Response(payload={"result": {"TTL": "4"}}), + _Response(), + _Response(), + _Response(payload={"kvs": [{"value": value}]}), + _Response(payload={}), + ] + ) + monkeypatch.setattr(requests, "Session", lambda: session) + client = EtcdGatewayClient("127.0.0.1:2379", username="u", password="p") + + assert client._session.headers["Authorization"] == "tok" + assert client.lease_grant(6) == "lease-1" + assert client.lease_keepalive("lease-1") == 4 + client.lease_revoke("lease-1") + client.put("key", "payload", "lease-1") + assert client.get("key") == "payload" + assert client.get("missing") is None + + monkeypatch.setattr( + requests, "Session", lambda: _Session([_Response(500, text="bad")]) + ) + with pytest.raises(EtcdError): + EtcdGatewayClient("127.0.0.1:2379").lease_grant(6) + with pytest.raises(ValueError): + EtcdGatewayClient("") + + +def test_etcd_gateway_endpoint_scheme_normalization(): + client = EtcdGatewayClient( + "127.0.0.1:2379, https://secure:2379, http://plain:2379/" + ) + assert client._bases == [ + "http://127.0.0.1:2379", + "https://secure:2379", + "http://plain:2379", + ] + + +def test_etcd_gateway_raises_on_non_json_200(monkeypatch): + class _BadJson: + status_code = 200 + text = "proxy error" + + def json(self): + raise ValueError("not json") + + monkeypatch.setattr(requests, "Session", lambda: _Session([_BadJson()])) + # A 200 with an undecodable body must surface as EtcdError, not a raw + # ValueError that would crash the sidecar. + with pytest.raises(EtcdError): + EtcdGatewayClient("127.0.0.1:2379").lease_grant(6) + + +def test_etcd_gateway_keepalive_handles_null_result(monkeypatch): + # etcd may serialize an expired lease as {"result": null}; that must read + # back as TTL 0, not raise AttributeError. + monkeypatch.setattr( + requests, "Session", lambda: _Session([_Response(payload={"result": None})]) + ) + assert EtcdGatewayClient("127.0.0.1:2379").lease_keepalive("lease-1") == 0 + + +def test_etcd_gateway_get_handles_empty_value(monkeypatch): + # proto3 JSON omits an empty "value" field; get() must return "" not crash. + monkeypatch.setattr( + requests, "Session", lambda: _Session([_Response(payload={"kvs": [{}]})]) + ) + assert EtcdGatewayClient("127.0.0.1:2379").get("key") == "" + + +def test_etcd_gateway_coerces_non_object_200_to_empty(monkeypatch): + # A 200 whose JSON body is not an object (e.g. a list/null) must not crash + # callers that rely on dict .get(); it surfaces as a normal EtcdError. + monkeypatch.setattr(requests, "Session", lambda: _Session([_Response(payload=[])])) + with pytest.raises(EtcdError): + EtcdGatewayClient("127.0.0.1:2379").lease_grant(6) + + +def test_etcd_gateway_keepalive_handles_null_ttl(monkeypatch): + monkeypatch.setattr( + requests, + "Session", + lambda: _Session([_Response(payload={"result": {"TTL": None}})]), + ) + assert EtcdGatewayClient("127.0.0.1:2379").lease_keepalive("lease-1") == 0 + + +def test_sidecar_registration_keepalive_and_deregister(monkeypatch): + etcd = _install_fakes(monkeypatch) + sc = sidecar_mod.Sidecar(_args()) + + assert sc._register() + assert sc._registered + key, meta, lease_id = etcd.puts[0] + assert key == "XLLM:DEFAULT:127.0.0.1:18000" + assert meta["backend_type"] == "vllm" + assert lease_id == "lease-1" + + sc._keepalive_or_reregister() + assert etcd.keepalives == ["lease-1"] + + sc._deregister() + assert not sc._registered + assert etcd.revoked == ["lease-1"] + + +def test_sidecar_reregisters_on_lost_or_failed_lease(monkeypatch): + etcd = _install_fakes(monkeypatch) + sc = sidecar_mod.Sidecar(_args()) + assert sc._register() + + etcd.keepalive_ttl = 0 + sc._keepalive_or_reregister() + assert len(etcd.puts) == 2 + + etcd.fail_keepalive = True + sc._keepalive_or_reregister() + assert len(etcd.puts) == 3 + + etcd.fail_grant = True + sc._lease_id = None + assert not sc._register() + assert not sc._registered + + +def test_sidecar_parser_and_main(monkeypatch): + ran = [] + + class FakeSidecar: + def __init__(self, args): + self.args = args + + def run(self): + ran.append((self.args.register_addr, self.args.log_level)) + + monkeypatch.setattr(sidecar_mod, "Sidecar", FakeSidecar) + assert sidecar_mod._derive_addr("http://host:18000/v1/models") == "host:18000" + assert ( + sidecar_mod.main(["--vllm-url", "http://host:18000", "--log-level", "DEBUG"]) + == 0 + ) + assert ran == [("host:18000", "DEBUG")] + + args = sidecar_mod.build_parser().parse_args(["--instance-type", "DEFAULT"]) + assert args.instance_type == InstanceType.DEFAULT.name diff --git a/xllm_service/examples/vllm_backend/README.md b/xllm_service/examples/vllm_backend/README.md index df4d543..6bc9913 100644 --- a/xllm_service/examples/vllm_backend/README.md +++ b/xllm_service/examples/vllm_backend/README.md @@ -13,11 +13,13 @@ OpenAI client xllm-service master │ HTTP forwarding for backend_type=vllm ▼ -vLLM backend (:18000) +vLLM backend (:18000) ◀── vLLM sidecar registers it in etcd ``` vLLM instances are discovered through etcd keys such as -`XLLM:DEFAULT:127.0.0.1:18000` with `backend_type="vllm"`. +`XLLM:DEFAULT:127.0.0.1:18000` with `backend_type="vllm"`. The demo uses +`vllm_sidecar` to create and keep this registration under an etcd +lease; if the sidecar exits, the lease expires and the instance is removed. ## Prerequisites @@ -76,6 +78,7 @@ Set `STOP_VLLM=0` to leave an externally managed vLLM process running. ## Scope This demo covers the minimal vLLM backend path: OpenAI-compatible request -forwarding, model listing, and streaming/non-streaming responses. Cache-aware +forwarding, model listing, streaming/non-streaming responses, and sidecar-based +etcd lease registration. Cache-aware routing, disaggregated PD, and mixed-backend clusters are out of scope for this demo. diff --git a/xllm_service/examples/vllm_backend/start_demo.sh b/xllm_service/examples/vllm_backend/start_demo.sh index a67a52c..b53b726 100755 --- a/xllm_service/examples/vllm_backend/start_demo.sh +++ b/xllm_service/examples/vllm_backend/start_demo.sh @@ -93,7 +93,6 @@ if [ -d "$ROOT/vllm_sidecar" ]; then --etcd-endpoints "$ETCD" \ --vllm-url "http://127.0.0.1:$VLLM_PORT" \ --register-addr "$INSTANCE_ADDR" \ - --xllm-service-url "http://127.0.0.1:$HTTP_PORT" \ > "$SIDECAR_LOG" 2>&1 & echo $! > "$SIDECAR_PID" ok "sidecar started (pid=$(cat "$SIDECAR_PID"))"