Eric Allam
4eb9292cbe
## Summary
Three related changes, each independently gated:
**Queue metrics and health.** Per-queue depth, throughput (enqueued,
started, completed), concurrency, whether a queue is throttled, and
scheduling delay (how long a run waits between becoming eligible and
actually starting), plus a per concurrency-key breakdown for keyed
queues. Collected from inside the run queue itself, stored in
ClickHouse, and surfaced on the Queues list, a new per-queue detail
page, the task pages, and the run inspector. The question it answers is
"does this queue have enough concurrency to keep up, and if not, which
key or which limit is the constraint".
**Percent-based queue concurrency limits.** A queue's concurrency
override can now be expressed as a percentage of the environment limit,
stored as the source of truth and re-materialized whenever the
environment limit changes. Absolute overrides above the environment
limit are now **rejected with a 400** instead of being silently capped,
which is a behavior change on `POST
/api/v1/queues/:queue/concurrency/override`.
**The `health` report.** A server-computed verdict on whether work is
flowing, whether the runs that do start are healthy, and whether
telemetry is fresh, rendered as text with sparklines. Available as `GET
/api/v1/reports/:key`, `trigger report`, and the `get_report` MCP tool
(plus a `report` MCP prompt, which shows up as a slash command in hosts
that support prompts).
With the flags off, the Queues page renders the pre-metrics component
verbatim, nothing is emitted, and nothing is written to ClickHouse.
## Configuration
Two independent gates, on purpose. Emission is global so data accrues
for everyone before anyone can look at it; the view is per organization
so it can be turned on for one org at a time without a deploy.
**Runtime flags (no restart)**
| Flag | Store | Gates |
| --- | --- | --- |
| `queue_metrics:enabled` | run-queue Redis key (`"1"`/`"0"`, off by
default) | All emission, gauges and counters. Cached in-process for 10s
with stale-while-revalidate, warmed eagerly at boot so the first op
after a deploy is not dropped. |
| `queue_metrics:gauge_sample_rate` | run-queue Redis key, `0..1` |
Fraction of queue ops that emit a gauge. Counters are never sampled, so
throughput stays exact at any rate. |
| `queueMetricsUiEnabled` | feature-flag catalog: global `FeatureFlag`
row, per-org `Organization.featureFlags` override wins | Whether an org
sees the metrics view at all: the Queues list variant, the queue detail
route, the built-in Queues dashboard, the concurrency-keys endpoint, and
the metrics blocks on task pages and the run inspector. Off by default;
a gated org gets a 404 on the detail route rather than an empty page. |
Both Redis keys are readable and writable from `/admin/queue-metrics`
(super-admin UI, with a live per-shard stream-health table) and
`GET`/`POST /admin/api/v1/queue-metrics` (admin PAT). The admin surface
uses its own Redis client, so it works on any instance regardless of
whether that instance runs the emitter or the consumer.
**Environment variables (boot time)**
| Variable | Default | Notes |
| --- | --- | --- |
| `QUEUE_METRICS_EMIT_ENABLED` | `0` | Constructs the emitter and
injects it into the run engine. Without it the run queue has no emitter
at all. |
| `QUEUE_METRICS_CONSUMER_ENABLED` | `0` | Boots the stream consumer on
this instance. Independent of emission, so consumers can be sized
separately from the API. |
| `QUEUE_METRICS_STREAM_SHARD_COUNT` | `4` | Stream shards, hashed per
queue. |
| `QUEUE_METRICS_CONSUMER_BATCH_SIZE` | `1000` | Poll batch equals
insert batch, so an ack can never outrun a write. |
| `QUEUE_METRICS_REDIS_{HOST,PORT,USERNAME,PASSWORD,TLS_DISABLED}` |
falls back to the run-queue Redis | Set `HOST` to move the metrics
stream onto a dedicated instance so a metrics backlog cannot compete
with the run queue for memory. Self-hosters can leave it unset and get a
single-Redis deployment. |
| `QUEUE_METRICS_COUNTER_STREAM_MAXLEN` | `2000000` shared, `8000000`
dedicated | Bound on how much a stalled consumer can hold. The default
is deliberately lower when the stream shares the queue-critical Redis. |
| `QUEUE_METRICS_COUNTER_ODOMETER_TTL_SECONDS` | `604800` | TTL on the
per-queue cumulative counter key, refreshed on every write, so only
queues idle for the whole window are purged. |
| `QUEUE_METRICS_MAX_QUEUE_NAMES_PER_ENV` | `1000` | Distinct queue
names tracked per environment; overflow collapses into `__overflow__`. |
| `QUEUE_METRICS_MAX_CONCURRENCY_KEYS_PER_QUEUE` | `10000` | Same idea
one level down, per queue. |
| `QUEUE_METRICS_GAUGE_SAMPLE_RATE` | `1` | Default for the live
sample-rate key above. |
| `QUEUE_METRICS_QUERY_TABLES_VISIBLE` | `0` | Lists the queue-metrics
tables in the Query page, its schema docs, the schema API and the AI
query context. Off keeps them unlisted while the feature is dark; a
query naming them still runs either way. |
| `QUEUE_METRICS_CLICKHOUSE_URL` | falls back to the shared wiring |
Runs queue metrics on their own ClickHouse service: the consumer's
inserts and every queue-metrics read go through it, so a metrics-heavy
chart refresh never competes with runs-list or trace reads. Unset
reproduces the previous split exactly (inserts on `CLICKHOUSE_URL`,
reads on the query pool). |
| `QUEUE_METRICS_CLICKHOUSE_READER_URL` | the write URL | Reader split,
so the consumer's inserts can never land on a read endpoint. |
|
`QUEUE_METRICS_CLICKHOUSE_{KEEP_ALIVE_ENABLED,KEEP_ALIVE_IDLE_SOCKET_TTL_MS,MAX_OPEN_CONNECTIONS,LOG_LEVEL,COMPRESSION_REQUEST}`
| `1`, unset, `10`, `info`, `1` | Pool tuning, matching the other
per-workload ClickHouse clients. |
Migrations to apply: ClickHouse `036_create_queue_metrics_v1.sql`, and a
Postgres migration adding the nullable
`TaskQueue.concurrencyLimitOverridePercent`. Both are additive.
## How collection works
Queue operations produce two kinds of signal, and they have opposite
failure modes, so they are handled differently.
**Gauges** (queued, running, queue limit, env queued, env running, env
limit, throttled, plus keys-with-backlog and worst-key wait on keyed
queues) are read *inside* the same Redis script that performs the
enqueue or dequeue, so the reading is atomic with the operation it
describes rather than a racy follow-up read. The script returns them on
its reply and the app forwards them to the stream. Gauges are sampled
and drop-tolerant: they are aggregated with `max`, so a lost reading
costs resolution, never correctness.
**Counters** (enqueued, started, completed, plus nack and dead-lettered)
are cumulative odometers. Each event increments a per-queue key on the
metrics Redis and emits the absolute total, and ClickHouse takes the
difference across buckets at read time. This is the important property
of the design: a summed-delta counter undercounts permanently on any
lost event, while a cumulative one self-heals, because the next
surviving reading restates the whole total. Only bucket granularity can
be lost, never the total. A queue returning after its odometer TTL
expired restarts at 1 and reset detection handles it, which is safe
precisely because expiry only spans a window with no activity.
Both land on one sharded Redis stream. A consumer reads it with a
consumer group, reclaims stale pending entries on a 15s interval rather
than on every poll, maps one entry to one or two ClickHouse rows
(whole-queue and, for keyed queues, per-key), and acks only after the
insert lands. Each batch carries a dedup token derived from its
stream-entry ids, and the target tables set
`non_replicated_deduplication_window`, so a retried batch cannot
double-count either the raw rows or the aggregates that hang off them.
Consumer and emitter both emit OTel metrics
(`queue_metrics.emitter.emitted`,
`queue_metrics.consumer.{entries,rows_inserted,insert_errors,insert_duration,stream_depth,group_lag,pending,lag_unknown}`);
stream depth and group lag are the two worth alerting on, and
`lag_unknown` exists because Redis can report a null lag after a trim,
which must not be read as zero.
## Storage and read path
`queue_metrics_raw_v1` is a short landing table with a 6 hour TTL. Four
aggregate tiers are materialized straight from raw, never cascaded off
each other, each with a 30 day TTL:
- `queue_metrics_v1`, 10 second buckets per queue, the default read path
- `queue_metrics_5m_v1`, 5 minute buckets per queue, for wide ranges and
cross-queue ranking
- `env_metrics_v1`, 10 second buckets per environment, queue-independent
so it stays cheap at any range
- `queue_metrics_ck_v1`, 10 second buckets per concurrency key
Every tier is an MV from raw because the counter states do not survive a
cascade: their merge is order sensitive, so a `-MergeState` chain off
the 10s table inflates the result, and the same property means an
aggregate state may only be merged inside one queue. That constraint is
now enforced by the query engine rather than by reviewer discipline: a
column can declare a `mergeGroupKey`, and any query that references it
without grouping by, or pinning to a single value of, every named key
fails to compile with an actionable message.
On the read side, TRQL gains three tables (`queue_metrics`,
`env_metrics`, and a `queue_metrics_by_key` that is hidden from the
editor, schema docs and schema API but still queryable, so per-key rows
can never silently merge into a plain per-queue query), plus
`deltaSumTimestampMerge` and `quantilesTDigestMerge`. Two schema-level
optimizations ride along: a table can declare coarser rollups, so a
query whose bucket interval is 5 minutes or wider is routed to the 5m
table with no change to the query itself, and it can opt into the
ClickHouse query cache with time bounds floored to a fixed grid, so the
auto-refreshing dashboards actually share cache entries instead of
missing on every tick. Both are caller-side substitutions, so the
printer stays unaware of physical layout.
All of this can also live on its own ClickHouse service. A table
declares the pool its reads run on, the three queue-metrics tables name
the dedicated one, and the ingestion consumer writes through the same
client, so both directions move together with one env var and nothing
else routes differently.
The other engine change is opt-in gap filling: charts can request rows
for empty buckets, where counters zero-fill and gauges carry forward.
Grouped gauge series are densified per group and carried inside a
partition, so a quiet queue's line holds its last value without bleeding
another queue's value into it.
## Queue concurrency limits
`concurrencyLimitOverridePercent` on `TaskQueue` is the source of truth
when an override is set as a percentage; the absolute `concurrencyLimit`
is materialized from it (floored, clamped to at least 1 so a percentage
can never act as a pause, and never above the environment limit). Every
path that changes an environment limit now recalculates the
environment's percent-based overrides afterwards, outside the
transaction, and pushes changed limits to the engine. The push is
attempted even when the stored value did not change, so a previously
failed sync self-heals rather than leaving the database and the engine
diverged; paused queues are skipped so a recalculation cannot
effectively unpause one.
The API accepts exactly one of `concurrencyLimit` or `percent`, and the
reject-instead-of-clamp change above means a request asking for more
than the environment allows now fails loudly. The percent bound (greater
than 0, at most 100) is defined once and shared by the zod schema, the
dashboard mutation handler and the service, so the three cannot drift.
The concurrency-keys table on a queue is now paginated against the
ClickHouse per-key tier, ranked by peak backlog with the total on every
row from a single scan, and only the keys on the current page are
enriched with live counts from Redis. That replaces a hard top-50 cap
with something whose cost is a function of page size rather than key
cardinality.
## The health report
`GET /api/v1/reports/:key?period=&format=markdown|ansi|json`. The
verdict is computed on the server and is deterministic, not
model-generated. Three independent analyzers run over one input
snapshot: flow (is work moving, and if not, is the cause a limit,
throttling, one bad queue, or dead-lettering), execution (are the runs
that start succeeding, and at what latency), and liveness (how fresh is
the telemetry). When telemetry is genuinely stale, the first two are
forced to unknown and every actionable field is stripped, so no surface
ever advises action off stale data.
Authorization is per query table rather than a blanket query grant: a
JWT must be scoped to every table the report reads (`runs`,
`env_metrics`, `queue_metrics`), so a narrowly scoped token cannot pull
a report that reads more than it was granted. `period` is validated as a
shorthand with a 90 day ceiling at the edge. The report catalog is a
registry of `{ load, interpret }` entries, so the next report is a new
entry and no change to the route, the view model, the renderers, the CLI
or the MCP tool.
`trigger mcp` no longer launches the install wizard when stdout is a
TTY, which fixed a real failure: hosts spawn the server over a PTY, so
the wizard would open and the client would time out waiting for a server
that never started. The wizard now needs `trigger mcp --install`.
## The part that is live regardless of every flag
The enqueue and dequeue scripts now return a 2-tuple so a gauge reading
can ride back on the reply. Every return site in the eight affected
scripts is wrapped, and a `nil` original is converted to `false` on the
way out, because a raw `nil` in the first slot would make Lua truncate
the multi-bulk reply and silently drop the gauge on the throttled and
empty-queue paths. The reply shape and the destructuring on the app side
are exercised on every queue operation whether or not metrics are
enabled, so that is the part of `run-engine` worth the closest review.
One behavior fix in the same area: the scheduling-delay anchor is set
only on a run's first entry into the queue. Anchoring it to trigger time
on re-enqueues made waitpoint and checkpoint resumes report the entire
wait as scheduling delay. Queue ordering is untouched, so a re-enqueued
run keeps its position, and nacks deliberately keep the original anchor
because a rolled-back dequeue is the same continuous wait.
A pending-version promotion still anchors to trigger time, on purpose:
that promotion is the run's first real entry into the queue, since the
trigger deliberately held it back waiting for a worker version, and the
TTL is armed at the same point for the same reason. The consequence is
worth naming, because it is a judgement call: a run that waits on a
deployment reports that wait as scheduling delay on its queue, which is
time unrelated to queue capacity.
## Verification
Unit and integration suites across the new package, the run queue, the
mapping layer, the query engine and ClickHouse (including a test that
applies migration 036 through the same splitter CI uses, and a
regression test that inserts the same batch three times to prove the
aggregates do not inflate). Beyond that, the whole path was driven end
to end against a live stack with real runs: emitter to Redis stream to
consumer to ClickHouse to the dashboards, for both the local dev path
and the deployed path where a supervisor drives the dequeue, with
assertions on exact counter reconstruction per queue and per concurrency
key, throttling, environment saturation, scheduling delay, and a
deliberate mid-stream reading drop to confirm the cumulative counters
still reconstruct the correct total. The gated-off state was checked on
every touched surface.
The dedicated ClickHouse service was verified against a second,
separately-schema'd instance: with it configured, the driven counters
reconstruct exactly on the dedicated instance, the shared instance gains
no rows for that window, a read through the query API returns the value
that exists only on the dedicated instance, and a `runs` query still
succeeds (it would fail outright if it were mis-routed to a service
without that table). With the variable unset, the full suite passes
unchanged.
---------
Co-authored-by: Katia Bulatova <katia@trigger.dev>
Co-authored-by: Katia Bulatova <katherine.bulatova@gmail.com>
Co-authored-by: James Ritchie <james@trigger.dev>
1478 lines
48 KiB
TypeScript
1478 lines
48 KiB
TypeScript
import { column, type BucketThreshold, type TableSchema } from "@internal/tsql";
|
||
import { z } from "zod";
|
||
import { autoFormatSQL } from "~/components/code/TSQLEditor";
|
||
import { runFriendlyStatus, runStatusTitleFromStatus } from "~/components/runs/v3/TaskRunStatus";
|
||
|
||
export const QueryScopeSchema = z.enum(["organization", "project", "environment"]);
|
||
export type QueryScope = z.infer<typeof QueryScopeSchema>;
|
||
|
||
/**
|
||
* Environment type values
|
||
*/
|
||
const ENVIRONMENT_TYPES = ["PRODUCTION", "STAGING", "DEVELOPMENT", "PREVIEW"] as const;
|
||
|
||
/**
|
||
* Machine preset values
|
||
*/
|
||
const MACHINE_PRESETS = [
|
||
"micro",
|
||
"small-1x",
|
||
"small-2x",
|
||
"medium-1x",
|
||
"medium-2x",
|
||
"large-1x",
|
||
"large-2x",
|
||
] as const;
|
||
|
||
/**
|
||
* Schema definition for the runs table (trigger_dev.task_runs_v2)
|
||
*/
|
||
export const runsSchema: TableSchema = {
|
||
name: "runs",
|
||
clickhouseName: "trigger_dev.task_runs_v2",
|
||
description: "Task runs - stores all task execution records",
|
||
timeConstraint: "triggered_at",
|
||
useFinal: true,
|
||
tenantColumns: {
|
||
organizationId: "organization_id",
|
||
projectId: "project_id",
|
||
environmentId: "environment_id",
|
||
},
|
||
requiredFilters: [{ column: "engine", value: "V2" }],
|
||
columns: {
|
||
run_id: {
|
||
name: "run_id",
|
||
clickhouseName: "friendly_id",
|
||
...column("String", {
|
||
description:
|
||
"A unique ID for a run. They always start with `run_`, e.g., run_cm1a2b3c4d5e6f7g8h9i",
|
||
customRenderType: "runId",
|
||
example: "run_cm1a2b3c4d5e6f7g8h9i",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
environment: {
|
||
name: "environment",
|
||
clickhouseName: "environment_id",
|
||
...column("String", { description: "The environment slug", example: "prod" }),
|
||
fieldMapping: "environment",
|
||
customRenderType: "environment",
|
||
},
|
||
project: {
|
||
name: "project",
|
||
clickhouseName: "project_id",
|
||
...column("String", {
|
||
description: "The project reference, they always start with `proj_`.",
|
||
example: "proj_howcnaxbfxdmwmxazktx",
|
||
}),
|
||
fieldMapping: "project",
|
||
customRenderType: "project",
|
||
},
|
||
environment_type: {
|
||
name: "environment_type",
|
||
...column("LowCardinality(String)", {
|
||
description: "Environment type",
|
||
allowedValues: [...ENVIRONMENT_TYPES],
|
||
customRenderType: "environmentType",
|
||
example: "PRODUCTION",
|
||
}),
|
||
},
|
||
attempt_count: {
|
||
name: "attempt_count",
|
||
clickhouseName: "attempt",
|
||
...column("UInt8", {
|
||
description: "Number of attempts (starts at 1)",
|
||
example: "1",
|
||
customRenderType: "number",
|
||
}),
|
||
},
|
||
status: {
|
||
name: "status",
|
||
...column("LowCardinality(String)", {
|
||
description: "Run status",
|
||
allowedValues: [...runFriendlyStatus],
|
||
valueMap: runStatusTitleFromStatus,
|
||
customRenderType: "runStatus",
|
||
example: "Completed",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
is_finished: {
|
||
name: "is_finished",
|
||
...column("UInt8", {
|
||
description:
|
||
"Whether the run is finished. This includes failed and successful runs. (0 or 1)",
|
||
example: "0",
|
||
}),
|
||
expression:
|
||
"if(status IN ('COMPLETED_SUCCESSFULLY', 'COMPLETED_WITH_ERRORS', 'CANCELED', 'TIMED_OUT', 'CRASHED', 'SYSTEM_FAILURE', 'EXPIRED', 'PAUSED'), true, false)",
|
||
},
|
||
|
||
// Task & queue
|
||
task_identifier: {
|
||
name: "task_identifier",
|
||
...column("String", {
|
||
description: "Task identifier/slug",
|
||
example: "my-background-task",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
queue: {
|
||
name: "queue",
|
||
...column("String", {
|
||
description: "Queue name",
|
||
example: "task/my-background-task",
|
||
customRenderType: "queue",
|
||
}),
|
||
},
|
||
|
||
batch_id: {
|
||
name: "batch_id",
|
||
...column("String", {
|
||
description: "Batch ID (if part of a batch)",
|
||
example: "batch_5678efgh",
|
||
expression: "if(batch_id = '', NULL, 'batch_' || batch_id)",
|
||
}),
|
||
whereTransform: (value: string) => value.replace(/^batch_/, ""),
|
||
},
|
||
|
||
// Related runs
|
||
root_run_id: {
|
||
name: "root_run_id",
|
||
...column("String", {
|
||
description: "Root run ID (for child runs)",
|
||
example: "run_cm1a2b3c4d5e6f7g8h9i",
|
||
customRenderType: "runId",
|
||
expression: "if(root_run_id = '', NULL, 'run_' || root_run_id)",
|
||
}),
|
||
whereTransform: (value: string) => value.replace(/^run_/, ""),
|
||
},
|
||
parent_run_id: {
|
||
name: "parent_run_id",
|
||
...column("String", {
|
||
description: "Parent run ID (for child runs)",
|
||
example: "run_cm1a2b3c4d5e6f7g8h9i",
|
||
customRenderType: "runId",
|
||
expression: "if(parent_run_id = '', NULL, 'run_' || parent_run_id)",
|
||
}),
|
||
whereTransform: (value: string) => value.replace(/^run_/, ""),
|
||
},
|
||
depth: {
|
||
name: "depth",
|
||
...column("UInt8", { description: "Nesting depth (0 for root runs)", example: "0" }),
|
||
},
|
||
is_root_run: {
|
||
name: "is_root_run",
|
||
...column("UInt8", { description: "Whether this is a root run (0 or 1)", example: "0" }),
|
||
expression: "if(depth = 0, true, false)",
|
||
},
|
||
is_child_run: {
|
||
name: "is_child_run",
|
||
...column("UInt8", { description: "Whether this is a child run (0 or 1)", example: "0" }),
|
||
expression: "if(depth > 0, true, false)",
|
||
},
|
||
|
||
idempotency_key: {
|
||
name: "idempotency_key",
|
||
clickhouseName: "idempotency_key_user",
|
||
...column("String", {
|
||
description: "Idempotency key (available from 4.3.3)",
|
||
example: "user-123-action-456",
|
||
}),
|
||
},
|
||
idempotency_key_scope: {
|
||
name: "idempotency_key_scope",
|
||
...column("String", {
|
||
description:
|
||
"The idempotency key scope determines whether a task should be considered unique within a parent run, a specific attempt, or globally. An empty value means there's no idempotency key set (available from 4.3.3).",
|
||
example: "run",
|
||
allowedValues: ["global", "run", "attempt"],
|
||
}),
|
||
},
|
||
region: {
|
||
name: "region",
|
||
clickhouseName: "region",
|
||
...column("String", {
|
||
description: "Region",
|
||
example: "us-east-1",
|
||
}),
|
||
// No whereTransform: the expression drives WHERE too, so pre-region rows still match.
|
||
expression:
|
||
"multiIf(region != '', region, startsWith(worker_queue, 'cm'), NULL, worker_queue)",
|
||
},
|
||
|
||
// Timing
|
||
triggered_at: {
|
||
name: "triggered_at",
|
||
clickhouseName: "created_at",
|
||
...column("DateTime64", {
|
||
description: "When the run was triggered.",
|
||
example: "2024-01-15 09:30:00.000",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
queued_at: {
|
||
name: "queued_at",
|
||
...column("Nullable(DateTime64)", {
|
||
description:
|
||
"When the run was added to the queue. This is normally the same time as the triggered_at time, unless a delay is passed in or it's a scheduled run.",
|
||
example: "2024-01-15 09:30:01.000",
|
||
}),
|
||
},
|
||
dequeued_at: {
|
||
name: "dequeued_at",
|
||
clickhouseName: "started_at",
|
||
...column("Nullable(DateTime64)", {
|
||
description:
|
||
"When the run was dequeued for execution. This happens when there is available concurrency to execute your run.",
|
||
example: "2024-01-15 09:30:01.000",
|
||
}),
|
||
},
|
||
executed_at: {
|
||
name: "executed_at",
|
||
...column("Nullable(DateTime64)", {
|
||
description: "When execution of the run began.",
|
||
example: "2024-01-15 09:30:01.500",
|
||
}),
|
||
},
|
||
completed_at: {
|
||
name: "completed_at",
|
||
...column("Nullable(DateTime64)", {
|
||
description: "When the run completed",
|
||
example: "2024-01-15 09:30:05.000",
|
||
}),
|
||
},
|
||
delay_until: {
|
||
name: "delay_until",
|
||
...column("Nullable(DateTime64)", {
|
||
description: "Delayed execution until this time",
|
||
example: "2024-01-15 10:00:00.000",
|
||
}),
|
||
},
|
||
has_delay: {
|
||
name: "has_delay",
|
||
...column("UInt8", { description: "Whether the run had a delay passed in", example: "1" }),
|
||
expression: "if(isNotNull(delay_until), true, false)",
|
||
},
|
||
expired_at: {
|
||
name: "expired_at",
|
||
...column("Nullable(DateTime64)", {
|
||
description:
|
||
'If there was a TTL on the run, this is when the run "expired". By default dev runs have a TTL of 10 minutes.',
|
||
example: "2024-01-15 09:35:00.000",
|
||
}),
|
||
},
|
||
ttl: {
|
||
name: "ttl",
|
||
clickhouseName: "expiration_ttl",
|
||
...column("String", {
|
||
description: "The TTL string for expiration – by default dev runs have a TTL of '10m'.",
|
||
example: "10m",
|
||
}),
|
||
},
|
||
|
||
// Useful time periods
|
||
execution_duration: {
|
||
name: "execution_duration",
|
||
...column("Nullable(Int64)", {
|
||
description:
|
||
"The time between starting to execute and completing. This includes any time spent waiting (it is not compute time, use `usage_duration` for that).",
|
||
customRenderType: "duration",
|
||
example: "4000",
|
||
}),
|
||
expression: "dateDiff('millisecond', executed_at, completed_at)",
|
||
},
|
||
total_duration: {
|
||
name: "total_duration",
|
||
...column("Nullable(Int64)", {
|
||
description:
|
||
"The time between being triggered and completing (if it has). This includes any time spent waiting (it is not compute time, use `usage_duration` for that).",
|
||
customRenderType: "duration",
|
||
example: "4000",
|
||
}),
|
||
expression: "dateDiff('millisecond', created_at, completed_at)",
|
||
},
|
||
queued_duration: {
|
||
name: "queued_duration",
|
||
...column("Nullable(Int64)", {
|
||
description:
|
||
"The time between being queued and dequeued. Remember you need enough available concurrency for runs to be dequeued and start executing.",
|
||
customRenderType: "duration",
|
||
example: "4000",
|
||
}),
|
||
expression: "dateDiff('millisecond', queued_at, started_at)",
|
||
},
|
||
|
||
// Cost & usage
|
||
usage_duration: {
|
||
name: "usage_duration",
|
||
clickhouseName: "usage_duration_ms",
|
||
...column("UInt32", {
|
||
description: "Compute usage duration in milliseconds.",
|
||
customRenderType: "duration",
|
||
example: "3500",
|
||
}),
|
||
},
|
||
compute_cost: {
|
||
name: "compute_cost",
|
||
...column("Float64", {
|
||
description: "Compute cost in dollars",
|
||
customRenderType: "costInDollars",
|
||
example: "0.000676",
|
||
}),
|
||
expression: "cost_in_cents / 100.0",
|
||
},
|
||
invocation_cost: {
|
||
name: "invocation_cost",
|
||
...column("Float64", {
|
||
description: "Invocation cost in dollars – the cost to start a run.",
|
||
customRenderType: "costInDollars",
|
||
example: "0.000025",
|
||
}),
|
||
expression: "base_cost_in_cents / 100.0",
|
||
},
|
||
total_cost: {
|
||
name: "total_cost",
|
||
...column("Float64", {
|
||
description: "Total cost in dollars (compute_cost + invocation_cost)",
|
||
customRenderType: "costInDollars",
|
||
example: "0.000701",
|
||
}),
|
||
expression: "(cost_in_cents + base_cost_in_cents) / 100.0",
|
||
},
|
||
|
||
// Output & error (JSON columns)
|
||
// For JSON columns, NULL checks are transformed to check for empty object '{}'
|
||
// So `error IS NULL` becomes `error = '{}'` and `error IS NOT NULL` becomes `error != '{}'`
|
||
// textColumn uses the pre-materialized text columns for better performance
|
||
// dataPrefix handles the internal {"data": ...} wrapper transparently
|
||
output: {
|
||
name: "output",
|
||
...column("JSON", {
|
||
description: "The data you returned from the task.",
|
||
example: '{"result": "success"}',
|
||
}),
|
||
nullValue: "'{}'", // Transform NULL checks to compare against empty object
|
||
textColumn: "output_text", // Use output_text for full JSON value queries
|
||
dataPrefix: "data", // Internal data is wrapped in {"data": ...}
|
||
},
|
||
error: {
|
||
name: "error",
|
||
...column("JSON", {
|
||
description:
|
||
"If a run completely failed (after all attempts) then this error will be populated.",
|
||
example: '{"message": "Task failed"}',
|
||
}),
|
||
nullValue: "'{}'", // Transform NULL checks to compare against empty object
|
||
textColumn: "error_text", // Use error_text for full JSON value queries
|
||
dataPrefix: "data", // Internal data is wrapped in {"data": ...}
|
||
},
|
||
|
||
// Tags & versions
|
||
tags: {
|
||
name: "tags",
|
||
...column("Array(String)", {
|
||
description: "Tags you have added to the run.",
|
||
customRenderType: "tags",
|
||
example: '["user:123", "priority:high"]',
|
||
}),
|
||
},
|
||
task_version: {
|
||
name: "task_version",
|
||
...column("String", {
|
||
description: "The version of your code in reverse date format.",
|
||
example: "20240115.1",
|
||
}),
|
||
},
|
||
sdk_version: {
|
||
name: "sdk_version",
|
||
...column("String", {
|
||
description: "The SDK package version for this run.",
|
||
example: "3.3.0",
|
||
}),
|
||
},
|
||
cli_version: {
|
||
name: "cli_version",
|
||
...column("String", {
|
||
description: "The CLI package version for this run.",
|
||
example: "3.3.0",
|
||
}),
|
||
},
|
||
machine: {
|
||
name: "machine",
|
||
clickhouseName: "machine_preset",
|
||
...column("LowCardinality(String)", {
|
||
description: "The machine that the run executed on.",
|
||
allowedValues: [...MACHINE_PRESETS],
|
||
customRenderType: "machine",
|
||
example: "small-1x",
|
||
}),
|
||
},
|
||
is_test: {
|
||
name: "is_test",
|
||
...column("UInt8", { description: "Whether this is a test run (0 or 1)", example: "0" }),
|
||
expression: "if(is_test > 0, true, false)",
|
||
},
|
||
is_warm_start: {
|
||
name: "is_warm_start",
|
||
...column("Nullable(UInt8)", {
|
||
description: "Whether this run used a warm start vs a cold start.",
|
||
example: "1",
|
||
}),
|
||
},
|
||
concurrency_key: {
|
||
name: "concurrency_key",
|
||
...column("String", {
|
||
description: "The concurrency key you passed in when triggering the run.",
|
||
example: "user:1234567",
|
||
}),
|
||
},
|
||
max_duration: {
|
||
name: "max_duration",
|
||
clickhouseName: "max_duration_in_seconds",
|
||
...column("Nullable(UInt32)", {
|
||
description:
|
||
"The maximum allowed compute duration for this run in seconds. If the run exceeds this duration, the run will fail with an error. Can be set on an individual task, in the trigger.config, or per-run when triggering.",
|
||
example: "300",
|
||
customRenderType: "durationSeconds",
|
||
}),
|
||
},
|
||
bulk_action_group_ids: {
|
||
name: "bulk_action_group_ids",
|
||
...column("Array(String)", {
|
||
description: "Any bulk actions that operated on this run.",
|
||
example: '["bulk_12345678", "bulk_34567890"]',
|
||
whereTransform: (value: string) => value.replace(/^bulk_/, ""),
|
||
}),
|
||
},
|
||
},
|
||
};
|
||
|
||
/**
|
||
* Schema definition for the metrics table (trigger_dev.metrics_v1)
|
||
*/
|
||
export const metricsSchema: TableSchema = {
|
||
name: "metrics",
|
||
clickhouseName: "trigger_dev.metrics_v1",
|
||
description: "Host and runtime metrics collected during task execution",
|
||
timeConstraint: "bucket_start",
|
||
tenantColumns: {
|
||
organizationId: "organization_id",
|
||
projectId: "project_id",
|
||
environmentId: "environment_id",
|
||
},
|
||
columns: {
|
||
environment: {
|
||
name: "environment",
|
||
clickhouseName: "environment_id",
|
||
...column("String", { description: "The environment slug", example: "prod" }),
|
||
fieldMapping: "environment",
|
||
customRenderType: "environment",
|
||
},
|
||
project: {
|
||
name: "project",
|
||
clickhouseName: "project_id",
|
||
...column("String", {
|
||
description: "The project reference, they always start with `proj_`.",
|
||
example: "proj_howcnaxbfxdmwmxazktx",
|
||
}),
|
||
fieldMapping: "project",
|
||
customRenderType: "project",
|
||
},
|
||
metric_name: {
|
||
name: "metric_name",
|
||
...column("LowCardinality(String)", {
|
||
description: "The name of the metric (e.g. process.cpu.utilization, system.memory.usage)",
|
||
example: "process.cpu.utilization",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
metric_type: {
|
||
name: "metric_type",
|
||
...column("LowCardinality(String)", {
|
||
description: "The type of metric",
|
||
allowedValues: ["gauge", "sum", "histogram"],
|
||
example: "gauge",
|
||
}),
|
||
},
|
||
machine_id: {
|
||
name: "machine_id",
|
||
clickhouseName: "metric_subject",
|
||
...column("String", {
|
||
description: "The machine ID that produced this metric",
|
||
example: "machine-abc123",
|
||
}),
|
||
},
|
||
bucket_start: {
|
||
name: "bucket_start",
|
||
...column("DateTime", {
|
||
description: "The start of the 10-second aggregation bucket",
|
||
example: "2024-01-15 09:30:00",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
metric_value: {
|
||
name: "metric_value",
|
||
clickhouseName: "value",
|
||
...column("Float64", {
|
||
description: "The metric value",
|
||
example: "0.75",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
|
||
// Attributes (JSON column for user-defined and system attributes)
|
||
attributes: {
|
||
name: "attributes",
|
||
...column("JSON", {
|
||
description: "JSON attributes attached to the metric data point.",
|
||
example: '{"region": "us-east-1"}',
|
||
}),
|
||
},
|
||
|
||
// Trigger context columns (from attributes.trigger.* JSON subpaths)
|
||
run_id: {
|
||
name: "run_id",
|
||
...column("String", {
|
||
description: "The run ID associated with this metric",
|
||
customRenderType: "runId",
|
||
example: "run_cm1a2b3c4d5e6f7g8h9i",
|
||
coreColumn: true,
|
||
}),
|
||
expression: "attributes.trigger.run_id",
|
||
},
|
||
task_identifier: {
|
||
name: "task_identifier",
|
||
...column("String", {
|
||
description: "Task identifier/slug",
|
||
example: "my-background-task",
|
||
coreColumn: true,
|
||
}),
|
||
expression: "attributes.trigger.task_slug",
|
||
},
|
||
attempt_number: {
|
||
name: "attempt_number",
|
||
...column("UInt64", {
|
||
description: "The attempt number for this metric",
|
||
example: "1",
|
||
}),
|
||
expression: "attributes.trigger.attempt_number",
|
||
},
|
||
machine_name: {
|
||
name: "machine_name",
|
||
...column("String", {
|
||
description: "The machine preset used for execution",
|
||
allowedValues: [...MACHINE_PRESETS],
|
||
example: "small-1x",
|
||
}),
|
||
expression: "attributes.trigger.machine_name",
|
||
},
|
||
environment_type: {
|
||
name: "environment_type",
|
||
...column("String", {
|
||
description: "Environment type",
|
||
allowedValues: [...ENVIRONMENT_TYPES],
|
||
customRenderType: "environmentType",
|
||
example: "PRODUCTION",
|
||
}),
|
||
expression: "attributes.trigger.environment_type",
|
||
},
|
||
worker_id: {
|
||
name: "worker_id",
|
||
...column("String", {
|
||
description: "The worker ID that produced this metric",
|
||
customRenderType: "deploymentId",
|
||
example: "deployment_cm1a2b3c4d5e",
|
||
}),
|
||
expression: "attributes.trigger.worker_id",
|
||
},
|
||
worker_version: {
|
||
name: "worker_version",
|
||
...column("String", {
|
||
description: "The worker version that produced this metric",
|
||
example: "20240115.1",
|
||
}),
|
||
expression: "attributes.trigger.worker_version",
|
||
},
|
||
},
|
||
timeBucketThresholds: [
|
||
// Metrics are pre-aggregated into 10-second buckets, so 10s is the most granular interval.
|
||
// All thresholds are shifted coarser compared to the runs table defaults.
|
||
{ maxRangeSeconds: 3 * 60 * 60, interval: { value: 10, unit: "SECOND" } },
|
||
{ maxRangeSeconds: 12 * 60 * 60, interval: { value: 1, unit: "MINUTE" } },
|
||
{ maxRangeSeconds: 2 * 24 * 60 * 60, interval: { value: 5, unit: "MINUTE" } },
|
||
{ maxRangeSeconds: 7 * 24 * 60 * 60, interval: { value: 15, unit: "MINUTE" } },
|
||
{ maxRangeSeconds: 30 * 24 * 60 * 60, interval: { value: 1, unit: "HOUR" } },
|
||
{ maxRangeSeconds: 90 * 24 * 60 * 60, interval: { value: 6, unit: "HOUR" } },
|
||
{ maxRangeSeconds: 180 * 24 * 60 * 60, interval: { value: 1, unit: "DAY" } },
|
||
{ maxRangeSeconds: 365 * 24 * 60 * 60, interval: { value: 1, unit: "WEEK" } },
|
||
] satisfies BucketThreshold[],
|
||
};
|
||
|
||
/**
|
||
* Schema definition for the queue_metrics table (trigger_dev.queue_metrics_v1).
|
||
* Pre-aggregated into 10-second buckets. Counter columns re-aggregate with sum(),
|
||
* gauges with max(), and wait_quantiles with quantilesMerge() — never FINAL.
|
||
*/
|
||
export const queueMetricsSchema: TableSchema = {
|
||
name: "queue_metrics",
|
||
clickhouseName: "trigger_dev.queue_metrics_v1",
|
||
description: "Per-queue depth, concurrency, throttling, and scheduling-delay metrics",
|
||
timeConstraint: "bucket_start",
|
||
tenantColumns: {
|
||
organizationId: "organization_id",
|
||
projectId: "project_id",
|
||
environmentId: "environment_id",
|
||
},
|
||
columns: {
|
||
environment: {
|
||
name: "environment",
|
||
clickhouseName: "environment_id",
|
||
...column("String", { description: "The environment slug", example: "prod" }),
|
||
fieldMapping: "environment",
|
||
customRenderType: "environment",
|
||
},
|
||
project: {
|
||
name: "project",
|
||
clickhouseName: "project_id",
|
||
...column("String", {
|
||
description: "The project reference, they always start with `proj_`.",
|
||
example: "proj_howcnaxbfxdmwmxazktx",
|
||
}),
|
||
fieldMapping: "project",
|
||
customRenderType: "project",
|
||
},
|
||
queue: {
|
||
name: "queue",
|
||
clickhouseName: "queue_name",
|
||
...column("LowCardinality(String)", {
|
||
description: "The queue name",
|
||
example: "my-queue",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
bucket_start: {
|
||
name: "bucket_start",
|
||
...column("DateTime", {
|
||
description: "The start of the 10-second aggregation bucket",
|
||
example: "2024-01-15 09:30:00",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
// Cumulative-counter delta states. Read with deltaSumTimestampMerge(<col>) (loss-tolerant,
|
||
// reset-safe), never sum(); opaque like wait_quantiles. Merging across queues is
|
||
// invalid (mixes unrelated odometers): totals must GROUP BY queue, then sum outside.
|
||
enqueue_delta: {
|
||
name: "enqueue_delta",
|
||
mergeGroupKey: "queue",
|
||
...column("String", {
|
||
description:
|
||
"Runs enqueued (cumulative-counter delta). Read with deltaSumTimestampMerge(enqueue_delta) grouped by queue. For totals across queues, sum the per-queue results in an outer query, never merge across queues. Per-bucket values can undercount by one inter-reading delta at bucket boundaries (the bridge lives in the prior bucket's state); totals over the whole range are exact.",
|
||
}),
|
||
groupable: false,
|
||
sortable: false,
|
||
filterable: false,
|
||
},
|
||
started_delta: {
|
||
name: "started_delta",
|
||
mergeGroupKey: "queue",
|
||
...column("String", {
|
||
description:
|
||
"Runs dequeued/started (throughput). Read with deltaSumTimestampMerge(started_delta) grouped by queue. For totals across queues, sum the per-queue results in an outer query, never merge across queues. Per-bucket values can undercount by one inter-reading delta at bucket boundaries (the bridge lives in the prior bucket's state); totals over the whole range are exact.",
|
||
coreColumn: true,
|
||
}),
|
||
groupable: false,
|
||
sortable: false,
|
||
filterable: false,
|
||
},
|
||
ack_delta: {
|
||
name: "ack_delta",
|
||
mergeGroupKey: "queue",
|
||
...column("String", {
|
||
description:
|
||
"Runs acked (completed). Read with deltaSumTimestampMerge(ack_delta) grouped by queue; sum per-queue results for totals.",
|
||
}),
|
||
groupable: false,
|
||
sortable: false,
|
||
filterable: false,
|
||
},
|
||
nack_delta: {
|
||
name: "nack_delta",
|
||
mergeGroupKey: "queue",
|
||
...column("String", {
|
||
description:
|
||
"Runs nacked. Read with deltaSumTimestampMerge(nack_delta) grouped by queue; sum per-queue results for totals.",
|
||
}),
|
||
groupable: false,
|
||
sortable: false,
|
||
filterable: false,
|
||
},
|
||
dlq_delta: {
|
||
name: "dlq_delta",
|
||
mergeGroupKey: "queue",
|
||
...column("String", {
|
||
description:
|
||
"Runs dead-lettered. Read with deltaSumTimestampMerge(dlq_delta) grouped by queue; sum per-queue results for totals.",
|
||
}),
|
||
groupable: false,
|
||
sortable: false,
|
||
filterable: false,
|
||
},
|
||
throttled_count: {
|
||
name: "throttled_count",
|
||
...column("UInt64", {
|
||
description: "Gauge emissions where running>=limit and queued>0. Aggregate with sum().",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
max_queued: {
|
||
name: "max_queued",
|
||
...column("UInt32", {
|
||
description: "Peak queue depth in the bucket. Aggregate with max().",
|
||
coreColumn: true,
|
||
fillMode: "carry",
|
||
}),
|
||
},
|
||
max_running: {
|
||
name: "max_running",
|
||
...column("UInt32", {
|
||
description: "Peak running (concurrency) in the bucket. Aggregate with max().",
|
||
coreColumn: true,
|
||
fillMode: "carry",
|
||
}),
|
||
},
|
||
max_limit: {
|
||
name: "max_limit",
|
||
...column("UInt32", {
|
||
description: "The queue concurrency limit. Aggregate with max().",
|
||
coreColumn: true,
|
||
fillMode: "carry",
|
||
}),
|
||
},
|
||
max_env_queued: {
|
||
name: "max_env_queued",
|
||
...column("UInt32", {
|
||
description: "Peak environment-wide queued in the bucket. Aggregate with max().",
|
||
fillMode: "carry",
|
||
}),
|
||
},
|
||
max_env_running: {
|
||
name: "max_env_running",
|
||
...column("UInt32", {
|
||
description: "Peak environment-wide running in the bucket. Aggregate with max().",
|
||
fillMode: "carry",
|
||
}),
|
||
},
|
||
max_env_limit: {
|
||
name: "max_env_limit",
|
||
...column("UInt32", {
|
||
description: "The environment concurrency limit. Aggregate with max().",
|
||
fillMode: "carry",
|
||
}),
|
||
},
|
||
max_ck_backlogged: {
|
||
name: "max_ck_backlogged",
|
||
...column("UInt32", {
|
||
description:
|
||
"Peak number of distinct concurrency keys with queued runs in the bucket. Aggregate with max(). Zero for queues that do not use concurrency keys.",
|
||
fillMode: "carry",
|
||
}),
|
||
},
|
||
max_ck_wait_ms: {
|
||
name: "max_ck_wait_ms",
|
||
...column("UInt32", {
|
||
description:
|
||
"Worst head-of-line wait (ms) across concurrency keys in the bucket: how long the most-starved key's oldest queued run has been waiting. Aggregate with max(). Zero for queues that do not use concurrency keys.",
|
||
fillMode: "carry",
|
||
}),
|
||
},
|
||
wait_ms_sum: {
|
||
name: "wait_ms_sum",
|
||
...column("UInt64", {
|
||
description: "Sum of scheduling delays (ms). Mean = wait_ms_sum/wait_ms_count.",
|
||
}),
|
||
},
|
||
wait_ms_count: {
|
||
name: "wait_ms_count",
|
||
...column("UInt64", {
|
||
description: "Count of scheduling-delay samples. Aggregate with sum().",
|
||
}),
|
||
},
|
||
wait_quantiles: {
|
||
name: "wait_quantiles",
|
||
...column("String", {
|
||
description:
|
||
"Scheduling-delay (dequeue minus eligible-at) quantile state. Read with quantilesMerge(0.5,0.9,0.95,0.99)(wait_quantiles)[n].",
|
||
}),
|
||
groupable: false,
|
||
sortable: false,
|
||
filterable: false,
|
||
},
|
||
},
|
||
timeBucketThresholds: [
|
||
{ maxRangeSeconds: 3 * 60 * 60, interval: { value: 10, unit: "SECOND" } },
|
||
{ maxRangeSeconds: 12 * 60 * 60, interval: { value: 1, unit: "MINUTE" } },
|
||
{ maxRangeSeconds: 2 * 24 * 60 * 60, interval: { value: 5, unit: "MINUTE" } },
|
||
{ maxRangeSeconds: 7 * 24 * 60 * 60, interval: { value: 15, unit: "MINUTE" } },
|
||
{ maxRangeSeconds: 30 * 24 * 60 * 60, interval: { value: 1, unit: "HOUR" } },
|
||
{ maxRangeSeconds: 90 * 24 * 60 * 60, interval: { value: 6, unit: "HOUR" } },
|
||
{ maxRangeSeconds: 180 * 24 * 60 * 60, interval: { value: 1, unit: "DAY" } },
|
||
{ maxRangeSeconds: 365 * 24 * 60 * 60, interval: { value: 1, unit: "WEEK" } },
|
||
] satisfies BucketThreshold[],
|
||
// Ranges whose bucket interval is >= 5 minutes read the 5m rollup instead (same
|
||
// logical columns, ~30x fewer rows).
|
||
rollups: [{ minIntervalSeconds: 300, clickhouseName: "trigger_dev.queue_metrics_5m_v1" }],
|
||
queryCache: { ttlSeconds: 30, alignSeconds: 30 },
|
||
queryClient: "queueMetrics",
|
||
};
|
||
|
||
/**
|
||
* Schema definition for the env_metrics table (trigger_dev.env_metrics_v1).
|
||
* Environment-level rollup of queue_metrics with the queue dimension dropped, so
|
||
* header tiles and saturation charts cost the same regardless of how many queues
|
||
* the environment has. Keeps the full 10-second granularity: row count is
|
||
* queue-independent, so even 30-day ranges stay small.
|
||
*/
|
||
export const envMetricsSchema: TableSchema = {
|
||
name: "env_metrics",
|
||
clickhouseName: "trigger_dev.env_metrics_v1",
|
||
description:
|
||
"Environment-level concurrency, saturation, throttling, and scheduling-delay metrics (10-second buckets)",
|
||
timeConstraint: "bucket_start",
|
||
tenantColumns: {
|
||
organizationId: "organization_id",
|
||
projectId: "project_id",
|
||
environmentId: "environment_id",
|
||
},
|
||
columns: {
|
||
environment: {
|
||
name: "environment",
|
||
clickhouseName: "environment_id",
|
||
...column("String", { description: "The environment slug", example: "prod" }),
|
||
fieldMapping: "environment",
|
||
customRenderType: "environment",
|
||
},
|
||
project: {
|
||
name: "project",
|
||
clickhouseName: "project_id",
|
||
...column("String", {
|
||
description: "The project reference, they always start with `proj_`.",
|
||
example: "proj_howcnaxbfxdmwmxazktx",
|
||
}),
|
||
fieldMapping: "project",
|
||
customRenderType: "project",
|
||
},
|
||
bucket_start: {
|
||
name: "bucket_start",
|
||
...column("DateTime", {
|
||
description: "The start of the 10-second aggregation bucket",
|
||
example: "2024-01-15 09:30:00",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
max_env_queued: {
|
||
name: "max_env_queued",
|
||
...column("UInt32", {
|
||
description: "Peak environment-wide queued in the bucket. Aggregate with max().",
|
||
coreColumn: true,
|
||
fillMode: "carry",
|
||
}),
|
||
},
|
||
max_env_running: {
|
||
name: "max_env_running",
|
||
...column("UInt32", {
|
||
description: "Peak environment-wide running in the bucket. Aggregate with max().",
|
||
coreColumn: true,
|
||
fillMode: "carry",
|
||
}),
|
||
},
|
||
max_env_limit: {
|
||
name: "max_env_limit",
|
||
...column("UInt32", {
|
||
description: "The environment concurrency limit. Aggregate with max().",
|
||
coreColumn: true,
|
||
fillMode: "carry",
|
||
}),
|
||
},
|
||
throttled_count: {
|
||
name: "throttled_count",
|
||
...column("UInt64", {
|
||
description:
|
||
"Gauge emissions where a queue was at its limit with work queued. Aggregate with sum().",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
wait_ms_sum: {
|
||
name: "wait_ms_sum",
|
||
...column("UInt64", {
|
||
description: "Sum of scheduling delays (ms). Mean = wait_ms_sum/wait_ms_count.",
|
||
}),
|
||
},
|
||
wait_ms_count: {
|
||
name: "wait_ms_count",
|
||
...column("UInt64", {
|
||
description: "Count of scheduling-delay samples. Aggregate with sum().",
|
||
}),
|
||
},
|
||
wait_quantiles: {
|
||
name: "wait_quantiles",
|
||
...column("String", {
|
||
description:
|
||
"Scheduling-delay quantile state (TDigest). Read with quantilesTDigestMerge(0.5,0.9,0.95,0.99)(wait_quantiles)[n].",
|
||
}),
|
||
groupable: false,
|
||
sortable: false,
|
||
filterable: false,
|
||
},
|
||
},
|
||
timeBucketThresholds: [
|
||
{ maxRangeSeconds: 3 * 60 * 60, interval: { value: 10, unit: "SECOND" } },
|
||
{ maxRangeSeconds: 12 * 60 * 60, interval: { value: 1, unit: "MINUTE" } },
|
||
{ maxRangeSeconds: 2 * 24 * 60 * 60, interval: { value: 5, unit: "MINUTE" } },
|
||
{ maxRangeSeconds: 7 * 24 * 60 * 60, interval: { value: 15, unit: "MINUTE" } },
|
||
{ maxRangeSeconds: 30 * 24 * 60 * 60, interval: { value: 1, unit: "HOUR" } },
|
||
{ maxRangeSeconds: 90 * 24 * 60 * 60, interval: { value: 6, unit: "HOUR" } },
|
||
{ maxRangeSeconds: 180 * 24 * 60 * 60, interval: { value: 1, unit: "DAY" } },
|
||
{ maxRangeSeconds: 365 * 24 * 60 * 60, interval: { value: 1, unit: "WEEK" } },
|
||
] satisfies BucketThreshold[],
|
||
queryCache: { ttlSeconds: 30, alignSeconds: 30 },
|
||
queryClient: "queueMetrics",
|
||
};
|
||
|
||
/**
|
||
* Schema definition for the llm_metrics table (trigger_dev.llm_metrics_v1)
|
||
*/
|
||
export const llmMetricsSchema: TableSchema = {
|
||
name: "llm_metrics",
|
||
clickhouseName: "trigger_dev.llm_metrics_v1",
|
||
description: "LLM metrics: token usage, cost, performance, and behavior from GenAI spans",
|
||
timeConstraint: "start_time",
|
||
tenantColumns: {
|
||
organizationId: "organization_id",
|
||
projectId: "project_id",
|
||
environmentId: "environment_id",
|
||
},
|
||
columns: {
|
||
environment: {
|
||
name: "environment",
|
||
clickhouseName: "environment_id",
|
||
...column("String", { description: "The environment slug", example: "prod" }),
|
||
fieldMapping: "environment",
|
||
customRenderType: "environment",
|
||
},
|
||
project: {
|
||
name: "project",
|
||
clickhouseName: "project_id",
|
||
...column("String", {
|
||
description: "The project reference, they always start with `proj_`.",
|
||
example: "proj_howcnaxbfxdmwmxazktx",
|
||
}),
|
||
fieldMapping: "project",
|
||
customRenderType: "project",
|
||
},
|
||
run_id: {
|
||
name: "run_id",
|
||
...column("String", {
|
||
description: "The run ID",
|
||
customRenderType: "runId",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
trace_id: {
|
||
name: "trace_id",
|
||
...column("String", {
|
||
description: "The trace ID",
|
||
}),
|
||
},
|
||
span_id: {
|
||
name: "span_id",
|
||
...column("String", {
|
||
description: "The span ID",
|
||
}),
|
||
},
|
||
task_identifier: {
|
||
name: "task_identifier",
|
||
...column("LowCardinality(String)", {
|
||
description: "The task identifier",
|
||
example: "my-task",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
gen_ai_system: {
|
||
name: "gen_ai_system",
|
||
...column("LowCardinality(String)", {
|
||
description: "AI provider (e.g. openai, anthropic)",
|
||
example: "openai",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
request_model: {
|
||
name: "request_model",
|
||
...column("String", {
|
||
description: "The model name requested",
|
||
example: "gpt-4o",
|
||
}),
|
||
},
|
||
response_model: {
|
||
name: "response_model",
|
||
...column("String", {
|
||
description: "The model name returned by the provider",
|
||
example: "gpt-4o-2024-08-06",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
operation_id: {
|
||
name: "operation_id",
|
||
...column("LowCardinality(String)", {
|
||
description: "Operation type (e.g. ai.streamText.doStream, ai.generateText.doGenerate)",
|
||
example: "ai.streamText.doStream",
|
||
}),
|
||
},
|
||
finish_reason: {
|
||
name: "finish_reason",
|
||
...column("LowCardinality(String)", {
|
||
description: "Why the LLM stopped generating (e.g. stop, tool-calls, length)",
|
||
example: "stop",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
cost_source: {
|
||
name: "cost_source",
|
||
...column("LowCardinality(String)", {
|
||
description: "Where cost data came from (registry, gateway, openrouter)",
|
||
example: "registry",
|
||
}),
|
||
},
|
||
input_tokens: {
|
||
name: "input_tokens",
|
||
...column("UInt64", {
|
||
description: "Number of input tokens",
|
||
example: "702",
|
||
}),
|
||
},
|
||
output_tokens: {
|
||
name: "output_tokens",
|
||
...column("UInt64", {
|
||
description: "Number of output tokens",
|
||
example: "22",
|
||
}),
|
||
},
|
||
total_tokens: {
|
||
name: "total_tokens",
|
||
...column("UInt64", {
|
||
description: "Total token count",
|
||
example: "724",
|
||
}),
|
||
},
|
||
cached_read_tokens: {
|
||
name: "cached_read_tokens",
|
||
...column("UInt64", {
|
||
description:
|
||
"Input tokens served from the provider's prompt cache (cheaper than regular input tokens). Supported by Anthropic and OpenAI.",
|
||
example: "8200",
|
||
}),
|
||
expression: "usage_details['input_cached_tokens']",
|
||
},
|
||
cache_creation_tokens: {
|
||
name: "cache_creation_tokens",
|
||
...column("UInt64", {
|
||
description:
|
||
"Input tokens written to create a new prompt cache entry. Supported by Anthropic.",
|
||
example: "1751",
|
||
}),
|
||
expression: "usage_details['cache_creation_input_tokens']",
|
||
},
|
||
reasoning_tokens: {
|
||
name: "reasoning_tokens",
|
||
...column("UInt64", {
|
||
description:
|
||
"Tokens used for chain-of-thought reasoning (e.g. OpenAI o-series, DeepSeek R1). These count toward output but are not visible in the response.",
|
||
example: "512",
|
||
}),
|
||
expression: "usage_details['reasoning_tokens']",
|
||
},
|
||
input_cost: {
|
||
name: "input_cost",
|
||
...column("Decimal64(12)", {
|
||
description: "Input cost in USD (from pricing registry)",
|
||
customRenderType: "costInDollars",
|
||
}),
|
||
},
|
||
output_cost: {
|
||
name: "output_cost",
|
||
...column("Decimal64(12)", {
|
||
description: "Output cost in USD (from pricing registry)",
|
||
customRenderType: "costInDollars",
|
||
}),
|
||
},
|
||
total_cost: {
|
||
name: "total_cost",
|
||
...column("Decimal64(12)", {
|
||
description: "Total cost in USD",
|
||
customRenderType: "costInDollars",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
cached_read_cost: {
|
||
name: "cached_read_cost",
|
||
...column("Decimal64(12)", {
|
||
description:
|
||
"Cost of cached input tokens (discounted vs regular input). Only present when the pricing tier has a separate cached input price.",
|
||
customRenderType: "costInDollars",
|
||
}),
|
||
expression: "cost_details['input_cached_tokens']",
|
||
},
|
||
cache_creation_cost: {
|
||
name: "cache_creation_cost",
|
||
...column("Decimal64(12)", {
|
||
description: "Cost of tokens written to create a prompt cache entry.",
|
||
customRenderType: "costInDollars",
|
||
}),
|
||
expression: "cost_details['cache_creation_input_tokens']",
|
||
},
|
||
provider_cost: {
|
||
name: "provider_cost",
|
||
...column("Decimal64(12)", {
|
||
description: "Provider-reported cost in USD (from gateway or openrouter)",
|
||
customRenderType: "costInDollars",
|
||
}),
|
||
},
|
||
ms_to_first_chunk: {
|
||
name: "ms_to_first_chunk",
|
||
...column("Float64", {
|
||
description: "Time to first chunk in milliseconds (TTFC)",
|
||
example: "245.3",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
tokens_per_second: {
|
||
name: "tokens_per_second",
|
||
...column("Float64", {
|
||
description: "Average output tokens per second",
|
||
example: "72.5",
|
||
}),
|
||
},
|
||
pricing_tier_name: {
|
||
name: "pricing_tier_name",
|
||
...column("LowCardinality(String)", {
|
||
description: "The matched pricing tier name",
|
||
example: "Standard",
|
||
}),
|
||
},
|
||
start_time: {
|
||
name: "start_time",
|
||
...column("DateTime64(9)", {
|
||
description: "When the LLM call started",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
duration: {
|
||
name: "duration",
|
||
...column("UInt64", {
|
||
description: "Span duration in nanoseconds",
|
||
customRenderType: "durationNs",
|
||
}),
|
||
},
|
||
prompt_slug: {
|
||
name: "prompt_slug",
|
||
...column("LowCardinality(String)", {
|
||
description: "The managed prompt slug used for this LLM call",
|
||
example: "customer-support",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
prompt_version: {
|
||
name: "prompt_version",
|
||
...column("UInt32", {
|
||
description: "The managed prompt version number used for this LLM call",
|
||
example: "3",
|
||
}),
|
||
},
|
||
metadata: {
|
||
name: "metadata",
|
||
...column("Map(LowCardinality(String), String)", {
|
||
description:
|
||
"Key-value metadata from run tags (key:value format) and AI SDK telemetry metadata. Access keys with dot notation (metadata.userId) or bracket syntax (metadata['userId']).",
|
||
example: "{'userId':'user_123','org':'acme'}",
|
||
}),
|
||
},
|
||
},
|
||
};
|
||
|
||
/**
|
||
* Schema definition for the llm_models table (trigger_dev.llm_model_aggregates_v1)
|
||
* Global table — no tenant columns. Contains anonymized cross-tenant model performance data.
|
||
*/
|
||
export const llmModelsSchema: TableSchema = {
|
||
name: "llm_models",
|
||
clickhouseName: "trigger_dev.llm_model_aggregates_v1",
|
||
description:
|
||
"Cross-tenant model performance aggregates: calls, cost, latency, and throughput per model per minute. No tenant-specific data.",
|
||
timeConstraint: "minute",
|
||
// No tenantColumns — this is a global table with anonymized data
|
||
columns: {
|
||
response_model: {
|
||
name: "response_model",
|
||
...column("String", {
|
||
description: "The model name as returned by the provider",
|
||
example: "gpt-4o-2024-08-06",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
base_response_model: {
|
||
name: "base_response_model",
|
||
...column("String", {
|
||
description: "The base model name with dated variants grouped",
|
||
example: "gpt-4o",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
gen_ai_system: {
|
||
name: "gen_ai_system",
|
||
...column("String", {
|
||
description: "The AI provider system identifier",
|
||
example: "openai.responses",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
minute: {
|
||
name: "minute",
|
||
...column("DateTime", {
|
||
description: "Aggregation time bucket (per minute)",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
call_count: {
|
||
name: "call_count",
|
||
...column("UInt64", {
|
||
description: "Number of LLM calls in this time bucket",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
total_input_tokens: {
|
||
name: "total_input_tokens",
|
||
...column("UInt64", {
|
||
description: "Total input tokens consumed",
|
||
}),
|
||
},
|
||
total_output_tokens: {
|
||
name: "total_output_tokens",
|
||
...column("UInt64", {
|
||
description: "Total output tokens generated",
|
||
}),
|
||
},
|
||
total_cost: {
|
||
name: "total_cost",
|
||
...column("Float64", {
|
||
description: "Total cost in USD",
|
||
customRenderType: "costInDollars",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
// Aggregate state columns — use quantilesMerge() in queries to extract values
|
||
// Example: quantilesMerge(0.5)(ttfc_quantiles)[1] AS ttfc_p50
|
||
ttfc_quantiles: {
|
||
name: "ttfc_quantiles",
|
||
...column("String", {
|
||
description:
|
||
"Time to first chunk quantile state. Use quantilesMerge(0.5)(ttfc_quantiles)[1] AS ttfc_p50 in queries.",
|
||
example: "quantilesMerge(0.5)(ttfc_quantiles)[1]",
|
||
}),
|
||
},
|
||
tps_quantiles: {
|
||
name: "tps_quantiles",
|
||
...column("String", {
|
||
description:
|
||
"Tokens per second quantile state. Use quantilesMerge(0.5)(tps_quantiles)[1] AS tps_p50 in queries.",
|
||
example: "quantilesMerge(0.5)(tps_quantiles)[1]",
|
||
}),
|
||
},
|
||
duration_quantiles: {
|
||
name: "duration_quantiles",
|
||
...column("String", {
|
||
description:
|
||
"Duration quantile state. Use quantilesMerge(0.5)(duration_quantiles)[1] AS duration_p50 in queries.",
|
||
example: "quantilesMerge(0.5)(duration_quantiles)[1]",
|
||
}),
|
||
},
|
||
},
|
||
};
|
||
|
||
/**
|
||
* Per-concurrency-key drill-down for queues that shard work with `concurrencyKey`
|
||
* (e.g. per-tenant fairness). Rows are activity-bound: a (queue, key, bucket) row exists
|
||
* only when that key had events, so key cardinality cannot inflate the table.
|
||
*/
|
||
export const queueMetricsByKeySchema: TableSchema = {
|
||
name: "queue_metrics_by_key",
|
||
clickhouseName: "trigger_dev.queue_metrics_ck_v1",
|
||
description: "Per-concurrency-key queue metrics: backlog, throughput, and wait by key",
|
||
hidden: true,
|
||
timeConstraint: "bucket_start",
|
||
tenantColumns: {
|
||
organizationId: "organization_id",
|
||
projectId: "project_id",
|
||
environmentId: "environment_id",
|
||
},
|
||
columns: {
|
||
environment: {
|
||
name: "environment",
|
||
clickhouseName: "environment_id",
|
||
...column("String", { description: "The environment slug", example: "prod" }),
|
||
fieldMapping: "environment",
|
||
customRenderType: "environment",
|
||
},
|
||
project: {
|
||
name: "project",
|
||
clickhouseName: "project_id",
|
||
...column("String", {
|
||
description: "The project reference, they always start with `proj_`.",
|
||
example: "proj_howcnaxbfxdmwmxazktx",
|
||
}),
|
||
fieldMapping: "project",
|
||
customRenderType: "project",
|
||
},
|
||
queue: {
|
||
name: "queue",
|
||
clickhouseName: "queue_name",
|
||
...column("LowCardinality(String)", {
|
||
description: "The queue name",
|
||
example: "my-queue",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
concurrency_key: {
|
||
name: "concurrency_key",
|
||
...column("String", {
|
||
description: "The concurrency key the run was sharded by (e.g. a tenant id)",
|
||
example: "tenant-42",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
bucket_start: {
|
||
name: "bucket_start",
|
||
...column("DateTime", {
|
||
description: "The start of the 10-second aggregation bucket",
|
||
example: "2024-01-15 09:30:00",
|
||
coreColumn: true,
|
||
}),
|
||
},
|
||
enqueue_delta: {
|
||
name: "enqueue_delta",
|
||
mergeGroupKey: ["queue", "concurrency_key"],
|
||
...column("String", {
|
||
description:
|
||
"Runs enqueued for this key (cumulative-counter delta). Read with deltaSumTimestampMerge(enqueue_delta) grouped by queue and concurrency_key, or with both pinned; never merge across keys.",
|
||
}),
|
||
groupable: false,
|
||
sortable: false,
|
||
filterable: false,
|
||
},
|
||
started_delta: {
|
||
name: "started_delta",
|
||
mergeGroupKey: ["queue", "concurrency_key"],
|
||
...column("String", {
|
||
description:
|
||
"Runs dequeued/started for this key (throughput). Read with deltaSumTimestampMerge(started_delta) grouped by queue and concurrency_key, or with both pinned; never merge across keys.",
|
||
coreColumn: true,
|
||
}),
|
||
groupable: false,
|
||
sortable: false,
|
||
filterable: false,
|
||
},
|
||
ack_delta: {
|
||
name: "ack_delta",
|
||
mergeGroupKey: ["queue", "concurrency_key"],
|
||
...column("String", {
|
||
description:
|
||
"Runs acked (completed) for this key. Read with deltaSumTimestampMerge(ack_delta) grouped by queue and concurrency_key, or with both pinned.",
|
||
}),
|
||
groupable: false,
|
||
sortable: false,
|
||
filterable: false,
|
||
},
|
||
max_queued: {
|
||
name: "max_queued",
|
||
...column("UInt32", {
|
||
description: "Peak backlog for this key in the bucket. Aggregate with max().",
|
||
coreColumn: true,
|
||
fillMode: "carry",
|
||
}),
|
||
},
|
||
max_running: {
|
||
name: "max_running",
|
||
...column("UInt32", {
|
||
description: "Peak running for this key in the bucket. Aggregate with max().",
|
||
fillMode: "carry",
|
||
}),
|
||
},
|
||
wait_ms_sum: {
|
||
name: "wait_ms_sum",
|
||
...column("UInt64", {
|
||
description:
|
||
"Sum of scheduling delays (ms) for this key. Mean = wait_ms_sum/wait_ms_count.",
|
||
}),
|
||
},
|
||
wait_ms_count: {
|
||
name: "wait_ms_count",
|
||
...column("UInt64", {
|
||
description: "Count of scheduling-delay samples for this key. Aggregate with sum().",
|
||
}),
|
||
},
|
||
},
|
||
timeBucketThresholds: [
|
||
{ maxRangeSeconds: 3 * 60 * 60, interval: { value: 10, unit: "SECOND" } },
|
||
{ maxRangeSeconds: 12 * 60 * 60, interval: { value: 1, unit: "MINUTE" } },
|
||
{ maxRangeSeconds: 2 * 24 * 60 * 60, interval: { value: 5, unit: "MINUTE" } },
|
||
{ maxRangeSeconds: 7 * 24 * 60 * 60, interval: { value: 15, unit: "MINUTE" } },
|
||
{ maxRangeSeconds: 30 * 24 * 60 * 60, interval: { value: 1, unit: "HOUR" } },
|
||
{ maxRangeSeconds: 90 * 24 * 60 * 60, interval: { value: 6, unit: "HOUR" } },
|
||
{ maxRangeSeconds: 180 * 24 * 60 * 60, interval: { value: 1, unit: "DAY" } },
|
||
{ maxRangeSeconds: 365 * 24 * 60 * 60, interval: { value: 1, unit: "WEEK" } },
|
||
] satisfies BucketThreshold[],
|
||
queryCache: { ttlSeconds: 30, alignSeconds: 30 },
|
||
queryClient: "queueMetrics",
|
||
};
|
||
|
||
export const querySchemas: TableSchema[] = [
|
||
runsSchema,
|
||
metricsSchema,
|
||
llmMetricsSchema,
|
||
llmModelsSchema,
|
||
queueMetricsSchema,
|
||
envMetricsSchema,
|
||
queueMetricsByKeySchema,
|
||
];
|
||
|
||
/** Tables whose listing is deferred until queue metrics are rolled out. */
|
||
const QUEUE_METRICS_TABLE_NAMES = new Set([
|
||
queueMetricsSchema.name,
|
||
envMetricsSchema.name,
|
||
queueMetricsByKeySchema.name,
|
||
]);
|
||
|
||
/**
|
||
* Schemas shown in user-facing listings (editor autocomplete, schema docs, schema API, AI query
|
||
* context). Listing only: `querySchemas` stays the compile-time set, so a query naming an unlisted
|
||
* table still runs, with tenancy enforced as usual.
|
||
*
|
||
* Server callers pass `env.QUEUE_METRICS_QUERY_TABLES_VISIBLE === "1"`; client callers read
|
||
* `queueMetricsQueryTables` from `useFeatures()`. This module is imported by browser code, so it
|
||
* must not reach for `env.server` itself.
|
||
*/
|
||
export function listableQuerySchemas(options: { includeQueueMetrics: boolean }): TableSchema[] {
|
||
return querySchemas.filter((s) => {
|
||
if (s.hidden) return false;
|
||
if (!options.includeQueueMetrics && QUEUE_METRICS_TABLE_NAMES.has(s.name)) return false;
|
||
return true;
|
||
});
|
||
}
|
||
|
||
/**
|
||
* Default query for the query editor
|
||
*/
|
||
export const defaultQuery = autoFormatSQL(`SELECT run_id, task_identifier, triggered_at, status
|
||
FROM runs
|
||
ORDER BY triggered_at DESC
|
||
LIMIT 100`);
|