diff --git a/apps/api/src/routes/queries.ts b/apps/api/src/routes/queries.ts index c24e311b1..a84de8193 100644 --- a/apps/api/src/routes/queries.ts +++ b/apps/api/src/routes/queries.ts @@ -3,6 +3,7 @@ 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 { + HostInfraTimeseriesRequest, CloudflareInfraZoneFacetsRequest, CloudflareInfraZoneDetailRequest, ServicePlanetScaleStatsRequest, @@ -24,6 +25,7 @@ import type { WorkloadInfraTimeseriesRequest, } from "@maple/domain/http" import { + hostMetricSpec, nodeMetricSpec, partitionWindowAround, podMetricSpec, @@ -469,6 +471,49 @@ const cloudflareInfraZoneFacets = defineQuery({ }), }) +// --- hostInfraTimeseries: two query families behind one endpoint ---------- +// +// Network reads a counter family, everything else a gauge family, so they are +// separate defs rather than one def with a branch — the row shapes differ and +// the handler maps them differently. Both keep the id "hostInfraTimeseries", +// which is what their spans already report; renaming would break continuity of +// existing telemetry for no gain. + +const hostInfraNetworkTimeseries = defineQuery({ + id: "hostInfraTimeseries", + profile: "aggregation", + cache: undefined, + compile: (payload: HostInfraTimeseriesRequest, orgId: string) => + CH.compile(CH.hostNetworkTimeseriesQuery({ hostName: payload.hostName }), { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + bucketSeconds: payload.bucketSeconds ?? 60, + }), +}) + +const hostInfraGaugeTimeseries = defineQuery({ + id: "hostInfraTimeseries", + profile: "aggregation", + cache: undefined, + compile: (payload: HostInfraTimeseriesRequest, orgId: string) => { + const spec = hostMetricSpec(payload.metric) + return CH.compile( + CH.hostGaugeTimeseriesQuery({ + hostName: payload.hostName, + metricName: spec.metricName, + groupByAttributeKey: spec.groupByAttributeKey, + }), + { + orgId, + startTime: payload.startTime, + endTime: payload.endTime, + bucketSeconds: payload.bucketSeconds ?? 60, + }, + ) + }, +}) + export const Queries = { ...Core, @@ -651,4 +696,6 @@ export const Queries = { planetscaleServiceConnections, planetscaleServiceStorage, cloudflareInfraZoneFacets, + hostInfraNetworkTimeseries, + hostInfraGaugeTimeseries, } as const diff --git a/apps/api/src/routes/query-helpers.ts b/apps/api/src/routes/query-helpers.ts index c5278bde1..409075a40 100644 --- a/apps/api/src/routes/query-helpers.ts +++ b/apps/api/src/routes/query-helpers.ts @@ -1,6 +1,7 @@ import * as Integrations from "@maple/query-engine-integrations" import { formatWarehouseDateTime, parseWarehouseDateTime } from "@maple/query-engine" import type { + HostInfraTimeseriesRequest, NodeInfraTimeseriesRequest, PodInfraTimeseriesRequest, WorkloadInfraTimeseriesRequest, @@ -111,3 +112,50 @@ export const workloadMetricSpec = (metric: WorkloadInfraTimeseriesRequest["metri } } } + +/** + * Metric name, grouping key, unit and query-family flag for a host metric. + * + * Shared like the pod/node/workload specs: the registry needs `metricName` and + * `isNetwork` to build the query, the handler needs `unit` and + * `groupByAttributeKey` for its response. + */ +export const hostMetricSpec = (metric: HostInfraTimeseriesRequest["metric"]) => { + switch (metric) { + case "cpu": + return { + metricName: "system.cpu.utilization", + groupByAttributeKey: "state", + unit: "percent" as const, + isNetwork: false, + } + case "memory": + return { + metricName: "system.memory.utilization", + groupByAttributeKey: "state", + unit: "percent" as const, + isNetwork: false, + } + case "filesystem": + return { + metricName: "system.filesystem.utilization", + groupByAttributeKey: "mountpoint", + unit: "percent" as const, + isNetwork: false, + } + case "load15": + return { + metricName: "system.cpu.load_average.15m", + groupByAttributeKey: undefined, + unit: "load" as const, + isNetwork: false, + } + case "network": + return { + metricName: "system.network.io", + groupByAttributeKey: "direction", + unit: "bytes_per_second" as const, + isNetwork: true, + } + } +} diff --git a/apps/api/src/routes/query-runner.ts b/apps/api/src/routes/query-runner.ts index db7e788dc..a9c8170ae 100644 --- a/apps/api/src/routes/query-runner.ts +++ b/apps/api/src/routes/query-runner.ts @@ -2,8 +2,8 @@ import type { QueryDef } from "@maple/query-engine/registry" import type { QueryEngineDirectError } from "@maple/query-engine/runtime" 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" +import type { QueryEngineServiceShape } from "@/services/warehouse/QueryEngineService" +import type { WarehouseQueryServiceShape } from "@/services/warehouse/WarehouseQueryService" /** * Execute registry-declared warehouse queries. @@ -15,72 +15,81 @@ import { WarehouseQueryService } from "@/services/warehouse/WarehouseQueryServic * * Handlers keep their own row-to-response mapping. What they lose is the * plumbing, not the presentation. + * + * The services are taken as VALUES, not read from context, and the runners are + * built once per handler group. That is load-bearing rather than stylistic: + * `QueryEngineService.cachedDirect` accepts an `Effect` with no requirements, so + * a context-reading runner could never be composed inside a cache wrapper — + * which `spanHierarchy` needs, since its probe must only fire on a cache miss. + * Currying here keeps every call site written as `runQuery(def, tenant, payload)` + * while the effects it returns carry `R = never`. */ - -/** - * 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 } +export interface QueryRunnerDeps { + readonly warehouse: WarehouseQueryServiceShape + readonly queryEngine: QueryEngineServiceShape } -/** - * 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 - // 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 - // disagree with the span context beside it. - Effect.tapError(() => - Effect.annotateCurrentSpan({ - "maple.query_engine.failed_step": `${def.id} query failed`, - }), - ), - ) - if (cache === undefined) { - return yield* labelled - } - return yield* queryEngine.cachedDirect(tenant, def.id, payload, labelled, cache) - }) +export const makeQueryRunners = ({ warehouse, queryEngine }: QueryRunnerDeps) => { + /** + * Settings may be static or payload-dependent. Resolved to a spread so an + * absent result omits the key entirely rather than passing an explicit + * `settings: undefined`, which would read as "clear the profile defaults". + */ + const resolveSettings = (def: QueryDef, payload: Payload) => { + const settings = typeof def.settings === "function" ? def.settings(payload) : def.settings + return settings === undefined ? {} : { settings } + } -/** - * 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( + /** + * 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* () { + // 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 disagree with the span context beside it. + Effect.tapError(() => + Effect.annotateCurrentSpan({ + "maple.query_engine.failed_step": `${def.id} query failed`, + }), + ), + ) + if (cache === undefined) { + return yield* labelled + } + return yield* queryEngine.cachedDirect(tenant, def.id, payload, labelled, 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. + */ + const runQuery = ( + def: QueryDef, + tenant: TenantContext, + payload: Payload, + ) => + withPolicy( def, tenant, payload, @@ -90,22 +99,19 @@ export const runQuery = ( 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( + /** + * 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. + */ + const runQueryFirst = ( + def: QueryDef, + tenant: TenantContext, + payload: Payload, + ) => + withPolicy( def, tenant, payload, @@ -117,4 +123,6 @@ export const runQueryFirst = ( }) .pipe(Effect.map(Option.getOrNull)), ) - }) + + return { runQuery, runQueryFirst } as const +} diff --git a/apps/api/src/routes/v1/query-engine.http.ts b/apps/api/src/routes/v1/query-engine.http.ts index c1098dd48..6258bab74 100644 --- a/apps/api/src/routes/v1/query-engine.http.ts +++ b/apps/api/src/routes/v1/query-engine.http.ts @@ -84,6 +84,7 @@ import { } from "@maple/query-engine" import { LOGS_BODY_SEARCH_SETTINGS } from "@maple/query-engine/profiles" import { + hostMetricSpec, nodeMetricSpec, partitionWindowAround, podMetricSpec, @@ -91,7 +92,7 @@ import { workloadMetricSpec, } from "@/routes/query-helpers" import { Queries } from "@/routes/queries" -import { runQuery, runQueryFirst } from "@/routes/query-runner" +import { makeQueryRunners } 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" @@ -152,6 +153,8 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", Effect.gen(function* () { const queryEngine = yield* QueryEngineService const warehouse = yield* WarehouseQueryService + const { runQuery, runQueryFirst } = makeQueryRunners({ warehouse, queryEngine }) + const serviceOperationsRollupEnabled = yield* Config.boolean( "SERVICE_OPERATIONS_ROLLUP_ENABLED", ).pipe(Config.withDefault(false)) @@ -186,60 +189,23 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", let startTime = payload.startTime let endTime = payload.endTime if (startTime == null || endTime == null) { - const runProbe = (narrowByTime: boolean) => - mapExecError( - warehouse - .compiledQueryFirst( - tenant, - CH.compile( - CH.traceTimeProbeQuery({ - traceId: payload.traceId, - narrowByTime, - }), - narrowByTime - ? { - orgId: tenant.orgId, - startTime: formatWarehouseDateTime( - nowMs - PROBE_RECENT_WINDOW_MS, - ), - } - : { orgId: tenant.orgId }, - ), - { - profile: "discovery", - context: narrowByTime - ? "spanHierarchyProbeRecent" - : "spanHierarchyProbe", - }, - ) - .pipe(Effect.map(Option.getOrNull)), - "spanHierarchy probe failed", - ) - const probe = (yield* runProbe(true)) ?? (yield* runProbe(false)) + const probe = + (yield* runQueryFirst(Queries.spanHierarchyProbeRecent, tenant, { + traceId: payload.traceId, + startTime: formatWarehouseDateTime(nowMs - PROBE_RECENT_WINDOW_MS), + })) ?? (yield* runQueryFirst(Queries.spanHierarchyProbe, tenant, payload)) if (probe?.timestamp != null) { const window = partitionWindowAround(probe.timestamp) startTime = window.startTime endTime = window.endTime } } - const narrowByTime = startTime != null && endTime != null - const compiled = CH.compile( - CH.spanHierarchyQuery({ - traceId: payload.traceId, - spanId: payload.spanId, - narrowByTime, - }), - narrowByTime - ? { orgId: tenant.orgId, startTime, endTime } - : { orgId: tenant.orgId }, - ) - return yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "list", - context: "spanHierarchy", - }), - "spanHierarchy query failed", - ) + return yield* runQuery(Queries.spanHierarchy, tenant, { + traceId: payload.traceId, + spanId: payload.spanId, + startTime, + endTime, + }) }), traceCacheTtlSeconds(payload.endTime, nowMs), ) @@ -1536,66 +1502,12 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", const tenant = yield* CurrentTenant.Context const bucketSeconds = payload.bucketSeconds ?? 60 - const spec = (() => { - switch (payload.metric) { - case "cpu": - return { - metricName: "system.cpu.utilization", - groupByAttributeKey: "state", - unit: "percent" as const, - isNetwork: false, - } - case "memory": - return { - metricName: "system.memory.utilization", - groupByAttributeKey: "state", - unit: "percent" as const, - isNetwork: false, - } - case "filesystem": - return { - metricName: "system.filesystem.utilization", - groupByAttributeKey: "mountpoint", - unit: "percent" as const, - isNetwork: false, - } - case "load15": - return { - metricName: "system.cpu.load_average.15m", - groupByAttributeKey: undefined, - unit: "load" as const, - isNetwork: false, - } - case "network": - return { - metricName: "system.network.io", - groupByAttributeKey: "direction", - unit: "bytes_per_second" as const, - isNetwork: true, - } - } - })() + const spec = hostMetricSpec(payload.metric) if (spec.isNetwork) { - const compiled = CH.compile( - CH.hostNetworkTimeseriesQuery({ hostName: payload.hostName }), - { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - bucketSeconds, - }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "hostInfraTimeseries", - }), - "hostInfraTimeseries (network) query failed", - ) - const typedRows = rows + const rows = yield* runQuery(Queries.hostInfraNetworkTimeseries, tenant, payload) return new HostInfraTimeseriesResponse({ - data: typedRows.map((row) => ({ + data: rows.map((row) => ({ bucket: String(row.bucket), attributeValue: String(row.attributeValue ?? ""), value: Number(row.sumValue) || 0, @@ -1605,29 +1517,9 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", }) } - const compiled = CH.compile( - CH.hostGaugeTimeseriesQuery({ - hostName: payload.hostName, - metricName: spec.metricName, - groupByAttributeKey: spec.groupByAttributeKey, - }), - { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - bucketSeconds, - }, - ) - const rows = yield* mapExecError( - warehouse.compiledQuery(tenant, compiled, { - profile: "aggregation", - context: "hostInfraTimeseries", - }), - "hostInfraTimeseries query failed", - ) - const typedRows = rows + const rows = yield* runQuery(Queries.hostInfraGaugeTimeseries, tenant, payload) return new HostInfraTimeseriesResponse({ - 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/queries.ts b/packages/query-engine/src/registry/queries.ts index 51aefe969..a49236480 100644 --- a/packages/query-engine/src/registry/queries.ts +++ b/packages/query-engine/src/registry/queries.ts @@ -749,3 +749,63 @@ export const listPodsCount = defineQuery({ { rowSchema: CH.ListPodsSummaryOutputSchema }, ), }) + +// --- spanHierarchy: probe, then the pruned hierarchy read ----------------- +// +// `trace_detail_spans` is partitioned by toDate(Timestamp); without a time +// predicate the hierarchy query seeks every daily partition (~30) — p95 ~8.8s +// vs ~2.3s when pruned to one. When the caller has no timestamp (direct URL, +// shared link, AI link) a cheap LIMIT-1 probe resolves one. +// +// All three carry `cache: undefined` ON PURPOSE. The handler wraps the whole +// probe-then-read sequence in a single `cachedDirect`, so the probe only fires +// on an outer cache miss. Caching them individually would run the probe on +// every request and cache a result nobody asked for. + +/** Probe restricted to the recent window — tries ~2 daily partitions first. */ +export const spanHierarchyProbeRecent = defineQuery({ + id: "spanHierarchyProbeRecent", + profile: "discovery", + cache: undefined, + compile: (payload: { readonly traceId: string; readonly startTime: string }, orgId: string) => + CH.compile(CH.traceTimeProbeQuery({ traceId: payload.traceId, narrowByTime: true }), { + orgId, + startTime: payload.startTime, + }), +}) + +/** Unbounded fallback probe: every partition, only when the recent one missed. */ +export const spanHierarchyProbe = defineQuery({ + id: "spanHierarchyProbe", + profile: "discovery", + cache: undefined, + compile: (payload: { readonly traceId: string }, orgId: string) => + CH.compile(CH.traceTimeProbeQuery({ traceId: payload.traceId, narrowByTime: false }), { + orgId, + }), +}) + +export const spanHierarchy = defineQuery({ + id: "spanHierarchy", + profile: "list", + cache: undefined, + compile: ( + payload: { + readonly traceId: string + readonly spanId?: string | undefined + readonly startTime?: string | undefined + readonly endTime?: string | undefined + }, + orgId: string, + ) => { + const narrowByTime = payload.startTime != null && payload.endTime != null + return CH.compile( + CH.spanHierarchyQuery({ + traceId: payload.traceId, + spanId: payload.spanId, + narrowByTime, + }), + narrowByTime ? { orgId, startTime: payload.startTime, endTime: payload.endTime } : { orgId }, + ) + }, +})