diff --git a/docs/data-pipeline.md b/docs/data-pipeline.md index 9d1ce61cc..24fce8097 100644 --- a/docs/data-pipeline.md +++ b/docs/data-pipeline.md @@ -89,6 +89,18 @@ 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, 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 (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 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/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 3cc023de4..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 @@ -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 ( 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: `DP ${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 749241713..e1bd7ea58 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,347 @@ 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]], + ), + ), + ), + ); + // 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. One role, so the + // rank alone is the base and the worker id disambiguates. + 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(['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([ + '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 () => { + // 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..d719e0e60 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. */ @@ -111,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. @@ -303,6 +322,400 @@ 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 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). + */ +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', +}; + +/** 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 labelOrNull(labels, 'engine', 'engine_idx', 'dp_rank'); +} + +function engineRoleLabel(labels: Record): string | 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, + preferEndpoint = false, +): string | null { + const worker = preferEndpoint ? null : labelOrNull(labels, 'worker_id'); + if (worker) return worker.length > 4 ? worker.slice(-4) : worker; + return endpointHostPort(endpointUrl); +} + +/** + * 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; + +interface LogicalEngine { + engineLabel: string; + points: TimeSeriesPoint[]; +} + +/** One endpoint's samples for one identity, keyed by scrape instant. */ +type ScrapeMap = Map; + +interface EngineGroup { + labels: Record; + 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): 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, + 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' || !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; + at.count++; + } else { + scrapes.set(ts.start_ns, { sum: ts.avg, count: 1 }); + } + } + } + + const resolved: ResolvedEngine[] = []; + for (const group of groups.values()) { + 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 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); + } + + // 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 line. + const baseCounts = new Map(); + for (const engine of withBase) { + baseCounts.set(engine.base, (baseCounts.get(engine.base) ?? 0) + 1); + } + const used = new Set(); + return withBase.map((engine) => { + let label = engine.base; + if ((baseCounts.get(engine.base) ?? 0) > 1 && engine.discriminator) { + label = `${engine.base} (${engine.discriminator})`; + } + // 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 }; + }); +} + +/** + * 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. 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); + 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 carryLimit = active.map((engine) => carryLimitSeconds(engine.points)); + 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; + // 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++; + } + if (n > 0) out.push({ t, value: sum / n }); + } + 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, @@ -347,33 +760,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.