Skip to content

Commit c28b6cf

Browse files
ericallamericallam
committed
feat(webapp): serve queue metrics reads from rollups and fix env totals
The built-in queues dashboard's enqueued vs started chart merged counter states across queues, which mixes unrelated cumulative counters and returns wrong totals; it now merges per queue and sums outside. Env header tiles and saturation charts read the environment rollup, so their cost no longer scales with queue count, and coarse-bucket ranges are served from the 5m rollup automatically. Queue list ranking runs as one query, time bounds are aligned to the bucket grid, and repeated auto-refresh reads share ClickHouse query-cache entries.
1 parent d60696c commit c28b6cf

6 files changed

Lines changed: 240 additions & 46 deletions

File tree

‎apps/webapp/app/presenters/v3/BuiltInDashboards.server.ts‎

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -575,22 +575,22 @@ const queuesDashboard: BuiltInDashboard = {
575575
widgets: {
576576
"env-used": {
577577
title: "Concurrency in use",
578-
query: `SELECT argMax(max_env_running, bucket_start) AS in_use\nFROM queue_metrics`,
578+
query: `SELECT argMax(max_env_running, bucket_start) AS in_use\nFROM env_metrics`,
579579
display: { type: "bignumber", column: "in_use", aggregation: "max", abbreviate: false },
580580
},
581581
"env-limit": {
582582
title: "Environment limit",
583-
query: `SELECT argMax(max_env_limit, bucket_start) AS env_limit\nFROM queue_metrics`,
583+
query: `SELECT argMax(max_env_limit, bucket_start) AS env_limit\nFROM env_metrics`,
584584
display: { type: "bignumber", column: "env_limit", aggregation: "max", abbreviate: false },
585585
},
586586
"env-avail": {
587587
title: "Available slots",
588-
query: `SELECT argMax(max_env_limit, bucket_start) - argMax(max_env_running, bucket_start) AS available\nFROM queue_metrics`,
588+
query: `SELECT argMax(max_env_limit, bucket_start) - argMax(max_env_running, bucket_start) AS available\nFROM env_metrics`,
589589
display: { type: "bignumber", column: "available", aggregation: "max", abbreviate: false },
590590
},
591591
"env-sat": {
592592
title: "Env saturation",
593-
query: `SELECT round(argMax(max_env_running, bucket_start) * 100.0 / nullIf(argMax(max_env_limit, bucket_start), 0), 1) AS saturation\nFROM queue_metrics`,
593+
query: `SELECT round(argMax(max_env_running, bucket_start) * 100.0 / nullIf(argMax(max_env_limit, bucket_start), 0), 1) AS saturation\nFROM env_metrics`,
594594
display: {
595595
type: "bignumber",
596596
column: "saturation",
@@ -601,7 +601,7 @@ const queuesDashboard: BuiltInDashboard = {
601601
},
602602
"sat-time": {
603603
title: "Environment saturation over time",
604-
query: `SELECT timeBucket() AS t,\n round(max(max_env_running) * 100.0 / nullIf(max(max_env_limit), 0), 1) AS saturation\nFROM queue_metrics\nGROUP BY t\nORDER BY t`,
604+
query: `SELECT timeBucket() AS t,\n round(max(max_env_running) * 100.0 / nullIf(max(max_env_limit), 0), 1) AS saturation\nFROM env_metrics\nGROUP BY t\nORDER BY t`,
605605
display: {
606606
type: "chart",
607607
chartType: "line",
@@ -616,7 +616,7 @@ const queuesDashboard: BuiltInDashboard = {
616616
},
617617
"used-limit": {
618618
title: "Concurrency used vs limit",
619-
query: `SELECT timeBucket() AS t,\n max(max_env_running) AS used,\n max(max_env_limit) AS limit\nFROM queue_metrics\nGROUP BY t\nORDER BY t`,
619+
query: `SELECT timeBucket() AS t,\n max(max_env_running) AS used,\n max(max_env_limit) AS limit\nFROM env_metrics\nGROUP BY t\nORDER BY t`,
620620
// Single-series gauge: carry the last known used/limit across idle buckets instead of dropping to 0.
621621
fillGaps: true,
622622
display: {
@@ -695,9 +695,9 @@ const queuesDashboard: BuiltInDashboard = {
695695
},
696696
throughput: {
697697
title: "Enqueued vs started",
698-
query: `SELECT timeBucket() AS t,\n deltaSumTimestampMerge(enqueue_delta) AS enqueued,\n deltaSumTimestampMerge(started_delta) AS started\nFROM queue_metrics\nGROUP BY t\nORDER BY t`,
699-
// Single-series counters: zero-fill idle buckets so the line returns to 0 rather than interpolating across gaps.
700-
fillGaps: true,
698+
// Counter states merge per queue, then sum outside: a single merge across queues
699+
// mixes unrelated odometers and returns wrong totals.
700+
query: `SELECT t, sum(enq) AS enqueued, sum(st) AS started\nFROM (\n SELECT timeBucket() AS t, queue,\n deltaSumTimestampMerge(enqueue_delta) AS enq,\n deltaSumTimestampMerge(started_delta) AS st\n FROM queue_metrics\n GROUP BY t, queue\n)\nGROUP BY t\nORDER BY t`,
701701
display: {
702702
type: "chart",
703703
chartType: "line",
@@ -712,7 +712,7 @@ const queuesDashboard: BuiltInDashboard = {
712712
},
713713
"wait-pct": {
714714
title: "Scheduling delay p50/p95/p99 (ms)",
715-
query: `SELECT timeBucket() AS t,\n round(quantilesMerge(0.5, 0.95, 0.99)(wait_quantiles)[1]) AS p50,\n round(quantilesMerge(0.5, 0.95, 0.99)(wait_quantiles)[2]) AS p95,\n round(quantilesMerge(0.5, 0.95, 0.99)(wait_quantiles)[3]) AS p99\nFROM queue_metrics\nGROUP BY t\nORDER BY t`,
715+
query: `SELECT timeBucket() AS t,\n round(quantilesTDigestMerge(0.5, 0.95, 0.99)(wait_quantiles)[1]) AS p50,\n round(quantilesTDigestMerge(0.5, 0.95, 0.99)(wait_quantiles)[2]) AS p95,\n round(quantilesTDigestMerge(0.5, 0.95, 0.99)(wait_quantiles)[3]) AS p99\nFROM env_metrics\nGROUP BY t\nORDER BY t`,
716716
display: {
717717
type: "chart",
718718
chartType: "line",

‎apps/webapp/app/presenters/v3/QueueListPresenter.server.ts‎

Lines changed: 28 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -217,40 +217,49 @@ export class QueueListPresenter extends BasePresenter {
217217
"query"
218218
);
219219

220+
// The window start is aligned to the minute so repeated page loads produce identical
221+
// query text and can share ClickHouse query-cache entries.
222+
const windowStartMs =
223+
Math.floor((Date.now() - QUEUE_RANKING_WINDOW_MINUTES * 60 * 1000) / 60_000) * 60_000;
220224
const rankingArgs = {
221225
organizationId: environment.organizationId,
222226
projectId: environment.projectId,
223227
environmentId: environment.id,
224-
startTime: formatClickhouseDateTime(
225-
new Date(Date.now() - QUEUE_RANKING_WINDOW_MINUTES * 60 * 1000)
226-
),
228+
startTime: formatClickhouseDateTime(new Date(windowStartMs)),
227229
nameContains: query?.trim() ?? "",
228230
};
229231

230-
const [countError, countRows] = await clickhouse.queueMetrics.rankingCount(rankingArgs);
231-
if (countError) {
232-
throw countError;
232+
const offset = (page - 1) * this.perPage;
233+
234+
// One scan returns the page and the total ranked count (window function).
235+
const [pageError, pageRows] = await clickhouse.queueMetrics.ranking({
236+
...rankingArgs,
237+
byQueuedOnly: sort === "queued" ? 1 : 0,
238+
limit: this.perPage,
239+
offset,
240+
});
241+
if (pageError) {
242+
throw pageError;
243+
}
244+
245+
let ranked = pageRows?.[0]?.ranked_total ?? 0;
246+
if (ranked === 0 && offset > 0) {
247+
// Empty page past the ranked head: fetch the count alone for the tail slot math.
248+
const [countError, countRows] = await clickhouse.queueMetrics.rankingCount(rankingArgs);
249+
if (countError) {
250+
throw countError;
251+
}
252+
ranked = countRows?.[0]?.ranked ?? 0;
233253
}
234-
const ranked = countRows?.[0]?.ranked ?? 0;
235254
if (ranked > MAX_RANKED_QUEUES) {
236255
return null;
237256
}
238257

239258
const where = buildQueueListWhere(environment.id, query, type);
240259
const totalQueues = await this._replica.taskQueue.count({ where });
241-
const offset = (page - 1) * this.perPage;
242260

243261
let rankedPageQueues: QueueListRow[] = [];
244-
if (offset < ranked) {
245-
const [pageError, pageRows] = await clickhouse.queueMetrics.rankingPage({
246-
...rankingArgs,
247-
byQueuedOnly: sort === "queued" ? 1 : 0,
248-
limit: this.perPage,
249-
offset,
250-
});
251-
if (pageError) {
252-
throw pageError;
253-
}
262+
if ((pageRows?.length ?? 0) > 0) {
254263
const rankedNames = (pageRows ?? []).map((row) => row.queue_name);
255264
rankedPageQueues = await this.findQueuesByNames(where, rankedNames);
256265
}
@@ -263,11 +272,9 @@ export class QueueListPresenter extends BasePresenter {
263272
if (tailNeeded > 0) {
264273
let excludedNames: string[] = [];
265274
if (ranked > 0) {
266-
const [allError, allRows] = await clickhouse.queueMetrics.rankingPage({
275+
const [allError, allRows] = await clickhouse.queueMetrics.rankingNames({
267276
...rankingArgs,
268-
byQueuedOnly: 0,
269277
limit: MAX_RANKED_QUEUES,
270-
offset: 0,
271278
});
272279
if (allError) {
273280
throw allError;

‎apps/webapp/app/presenters/v3/QueueMetricsPresenter.server.ts‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -67,13 +67,16 @@ export class QueueMetricsPresenter {
6767
"query"
6868
);
6969

70+
// End bound snaps up to the bucket grid so repeated loads within a bucket produce
71+
// identical params and share ClickHouse query-cache entries.
72+
const endMs = Math.ceil(to.getTime() / bucketIntervalMs) * bucketIntervalMs;
7073
const ids = {
7174
organizationId: environment.organizationId,
7275
projectId: environment.projectId,
7376
environmentId: environment.id,
7477
queueNames,
7578
startTime: formatClickhouseDateTime(new Date(bucketStartMs)),
76-
endTime: formatClickhouseDateTime(to),
79+
endTime: formatClickhouseDateTime(new Date(endMs)),
7780
};
7881

7982
const [summaryResult, sparklineResult] = await Promise.all([

‎apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1327,7 +1327,7 @@ const QUEUE_HEADER_TILES: QueueHeaderTile[] = [
13271327
id: "saturation",
13281328
label: "Env saturation",
13291329
color: "#6366F1",
1330-
query: `SELECT timeBucket() AS t,\n max(max_env_running) AS used,\n max(max_env_limit) AS env_limit\nFROM queue_metrics\nGROUP BY t\nORDER BY t`,
1330+
query: `SELECT timeBucket() AS t,\n max(max_env_running) AS used,\n max(max_env_limit) AS env_limit\nFROM env_metrics\nGROUP BY t\nORDER BY t`,
13311331
formatValue: (v) => `${v}%`,
13321332
derive: (rows) => {
13331333
const sparkline = rows.map((r) => {
@@ -1342,7 +1342,7 @@ const QUEUE_HEADER_TILES: QueueHeaderTile[] = [
13421342
id: "backlog",
13431343
label: "Backlog",
13441344
color: "#A78BFA",
1345-
query: `SELECT timeBucket() AS t,\n max(max_env_queued) AS queued\nFROM queue_metrics\nGROUP BY t\nORDER BY t`,
1345+
query: `SELECT timeBucket() AS t,\n max(max_env_queued) AS queued\nFROM env_metrics\nGROUP BY t\nORDER BY t`,
13461346
derive: (rows) => {
13471347
const sparkline = rows.map((r) => tileNumber(r.queued));
13481348
const peak = sparkline.reduce((max, v) => Math.max(max, v), 0);
@@ -1353,7 +1353,7 @@ const QUEUE_HEADER_TILES: QueueHeaderTile[] = [
13531353
id: "p95",
13541354
label: "Scheduling delay p95",
13551355
color: "#F59E0B",
1356-
query: `SELECT timeBucket() AS t,\n round(quantilesMerge(0.5, 0.95, 0.99)(wait_quantiles)[2]) AS p95\nFROM queue_metrics\nGROUP BY t\nORDER BY t`,
1356+
query: `SELECT timeBucket() AS t,\n round(quantilesTDigestMerge(0.5, 0.95, 0.99)(wait_quantiles)[2]) AS p95\nFROM env_metrics\nGROUP BY t\nORDER BY t`,
13571357
formatValue: formatWaitMs,
13581358
derive: (rows) => {
13591359
const sparkline = rows.map((r) => tileNumber(r.p95));
@@ -1370,7 +1370,7 @@ const QUEUE_HEADER_TILES: QueueHeaderTile[] = [
13701370
id: "throttled",
13711371
label: "Throttled",
13721372
color: "#F59E0B",
1373-
query: `SELECT timeBucket() AS t,\n sum(throttled_count) AS throttled\nFROM queue_metrics\nGROUP BY t\nORDER BY t`,
1373+
query: `SELECT timeBucket() AS t,\n sum(throttled_count) AS throttled\nFROM env_metrics\nGROUP BY t\nORDER BY t`,
13741374
derive: (rows) => {
13751375
const sparkline = rows.map((r) => tileNumber(r.throttled));
13761376
const total = sparkline.reduce((sum, v) => sum + v, 0);

‎apps/webapp/app/services/queryService.server.ts‎

Lines changed: 71 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,12 @@ import {
77
type TSQLQueryResult,
88
} from "@internal/clickhouse";
99
import type { CustomerQuerySource } from "@trigger.dev/database";
10-
import type { TableSchema, WhereClauseCondition } from "@internal/tsql";
10+
import {
11+
calculateTimeBucketInterval,
12+
type TableSchema,
13+
type TimeBucketInterval,
14+
type WhereClauseCondition,
15+
} from "@internal/tsql";
1116
import { z } from "zod";
1217
import { prisma } from "~/db.server";
1318
import { env } from "~/env.server";
@@ -110,6 +115,41 @@ export type ExecuteQueryResult<T> =
110115
}
111116
| { success: false; error: Error };
112117

118+
const INTERVAL_UNIT_SECONDS: Record<TimeBucketInterval["unit"], number> = {
119+
SECOND: 1,
120+
MINUTE: 60,
121+
HOUR: 3_600,
122+
DAY: 86_400,
123+
WEEK: 604_800,
124+
MONTH: 2_592_000,
125+
};
126+
127+
function floorToSeconds(date: Date, alignSeconds: number): Date {
128+
const ms = alignSeconds * 1000;
129+
return new Date(Math.floor(date.getTime() / ms) * ms);
130+
}
131+
132+
/**
133+
* Swap a table for one of its rollups when the query's bucket interval is at least the
134+
* rollup's granularity. The rollup has identical logical columns, so only the physical
135+
* table (and therefore rows read) changes.
136+
*/
137+
function resolveRollup(schema: TableSchema, timeRange: { from: Date; to: Date }): TableSchema {
138+
if (!schema.rollups || schema.rollups.length === 0) {
139+
return schema;
140+
}
141+
const interval = calculateTimeBucketInterval(
142+
timeRange.from,
143+
timeRange.to,
144+
schema.timeBucketThresholds
145+
);
146+
const intervalSeconds = interval.value * INTERVAL_UNIT_SECONDS[interval.unit];
147+
const best = [...schema.rollups]
148+
.sort((a, b) => b.minIntervalSeconds - a.minIntervalSeconds)
149+
.find((r) => r.minIntervalSeconds <= intervalSeconds);
150+
return best ? { ...schema, clickhouseName: best.clickhouseName } : schema;
151+
}
152+
113153
export async function getDefaultPeriod(organizationId: string): Promise<string> {
114154
const idealDefaultPeriodDays = 7;
115155
const maxQueryPeriod = await getLimit(organizationId, "queryPeriodDays", 30);
@@ -183,6 +223,14 @@ export async function executeQuery<TOut extends z.ZodSchema>(
183223
defaultPeriod,
184224
});
185225

226+
// Align the time bounds so repeated auto-refresh queries produce identical query
227+
// params and can share ClickHouse query-cache entries (params are part of the key).
228+
const alignSeconds = matchedSchema?.queryCache?.alignSeconds;
229+
if (alignSeconds) {
230+
if (timeFilter.from) timeFilter.from = floorToSeconds(timeFilter.from, alignSeconds);
231+
if (timeFilter.to) timeFilter.to = floorToSeconds(timeFilter.to, alignSeconds);
232+
}
233+
186234
// Calculate the effective "from" date the user is requesting (for period clipping check)
187235
// This is null only when the user specifies just a "to" date (rare case)
188236
let requestedFromDate: Date | null = null;
@@ -192,6 +240,9 @@ export async function executeQuery<TOut extends z.ZodSchema>(
192240
// Period specified (or default) - calculate from now
193241
const periodMs = parse(timeFilter.period ?? defaultPeriod) ?? 7 * 24 * 60 * 60 * 1000;
194242
requestedFromDate = new Date(Date.now() - periodMs);
243+
if (alignSeconds) {
244+
requestedFromDate = floorToSeconds(requestedFromDate, alignSeconds);
245+
}
195246
}
196247

197248
// Build the fallback WHERE condition based on what the user specified
@@ -207,7 +258,10 @@ export async function executeQuery<TOut extends z.ZodSchema>(
207258
}
208259

209260
const maxQueryPeriod = await getLimit(organizationId, "queryPeriodDays", 30);
210-
const maxQueryPeriodDate = new Date(Date.now() - maxQueryPeriod * 24 * 60 * 60 * 1000);
261+
let maxQueryPeriodDate = new Date(Date.now() - maxQueryPeriod * 24 * 60 * 60 * 1000);
262+
if (alignSeconds) {
263+
maxQueryPeriodDate = floorToSeconds(maxQueryPeriodDate, alignSeconds);
264+
}
211265

212266
// Check if the requested time period exceeds the plan limit
213267
const periodClipped = requestedFromDate !== null && requestedFromDate < maxQueryPeriodDate;
@@ -255,6 +309,10 @@ export async function executeQuery<TOut extends z.ZodSchema>(
255309
to: to ?? undefined,
256310
defaultPeriod,
257311
});
312+
if (alignSeconds) {
313+
timeRange.from = floorToSeconds(timeRange.from, alignSeconds);
314+
timeRange.to = floorToSeconds(timeRange.to, alignSeconds);
315+
}
258316

259317
try {
260318
// Build field mappings for project_ref → project_id and environment_id → slug translation
@@ -277,10 +335,19 @@ export async function executeQuery<TOut extends z.ZodSchema>(
277335
organizationId,
278336
"query"
279337
);
338+
// Serve coarse-bucket queries from the table's rollup when one qualifies.
339+
const effectiveSchemas = matchedSchema?.rollups
340+
? querySchemas.map((s) => (s === matchedSchema ? resolveRollup(s, timeRange) : s))
341+
: querySchemas;
342+
343+
const queryCacheSettings: ClickHouseSettings = matchedSchema?.queryCache
344+
? { use_query_cache: 1, query_cache_ttl: matchedSchema.queryCache.ttlSeconds }
345+
: {};
346+
280347
const result = await executeTSQL(queryClickhouse.reader, {
281348
...baseOptions,
282349
schema: z.record(z.any()),
283-
tableSchema: querySchemas,
350+
tableSchema: effectiveSchemas,
284351
transformValues: true,
285352
enforcedWhereClause,
286353
fieldMappings,
@@ -290,6 +357,7 @@ export async function executeQuery<TOut extends z.ZodSchema>(
290357
timeRange,
291358
clickhouseSettings: {
292359
...getDefaultClickhouseSettings(),
360+
...queryCacheSettings,
293361
...baseOptions.clickhouseSettings, // Allow caller overrides if needed
294362
},
295363
querySettings: {

0 commit comments

Comments
 (0)