diff --git a/apps/ai/src/mcp/__evals__/infra-world.ts b/apps/ai/src/mcp/__evals__/infra-world.ts index 02603e40a7..1bda90d12a 100644 --- a/apps/ai/src/mcp/__evals__/infra-world.ts +++ b/apps/ai/src/mcp/__evals__/infra-world.ts @@ -545,6 +545,17 @@ const workloadRows = (sql: string) => { avgCpuLimitPct: mean((s) => s.cpuLimitPct), avgMemoryLimitPct: mean((s) => s.memoryLimitPct), avgCpuUsage: mean((s) => s.cpuUsage), + peakCpuLimitPct: Math.max(...stats.map((s) => s.cpuLimitPctPeak)), + peakMemoryLimitPct: Math.max(...stats.map((s) => s.memoryLimitPctPeak)), + saturation: Math.max(...stats.map((s) => Math.max(s.cpuLimitPctPeak, s.memoryLimitPctPeak))), + cpuUsage: stats.reduce((sum, s) => sum + s.cpuUsage, 0), + desiredReplicas: pods.length, + readyReplicas: pods.length, + replicaSamples: 1, + shortfall: 0, + failedPods: 0, + restarts: 0, + lastTerminatedReason: "", } }) } diff --git a/apps/ai/src/mcp/lib/infra.ts b/apps/ai/src/mcp/lib/infra.ts index db92975094..fe373ae969 100644 --- a/apps/ai/src/mcp/lib/infra.ts +++ b/apps/ai/src/mcp/lib/infra.ts @@ -8,8 +8,8 @@ import { CH, formatWarehouseDateTime, parseWarehouseDateTime } from "@maple/quer import { WarehouseExecutor } from "@maple/query-engine/observability" import type { InfraEntityRow, InfraKind, InfraKindSection, InfraSeriesStats } from "@maple/domain/mcp-outputs" -export type WorkloadKind = "deployment" | "statefulset" | "daemonset" -export const WORKLOAD_KINDS: ReadonlyArray = ["deployment", "statefulset", "daemonset"] +export type WorkloadKind = "deployment" | "statefulset" | "daemonset" | "job" +export const WORKLOAD_KINDS: ReadonlyArray = ["deployment", "statefulset", "daemonset", "job"] export interface InfraWindow { readonly startTime: string @@ -225,11 +225,11 @@ const workloadRow = (kind: WorkloadKind, row: CH.ListWorkloadsOutput): InfraEnti workloadKind: kind, ...placement(row), podCount: row.podCount, - cpuCores: row.avgCpuUsage, - cpu: row.avgCpuLimitPct, - memory: row.avgMemoryLimitPct, - saturation: Math.max(row.avgCpuLimitPct, row.avgMemoryLimitPct), - ...(row.avgCpuLimitPct === 0 && row.avgMemoryLimitPct === 0 ? { unbounded: true } : undefined), + cpuCores: row.cpuUsage, + cpu: row.peakCpuLimitPct, + memory: row.peakMemoryLimitPct, + saturation: row.saturation, + ...(row.peakCpuLimitPct === 0 && row.peakMemoryLimitPct === 0 ? { unbounded: true } : undefined), }) const containerRow = (row: CH.ListContainersOutput): InfraEntityRow => ({ diff --git a/apps/api/src/routes/internal/query-engine.http.ts b/apps/api/src/routes/internal/query-engine.http.ts index 061b4cf070..7a71bf8860 100644 --- a/apps/api/src/routes/internal/query-engine.http.ts +++ b/apps/api/src/routes/internal/query-engine.http.ts @@ -1847,9 +1847,14 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleInternalApi, "query environment: row.environment, podCount: row.podCount || 0, lastSeen: DateTime.formatIso(row.lastSeen), - avgCpuLimitPct: row.avgCpuLimitPct || 0, - avgMemoryLimitPct: row.avgMemoryLimitPct || 0, - avgCpuUsage: row.avgCpuUsage || 0, + desiredReplicas: row.replicaSamples > 0 ? row.desiredReplicas : null, + readyReplicas: row.replicaSamples > 0 ? row.readyReplicas : null, + failedPods: row.failedPods || 0, + restarts: row.restarts || 0, + lastTerminatedReason: row.lastTerminatedReason, + peakCpuLimitPct: row.peakCpuLimitPct || 0, + peakMemoryLimitPct: row.peakMemoryLimitPct || 0, + cpuUsage: row.cpuUsage || 0, })), }) }), diff --git a/apps/web/src/api/warehouse/infra.ts b/apps/web/src/api/warehouse/infra.ts index fe939f42c1..b4ab5d2598 100644 --- a/apps/web/src/api/warehouse/infra.ts +++ b/apps/web/src/api/warehouse/infra.ts @@ -50,7 +50,7 @@ import { Effect } from "effect" import { MapleInternalAtomClient } from "@/lib/services/common/internal-atom-client" import { runWarehouseQuery } from "./effect-utils" -export type WorkloadKind = "deployment" | "statefulset" | "daemonset" +export type WorkloadKind = "deployment" | "statefulset" | "daemonset" | "job" /** Mirrors PodSortKeyLiteral in @maple/domain — `saturation` is peak-of-either-limit. */ export type PodSortKey = "saturation" | "cpuUsage" | "cpuLimitPct" | "memoryLimitPct" | "podName" | "lastSeen" diff --git a/apps/web/src/components/infra/workload-table.tsx b/apps/web/src/components/infra/workload-table.tsx index 356dc1172f..2308f69eb9 100644 --- a/apps/web/src/components/infra/workload-table.tsx +++ b/apps/web/src/components/infra/workload-table.tsx @@ -2,6 +2,8 @@ import { Link } from "@tanstack/react-router" import { TruncatedText } from "@maple/ui/components/ui/truncated-text" import { Skeleton } from "@maple/ui/components/ui/skeleton" +import { cn } from "@maple/ui/lib/utils" +import { TONE_TEXT, type Tone } from "@maple/ui/lib/tone" import type { ListWorkloadsResponse } from "@maple/domain/http" import type { WorkloadKind } from "@/api/warehouse/infra" @@ -9,6 +11,7 @@ import type { WorkloadKind } from "@/api/warehouse/infra" import { HostStatusBadge } from "./status-badge" import { MeterRows } from "./primitives/meter-rows" import { MetaLine } from "./primitives/meta-line" +import { restartTone } from "./pod-restarts" import { ColumnHead, DataTable, @@ -20,7 +23,13 @@ import { RelativeTime } from "@/components/common/relative-time" export type WorkloadRow = ListWorkloadsResponse["data"][number] -type SortKey = "workloadName" | "namespace" | "podCount" | "avgCpuLimitPct" | "avgMemoryLimitPct" | "lastSeen" +type SortKey = + | "workloadName" + | "podCount" + | "restarts" + | "peakCpuLimitPct" + | "peakMemoryLimitPct" + | "lastSeen" interface WorkloadTableProps { workloads: ReadonlyArray @@ -30,7 +39,7 @@ interface WorkloadTableProps { } /** Declared once and rendered by both the table and its skeleton so widths cannot drift. */ -function WorkloadColumns({ sort }: { sort?: SortControls }) { +function WorkloadColumns({ sort, kind }: { sort?: SortControls; kind?: WorkloadKind }) { return ( <> @@ -39,21 +48,29 @@ function WorkloadColumns({ sort }: { sort?: SortControls }) { {...sort} width="w-0 flex-1 min-w-[260px]" /> + label={kind === "job" ? "Completions" : "Ready"} width="w-[110px]" /> label="Pods" sortKey="podCount" {...sort} align="right" width="w-[60px]" /> - label="Avg CPU" - sortKey="avgCpuLimitPct" + label="Restarts" + sortKey="restarts" {...sort} align="right" - width="w-[160px]" + width="w-[140px]" + /> + + label="CPU peak" + sortKey="peakCpuLimitPct" + {...sort} + align="right" + width="w-[140px]" hidden="hidden md:flex" /> - label="Avg memory" - sortKey="avgMemoryLimitPct" + label="Memory peak" + sortKey="peakMemoryLimitPct" {...sort} align="right" - width="w-[160px]" + width="w-[140px]" hidden="hidden lg:flex" /> @@ -78,25 +95,59 @@ export function WorkloadTableLoading() { + - - + + + ) } +/** + * Ready against desired, from the cluster collector. A job counts completions, so a + * running job is not "short"; its failed pods are what reads as trouble. + */ +function ReplicaCell({ workload, kind }: { workload: WorkloadRow; kind: WorkloadKind }) { + const { desiredReplicas: desired, readyReplicas: ready, failedPods } = workload + if (desired === null || ready === null) { + return — + } + const tone: Tone = + kind === "job" + ? failedPods > 0 + ? "crit" + : "neutral" + : ready === 0 && desired > 0 + ? "crit" + : ready < desired + ? "warn" + : "neutral" + return ( + + {ready} / {desired} + {kind === "job" && failedPods > 0 && ` · ${failedPods} failed`} + + ) +} + export function WorkloadTable({ workloads, kind, waiting, referenceTime }: WorkloadTableProps) { + // No initial key: the server already sorts worst first (missing replicas, failures, restarts, saturation). const { sorted, sortKey, sortDir, handleSort } = useTableSort(workloads, { - initialKey: "avgCpuLimitPct", - stringKeys: ["workloadName", "namespace"], + initialKey: null, + stringKeys: ["workloadName"], + resettable: true, }) return ( - + {sorted.length === 0 && No workloads match your filter.} @@ -119,14 +170,28 @@ export function WorkloadTable({ workloads, kind, waiting, referenceTime }: Workl +
+ +
{wl.podCount.toLocaleString()}
-
- +
+ {wl.restarts.toLocaleString()} + {wl.restarts > 0 && wl.lastTerminatedReason && ( + {wl.lastTerminatedReason} + )} +
+
+
-
- +
+
= { deployment: "Deployment", statefulset: "StatefulSet", daemonset: "DaemonSet", + job: "Job", } satisfies Record function WorkloadDetailPage() { diff --git a/apps/web/src/routes/infra/kubernetes/workloads/index.tsx b/apps/web/src/routes/infra/kubernetes/workloads/index.tsx index 72c4af36fd..9b1c7c404a 100644 --- a/apps/web/src/routes/infra/kubernetes/workloads/index.tsx +++ b/apps/web/src/routes/infra/kubernetes/workloads/index.tsx @@ -34,10 +34,11 @@ import { sessionTimeRangeSearchMiddleware } from "@/components/time-range-picker import { TONE_FILL } from "@maple/ui/lib/tone" const DEFAULT_PRESET = "12h" -// The list query returns the most recently seen N workloads; the toolbar says so at the cap. +// The list query returns the N worst workloads (missing replicas, failures, restarts, saturation); +// the toolbar says so at the cap. const WORKLOAD_LIST_LIMIT = 200 -const WorkloadKindLiteral = Schema.Literals(["deployment", "statefulset", "daemonset"]) +const WorkloadKindLiteral = Schema.Literals(["deployment", "statefulset", "daemonset", "job"]) /** A one-click scope from the band. `undefined` means every workload. */ type WorkloadScope = "saturated" | "elevated" | "stale" @@ -68,17 +69,13 @@ const KIND_OPTIONS = [ { value: "deployment", label: "Deployments" }, { value: "statefulset", label: "StatefulSets" }, { value: "daemonset", label: "DaemonSets" }, + { value: "job", label: "Jobs" }, ] as const satisfies ReadonlyArray<{ value: WorkloadKind; label: string }> -/** - * The workload list is average-over-window, not peak: the query aggregates per - * workload and only carries the mean. So the band says "average", and a - * workload that spikes without moving its mean stays out of the red here — the - * pods view is where a spike shows. - */ +/** Saturation is the worst pod's peak, so one pod at its limit marks the whole workload. */ function scopeOf(workload: WorkloadRow, referenceTime: string): WorkloadScope | null { if (deriveHostStatus(workload.lastSeen, referenceTime) === "ended") return "stale" - const level = severityLevel(Math.max(workload.avgCpuLimitPct, workload.avgMemoryLimitPct)) + const level = severityLevel(Math.max(workload.peakCpuLimitPct, workload.peakMemoryLimitPct)) if (level === "crit") return "saturated" if (level === "warn") return "elevated" return null @@ -203,7 +200,7 @@ function WorkloadsPage() { } title={`No ${kindOption.label} reporting yet`} - description="Maple aggregates pod metrics by k8s.deployment.name, k8s.statefulset.name, and k8s.daemonset.name. If the Helm chart is not installed yet, install it so the k8sattributes processor can tag pod metrics with their workload. If it is, try another workload kind." + description="Maple aggregates pod metrics by k8s.deployment.name, k8s.statefulset.name, k8s.daemonset.name, and k8s.job.name. If the Helm chart is not installed yet, install it so the k8sattributes processor can tag pod metrics with their workload. If it is, try another workload kind." installTab="kubernetes" actionLabel="Install the Helm chart" docs="kubernetes" @@ -230,14 +227,14 @@ function WorkloadsPage() { { scope: "saturated", label: "Saturated", - hint: "avg ≥90%", + hint: "a pod ≥90%", value: saturated, tone: "crit", }, { scope: "elevated", label: "Elevated", - hint: "avg ≥60%", + hint: "a pod ≥60%", value: elevated, tone: "warn", }, @@ -264,7 +261,7 @@ function WorkloadsPage() { = WORKLOAD_LIST_LIMIT - ? `${formatNumber(filtered.length)} of the ${formatNumber(WORKLOAD_LIST_LIMIT)} most recently seen workloads` + ? `${formatNumber(filtered.length)} of the ${formatNumber(WORKLOAD_LIST_LIMIT)} least healthy workloads` : countLabel(filtered.length, filtered.length, "workload") } > diff --git a/packages/backend/src/services/warehouse/k8s-workloads.clickhouse.e2e.test.ts b/packages/backend/src/services/warehouse/k8s-workloads.clickhouse.e2e.test.ts new file mode 100644 index 0000000000..856ffd6f70 --- /dev/null +++ b/packages/backend/src/services/warehouse/k8s-workloads.clickhouse.e2e.test.ts @@ -0,0 +1,247 @@ +// SAFETY-FILE: JSON is emitted by the isolated ClickHouse fixture and decoded by the real query schema. +// The workload list joins three collectors' series (replica gauges, container restarts, +// pod usage) per workload. These seed each against the real schema and read the rows back. + +import { afterAll, assert, beforeAll, describe, it } from "@effect/vitest" +import { Schema } from "effect" +import { compileUnsafe } from "@maple-dev/effect-orm/clickhouse" +import { OrgId } from "@maple/domain" +import { listWorkloadsQuery, type ListWorkloadsOpts } from "../../../../query-engine/src/ch/queries/infra" +import { normalizeSqlForClickHouseClient } from "@maple/query-engine/execution" +import { + applyRealMigrations, + clickhouseE2eEnabled, + clickhouseExec, + uniqueDatabase, +} from "./clickhouse-e2e-support" + +const database = uniqueDatabase("maple_k8s_workloads_e2e") +const ORG = "org_k8s" + +const MINUTE_MS = 60_000 +// Anchored to now so the table TTL never drops the seed. +const END_MS = Math.floor(Date.now() / MINUTE_MS) * MINUTE_MS - 10 * MINUTE_MS +const START_MS = END_MS - 60 * MINUTE_MS +const chDateTime = (epochMs: number): string => new Date(epochMs).toISOString().replace("T", " ").slice(0, 19) +const minutesBeforeEnd = (minutes: number) => chDateTime(END_MS - minutes * MINUTE_MS) + +const quote = (value: string): string => `'${value.replace(/\\/g, "\\\\").replace(/'/g, "\\'")}'` + +interface Sample { + readonly org?: string + readonly metric: string + readonly minutesBeforeEnd: number + readonly value: number + readonly attrs: Record +} + +const cluster = { "k8s.cluster.name": "prod", "k8s.namespace.name": "payments" } +const deploy = (name: string) => ({ ...cluster, "k8s.deployment.name": name }) +const pod = (deployment: string, uid: string, extra: Record = {}) => ({ + ...deploy(deployment), + "k8s.pod.name": `${deployment}-${uid}`, + "k8s.pod.uid": uid, + ...extra, +}) +const fargatePod = (deployment: string, uid: string) => + pod(deployment, uid, { "eks.amazonaws.com/compute-type": "fargate" }) +const job = (name: string) => ({ ...cluster, "k8s.job.name": name }) + +const samples: ReadonlyArray = [ + // checkout-api: a rollout left 2 of 5 available, and two pods restart. + { metric: "k8s.deployment.desired", minutesBeforeEnd: 50, value: 5, attrs: deploy("checkout-api") }, + { metric: "k8s.deployment.desired", minutesBeforeEnd: 1, value: 5, attrs: deploy("checkout-api") }, + { metric: "k8s.deployment.available", minutesBeforeEnd: 50, value: 5, attrs: deploy("checkout-api") }, + { metric: "k8s.deployment.available", minutesBeforeEnd: 1, value: 2, attrs: deploy("checkout-api") }, + { metric: "k8s.pod.cpu.usage", minutesBeforeEnd: 40, value: 0.6, attrs: pod("checkout-api", "a") }, + { metric: "k8s.pod.cpu.usage", minutesBeforeEnd: 1, value: 0.2, attrs: pod("checkout-api", "a") }, + { + metric: "k8s.pod.memory_limit_utilization", + minutesBeforeEnd: 20, + value: 0.38, + attrs: pod("checkout-api", "a"), + }, + // Pod b stopped reporting half an hour ago: it counts as a pod, not as cores in use now. + { metric: "k8s.pod.cpu.usage", minutesBeforeEnd: 30, value: 0.9, attrs: pod("checkout-api", "b") }, + ...restarts("checkout-api", "a", "app", [0, 1, 4], "OOMKilled"), + ...restarts("checkout-api", "b", "app", [10, 12], ""), + ...restarts("checkout-api", "b", "sidecar", [5, 5], "Completed"), + // Errored 20 minutes before pod a's OOM kill, so the OOM kill is the workload's latest reason. + ...restarts("checkout-api", "b", "worker", [3, 8, 8], "Error"), + + // api: fully available. A Fargate pod near its CPU limit restarts twice; a node pod with + // no usage series restarts 7 times, which a Fargate filter must not count. + // The Fargate pod's Error is the more recent restart; the node pod's OOM kills came first. + ...restarts("api", "c", "app", [0, 0, 2], "Error", { computeType: "fargate" }), + ...restarts("api", "d", "app", [0, 7, 7], "OOMKilled"), + { metric: "k8s.deployment.desired", minutesBeforeEnd: 1, value: 3, attrs: deploy("api") }, + { metric: "k8s.deployment.available", minutesBeforeEnd: 1, value: 3, attrs: deploy("api") }, + { + metric: "k8s.pod.cpu_limit_utilization", + minutesBeforeEnd: 15, + value: 0.93, + attrs: fargatePod("api", "c"), + }, + { + metric: "k8s.pod.cpu_limit_utilization", + minutesBeforeEnd: 1, + value: 0.4, + attrs: fargatePod("api", "c"), + }, + { metric: "k8s.pod.cpu.usage", minutesBeforeEnd: 1, value: 0.5, attrs: fargatePod("api", "c") }, + + // orphan: wants 2 pods, none ever scheduled, so there is no pod usage at all. + { metric: "k8s.deployment.desired", minutesBeforeEnd: 1, value: 2, attrs: deploy("orphan") }, + { metric: "k8s.deployment.available", minutesBeforeEnd: 1, value: 0, attrs: deploy("orphan") }, + + // Another tenant's checkout-api must not leak into this org's row. + { + org: "org_other", + metric: "k8s.deployment.desired", + minutesBeforeEnd: 1, + value: 99, + attrs: deploy("checkout-api"), + }, + ...restarts("checkout-api", "z", "app", [0, 500], "Error", { org: "org_other" }), + + // Jobs: one finished cleanly, one keeps failing. + { metric: "k8s.job.desired_successful_pods", minutesBeforeEnd: 1, value: 1, attrs: job("nightly") }, + { metric: "k8s.job.successful_pods", minutesBeforeEnd: 1, value: 1, attrs: job("nightly") }, + { metric: "k8s.job.failed_pods", minutesBeforeEnd: 1, value: 0, attrs: job("nightly") }, + { metric: "k8s.job.desired_successful_pods", minutesBeforeEnd: 1, value: 1, attrs: job("broken") }, + { metric: "k8s.job.successful_pods", minutesBeforeEnd: 1, value: 0, attrs: job("broken") }, + { metric: "k8s.job.failed_pods", minutesBeforeEnd: 1, value: 2, attrs: job("broken") }, +] + +function restarts( + deployment: string, + uid: string, + container: string, + totals: ReadonlyArray, + reason: string, + { org, computeType }: { org?: string; computeType?: string } = {}, +): ReadonlyArray { + const baseAttrs = { + "k8s.container.name": container, + "k8s.container.status.last_terminated_reason": reason, + } + const containerAttrs = + computeType === undefined + ? baseAttrs + : { ...baseAttrs, "eks.amazonaws.com/compute-type": computeType } + return totals.map((value, index) => ({ + org, + metric: "k8s.container.restarts", + minutesBeforeEnd: 45 - index * 20, + value, + attrs: pod(deployment, uid, containerAttrs), + })) +} + +const mapLiteral = (attrs: Record) => + `map(${Object.entries(attrs) + .flatMap(([key, value]) => [quote(key), quote(value)]) + .join(", ")})` + +describe.skipIf(!clickhouseE2eEnabled)("listWorkloadsQuery against ClickHouse", () => { + beforeAll(async () => { + await clickhouseExec(`CREATE DATABASE ${database}`) + await applyRealMigrations(database) + const rows = samples + .map((sample) => { + const at = quote(minutesBeforeEnd(sample.minutesBeforeEnd)) + return `(${quote(sample.org ?? ORG)}, '', ${quote(sample.metric)}, ${at}, ${at}, ${sample.value}, ${mapLiteral(sample.attrs)})` + }) + .join(",\n") + await clickhouseExec( + `INSERT INTO metrics_gauge (OrgId, ServiceName, MetricName, StartTimeUnix, TimeUnix, Value, ResourceAttributes) VALUES\n${rows}`, + database, + ) + }, 120_000) + + afterAll(async () => { + await clickhouseExec(`DROP DATABASE IF EXISTS ${database}`) + }) + + const run = async (opts: ListWorkloadsOpts) => { + const compiled = compileUnsafe(listWorkloadsQuery(opts), { + orgId: OrgId.make(ORG), + startTime: chDateTime(START_MS), + endTime: chDateTime(END_MS), + }) + const body = JSON.parse( + await clickhouseExec(normalizeSqlForClickHouseClient(compiled.sql), database, { + default_format: "JSON", + output_format_json_quote_64bit_integers: "0", + }), + ) + assert.isDefined(compiled.rowSchema) + return Schema.decodeUnknownSync(Schema.Array(compiled.rowSchema!))(body.data) + } + + it("ranks missing replicas first, including a workload with no pods at all", async () => { + const rows = await run({ kind: "deployment" }) + assert.deepEqual( + rows.map((row) => row.workloadName), + ["checkout-api", "orphan", "api"], + ) + const orphan = rows[1]! + assert.strictEqual(orphan.podCount, 0) + assert.strictEqual(orphan.shortfall, 2) + }) + + it("reads the latest replica gauges and sums restarts per container, not per sample", async () => { + const [checkout] = await run({ kind: "deployment", workloadNames: ["checkout-api"] }) + assert.isDefined(checkout) + assert.strictEqual(checkout!.desiredReplicas, 5) + assert.strictEqual(checkout!.readyReplicas, 2) + assert.strictEqual(checkout!.shortfall, 3) + assert.isAbove(checkout!.replicaSamples, 0) + // 0 → 4 on pod a's app, 10 → 12 on b's app, 3 → 8 on b's worker, 5 → 5 on the sidecar; + // the other org's 500 is excluded. + assert.strictEqual(checkout!.restarts, 11) + // a's OOM kill is the latest restart; b's worker errored earlier. + assert.strictEqual(checkout!.lastTerminatedReason, "OOMKilled") + assert.strictEqual(checkout!.podCount, 2) + // Only pod a still reports at the window's end. + assert.closeTo(checkout!.cpuUsage, 0.2, 1e-9) + assert.closeTo(checkout!.peakMemoryLimitPct, 0.38, 1e-9) + }) + + it("reports the worst pod's peak, not where it ended", async () => { + const [api] = await run({ kind: "deployment", workloadNames: ["api"] }) + assert.closeTo(api!.peakCpuLimitPct, 0.93, 1e-9) + assert.closeTo(api!.saturation, 0.93, 1e-9) + assert.strictEqual(api!.shortfall, 0) + assert.strictEqual(api!.restarts, 9) + // c's Error is newer than d's OOM kills, whatever order the series are read in. + assert.strictEqual(api!.lastTerminatedReason, "Error") + }) + + it("lists jobs by failures, without calling a running job short of replicas", async () => { + const rows = await run({ kind: "job" }) + assert.deepEqual( + rows.map((row) => [row.workloadName, row.failedPods, row.readyReplicas, row.shortfall]), + [ + ["broken", 2, 0, 0], + ["nightly", 0, 1, 0], + ], + ) + }) + + it("lists only workloads with a matching pod when filtering by compute type, keeping their replica gauges", async () => { + const rows = await run({ kind: "deployment", computeTypes: ["fargate"] }) + // orphan has no pods and checkout-api's pods carry no compute type: neither may leak in. + // api's restarts count only its Fargate pod, not the node pod's OOM kills. + assert.deepEqual( + rows.map((row) => [ + row.workloadName, + row.desiredReplicas, + row.podCount, + row.restarts, + row.lastTerminatedReason, + ]), + [["api", 3, 1, 2, "Error"]], + ) + }) +}) diff --git a/packages/domain/src/http/query-engine.ts b/packages/domain/src/http/query-engine.ts index 04307a6057..4ace0d12d3 100644 --- a/packages/domain/src/http/query-engine.ts +++ b/packages/domain/src/http/query-engine.ts @@ -1305,7 +1305,7 @@ export class HostInfraTimeseriesResponse extends Schema.Class("ListWorkloadsResponse")({ diff --git a/packages/query-engine/src/ch/queries/infra.ts b/packages/query-engine/src/ch/queries/infra.ts index f395726096..e2226f79b4 100644 --- a/packages/query-engine/src/ch/queries/infra.ts +++ b/packages/query-engine/src/ch/queries/infra.ts @@ -342,7 +342,7 @@ export interface ListPodsOpts { excludedComputeTypes?: ReadonlyArray // Single-value filters retained for backward compat with the workload detail // page, which still narrows by a single workload owner. - workloadKind?: "deployment" | "statefulset" | "daemonset" + workloadKind?: WorkloadKind workloadName?: string /** Defaults to `saturation` — see PodSortKey. */ sortBy?: PodSortKey @@ -409,12 +409,14 @@ export interface ListPodsOutput { readonly saturation: number } -const workloadAttrKey = (kind: "deployment" | "statefulset" | "daemonset") => - kind === "deployment" - ? "k8s.deployment.name" - : kind === "statefulset" - ? "k8s.statefulset.name" - : "k8s.daemonset.name" +const WORKLOAD_ATTR_KEY = { + deployment: "k8s.deployment.name", + statefulset: "k8s.statefulset.name", + daemonset: "k8s.daemonset.name", + job: "k8s.job.name", +} as const satisfies Record + +const workloadAttrKey = (kind: WorkloadKind) => WORKLOAD_ATTR_KEY[kind] const podBaseConditions = ( $: ColumnAccessor, @@ -742,7 +744,7 @@ export function podDetailSummaryQuery(opts: PodDetailSummaryOpts) { export interface PodRestartsOpts { podNames?: ReadonlyArray namespace?: string - workloadKind?: "deployment" | "statefulset" | "daemonset" + workloadKind?: WorkloadKind workloadName?: string limit?: number } @@ -1033,10 +1035,10 @@ export function nodeGaugeTimeseriesQuery(opts: NodeGaugeTimeseriesOpts) { .format("JSON") } -// Kubernetes — workload aggregations (Deployment / StatefulSet / DaemonSet). +// Kubernetes — workload aggregations (Deployment / StatefulSet / DaemonSet / Job). // Walks over k8s.pod.* metrics and groups by workload-name + namespace. -export type WorkloadKind = "deployment" | "statefulset" | "daemonset" +export type WorkloadKind = "deployment" | "statefulset" | "daemonset" | "job" export interface ListWorkloadsOpts { kind: WorkloadKind @@ -1057,11 +1059,41 @@ export interface ListWorkloadsOutput { readonly environment: string readonly podCount: number readonly lastSeen: DateTime.Utc - readonly avgCpuLimitPct: number - readonly avgMemoryLimitPct: number - readonly avgCpuUsage: number + /** + * The latest desired / ready gauges from the k8s_cluster receiver. Jobs report + * completions: desired successful pods and pods that succeeded. Both read 0 when + * `replicaSamples` is 0, i.e. the cluster collector isn't running. + */ + readonly desiredReplicas: number + readonly readyReplicas: number + readonly replicaSamples: number + /** Desired minus ready, never negative; always 0 for jobs, whose shortfall is just "still running". */ + readonly shortfall: number + /** Jobs only: pods that failed. */ + readonly failedPods: number + /** Container restarts in the window, summed over every pod. */ + readonly restarts: number + /** The reason a restarting container last terminated (OOMKilled, Error, ...); empty with no restarts. */ + readonly lastTerminatedReason: string + /** The worst pod's peak, not the average: the average hides the one pod at its limit. */ + readonly peakCpuLimitPct: number + readonly peakMemoryLimitPct: number + readonly saturation: number + /** Cores in use now, summed over the pods still reporting at the window's end. */ + readonly cpuUsage: number } +/** The k8s_cluster receiver's gauges for each kind; job readiness counts completions. */ +const WORKLOAD_REPLICA_METRICS = { + deployment: { desired: "k8s.deployment.desired", ready: "k8s.deployment.available" }, + statefulset: { desired: "k8s.statefulset.desired_pods", ready: "k8s.statefulset.ready_pods" }, + daemonset: { desired: "k8s.daemonset.desired_scheduled_nodes", ready: "k8s.daemonset.ready_nodes" }, + job: { desired: "k8s.job.desired_successful_pods", ready: "k8s.job.successful_pods" }, +} as const satisfies Record + +const JOB_FAILED_METRIC = "k8s.job.failed_pods" as const +const CONTAINER_RESTARTS_METRIC = "k8s.container.restarts" as const + const workloadFilterConditions = ( $: ColumnAccessor, opts: ListWorkloadsOpts, @@ -1080,36 +1112,140 @@ const workloadFilterConditions = ( opts.environments?.length ? CH.inList(deploymentEnvExpr($.ResourceAttributes), opts.environments) : undefined, + // Pod and restart series carry the compute-type label (k8sattributes copies it from the node); + // the replica gauges don't, so they stay in the scan and listWorkloadsQuery drops workloads + // with no matching pod series. opts.computeTypes?.length - ? CH.inList($.ResourceAttributes.get("eks.amazonaws.com/compute-type"), opts.computeTypes) + ? CH.notInList($.MetricName, [...POD_METRIC_NAMES, CONTAINER_RESTARTS_METRIC]).or( + CH.inList($.ResourceAttributes.get("eks.amazonaws.com/compute-type"), opts.computeTypes), + ) : undefined, ] +/** + * One row per workload: replica health from the k8s_cluster gauges, restarts from + * `k8s.container.restarts`, and the worst pod's saturation from kubeletstats. + * + * The inner pass reduces each series (one pod, one container, one metric) to its + * first, last and peak value, because restarts are a running total per container + * and only the series' delta is meaningful. The outer pass rolls the series up per + * workload. Sorted worst first: missing replicas, failed job pods, restarts, saturation. + */ export function listWorkloadsQuery(opts: ListWorkloadsOpts) { const attrKey = workloadAttrKey(opts.kind) - return from(MetricsGauge) + const replica = WORKLOAD_REPLICA_METRICS[opts.kind] + const metricNames = [ + ...POD_METRIC_NAMES, + CONTAINER_RESTARTS_METRIC, + replica.desired, + replica.ready, + ...(opts.kind === "job" ? [JOB_FAILED_METRIC] : []), + ] + + const perSeries = from(MetricsGauge) .select(($) => ({ workloadName: $.ResourceAttributes.get(attrKey), // Same-named workloads in different namespaces/clusters are distinct rows. namespace: $.ResourceAttributes.get("k8s.namespace.name"), clusterName: $.ResourceAttributes.get("k8s.cluster.name"), + podUid: $.ResourceAttributes.get("k8s.pod.uid"), + containerName: $.ResourceAttributes.get("k8s.container.name"), + metric: $.MetricName, environment: CH.any_(deploymentEnvExpr($.ResourceAttributes)), - podCount: CH.uniq($.ResourceAttributes.get("k8s.pod.uid")), + earliest: CH.argMin($.Value, $.TimeUnix), + latest: CH.argMax($.Value, $.TimeUnix), + peak: CH.max_($.Value), lastSeen: CH.max_($.TimeUnix), - avgCpuLimitPct: avgIfOrZero($.Value, $.MetricName.eq("k8s.pod.cpu_limit_utilization")), - avgMemoryLimitPct: avgIfOrZero($.Value, $.MetricName.eq("k8s.pod.memory_limit_utilization")), - avgCpuUsage: avgIfOrZero($.Value, $.MetricName.eq("k8s.pod.cpu.usage")), + lastReason: CH.argMax( + $.ResourceAttributes.get("k8s.container.status.last_terminated_reason"), + $.TimeUnix, + ), + // When a restart counter first reached its high-water mark, i.e. its last restart. + // The key ranks by value, then by earliest time; both fit exactly in a Float64. + restartedAt: CH.toUnixTimestamp( + CH.argMax($.TimeUnix, $.Value.mul(4294967296).sub(CH.toUnixTimestamp($.TimeUnix))), + ), })) .where(($) => [ $.OrgId.eq(orgIdParam), $.TimeUnix.gte(param.dateTime("startTime")), $.TimeUnix.lte(param.dateTime("endTime")), $.ResourceAttributes.get(attrKey).neq(""), - $.MetricName.in_(...POD_METRIC_NAMES), + $.MetricName.in_(...metricNames), ...workloadFilterConditions($, opts, attrKey), ]) + .groupBy("workloadName", "namespace", "clusterName", "podUid", "containerName", "metric") + + const perWorkload = fromQuery(perSeries, "series") + .select(($) => { + const isRestarts = $.metric.eq(CONTAINER_RESTARTS_METRIC) + const restartDelta = CH.greatest_($.latest.sub($.earliest), CH.lit(0)) + const restartedWithReason = isRestarts.and($.latest.gt($.earliest)).and($.lastReason.neq("")) + const desiredReplicas = maxIfOrZero($.latest, $.metric.eq(replica.desired)) + const readyReplicas = maxIfOrZero($.latest, $.metric.eq(replica.ready)) + const peakCpuLimitPct = maxIfOrZero($.peak, $.metric.eq("k8s.pod.cpu_limit_utilization")) + const peakMemoryLimitPct = maxIfOrZero($.peak, $.metric.eq("k8s.pod.memory_limit_utilization")) + return { + workloadName: $.workloadName, + namespace: $.namespace, + clusterName: $.clusterName, + environment: CH.any_($.environment), + podCount: CH.uniqIf($.podUid, CH.inList($.metric, POD_METRIC_NAMES)), + lastSeen: CH.max_($.lastSeen), + desiredReplicas, + readyReplicas, + replicaSamples: CH.countIf($.metric.eq(replica.desired)), + shortfall: + opts.kind === "job" + ? CH.lit(0) + : CH.greatest_(desiredReplicas.sub(readyReplicas), CH.lit(0)), + failedPods: maxIfOrZero($.latest, $.metric.eq(JOB_FAILED_METRIC)), + restarts: finiteOrZero(CH.sumIf(restartDelta, isRestarts)), + // The reason of whichever container restarted most recently. + lastTerminatedReason: CH.argMax( + CH.if_(restartedWithReason, $.lastReason, CH.lit("")), + CH.if_(restartedWithReason, $.restartedAt, CH.lit(0)), + ), + peakCpuLimitPct, + peakMemoryLimitPct, + saturation: CH.greatest_(peakCpuLimitPct, peakMemoryLimitPct), + cpuUsage: finiteOrZero( + CH.sumIf($.latest, $.metric.eq("k8s.pod.cpu.usage").and($.lastSeen.gte(endedCutoff()))), + ), + } + }) .groupBy("workloadName", "namespace", "clusterName") - .orderBy(["lastSeen", "desc"]) + + // The builder has no HAVING, so the compute-type gate filters outside the grouping: + // a workload qualifies only if at least one of its pod series matched the filter. + return fromQuery(perWorkload, "workloads") + .select(($) => ({ + workloadName: $.workloadName, + namespace: $.namespace, + clusterName: $.clusterName, + environment: $.environment, + podCount: $.podCount, + lastSeen: $.lastSeen, + desiredReplicas: $.desiredReplicas, + readyReplicas: $.readyReplicas, + replicaSamples: $.replicaSamples, + shortfall: $.shortfall, + failedPods: $.failedPods, + restarts: $.restarts, + lastTerminatedReason: $.lastTerminatedReason, + peakCpuLimitPct: $.peakCpuLimitPct, + peakMemoryLimitPct: $.peakMemoryLimitPct, + saturation: $.saturation, + cpuUsage: $.cpuUsage, + })) + .where(($) => [opts.computeTypes?.length ? $.podCount.gt(0) : undefined]) + .orderBy( + ["shortfall", "desc"], + ["failedPods", "desc"], + ["restarts", "desc"], + ["saturation", "desc"], + ["workloadName", "asc"], + ) .limit(opts.limit ?? 200) .offset(opts.offset ?? 0) .format("JSON")