You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tooling.apache.org by gs...@apache.org on 2026/10/18 11:17:39 UTC
(tooling-llmao) branch main updated (2c0c252 -> 759a6e3)
This is an automated email from the ASF dual-hosted git repository.
gstein pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/tooling-llmao.git
from 2c0c252 models: drop the derived per-token pricing for self-hosted models
new 0135d40 capacity: hourly rollup of vLLM readings, kept in LiteLLM's memory store
new 759a6e3 sampler: one reader of the boxes' /metrics, feeding the hourly rollup
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
config.yaml.example | 20 +--
docs/capacity-design.md | 117 +++++++++++++++++
llmao/capacity.py | 335 +++++++++++++++++++++++++++++++++++++++++++++++
llmao/contention.py | 183 --------------------------
llmao/proxy/client.py | 3 +-
llmao/sampler.py | 195 +++++++++++++++++++++++++++
main.py | 34 ++---
tests/test_capacity.py | 174 ++++++++++++++++++++++++
tests/test_contention.py | 47 -------
tests/test_sampler.py | 136 +++++++++++++++++++
10 files changed, 983 insertions(+), 261 deletions(-)
create mode 100644 docs/capacity-design.md
create mode 100644 llmao/capacity.py
delete mode 100644 llmao/contention.py
create mode 100644 llmao/sampler.py
create mode 100644 tests/test_capacity.py
delete mode 100644 tests/test_contention.py
create mode 100644 tests/test_sampler.py
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org
(tooling-llmao) 02/02: sampler: one reader of the boxes' /metrics, feeding the hourly rollup
Posted by gs...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gstein pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/tooling-llmao.git
commit 759a6e33e3cb377f972c5221b3740c0f23161fe0
Author: Greg Stein <gs...@gmail.com>
AuthorDate: Sun Oct 18 05:26:28 2026 -0500
sampler: one reader of the boxes' /metrics, feeding the hourly rollup
The contention sampler becomes llmao/sampler.py, the one reader of each
self-hosted box's vLLM /metrics, and feeds the capacity rollup instead of
writing a JSON log line per box per interval. The rollup's waiting
fraction answers what those lines were for: whether requests queue.
- Servers are named host:listen port, as everywhere else. They were
named by model, so every box serving one model logged as the same box.
- KV cache use is read from kv_cache_usage_perc, vLLM's current name;
the old name read nothing, so every logged KV figure was 0.
- It also reads the token, request, preemption and queue-time counters
the rollup needs, and reads all boxes at once rather than in turn.
main.py runs three runners: metrics-sampler, capacity-writer and
capacity-prune. The capacity store is built there over the backend's
memory store, which the backend now exposes; the proxy layer knows
nothing of capacity.
sample_interval_s and capacity_retention_days are required, and a
missing or non-positive value stops startup: sampling is never off.
contention_interval_s, the old name, is still read with a warning,
since p6 writes config.yaml on its own schedule. config.yaml.example
carries the suggested values, 15 and 90.
Co-Authored-By: Claude Opus 5.5 <no...@anthropic.com>
---
config.yaml.example | 20 ++---
docs/capacity-design.md | 7 +-
llmao/capacity.py | 18 +++++
llmao/contention.py | 183 --------------------------------------------
llmao/proxy/client.py | 3 +-
llmao/sampler.py | 195 +++++++++++++++++++++++++++++++++++++++++++++++
main.py | 34 ++++-----
tests/test_capacity.py | 13 ++++
tests/test_contention.py | 47 ------------
tests/test_sampler.py | 136 +++++++++++++++++++++++++++++++++
10 files changed, 392 insertions(+), 264 deletions(-)
diff --git a/config.yaml.example b/config.yaml.example
index 1aeedc0..bab0369 100644
--- a/config.yaml.example
+++ b/config.yaml.example
@@ -104,17 +104,17 @@ entitlements:
tpm_limit: 300000
max_parallel_requests: 8
-# How often to sample each box's queue depth, in seconds. 0 turns it off.
+# Capacity planning (docs/capacity-design.md). Both required.
#
-# The spend log records what RAN and cannot record what WAITED -- queue depth
-# only exists while it is happening. Without this, the question of whether
-# free-tier and project work ever actually contend for a GPU is unanswerable
-# after the fact.
-#
-# One HTTP GET per self-hosted box per interval, one JSON line per box on the
-# llmao.contention logger. At three boxes and fifteen seconds that is about
-# 17,000 lines a day; jq answers every question we currently have of it.
-contention_interval_s: 15
+# Seconds between readings of every self-hosted box's /metrics. Never off:
+# queue depth exists only while it is happening -- the spend log records
+# what RAN, not what WAITED -- so an unsampled stretch is lost for good.
+# One HTTP GET per box per interval. (Formerly contention_interval_s, still
+# read with a warning.)
+sample_interval_s: 15
+# Days of hourly rows kept in LiteLLM's memory store. 90 days at twenty
+# boxes is about 43,000 rows.
+capacity_retention_days: 90
# uids that may view any project's activity, and reach /admin and the
# Deployments on /models (plus asfquart isRoot).
diff --git a/docs/capacity-design.md b/docs/capacity-design.md
index 000e587..623e474 100644
--- a/docs/capacity-design.md
+++ b/docs/capacity-design.md
@@ -22,7 +22,7 @@ hand, and this page is not it.
| Step | What | State |
|---|---|---|
| 1 | Vast and RunPod keys under `providers:` | done |
-| 2 | Hourly rollups, stored | in progress |
+| 2 | Hourly rollups, stored | done |
| 3 | Rental prices from the Vast and RunPod APIs | planned |
| 4 | Provider spend from LiteLLM's daily activity | planned |
| 5 | The page | planned |
@@ -40,7 +40,7 @@ Four parts, each knowing one thing:
| Part | Knows | Runs |
|---|---|---|
-| **Sampler** (`llmao/sampler.py`) | which boxes are serving; fetching and parsing vLLM's `/metrics` | every `sample_interval_s` (15s) |
+| **Sampler** (`llmao/sampler.py`) | which boxes are serving; fetching and parsing vLLM's `/metrics` | every `sample_interval_s` |
| **Rollup** (`HourlyRollup`) | turning readings into one row per server per clock hour | no; fed by the sampler |
| **Writer** (`CapacityWriter`) | when an hour is finished; handing it to the store | about every minute |
| **Store** (`CapacityStore`) | the `/v1/memory` key layout; write, read, prune | no; used by the writer, pruner and page |
@@ -91,7 +91,8 @@ a Stopwatch. If the 90-day view proves too slow even for two admins, rows
older than a week can be merged into one per server per day, about 5,000
rows for 90 days. Not built until the timings say so.
-Rows older than `capacity_retention_days` (90) are deleted daily. Each day
+Rows older than `capacity_retention_days` (required; config.yaml.example
+suggests 90) are deleted daily. Each day
past retention is swept seven times, so a few days with llmao down still
get cleared.
diff --git a/llmao/capacity.py b/llmao/capacity.py
index 4e02924..386a069 100644
--- a/llmao/capacity.py
+++ b/llmao/capacity.py
@@ -82,6 +82,24 @@ class Sample:
buckets: dict[float, float] = field(default_factory=dict)
+def retention_days(cfg: Any) -> int:
+ """config.yaml capacity_retention_days: how long hourly rows are kept. Call at startup.
+
+ Required, like every other setting with an operational effect: a
+ missing value is a startup error, not a quiet default.
+ """
+ if "capacity_retention_days" not in cfg:
+ raise ValueError("config.yaml: capacity_retention_days is required")
+ raw = cfg.capacity_retention_days
+ try:
+ days = int(raw)
+ except (TypeError, ValueError) as e:
+ raise ValueError("config.yaml: capacity_retention_days must be a whole number of days") from e
+ if days <= 0:
+ raise ValueError("config.yaml: capacity_retention_days must be > 0")
+ return days
+
+
def _growth(now: float, before: float) -> float:
"""How far a counter moved between two readings. A drop is a restart, counted from zero."""
return now - before if now >= before else now
diff --git a/llmao/contention.py b/llmao/contention.py
deleted file mode 100644
index 93e7645..0000000
--- a/llmao/contention.py
+++ /dev/null
@@ -1,183 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you 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
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# 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.
-
-"""Sample fleet contention, so the pilot answers a question.
-
-WHY THIS EXISTS
-
-The weekly token allowance is a proxy. What we actually want to know is
-whether free-tier work and project work ever contend for a GPU -- because if
-they rarely do, the allowance is solving a problem that is not there, and the
-replacement design (defer a free-tier call only when projects need the box)
-needs no allowance at all.
-
-Nobody can answer that from the spend log: it records what ran, not what
-waited. Queue depth is only visible while it is happening, so it has to be
-sampled.
-
-WHAT IT RECORDS
-
-One line per box per interval, as structured JSON on the logger. Not a
-database: a pilot is weeks, the volume is a few thousand lines, and grep and
-jq answer every question we currently have. A table would be a schema to
-design before knowing what the questions are.
-
- {"ts": ..., "box": "gemma4-26b", "running": 4, "waiting": 2,
- "kv_pct": 31.0, "preempted": 0}
-
-The three that matter:
-
- waiting > 0 while running > 0 contention is real and measurable
- waiting == 0 throughout the allowance is unnecessary; capacity is
- not the scarce thing
- preempted climbing vLLM is evicting under memory pressure,
- which is a different problem from queueing
- and wants a different fix
-"""
-
-from __future__ import annotations
-
-import asyncio
-import json
-import logging
-import time
-from dataclasses import dataclass
-from typing import Any
-
-import httpx
-
-_LOGGER = logging.getLogger("llmao.contention")
-
-# Metric names, with the vllm: prefix stripped at parse time.
-_WANTED = {
- "num_requests_running": "running",
- "num_requests_waiting": "waiting",
- "gpu_cache_usage_perc": "kv_pct",
- "num_preemptions": "preempted",
- "num_preemptions_total": "preempted",
-}
-
-
-@dataclass
-class Sample:
- ts: float
- box: str
- running: float = 0.0
- waiting: float = 0.0
- kv_pct: float = 0.0
- preempted: float = 0.0
-
- def as_dict(self) -> dict[str, Any]:
- return {
- "ts": round(self.ts, 1),
- "box": self.box,
- "running": self.running,
- "waiting": self.waiting,
- "kv_pct": round(self.kv_pct * 100, 1) if self.kv_pct <= 1 else round(self.kv_pct, 1),
- "preempted": self.preempted,
- }
-
-
-def parse_metrics(box: str, text: str, *, now: float | None = None) -> Sample:
- """Pull the four figures out of a Prometheus exposition response.
-
- Hand-parsed rather than via a client library: four metrics, one format,
- and a dependency for this would be heavier than the code.
- """
- out = Sample(ts=now if now is not None else time.time(), box=box)
- for line in text.splitlines():
- if not line or line.startswith("#"):
- continue
- name, _, rest = line.partition("{")
- if not rest:
- name, _, value = line.partition(" ")
- else:
- _, _, value = rest.partition("} ")
- field = _WANTED.get(name.removeprefix("vllm:"))
- if not field:
- continue
- try:
- setattr(out, field, float(value.strip()))
- except (TypeError, ValueError):
- continue
- return out
-
-
-class ContentionSampler:
- """Polls each box's /metrics and logs a line per sample.
-
- Failures are logged once per box per outage rather than every interval:
- a box being down is worth knowing and is not worth a line every fifteen
- seconds for an hour.
- """
-
- def __init__(self, fleet: Any, *, interval_s: int = 15, timeout_s: float = 3.0):
- self._fleet = fleet
- self._interval = interval_s
- self._timeout = timeout_s
- self._down: set[str] = set()
- self._task: asyncio.Task | None = None
-
- def _targets(self) -> list[tuple[str, str]]:
- """(box name, metrics url) for every self-hosted deployment serving."""
- out = []
- for dep in getattr(self._fleet, "deployments", None) or []:
- if not getattr(dep, "self_hosted", False):
- continue
- base = getattr(dep, "api_base", None)
- if not base:
- continue
- root = base.rstrip("/")
- if root.endswith("/v1"):
- root = root[: -len("/v1")]
- out.append((getattr(dep, "model_name", getattr(dep, "name", "?")), f"{root}/metrics"))
- return out
-
- async def sample_once(self, client: httpx.AsyncClient) -> list[Sample]:
- samples = []
- for box, url in self._targets():
- try:
- resp = await client.get(url, timeout=self._timeout)
- resp.raise_for_status()
- except Exception as e:
- if box not in self._down:
- self._down.add(box)
- _LOGGER.warning("contention: %s unreachable (%s)", box, type(e).__name__)
- continue
- self._down.discard(box)
- samples.append(parse_metrics(box, resp.text))
- return samples
-
- async def run(self) -> None:
- async with httpx.AsyncClient() as client:
- while True:
- try:
- for s in await self.sample_once(client):
- # One JSON object per line: greppable, jq-able, and
- # loadable into pandas without a parser.
- _LOGGER.info("contention %s", json.dumps(s.as_dict()))
- except asyncio.CancelledError:
- raise
- except Exception:
- _LOGGER.exception("contention: sample failed")
- await asyncio.sleep(self._interval)
-
- def start(self) -> asyncio.Task:
- """Standalone use. Under the app, pass `run` to add_runner instead --
- it owns the task and cancels it on shutdown, which this does not."""
- self._task = asyncio.create_task(self.run(), name="contention-sampler")
- return self._task
diff --git a/llmao/proxy/client.py b/llmao/proxy/client.py
index 60d8b89..9cb969b 100644
--- a/llmao/proxy/client.py
+++ b/llmao/proxy/client.py
@@ -339,7 +339,8 @@ class LiteLLMBackend:
self._cfg = cfg
self._transport = transport or LiteLLMTransport(cfg)
self.deployments = DeploymentManager(cfg, fleet, self._transport)
- memory = MemoryStore(self._transport)
+ # Public: other parts of llmao keep their own rows there too.
+ self.memory = memory = MemoryStore(self._transport)
self.requests = AccessRequestManager(memory)
self.skew = SkewManager(memory)
self.log = LogManager(memory)
diff --git a/llmao/sampler.py b/llmao/sampler.py
new file mode 100644
index 0000000..51490f7
--- /dev/null
+++ b/llmao/sampler.py
@@ -0,0 +1,195 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# 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.
+
+"""Read every serving box's vLLM /metrics, and hand each reading on.
+
+The one reader of /metrics on our boxes. Every sample_interval_s it fetches
+all of them at once, parses what capacity planning needs, and gives each
+reading to the hourly rollup (llmao/capacity.py), which knows nothing of
+HTTP. Nothing here is stored or logged per reading.
+
+Queue depth is only visible while it is happening -- the spend log records
+what ran, not what waited -- so it has to be sampled or it is gone. The
+rollup's waiting fraction is what shows whether requests actually queue,
+and so whether a model needs another box.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from typing import Any
+
+import httpx
+
+from llmao.capacity import HourlyRollup, Sample
+
+_LOGGER = logging.getLogger(__name__)
+
+# Instant values. Metric names with the vllm: prefix stripped.
+_GAUGES = {
+ "num_requests_running": "running",
+ "num_requests_waiting": "waiting",
+ "kv_cache_usage_perc": "kv",
+}
+# vLLM's totals since it started. request_success_total has a line per
+# finish reason, so lines of one name are summed.
+_COUNTERS = {
+ "prompt_tokens_total": "prompt_tokens",
+ "generation_tokens_total": "generation_tokens",
+ "request_success_total": "requests",
+ "num_preemptions_total": "preemptions",
+ "request_queue_time_seconds_sum": "queue_s",
+ "request_queue_time_seconds_count": "queued",
+}
+_QUEUE_BUCKET = "request_queue_time_seconds_bucket"
+
+
+def sample_interval(cfg: Any) -> int:
+ """Seconds between readings: config.yaml sample_interval_s. Call at startup.
+
+ Sampling is never off, so a missing or non-positive value is a startup
+ error. contention_interval_s, the old name, is still read, with a
+ warning: p6 writes config.yaml and moves on its own schedule.
+ """
+ if "sample_interval_s" in cfg:
+ raw = cfg.sample_interval_s
+ elif "contention_interval_s" in cfg:
+ _LOGGER.warning("config.yaml: contention_interval_s is deprecated; rename it to sample_interval_s")
+ raw = cfg.contention_interval_s
+ else:
+ raise ValueError("config.yaml: sample_interval_s is required")
+ try:
+ interval = int(raw)
+ except (TypeError, ValueError) as e:
+ raise ValueError("config.yaml: sample_interval_s must be a whole number of seconds") from e
+ if interval <= 0:
+ raise ValueError("config.yaml: sample_interval_s must be > 0; sampling is never off")
+ return interval
+
+
+def _labels(inner: str) -> dict[str, str]:
+ labels = {}
+ for part in inner.split(","):
+ key, _, value = part.partition("=")
+ if key:
+ labels[key.strip()] = value.strip().strip('"')
+ return labels
+
+
+def parse_metrics(text: str, *, now: float | None = None) -> Sample:
+ """What the rollup needs, from vLLM's Prometheus exposition text.
+
+ Hand-parsed rather than with a client library: a dozen metrics in one
+ format. Anything unreadable is skipped, so a box answering with HTML or
+ nothing gives an empty reading rather than an error.
+ """
+ out = Sample(ts=now if now is not None else time.time())
+ for line in text.splitlines():
+ if not line or line.startswith("#"):
+ continue
+ name, brace, rest = line.partition("{")
+ if brace:
+ inner, _, value = rest.partition("} ")
+ else:
+ inner = ""
+ name, _, value = line.partition(" ")
+ name = name.strip().removeprefix("vllm:")
+ try:
+ number = float(value.strip())
+ except (TypeError, ValueError):
+ continue
+ if name in _GAUGES:
+ setattr(out, _GAUGES[name], getattr(out, _GAUGES[name]) + number)
+ elif name in _COUNTERS:
+ key = _COUNTERS[name]
+ out.counters[key] = out.counters.get(key, 0.0) + number
+ elif name == _QUEUE_BUCKET:
+ try:
+ bound = float(_labels(inner).get("le", ""))
+ except ValueError:
+ continue
+ out.buckets[bound] = out.buckets.get(bound, 0.0) + number
+ return out
+
+
+class MetricsSampler:
+ """Runner: reads every serving box on an interval and feeds the rollup.
+
+ A box that does not answer is logged once per outage, not every
+ interval: worth knowing, and not worth a line every fifteen seconds for
+ an hour.
+ """
+
+ def __init__(self, fleet: Any, rollup: HourlyRollup, *, interval_s: int, timeout_s: float = 3.0):
+ self._fleet = fleet
+ self._rollup = rollup
+ self._interval = interval_s
+ self._timeout = timeout_s
+ self._down: set[str] = set()
+
+ def targets(self) -> list[tuple[str, str, str]]:
+ """(server, model_name, metrics URL) for each self-hosted server with an address.
+
+ A server is host:listen port, as everywhere else in llmao. The URL
+ is its public address, the one LiteLLM uses.
+ """
+ out = []
+ for dep in self._fleet.deployments:
+ srv = dep.vllm
+ if not dep.self_hosted or srv is None or not dep.api_base:
+ continue
+ root = dep.api_base.rstrip("/").removesuffix("/v1")
+ out.append((f"{srv.host}:{srv.listen_port}", dep.model_name, f"{root}/metrics"))
+ return out
+
+ async def _read(self, client: httpx.AsyncClient, server: str, url: str) -> Sample | None:
+ try:
+ resp = await client.get(url, timeout=self._timeout)
+ resp.raise_for_status()
+ except httpx.HTTPError as e:
+ if server not in self._down:
+ self._down.add(server)
+ _LOGGER.warning("sampler: %s unreachable (%s)", server, type(e).__name__)
+ return None
+ if server in self._down:
+ self._down.discard(server)
+ _LOGGER.info("sampler: %s answering again", server)
+ return parse_metrics(resp.text)
+
+ async def sample_once(self, client: httpx.AsyncClient) -> None:
+ """One reading of every box, all at once: twenty boxes in turn could
+ outlast the interval."""
+ targets = self.targets()
+ readings = await asyncio.gather(*(self._read(client, server, url) for server, _model, url in targets))
+ for (server, model_name, _url), sample in zip(targets, readings, strict=True):
+ if sample is None:
+ self._rollup.missed(server)
+ else:
+ self._rollup.record(server, model_name, sample)
+
+ async def run(self) -> None:
+ async with httpx.AsyncClient() as client:
+ while True:
+ try:
+ await self.sample_once(client)
+ except asyncio.CancelledError:
+ raise
+ except Exception:
+ _LOGGER.exception("sampler: reading the boxes failed")
+ await asyncio.sleep(self._interval)
diff --git a/main.py b/main.py
index 430cfc5..bfa11ba 100644
--- a/main.py
+++ b/main.py
@@ -68,16 +68,20 @@ def create_app():
from llmao.auth import make_token_handler
# Fail-fast: models.yaml is required (same presumption as config.yaml).
- from llmao.contention import ContentionSampler
+ from llmao.capacity import CapacityPruner, CapacityStore, CapacityWriter, HourlyRollup, retention_days
from llmao.fleet import Fleet, validate_fleet
# Config is app.cfg (EasyDict from config.yaml) — dotted access throughout.
from llmao.models import load_models
from llmao.proxy.client import LiteLLMBackend
+ from llmao.sampler import MetricsSampler, sample_interval
from llmao.seam import Seam
definitions = load_models(cfg=app.cfg)
validate_fleet(app.cfg, models=definitions)
+ # Required, and checked before anything starts.
+ interval = sample_interval(app.cfg)
+ keep_days = retention_days(app.cfg)
fleet = Fleet.from_cfg(app.cfg, models=definitions)
backend = LiteLLMBackend(app.cfg, fleet)
@@ -87,25 +91,15 @@ def create_app():
app.add_runner(backend.run_skew, name="litellm-skew")
app.add_runner(backend.watch_outside_edits, name="litellm-outside-edits")
- # Sample queue depth on each box.
- #
- # The spend log records what RAN; it cannot record what waited. Queue
- # depth is only visible while it is happening, so it has to be sampled or
- # it is gone.
- #
- # That matters because the question the pilot exists to answer is whether
- # free-tier and project work ever actually contend for a GPU. If nothing
- # ever waits, the allowances are rationing a resource that is not scarce,
- # and the honest conclusion is to stop rationing rather than to tune the
- # numbers.
- #
- # Cheap: one HTTP GET per box per interval, one log line each.
- interval = int(getattr(app.cfg, "contention_interval_s", 0) or 15)
- if interval > 0:
- app.add_runner(
- ContentionSampler(fleet, interval_s=interval).run,
- name="contention-sampler",
- )
+ # Capacity: what each box did, hour by hour (docs/capacity-design.md).
+ # The sampler reads every box's /metrics and feeds the rollup; the writer
+ # stores finished hours in LiteLLM, apart from the sampler so a slow
+ # LiteLLM cannot stall the readings; the pruner drops old days.
+ rollup = HourlyRollup(interval_s=interval)
+ app.capacity = CapacityStore(backend.memory)
+ app.add_runner(MetricsSampler(fleet, rollup, interval_s=interval).run, name="metrics-sampler")
+ app.add_runner(CapacityWriter(rollup, app.capacity).run, name="capacity-writer")
+ app.add_runner(CapacityPruner(app.capacity, keep_days).run, name="capacity-prune")
seam = Seam(app.cfg, backend)
app.token_handler = make_token_handler(app.cfg)
app.seam = seam
diff --git a/tests/test_capacity.py b/tests/test_capacity.py
index 82c3de0..c3d8ba1 100644
--- a/tests/test_capacity.py
+++ b/tests/test_capacity.py
@@ -22,6 +22,9 @@ import asyncio
import datetime as dt
import json
+import pytest
+from easydict import EasyDict as edict # noqa: N813
+
from llmao.capacity import (
HOUR_S,
CapacityStore,
@@ -29,6 +32,7 @@ from llmao.capacity import (
HourlyRollup,
Sample,
hour_label,
+ retention_days,
)
from llmao.proxy.client import MemoryStore
from tests.fake_transport import fake_memory
@@ -159,3 +163,12 @@ def test_an_hour_that_cannot_be_stored_waits_for_the_next_pass():
transport.down = False
asyncio.run(writer.write_finished(T0 + HOUR_S + 60))
assert list(rows) == [f"llmao:capacity:2026-10-18T09:{SERVER}"]
+
+
+def test_retention_is_required():
+ assert retention_days(edict(capacity_retention_days=90)) == 90
+ with pytest.raises(ValueError, match="capacity_retention_days is required"):
+ retention_days(edict())
+ for bad in (0, -1, "ninety"):
+ with pytest.raises(ValueError, match="capacity_retention_days must be"):
+ retention_days(edict(capacity_retention_days=bad))
diff --git a/tests/test_contention.py b/tests/test_contention.py
deleted file mode 100644
index 1764edd..0000000
--- a/tests/test_contention.py
+++ /dev/null
@@ -1,47 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you 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
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# 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.
-"""Contention sampling (llmao/contention.py).
-
-The pilot's one unanswerable-after-the-fact question: did anyone wait?
-"""
-
-from llmao.contention import parse_metrics
-
-
-def test_parse_metrics_reads_the_four_figures():
- text = "\n".join(
- [
- "# HELP vllm:num_requests_running whatever",
- 'vllm:num_requests_running{engine="0",model_name="x"} 4.0',
- 'vllm:num_requests_waiting{engine="0",model_name="x"} 2.0',
- 'vllm:gpu_cache_usage_perc{engine="0",model_name="x"} 0.31',
- 'vllm:num_preemptions_total{engine="0",model_name="x"} 3.0',
- ]
- )
- got = parse_metrics("gemma4-26b", text, now=1.0).as_dict()
- assert got["running"] == 4.0
- assert got["waiting"] == 2.0
- assert got["preempted"] == 3.0
- # exposed as a fraction, reported as a percentage
- assert got["kv_pct"] == 31.0
-
-
-def test_parse_metrics_survives_an_unexpected_body():
- """A box returning HTML, or nothing, must not stop the sampler."""
- for body in ("", "<html>503</html>", "vllm:num_requests_running{} notanumber"):
- got = parse_metrics("x", body, now=1.0)
- assert got.running == 0.0
diff --git a/tests/test_sampler.py b/tests/test_sampler.py
new file mode 100644
index 0000000..bb5fcf2
--- /dev/null
+++ b/tests/test_sampler.py
@@ -0,0 +1,136 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# 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.
+"""Reading the boxes' /metrics (llmao/sampler.py)."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+
+import httpx
+import pytest
+from easydict import EasyDict as edict # noqa: N813
+
+from llmao.capacity import HOUR_S, HourlyRollup
+from llmao.fleet import Fleet, FleetDeployment, VllmServer
+from llmao.sampler import MetricsSampler, parse_metrics, sample_interval
+
+# As vLLM 0.27 writes it, trimmed.
+METRICS = "\n".join(
+ [
+ "# HELP vllm:num_requests_running Number of requests in model execution batches.",
+ 'vllm:num_requests_running{engine="0",model_name="qwen3-8b"} 2.0',
+ 'vllm:num_requests_waiting{engine="0",model_name="qwen3-8b"} 1.0',
+ 'vllm:kv_cache_usage_perc{engine="0",model_name="qwen3-8b"} 0.25',
+ 'vllm:num_preemptions_total{engine="0",model_name="qwen3-8b"} 3.0',
+ 'vllm:prompt_tokens_total{engine="0",model_name="qwen3-8b"} 212.0',
+ 'vllm:generation_tokens_total{engine="0",model_name="qwen3-8b"} 131.0',
+ 'vllm:request_success_total{engine="0",finished_reason="stop",model_name="qwen3-8b"} 7.0',
+ 'vllm:request_success_total{engine="0",finished_reason="length",model_name="qwen3-8b"} 4.0',
+ 'vllm:request_queue_time_seconds_bucket{engine="0",le="0.3",model_name="qwen3-8b"} 10.0',
+ 'vllm:request_queue_time_seconds_bucket{engine="0",le="+Inf",model_name="qwen3-8b"} 11.0',
+ 'vllm:request_queue_time_seconds_sum{engine="0",model_name="qwen3-8b"} 0.9',
+ 'vllm:request_queue_time_seconds_count{engine="0",model_name="qwen3-8b"} 11.0',
+ ]
+)
+
+
+def test_parse_metrics_reads_what_the_rollup_needs():
+ got = parse_metrics(METRICS, now=1.0)
+ assert (got.running, got.waiting, got.kv) == (2.0, 1.0, 0.25)
+ assert got.counters == {
+ "preemptions": 3.0,
+ "prompt_tokens": 212.0,
+ "generation_tokens": 131.0,
+ # one line per finish reason, summed
+ "requests": 11.0,
+ "queue_s": 0.9,
+ "queued": 11.0,
+ }
+ assert got.buckets == {0.3: 10.0, float("inf"): 11.0}
+
+
+def test_parse_metrics_survives_an_unexpected_body():
+ """A box answering with HTML, or nothing, must not stop the sampler."""
+ for body in ("", "<html>503</html>", "vllm:num_requests_running{} notanumber"):
+ got = parse_metrics(body, now=1.0)
+ assert got.running == 0.0
+ assert got.counters == {}
+
+
+def test_the_interval_is_required_and_never_off():
+ assert sample_interval(edict(sample_interval_s=15)) == 15
+ with pytest.raises(ValueError, match="sample_interval_s is required"):
+ sample_interval(edict())
+ for bad in (0, -5):
+ with pytest.raises(ValueError, match="never off"):
+ sample_interval(edict(sample_interval_s=bad))
+ with pytest.raises(ValueError, match="whole number"):
+ sample_interval(edict(sample_interval_s="soon"))
+
+
+def test_the_old_name_still_works_with_a_warning(caplog):
+ with caplog.at_level(logging.WARNING, logger="llmao.sampler"):
+ assert sample_interval(edict(contention_interval_s=30)) == 30
+ assert "contention_interval_s is deprecated; rename it to sample_interval_s" in caplog.text
+ # The new name wins, silently.
+ caplog.clear()
+ assert sample_interval(edict(sample_interval_s=15, contention_interval_s=30)) == 15
+ assert caplog.text == ""
+
+
+def _server(host, *, public_port=8001):
+ return VllmServer(
+ model_name="qwen3-8b",
+ name="qwen3-8b",
+ host=host,
+ listen_port=8001,
+ hf_model="Qwen/Qwen3-8B",
+ api_key="sk-x",
+ args=[],
+ public_port=public_port,
+ )
+
+
+def test_each_server_is_named_by_host_and_listen_port():
+ fleet = Fleet(None, [_server("203.0.113.10", public_port=22018), _server("198.51.100.8", public_port=None)])
+ fleet.deployments.append(FleetDeployment(model_name="grok-4", name="grok-4", self_hosted=False))
+ sampler = MetricsSampler(fleet, HourlyRollup(15), interval_s=15)
+ # Read at the public port; named by the listen port. No address, no reading.
+ assert sampler.targets() == [("203.0.113.10:8001", "qwen3-8b", "http://203.0.113.10:22018/metrics")]
+
+
+def test_a_pass_feeds_the_rollup_and_notes_a_box_that_did_not_answer():
+ fleet = Fleet(None, [_server("203.0.113.10"), _server("198.51.100.8")])
+ rollup = HourlyRollup(15)
+ sampler = MetricsSampler(fleet, rollup, interval_s=15)
+
+ def answer(request):
+ if request.url.host == "198.51.100.8":
+ raise httpx.ConnectError("refused")
+ return httpx.Response(200, text=METRICS)
+
+ async def one_pass():
+ async with httpx.AsyncClient(transport=httpx.MockTransport(answer)) as client:
+ await sampler.sample_once(client)
+
+ asyncio.run(one_pass())
+ # An hour on, the reading's hour is finished.
+ (hour,) = rollup.take_finished(time.time() + HOUR_S)
+ assert hour.server == "203.0.113.10:8001"
+ assert hour.readings == 1
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org
(tooling-llmao) 01/02: capacity: hourly rollup of vLLM readings, kept in LiteLLM's memory store
Posted by gs...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gstein pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/tooling-llmao.git
commit 0135d40a449fad97207a45c85cacd95769f637f4
Author: Greg Stein <gs...@gmail.com>
AuthorDate: Sun Oct 18 04:52:38 2026 -0500
capacity: hourly rollup of vLLM readings, kept in LiteLLM's memory store
The first piece of the capacity report: what each self-hosted server
did, hour by hour, so admins can see over weeks whether to add a box or
return one. Not wired in yet; the sampler feeds it in the next commit.
- HourlyRollup turns readings into one row per server per UTC clock
hour: busy and waiting fractions, peak waiting, KV cache average and
peak, and the growth of vLLM's token, request, preemption and
queue-time counters. A counter that drops is a restarted server,
counted from zero; a missed reading breaks the chain rather than
crediting the gap to one hour. No I/O, so the arithmetic is tested on
its own.
- CapacityStore is the only part that knows the /v1/memory keys
(llmao:capacity:<hour>:<host:port>). It reads one day per request,
all at once, under a Stopwatch, and prunes the days just past
retention.
- CapacityWriter and CapacityPruner are the runners: store finished
hours about once a minute, keeping any LiteLLM refuses for the next
pass, and prune once a day.
docs/capacity-design.md sets out the report, the parts, the row, its
size, and the accepted gaps: an llmao restart loses the hour in
progress, which the report, read for trends over weeks, can afford.
Co-Authored-By: Claude Opus 5.5 <no...@anthropic.com>
---
docs/capacity-design.md | 116 ++++++++++++++++++
llmao/capacity.py | 317 ++++++++++++++++++++++++++++++++++++++++++++++++
tests/test_capacity.py | 161 ++++++++++++++++++++++++
3 files changed, 594 insertions(+)
diff --git a/docs/capacity-design.md b/docs/capacity-design.md
new file mode 100644
index 0000000..000e587
--- /dev/null
+++ b/docs/capacity-design.md
@@ -0,0 +1,116 @@
+# Capacity and spend report: design
+
+A page for llmao's site admins (the people who see Deployments on `/models`),
+at `/admin/capacity`, linked from `/admin`. It answers, per model:
+
+1. **Is there enough capacity?** How often requests queue, how full the KV
+ cache gets at peak, and whether vLLM is preempting.
+2. **Is there too much?** How busy each box is, and tokens served against
+ what the box can do.
+3. **What does it cost?** Rental dollars per day, and the real cost per
+ million tokens: rental divided by tokens served.
+4. **Which way is demand going?** Tokens per day, and the busiest hours.
+
+And, for the model providers we pay (xAI, OpenRouter), what we spend: per
+day, month to date, a projection to month end, and which projects spend it.
+
+Some figures may later go into a Board report; that report is curated by
+hand, and this page is not it.
+
+## Status
+
+| Step | What | State |
+|---|---|---|
+| 1 | Vast and RunPod keys under `providers:` | done |
+| 2 | Hourly rollups, stored | in progress |
+| 3 | Rental prices from the Vast and RunPod APIs | planned |
+| 4 | Provider spend from LiteLLM's daily activity | planned |
+| 5 | The page | planned |
+| 6 | "Add a box / remove a box" suggestions | after a few weeks of data |
+
+**Future, on demand only:** a CSV download of the report. Not built until
+someone asks for it.
+
+## Measuring the boxes
+
+llmao has no database (see ARCHITECTURE.md), so the measurements live in
+LiteLLM's `/v1/memory`, like access requests and tier changes.
+
+Four parts, each knowing one thing:
+
+| Part | Knows | Runs |
+|---|---|---|
+| **Sampler** (`llmao/sampler.py`) | which boxes are serving; fetching and parsing vLLM's `/metrics` | every `sample_interval_s` (15s) |
+| **Rollup** (`HourlyRollup`) | turning readings into one row per server per clock hour | no; fed by the sampler |
+| **Writer** (`CapacityWriter`) | when an hour is finished; handing it to the store | about every minute |
+| **Store** (`CapacityStore`) | the `/v1/memory` key layout; write, read, prune | no; used by the writer, pruner and page |
+
+Plus the **pruner** (`CapacityPruner`), once a day.
+
+The sampler is the only reader of each box's `/metrics`. The writer is
+separate from the sampler so that a slow or absent LiteLLM delays storing
+rows without stalling the readings; finished hours wait in memory until it
+answers.
+
+### What a row holds
+
+One row per server per UTC clock hour, keyed by hour then server:
+
+ llmao:capacity:2026-10-18T09:203.0.113.10:8001
+
+| Field | Meaning |
+|---|---|
+| `hour`, `server`, `model_name` | which hour, which server (host:listen port), which model |
+| `covered_s` | how much of the hour was sampled |
+| `busy_frac` | share of readings with any request running |
+| `waiting_frac`, `waiting_peak` | share of readings with any request waiting; the most waiting at once |
+| `kv_avg`, `kv_peak` | KV cache use, as a fraction |
+| `prompt_tokens`, `generation_tokens`, `requests`, `preemptions` | growth of vLLM's counters over the hour |
+| `queue_avg_s`, `queue_p95_s` | time requests spent queued, from vLLM's histogram |
+| `usd_per_hour` | the box's rental price (step 3; until then empty) |
+
+Counters are vLLM's totals since it started. A total that goes down means
+the server restarted, and the new total is counted from zero. A reading the
+box did not answer breaks the chain: the next reading starts afresh rather
+than crediting the gap to one hour.
+
+`waiting_frac` is the evidence for "add a box": it shows whether requests
+actually queue, which the spend log cannot (it records what ran, not what
+waited). It replaces the contention log lines the sampler used to write.
+
+The price is on each row because a box's rental price is fixed for as long
+as we hold it, and a returned box vanishes from its provider's API. Kept on
+the row, a returned box keeps its price in history.
+
+### Size
+
+20 boxes x 24 hours x 90 days is about 43,000 rows, roughly 17 MB to read
+for a 90-day view. Reads go one day per request (24 rows per server, within
+the memory API's page of 500 up to twenty servers), all at once, timed with
+a Stopwatch. If the 90-day view proves too slow even for two admins, rows
+older than a week can be merged into one per server per day, about 5,000
+rows for 90 days. Not built until the timings say so.
+
+Rows older than `capacity_retention_days` (90) are deleted daily. Each day
+past retention is swept seven times, so a few days with llmao down still
+get cleared.
+
+### Accepted gaps
+
+- **An llmao restart loses the hour in progress.** Each row's `covered_s`
+ shows how much of its hour was seen. We accept this: the report is read
+ for trends over weeks, and an hour against 90 days is noise. Keeping the
+ hour across restarts would need a second store.
+- **Bandwidth** is billed by Vast per GB, separately from the hourly price,
+ and is left out.
+
+## Cost
+
+- **Self-hosted:** rental dollars per hour (Vast `dph_total`, RunPod
+ `costPerHr`; `usd_per_hour` on a `fleet.hosts` entry overrides), set
+ against tokens served. LiteLLM records self-hosted requests at $0.00 on
+ purpose, since a price there would count against team budgets.
+- **Model providers:** LiteLLM prices each request from its model cost map,
+ reloaded every 24 hours. Spend is fixed when a request is made, so it is
+ LiteLLM's estimate at the time, not the provider's bill. Close enough for
+ planning.
diff --git a/llmao/capacity.py b/llmao/capacity.py
new file mode 100644
index 0000000..4e02924
--- /dev/null
+++ b/llmao/capacity.py
@@ -0,0 +1,317 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# 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.
+
+"""What each self-hosted server did, hour by hour, for capacity planning.
+
+Whether to add a box or return one is a question about weeks: how busy each
+server was, how often requests queued, how full its KV cache got, and how
+many tokens it served. See docs/capacity-design.md for the whole design.
+
+Four parts, each knowing one thing:
+
+ sampler (llmao/sampler.py) reads every serving box's /metrics and hands
+ each reading to the rollup
+ HourlyRollup turns readings into one row per server per
+ clock hour; no I/O, so all the arithmetic is
+ testable on its own
+ CapacityStore the only part that knows the /v1/memory keys:
+ writes rows, reads days, prunes old days
+ CapacityWriter, ...Pruner the two runners: write finished hours, and
+ delete days past retention
+
+The writer is not part of the sampler's loop on purpose. Writing goes to
+LiteLLM; a slow or absent LiteLLM must not stall sampling, so finished hours
+wait in memory until it answers.
+
+An llmao restart loses the hour in progress. Each row says how much of its
+hour was sampled (covered_s), and the loss is accepted: the report is read
+for trends over weeks.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import datetime as dt
+import json
+import logging
+import time
+from dataclasses import dataclass, field
+from typing import Any
+
+from asfpy.stopwatch import Stopwatch
+from easydict import EasyDict as edict # noqa: N813
+
+_LOGGER = logging.getLogger(__name__)
+
+KEY_PREFIX = "llmao:capacity:"
+HOUR_S = 3600
+DAY_S = 86400
+
+
+@dataclass
+class Sample:
+ """One reading of one server's /metrics, as the sampler parsed it.
+
+ Gauges are the value at the moment of reading. Counters and the
+ queue-time histogram are vLLM's totals since the server started; the
+ rollup differences them.
+ """
+
+ ts: float
+ running: float = 0.0
+ waiting: float = 0.0
+ # KV cache in use, as a fraction.
+ kv: float = 0.0
+ # prompt_tokens, generation_tokens, requests, preemptions, queue_s, queued
+ counters: dict[str, float] = field(default_factory=dict)
+ # Queue-time histogram: upper bound in seconds => cumulative count.
+ buckets: dict[float, float] = field(default_factory=dict)
+
+
+def _growth(now: float, before: float) -> float:
+ """How far a counter moved between two readings. A drop is a restart, counted from zero."""
+ return now - before if now >= before else now
+
+
+def hour_of(ts: float) -> int:
+ """The start of the clock hour `ts` falls in, as epoch seconds."""
+ return int(ts // HOUR_S) * HOUR_S
+
+
+def hour_label(hour: int) -> str:
+ """2026-10-18T09, in UTC. Sorts as text, and a day is a prefix of it."""
+ return dt.datetime.fromtimestamp(hour, dt.UTC).strftime("%Y-%m-%dT%H")
+
+
+@dataclass
+class Hour:
+ """One server's clock hour, built up a reading at a time."""
+
+ hour: int
+ server: str
+ model_name: str
+ readings: int = 0
+ busy: int = 0
+ queueing: int = 0
+ waiting_peak: float = 0.0
+ kv_sum: float = 0.0
+ kv_peak: float = 0.0
+ counters: dict[str, float] = field(default_factory=dict)
+ buckets: dict[float, float] = field(default_factory=dict)
+ # The box's rental price, filled in when prices are read (a later step).
+ usd_per_hour: float | None = None
+
+ def add(self, sample: Sample, before: Sample | None) -> None:
+ self.readings += 1
+ self.busy += sample.running > 0
+ self.queueing += sample.waiting > 0
+ self.waiting_peak = max(self.waiting_peak, sample.waiting)
+ self.kv_sum += sample.kv
+ self.kv_peak = max(self.kv_peak, sample.kv)
+ if before is None:
+ # Nothing to difference against: what the counters held before
+ # this reading accrued before llmao was looking.
+ return
+ for key, value in sample.counters.items():
+ self.counters[key] = self.counters.get(key, 0.0) + _growth(value, before.counters.get(key, 0.0))
+ for bound, value in sample.buckets.items():
+ self.buckets[bound] = self.buckets.get(bound, 0.0) + _growth(value, before.buckets.get(bound, 0.0))
+
+ def queue_p95_s(self) -> float | None:
+ """The histogram bound under which 95% of this hour's queued requests fell."""
+ total = self.counters.get("queued", 0.0)
+ if total <= 0:
+ return None
+ for bound in sorted(self.buckets):
+ if self.buckets[bound] >= 0.95 * total:
+ return bound
+ return None
+
+ def row(self, interval_s: float) -> edict:
+ """The stored form, one per server per hour."""
+ n = max(self.readings, 1)
+ queued = self.counters.get("queued", 0.0)
+
+ def count(key: str) -> int:
+ return round(self.counters.get(key, 0.0))
+
+ return edict(
+ hour=hour_label(self.hour),
+ server=self.server,
+ model_name=self.model_name,
+ # How much of the hour was sampled: short after an llmao
+ # restart, or while the box did not answer.
+ covered_s=int(min(self.readings * interval_s, HOUR_S)),
+ busy_frac=round(self.busy / n, 3),
+ waiting_frac=round(self.queueing / n, 3),
+ waiting_peak=int(self.waiting_peak),
+ kv_avg=round(self.kv_sum / n, 3),
+ kv_peak=round(self.kv_peak, 3),
+ prompt_tokens=count("prompt_tokens"),
+ generation_tokens=count("generation_tokens"),
+ requests=count("requests"),
+ preemptions=count("preemptions"),
+ queue_avg_s=round(self.counters.get("queue_s", 0.0) / queued, 3) if queued else None,
+ queue_p95_s=self.queue_p95_s(),
+ usd_per_hour=self.usd_per_hour,
+ )
+
+
+class HourlyRollup:
+ """Readings in, finished hours out. Fed by the sampler, drained by the writer."""
+
+ def __init__(self, interval_s: float):
+ self.interval_s = interval_s
+ self._hours: dict[str, Hour] = {}
+ self._before: dict[str, Sample] = {}
+ self._finished: list[Hour] = []
+
+ def record(self, server: str, model_name: str, sample: Sample) -> None:
+ hour = hour_of(sample.ts)
+ current = self._hours.get(server)
+ if current is None or current.hour != hour:
+ if current is not None:
+ self._finished.append(current)
+ current = self._hours[server] = Hour(hour=hour, server=server, model_name=model_name)
+ current.add(sample, self._before.get(server))
+ self._before[server] = sample
+
+ def missed(self, server: str) -> None:
+ """The box did not answer. Its counters will have moved by the time it
+ does, and that growth cannot be placed in any one hour, so the next
+ reading starts afresh."""
+ self._before.pop(server, None)
+
+ def take_finished(self, now: float) -> list[Hour]:
+ """Every hour that is over, handed to the caller and forgotten here.
+
+ An hour is over when a later one replaced it, or when the clock has
+ moved past it for a server that has gone quiet.
+ """
+ this_hour = hour_of(now)
+ for server, current in list(self._hours.items()):
+ if current.hour < this_hour:
+ self._finished.append(current)
+ del self._hours[server]
+ done, self._finished = self._finished, []
+ return done
+
+ def put_back(self, hours: list[Hour]) -> None:
+ """Hours the writer could not store, kept for its next attempt."""
+ self._finished = hours + self._finished
+
+
+class CapacityStore:
+ """The hourly rows in LiteLLM's /v1/memory, keyed by hour then server:
+
+ llmao:capacity:2026-10-18T09:203.0.113.10:8001
+
+ Handed the MemoryStore.
+ """
+
+ def __init__(self, memory: Any):
+ self._memory = memory
+
+ async def write(self, row: edict) -> None:
+ await self._memory.put(f"{KEY_PREFIX}{row.hour}:{row.server}", json.dumps(row, sort_keys=True))
+
+ async def _day(self, day: dt.date) -> list[edict]:
+ return [edict(item) for item in await self._memory.find(f"{KEY_PREFIX}{day.isoformat()}T")]
+
+ async def read(self, days: int, *, today: dt.date | None = None) -> list[edict]:
+ """Every row of the last `days` days, today included.
+
+ One request per day, all at once. A day holds 24 rows per server, so
+ it fits one page of the memory API (500) up to twenty servers.
+ """
+ today = today or dt.datetime.now(dt.UTC).date()
+ wanted = [today - dt.timedelta(days=n) for n in range(days)]
+ with Stopwatch(name=f"capacity read of {days} days"):
+ found = await asyncio.gather(*(self._day(day) for day in wanted))
+ rows = []
+ for item in (item for day in found for item in day):
+ try:
+ rows.append(edict(json.loads(item.get("value") or "")))
+ except (TypeError, ValueError):
+ continue
+ return rows
+
+ async def prune(self, keep_days: int, *, today: dt.date | None = None, sweep_days: int = 7) -> int:
+ """Delete the rows of the `sweep_days` days just past retention.
+
+ Run daily, each day is swept seven times before it slips out of
+ reach, so a few days with llmao down still get cleared.
+ """
+ today = today or dt.datetime.now(dt.UTC).date()
+ oldest_kept = today - dt.timedelta(days=keep_days - 1)
+ gone = 0
+ for n in range(1, sweep_days + 1):
+ for item in await self._day(oldest_kept - dt.timedelta(days=n)):
+ if item.get("key"):
+ await self._memory.delete(item.key)
+ gone += 1
+ return gone
+
+
+class CapacityWriter:
+ """Runner: stores each finished hour. A failed write is kept and retried."""
+
+ def __init__(self, rollup: HourlyRollup, store: CapacityStore, *, every_s: float = 60):
+ self._rollup = rollup
+ self._store = store
+ self._every = every_s
+
+ async def write_finished(self, now: float) -> None:
+ pending = self._rollup.take_finished(now)
+ for i, hour in enumerate(pending):
+ try:
+ await self._store.write(hour.row(self._rollup.interval_s))
+ except Exception as e:
+ # LiteLLM away: keep this hour and the rest for the next pass.
+ _LOGGER.warning("capacity: could not store %s %s (%s)", hour.server, hour_label(hour.hour), e)
+ self._rollup.put_back(pending[i:])
+ return
+
+ async def run(self) -> None:
+ while True:
+ try:
+ await self.write_finished(time.time())
+ except asyncio.CancelledError:
+ raise
+ except Exception:
+ _LOGGER.exception("capacity: writing failed")
+ await asyncio.sleep(self._every)
+
+
+class CapacityPruner:
+ """Runner: once a day, deletes rows older than the retention period."""
+
+ def __init__(self, store: CapacityStore, keep_days: int):
+ self._store = store
+ self._keep_days = keep_days
+
+ async def run(self) -> None:
+ while True:
+ try:
+ gone = await self._store.prune(self._keep_days)
+ if gone:
+ _LOGGER.info("capacity: pruned %d rows older than %d days", gone, self._keep_days)
+ except asyncio.CancelledError:
+ raise
+ except Exception:
+ _LOGGER.exception("capacity: pruning failed")
+ await asyncio.sleep(DAY_S)
diff --git a/tests/test_capacity.py b/tests/test_capacity.py
new file mode 100644
index 0000000..82c3de0
--- /dev/null
+++ b/tests/test_capacity.py
@@ -0,0 +1,161 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# 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.
+"""The hourly capacity rollup and where its rows are kept (llmao/capacity.py)."""
+
+from __future__ import annotations
+
+import asyncio
+import datetime as dt
+import json
+
+from llmao.capacity import (
+ HOUR_S,
+ CapacityStore,
+ CapacityWriter,
+ HourlyRollup,
+ Sample,
+ hour_label,
+)
+from llmao.proxy.client import MemoryStore
+from tests.fake_transport import fake_memory
+
+# 2026-10-18T09:00:00Z, a clock hour's start.
+T0 = 1792314000.0
+SERVER = "203.0.113.10:8001"
+
+
+def _sample(ts, *, running=0, waiting=0, kv=0.0, **counters):
+ buckets = counters.pop("buckets", {})
+ return Sample(ts=ts, running=running, waiting=waiting, kv=kv, counters=counters, buckets=buckets)
+
+
+def _store():
+ transport, rows = fake_memory()
+ return CapacityStore(MemoryStore(transport)), rows
+
+
+def test_t0_is_on_the_hour():
+ assert T0 % HOUR_S == 0
+ assert hour_label(int(T0)) == "2026-10-18T09"
+
+
+def test_an_hour_of_readings_becomes_one_row():
+ rollup = HourlyRollup(interval_s=15)
+ rollup.record(SERVER, "qwen3-8b", _sample(T0, running=2, kv=0.2, prompt_tokens=100, generation_tokens=10))
+ rollup.record(
+ SERVER, "qwen3-8b", _sample(T0 + 15, running=1, waiting=3, kv=0.6, prompt_tokens=400, generation_tokens=60)
+ )
+ rollup.record(SERVER, "qwen3-8b", _sample(T0 + 30, kv=0.1, prompt_tokens=450, generation_tokens=70))
+ (hour,) = rollup.take_finished(T0 + HOUR_S)
+ row = hour.row(15)
+ assert row.hour == "2026-10-18T09"
+ assert row.server == SERVER
+ assert row.model_name == "qwen3-8b"
+ assert row.covered_s == 45
+ assert row.busy_frac == round(2 / 3, 3)
+ assert row.waiting_frac == round(1 / 3, 3)
+ assert row.waiting_peak == 3
+ assert row.kv_avg == 0.3
+ assert row.kv_peak == 0.6
+ # Growth since the first reading; what came before it is not this hour's.
+ assert row.prompt_tokens == 350
+ assert row.generation_tokens == 60
+ assert row.usd_per_hour is None
+
+
+def test_a_restarted_server_counts_from_zero():
+ rollup = HourlyRollup(interval_s=15)
+ rollup.record(SERVER, "m", _sample(T0, requests=50))
+ rollup.record(SERVER, "m", _sample(T0 + 15, requests=60))
+ rollup.record(SERVER, "m", _sample(T0 + 30, requests=4)) # vLLM restarted
+ (hour,) = rollup.take_finished(T0 + HOUR_S)
+ assert hour.row(15).requests == 14
+
+
+def test_a_missed_reading_does_not_credit_the_gap():
+ rollup = HourlyRollup(interval_s=15)
+ rollup.record(SERVER, "m", _sample(T0, requests=10))
+ rollup.missed(SERVER)
+ rollup.record(SERVER, "m", _sample(T0 + 600, requests=500))
+ (hour,) = rollup.take_finished(T0 + HOUR_S)
+ assert hour.row(15).requests == 0
+ assert hour.row(15).covered_s == 30
+
+
+def test_the_hour_is_finished_by_the_next_reading_or_by_the_clock():
+ rollup = HourlyRollup(interval_s=15)
+ rollup.record(SERVER, "m", _sample(T0))
+ rollup.record("198.51.100.8:8001", "m", _sample(T0))
+ assert rollup.take_finished(T0 + 60) == []
+ # The first server moves into the next hour; the second goes quiet.
+ rollup.record(SERVER, "m", _sample(T0 + HOUR_S))
+ assert [h.server for h in rollup.take_finished(T0 + HOUR_S + 1)] == [SERVER, "198.51.100.8:8001"]
+ # The new hour is still open.
+ assert rollup.take_finished(T0 + HOUR_S + 2) == []
+
+
+def test_queue_time_p95_comes_from_the_histogram():
+ rollup = HourlyRollup(interval_s=15)
+ rollup.record(SERVER, "m", _sample(T0, queued=0, queue_s=0, buckets={0.3: 0, 1.0: 0, 5.0: 0}))
+ rollup.record(SERVER, "m", _sample(T0 + 15, queued=100, queue_s=50, buckets={0.3: 80, 1.0: 96, 5.0: 100}))
+ (hour,) = rollup.take_finished(T0 + HOUR_S)
+ row = hour.row(15)
+ assert row.queue_p95_s == 1.0
+ assert row.queue_avg_s == 0.5
+
+
+def test_a_quiet_hour_has_no_queue_figures():
+ rollup = HourlyRollup(interval_s=15)
+ rollup.record(SERVER, "m", _sample(T0))
+ (hour,) = rollup.take_finished(T0 + HOUR_S)
+ assert hour.row(15).queue_p95_s is None
+ assert hour.row(15).queue_avg_s is None
+
+
+def test_rows_are_kept_by_hour_then_server_and_read_by_day():
+ store, rows = _store()
+ rollup = HourlyRollup(interval_s=15)
+ rollup.record(SERVER, "m", _sample(T0))
+ (hour,) = rollup.take_finished(T0 + HOUR_S)
+ asyncio.run(store.write(hour.row(15)))
+ assert list(rows) == [f"llmao:capacity:2026-10-18T09:{SERVER}"]
+ today = dt.date(2026, 10, 18)
+ assert [r.server for r in asyncio.run(store.read(1, today=today))] == [SERVER]
+ assert asyncio.run(store.read(1, today=today + dt.timedelta(days=1))) == []
+ assert len(asyncio.run(store.read(2, today=today + dt.timedelta(days=1)))) == 1
+
+
+def test_pruning_drops_only_the_days_past_retention():
+ store, rows = _store()
+ for day in ("2026-07-19", "2026-07-20", "2026-07-21"):
+ rows[f"llmao:capacity:{day}T00:{SERVER}"] = json.dumps({"hour": f"{day}T00"})
+ # Keeping 90 days on 2026-10-18 keeps 2026-07-21 onwards.
+ gone = asyncio.run(store.prune(90, today=dt.date(2026, 10, 18)))
+ assert gone == 2
+ assert list(rows) == [f"llmao:capacity:2026-07-21T00:{SERVER}"]
+
+
+def test_an_hour_that_cannot_be_stored_waits_for_the_next_pass():
+ transport, rows = fake_memory(down=True)
+ rollup = HourlyRollup(interval_s=15)
+ writer = CapacityWriter(rollup, CapacityStore(MemoryStore(transport)))
+ rollup.record(SERVER, "m", _sample(T0))
+ asyncio.run(writer.write_finished(T0 + HOUR_S))
+ assert rows == {}
+ transport.down = False
+ asyncio.run(writer.write_finished(T0 + HOUR_S + 60))
+ assert list(rows) == [f"llmao:capacity:2026-10-18T09:{SERVER}"]
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org