From 43f113dbd6ab6f868fc89a6de0c6738b7467b045 Mon Sep 17 00:00:00 2001 From: Cam Quilici Date: Fri, 14 Aug 2026 10:45:05 -0500 Subject: [PATCH 1/3] fix(agentic): sum server metrics on a canonical scrape grid MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit v13 gave the AVERAGED kv-cache series a real per-engine identity. The summed series never got the same treatment: prefill/decode/prefix-hit rates, queue depth, host KV usage and the prompt-token source breakdown still grouped on an exact `start_ns`, so each output tick carried whichever components happened to share that nanosecond rather than the cluster total. Components of one metric are not scraped in lockstep. Two shapes produce it: 1. Disaggregated runs put every worker on its own /metrics endpoint and its own sub-second offset, so an N-worker run emitted N partial-sum points per tick. 2. SGLang splits its token counters by `is_streaming`. On a fully streaming benchmark the "false" series is a full-length run of zeros on a grid ~16 ms off the real one. Downstream this reads as a comb. `rollingAverage` is a sample-count mean, so it averaged real samples together with the other components' structural gaps and drew a fraction of the actual throughput — measured 2.2x low on the 8-rank SGLang DP row (point 439817) and ~7x low on the 7-worker disaggregated rows. 223 of 599 stored rows carry a fragmented prefill series. Every summed series now goes through `sumOntoGrid`: one canonical grid at the blob's native cadence (`canonicalTickNs`), each component holding its last sample between its own scrapes and contributing only inside its observed window. The grid must be uniform rather than the union of scrape instants the way `averageAcrossEngines` does it — a mean is scale-free but a sum is not, and `cumulativeUniqueInputTokens` turns these rates into token totals with `sum += value`, which only stays correct at one point per scrape bucket. The lattice is anchored at t=0 and shared across metrics so the pairs that are divided or added downstream (hits/queries, used/total, running+waiting) land on identical `t`. Anchoring per-metric instead emptied the prefix-cache-hit-rate chart outright, because sglang:cached_tokens starts ~0.18 s off the grid sglang:prompt_tokens starts on. Mirrored API-server endpoints are collapsed on the summed path too, using a relative tolerance — a throughput mean is O(10^5), so v13's absolute gauge threshold would never fire. That is a correction as well as a de-duplication: the two-API-server vLLM rows were double-counting queue depth and token totals exactly 2x. Measured on real blobs (stored v12 -> v14): 439817 sglang/b200 DP8 decode mean 393 -> 859 tok/s; token totals byte-identical 439201 vllm/b300 DP8 queue depth sum 25,680 -> 12,831 (mirror double-count) 439263 vllm/b200 DP4 queue depth sum 155,011 -> 77,507 (mirror double-count) `aggregateByStart` is gone; nothing groups on an exact `start_ns` any more. CHART_SERIES_VERSION 13 -> 14. Run `bun run --cwd packages/db db:backfill-chart-series` after merge — until then the API serves these points through the slower recompute path. 中文：v13 只为「取平均」的 KV 缓存序列建立了真正的按引擎身份，「求和」类序列 （prefill/decode/前缀缓存命中速率、队列深度、host KV 使用率、prompt token 来源 拆分）仍按精确的 start_ns 分组。同一指标的各个组件并非同步采样：分离式部署下每 个 worker 有各自的 /metrics 端点与亚秒级偏移；SGLang 还会按 is_streaming 把 token 计数器拆成两条序列，全流式压测时 "false" 那条是一整串零、且与真实网格相差约 16 ms。 结果是每个输出点只携带恰好落在该纳秒上的那部分组件，前端的 rollingAverage 又是按 样本计数取平均，于是把真实样本和其它组件的结构性空档一起平均，画出来只有实际吞吐量 的一小部分——在 8 路 DP 的 SGLang 行（439817）上偏低 2.2 倍，在 7 worker 的分离式 行上偏低约 7 倍；599 行已存数据中有 223 行的 prefill 序列存在此问题。 现在所有求和序列都经由 sumOntoGrid：以 blob 原生采样周期构建统一网格 （canonicalTickNs），每个组件在自身两次采样之间保持上一个取值，且只在自己的观测 窗口内参与求和。网格必须是均匀的，不能像 averageAcrossEngines 那样取采样时刻的并 集——平均值与尺度无关，求和则不然，而 cumulativeUniqueInputTokens 是用 sum += value 把速率累加成 token 总量的，只有「每个采样桶一个点」时才成立。网格锚定 在 t=0 并被所有指标共享，这样下游需要相除或相加的成对指标才能落在相同的 t 上；否则 前缀缓存命中率图会直接变空。求和路径同样合并镜像端点，但改用相对容差（吞吐量均值量 级为 10^5，v13 的绝对阈值永远不会触发）——这同时修正了双 API server 的 vLLM 行此前 队列深度与 token 总量正好翻倍的问题。aggregateByStart 已删除。 CHART_SERIES_VERSION 13 → 14，合并后需运行 db:backfill-chart-series。 Co-Authored-By: Claude Opus 5 (1M context) --- docs/data-pipeline.md | 12 + .../db/src/etl/compute-chart-series.test.ts | 196 +++++++++ packages/db/src/etl/compute-chart-series.ts | 377 ++++++++++++++---- 3 files changed, 497 insertions(+), 88 deletions(-) diff --git a/docs/data-pipeline.md b/docs/data-pipeline.md index 24fce809..beb5d151 100644 --- a/docs/data-pipeline.md +++ b/docs/data-pipeline.md @@ -101,6 +101,18 @@ The cluster average is then a mean across those logical engines on the union of Engines are ordered by role, then numeric rank, then worker — never by the composed display string, which would sort `"decode 10"` before `"decode 2"` and scramble DP ranks. The role is only shown when engines actually differ in role, so an aggregated deployment reads `DP 0…DP 3` rather than `decode 0…decode 3`. +### Summed Series and the Canonical Grid + +The same "components aren't scraped in lockstep" problem hits the **summed** series — prefill/decode/prefix-hit rates, queue depth, host KV usage, and the prompt-token source breakdown — but it cannot be solved the same way. A mean is scale-free, so `averageAcrossEngines` can evaluate on the union of scrape instants; a sum is not. `cumulativeUniqueInputTokens` turns these rates into token totals with `sum += value` and `rollingAverage` is a sample-count mean, so both only stay correct at **one point per scrape bucket**. + +Summing on an exact `start_ns` emitted one point per component per tick, each holding that component's share alone. Downstream that reads as a comb: the rolling average mixed the real samples with the other components' structural gaps and drew roughly 1/N of the cluster total. Two shapes produced it in the corpus — a disaggregated run puts every worker on its own `/metrics` endpoint and its own sub-second offset (~7× low on the 7-worker rows), and even a single-endpoint SGLang run splits its token counters into `is_streaming="true"`/`"false"` series ~16 ms apart, where the `"false"` half is an all-zero run for a fully streaming benchmark (2.2× low). + +Every summed series is therefore evaluated by `sumOntoGrid` on one canonical grid at the blob's native scrape cadence (`canonicalTickNs`, the median gap of the best-sampled series), with each component holding its last sample between its own scrapes and contributing only inside its observed window. The lattice is anchored at `t=0` and shared by every metric, so the pairs that get divided or added downstream (hits/queries, used/total, running+waiting) land on identical `t` values — anchoring each metric at its own first sample instead silently emptied the prefix-cache-hit-rate chart, because `sglang:cached_tokens` starts ~0.18 s off the grid `sglang:prompt_tokens` starts on. + +Mirrored endpoints are collapsed here too, on a **relative** tolerance rather than the gauge path's absolute one — a throughput mean is O(10⁵), so an absolute threshold would never fire and every mirror would be counted twice. This is a correction as well as a de-duplication: the two-API-server vLLM rows were double-counting their queue depth and token totals exactly 2× before v14. + +Nothing groups on an exact `start_ns` any more; `aggregateByStart` was removed in v14. + ### Agentic Dataset Provenance AIPerf exports public-dataset provenance in `metadata.dataset`, including the Hugging Face dataset ID. InferenceX preserves that object as `dataset` on each agentic aggregate benchmark row. During benchmark ingest, `ingest-ci-run.ts` derives the dashboard slug from `hf_dataset_name` (for example, `semianalysisai/cc-traces-weka-062126` becomes `cc-traces-weka-062126`) and upserts `run_datasets` for the workflow run. diff --git a/packages/db/src/etl/compute-chart-series.test.ts b/packages/db/src/etl/compute-chart-series.test.ts index e1bd7ea5..77e2a2b2 100644 --- a/packages/db/src/etl/compute-chart-series.test.ts +++ b/packages/db/src/etl/compute-chart-series.test.ts @@ -707,3 +707,199 @@ describe('computeChartSeries', () => { expect(result?.metricSources).toEqual([]); }); }); + +// ── Summed series on the canonical grid (v14) ─────────────────────────── +// +// Components of one metric are not scraped in lockstep. Summing them on an +// exact `start_ns` emitted one point per component per tick, each holding that +// component's share alone — a comb that `rollingAverage` (a sample-count mean) +// then read as a fraction of the real cluster total. + +/** A counter series for one component, sampled at `1 Hz` from `startNs`. */ +function rateSeriesFor( + endpoint_url: string, + labels: Record, + startNs: number, + rates: number[], +) { + return { + endpoint_url, + labels, + timeslices: rates.map((rate, i) => ({ + start_ns: startNs + i * 1e9, + end_ns: startNs + (i + 1) * 1e9, + rate, + })), + }; +} + +function rateBlob(metricName: string, series: unknown[]) { + return gzipSync(Buffer.from(JSON.stringify({ metrics: { [metricName]: { series } } }))); +} + +/** Σ of a rate series ≈ total tokens, which is what the cumulative charts read. */ +function total(points: { value: number }[] | undefined): number { + return (points ?? []).reduce((a, p) => a + p.value, 0); +} + +describe('computeChartSeries summed series', () => { + it('sums components scraped on offset grids instead of emitting a comb', async () => { + // Two workers at 1 Hz, 16 ms out of phase — the real SGLang/vLLM pattern. + // Grouping on an exact start_ns produced 6 points alternating 100 and 20; + // the cluster was never once reported as its actual 120 tok/s. + const cs = await computeChartSeries( + rateBlob('vllm:prompt_tokens', [ + rateSeriesFor('http://a:8000/metrics', { worker_id: 'a' }, 0, [100, 100, 100]), + rateSeriesFor('http://b:8000/metrics', { worker_id: 'b' }, 0.016e9, [20, 20, 20]), + ]), + ); + expect(cs?.prefillTps).toEqual([ + { t: 0, value: 100 }, + { t: 1, value: 120 }, + { t: 2, value: 120 }, + ]); + }); + + it('ignores a label split that reports nothing rather than halving the rate', async () => { + // SGLang splits its token counters by `is_streaming`. When every request + // streams, the "false" series is a full-length run of zeros on its own + // grid, and interleaving it dragged the rolling average down ~2.2x. + const cs = await computeChartSeries( + rateBlob('sglang:generation_tokens', [ + rateSeriesFor('http://a:8000/metrics', { is_streaming: 'false' }, 0, [0, 0, 0, 0]), + rateSeriesFor('http://a:8000/metrics', { is_streaming: 'true' }, 0.016e9, [80, 90, 70, 60]), + ]), + ); + expect(cs?.decodeTps.map((p) => p.value)).toEqual([0, 80, 90, 70]); + }); + + it('preserves the token total, which the cumulative charts read as a sum', async () => { + // `cumulativeUniqueInputTokens` does `sum += value`, so regridding must not + // move Σ — one point per one-second bucket, no more and no less. Components + // already on the grid round-trip exactly. + const rates = [10, 40, 0, 25, 5]; + const aligned = await computeChartSeries( + rateBlob('vllm:prompt_tokens', [ + rateSeriesFor('http://a:8000/metrics', { worker_id: 'a' }, 0, rates), + rateSeriesFor('http://b:8000/metrics', { worker_id: 'b' }, 0, rates), + ]), + ); + expect(total(aligned?.prefillTps)).toBe(160); + expect(aligned?.prefillTps.map((p) => p.t)).toEqual([0, 1, 2, 3, 4]); + + // An off-grid component is read at each tick through its own step + // function, so Σ can differ by at most the one trailing bucket that falls + // past the last whole tick (here worker b's final 5). On a real 4000-tick + // row that edge is invisible: points 439817's prefill and decode totals + // both came out byte-identical to the pre-v14 values. + const offset = await computeChartSeries( + rateBlob('vllm:prompt_tokens', [ + rateSeriesFor('http://a:8000/metrics', { worker_id: 'a' }, 0, rates), + rateSeriesFor('http://b:8000/metrics', { worker_id: 'b' }, 0.4e9, rates), + ]), + ); + expect(total(offset?.prefillTps)).toBe(155); + expect(offset?.prefillTps.map((p) => p.t)).toEqual([0, 1, 2, 3, 4]); + }); + + it('counts mirrored API-server frontends once, not twice', async () => { + // vLLM with two API servers exposes the same engine on both /metrics + // endpoints. Summing both double-counted every token — measured at exactly + // 2x on the stored rows for points 439201 and 439263. + const cs = await computeChartSeries( + rateBlob('vllm:prompt_tokens', [ + rateSeriesFor('http://localhost:8888/metrics', { engine: '0' }, 0, [500, 500, 500]), + rateSeriesFor('http://localhost:8889/metrics', { engine: '0' }, 0.019e9, [500, 500, 500]), + ]), + ); + expect(cs?.prefillTps.map((p) => p.value)).toEqual([500, 500, 500]); + }); + + it('keeps endpoints that disagree, which are engines rather than mirrors', async () => { + // Same label set on two endpoints but very different levels: a router in + // front of two replicas. Dropping one would silently lose half the load. + const cs = await computeChartSeries( + rateBlob('vllm:prompt_tokens', [ + rateSeriesFor('http://localhost:8888/metrics', { engine: '0' }, 0, [100, 100, 100]), + rateSeriesFor('http://localhost:8889/metrics', { engine: '0' }, 0, [900, 900, 900]), + ]), + ); + expect(cs?.prefillTps.map((p) => p.value)).toEqual([1000, 1000, 1000]); + }); + + it('puts divided metrics on one lattice so the hit rate survives', async () => { + // Regression guard: anchoring each metric's grid at its own first sample + // put hits on ...x.18 and queries on ...x.00, so the join found no shared + // instant and the prefix-cache-hit-rate chart came out empty. + const cs = await computeChartSeries( + gzipSync( + Buffer.from( + JSON.stringify({ + metrics: { + 'vllm:prefix_cache_hits': { + series: [ + rateSeriesFor('http://a:8000/metrics', { engine: '0' }, 0.18e9, [75, 75]), + rateSeriesFor('http://a:8000/metrics', { engine: '1' }, 0.34e9, [75, 75]), + ], + }, + 'vllm:prefix_cache_queries': { + series: [ + rateSeriesFor('http://a:8000/metrics', { engine: '0' }, 0, [100, 100]), + rateSeriesFor('http://a:8000/metrics', { engine: '1' }, 0.5e9, [100, 100]), + ], + }, + }, + }), + ), + ), + ); + expect(cs?.prefixCacheHitRate.length).toBeGreaterThan(0); + for (const point of cs!.prefixCacheHitRate) { + expect(Number.isInteger(point.t)).toBe(true); + expect(point.value).toBeCloseTo(0.75, 5); + } + }); + + it('sums queue depth across workers on their own scrape offsets', async () => { + const json = JSON.stringify({ + metrics: { + 'vllm:num_requests_running': { + series: [0, 1, 2, 3].map((w) => ({ + endpoint_url: `http://10.0.0.${w}:7500/metrics`, + labels: { worker_id: `w${w}` }, + timeslices: [ + { start_ns: w * 0.01e9, end_ns: w * 0.01e9 + 1e9, avg: 2 }, + { start_ns: w * 0.01e9 + 1e9, end_ns: w * 0.01e9 + 2e9, avg: 2 }, + ], + })), + }, + 'vllm:num_requests_waiting': { + series: [0, 1, 2, 3].map((w) => ({ + endpoint_url: `http://10.0.0.${w}:7500/metrics`, + labels: { worker_id: `w${w}` }, + timeslices: [ + { start_ns: w * 0.01e9, end_ns: w * 0.01e9 + 1e9, avg: 1 }, + { start_ns: w * 0.01e9 + 1e9, end_ns: w * 0.01e9 + 2e9, avg: 1 }, + ], + })), + }, + }, + }); + const cs = await computeChartSeries(gzipSync(Buffer.from(json))); + // Four workers x (2 running + 1 waiting) once the last one has reported. + expect(cs?.queueDepth.at(-1)).toEqual({ t: 1, running: 8, waiting: 4, total: 12 }); + }); + + it('leaves a single-component metric on a plain one-per-tick grid', async () => { + const cs = await computeChartSeries( + rateBlob('vllm:generation_tokens', [ + rateSeriesFor('http://a:8000/metrics', { engine: '0' }, 0, [10, 20, 30]), + ]), + ); + expect(cs?.decodeTps).toEqual([ + { t: 0, value: 10 }, + { t: 1, value: 20 }, + { t: 2, value: 30 }, + ]); + }); +}); diff --git a/packages/db/src/etl/compute-chart-series.ts b/packages/db/src/etl/compute-chart-series.ts index d719e0e6..3e5e7a3e 100644 --- a/packages/db/src/etl/compute-chart-series.ts +++ b/packages/db/src/etl/compute-chart-series.ts @@ -83,8 +83,30 @@ import { * `seriesIdentityKey`), and the cluster average is a real mean across those * logical engines (see `averageAcrossEngines`) rather than a mean over * whichever engines happened to share an exact `start_ns`. + * + * v14: give the SUMMED series the same treatment v13 gave the averaged one. + * Components of a metric are not scraped in lockstep — a disaggregated run + * puts every worker on its own `/metrics` endpoint and its own sub-second + * offset, and even a single-endpoint SGLang run splits its token counters into + * `is_streaming="true"`/`"false"` series ~16 ms apart — so summing on an exact + * `start_ns` emitted one point per component per tick, each carrying that + * component's share alone instead of the cluster total. The detail page reads + * those as a comb: `rollingAverage` is a sample-count mean, so it averaged the + * real samples together with the other components' structural gaps and drew + * roughly 1/N of the actual throughput (measured: 2.2x low on an 8-rank SGLang + * DP run, ~7x low on the 7-worker disaggregated rows). + * + * Every summed series — prefill/decode/prefix-hit rates, queue depth, host KV + * usage and the prompt-token source breakdown — is now evaluated on one + * canonical grid at the blob's native scrape cadence (`canonicalTickNs`), with + * each component holding its last sample between its own scrapes + * (`sumOntoGrid`). The grid has to be uniform, not the union of scrape + * instants the way `averageAcrossEngines` does it: a mean is scale-free but a + * sum is not, and `cumulativeUniqueInputTokens` turns these rates into token + * totals with `sum += value`, which only stays correct at one point per + * scrape bucket. */ -export const CHART_SERIES_VERSION = 13; +export const CHART_SERIES_VERSION = 14; export interface TimeSeriesPoint { /** Seconds from benchmark start. */ @@ -289,38 +311,11 @@ export function computeChartSeriesFromMetricPhases( return buildSeriesFromMetrics(mergePhaseMetrics(profiling, warmup), context); } -/** - * Aggregate one timeslice field across all series of a metric, indexed by - * `start_ns`. Multi-engine vllm deployments report one series per engine — - * the cluster value is the sum (for running/waiting/throughput counters) - * or the average (for kv_cache_usage_perc, a per-engine fraction). - */ -function aggregateByStart( - series: readonly RawSeries[] | undefined, - field: 'avg' | 'rate', - combine: 'sum' | 'avg', -): Map { - const sums = new Map(); - const counts = new Map(); - for (const s of series ?? []) { - for (const ts of s.timeslices ?? []) { - if (typeof ts.start_ns !== 'number') continue; - const v = ts[field]; - if (typeof v !== 'number' || !Number.isFinite(v)) continue; - sums.set(ts.start_ns, (sums.get(ts.start_ns) ?? 0) + v); - counts.set(ts.start_ns, (counts.get(ts.start_ns) ?? 0) + 1); - } - } - if (combine === 'sum') return sums; - const out = new Map(); - for (const [t, s] of sums) out.set(t, s / (counts.get(t) ?? 1)); - return out; -} - -/** Stable order: emit one point per unique start_ns, chronologically. */ -function sortedEntries(m: Map): [number, number][] { - return [...m.entries()].toSorted((a, b) => a[0] - b[0]); -} +// Note: v14 removed `aggregateByStart`/`sortedEntries`. Nothing groups on an +// exact `start_ns` any more — averaged series go through `averageAcrossEngines` +// and summed ones through `sumOntoGrid`, both of which treat a component's +// samples as a step function rather than requiring components to share a +// nanosecond. // ── Per-engine identity (v13) ─────────────────────────────────────────── // @@ -456,6 +451,38 @@ function engineDiscriminator( */ const MIRROR_MEAN_TOLERANCE = 0.02; +/** + * Same idea as `MIRROR_MEAN_TOLERANCE`, but for counter rates (v14), where an + * absolute threshold is meaningless — a token-throughput mean is O(10^5), so + * the gauge tolerance would never fire and every mirror would be counted twice. + * + * Expressed as a fraction of the larger mean. The measured mirrors in the + * corpus differ by 0.03%-2.19% of their means, and a router endpoint that + * aggregates several workers differs from any single worker by far more than + * 5%, so this sits above the observed mirror spread with room to spare while + * staying well below a genuine difference. + */ +const MIRROR_RATE_RELATIVE_TOLERANCE = 0.05; + +/** + * Do these endpoints' whole-run means agree closely enough to be two views of + * one series rather than two different ones? + * + * Gauges keep v13's absolute test unchanged — they are 0..1 fractions (or small + * request counts) whose mirrors agree almost exactly. Rates need a relative + * test because their magnitude is unbounded. + */ +function looksMirrored(means: readonly number[], field: 'avg' | 'rate'): boolean { + const spread = Math.max(...means) - Math.min(...means); + if (field === 'avg') return spread <= MIRROR_MEAN_TOLERANCE; + const scale = Math.max(...means.map((m) => Math.abs(m))); + // Two endpoints that both read ~0 are mirrors of an idle series, not two + // engines: fall back to the absolute test so the relative one can't divide + // by a vanishing scale. + if (scale === 0) return true; + return spread <= MIRROR_RATE_RELATIVE_TOLERANCE * scale; +} + interface LogicalEngine { engineLabel: string; points: TimeSeriesPoint[]; @@ -522,10 +549,11 @@ function spanOf(scrapes: ScrapeMap): number { * behind one label (a router in front of several replicas), and dropping * one would silently lose an engine, so those are kept separate instead. */ -function resolveLogicalEngines( +function resolveComponents( series: readonly RawSeries[] | undefined, tOf: (ns: number) => number, -): LogicalEngine[] { + field: 'avg' | 'rate' = 'avg', +): ResolvedEngine[] { const groups = new Map(); for (const s of series ?? []) { const key = seriesIdentityKey(s); @@ -542,13 +570,14 @@ function resolveLogicalEngines( } for (const ts of s.timeslices ?? []) { if (typeof ts.start_ns !== 'number' || !Number.isFinite(ts.start_ns)) continue; - if (typeof ts.avg !== 'number' || !Number.isFinite(ts.avg)) continue; + const value = ts[field]; + if (typeof value !== 'number' || !Number.isFinite(value)) continue; const at = scrapes.get(ts.start_ns); if (at) { - at.sum += ts.avg; + at.sum += value; at.count++; } else { - scrapes.set(ts.start_ns, { sum: ts.avg, count: 1 }); + scrapes.set(ts.start_ns, { sum: value, count: 1 }); } } } @@ -579,7 +608,7 @@ function resolveLogicalEngines( } const means = endpoints.map(([, scrapes]) => meanOf(scrapes)); - const mirrored = Math.max(...means) - Math.min(...means) <= MIRROR_MEAN_TOLERANCE; + const mirrored = looksMirrored(means, field); if (!mirrored) { // Same labels, different measurements: distinct engines, not mirrors. for (const [endpointUrl, scrapes] of endpoints) push(endpointUrl, scrapes, true); @@ -602,13 +631,21 @@ function resolveLogicalEngines( // first, then numeric rank, then worker, then blob order. Sorting a label // like "decode 3" lexically (or falling back to array index) scrambled DP // ranks on multi-worker runs. - const ordered = resolved.toSorted( + return resolved.toSorted( (a, b) => engineRoleSortKey(a.role) - engineRoleSortKey(b.role) || engineRankSortKey(a.rank) - engineRankSortKey(b.rank) || (a.discriminator ?? '').localeCompare(b.discriminator ?? '') || a.order - b.order, ); +} + +/** Collapse a gauge's raw series into one labelled entry per logical engine. */ +function resolveLogicalEngines( + series: readonly RawSeries[] | undefined, + tOf: (ns: number) => number, +): LogicalEngine[] { + const ordered = resolveComponents(series, tOf, 'avg'); // Only name the role when there is more than one, otherwise every engine on // an aggregated deployment reads "decode 0", "decode 1", ... for no reason. @@ -716,6 +753,173 @@ function carryLimitSeconds(points: readonly TimeSeriesPoint[]): number { return 5 * gaps[gaps.length >> 1]!; } +// ── Canonical scrape grid for summed series (v14) ─────────────────────── +// +// `averageAcrossEngines` can evaluate on the union of scrape instants because +// a mean is scale-free: adding more evaluation points doesn't change the level. +// A SUM cannot. Its consumers on the detail page treat one point as one +// one-second bucket — `cumulativeUniqueInputTokens` does `sum += value` to turn +// a token *rate* into a token *total*, and `rollingAverage` is a sample-count +// mean — so the emitted grid has to be uniform and at the native scrape +// cadence, or the totals and the levels both move. + +/** Widest interval we will ever call a scrape cadence, in ns (5 minutes). */ +const MAX_TICK_NS = 300 * 1e9; +/** Hard cap on emitted ticks, so a malformed cadence can't blow up the row. */ +const MAX_GRID_TICKS = 500_000; +/** + * Slack when snapping a component's first/last sample to the lattice, as a + * fraction of a tick. A sample sitting a floating-point hair outside the grid + * should still get its tick rather than losing an endpoint. + */ +const GRID_EPSILON = 1e-6; + +/** + * The blob's native scrape interval, in nanoseconds. + * + * Taken as the median gap of the single series with the most timeslices: that + * series is by construction the best-sampled view of the run, and a median + * shrugs off the one-off boundary gap that the profiling block leaves where + * warmup used to be. Returns null when nothing in the blob has two samples to + * compare, in which case callers fall back to the pre-v14 exact-instant path. + */ +function canonicalTickNs(metrics: MetricsMap): number | null { + let best: readonly RawSlice[] | null = null; + for (const metricMeta of Object.values(metrics)) { + for (const s of metricMeta?.series ?? []) { + const ts = s.timeslices ?? []; + if (ts.length > (best?.length ?? 0)) best = ts; + } + } + if (!best || best.length < 2) return null; + const gaps: number[] = []; + for (let i = 1; i < best.length; i++) { + const a = best[i - 1]!.start_ns; + const b = best[i]!.start_ns; + if (typeof a !== 'number' || typeof b !== 'number') continue; + const gap = b - a; + if (gap > 0 && gap <= MAX_TICK_NS) gaps.push(gap); + } + if (gaps.length === 0) return null; + gaps.sort((a, b) => a - b); + const median = gaps[gaps.length >> 1]!; + return median > 0 ? median : null; +} + +/** + * Evaluate a step-held component at `t`, or null when it has nothing to say. + * + * `cursor` is carried across calls so the whole sweep stays linear; callers + * must visit `t` in ascending order. + */ +interface StepCursor { + points: readonly TimeSeriesPoint[]; + carryLimit: number; + cursor: number; +} + +function stepValueAt(state: StepCursor, t: number): number | null { + const { points } = state; + while (state.cursor + 1 < points.length && points[state.cursor + 1]!.t <= t) state.cursor++; + // Outside the component's own observed window it contributes nothing, so a + // worker that starts late or stops early neither invents load nor forces a + // zero into the cluster total. + if (state.cursor < 0 || t > points.at(-1)!.t) return null; + // Inside the window but long past the last sample: the component stopped + // reporting, so don't carry a stale reading across the hole. + if (t - points[state.cursor]!.t > state.carryLimit) return null; + return points[state.cursor]!.value; +} + +/** + * Sum components onto a uniform grid at the native scrape cadence. + * + * Components of one metric are NOT scraped in lockstep: a disaggregated run + * gives every worker its own `/metrics` endpoint on its own sub-second offset, + * and even a single-endpoint SGLang run splits its token counters into + * `is_streaming="true"`/`"false"` series that sit ~16 ms apart. Summing on an + * exact `start_ns` therefore emitted one point per component per tick, each + * holding that component's share alone. Downstream that reads as a comb — + * `rollingAverage` averages the real samples together with the other + * components' structural gaps, so an 8-worker run's throughput chart drew + * roughly an eighth of the actual tokens/sec. + * + * Each component instead holds its last sample between scrapes (its rate is a + * step function over its bucket) and the grid samples the sum once per tick. + * Returns an empty array when there is nothing to place on a grid. + */ +function sumOntoGrid( + components: readonly (readonly TimeSeriesPoint[])[], + tickS: number | null, +): TimeSeriesPoint[] { + const active = components.filter((points) => points.length > 0); + if (active.length === 0) return []; + const first = Math.min(...active.map((points) => points[0]!.t)); + const last = Math.max(...active.map((points) => points.at(-1)!.t)); + if (!Number.isFinite(first) || !Number.isFinite(last)) return []; + + // Every summed series shares one lattice anchored at t=0, so metrics that + // are divided by one another downstream (hit rate = hits/queries, host KV + // usage = used/total, queue depth = running+waiting) land on identical `t` + // values and can be joined. Anchoring each metric at its OWN first sample + // instead silently emptied the hit-rate chart, because `sglang:cached_tokens` + // starts ~0.18 s off the grid `sglang:prompt_tokens` starts on and the two + // lattices then never share a point. + // + // With no inferable cadence (every series carrying a single timeslice, as in + // the smallest fixtures) fall back to the union of the components' own + // instants: still a step-held sum rather than an exact-instant one, just + // without a regular grid to place it on. + let timeline: number[]; + if (tickS && tickS > 0) { + const firstTick = Math.ceil(first / tickS - GRID_EPSILON); + const lastTick = Math.floor(last / tickS + GRID_EPSILON); + const ticks = lastTick - firstTick + 1; + if (ticks < 1 || ticks > MAX_GRID_TICKS) return []; + timeline = Array.from({ length: ticks }, (_, i) => (firstTick + i) * tickS); + } else { + timeline = [...new Set(active.flatMap((points) => points.map((p) => p.t)))].toSorted( + (a, b) => a - b, + ); + } + + const states: StepCursor[] = active.map((points) => ({ + points, + carryLimit: carryLimitSeconds(points), + cursor: -1, + })); + const out: TimeSeriesPoint[] = []; + for (const t of timeline) { + let sum = 0; + let n = 0; + for (const state of states) { + const value = stepValueAt(state, t); + if (value === null) continue; + sum += value; + n++; + } + // No component covers this tick — emit nothing rather than a false zero. + if (n > 0) out.push({ t, value: sum }); + } + return out; +} + +/** `sumOntoGrid` over a metric's raw series, resolved to logical components. */ +function summedSeries( + series: readonly RawSeries[] | undefined, + tOf: (ns: number) => number, + field: 'avg' | 'rate', + tickS: number | null, +): TimeSeriesPoint[] { + const components = resolveComponents(series, tOf, field).map((c) => c.points); + return sumOntoGrid(components, tickS); +} + +/** Index a grid series by `t` so two of them can be divided tick-for-tick. */ +function byT(points: readonly TimeSeriesPoint[]): Map { + return new Map(points.map((p) => [p.t, p.value])); +} + function buildSeriesFromMetrics( metrics: MetricsMap, context: ServerMetricsContext, @@ -742,6 +946,12 @@ function buildSeriesFromMetrics( } if (!Number.isFinite(startNs)) startNs = 0; const tOf = (ns: number) => (ns - (originStartNs ?? startNs)) / 1e9; + // Grid spacing for every summed series (v14). Null when the blob has nothing + // with two comparable samples, in which case `sumOntoGrid` emits only the + // single-component case and multi-component metrics come out empty rather + // than fragmented. + const tickNs = canonicalTickNs(metrics); + const tickS = tickNs === null ? null : tickNs / 1e9; // Pick the first metric name whose series array has any data; fallback // chain lets the same code path serve both vllm:* and sglang:* blobs. @@ -778,19 +988,21 @@ function buildSeriesFromMetrics( 'vllm:prompt_tokens', 'sglang:prompt_tokens', ); - const hitsByT = aggregateByStart(hitsSeries, 'rate', 'sum'); - const qsByT = aggregateByStart(qsSeries, 'rate', 'sum'); + const hitsOnGrid = summedSeries(hitsSeries, tOf, 'rate', tickS); + const qsOnGrid = summedSeries(qsSeries, tOf, 'rate', tickS); + const qsByT = byT(qsOnGrid); const prefixCacheHitRate: TimeSeriesPoint[] = []; - for (const [t, h] of sortedEntries(hitsByT)) { + for (const { t, value: h } of hitsOnGrid) { const q = qsByT.get(t); - if (q !== undefined && q > 0) prefixCacheHitRate.push({ t: tOf(t), value: h / q }); + if (q !== undefined && q > 0) prefixCacheHitRate.push({ t, value: h / q }); } // Queue depth: sum running + waiting across engines per timeslice. const runSeries = pickSeries('vllm:num_requests_running', 'sglang:num_running_reqs'); const waitSeries = pickSeries('vllm:num_requests_waiting', 'sglang:num_queue_reqs'); - const runByT = aggregateByStart(runSeries, 'avg', 'sum'); - const waitByT = aggregateByStart(waitSeries, 'avg', 'sum'); + const runOnGrid = summedSeries(runSeries, tOf, 'avg', tickS); + const waitByT = byT(summedSeries(waitSeries, tOf, 'avg', tickS)); + const runByT = byT(runOnGrid); const queueDepth: QueueDepthPoint[] = []; // Union of timestamps so we surface activity even if one of the gauges // didn't report a sample on a given tick. @@ -798,18 +1010,14 @@ function buildSeriesFromMetrics( for (const t of [...allTimes].toSorted((a, b) => a - b)) { const running = runByT.get(t) ?? 0; const waiting = waitByT.get(t) ?? 0; - queueDepth.push({ t: tOf(t), running, waiting, total: running + waiting }); + queueDepth.push({ t, running, waiting, total: running + waiting }); } - // Throughput: sum the counter `rate` (already per-second) across engines. - // Takes a fallback chain so vllm:* and sglang:* both work. - const counterRate = (...names: string[]): TimeSeriesPoint[] => { - const s = pickSeries(...names); - return sortedEntries(aggregateByStart(s, 'rate', 'sum')).map(([t, v]) => ({ - t: tOf(t), - value: v, - })); - }; + // Throughput: sum the counter `rate` (already per-second) across engines, + // on the canonical grid. Takes a fallback chain so vllm:* and sglang:* both + // work. + const counterRate = (...names: string[]): TimeSeriesPoint[] => + summedSeries(pickSeries(...names), tOf, 'rate', tickS); const prefillTps = counterRate('vllm:prompt_tokens', 'sglang:prompt_tokens'); const decodeTps = counterRate('vllm:generation_tokens', 'sglang:generation_tokens'); // Tokens served from prefix cache per scrape. Lets the frontend derive @@ -819,21 +1027,19 @@ function buildSeriesFromMetrics( // SGLang hicache: host-pool KV cache utilization as used/total per // timeslice. Both metrics are gauges in absolute tokens. Total stays // constant (it's the pool size), used fluctuates. - const hostUsedByT = aggregateByStart( - metrics['sglang:hicache_host_used_tokens']?.series, - 'avg', - 'sum', - ); - const hostTotalByT = aggregateByStart( - metrics['sglang:hicache_host_total_tokens']?.series, - 'avg', - 'sum', + const hostTotalByT = byT( + summedSeries(metrics['sglang:hicache_host_total_tokens']?.series, tOf, 'avg', tickS), ); const hostKvCacheUsage: TimeSeriesPoint[] = []; - for (const [t, used] of sortedEntries(hostUsedByT)) { + for (const { t, value: used } of summedSeries( + metrics['sglang:hicache_host_used_tokens']?.series, + tOf, + 'avg', + tickS, + )) { const total = hostTotalByT.get(t); if (total !== undefined && total > 0) { - hostKvCacheUsage.push({ t: tOf(t), value: used / total }); + hostKvCacheUsage.push({ t, value: used / total }); } } @@ -844,22 +1050,18 @@ function buildSeriesFromMetrics( // sglang: sglang:realtime_tokens uses a `mode` label with values // {prefill_cache, prefill_compute, decode}. Filter to prefill_* // since decode isn't prompt-token volume. - const promptBySrcByT = new Map>(); - // Sum a series' per-scrape rates into the bucket for `label`. The bucket is - // created even when the series has no valid timeslices — the SGLang fallback - // below is gated on `promptBySrcByT.size === 0`, so an empty vllm breakdown - // must still suppress it. + // Collect the raw series per source label first, then sum each source onto + // the canonical grid — same reason as the other summed metrics: a source's + // per-worker series sit on their own scrape offsets, so adding them by exact + // instant would emit one partial-sum point per worker per tick. + const promptBySrc = new Map(); + // The bucket is created even when the series has no valid timeslices — the + // SGLang fallback below is gated on `promptBySrc.size === 0`, so an empty + // vllm breakdown must still suppress it. const addSeriesRates = (label: string, series: RawSeries): void => { - let byT = promptBySrcByT.get(label); - if (!byT) { - byT = new Map(); - promptBySrcByT.set(label, byT); - } - for (const ts of series.timeslices ?? []) { - if (typeof ts.rate === 'number' && typeof ts.start_ns === 'number') { - byT.set(ts.start_ns, (byT.get(ts.start_ns) ?? 0) + ts.rate); - } - } + const existing = promptBySrc.get(label); + if (existing) existing.push(series); + else promptBySrc.set(label, [series]); }; for (const series of metrics['vllm:prompt_tokens_by_source']?.series ?? []) { const labels = series.labels ?? {}; @@ -872,7 +1074,7 @@ function buildSeriesFromMetrics( // series carries a `cache_source` label ("device" = HBM, "host" = CPU // offload via hicache). Current runs have only `device`; when hicache // runs land, additional series will appear and the chart will split. - if (promptBySrcByT.size === 0) { + if (promptBySrc.size === 0) { for (const series of metrics['sglang:realtime_tokens']?.series ?? []) { const labels = series.labels ?? {}; const mode = labels['mode'] ?? 'unknown'; @@ -900,11 +1102,10 @@ function buildSeriesFromMetrics( } } const promptTokensBySource: Record = {}; - for (const [source, byT] of promptBySrcByT) { - const arr: TimeSeriesPoint[] = []; - for (const [t, v] of sortedEntries(byT)) { - if (v > 0) arr.push({ t: tOf(t), value: v }); - } + for (const [source, seriesForSource] of promptBySrc) { + // Idle ticks are dropped rather than emitted as zeros: this feeds a + // cumulative stacked area, so a zero adds nothing but payload. + const arr = summedSeries(seriesForSource, tOf, 'rate', tickS).filter((p) => p.value > 0); if (arr.length > 0) promptTokensBySource[source] = arr; } From 5b8985596752d398f9eea82598970ba92758c3c4 Mon Sep 17 00:00:00 2001 From: Cam Quilici Date: Fri, 14 Aug 2026 12:07:23 -0500 Subject: [PATCH 2/3] chore: redeploy preview against chart-series v14 backfill test DB MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Empty commit to pick up the branch-scoped DATABASE_READONLY_URL / DATABASE_WRITE_URL / CACHE_NAMESPACE that point this preview at an expiring Neon branch of prod, for validating the v14 backfill before merge. 中文：空提交，用于让预览部署重新读取分支级环境变量（指向 prod 的临时 Neon 分支）， 以便在合并前验证 v14 回填。 Co-Authored-By: Claude Opus 5 (1M context) From 7be590bf75a2a8dcfd78bd4f52a557c6a4e6ad91 Mon Sep 17 00:00:00 2001 From: Cam Quilici Date: Sun, 16 Aug 2026 14:38:56 -0500 Subject: [PATCH 3/3] chore: redeploy preview against 7-day v14 backfill branch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Empty commit so the preview picks up the branch-scoped DATABASE_READONLY_URL / DATABASE_WRITE_URL / CACHE_NAMESPACE now pointing at Neon branch chart-series-v14-preview (expires 2026-08-23). 中文：空提交，让预览部署重新读取指向 Neon 分支 chart-series-v14-preview 的 分支级环境变量（该分支 2026-08-23 到期）。 Co-Authored-By: Claude Opus 5 (1M context)