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:40 UTC
(tooling-llmao) 01/02: capacity: hourly rollup of vLLM readings, kept in LiteLLM's memory store
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