From 1327c6760bb5bd0cdcec0a3fefa38a67b69134f8 Mon Sep 17 00:00:00 2001 From: Cam Quilici Date: Tue, 4 Aug 2026 17:11:32 -0500 Subject: [PATCH] Optimize agentic trace ingestion --- .../db/src/etl/compute-aggregate-stats.ts | 30 +++- packages/db/src/etl/compute-chart-series.ts | 19 ++- .../db/src/etl/compute-trace-derived.test.ts | 142 ++++++++++++++++++ packages/db/src/etl/compute-trace-derived.ts | 95 ++++++++++++ packages/db/src/etl/gzip-json-stream.test.ts | 50 +++++- packages/db/src/etl/gzip-json-stream.ts | 83 +++++++++- .../src/etl/trace-artifact-discovery.test.ts | 53 +++++++ .../db/src/etl/trace-artifact-discovery.ts | 37 ++++- packages/db/src/etl/trace-replay-ingest.ts | 14 +- 9 files changed, 498 insertions(+), 25 deletions(-) create mode 100644 packages/db/src/etl/compute-trace-derived.test.ts create mode 100644 packages/db/src/etl/compute-trace-derived.ts diff --git a/packages/db/src/etl/compute-aggregate-stats.ts b/packages/db/src/etl/compute-aggregate-stats.ts index d4a76e27b..503d75631 100644 --- a/packages/db/src/etl/compute-aggregate-stats.ts +++ b/packages/db/src/etl/compute-aggregate-stats.ts @@ -61,7 +61,7 @@ export function mergeProfileStatsUpgrade( } /** Metric subtrees we extract via stream-parse on oversized server blobs. */ -const TARGET_METRIC_KEYS = new Set([ +export const AGGREGATE_SERVER_METRIC_KEYS = new Set([ 'vllm:kv_cache_usage_perc', 'vllm:gpu_cache_usage_perc', 'vllm:prefix_cache_hits', @@ -78,10 +78,36 @@ const TARGET_METRIC_KEYS = new Set([ async function streamExtractServer( buffer: Buffer, ): Promise<{ kvCacheUtil: number[]; prefixCacheHitRate: number[] }> { - const collected = await streamCollectKeys(buffer, 'metrics', TARGET_METRIC_KEYS); + const collected = await streamCollectKeys( + buffer, + 'metrics', + AGGREGATE_SERVER_METRIC_KEYS, + ); return extractServerMetricSamples(JSON.stringify({ metrics: collected })); } +/** + * Add server-derived distributions to profile stats using an already parsed + * profiling metric map. Ingest uses this to share one server JSON parse with + * chart-series generation; the output shape and ordering match + * `computeAggregateStats()` exactly. + */ +export function withServerMetricAggregateStats( + profileStats: AggregateStats, + metrics: Record, +): AggregateStats { + try { + const server = extractServerMetricSamples(JSON.stringify({ metrics })); + return { + ...profileStats, + kvCacheUtil: percentilesOf(server.kvCacheUtil), + prefixCacheHitRate: percentilesOf(server.prefixCacheHitRate), + }; + } catch { + return profileStats; + } +} + /** * Compute the full versioned stats bundle from a (profile, server-metrics) * blob pair. Either blob may be null (e.g. only the server file existed) — diff --git a/packages/db/src/etl/compute-chart-series.ts b/packages/db/src/etl/compute-chart-series.ts index 5536d15ac..ba2661296 100644 --- a/packages/db/src/etl/compute-chart-series.ts +++ b/packages/db/src/etl/compute-chart-series.ts @@ -156,18 +156,18 @@ interface RawSeries { timeslices?: RawSlice[]; } -interface RawMetric { +export interface RawMetric { series?: RawSeries[]; } -type MetricsMap = Record; +export type MetricsMap = Record; /** * The set of metric subtrees the chart consumes. Includes both vllm:* and * sglang:* names so the stream-parse fallback collects whichever framework * the blob was emitted by — `buildSeriesFromMetrics` then picks per metric. */ -const CHART_METRIC_KEYS = new Set([ +export const CHART_METRIC_KEYS = new Set([ // vLLM 'vllm:kv_cache_usage_perc', 'vllm:gpu_cache_usage_perc', @@ -257,6 +257,19 @@ export async function computeChartSeries( return buildSeriesFromMetrics(metrics, context); } +/** + * Build the chart payload from already parsed phase maps. This is the same + * merge + projection used by `computeChartSeries()`, exposed so ingest can + * share one server JSON parse with aggregate-stat computation. + */ +export function computeChartSeriesFromMetricPhases( + profiling: MetricsMap, + warmup: MetricsMap, + context: ServerMetricsContext = {}, +): ChartSeries { + return buildSeriesFromMetrics(mergePhaseMetrics(profiling, warmup), context); +} + /** * Aggregate one timeslice field across all series of a metric, indexed by * `start_ns`. Multi-engine vllm deployments report one series per engine — diff --git a/packages/db/src/etl/compute-trace-derived.test.ts b/packages/db/src/etl/compute-trace-derived.test.ts new file mode 100644 index 000000000..ca1227611 --- /dev/null +++ b/packages/db/src/etl/compute-trace-derived.test.ts @@ -0,0 +1,142 @@ +import { gzipSync } from 'node:zlib'; + +import { describe, expect, it } from 'vitest'; + +import { computeAggregateStats } from './compute-aggregate-stats.js'; +import { computeChartSeries } from './compute-chart-series.js'; +import { computeRequestTimeline } from './compute-request-timeline.js'; +import { computeTraceDerivedPayloads } from './compute-trace-derived.js'; + +function makeProfileBlob(): Buffer { + return gzipSync( + Buffer.from( + [ + { + metadata: { + conversation_id: 'conv-1', + turn_index: 0, + worker_id: 'worker-1', + agent_depth: 0, + benchmark_phase: 'profiling', + credit_issued_ns: 1_000, + request_start_ns: 2_000, + request_end_ns: 5_000, + }, + metrics: { + input_sequence_length: { value: 128, unit: 'tokens' }, + output_sequence_length: { value: 64, unit: 'tokens' }, + time_to_first_token: { value: 20, unit: 'ms' }, + inter_token_latency: { value: 5, unit: 'ms' }, + }, + }, + { + metadata: { + conversation_id: 'conv-1', + turn_index: 1, + worker_id: 'worker-1', + agent_depth: 0, + benchmark_phase: 'profiling', + credit_issued_ns: 6_000, + request_start_ns: 7_000, + request_end_ns: 10_000, + }, + metrics: { + input_sequence_length: { value: 256, unit: 'tokens' }, + output_sequence_length: { value: 32, unit: 'tokens' }, + time_to_first_token: { value: 30, unit: 'ms' }, + inter_token_latency: { value: 6, unit: 'ms' }, + }, + }, + ] + .map((record) => JSON.stringify(record)) + .join('\n'), + ), + ); +} + +function metric( + values: { start_ns: number; end_ns: number; avg?: number; rate?: number }[], + labels: Record = {}, +) { + return { series: [{ endpoint_url: 'worker.test:8000', labels, timeslices: values }] }; +} + +function makeServerBlob(): Buffer { + return gzipSync( + Buffer.from( + JSON.stringify({ + warmup_metrics: { + 'vllm:kv_cache_usage_perc': metric([{ start_ns: 0, end_ns: 1e9, avg: 0.1 }]), + 'vllm:prompt_tokens': metric([{ start_ns: 0, end_ns: 1e9, rate: 100 }]), + }, + metrics: { + 'vllm:kv_cache_usage_perc': metric([ + { start_ns: 10e9, end_ns: 11e9, avg: 0.4 }, + { start_ns: 11e9, end_ns: 12e9, avg: 0.6 }, + ]), + 'vllm:prefix_cache_hits': metric([{ start_ns: 10e9, end_ns: 11e9, rate: 80 }]), + 'vllm:prefix_cache_queries': metric([{ start_ns: 10e9, end_ns: 11e9, rate: 100 }]), + 'vllm:gpu_prefix_cache_hits': metric([{ start_ns: 10e9, end_ns: 11e9, rate: 80 }]), + 'vllm:gpu_prefix_cache_queries': metric([{ start_ns: 10e9, end_ns: 11e9, rate: 100 }]), + 'vllm:num_requests_running': metric([{ start_ns: 10e9, end_ns: 11e9, avg: 3 }]), + 'vllm:num_requests_waiting': metric([{ start_ns: 10e9, end_ns: 11e9, avg: 2 }]), + 'vllm:prompt_tokens': metric([{ start_ns: 10e9, end_ns: 11e9, rate: 900 }]), + 'vllm:generation_tokens': metric([{ start_ns: 10e9, end_ns: 11e9, rate: 450 }]), + }, + }), + ), + ); +} + +describe('computeTraceDerivedPayloads', () => { + it('is byte-for-byte JSON equivalent to the independent upload computations', async () => { + const profileBlob = makeProfileBlob(); + const serverBlob = makeServerBlob(); + const context = { framework: 'dynamo-vllm', disagg: true } as const; + + const [aggregateStats, chartSeries, requestTimeline] = await Promise.all([ + computeAggregateStats({ profileBlob, serverBlob }), + computeChartSeries(serverBlob, context), + Promise.resolve(computeRequestTimeline(profileBlob)), + ]); + const optimized = await computeTraceDerivedPayloads(profileBlob, serverBlob, context); + + expect(Buffer.from(JSON.stringify(optimized.aggregateStats))).toEqual( + Buffer.from(JSON.stringify(aggregateStats)), + ); + expect(Buffer.from(JSON.stringify(optimized.chartSeries))).toEqual( + Buffer.from(JSON.stringify(chartSeries)), + ); + expect(Buffer.from(JSON.stringify(optimized.requestTimeline))).toEqual( + Buffer.from(JSON.stringify(requestTimeline)), + ); + }); + + it('produces the same payloads through the oversized streaming path', async () => { + const profileBlob = makeProfileBlob(); + const serverBlob = makeServerBlob(); + const bounded = await computeTraceDerivedPayloads(profileBlob, serverBlob); + const streamed = await computeTraceDerivedPayloads( + profileBlob, + serverBlob, + {}, + { + maxInMemoryBytes: 1, + }, + ); + + expect(streamed).toEqual(bounded); + }); + + it('preserves independent malformed-input fallbacks', async () => { + const profileBlob = makeProfileBlob(); + const malformedServer = Buffer.from('not-gzip'); + const optimized = await computeTraceDerivedPayloads(profileBlob, malformedServer); + + expect(optimized.aggregateStats).toEqual( + await computeAggregateStats({ profileBlob, serverBlob: malformedServer }), + ); + expect(optimized.chartSeries).toBeNull(); + expect(optimized.requestTimeline).toEqual(computeRequestTimeline(profileBlob)); + }); +}); diff --git a/packages/db/src/etl/compute-trace-derived.ts b/packages/db/src/etl/compute-trace-derived.ts new file mode 100644 index 000000000..9cde30fd3 --- /dev/null +++ b/packages/db/src/etl/compute-trace-derived.ts @@ -0,0 +1,95 @@ +/** + * Compute every derived trace-replay payload while parsing server metrics only + * once. The standalone aggregate/chart helpers remain the compatibility path + * for backfills; ingest uses this coordinator to avoid repeated GiB-scale + * decompression and JSON tokenization. + */ + +import { + AGGREGATE_SERVER_METRIC_KEYS, + computeAggregateStats, + withServerMetricAggregateStats, + type AggregateStats, +} from './compute-aggregate-stats.js'; +import { + CHART_METRIC_KEYS, + computeChartSeriesFromMetricPhases, + type ChartSeries, + type MetricsMap, + type RawMetric, +} from './compute-chart-series.js'; +import { computeRequestTimeline, type RequestTimeline } from './compute-request-timeline.js'; +import { collectMetricPhases } from './gzip-json-stream.js'; +import type { ServerMetricsContext } from './server-metrics-adapters.js'; + +export interface TraceDerivedPayloads { + aggregateStats: AggregateStats; + chartSeries: ChartSeries | null; + requestTimeline: RequestTimeline | null; +} + +export interface TraceDerivedComputeOptions { + /** Override the bounded fast-path threshold, primarily for streaming tests. */ + maxInMemoryBytes?: number; +} + +const DERIVED_SERVER_METRIC_KEYS = new Set([...CHART_METRIC_KEYS, ...AGGREGATE_SERVER_METRIC_KEYS]); + +function selectMetrics(metrics: MetricsMap, wanted: ReadonlySet): MetricsMap { + const selected: MetricsMap = {}; + for (const [name, metric] of Object.entries(metrics)) { + if (wanted.has(name)) selected[name] = metric; + } + return selected; +} + +/** + * Produce the same three JSON values previously computed independently in + * `insertTraceReplay()`. Malformed inputs retain the old failure isolation: + * profile stats/timeline can succeed without server metrics, and a chart + * projection failure does not discard aggregate stats. + */ +export async function computeTraceDerivedPayloads( + profileBlob: Buffer | null, + serverBlob: Buffer | null, + metricsContext: ServerMetricsContext = {}, + options: TraceDerivedComputeOptions = {}, +): Promise { + const profileStatsPromise = computeAggregateStats({ profileBlob, serverBlob: null }); + const requestTimeline = computeRequestTimeline(profileBlob); + + const phases = serverBlob + ? await collectMetricPhases( + serverBlob, + DERIVED_SERVER_METRIC_KEYS, + options.maxInMemoryBytes, + ).catch(() => null) + : null; + let aggregateStats = await profileStatsPromise; + let chartSeries: ChartSeries | null = null; + + if (phases) { + const aggregateMetrics = phases.complete + ? phases.metrics + : selectMetrics(phases.metrics, AGGREGATE_SERVER_METRIC_KEYS); + aggregateStats = withServerMetricAggregateStats(aggregateStats, aggregateMetrics); + + // The historical in-memory path builds chart timing metadata from every + // metric, while its oversized streaming fallback retains only chart keys. + // Preserve that distinction exactly even though the shared streaming pass + // also collects the aggregate-only GPU prefix-cache aliases. + const profiling = phases.complete + ? phases.metrics + : selectMetrics(phases.metrics, CHART_METRIC_KEYS); + const warmup = phases.complete + ? phases.warmupMetrics + : selectMetrics(phases.warmupMetrics, CHART_METRIC_KEYS); + try { + chartSeries = computeChartSeriesFromMetricPhases(profiling, warmup, metricsContext); + } catch { + chartSeries = null; + } + } + + return { aggregateStats, chartSeries, requestTimeline }; +} diff --git a/packages/db/src/etl/gzip-json-stream.test.ts b/packages/db/src/etl/gzip-json-stream.test.ts index 5fbb7dc81..30aa9fea7 100644 --- a/packages/db/src/etl/gzip-json-stream.test.ts +++ b/packages/db/src/etl/gzip-json-stream.test.ts @@ -2,7 +2,11 @@ import { gzipSync } from 'node:zlib'; import { describe, expect, it } from 'vitest'; -import { gunzipJsonWithinLimit, streamCollectKeys } from './gzip-json-stream.js'; +import { + collectMetricPhases, + gunzipJsonWithinLimit, + streamCollectKeys, +} from './gzip-json-stream.js'; describe('gunzipJsonWithinLimit', () => { const json = JSON.stringify({ metrics: { value: 1 } }); @@ -61,3 +65,47 @@ describe('streamCollectKeys', () => { ).rejects.toThrow(); }); }); + +describe('collectMetricPhases', () => { + const blob = gzipSync( + JSON.stringify({ + metadata: { ignored: true }, + metrics: { + wanted: { series: [{ timeslices: [{ start_ns: 1, rate: 2 }] }] }, + ignored: { series: [{ timeslices: [{ start_ns: 3, rate: 4 }] }] }, + }, + warmup_metrics: { + wanted: { series: [{ timeslices: [{ start_ns: 0, rate: 1 }] }] }, + ignored: { series: [] }, + }, + }), + ); + + it('retains the complete phase maps on the bounded fast path', async () => { + const phases = await collectMetricPhases(blob, new Set(['wanted'])); + + expect(phases.complete).toBe(true); + expect(Object.keys(phases.metrics)).toEqual(['wanted', 'ignored']); + expect(Object.keys(phases.warmupMetrics)).toEqual(['wanted', 'ignored']); + }); + + it('collects both filtered phase maps from one streaming parse', async () => { + const phases = await collectMetricPhases(blob, new Set(['wanted']), 1); + + expect(phases).toEqual({ + metrics: { + wanted: { series: [{ timeslices: [{ start_ns: 1, rate: 2 }] }] }, + }, + warmupMetrics: { + wanted: { series: [{ timeslices: [{ start_ns: 0, rate: 1 }] }] }, + }, + complete: false, + }); + }); + + it('rejects malformed gzip input on both paths', async () => { + await expect( + collectMetricPhases(Buffer.from('not gzip'), new Set(['wanted']), 1), + ).rejects.toThrow(); + }); +}); diff --git a/packages/db/src/etl/gzip-json-stream.ts b/packages/db/src/etl/gzip-json-stream.ts index 1fc4b6ee6..48723f572 100644 --- a/packages/db/src/etl/gzip-json-stream.ts +++ b/packages/db/src/etl/gzip-json-stream.ts @@ -7,7 +7,8 @@ * stream-json pipeline collects only the top-level subtrees callers need. */ -import { Readable } from 'node:stream'; +import { PassThrough, Readable } from 'node:stream'; +import { pipeline } from 'node:stream/promises'; import { createGunzip, gunzipSync } from 'node:zlib'; import { chain } from 'stream-chain'; @@ -56,7 +57,7 @@ export async function streamCollectKeys( wanted: ReadonlySet, ): Promise> { const collected: Record = {}; - const pipeline = chain([ + const metricStream = chain([ Readable.from(buffer), createGunzip(), parser(), @@ -64,12 +65,84 @@ export async function streamCollectKeys( streamObject(), ]); await new Promise((resolve, reject) => { - pipeline.on('data', (chunk: unknown) => { + metricStream.on('data', (chunk: unknown) => { const { key, value } = chunk as { key: string; value: T }; if (wanted.has(key)) collected[key] = value; }); - pipeline.on('end', resolve); - pipeline.on('error', reject); + metricStream.on('end', resolve); + metricStream.on('error', reject); }); return collected; } + +export interface MetricPhaseMaps { + metrics: Record; + warmupMetrics: Record; + /** True when the bounded fast path retained every metric in the document. */ + complete: boolean; +} + +async function collectTokenBranch( + input: PassThrough, + filter: 'metrics' | 'warmup_metrics', + wanted: ReadonlySet, +): Promise> { + const collected: Record = {}; + const output = chain([input, pick({ filter }), streamObject()]); + for await (const chunk of output) { + const { key, value } = chunk as { key: string; value: T }; + if (wanted.has(key)) collected[key] = value; + } + return collected; +} + +/** + * Gunzip and parse both server-metric phase blocks once. Large documents fan + * the parser's token stream out to two lightweight selectors, avoiding one + * complete decompression + JSON tokenization pass per phase. + */ +export async function collectMetricPhases( + buffer: Buffer, + wanted: ReadonlySet, + maxInMemoryBytes = MAX_IN_MEMORY_JSON_BYTES, +): Promise> { + const json = gunzipJsonWithinLimit(buffer, maxInMemoryBytes); + if (json !== null) { + const parsed = JSON.parse(json) as { + metrics?: Record; + warmup_metrics?: Record; + }; + return { + metrics: parsed.metrics ?? {}, + warmupMetrics: parsed.warmup_metrics ?? {}, + complete: true, + }; + } + + // Attach every branch before starting the source pipeline so no parser + // tokens can be missed. PassThrough backpressure keeps the two consumers in + // lockstep without buffering the full document. + const profilingInput = new PassThrough({ objectMode: true }); + const warmupInput = new PassThrough({ objectMode: true }); + const tokenTee = new PassThrough({ objectMode: true }); + tokenTee.pipe(profilingInput); + tokenTee.pipe(warmupInput); + + const profiling = collectTokenBranch(profilingInput, 'metrics', wanted); + const warmup = collectTokenBranch(warmupInput, 'warmup_metrics', wanted); + const tokens = chain([Readable.from(buffer), createGunzip(), parser()]); + + try { + const [, metrics, warmupMetrics] = await Promise.all([ + pipeline(tokens, tokenTee), + profiling, + warmup, + ]); + return { metrics, warmupMetrics, complete: false }; + } catch (error) { + tokenTee.destroy(); + profilingInput.destroy(); + warmupInput.destroy(); + throw error; + } +} diff --git a/packages/db/src/etl/trace-artifact-discovery.test.ts b/packages/db/src/etl/trace-artifact-discovery.test.ts index 2bb1d51bf..07fa2e6cb 100644 --- a/packages/db/src/etl/trace-artifact-discovery.test.ts +++ b/packages/db/src/etl/trace-artifact-discovery.test.ts @@ -63,4 +63,57 @@ describe('discoverTraceReplayArtifacts', () => { 'multinode_server_logs/agentic/conc_96/aiperf_artifacts/profile_export.jsonl', ); }); + + it('does not extract a duplicate archive when a complete per-concurrency trace exists', () => { + const root = tempDir(); + const suffix = 'config-c_conc96_variant'; + writeTraceFiles(path.join(root, `agentic_${suffix}`, 'conc_96')); + + const artifactDir = path.join(root, `multinode_server_logs_${suffix}`); + const archiveSource = path.join(root, 'duplicate-archive-source'); + writeTraceFiles(path.join(archiveSource, 'agentic', 'conc_96')); + fs.mkdirSync(artifactDir, { recursive: true }); + execFileSync('tar', [ + '-czf', + path.join(artifactDir, 'multinode_server_logs.tar.gz'), + '-C', + archiveSource, + '.', + ]); + fs.rmSync(archiveSource, { recursive: true, force: true }); + + const found = discoverTraceReplayArtifacts(root); + + expect(found.get(`${suffix}|96`)?.profileJsonl).toContain( + `agentic_${suffix}/conc_96/aiperf_artifacts`, + ); + expect(fs.existsSync(path.join(artifactDir, 'multinode_server_logs'))).toBe(false); + }); + + it('retains the archive fallback when the direct trace is incomplete', () => { + const root = tempDir(); + const suffix = 'config-d_conc96_variant'; + const directDir = path.join(root, `agentic_${suffix}`, 'conc_96'); + writeTraceFiles(directDir); + fs.rmSync(path.join(directDir, 'aiperf_artifacts', 'server_metrics_export.json')); + + const artifactDir = path.join(root, `multinode_server_logs_${suffix}`); + const archiveSource = path.join(root, 'fallback-archive-source'); + writeTraceFiles(path.join(archiveSource, 'agentic', 'conc_96')); + fs.mkdirSync(artifactDir, { recursive: true }); + execFileSync('tar', [ + '-czf', + path.join(artifactDir, 'multinode_server_logs.tar.gz'), + '-C', + archiveSource, + '.', + ]); + fs.rmSync(archiveSource, { recursive: true, force: true }); + + const found = discoverTraceReplayArtifacts(root); + + expect(found.get(`${suffix}|96`)?.serverMetricsJson).toContain( + 'multinode_server_logs/agentic/conc_96/aiperf_artifacts/server_metrics_export.json', + ); + }); }); diff --git a/packages/db/src/etl/trace-artifact-discovery.ts b/packages/db/src/etl/trace-artifact-discovery.ts index 71ee74dff..58030305b 100644 --- a/packages/db/src/etl/trace-artifact-discovery.ts +++ b/packages/db/src/etl/trace-artifact-discovery.ts @@ -13,6 +13,7 @@ const TRACE_SUBDIRS = ['aiperf_artifacts', 'trace_replay']; const AGENTIC_PREFIX = /^agentic_/u; const MULTINODE_PREFIX = /^multinode_server_logs_/u; const CONC_DIR_PATTERN = /^conc_(?\d+)$/u; +const SUFFIX_CONC_PATTERN = /(?:^|_)conc(?\d+)(?:_|$)/u; function traceFilesIn(dir: string): TraceReplayArtifactPaths | null { let profileJsonl: string | null = null; @@ -51,6 +52,7 @@ function extractMultinodeArchive(artifactDir: string): string | null { * Discover trace-replay siblings in both artifact layouts: * * - Single-node: `agentic_/aiperf_artifacts/*` + * - Direct multinode upload: `agentic_/conc_/aiperf_artifacts/*` * - Multinode: `multinode_server_logs_/multinode_server_logs.tar.gz`, * containing `agentic/conc_/aiperf_artifacts/*` * @@ -63,24 +65,47 @@ export function discoverTraceReplayArtifacts( const discovered = new Map(); if (!fs.existsSync(artifactsDir)) return discovered; - for (const entry of fs.readdirSync(artifactsDir)) { + const entries = fs.readdirSync(artifactsDir); + + // Current multinode jobs upload the same completed aiperf directory twice: + // directly as `agentic_*`, and inside the server-log tarball. Index complete + // direct copies first so matching archives do not expand a second GiB-scale + // copy. Incomplete/legacy direct artifacts still fall through to the archive. + for (const entry of entries) { + if (!entry.startsWith('agentic_')) continue; const artifactDir = path.join(artifactsDir, entry); if (!fs.statSync(artifactDir).isDirectory()) continue; + const suffix = entry.replace(AGENTIC_PREFIX, ''); + const trace = traceFilesIn(artifactDir); + if (trace) discovered.set(suffix, trace); - if (entry.startsWith('agentic_')) { - const trace = traceFilesIn(artifactDir); - if (trace) discovered.set(entry.replace(AGENTIC_PREFIX, ''), trace); - continue; + for (const concEntry of fs.readdirSync(artifactDir)) { + const match = concEntry.match(CONC_DIR_PATTERN); + if (!match?.groups?.conc) continue; + const concTrace = traceFilesIn(path.join(artifactDir, concEntry)); + if (concTrace) discovered.set(`${suffix}|${match.groups.conc}`, concTrace); } + } + for (const entry of entries) { if (!entry.startsWith('multinode_server_logs_')) continue; + const suffix = entry.replace(MULTINODE_PREFIX, ''); + const suffixConc = suffix.match(SUFFIX_CONC_PATTERN)?.groups?.conc; + const direct = + (suffixConc ? discovered.get(`${suffix}|${suffixConc}`) : undefined) ?? + discovered.get(suffix); + if (suffixConc && direct?.profileJsonl && direct.serverMetricsCsv && direct.serverMetricsJson) { + continue; + } + + const artifactDir = path.join(artifactsDir, entry); + if (!fs.statSync(artifactDir).isDirectory()) continue; const extractedDir = extractMultinodeArchive(artifactDir); if (!extractedDir) continue; const agenticDir = path.join(extractedDir, 'agentic'); if (!fs.existsSync(agenticDir) || !fs.statSync(agenticDir).isDirectory()) continue; - const suffix = entry.replace(MULTINODE_PREFIX, ''); for (const concEntry of fs.readdirSync(agenticDir)) { const match = concEntry.match(CONC_DIR_PATTERN); if (!match?.groups?.conc) continue; diff --git a/packages/db/src/etl/trace-replay-ingest.ts b/packages/db/src/etl/trace-replay-ingest.ts index 0b8781922..5e6fa65e5 100644 --- a/packages/db/src/etl/trace-replay-ingest.ts +++ b/packages/db/src/etl/trace-replay-ingest.ts @@ -14,9 +14,7 @@ import { createGzip, gzipSync } from 'node:zlib'; import type postgres from 'postgres'; -import { computeAggregateStats } from './compute-aggregate-stats.js'; -import { computeChartSeries } from './compute-chart-series.js'; -import { computeRequestTimeline } from './compute-request-timeline.js'; +import { computeTraceDerivedPayloads } from './compute-trace-derived.js'; import type { ServerMetricsContext } from './server-metrics-adapters'; type Sql = ReturnType; @@ -195,11 +193,11 @@ export async function insertTraceReplay( // a streaming parser for oversized server_metrics blobs. const computeStart = Date.now(); log('computing aggregate stats, chart series, and request timeline'); - const [aggregateStats, chartSeries, requestTimeline] = await Promise.all([ - computeAggregateStats({ profileBlob: profileGz, serverBlob: metricsJsonGz }), - computeChartSeries(metricsJsonGz, metricsContext), - Promise.resolve(computeRequestTimeline(profileGz)), - ]); + const { aggregateStats, chartSeries, requestTimeline } = await computeTraceDerivedPayloads( + profileGz, + metricsJsonGz, + metricsContext, + ); log( `computed derived JSON: chart_windows=${chartSeries?.timeslicesCount ?? 0}, ` + `timeline_requests=${requestTimeline?.requests.length ?? 0} (${elapsed(computeStart)})`,