From 8deea1f7ee2fd47a89de5608c61b59e56047557e Mon Sep 17 00:00:00 2001 From: Makisuo Date: Wed, 5 Aug 2026 01:29:42 +0200 Subject: [PATCH 1/3] feat(query-engine): migrate spanHierarchy (55/61) Three defs: the recent-window probe, the unbounded fallback probe, and the pruned hierarchy read. All three carry cache: undefined on purpose. The handler keeps a single cachedDirect around the whole probe-then-read sequence, 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 -- and the probe exists precisely because trace_detail_spans is partitioned by toDate(Timestamp), where an unpruned hierarchy read is p95 ~8.8s against ~2.3s pruned. Adds a `withDeps` helper. cachedDirect takes an Effect with no requirements, while runQuery reads WarehouseQueryService and QueryEngineService from context; withDeps supplies the instances already bound at the top of the group so a registry query can run INSIDE a cache wrapper. Without it the only options were hoisting the probe out of the cache (losing the miss-only property) or threading services through every call site. Verified: apps/api typecheck; SQL baseline byte-identical; routes + warehouse suites 310 pass. --- apps/api/src/routes/v1/query-engine.http.ts | 90 +++++++------------ packages/query-engine/src/registry/queries.ts | 60 +++++++++++++ 2 files changed, 94 insertions(+), 56 deletions(-) diff --git a/apps/api/src/routes/v1/query-engine.http.ts b/apps/api/src/routes/v1/query-engine.http.ts index c1098dd48..4dec71935 100644 --- a/apps/api/src/routes/v1/query-engine.http.ts +++ b/apps/api/src/routes/v1/query-engine.http.ts @@ -152,6 +152,16 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", Effect.gen(function* () { const queryEngine = yield* QueryEngineService const warehouse = yield* WarehouseQueryService + + // `cachedDirect` takes an Effect with no requirements, but `runQuery` reads + // its services from context. Supplying the ones already bound above lets a + // registry query run INSIDE a cache wrapper — needed by spanHierarchy, + // where the probe must only fire on an outer cache miss. + const withDeps = (effect: Effect.Effect) => + effect.pipe( + Effect.provideService(WarehouseQueryService, warehouse), + Effect.provideService(QueryEngineService, queryEngine), + ) const serviceOperationsRollupEnabled = yield* Config.boolean( "SERVICE_OPERATIONS_ROLLUP_ENABLED", ).pipe(Config.withDefault(false)) @@ -182,65 +192,33 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", // shared link, AI link), resolve one via a cheap LIMIT-1 probe and // derive a ±1h window so the main query can prune. The probe itself // tries the recent window first (see PROBE_RECENT_WINDOW_MS). - Effect.gen(function* () { - 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)) - if (probe?.timestamp != null) { - const window = partitionWindowAround(probe.timestamp) - startTime = window.startTime - endTime = window.endTime + withDeps( + Effect.gen(function* () { + let startTime = payload.startTime + let endTime = payload.endTime + if (startTime == null || endTime == null) { + 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({ + return yield* runQuery(Queries.spanHierarchy, tenant, { 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", - ) - }), + startTime, + endTime, + }) + }), + ), traceCacheTtlSeconds(payload.endTime, nowMs), ) const typedRows = rows.map((row) => ({ 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 }, + ) + }, +}) From 301a4ad669317b14200bbf2ef5c3d836c579db4c Mon Sep 17 00:00:00 2001 From: Makisuo Date: Wed, 5 Aug 2026 01:32:03 +0200 Subject: [PATCH 2/3] refactor(query-engine): curry runner deps instead of laundering context withDeps was an anti-pattern. It re-provided WarehouseQueryService and QueryEngineService into an Effect that read them from context, using the very instances already bound as values three lines above -- laundering a requirement purely to satisfy cachedDirect's R = never. makeQueryRunners takes those services as values once per handler group and returns runQuery/runQueryFirst closed over them. Every call site is unchanged (`runQuery(def, tenant, payload)`), but the effects now carry R = never, so a registry query composes inside a cache wrapper with no ceremony. spanHierarchy's probe-inside-cachedDirect works for the right reason rather than by re-injection. Verified: apps/api typecheck; SQL baseline byte-identical; routes + warehouse suites 310 pass. --- apps/api/src/routes/query-helpers.ts | 48 ++++++ apps/api/src/routes/query-runner.ts | 168 ++++++++++---------- apps/api/src/routes/v1/query-engine.http.ts | 59 +++---- 3 files changed, 159 insertions(+), 116 deletions(-) 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 4dec71935..353909083 100644 --- a/apps/api/src/routes/v1/query-engine.http.ts +++ b/apps/api/src/routes/v1/query-engine.http.ts @@ -91,7 +91,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,16 +152,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 }) - // `cachedDirect` takes an Effect with no requirements, but `runQuery` reads - // its services from context. Supplying the ones already bound above lets a - // registry query run INSIDE a cache wrapper — needed by spanHierarchy, - // where the probe must only fire on an outer cache miss. - const withDeps = (effect: Effect.Effect) => - effect.pipe( - Effect.provideService(WarehouseQueryService, warehouse), - Effect.provideService(QueryEngineService, queryEngine), - ) const serviceOperationsRollupEnabled = yield* Config.boolean( "SERVICE_OPERATIONS_ROLLUP_ENABLED", ).pipe(Config.withDefault(false)) @@ -192,33 +184,28 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", // shared link, AI link), resolve one via a cheap LIMIT-1 probe and // derive a ±1h window so the main query can prune. The probe itself // tries the recent window first (see PROBE_RECENT_WINDOW_MS). - withDeps( - Effect.gen(function* () { - let startTime = payload.startTime - let endTime = payload.endTime - if (startTime == null || endTime == null) { - 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 - } + Effect.gen(function* () { + let startTime = payload.startTime + let endTime = payload.endTime + if (startTime == null || endTime == null) { + 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 } - return yield* runQuery(Queries.spanHierarchy, tenant, { - traceId: payload.traceId, - spanId: payload.spanId, - startTime, - endTime, - }) - }), - ), + } + return yield* runQuery(Queries.spanHierarchy, tenant, { + traceId: payload.traceId, + spanId: payload.spanId, + startTime, + endTime, + }) + }), traceCacheTtlSeconds(payload.endTime, nowMs), ) const typedRows = rows.map((row) => ({ From d5987c4243a649f789005abe8cb5ea2630df886b Mon Sep 17 00:00:00 2001 From: Makisuo Date: Wed, 5 Aug 2026 01:34:12 +0200 Subject: [PATCH 3/3] feat(query-engine): migrate hostInfraTimeseries (56/61) Network reads a counter family, everything else a gauge family, so they are two 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. hostMetricSpec joins the pod/node/workload specs in query-helpers: the def needs metricName and groupByAttributeKey, the handler needs unit and isNetwork. Verified: apps/api typecheck; SQL baseline byte-identical. Local test suites not re-run for this commit -- CI covers them. --- apps/api/src/routes/queries.ts | 47 ++++++++++++ apps/api/src/routes/v1/query-engine.http.ts | 85 ++------------------- 2 files changed, 53 insertions(+), 79 deletions(-) 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/v1/query-engine.http.ts b/apps/api/src/routes/v1/query-engine.http.ts index 353909083..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, @@ -1501,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, @@ -1570,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,