diff --git a/apps/api/src/routes/queries.ts b/apps/api/src/routes/queries.ts index c24e311b1..080bcd70f 100644 --- a/apps/api/src/routes/queries.ts +++ b/apps/api/src/routes/queries.ts @@ -3,6 +3,8 @@ 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 { + CloudflareInfraZoneBreakdownRequest, + HostInfraTimeseriesRequest, CloudflareInfraZoneFacetsRequest, CloudflareInfraZoneDetailRequest, ServicePlanetScaleStatsRequest, @@ -24,6 +26,7 @@ import type { WorkloadInfraTimeseriesRequest, } from "@maple/domain/http" import { + hostMetricSpec, nodeMetricSpec, partitionWindowAround, podMetricSpec, @@ -469,6 +472,133 @@ 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, + }, + ) + }, +}) + +// --- cloudflareInfraZoneBreakdown: three parallel, then one dependent ----- + +const zoneBreakdownParams = (payload: CloudflareInfraZoneBreakdownRequest, orgId: string) => ({ + orgId, + serviceName: payload.serviceName, + startTime: payload.startTime, + endTime: payload.endTime, +}) + +const cloudflareInfraZoneBreakdownTotals = defineQuery({ + id: "cloudflareInfraZoneBreakdownTotals", + profile: "aggregation", + cache: undefined, + compile: (payload: CloudflareInfraZoneBreakdownRequest, orgId: string) => + CH.compile( + Integrations.cloudflareZoneBreakdownTotalsSQL( + payload.dimension, + toCloudflareFilters(payload), + payload.limit ?? 100, + ), + zoneBreakdownParams(payload, orgId), + { rowSchema: Integrations.cloudflareZoneBreakdownTotalsRowSchema }, + ), +}) + +/** + * Coverage is deliberately UNFILTERED: it answers "what did the poller collect + * here", which the UI needs in order to say "not collected yet" rather than "no + * traffic" for a window that predates the dataset. Do not thread filters in. + */ +const cloudflareInfraZoneBreakdownCoverage = defineQuery({ + id: "cloudflareInfraZoneBreakdownCoverage", + profile: "aggregation", + cache: undefined, + compile: (payload: CloudflareInfraZoneBreakdownRequest, orgId: string) => + CH.compile( + Integrations.cloudflareZoneBreakdownCoverageSQL(payload.dimension), + zoneBreakdownParams(payload, orgId), + { rowSchema: Integrations.cloudflareZoneBreakdownCoverageRowSchema }, + ), +}) + +const cloudflareInfraZoneBreakdownZoneTotal = defineQuery({ + id: "cloudflareInfraZoneBreakdownZoneTotal", + profile: "aggregation", + cache: undefined, + compile: (payload: CloudflareInfraZoneBreakdownRequest, orgId: string) => + CH.compile( + Integrations.cloudflareZoneCountersSQL(toCloudflareFilters(payload)), + zoneBreakdownParams(payload, orgId), + { rowSchema: Integrations.cloudflareZoneCountersRowSchema }, + ), +}) + +/** + * The chart runs AFTER the totals rather than beside them: totals are already + * ranked by requests, so they name the series worth plotting. Without that the + * grouping is unbounded — a zone taking scanner traffic returns a distinct path + * per probe, and the response grows to buckets x thousands of keys. One extra + * round trip over the same warm scan buys a payload that can't blow up. + * + * `topKeys` therefore rides in the PAYLOAD rather than being derived inside + * `compile`: it is the output of a previous query, which a def has no way to + * see. The caller must also skip this entirely when `topKeys` is empty. + */ +const cloudflareInfraZoneBreakdownTimeseries = defineQuery({ + id: "cloudflareInfraZoneBreakdownTimeseries", + profile: "aggregation", + cache: undefined, + compile: ( + payload: CloudflareInfraZoneBreakdownRequest & { readonly topKeys: ReadonlyArray }, + orgId: string, + ) => + CH.compile( + Integrations.cloudflareZoneBreakdownTimeseriesSQL( + payload.dimension, + toCloudflareFilters(payload), + payload.topKeys, + ), + { ...zoneBreakdownParams(payload, orgId), bucketSeconds: payload.bucketSeconds }, + { rowSchema: Integrations.cloudflareZoneBreakdownTimeseriesRowSchema }, + ), +}) + export const Queries = { ...Core, @@ -651,4 +781,10 @@ export const Queries = { planetscaleServiceConnections, planetscaleServiceStorage, cloudflareInfraZoneFacets, + hostInfraNetworkTimeseries, + hostInfraGaugeTimeseries, + cloudflareInfraZoneBreakdownTotals, + cloudflareInfraZoneBreakdownCoverage, + cloudflareInfraZoneBreakdownZoneTotal, + cloudflareInfraZoneBreakdownTimeseries, } 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..edc7d60b5 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), ) @@ -550,23 +516,6 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", .handle("planetscaleInfraTimeseries", ({ payload }) => Effect.gen(function* () { const tenant = yield* CurrentTenant.Context - const base = { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - bucketSeconds: Math.max(60, Math.floor(payload.bucketSeconds)), - database: payload.database, - } - const compiled = - payload.branch === undefined - ? CH.compile(Integrations.planetscaleInfraTimeseriesSQL(), base, { - rowSchema: Integrations.planetscaleInfraTimeseriesRowSchema, - }) - : CH.compile( - Integrations.planetscaleBranchInfraTimeseriesSQL(), - { ...base, branch: payload.branch }, - { rowSchema: Integrations.planetscaleInfraTimeseriesRowSchema }, - ) const rows = yield* runQuery(Queries.planetscaleInfraTimeseries, tenant, payload) return new PlanetScaleInfraTimeseriesResponse({ data: rows.map((row) => ({ ...row })) }) }), @@ -718,70 +667,14 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", Effect.gen(function* () { const tenant = yield* CurrentTenant.Context const filters = toCloudflareFilters(payload) - const params = { - orgId: tenant.orgId, - serviceName: payload.serviceName, - startTime: payload.startTime, - endTime: payload.endTime, - } - const totalsCompiled = CH.compile( - Integrations.cloudflareZoneBreakdownTotalsSQL( - payload.dimension, - filters, - payload.limit ?? 100, - ), - params, - { rowSchema: Integrations.cloudflareZoneBreakdownTotalsRowSchema }, - ) - // Coverage is deliberately unfiltered: it answers "what did the poller collect - // here", which the UI needs in order to say "not collected yet" rather than - // "no traffic" for a window that predates the dataset. - const coverageCompiled = CH.compile( - Integrations.cloudflareZoneBreakdownCoverageSQL(payload.dimension), - params, - { rowSchema: Integrations.cloudflareZoneBreakdownCoverageRowSchema }, - ) - const zoneTotalCompiled = CH.compile( - Integrations.cloudflareZoneCountersSQL(filters), - params, - { - rowSchema: Integrations.cloudflareZoneCountersRowSchema, - }, - ) const [totalRows, coverageRows, zoneRows] = yield* Effect.all( [ - mapExecError( - warehouse.compiledQuery(tenant, totalsCompiled, { - profile: "aggregation", - context: "cloudflareInfraZoneBreakdownTotals", - }), - "cloudflareInfraZoneBreakdownTotals query failed", - ), - mapExecError( - warehouse.compiledQuery(tenant, coverageCompiled, { - profile: "aggregation", - context: "cloudflareInfraZoneBreakdownCoverage", - }), - "cloudflareInfraZoneBreakdownCoverage query failed", - ), - mapExecError( - warehouse.compiledQuery(tenant, zoneTotalCompiled, { - profile: "aggregation", - context: "cloudflareInfraZoneBreakdownZoneTotal", - }), - "cloudflareInfraZoneBreakdownZoneTotal query failed", - ), + runQuery(Queries.cloudflareInfraZoneBreakdownTotals, tenant, payload), + runQuery(Queries.cloudflareInfraZoneBreakdownCoverage, tenant, payload), + runQuery(Queries.cloudflareInfraZoneBreakdownZoneTotal, tenant, payload), ], { concurrency: 3 }, ) - // The chart runs after the totals rather than beside them: totals are already - // ranked by requests, so they name the series worth plotting. Without that the - // grouping is unbounded — a zone taking scanner traffic returns a distinct path - // per probe, and the response grows to buckets × thousands of keys. One extra - // round trip over the same warm scan buys a payload that can't blow up. - // The poller's own tail bucket is dropped from the picks, not plotted as a peer — - // it means the same thing as the fold, so it merges into it and leaves the slot - // for a real key. const topKeys = totalRows .filter((row) => row.key !== Integrations.CLOUDFLARE_BREAKDOWN_OTHER_KEY) .slice(0, Integrations.CLOUDFLARE_BREAKDOWN_SERIES_LIMIT) @@ -789,28 +682,10 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", const bucketRows: ReadonlyArray = topKeys.length === 0 ? [] - : yield* mapExecError( - warehouse.compiledQuery( - tenant, - CH.compile( - Integrations.cloudflareZoneBreakdownTimeseriesSQL( - payload.dimension, - filters, - topKeys, - ), - { ...params, bucketSeconds: payload.bucketSeconds }, - { - rowSchema: - Integrations.cloudflareZoneBreakdownTimeseriesRowSchema, - }, - ), - { - profile: "aggregation", - context: "cloudflareInfraZoneBreakdownTimeseries", - }, - ), - "cloudflareInfraZoneBreakdownTimeseries query failed", - ) + : yield* runQuery(Queries.cloudflareInfraZoneBreakdownTimeseries, tenant, { + ...payload, + topKeys, + }) const coverage = coverageRows[0] const zoneRequests = zoneRows.find((row) => row.serviceName === payload.serviceName) // Breakdown metrics are a per-window top-N fold of what Cloudflare returned, so @@ -1140,11 +1015,6 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", "serviceOperations", payload, Effect.gen(function* () { - const params = { - orgId: tenant.orgId, - startTime: payload.startTime, - endTime: payload.endTime, - } yield* Effect.annotateCurrentSpan( "query.rollup.enabled", serviceOperationsRollupEnabled, @@ -1152,50 +1022,27 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", // The rollout flag stays off until migration 0008 is deployed, // backfilled, and parity-checked. Disabling it restores the all-raw // rollback path without changing the endpoint contract. - const summaryOptions = { - serviceName: payload.serviceName, - environments: payload.environments, - limit: payload.limit, - } const runRawSummary = () => - warehouse.compiledQuery( - tenant, - CH.compile(CH.serviceOperationsSummaryRawQuery(summaryOptions), params, { - rowSchema: CH.serviceOperationsSummaryRowSchema, - }), - { profile: "aggregation", context: "serviceOperations" }, - ) + runQuery(Queries.serviceOperationsSummaryRaw, tenant, payload) let useRollup = serviceOperationsRollupEnabled const summaryEffect = useRollup - ? warehouse - .compiledQuery( - tenant, - CH.compile( - CH.serviceOperationsSummaryQuery(summaryOptions), - params, - { - rowSchema: CH.serviceOperationsSummaryRowSchema, - }, - ), - { profile: "aggregation", context: "serviceOperations" }, - ) - .pipe( - Effect.catch((error) => { - if (!isMissingServiceOperationsRollup(error)) - return Effect.fail(error) - useRollup = false - return Effect.gen(function* () { - yield* Effect.logWarning( - "Service operations rollup is unavailable; using raw rollback path", - ).pipe(Effect.annotateLogs({ orgId: tenant.orgId })) - yield* Effect.annotateCurrentSpan( - "query.rollup.fallback", - true, - ) - return yield* runRawSummary() - }) - }), - ) + ? runQuery(Queries.serviceOperationsSummary, tenant, payload).pipe( + Effect.catch((error) => { + if (!isMissingServiceOperationsRollup(error)) + return Effect.fail(error) + useRollup = false + return Effect.gen(function* () { + yield* Effect.logWarning( + "Service operations rollup is unavailable; using raw rollback path", + ).pipe(Effect.annotateLogs({ orgId: tenant.orgId })) + yield* Effect.annotateCurrentSpan( + "query.rollup.fallback", + true, + ) + return yield* runRawSummary() + }) + }), + ) : runRawSummary() const summaryRows = yield* mapExecError( summaryEffect, @@ -1216,50 +1063,23 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleApi, "queryEngine", ) const requestedBucketSeconds = payload.bucketSeconds ?? windowSeconds / 50 const bucketSeconds = Math.max(1, Math.round(requestedBucketSeconds / 60)) * 60 - const timeseriesOptions = { - serviceName: payload.serviceName, - environments: payload.environments, - spanNames, - bucketSeconds, - } - const timeseriesParams = { ...params, bucketSeconds } + const timeseriesInput = { ...payload, spanNames, bucketSeconds } const runRawTimeseries = () => - warehouse.compiledQuery( - tenant, - CH.compile( - CH.serviceOperationsTimeseriesRawQuery(timeseriesOptions), - timeseriesParams, - { rowSchema: CH.serviceOperationsTimeseriesRowSchema }, - ), - { profile: "aggregation", context: "serviceOperationsTimeseries" }, - ) + runQuery(Queries.serviceOperationsTimeseriesRaw, tenant, timeseriesInput) const timeseriesEffect = useRollup - ? warehouse - .compiledQuery( - tenant, - CH.compile( - CH.serviceOperationsTimeseriesQuery(timeseriesOptions), - timeseriesParams, - { rowSchema: CH.serviceOperationsTimeseriesRowSchema }, - ), - { - profile: "aggregation", - context: "serviceOperationsTimeseries", - }, - ) - .pipe( - Effect.catch((error) => - isMissingServiceOperationsRollup(error) - ? Effect.gen(function* () { - yield* Effect.annotateCurrentSpan( - "query.rollup.fallback", - true, - ) - return yield* runRawTimeseries() - }) - : Effect.fail(error), - ), - ) + ? runQuery(Queries.serviceOperationsTimeseries, tenant, timeseriesInput).pipe( + Effect.catch((error) => + isMissingServiceOperationsRollup(error) + ? Effect.gen(function* () { + yield* Effect.annotateCurrentSpan( + "query.rollup.fallback", + true, + ) + return yield* runRawTimeseries() + }) + : Effect.fail(error), + ), + ) : runRawTimeseries() const timeseriesRows = yield* mapExecError( timeseriesEffect, @@ -1536,66 +1356,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 +1371,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/apps/web/src/hooks/use-widget-data.ts b/apps/web/src/hooks/use-widget-data.ts index 8addacf1d..227ae5d91 100644 --- a/apps/web/src/hooks/use-widget-data.ts +++ b/apps/web/src/hooks/use-widget-data.ts @@ -23,7 +23,8 @@ export type WidgetDataSourceLike = { import { disabledResultAtom } from "@/lib/services/atoms/disabled-result-atom" import { MapleApiAtomClient } from "@/lib/services/common/atom-client" import type { WidgetDataState } from "@/components/dashboard-builder/types" -import { encodeKey } from "@/lib/cache-key" +import { encodeKey, encodeOrgScopedKey, orgScopedKeyPayload } from "@/lib/cache-key" +import { getActiveOrgId } from "@/lib/services/common/auth-headers" import { formatBackendError } from "@/lib/error-messages" import { Cause, Option } from "effect" import { WarehouseDecodeError, type BackendError } from "@/api/warehouse/effect-utils" @@ -318,8 +319,9 @@ const toWidgetDataAtomError = (error: unknown): WidgetFetchError => { const fetchWidgetData = Effect.fnUntraced( function* (key: string) { const parsed = yield* Effect.try({ + // The key is org-scoped; only the payload after the separator is JSON. try: () => - JSON.parse(key) as { + JSON.parse(orgScopedKeyPayload(key)) as { endpoint: string params: Record }, @@ -355,7 +357,7 @@ const widgetFetchFamily = Atom.family((key: string) => ) const widgetFetchAtom = (input: { endpoint: string; params: Record }) => - widgetFetchFamily(encodeKey(input)) + widgetFetchFamily(encodeOrgScopedKey(getActiveOrgId(), input)) /** * Fetches and transforms data for a single data source. Powers both whole diff --git a/apps/web/src/lib/cache-key.ts b/apps/web/src/lib/cache-key.ts index 2518f9b58..4b31bcea2 100644 --- a/apps/web/src/lib/cache-key.ts +++ b/apps/web/src/lib/cache-key.ts @@ -26,3 +26,31 @@ export function encodeKey(value: unknown): string { const normalized = normalizeForKey(value) return JSON.stringify(normalized === undefined ? null : normalized) } + +/** + * Separator between the active org and the encoded input in an atom family key. + * + * Atom keys used to carry only filters and time range. The org rode along + * invisibly in the auth header, so switching orgs left every cached entry + * addressable by the new org: the UI re-rendered the previous org's rows until + * the idle TTL expired — up to 30 minutes on some atoms. Electric collections + * never had this bug because their ids already embed the org. + * + * The org is prefixed rather than folded into the encoded object because atoms + * decode the key back into the query input; an extra field there would travel + * to the server as part of the request payload. + * + * NUL cannot occur in `encodeKey` output (it is JSON) or in a Clerk org id, so + * the first occurrence always marks the boundary. + */ +const ORG_KEY_SEPARATOR = "\u0000" + +/** Build an org-scoped atom family key. */ +export function encodeOrgScopedKey(orgId: string | null | undefined, value: unknown): string { + return `${orgId ?? ""}${ORG_KEY_SEPARATOR}${encodeKey(value)}` +} + +/** Recover just the encoded input from a key built by `encodeOrgScopedKey`. */ +export function orgScopedKeyPayload(key: string): string { + return key.slice(key.indexOf(ORG_KEY_SEPARATOR) + 1) +} diff --git a/apps/web/src/lib/services/atoms/warehouse-query-atoms.ts b/apps/web/src/lib/services/atoms/warehouse-query-atoms.ts index 30b34c0b2..d2269e938 100644 --- a/apps/web/src/lib/services/atoms/warehouse-query-atoms.ts +++ b/apps/web/src/lib/services/atoms/warehouse-query-atoms.ts @@ -1,6 +1,7 @@ import { Atom } from "@/lib/effect-atom" import { Effect, Schema } from "effect" -import { encodeKey } from "@/lib/cache-key" +import { encodeOrgScopedKey, orgScopedKeyPayload } from "@/lib/cache-key" +import { getActiveOrgId } from "@/lib/services/common/auth-headers" import { MapleApiAtomClient } from "@/lib/services/common/atom-client" import type { BackendError, WarehouseApiError } from "@/api/warehouse/effect-utils" import { @@ -161,7 +162,7 @@ function makeQueryAtomFamily(query: QueryEffect, o // shipped). The inner query spans already export by re-providing this same // (memoized) layer; this lifts the parent onto the same tracer. let resultAtom = MapleApiAtomClient.runtime.atom( - Schema.decodeUnknownEffect(UnknownFromJson)(key).pipe( + Schema.decodeUnknownEffect(UnknownFromJson)(orgScopedKeyPayload(key)).pipe( Effect.flatMap((input) => query(input as Input)), Effect.mapError(toQueryAtomError), ), @@ -174,7 +175,7 @@ function makeQueryAtomFamily(query: QueryEffect, o return resultAtom }) - return (input: Input) => family(encodeKey(input)) + return (input: Input) => family(encodeOrgScopedKey(getActiveOrgId(), input)) } export const getServiceUsageResultAtom = makeQueryAtomFamily(getServiceUsage, { diff --git a/packages/query-engine/src/registry/queries.ts b/packages/query-engine/src/registry/queries.ts index 51aefe969..761f1bd65 100644 --- a/packages/query-engine/src/registry/queries.ts +++ b/packages/query-engine/src/registry/queries.ts @@ -1,4 +1,5 @@ import type { + ServiceOperationsRequest, ListPodsRequest, NodeFacetsRequest, PodFacetsRequest, @@ -749,3 +750,156 @@ 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 }, + ) + }, +}) + +// --- serviceOperations: rollup and raw variants --------------------------- +// +// Four defs, two logical queries. Each has a rollup form and a raw form, and +// the CHOICE between them stays in the handler because it is policy, not query +// construction: a feature flag selects the rollup, and a typed +// `isMissingServiceOperationsRollup` error falls back to raw at runtime, +// flipping a flag that the timeseries query then honors too. +// +// Rollup/raw pairs share an id, matching the context their spans already +// report — the fallback is recorded separately as `query.rollup.fallback`. +// +// All four are `cache: undefined`; the handler wraps the whole sequence in one +// `cachedDirect`. + +const serviceOperationsSummaryOptions = (payload: ServiceOperationsRequest) => ({ + serviceName: payload.serviceName, + environments: payload.environments, + limit: payload.limit, +}) + +const serviceOperationsParams = (payload: ServiceOperationsRequest, orgId: string) => ({ + orgId, + startTime: payload.startTime, + endTime: payload.endTime, +}) + +export const serviceOperationsSummary = defineQuery({ + id: "serviceOperations", + profile: "aggregation", + cache: undefined, + compile: (payload: ServiceOperationsRequest, orgId: string) => + CH.compile( + CH.serviceOperationsSummaryQuery(serviceOperationsSummaryOptions(payload)), + serviceOperationsParams(payload, orgId), + { rowSchema: CH.serviceOperationsSummaryRowSchema }, + ), +}) + +/** Rollback path, used when the rollup table is absent or the flag is off. */ +export const serviceOperationsSummaryRaw = defineQuery({ + id: "serviceOperations", + profile: "aggregation", + cache: undefined, + compile: (payload: ServiceOperationsRequest, orgId: string) => + CH.compile( + CH.serviceOperationsSummaryRawQuery(serviceOperationsSummaryOptions(payload)), + serviceOperationsParams(payload, orgId), + { rowSchema: CH.serviceOperationsSummaryRowSchema }, + ), +}) + +/** + * `spanNames` and `bucketSeconds` ride in the payload: both are derived from the + * summary rows and the requested window, so `compile` cannot see them. The + * rollup is minute-grain, which is why the caller rounds the bucket to a whole + * minute before passing it here. + */ +type ServiceOperationsTimeseriesInput = ServiceOperationsRequest & { + readonly spanNames: ReadonlyArray + readonly bucketSeconds: number +} + +const serviceOperationsTimeseriesOptions = (payload: ServiceOperationsTimeseriesInput) => ({ + serviceName: payload.serviceName, + environments: payload.environments, + spanNames: payload.spanNames, + bucketSeconds: payload.bucketSeconds, +}) + +export const serviceOperationsTimeseries = defineQuery({ + id: "serviceOperationsTimeseries", + profile: "aggregation", + cache: undefined, + compile: (payload: ServiceOperationsTimeseriesInput, orgId: string) => + CH.compile( + CH.serviceOperationsTimeseriesQuery(serviceOperationsTimeseriesOptions(payload)), + { ...serviceOperationsParams(payload, orgId), bucketSeconds: payload.bucketSeconds }, + { rowSchema: CH.serviceOperationsTimeseriesRowSchema }, + ), +}) + +export const serviceOperationsTimeseriesRaw = defineQuery({ + id: "serviceOperationsTimeseries", + profile: "aggregation", + cache: undefined, + compile: (payload: ServiceOperationsTimeseriesInput, orgId: string) => + CH.compile( + CH.serviceOperationsTimeseriesRawQuery(serviceOperationsTimeseriesOptions(payload)), + { ...serviceOperationsParams(payload, orgId), bucketSeconds: payload.bucketSeconds }, + { rowSchema: CH.serviceOperationsTimeseriesRowSchema }, + ), +})