From b9c7b3581f2966cd8227e683aeb0658e6732726e Mon Sep 17 00:00:00 2001 From: Makisuo Date: Sun, 11 Oct 2026 01:20:08 +0200 Subject: [PATCH] feat(infra): show allocatable capacity and utilization for Kubernetes nodes The Nodes views only showed CPU cores in use, so there was no way to tell whether a node was full. Node queries now merge k8s_cluster allocatable CPU/memory with kubelet usage and return usage/allocatable fractions. - listNodesQuery / nodeDetailSummaryQuery: cpuAllocatable, memoryUsage (working set), memoryAllocatable, cpuUtilization, memoryUtilization - nodeGaugeTimeseriesQuery: optional capacityMetricName for per-bucket utilization; node charts gain CPU / allocatable, Memory, Memory / allocatable - string attributes use anyIf(!= '') so kubelet version and node uid (set only by k8s_cluster) are no longer blank - the list keeps requiring kubelet usage (HAVING) so cluster-only nodes stay out, matching the facets - benchmark fixtures for listNodesQuery and nodeDetailSummaryQuery --- .../src/routes/internal/query-engine.http.ts | 16 ++- apps/api/src/routes/queries.ts | 2 +- apps/web/src/api/warehouse/infra.ts | 2 +- apps/web/src/components/infra/format.test.ts | 28 ++++- apps/web/src/components/infra/format.ts | 19 ++- .../components/infra/infra-correlations.ts | 3 +- .../src/components/infra/k8s-detail-chart.tsx | 3 + apps/web/src/components/infra/node-table.tsx | 47 ++++++-- .../infra/kubernetes/nodes/$nodeName.tsx | 49 ++++++-- deploy/k8s-infra/values.yaml | 2 + .../backend/src/queries/query-helpers.test.ts | 25 ++++ packages/backend/src/queries/query-helpers.ts | 14 +++ packages/domain/src/http/query-engine.ts | 22 +++- .../src/__sql_baseline__/catalog.sql | 70 +++++++++++ .../query-engine/src/benchmark/builders.ts | 27 +++++ .../src/benchmark/catalog.test.ts | 2 - packages/query-engine/src/ch/index.ts | 1 + .../query-engine/src/ch/queries/infra.test.ts | 61 ++++++++++ packages/query-engine/src/ch/queries/infra.ts | 111 ++++++++++++++---- 19 files changed, 446 insertions(+), 58 deletions(-) create mode 100644 packages/backend/src/queries/query-helpers.test.ts diff --git a/apps/api/src/routes/internal/query-engine.http.ts b/apps/api/src/routes/internal/query-engine.http.ts index 5bd9585d76..8875d2e84f 100644 --- a/apps/api/src/routes/internal/query-engine.http.ts +++ b/apps/api/src/routes/internal/query-engine.http.ts @@ -297,6 +297,16 @@ const toReleaseTimelinePoint = (row: CH.ReleasesTimelineOutput) => ({ count: row.count, }) +const nodeCapacity = (row: CH.NodeCapacityOutput) => ({ + cpuUsage: row.cpuUsage || 0, + cpuAllocatable: row.cpuAllocatable || 0, + cpuUtilization: row.cpuUtilization || 0, + memoryUsage: row.memoryUsage || 0, + memoryAllocatable: row.memoryAllocatable || 0, + memoryUtilization: row.memoryUtilization || 0, + uptime: row.uptime || 0, +}) + const toServiceWorkloadRow = (row: CH.ServiceWorkloadsOutput) => ({ serviceName: decodeServiceName(row.serviceName), workloadKind: row.workloadKind, @@ -1741,8 +1751,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleInternalApi, "query environment: row.environment, kubeletVersion: row.kubeletVersion, lastSeen: DateTime.formatIso(row.lastSeen), - cpuUsage: row.cpuUsage || 0, - uptime: row.uptime || 0, + ...nodeCapacity(row), })), }) }), @@ -1760,8 +1769,7 @@ export const HttpQueryEngineLive = HttpApiBuilder.group(MapleInternalApi, "query containerRuntime: row.containerRuntime, firstSeen: DateTime.formatIso(row.firstSeen), lastSeen: DateTime.formatIso(row.lastSeen), - cpuUsage: row.cpuUsage || 0, - uptime: row.uptime || 0, + ...nodeCapacity(row), } : null, }) diff --git a/apps/api/src/routes/queries.ts b/apps/api/src/routes/queries.ts index a844c89eb7..7fa9529614 100644 --- a/apps/api/src/routes/queries.ts +++ b/apps/api/src/routes/queries.ts @@ -651,7 +651,7 @@ export const Queries = { CH.compile( CH.nodeGaugeTimeseriesQuery({ nodeName: payload.nodeName, - metricName: nodeMetricSpec(payload.metric).metricName, + ...nodeMetricSpec(payload.metric), }), { orgId, diff --git a/apps/web/src/api/warehouse/infra.ts b/apps/web/src/api/warehouse/infra.ts index a329181dc9..3aa6fae175 100644 --- a/apps/web/src/api/warehouse/infra.ts +++ b/apps/web/src/api/warehouse/infra.ts @@ -466,7 +466,7 @@ export function nodeDetailSummary({ data }: { data: NodeDetailSummaryInput }) { ) } -export type NodeInfraMetric = "cpu_usage" | "uptime" +export type NodeInfraMetric = NodeInfraTimeseriesRequest["metric"] export interface NodeInfraTimeseriesInput { startTime: string diff --git a/apps/web/src/components/infra/format.test.ts b/apps/web/src/components/infra/format.test.ts index 0e56f3ec19..9d3ce38cd8 100644 --- a/apps/web/src/components/infra/format.test.ts +++ b/apps/web/src/components/infra/format.test.ts @@ -1,5 +1,11 @@ import { describe, expect, it } from "vitest" -import { deriveHostStatus, severityLevel } from "./format" +import { + capacityFraction, + deriveHostStatus, + formatBytesOfAllocatable, + formatCoresOfAllocatable, + severityLevel, +} from "./format" import { formatBytesPerSecond, formatLoad, formatPercent, formatUptime } from "@maple/ui/lib/format" describe("deriveHostStatus", () => { @@ -107,3 +113,23 @@ describe("severityLevel", () => { expect(severityLevel(Number.NaN)).toBe("ok") }) }) + +describe("node capacity", () => { + it("keeps the utilization when allocatable was collected", () => { + expect(capacityFraction(0.25, 4)).toBe(0.25) + }) + + it("is NaN when allocatable was not collected, so it renders a dash", () => { + expect(capacityFraction(0, 0)).toBeNaN() + }) + + it("formats cores against allocatable", () => { + expect(formatCoresOfAllocatable(0.4231, 3.92)).toBe("0.42 / 3.92 cores") + expect(formatCoresOfAllocatable(0.4231, 0)).toBe("0.42 cores") + }) + + it("formats memory against allocatable", () => { + expect(formatBytesOfAllocatable(3 * 1024 ** 3, 16 * 1024 ** 3)).toBe("3.0 GB / 16.0 GB") + expect(formatBytesOfAllocatable(3 * 1024 ** 3, 0)).toBe("3.0 GB") + }) +}) diff --git a/apps/web/src/components/infra/format.ts b/apps/web/src/components/infra/format.ts index b3cac51877..c53139a39c 100644 --- a/apps/web/src/components/infra/format.ts +++ b/apps/web/src/components/infra/format.ts @@ -1,4 +1,4 @@ -import { EMPTY_VALUE } from "@maple/ui/lib/format" +import { EMPTY_VALUE, formatBytes } from "@maple/ui/lib/format" import { toEpochMs } from "@maple/ui/lib/time-format" import { type UtilizationLevel, utilizationLevel } from "@maple/ui/lib/utilization" @@ -40,3 +40,20 @@ export const formatWholePercent = (fraction: number) => /** CPU usage in cores, two decimals. */ export const formatCores = (cores: number) => (Number.isFinite(cores) ? cores.toFixed(2) : EMPTY_VALUE) + +/** + * A node's usage / allocatable fraction, or NaN when allocatable was not collected + * (the API sends 0 for both), so meters and percents render a dash instead of 0%. + */ +export const capacityFraction = (utilization: number, allocatable: number) => + allocatable > 0 ? utilization : Number.NaN + +/** "0.42 / 3.92 cores"; usage alone when allocatable was not collected. */ +export const formatCoresOfAllocatable = (usage: number, allocatable: number) => + allocatable > 0 + ? `${formatCores(usage)} / ${formatCores(allocatable)} cores` + : `${formatCores(usage)} cores` + +/** "3.1 GB / 15.6 GB"; usage alone when allocatable was not collected. */ +export const formatBytesOfAllocatable = (usage: number, allocatable: number) => + allocatable > 0 ? `${formatBytes(usage)} / ${formatBytes(allocatable)}` : formatBytes(usage) diff --git a/apps/web/src/components/infra/infra-correlations.ts b/apps/web/src/components/infra/infra-correlations.ts index ed52fa4ddd..efbcebd28f 100644 --- a/apps/web/src/components/infra/infra-correlations.ts +++ b/apps/web/src/components/infra/infra-correlations.ts @@ -74,7 +74,8 @@ const POD_CHARTS: ReadonlyArray<{ label: string; metric: PodInfraMetric }> = [ const NODE_CHARTS: ReadonlyArray<{ label: string; metric: NodeInfraMetric }> = [ { label: "CPU cores", metric: "cpu_usage" }, - { label: "Uptime", metric: "uptime" }, + { label: "CPU / allocatable", metric: "cpu_utilization" }, + { label: "Memory / allocatable", metric: "memory_utilization" }, ] const HOST_CHARTS: ReadonlyArray<{ label: string; metric: HostInfraMetric }> = [ diff --git a/apps/web/src/components/infra/k8s-detail-chart.tsx b/apps/web/src/components/infra/k8s-detail-chart.tsx index 8f19449f02..557e837f38 100644 --- a/apps/web/src/components/infra/k8s-detail-chart.tsx +++ b/apps/web/src/components/infra/k8s-detail-chart.tsx @@ -41,6 +41,9 @@ const POD_METRIC_LABELS: Record = { const NODE_METRIC_LABELS: Record = { cpu_usage: "CPU usage", + cpu_utilization: "CPU / allocatable", + memory_usage: "Memory working set", + memory_utilization: "Memory / allocatable", uptime: "Uptime", } satisfies Record diff --git a/apps/web/src/components/infra/node-table.tsx b/apps/web/src/components/infra/node-table.tsx index f393ef6661..2a20d87e4e 100644 --- a/apps/web/src/components/infra/node-table.tsx +++ b/apps/web/src/components/infra/node-table.tsx @@ -13,13 +13,14 @@ import { useTableSort, } from "@/components/common/data-table" import { MetaLine } from "./primitives/meta-line" -import { formatCores } from "./format" +import { MeterRows } from "./primitives/meter-rows" +import { capacityFraction, formatBytesOfAllocatable, formatCoresOfAllocatable } from "./format" import { formatUptime } from "@maple/ui/lib/format" import { RelativeTime } from "@/components/common/relative-time" export type NodeRow = ListNodesResponse["data"][number] -type SortKey = "nodeName" | "cpuUsage" | "uptime" | "lastSeen" +type SortKey = "nodeName" | "cpuUtilization" | "memoryUtilization" | "uptime" | "lastSeen" interface NodeTableProps { nodes: ReadonlyArray @@ -33,13 +34,19 @@ function NodeColumns({ sort }: { sort?: SortControls }) { <> label="Node" sortKey="nodeName" {...sort} width="w-0 flex-1 min-w-[260px]" /> - label="CPU cores" - sortKey="cpuUsage" + label="CPU / allocatable" + sortKey="cpuUtilization" {...sort} - align="right" - width="w-[110px]" + width="w-[176px]" hidden="hidden md:flex" /> + + label="Memory / allocatable" + sortKey="memoryUtilization" + {...sort} + width="w-[176px]" + hidden="hidden lg:flex" + /> label="Uptime" sortKey="uptime" @@ -70,7 +77,8 @@ export function NodeTableLoading() { - + + @@ -80,7 +88,7 @@ export function NodeTableLoading() { export function NodeTable({ nodes, waiting, referenceTime }: NodeTableProps) { const { sorted, sortKey, sortDir, handleSort } = useTableSort(nodes, { - initialKey: "cpuUsage", + initialKey: "cpuUtilization", stringKeys: ["nodeName"], }) @@ -107,8 +115,17 @@ export function NodeTable({ nodes, waiting, referenceTime }: NodeTableProps) { -
- {formatCores(node.cpuUsage)} +
+ +
+
+
{formatUptime(node.uptime)} @@ -125,3 +142,13 @@ export function NodeTable({ nodes, waiting, referenceTime }: NodeTableProps) { ) } + +/** Usage against allocatable capacity: the meter answers "is this node full", the caption says of what. */ +function CapacityCell({ fraction, caption }: { fraction: number; caption: string }) { + return ( +
+ +
{caption}
+
+ ) +} diff --git a/apps/web/src/routes/infra/kubernetes/nodes/$nodeName.tsx b/apps/web/src/routes/infra/kubernetes/nodes/$nodeName.tsx index 5c7679fcf4..054a070f24 100644 --- a/apps/web/src/routes/infra/kubernetes/nodes/$nodeName.tsx +++ b/apps/web/src/routes/infra/kubernetes/nodes/$nodeName.tsx @@ -11,7 +11,7 @@ import { ResourceAttributesCardSkeleton, } from "@/components/infra/primitives/resource-attributes-card" import { NoMetricsMessage } from "@/components/infra/primitives/no-metrics-message" -import { EMPTY_VALUE, formatUptime } from "@maple/ui/lib/format" +import { EMPTY_VALUE, formatPercent, formatUptime } from "@maple/ui/lib/format" import type { NodeInfraMetric } from "@/api/warehouse/infra" import { ServerIcon } from "@/components/icons" @@ -19,7 +19,12 @@ import { KubernetesShell } from "@/components/infra/kubernetes/kubernetes-shell" import { NodeDetailChart } from "@/components/infra/k8s-detail-chart" import { PodTable } from "@/components/infra/pod-table" import { bucketSecondsForRange } from "@/components/infra/constants" -import { formatCores } from "@/components/infra/format" +import { + capacityFraction, + formatBytesOfAllocatable, + formatCoresOfAllocatable, + severityLevel, +} from "@/components/infra/format" import { PageHero, HeroChip } from "@/components/common/page-hero" import { SegmentPivot } from "@/components/infra/primitives/segment-pivot" import { StatRail, StatRailItem } from "@/components/common/stat-rail" @@ -44,6 +49,9 @@ export const Route = createFileRoute("/infra/kubernetes/nodes/$nodeName")({ const METRIC_OPTIONS = [ { value: "cpu_usage", label: "CPU cores" }, + { value: "cpu_utilization", label: "CPU / allocatable" }, + { value: "memory_usage", label: "Memory" }, + { value: "memory_utilization", label: "Memory / allocatable" }, { value: "uptime", label: "Uptime" }, ] as const satisfies ReadonlyArray<{ value: NodeInfraMetric; label: string }> @@ -101,7 +109,7 @@ function NodeDetailPage() {
{nodeName}} - description="Node metrics from the kubelet stats receiver." + description="Usage from the kubelet stats receiver, allocatable capacity from the cluster receiver." meta={ summary ? ( <> @@ -117,11 +125,16 @@ function NodeDetailPage() { /> {summary ? ( - - + + ) } + +function CapacityStat({ + eyebrow, + fraction, + subline, +}: { + eyebrow: string + fraction: number + subline: string +}) { + return ( + + ) +} diff --git a/deploy/k8s-infra/values.yaml b/deploy/k8s-infra/values.yaml index 0391527a0c..7539f3ed33 100644 --- a/deploy/k8s-infra/values.yaml +++ b/deploy/k8s-infra/values.yaml @@ -301,6 +301,8 @@ presets: - DiskPressure - PIDPressure - NetworkUnavailable + # Emits k8s.node.allocatable_cpu (cores) and k8s.node.allocatable_memory (bytes); + # the Nodes view divides kubelet usage by these for CPU/memory utilization. allocatableTypesToReport: - cpu - memory diff --git a/packages/backend/src/queries/query-helpers.test.ts b/packages/backend/src/queries/query-helpers.test.ts new file mode 100644 index 0000000000..073a6e80fb --- /dev/null +++ b/packages/backend/src/queries/query-helpers.test.ts @@ -0,0 +1,25 @@ +import { describe, expect, it } from "vitest" +import { nodeMetricSpec } from "./query-helpers" + +describe("nodeMetricSpec", () => { + it("maps utilization metrics to usage over k8s_cluster allocatable", () => { + expect(nodeMetricSpec("cpu_utilization")).toEqual({ + metricName: "k8s.node.cpu.usage", + capacityMetricName: "k8s.node.allocatable_cpu", + unit: "percent", + }) + expect(nodeMetricSpec("memory_utilization")).toEqual({ + metricName: "k8s.node.memory.working_set", + capacityMetricName: "k8s.node.allocatable_memory", + unit: "percent", + }) + }) + + it("leaves raw gauges without a capacity metric", () => { + expect(nodeMetricSpec("memory_usage")).toEqual({ + metricName: "k8s.node.memory.working_set", + unit: "bytes", + }) + expect(nodeMetricSpec("cpu_usage")).toEqual({ metricName: "k8s.node.cpu.usage", unit: "cores" }) + }) +}) diff --git a/packages/backend/src/queries/query-helpers.ts b/packages/backend/src/queries/query-helpers.ts index 923cfeb1fe..c45a2f9537 100644 --- a/packages/backend/src/queries/query-helpers.ts +++ b/packages/backend/src/queries/query-helpers.ts @@ -93,6 +93,20 @@ export const nodeMetricSpec = (metric: NodeInfraTimeseriesRequest["metric"]) => switch (metric) { case "cpu_usage": return { metricName: "k8s.node.cpu.usage", unit: "cores" as const } + case "cpu_utilization": + return { + metricName: "k8s.node.cpu.usage", + capacityMetricName: "k8s.node.allocatable_cpu", + unit: "percent" as const, + } + case "memory_usage": + return { metricName: "k8s.node.memory.working_set", unit: "bytes" as const } + case "memory_utilization": + return { + metricName: "k8s.node.memory.working_set", + capacityMetricName: "k8s.node.allocatable_memory", + unit: "percent" as const, + } case "uptime": return { metricName: "k8s.node.uptime", unit: "seconds" as const } } diff --git a/packages/domain/src/http/query-engine.ts b/packages/domain/src/http/query-engine.ts index 96123840dd..e79a2f128f 100644 --- a/packages/domain/src/http/query-engine.ts +++ b/packages/domain/src/http/query-engine.ts @@ -2153,6 +2153,18 @@ export class ListNodesRequest extends Schema.Class("ListNodesR offset: Schema.optional(Schema.Number), }) {} +// Allocatable is 0 when the k8s_cluster receiver did not report it; utilizations are +// usage / allocatable fractions (0-1) and are 0 in that case. +const NodeCapacityFields = { + cpuUsage: Schema.Number, + cpuAllocatable: Schema.Number, + cpuUtilization: Schema.Number, + memoryUsage: Schema.Number, + memoryAllocatable: Schema.Number, + memoryUtilization: Schema.Number, + uptime: Schema.Number, +} + const NodeRow = Schema.Struct({ nodeName: Schema.String, nodeUid: Schema.String, @@ -2160,8 +2172,7 @@ const NodeRow = Schema.Struct({ environment: Schema.String, kubeletVersion: Schema.String, lastSeen: Schema.String, - cpuUsage: Schema.Number, - uptime: Schema.Number, + ...NodeCapacityFields, }) export class ListNodesResponse extends Schema.Class("ListNodesResponse")({ @@ -2204,8 +2215,7 @@ export class NodeDetailSummaryResponse extends Schema.Class("ListWorkloadsRequest")({ diff --git a/packages/query-engine/src/__sql_baseline__/catalog.sql b/packages/query-engine/src/__sql_baseline__/catalog.sql index 251b75e0df..4955428c6e 100644 --- a/packages/query-engine/src/__sql_baseline__/catalog.sql +++ b/packages/query-engine/src/__sql_baseline__/catalog.sql @@ -1137,6 +1137,60 @@ SELECT LIMIT 1 FORMAT JSON +-- builder:infra:listNodesQuery:default [6fe8973a] +SELECT + metrics_gauge.ResourceAttributes['k8s.node.name'] AS nodeName, + anyIf(metrics_gauge.ResourceAttributes['k8s.node.uid'], metrics_gauge.ResourceAttributes['k8s.node.uid'] != '') AS nodeUid, + anyIf(metrics_gauge.ResourceAttributes['k8s.cluster.name'], metrics_gauge.ResourceAttributes['k8s.cluster.name'] != '') AS clusterName, + anyIf(coalesce(nullIf(metrics_gauge.ResourceAttributes['deployment.environment.name'], ''), metrics_gauge.ResourceAttributes['deployment.environment']), coalesce(nullIf(metrics_gauge.ResourceAttributes['deployment.environment.name'], ''), metrics_gauge.ResourceAttributes['deployment.environment']) != '') AS environment, + anyIf(metrics_gauge.ResourceAttributes['k8s.kubelet.version'], metrics_gauge.ResourceAttributes['k8s.kubelet.version'] != '') AS kubeletVersion, + max(metrics_gauge.TimeUnix) AS lastSeen, + ifNull(ifNotFinite(avgIf(metrics_gauge.Value, metrics_gauge.MetricName = 'k8s.node.cpu.usage'), 0), 0) AS cpuUsage, + ifNotFinite(maxIf(metrics_gauge.Value, metrics_gauge.MetricName = 'k8s.node.allocatable_cpu'), 0) AS cpuAllocatable, + ifNull(ifNotFinite(ifNull(ifNotFinite(avgIf(metrics_gauge.Value, metrics_gauge.MetricName = 'k8s.node.cpu.usage'), 0), 0) / ifNotFinite(maxIf(metrics_gauge.Value, metrics_gauge.MetricName = 'k8s.node.allocatable_cpu'), 0), 0), 0) AS cpuUtilization, + ifNull(ifNotFinite(avgIf(metrics_gauge.Value, metrics_gauge.MetricName = 'k8s.node.memory.working_set'), 0), 0) AS memoryUsage, + ifNotFinite(maxIf(metrics_gauge.Value, metrics_gauge.MetricName = 'k8s.node.allocatable_memory'), 0) AS memoryAllocatable, + ifNull(ifNotFinite(ifNull(ifNotFinite(avgIf(metrics_gauge.Value, metrics_gauge.MetricName = 'k8s.node.memory.working_set'), 0), 0) / ifNotFinite(maxIf(metrics_gauge.Value, metrics_gauge.MetricName = 'k8s.node.allocatable_memory'), 0), 0), 0) AS memoryUtilization, + ifNotFinite(maxIf(metrics_gauge.Value, metrics_gauge.MetricName = 'k8s.node.uptime'), 0) AS uptime + FROM metrics_gauge + WHERE metrics_gauge.OrgId = 'org_sql_catalog' + AND metrics_gauge.TimeUnix >= '2026-01-01 10:30:00' + AND metrics_gauge.TimeUnix <= '2026-01-03 14:15:00' + AND metrics_gauge.ResourceAttributes['k8s.node.name'] != '' + AND metrics_gauge.ResourceAttributes['k8s.pod.name'] = '' + AND metrics_gauge.MetricName IN ('k8s.node.cpu.usage', 'k8s.node.uptime', 'k8s.node.memory.working_set', 'k8s.node.allocatable_cpu', 'k8s.node.allocatable_memory') + GROUP BY nodeName + HAVING countIf(metrics_gauge.MetricName = 'k8s.node.cpu.usage') > 0 + ORDER BY lastSeen DESC + LIMIT 200 + OFFSET 0 + FORMAT JSON + +-- builder:infra:nodeDetailSummaryQuery:default [33fb2081] +SELECT + metrics_gauge.ResourceAttributes['k8s.node.name'] AS nodeName, + anyIf(metrics_gauge.ResourceAttributes['k8s.node.uid'], metrics_gauge.ResourceAttributes['k8s.node.uid'] != '') AS nodeUid, + anyIf(metrics_gauge.ResourceAttributes['k8s.kubelet.version'], metrics_gauge.ResourceAttributes['k8s.kubelet.version'] != '') AS kubeletVersion, + anyIf(coalesce(nullIf(metrics_gauge.ResourceAttributes['container.runtime.name'], ''), metrics_gauge.ResourceAttributes['container.runtime']), coalesce(nullIf(metrics_gauge.ResourceAttributes['container.runtime.name'], ''), metrics_gauge.ResourceAttributes['container.runtime']) != '') AS containerRuntime, + min(metrics_gauge.TimeUnix) AS firstSeen, + max(metrics_gauge.TimeUnix) AS lastSeen, + ifNull(ifNotFinite(avgIf(metrics_gauge.Value, metrics_gauge.MetricName = 'k8s.node.cpu.usage'), 0), 0) AS cpuUsage, + ifNotFinite(maxIf(metrics_gauge.Value, metrics_gauge.MetricName = 'k8s.node.allocatable_cpu'), 0) AS cpuAllocatable, + ifNull(ifNotFinite(ifNull(ifNotFinite(avgIf(metrics_gauge.Value, metrics_gauge.MetricName = 'k8s.node.cpu.usage'), 0), 0) / ifNotFinite(maxIf(metrics_gauge.Value, metrics_gauge.MetricName = 'k8s.node.allocatable_cpu'), 0), 0), 0) AS cpuUtilization, + ifNull(ifNotFinite(avgIf(metrics_gauge.Value, metrics_gauge.MetricName = 'k8s.node.memory.working_set'), 0), 0) AS memoryUsage, + ifNotFinite(maxIf(metrics_gauge.Value, metrics_gauge.MetricName = 'k8s.node.allocatable_memory'), 0) AS memoryAllocatable, + ifNull(ifNotFinite(ifNull(ifNotFinite(avgIf(metrics_gauge.Value, metrics_gauge.MetricName = 'k8s.node.memory.working_set'), 0), 0) / ifNotFinite(maxIf(metrics_gauge.Value, metrics_gauge.MetricName = 'k8s.node.allocatable_memory'), 0), 0), 0) AS memoryUtilization, + ifNotFinite(maxIf(metrics_gauge.Value, metrics_gauge.MetricName = 'k8s.node.uptime'), 0) AS uptime + FROM metrics_gauge + WHERE metrics_gauge.OrgId = 'org_sql_catalog' + AND metrics_gauge.TimeUnix >= '2026-01-01 10:30:00' + AND metrics_gauge.TimeUnix <= '2026-01-03 14:15:00' + AND metrics_gauge.ResourceAttributes['k8s.node.name'] = 'ip-10-0-1-42.ec2.internal' + AND metrics_gauge.ResourceAttributes['k8s.pod.name'] = '' + AND metrics_gauge.MetricName IN ('k8s.node.cpu.usage', 'k8s.node.uptime', 'k8s.node.memory.working_set', 'k8s.node.allocatable_cpu', 'k8s.node.allocatable_memory') + GROUP BY nodeName + FORMAT JSON + -- builder:infra:nodeFacetsQuery:default [0dff8530] SELECT metrics_gauge.ResourceAttributes['k8s.node.name'] AS name, @@ -1203,6 +1257,22 @@ SELECT ORDER BY bucket ASC FORMAT JSON +-- builder:infra:nodeGaugeTimeseriesQuery:utilization [f4c7789a] +SELECT + toStartOfInterval(metrics_gauge.TimeUnix, INTERVAL 300 SECOND) AS bucket, + '' AS attributeValue, + ifNull(ifNotFinite(ifNull(ifNotFinite(avgIf(metrics_gauge.Value, metrics_gauge.MetricName = 'k8s.node.memory.working_set'), 0), 0) / ifNotFinite(maxIf(metrics_gauge.Value, metrics_gauge.MetricName = 'k8s.node.allocatable_memory'), 0), 0), 0) AS avgValue + FROM metrics_gauge + WHERE metrics_gauge.OrgId = 'org_sql_catalog' + AND metrics_gauge.TimeUnix >= '2026-01-01 10:30:00' + AND metrics_gauge.TimeUnix <= '2026-01-03 14:15:00' + AND metrics_gauge.ResourceAttributes['k8s.node.name'] = 'ip-10-0-1-42.ec2.internal' + AND metrics_gauge.ResourceAttributes['k8s.pod.name'] = '' + AND metrics_gauge.MetricName IN ('k8s.node.memory.working_set', 'k8s.node.allocatable_memory') + GROUP BY bucket + ORDER BY bucket ASC + FORMAT JSON + -- builder:infra:podFacetsQuery:default [cc0c4122] SELECT metrics_gauge.ResourceAttributes['k8s.pod.name'] AS name, diff --git a/packages/query-engine/src/benchmark/builders.ts b/packages/query-engine/src/benchmark/builders.ts index e13885d509..805f051683 100644 --- a/packages/query-engine/src/benchmark/builders.ts +++ b/packages/query-engine/src/benchmark/builders.ts @@ -1268,6 +1268,33 @@ export const builderFixtures: ReadonlyArray = [ { ...window, bucketSeconds: 300 }, ), }, + { + module: "infra", + name: "listNodesQuery", + label: "default", + compile: () => CH.compileUnsafe(CH.listNodesQuery(), window), + }, + { + module: "infra", + name: "nodeDetailSummaryQuery", + label: "default", + compile: () => + CH.compileUnsafe(CH.nodeDetailSummaryQuery({ nodeName: "ip-10-0-1-42.ec2.internal" }), window), + }, + { + module: "infra", + name: "nodeGaugeTimeseriesQuery", + label: "utilization", + compile: () => + CH.compileUnsafe( + CH.nodeGaugeTimeseriesQuery({ + nodeName: "ip-10-0-1-42.ec2.internal", + metricName: "k8s.node.memory.working_set", + capacityMetricName: "k8s.node.allocatable_memory", + }), + { ...window, bucketSeconds: 300 }, + ), + }, { module: "infra", name: "nodeGaugeTimeseriesQuery", diff --git a/packages/query-engine/src/benchmark/catalog.test.ts b/packages/query-engine/src/benchmark/catalog.test.ts index 80111b3f45..387ae37ae9 100644 --- a/packages/query-engine/src/benchmark/catalog.test.ts +++ b/packages/query-engine/src/benchmark/catalog.test.ts @@ -366,8 +366,6 @@ const EXEMPT_BUILDERS: ReadonlySet = new Set([ "infra/listPodsQuery", "infra/listPodsSummaryQuery", "infra/podDetailSummaryQuery", - "infra/listNodesQuery", - "infra/nodeDetailSummaryQuery", "infra/listWorkloadsQuery", "infra/workloadDetailSummaryQuery", "service-infra/serviceWorkloadsSQL", diff --git a/packages/query-engine/src/ch/index.ts b/packages/query-engine/src/ch/index.ts index 7ece54c1ab..125c8ef8ab 100644 --- a/packages/query-engine/src/ch/index.ts +++ b/packages/query-engine/src/ch/index.ts @@ -622,6 +622,7 @@ export { type PodFacetsOutput, type ListNodesOpts, type ListNodesOutput, + type NodeCapacityOutput, type NodeDetailSummaryOpts, type NodeDetailSummaryOutput, type NodeGaugeTimeseriesOpts, diff --git a/packages/query-engine/src/ch/queries/infra.test.ts b/packages/query-engine/src/ch/queries/infra.test.ts index ead6df77dd..98ebee26ec 100644 --- a/packages/query-engine/src/ch/queries/infra.test.ts +++ b/packages/query-engine/src/ch/queries/infra.test.ts @@ -324,6 +324,37 @@ describe("listNodesQuery", () => { expect(sql).not.toMatch(/__PARAM_\w+__/) }) + it("joins kubelet usage with k8s_cluster allocatable capacity", () => { + const { sql } = compileUnsafe(listNodesQuery({}), baseParams) + expect(sql).toContain( + "MetricName IN ('k8s.node.cpu.usage', 'k8s.node.uptime', 'k8s.node.memory.working_set', 'k8s.node.allocatable_cpu', 'k8s.node.allocatable_memory')", + ) + expect(sql).toContain( + "ifNotFinite(maxIf(metrics_gauge.Value, metrics_gauge.MetricName = 'k8s.node.allocatable_cpu'), 0) AS cpuAllocatable", + ) + expect(sql).toContain( + "ifNotFinite(maxIf(metrics_gauge.Value, metrics_gauge.MetricName = 'k8s.node.allocatable_memory'), 0) AS memoryAllocatable", + ) + expect(sql).toMatch( + /k8s\.node\.cpu\.usage'\), 0\), 0\) \/ ifNotFinite\(maxIf\([^)]*'k8s\.node\.allocatable_cpu'\), 0\)[^\n]*AS cpuUtilization/, + ) + expect(sql).toMatch( + /k8s\.node\.memory\.working_set'\), 0\), 0\) \/ ifNotFinite\(maxIf\([^)]*'k8s\.node\.allocatable_memory'\), 0\)[^\n]*AS memoryUtilization/, + ) + }) + + it("reads string attributes from whichever receiver set them", () => { + const { sql } = compileUnsafe(listNodesQuery({}), baseParams) + expect(sql).toContain( + "anyIf(metrics_gauge.ResourceAttributes['k8s.kubelet.version'], metrics_gauge.ResourceAttributes['k8s.kubelet.version'] != '') AS kubeletVersion", + ) + }) + + it("drops nodes that only k8s_cluster reports", () => { + const { sql } = compileUnsafe(listNodesQuery({}), baseParams) + expect(sql).toContain("HAVING countIf(metrics_gauge.MetricName = 'k8s.node.cpu.usage') > 0") + }) + it("applies cluster/environment array filters", () => { const { sql } = compileUnsafe( listNodesQuery({ @@ -363,6 +394,21 @@ describe("nodeDetailSummaryQuery", () => { expect(sql).toContain("'node-7'") expect(sql).toContain("ResourceAttributes['k8s.pod.name'] = ''") }) + + it("selects allocatable capacity and utilization", () => { + const { sql } = compileUnsafe(nodeDetailSummaryQuery({ nodeName: "node-7" }), baseParams) + expect(sql).toContain("'k8s.node.allocatable_cpu'") + expect(sql).toContain("'k8s.node.allocatable_memory'") + for (const alias of [ + "cpuAllocatable", + "cpuUtilization", + "memoryUsage", + "memoryAllocatable", + "memoryUtilization", + ]) { + expect(sql).toContain(`AS ${alias}`) + } + }) }) describe("nodeGaugeTimeseriesQuery", () => { @@ -378,6 +424,21 @@ describe("nodeGaugeTimeseriesQuery", () => { expect(sql).toContain("MetricName = 'k8s.node.cpu.usage'") expect(sql).toContain("'node-7'") }) + + it("divides usage by allocatable per bucket when given a capacity metric", () => { + const { sql } = compileUnsafe( + nodeGaugeTimeseriesQuery({ + nodeName: "node-7", + metricName: "k8s.node.memory.working_set", + capacityMetricName: "k8s.node.allocatable_memory", + }), + baseParams, + ) + expect(sql).toContain("MetricName IN ('k8s.node.memory.working_set', 'k8s.node.allocatable_memory')") + expect(sql).toMatch( + /working_set'\), 0\), 0\) \/ ifNotFinite\(maxIf\([^)]*'k8s\.node\.allocatable_memory'\), 0\), 0\), 0\) AS avgValue/, + ) + }) }) // Workloads diff --git a/packages/query-engine/src/ch/queries/infra.ts b/packages/query-engine/src/ch/queries/infra.ts index 15c27ee3f6..05720f5e10 100644 --- a/packages/query-engine/src/ch/queries/infra.ts +++ b/packages/query-engine/src/ch/queries/infra.ts @@ -19,6 +19,7 @@ import { unionAll, type CHUnionQuery } from "@maple-dev/effect-orm/clickhouse" import { MetricsGauge, MetricsSum, orgIdParam } from "../tables" import { containerRuntimeExpr, deploymentEnvExpr } from "@maple/domain/tinybird/semconv-renames" import { avgIfOrZero, facetAttrExpr, maxIfOrZero, type FacetOutput } from "./query-helpers" +import { finiteOrZero } from "./format" const HOSTMETRIC_NAMES = [ "system.cpu.utilization", @@ -710,10 +711,22 @@ export function podGaugeTimeseriesQuery(opts: PodGaugeTimeseriesOpts) { // Kubernetes — node aggregations over k8s.node.* metrics from the kubelet // stats + k8s_cluster receivers. -// k8s.node.cpu.usage gauge, cores -// k8s.node.uptime gauge, seconds - -const NODE_METRIC_NAMES = ["k8s.node.cpu.usage", "k8s.node.uptime"] as const +// k8s.node.cpu.usage gauge, cores (kubeletstats) +// k8s.node.uptime gauge, seconds (kubeletstats) +// k8s.node.memory.working_set gauge, bytes (kubeletstats) +// k8s.node.allocatable_cpu gauge, cores (k8s_cluster, allocatable_types_to_report) +// k8s.node.allocatable_memory gauge, bytes (k8s_cluster, allocatable_types_to_report) + +const NODE_ALLOCATABLE_CPU_METRIC = "k8s.node.allocatable_cpu" as const +const NODE_ALLOCATABLE_MEMORY_METRIC = "k8s.node.allocatable_memory" as const + +const NODE_METRIC_NAMES = [ + "k8s.node.cpu.usage", + "k8s.node.uptime", + "k8s.node.memory.working_set", + NODE_ALLOCATABLE_CPU_METRIC, + NODE_ALLOCATABLE_MEMORY_METRIC, +] as const // Single representative metric for node facets — see POD_FACET_PROBE_METRIC. const NODE_FACET_PROBE_METRIC = "k8s.node.cpu.usage" as const @@ -727,17 +740,56 @@ export interface ListNodesOpts { offset?: number } -export interface ListNodesOutput { +/** Usage, allocatable capacity, usage/allocatable fractions and uptime; 0 when not collected. */ +export interface NodeCapacityOutput { + readonly cpuUsage: number + readonly cpuAllocatable: number + readonly cpuUtilization: number + readonly memoryUsage: number + readonly memoryAllocatable: number + readonly memoryUtilization: number + readonly uptime: number +} + +export interface ListNodesOutput extends NodeCapacityOutput { readonly nodeName: string readonly nodeUid: string readonly clusterName: string readonly environment: string readonly kubeletVersion: string readonly lastSeen: DateTime.Utc - readonly cpuUsage: number - readonly uptime: number } +/** `usage / allocatable`; 0 when allocatable was not collected (x/0 is non-finite). */ +const ratioOrZero = (usage: CH.Expr, allocatable: CH.Expr): CH.Expr => + finiteOrZero(usage.div(allocatable)) + +// The node row merges two receivers: kubeletstats (usage) and k8s_cluster (allocatable, +// kubelet version). String attributes skip the receiver that leaves them empty. +const nodeCapacityColumns = ($: ColumnAccessor) => { + const cpuUsage = avgIfOrZero($.Value, $.MetricName.eq("k8s.node.cpu.usage")) + const cpuAllocatable = maxIfOrZero($.Value, $.MetricName.eq(NODE_ALLOCATABLE_CPU_METRIC)) + const memoryUsage = avgIfOrZero($.Value, $.MetricName.eq("k8s.node.memory.working_set")) + const memoryAllocatable = maxIfOrZero($.Value, $.MetricName.eq(NODE_ALLOCATABLE_MEMORY_METRIC)) + return { + cpuUsage, + cpuAllocatable, + cpuUtilization: ratioOrZero(cpuUsage, cpuAllocatable), + memoryUsage, + memoryAllocatable, + memoryUtilization: ratioOrZero(memoryUsage, memoryAllocatable), + uptime: maxIfOrZero($.Value, $.MetricName.eq("k8s.node.uptime")), + } +} + +const nonEmptyAttr = ($: ColumnAccessor, key: string) => + CH.anyIf($.ResourceAttributes.get(key), $.ResourceAttributes.get(key).neq("")) + +// Nodes that only k8s_cluster reports (EKS Fargate virtual nodes, nodes the DaemonSet +// does not run on) carry capacity but no usage; the list and facets both require kubelet data. +const hasKubeletUsage = ($: ColumnAccessor) => + CH.countIf($.MetricName.eq(NODE_FACET_PROBE_METRIC)).gt(0) + const nodeBaseConditions = ( $: ColumnAccessor, metricNames: ReadonlyArray = NODE_METRIC_NAMES, @@ -770,16 +822,19 @@ export function listNodesQuery(opts: ListNodesOpts = {}) { return from(MetricsGauge) .select(($) => ({ nodeName: $.ResourceAttributes.get("k8s.node.name"), - nodeUid: CH.any_($.ResourceAttributes.get("k8s.node.uid")), - clusterName: CH.any_($.ResourceAttributes.get("k8s.cluster.name")), - environment: CH.any_(deploymentEnvExpr($.ResourceAttributes)), - kubeletVersion: CH.any_($.ResourceAttributes.get("k8s.kubelet.version")), + nodeUid: nonEmptyAttr($, "k8s.node.uid"), + clusterName: nonEmptyAttr($, "k8s.cluster.name"), + environment: CH.anyIf( + deploymentEnvExpr($.ResourceAttributes), + deploymentEnvExpr($.ResourceAttributes).neq(""), + ), + kubeletVersion: nonEmptyAttr($, "k8s.kubelet.version"), lastSeen: CH.max_($.TimeUnix), - cpuUsage: avgIfOrZero($.Value, $.MetricName.eq("k8s.node.cpu.usage")), - uptime: maxIfOrZero($.Value, $.MetricName.eq("k8s.node.uptime")), + ...nodeCapacityColumns($), })) .where(($) => [...nodeBaseConditions($), ...nodeFilterConditions($, opts)]) .groupBy("nodeName") + .having(($) => [hasKubeletUsage($)]) .orderBy(["lastSeen", "desc"]) .limit(opts.limit ?? 200) .offset(opts.offset ?? 0) @@ -790,28 +845,28 @@ export interface NodeDetailSummaryOpts { nodeName: string } -export interface NodeDetailSummaryOutput { +export interface NodeDetailSummaryOutput extends NodeCapacityOutput { readonly nodeName: string readonly nodeUid: string readonly kubeletVersion: string readonly containerRuntime: string readonly firstSeen: DateTime.Utc readonly lastSeen: DateTime.Utc - readonly cpuUsage: number - readonly uptime: number } export function nodeDetailSummaryQuery(opts: NodeDetailSummaryOpts) { return from(MetricsGauge) .select(($) => ({ nodeName: $.ResourceAttributes.get("k8s.node.name"), - nodeUid: CH.any_($.ResourceAttributes.get("k8s.node.uid")), - kubeletVersion: CH.any_($.ResourceAttributes.get("k8s.kubelet.version")), - containerRuntime: CH.any_(containerRuntimeExpr($.ResourceAttributes)), + nodeUid: nonEmptyAttr($, "k8s.node.uid"), + kubeletVersion: nonEmptyAttr($, "k8s.kubelet.version"), + containerRuntime: CH.anyIf( + containerRuntimeExpr($.ResourceAttributes), + containerRuntimeExpr($.ResourceAttributes).neq(""), + ), firstSeen: CH.min_($.TimeUnix), lastSeen: CH.max_($.TimeUnix), - cpuUsage: avgIfOrZero($.Value, $.MetricName.eq("k8s.node.cpu.usage")), - uptime: maxIfOrZero($.Value, $.MetricName.eq("k8s.node.uptime")), + ...nodeCapacityColumns($), })) .where(($) => [ $.OrgId.eq(orgIdParam), @@ -828,14 +883,22 @@ export function nodeDetailSummaryQuery(opts: NodeDetailSummaryOpts) { export interface NodeGaugeTimeseriesOpts { nodeName: string metricName: string + /** Plot `metricName / capacityMetricName` per bucket (a 0-1 fraction) instead of the raw gauge. */ + capacityMetricName?: string } export function nodeGaugeTimeseriesQuery(opts: NodeGaugeTimeseriesOpts) { + const capacityMetricName = opts.capacityMetricName return from(MetricsGauge) .select(($) => ({ bucket: CH.toStartOfInterval($.TimeUnix, param.int("bucketSeconds")), attributeValue: CH.lit(""), - avgValue: CH.avg($.Value), + avgValue: capacityMetricName + ? ratioOrZero( + avgIfOrZero($.Value, $.MetricName.eq(opts.metricName)), + maxIfOrZero($.Value, $.MetricName.eq(capacityMetricName)), + ) + : CH.avg($.Value), })) .where(($) => [ $.OrgId.eq(orgIdParam), @@ -843,7 +906,9 @@ export function nodeGaugeTimeseriesQuery(opts: NodeGaugeTimeseriesOpts) { $.TimeUnix.lte(param.dateTime("endTime")), $.ResourceAttributes.get("k8s.node.name").eq(opts.nodeName), $.ResourceAttributes.get("k8s.pod.name").eq(""), - $.MetricName.eq(opts.metricName), + capacityMetricName + ? $.MetricName.in_(opts.metricName, capacityMetricName) + : $.MetricName.eq(opts.metricName), ]) .groupBy("bucket") .orderBy(["bucket", "asc"])