From 7ddc85c1699b419eff400e54ba773fc87f1f43b4 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Wed, 5 Aug 2026 00:37:35 +0200 Subject: [PATCH 1/5] feat(query-engine): add QueryDef registry, migrate 5 pilot handlers Every handler in query-engine.http.ts wires its own cost profile, span context, error label and -- if the author remembered -- caching. Caching being opt-in per call site meant silence read as "off": 11 of 61 handlers called cachedDirect, the other 50 are uncached by omission rather than by decision. Nothing in the type system asked. QueryDef makes those choices data instead of boilerplate. The load-bearing field is `cache`, which is required-but-nullable: `cache: undefined` is a decision someone made, a missing `cache` is a compile error. That is the mechanism by which the remaining ~50 handlers become reviewed decisions as they migrate. runQuery/runQueryFirst apply the declared policy in one place. Rows-vs-first -row stays a call-site concern rather than a field on the def, because the same compiled query legitimately supports both and compiledQueryFirst takes the identical CompiledQuery -- encoding it would only let a caller disagree with it. Deliberately out of scope: row decoding. Handlers keep their own hand-coercion and response mapping. Folding that into a declared rowSchema is a real improvement but a riskier, separate change -- it turns ClickHouse's UInt64-as-JSON-string into a hard failure where it is currently a silent Number("123"), and every migrated query has to clear the DESCRIBE sweep. Keeping them apart means a decode regression can never be tangled up with a caching change. Pilot covers 5 handlers spanning both shapes and both cache states: errorsByType, errorsTimeseries, errorRateByService (rows, uncached), errorsSummary (first-row, uncached), serviceOverview (rows, cached ttl=15 version=2). Cache settings are carried over verbatim, so this changes no caching behaviour. 56 handlers keep their inline wiring; the two styles coexist while migration proceeds. Registry lives behind the ./registry subpath, not the root barrel, since entries pull in ./runtime for cache-policy types and the root barrel stays driver-free for web/cli. Verified: query-engine + apps/api typecheck; SQL baseline byte-identical (14 tests) proving the migrated queries compile to the same SQL; apps/api warehouse + v2 route suites 254 pass. --- apps/api/src/routes/query-runner.ts | 103 ++++++++++++++++ apps/api/src/routes/v1/query-engine.http.ts | 105 ++-------------- packages/query-engine/package.json | 1 + packages/query-engine/src/registry/index.ts | 9 ++ packages/query-engine/src/registry/queries.ts | 116 ++++++++++++++++++ .../query-engine/src/registry/query-def.ts | 86 +++++++++++++ 6 files changed, 325 insertions(+), 95 deletions(-) create mode 100644 apps/api/src/routes/query-runner.ts create mode 100644 packages/query-engine/src/registry/index.ts create mode 100644 packages/query-engine/src/registry/queries.ts create mode 100644 packages/query-engine/src/registry/query-def.ts diff --git a/apps/api/src/routes/query-runner.ts b/apps/api/src/routes/query-runner.ts new file mode 100644 index 000000000..9f38626a8 --- /dev/null +++ b/apps/api/src/routes/query-runner.ts @@ -0,0 +1,103 @@ +import type { QueryDef } from "@maple/query-engine/registry" +import type { QueryEngineDirectError } from "@maple/query-engine/runtime" +import { Effect, Option } from "effect" +import type { TenantContext } from "@/services/auth/AuthService" +import { QueryEngineService } from "@/services/warehouse/QueryEngineService" +import { WarehouseQueryService } from "@/services/warehouse/WarehouseQueryService" + +/** + * Execute registry-declared warehouse queries. + * + * This is the single place that applies a `QueryDef`'s cost profile, settings, + * span context, error label and cache policy — the five things that used to be + * hand-repeated in each of 61 handlers, where the cache in particular was + * silently omitted 50 times. + * + * Handlers keep their own row-to-response mapping. What they lose is the + * plumbing, not the presentation. + */ + +/** + * Shared tail: annotate failures with the query id, then apply the declared + * cache policy (or don't). `cachedDirect` wraps the whole execution so a hit + * skips the warehouse entirely, and it takes the raw payload as key input — + * it snaps timestamps and sorts set-valued keys itself, which is why the + * payload goes in unnormalized. + */ +const withPolicy = ( + def: QueryDef, + tenant: TenantContext, + payload: Payload, + execute: Effect.Effect, +) => + Effect.gen(function* () { + const queryEngine = yield* QueryEngineService + const labelled = execute.pipe( + // Same annotation the old inline `mapExecError` produced, with the label + // derived from `def.id` rather than a hand-written string that could + // disagree with the span context beside it. + Effect.tapError(() => + Effect.annotateCurrentSpan({ + "maple.query_engine.failed_step": `${def.id} query failed`, + }), + ), + ) + if (def.cache === undefined) { + return yield* labelled + } + return yield* queryEngine.cachedDirect(tenant, def.id, payload, labelled, def.cache) + }) + +/** + * Run a `QueryDef` that returns many rows. + * + * Rows-vs-first-row is a call-site concern rather than a field on the def: the + * same compiled query legitimately supports both, and `compiledQueryFirst` + * takes the identical `CompiledQuery`. Putting it in the def would only let a + * caller disagree with it. + */ +export const runQuery = ( + def: QueryDef, + tenant: TenantContext, + payload: Payload, +) => + Effect.gen(function* () { + const warehouse = yield* WarehouseQueryService + return yield* withPolicy( + def, + tenant, + payload, + warehouse.compiledQuery(tenant, def.compile(payload, tenant.orgId), { + profile: def.profile, + ...(def.settings === undefined ? {} : { settings: def.settings }), + context: def.id, + }), + ) + }) + +/** + * Run a `QueryDef` that returns at most one row, as `Row | null`. + * + * Null rather than `Option` because every current caller immediately does + * `Option.getOrNull` to build a nullable response field. + */ +export const runQueryFirst = ( + def: QueryDef, + tenant: TenantContext, + payload: Payload, +) => + Effect.gen(function* () { + const warehouse = yield* WarehouseQueryService + return yield* withPolicy( + def, + tenant, + payload, + warehouse + .compiledQueryFirst(tenant, def.compile(payload, tenant.orgId), { + profile: def.profile, + ...(def.settings === undefined ? {} : { settings: def.settings }), + context: def.id, + }) + .pipe(Effect.map(Option.getOrNull)), + ) + }) diff --git a/apps/api/src/routes/v1/query-engine.http.ts b/apps/api/src/routes/v1/query-engine.http.ts index d738c6caa..334950953 100644 --- a/apps/api/src/routes/v1/query-engine.http.ts +++ b/apps/api/src/routes/v1/query-engine.http.ts @@ -83,6 +83,8 @@ import { parseWarehouseDateTime, } from "@maple/query-engine" import { LOGS_BODY_SEARCH_SETTINGS } from "@maple/query-engine/profiles" +import { Queries } from "@maple/query-engine/registry" +import { runQuery, runQueryFirst } from "@/routes/query-runner" import type { ExecutionTenant, WarehouseSqlError } from "@maple/query-engine/execution" import { buildBreakdownQuerySpec, buildTimeseriesQuerySpec } from "@maple/query-engine/query-builder" import * as Integrations from "@maple/query-engine-integrations" @@ -329,26 +331,9 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("errorsByType", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.errorsByTypeQuery({ - rootOnly: payload.rootOnly, - services: payload.services, - deploymentEnvs: payload.deploymentEnvs, - fingerprintHashes: payload.fingerprintHashes, - limit: payload.limit, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "errorsByType", - }), - "errorsByType query failed", - ) - const typedRows = rows + const rows = yield* runQuery(Queries.errorsByType, tenant, payload) return new ErrorsByTypeResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ fingerprintHash: decodeFingerprintHash(row.fingerprintHash), errorLabel: row.errorLabel, sampleMessage: row.sampleMessage, @@ -363,28 +348,9 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("errorsTimeseries", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.errorsTimeseriesQuery({ - fingerprintHash: payload.fingerprintHash, - services: payload.services, - }), - { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - bucketSeconds: payload.bucketSeconds ?? 3600, - }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "errorsTimeseries", - }), - "errorsTimeseries query failed", - ) - const typedRows = rows + const rows = yield* runQuery(Queries.errorsTimeseries, tenant, payload) return new ErrorsTimeseriesResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ bucket: String(row.bucket), count: Number(row.count), })), @@ -394,24 +360,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("errorsSummary", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.errorsSummaryQuery({ - rootOnly: payload.rootOnly, - services: payload.services, - deploymentEnvs: payload.deploymentEnvs, - fingerprintHashes: payload.fingerprintHashes, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const row = yield* mapExecError( - warehouse - .compiledQueryFirst(tenant, compiled, { - profile: "aggregation", - context: "errorsSummary", - }) - .pipe(Effect.map(Option.getOrNull)), - "errorsSummary query failed", - ) + const row = yield* runQueryFirst(Queries.errorsSummary, tenant, payload) return new ErrorsSummaryResponse({ data: row ? { @@ -461,21 +410,9 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("errorRateByService", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile(CH.errorRateByServiceQuery(), { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - }) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "errorRateByService", - }), - "errorRateByService query failed", - ) - const typedRows = rows + const rows = yield* runQuery(Queries.errorRateByService, tenant, payload) return new ErrorRateByServiceResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ serviceName: decodeServiceName(row.serviceName), totalLogs: Number(row.totalLogs), errorLogs: Number(row.errorLogs), @@ -487,29 +424,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("serviceOverview", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.serviceOverviewQuery({ - environments: payload.environments, - namespaces: payload.namespaces, - commitShas: payload.commitShas, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const rows = yield* queryEngine.cachedDirect( - tenant, - "serviceOverview", - payload, - mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "serviceOverview", - }), - "serviceOverview query failed", - ), - // v2: rows gained per-commit `firstSeen`; the version bump keeps - // pre-upgrade cached rows (missing the field) from being served. - makeDirectRouteCachePolicy({ ttlSeconds: 15, version: 2 }), - ) + const rows = yield* runQuery(Queries.serviceOverview, tenant, payload) return new ServiceOverviewResponse({ data: rows }) }), ) diff --git a/packages/query-engine/package.json b/packages/query-engine/package.json index 9e3cab148..a3b2d9e03 100644 --- a/packages/query-engine/package.json +++ b/packages/query-engine/package.json @@ -14,6 +14,7 @@ "./caching": "./src/caching/index.ts", "./execution": "./src/execution/index.ts", "./runtime": "./src/runtime/index.ts", + "./registry": "./src/registry/index.ts", "./observability": "./src/observability/index.ts", "./local": "./src/local.ts", "./sql-catalog": "./src/sql-catalog.ts", diff --git a/packages/query-engine/src/registry/index.ts b/packages/query-engine/src/registry/index.ts new file mode 100644 index 000000000..74a49658a --- /dev/null +++ b/packages/query-engine/src/registry/index.ts @@ -0,0 +1,9 @@ +/** + * Declarative warehouse query registry. + * + * Kept out of the root barrel deliberately: entries pull in `./runtime` for the + * cache-policy types, and the root barrel stays driver-free so web and cli can + * import it. Only `apps/api` should reach for this subpath. + */ +export { defineQuery, type QueryDef } from "./query-def" +export * as Queries from "./queries" diff --git a/packages/query-engine/src/registry/queries.ts b/packages/query-engine/src/registry/queries.ts new file mode 100644 index 000000000..c898aaa1b --- /dev/null +++ b/packages/query-engine/src/registry/queries.ts @@ -0,0 +1,116 @@ +import type { + ErrorRateByServiceRequest, + ErrorsByTypeRequest, + ErrorsSummaryRequest, + ErrorsTimeseriesRequest, + ServiceOverviewRequest, +} from "@maple/domain/http" +import * as CH from "../ch" +import { makeDirectRouteCachePolicy } from "../runtime/query-engine" +import { defineQuery } from "./query-def" + +/** + * The declarative warehouse query registry. + * + * Each entry replaces the profile/context/error-label/cache wiring that used to + * be repeated inline in every handler in `apps/api/src/routes/v1/query-engine.http.ts`. + * Handlers keep their own row-to-response mapping; see `QueryDef` for why + * decoding is deliberately out of scope here. + * + * Migration is incremental and the two surfaces coexist: a handler either takes + * a `QueryDef` through `runQuery` or keeps its inline wiring. Nothing breaks + * while entries are added. + * + * Cache values below are carried over EXACTLY as the handlers had them, so this + * pilot changes no caching behaviour — `cache: undefined` here means the handler + * was uncached before, not that being uncached is correct. Turning any of those + * on is a separate, separately-reviewed change with a justified TTL. + */ + +export const errorsByType = defineQuery({ + id: "errorsByType", + profile: "aggregation", + // Was uncached inline. Preserved as-is: changing it belongs in its own commit. + cache: undefined, + compile: (payload: ErrorsByTypeRequest, orgId: string) => + CH.compile( + CH.errorsByTypeQuery({ + rootOnly: payload.rootOnly, + services: payload.services, + deploymentEnvs: payload.deploymentEnvs, + fingerprintHashes: payload.fingerprintHashes, + limit: payload.limit, + }), + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) + +export const errorsTimeseries = defineQuery({ + id: "errorsTimeseries", + profile: "aggregation", + cache: undefined, + compile: (payload: ErrorsTimeseriesRequest, orgId: string) => + CH.compile( + CH.errorsTimeseriesQuery({ + fingerprintHash: payload.fingerprintHash, + services: payload.services, + }), + { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + // Matches the handler's previous inline default. The builder needs a + // bucket width and the request treats it as optional. + bucketSeconds: payload.bucketSeconds ?? 3600, + }, + ), +}) + +/** Single-row: the handler reads this through `runQueryFirst`. */ +export const errorsSummary = defineQuery({ + id: "errorsSummary", + profile: "aggregation", + cache: undefined, + compile: (payload: ErrorsSummaryRequest, orgId: string) => + CH.compile( + CH.errorsSummaryQuery({ + rootOnly: payload.rootOnly, + services: payload.services, + deploymentEnvs: payload.deploymentEnvs, + fingerprintHashes: payload.fingerprintHashes, + }), + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) + +export const errorRateByService = defineQuery({ + id: "errorRateByService", + profile: "aggregation", + cache: undefined, + // The builder takes no options — this query is scoped entirely by org and + // time range. The payload still carries the range. + compile: (payload: ErrorRateByServiceRequest, orgId: string) => + CH.compile(CH.errorRateByServiceQuery(), { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + }), +}) + +export const serviceOverview = defineQuery({ + id: "serviceOverview", + profile: "aggregation", + // v2: rows gained per-commit `firstSeen`; the version bump keeps pre-upgrade + // cached rows (missing the field) from being served. Carried over verbatim + // from the handler — do not renumber without the same reasoning. + cache: makeDirectRouteCachePolicy({ ttlSeconds: 15, version: 2 }), + compile: (payload: ServiceOverviewRequest, orgId: string) => + CH.compile( + CH.serviceOverviewQuery({ + environments: payload.environments, + namespaces: payload.namespaces, + commitShas: payload.commitShas, + }), + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) diff --git a/packages/query-engine/src/registry/query-def.ts b/packages/query-engine/src/registry/query-def.ts new file mode 100644 index 000000000..d3e3ffc8a --- /dev/null +++ b/packages/query-engine/src/registry/query-def.ts @@ -0,0 +1,86 @@ +import type { CompiledQuery } from "@maple-dev/clickhouse-builder" +import type { QueryProfileName, WarehouseQuerySettings } from "../profiles/query-profile" +import type { DirectRouteCachePolicyInput } from "../runtime/query-engine" + +/** + * One warehouse query, declared in one place. + * + * The problem this solves: every handler in `query-engine.http.ts` wires its own + * cost profile, span context, error label and — if the author remembered — + * caching. Caching being opt-in per call site meant silence read as "off": 11 of + * 61 handlers called `cachedDirect`, the other 50 were uncached by omission + * rather than by decision. Nothing in the type system asked. + * + * A `QueryDef` makes those choices data instead of boilerplate, so `runQuery` + * can apply them uniformly and a reviewer can see all of them at once. The + * single most important field is `cache`, which is REQUIRED and nullable rather + * than optional: `cache: undefined` is a decision someone made, a missing + * `cache` is a compile error. + * + * Scope note: this deliberately does NOT own row decoding or response shaping. + * Handlers still map rows into their own response class. Folding hand-coercion + * (`Number(row.count)`, `decodeFingerprintHash(...)`) into a declared + * `rowSchema` is a real improvement but a separate, riskier change — it turns + * ClickHouse's UInt64-as-JSON-string behaviour into a hard failure where it is + * currently a silent `Number("123")`, and every migrated query has to go + * through the DESCRIBE sweep (`bun run ch:up && bun run ch:test`) to prove it. + * Keep the two changes apart so a decode regression is never tangled up with a + * caching change. + */ +export interface QueryDef { + /** + * Stable identifier. Used as the warehouse span's `query.context`, as the + * `cachedDirect` route name (and therefore part of the cache key), and as the + * error label. One id, so those three can never drift apart the way they can + * when each is passed as its own string literal. + * + * Where a pipe name already exists for the same logical query, use it — the + * goal is one vocabulary, not a sixth. + */ + readonly id: string + + /** + * Cost profile, applied as ClickHouse `SETTINGS`. Defaults are NOT inherited + * here: state it, because "which budget does this query get" is exactly the + * kind of decision that goes stale silently when it is implicit. + */ + readonly profile: QueryProfileName + + /** + * Extra settings merged over the profile — e.g. `LOGS_BODY_SEARCH_SETTINGS` + * for log body search, whose `maxBlockSize: 512` is the difference between a + * sub-2s query and an OOM on a large-body org. + */ + readonly settings?: WarehouseQuerySettings + + /** + * Edge cache policy, or `undefined` for deliberately uncached. + * + * Required-but-nullable on purpose. This is the field that turns ~50 silent + * opt-outs into ~50 reviewed decisions, and it only works if omitting it + * fails to compile. + * + * Be careful raising a TTL on anything live-tailing: a stale entry there is a + * user-visible correctness bug that presents as "the dashboard is frozen". + */ + readonly cache: DirectRouteCachePolicyInput | undefined + + /** + * Build the compiled query from the request payload. + * + * `orgId` is passed separately rather than read off the payload because it + * comes from the authenticated tenant, never from user input — every query + * must filter `OrgId`, and that guarantee should not depend on a client + * sending the right value. + */ + readonly compile: (payload: Payload, orgId: string) => CompiledQuery +} + +/** + * Identity helper that pins `Payload`/`Row` from the `compile` function while + * still checking the object against `QueryDef`. + * + * Without it, annotating an entry as `QueryDef` forces you + * to restate the row type that `CH.compile` already knows, and those two drift. + */ +export const defineQuery = (def: QueryDef): QueryDef => def From f4facdb8507cea28f22db7da14995f763b1e4e33 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Wed, 5 Aug 2026 00:43:28 +0200 Subject: [PATCH 2/5] feat(query-engine): migrate 17 more handlers to the QueryDef registry Takes the registry from 5 to 22 of 61 handlers. query-engine.http.ts loses 327 lines; the remaining handlers keep their inline wiring and the two styles continue to coexist. Entries were generated from the handler source rather than written by hand, after the pilot showed that transcribing them from memory silently dropped builder options. The generator asserts, per handler, that the region it replaces contains exactly one CH.compile, no Effect.all, and no statements other than the query plumbing -- so nothing but boilerplate is deleted. Two type changes fell out of real handlers rather than speculation: * `settings` may now be a function of the payload. listLogs only wants LOGS_BODY_SEARCH_SETTINGS when the caller actually passed a search term, and a static field could not express that. Resolved to a spread so an undefined result omits the key instead of passing `settings: undefined`, which downstream would read as "clear the profile defaults". * serviceHealthSnapshot and serviceApdex called cachedDirect with no explicit policy, i.e. its default of CACHE_SNAP_S. That constant is not exported, but DirectRouteCachePolicyInput accepts a bare number and the value is 15, so `cache: 15` is byte-equivalent to passing nothing. Seven candidates were deliberately NOT migrated: * cloudflareInfraZoneTimeseries / cloudflareInfraWorkerTimeseries build on @maple/query-engine-integrations, which depends on query-engine -- putting them in this registry would invert that dependency. * getLog, fleetUtilizationTimeseries, podInfraTimeseries, nodeInfraTimeseries and workloadInfraTimeseries compute values in the handler body (a spec, a bucket width, a time range) that `compile` cannot see. Moving that logic in is worthwhile but is per-handler judgement, not a mechanical rewrite. Caching behaviour is unchanged everywhere: every policy is carried over verbatim, and `cache: undefined` records "was uncached", not "should be". Verified: query-engine + apps/api typecheck; SQL baseline byte-identical (14 tests), proving all 22 migrated handlers emit exactly the SQL they did before; apps/api routes + warehouse suites 310 pass. --- apps/api/src/routes/query-runner.ts | 15 +- apps/api/src/routes/v1/query-engine.http.ts | 350 ++---------------- packages/query-engine/src/registry/queries.ts | 299 +++++++++++++++ .../query-engine/src/registry/query-def.ts | 6 +- 4 files changed, 340 insertions(+), 330 deletions(-) diff --git a/apps/api/src/routes/query-runner.ts b/apps/api/src/routes/query-runner.ts index 9f38626a8..4aaff38b5 100644 --- a/apps/api/src/routes/query-runner.ts +++ b/apps/api/src/routes/query-runner.ts @@ -17,6 +17,17 @@ import { WarehouseQueryService } from "@/services/warehouse/WarehouseQueryServic * plumbing, not the presentation. */ +/** + * Settings may be static or payload-dependent. Resolved to a spread so an + * absent/undefined result omits the key entirely rather than passing an + * explicit `settings: undefined`, which would read as "clear the profile + * defaults" downstream. + */ +const resolveSettings = (def: QueryDef, payload: Payload) => { + const settings = typeof def.settings === "function" ? def.settings(payload) : def.settings + return settings === undefined ? {} : { settings } +} + /** * Shared tail: annotate failures with the query id, then apply the declared * cache policy (or don't). `cachedDirect` wraps the whole execution so a hit @@ -69,7 +80,7 @@ export const runQuery = ( payload, warehouse.compiledQuery(tenant, def.compile(payload, tenant.orgId), { profile: def.profile, - ...(def.settings === undefined ? {} : { settings: def.settings }), + ...resolveSettings(def, payload), context: def.id, }), ) @@ -95,7 +106,7 @@ export const runQueryFirst = ( warehouse .compiledQueryFirst(tenant, def.compile(payload, tenant.orgId), { profile: def.profile, - ...(def.settings === undefined ? {} : { settings: def.settings }), + ...resolveSettings(def, payload), context: def.id, }) .pipe(Effect.map(Option.getOrNull)), diff --git a/apps/api/src/routes/v1/query-engine.http.ts b/apps/api/src/routes/v1/query-engine.http.ts index 334950953..6a04e4d56 100644 --- a/apps/api/src/routes/v1/query-engine.http.ts +++ b/apps/api/src/routes/v1/query-engine.http.ts @@ -377,25 +377,9 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("errorDetailTraces", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.errorDetailTracesQuery({ - fingerprintHash: payload.fingerprintHash, - rootOnly: payload.rootOnly, - services: payload.services, - limit: payload.limit, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "list", - context: "errorDetailTraces", - }), - "errorDetailTraces query failed", - ) - const typedRows = rows + const rows = yield* runQuery(Queries.errorDetailTraces, tenant, payload) return new ErrorDetailTracesResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ traceId: decodeTraceId(row.traceId), startTime: String(row.startTime), durationMicros: Number(row.durationMicros), @@ -431,23 +415,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("serviceHealthSnapshot", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.serviceHealthSnapshotQuery({ environments: payload.environments }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - { rowSchema: CH.serviceHealthSnapshotRowSchema }, - ) - const rows = yield* queryEngine.cachedDirect( - tenant, - "serviceHealthSnapshot", - payload, - mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "serviceHealthSnapshot", - }), - "serviceHealthSnapshot query failed", - ), - ) + const rows = yield* runQuery(Queries.serviceHealthSnapshot, tenant, payload) return new ServiceHealthSnapshotResponse({ data: rows.map((row) => ({ serviceName: decodeServiceName(row.serviceName), @@ -462,31 +430,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("serviceHealthBaseline", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.serviceHealthBaselineQuery({ - environments: payload.environments, - namespaces: payload.namespaces, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const rows = yield* queryEngine.cachedDirect( - tenant, - "serviceHealthBaseline", - payload, - mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "serviceHealthBaseline", - }), - "serviceHealthBaseline query failed", - ), - // The payload's start/end are floored to the hour upstream - // (`floorToHour`) and this is a trailing 7-day baseline that - // changes at most hourly, so the cache key already rotates once - // an hour — a 1h TTL yields ≤1 recompute/hour per (org, env, ns) - // instead of every 15s for an ~900ms query. - 3600, - ) + const rows = yield* runQuery(Queries.serviceHealthBaseline, tenant, payload) return new ServiceHealthBaselineResponse({ data: rows.map((row) => ({ serviceName: decodeServiceName(String(row.serviceName ?? "")), @@ -501,33 +445,9 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("serviceApdex", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.serviceApdexTimeseriesQuery({ - serviceName: payload.serviceName, - apdexThresholdMs: payload.apdexThresholdMs, - }), - { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - bucketSeconds: payload.bucketSeconds ?? 60, - }, - ) - const rows = yield* queryEngine.cachedDirect( - tenant, - "serviceApdex", - payload, - mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "serviceApdex", - }), - "serviceApdex query failed", - ), - ) - const typedRows = rows + const rows = yield* runQuery(Queries.serviceApdex, tenant, payload) return new ServiceApdexResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ bucket: String(row.bucket), totalCount: Number(row.totalCount), satisfiedCount: Number(row.satisfiedCount), @@ -557,24 +477,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("serviceDependenciesForService", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.serviceDependenciesForServiceQuery({ - serviceName: payload.serviceName, - deploymentEnv: payload.deploymentEnv, - }), - { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "serviceDependenciesForService", - }), - "serviceDependenciesForService query failed", - ) + const rows = yield* runQuery(Queries.serviceDependenciesForService, tenant, payload) return new ServiceDependenciesResponse({ data: rows }) }), ) @@ -598,24 +501,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("serviceDbEdgesForService", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.serviceDbEdgesForServiceQuery({ - serviceName: payload.serviceName, - deploymentEnv: payload.deploymentEnv, - }), - { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "serviceDbEdgesForService", - }), - "serviceDbEdgesForService query failed", - ) + const rows = yield* runQuery(Queries.serviceDbEdgesForService, tenant, payload) return new ServiceDbEdgesResponse({ data: rows }) }), ) @@ -1991,47 +1877,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("listLogs", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.logsListQuery({ - serviceName: payload.service, - severity: payload.severity, - minSeverity: payload.minSeverity, - traceId: payload.traceId, - spanId: payload.spanId, - cursor: payload.cursor, - search: payload.search, - environments: payload.deploymentEnv ? [payload.deploymentEnv] : undefined, - namespaces: payload.namespace ? [payload.namespace] : undefined, - matchModes: Match.value([ - payload.deploymentEnvMatchMode, - payload.namespaceMatchMode, - ] as const).pipe( - Match.when([undefined, undefined], () => undefined), - Match.orElse(([deploymentEnv, serviceNamespace]) => ({ - deploymentEnv, - serviceNamespace, - })), - ), - limit: payload.limit, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const rows = yield* queryEngine.cachedDirect( - tenant, - "listLogs", - payload, - mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "list", - context: "listLogs", - // Body search reads the wide Body column for the ILIKE - // filter — cap the read block size (see - // WarehouseQuerySettings.maxBlockSize). - settings: payload.search ? LOGS_BODY_SEARCH_SETTINGS : undefined, - }), - "listLogs query failed", - ), - ) + const rows = yield* runQuery(Queries.listLogs, tenant, payload) return new ListLogsResponse({ data: rows }) }), ) @@ -2063,44 +1909,16 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("listMetrics", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.listMetricsQuery({ - serviceName: payload.service, - metricType: payload.metricType, - search: payload.search, - limit: payload.limit, - offset: payload.offset, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "discovery", - context: "listMetrics", - }), - "listMetrics query failed", - ) + const rows = yield* runQuery(Queries.listMetrics, tenant, payload) return new ListMetricsResponse({ data: rows }) }), ) .handle("metricsSummary", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile(CH.metricsSummaryQuery({ serviceName: payload.service }), { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - }) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "discovery", - context: "metricsSummary", - }), - "metricsSummary query failed", - ) - const typedRows = rows + const rows = yield* runQuery(Queries.metricsSummary, tenant, payload) return new MetricsSummaryResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ metricType: row.metricType, metricCount: Number(row.metricCount), dataPointCount: Number(row.dataPointCount), @@ -2260,21 +2078,9 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("listHosts", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.listHostsQuery({ - search: payload.search, - limit: payload.limit, - offset: payload.offset, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { profile: "list", context: "listHosts" }), - "listHosts query failed", - ) - const typedRows = rows + const rows = yield* runQuery(Queries.listHosts, tenant, payload) return new ListHostsResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ hostName: row.hostName, osType: row.osType, hostArch: row.hostArch, @@ -2291,20 +2097,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("hostDetailSummary", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile(CH.hostDetailSummaryQuery({ hostName: payload.hostName }), { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - }) - const row = yield* mapExecError( - warehouse - .compiledQueryFirst(tenant, compiled, { - profile: "aggregation", - context: "hostDetailSummary", - }) - .pipe(Effect.map(Option.getOrNull)), - "hostDetailSummary query failed", - ) + const row = yield* runQueryFirst(Queries.hostDetailSummary, tenant, payload) return new HostDetailSummaryResponse({ data: row ? { @@ -2574,27 +2367,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("podsSummary", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - // Scope only — no row filters. The browse band is what tells you how - // much of the fleet your filters just hid, so narrowing it too would - // defeat the point. - const compiled = CH.compile( - CH.listPodsSummaryQuery({ - namespaces: payload.namespaces, - clusters: payload.clusters, - environments: payload.environments, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - { rowSchema: CH.ListPodsSummaryOutputSchema }, - ) - const row = yield* mapExecError( - warehouse - .compiledQueryFirst(tenant, compiled, { - profile: "aggregation", - context: "podsSummary", - }) - .pipe(Effect.map(Option.getOrNull)), - "podsSummary query failed", - ) + const row = yield* runQueryFirst(Queries.podsSummary, tenant, payload) return new PodsSummaryResponse({ totalPods: Number(row?.totalPods) || 0, saturatedPods: Number(row?.saturatedPods) || 0, @@ -2607,19 +2380,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("podDetailSummary", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.podDetailSummaryQuery({ podName: payload.podName, namespace: payload.namespace }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const row = yield* mapExecError( - warehouse - .compiledQueryFirst(tenant, compiled, { - profile: "aggregation", - context: "podDetailSummary", - }) - .pipe(Effect.map(Option.getOrNull)), - "podDetailSummary query failed", - ) + const row = yield* runQueryFirst(Queries.podDetailSummary, tenant, payload) return new PodDetailSummaryResponse({ data: row ? { @@ -2711,24 +2472,9 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("listNodes", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.listNodesQuery({ - search: payload.search, - nodeNames: payload.nodeNames, - clusters: payload.clusters, - environments: payload.environments, - limit: payload.limit, - offset: payload.offset, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { profile: "list", context: "listNodes" }), - "listNodes query failed", - ) - const typedRows = rows + const rows = yield* runQuery(Queries.listNodes, tenant, payload) return new ListNodesResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ nodeName: row.nodeName, nodeUid: row.nodeUid, clusterName: row.clusterName, @@ -2744,20 +2490,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("nodeDetailSummary", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile(CH.nodeDetailSummaryQuery({ nodeName: payload.nodeName }), { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - }) - const row = yield* mapExecError( - warehouse - .compiledQueryFirst(tenant, compiled, { - profile: "aggregation", - context: "nodeDetailSummary", - }) - .pipe(Effect.map(Option.getOrNull)), - "nodeDetailSummary query failed", - ) + const row = yield* runQueryFirst(Queries.nodeDetailSummary, tenant, payload) return new NodeDetailSummaryResponse({ data: row ? { @@ -2821,30 +2554,9 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("listWorkloads", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.listWorkloadsQuery({ - kind: payload.kind, - search: payload.search, - workloadNames: payload.workloadNames, - namespaces: payload.namespaces, - clusters: payload.clusters, - environments: payload.environments, - computeTypes: payload.computeTypes, - limit: payload.limit, - offset: payload.offset, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "list", - context: "listWorkloads", - }), - "listWorkloads query failed", - ) - const typedRows = rows + const rows = yield* runQuery(Queries.listWorkloads, tenant, payload) return new ListWorkloadsResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ workloadName: row.workloadName, namespace: row.namespace, clusterName: row.clusterName, @@ -2861,23 +2573,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("workloadDetailSummary", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - CH.workloadDetailSummaryQuery({ - kind: payload.kind, - workloadName: payload.workloadName, - namespace: payload.namespace, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const row = yield* mapExecError( - warehouse - .compiledQueryFirst(tenant, compiled, { - profile: "aggregation", - context: "workloadDetailSummary", - }) - .pipe(Effect.map(Option.getOrNull)), - "workloadDetailSummary query failed", - ) + const row = yield* runQueryFirst(Queries.workloadDetailSummary, tenant, payload) return new WorkloadDetailSummaryResponse({ data: row ? { diff --git a/packages/query-engine/src/registry/queries.ts b/packages/query-engine/src/registry/queries.ts index c898aaa1b..4c01320cf 100644 --- a/packages/query-engine/src/registry/queries.ts +++ b/packages/query-engine/src/registry/queries.ts @@ -1,11 +1,30 @@ import type { + ErrorDetailTracesRequest, ErrorRateByServiceRequest, ErrorsByTypeRequest, ErrorsSummaryRequest, ErrorsTimeseriesRequest, + HostDetailSummaryRequest, + ListHostsRequest, + ListLogsRequest, + ListMetricsRequest, + ListNodesRequest, + ListWorkloadsRequest, + MetricsSummaryRequest, + NodeDetailSummaryRequest, + PodDetailSummaryRequest, + PodsSummaryRequest, + ServiceApdexRequest, + ServiceDbEdgesForServiceRequest, + ServiceDependenciesForServiceRequest, + ServiceHealthBaselineRequest, + ServiceHealthSnapshotRequest, ServiceOverviewRequest, + WorkloadDetailSummaryRequest, } from "@maple/domain/http" +import { Match } from "effect" import * as CH from "../ch" +import { LOGS_BODY_SEARCH_SETTINGS } from "../profiles" import { makeDirectRouteCachePolicy } from "../runtime/query-engine" import { defineQuery } from "./query-def" @@ -114,3 +133,283 @@ export const serviceOverview = defineQuery({ { orgId, startTime: payload.startTime, endTime: payload.endTime }, ), }) + +export const errorDetailTraces = defineQuery({ + id: "errorDetailTraces", + profile: "list", + cache: undefined, + compile: (payload: ErrorDetailTracesRequest, orgId: string) => + CH.compile( + CH.errorDetailTracesQuery({ + fingerprintHash: payload.fingerprintHash, + rootOnly: payload.rootOnly, + services: payload.services, + limit: payload.limit, + }), + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) + +export const serviceHealthSnapshot = defineQuery({ + id: "serviceHealthSnapshot", + profile: "aggregation", + cache: 15, + compile: (payload: ServiceHealthSnapshotRequest, orgId: string) => + CH.compile( + CH.serviceHealthSnapshotQuery({ environments: payload.environments }), + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + { rowSchema: CH.serviceHealthSnapshotRowSchema }, + ), +}) + +export const serviceHealthBaseline = defineQuery({ + id: "serviceHealthBaseline", + profile: "aggregation", + cache: 3600, + compile: (payload: ServiceHealthBaselineRequest, orgId: string) => + CH.compile( + CH.serviceHealthBaselineQuery({ + environments: payload.environments, + namespaces: payload.namespaces, + }), + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) + +export const serviceApdex = defineQuery({ + id: "serviceApdex", + profile: "aggregation", + cache: 15, + compile: (payload: ServiceApdexRequest, orgId: string) => + CH.compile( + CH.serviceApdexTimeseriesQuery({ + serviceName: payload.serviceName, + apdexThresholdMs: payload.apdexThresholdMs, + }), + { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + bucketSeconds: payload.bucketSeconds ?? 60, + }, + ), +}) + +export const serviceDependenciesForService = defineQuery({ + id: "serviceDependenciesForService", + profile: "aggregation", + cache: undefined, + compile: (payload: ServiceDependenciesForServiceRequest, orgId: string) => + CH.compile( + CH.serviceDependenciesForServiceQuery({ + serviceName: payload.serviceName, + deploymentEnv: payload.deploymentEnv, + }), + { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + }, + ), +}) + +export const serviceDbEdgesForService = defineQuery({ + id: "serviceDbEdgesForService", + profile: "aggregation", + cache: undefined, + compile: (payload: ServiceDbEdgesForServiceRequest, orgId: string) => + CH.compile( + CH.serviceDbEdgesForServiceQuery({ + serviceName: payload.serviceName, + deploymentEnv: payload.deploymentEnv, + }), + { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + }, + ), +}) + +export const listLogs = defineQuery({ + id: "listLogs", + profile: "list", + settings: (payload) => (payload.search ? LOGS_BODY_SEARCH_SETTINGS : undefined), + cache: 15, + compile: (payload: ListLogsRequest, orgId: string) => + CH.compile( + CH.logsListQuery({ + serviceName: payload.service, + severity: payload.severity, + minSeverity: payload.minSeverity, + traceId: payload.traceId, + spanId: payload.spanId, + cursor: payload.cursor, + search: payload.search, + environments: payload.deploymentEnv ? [payload.deploymentEnv] : undefined, + namespaces: payload.namespace ? [payload.namespace] : undefined, + matchModes: Match.value([ + payload.deploymentEnvMatchMode, + payload.namespaceMatchMode, + ] as const).pipe( + Match.when([undefined, undefined], () => undefined), + Match.orElse(([deploymentEnv, serviceNamespace]) => ({ + deploymentEnv, + serviceNamespace, + })), + ), + limit: payload.limit, + }), + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) + +export const listMetrics = defineQuery({ + id: "listMetrics", + profile: "discovery", + cache: undefined, + compile: (payload: ListMetricsRequest, orgId: string) => + CH.compile( + CH.listMetricsQuery({ + serviceName: payload.service, + metricType: payload.metricType, + search: payload.search, + limit: payload.limit, + offset: payload.offset, + }), + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) + +export const metricsSummary = defineQuery({ + id: "metricsSummary", + profile: "discovery", + cache: undefined, + compile: (payload: MetricsSummaryRequest, orgId: string) => + CH.compile(CH.metricsSummaryQuery({ serviceName: payload.service }), { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + }), +}) + +export const listHosts = defineQuery({ + id: "listHosts", + profile: "list", + cache: undefined, + compile: (payload: ListHostsRequest, orgId: string) => + CH.compile( + CH.listHostsQuery({ + search: payload.search, + limit: payload.limit, + offset: payload.offset, + }), + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) + +export const hostDetailSummary = defineQuery({ + id: "hostDetailSummary", + profile: "aggregation", + cache: undefined, + compile: (payload: HostDetailSummaryRequest, orgId: string) => + CH.compile(CH.hostDetailSummaryQuery({ hostName: payload.hostName }), { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + }), +}) + +export const podsSummary = defineQuery({ + id: "podsSummary", + profile: "aggregation", + cache: undefined, + compile: (payload: PodsSummaryRequest, orgId: string) => + CH.compile( + CH.listPodsSummaryQuery({ + namespaces: payload.namespaces, + clusters: payload.clusters, + environments: payload.environments, + }), + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + { rowSchema: CH.ListPodsSummaryOutputSchema }, + ), +}) + +export const podDetailSummary = defineQuery({ + id: "podDetailSummary", + profile: "aggregation", + cache: undefined, + compile: (payload: PodDetailSummaryRequest, orgId: string) => + CH.compile(CH.podDetailSummaryQuery({ podName: payload.podName, namespace: payload.namespace }), { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + }), +}) + +export const listNodes = defineQuery({ + id: "listNodes", + profile: "list", + cache: undefined, + compile: (payload: ListNodesRequest, orgId: string) => + CH.compile( + CH.listNodesQuery({ + search: payload.search, + nodeNames: payload.nodeNames, + clusters: payload.clusters, + environments: payload.environments, + limit: payload.limit, + offset: payload.offset, + }), + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) + +export const nodeDetailSummary = defineQuery({ + id: "nodeDetailSummary", + profile: "aggregation", + cache: undefined, + compile: (payload: NodeDetailSummaryRequest, orgId: string) => + CH.compile(CH.nodeDetailSummaryQuery({ nodeName: payload.nodeName }), { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + }), +}) + +export const listWorkloads = defineQuery({ + id: "listWorkloads", + profile: "list", + cache: undefined, + compile: (payload: ListWorkloadsRequest, orgId: string) => + CH.compile( + CH.listWorkloadsQuery({ + kind: payload.kind, + search: payload.search, + workloadNames: payload.workloadNames, + namespaces: payload.namespaces, + clusters: payload.clusters, + environments: payload.environments, + computeTypes: payload.computeTypes, + limit: payload.limit, + offset: payload.offset, + }), + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) + +export const workloadDetailSummary = defineQuery({ + id: "workloadDetailSummary", + profile: "aggregation", + cache: undefined, + compile: (payload: WorkloadDetailSummaryRequest, orgId: string) => + CH.compile( + CH.workloadDetailSummaryQuery({ + kind: payload.kind, + workloadName: payload.workloadName, + namespace: payload.namespace, + }), + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) diff --git a/packages/query-engine/src/registry/query-def.ts b/packages/query-engine/src/registry/query-def.ts index d3e3ffc8a..bde8d2668 100644 --- a/packages/query-engine/src/registry/query-def.ts +++ b/packages/query-engine/src/registry/query-def.ts @@ -50,8 +50,12 @@ export interface QueryDef { * Extra settings merged over the profile — e.g. `LOGS_BODY_SEARCH_SETTINGS` * for log body search, whose `maxBlockSize: 512` is the difference between a * sub-2s query and an OOM on a large-body org. + * + * May be a function of the payload, because whether a query needs them can + * depend on the request: `listLogs` only wants the body-search settings when + * the caller actually passed a search term. */ - readonly settings?: WarehouseQuerySettings + readonly settings?: WarehouseQuerySettings | ((payload: Payload) => WarehouseQuerySettings | undefined) /** * Edge cache policy, or `undefined` for deliberately uncached. From ac95f7e9f5a89067a80c823b2c9b1534d55f8bc9 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Wed, 5 Aug 2026 00:51:05 +0200 Subject: [PATCH 3/5] feat(query-engine): app-side registry unblocks 8 more handlers (30/61) Three of the seven handlers skipped last round were skipped for a dependency reason, not a difficulty one: the Cloudflare queries come from @maple/query-engine-integrations, which itself depends on @maple/query-engine, so declaring them in the core registry would invert that edge. The rest needed helpers owned by the API app. Adds apps/api/src/routes/queries.ts, which spreads the core registry and adds entries that need app-side dependencies. Handlers import `Queries` from there, so the split is invisible at the call site and an entry can move between halves without touching a handler. Extracts apps/api/src/routes/query-helpers.ts for logic both sides need. The pod/node/workload metric switches are the interesting case: `compile` needs the metric name while the handler needs the unit for its response, so the switch has to be shared. Duplicating it would let the two drift -- the exact failure this registry exists to prevent. `cache` may now be a function of payload and current time. spanDetail needs it: a finished trace is immutable and cacheable, one still receiving spans is not, so its TTL comes from the requested end time against now. `nowMs` is supplied by the runner from the Effect Clock rather than read in the def, and the clock is only read when a def actually asks for it. cloudflareInfraZoneTimeseries keeps its local `filters` binding: the response reports which filters the metric family could not honor, which is presentation, not query construction. Verified: apps/api typecheck; SQL baseline byte-identical (14 tests) across all 30 migrated handlers; apps/api routes + warehouse suites 310 pass. --- apps/api/src/routes/queries.ts | 197 ++++++++++++ apps/api/src/routes/query-helpers.ts | 113 +++++++ apps/api/src/routes/query-runner.ts | 12 +- apps/api/src/routes/v1/query-engine.http.ts | 286 ++---------------- .../query-engine/src/registry/query-def.ts | 11 +- 5 files changed, 352 insertions(+), 267 deletions(-) create mode 100644 apps/api/src/routes/queries.ts create mode 100644 apps/api/src/routes/query-helpers.ts diff --git a/apps/api/src/routes/queries.ts b/apps/api/src/routes/queries.ts new file mode 100644 index 000000000..072b13cdc --- /dev/null +++ b/apps/api/src/routes/queries.ts @@ -0,0 +1,197 @@ +import * as CH from "@maple/query-engine/ch" +import * as Integrations from "@maple/query-engine-integrations" +import { defineQuery } from "@maple/query-engine/registry" +import { Queries as Core } from "@maple/query-engine/registry" +import type { + CloudflareInfraWorkerTimeseriesRequest, + CloudflareInfraZoneTimeseriesRequest, + FleetUtilizationTimeseriesRequest, + GetLogRequest, + NodeInfraTimeseriesRequest, + PodInfraTimeseriesRequest, + SpanDetailRequest, + WorkloadInfraTimeseriesRequest, +} from "@maple/domain/http" +import { + nodeMetricSpec, + partitionWindowAround, + podMetricSpec, + toCloudflareFilters, + workloadMetricSpec, +} from "@/routes/query-helpers" +import { traceCacheTtlSeconds } from "@/services/warehouse/trace-detail-cache" + +/** + * App-side half of the warehouse query registry. + * + * Most entries live in `@maple/query-engine/registry`. These do not, for two + * reasons that are both about dependency direction rather than taste: + * + * * The Cloudflare and PlanetScale queries are built by + * `@maple/query-engine-integrations`, which itself depends on + * `@maple/query-engine`. Declaring them in the core registry would invert + * that edge. + * * A few queries need helpers or services that belong to the API app — + * `partitionWindowAround`, `traceCacheTtlSeconds` — and pulling those down + * into the query-engine package would drag app concerns into a shared lib. + * + * Handlers import `Queries` from here, so the split is invisible at the call + * site and an entry can move between the two halves without touching handlers. + */ +export const Queries = { + ...Core, + + /** + * Bounded to a ±1h window around the requested log so ClickHouse can prune + * partitions instead of reading every retained daily partition for an + * exact-timestamp match. That window used to be computed in the handler. + */ + getLog: defineQuery({ + id: "getLog", + profile: "list", + cache: undefined, + compile: (payload: GetLogRequest, orgId: string) => { + const { startTime, endTime } = partitionWindowAround(payload.timestamp) + return CH.compile( + CH.getLogByKeyQuery({ + serviceName: payload.serviceName, + traceId: payload.traceId, + spanId: payload.spanId, + }), + { orgId, startTime, endTime, timestamp: payload.timestamp }, + ) + }, + }), + + /** + * A finished trace is immutable and cacheable; one still receiving spans is + * not. `traceCacheTtlSeconds` decides from the requested end time against + * now, which is why this def takes the dynamic-cache form. + */ + spanDetail: defineQuery({ + id: "spanDetail", + profile: "discovery", + cache: (payload: SpanDetailRequest, nowMs: number) => traceCacheTtlSeconds(payload.endTime, nowMs), + compile: (payload: SpanDetailRequest, orgId: string) => { + // Without both bounds there is no window to narrow to, and passing a + // half-open range would widen the scan rather than prune it. + const narrowByTime = payload.startTime != null && payload.endTime != null + return CH.compile( + CH.spanDetailQuery({ + traceId: payload.traceId, + spanId: payload.spanId, + narrowByTime, + }), + narrowByTime ? { orgId, startTime: payload.startTime, endTime: payload.endTime } : { orgId }, + ) + }, + }), + + fleetUtilizationTimeseries: defineQuery({ + id: "fleetUtilizationTimeseries", + profile: "aggregation", + cache: undefined, + compile: (payload: FleetUtilizationTimeseriesRequest, orgId: string) => + CH.compile(CH.fleetUtilizationTimeseriesQuery(), { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + bucketSeconds: payload.bucketSeconds ?? 300, + }), + }), + + podInfraTimeseries: defineQuery({ + id: "podInfraTimeseries", + profile: "aggregation", + cache: undefined, + compile: (payload: PodInfraTimeseriesRequest, orgId: string) => + CH.compile( + CH.podGaugeTimeseriesQuery({ + podName: payload.podName, + namespace: payload.namespace, + metricName: podMetricSpec(payload.metric).metricName, + }), + { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + bucketSeconds: payload.bucketSeconds ?? 60, + }, + ), + }), + + nodeInfraTimeseries: defineQuery({ + id: "nodeInfraTimeseries", + profile: "aggregation", + cache: undefined, + compile: (payload: NodeInfraTimeseriesRequest, orgId: string) => + CH.compile( + CH.nodeGaugeTimeseriesQuery({ + nodeName: payload.nodeName, + metricName: nodeMetricSpec(payload.metric).metricName, + }), + { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + bucketSeconds: payload.bucketSeconds ?? 60, + }, + ), + }), + + workloadInfraTimeseries: defineQuery({ + id: "workloadInfraTimeseries", + profile: "aggregation", + cache: undefined, + compile: (payload: WorkloadInfraTimeseriesRequest, orgId: string) => + CH.compile( + CH.workloadGaugeTimeseriesQuery({ + kind: payload.kind, + workloadName: payload.workloadName, + namespace: payload.namespace, + metricName: workloadMetricSpec(payload.metric).metricName, + groupByPod: payload.groupByPod, + }), + { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + bucketSeconds: payload.bucketSeconds ?? 60, + }, + ), + }), + + cloudflareInfraZoneTimeseries: defineQuery({ + id: "cloudflareInfraZoneTimeseries", + profile: "aggregation", + cache: undefined, + compile: (payload: CloudflareInfraZoneTimeseriesRequest, orgId: string) => + CH.compile( + Integrations.cloudflareZoneTimeseriesSQL(toCloudflareFilters(payload)), + { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + bucketSeconds: payload.bucketSeconds, + }, + { rowSchema: Integrations.cloudflareZoneTimeseriesRowSchema }, + ), + }), + + cloudflareInfraWorkerTimeseries: defineQuery({ + id: "cloudflareInfraWorkerTimeseries", + profile: "aggregation", + cache: undefined, + compile: (payload: CloudflareInfraWorkerTimeseriesRequest, orgId: string) => + CH.compile( + Integrations.cloudflareWorkerTimeseriesSQL(), + { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + bucketSeconds: payload.bucketSeconds, + }, + { rowSchema: Integrations.cloudflareWorkerTimeseriesRowSchema }, + ), + }), +} as const diff --git a/apps/api/src/routes/query-helpers.ts b/apps/api/src/routes/query-helpers.ts new file mode 100644 index 000000000..c5278bde1 --- /dev/null +++ b/apps/api/src/routes/query-helpers.ts @@ -0,0 +1,113 @@ +import * as Integrations from "@maple/query-engine-integrations" +import { formatWarehouseDateTime, parseWarehouseDateTime } from "@maple/query-engine" +import type { + NodeInfraTimeseriesRequest, + PodInfraTimeseriesRequest, + WorkloadInfraTimeseriesRequest, +} from "@maple/domain/http" + +/** + * Helpers shared between the query-engine handlers and the app-side query + * registry (`./queries`). + * + * They live here rather than in either caller because both need them: the + * registry's `compile` needs the metric name, while the handler needs the unit + * for the response. Duplicating the switch would let those two drift, which is + * exactly the failure the registry exists to prevent. + */ + +export const toCloudflareFilters = (payload: { + readonly hosts?: ReadonlyArray | undefined + readonly cacheStatuses?: ReadonlyArray | undefined + readonly statusClasses?: ReadonlyArray | undefined + readonly paths?: ReadonlyArray | undefined + readonly pathContains?: string | undefined + readonly countries?: ReadonlyArray | undefined + readonly methods?: ReadonlyArray | undefined + readonly protocols?: ReadonlyArray | undefined + readonly deviceTypes?: ReadonlyArray | undefined + readonly firewallActions?: ReadonlyArray | undefined + readonly firewallSources?: ReadonlyArray | undefined + readonly firewallRuleIds?: ReadonlyArray | undefined + readonly dnsQueryNames?: ReadonlyArray | undefined + readonly dnsResponseCodes?: ReadonlyArray | undefined +}): Integrations.CloudflareFilterOpts => ({ + hosts: payload.hosts, + cacheStatuses: payload.cacheStatuses, + statusClasses: payload.statusClasses, + paths: payload.paths, + pathContains: payload.pathContains, + countries: payload.countries, + methods: payload.methods, + protocols: payload.protocols, + deviceTypes: payload.deviceTypes, + firewallActions: payload.firewallActions, + firewallSources: payload.firewallSources, + firewallRuleIds: payload.firewallRuleIds, + dnsQueryNames: payload.dnsQueryNames, + dnsResponseCodes: payload.dnsResponseCodes, +}) + +export const partitionWindowAround = (timestamp: string): { startTime: string; endTime: string } => { + const ms = parseWarehouseDateTime(timestamp) + return { + startTime: formatWarehouseDateTime(ms - 3_600_000), + endTime: formatWarehouseDateTime(ms + 3_600_000), + } +} + +/** Metric name + response unit for a pod infra metric. */ +export const podMetricSpec = (metric: PodInfraTimeseriesRequest["metric"]) => { + switch (metric) { + case "cpu_usage": + return { metricName: "k8s.pod.cpu.usage", unit: "cores" as const } + case "cpu_limit": + return { + metricName: "k8s.pod.cpu_limit_utilization", + unit: "percent" as const, + } + case "cpu_request": + return { + metricName: "k8s.pod.cpu_request_utilization", + unit: "percent" as const, + } + case "memory_limit": + return { + metricName: "k8s.pod.memory_limit_utilization", + unit: "percent" as const, + } + case "memory_request": + return { + metricName: "k8s.pod.memory_request_utilization", + unit: "percent" as const, + } + } +} + +/** Metric name + response unit for a node infra metric. */ +export const nodeMetricSpec = (metric: NodeInfraTimeseriesRequest["metric"]) => { + switch (metric) { + case "cpu_usage": + return { metricName: "k8s.node.cpu.usage", unit: "cores" as const } + case "uptime": + return { metricName: "k8s.node.uptime", unit: "seconds" as const } + } +} + +/** Metric name + response unit for a workload infra metric. */ +export const workloadMetricSpec = (metric: WorkloadInfraTimeseriesRequest["metric"]) => { + switch (metric) { + case "cpu_usage": + return { metricName: "k8s.pod.cpu.usage", unit: "cores" as const } + case "cpu_limit": + return { + metricName: "k8s.pod.cpu_limit_utilization", + unit: "percent" as const, + } + case "memory_limit": + return { + metricName: "k8s.pod.memory_limit_utilization", + unit: "percent" as const, + } + } +} diff --git a/apps/api/src/routes/query-runner.ts b/apps/api/src/routes/query-runner.ts index 4aaff38b5..db7e788dc 100644 --- a/apps/api/src/routes/query-runner.ts +++ b/apps/api/src/routes/query-runner.ts @@ -1,6 +1,6 @@ import type { QueryDef } from "@maple/query-engine/registry" import type { QueryEngineDirectError } from "@maple/query-engine/runtime" -import { Effect, Option } from "effect" +import { Clock, Effect, Option } from "effect" import type { TenantContext } from "@/services/auth/AuthService" import { QueryEngineService } from "@/services/warehouse/QueryEngineService" import { WarehouseQueryService } from "@/services/warehouse/WarehouseQueryService" @@ -43,6 +43,12 @@ const withPolicy = ( ) => Effect.gen(function* () { const queryEngine = yield* QueryEngineService + // Only read the clock when a def actually needs it — a static policy must + // not pay for, or depend on, a Clock read. + const cache = + typeof def.cache === "function" + ? def.cache(payload, yield* Clock.currentTimeMillis) + : def.cache const labelled = execute.pipe( // Same annotation the old inline `mapExecError` produced, with the label // derived from `def.id` rather than a hand-written string that could @@ -53,10 +59,10 @@ const withPolicy = ( }), ), ) - if (def.cache === undefined) { + if (cache === undefined) { return yield* labelled } - return yield* queryEngine.cachedDirect(tenant, def.id, payload, labelled, def.cache) + return yield* queryEngine.cachedDirect(tenant, def.id, payload, labelled, cache) }) /** diff --git a/apps/api/src/routes/v1/query-engine.http.ts b/apps/api/src/routes/v1/query-engine.http.ts index 6a04e4d56..8107700e7 100644 --- a/apps/api/src/routes/v1/query-engine.http.ts +++ b/apps/api/src/routes/v1/query-engine.http.ts @@ -83,7 +83,14 @@ import { parseWarehouseDateTime, } from "@maple/query-engine" import { LOGS_BODY_SEARCH_SETTINGS } from "@maple/query-engine/profiles" -import { Queries } from "@maple/query-engine/registry" +import { + nodeMetricSpec, + partitionWindowAround, + podMetricSpec, + toCloudflareFilters, + workloadMetricSpec, +} from "@/routes/query-helpers" +import { Queries } from "@/routes/queries" import { runQuery, runQueryFirst } from "@/routes/query-runner" import type { ExecutionTenant, WarehouseSqlError } from "@maple/query-engine/execution" import { buildBreakdownQuerySpec, buildTimeseriesQuerySpec } from "@maple/query-engine/query-builder" @@ -105,37 +112,6 @@ const mapExecError = (effect: Effect.Effect, context: string): * `Integrations.cloudflareIgnoredFiltersFor` answers that, and the answer ships in the response so the UI can * mark a panel zone-wide instead of pretending the filter applied. */ -const toCloudflareFilters = (payload: { - readonly hosts?: ReadonlyArray | undefined - readonly cacheStatuses?: ReadonlyArray | undefined - readonly statusClasses?: ReadonlyArray | undefined - readonly paths?: ReadonlyArray | undefined - readonly pathContains?: string | undefined - readonly countries?: ReadonlyArray | undefined - readonly methods?: ReadonlyArray | undefined - readonly protocols?: ReadonlyArray | undefined - readonly deviceTypes?: ReadonlyArray | undefined - readonly firewallActions?: ReadonlyArray | undefined - readonly firewallSources?: ReadonlyArray | undefined - readonly firewallRuleIds?: ReadonlyArray | undefined - readonly dnsQueryNames?: ReadonlyArray | undefined - readonly dnsResponseCodes?: ReadonlyArray | undefined -}): Integrations.CloudflareFilterOpts => ({ - hosts: payload.hosts, - cacheStatuses: payload.cacheStatuses, - statusClasses: payload.statusClasses, - paths: payload.paths, - pathContains: payload.pathContains, - countries: payload.countries, - methods: payload.methods, - protocols: payload.protocols, - deviceTypes: payload.deviceTypes, - firewallActions: payload.firewallActions, - firewallSources: payload.firewallSources, - firewallRuleIds: payload.firewallRuleIds, - dnsQueryNames: payload.dnsQueryNames, - dnsResponseCodes: payload.dnsResponseCodes, -}) const isMissingServiceOperationsRollup = (error: unknown): boolean => { if (typeof error !== "object" || error === null) return false @@ -165,15 +141,6 @@ const decodeStatusCodeOption = Schema.decodeUnknownOption(StatusCode) const coerceStatusCode = (value: string): StatusCode => Option.getOrElse(decodeStatusCodeOption(value), () => "Unset" as const) -// Build a ±1h partition-pruning window around a ClickHouse datetime string. -const partitionWindowAround = (timestamp: string): { startTime: string; endTime: string } => { - const ms = parseWarehouseDateTime(timestamp) - return { - startTime: formatWarehouseDateTime(ms - 3_600_000), - endTime: formatWarehouseDateTime(ms + 3_600_000), - } -} - // Most traces opened without a timestamp are still recent (list rows carry // `?t=`; it's direct/shared/AI links that don't, and those overwhelmingly // point at fresh traces). Probing the last 48h first prunes to ~2 daily @@ -290,33 +257,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("spanDetail", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const nowMs = yield* Clock.currentTimeMillis - const narrowByTime = payload.startTime != null && payload.endTime != null - const compiled = CH.compile( - CH.spanDetailQuery({ - traceId: payload.traceId, - spanId: payload.spanId, - narrowByTime, - }), - narrowByTime - ? { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime } - : { orgId: tenant.orgId }, - ) - const row = yield* queryEngine.cachedDirect( - tenant, - "spanDetail", - payload, - mapExecError( - warehouse - .compiledQueryFirst(tenant, compiled, { - profile: "discovery", - context: "spanDetail", - }) - .pipe(Effect.map(Option.getOrNull)), - "spanDetail query failed", - ), - traceCacheTtlSeconds(payload.endTime, nowMs), - ) + const row = yield* runQueryFirst(Queries.spanDetail, tenant, payload) return new SpanDetailResponse({ data: row ? { @@ -799,23 +740,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", Effect.gen(function* () { const tenant = yield* CurrentTenant.Context const filters = toCloudflareFilters(payload) - const compiled = CH.compile( - Integrations.cloudflareZoneTimeseriesSQL(filters), - { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - bucketSeconds: payload.bucketSeconds, - }, - { rowSchema: Integrations.cloudflareZoneTimeseriesRowSchema }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "cloudflareInfraZoneTimeseries", - }), - "cloudflareInfraZoneTimeseries query failed", - ) + const rows = yield* runQuery(Queries.cloudflareInfraZoneTimeseries, tenant, payload) return new CloudflareInfraZoneTimeseriesResponse({ data: rows.map((row) => ({ ...row })), ignoredFilters: Integrations.cloudflareIgnoredFiltersFor(filters, [ @@ -1325,23 +1250,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("cloudflareInfraWorkerTimeseries", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.compile( - Integrations.cloudflareWorkerTimeseriesSQL(), - { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - bucketSeconds: payload.bucketSeconds, - }, - { rowSchema: Integrations.cloudflareWorkerTimeseriesRowSchema }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "cloudflareInfraWorkerTimeseries", - }), - "cloudflareInfraWorkerTimeseries query failed", - ) + const rows = yield* runQuery(Queries.cloudflareInfraWorkerTimeseries, tenant, payload) return new CloudflareInfraWorkerTimeseriesResponse({ data: rows.map((row) => ({ ...row })), }) @@ -1884,25 +1793,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("getLog", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - // Bound the scan to a ±1h window around the requested log so - // ClickHouse can prune partitions instead of reading every - // retained daily partition for an exact-timestamp match. - const { startTime, endTime } = partitionWindowAround(payload.timestamp) - const compiled = CH.compile( - CH.getLogByKeyQuery({ - serviceName: payload.serviceName, - traceId: payload.traceId, - spanId: payload.spanId, - }), - { orgId: tenant.orgId, startTime, endTime, timestamp: payload.timestamp }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "list", - context: "getLog", - }), - "getLog query failed", - ) + const rows = yield* runQuery(Queries.getLog, tenant, payload) return new GetLogResponse({ data: rows }) }), ) @@ -2120,23 +2011,9 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("fleetUtilizationTimeseries", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const bucketSeconds = payload.bucketSeconds ?? 300 - const compiled = CH.compile(CH.fleetUtilizationTimeseriesQuery(), { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - bucketSeconds, - }) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "fleetUtilizationTimeseries", - }), - "fleetUtilizationTimeseries query failed", - ) - const typedRows = rows + const rows = yield* runQuery(Queries.fleetUtilizationTimeseries, tenant, payload) return new FleetUtilizationTimeseriesResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ bucket: String(row.bucket), avgCpu: Number(row.avgCpu) || 0, avgMemory: Number(row.avgMemory) || 0, @@ -2409,58 +2286,10 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("podInfraTimeseries", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const bucketSeconds = payload.bucketSeconds ?? 60 - - const spec = (() => { - switch (payload.metric) { - case "cpu_usage": - return { metricName: "k8s.pod.cpu.usage", unit: "cores" as const } - case "cpu_limit": - return { - metricName: "k8s.pod.cpu_limit_utilization", - unit: "percent" as const, - } - case "cpu_request": - return { - metricName: "k8s.pod.cpu_request_utilization", - unit: "percent" as const, - } - case "memory_limit": - return { - metricName: "k8s.pod.memory_limit_utilization", - unit: "percent" as const, - } - case "memory_request": - return { - metricName: "k8s.pod.memory_request_utilization", - unit: "percent" as const, - } - } - })() - - const compiled = CH.compile( - CH.podGaugeTimeseriesQuery({ - podName: payload.podName, - namespace: payload.namespace, - metricName: spec.metricName, - }), - { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - bucketSeconds, - }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "podInfraTimeseries", - }), - "podInfraTimeseries query failed", - ) - const typedRows = rows + const spec = podMetricSpec(payload.metric) + const rows = yield* runQuery(Queries.podInfraTimeseries, tenant, payload) return new PodInfraTimeseriesResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ bucket: String(row.bucket), attributeValue: String(row.attributeValue ?? ""), value: Number(row.avgValue) || 0, @@ -2510,39 +2339,10 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("nodeInfraTimeseries", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const bucketSeconds = payload.bucketSeconds ?? 60 - - const spec = (() => { - switch (payload.metric) { - case "cpu_usage": - return { metricName: "k8s.node.cpu.usage", unit: "cores" as const } - case "uptime": - return { metricName: "k8s.node.uptime", unit: "seconds" as const } - } - })() - - const compiled = CH.compile( - CH.nodeGaugeTimeseriesQuery({ - nodeName: payload.nodeName, - metricName: spec.metricName, - }), - { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - bucketSeconds, - }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "nodeInfraTimeseries", - }), - "nodeInfraTimeseries query failed", - ) - const typedRows = rows + const spec = nodeMetricSpec(payload.metric) + const rows = yield* runQuery(Queries.nodeInfraTimeseries, tenant, payload) return new NodeInfraTimeseriesResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ bucket: String(row.bucket), attributeValue: String(row.attributeValue ?? ""), value: Number(row.avgValue) || 0, @@ -2594,50 +2394,10 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("workloadInfraTimeseries", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const bucketSeconds = payload.bucketSeconds ?? 60 - - const spec = (() => { - switch (payload.metric) { - case "cpu_usage": - return { metricName: "k8s.pod.cpu.usage", unit: "cores" as const } - case "cpu_limit": - return { - metricName: "k8s.pod.cpu_limit_utilization", - unit: "percent" as const, - } - case "memory_limit": - return { - metricName: "k8s.pod.memory_limit_utilization", - unit: "percent" as const, - } - } - })() - - const compiled = CH.compile( - CH.workloadGaugeTimeseriesQuery({ - kind: payload.kind, - workloadName: payload.workloadName, - namespace: payload.namespace, - metricName: spec.metricName, - groupByPod: payload.groupByPod, - }), - { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - bucketSeconds, - }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "workloadInfraTimeseries", - }), - "workloadInfraTimeseries query failed", - ) - const typedRows = rows + const spec = workloadMetricSpec(payload.metric) + const rows = yield* runQuery(Queries.workloadInfraTimeseries, tenant, payload) return new WorkloadInfraTimeseriesResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ bucket: String(row.bucket), attributeValue: String(row.attributeValue ?? ""), value: Number(row.avgValue) || 0, diff --git a/packages/query-engine/src/registry/query-def.ts b/packages/query-engine/src/registry/query-def.ts index bde8d2668..c6a5ddedf 100644 --- a/packages/query-engine/src/registry/query-def.ts +++ b/packages/query-engine/src/registry/query-def.ts @@ -66,8 +66,17 @@ export interface QueryDef { * * Be careful raising a TTL on anything live-tailing: a stale entry there is a * user-visible correctness bug that presents as "the dashboard is frozen". + * + * May be a function of the payload and the current time. `spanDetail` needs + * that: a trace that finished long ago is immutable and can be cached hard, + * while one still receiving spans must not be — so its TTL is computed from + * the requested end time against now. `nowMs` is supplied by the runner from + * the Effect `Clock` rather than read here, so it stays testable. */ - readonly cache: DirectRouteCachePolicyInput | undefined + readonly cache: + | DirectRouteCachePolicyInput + | undefined + | ((payload: Payload, nowMs: number) => DirectRouteCachePolicyInput | undefined) /** * Build the compiled query from the request payload. From ca6e872848f06b2cb265e32aa4cceb26e5b5ac51 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Wed, 5 Aug 2026 00:53:21 +0200 Subject: [PATCH 4/5] feat(query-engine): migrate composite handlers (33/61) serviceUsage, serviceDependenciesBundle and serviceDetailOverview. serviceDependenciesBundle is the first payoff of the unification rather than just a refactor of it: two of its three sub-queries are the SAME builders as the standalone serviceDependenciesForService and serviceDbEdgesForService handlers, so it now reuses their defs instead of re-deriving the compile inline. That is one definition where there were two, which is the drift this registry exists to stop. Sub-queries of bundle endpoints keep their own ids -- an id is both the span context and the cache-key prefix, so folding them under the bundle name would merge unrelated cache entries. Their payload types are the MINIMAL input each needs, not the bundle's full payload. That is load-bearing rather than tidiness: runQuery keys the cache on whatever payload it is handed, and serviceEnvironments was hand-cached on {serviceName, startTime, endTime} rather than the whole request. Typing the def narrowly reproduces that key byte-for-byte instead of silently moving every entry. serviceUsage folds its with/without-previous-window branch into compile, so the id, profile and 60s TTL stay a single decision rather than being restated on both arms. Verified: apps/api typecheck; SQL baseline byte-identical (14 tests); apps/api routes + warehouse suites 310 pass. --- apps/api/src/routes/v1/query-engine.http.ts | 122 ++---------------- packages/query-engine/src/registry/queries.ts | 110 ++++++++++++++++ 2 files changed, 120 insertions(+), 112 deletions(-) diff --git a/apps/api/src/routes/v1/query-engine.http.ts b/apps/api/src/routes/v1/query-engine.http.ts index 8107700e7..058b69b37 100644 --- a/apps/api/src/routes/v1/query-engine.http.ts +++ b/apps/api/src/routes/v1/query-engine.http.ts @@ -1259,21 +1259,6 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("serviceDetailOverview", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - - const releasesCompiled = CH.compile( - CH.serviceReleasesTimelineQuery({ serviceName: payload.serviceName }), - { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - bucketSeconds: payload.releasesBucketSeconds ?? 300, - }, - ) - const environmentsCompiled = CH.compile( - CH.serviceEnvironmentsQuery({ serviceName: payload.serviceName }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - // One Worker invocation for the whole Overview tab: per-org config // resolves once (the first sub-query warms the in-isolate memo) and // the three queries run concurrently, replacing three separate @@ -1283,33 +1268,15 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", const [timeseries, releaseRows, environmentRows] = yield* Effect.all( [ queryEngine.execute(tenant, payload.timeseries), - mapExecError( - warehouse.compiledQuery(tenant, releasesCompiled, { - profile: "list", - context: "serviceReleases", - }), - "serviceReleases query failed", - ), - queryEngine.cachedDirect( - tenant, - "serviceEnvironments", - { - serviceName: payload.serviceName, - startTime: payload.startTime, - endTime: payload.endTime, - }, - mapExecError( - warehouse.compiledQuery(tenant, environmentsCompiled, { - profile: "discovery", - context: "serviceEnvironments", - }), - "serviceEnvironments query failed", - ), - ), + runQuery(Queries.serviceReleases, tenant, payload), + runQuery(Queries.serviceEnvironments, tenant, { + serviceName: payload.serviceName, + startTime: payload.startTime, + endTime: payload.endTime, + }), ], { concurrency: 3 }, ) - return new ServiceDetailOverviewResponse({ timeseries, releases: releaseRows.map((row) => ({ @@ -1327,56 +1294,17 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("serviceDependenciesBundle", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - - const dependenciesCompiled = CH.compile( - CH.serviceDependenciesForServiceQuery({ - serviceName: payload.serviceName, - deploymentEnv: payload.deploymentEnv, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const dbEdgesCompiled = CH.compile( - CH.serviceDbEdgesForServiceQuery({ - serviceName: payload.serviceName, - deploymentEnv: payload.deploymentEnv, - }), - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const externalEdgesCompiled = CH.serviceExternalEdgesSQL( - { deploymentEnv: payload.deploymentEnv, serviceName: payload.serviceName }, - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - // Dependencies tab in one Worker invocation: the three service-map // edge queries run concurrently and share a single config // resolution, replacing three independent round-trips. const [dependencyRows, dbEdgeRows, externalEdgeRows] = yield* Effect.all( [ - mapExecError( - warehouse.compiledQuery(tenant, dependenciesCompiled, { - profile: "aggregation", - context: "serviceDependenciesForService", - }), - "serviceDependenciesForService query failed", - ), - mapExecError( - warehouse.compiledQuery(tenant, dbEdgesCompiled, { - profile: "aggregation", - context: "serviceDbEdgesForService", - }), - "serviceDbEdgesForService query failed", - ), - mapExecError( - warehouse.compiledQuery(tenant, externalEdgesCompiled, { - profile: "aggregation", - context: "serviceExternalEdges", - }), - "serviceExternalEdges query failed", - ), + runQuery(Queries.serviceDependenciesForService, tenant, payload), + runQuery(Queries.serviceDbEdgesForService, tenant, payload), + runQuery(Queries.serviceExternalEdges, tenant, payload), ], { concurrency: 3 }, ) - return new ServiceDependenciesBundleResponse({ dependencies: dependencyRows.map((row) => ({ ...row })), dbEdges: dbEdgeRows.map((row) => ({ ...row })), @@ -1588,37 +1516,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("serviceUsage", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const prevStart = payload.previousStartTime - const prevEnd = payload.previousEndTime - const compiled = - prevStart != null && prevEnd != null - ? CH.compile(CH.serviceUsageWithPreviousQuery({ serviceName: payload.service }), { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - previousStartTime: prevStart, - previousEndTime: prevEnd, - }) - : CH.compile(CH.serviceUsageQuery({ serviceName: payload.service }), { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - }) - const rows = yield* queryEngine.cachedDirect( - tenant, - "serviceUsage", - payload, - mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "serviceUsage", - }), - "serviceUsage query failed", - ), - // Usage totals (GB / session counts) tolerate a minute of - // staleness; a 60s TTL cuts repeat-load recomputes ~4× vs 15s. - 60, - ) + const rows = yield* runQuery(Queries.serviceUsage, tenant, payload) return new ServiceUsageResponse({ data: rows }) }), ) diff --git a/packages/query-engine/src/registry/queries.ts b/packages/query-engine/src/registry/queries.ts index 4c01320cf..5bfe2c016 100644 --- a/packages/query-engine/src/registry/queries.ts +++ b/packages/query-engine/src/registry/queries.ts @@ -1,4 +1,5 @@ import type { + ServiceUsageRequest, ErrorDetailTracesRequest, ErrorRateByServiceRequest, ErrorsByTypeRequest, @@ -413,3 +414,112 @@ export const workloadDetailSummary = defineQuery({ { orgId, startTime: payload.startTime, endTime: payload.endTime }, ), }) + +// --- Sub-queries of composite (bundle) handlers --------------------------- +// +// Bundle endpoints run several queries in one Worker invocation so per-org +// config resolves once and the browser makes one round-trip instead of three. +// Each sub-query keeps its own id, because that id is both its span context and +// its cache-key prefix -- collapsing them under the bundle's name would merge +// unrelated cache entries. +// +// Their payload types are the MINIMAL input each needs rather than the bundle's +// full payload. That is load-bearing: `runQuery` keys the cache on whatever +// payload it is handed, so typing these narrowly reproduces the exact key the +// hand-written `cachedDirect` calls used. + +/** Release markers for the service detail overview chart. Uncached, mirroring the standalone path. */ +export const serviceReleases = defineQuery({ + id: "serviceReleases", + profile: "list", + cache: undefined, + compile: ( + payload: { + readonly serviceName: string + readonly startTime: string + readonly endTime: string + readonly releasesBucketSeconds?: number | undefined + }, + orgId: string, + ) => + CH.compile(CH.serviceReleasesTimelineQuery({ serviceName: payload.serviceName }), { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + bucketSeconds: payload.releasesBucketSeconds ?? 300, + }), +}) + +/** Environments a service reported in the window. Edge-cached on a service-scoped key. */ +export const serviceEnvironments = defineQuery({ + id: "serviceEnvironments", + profile: "discovery", + cache: 15, + compile: ( + payload: { readonly serviceName: string; readonly startTime: string; readonly endTime: string }, + orgId: string, + ) => + CH.compile(CH.serviceEnvironmentsQuery({ serviceName: payload.serviceName }), { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + }), +}) + +/** + * External (non-service) edges for the dependencies tab. + * + * Built by `serviceExternalEdgesSQL`, which returns a CompiledQuery directly + * rather than going through `CH.compile`. + */ +export const serviceExternalEdges = defineQuery({ + id: "serviceExternalEdges", + profile: "aggregation", + cache: undefined, + compile: ( + payload: { + readonly serviceName: string + readonly deploymentEnv?: string | undefined + readonly startTime: string + readonly endTime: string + }, + orgId: string, + ) => + CH.serviceExternalEdgesSQL( + { deploymentEnv: payload.deploymentEnv, serviceName: payload.serviceName }, + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) + +/** + * Service usage totals. + * + * Two builders behind one id: with both previous-window bounds the query + * returns current-vs-previous in a single scan, otherwise just the current + * window. The branch lives in `compile` so the id, profile and TTL stay one + * decision. + */ +export const serviceUsage = defineQuery({ + id: "serviceUsage", + profile: "aggregation", + // Usage totals (GB / session counts) tolerate a minute of staleness; a 60s + // TTL cuts repeat-load recomputes ~4x vs 15s. + cache: 60, + compile: (payload: ServiceUsageRequest, orgId: string) => { + const prevStart = payload.previousStartTime + const prevEnd = payload.previousEndTime + return prevStart != null && prevEnd != null + ? CH.compile(CH.serviceUsageWithPreviousQuery({ serviceName: payload.service }), { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + previousStartTime: prevStart, + previousEndTime: prevEnd, + }) + : CH.compile(CH.serviceUsageQuery({ serviceName: payload.service }), { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + }) + }, +}) From 42f29b8b41abc5661fc6d80dbef0f1eeacdcf952 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Wed, 5 Aug 2026 00:55:13 +0200 Subject: [PATCH 5/5] feat(query-engine): migrate the service-map edge handlers (37/61) serviceDependencies, serviceDbEdges, serviceExternalEdges and serviceWorkloads. All four use *SQL(opts, params) builders, which return a CompiledQuery directly rather than going through CH.compile. serviceWorkloads keeps its empty-service-list guard in the handler: that path returns without issuing a query at all, which a def cannot express. The rewrite anchors on the handler's `const compiled` rather than on its response constructor. Anchoring on the response silently clobbered serviceWorkloads, whose early-return response appears BEFORE its query. Verified: apps/api typecheck; SQL baseline byte-identical; routes + warehouse suites 310 pass. --- apps/api/src/routes/v1/query-engine.http.ts | 51 ++----------------- packages/query-engine/src/registry/queries.ts | 47 +++++++++++++++++ 2 files changed, 51 insertions(+), 47 deletions(-) diff --git a/apps/api/src/routes/v1/query-engine.http.ts b/apps/api/src/routes/v1/query-engine.http.ts index 058b69b37..6c430d6aa 100644 --- a/apps/api/src/routes/v1/query-engine.http.ts +++ b/apps/api/src/routes/v1/query-engine.http.ts @@ -401,17 +401,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("serviceDependencies", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.serviceDependenciesSQL( - { deploymentEnv: payload.deploymentEnv }, - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "serviceDependencies", - }), - "serviceDependencies query failed", - ) + const rows = yield* runQuery(Queries.serviceDependencies, tenant, payload) return new ServiceDependenciesResponse({ data: rows.map((row) => ({ ...row })) }) }), ) @@ -425,17 +415,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("serviceDbEdges", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.serviceDbEdgesSQL( - { deploymentEnv: payload.deploymentEnv }, - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "serviceDbEdges", - }), - "serviceDbEdges query failed", - ) + const rows = yield* runQuery(Queries.serviceDbEdges, tenant, payload) return new ServiceDbEdgesResponse({ data: rows.map((row) => ({ ...row })) }) }), ) @@ -1406,20 +1386,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("serviceExternalEdges", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const compiled = CH.serviceExternalEdgesSQL( - { - deploymentEnv: payload.deploymentEnv, - serviceName: payload.serviceName, - }, - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "serviceExternalEdges", - }), - "serviceExternalEdges query failed", - ) + const rows = yield* runQuery(Queries.serviceExternalEdges, tenant, payload) return new ServiceExternalEdgesResponse({ data: rows.map((row) => ({ ...row })) }) }), ) @@ -1482,17 +1449,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", if (payload.services.length === 0) { return new ServiceWorkloadsResponse({ data: [] }) } - const compiled = CH.serviceWorkloadsSQL( - { services: payload.services }, - { orgId: tenant.orgId, startTime: payload.startTime, endTime: payload.endTime }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "serviceWorkloads", - }), - "serviceWorkloads query failed", - ) + const rows = yield* runQuery(Queries.serviceWorkloads, tenant, payload) return new ServiceWorkloadsResponse({ data: rows.map((row) => ({ serviceName: decodeServiceName(String(row.serviceName ?? "")), diff --git a/packages/query-engine/src/registry/queries.ts b/packages/query-engine/src/registry/queries.ts index 5bfe2c016..ae46161c3 100644 --- a/packages/query-engine/src/registry/queries.ts +++ b/packages/query-engine/src/registry/queries.ts @@ -1,4 +1,7 @@ import type { + ServiceDbEdgesRequest, + ServiceDependenciesRequest, + ServiceWorkloadsRequest, ServiceUsageRequest, ErrorDetailTracesRequest, ErrorRateByServiceRequest, @@ -523,3 +526,47 @@ export const serviceUsage = defineQuery({ }) }, }) + +// --- Service-map edge queries --------------------------------------------- +// These use `*SQL(opts, params)` builders, which return a CompiledQuery +// directly instead of going through `CH.compile`. + +export const serviceDependencies = defineQuery({ + id: "serviceDependencies", + profile: "aggregation", + cache: undefined, + compile: (payload: ServiceDependenciesRequest, orgId: string) => + CH.serviceDependenciesSQL( + { deploymentEnv: payload.deploymentEnv }, + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) + +export const serviceDbEdges = defineQuery({ + id: "serviceDbEdges", + profile: "aggregation", + cache: undefined, + compile: (payload: ServiceDbEdgesRequest, orgId: string) => + CH.serviceDbEdgesSQL( + { deploymentEnv: payload.deploymentEnv }, + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +}) + +/** + * Workloads backing a set of services. + * + * The caller must skip this entirely for an empty service list — that guard + * stays in the handler because it avoids issuing a query at all, which a def + * cannot express. + */ +export const serviceWorkloads = defineQuery({ + id: "serviceWorkloads", + profile: "aggregation", + cache: undefined, + compile: (payload: ServiceWorkloadsRequest, orgId: string) => + CH.serviceWorkloadsSQL( + { services: payload.services }, + { orgId, startTime: payload.startTime, endTime: payload.endTime }, + ), +})