diff --git a/.changeset/queue-total-concurrency-stats.md b/.changeset/queue-total-concurrency-stats.md new file mode 100644 index 00000000000..a70da24d1fb --- /dev/null +++ b/.changeset/queue-total-concurrency-stats.md @@ -0,0 +1,5 @@ +--- +"@trigger.dev/core": patch +--- + +Queue retrieve and list API responses now report total concurrency usage. When a queue has a `totalConcurrencyLimit`, `concurrency.total` includes the effective cap, the declared base, any active override, and how many runs are in flight across all concurrency keys. diff --git a/apps/webapp/app/presenters/v3/QueueListPresenter.server.ts b/apps/webapp/app/presenters/v3/QueueListPresenter.server.ts index 0dc3daa9856..d78e98ade4c 100644 --- a/apps/webapp/app/presenters/v3/QueueListPresenter.server.ts +++ b/apps/webapp/app/presenters/v3/QueueListPresenter.server.ts @@ -9,7 +9,10 @@ import { engine } from "~/v3/runEngine.server"; import { BasePresenter } from "./basePresenter.server"; import { toQueueItem } from "./QueueRetrievePresenter.server"; -type QueueListEngine = Pick; +type QueueListEngine = Pick< + RunEngine, + "lengthOfQueues" | "currentConcurrencyOfQueues" | "totalConcurrencyOfQueues" +>; export const QUEUE_LIST_DEFAULT_ITEMS_PER_PAGE = 25; const MAX_ITEMS_PER_PAGE = 100; @@ -34,6 +37,9 @@ const queueListSelect = { concurrencyLimitOverriddenAt: true, concurrencyLimitOverriddenBy: true, concurrencyLimitOverridePercent: true, + totalConcurrencyLimit: true, + totalConcurrencyLimitBase: true, + totalConcurrencyLimitOverriddenAt: true, type: true, paused: true, } satisfies Prisma.TaskQueueSelect; @@ -333,11 +339,15 @@ export class QueueListPresenter extends BasePresenter { concurrencyLimitOverriddenAt: Date | null; concurrencyLimitOverriddenBy: string | null; concurrencyLimitOverridePercent: Prisma.Decimal | null; + totalConcurrencyLimit: number | null; + totalConcurrencyLimitBase: number | null; + totalConcurrencyLimitOverriddenAt: Date | null; type: TaskQueueType; paused: boolean; }[] ): Promise { - const [queuedByQueue, runningByQueue] = await Promise.all([ + const queuesWithTotalCap = queues.filter((q) => q.totalConcurrencyLimit !== null); + const [queuedByQueue, runningByQueue, totalRunningByQueue] = await Promise.all([ this.engineClient.lengthOfQueues( environment, queues.map((q) => q.name) @@ -346,6 +356,12 @@ export class QueueListPresenter extends BasePresenter { environment, queues.map((q) => q.name) ), + queuesWithTotalCap.length > 0 + ? this.engineClient.totalConcurrencyOfQueues( + environment, + queuesWithTotalCap.map((q) => q.name) + ) + : Promise.resolve({} as Record), ]); // Manually "join" the overridden users because there is no way to implement the relationship @@ -373,6 +389,11 @@ export class QueueListPresenter extends BasePresenter { ? (overriddenByMap.get(queue.concurrencyLimitOverriddenBy) ?? null) : null, paused: queue.paused, + totalConcurrencyLimit: queue.totalConcurrencyLimit, + totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase, + totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt, + totalRunning: + queue.totalConcurrencyLimit !== null ? (totalRunningByQueue[queue.name] ?? 0) : null, }), // Prisma returns Decimal; the client only needs a plain number (null for absolute overrides). concurrencyLimitOverridePercent: diff --git a/apps/webapp/app/presenters/v3/QueueRetrievePresenter.server.ts b/apps/webapp/app/presenters/v3/QueueRetrievePresenter.server.ts index f6918394e5c..1ce08e4c628 100644 --- a/apps/webapp/app/presenters/v3/QueueRetrievePresenter.server.ts +++ b/apps/webapp/app/presenters/v3/QueueRetrievePresenter.server.ts @@ -90,6 +90,9 @@ export class QueueRetrievePresenter extends BasePresenter { const results = await Promise.all([ engine.lengthOfQueues(environment, [queue.name]), engine.currentConcurrencyOfQueues(environment, [queue.name]), + queue.totalConcurrencyLimit != null + ? engine.totalConcurrencyOfQueues(environment, [queue.name]) + : undefined, ]); // Transform queues to include running and queued counts @@ -107,6 +110,11 @@ export class QueueRetrievePresenter extends BasePresenter { concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt ?? null, concurrencyLimitOverriddenBy: queue.concurrencyLimitOverriddenBy ?? null, paused: queue.paused, + totalConcurrencyLimit: queue.totalConcurrencyLimit ?? null, + totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase ?? null, + totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt ?? null, + totalRunning: + queue.totalConcurrencyLimit != null ? (results[2]?.[queue.name] ?? 0) : null, }), // The percent source-of-truth for percent-based overrides isn't part of the shared // `QueueItem` schema (that's a public contract), so we surface it as an extra field on @@ -148,6 +156,10 @@ export function toQueueItem(data: { concurrencyLimitOverriddenAt: Date | null; concurrencyLimitOverriddenBy: User | null; paused: boolean; + totalConcurrencyLimit?: number | null; + totalConcurrencyLimitBase?: number | null; + totalConcurrencyLimitOverriddenAt?: Date | null; + totalRunning?: number | null; }): QueueItem & { releaseConcurrencyOnWaitpoint: boolean } { return { id: data.friendlyId, @@ -164,6 +176,16 @@ export function toQueueItem(data: { override: data.concurrencyLimitOverriddenAt ? data.concurrencyLimit : null, overriddenBy: toQueueConcurrencyOverriddenBy(data.concurrencyLimitOverriddenBy), overriddenAt: data.concurrencyLimitOverriddenAt, + combined: + data.totalConcurrencyLimit !== undefined + ? { + current: data.totalConcurrencyLimit, + base: data.totalConcurrencyLimitBase ?? null, + override: data.totalConcurrencyLimitOverriddenAt ? data.totalConcurrencyLimit : null, + overriddenAt: data.totalConcurrencyLimitOverriddenAt ?? null, + running: data.totalRunning ?? null, + } + : undefined, }, // TODO: This needs to be removed but keeping this here for now to avoid breaking existing clients releaseConcurrencyOnWaitpoint: true, diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx index 4b5673c6f0e..7b10b8d6d14 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.env.$envParam.queues/route.tsx @@ -697,7 +697,14 @@ function QueuesWithMetricsView() { Name Queued Running - Limit + + Limit + 0 && "text-text-bright" + queue.concurrency?.combined?.current != null && + (queue.concurrency.combined.running ?? 0) >= + Math.min( + queue.concurrency.combined.current, + environment.concurrencyLimit + ) + ? "text-warning" + : queue.running > 0 && "text-text-bright" )} > {queue.running} @@ -854,6 +868,39 @@ function QueuesWithMetricsView() { queue.paused ? "opacity-50" : undefined, queue.concurrency?.overriddenAt && "font-medium text-text-bright" )} + // The combined-limit hint is a tooltip button, so it renders beside the + // link (trailing) rather than nested inside the ; the number stays the + // link. + trailingContent={ + queue.concurrency?.combined?.current != null ? ( + + ( + {Math.min( + queue.concurrency.combined.current, + environment.concurrencyLimit + )} + ) + + } + content={ + <> + Combined limit: at most{" "} + {Math.min( + queue.concurrency.combined.current, + environment.concurrencyLimit + )}{" "} + runs across all concurrency keys of this queue. The main limit + applies to each key separately. + + } + className="max-w-[260px]" + /> + ) : undefined + } > {queue.concurrencyLimitOverridePercent !== null ? ( <> @@ -1775,7 +1822,13 @@ function ClassicQueuesView() { Name Queued Running - Limit + + Limit + 0 && "text-text-bright", + queue.concurrency?.combined?.current != null && + (queue.concurrency.combined.running ?? 0) >= + Math.min( + queue.concurrency.combined.current, + environment.concurrencyLimit + ) + ? "text-warning" + : queue.running > 0 && "text-text-bright", isAtConcurrencyLimit && "text-warning" )} > @@ -1897,6 +1957,34 @@ function ClassicQueuesView() { )} > {limit} + {queue.concurrency?.combined?.current != null ? ( + + ( + {Math.min( + queue.concurrency.combined.current, + environment.concurrencyLimit + )} + ) + + } + content={ + <> + Combined limit: at most{" "} + {Math.min( + queue.concurrency.combined.current, + environment.concurrencyLimit + )}{" "} + runs across all concurrency keys of this queue. The main limit + applies to each key separately. + + } + className="max-w-[260px]" + /> + ) : null} ) ) : ( - + )} @@ -402,7 +407,13 @@ export default function Page() { {view === "keys" && hasKeys ? ( <> - + {selectedKey ? ( @@ -436,10 +447,12 @@ function OverviewCharts({ ids, timeRange, queueName, + hasTotalLimit, }: { ids: Ids; timeRange: TimeRangeParams; queueName: string; + hasTotalLimit: boolean; }) { const zoomToTimeFilter = useZoomToTimeFilter(); return ( @@ -479,6 +492,37 @@ function OverviewCharts({ // leading zeros so the reference line doesn't start with a false 0→limit step. carryBackfill={["limit"]} /> + {hasTotalLimit ? ( + + Runs in flight across ALL concurrency keys ( + ) versus the queue's combined limit ( + + ). + + } + showLegend + className="aspect-[2/1]" + query={`SELECT timeBucket() AS t, max(max_combined_running) AS running, least(max(max_combined_limit), max(max_env_limit)) AS cap\nFROM queue_metrics\nGROUP BY t\nORDER BY t`} + fillGaps + minBucketSeconds={SYNCED_CHART_MIN_BUCKET_SECONDS} + ids={ids} + timeRange={timeRange} + queueName={queueName} + series={[ + { key: "cap", label: "Combined limit", color: COLORS.limit }, + { key: "running", label: "Running", color: COLORS.running }, + ]} + thresholdStroke={{ + series: "running", + valueFromSeries: "cap", + aboveColor: "var(--color-warning)", + }} + carryBackfill={["cap"]} + /> + ) : null} Key Queued now Running now + + Limit + Oldest wait Started Peak backlog @@ -976,11 +1031,11 @@ function KeyStatsTable({ {showLoading ? ( - + Loading… ) : rows.length === 0 ? ( - + {search ? `No keys match “${search}”` : "No concurrency keys"} ) : ( @@ -994,6 +1049,12 @@ function KeyStatsTable({ {row.key} {row.queued.toLocaleString()} {row.running.toLocaleString()} + + {Math.min(row.limitOverride ?? defaultKeyLimit, envLimit).toLocaleString()} + {row.oldestWaitMs === null ? "–" : formatWaitMs(row.oldestWaitMs)} diff --git a/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.combined.override.ts b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.combined.override.ts index c643b77965a..77688a9fcc5 100644 --- a/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.combined.override.ts +++ b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.combined.override.ts @@ -46,6 +46,9 @@ const route = createActionApiRoute( concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt, concurrencyLimitOverriddenBy: null, paused: queue.paused, + totalConcurrencyLimit: queue.totalConcurrencyLimit, + totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase, + totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt, }), { status: 200 } ); diff --git a/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.combined.reset.ts b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.combined.reset.ts index b2841f1efe6..0e588716658 100644 --- a/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.combined.reset.ts +++ b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.combined.reset.ts @@ -45,6 +45,9 @@ const route = createActionApiRoute( concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt, concurrencyLimitOverriddenBy: null, paused: queue.paused, + totalConcurrencyLimit: queue.totalConcurrencyLimit, + totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase, + totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt, }), { status: 200 } ); diff --git a/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.key.override.ts b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.key.override.ts index 5a37b4526ec..3758a882c93 100644 --- a/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.key.override.ts +++ b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.key.override.ts @@ -52,6 +52,9 @@ const route = createActionApiRoute( concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt, concurrencyLimitOverriddenBy: null, paused: queue.paused, + totalConcurrencyLimit: queue.totalConcurrencyLimit, + totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase, + totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt, }), { status: 200 } ); diff --git a/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.key.reset.ts b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.key.reset.ts index 51d14642e2c..e30bf3f6360 100644 --- a/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.key.reset.ts +++ b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.key.reset.ts @@ -46,6 +46,9 @@ const route = createActionApiRoute( concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt, concurrencyLimitOverriddenBy: null, paused: queue.paused, + totalConcurrencyLimit: queue.totalConcurrencyLimit, + totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase, + totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt, }), { status: 200 } ); diff --git a/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.override.ts b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.override.ts index 90f5772c5d3..42bb2008682 100644 --- a/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.override.ts +++ b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.override.ts @@ -61,6 +61,9 @@ const route = createActionApiRoute( concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt, concurrencyLimitOverriddenBy: null, paused: queue.paused, + totalConcurrencyLimit: queue.totalConcurrencyLimit, + totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase, + totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt, }), { status: 200 } ); diff --git a/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.reset.ts b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.reset.ts index 503d875e471..3f36e629f09 100644 --- a/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.reset.ts +++ b/apps/webapp/app/routes/api.v1.queues.$queueParam.concurrency.reset.ts @@ -43,6 +43,9 @@ const route = createActionApiRoute( concurrencyLimitOverriddenAt: queue.concurrencyLimitOverriddenAt, concurrencyLimitOverriddenBy: null, paused: queue.paused, + totalConcurrencyLimit: queue.totalConcurrencyLimit, + totalConcurrencyLimitBase: queue.totalConcurrencyLimitBase, + totalConcurrencyLimitOverriddenAt: queue.totalConcurrencyLimitOverriddenAt, }), { status: 200 } ); diff --git a/apps/webapp/app/routes/resources.queues.concurrency-keys.ts b/apps/webapp/app/routes/resources.queues.concurrency-keys.ts index 67c2b9f500a..8c590554e51 100644 --- a/apps/webapp/app/routes/resources.queues.concurrency-keys.ts +++ b/apps/webapp/app/routes/resources.queues.concurrency-keys.ts @@ -43,6 +43,8 @@ export type ConcurrencyKeyRow = { peakBacklog: number; peakRunning: number; meanWaitMs: number; + /** Per-key concurrency limit override, when one is set for this key (null = inherits the queue limit). */ + limitOverride: number | null; }; export type ConcurrencyKeysResponse = @@ -151,8 +153,11 @@ export const action = async ({ request }: ActionFunctionArgs) => { const total = rankingRows?.[0]?.ranked_total ?? 0; const keys = (rankingRows ?? []).map((r) => r.concurrency_key); - // Enrich just this page's keys with live "now" counts from Redis. - const live = await engine.concurrencyKeyLiveStats(environment, queueName, keys); + // Enrich just this page's keys with live "now" counts and any per-key limit overrides from Redis. + const [live, keyLimitOverrides] = await Promise.all([ + engine.concurrencyKeyLiveStats(environment, queueName, keys), + engine.runQueue.getQueueConcurrencyKeyLimitsForKeys(environment, queueName, keys), + ]); const loadedAt = Date.now(); const rows: ConcurrencyKeyRow[] = (rankingRows ?? []).map((r) => { @@ -168,6 +173,7 @@ export const action = async ({ request }: ActionFunctionArgs) => { peakBacklog: r.peak_backlog, peakRunning: r.peak_running, meanWaitMs: r.mean_wait_ms, + limitOverride: keyLimitOverrides[r.concurrency_key] ?? null, }; }); diff --git a/apps/webapp/app/v3/querySchemas.ts b/apps/webapp/app/v3/querySchemas.ts index 690bbaf5396..05cd7f0b394 100644 --- a/apps/webapp/app/v3/querySchemas.ts +++ b/apps/webapp/app/v3/querySchemas.ts @@ -770,6 +770,22 @@ const queueMetricsSchema: TableSchema = { fillMode: "carry", }), }, + max_combined_running: { + name: "max_combined_running", + ...column("UInt32", { + description: + "Peak in-flight runs across ALL concurrency keys of the queue in the bucket (only emitted for keyed queues). Aggregate with max().", + fillMode: "carry", + }), + }, + max_combined_limit: { + name: "max_combined_limit", + ...column("UInt32", { + description: + "The queue's combined concurrency limit across all keys, as stored (0 = no cap; clamp against max_env_limit). Aggregate with max().", + fillMode: "carry", + }), + }, max_ck_backlogged: { name: "max_ck_backlogged", ...column("UInt32", { @@ -1406,6 +1422,14 @@ const queueMetricsByKeySchema: TableSchema = { fillMode: "carry", }), }, + max_limit: { + name: "max_limit", + ...column("UInt32", { + description: + "The effective concurrency limit for this key (the queue limit, or its per-key override). Aggregate with max().", + fillMode: "carry", + }), + }, wait_ms_sum: { name: "wait_ms_sum", ...column("UInt64", { diff --git a/apps/webapp/app/v3/queueMetricsMapping.ts b/apps/webapp/app/v3/queueMetricsMapping.ts index 9433b361a88..f093dc3f027 100644 --- a/apps/webapp/app/v3/queueMetricsMapping.ts +++ b/apps/webapp/app/v3/queueMetricsMapping.ts @@ -131,6 +131,8 @@ export function mapEntryToRows( throttled: num(f.thr), ck_backlogged: num(f.ckq), ck_max_wait_ms: num(f.ckw), + combined_running: num(f.tcc), + combined_limit: num(f.tlim), }, ]; } diff --git a/internal-packages/clickhouse/schema/042_add_queue_metrics_combined_concurrency.sql b/internal-packages/clickhouse/schema/042_add_queue_metrics_combined_concurrency.sql new file mode 100644 index 00000000000..03cb133799a --- /dev/null +++ b/internal-packages/clickhouse/schema/042_add_queue_metrics_combined_concurrency.sql @@ -0,0 +1,177 @@ +-- +goose Up + +-- Total-concurrency gauges: combined_running is the in-flight count across ALL +-- concurrency-key variants of a queue (the groupConcurrency set), combined_limit the +-- RAW stored total cap (0 = none, readers clamp against max_env_limit). Emitted on +-- base-queue gauge rows only. Per-key gauge rows now carry the EFFECTIVE per-key +-- limit in queue_limit (override-aware), surfaced in the ck tier as max_limit. + +ALTER TABLE trigger_dev.queue_metrics_raw_v1 + ADD COLUMN IF NOT EXISTS combined_running UInt32 DEFAULT 0, + ADD COLUMN IF NOT EXISTS combined_limit UInt32 DEFAULT 0; + +ALTER TABLE trigger_dev.queue_metrics_v1 + ADD COLUMN IF NOT EXISTS max_combined_running SimpleAggregateFunction(max, UInt32), + ADD COLUMN IF NOT EXISTS max_combined_limit SimpleAggregateFunction(max, UInt32); + +ALTER TABLE trigger_dev.queue_metrics_5m_v1 + ADD COLUMN IF NOT EXISTS max_combined_running SimpleAggregateFunction(max, UInt32), + ADD COLUMN IF NOT EXISTS max_combined_limit SimpleAggregateFunction(max, UInt32); + +ALTER TABLE trigger_dev.queue_metrics_ck_v1 + ADD COLUMN IF NOT EXISTS max_limit SimpleAggregateFunction(max, UInt32); + +-- Materialized views cannot be altered: recreate them with the new columns. The 5m +-- MV MUST keep reading raw, never cascade off queue_metrics_v1 (out-of-time-order +-- deltaSumTimestamp merges double-count bridging spans). + +DROP VIEW IF EXISTS trigger_dev.queue_metrics_mv_v1; +CREATE MATERIALIZED VIEW IF NOT EXISTS trigger_dev.queue_metrics_mv_v1 +TO trigger_dev.queue_metrics_v1 AS +SELECT + organization_id, project_id, environment_id, queue_name, + toStartOfInterval(event_time, INTERVAL 10 SECOND) AS bucket_start, + deltaSumTimestampStateIf(cumulative, order_key, op = 'enqueue' AND concurrency_key = '') AS enqueue_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'started' AND concurrency_key = '') AS started_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'ack' AND concurrency_key = '') AS ack_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'nack' AND concurrency_key = '') AS nack_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'dlq' AND concurrency_key = '') AS dlq_delta, + sum(throttled) AS throttled_count, + max(queued) AS max_queued, + max(running) AS max_running, + max(queue_limit) AS max_limit, + max(env_queued) AS max_env_queued, + max(env_running) AS max_env_running, + max(env_limit) AS max_env_limit, + max(ck_backlogged) AS max_ck_backlogged, + max(ck_max_wait_ms) AS max_ck_wait_ms, + max(combined_running) AS max_combined_running, + max(combined_limit) AS max_combined_limit, + sumIf(wait_ms, op = 'started' AND concurrency_key = '') AS wait_ms_sum, + countIf(op = 'started' AND wait_ms > 0 AND concurrency_key = '') AS wait_ms_count, + quantilesStateIf(0.5, 0.9, 0.95, 0.99)(wait_ms, op = 'started' AND wait_ms > 0 AND concurrency_key = '') AS wait_quantiles +FROM trigger_dev.queue_metrics_raw_v1 +GROUP BY organization_id, project_id, environment_id, queue_name, bucket_start; + +DROP VIEW IF EXISTS trigger_dev.queue_metrics_5m_mv_v1; +CREATE MATERIALIZED VIEW IF NOT EXISTS trigger_dev.queue_metrics_5m_mv_v1 +TO trigger_dev.queue_metrics_5m_v1 AS +SELECT + organization_id, project_id, environment_id, queue_name, + toStartOfInterval(event_time, INTERVAL 5 MINUTE) AS bucket_start, + deltaSumTimestampStateIf(cumulative, order_key, op = 'enqueue' AND concurrency_key = '') AS enqueue_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'started' AND concurrency_key = '') AS started_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'ack' AND concurrency_key = '') AS ack_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'nack' AND concurrency_key = '') AS nack_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'dlq' AND concurrency_key = '') AS dlq_delta, + sum(throttled) AS throttled_count, + max(queued) AS max_queued, + max(running) AS max_running, + max(queue_limit) AS max_limit, + max(env_queued) AS max_env_queued, + max(env_running) AS max_env_running, + max(env_limit) AS max_env_limit, + max(ck_backlogged) AS max_ck_backlogged, + max(ck_max_wait_ms) AS max_ck_wait_ms, + max(combined_running) AS max_combined_running, + max(combined_limit) AS max_combined_limit, + sumIf(wait_ms, op = 'started' AND concurrency_key = '') AS wait_ms_sum, + countIf(op = 'started' AND wait_ms > 0 AND concurrency_key = '') AS wait_ms_count, + quantilesStateIf(0.5, 0.9, 0.95, 0.99)(wait_ms, op = 'started' AND wait_ms > 0 AND concurrency_key = '') AS wait_quantiles +FROM trigger_dev.queue_metrics_raw_v1 +GROUP BY organization_id, project_id, environment_id, queue_name, bucket_start; + +DROP VIEW IF EXISTS trigger_dev.queue_metrics_ck_mv_v1; +CREATE MATERIALIZED VIEW IF NOT EXISTS trigger_dev.queue_metrics_ck_mv_v1 +TO trigger_dev.queue_metrics_ck_v1 AS +SELECT + organization_id, project_id, environment_id, queue_name, concurrency_key, + toStartOfInterval(event_time, INTERVAL 10 SECOND) AS bucket_start, + deltaSumTimestampStateIf(cumulative, order_key, op = 'enqueue') AS enqueue_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'started') AS started_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'ack') AS ack_delta, + maxIf(queued, op = 'gauge') AS max_queued, + maxIf(running, op = 'gauge') AS max_running, + maxIf(queue_limit, op = 'gauge') AS max_limit, + sumIf(wait_ms, op = 'started') AS wait_ms_sum, + countIf(op = 'started' AND wait_ms > 0) AS wait_ms_count +FROM trigger_dev.queue_metrics_raw_v1 +WHERE concurrency_key != '' +GROUP BY organization_id, project_id, environment_id, queue_name, concurrency_key, bucket_start; + +-- +goose Down +DROP VIEW IF EXISTS trigger_dev.queue_metrics_ck_mv_v1; +DROP VIEW IF EXISTS trigger_dev.queue_metrics_5m_mv_v1; +DROP VIEW IF EXISTS trigger_dev.queue_metrics_mv_v1; +ALTER TABLE trigger_dev.queue_metrics_ck_v1 DROP COLUMN IF EXISTS max_limit; +ALTER TABLE trigger_dev.queue_metrics_5m_v1 DROP COLUMN IF EXISTS max_combined_running, DROP COLUMN IF EXISTS max_combined_limit; +ALTER TABLE trigger_dev.queue_metrics_v1 DROP COLUMN IF EXISTS max_combined_running, DROP COLUMN IF EXISTS max_combined_limit; +ALTER TABLE trigger_dev.queue_metrics_raw_v1 DROP COLUMN IF EXISTS combined_running, DROP COLUMN IF EXISTS combined_limit; + +-- Recreate the pre-042 materialized views (the definitions from 036) so ingestion keeps +-- feeding every aggregate table after a rollback. +CREATE MATERIALIZED VIEW IF NOT EXISTS trigger_dev.queue_metrics_mv_v1 +TO trigger_dev.queue_metrics_v1 AS +SELECT + organization_id, project_id, environment_id, queue_name, + toStartOfInterval(event_time, INTERVAL 10 SECOND) AS bucket_start, + deltaSumTimestampStateIf(cumulative, order_key, op = 'enqueue' AND concurrency_key = '') AS enqueue_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'started' AND concurrency_key = '') AS started_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'ack' AND concurrency_key = '') AS ack_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'nack' AND concurrency_key = '') AS nack_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'dlq' AND concurrency_key = '') AS dlq_delta, + sum(throttled) AS throttled_count, + max(queued) AS max_queued, + max(running) AS max_running, + max(queue_limit) AS max_limit, + max(env_queued) AS max_env_queued, + max(env_running) AS max_env_running, + max(env_limit) AS max_env_limit, + max(ck_backlogged) AS max_ck_backlogged, + max(ck_max_wait_ms) AS max_ck_wait_ms, + sumIf(wait_ms, op = 'started' AND concurrency_key = '') AS wait_ms_sum, + countIf(op = 'started' AND wait_ms > 0 AND concurrency_key = '') AS wait_ms_count, + quantilesStateIf(0.5, 0.9, 0.95, 0.99)(wait_ms, op = 'started' AND wait_ms > 0 AND concurrency_key = '') AS wait_quantiles +FROM trigger_dev.queue_metrics_raw_v1 +GROUP BY organization_id, project_id, environment_id, queue_name, bucket_start; + +CREATE MATERIALIZED VIEW IF NOT EXISTS trigger_dev.queue_metrics_5m_mv_v1 +TO trigger_dev.queue_metrics_5m_v1 AS +SELECT + organization_id, project_id, environment_id, queue_name, + toStartOfInterval(event_time, INTERVAL 5 MINUTE) AS bucket_start, + deltaSumTimestampStateIf(cumulative, order_key, op = 'enqueue' AND concurrency_key = '') AS enqueue_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'started' AND concurrency_key = '') AS started_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'ack' AND concurrency_key = '') AS ack_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'nack' AND concurrency_key = '') AS nack_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'dlq' AND concurrency_key = '') AS dlq_delta, + sum(throttled) AS throttled_count, + max(queued) AS max_queued, + max(running) AS max_running, + max(queue_limit) AS max_limit, + max(env_queued) AS max_env_queued, + max(env_running) AS max_env_running, + max(env_limit) AS max_env_limit, + max(ck_backlogged) AS max_ck_backlogged, + max(ck_max_wait_ms) AS max_ck_wait_ms, + sumIf(wait_ms, op = 'started' AND concurrency_key = '') AS wait_ms_sum, + countIf(op = 'started' AND wait_ms > 0 AND concurrency_key = '') AS wait_ms_count, + quantilesStateIf(0.5, 0.9, 0.95, 0.99)(wait_ms, op = 'started' AND wait_ms > 0 AND concurrency_key = '') AS wait_quantiles +FROM trigger_dev.queue_metrics_raw_v1 +GROUP BY organization_id, project_id, environment_id, queue_name, bucket_start; + +CREATE MATERIALIZED VIEW IF NOT EXISTS trigger_dev.queue_metrics_ck_mv_v1 +TO trigger_dev.queue_metrics_ck_v1 AS +SELECT + organization_id, project_id, environment_id, queue_name, concurrency_key, + toStartOfInterval(event_time, INTERVAL 10 SECOND) AS bucket_start, + deltaSumTimestampStateIf(cumulative, order_key, op = 'enqueue') AS enqueue_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'started') AS started_delta, + deltaSumTimestampStateIf(cumulative, order_key, op = 'ack') AS ack_delta, + maxIf(queued, op = 'gauge') AS max_queued, + maxIf(running, op = 'gauge') AS max_running, + sumIf(wait_ms, op = 'started') AS wait_ms_sum, + countIf(op = 'started' AND wait_ms > 0) AS wait_ms_count +FROM trigger_dev.queue_metrics_raw_v1 +WHERE concurrency_key != '' +GROUP BY organization_id, project_id, environment_id, queue_name, concurrency_key, bucket_start; diff --git a/internal-packages/clickhouse/src/queueMetrics.ts b/internal-packages/clickhouse/src/queueMetrics.ts index 39576b4a0a3..f3a6be695e4 100644 --- a/internal-packages/clickhouse/src/queueMetrics.ts +++ b/internal-packages/clickhouse/src/queueMetrics.ts @@ -21,6 +21,8 @@ export const QueueMetricsRawV1Input = z.object({ throttled: z.number().optional(), ck_backlogged: z.number().optional(), ck_max_wait_ms: z.number().optional(), + combined_running: z.number().optional(), + combined_limit: z.number().optional(), wait_ms: z.number().optional(), cumulative: z.number().optional(), }); diff --git a/internal-packages/metrics-pipeline/src/lua.ts b/internal-packages/metrics-pipeline/src/lua.ts index 64f3b896c0d..701f608308a 100644 --- a/internal-packages/metrics-pipeline/src/lua.ts +++ b/internal-packages/metrics-pipeline/src/lua.ts @@ -17,6 +17,10 @@ export type GaugeComputeLuaParams = { // CK-health extras (both or neither): appended as an optional gauge tail, gauge[8]/gauge[9]. ckBacklogged?: string; ckMaxWaitMs?: string; + // Total-concurrency extras (both or neither, and only with the CK extras): appended as + // gauge[10]/gauge[11]. totalLimit is the RAW stored limit (0 = none); readers clamp. + totalRunning?: string; + totalLimit?: string; }; // Computes an op=gauge snapshot into the enclosing script's `__qm_g` local (a flat @@ -26,11 +30,21 @@ export type GaugeComputeLuaParams = { export function createMetricsGaugeComputeLua(params: GaugeComputeLuaParams): string { const throttled = params.throttledExpr ?? "__cc >= __lim and __ql > 0"; const hasCk = params.ckBacklogged != null && params.ckMaxWaitMs != null; - const gauge = hasCk + const hasTotal = params.totalRunning != null && params.totalLimit != null; + if (hasTotal && !hasCk) { + throw new Error("gauge totalRunning/totalLimit extras require the CK extras"); + } + const gauge = hasTotal ? ` local __ckq = tonumber(${params.ckBacklogged}) or 0 local __ckw = tonumber(${params.ckMaxWaitMs}) or 0 + local __tcc = tonumber(${params.totalRunning}) or 0 + local __tlim = tonumber(${params.totalLimit}) or 0 + __qm_g = {__ql, __cc, __lim, __eql, __ec, __elim, __thr, __ckq, __ckw, __tcc, __tlim}` + : hasCk + ? ` local __ckq = tonumber(${params.ckBacklogged}) or 0 + local __ckw = tonumber(${params.ckMaxWaitMs}) or 0 __qm_g = {__ql, __cc, __lim, __eql, __ec, __elim, __thr, __ckq, __ckw}` - : ` __qm_g = {__ql, __cc, __lim, __eql, __ec, __elim, __thr}`; + : ` __qm_g = {__ql, __cc, __lim, __eql, __ec, __elim, __thr}`; return ` if ${params.enabledArg} then diff --git a/internal-packages/run-engine/src/engine/index.ts b/internal-packages/run-engine/src/engine/index.ts index 9d7bb8ff947..504673bf052 100644 --- a/internal-packages/run-engine/src/engine/index.ts +++ b/internal-packages/run-engine/src/engine/index.ts @@ -1740,6 +1740,20 @@ export class RunEngine { return this.runQueue.currentConcurrencyOfQueues(environment, queues); } + async totalConcurrencyOfQueues( + environment: MinimalAuthenticatedEnvironment, + queues: string[] + ): Promise> { + return this.runQueue.totalConcurrencyOfQueues(environment, queues); + } + + async totalConcurrencyLimitsOfQueues( + environment: MinimalAuthenticatedEnvironment, + queues: string[] + ): Promise> { + return this.runQueue.totalConcurrencyLimitsOfQueues(environment, queues); + } + async concurrencyKeyBreakdown( environment: MinimalAuthenticatedEnvironment, queue: string, diff --git a/internal-packages/run-engine/src/run-queue/index.ts b/internal-packages/run-engine/src/run-queue/index.ts index 7f6493dfc2b..f873e2f5f68 100644 --- a/internal-packages/run-engine/src/run-queue/index.ts +++ b/internal-packages/run-engine/src/run-queue/index.ts @@ -208,16 +208,26 @@ const QUEUE_METRICS_CK_GAUGE_EXTRAS = { ckMaxWaitMs: "__ckwait", }; +// Total-concurrency tail (gauge[10]/gauge[11]): live group cardinality + raw stored cap. +// Requires groupConcurrencyKey/totalConcurrencyLimitKey locals; the CK scripts that actually +// run (the Tracked variants and the CK dequeue) all declare them for the total-cap gate. +const QUEUE_METRICS_TOTAL_GAUGE_EXTRAS = { + totalRunning: "redis.call('SCARD', groupConcurrencyKey)", + totalLimit: "redis.call('GET', totalConcurrencyLimitKey) or '0'", +}; + // CK enqueue variants of the two gauges above, extended with the CK-health tail. const QUEUE_METRICS_CK_ENQUEUE_GAUGE_LUA = createMetricsGaugeComputeLua({ enabledArg: "ARGV[#ARGV] == '1'", queued: "redis.call('ZCARD', queueKey)", running: "redis.call('SCARD', queueCurrentConcurrencyKey)", - queueLimit: "redis.call('GET', queueConcurrencyLimitKey) or '1000000'", + queueLimit: + "redis.call('HGET', ckLimitsKey, queueName) or redis.call('GET', queueConcurrencyLimitKey) or '1000000'", envQueued: "redis.call('ZCARD', envQueueKey)", envRunning: "redis.call('SCARD', envCurrentConcurrencyKey)", envLimit: "redis.call('GET', envConcurrencyLimitKey) or defaultEnvConcurrencyLimit", ...QUEUE_METRICS_CK_GAUGE_EXTRAS, + ...QUEUE_METRICS_TOTAL_GAUGE_EXTRAS, }); const QUEUE_METRICS_CK_ENQUEUE_FASTPATH_GAUGE_LUA = createMetricsGaugeComputeLua({ @@ -229,6 +239,7 @@ const QUEUE_METRICS_CK_ENQUEUE_FASTPATH_GAUGE_LUA = createMetricsGaugeComputeLua envRunning: "envCurrent", envLimit: "envLimit", ...QUEUE_METRICS_CK_GAUGE_EXTRAS, + ...QUEUE_METRICS_TOTAL_GAUGE_EXTRAS, }); // CK dequeue: depth/running from the per-base-queue aggregate counters the run-queue already @@ -244,6 +255,7 @@ const QUEUE_METRICS_CK_DEQUEUE_GAUGE_LUA = createMetricsGaugeComputeLua({ envLimit: "redis.call('GET', envConcurrencyLimitKey) or defaultEnvConcurrencyLimit", throttledExpr: "false", ...QUEUE_METRICS_CK_GAUGE_EXTRAS, + ...QUEUE_METRICS_TOTAL_GAUGE_EXTRAS, }); /** Injected queue-metrics stream emitter; all calls are no-ops when metrics are disabled. */ @@ -702,6 +714,69 @@ export class RunQueue { return limits; } + /** Per-key limit overrides for just the given keys: one HMGET, O(keys) not O(overrides). */ + public async getQueueConcurrencyKeyLimitsForKeys( + env: MinimalAuthenticatedEnvironment, + queue: string, + concurrencyKeys: string[] + ): Promise> { + if (concurrencyKeys.length === 0) { + return {}; + } + + const fields = concurrencyKeys.map((key) => this.keys.queueKey(env, queue, key)); + const values = await this.redis.hmget(this.keys.queueCkLimitsKey(env, queue), ...fields); + + const limits: Record = {}; + concurrencyKeys.forEach((key, index) => { + const value = values[index]; + if (value != null) { + limits[key] = Number(value); + } + }); + return limits; + } + + /** Batch variant of totalConcurrencyOfQueue: one pipeline of group SCARDs. */ + public async totalConcurrencyOfQueues( + env: MinimalAuthenticatedEnvironment, + queues: string[] + ): Promise> { + const pipeline = this.redis.pipeline(); + queues.forEach((queue) => { + pipeline.scard(this.keys.queueGroupConcurrencyKey(env, queue)); + }); + + const results = await pipeline.exec(); + + return queues.reduce( + (acc, queue, index) => { + const value = results?.[index]?.[1]; + acc[queue] = typeof value === "number" ? value : 0; + return acc; + }, + {} as Record + ); + } + + /** Batch read of the RAW stored total concurrency limits (undefined = no cap). */ + public async totalConcurrencyLimitsOfQueues( + env: MinimalAuthenticatedEnvironment, + queues: string[] + ): Promise> { + const keys = queues.map((queue) => this.keys.queueTotalConcurrencyLimitKey(env, queue)); + const values = keys.length > 0 ? await this.redis.mget(...keys) : []; + + return queues.reduce( + (acc, queue, index) => { + const value = values[index]; + acc[queue] = value != null ? Number(value) : undefined; + return acc; + }, + {} as Record + ); + } + public async updateEnvConcurrencyLimits(env: MinimalAuthenticatedEnvironment) { await this.#callUpdateEnvironmentConcurrencyLimits({ envConcurrencyLimitKey: this.keys.envConcurrencyLimitKey(env), @@ -2333,6 +2408,10 @@ export class RunQueue { fields.ckq = ckq; fields.ckw = ckw; } + if (gauge.length >= 11) { + fields.tcc = gauge[9]; + fields.tlim = gauge[10]; + } this.options.queueMetrics?.emitGauge(queue, fields); } diff --git a/packages/core/src/v3/schemas/queues.ts b/packages/core/src/v3/schemas/queues.ts index 34a47b34e3e..9ca282fb33d 100644 --- a/packages/core/src/v3/schemas/queues.ts +++ b/packages/core/src/v3/schemas/queues.ts @@ -45,6 +45,21 @@ export const QueueItem = z.object({ overriddenAt: z.coerce.date().nullable(), /** Who overrode the concurrency limit (will be null if overridden via the API) */ overriddenBy: z.string().nullable(), + /** The combined concurrency cap across all concurrencyKey values of the queue */ + combined: z + .object({ + /** The effective/current combined concurrency limit (null = no cap) */ + current: z.number().nullable(), + /** The declared combined limit an override reverts to on reset */ + base: z.number().nullable(), + /** The overridden combined limit, when an override is active */ + override: z.number().nullable(), + /** When the combined override was applied */ + overriddenAt: z.coerce.date().nullable(), + /** Runs currently in flight across all concurrencyKey values */ + running: z.number().nullable(), + }) + .optional(), }) .optional(), });