Skip to content

Commit a7cadd6

Browse files
codexcodex
committed
perf(control-plane): optimize long-range analytics
1 parent e05a455 commit a7cadd6

6 files changed

Lines changed: 449 additions & 253 deletions

File tree

‎services/control-plane/index.ts‎

Lines changed: 102 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -28,16 +28,11 @@ import { selectRegions } from './src/regions';
2828
import { SessionService } from './src/sessions';
2929
import { buildSessionAllocationEvent, buildSessionAnalyticsMetadata, normalizeViewerRttSummary } from './src/session-analytics';
3030
import {
31-
getActiveSessionCount,
32-
getSessionsByRegion,
33-
getSessionAllocationStats,
34-
getSessionRttStats,
31+
getActiveSessionStats,
32+
getAnalyticsTimeSeries,
33+
getSessionDimensions,
34+
getSessionOverviewStats,
3535
getSessionRttStatsByRegion,
36-
getSessionTimeSeries,
37-
getSessionWindowStats,
38-
getViewerRttTimeSeries,
39-
getStaleActiveSessionCount,
40-
getTopClients,
4136
} from './src/stats';
4237
import { expiresAtFromTtlSeconds, extendExpiresAt, readOptionalSeconds, validateTtlSeconds } from './src/ttl';
4338
import { X402PaymentGateway } from './src/x402-payment';
@@ -775,7 +770,7 @@ function normalizeWindowHours(raw: string | number | undefined): number {
775770
}
776771

777772
// Pick a trend-chart bucket count so longer ranges keep useful granularity
778-
// without over-crowding the x-axis (getSessionTimeSeries caps at 48).
773+
// without over-crowding the x-axis (getAnalyticsTimeSeries caps at 48).
779774
function bucketsForWindow(windowHours: number): number {
780775
if (windowHours <= 24) return 12; // 1h→5m, 6h→30m, 24h→2h
781776
if (windowHours <= 72) return 12; // 2d→4h, 3d→6h
@@ -784,6 +779,87 @@ function bucketsForWindow(windowHours: number): number {
784779
return 15; // 30d→2d
785780
}
786781

782+
type HistoricalSessionAnalytics = {
783+
overview: Awaited<ReturnType<typeof getSessionOverviewStats>>;
784+
rttByRegion: Awaited<ReturnType<typeof getSessionRttStatsByRegion>>;
785+
timeSeries: Awaited<ReturnType<typeof getAnalyticsTimeSeries>>;
786+
dimensions: Awaited<ReturnType<typeof getSessionDimensions>>;
787+
};
788+
789+
type HistoricalSessionAnalyticsCacheEntry = {
790+
value?: HistoricalSessionAnalytics;
791+
expiresAt: number;
792+
inFlight?: Promise<HistoricalSessionAnalytics>;
793+
};
794+
795+
const HISTORICAL_ANALYTICS_CACHE_MAX_ENTRIES = 16;
796+
const historicalAnalyticsCache = new Map<number, HistoricalSessionAnalyticsCacheEntry>();
797+
798+
function historicalAnalyticsTtlMs(windowHours: number) {
799+
if (windowHours >= 168) return 60_000;
800+
if (windowHours >= 24) return 30_000;
801+
return 15_000;
802+
}
803+
804+
async function loadHistoricalSessionAnalytics(windowHours: number): Promise<HistoricalSessionAnalytics> {
805+
// Keep the two most expensive percentile phases separate to bound peak
806+
// memory on PostgreSQL. Queries inside each phase reuse covering indexes.
807+
const [overview, rttByRegion] = await Promise.all([
808+
getSessionOverviewStats(windowHours),
809+
getSessionRttStatsByRegion(windowHours),
810+
]);
811+
const [timeSeries, dimensions] = await Promise.all([
812+
getAnalyticsTimeSeries(windowHours, bucketsForWindow(windowHours)),
813+
getSessionDimensions(windowHours),
814+
]);
815+
return { overview, rttByRegion, timeSeries, dimensions };
816+
}
817+
818+
function startHistoricalAnalyticsRefresh(
819+
windowHours: number,
820+
entry: HistoricalSessionAnalyticsCacheEntry,
821+
): Promise<HistoricalSessionAnalytics> {
822+
const inFlight = loadHistoricalSessionAnalytics(windowHours)
823+
.then((value) => {
824+
historicalAnalyticsCache.delete(windowHours);
825+
historicalAnalyticsCache.set(windowHours, {
826+
value,
827+
expiresAt: Date.now() + historicalAnalyticsTtlMs(windowHours),
828+
});
829+
while (historicalAnalyticsCache.size > HISTORICAL_ANALYTICS_CACHE_MAX_ENTRIES) {
830+
const oldestKey = historicalAnalyticsCache.keys().next().value;
831+
if (oldestKey === undefined) break;
832+
historicalAnalyticsCache.delete(oldestKey);
833+
}
834+
return value;
835+
})
836+
.catch((error) => {
837+
if (entry.value) {
838+
entry.inFlight = undefined;
839+
historicalAnalyticsCache.set(windowHours, entry);
840+
} else {
841+
historicalAnalyticsCache.delete(windowHours);
842+
}
843+
throw error;
844+
});
845+
entry.inFlight = inFlight;
846+
historicalAnalyticsCache.set(windowHours, entry);
847+
return inFlight;
848+
}
849+
850+
async function getHistoricalSessionAnalytics(windowHours: number): Promise<HistoricalSessionAnalytics> {
851+
const entry = historicalAnalyticsCache.get(windowHours) ?? { expiresAt: 0 };
852+
if (entry.value && entry.expiresAt > Date.now()) return entry.value;
853+
if (entry.value) {
854+
if (!entry.inFlight) {
855+
void startHistoricalAnalyticsRefresh(windowHours, entry)
856+
.catch((error) => console.error('Failed to refresh historical analytics cache:', error));
857+
}
858+
return entry.value;
859+
}
860+
return entry.inFlight ?? startHistoricalAnalyticsRefresh(windowHours, entry);
861+
}
862+
787863
// Combines live Agones gauges (via pool managers) with cumulative Postgres
788864
// session stats into a single payload shared by the API and the admin UI.
789865
// USDC and the other supported x402 settlement assets use 6 decimals; the
@@ -803,19 +879,24 @@ async function buildX402AnalyticsPayload(windowHours: number): Promise<X402Analy
803879
}
804880

805881
async function buildStatsPayload(windowHours: number): Promise<AnalyticsData> {
806-
const [regions, windowStats, allocationStats, rttStats, rttByRegion, rttSeries, activeSessions, staleActiveSessions, series, regionSessions, topClients] = await Promise.all([
882+
// Live fleet state remains uncached. Historical aggregates use a short,
883+
// bounded stale-while-refresh cache so manual refreshes do not repeatedly
884+
// sort the same 30-day percentile data.
885+
const [regions, historical, activeSessionStats] = await Promise.all([
807886
loadAdminRegions(),
808-
getSessionWindowStats(windowHours),
809-
getSessionAllocationStats(windowHours),
810-
getSessionRttStats(windowHours),
811-
getSessionRttStatsByRegion(windowHours),
812-
getViewerRttTimeSeries(windowHours, bucketsForWindow(windowHours)),
813-
getActiveSessionCount(),
814-
getStaleActiveSessionCount(),
815-
getSessionTimeSeries(windowHours, bucketsForWindow(windowHours)),
816-
getSessionsByRegion(windowHours),
817-
getTopClients(windowHours),
887+
getHistoricalSessionAnalytics(windowHours),
888+
getActiveSessionStats(),
818889
]);
890+
const { overview, rttByRegion, timeSeries, dimensions } = historical;
891+
const windowStats = overview.window;
892+
const allocationStats = overview.allocation;
893+
const rttStats = overview.viewerRtt;
894+
const series = timeSeries.sessions;
895+
const rttSeries = timeSeries.viewerRtt;
896+
const activeSessions = activeSessionStats.active;
897+
const staleActiveSessions = activeSessionStats.stale;
898+
const regionSessions = dimensions.byRegion;
899+
const topClients = dimensions.topClients;
819900

820901
const enabledRegions = regions.filter((region) => region.enabled);
821902
const servers = enabledRegions.flatMap((region) => region.servers || []);

‎services/control-plane/migrations/0007_add_session_analytics_projections.sql‎

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,66 @@
1+
-- Keep index construction comfortably below the standalone database's legacy
2+
-- 512 MiB limit; the setting is transaction-local to this migration job.
3+
SET LOCAL maintenance_work_mem = '16MB';
4+
--> statement-breakpoint
5+
ALTER TABLE "sessions"
6+
ADD COLUMN IF NOT EXISTS "allocation_latency_ms" double precision
7+
GENERATED ALWAYS AS (
8+
CASE
9+
WHEN jsonb_typeof("metadata"->'allocationLatencyMs') = 'number'
10+
THEN ("metadata"->>'allocationLatencyMs')::double precision
11+
END
12+
) STORED,
13+
ADD COLUMN IF NOT EXISTS "viewer_rtt_avg_ms" double precision
14+
GENERATED ALWAYS AS (
15+
CASE
16+
WHEN jsonb_typeof("metadata"->'viewerRtt') = 'object'
17+
AND jsonb_typeof("metadata"->'viewerRtt'->'avgMs') = 'number'
18+
THEN ("metadata"->'viewerRtt'->>'avgMs')::double precision
19+
END
20+
) STORED,
21+
ADD COLUMN IF NOT EXISTS "viewer_rtt_p50_ms" double precision
22+
GENERATED ALWAYS AS (
23+
CASE
24+
WHEN jsonb_typeof("metadata"->'viewerRtt') = 'object'
25+
AND jsonb_typeof("metadata"->'viewerRtt'->'avgMs') = 'number'
26+
THEN ("metadata"->'viewerRtt'->>'p50Ms')::double precision
27+
END
28+
) STORED,
29+
ADD COLUMN IF NOT EXISTS "viewer_rtt_p95_ms" double precision
30+
GENERATED ALWAYS AS (
31+
CASE
32+
WHEN jsonb_typeof("metadata"->'viewerRtt') = 'object'
33+
AND jsonb_typeof("metadata"->'viewerRtt'->'avgMs') = 'number'
34+
THEN ("metadata"->'viewerRtt'->>'p95Ms')::double precision
35+
END
36+
) STORED,
37+
ADD COLUMN IF NOT EXISTS "viewer_rtt_sample_count" double precision
38+
GENERATED ALWAYS AS (
39+
CASE
40+
WHEN jsonb_typeof("metadata"->'viewerRtt') = 'object'
41+
AND jsonb_typeof("metadata"->'viewerRtt'->'avgMs') = 'number'
42+
THEN ("metadata"->'viewerRtt'->>'sampleCount')::double precision
43+
END
44+
) STORED;
45+
--> statement-breakpoint
46+
CREATE INDEX IF NOT EXISTS "sessions_created_analytics_idx"
47+
ON "sessions" (
48+
"created_at",
49+
"region",
50+
"client_name",
51+
"allocation_latency_ms",
52+
"viewer_rtt_avg_ms",
53+
"viewer_rtt_p50_ms",
54+
"viewer_rtt_p95_ms",
55+
"viewer_rtt_sample_count"
56+
);
57+
--> statement-breakpoint
58+
CREATE INDEX IF NOT EXISTS "sessions_ended_analytics_idx"
59+
ON "sessions" (
60+
"ended_at",
61+
"created_at",
62+
"status",
63+
"viewer_rtt_avg_ms",
64+
"viewer_rtt_p50_ms",
65+
"viewer_rtt_p95_ms"
66+
);

‎services/control-plane/migrations/meta/_journal.json‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,13 @@
5050
"when": 1785843000000,
5151
"tag": "0006_replace_x402_management_token",
5252
"breakpoints": true
53+
},
54+
{
55+
"idx": 7,
56+
"version": "7",
57+
"when": 1788105600000,
58+
"tag": "0007_add_session_analytics_projections",
59+
"breakpoints": true
5360
}
5461
]
5562
}

‎services/control-plane/src/schema.ts‎

Lines changed: 59 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
import { sql } from 'drizzle-orm';
2-
import { pgTable, uuid, text, timestamp, boolean, jsonb, index, integer, bigint, uniqueIndex, check } from 'drizzle-orm/pg-core';
2+
import { pgTable, uuid, text, timestamp, boolean, jsonb, index, integer, bigint, doublePrecision, uniqueIndex, check } from 'drizzle-orm/pg-core';
33

44
export const clients = pgTable('clients', {
55
id: text('id').primaryKey(), // e.g., "client_abc123"
@@ -29,6 +29,44 @@ export const sessions = pgTable('sessions', {
2929

3030
// Additional metadata
3131
metadata: jsonb('metadata'),
32+
33+
// Frequently aggregated values are projected once when metadata changes.
34+
// Keeping them as generated columns avoids repeatedly decoding the much
35+
// wider JSON document during analytics reads.
36+
allocationLatencyMs: doublePrecision('allocation_latency_ms').generatedAlwaysAs(sql`
37+
CASE
38+
WHEN jsonb_typeof(metadata->'allocationLatencyMs') = 'number'
39+
THEN (metadata->>'allocationLatencyMs')::double precision
40+
END
41+
`),
42+
viewerRttAvgMs: doublePrecision('viewer_rtt_avg_ms').generatedAlwaysAs(sql`
43+
CASE
44+
WHEN jsonb_typeof(metadata->'viewerRtt') = 'object'
45+
AND jsonb_typeof(metadata->'viewerRtt'->'avgMs') = 'number'
46+
THEN (metadata->'viewerRtt'->>'avgMs')::double precision
47+
END
48+
`),
49+
viewerRttP50Ms: doublePrecision('viewer_rtt_p50_ms').generatedAlwaysAs(sql`
50+
CASE
51+
WHEN jsonb_typeof(metadata->'viewerRtt') = 'object'
52+
AND jsonb_typeof(metadata->'viewerRtt'->'avgMs') = 'number'
53+
THEN (metadata->'viewerRtt'->>'p50Ms')::double precision
54+
END
55+
`),
56+
viewerRttP95Ms: doublePrecision('viewer_rtt_p95_ms').generatedAlwaysAs(sql`
57+
CASE
58+
WHEN jsonb_typeof(metadata->'viewerRtt') = 'object'
59+
AND jsonb_typeof(metadata->'viewerRtt'->'avgMs') = 'number'
60+
THEN (metadata->'viewerRtt'->>'p95Ms')::double precision
61+
END
62+
`),
63+
viewerRttSampleCount: doublePrecision('viewer_rtt_sample_count').generatedAlwaysAs(sql`
64+
CASE
65+
WHEN jsonb_typeof(metadata->'viewerRtt') = 'object'
66+
AND jsonb_typeof(metadata->'viewerRtt'->'avgMs') = 'number'
67+
THEN (metadata->'viewerRtt'->>'sampleCount')::double precision
68+
END
69+
`),
3270
}, (table) => ({
3371
// Indexes for common query patterns
3472
clientIdx: index('sessions_client_idx').on(table.clientId),
@@ -41,6 +79,26 @@ export const sessions = pgTable('sessions', {
4179
// Composite indexes for common query combinations
4280
clientTimeIdx: index('sessions_client_time_idx').on(table.clientId, table.createdAt),
4381
clusterTimeIdx: index('sessions_cluster_time_idx').on(table.clusterName, table.createdAt),
82+
// Cover the full historical-dashboard projections so old, all-visible rows
83+
// can be answered without fetching the 222 MB sessions heap.
84+
createdAnalyticsIdx: index('sessions_created_analytics_idx').on(
85+
table.createdAt,
86+
table.region,
87+
table.clientName,
88+
table.allocationLatencyMs,
89+
table.viewerRttAvgMs,
90+
table.viewerRttP50Ms,
91+
table.viewerRttP95Ms,
92+
table.viewerRttSampleCount,
93+
),
94+
endedAnalyticsIdx: index('sessions_ended_analytics_idx').on(
95+
table.endedAt,
96+
table.createdAt,
97+
table.status,
98+
table.viewerRttAvgMs,
99+
table.viewerRttP50Ms,
100+
table.viewerRttP95Ms,
101+
),
44102
}));
45103

46104
// Event audit log - for detailed tracking (optional, can be disabled for performance)

‎services/control-plane/src/sessions.ts‎

Lines changed: 17 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,20 @@ import { db } from './db';
22
import { sessions, sessionEvents } from './schema';
33
import { and, count, desc, eq, isNull, sql } from 'drizzle-orm';
44

5+
// Generated analytics projections are intentionally internal to aggregate
6+
// queries and must not expand the existing session API response shape.
7+
const publicSessionColumns = {
8+
sessionId: sessions.sessionId,
9+
clientId: sessions.clientId,
10+
clientName: sessions.clientName,
11+
clusterName: sessions.clusterName,
12+
region: sessions.region,
13+
createdAt: sessions.createdAt,
14+
endedAt: sessions.endedAt,
15+
status: sessions.status,
16+
metadata: sessions.metadata,
17+
};
18+
519
export const SessionService = {
620
// Create a new session
721
async createSession(sessionId: string, clientId: string, clientName: string, clusterName: string, region?: string, metadata?: Record<string, unknown>): Promise<void> {
@@ -89,20 +103,20 @@ export const SessionService = {
89103

90104
// Get session info
91105
async getSession(sessionId: string) {
92-
return await db.select().from(sessions).where(eq(sessions.sessionId, sessionId)).limit(1);
106+
return await db.select(publicSessionColumns).from(sessions).where(eq(sessions.sessionId, sessionId)).limit(1);
93107
},
94108

95109
async listSessions(limit = 100, clientId?: string, offset = 0) {
96110
if (clientId) {
97-
return await db.select()
111+
return await db.select(publicSessionColumns)
98112
.from(sessions)
99113
.where(eq(sessions.clientId, clientId))
100114
.orderBy(desc(sessions.createdAt))
101115
.limit(limit)
102116
.offset(offset);
103117
}
104118

105-
return await db.select()
119+
return await db.select(publicSessionColumns)
106120
.from(sessions)
107121
.orderBy(desc(sessions.createdAt))
108122
.limit(limit)

0 commit comments

Comments
 (0)