From c5999778ab5c30b8e3a33b7c622e918f31fea831 Mon Sep 17 00:00:00 2001 From: Cam Quilici Date: Sun, 9 Aug 2026 16:08:54 -0500 Subject: [PATCH 1/3] fix(agentic): give KV cache series a real per-engine identity MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The KV cache utilization chart drew one line per raw metric series rather than one per engine, so DEP points rendered a tangle of duplicate lines with an unreadable legend — the 8-rank DP run at /inference/agentic/439261 drew 32 lines labelled "DP 0 DP 0 DP 0 DP 0 DP 1 ...". Three independent duplications were folded into `kvCacheUsageByEngine`: 1. v12's warmup merge concatenates each engine's warmup and profiling series, so every engine appeared at least twice. Single-engine points also began drawing a spurious two-line "per-engine" overlay. 2. vLLM run with several API-server frontends exposes the *same* engine set on every /metrics endpoint, ~176 ms apart, so 8 DP ranks became 16. 3. Tensor-/pipeline-/expert-parallel ranks each report the one KV pool they share, so a TP8 SGLang worker looked like 8 engines holding identical values. The same fragmentation corrupted the cluster-average line, which grouped on an exact `start_ns`: each tick averaged only the engines that happened to share that nanosecond. On disaggregated runs that alternates between "prefill only" and "decode only" — a full-scale sawtooth, not an average. Series are now grouped by their Prometheus label set (endpoint_url is transport, not identity; intra-engine shard ranks are excluded), and the average is a true mean across logical engines on the union of their scrape instants, each engine holding its last sample only inside its own observed window. CHART_SERIES_VERSION 12 -> 13; run db:backfill-chart-series. Measured on real blobs (roughness = mean |tick delta| / stddev; a clean 1 Hz single-grid row sits near 0.18): point 439261 vllm/mi355x/dsv4 DEP8 c=64 32 -> 8 engines 0.59 -> 0.05 point 439292 dynamo-sglang/h200/glm5.2 64 -> 18 engines 1.00 -> 0.02 point 439312 dynamo-sglang/gb300/dsv4 10 -> 5 engines 1.29 -> 0.07 point 436497 vllm/b200/dsv4 c=8 16 -> 8 engines unchanged point 437312 dynamo-vllm/b200/kimik3 2 -> 0 engines unchanged 229 of the 436 stored rows carrying a per-engine breakdown were affected. Single-endpoint rows come out byte-identical; disaggregated rows shift level because the mean now weights every engine once instead of over-weighting whichever subset shared a timestamp. Sum-combined metrics (queue depth, prefill/decode TPS) still group on exact start_ns and are untouched here: their point count feeds running-sum cumulative charts, so resampling them needs its own change. Co-Authored-By: Claude Opus 5 (1M context) --- docs/data-pipeline.md | 10 + .../e2e/agentic-point-time-series.cy.ts | 60 ++++ .../agentic-point/server-metric-cards.tsx | 17 +- .../db/src/etl/compute-chart-series.test.ts | 245 +++++++++++++++ packages/db/src/etl/compute-chart-series.ts | 295 ++++++++++++++++-- 5 files changed, 597 insertions(+), 30 deletions(-) diff --git a/docs/data-pipeline.md b/docs/data-pipeline.md index 9d1ce61cc..cc50d2996 100644 --- a/docs/data-pipeline.md +++ b/docs/data-pipeline.md @@ -89,6 +89,16 @@ AIPerf defines the `server_metrics_export.json` envelope, but labels such as wor Adapters are selected from the benchmark's canonical framework, and per-worker series are only emitted for disaggregated configs with a recognized adapter. Unknown orchestrators and non-disaggregated configs retain their aggregate-only series; roles are never guessed from ports or metric names. The frontend only consumes the canonical source identity and never interprets orchestrator-native labels. +### Logical Engines vs Raw Series + +A raw series in the blob is one `(scrape endpoint × phase block × label set)` tuple, which is **not** the same as one engine. The KV-cache chart needs one entry per _logical engine_ — one KV pool — so `compute-chart-series.ts` groups series by their Prometheus label set (`seriesIdentityKey`) rather than emitting one entry per raw series. Three kinds of duplication collapse there: + +- **Phase blocks.** The warmup and profiling blocks each carry their own series for the same engine. They cover disjoint time ranges, so keying by scrape instant unions them into one continuous line. +- **Mirrored API-server frontends.** vLLM run with several API servers exposes the _same_ engine set on every `/metrics` endpoint, a few hundred ms apart. Identity ignores `endpoint_url` (Prometheus treats the label set as the series), and the endpoint with the most complete coverage wins — merging the mirrors instead would interleave near-duplicate samples and halve the effective span of the frontend's fixed-width rolling average. +- **Intra-engine shard ranks.** `tp_rank` / `pp_rank` / `ep_rank` / `moe_ep_rank` shard one pool and all report the same utilization, so they are excluded from the identity and averaged. `engine` / `engine_idx` / `dp_rank` are _not_ excluded — those do name distinct pools. + +The cluster average is then a mean across those logical engines on the union of their scrape instants, with each engine holding its last sample until its next one and contributing only inside its own observed window. Grouping on an exact `start_ns` instead would average whichever engines happened to share that nanosecond — on a disaggregated run that alternates between "prefill only" and "decode only" and reads as a full-scale sawtooth. + ### 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/app/cypress/e2e/agentic-point-time-series.cy.ts b/packages/app/cypress/e2e/agentic-point-time-series.cy.ts index ebb7bba3c..00794b6c8 100644 --- a/packages/app/cypress/e2e/agentic-point-time-series.cy.ts +++ b/packages/app/cypress/e2e/agentic-point-time-series.cy.ts @@ -353,3 +353,63 @@ describe('Agentic point orchestrator metric sources', () => { cy.get('[data-testid="throughput-series-decode"]').should('have.attr', 'aria-pressed', 'false'); }); }); + +const engineSeries = (engineLabel: string, value: number) => ({ + engineLabel, + points: [ + { t: 0, value }, + { t: 1, value: value + 0.05 }, + ], +}); + +describe('Agentic point per-engine KV overlay', () => { + beforeEach(() => { + cy.intercept('GET', '/api/v1/trace-histograms*', { body: {} }); + cy.intercept('GET', '/api/v1/benchmark-siblings*', { statusCode: 404 }); + cy.intercept('GET', '/api/v1/request-timeline*', { statusCode: 404 }); + cy.intercept('GET', '/api/v1/trace-server-metrics*', { + body: { + meta: pointMeta, + startNs: 0, + endNs: 2_000_000_000, + durationS: 2, + timeslicesCount: 2, + kvCacheUsage: [ + { t: 0, value: 0.3 }, + { t: 1, value: 0.35 }, + ], + prefixCacheHitRate: [], + queueDepth: [], + promptTokensBySource: {}, + prefillTps: [], + decodeTps: [], + prefixCacheHitsTps: [], + hostKvCacheUsage: [], + // Bare DP ranks plus a role-qualified engine, as the ETL emits for a + // disaggregated run where the decode worker reports no rank label. + kvCacheUsageByEngine: [ + engineSeries('0', 0.2), + engineSeries('1', 0.4), + engineSeries('decode', 0.6), + ], + metricSources: [], + }, + }); + cy.visit('/inference/agentic/206885', { onBeforeLoad: unlockAgenticGate }); + }); + + it('draws one legend entry per engine and leaves named engines unprefixed', () => { + cy.contains('svg', 'KV cache (%)') + .first() + .within(() => { + cy.contains('text', 'DP 0').should('be.visible'); + cy.contains('text', 'DP 1').should('be.visible'); + // Already self-describing — must NOT come out as "DP decode". + cy.contains('text', 'decode').should('be.visible'); + cy.contains('text', 'DP decode').should('not.exist'); + cy.contains('text', 'Avg').should('be.visible'); + // One line per engine plus the average, and no duplicate chips. + cy.get('path[fill="none"]').should('have.length', 4); + }); + }); +}); diff --git a/packages/app/src/components/inference/agentic-point/server-metric-cards.tsx b/packages/app/src/components/inference/agentic-point/server-metric-cards.tsx index 3cc023de4..6d92f1501 100644 --- a/packages/app/src/components/inference/agentic-point/server-metric-cards.tsx +++ b/packages/app/src/components/inference/agentic-point/server-metric-cards.tsx @@ -62,6 +62,15 @@ const DP_RANK_PALETTE = [ '#eab308', ]; +/** + * Bare DP ranks read as "DP 3"; disaggregated runs get role- or worker- + * qualified labels from the ETL ("decode", "prefill 0", "0 (a01a)") that + * already name themselves, so those pass through untouched. + */ +function engineSeriesName(engineLabel: string): string { + return /^\d+$/u.test(engineLabel) ? `DP ${engineLabel}` : engineLabel; +} + export function KvCacheUtilizationCard({ sliced }: { sliced: SlicedServerSeries }) { return ( e.points.length > 0, + ); const hasPerEngine = perEngine.length > 1; // Render order matters: per-engine first → average drawn on top. const series = [ ...(hasPerEngine ? perEngine.map((e, i) => ({ - name: `DP ${e.engineLabel}`, + name: engineSeriesName(e.engineLabel), data: rollingAverage(e.points, 50), color: DP_RANK_PALETTE[i % DP_RANK_PALETTE.length]!, // Thin + translucent so the Avg line on top reads as diff --git a/packages/db/src/etl/compute-chart-series.test.ts b/packages/db/src/etl/compute-chart-series.test.ts index 749241713..4d10dbc24 100644 --- a/packages/db/src/etl/compute-chart-series.test.ts +++ b/packages/db/src/etl/compute-chart-series.test.ts @@ -119,6 +119,33 @@ function buildDynamoSeries( }; } +/** A kv_cache_usage_perc series for one engine as seen from one endpoint. */ +function kvSeriesFor( + endpoint_url: string, + labels: Record, + samples: [startNs: number, avg: number][], +) { + return { + endpoint_url, + labels, + timeslices: samples.map(([start_ns, avg]) => ({ start_ns, end_ns: start_ns + 1e9, avg })), + }; +} + +/** Gzip a blob carrying only kv_cache_usage_perc, optionally in both phases. */ +function kvBlob(profiling: unknown[], warmup: unknown[] = []) { + return gzipSync( + Buffer.from( + JSON.stringify({ + metrics: { 'vllm:kv_cache_usage_perc': { series: profiling } }, + ...(warmup.length > 0 + ? { warmup_metrics: { 'vllm:kv_cache_usage_perc': { series: warmup } } } + : {}), + }), + ), + ); +} + describe('computeChartSeries', () => { it('returns null when the blob is null', async () => { expect(await computeChartSeries(null)).toBeNull(); @@ -316,6 +343,224 @@ describe('computeChartSeries', () => { expect(nonDisagg?.metricSources).toEqual([]); }); + // ── Per-engine identity (v13) ───────────────────────────────────────── + // + // The blob stores one series per (scrape endpoint x phase block x label + // set), which is not one series per engine. These cover the three ways + // that mismatch used to inflate `kvCacheUsageByEngine` and fragment the + // cluster average. + + it('collapses API-server frontends that mirror the same engines', async () => { + // vLLM with two API servers exposes every DP rank on BOTH /metrics + // endpoints, scraped a fraction of a second apart. That is 4 series for + // 2 engines, not 4 engines. + const cs = await computeChartSeries( + kvBlob([ + kvSeriesFor('http://localhost:8895/metrics', { engine: '0', model_name: 'm' }, [ + [0, 0.2], + [1e9, 0.3], + ]), + kvSeriesFor('http://localhost:8895/metrics', { engine: '1', model_name: 'm' }, [ + [0, 0.6], + [1e9, 0.7], + ]), + kvSeriesFor('http://localhost:8896/metrics', { engine: '0', model_name: 'm' }, [ + [0.176e9, 0.2], + [1.176e9, 0.3], + ]), + kvSeriesFor('http://localhost:8896/metrics', { engine: '1', model_name: 'm' }, [ + [0.176e9, 0.6], + [1.176e9, 0.7], + ]), + ]), + ); + expect(cs?.kvCacheUsageByEngine.map((e) => e.engineLabel)).toEqual(['0', '1']); + // The mean is over the two real engines, once per scrape tick — not four + // ticks averaging one endpoint's subset each. + expect(cs?.kvCacheUsage).toEqual([ + { t: 0, value: 0.4 }, + { t: 1, value: 0.5 }, + ]); + }); + + it('joins an engine warmup and profiling scrapes into one line', async () => { + const cs = await computeChartSeries( + kvBlob( + [ + kvSeriesFor('http://localhost:8000/metrics', { engine: '0' }, [[10e9, 0.8]]), + kvSeriesFor('http://localhost:8000/metrics', { engine: '1' }, [[10e9, 0.4]]), + ], + [ + kvSeriesFor('http://localhost:8000/metrics', { engine: '0' }, [[0, 0.2]]), + kvSeriesFor('http://localhost:8000/metrics', { engine: '1' }, [[0, 0.1]]), + ], + ), + ); + expect(cs?.kvCacheUsageByEngine).toEqual([ + { + engineLabel: '0', + points: [ + { t: 0, value: 0.2 }, + { t: 10, value: 0.8 }, + ], + }, + { + engineLabel: '1', + points: [ + { t: 0, value: 0.1 }, + { t: 10, value: 0.4 }, + ], + }, + ]); + }); + + it('suppresses the per-engine overlay for a single-engine deployment', async () => { + // Warmup + profiling is two series but still one engine, so there is + // nothing for a per-rank overlay to compare. + const cs = await computeChartSeries( + kvBlob( + [kvSeriesFor('http://localhost:8000/metrics', { engine: '0' }, [[10e9, 0.8]])], + [kvSeriesFor('http://localhost:8000/metrics', { engine: '0' }, [[0, 0.2]])], + ), + ); + expect(cs?.kvCacheUsageByEngine).toEqual([]); + expect(cs?.kvCacheUsage).toEqual([ + { t: 0, value: 0.2 }, + { t: 10, value: 0.8 }, + ]); + }); + + it('keeps disaggregated workers apart even when their ranks collide', async () => { + // Prefill rank 0 and decode rank 0 are different engines; `worker_id` + // is what says so. + const cs = await computeChartSeries( + kvBlob([ + kvSeriesFor( + 'http://10.0.0.1:7500/metrics', + { dp_rank: '0', engine_type: 'prefill', worker_id: 'aaaaaaaawork0001' }, + [[0, 0.02]], + ), + kvSeriesFor( + 'http://10.0.0.2:7502/metrics', + { dp_rank: '0', engine_type: 'decode', worker_id: 'bbbbbbbbwork0002' }, + [[0, 0.5]], + ), + ]), + ); + expect(cs?.kvCacheUsageByEngine.map((e) => e.engineLabel)).toEqual(['prefill 0', 'decode 0']); + expect(cs?.kvCacheUsage).toEqual([{ t: 0, value: 0.26 }]); + }); + + it('collapses tensor-parallel shard ranks that share one KV pool', async () => { + // A TP4 SGLang worker reports kv_cache_usage_perc once per tp_rank, but + // the four ranks shard a single pool and track each other to ~4dp. They + // are one engine, not four. + const shards = [0, 1, 2, 3].map((tp) => + kvSeriesFor( + 'http://10.0.0.1:7500/metrics', + { + tp_rank: String(tp), + pp_rank: '0', + moe_ep_rank: String(tp), + engine_type: 'prefill', + worker_id: 'worker-aaaa', + }, + [ + [0, 0.4 + tp * 0.0001], + [1e9, 0.6 + tp * 0.0001], + ], + ), + ); + const cs = await computeChartSeries(kvBlob(shards)); + // One pool -> one engine, so the per-rank overlay stays off entirely. + expect(cs?.kvCacheUsageByEngine).toEqual([]); + expect(cs?.kvCacheUsage).toEqual([ + { t: 0, value: 0.40015 }, + { t: 1, value: 0.60015 }, + ]); + }); + + it('keeps DP ranks apart even when shard ranks co-vary with them', async () => { + // Same worker, but now dp_rank names a real per-rank pool and tp_rank + // happens to move with it. Dropping shard labels must not fuse these. + const cs = await computeChartSeries( + kvBlob( + [0, 1].map((dp) => + kvSeriesFor( + 'http://10.0.0.1:7500/metrics', + { + dp_rank: String(dp), + tp_rank: String(dp), + engine_type: 'prefill', + worker_id: 'worker-aaaa', + }, + [[0, dp === 0 ? 0.2 : 0.8]], + ), + ), + ), + ); + expect(cs?.kvCacheUsageByEngine.map((e) => e.engineLabel)).toEqual(['prefill 0', 'prefill 1']); + expect(cs?.kvCacheUsage).toEqual([{ t: 0, value: 0.5 }]); + }); + + it('qualifies engines whose display label would otherwise collide', async () => { + // Two decode workers that each number their ranks from 0. + const cs = await computeChartSeries( + kvBlob([ + kvSeriesFor( + 'http://10.0.0.1:7502/metrics', + { dp_rank: '0', engine_type: 'decode', worker_id: 'worker-a01a' }, + [[0, 0.4]], + ), + kvSeriesFor( + 'http://10.0.0.2:7503/metrics', + { dp_rank: '0', engine_type: 'decode', worker_id: 'worker-b01b' }, + [[0, 0.6]], + ), + ]), + ); + expect(cs?.kvCacheUsageByEngine.map((e) => e.engineLabel)).toEqual([ + 'decode 0 (a01a)', + 'decode 0 (b01b)', + ]); + }); + + it('averages engines on unaligned scrape grids without sawtoothing', async () => { + // A prefill worker and a decode worker on their own sub-second grids. + // Grouping on an exact start_ns made every tick "prefill only" (0.0) or + // "decode only" (1.0), i.e. a full-scale sawtooth where the real cluster + // mean is a flat 0.5. + const prefill: [number, number][] = []; + const decode: [number, number][] = []; + for (let i = 0; i < 10; i++) { + prefill.push([i * 1e9, 0]); + decode.push([i * 1e9 + 0.5e9, 1]); + } + const cs = await computeChartSeries( + kvBlob([ + kvSeriesFor( + 'http://10.0.0.1:7500/metrics', + { worker_id: 'p', engine_type: 'prefill' }, + prefill, + ), + kvSeriesFor( + 'http://10.0.0.2:7502/metrics', + { worker_id: 'd', engine_type: 'decode' }, + decode, + ), + ]), + ); + // The edge ticks fall outside one engine's observed window — t=0 predates + // decode's first scrape, t=9.5 postdates prefill's last — so they report + // the one engine that was actually running rather than a carried-forward + // stale value. Every tick in between averages both and stays flat. + expect(cs?.kvCacheUsage.at(0)).toEqual({ t: 0, value: 0 }); + expect(cs?.kvCacheUsage.at(-1)).toEqual({ t: 9.5, value: 1 }); + const both = cs!.kvCacheUsage.slice(1, -1); + expect(both).toHaveLength(18); + expect(both.every((p) => p.value === 0.5)).toBe(true); + }); + it('does not interpret Dynamo-native labels without selecting the Dynamo adapter', async () => { const json = JSON.stringify({ metrics: { diff --git a/packages/db/src/etl/compute-chart-series.ts b/packages/db/src/etl/compute-chart-series.ts index ba2661296..203c87ed4 100644 --- a/packages/db/src/etl/compute-chart-series.ts +++ b/packages/db/src/etl/compute-chart-series.ts @@ -65,8 +65,26 @@ import { * warmup block are unaffected. (v11 was a short-lived, since-reverted attempt to * carry kvCachePoolTokens in chart_series; that value now lives in * benchmark_results.metrics, derived from the server log — unrelated to this.) + * + * v13: give the KV-cache series a real per-engine identity instead of "one + * entry per raw series". Three independent sources of duplication were + * inflating `kvCacheUsageByEngine` (up to 8x — 64 lines for a run with 18 + * real engines) and fragmenting the cluster-average `kvCacheUsage`: + * 1. v12's warmup merge concatenates each engine's warmup and profiling + * series, so every engine appeared at least twice — and single-engine + * deployments started drawing a spurious two-line "per-engine" overlay. + * 2. vLLM run with multiple API-server frontends exposes the *same* engine + * set on every `/metrics` endpoint, so an 8-rank DP deployment scraped + * from two frontends yielded 16 series for 8 engines. + * 3. Tensor-/pipeline-/expert-parallel ranks each report the one KV pool + * they share, so a TP8 worker looked like 8 engines holding identical + * values. + * Series are now grouped by their Prometheus label set (see + * `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`. */ -export const CHART_SERIES_VERSION = 12; +export const CHART_SERIES_VERSION = 13; export interface TimeSeriesPoint { /** Seconds from benchmark start. */ @@ -303,6 +321,245 @@ function sortedEntries(m: Map): [number, number][] { return [...m.entries()].toSorted((a, b) => a[0] - b[0]); } +// ── Per-engine identity (v13) ─────────────────────────────────────────── +// +// A "logical engine" is one KV-cache pool: one DP rank of one worker. The blob +// stores one `RawSeries` per (scrape endpoint × phase block × label set), which +// is NOT the same thing — see the v13 note at the top of the file. + +/** + * Ranks that shard ONE engine rather than naming a separate one. A KV cache is + * allocated per engine and shared by its tensor-, pipeline- and expert-parallel + * ranks, so every such rank reports the same pool: on a TP8 SGLang prefill + * worker all eight `tp_rank` series track each other to four decimal places. + * Treating them as separate engines drew eight identical lines for one pool. + * + * `engine` / `engine_idx` / `dp_rank` are deliberately NOT here — those DO name + * distinct pools (one per DP rank / engine core). + */ +const INTRA_ENGINE_SHARD_LABELS = new Set(['tp_rank', 'pp_rank', 'ep_rank', 'moe_ep_rank']); + +/** + * Identity of a metric series, following Prometheus semantics: the label set + * IS the series, and the scrape endpoint is transport rather than identity. + * Two `/metrics` endpoints exposing `{engine="3", model_name="X"}` are two + * views of one engine — which is exactly what vLLM does when it runs several + * API-server frontends over one DP engine group. + * + * Deployments whose endpoints really are distinct engines say so in the + * labels: Dynamo tags every series with `worker_id` (plus `dynamo_component` + * / `engine_type`), so prefill worker rank 0 and decode worker rank 0 keep + * separate identities here. + * + * Shard ranks are excluded (see `INTRA_ENGINE_SHARD_LABELS`). Series left with + * no labels at all fall back to the endpoint, which keeps label-less workers + * apart rather than silently fusing them. + */ +function seriesIdentityKey(s: RawSeries): string { + const labels = s.labels ?? {}; + const names = Object.keys(labels) + .filter((name) => !INTRA_ENGINE_SHARD_LABELS.has(name)) + .toSorted(); + if (names.length === 0) return `@${s.endpoint_url ?? ''}`; + // Join on control characters so a name or value containing '=' or ',' + // cannot forge another label set's key. + return names.map((name) => `${name}\u0001${labels[name]}`).join('\u0002'); +} + +/** Dynamo/SGLang name their roles differently; normalize for display. */ +const ENGINE_ROLE_BY_NATIVE_LABEL: Record = { + prefill: 'prefill', + decode: 'decode', + backend: 'decode', +}; + +/** DP-rank-ish label under any of the names the frameworks emit. */ +function engineRankLabel(labels: Record): string | null { + return labels['engine'] ?? labels['engine_idx'] ?? labels['dp_rank'] ?? null; +} + +function engineRoleLabel(labels: Record): string | null { + const native = labels['engine_type'] ?? labels['dynamo_component']; + return native ? (ENGINE_ROLE_BY_NATIVE_LABEL[native] ?? null) : null; +} + +/** + * Short, human-readable tiebreaker for engines that would otherwise share a + * display label (e.g. two decode workers that each number their ranks 0..7). + */ +function engineDiscriminator(labels: Record, endpointUrl: string): string | null { + const worker = labels['worker_id']; + if (worker) return worker.length > 4 ? worker.slice(-4) : worker; + // Fall back to the endpoint's host:port, which is what distinguishes + // workers when the orchestrator doesn't emit a worker id. + const hostPort = /^\w+:\/\/(?[^/]+)/u.exec(endpointUrl)?.groups?.['hostPort']; + return hostPort ?? (endpointUrl || null); +} + +interface LogicalEngine { + engineLabel: string; + points: TimeSeriesPoint[]; +} + +interface EngineGroup { + labels: Record; + /** Per endpoint, the engine's samples keyed by scrape instant. */ + byEndpoint: Map>; +} + +/** + * Collapse a gauge's raw series into one entry per logical engine. + * + * The three kinds of duplication need three different treatments: + * - Same endpoint, different phase blocks (v12's warmup merge): disjoint + * time ranges, so keying by scrape instant simply unions them into one + * continuous series. + * - Same endpoint, same instant: intra-engine shard ranks reporting the + * one pool they share, so they collapse to their mean (identical in + * practice, to four decimal places). + * - Different endpoints (mirrored API-server frontends): overlapping time + * ranges carrying the same measurement a few hundred ms apart. Merging + * them would interleave near-duplicate samples and silently halve the + * span of the frontend's fixed-width rolling average, so we keep the + * endpoint with the most complete coverage and drop the rest. + */ +function resolveLogicalEngines( + series: readonly RawSeries[] | undefined, + tOf: (ns: number) => number, +): LogicalEngine[] { + const groups = new Map(); + for (const s of series ?? []) { + const key = seriesIdentityKey(s); + let group = groups.get(key); + if (!group) { + group = { labels: s.labels ?? {}, byEndpoint: new Map() }; + groups.set(key, group); + } + const endpoint = s.endpoint_url ?? ''; + let scrapes = group.byEndpoint.get(endpoint); + if (!scrapes) { + scrapes = new Map(); + group.byEndpoint.set(endpoint, scrapes); + } + for (const ts of s.timeslices ?? []) { + if (typeof ts.start_ns !== 'number' || typeof ts.avg !== 'number') continue; + if (!Number.isFinite(ts.avg)) continue; + const at = scrapes.get(ts.start_ns); + if (at) { + at.sum += ts.avg; + at.count++; + } else { + scrapes.set(ts.start_ns, { sum: ts.avg, count: 1 }); + } + } + } + + // Insertion order = first appearance in the blob, which is the engine order + // the exporter emitted; `rank` sorting below refines it when ranks exist. + const resolved: { label: string; discriminator: string | null; points: TimeSeriesPoint[] }[] = []; + for (const group of groups.values()) { + // Deterministic pick: most scrape instants wins, endpoint URL breaks ties. + let chosenEndpoint = ''; + let chosen: Map | null = null; + for (const [endpoint, scrapes] of [...group.byEndpoint].toSorted((a, b) => + a[0].localeCompare(b[0]), + )) { + if (scrapes.size > (chosen?.size ?? 0)) { + chosen = scrapes; + chosenEndpoint = endpoint; + } + } + if (!chosen || chosen.size === 0) continue; + const rank = engineRankLabel(group.labels); + const role = engineRoleLabel(group.labels); + const discriminator = engineDiscriminator(group.labels, chosenEndpoint); + const named = role ? (rank === null ? role : `${role} ${rank}`) : rank; + resolved.push({ + // With no rank- or role-like label the worker/endpoint is the only thing + // that names this engine, so lead with it instead of a bare index. + label: named ?? discriminator ?? '', + discriminator, + points: [...chosen.entries()] + .toSorted((a, b) => a[0] - b[0]) + .map(([startNs, { sum, count }]) => ({ t: tOf(startNs), value: sum / count })), + }); + } + + // Plain numeric ranks render in 0..N order; role-qualified and endpoint-named + // engines keep the order the exporter emitted them in. + const ordered = resolved + .map((engine, idx) => { + const numeric = Number(engine.label); + return { ...engine, idx, sortKey: engine.label && Number.isFinite(numeric) ? numeric : idx }; + }) + .toSorted((a, b) => a.sortKey - b.sortKey || a.idx - b.idx) + .map((engine, idx) => ({ ...engine, base: engine.label || `#${idx}` })); + + // Qualify collisions (e.g. two decode workers that each number their ranks + // 0..7) so every legend entry names exactly one engine. + const baseCounts = new Map(); + for (const engine of ordered) baseCounts.set(engine.base, (baseCounts.get(engine.base) ?? 0) + 1); + const used = new Set(); + return ordered.map((engine, idx) => { + let engineLabel = engine.base; + if ((baseCounts.get(engine.base) ?? 0) > 1) { + const qualifier = + engine.discriminator && engine.discriminator !== engine.base ? engine.discriminator : idx; + engineLabel = `${engine.base} (${qualifier})`; + } + // Last resort, so a legend entry never stands for two lines. + while (used.has(engineLabel)) engineLabel = `${engineLabel}'`; + used.add(engineLabel); + return { engineLabel, points: engine.points }; + }); +} + +/** + * Mean utilization across logical engines on the union of their scrape times. + * + * Engines are not scraped in lockstep — different workers (and occasionally a + * single lagging rank) sit on their own sub-second grid — so grouping on an + * exact `start_ns` would average whichever subset happened to share that + * nanosecond. On a disaggregated run that means alternating between + * "prefill only" and "decode only", which reads as a full-scale sawtooth + * rather than a cluster average. + * + * Each engine therefore holds its last scrape until its next one (a gauge + * keeps its value between scrapes) and contributes only inside its own + * observed window, so an engine that starts late or stops early neither + * pulls the mean toward a stale value nor drops it to zero. + */ +function averageAcrossEngines(engines: readonly LogicalEngine[]): TimeSeriesPoint[] { + const active = engines.filter((engine) => engine.points.length > 0); + if (active.length === 0) return []; + // Single engine: its own samples already are the cluster average. + if (active.length === 1) return active[0]!.points; + + const timeline = [...new Set(active.flatMap((e) => e.points.map((p) => p.t)))].toSorted( + (a, b) => a - b, + ); + const cursors: number[] = Array.from({ length: active.length }, () => -1); + const lastT = active.map((engine) => engine.points.at(-1)!.t); + const out: TimeSeriesPoint[] = []; + for (const t of timeline) { + let sum = 0; + let n = 0; + for (const [i, engine] of active.entries()) { + const points = engine.points; + let cursor = cursors[i]!; + while (cursor + 1 < points.length && points[cursor + 1]!.t <= t) cursor++; + cursors[i] = cursor; + // Before this engine's first scrape or after its last — no value to + // carry, so it sits out of this tick's mean entirely. + if (cursor < 0 || t > lastT[i]!) continue; + sum += points[cursor]!.value; + n++; + } + if (n > 0) out.push({ t, value: sum / n }); + } + return out; +} + function buildSeriesFromMetrics( metrics: MetricsMap, context: ServerMetricsContext, @@ -347,33 +604,15 @@ function buildSeriesFromMetrics( 'vllm:gpu_cache_usage_perc', 'sglang:token_usage', ); - const kvCacheUsage: TimeSeriesPoint[] = sortedEntries( - aggregateByStart(kvSeries, 'avg', 'avg'), - ).map(([t, v]) => ({ t: tOf(t), value: v })); - // Per-engine breakdown of the same metric. We only emit it when there's - // more than one series — single-engine deployments would just duplicate - // the cluster-average line. - const kvCacheUsageByEngine: { engineLabel: string; points: TimeSeriesPoint[] }[] = []; - if (kvSeries && kvSeries.length > 1) { - // Sort by numeric engine label when present so rank 0..N renders in - // order; fall back to series-array index otherwise. - const decorated = kvSeries.map((s, idx) => { - const raw = - s.labels?.['engine'] ?? s.labels?.['engine_idx'] ?? s.labels?.['dp_rank'] ?? String(idx); - const numeric = Number(raw); - return { series: s, idx, label: raw, sortKey: Number.isFinite(numeric) ? numeric : idx }; - }); - decorated.sort((a, b) => a.sortKey - b.sortKey); - for (const { series, label } of decorated) { - const pts: TimeSeriesPoint[] = []; - for (const ts of series.timeslices ?? []) { - if (typeof ts.start_ns !== 'number' || typeof ts.avg !== 'number') continue; - if (!Number.isFinite(ts.avg)) continue; - pts.push({ t: tOf(ts.start_ns), value: ts.avg }); - } - if (pts.length > 0) kvCacheUsageByEngine.push({ engineLabel: label, points: pts }); - } - } + // One entry per logical engine (v13) — mirrored API-server frontends and the + // warmup/profiling phase split are collapsed here rather than showing up as + // extra "engines". + const engines = resolveLogicalEngines(kvSeries, tOf); + const kvCacheUsage: TimeSeriesPoint[] = averageAcrossEngines(engines); + // Per-engine breakdown of the same metric. Emitted only for genuinely + // multi-engine deployments — with one engine it would just duplicate the + // cluster-average line. + const kvCacheUsageByEngine = engines.length > 1 ? engines : []; // Prefix cache hit rate per scrape: Σhits.rate / Σqueries.rate across // engines, joined on start_ns. SGLang names: cached_tokens / prompt_tokens. From 6c4200637173f6fd7c0e11e8b6063eaac049620a Mon Sep 17 00:00:00 2001 From: Cam Quilici Date: Sun, 9 Aug 2026 19:35:39 -0500 Subject: [PATCH 2/3] fix(agentic): correct engine ordering and harden the KV identity rules Follow-up to the per-engine identity change in this PR, from three independent audits of the diff against the full production corpus. Fixes a regression the first commit introduced, proven on replay 654 (dynamo-vllm gb200, 4 decode workers, engine=0..3): v12 -> [0, 0, 1, 1, 2, 2, 3, 3] (duplicated per phase) was -> [decode 3, decode 1, decode 0, decode 2] now -> [0, 1, 2, 3] Two causes, both fixed: - The role prefix was applied even when every engine shared one role, so an aggregated deployment read "decode 0..3" for no reason. The role is now shown only when engines actually differ in role. - Sorting keyed off the COMPOSED label, so role-qualified names fell back to blob order and scrambled the ranks (and with them the palette, which is indexed by array position). Ordering is now a tuple over the identity components: role, then numeric rank, then worker, then blob order. Hardening, each with a test: - Same-label endpoints are only fused when their values agree. Plain sglang emits an identity with no distinguishing field at all ({engine_type: unified, model_name, tp/pp/moe_ep_rank all 0}), so two replicas behind a router would previously have collapsed into one with the other's samples discarded silently. All 169 sglang rows are single-endpoint today, so this is a latent hazard, not a live bug. - The surviving mirror is picked by wall-clock coverage, then sample count. On replay 815 the old count-only rule was decided by one sample out of 3865, and a dense-but-truncated mirror could have shortened the engine's whole series. - Carry-forward is capped at 5x an engine's own median scrape gap, so a reporting hole drops the engine out of the mean instead of pinning it to a stale reading. Real runs sit at 1 Hz with gaps never above ~1 s. - engineRoleLabel falls through to dynamo_component instead of stopping at a present-but-unmapped engine_type. Aggregated dynamo-sglang workers carry engine_type="unified" alongside dynamo_component="backend". - Blank label values are treated as absent, and the collision qualifier falls back to a counter rather than appending apostrophes. Frontend: the average line's name, color, stroke and scatter are driven by whether the point has multiple engines, so keying that off the phase-sliced array made the chart change identity between the Warmup and Profiling tabs. It now keys off the unsliced count, and engine colors come from the unsliced position so a rank keeps its color across phases. Ops: the trace-server-metrics route gets maxDuration=300. Every stored row is stale until the backfill drains it, and the slow path re-parses blobs up to 448 MB compressed; the platform default would cut those off mid-parse, and the blob cache only populates on success, so every visitor would re-pay. Corrections to comments and docs that the audits disproved: - Warmup and profiling blocks were documented as covering disjoint time ranges. Their first/last bounds do overlap (replay 820 by 67 s), but measurement shows only ONE profiling sample lands inside the warmup window and there are zero exact-instant collisions, so the union is still sound. Both the code comment and the doc now say that precisely. - "TP ranks track each other to four decimal places" overstated it: they agree exactly at 99.7-99.8% of instants and their whole-run means agree to within 0.25%, with rare single-scrape transients. - The kvCacheUsageByEngine type doc still described the v12 contract. Measured after these changes (roughness = mean |tick delta| / stddev): replay 654 dynamo-vllm/gb200/kimik3 8 -> 4 engines 0.734 -> 0.009 replay 839 dynamo-sglang/h200/glm5.2 64 -> 18 engines 1.003 -> 0.023 replay 813 vllm/mi355x/dsv4 DEP8 8 engines 0.048 (unchanged) Co-Authored-By: Claude Opus 5 (1M context) --- docs/data-pipeline.md | 8 +- .../app/api/v1/trace-server-metrics/route.ts | 9 + .../agentic-point/server-metric-cards.tsx | 38 ++- .../db/src/etl/compute-chart-series.test.ts | 131 +++++++- packages/db/src/etl/compute-chart-series.ts | 296 +++++++++++++----- 5 files changed, 382 insertions(+), 100 deletions(-) diff --git a/docs/data-pipeline.md b/docs/data-pipeline.md index cc50d2996..24fce8097 100644 --- a/docs/data-pipeline.md +++ b/docs/data-pipeline.md @@ -93,11 +93,13 @@ Adapters are selected from the benchmark's canonical framework, and per-worker s A raw series in the blob is one `(scrape endpoint × phase block × label set)` tuple, which is **not** the same as one engine. The KV-cache chart needs one entry per _logical engine_ — one KV pool — so `compute-chart-series.ts` groups series by their Prometheus label set (`seriesIdentityKey`) rather than emitting one entry per raw series. Three kinds of duplication collapse there: -- **Phase blocks.** The warmup and profiling blocks each carry their own series for the same engine. They cover disjoint time ranges, so keying by scrape instant unions them into one continuous line. -- **Mirrored API-server frontends.** vLLM run with several API servers exposes the _same_ engine set on every `/metrics` endpoint, a few hundred ms apart. Identity ignores `endpoint_url` (Prometheus treats the label set as the series), and the endpoint with the most complete coverage wins — merging the mirrors instead would interleave near-duplicate samples and halve the effective span of the frontend's fixed-width rolling average. +- **Phase blocks.** The warmup and profiling blocks each carry their own series for the same engine, so keying by scrape instant unions them into one continuous line. The two blocks' first/last bounds can look overlapping — the profiling series often emits a single boundary sample and then gaps until warmup ends — but they never share a scrape instant, so the union neither drops nor double-counts a sample. +- **Mirrored API-server frontends.** vLLM run with several API servers exposes the _same_ engine set on every `/metrics` endpoint, a few hundred ms apart. Identity ignores `endpoint_url` (Prometheus treats the label set as the series), and the endpoint covering the most wall-clock wins — merging the mirrors instead would interleave near-duplicate samples and halve the effective span of the frontend's fixed-width rolling average. Endpoints are only fused when their values actually agree; same-label endpoints whose readings diverge are independent replicas behind a router, and are kept as separate engines so none is silently discarded. - **Intra-engine shard ranks.** `tp_rank` / `pp_rank` / `ep_rank` / `moe_ep_rank` shard one pool and all report the same utilization, so they are excluded from the identity and averaged. `engine` / `engine_idx` / `dp_rank` are _not_ excluded — those do name distinct pools. -The cluster average is then a mean across those logical engines on the union of their scrape instants, with each engine holding its last sample until its next one and contributing only inside its own observed window. Grouping on an exact `start_ns` instead would average whichever engines happened to share that nanosecond — on a disaggregated run that alternates between "prefill only" and "decode only" and reads as a full-scale sawtooth. +The cluster average is then a mean across those logical engines on the union of their scrape instants, with each engine holding its last sample until its next one and contributing only inside its own observed window (and only while that sample is fresher than 5× the engine's own median scrape gap, so a reporting hole drops the engine out of the mean rather than pinning it to a stale reading). Grouping on an exact `start_ns` instead would average whichever engines happened to share that nanosecond — on a disaggregated run that alternates between "prefill only" and "decode only" and reads as a full-scale sawtooth. + +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`. ### Agentic Dataset Provenance diff --git a/packages/app/src/app/api/v1/trace-server-metrics/route.ts b/packages/app/src/app/api/v1/trace-server-metrics/route.ts index 1334b580e..a2adf3d54 100644 --- a/packages/app/src/app/api/v1/trace-server-metrics/route.ts +++ b/packages/app/src/app/api/v1/trace-server-metrics/route.ts @@ -12,6 +12,15 @@ import { idQueryRoute } from '../id-routes'; export const dynamic = 'force-dynamic'; +// The slow path re-parses `server_metrics_json_gz`, which reaches 448 MB +// compressed (several GB of JSON) on the largest agentic points and takes +// minutes to stream. That only happens for rows whose stored `chart_series` +// predates the current version — i.e. after a version bump, until +// `db:backfill-chart-series` drains them — but the platform default would +// cut those requests off mid-parse and the blob cache only populates on +// success, so every visitor would re-pay. Ask for the maximum instead. +export const maxDuration = 300; + // Key derived from TRACE_SERVER_METRICS_VERSION (governs chart_series plus // the separately queried point-metadata payload). // The blob cache is write-once with no post-backfill purge, so the diff --git a/packages/app/src/components/inference/agentic-point/server-metric-cards.tsx b/packages/app/src/components/inference/agentic-point/server-metric-cards.tsx index 6d92f1501..6442f15be 100644 --- a/packages/app/src/components/inference/agentic-point/server-metric-cards.tsx +++ b/packages/app/src/components/inference/agentic-point/server-metric-cards.tsx @@ -86,25 +86,29 @@ export function KvCacheUtilizationCard({ sliced }: { sliced: SlicedServerSeries // than one, draw one line per rank in distinct colors so // load skew is visible at a glance; cluster-average sits on // top in white so it stands out. - // Phase slicing can empty an engine that only reported in the other - // phase — drop those so they don't take a legend slot with no line. - const perEngine = (serverSeries.kvCacheUsageByEngine ?? []).filter( - (e) => e.points.length > 0, - ); - const hasPerEngine = perEngine.length > 1; + const allEngines = serverSeries.kvCacheUsageByEngine ?? []; + // Decide off the point's own engine count, not the phase-sliced one: + // this also drives the average line's name, color and stroke, so + // keying it to the slice would make the chart change identity when + // you switch between the Warmup and Profiling tabs. + const hasPerEngine = allEngines.length > 1; + // Colors come from the unsliced position so a rank keeps its color + // across phases; engines with no points in this phase are dropped + // afterwards so they don't take a legend slot with no line. + const perEngine = allEngines + .map((e, i) => ({ + name: engineSeriesName(e.engineLabel), + data: rollingAverage(e.points, 50), + color: DP_RANK_PALETTE[i % DP_RANK_PALETTE.length]!, + // Thin + translucent so the Avg line on top reads as + // the headline number, not just one more series. + strokeWidth: 1, + strokeOpacity: 0.5, + })) + .filter((s) => s.data.length > 0); // Render order matters: per-engine first → average drawn on top. const series = [ - ...(hasPerEngine - ? perEngine.map((e, i) => ({ - name: engineSeriesName(e.engineLabel), - data: rollingAverage(e.points, 50), - color: DP_RANK_PALETTE[i % DP_RANK_PALETTE.length]!, - // Thin + translucent so the Avg line on top reads as - // the headline number, not just one more series. - strokeWidth: 1, - strokeOpacity: 0.5, - })) - : []), + ...(hasPerEngine ? perEngine : []), { name: hasHost ? 'Chip HBM (avg n=50)' diff --git a/packages/db/src/etl/compute-chart-series.test.ts b/packages/db/src/etl/compute-chart-series.test.ts index 4d10dbc24..e1bd7ea58 100644 --- a/packages/db/src/etl/compute-chart-series.test.ts +++ b/packages/db/src/etl/compute-chart-series.test.ts @@ -499,12 +499,76 @@ describe('computeChartSeries', () => { ), ), ); - expect(cs?.kvCacheUsageByEngine.map((e) => e.engineLabel)).toEqual(['prefill 0', 'prefill 1']); + // Both engines are prefill, so naming the role would add nothing. + expect(cs?.kvCacheUsageByEngine.map((e) => e.engineLabel)).toEqual(['0', '1']); expect(cs?.kvCacheUsage).toEqual([{ t: 0, value: 0.5 }]); }); + it('keeps DP ranks in rank order regardless of the order the exporter emits', async () => { + // Regression: role-qualified labels used to sort by blob position, so a + // multi-worker aggregated deployment came out as [decode 3, decode 1, + // decode 0, decode 2] with the palette indexed by that scrambled order. + const cs = await computeChartSeries( + kvBlob( + [3, 1, 0, 2].map((rank) => + kvSeriesFor( + `http://10.30.1.${100 + rank}:7500/metrics`, + { engine: String(rank), dynamo_component: 'backend', worker_id: `w${rank}` }, + [[0, [0.25, 0.5, 0.75, 1][rank]!]], + ), + ), + ), + { framework: 'dynamo-vllm', disagg: false }, + ); + // Every engine is a decode worker, so the role is dropped and the bare + // ranks render in 0..N order — with each rank's own data still attached. + expect(cs?.kvCacheUsageByEngine.map((e) => e.engineLabel)).toEqual(['0', '1', '2', '3']); + expect(cs?.kvCacheUsageByEngine.map((e) => e.points[0]!.value)).toEqual([0.25, 0.5, 0.75, 1]); + }); + + it('names the role only when engines actually differ in role', async () => { + const cs = await computeChartSeries( + kvBlob([ + kvSeriesFor( + 'http://10.0.0.2:7502/metrics', + { dp_rank: '0', engine_type: 'decode', worker_id: 'dec' }, + [[0, 0.6]], + ), + kvSeriesFor( + 'http://10.0.0.1:7500/metrics', + { dp_rank: '0', engine_type: 'prefill', worker_id: 'pre' }, + [[0, 0.2]], + ), + ]), + ); + // Mixed roles -> role is shown, and prefill sorts ahead of decode. + expect(cs?.kvCacheUsageByEngine.map((e) => e.engineLabel)).toEqual(['prefill 0', 'decode 0']); + }); + + it('falls through to dynamo_component when engine_type has no role mapping', async () => { + // Aggregated dynamo-sglang workers carry engine_type="unified" (no role) + // alongside dynamo_component. Stopping at the first present label would + // resolve the role to null and lose the prefill/decode distinction. + const cs = await computeChartSeries( + kvBlob([ + kvSeriesFor( + 'http://10.0.0.1:7500/metrics', + { engine_type: 'unified', dynamo_component: 'prefill', worker_id: 'pre', dp_rank: '0' }, + [[0, 0.2]], + ), + kvSeriesFor( + 'http://10.0.0.2:7502/metrics', + { engine_type: 'unified', dynamo_component: 'backend', worker_id: 'dec', dp_rank: '0' }, + [[0, 0.6]], + ), + ]), + ); + expect(cs?.kvCacheUsageByEngine.map((e) => e.engineLabel)).toEqual(['prefill 0', 'decode 0']); + }); + it('qualifies engines whose display label would otherwise collide', async () => { - // Two decode workers that each number their ranks from 0. + // Two decode workers that each number their ranks from 0. One role, so the + // rank alone is the base and the worker id disambiguates. const cs = await computeChartSeries( kvBlob([ kvSeriesFor( @@ -519,10 +583,69 @@ describe('computeChartSeries', () => { ), ]), ); + expect(cs?.kvCacheUsageByEngine.map((e) => e.engineLabel)).toEqual(['0 (a01a)', '0 (b01b)']); + }); + + it('keeps same-label endpoints apart when their values disagree', async () => { + // Two independent replicas behind a router share an identical label set. + // Treating them as mirrors and dropping one would silently lose an engine. + const labels = { engine_type: 'unified', model_name: 'm', tp_rank: '0' }; + const cs = await computeChartSeries( + kvBlob([ + kvSeriesFor('http://node-a:8888/metrics', labels, [ + [0, 0.1], + [1e9, 0.1], + ]), + kvSeriesFor('http://node-b:8888/metrics', labels, [ + [0, 0.9], + [1e9, 0.9], + ]), + ]), + ); expect(cs?.kvCacheUsageByEngine.map((e) => e.engineLabel)).toEqual([ - 'decode 0 (a01a)', - 'decode 0 (b01b)', + 'node-a:8888', + 'node-b:8888', ]); + expect(cs?.kvCacheUsage).toEqual([ + { t: 0, value: 0.5 }, + { t: 1, value: 0.5 }, + ]); + }); + + it('prefers the mirror that covers the most wall-clock, not the densest', async () => { + // A truncated but high-frequency mirror must not shorten the engine. + const labels = { engine: '0', model_name: 'm' }; + const long: [number, number][] = Array.from({ length: 11 }, (_, i) => [i * 1e9, 0.5]); + const dense: [number, number][] = Array.from({ length: 12 }, (_, i) => [i * 1e8, 0.5]); + const cs = await computeChartSeries( + kvBlob([ + kvSeriesFor('http://a:8000/metrics', labels, dense), + kvSeriesFor('http://b:8000/metrics', labels, long), + ]), + ); + // Same values -> mirrors -> one engine, and it must span the full 10 s. + expect(cs?.kvCacheUsage).toHaveLength(11); + expect(cs?.kvCacheUsage.at(-1)?.t).toBe(10); + }); + + it('drops an engine from the mean while it stops reporting', async () => { + // A carries 1 for t=0..9, goes silent for 300 s, then returns. B reports 0 + // throughout. During the hole the mean must be B alone, not (1+0)/2. + const a: [number, number][] = []; + const b: [number, number][] = []; + for (let i = 0; i < 10; i++) a.push([i * 1e9, 1]); + for (let i = 310; i < 320; i++) a.push([i * 1e9, 1]); + for (let i = 0; i < 320; i++) b.push([i * 1e9, 0]); + const cs = await computeChartSeries( + kvBlob([ + kvSeriesFor('http://a:8000/metrics', { engine: '0' }, a), + kvSeriesFor('http://a:8000/metrics', { engine: '1' }, b), + ]), + ); + const at = (t: number) => cs?.kvCacheUsage.find((p) => p.t === t)?.value; + expect(at(5)).toBe(0.5); // both reporting + expect(at(150)).toBe(0); // A silent -> excluded, B alone + expect(at(315)).toBe(0.5); // A back }); it('averages engines on unaligned scrape grids without sawtoothing', async () => { diff --git a/packages/db/src/etl/compute-chart-series.ts b/packages/db/src/etl/compute-chart-series.ts index 203c87ed4..debb7efca 100644 --- a/packages/db/src/etl/compute-chart-series.ts +++ b/packages/db/src/etl/compute-chart-series.ts @@ -129,9 +129,10 @@ export interface ChartSeries { */ hostKvCacheUsage: TimeSeriesPoint[]; /** - * Per-DP-rank KV cache utilization (0..1 each). One entry per engine - * series found in the raw metric, ordered by the `engine` label when - * present and by series-array index otherwise. Empty for single-engine + * Per-DP-rank KV cache utilization (0..1 each). One entry per LOGICAL + * engine — one KV pool — not per raw series; see `resolveLogicalEngines` + * for how mirrored endpoints, phase blocks and shard ranks collapse. + * Ordered by role, then numeric rank, then worker. Empty for single-engine * deployments — the average `kvCacheUsage` line covers that case alone. * The detail page overlays these on the same chart so DEP load skew is * visible without changing the headline number. @@ -331,8 +332,9 @@ function sortedEntries(m: Map): [number, number][] { * Ranks that shard ONE engine rather than naming a separate one. A KV cache is * allocated per engine and shared by its tensor-, pipeline- and expert-parallel * ranks, so every such rank reports the same pool: on a TP8 SGLang prefill - * worker all eight `tp_rank` series track each other to four decimal places. - * Treating them as separate engines drew eight identical lines for one pool. + * worker the eight `tp_rank` series agree to ~4 decimal places on average + * (with rare single-scrape transients where one rank spikes alone). + * Treating them as separate engines drew eight near-identical lines per pool. * * `engine` / `engine_idx` / `dp_rank` are deliberately NOT here — those DO name * distinct pools (one per DP rank / engine core). @@ -373,55 +375,140 @@ const ENGINE_ROLE_BY_NATIVE_LABEL: Record = { backend: 'decode', }; +/** Label lookup that treats blank values as absent. */ +function labelOrNull(labels: Record, ...names: string[]): string | null { + for (const name of names) { + const value = labels[name]?.trim(); + if (value) return value; + } + return null; +} + /** DP-rank-ish label under any of the names the frameworks emit. */ function engineRankLabel(labels: Record): string | null { - return labels['engine'] ?? labels['engine_idx'] ?? labels['dp_rank'] ?? null; + return labelOrNull(labels, 'engine', 'engine_idx', 'dp_rank'); } function engineRoleLabel(labels: Record): string | null { - const native = labels['engine_type'] ?? labels['dynamo_component']; - return native ? (ENGINE_ROLE_BY_NATIVE_LABEL[native] ?? null) : null; + // Try each source in turn rather than taking the first that EXISTS: an + // aggregated dynamo-sglang worker carries engine_type="unified" (which maps + // to no role) alongside dynamo_component="backend", and stopping at the + // first present label would resolve the role to null. + for (const name of ['engine_type', 'dynamo_component']) { + const native = labelOrNull(labels, name); + const role = native ? ENGINE_ROLE_BY_NATIVE_LABEL[native] : undefined; + if (role) return role; + } + return null; +} + +/** Rank as a sort key. Only plain digit strings sort numerically. */ +function engineRankSortKey(rank: string | null): number { + return rank !== null && /^\d+$/u.test(rank) ? Number(rank) : Number.POSITIVE_INFINITY; +} + +/** Prefill before decode; unroled engines last. Stable across runs. */ +const ROLE_SORT_ORDER: Record = { prefill: 0, decode: 1 }; +function engineRoleSortKey(role: string | null): number { + return role === null ? 2 : (ROLE_SORT_ORDER[role] ?? 2); +} + +/** `scheme://host:port/path` -> `host:port`, the human-facing part. */ +function endpointHostPort(endpointUrl: string): string | null { + const hostPort = /^\w+:\/\/(?[^/]+)/u.exec(endpointUrl)?.groups?.['hostPort']; + return hostPort ?? (endpointUrl || null); } /** * Short, human-readable tiebreaker for engines that would otherwise share a * display label (e.g. two decode workers that each number their ranks 0..7). + * `preferEndpoint` is set when the engines being separated came from the same + * worker, so the worker id cannot tell them apart. */ -function engineDiscriminator(labels: Record, endpointUrl: string): string | null { - const worker = labels['worker_id']; +function engineDiscriminator( + labels: Record, + endpointUrl: string, + preferEndpoint = false, +): string | null { + const worker = preferEndpoint ? null : labelOrNull(labels, 'worker_id'); if (worker) return worker.length > 4 ? worker.slice(-4) : worker; - // Fall back to the endpoint's host:port, which is what distinguishes - // workers when the orchestrator doesn't emit a worker id. - const hostPort = /^\w+:\/\/(?[^/]+)/u.exec(endpointUrl)?.groups?.['hostPort']; - return hostPort ?? (endpointUrl || null); + return endpointHostPort(endpointUrl); } +/** + * Mirrored frontends report the same pool, so their means agree to a few + * thousandths in practice; genuinely different workers on the same labels + * (a router in front of several replicas) sit far apart. This threshold is + * an absolute gap on a 0..1 gauge, comfortably above scrape jitter and well + * below any real load difference. + */ +const MIRROR_MEAN_TOLERANCE = 0.02; + interface LogicalEngine { engineLabel: string; points: TimeSeriesPoint[]; } +/** One endpoint's samples for one identity, keyed by scrape instant. */ +type ScrapeMap = Map; + interface EngineGroup { labels: Record; - /** Per endpoint, the engine's samples keyed by scrape instant. */ - byEndpoint: Map>; + byEndpoint: Map; +} + +/** An engine after endpoint resolution, before its display label is composed. */ +interface ResolvedEngine { + rank: string | null; + role: string | null; + discriminator: string | null; + order: number; + points: TimeSeriesPoint[]; +} + +function scrapesToPoints(scrapes: ScrapeMap, tOf: (ns: number) => number): TimeSeriesPoint[] { + return [...scrapes.entries()] + .toSorted((a, b) => a[0] - b[0]) + .map(([startNs, { sum, count }]) => ({ t: tOf(startNs), value: sum / count })); +} + +function meanOf(scrapes: ScrapeMap): number { + let total = 0; + for (const { sum, count } of scrapes.values()) total += sum / count; + return scrapes.size === 0 ? 0 : total / scrapes.size; +} + +/** Wall-clock span covered, so a dense-but-truncated mirror can't win. */ +function spanOf(scrapes: ScrapeMap): number { + let lo = Number.POSITIVE_INFINITY; + let hi = Number.NEGATIVE_INFINITY; + for (const ns of scrapes.keys()) { + if (ns < lo) lo = ns; + if (ns > hi) hi = ns; + } + return hi >= lo ? hi - lo : 0; } /** * Collapse a gauge's raw series into one entry per logical engine. * * The three kinds of duplication need three different treatments: - * - Same endpoint, different phase blocks (v12's warmup merge): disjoint - * time ranges, so keying by scrape instant simply unions them into one - * continuous series. - * - Same endpoint, same instant: intra-engine shard ranks reporting the - * one pool they share, so they collapse to their mean (identical in - * practice, to four decimal places). - * - Different endpoints (mirrored API-server frontends): overlapping time - * ranges carrying the same measurement a few hundred ms apart. Merging - * them would interleave near-duplicate samples and silently halve the - * span of the frontend's fixed-width rolling average, so we keep the - * endpoint with the most complete coverage and drop the rest. + * - Same endpoint, different phase blocks (v12's warmup merge): keying by + * scrape instant unions them into one series. The blocks' first/last + * bounds can look overlapping (profiling often emits one boundary sample + * then gaps until warmup ends) but they never share an instant, so the + * union neither drops nor double-counts a scrape. + * - Same endpoint, same instant: intra-engine shard ranks reporting the one + * pool they share, so they collapse to their mean (they agree to ~4 decimal + * places on average, with rare single-scrape transients). + * - Different endpoints reporting the same identity: usually mirrored + * API-server frontends carrying the same measurement a few hundred ms + * apart. Merging those would interleave near-duplicate samples and halve + * the effective span of the frontend's fixed-width rolling average, so the + * best-covered endpoint wins and the rest are dropped — but ONLY when their + * values agree. Endpoints that disagree are genuinely different engines + * 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( series: readonly RawSeries[] | undefined, @@ -442,8 +529,8 @@ function resolveLogicalEngines( group.byEndpoint.set(endpoint, scrapes); } for (const ts of s.timeslices ?? []) { - if (typeof ts.start_ns !== 'number' || typeof ts.avg !== 'number') continue; - if (!Number.isFinite(ts.avg)) continue; + if (typeof ts.start_ns !== 'number' || !Number.isFinite(ts.start_ns)) continue; + if (typeof ts.avg !== 'number' || !Number.isFinite(ts.avg)) continue; const at = scrapes.get(ts.start_ns); if (at) { at.sum += ts.avg; @@ -454,63 +541,97 @@ function resolveLogicalEngines( } } - // Insertion order = first appearance in the blob, which is the engine order - // the exporter emitted; `rank` sorting below refines it when ranks exist. - const resolved: { label: string; discriminator: string | null; points: TimeSeriesPoint[] }[] = []; + const resolved: ResolvedEngine[] = []; for (const group of groups.values()) { - // Deterministic pick: most scrape instants wins, endpoint URL breaks ties. - let chosenEndpoint = ''; - let chosen: Map | null = null; - for (const [endpoint, scrapes] of [...group.byEndpoint].toSorted((a, b) => - a[0].localeCompare(b[0]), - )) { - if (scrapes.size > (chosen?.size ?? 0)) { - chosen = scrapes; - chosenEndpoint = endpoint; - } - } - if (!chosen || chosen.size === 0) continue; + const endpoints = [...group.byEndpoint] + .filter(([, scrapes]) => scrapes.size > 0) + // Endpoint URL keeps the walk deterministic before any ranking. + .toSorted((a, b) => (a[0] < b[0] ? -1 : a[0] > b[0] ? 1 : 0)); + if (endpoints.length === 0) continue; + const rank = engineRankLabel(group.labels); const role = engineRoleLabel(group.labels); - const discriminator = engineDiscriminator(group.labels, chosenEndpoint); - const named = role ? (rank === null ? role : `${role} ${rank}`) : rank; - resolved.push({ - // With no rank- or role-like label the worker/endpoint is the only thing - // that names this engine, so lead with it instead of a bare index. - label: named ?? discriminator ?? '', - discriminator, - points: [...chosen.entries()] - .toSorted((a, b) => a[0] - b[0]) - .map(([startNs, { sum, count }]) => ({ t: tOf(startNs), value: sum / count })), + const push = (endpointUrl: string, scrapes: ScrapeMap, preferEndpoint: boolean) => { + resolved.push({ + rank, + role, + discriminator: engineDiscriminator(group.labels, endpointUrl, preferEndpoint), + order: resolved.length, + points: scrapesToPoints(scrapes, tOf), + }); + }; + + if (endpoints.length === 1) { + push(endpoints[0]![0], endpoints[0]![1], false); + continue; + } + + const means = endpoints.map(([, scrapes]) => meanOf(scrapes)); + const mirrored = Math.max(...means) - Math.min(...means) <= MIRROR_MEAN_TOLERANCE; + if (!mirrored) { + // Same labels, different measurements: distinct engines, not mirrors. + for (const [endpointUrl, scrapes] of endpoints) push(endpointUrl, scrapes, true); + continue; + } + // Mirrors: keep the endpoint that covers the most wall-clock, breaking + // ties on sample count and then URL, so a dense but truncated mirror + // cannot shorten the engine's series. + const best = endpoints.reduce((a, b) => { + const sa = spanOf(a[1]); + const sb = spanOf(b[1]); + if (sb !== sa) return sb > sa ? b : a; + if (b[1].size !== a[1].size) return b[1].size > a[1].size ? b : a; + return a; }); + push(best[0], best[1], false); } - // Plain numeric ranks render in 0..N order; role-qualified and endpoint-named - // engines keep the order the exporter emitted them in. - const ordered = resolved - .map((engine, idx) => { - const numeric = Number(engine.label); - return { ...engine, idx, sortKey: engine.label && Number.isFinite(numeric) ? numeric : idx }; - }) - .toSorted((a, b) => a.sortKey - b.sortKey || a.idx - b.idx) - .map((engine, idx) => ({ ...engine, base: engine.label || `#${idx}` })); + // Sort on the identity's components, never on the composed string: role + // 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( + (a, b) => + engineRoleSortKey(a.role) - engineRoleSortKey(b.role) || + engineRankSortKey(a.rank) - engineRankSortKey(b.rank) || + (a.discriminator ?? '').localeCompare(b.discriminator ?? '') || + a.order - b.order, + ); + + // 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. + const roles = new Set(ordered.map((e) => e.role).filter((r) => r !== null)); + const showRole = roles.size > 1; + const withBase = ordered.map((engine, idx) => { + const named = + showRole && engine.role + ? engine.rank === null + ? engine.role + : `${engine.role} ${engine.rank}` + : engine.rank; + // Nothing rank- or role-like to go on: the worker/endpoint is the only + // thing that names this engine, so lead with it rather than a bare index. + return { ...engine, base: named ?? engine.discriminator ?? `#${idx}` }; + }); // Qualify collisions (e.g. two decode workers that each number their ranks - // 0..7) so every legend entry names exactly one engine. + // 0..7) so every legend entry names exactly one line. const baseCounts = new Map(); - for (const engine of ordered) baseCounts.set(engine.base, (baseCounts.get(engine.base) ?? 0) + 1); + for (const engine of withBase) { + baseCounts.set(engine.base, (baseCounts.get(engine.base) ?? 0) + 1); + } const used = new Set(); - return ordered.map((engine, idx) => { - let engineLabel = engine.base; - if ((baseCounts.get(engine.base) ?? 0) > 1) { - const qualifier = - engine.discriminator && engine.discriminator !== engine.base ? engine.discriminator : idx; - engineLabel = `${engine.base} (${qualifier})`; + return withBase.map((engine) => { + let label = engine.base; + if ((baseCounts.get(engine.base) ?? 0) > 1 && engine.discriminator) { + label = `${engine.base} (${engine.discriminator})`; } - // Last resort, so a legend entry never stands for two lines. - while (used.has(engineLabel)) engineLabel = `${engineLabel}'`; - used.add(engineLabel); - return { engineLabel, points: engine.points }; + // The discriminator isn't guaranteed unique either; fall back to a counter + // so a legend entry never stands for two lines. + let candidate = label; + for (let n = 2; used.has(candidate); n++) candidate = `${label} #${n}`; + used.add(candidate); + return { engineLabel: candidate, points: engine.points }; }); } @@ -527,7 +648,8 @@ function resolveLogicalEngines( * Each engine therefore holds its last scrape until its next one (a gauge * keeps its value between scrapes) and contributes only inside its own * observed window, so an engine that starts late or stops early neither - * pulls the mean toward a stale value nor drops it to zero. + * pulls the mean toward a stale value nor drops it to zero. A hole in the + * middle of that window is bounded too — see `carryLimitSeconds`. */ function averageAcrossEngines(engines: readonly LogicalEngine[]): TimeSeriesPoint[] { const active = engines.filter((engine) => engine.points.length > 0); @@ -540,6 +662,7 @@ function averageAcrossEngines(engines: readonly LogicalEngine[]): TimeSeriesPoin ); const cursors: number[] = Array.from({ length: active.length }, () => -1); const lastT = active.map((engine) => engine.points.at(-1)!.t); + const carryLimit = active.map((engine) => carryLimitSeconds(engine.points)); const out: TimeSeriesPoint[] = []; for (const t of timeline) { let sum = 0; @@ -552,6 +675,9 @@ function averageAcrossEngines(engines: readonly LogicalEngine[]): TimeSeriesPoin // Before this engine's first scrape or after its last — no value to // carry, so it sits out of this tick's mean entirely. if (cursor < 0 || t > lastT[i]!) continue; + // Inside the window but far past the last sample: the engine stopped + // reporting for a while, so don't average in a stale reading. + if (t - points[cursor]!.t > carryLimit[i]!) continue; sum += points[cursor]!.value; n++; } @@ -560,6 +686,24 @@ function averageAcrossEngines(engines: readonly LogicalEngine[]): TimeSeriesPoin return out; } +/** + * How long one engine's last sample may stand in for it: 5x its own median + * scrape gap. A dropped scrape or two still carries (real runs sit at 1 Hz + * with gaps never above ~1 s), but a long reporting hole drops the engine + * out of the mean instead of pinning it to a minutes-old value. + */ +function carryLimitSeconds(points: readonly TimeSeriesPoint[]): number { + if (points.length < 2) return Number.POSITIVE_INFINITY; + const gaps: number[] = []; + for (let i = 1; i < points.length; i++) { + const gap = points[i]!.t - points[i - 1]!.t; + if (gap > 0) gaps.push(gap); + } + if (gaps.length === 0) return Number.POSITIVE_INFINITY; + gaps.sort((a, b) => a - b); + return 5 * gaps[gaps.length >> 1]!; +} + function buildSeriesFromMetrics( metrics: MetricsMap, context: ServerMetricsContext, From 1aedd4da39a27d23491634fa096f4e75724c481b Mon Sep 17 00:00:00 2001 From: Cam Quilici Date: Sun, 9 Aug 2026 19:37:49 -0500 Subject: [PATCH 3/3] docs(agentic): state the mirror-detection threshold's residual limitation Two independent replicas behind a round-robin router would have similar means by design and would still be fused into one engine. That degrades the per-engine overlay but not the cluster average, whereas the uneven-load case the threshold does catch is the one that would make the average wrong. Say so at the constant rather than implying the check is airtight. Co-Authored-By: Claude Opus 5 (1M context) --- packages/db/src/etl/compute-chart-series.ts | 22 ++++++++++++++++----- 1 file changed, 17 insertions(+), 5 deletions(-) diff --git a/packages/db/src/etl/compute-chart-series.ts b/packages/db/src/etl/compute-chart-series.ts index debb7efca..d719e0e60 100644 --- a/packages/db/src/etl/compute-chart-series.ts +++ b/packages/db/src/etl/compute-chart-series.ts @@ -436,11 +436,23 @@ function engineDiscriminator( } /** - * Mirrored frontends report the same pool, so their means agree to a few - * thousandths in practice; genuinely different workers on the same labels - * (a router in front of several replicas) sit far apart. This threshold is - * an absolute gap on a 0..1 gauge, comfortably above scrape jitter and well - * below any real load difference. + * Absolute gap between two endpoints' whole-run means, on a 0..1 gauge, below + * which they are treated as mirrors of one engine rather than two engines + * sharing a label set. + * + * Measured mirrors sit far under this: the three two-endpoint vLLM configs in + * the corpus differ by 0.03%-2.19% of their means (under 0.001 absolute, even + * on the heavily loaded rows), while a genuinely distinct prefill and decode + * worker differ by ~0.08. So both sides have roughly 4x of margin. + * + * Residual limitation, accepted deliberately: two independent replicas behind + * a round-robin router would have similar means BY DESIGN and would still be + * fused, showing one line instead of two. That case degrades the per-engine + * overlay but not the cluster average — averaging two engines that track each + * other gives the same number either way — whereas the case this does catch + * (replicas under uneven load) is the one where fusing would make the average + * itself wrong. Distinguishing the former needs lag-aligned pointwise + * comparison, which no data in the corpus currently justifies. */ const MIRROR_MEAN_TOLERANCE = 0.02;