From fa3ad2bc96b99fc7b464c5646e92637b531a39ec Mon Sep 17 00:00:00 2001 From: Makisuo Date: Fri, 23 Oct 2026 01:22:55 +0200 Subject: [PATCH] perf(ci): read the CI analytics from ci_runs and ci_jobs Every CI query (the CI page, the pull request sheet's runs, the ci_runs MCP tool) reads the projections instead of raw traces' Map columns. The week-long default on a Maple-sized org reads 10 MB instead of 1.3 GB. Attempt and pull request come back as numbers, so the string parsing goes. The query tests move to ClickHouse: the e2e runs every builder on seeded runs and jobs. A self-hosted cluster without migration 0047 gets an explicit apply-the-schema answer from the MCP tool. --- .github/workflows/ci.yml | 2 +- .../ai/src/mcp/__evals__/integration-world.ts | 18 +- apps/ai/src/mcp/tools/ci-runs.ts | 34 +- apps/api/scripts/bench-suites/ci.ts | 64 ++++ .../services/integrations/ci-traces/otlp.ts | 2 +- .../pr-review/CiAnalyticsService.test.ts | 21 +- .../services/pr-review/CiAnalyticsService.ts | 15 +- ...test.ts => ci-runs.clickhouse.e2e.test.ts} | 150 +++++++- .../src/services/warehouse/missing-table.ts | 3 + packages/domain/src/mcp-outputs/ci.ts | 2 +- packages/domain/src/tinybird/datasources.ts | 3 + .../domain/src/tinybird/materializations.ts | 4 +- .../src/__sql_baseline__/catalog.sql | 325 ++++++++---------- packages/query-engine/src/ch/index.ts | 3 - .../src/ch/queries/ci-runs.test.ts | 117 +------ .../query-engine/src/ch/queries/ci-runs.ts | 212 +++++------- 16 files changed, 499 insertions(+), 476 deletions(-) create mode 100644 apps/api/scripts/bench-suites/ci.ts rename packages/backend/src/services/warehouse/{ci-runs-materialization.clickhouse.e2e.test.ts => ci-runs.clickhouse.e2e.test.ts} (54%) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 15bc4097c5..03b4c4ac7f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -395,7 +395,7 @@ jobs: src/services/warehouse/alert-window-alignment.clickhouse.e2e.test.ts src/services/warehouse/web-analytics-parity.clickhouse.e2e.test.ts src/services/warehouse/trace-facets-hourly-materialization.clickhouse.e2e.test.ts - src/services/warehouse/ci-runs-materialization.clickhouse.e2e.test.ts + src/services/warehouse/ci-runs.clickhouse.e2e.test.ts # Every SQL shape the product emits, against the real analyzer: unit tests # only compare SQL text, which is how a NO_COMMON_TYPE reached production. diff --git a/apps/ai/src/mcp/__evals__/integration-world.ts b/apps/ai/src/mcp/__evals__/integration-world.ts index 0131f94acf..778925c717 100644 --- a/apps/ai/src/mcp/__evals__/integration-world.ts +++ b/apps/ai/src/mcp/__evals__/integration-world.ts @@ -245,7 +245,13 @@ export const seedIntegrationWorld = async (rt: EvalRuntime) => { FROM scrape_targets t WHERE t.source_id = s.id AND s.platform = 'prometheus'`, [], ) - const reported = (platform: string, okMinutesAgo: number, error: string | null, total: number, unhealthy: number) => + const reported = ( + platform: string, + okMinutesAgo: number, + error: string | null, + total: number, + unhealthy: number, + ) => sql( `UPDATE sources SET last_poll_at = $2, last_poll_ok_at = $2, last_poll_error = $3, poll_resources = $4::jsonb WHERE platform = $1`, @@ -410,13 +416,13 @@ const inWindow = (sql: string, hoursAgo: number) => { export const integrationFixtureRules = (): FixtureRule[] => [ { - match: (sql) => sql.includes(" AS pipeline") && sql.includes("cicd.pipeline.name"), + match: (sql) => sql.includes("FROM ci_runs"), rows: (sql) => CI_RUNS.filter((run) => inWindow(sql, run.hoursAgo)) .filter((run) => !sql.includes("'failure'") || run.result === "failure") .filter( (run) => - !sql.includes("vcs.repository.name'] = '") || sql.includes(`= '${run.repository}'`), + !sql.includes("Repository = '") || sql.includes(`Repository = '${run.repository}'`), ) .map((run) => ({ traceId: run.traceId, @@ -427,16 +433,16 @@ export const integrationFixtureRules = (): FixtureRule[] => [ repository: run.repository, branch: "main", revision: run.revision, - attempt: "1", + attempt: 1, trigger: "push", actor: "ada", - pullRequest: "", + pullRequest: 0, durationMs: run.minutes * 60_000, startedAt: wire(Date.now() - run.hoursAgo * 3_600_000), })), }, { - match: (sql) => sql.includes("maple.cicd.failure.reason"), + match: (sql) => sql.includes("FROM ci_jobs"), rows: CI_RUNS.filter((run) => run.result === "failure").map((run) => ({ traceId: run.traceId, job: CI_WORLD.failingJob, diff --git a/apps/ai/src/mcp/tools/ci-runs.ts b/apps/ai/src/mcp/tools/ci-runs.ts index 144a61c521..6bd19be5c4 100644 --- a/apps/ai/src/mcp/tools/ci-runs.ts +++ b/apps/ai/src/mcp/tools/ci-runs.ts @@ -2,7 +2,8 @@ import { DateTime, Effect, Schema } from "effect" import { CiRunsOutput } from "@maple/domain/mcp-outputs" import { CH, formatWarehouseDateTime } from "@maple/query-engine" import { WarehouseExecutor } from "@maple/query-engine/observability" -import type { McpToolRegistrar } from "./types" +import { isMissingCiTables } from "@maple/backend/services/warehouse/missing-table" +import { McpUnavailableError, type McpToolRegistrar } from "./types" import { warehouseToMcpHandlers } from "../lib/map-warehouse-error" import { withTenantExecutor, CurrentMcpTenant } from "../lib/query-warehouse" import { MCP_DISCOVERY_MAX_HOURS } from "../lib/time" @@ -29,8 +30,17 @@ const formatRunDuration = (ms: number) => { const quantile = (sorted: ReadonlyArray, q: number) => sorted.length === 0 ? 0 : (sorted[Math.min(sorted.length - 1, Math.floor(q * sorted.length))] ?? 0) -const summarize = (runs: ReadonlyArray): Output["pipelines"] => { - const groups = new Map>() +interface SummarizedRun { + readonly repository: string + readonly pipeline: string + readonly result: string + readonly revision: string + readonly durationMs: number + readonly startedAt: DateTime.Utc +} + +const summarize = (runs: ReadonlyArray): Output["pipelines"] => { + const groups = new Map>() for (const run of runs) { const key = `${run.repository}\u0000${run.pipeline}` groups.set(key, [...(groups.get(key) ?? []), run]) @@ -146,7 +156,19 @@ export function registerCiRunsTool(server: McpToolRegistrar) { failedJobsCap: failedIds.length * FAILED_JOBS_PER_RUN, } }), - ).pipe(Effect.catchTags(warehouseToMcpHandlers(TOOL))) + ).pipe( + // A self-hosted ClickHouse gets the CI tables only once an admin applies the schema. + Effect.catchIf(isMissingCiTables, () => + Effect.fail( + new McpUnavailableError({ + message: + "This org's warehouse has no CI tables yet. On a self-hosted ClickHouse, apply the schema from its ClickHouse settings page.", + capability: "ci_runs", + }), + ), + ), + Effect.catchTags(warehouseToMcpHandlers(TOOL)), + ) yield* Effect.annotateCurrentSpan("result.rowCount", runs.length) return { @@ -164,7 +186,7 @@ export function registerCiRunsTool(server: McpToolRegistrar) { branch: run.branch, revision: run.revision, runId: run.runId, - attempt: run.attempt, + attempt: Math.max(run.attempt, 1), trigger: run.trigger, result: run.result, durationMs: run.durationMs, @@ -316,7 +338,7 @@ export function registerCiRunsTool(server: McpToolRegistrar) { run.pipeline, run.branch, run.revision, - run.attempt !== "" && run.attempt !== "1" + run.attempt > 1 ? `${run.result} (attempt ${run.attempt})` : run.result, formatRunDuration(run.durationMs), diff --git a/apps/api/scripts/bench-suites/ci.ts b/apps/api/scripts/bench-suites/ci.ts new file mode 100644 index 0000000000..34ca72fc60 --- /dev/null +++ b/apps/api/scripts/bench-suites/ci.ts @@ -0,0 +1,64 @@ +/** + * The CI page's reads (`ci_runs` / `ci_jobs`) over a week and a month. Edit the org and windows + * to a populated snapshot; the 2026-10-22 table layout choice was measured on this shape. + */ +import * as CH from "@maple/query-engine/ch" +import * as Bench from "@maple-dev/effect-orm/benchmark" +import { OrgId } from "@maple/domain" + +const END = "2026-10-22 22:58:00" +const WINDOWS = { "7d": "2026-10-15 22:58:00", "30d": "2026-09-22 22:58:00" } as const + +const cases = (label: keyof typeof WINDOWS) => { + const inputs = { orgId: OrgId.make("org_sql_catalog"), startTime: WINDOWS[label], endTime: END } + return [ + Bench.query({ + id: `ci/run-totals/${label}`, + inputs, + compile: (i) => CH.compile(CH.ciRunTotalsQuery({}), i), + results: "unordered", + }), + Bench.query({ + id: `ci/run-series/${label}`, + inputs: { ...inputs, bucketSeconds: 21_600 }, + compile: (i) => CH.compile(CH.ciRunSeriesQuery({}), i), + results: "unordered", + }), + Bench.query({ + id: `ci/pipeline-stats/${label}`, + inputs, + compile: (i) => CH.compile(CH.ciPipelineStatsQuery({}), i), + results: "unordered", + }), + Bench.query({ + id: `ci/job-stats/${label}`, + inputs, + compile: (i) => CH.compile(CH.ciJobStatsQuery({}), i), + results: "unordered", + }), + Bench.query({ + id: `ci/queue-totals/${label}`, + inputs, + compile: (i) => CH.compile(CH.ciQueueTotalsQuery({}), i), + results: "unordered", + }), + Bench.query({ + id: `ci/runs-list/${label}`, + inputs, + compile: (i) => CH.compile(CH.ciRunsQuery({ limit: 51 }), i), + results: "ordered", + }), + Bench.query({ + id: `ci/pull-request-runs/${label}`, + inputs, + compile: (i) => CH.compile(CH.ciRunsQuery({ pullRequest: 1234, limit: 100 }), i), + results: "ordered", + }), + ] +} + +export default Bench.defineSuite({ + name: "ci-workload", + dataset: "replace-with-your-snapshot-revision", + cases: [...cases("7d"), ...cases("30d")], +}) diff --git a/packages/backend/src/services/integrations/ci-traces/otlp.ts b/packages/backend/src/services/integrations/ci-traces/otlp.ts index 49187ad319..2fad857d0a 100644 --- a/packages/backend/src/services/integrations/ci-traces/otlp.ts +++ b/packages/backend/src/services/integrations/ci-traces/otlp.ts @@ -14,8 +14,8 @@ import { type WorkflowRunJob, type WorkflowStep, } from "@maple/domain/http" +import { CI_SERVICE_NAME } from "@maple/domain/tinybird" -export const CI_SERVICE_NAME = "github-actions" export const SCOPE_NAME = "@maple/ci-traces" const SPAN_KIND_INTERNAL = 1 diff --git a/packages/backend/src/services/pr-review/CiAnalyticsService.test.ts b/packages/backend/src/services/pr-review/CiAnalyticsService.test.ts index 7ab56a7213..bc802ab64c 100644 --- a/packages/backend/src/services/pr-review/CiAnalyticsService.test.ts +++ b/packages/backend/src/services/pr-review/CiAnalyticsService.test.ts @@ -40,10 +40,10 @@ const run = (n: number, result: string, startedAt: number, extra: Record { }) assert.isAbove(reads.length, 0) for (const read of reads) { - assert.include(read.sql, "ResourceAttributes['vcs.owner.name'] = 'acme'") - assert.include(read.sql, "ResourceAttributes['vcs.repository.name'] = 'web'") + assert.include(read.sql, ".Owner = 'acme'") + assert.include(read.sql, ".Repository = 'web'") } }), ), @@ -205,11 +205,8 @@ describe("CiAnalyticsService.analytics", () => { }) // The red workflow's own runs are read, skipped ones left out. const runsRead = reads.find((read) => read.context === "ciWorkflowRuns") - assert.include(runsRead?.sql ?? "", "traces.SpanAttributes['cicd.pipeline.name'] = 'CI'") - assert.include( - runsRead?.sql ?? "", - "traces.ResourceAttributes['vcs.owner.name'] = 'acme'", - ) + assert.include(runsRead?.sql ?? "", "ci_runs.Pipeline = 'CI'") + assert.include(runsRead?.sql ?? "", "ci_runs.Owner = 'acme'") assert.include(runsRead?.sql ?? "", "!= 'skip'") const red = analytics.pipelines[0]?.red assert.strictEqual(red?.failedRuns, 2) @@ -296,7 +293,7 @@ describe("CiAnalyticsService.listRuns", () => { withService( { ciRuns: [ - run(3, "failure", END - HOUR, { pullRequest: "42", attempt: "2" }), + run(3, "failure", END - HOUR, { pullRequest: 42, attempt: 2 }), run(2, "success", END - 2 * HOUR), run(1, "success", END - 3 * HOUR), ], @@ -337,7 +334,7 @@ describe("CiAnalyticsService.listRuns", () => { assert.notInclude(jobsRead?.sql ?? "", traceId(2)) // A failed run's jobs are read up to the end of the run, not the window. - assert.include(jobsRead?.sql ?? "", `traces.Timestamp <= '${wire(END - HOUR + 61_000)}'`) + assert.include(jobsRead?.sql ?? "", `ci_jobs.Timestamp <= '${wire(END - HOUR + 61_000)}'`) yield* service.listRuns(tenant, { startTime: END - 24 * HOUR, @@ -346,7 +343,7 @@ describe("CiAnalyticsService.listRuns", () => { cursor: page.nextCursor ?? undefined, }) const lastRead = reads.filter((read) => read.context === "ciRuns").at(-1) - assert.include(lastRead?.sql ?? "", `traces.TraceId < '${traceId(2)}'`) + assert.include(lastRead?.sql ?? "", `ci_runs.TraceId < '${traceId(2)}'`) }), ), ) diff --git a/packages/backend/src/services/pr-review/CiAnalyticsService.ts b/packages/backend/src/services/pr-review/CiAnalyticsService.ts index 79d0bb976a..37c444fd41 100644 --- a/packages/backend/src/services/pr-review/CiAnalyticsService.ts +++ b/packages/backend/src/services/pr-review/CiAnalyticsService.ts @@ -40,7 +40,7 @@ export interface CiAnalyticsServiceApi { const HOUR = 3_600 const DAY = 86_400 -/** Raw CI traces are scanned per read; past two weeks a big org's scan outlasts the query budget. */ +/** The longest window the CI page offers: two weeks. */ export const CI_MAX_SPAN_MS = 14 * DAY * 1000 /** Runs read per possibly red workflow to find where its streak began. */ const STREAK_RUNS = 200 @@ -72,14 +72,6 @@ const decodeCursor = (cursor: string) => { /** A quantile over no rows is NaN or null; the contract says null. */ const finiteOrNull = (value: number | null) => (value !== null && Number.isFinite(value) ? value : null) -const attemptOf = (attempt: string) => { - const parsed = Number(attempt) - return Number.isInteger(parsed) && parsed > 0 ? parsed : 1 -} -const pullRequestOf = (value: string) => { - const parsed = Number(value) - return value !== "" && Number.isInteger(parsed) && parsed > 0 ? parsed : null -} type CiSeriesRow = { readonly bucket: DateTime.Utc @@ -451,10 +443,11 @@ export class CiAnalyticsService extends Context.Service = [ }, ] -const insert = (spans: ReadonlyArray) => +const insert = (spans: ReadonlyArray, orgId = ORG_ID) => clickhouseExec( `INSERT INTO traces (OrgId, Timestamp, TraceId, SpanId, ParentSpanId, SpanName, SpanKind, ServiceName, Duration, StatusCode, StatusMessage, SampleRate, SpanAttributes, ResourceAttributes) @@ -130,7 +130,7 @@ const insert = (spans: ReadonlyArray) => const attributes = Object.entries(span.attributes) .map(([key, value]) => `'${key}', '${value}'`) .join(", ") - return `('${ORG_ID}', '${chDateTime(BASE_MS + index * 1000)}', '${span.traceId}', '${span.spanId}', '${span.parentSpanId}', '${span.name}', 'Internal', '${span.service ?? "github-actions"}', ${span.durationNs}, '${span.status ?? "Unset"}', '', 1, map(${attributes}), ${REPO})` + return `('${orgId}', '${chDateTime(BASE_MS + index * 1000)}', '${span.traceId}', '${span.spanId}', '${span.parentSpanId}', '${span.name}', 'Internal', '${span.service ?? "github-actions"}', ${span.durationNs}, '${span.status ?? "Unset"}', '', 1, map(${attributes}), ${REPO})` }) .join("\n,")}`, database, @@ -155,15 +155,56 @@ const jobs = () => FROM ci_jobs WHERE OrgId = '${ORG_ID}' ORDER BY TraceId, Job`, ) -const backfills = migrations - .find((migration) => migration.version === 47)! - .statements.filter((statement) => typeof statement !== "string") +const pipelineRun = ( + traceId: string, + pipeline: string, + result: string, + revision: string, + pullRequest: string, + attempt = "1", +) => + run(traceId, result, { + "cicd.pipeline.name": pipeline, + "vcs.ref.head.revision": revision, + "vcs.change.id": pullRequest, + "maple.cicd.run_attempt": attempt, + }) + +// Oldest first: each span starts a second after the one before. +const READ_SPANS: ReadonlyArray = [ + pipelineRun("r1", "CI", "success", "a", "5"), + job("r1", "build", "success", { "maple.cicd.queue_duration_ms": "1000" }), + pipelineRun("r2", "CI", "failure", "b", "5"), + job("r2", "build", "failure", { + "maple.cicd.queue_duration_ms": "2000", + "maple.cicd.failure.reason": "boom", + }), + // Cancelled before a runner took it: a zero-length job never ran. + { ...job("r2", "lint", "cancellation"), durationNs: 0 }, + pipelineRun("r3", "Deploy", "success", "b", ""), + pipelineRun("r4", "CI", "skip", "c", "6"), + pipelineRun("r5", "CI", "timeout", "c", "6", "2"), + job("r5", "build", "failure", { "maple.cicd.queue_duration_ms": "3000", "maple.cicd.run_attempt": "2" }), +] + +const readWindow = { + orgId: READS_ORG_ID, + startTime: chDateTime(BASE_MS - 60_000), + endTime: chDateTime(BASE_MS + 60_000), +} +const read = (query: Parameters[0], params: Record = readWindow) => + runJson(CH.compileUnsafe(query, params).sql.replace(/\s*FORMAT JSON\s*$/, "")) +const traceIds = (rows: ReadonlyArray>) => rows.map((row) => row.traceId) + +const ciMigration = migrations.filter((migration) => migration.version === 47) +const backfills = ciMigration.flatMap((migration) => migration.statements.filter(isBackfill)) describe.skipIf(!clickhouseE2eEnabled)("ci_runs / ci_jobs materialization", () => { beforeAll(async () => { await clickhouseExec(`CREATE DATABASE ${database}`) await applyRealMigrations(database) await insert(SPANS) + await insert(READ_SPANS, READS_ORG_ID) }, 180_000) afterAll(async () => { @@ -256,4 +297,91 @@ describe.skipIf(!clickhouseE2eEnabled)("ci_runs / ci_jobs materialization", () = assert.deepStrictEqual(await runs(), liveRuns) assert.deepStrictEqual(await jobs(), liveJobs) }) + + it("keeps runs older than raw retention when the whole migration is applied again", async () => { + const expired = "org_ci_expired_e2e" + await clickhouseExec( + `INSERT INTO ci_runs (OrgId, Timestamp, TraceId, SpanId, Pipeline, Result) VALUES ('${expired}', now64(9) - INTERVAL 200 DAY, 'gone', 'run-gone', 'CI', 'success')`, + database, + ) + const before = await runs() + for (const statement of ciMigration.flatMap((migration) => migration.statements)) + await clickhouseExec(renderStatementFull(statement, database), database) + assert.deepStrictEqual(await runs(), before) + const kept = await runJson(`SELECT TraceId FROM ci_runs WHERE OrgId = '${expired}'`) + assert.deepStrictEqual(kept, [{ TraceId: "gone" }]) + }) + + it("counts runs that ran, failures apart from cancellations", async () => { + const [totals] = await read(CH.ciRunTotalsQuery({})) + assert.deepInclude(totals, { runs: 4, succeeded: 2, failed: 2, cancelled: 0 }) + const series = await read(CH.ciRunSeriesQuery({}), { ...readWindow, bucketSeconds: 86_400 }) + assert.strictEqual( + series.reduce((sum, row) => sum + Number(row.runs), 0), + 4, + ) + }) + + it("keeps each workflow's run counts and latest result", async () => { + const rows = await read(CH.ciPipelineStatsQuery({})) + assert.deepStrictEqual( + rows.map(({ pipeline, runs, failed, lastResult }) => ({ pipeline, runs, failed, lastResult })), + [ + { pipeline: "CI", runs: 3, failed: 2, lastResult: "timeout" }, + { pipeline: "Deploy", runs: 1, failed: 0, lastResult: "success" }, + ], + ) + const [ci] = await read( + CH.ciPipelineStatsQuery({ owner: "acme", repository: "api", pipeline: "Deploy" }), + ) + assert.strictEqual(ci?.pipeline, "Deploy") + }) + + it("counts jobs that ran, with reruns and queue time", async () => { + const [build, ...others] = await read(CH.ciJobStatsQuery({})) + assert.deepInclude(build, { job: "build", runs: 3, failed: 2, retried: 1, p50QueueMs: 2000 }) + assert.deepStrictEqual(others, []) + const [queue] = await read(CH.ciQueueTotalsQuery({})) + assert.deepInclude(queue, { jobs: 3, p50QueueMs: 2000 }) + }) + + it("lists runs newest first, pages after a cursor, and filters by pull request", async () => { + const all = await read(CH.ciRunsQuery({})) + assert.deepStrictEqual(traceIds(all), ["r5", "r4", "r3", "r2", "r1"]) + assert.deepInclude(all[0], { attempt: 2, pullRequest: 6, durationMs: 120_000 }) + assert.strictEqual(all[2]?.pullRequest, 0) + + const startedAt = DateTime.makeUnsafe(`${String(all[1]?.startedAt).replace(" ", "T")}Z`) + const after = await read(CH.ciRunsQuery({ before: { startedAt, traceId: "r4" } }), { + ...readWindow, + beforeStartedAt: startedAt, + beforeTraceId: "r4", + }) + assert.deepStrictEqual(traceIds(after), ["r3", "r2", "r1"]) + assert.deepStrictEqual(traceIds(await read(CH.ciRunsQuery({ pullRequest: 5 }))), ["r2", "r1"]) + }) + + it("finds a workflow's red streak and the failed jobs of its runs", async () => { + const ran = await read(CH.ciRunsQuery({ pipeline: "CI", ranOnly: true })) + assert.deepStrictEqual(traceIds(ran), ["r5", "r2", "r1"]) + const streak = CH.ciRedStreak( + ran.map((row) => ({ result: String(row.result), revision: String(row.revision) })), + ) + assert.deepStrictEqual( + { + failedRuns: streak?.failedRuns, + first: streak?.firstFailed.revision, + green: streak?.lastGreen?.revision, + }, + { failedRuns: 2, first: "b", green: "a" }, + ) + const failed = await read(CH.ciFailedJobsQuery({ traceIds: ["r2", "r5"] })) + assert.deepStrictEqual( + failed.map(({ traceId, job, reason }) => ({ traceId, job, reason })), + [ + { traceId: "r2", job: "build", reason: "boom" }, + { traceId: "r5", job: "build", reason: "" }, + ], + ) + }) }) diff --git a/packages/backend/src/services/warehouse/missing-table.ts b/packages/backend/src/services/warehouse/missing-table.ts index 789c1db9da..a3dbf5be0f 100644 --- a/packages/backend/src/services/warehouse/missing-table.ts +++ b/packages/backend/src/services/warehouse/missing-table.ts @@ -39,4 +39,7 @@ export const isMissingProductEvents = isMissingTable(/product_events/i) /** `ai_crawler_requests` ships in a later migration; only its absence may drop the crawler stats. */ export const isMissingAiCrawlerRequests = isMissingTable(/ai_crawler_requests/i) +/** `ci_runs` / `ci_jobs` ship in migration 0047; the CI reads have no raw fallback. */ +export const isMissingCiTables = isMissingTable(/ci_(?:runs|jobs)/i) + export const isMissingServiceOperationsRollup = isMissingTable(/service_operations_(?:minutely|hourly)/i) diff --git a/packages/domain/src/mcp-outputs/ci.ts b/packages/domain/src/mcp-outputs/ci.ts index 02d6d7b332..dae94d8c4a 100644 --- a/packages/domain/src/mcp-outputs/ci.ts +++ b/packages/domain/src/mcp-outputs/ci.ts @@ -19,7 +19,7 @@ export const CiRunRow = Schema.Struct({ branch: Schema.String, revision: Schema.String, runId: Schema.String, - attempt: Schema.String, + attempt: Schema.Number, trigger: Schema.String, /** `cicd.pipeline.result`: success, failure, timeout, cancellation, skip, error. */ result: Schema.String, diff --git a/packages/domain/src/tinybird/datasources.ts b/packages/domain/src/tinybird/datasources.ts index 1306d67d28..92db8d32cb 100644 --- a/packages/domain/src/tinybird/datasources.ts +++ b/packages/domain/src/tinybird/datasources.ts @@ -1314,6 +1314,9 @@ export const aiCrawlerRequests = defineDatasource("ai_crawler_requests", { export type AiCrawlerRequestsRow = InferRow +/** `service.name` of the CI traces the GitHub App's workflow webhooks become. */ +export const CI_SERVICE_NAME = "github-actions" + /** * CI workflow runs: the root span of each `github-actions` trace (CiTraceService), with its * `cicd.*` attributes as columns. Skipped runs are kept; the reads leave them out. diff --git a/packages/domain/src/tinybird/materializations.ts b/packages/domain/src/tinybird/materializations.ts index 6be192a4d5..de1d8f768d 100644 --- a/packages/domain/src/tinybird/materializations.ts +++ b/packages/domain/src/tinybird/materializations.ts @@ -33,6 +33,7 @@ import { errorFingerprintsMinutely, aiTraceIndex, aiCrawlerRequests, + CI_SERVICE_NAME, ciJobs, ciRuns, TRACE_BUCKET_EXPR, @@ -1238,9 +1239,6 @@ export const aiCrawlerRequestsMv = defineMaterializedView("ai_crawler_requests_m ], }) -/** `service.name` of the CI traces CiTraceService writes. */ -const CI_SERVICE_NAME = "github-actions" - /** Populates `ci_runs` with the root span of each CI trace: one row per workflow run attempt. */ export const ciRunsMv = defineMaterializedView("ci_runs_mv", { description: diff --git a/packages/query-engine/src/__sql_baseline__/catalog.sql b/packages/query-engine/src/__sql_baseline__/catalog.sql index 88ea530c4b..b3fc2bca9a 100644 --- a/packages/query-engine/src/__sql_baseline__/catalog.sql +++ b/packages/query-engine/src/__sql_baseline__/catalog.sql @@ -386,221 +386,200 @@ SELECT OFFSET 50 FORMAT JSON --- builder:ci-runs:ciFailedJobsQuery:default [85749449] -SELECT - traces.TraceId AS traceId, - traces.SpanAttributes['cicd.pipeline.task.name'] AS job, - traces.SpanAttributes['cicd.pipeline.task.run.result'] AS result, - traces.SpanAttributes['maple.cicd.failure.reason'] AS reason, - traces.SpanAttributes['cicd.pipeline.task.run.url.full'] AS url, - traces.Duration / 1000000 AS durationMs - FROM traces - WHERE traces.OrgId = 'org_sql_catalog' - AND traces.Timestamp >= '2026-01-01 10:30:00' - AND traces.Timestamp <= '2026-01-03 14:15:00' - AND traces.ServiceName = 'github-actions' - AND traces.TraceId IN ('abc123') - AND traces.StatusCode = 'Error' - AND traces.SpanAttributes['cicd.pipeline.task.name'] != '' - AND traces.SpanAttributes['maple.cicd.step.number'] = '' +-- builder:ci-runs:ciFailedJobsQuery:default [fc2d17c0] +SELECT + ci_jobs.TraceId AS traceId, + ci_jobs.Job AS job, + ci_jobs.Result AS result, + ci_jobs.FailureReason AS reason, + ci_jobs.Url AS url, + ci_jobs.Duration / 1000000 AS durationMs + FROM ci_jobs + WHERE ci_jobs.OrgId = 'org_sql_catalog' + AND ci_jobs.Timestamp >= '2026-01-01 10:30:00' + AND ci_jobs.Timestamp <= '2026-01-03 14:15:00' + AND ci_jobs.TraceId IN ('abc123') + AND ci_jobs.StatusCode = 'Error' ORDER BY traceId ASC LIMIT 200 FORMAT JSON --- builder:ci-runs:ciJobStatsQuery:allWorkflows [6cc8534e] +-- builder:ci-runs:ciJobStatsQuery:allWorkflows [d9988daa] SELECT - traces.ResourceAttributes['vcs.owner.name'] AS owner, - traces.ResourceAttributes['vcs.repository.name'] AS repository, - traces.SpanAttributes['cicd.pipeline.name'] AS pipeline, - traces.SpanAttributes['cicd.pipeline.task.name'] AS job, + ci_jobs.Owner AS owner, + ci_jobs.Repository AS repository, + ci_jobs.Pipeline AS pipeline, + ci_jobs.Job AS job, count() AS runs, - countIf(traces.SpanAttributes['cicd.pipeline.task.run.result'] IN ('failure', 'timeout', 'error')) AS failed, - countIf(traces.SpanAttributes['maple.cicd.run_attempt'] != '1') AS retried, - quantile(0.5)(traces.Duration) / 1000000 AS p50DurationMs, - quantile(0.95)(traces.Duration) / 1000000 AS p95DurationMs, - sum(traces.Duration) / 1000000 AS totalDurationMs, - quantile(0.5)(toFloat64OrZero(traces.SpanAttributes['maple.cicd.queue_duration_ms'])) AS p50QueueMs, - quantile(0.95)(toFloat64OrZero(traces.SpanAttributes['maple.cicd.queue_duration_ms'])) AS p95QueueMs - FROM traces - WHERE traces.OrgId = 'org_sql_catalog' - AND traces.Timestamp >= '2026-01-01 10:30:00' - AND traces.Timestamp <= '2026-01-03 14:15:00' - AND traces.ServiceName = 'github-actions' - AND traces.SpanAttributes['cicd.pipeline.task.name'] != '' - AND traces.SpanAttributes['maple.cicd.step.number'] = '' - AND traces.SpanAttributes['cicd.pipeline.task.run.result'] != 'skip' - AND traces.Duration > 0 + countIf(ci_jobs.Result IN ('failure', 'timeout', 'error')) AS failed, + countIf(ci_jobs.Attempt > 1) AS retried, + quantile(0.5)(ci_jobs.Duration) / 1000000 AS p50DurationMs, + quantile(0.95)(ci_jobs.Duration) / 1000000 AS p95DurationMs, + sum(ci_jobs.Duration) / 1000000 AS totalDurationMs, + quantile(0.5)(ci_jobs.QueueMs) AS p50QueueMs, + quantile(0.95)(ci_jobs.QueueMs) AS p95QueueMs + FROM ci_jobs + WHERE ci_jobs.OrgId = 'org_sql_catalog' + AND ci_jobs.Timestamp >= '2026-01-01 10:30:00' + AND ci_jobs.Timestamp <= '2026-01-03 14:15:00' + AND ci_jobs.Result != 'skip' + AND ci_jobs.Duration > 0 GROUP BY owner, repository, pipeline, job ORDER BY totalDurationMs DESC LIMIT 50 FORMAT JSON --- builder:ci-runs:ciPipelineStatsQuery:oneRepository [c7557d93] +-- builder:ci-runs:ciPipelineStatsQuery:oneRepository [d6ae13bb] SELECT - traces.ResourceAttributes['vcs.owner.name'] AS owner, - traces.ResourceAttributes['vcs.repository.name'] AS repository, - traces.SpanAttributes['cicd.pipeline.name'] AS pipeline, + ci_runs.Owner AS owner, + ci_runs.Repository AS repository, + ci_runs.Pipeline AS pipeline, count() AS runs, - countIf(traces.SpanAttributes['cicd.pipeline.result'] = 'success') AS succeeded, - countIf(traces.SpanAttributes['cicd.pipeline.result'] IN ('failure', 'timeout', 'error')) AS failed, - countIf(traces.SpanAttributes['cicd.pipeline.result'] = 'cancellation') AS cancelled, - quantile(0.5)(traces.Duration) / 1000000 AS p50DurationMs, - quantile(0.95)(traces.Duration) / 1000000 AS p95DurationMs, - sum(traces.Duration) / 1000000 AS totalDurationMs, - argMax(traces.SpanAttributes['cicd.pipeline.result'], traces.Timestamp) AS lastResult, - max(traces.Timestamp) AS lastStartedAt - FROM traces - WHERE traces.OrgId = 'org_sql_catalog' - AND traces.Timestamp >= '2026-01-01 10:30:00' - AND traces.Timestamp <= '2026-01-03 14:15:00' - AND traces.ServiceName = 'github-actions' - AND traces.ResourceAttributes['vcs.owner.name'] = 'acme' - AND traces.ResourceAttributes['vcs.repository.name'] = 'api' - AND traces.SpanAttributes['cicd.pipeline.name'] = 'CI' - AND traces.ParentSpanId = '' - AND traces.SpanAttributes['cicd.pipeline.result'] != 'skip' + countIf(ci_runs.Result = 'success') AS succeeded, + countIf(ci_runs.Result IN ('failure', 'timeout', 'error')) AS failed, + countIf(ci_runs.Result = 'cancellation') AS cancelled, + quantile(0.5)(ci_runs.Duration) / 1000000 AS p50DurationMs, + quantile(0.95)(ci_runs.Duration) / 1000000 AS p95DurationMs, + sum(ci_runs.Duration) / 1000000 AS totalDurationMs, + argMax(ci_runs.Result, ci_runs.Timestamp) AS lastResult, + max(ci_runs.Timestamp) AS lastStartedAt + FROM ci_runs + WHERE ci_runs.OrgId = 'org_sql_catalog' + AND ci_runs.Timestamp >= '2026-01-01 10:30:00' + AND ci_runs.Timestamp <= '2026-01-03 14:15:00' + AND ci_runs.Owner = 'acme' + AND ci_runs.Repository = 'api' + AND ci_runs.Pipeline = 'CI' + AND ci_runs.Result != 'skip' GROUP BY owner, repository, pipeline ORDER BY runs DESC LIMIT 100 FORMAT JSON --- builder:ci-runs:ciQueueTotalsQuery:allWorkflows [57f8fbf3] +-- builder:ci-runs:ciQueueTotalsQuery:allWorkflows [df1871ea] SELECT count() AS jobs, - quantile(0.5)(toFloat64OrZero(traces.SpanAttributes['maple.cicd.queue_duration_ms'])) AS p50QueueMs, - quantile(0.95)(toFloat64OrZero(traces.SpanAttributes['maple.cicd.queue_duration_ms'])) AS p95QueueMs - FROM traces - WHERE traces.OrgId = 'org_sql_catalog' - AND traces.Timestamp >= '2026-01-01 10:30:00' - AND traces.Timestamp <= '2026-01-03 14:15:00' - AND traces.ServiceName = 'github-actions' - AND traces.SpanAttributes['cicd.pipeline.task.name'] != '' - AND traces.SpanAttributes['maple.cicd.step.number'] = '' - AND traces.SpanAttributes['cicd.pipeline.task.run.result'] != 'skip' - AND traces.Duration > 0 + quantile(0.5)(ci_jobs.QueueMs) AS p50QueueMs, + quantile(0.95)(ci_jobs.QueueMs) AS p95QueueMs + FROM ci_jobs + WHERE ci_jobs.OrgId = 'org_sql_catalog' + AND ci_jobs.Timestamp >= '2026-01-01 10:30:00' + AND ci_jobs.Timestamp <= '2026-01-03 14:15:00' + AND ci_jobs.Result != 'skip' + AND ci_jobs.Duration > 0 FORMAT JSON --- builder:ci-runs:ciRunSeriesQuery:sixHourly [ff52eea3] +-- builder:ci-runs:ciRunSeriesQuery:sixHourly [50d0464c] SELECT - toStartOfInterval(traces.Timestamp, INTERVAL 21600 SECOND) AS bucket, + toStartOfInterval(ci_runs.Timestamp, INTERVAL 21600 SECOND) AS bucket, count() AS runs, - countIf(traces.SpanAttributes['cicd.pipeline.result'] = 'success') AS succeeded, - countIf(traces.SpanAttributes['cicd.pipeline.result'] IN ('failure', 'timeout', 'error')) AS failed, - countIf(traces.SpanAttributes['cicd.pipeline.result'] = 'cancellation') AS cancelled, - quantile(0.5)(traces.Duration) / 1000000 AS p50DurationMs, - quantile(0.95)(traces.Duration) / 1000000 AS p95DurationMs, - sum(traces.Duration) / 1000000 AS totalDurationMs - FROM traces - WHERE traces.OrgId = 'org_sql_catalog' - AND traces.Timestamp >= '2026-01-01 10:30:00' - AND traces.Timestamp <= '2026-01-03 14:15:00' - AND traces.ServiceName = 'github-actions' - AND traces.ParentSpanId = '' - AND traces.SpanAttributes['cicd.pipeline.result'] != 'skip' + countIf(ci_runs.Result = 'success') AS succeeded, + countIf(ci_runs.Result IN ('failure', 'timeout', 'error')) AS failed, + countIf(ci_runs.Result = 'cancellation') AS cancelled, + quantile(0.5)(ci_runs.Duration) / 1000000 AS p50DurationMs, + quantile(0.95)(ci_runs.Duration) / 1000000 AS p95DurationMs, + sum(ci_runs.Duration) / 1000000 AS totalDurationMs + FROM ci_runs + WHERE ci_runs.OrgId = 'org_sql_catalog' + AND ci_runs.Timestamp >= '2026-01-01 10:30:00' + AND ci_runs.Timestamp <= '2026-01-03 14:15:00' + AND ci_runs.Result != 'skip' GROUP BY bucket ORDER BY bucket ASC FORMAT JSON --- builder:ci-runs:ciRunsQuery:allRuns [b8a0ffdd] -SELECT - traces.TraceId AS traceId, - traces.SpanAttributes['cicd.pipeline.name'] AS pipeline, - traces.SpanAttributes['cicd.pipeline.run.id'] AS runId, - traces.SpanAttributes['cicd.pipeline.result'] AS result, - traces.SpanAttributes['cicd.pipeline.run.url.full'] AS url, - traces.ResourceAttributes['vcs.repository.name'] AS repository, - traces.SpanAttributes['vcs.ref.head.name'] AS branch, - traces.SpanAttributes['vcs.ref.head.revision'] AS revision, - traces.SpanAttributes['maple.cicd.run_attempt'] AS attempt, - traces.SpanAttributes['maple.cicd.trigger_event'] AS trigger, - traces.SpanAttributes['maple.cicd.actor'] AS actor, - traces.SpanAttributes['vcs.change.id'] AS pullRequest, - traces.Duration / 1000000 AS durationMs, - traces.Timestamp AS startedAt - FROM traces - WHERE traces.OrgId = 'org_sql_catalog' - AND traces.Timestamp >= '2026-01-01 10:30:00' - AND traces.Timestamp <= '2026-01-03 14:15:00' - AND traces.ServiceName = 'github-actions' - AND traces.ParentSpanId = '' +-- builder:ci-runs:ciRunsQuery:allRuns [fa2dede5] +SELECT + ci_runs.TraceId AS traceId, + ci_runs.Pipeline AS pipeline, + ci_runs.RunId AS runId, + ci_runs.Result AS result, + ci_runs.Url AS url, + ci_runs.Repository AS repository, + ci_runs.Branch AS branch, + ci_runs.Revision AS revision, + ci_runs.Attempt AS attempt, + ci_runs.TriggerEvent AS trigger, + ci_runs.Actor AS actor, + ci_runs.PullRequest AS pullRequest, + ci_runs.Duration / 1000000 AS durationMs, + ci_runs.Timestamp AS startedAt + FROM ci_runs + WHERE ci_runs.OrgId = 'org_sql_catalog' + AND ci_runs.Timestamp >= '2026-01-01 10:30:00' + AND ci_runs.Timestamp <= '2026-01-03 14:15:00' ORDER BY startedAt DESC, traceId DESC LIMIT 200 FORMAT JSON --- builder:ci-runs:ciRunsQuery:filtered [9a080dbd] -SELECT - traces.TraceId AS traceId, - traces.SpanAttributes['cicd.pipeline.name'] AS pipeline, - traces.SpanAttributes['cicd.pipeline.run.id'] AS runId, - traces.SpanAttributes['cicd.pipeline.result'] AS result, - traces.SpanAttributes['cicd.pipeline.run.url.full'] AS url, - traces.ResourceAttributes['vcs.repository.name'] AS repository, - traces.SpanAttributes['vcs.ref.head.name'] AS branch, - traces.SpanAttributes['vcs.ref.head.revision'] AS revision, - traces.SpanAttributes['maple.cicd.run_attempt'] AS attempt, - traces.SpanAttributes['maple.cicd.trigger_event'] AS trigger, - traces.SpanAttributes['maple.cicd.actor'] AS actor, - traces.SpanAttributes['vcs.change.id'] AS pullRequest, - traces.Duration / 1000000 AS durationMs, - traces.Timestamp AS startedAt - FROM traces - WHERE traces.OrgId = 'org_sql_catalog' - AND traces.Timestamp >= '2026-01-01 10:30:00' - AND traces.Timestamp <= '2026-01-03 14:15:00' - AND traces.ServiceName = 'github-actions' - AND traces.ParentSpanId = '' - AND traces.SpanAttributes['cicd.pipeline.name'] = 'CI' - AND traces.ResourceAttributes['vcs.repository.name'] = 'api' - AND traces.SpanAttributes['vcs.ref.head.name'] = 'main' - AND traces.SpanAttributes['cicd.pipeline.result'] = 'failure' +-- builder:ci-runs:ciRunsQuery:filtered [0d7930c2] +SELECT + ci_runs.TraceId AS traceId, + ci_runs.Pipeline AS pipeline, + ci_runs.RunId AS runId, + ci_runs.Result AS result, + ci_runs.Url AS url, + ci_runs.Repository AS repository, + ci_runs.Branch AS branch, + ci_runs.Revision AS revision, + ci_runs.Attempt AS attempt, + ci_runs.TriggerEvent AS trigger, + ci_runs.Actor AS actor, + ci_runs.PullRequest AS pullRequest, + ci_runs.Duration / 1000000 AS durationMs, + ci_runs.Timestamp AS startedAt + FROM ci_runs + WHERE ci_runs.OrgId = 'org_sql_catalog' + AND ci_runs.Timestamp >= '2026-01-01 10:30:00' + AND ci_runs.Timestamp <= '2026-01-03 14:15:00' + AND ci_runs.Pipeline = 'CI' + AND ci_runs.Repository = 'api' + AND ci_runs.Branch = 'main' + AND ci_runs.Result = 'failure' ORDER BY startedAt DESC, traceId DESC LIMIT 50 FORMAT JSON --- builder:ci-runs:ciRunsQuery:pageAfterCursor [d212a152] -SELECT - traces.TraceId AS traceId, - traces.SpanAttributes['cicd.pipeline.name'] AS pipeline, - traces.SpanAttributes['cicd.pipeline.run.id'] AS runId, - traces.SpanAttributes['cicd.pipeline.result'] AS result, - traces.SpanAttributes['cicd.pipeline.run.url.full'] AS url, - traces.ResourceAttributes['vcs.repository.name'] AS repository, - traces.SpanAttributes['vcs.ref.head.name'] AS branch, - traces.SpanAttributes['vcs.ref.head.revision'] AS revision, - traces.SpanAttributes['maple.cicd.run_attempt'] AS attempt, - traces.SpanAttributes['maple.cicd.trigger_event'] AS trigger, - traces.SpanAttributes['maple.cicd.actor'] AS actor, - traces.SpanAttributes['vcs.change.id'] AS pullRequest, - traces.Duration / 1000000 AS durationMs, - traces.Timestamp AS startedAt - FROM traces - WHERE traces.OrgId = 'org_sql_catalog' - AND traces.Timestamp >= '2026-01-01 10:30:00' - AND traces.Timestamp <= '2026-01-03 14:15:00' - AND traces.ServiceName = 'github-actions' - AND traces.ParentSpanId = '' - AND traces.SpanAttributes['cicd.pipeline.result'] != 'skip' - AND (traces.Timestamp < '2026-01-01 10:30:00' OR (traces.Timestamp = '2026-01-01 10:30:00' AND traces.TraceId < 'abc123')) +-- builder:ci-runs:ciRunsQuery:pageAfterCursor [e63082f3] +SELECT + ci_runs.TraceId AS traceId, + ci_runs.Pipeline AS pipeline, + ci_runs.RunId AS runId, + ci_runs.Result AS result, + ci_runs.Url AS url, + ci_runs.Repository AS repository, + ci_runs.Branch AS branch, + ci_runs.Revision AS revision, + ci_runs.Attempt AS attempt, + ci_runs.TriggerEvent AS trigger, + ci_runs.Actor AS actor, + ci_runs.PullRequest AS pullRequest, + ci_runs.Duration / 1000000 AS durationMs, + ci_runs.Timestamp AS startedAt + FROM ci_runs + WHERE ci_runs.OrgId = 'org_sql_catalog' + AND ci_runs.Timestamp >= '2026-01-01 10:30:00' + AND ci_runs.Timestamp <= '2026-01-03 14:15:00' + AND ci_runs.Result != 'skip' + AND (ci_runs.Timestamp < '2026-01-01 10:30:00' OR (ci_runs.Timestamp = '2026-01-01 10:30:00' AND ci_runs.TraceId < 'abc123')) ORDER BY startedAt DESC, traceId DESC LIMIT 51 FORMAT JSON --- builder:ci-runs:ciRunTotalsQuery:allWorkflows [10bab54f] +-- builder:ci-runs:ciRunTotalsQuery:allWorkflows [ede05f69] SELECT count() AS runs, - countIf(traces.SpanAttributes['cicd.pipeline.result'] = 'success') AS succeeded, - countIf(traces.SpanAttributes['cicd.pipeline.result'] IN ('failure', 'timeout', 'error')) AS failed, - countIf(traces.SpanAttributes['cicd.pipeline.result'] = 'cancellation') AS cancelled, - quantile(0.5)(traces.Duration) / 1000000 AS p50DurationMs, - quantile(0.95)(traces.Duration) / 1000000 AS p95DurationMs, - sum(traces.Duration) / 1000000 AS totalDurationMs - FROM traces - WHERE traces.OrgId = 'org_sql_catalog' - AND traces.Timestamp >= '2026-01-01 10:30:00' - AND traces.Timestamp <= '2026-01-03 14:15:00' - AND traces.ServiceName = 'github-actions' - AND traces.ParentSpanId = '' - AND traces.SpanAttributes['cicd.pipeline.result'] != 'skip' + countIf(ci_runs.Result = 'success') AS succeeded, + countIf(ci_runs.Result IN ('failure', 'timeout', 'error')) AS failed, + countIf(ci_runs.Result = 'cancellation') AS cancelled, + quantile(0.5)(ci_runs.Duration) / 1000000 AS p50DurationMs, + quantile(0.95)(ci_runs.Duration) / 1000000 AS p95DurationMs, + sum(ci_runs.Duration) / 1000000 AS totalDurationMs + FROM ci_runs + WHERE ci_runs.OrgId = 'org_sql_catalog' + AND ci_runs.Timestamp >= '2026-01-01 10:30:00' + AND ci_runs.Timestamp <= '2026-01-03 14:15:00' + AND ci_runs.Result != 'skip' FORMAT JSON -- builder:containers:containerCountersSummaryQuery:default [1b4fe1f9] diff --git a/packages/query-engine/src/ch/index.ts b/packages/query-engine/src/ch/index.ts index 40fa5dc1e3..6ea9fb403b 100644 --- a/packages/query-engine/src/ch/index.ts +++ b/packages/query-engine/src/ch/index.ts @@ -701,7 +701,6 @@ export { // Queries — CI pipeline runs (GitHub Actions traces) export { CI_FAILED_RESULTS, - CI_TRACES_SERVICE_NAME, ciFailedJobsQuery, ciJobStatsQuery, ciPipelineStatsQuery, @@ -713,9 +712,7 @@ export { ciRunTotalsQuery, ciRunsQuery, type CiFailedJobsOpts, - type CiFailedJobsOutput, type CiRedStreak, type CiRunsOpts, - type CiRunsOutput, type CiScopeOpts, } from "./queries/ci-runs" diff --git a/packages/query-engine/src/ch/queries/ci-runs.test.ts b/packages/query-engine/src/ch/queries/ci-runs.test.ts index 38bdb644da..bbdf7d5996 100644 --- a/packages/query-engine/src/ch/queries/ci-runs.test.ts +++ b/packages/query-engine/src/ch/queries/ci-runs.test.ts @@ -1,119 +1,8 @@ +// What the CI queries read is proven against ClickHouse in +// packages/backend/src/services/warehouse/ci-runs.clickhouse.e2e.test.ts. import { describe, expect, it } from "vitest" import { DateTime } from "effect" -import { compileUnsafe } from "@maple-dev/effect-orm/clickhouse" -import { OrgId } from "@maple/domain" -import { - ciJobStatsQuery, - ciPipelineStatsQuery, - ciQueueTotalsQuery, - ciRedStreak, - ciRunSeriesQuery, - ciRunsSpanMs, - ciRunTotalsQuery, - ciRunsQuery, -} from "./ci-runs" - -const baseParams = { - orgId: OrgId.make("org_1"), - startTime: "2026-10-01 00:00:00", - endTime: "2026-10-08 00:00:00", -} - -const scope = { owner: "acme", repository: "api", pipeline: "CI" } - -describe("ci run analytics", () => { - it.each([ - ["totals", () => compileUnsafe(ciRunTotalsQuery(scope), baseParams).sql], - ["workflow stats", () => compileUnsafe(ciPipelineStatsQuery(scope), baseParams).sql], - ])("reads only the org's ran workflow runs in the scope (%s)", (_name, compile) => { - const sql = compile() - expect(sql).toContain("OrgId = 'org_1'") - expect(sql).toContain("ServiceName = 'github-actions'") - expect(sql).toContain("traces.ParentSpanId = ''") - expect(sql).toContain("traces.SpanAttributes['cicd.pipeline.result'] != 'skip'") - expect(sql).toContain("traces.ResourceAttributes['vcs.owner.name'] = 'acme'") - expect(sql).toContain("traces.ResourceAttributes['vcs.repository.name'] = 'api'") - expect(sql).toContain("traces.SpanAttributes['cicd.pipeline.name'] = 'CI'") - }) - - it("counts timeouts and errors as failures, cancellations apart", () => { - const { sql } = compileUnsafe(ciRunTotalsQuery({}), baseParams) - expect(sql).toContain( - "countIf(traces.SpanAttributes['cicd.pipeline.result'] IN ('failure', 'timeout', 'error'))", - ) - expect(sql).toContain("countIf(traces.SpanAttributes['cicd.pipeline.result'] = 'cancellation')") - expect(sql).not.toContain("vcs.owner.name") - }) - - it("buckets the series by the requested interval", () => { - const { sql } = compileUnsafe(ciRunSeriesQuery({}), { ...baseParams, bucketSeconds: 86_400 }) - expect(sql).toContain("toStartOfInterval(traces.Timestamp, INTERVAL 86400 SECOND)") - expect(sql).toContain("GROUP BY bucket") - }) - - it("keeps each workflow's latest result", () => { - const { sql } = compileUnsafe(ciPipelineStatsQuery({ limit: 20 }), baseParams) - expect(sql).toContain("argMax(traces.SpanAttributes['cicd.pipeline.result'], traces.Timestamp)") - expect(sql).toContain("GROUP BY owner, repository, pipeline") - expect(sql).toContain("LIMIT 20") - }) -}) - -describe("ci job analytics", () => { - it.each([ - ["job stats", () => compileUnsafe(ciJobStatsQuery({}), baseParams).sql], - ["queue totals", () => compileUnsafe(ciQueueTotalsQuery({}), baseParams).sql], - ])("reads job spans that ran, not steps or runner waits (%s)", (_name, compile) => { - const sql = compile() - expect(sql).toContain("OrgId = 'org_1'") - expect(sql).toContain("traces.SpanAttributes['cicd.pipeline.task.name'] != ''") - expect(sql).toContain("traces.SpanAttributes['maple.cicd.step.number'] = ''") - expect(sql).toContain("traces.SpanAttributes['cicd.pipeline.task.run.result'] != 'skip'") - expect(sql).toContain("toFloat64OrZero(traces.SpanAttributes['maple.cicd.queue_duration_ms'])") - expect(sql).toContain("traces.Duration > 0") - }) - - it("ranks jobs by the CI time they take", () => { - const { sql } = compileUnsafe(ciJobStatsQuery({}), baseParams) - expect(sql).toContain("GROUP BY owner, repository, pipeline, job") - expect(sql).toContain("ORDER BY totalDurationMs DESC") - expect(sql).toContain("countIf(traces.SpanAttributes['maple.cicd.run_attempt'] != '1')") - }) -}) - -describe("ciRunsQuery", () => { - it("pages newest first after the cursor run", () => { - const { sql } = compileUnsafe( - ciRunsQuery({ - pullRequest: 42, - owner: "acme", - before: { startedAt: DateTime.makeUnsafe("2026-10-05T12:00:00Z"), traceId: "abc" }, - limit: 51, - }), - { - ...baseParams, - beforeStartedAt: DateTime.makeUnsafe("2026-10-05T12:00:00Z"), - beforeTraceId: "abc", - }, - ) - expect(sql).toContain("traces.SpanAttributes['vcs.change.id'] = '42'") - expect(sql).toContain("traces.ResourceAttributes['vcs.owner.name'] = 'acme'") - expect(sql).toContain("traces.TraceId < 'abc'") - expect(sql).toContain("ORDER BY startedAt DESC, traceId DESC") - expect(sql).toContain("LIMIT 51") - }) - - it("reads every run when no cursor is given", () => { - const { sql } = compileUnsafe(ciRunsQuery({}), baseParams) - expect(sql).not.toContain("traces.TraceId <") - expect(sql).not.toContain("!= 'skip'") - }) - - it("leaves skipped runs out when asked", () => { - const { sql } = compileUnsafe(ciRunsQuery({ ranOnly: true }), baseParams) - expect(sql).toContain("traces.SpanAttributes['cicd.pipeline.result'] != 'skip'") - }) -}) +import { ciRedStreak, ciRunsSpanMs } from "./ci-runs" describe("ciRunsSpanMs", () => { it("covers every run from the first start to a second past the last end", () => { diff --git a/packages/query-engine/src/ch/queries/ci-runs.ts b/packages/query-engine/src/ch/queries/ci-runs.ts index 870a8f54c3..0c3f5d416f 100644 --- a/packages/query-engine/src/ch/queries/ci-runs.ts +++ b/packages/query-engine/src/ch/queries/ci-runs.ts @@ -2,17 +2,15 @@ // // GitHub Actions runs land as traces under `service.name = github-actions` (CiTraceService): one // root SERVER span per workflow run carrying the `cicd.pipeline.*` attributes, one child per job, -// and step / runner-wait spans under each job. A failed job carries `maple.cicd.failure.reason` -// from its annotations. Both queries read raw `traces`, scoped to that one service. +// and step / runner-wait spans under each job. Migration 0047's views keep the runs in `ci_runs` +// and the jobs in `ci_jobs` with those attributes as columns; every read here goes to them. import { DateTime } from "effect" import * as CH from "@maple-dev/effect-orm/expr" import { from, param, type ColumnAccessor } from "@maple-dev/effect-orm/clickhouse" -import { Traces, orgIdParam, utcSecondsParam } from "../tables" +import { CiJobs, CiRuns, orgIdParam, utcSecondsParam } from "../tables" import { timeseriesBucket } from "./query-helpers" -export const CI_TRACES_SERVICE_NAME = "github-actions" - /** `cicd.pipeline.result` values that mean the run broke, as opposed to being cancelled or skipped. */ export const CI_FAILED_RESULTS = ["failure", "timeout", "error"] as const @@ -36,39 +34,22 @@ export interface CiRunsOpts { readonly limit?: number } -export interface CiRunsOutput { - readonly traceId: string - readonly pipeline: string - readonly runId: string - readonly result: string - readonly url: string - readonly repository: string - readonly branch: string - readonly revision: string - readonly attempt: string - readonly trigger: string - readonly actor: string - readonly pullRequest: string - readonly durationMs: number - readonly startedAt: DateTime.Utc -} - -/** Workflow runs, newest first: the root span of each CI trace. */ +/** Workflow runs, newest first. `pullRequest` is 0 for a run without one. */ export function ciRunsQuery(opts: CiRunsOpts) { - return from(Traces) + return from(CiRuns) .select(($) => ({ traceId: $.TraceId, - pipeline: $.SpanAttributes.get("cicd.pipeline.name"), - runId: $.SpanAttributes.get("cicd.pipeline.run.id"), - result: $.SpanAttributes.get("cicd.pipeline.result"), - url: $.SpanAttributes.get("cicd.pipeline.run.url.full"), - repository: $.ResourceAttributes.get("vcs.repository.name"), - branch: $.SpanAttributes.get("vcs.ref.head.name"), - revision: $.SpanAttributes.get("vcs.ref.head.revision"), - attempt: $.SpanAttributes.get("maple.cicd.run_attempt"), - trigger: $.SpanAttributes.get("maple.cicd.trigger_event"), - actor: $.SpanAttributes.get("maple.cicd.actor"), - pullRequest: $.SpanAttributes.get("vcs.change.id"), + pipeline: $.Pipeline, + runId: $.RunId, + result: $.Result, + url: $.Url, + repository: $.Repository, + branch: $.Branch, + revision: $.Revision, + attempt: $.Attempt, + trigger: $.TriggerEvent, + actor: $.Actor, + pullRequest: $.PullRequest, durationMs: $.Duration.div(1000000), startedAt: $.Timestamp, })) @@ -76,17 +57,13 @@ export function ciRunsQuery(opts: CiRunsOpts) { $.OrgId.eq(orgIdParam), $.Timestamp.gte(utcSecondsParam("startTime")), $.Timestamp.lte(utcSecondsParam("endTime")), - $.ServiceName.eq(CI_TRACES_SERVICE_NAME), - $.ParentSpanId.eq(""), - CH.when(opts.pipeline, (v: string) => $.SpanAttributes.get("cicd.pipeline.name").eq(v)), - CH.when(opts.repository, (v: string) => $.ResourceAttributes.get("vcs.repository.name").eq(v)), - CH.when(opts.branch, (v: string) => $.SpanAttributes.get("vcs.ref.head.name").eq(v)), - CH.when(opts.result, (v: string) => $.SpanAttributes.get("cicd.pipeline.result").eq(v)), - CH.when(opts.owner, (v: string) => $.ResourceAttributes.get("vcs.owner.name").eq(v)), - CH.whenTrue(opts.ranOnly === true, () => - $.SpanAttributes.get("cicd.pipeline.result").neq("skip"), - ), - CH.when(opts.pullRequest, (v: number) => $.SpanAttributes.get("vcs.change.id").eq(String(v))), + CH.when(opts.pipeline, (v: string) => $.Pipeline.eq(v)), + CH.when(opts.repository, (v: string) => $.Repository.eq(v)), + CH.when(opts.branch, (v: string) => $.Branch.eq(v)), + CH.when(opts.result, (v: string) => $.Result.eq(v)), + CH.when(opts.owner, (v: string) => $.Owner.eq(v)), + CH.whenTrue(opts.ranOnly === true, () => $.Result.neq("skip")), + CH.when(opts.pullRequest, (v: number) => $.PullRequest.eq(v)), CH.whenTrue(opts.before !== undefined, () => $.Timestamp.lt(param.dateTime("beforeStartedAt")).or( $.Timestamp.eq(param.dateTime("beforeStartedAt")).and( @@ -106,38 +83,23 @@ export interface CiFailedJobsOpts { readonly limit?: number } -export interface CiFailedJobsOutput { - readonly traceId: string - readonly job: string - readonly result: string - readonly reason: string - readonly url: string - readonly durationMs: number -} - -/** - * Failed jobs of the given runs, with the failure reason their annotations gave. Jobs are the run - * span's children that name a task but not a step. - */ +/** Failed jobs of the given runs, with the failure reason their annotations gave. */ export function ciFailedJobsQuery(opts: CiFailedJobsOpts) { - return from(Traces) + return from(CiJobs) .select(($) => ({ traceId: $.TraceId, - job: $.SpanAttributes.get("cicd.pipeline.task.name"), - result: $.SpanAttributes.get("cicd.pipeline.task.run.result"), - reason: $.SpanAttributes.get("maple.cicd.failure.reason"), - url: $.SpanAttributes.get("cicd.pipeline.task.run.url.full"), + job: $.Job, + result: $.Result, + reason: $.FailureReason, + url: $.Url, durationMs: $.Duration.div(1000000), })) .where(($) => [ $.OrgId.eq(orgIdParam), $.Timestamp.gte(utcSecondsParam("startTime")), $.Timestamp.lte(utcSecondsParam("endTime")), - $.ServiceName.eq(CI_TRACES_SERVICE_NAME), CH.inList($.TraceId, opts.traceIds), $.StatusCode.eq("Error"), - $.SpanAttributes.get("cicd.pipeline.task.name").neq(""), - $.SpanAttributes.get("maple.cicd.step.number").eq(""), ]) .orderBy(["traceId", "asc"]) .limit(opts.limit ?? 200) @@ -196,53 +158,41 @@ export interface CiScopeOpts { readonly pipeline?: string } -type TraceColumns = ColumnAccessor +type RunColumns = ColumnAccessor +type JobColumns = ColumnAccessor -const ciScope = ($: TraceColumns, opts: CiScopeOpts) => [ +const ciScope = ($: RunColumns | JobColumns, opts: CiScopeOpts) => [ $.OrgId.eq(orgIdParam), $.Timestamp.gte(utcSecondsParam("startTime")), $.Timestamp.lte(utcSecondsParam("endTime")), - $.ServiceName.eq(CI_TRACES_SERVICE_NAME), - CH.when(opts.owner, (v: string) => $.ResourceAttributes.get("vcs.owner.name").eq(v)), - CH.when(opts.repository, (v: string) => $.ResourceAttributes.get("vcs.repository.name").eq(v)), - CH.when(opts.pipeline, (v: string) => $.SpanAttributes.get("cicd.pipeline.name").eq(v)), + CH.when(opts.owner, (v: string) => $.Owner.eq(v)), + CH.when(opts.repository, (v: string) => $.Repository.eq(v)), + CH.when(opts.pipeline, (v: string) => $.Pipeline.eq(v)), ] -/** Run spans that did something: a skipped run takes a second and would drag every percentile down. */ -const ranRuns = ($: TraceColumns, opts: CiScopeOpts) => [ - ...ciScope($, opts), - $.ParentSpanId.eq(""), - $.SpanAttributes.get("cicd.pipeline.result").neq("skip"), -] +/** Runs that did something: a skipped run takes a second and would drag every percentile down. */ +const ranRuns = ($: RunColumns, opts: CiScopeOpts) => [...ciScope($, opts), $.Result.neq("skip")] -/** - * Job spans that ran: the run's children that name a task but not a step. A job skipped, or - * cancelled before a runner took it (a zero-length span), never ran. - */ -const ranJobs = ($: TraceColumns, opts: CiScopeOpts) => [ +/** Jobs that ran: a job skipped, or cancelled before a runner took it (zero length), never ran. */ +const ranJobs = ($: JobColumns, opts: CiScopeOpts) => [ ...ciScope($, opts), - $.SpanAttributes.get("cicd.pipeline.task.name").neq(""), - $.SpanAttributes.get("maple.cicd.step.number").eq(""), - $.SpanAttributes.get("cicd.pipeline.task.run.result").neq("skip"), + $.Result.neq("skip"), $.Duration.gt(0), ] -const runAggregates = ($: TraceColumns) => { - const result = $.SpanAttributes.get("cicd.pipeline.result") - return { - runs: CH.count(), - succeeded: CH.countIf(result.eq("success")), - failed: CH.countIf(CH.inList(result, CI_FAILED_RESULTS)), - cancelled: CH.countIf(result.eq("cancellation")), - p50DurationMs: CH.quantile(0.5)($.Duration).div(1000000), - p95DurationMs: CH.quantile(0.95)($.Duration).div(1000000), - totalDurationMs: CH.sum($.Duration).div(1000000), - } -} +const runAggregates = ($: RunColumns) => ({ + runs: CH.count(), + succeeded: CH.countIf($.Result.eq("success")), + failed: CH.countIf(CH.inList($.Result, CI_FAILED_RESULTS)), + cancelled: CH.countIf($.Result.eq("cancellation")), + p50DurationMs: CH.quantile(0.5)($.Duration).div(1000000), + p95DurationMs: CH.quantile(0.95)($.Duration).div(1000000), + totalDurationMs: CH.sum($.Duration).div(1000000), +}) /** Run counts and durations over the whole window. */ export function ciRunTotalsQuery(opts: CiScopeOpts) { - return from(Traces) + return from(CiRuns) .select(($) => runAggregates($)) .where(($) => ranRuns($, opts)) .format("JSON") @@ -250,7 +200,7 @@ export function ciRunTotalsQuery(opts: CiScopeOpts) { /** Run counts and durations per `bucketSeconds` bucket; empty buckets are absent. */ export function ciRunSeriesQuery(opts: CiScopeOpts) { - return from(Traces) + return from(CiRuns) .select(($) => ({ bucket: timeseriesBucket($.Timestamp, false), ...runAggregates($) })) .where(($) => ranRuns($, opts)) .groupBy("bucket") @@ -260,13 +210,13 @@ export function ciRunSeriesQuery(opts: CiScopeOpts) { /** Per repository and workflow: run counts, durations and the latest run's result. */ export function ciPipelineStatsQuery(opts: CiScopeOpts & { readonly limit?: number }) { - return from(Traces) + return from(CiRuns) .select(($) => ({ - owner: $.ResourceAttributes.get("vcs.owner.name"), - repository: $.ResourceAttributes.get("vcs.repository.name"), - pipeline: $.SpanAttributes.get("cicd.pipeline.name"), + owner: $.Owner, + repository: $.Repository, + pipeline: $.Pipeline, ...runAggregates($), - lastResult: CH.argMax($.SpanAttributes.get("cicd.pipeline.result"), $.Timestamp), + lastResult: CH.argMax($.Result, $.Timestamp), lastStartedAt: CH.max_($.Timestamp), })) .where(($) => ranRuns($, opts)) @@ -281,25 +231,22 @@ export function ciPipelineStatsQuery(opts: CiScopeOpts & { readonly limit?: numb * runner, so the durations include the queue that `p50QueueMs` and `p95QueueMs` break out. */ export function ciJobStatsQuery(opts: CiScopeOpts & { readonly limit?: number }) { - return from(Traces) - .select(($) => { - const result = $.SpanAttributes.get("cicd.pipeline.task.run.result") - const queueMs = CH.toFloat64OrZero($.SpanAttributes.get("maple.cicd.queue_duration_ms")) - return { - owner: $.ResourceAttributes.get("vcs.owner.name"), - repository: $.ResourceAttributes.get("vcs.repository.name"), - pipeline: $.SpanAttributes.get("cicd.pipeline.name"), - job: $.SpanAttributes.get("cicd.pipeline.task.name"), - runs: CH.count(), - failed: CH.countIf(CH.inList(result, CI_FAILED_RESULTS)), - retried: CH.countIf($.SpanAttributes.get("maple.cicd.run_attempt").neq("1")), - p50DurationMs: CH.quantile(0.5)($.Duration).div(1000000), - p95DurationMs: CH.quantile(0.95)($.Duration).div(1000000), - totalDurationMs: CH.sum($.Duration).div(1000000), - p50QueueMs: CH.quantile(0.5)(queueMs), - p95QueueMs: CH.quantile(0.95)(queueMs), - } - }) + return from(CiJobs) + .select(($) => ({ + owner: $.Owner, + repository: $.Repository, + pipeline: $.Pipeline, + job: $.Job, + runs: CH.count(), + failed: CH.countIf(CH.inList($.Result, CI_FAILED_RESULTS)), + // 0 when the span carried no attempt: that is a first run, not a rerun. + retried: CH.countIf($.Attempt.gt(1)), + p50DurationMs: CH.quantile(0.5)($.Duration).div(1000000), + p95DurationMs: CH.quantile(0.95)($.Duration).div(1000000), + totalDurationMs: CH.sum($.Duration).div(1000000), + p50QueueMs: CH.quantile(0.5)($.QueueMs), + p95QueueMs: CH.quantile(0.95)($.QueueMs), + })) .where(($) => ranJobs($, opts)) .groupBy("owner", "repository", "pipeline", "job") .orderBy(["totalDurationMs", "desc"]) @@ -309,15 +256,12 @@ export function ciJobStatsQuery(opts: CiScopeOpts & { readonly limit?: number }) /** How long jobs waited for a runner over the whole window. */ export function ciQueueTotalsQuery(opts: CiScopeOpts) { - return from(Traces) - .select(($) => { - const queueMs = CH.toFloat64OrZero($.SpanAttributes.get("maple.cicd.queue_duration_ms")) - return { - jobs: CH.count(), - p50QueueMs: CH.quantile(0.5)(queueMs), - p95QueueMs: CH.quantile(0.95)(queueMs), - } - }) + return from(CiJobs) + .select(($) => ({ + jobs: CH.count(), + p50QueueMs: CH.quantile(0.5)($.QueueMs), + p95QueueMs: CH.quantile(0.95)($.QueueMs), + })) .where(($) => ranJobs($, opts)) .format("JSON") }