From 4de37364324a3ec05072cf97db162a2a9a947ffa Mon Sep 17 00:00:00 2001 From: Makisuo Date: Sun, 11 Oct 2026 01:18:39 +0200 Subject: [PATCH 1/3] feat(ai): give agents MCP tools for the Infrastructure pages list_infra lists hosts, Kubernetes pods/nodes/workloads and Docker containers closest to a limit first, with fleet counts, status filters (hot, at_limit, no_limits), pod restarts and the metrics behind each view. inspect_infra reads one entity over a window: min/avg/max/last, peak time and trend, per-pod peaks for a workload, per-mount disk use for a host, node peers, and a note when memory drops after a peak. diagnose_service now shows the workload and pods a service runs on. create_dashboard accepts resource. group_by on metrics, which the query builder already supported. Adds hostFilesystemsQuery (pins state = used) and podRestartsQuery, a fixture infra world, and a tool playground that lets agents test the tools by using them (apps/ai/src/evals/playground/README.md). --- apps/ai/scripts/tool-playground.sh | 21 + apps/ai/src/evals/playground/README.md | 101 +++ .../src/evals/playground/tool.playground.ts | 76 ++ apps/ai/src/evals/tools/tasks.ts | 60 ++ apps/ai/src/evals/tools/tools.eval.ts | 7 +- apps/ai/src/mcp/__evals__/infra-world.ts | 750 ++++++++++++++++++ apps/ai/src/mcp/lib/infra-inspect.ts | 482 +++++++++++ apps/ai/src/mcp/lib/infra-params.ts | 59 ++ apps/ai/src/mcp/lib/infra-render.ts | 263 ++++++ apps/ai/src/mcp/lib/infra.ts | 571 +++++++++++++ apps/ai/src/mcp/resources/instructions.ts | 1 + .../mcp/tools/__tests__/infra-tools.test.ts | 163 ++++ .../ai/src/mcp/tools/create-dashboard.test.ts | 1 + apps/ai/src/mcp/tools/create-dashboard.ts | 16 +- apps/ai/src/mcp/tools/diagnose-service.ts | 83 +- apps/ai/src/mcp/tools/inspect-infra.ts | 324 ++++++++ apps/ai/src/mcp/tools/list-infra.ts | 271 +++++++ apps/ai/src/mcp/tools/registry.ts | 4 + apps/ai/vitest.playground.config.ts | 20 + packages/domain/src/mcp-outputs/catalog.ts | 3 + packages/domain/src/mcp-outputs/index.ts | 1 + packages/domain/src/mcp-outputs/infra.ts | 114 +++ packages/domain/src/mcp-outputs/services.ts | 22 + .../query-engine/src/benchmark/builders.ts | 20 + packages/query-engine/src/ch/index.ts | 6 + packages/query-engine/src/ch/queries/infra.ts | 83 ++ 26 files changed, 3511 insertions(+), 11 deletions(-) create mode 100755 apps/ai/scripts/tool-playground.sh create mode 100644 apps/ai/src/evals/playground/README.md create mode 100644 apps/ai/src/evals/playground/tool.playground.ts create mode 100644 apps/ai/src/mcp/__evals__/infra-world.ts create mode 100644 apps/ai/src/mcp/lib/infra-inspect.ts create mode 100644 apps/ai/src/mcp/lib/infra-params.ts create mode 100644 apps/ai/src/mcp/lib/infra-render.ts create mode 100644 apps/ai/src/mcp/lib/infra.ts create mode 100644 apps/ai/src/mcp/tools/__tests__/infra-tools.test.ts create mode 100644 apps/ai/src/mcp/tools/inspect-infra.ts create mode 100644 apps/ai/src/mcp/tools/list-infra.ts create mode 100644 apps/ai/vitest.playground.config.ts create mode 100644 packages/domain/src/mcp-outputs/infra.ts diff --git a/apps/ai/scripts/tool-playground.sh b/apps/ai/scripts/tool-playground.sh new file mode 100755 index 000000000..0dd13dd06 --- /dev/null +++ b/apps/ai/scripts/tool-playground.sh @@ -0,0 +1,21 @@ +#!/usr/bin/env bash +# Call Maple's MCP tools against a fixture world and print what a model would read. +# +# scripts/tool-playground.sh catalog +# scripts/tool-playground.sh schema inspect_infra +# scripts/tool-playground.sh call list_infra '{"kind":"pods"}' +# +# PLAYGROUND_LOG= appends every call for grading. See src/evals/playground/. +set -euo pipefail +cd "$(dirname "$0")/.." + +out="$(mktemp -t maple-playground)" +PLAYGROUND_CMD="${1:-}" PLAYGROUND_TOOL="${2:-}" PLAYGROUND_ARGS="${3:-{\}}" PLAYGROUND_OUT="$out" \ + bunx vitest run --config vitest.playground.config.ts --reporter=dot --silent >/dev/null 2>"$out.err" || { + echo "playground run failed:" >&2 + tail -40 "$out.err" >&2 + exit 1 +} +cat "$out" +echo +rm -f "$out" "$out.err" diff --git a/apps/ai/src/evals/playground/README.md b/apps/ai/src/evals/playground/README.md new file mode 100644 index 000000000..ec212568a --- /dev/null +++ b/apps/ai/src/evals/playground/README.md @@ -0,0 +1,101 @@ +# Tool playground: agents testing tools by using them + +The model-driven evals in `src/evals/` grade which tool a model picks for one request. This +playground answers a different question: given a realistic task and only Maple's tools, how +efficiently does an agent get to the right answer, and where do the tools slow it down? + +An agent (a Claude Code subagent, or any agent with a shell) plays the Maple agent. It may only +call tools through `scripts/tool-playground.sh`, which runs the real registry, decoding and +rendering against a fixture warehouse, and prints exactly what a model would read: + +```sh +cd apps/ai +scripts/tool-playground.sh catalog # every public tool, as a model sees it +scripts/tool-playground.sh schema inspect_infra # one tool's input schema +PLAYGROUND_LOG=/tmp/run.jsonl scripts/tool-playground.sh call list_infra '{"kind":"pods"}' +``` + +Each call is one vitest run (about 5s) on a fresh database, so writes do not persist between +calls: a dashboard created in one call cannot be read back in the next. Tell agents that in a +scenario that writes. The log records tool, arguments, outcome and size. + +## The infra world + +`src/mcp/__evals__/infra-world.ts`, one Kubernetes cluster (`prod-eu`) plus Docker hosts: + +| Fact | Where | +| ---------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------- | +| `checkout-6f9c7d-k2x8p` (deployment `checkout`, ns `shop`, node `ip-10-0-3-57`) peaks at 98% of its memory limit 70% through the window, then drops: an OOM kill | pods, workloads, `diagnose_service checkout` | +| `otel-agent` daemonset pods run with no CPU or memory limits | pods, workloads | +| `ip-10-0-3-57` is the busy node (3.62 cores in use, up 1 day) and hosts the OOM pod, `postgres-0`, an api-gateway pod and a cart pod | nodes | +| `ci-runner-01` host: CPU 91%, disk 93%, load 7.8; its `buildkitd` container peaks at 97% CPU and restarted twice | hosts, containers | +| `bastion-01` host is idle | hosts | + +Traces, logs and errors are empty in this world. A scenario that needs them is not ready. + +## Scenarios + +Each is written the way a user would ask, with no tool or parameter names. `Must find` is graded +against the final answer; `Efficient path` is the call count a well-designed tool set allows. + +| # | Ask | Must find | Efficient path | +| --- | ------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------- | -------------- | +| 1 | Is anything in our infrastructure running hot right now? | OOM-risk checkout pod (98% memory), buildkitd at 97% CPU, ci-runner-01 disk 93% | 1-2 | +| 2 | Checkout has been flaky for the last few hours. Could it be a resource problem? | checkout pod hit 98% of its memory limit, when it peaked, it is one of 3 pods | 2-3 | +| 3 | Which of our pods have no resource limits? | both otel-agent pods (daemonset, ns observability) | 1 | +| 4 | What runs on node ip-10-0-3-57, and is it overloaded? | its pods incl. the checkout pod at its memory limit; node uses 3.62 cores | 1-2 | +| 5 | Our CI builds got slow. Can you check the CI machine? | ci-runner-01: CPU 91%, disk 93%, buildkitd at 97% CPU with 2 restarts | 2-3 | +| 6 | How has the postgres pod's memory looked over the last day? | postgres-0, about 72% of its limit, peak about 79% | 1-2 | +| 7 | Build me a dashboard with CPU utilization per host. | create_dashboard on source=metrics, `system.cpu.utilization`, group by `resource.host.name` | 1-3 | +| 8 | How many 5xx errors did the API return today? (negative) | does not use the infra tools; traces/errors tools instead | n/a | + +## Grading + +Per scenario, from the agent's report and the JSONL log: + +- **Correct**: every `Must find` fact is in the answer, nothing invented. +- **Calls**: total calls, and wasted calls (invalid input, an empty result the agent then had to + widen, `run_sql` used where an infra tool fits, a call repeated with the same arguments). +- **Friction**: what the agent says confused it: a misleading description, a missing filter, + a value it could not interpret (fraction or percent? of what limit?), a `next` hint that led + nowhere. + +A tool set is done when every scenario is correct on both a strong and a small model, within its +efficient path, with no friction note pointing at the tool rather than the task. Fix the tool +(description, parameters, rendering, `next` hints), never the scenario, and rerun the round. + +## Running a round with subagents + +Give each subagent one scenario and these rules verbatim (subagents do not inherit them): + +1. You are an agent whose only tools are Maple's MCP tools. Call them only with + `apps/ai/scripts/tool-playground.sh call ''` (use `catalog` and `schema` first). + Do not read the repository's source, tests or fixtures: you would be grading the code, not + using the tools. +2. Run commands with a timeout of at most 150000 ms, one call at a time. +3. Answer the user's question, then report: every call you made and why, which calls were + wasted, and what in the tools' descriptions or output confused you or was missing. + +Set `PLAYGROUND_LOG` to a per-scenario file so the log can be graded alongside the report. +Run at most four subagents at once: each call starts a vitest process. + +## Results so far (Haiku unless noted; data calls, excluding `catalog`/`schema`) + +| # | Round 1 | Round 2 | Round 3 | Round 4 | Status | +| --- | --- | --- | --- | --- | --- | +| 1 | 2 | 2 | | 1 (Sonnet) | done | +| 2 | 3 | 4 | 3 | 3 | done; 2 is reachable (list_infra, then inspect_infra workload) | +| 3 | 1 | | 1 | | done | +| 4 | | 2 | 3 | 4 | correct; call count varies on Haiku | +| 5 | 3 | | 2 | | done | +| 6 | | 2 | 2 | | done | +| 7 | | | 10, then 7 (broken dashboard) | 3 | done after the sandbox modeled metric discovery | +| 8 | | | correct, no infra calls | | done | + +Every answer was correct from round 2 on. The open items the agents still raise: + +- Node capacity is not collected (`k8s.node.allocatable_cpu` from the k8s_cluster receiver would + answer "is this node full?"). +- Pod restarts have no termination reason; the OOM verdict is inferred from the memory curve. +- `create_dashboard` simple specs cannot filter `system.cpu.utilization` to non-idle states, so a + "CPU per host" chart averages across states; units for `widgets` are not enumerated. diff --git a/apps/ai/src/evals/playground/tool.playground.ts b/apps/ai/src/evals/playground/tool.playground.ts new file mode 100644 index 000000000..e0f469863 --- /dev/null +++ b/apps/ai/src/evals/playground/tool.playground.ts @@ -0,0 +1,76 @@ +/** + * One MCP tool call against a fixture world, for agents testing the tools by using them: the real + * registry, decoding and rendering, with only the warehouse faked. Driven by + * `scripts/tool-playground.sh`, which passes the request in the environment: + * + * PLAYGROUND_CMD catalog | schema | call + * PLAYGROUND_TOOL the tool, for schema and call + * PLAYGROUND_ARGS JSON arguments, for call + * PLAYGROUND_OUT where the text goes (what the model would read) + * PLAYGROUND_LOG optional JSONL file; every call is appended for grading + */ +import { appendFileSync, writeFileSync } from "node:fs" +import { afterAll, beforeAll, it } from "vitest" +import { installFakeWarehouse, restoreWarehouse } from "@/mcp/__evals__/fake-warehouse" +import { makeEvalRuntime, markdown, runToolDirect, type EvalRuntime } from "@/mcp/__evals__/eval-runtime" +import { infraFixtureRules } from "@/mcp/__evals__/infra-world" +import { inputSchemaOf, mapleToolCatalog } from "@/mcp/tools/registry" + +const env = process.env +const out = env.PLAYGROUND_OUT ?? "/tmp/maple-tool-playground.txt" +const publicTools = mapleToolCatalog.filter((entry) => entry.audience === "public") + +let rt: EvalRuntime | undefined +beforeAll(() => { + if (env.PLAYGROUND_CMD !== "call") return + installFakeWarehouse(infraFixtureRules(), undefined, "empty") + rt = makeEvalRuntime() +}) +afterAll(async () => { + restoreWarehouse() + await rt?.dispose() +}) + +const log = (entry: Record) => { + if (env.PLAYGROUND_LOG) + appendFileSync(env.PLAYGROUND_LOG, `${JSON.stringify({ at: new Date().toISOString(), ...entry })}\n`) +} + +it("playground", async () => { + const tool = env.PLAYGROUND_TOOL ?? "" + switch (env.PLAYGROUND_CMD) { + case "catalog": + return writeFileSync(out, publicTools.map((e) => `- ${e.name}: ${e.description}`).join("\n\n")) + case "schema": { + const entry = publicTools.find((candidate) => candidate.name === tool) + return writeFileSync( + out, + entry === undefined + ? `Unknown tool "${tool}".` + : JSON.stringify(inputSchemaOf(entry), null, 2), + ) + } + case "call": { + if (rt === undefined) return writeFileSync(out, "The playground runtime did not start.") + const started = Date.now() + const params: unknown = JSON.parse(env.PLAYGROUND_ARGS ?? "{}") + const result = await runToolDirect(rt, tool, params).then( + (value) => ({ text: markdown(value), ok: value?.isError !== true }), + (error: unknown) => ({ text: `Tool failed: ${String(error)}`, ok: false }), + ) + writeFileSync(out, result.text) + return log({ + tool, + args: params, + ok: result.ok, + chars: result.text.length, + ms: Date.now() - started, + }) + } + default: + return writeFileSync( + out, + "usage: tool-playground.sh catalog | schema | call ''", + ) + } +}) diff --git a/apps/ai/src/evals/tools/tasks.ts b/apps/ai/src/evals/tools/tasks.ts index f9d3f4419..e7d1243d2 100644 --- a/apps/ai/src/evals/tools/tasks.ts +++ b/apps/ai/src/evals/tools/tasks.ts @@ -11,6 +11,7 @@ * and move once they pass every trial on every target model. */ import { FIXTURES } from "../../mcp/__evals__/utils" +import { INFRA_WORLD } from "../../mcp/__evals__/infra-world" import type { EvalTask } from "../runner" import { answerMentions, call, calls, never, noTools, type Check } from "../targets" @@ -510,6 +511,65 @@ export const TOOL_TASKS: ReadonlyArray = [ call("find_errors", { service: SVC.api }), ), ), + // Infrastructure, in the world of `infra-world.ts`; see src/evals/playground/README.md. + withRoom( + capability( + "infra-running-hot", + ["infra", "fuzzy"], + "Is anything in our infrastructure running hot right now?", + calls(call("list_infra")), + answerMentions(INFRA_WORLD.oomPod, INFRA_WORLD.hotContainer), + ), + ), + capability( + "infra-no-limits", + ["infra"], + "Which of our pods have no resource limits set?", + calls( + call("list_infra", { kind: "pods" }), + call("list_infra", { kind: "workloads" }), + call("list_infra"), + ), + ), + capability( + "infra-node-contents", + ["infra"], + `What runs on node ${INFRA_WORLD.hotNode}, and is it overloaded?`, + calls( + call("inspect_infra", { kind: "node", name: INFRA_WORLD.hotNode }), + call("list_infra", { kind: "pods", node: INFRA_WORLD.hotNode }), + ), + ), + withRoom( + capability( + "infra-service-resources", + ["infra", "services", "fuzzy"], + "Our checkout service has been flaky for the last few hours. Could it be a resource problem?", + calls( + call("diagnose_service", { service: "checkout" }), + call("inspect_infra", { kind: "workload", name: "checkout" }), + call("inspect_infra", { kind: "pod", name: INFRA_WORLD.oomPod }), + call("list_infra", { kind: "pods" }), + ), + answerMentions(INFRA_WORLD.oomPod, "98"), + ), + ), + capability( + "infra-ci-host", + ["infra", "fuzzy"], + `Our CI builds got slow. Can you check ${INFRA_WORLD.hotHost}?`, + calls( + call("inspect_infra", { kind: "host", name: INFRA_WORLD.hotHost }), + call("list_infra", { kind: "hosts" }), + call("list_infra"), + ), + ), + capability( + "infra-not-for-errors", + ["infra", "negative"], + `How many server errors did the ${SVC.api} service return in the last hour?`, + never("list_infra", "inspect_infra"), + ), capability( "user-reported-failure", ["traces", "fuzzy"], diff --git a/apps/ai/src/evals/tools/tools.eval.ts b/apps/ai/src/evals/tools/tools.eval.ts index b3e9719a4..1c1ce4422 100644 --- a/apps/ai/src/evals/tools/tools.eval.ts +++ b/apps/ai/src/evals/tools/tools.eval.ts @@ -22,6 +22,7 @@ import { evalModel, hasEvalCredentials } from "../model" import { renderSummary, runSettings, runSuite, selectTasks } from "../runner" import { grade } from "../targets" import { seedWorld, worldRules } from "../world" +import { infraFixtureRules } from "../../mcp/__evals__/infra-world" import { TOOL_TASKS } from "./tasks" const DEFAULT_MAX_TOOL_CALLS = 6 @@ -32,7 +33,11 @@ let executor: McpToolExecutorApi | undefined beforeAll(async () => { if (!hasEvalCredentials()) return // The world answers the probes; the trace fixtures answer the trace tasks; anything else is empty. - installFakeWarehouse([...worldRules(), ...defaultTraceFixtures()], undefined, "empty") + installFakeWarehouse( + [...worldRules(), ...defaultTraceFixtures(), ...infraFixtureRules()], + undefined, + "empty", + ) rt = makeEvalRuntime() await seedWorld(rt) executor = await rt.runtime.runPromise(Effect.service(McpToolExecutor)) diff --git a/apps/ai/src/mcp/__evals__/infra-world.ts b/apps/ai/src/mcp/__evals__/infra-world.ts new file mode 100644 index 000000000..438b44262 --- /dev/null +++ b/apps/ai/src/mcp/__evals__/infra-world.ts @@ -0,0 +1,750 @@ +/** + * A small infrastructure world for the infra tools' tests and usability runs: one Kubernetes + * cluster where a checkout pod runs out of memory, a CI host whose build container pins the CPU, + * and a quiet bastion. Every entity is a curve over the window; list aggregates (avg, peak) and + * time series are both sampled from it, so the two views agree the way real data would. Rows + * answer the compiled SQL, honoring its name filters, sort and limit. + */ +import { parseWarehouseDateTime } from "@maple/query-engine" +import type { FixtureRule } from "./fake-warehouse" + +export const INFRA_WORLD = { + cluster: "prod-eu", + environment: "production", + /** Climbs to 98% of its memory limit 70% through the window, then drops: an OOM kill. */ + oomPod: "checkout-6f9c7d-k2x8p", + hotNode: "ip-10-0-3-57", + hotHost: "ci-runner-01", + hotContainer: "buildkitd", + unboundedWorkload: "otel-agent", +} as const + +/** Value at t in [0, 1) across the window. */ +type Curve = (t: number) => number + +const SAMPLES = 60 +const samples = (curve: Curve) => Array.from({ length: SAMPLES }, (_, i) => curve(i / SAMPLES)) +const avgOf = (curve: Curve) => samples(curve).reduce((sum, v) => sum + v, 0) / SAMPLES +const peakOf = (curve: Curve) => Math.max(...samples(curve)) + +/** Three full periods across the window, so the mean is `base` and the peak `base * (1 + amp)`. */ +const wave = + (base: number, amp = 0.08): Curve => + (t) => + base * (1 + amp * Math.sin(2 * Math.PI * 3 * t)) +const OOM_PEAK_AT = 0.7 +const oom: Curve = (t) => (t <= OOM_PEAK_AT ? 0.55 + (0.98 - 0.55) * (t / OOM_PEAK_AT) : 0.45) + +type OwnerKind = "deployment" | "statefulset" | "daemonset" + +interface Pod { + readonly podName: string + readonly namespace: string + readonly nodeName: string + readonly owner: readonly [OwnerKind, string] + readonly cpu: Curve + /** 0 when no limit is set. */ + readonly cpuLimitCores: number + /** Fraction of the memory limit; never sampled when no limit is set. */ + readonly memory: Curve + readonly unbounded: boolean +} + +const pod = ( + podName: string, + namespace: string, + nodeName: string, + owner: Pod["owner"], + cores: number, + cpuLimitCores: number, + memory: Curve, +): Pod => ({ + podName, + namespace, + nodeName, + owner, + cpu: wave(cores, 0.25), + cpuLimitCores, + memory, + unbounded: cpuLimitCores === 0, +}) + +const PODS: ReadonlyArray = [ + pod("checkout-6f9c7d-k2x8p", "shop", "ip-10-0-3-57", ["deployment", "checkout"], 0.41, 1, oom), + pod("checkout-6f9c7d-p9wq2", "shop", "ip-10-0-3-21", ["deployment", "checkout"], 0.33, 1, wave(0.58)), + pod("checkout-6f9c7d-z7m4c", "shop", "ip-10-0-3-21", ["deployment", "checkout"], 0.35, 1, wave(0.55)), + pod("cart-5b8d9-h3j5k", "shop", "ip-10-0-3-21", ["deployment", "cart"], 0.12, 0.5, wave(0.31)), + pod("cart-5b8d9-r2t6y", "shop", "ip-10-0-3-57", ["deployment", "cart"], 0.11, 0.5, wave(0.3)), + pod( + "api-gateway-7c4f-a1b2c", + "edge", + "ip-10-0-3-57", + ["deployment", "api-gateway"], + 0.62, + 1.25, + wave(0.44), + ), + pod( + "api-gateway-7c4f-d3e4f", + "edge", + "ip-10-0-3-21", + ["deployment", "api-gateway"], + 0.58, + 1.25, + wave(0.42), + ), + pod("postgres-0", "data", "ip-10-0-3-57", ["statefulset", "postgres"], 0.9, 2, wave(0.72)), + pod("otel-agent-x8k2l", "observability", "ip-10-0-3-21", ["daemonset", "otel-agent"], 0.22, 0, () => 0), + pod("otel-agent-q4w9e", "observability", "ip-10-0-3-57", ["daemonset", "otel-agent"], 0.27, 0, () => 0), +] + +const NODES = [ + { nodeName: "ip-10-0-3-21", nodeUid: "4f1c-21", kubeletVersion: "v1.31.2", uptime: 1_814_400 }, + { nodeName: "ip-10-0-3-57", nodeUid: "9a7e-57", kubeletVersion: "v1.31.2", uptime: 86_400 }, +] +/** System daemons outside any pod. */ +const NODE_OVERHEAD_CORES = 0.3 +const nodeCpu = + (nodeName: string): Curve => + (t) => + PODS.filter((p) => p.nodeName === nodeName).reduce((sum, p) => sum + p.cpu(t), NODE_OVERHEAD_CORES) + +const HOSTS = [ + { + hostName: "ci-runner-01", + osType: "linux", + hostArch: "amd64", + cpu: wave(0.91, 0.04), + memory: wave(0.64), + load15: wave(7.8, 0.1), + filesystems: [ + { mountpoint: "/var/lib/docker", device: "/dev/nvme1n1", used: 0.93 }, + { mountpoint: "/", device: "/dev/nvme0n1p1", used: 0.41 }, + ], + }, + { + hostName: "bastion-01", + osType: "linux", + hostArch: "arm64", + cpu: wave(0.04, 0.5), + memory: wave(0.21), + load15: wave(0.05, 0.5), + filesystems: [{ mountpoint: "/", device: "/dev/nvme0n1p1", used: 0.38 }], + }, +] + +const CONTAINERS = [ + { + containerName: "buildkitd", + hostName: "ci-runner-01", + imageName: "moby/buildkit:v0.17.0", + composeService: "buildkit", + cpu: wave(0.8, 0.22), + memory: wave(0.52, 0.12), + cpuLimitCores: 4, + restarts: 2, + }, + { + containerName: "registry-cache", + hostName: "ci-runner-01", + imageName: "registry:2", + composeService: "registry", + cpu: wave(0.06, 0.5), + memory: wave(0.4, 0.03), + cpuLimitCores: 0, + restarts: 0, + }, +] + +/** The window the SQL asked for, from its first two DateTime literals. */ +const windowOf = (sql: string): { start: number; end: number } => { + const stamps = [...sql.matchAll(/'(\d{4}-\d\d-\d\d \d\d:\d\d:\d\d)'/g)].map((m) => + parseWarehouseDateTime(m[1] ?? ""), + ) + const start = stamps[0] ?? parseWarehouseDateTime("2026-10-01 06:00:00") + const end = stamps[1] ?? start + 6 * 3_600_000 + return { start, end } +} + +const fmt = (ms: number): string => new Date(ms).toISOString().slice(0, 19).replace("T", " ") +const lastSeen = (sql: string) => fmt(windowOf(sql).end - 30_000) +const firstSeen = (sql: string) => fmt(windowOf(sql).start + 30_000) + +/** Values for `ResourceAttributes['key'] = 'v'` or `IN ('a', 'b')` in the SQL; undefined when unfiltered. */ +const filterValues = (sql: string, key: string): ReadonlyArray | undefined => { + const escaped = key.replace(/\./g, "\\.") + const eq = [...sql.matchAll(new RegExp(`\\['${escaped}'\\] = '([^']*)'`, "g"))].map((m) => m[1] ?? "") + const inList = [...sql.matchAll(new RegExp(`\\['${escaped}'\\] IN \\(([^)]*)\\)`, "g"))].flatMap((m) => + [...(m[1] ?? "").matchAll(/'([^']*)'/g)].map((v) => v[1] ?? ""), + ) + const all = [...eq, ...inList] + return all.length === 0 ? undefined : all +} + +const matches = (sql: string, key: string, value: string): boolean => { + const wanted = filterValues(sql, key) + return wanted === undefined || wanted.includes(value) +} + +const searched = (sql: string, name: string): boolean => { + const term = /positionCaseInsensitive\([^,]+, '([^']*)'\)/.exec(sql)?.[1]?.toLowerCase() + return term === undefined || name.toLowerCase().includes(term) +} + +const limitOf = (sql: string): number => Number([...sql.matchAll(/LIMIT (\d+)/g)].at(-1)?.[1] ?? 1000) + +const OWNER_KINDS: ReadonlyArray = ["deployment", "statefulset", "daemonset"] + +const podsFor = (sql: string): ReadonlyArray => + PODS.filter( + (p) => + matches(sql, "k8s.pod.name", p.podName) && + matches(sql, "k8s.namespace.name", p.namespace) && + matches(sql, "k8s.node.name", p.nodeName) && + matches(sql, "k8s.cluster.name", INFRA_WORLD.cluster) && + // An owner filter of any kind selects only pods owned by that kind and name. + OWNER_KINDS.every((kind) => { + const wanted = filterValues(sql, `k8s.${kind}.name`) + return wanted === undefined || (p.owner[0] === kind && wanted.includes(p.owner[1])) + }) && + searched(sql, p.podName), + ) + +const podStats = (p: Pod) => { + const cpuUsage = avgOf(p.cpu) + const cpuUsagePeak = peakOf(p.cpu) + const ofLimit = (cores: number) => (p.cpuLimitCores === 0 ? 0 : cores / p.cpuLimitCores) + const memoryLimitPct = p.unbounded ? 0 : avgOf(p.memory) + const memoryLimitPctPeak = p.unbounded ? 0 : peakOf(p.memory) + return { + cpuUsage, + cpuUsagePeak, + cpuLimitPct: ofLimit(cpuUsage), + cpuLimitPctPeak: ofLimit(cpuUsagePeak), + memoryLimitPct, + memoryLimitPctPeak, + saturation: Math.max(ofLimit(cpuUsagePeak), memoryLimitPctPeak), + } +} + +const podListRow = (sql: string, p: Pod) => { + const stats = podStats(p) + return { + podName: p.podName, + namespace: p.namespace, + nodeName: p.nodeName, + clusterName: INFRA_WORLD.cluster, + environment: INFRA_WORLD.environment, + deploymentName: p.owner[0] === "deployment" ? p.owner[1] : "", + statefulsetName: p.owner[0] === "statefulset" ? p.owner[1] : "", + daemonsetName: p.owner[0] === "daemonset" ? p.owner[1] : "", + jobName: "", + qosClass: p.unbounded ? "BestEffort" : "Burstable", + podUid: `uid-${p.podName}`, + computeType: "", + lastSeen: lastSeen(sql), + ...stats, + cpuRequestPct: stats.cpuLimitPct * 2, + memoryRequestPct: stats.memoryLimitPct * 1.5, + } +} + +const sortedPods = (sql: string) => { + const rows = podsFor(sql).map((p) => podListRow(sql, p)) + // The first key of the last ORDER BY decides; later keys are tiebreaks. + const key = /ORDER BY\s+(?:\w+\.)?(\w+)/i.exec(sql.slice(sql.lastIndexOf("ORDER BY")))?.[1] + if (key === "podName") return rows.sort((a, b) => a.podName.localeCompare(b.podName)) + if (key === "cpuUsage") return rows.sort((a, b) => b.cpuUsage - a.cpuUsage) + if (key === "memoryLimitPct") return rows.sort((a, b) => b.memoryLimitPct - a.memoryLimitPct) + return rows.sort((a, b) => b.saturation - a.saturation) +} + +/** Buckets across the SQL's window, sampled at the same points as the list aggregates. */ +const series = (sql: string, curve: Curve, attributeValue = "") => { + const { start, end } = windowOf(sql) + const step = (end - start) / SAMPLES + return Array.from({ length: SAMPLES }, (_, i) => ({ + bucket: fmt(start + i * step), + attributeValue, + avgValue: curve(i / SAMPLES), + })) +} + +const metricOf = (sql: string): string => /MetricName = '([^']+)'/.exec(sql)?.[1] ?? "" +const meanCurve = + (curves: ReadonlyArray): Curve => + (t) => + curves.reduce((sum, curve) => sum + curve(t), 0) / Math.max(1, curves.length) + +const podSeries = (sql: string): ReturnType => { + const pods = podsFor(sql) + if (pods.length === 0) return [] + // A workload chart grouped by pod: one series per pod, labeled with its name. + if (sql.includes("['k8s.pod.name'] AS attributeValue")) { + return pods.flatMap((p) => + podSeries(sql.replace("['k8s.pod.name'] AS attributeValue", "")).length === 0 + ? [] + : seriesForPod(sql, p), + ) + } + const bounded = pods.filter((p) => !p.unbounded) + switch (metricOf(sql)) { + case "k8s.pod.cpu.usage": + return series(sql, meanCurve(pods.map((p) => p.cpu))) + case "k8s.pod.cpu_limit_utilization": + return bounded.length === 0 + ? [] + : series(sql, meanCurve(bounded.map((p) => (t: number) => p.cpu(t) / p.cpuLimitCores))) + case "k8s.pod.memory_limit_utilization": + return bounded.length === 0 ? [] : series(sql, meanCurve(bounded.map((p) => p.memory))) + case "k8s.pod.memory.working_set": + return series(sql, meanCurve(pods.map(workingSet))) + default: + return [] + } +} + +const seriesForPod = (sql: string, p: Pod) => { + if (p.unbounded) return [] + switch (metricOf(sql)) { + case "k8s.pod.cpu.usage": + return series(sql, p.cpu, p.podName) + case "k8s.pod.cpu_limit_utilization": + return series(sql, (t) => p.cpu(t) / p.cpuLimitCores, p.podName) + case "k8s.pod.memory_limit_utilization": + return series(sql, p.memory, p.podName) + case "k8s.pod.memory.working_set": + return series(sql, workingSet(p), p.podName) + default: + return [] + } +} + +/** The metrics this world reports, for metric discovery (`list_metrics`) and the query builder. */ +interface WorldMetric { + readonly name: string + readonly type: "gauge" | "sum" + readonly unit: string + readonly service: string + /** Resource key that names an entity, and each entity's value over the window. */ + readonly key: string + readonly entities: ReadonlyArray +} + +const WORLD_METRICS: ReadonlyArray = [ + { + name: "system.cpu.utilization", + type: "gauge", + unit: "1", + service: "otel-collector", + key: "host.name", + entities: HOSTS.map((h) => [h.hostName, h.cpu]), + }, + { + name: "system.memory.utilization", + type: "gauge", + unit: "1", + service: "otel-collector", + key: "host.name", + entities: HOSTS.map((h) => [h.hostName, h.memory]), + }, + { + name: "system.memory.usage", + type: "gauge", + unit: "By", + service: "otel-collector", + key: "host.name", + entities: HOSTS.map((h) => [h.hostName, (t: number) => h.memory(t) * 16 * 2 ** 30]), + }, + { + name: "system.cpu.load_average.1m", + type: "gauge", + unit: "{thread}", + service: "otel-collector", + key: "host.name", + entities: HOSTS.map((h) => [h.hostName, wave(avgOf(h.load15), 0.3)]), + }, + { + name: "system.disk.io", + type: "sum", + unit: "By", + service: "otel-collector", + key: "host.name", + entities: HOSTS.map((h) => [h.hostName, wave(h.hostName === "ci-runner-01" ? 9e7 : 2e5, 0.5)]), + }, + { + name: "system.network.io", + type: "sum", + unit: "By", + service: "otel-collector", + key: "host.name", + entities: HOSTS.map((h) => [h.hostName, wave(h.hostName === "ci-runner-01" ? 4e7 : 1e5, 0.5)]), + }, + { + name: "system.cpu.load_average.15m", + type: "gauge", + unit: "{thread}", + service: "otel-collector", + key: "host.name", + entities: HOSTS.map((h) => [h.hostName, h.load15]), + }, + { + name: "system.filesystem.utilization", + type: "gauge", + unit: "1", + service: "otel-collector", + key: "host.name", + entities: HOSTS.map((h) => [h.hostName, () => Math.max(...h.filesystems.map((fs) => fs.used))]), + }, + { + name: "k8s.pod.cpu.usage", + type: "gauge", + unit: "{cpu}", + service: "k8s-agent", + key: "k8s.pod.name", + entities: PODS.map((p) => [p.podName, p.cpu]), + }, + { + name: "k8s.pod.memory_limit_utilization", + type: "gauge", + unit: "1", + service: "k8s-agent", + key: "k8s.pod.name", + entities: PODS.filter((p) => !p.unbounded).map((p) => [p.podName, p.memory]), + }, + { + name: "k8s.pod.cpu_limit_utilization", + type: "gauge", + unit: "1", + service: "k8s-agent", + key: "k8s.pod.name", + entities: PODS.filter((p) => !p.unbounded).map((p) => [ + p.podName, + (t: number) => p.cpu(t) / p.cpuLimitCores, + ]), + }, + { + name: "k8s.node.cpu.usage", + type: "gauge", + unit: "{cpu}", + service: "k8s-agent", + key: "k8s.node.name", + entities: NODES.map((n) => [n.nodeName, nodeCpu(n.nodeName)]), + }, + { + name: "container.cpu.utilization", + type: "gauge", + unit: "%", + service: "docker-agent", + key: "container.name", + entities: CONTAINERS.map((c) => [c.containerName, (t: number) => c.cpu(t) * 100]), + }, + { + name: "container.memory.percent", + type: "gauge", + unit: "%", + service: "docker-agent", + key: "container.name", + entities: CONTAINERS.map((c) => [c.containerName, (t: number) => c.memory(t) * 100]), + }, +] + +/** A query-builder metric series: one row per bucket and group, grouped by a resource key or not at all. */ +const builderSeries = (sql: string) => { + const metric = WORLD_METRICS.find((m) => m.name === metricOf(sql)) + if (metric === undefined) return [] + const groupKey = /ResourceAttributes\['([^']+)'\] AS groupName/.exec(sql)?.[1] + const groups: ReadonlyArray = + groupKey === metric.key + ? metric.entities.filter(([name]) => matches(sql, metric.key, name)) + : [["", meanCurve(metric.entities.map(([, curve]) => curve))]] + return groups.flatMap(([group, curve]) => + series(sql, curve, group).map((row) => ({ + bucket: row.bucket, + serviceName: metric.service, + attributeValue: group, + groupName: group, + avgValue: row.avgValue, + minValue: row.avgValue, + maxValue: row.avgValue, + sumValue: row.avgValue, + dataPointCount: 4, + })), + ) +} + +const metricCatalogRows = (sql: string) => { + const term = /ilike\([^,]+, '%([^%']*)%'\)/i.exec(sql)?.[1]?.toLowerCase() + return WORLD_METRICS.filter((m) => term === undefined || m.name.toLowerCase().includes(term)).map( + (m) => ({ + metricName: m.name, + metricType: m.type, + serviceName: m.service, + metricDescription: "", + metricUnit: m.unit, + dataPointCount: 14_400, + firstSeen: firstSeen(sql), + lastSeen: lastSeen(sql), + isMonotonic: 0, + }), + ) +} + +/** Memory limits: 512 MiB for checkout and cart, 4 GiB for postgres; unlimited pods sit near 150 MiB. */ +const MEMORY_LIMIT_BYTES = new Map([ + ["checkout", 512 * 2 ** 20], + ["cart", 512 * 2 ** 20], + ["postgres", 4 * 2 ** 30], + ["api-gateway", 2 ** 30], +]) +const workingSet = + (p: Pod): Curve => + (t) => + p.unbounded ? 150 * 2 ** 20 : p.memory(t) * (MEMORY_LIMIT_BYTES.get(p.owner[1]) ?? 2 ** 30) + +/** Container restarts: the OOM-killed checkout pod restarted once in the window. */ +const RESTARTS = new Map([[INFRA_WORLD.oomPod, 1]]) + +const has = (sql: string, alias: string) => sql.includes(` AS ${alias}`) + +const workloadRows = (sql: string) => { + const kind = OWNER_KINDS.find((k) => sql.includes(`['k8s.${k}.name']`)) + const owned = podsFor(sql).filter((p) => p.owner[0] === kind && searched(sql, p.owner[1])) + const names = [...new Set(owned.map((p) => p.owner[1]))] + return names.map((name) => { + const pods = owned.filter((p) => p.owner[1] === name) + const stats = pods.map(podStats) + const mean = (f: (s: (typeof stats)[number]) => number) => + stats.reduce((sum, s) => sum + f(s), 0) / stats.length + return { + workloadName: name, + kind, + namespace: pods[0]?.namespace ?? "", + clusterName: INFRA_WORLD.cluster, + environment: INFRA_WORLD.environment, + podCount: pods.length, + firstSeen: firstSeen(sql), + lastSeen: lastSeen(sql), + avgCpuLimitPct: mean((s) => s.cpuLimitPct), + avgMemoryLimitPct: mean((s) => s.memoryLimitPct), + avgCpuUsage: mean((s) => s.cpuUsage), + } + }) +} + +const containerRow = (sql: string, c: (typeof CONTAINERS)[number]) => ({ + containerName: c.containerName, + hostName: c.hostName, + imageName: c.imageName, + composeService: c.composeService, + containerId: `c0ffee${c.containerName.length}`, + composeProject: "ci", + runtime: "docker", + environment: "ci", + firstSeen: firstSeen(sql), + lastSeen: lastSeen(sql), + cpuPct: avgOf(c.cpu), + cpuPctPeak: peakOf(c.cpu), + memoryPct: avgOf(c.memory), + memoryPctPeak: peakOf(c.memory), + cpuLimitCores: c.cpuLimitCores, + uptimeSeconds: 5_400, + saturation: Math.max(peakOf(c.cpu), peakOf(c.memory)), +}) + +export const infraFixtureRules = (): FixtureRule[] => [ + // Metric discovery and the query builder, so infra metrics can be found and charted. + { match: (sql) => has(sql, "groupName") && has(sql, "dataPointCount"), rows: builderSeries }, + { match: (sql) => has(sql, "isMonotonic"), rows: metricCatalogRows }, + { + match: (sql) => has(sql, "metricCount"), + rows: [ + { + metricType: "gauge", + metricCount: WORLD_METRICS.length, + dataPointCount: 14_400 * WORLD_METRICS.length, + }, + ], + }, + { + match: (sql) => has(sql, "surface"), + rows: ["hosts", "containers", "k8sPods", "k8sNodes", "k8sWorkloads"].map((surface) => ({ surface })), + }, + // Pod detail before pod list: both lead with podName. + { + match: (sql) => has(sql, "totalRestarts"), + rows: (sql) => + podsFor(sql).map((p) => ({ + podName: p.podName, + containerName: p.owner[1], + restarts: RESTARTS.get(p.podName) ?? 0, + totalRestarts: (RESTARTS.get(p.podName) ?? 0) + 2, + })), + }, + { + match: (sql) => has(sql, "podStartTime"), + rows: (sql) => + podsFor(sql) + .slice(0, 1) + .map((p) => ({ + ...podListRow(sql, p), + podStartTime: new Date(windowOf(sql).start - 3 * 86_400_000).toISOString(), + firstSeen: firstSeen(sql), + })), + }, + { + match: (sql) => has(sql, "livePods"), + rows: (sql) => { + const pods = sortedPods(sql) + return [ + { + livePods: pods.length, + endedPods: 0, + saturatedPods: pods.filter((p) => p.saturation >= 0.9).length, + elevatedPods: pods.filter((p) => p.saturation >= 0.6 && p.saturation < 0.9).length, + unboundedPods: pods.filter((p) => p.cpuLimitPct === 0 && p.memoryLimitPct === 0).length, + }, + ] + }, + }, + { + match: (sql) => has(sql, "podName") && has(sql, "cpuUsagePeak"), + rows: (sql) => sortedPods(sql).slice(0, limitOf(sql)), + }, + { + match: (sql) => has(sql, "avgCpuLimitUtilization"), + rows: (sql) => { + if (!sql.includes("'checkout'")) return [] + const stats = PODS.filter((p) => p.owner[1] === "checkout").map(podStats) + const mean = (f: (s: (typeof stats)[number]) => number) => + stats.reduce((sum, s) => sum + f(s), 0) / stats.length + return [ + { + serviceName: "checkout", + workloadKind: "deployment", + workloadName: "checkout", + namespace: "shop", + clusterName: INFRA_WORLD.cluster, + podCount: stats.length, + avgCpuLimitUtilization: mean((s) => s.cpuLimitPct), + avgMemoryLimitUtilization: mean((s) => s.memoryLimitPct), + }, + ] + }, + }, + { match: (sql) => has(sql, "workloadName"), rows: workloadRows }, + { + match: (sql) => has(sql, "kubeletVersion"), + rows: (sql) => + NODES.filter((n) => matches(sql, "k8s.node.name", n.nodeName) && searched(sql, n.nodeName)).map( + (n) => ({ + ...n, + cpuUsage: avgOf(nodeCpu(n.nodeName)), + clusterName: INFRA_WORLD.cluster, + environment: INFRA_WORLD.environment, + containerRuntime: "containerd://1.7.22", + firstSeen: firstSeen(sql), + lastSeen: lastSeen(sql), + }), + ), + }, + { + match: (sql) => has(sql, "totalContainers"), + rows: (sql) => { + const rows = CONTAINERS.filter((c) => matches(sql, "host.name", c.hostName)).map((c) => + containerRow(sql, c), + ) + return [ + { + totalContainers: rows.length, + saturatedContainers: rows.filter((c) => c.saturation >= 0.9).length, + elevatedContainers: rows.filter((c) => c.saturation >= 0.6 && c.saturation < 0.9).length, + staleContainers: 0, + }, + ] + }, + }, + { + match: (sql) => has(sql, "restartsDelta"), + rows: (sql) => + CONTAINERS.filter((c) => matches(sql, "container.name", c.containerName)).map((c) => ({ + memoryBytesAvg: 1_200_000_000, + memoryLimitBytes: 2_147_483_648, + restartsDelta: c.restarts, + pidsAvg: 48, + })), + }, + { + match: (sql) => has(sql, "composeService") || has(sql, "imageName"), + rows: (sql) => + CONTAINERS.filter( + (c) => + matches(sql, "container.name", c.containerName) && + matches(sql, "host.name", c.hostName) && + matches(sql, "compose.service", c.composeService) && + searched(sql, c.containerName), + ).map((c) => containerRow(sql, c)), + }, + { + match: (sql) => has(sql, "mountpoint"), + rows: (sql) => + HOSTS.filter((h) => matches(sql, "host.name", h.hostName)).flatMap((h) => + h.filesystems.map((fs) => ({ + mountpoint: fs.mountpoint, + device: fs.device, + usedAvg: fs.used - 0.01, + usedMax: fs.used, + })), + ), + }, + { + match: (sql) => has(sql, "load15") || has(sql, "cloudRegion"), + rows: (sql) => + HOSTS.filter((h) => matches(sql, "host.name", h.hostName) && searched(sql, h.hostName)).map( + (h) => ({ + hostName: h.hostName, + osType: h.osType, + hostArch: h.hostArch, + cpuPct: avgOf(h.cpu), + memoryPct: avgOf(h.memory), + diskPct: Math.max(...h.filesystems.map((fs) => fs.used)), + load15: avgOf(h.load15), + cloudProvider: "aws", + cloudRegion: "eu-central-1", + firstSeen: firstSeen(sql), + lastSeen: lastSeen(sql), + }), + ), + }, + { + match: (sql) => has(sql, "avgValue"), + rows: (sql) => { + const metric = metricOf(sql) + if (metric.startsWith("k8s.pod.")) return podSeries(sql) + if (metric === "k8s.node.cpu.usage") { + const node = NODES.find((n) => n.nodeName === filterValues(sql, "k8s.node.name")?.[0]) + return node === undefined ? [] : series(sql, nodeCpu(node.nodeName)) + } + const container = CONTAINERS.find( + (c) => c.containerName === filterValues(sql, "container.name")?.[0], + ) + if (metric === "container.cpu.utilization" && container) return series(sql, container.cpu) + if (metric === "container.memory.percent" && container) return series(sql, container.memory) + const host = HOSTS.find((h) => h.hostName === filterValues(sql, "host.name")?.[0]) + if (host === undefined) return [] + if (metric === "system.cpu.utilization") + return [ + ...series(sql, (t) => 1 - host.cpu(t), "idle"), + ...series(sql, (t) => host.cpu(t) * 0.8, "user"), + ...series(sql, (t) => host.cpu(t) * 0.2, "system"), + ] + if (metric === "system.memory.utilization") + return [ + ...series(sql, host.memory, "used"), + ...series(sql, (t) => 1 - host.memory(t), "free"), + ] + if (metric === "system.cpu.load_average.15m") return series(sql, host.load15) + return [] + }, + }, +] diff --git a/apps/ai/src/mcp/lib/infra-inspect.ts b/apps/ai/src/mcp/lib/infra-inspect.ts new file mode 100644 index 000000000..5778ac2da --- /dev/null +++ b/apps/ai/src/mcp/lib/infra-inspect.ts @@ -0,0 +1,482 @@ +/** One infrastructure entity in depth: what the web app's host/pod/node/workload/container pages show. */ +import { Array as Arr, Effect } from "effect" +import { CH } from "@maple/query-engine" +import { formatPercent } from "./format" +import { WarehouseExecutor } from "@maple/query-engine/observability" +import type { + InfraEntityKind, + InfraEntityRow, + InfraKindSection, + InfraSeriesStats, +} from "@maple/domain/mcp-outputs" +import { + WORKLOAD_KINDS, + bucketSecondsFor, + busyFraction, + iso, + listInfraKind, + onlyGroup, + seriesStats, + ungrouped, + type InfraWindow, + type WorkloadKind, +} from "./infra" + +export interface InspectTarget { + readonly kind: InfraEntityKind + readonly name: string + readonly namespace?: string | undefined + readonly host?: string | undefined + readonly workloadKind?: WorkloadKind | undefined +} + +export interface InspectResult { + readonly found: boolean + readonly entity?: InfraEntityRow + readonly firstSeen?: string + readonly details?: Record + readonly series: ReadonlyArray + readonly children?: InfraKindSection +} + +const NOT_FOUND: InspectResult = { found: false, series: [] } +const CHILD_LIMIT = 10 +/** Pods of a workload whose own memory series is shown, highest peak first. */ +const PER_POD_SERIES = 3 + +/** Keeps only the details a collector actually reported. */ +const details = (entries: ReadonlyArray): Record => + Object.fromEntries(entries.filter(([, value]) => value !== "")) + +export const inspectInfra = (target: InspectTarget, window: InfraWindow) => + Effect.gen(function* () { + const executor = yield* WarehouseExecutor + const params = { orgId: executor.orgId, startTime: window.startTime, endTime: window.endTime } + const seriesParams = { ...params, bucketSeconds: bucketSecondsFor(window) } + const series = (query: CH.CompiledQueryInput, context: string) => + executor.compiledQuery(query, { profile: "aggregation", context }) + const first = (query: CH.CompiledQueryInput, context: string) => + executor + .compiledQuery(query, { profile: "aggregation", context }) + .pipe(Effect.map((rows) => rows[0])) + const children = ( + kind: "pods" | "containers", + filters: Parameters[1]["filters"], + ) => listInfraKind(kind, { window, filters, sort: "saturation", limit: CHILD_LIMIT }) + + switch (target.kind) { + case "host": { + const host = target.name + const [summary, cpu, memory, load, filesystems] = yield* Effect.all( + [ + first( + CH.compile(CH.hostDetailSummaryQuery({ hostName: host }), params), + "hostDetailSummary", + ), + series( + CH.compile( + CH.hostGaugeTimeseriesQuery({ + hostName: host, + metricName: "system.cpu.utilization", + groupByAttributeKey: "state", + }), + seriesParams, + ), + "hostInfraGaugeTimeseries", + ), + series( + CH.compile( + CH.hostGaugeTimeseriesQuery({ + hostName: host, + metricName: "system.memory.utilization", + groupByAttributeKey: "state", + }), + seriesParams, + ), + "hostInfraGaugeTimeseries", + ), + series( + CH.compile( + CH.hostGaugeTimeseriesQuery({ + hostName: host, + metricName: "system.cpu.load_average.15m", + }), + seriesParams, + ), + "hostInfraGaugeTimeseries", + ), + series( + CH.compile(CH.hostFilesystemsQuery({ hostName: host }), params), + "hostFilesystems", + ), + ], + { concurrency: 5 }, + ) + if (summary === undefined) return NOT_FOUND + const containers = yield* children("containers", { host }) + return { + found: true, + entity: { + name: summary.hostName, + lastSeen: iso(summary.lastSeen), + ...(summary.osType ? { os: summary.osType } : undefined), + cpu: summary.cpuPct, + memory: summary.memoryPct, + disk: summary.diskPct, + load15: summary.load15, + }, + firstSeen: iso(summary.firstSeen), + details: details([ + ["arch", summary.hostArch], + ["cloud.provider", summary.cloudProvider], + ["cloud.region", summary.cloudRegion], + ...filesystems.map((fs): readonly [string, string] => [ + `disk ${fs.mountpoint}`, + `${formatPercent(fs.usedMax)} full at most (avg ${formatPercent(fs.usedAvg)})${fs.device ? `, ${fs.device}` : ""}`, + ]), + ]), + series: [ + ...seriesStats({ label: "cpu busy", unit: "fraction" }, busyFraction(cpu)), + ...seriesStats({ label: "memory used", unit: "fraction" }, onlyGroup(memory, "used")), + ...seriesStats({ label: "load average (15m)", unit: "load" }, ungrouped(load)), + ], + ...(containers.rows.length > 0 ? { children: containers } : undefined), + } satisfies InspectResult + } + case "pod": { + const pod = { podName: target.name, namespace: target.namespace } + const podSeries = (metricName: string) => + series( + CH.compile(CH.podGaugeTimeseriesQuery({ ...pod, metricName }), seriesParams), + "podInfraTimeseries", + ) + const [summary, cores, cpuLimit, memoryLimit, workingSet, restarts] = yield* Effect.all( + [ + first(CH.compile(CH.podDetailSummaryQuery(pod), params), "podDetailSummary"), + podSeries("k8s.pod.cpu.usage"), + podSeries("k8s.pod.cpu_limit_utilization"), + podSeries("k8s.pod.memory_limit_utilization"), + podSeries("k8s.pod.memory.working_set"), + series( + CH.compile( + CH.podRestartsQuery({ podNames: [target.name], namespace: target.namespace }), + params, + ), + "podRestarts", + ), + ], + { concurrency: 6 }, + ) + if (summary === undefined) return NOT_FOUND + const restarted = restarts.reduce((sum, row) => sum + row.restarts, 0) + // The limit itself is not collected; working set over its share of the limit recovers it. + const workingSetAvg = seriesStats({ label: "", unit: "bytes" }, ungrouped(workingSet))[0]?.avg + const memoryLimitBytes = + workingSetAvg !== undefined && summary.memoryLimitPct > 0 + ? workingSetAvg / summary.memoryLimitPct + : undefined + const workload = summary.deploymentName || summary.statefulsetName || summary.daemonsetName + const workloadKind = summary.deploymentName + ? "deployment" + : summary.statefulsetName + ? "statefulset" + : summary.daemonsetName + ? "daemonset" + : undefined + return { + found: true, + entity: { + name: summary.podName, + lastSeen: iso(summary.lastSeen), + ...(summary.namespace ? { namespace: summary.namespace } : undefined), + ...(summary.nodeName ? { node: summary.nodeName } : undefined), + ...(workloadKind === undefined ? undefined : { workload, workloadKind }), + cpuCores: summary.cpuUsage, + cpu: summary.cpuLimitPct, + memory: summary.memoryLimitPct, + saturation: Math.max(summary.cpuLimitPct, summary.memoryLimitPct), + ...(summary.cpuLimitPct === 0 && summary.memoryLimitPct === 0 + ? { unbounded: true } + : undefined), + ...(restarts.length > 0 ? { restarts: restarted } : undefined), + }, + firstSeen: iso(summary.firstSeen), + details: details([ + ["qos class", summary.qosClass], + ["pod uid", summary.podUid], + ["pod started", summary.podStartTime], + ["compute type", summary.computeType], + [ + "memory limit", + memoryLimitBytes === undefined + ? "" + : `about ${Math.round(memoryLimitBytes / 2 ** 20)} MiB (working set ÷ % of limit)`, + ], + ...restarts.map((row): readonly [string, string] => [ + `restarts of container ${row.containerName}`, + `${row.restarts} in this window (${row.totalRestarts} since the pod started)`, + ]), + [ + "CPU vs request (avg)", + summary.cpuRequestPct > 0 + ? `${formatPercent(summary.cpuRequestPct)} (above 100% is allowed up to the limit)` + : "", + ], + [ + "memory vs request (avg)", + summary.memoryRequestPct > 0 + ? `${formatPercent(summary.memoryRequestPct)} (above 100% is allowed up to the limit)` + : "", + ], + ]), + series: [ + ...seriesStats({ label: "cpu cores", unit: "cores" }, ungrouped(cores)), + ...seriesStats({ label: "cpu % of limit", unit: "fraction" }, ungrouped(cpuLimit)), + ...seriesStats( + { label: "memory % of limit", unit: "fraction" }, + ungrouped(memoryLimit), + ), + ...seriesStats({ label: "memory working set", unit: "bytes" }, ungrouped(workingSet)), + ], + } satisfies InspectResult + } + case "node": { + const node = target.name + const [summary, cores] = yield* Effect.all( + [ + first( + CH.compile(CH.nodeDetailSummaryQuery({ nodeName: node }), params), + "nodeDetailSummary", + ), + series( + CH.compile( + CH.nodeGaugeTimeseriesQuery({ + nodeName: node, + metricName: "k8s.node.cpu.usage", + }), + seriesParams, + ), + "nodeInfraTimeseries", + ), + ], + { concurrency: 2 }, + ) + if (summary === undefined) return NOT_FOUND + const [pods, peers] = yield* Effect.all( + [ + children("pods", { node }), + listInfraKind("nodes", { window, filters: {}, sort: "cpu", limit: CHILD_LIMIT }), + ], + { concurrency: 2 }, + ) + const podCores = pods.rows.reduce((sum, pod) => sum + (pod.cpuCores ?? 0), 0) + const allPods = pods.summary === undefined || pods.rows.length >= pods.summary.total + const others = peers.rows.filter((peer) => peer.name !== summary.nodeName) + return { + found: true, + entity: { + name: summary.nodeName, + lastSeen: iso(summary.lastSeen), + cpuCores: summary.cpuUsage, + uptimeSeconds: summary.uptime, + }, + firstSeen: iso(summary.firstSeen), + details: details([ + ["node uid", summary.nodeUid], + ["kubelet", summary.kubeletVersion], + ["container runtime", summary.containerRuntime], + [ + "CPU by pods", + pods.rows.length === 0 + ? "" + : `${allPods ? "its" : `the ${pods.rows.length} busiest of its`} pods use ${podCores.toFixed(2)} of the node's ${summary.cpuUsage.toFixed(2)} cores (avg); the rest is the kubelet and system daemons${allPods ? "" : " and pods not listed"}`, + ], + [ + "other nodes (CPU cores, avg)", + others + .map((peer) => `${peer.name} ${(peer.cpuCores ?? 0).toFixed(2)}`) + .join(", "), + ], + [ + "capacity", + "not collected: compare with the other nodes, or the pods against their limits", + ], + ]), + series: seriesStats({ label: "cpu cores", unit: "cores" }, ungrouped(cores)), + ...(pods.rows.length > 0 ? { children: pods } : undefined), + } satisfies InspectResult + } + case "workload": + return yield* inspectWorkload(target, window) + case "container": { + const container = { containerName: target.name, hostName: target.host } + const [summary, counters, cpu, memory] = yield* Effect.all( + [ + first( + CH.compile(CH.containerDetailSummaryQuery(container), params), + "containerDetailSummary", + ), + first( + CH.compile(CH.containerCountersSummaryQuery(container), params), + "containerCountersSummary", + ), + series( + CH.compile( + CH.containerGaugeTimeseriesQuery({ + ...container, + metricName: "container.cpu.utilization", + divideBy: 100, + }), + seriesParams, + ), + "containerInfraTimeseries", + ), + series( + CH.compile( + CH.containerGaugeTimeseriesQuery({ + ...container, + metricName: "container.memory.percent", + divideBy: 100, + }), + seriesParams, + ), + "containerInfraTimeseries", + ), + ], + { concurrency: 4 }, + ) + if (summary === undefined) return NOT_FOUND + return { + found: true, + entity: { + name: summary.containerName, + lastSeen: iso(summary.lastSeen), + ...(summary.hostName ? { host: summary.hostName } : undefined), + ...(summary.imageName ? { image: summary.imageName } : undefined), + ...(summary.composeService + ? { workload: summary.composeService, workloadKind: "compose" } + : undefined), + cpu: summary.cpuPct, + memory: summary.memoryPct, + uptimeSeconds: summary.uptimeSeconds, + }, + firstSeen: iso(summary.firstSeen), + details: details([ + ["container id", summary.containerId], + ["runtime", summary.runtime], + ["compose project", summary.composeProject], + ["cpu limit (cores)", summary.cpuLimitCores > 0 ? String(summary.cpuLimitCores) : ""], + ["restarts in window", counters === undefined ? "" : String(counters.restartsDelta)], + [ + "memory limit (bytes)", + counters !== undefined && counters.memoryLimitBytes > 0 + ? String(counters.memoryLimitBytes) + : "", + ], + ]), + series: [ + ...seriesStats({ label: "cpu", unit: "fraction" }, ungrouped(cpu)), + ...seriesStats({ label: "memory", unit: "fraction" }, ungrouped(memory)), + ], + } satisfies InspectResult + } + } + }) + +/** A workload name does not say its kind, so an unqualified one is looked up under each. */ +const inspectWorkload = (target: InspectTarget, window: InfraWindow) => + Effect.gen(function* () { + const executor = yield* WarehouseExecutor + const params = { orgId: executor.orgId, startTime: window.startTime, endTime: window.endTime } + const kinds = target.workloadKind === undefined ? WORKLOAD_KINDS : [target.workloadKind] + const matches = yield* Effect.forEach( + kinds, + (kind) => + executor + .compiledQuery( + CH.compile( + CH.workloadDetailSummaryQuery({ + kind, + workloadName: target.name, + namespace: target.namespace, + }), + params, + ), + { profile: "aggregation", context: "workloadDetailSummary" }, + ) + .pipe(Effect.map((rows) => rows.map((row) => ({ kind, row })))), + { concurrency: 3 }, + ) + const match = matches.flat().sort((a, b) => b.row.podCount - a.row.podCount)[0] + if (match === undefined) return NOT_FOUND + const { kind, row } = match + const workload = { kind, workloadName: target.name, namespace: target.namespace } + const seriesParams = { ...params, bucketSeconds: bucketSecondsFor(window) } + const workloadSeries = (metricName: string) => + executor.compiledQuery( + CH.compile(CH.workloadGaugeTimeseriesQuery({ ...workload, metricName }), seriesParams), + { profile: "aggregation", context: "workloadInfraTimeseries" }, + ) + const [cores, cpuLimit, memoryLimit, memoryByPod, pods] = yield* Effect.all( + [ + workloadSeries("k8s.pod.cpu.usage"), + workloadSeries("k8s.pod.cpu_limit_utilization"), + workloadSeries("k8s.pod.memory_limit_utilization"), + executor.compiledQuery( + CH.compile( + CH.workloadGaugeTimeseriesQuery({ + ...workload, + metricName: "k8s.pod.memory_limit_utilization", + groupByPod: true, + }), + seriesParams, + ), + { profile: "aggregation", context: "workloadInfraTimeseries" }, + ), + listInfraKind("pods", { + window, + filters: { workloadKind: kind, workload: target.name, namespace: target.namespace }, + sort: "saturation", + limit: CHILD_LIMIT, + }), + ], + { concurrency: 5 }, + ) + const podPeaks = Object.entries(Arr.groupBy(memoryByPod, (r) => r.attributeValue)) + .flatMap(([pod, rows]) => + seriesStats({ label: "memory % of limit", unit: "fraction" }, ungrouped(rows), pod), + ) + .sort((a, b) => b.max - a.max) + .slice(0, PER_POD_SERIES) + return { + found: true, + entity: { + name: row.workloadName, + lastSeen: iso(row.lastSeen), + workloadKind: kind, + ...(row.namespace ? { namespace: row.namespace } : undefined), + 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), + }, + firstSeen: iso(row.firstSeen), + series: [ + ...seriesStats({ label: "cpu cores (avg per pod)", unit: "cores" }, ungrouped(cores)), + ...seriesStats( + { label: "cpu % of limit (avg per pod)", unit: "fraction" }, + ungrouped(cpuLimit), + ), + ...seriesStats( + { label: "memory % of limit (avg per pod)", unit: "fraction" }, + ungrouped(memoryLimit), + ), + ...podPeaks, + ], + ...(pods.rows.length > 0 ? { children: pods } : undefined), + } satisfies InspectResult + }) diff --git a/apps/ai/src/mcp/lib/infra-params.ts b/apps/ai/src/mcp/lib/infra-params.ts new file mode 100644 index 000000000..368e6d55b --- /dev/null +++ b/apps/ai/src/mcp/lib/infra-params.ts @@ -0,0 +1,59 @@ +/** + * The infra tools' `kind` parameters. list_infra lists kinds (plural) and inspect_infra names one + * entity (singular); each accepts the other spelling, since models carry one over to the other. + */ +import { Schema, SchemaTransformation } from "effect" + +export const INFRA_KINDS = ["hosts", "pods", "nodes", "workloads", "containers"] as const +export const ENTITY_KINDS = ["host", "pod", "node", "workload", "container"] as const + +type InfraKind = (typeof INFRA_KINDS)[number] +type EntityKind = (typeof ENTITY_KINDS)[number] + +const TO_ENTITY = { + hosts: "host", + pods: "pod", + nodes: "node", + workloads: "workload", + containers: "container", + host: "host", + pod: "pod", + node: "node", + workload: "workload", + container: "container", +} as const satisfies Record + +const TO_KIND = { + hosts: "hosts", + pods: "pods", + nodes: "nodes", + workloads: "workloads", + containers: "containers", + host: "hosts", + pod: "pods", + node: "nodes", + workload: "workloads", + container: "containers", +} as const satisfies Record + +const EITHER = Schema.Literals([...INFRA_KINDS, ...ENTITY_KINDS]) + +/** A list_infra kind; the singular is read as its plural. */ +export const optionalInfraKind = (description: string) => + Schema.optional( + EITHER.annotate({ description }).pipe( + Schema.decodeTo( + Schema.Literals(INFRA_KINDS), + SchemaTransformation.transform({ decode: (kind) => TO_KIND[kind], encode: (kind) => kind }), + ), + ), + ) + +/** An inspect_infra kind; the plural is read as its singular. */ +export const entityKind = (description: string) => + EITHER.annotate({ description }).pipe( + Schema.decodeTo( + Schema.Literals(ENTITY_KINDS), + SchemaTransformation.transform({ decode: (kind) => TO_ENTITY[kind], encode: (kind) => kind }), + ), + ) diff --git a/apps/ai/src/mcp/lib/infra-render.ts b/apps/ai/src/mcp/lib/infra-render.ts new file mode 100644 index 000000000..ed7c89580 --- /dev/null +++ b/apps/ai/src/mcp/lib/infra-render.ts @@ -0,0 +1,263 @@ +/** Markdown for infrastructure rows: one table layout per kind, shared by list_infra and inspect_infra. */ +import type { InfraEntityRow, InfraKind, InfraKindSection } from "@maple/domain/mcp-outputs" +import { formatNumber, formatPercent, tableCell } from "./format" +import { doc, type DocBlock } from "./tool-doc" +import { HOT_THRESHOLD, LIMIT_THRESHOLD, pressure } from "./infra" +import { parseWarehouseDateTime } from "@maple/query-engine" + +/** "21:54 to 22:54 UTC (1 hour)": the window with its length, so a default is visible. */ +export const windowText = (range: { readonly start: string; readonly end: string }): string => { + const minutes = Math.round( + (parseWarehouseDateTime(range.end) - parseWarehouseDateTime(range.start)) / 60_000, + ) + const length = + minutes % 1440 === 0 + ? `${minutes / 1440} day${minutes === 1440 ? "" : "s"}` + : minutes % 60 === 0 + ? `${minutes / 60} hour${minutes === 60 ? "" : "s"}` + : `${minutes} minutes` + return `${range.start} to ${range.end} UTC (${length})` +} + +export const KIND_NOUN = { + hosts: "hosts", + pods: "pods", + nodes: "nodes", + workloads: "workloads", + containers: "containers", +} as const satisfies Record + +/** The `inspect_infra` kind for a `list_infra` kind. */ +export const ENTITY_KIND = { + hosts: "host", + pods: "pod", + nodes: "node", + workloads: "workload", + containers: "container", +} as const satisfies Record + +const pct = (value: number | undefined): string => (value === undefined ? "" : formatPercent(value)) + +/** "41.00% → 97.00%": the window average, then its peak when the kind reports one. */ +const avgPeak = (avg: number | undefined, peak: number | undefined): string => + avg === undefined ? "" : peak === undefined ? pct(avg) : `${pct(avg)} → ${pct(peak)}` + +/** A share of a limit; "no limit" rather than a misleading 0% when none is set. */ +const limitPct = (row: InfraEntityRow, value: string): string => (row.unbounded ? "no limit" : value) + +const cores = (value: number | undefined, peak?: number): string => + value === undefined + ? "" + : peak === undefined + ? value.toFixed(2) + : `${value.toFixed(2)} → ${peak.toFixed(2)}` + +export const formatUptime = (seconds: number | undefined): string => { + if (seconds === undefined || seconds <= 0) return "" + if (seconds < 3600) return `${Math.round(seconds / 60)}m` + if (seconds < 86_400) return `${(seconds / 3600).toFixed(1)}h` + return `${(seconds / 86_400).toFixed(1)}d` +} + +/** Peaks are against a limit for pods and containers; hosts have no limit, so their averages and fullest disk. */ +const pressureText = (kind: InfraKind, resource: string, level: number): string => { + if (kind === "hosts") + return resource === "disk" ? `disk ${pct(level)} full` : `${resource} ${pct(level)} avg` + const ofLimit = kind === "containers" ? "" : " of limit" + return kind === "workloads" + ? `${resource} ${pct(level)}${ofLimit} avg` + : `${resource} peaked at ${pct(level)}${ofLimit}` +} + +/** Every resource at 60% or more, worst first, under the label of the worst. */ +export const statusOf = (row: InfraEntityRow, kind: InfraKind): string => { + if (row.unbounded) return "no limits set" + const resources: ReadonlyArray = [ + ["memory", row.memoryPeak ?? row.memory], + ["cpu", row.cpuPeak ?? row.cpu], + ["disk", row.disk], + ] + const hot = resources + .flatMap(([resource, level]) => + level !== undefined && level >= HOT_THRESHOLD ? [{ resource, level }] : [], + ) + .sort((a, b) => b.level - a.level) + const worst = hot[0] + const restartText = row.restarts ? `${row.restarts} restart${row.restarts === 1 ? "" : "s"}` : "" + if (worst === undefined) return restartText + // Hosts have no limit to be at; their 90% is just high. + const label = worst.level >= LIMIT_THRESHOLD ? (kind === "hosts" ? "HIGH" : "AT LIMIT") : "hot" + const restarts = restartText === "" ? "" : `; ${restartText}` + return `${label}: ${hot.map((h) => pressureText(kind, h.resource, h.level)).join(", ")}${restarts}` +} + +const owner = (row: InfraEntityRow): string => + row.workload === undefined ? "" : `${row.workloadKind ?? ""}/${row.workload}` + +export const kindTable = (section: InfraKindSection): DocBlock => { + const rows = section.rows + const status = (row: InfraEntityRow) => statusOf(row, section.kind) + switch (section.kind) { + case "hosts": + return doc.table( + [ + "Host", + "CPU busy (avg)", + "Memory used (avg)", + "Fullest disk", + "Load 15m (avg)", + "OS", + "Status", + "Last seen", + ], + rows.map((r) => [ + tableCell(r.name), + pct(r.cpu), + pct(r.memory), + pct(r.disk), + r.load15 === undefined ? "" : r.load15.toFixed(2), + r.os ?? "", + status(r), + r.lastSeen, + ]), + ) + case "pods": + return doc.table( + [ + "Pod", + "Namespace", + "Owner", + "Node", + "CPU cores avg → peak", + "CPU of limit avg → peak", + "Memory of limit avg → peak", + "Status", + ], + rows.map((r) => [ + tableCell(r.name), + r.namespace ?? "", + owner(r), + r.node ?? "", + cores(r.cpuCores, r.cpuCoresPeak), + limitPct(r, avgPeak(r.cpu, r.cpuPeak)), + limitPct(r, avgPeak(r.memory, r.memoryPeak)), + status(r), + ]), + ) + case "nodes": + return doc.table( + ["Node", "Cluster", "CPU cores in use (avg)", "Uptime", "Last seen"], + rows.map((r) => [ + tableCell(r.name), + r.cluster ?? "", + cores(r.cpuCores), + formatUptime(r.uptimeSeconds), + r.lastSeen, + ]), + ) + case "workloads": + return doc.table( + [ + "Workload", + "Kind", + "Namespace", + "Pods", + "CPU cores per pod (avg)", + "CPU of limit (avg)", + "Memory of limit (avg)", + "Status", + ], + rows.map((r) => [ + tableCell(r.name), + r.workloadKind ?? "", + r.namespace ?? "", + r.podCount === undefined ? "" : formatNumber(r.podCount), + cores(r.cpuCores), + limitPct(r, pct(r.cpu)), + limitPct(r, pct(r.memory)), + status(r), + ]), + ) + case "containers": + return doc.table( + ["Container", "Host", "Image", "CPU avg → peak", "Memory avg → peak", "Uptime", "Status"], + rows.map((r) => [ + tableCell(r.name), + r.host ?? "", + tableCell(r.image ?? "", 50), + avgPeak(r.cpu, r.cpuPeak), + avgPeak(r.memory, r.memoryPeak), + formatUptime(r.uptimeSeconds), + status(r), + ]), + ) + } +} + +/** The metrics behind each kind, and how to chart them: query_data or create_dashboard on source=metrics. */ +export const KIND_METRICS = { + hosts: "Metrics: system.cpu.utilization (by state; busy = 1 - idle), system.memory.utilization, system.filesystem.utilization, system.cpu.load_average.15m. Chart per host with group_by resource.host.name.", + pods: "Metrics: k8s.pod.cpu.usage (cores), k8s.pod.cpu_limit_utilization, k8s.pod.memory_limit_utilization. Chart per pod with group_by resource.k8s.pod.name.", + nodes: "Metrics: k8s.node.cpu.usage (cores). Chart per node with group_by resource.k8s.node.name.", + workloads: + "Metrics: the pod metrics (k8s.pod.*) of the workload's pods. Chart per workload with group_by resource.k8s.deployment.name (or statefulset/daemonset).", + containers: + "Metrics: container.cpu.utilization and container.memory.percent (0-100), container.restarts. Chart per container with group_by resource.container.name.", +} as const satisfies Record + +/** What the CPU and memory percentages of a kind are measured against. */ +export const KIND_BASIS = { + hosts: "CPU and memory are the host's own utilization averaged over the window; disk is the fullest filesystem.", + pods: "CPU and memory are measured against each pod's configured limit; status uses the window peak.", + nodes: "CPU is cores in use; node capacity is not reported, so compare nodes with each other.", + workloads: "CPU and memory are averaged across the workload's pods, against their limits.", + containers: + "CPU is a share of the host's CPUs; memory is a share of the container's memory limit (the host's memory when none is set); status uses the window peak.", +} as const satisfies Record + +export const summaryLine = (section: InfraKindSection, status?: string): string => { + const s = section.summary + const showing = section.rows.length + const total = s?.total + const noun = section.kind === "pods" ? "live pods" : KIND_NOUN[section.kind] + const head = + status !== undefined + ? `${formatNumber(showing)}${section.truncated ? "+" : ""} of ${total === undefined ? "the" : formatNumber(total)} ${noun} match status="${status}"` + : total === undefined + ? `${formatNumber(showing)} ${KIND_NOUN[section.kind]}${section.truncated ? " shown (more exist)" : ""}` + : `${formatNumber(total)} ${noun}${showing < total ? `, showing the top ${formatNumber(showing)}` : ""}` + const parts = [ + head, + s?.saturated === undefined || s.saturated === 0 + ? undefined + : `${formatNumber(s.saturated)} at a limit (peak ≥90%)`, + s?.elevated === undefined || s.elevated === 0 + ? undefined + : `${formatNumber(s.elevated)} hot (peak 60-90%)`, + s?.unbounded === undefined || s.unbounded === 0 + ? undefined + : `${formatNumber(s.unbounded)} running with no limits set (status="no_limits" lists them)`, + s?.stale === undefined || s.stale === 0 + ? undefined + : `${formatNumber(s.stale)} stale (no recent scrape)`, + s?.ended === undefined || s.ended === 0 + ? undefined + : `${formatNumber(s.ended)} ended in the window (include_ended=true lists them)`, + ] + // A cut list must say whether it cut anything that matters. + const flagged = (s?.saturated ?? 0) + (s?.elevated ?? 0) + (s?.unbounded ?? 0) + const flaggedShown = section.rows.filter((row) => statusOf(row, section.kind) !== "").length + const coverage = + status === undefined && total !== undefined && showing < total && flagged > 0 + ? flaggedShown >= flagged + ? " Every flagged one is shown." + : ` ${formatNumber(flagged - flaggedShown)} flagged not shown: pass kind="${section.kind}" to list them.` + : "" + return `${parts.filter((part) => part !== undefined).join(", ")}.${coverage} ${KIND_BASIS[section.kind]}` +} + +/** The row an agent should look at first: closest to a limit, when anything is hot. */ +export const hottest = (section: InfraKindSection): InfraEntityRow | undefined => + [...section.rows] + .filter((row) => !row.unbounded && pressure(row).level >= HOT_THRESHOLD) + .sort((a, b) => pressure(b).level - pressure(a).level)[0] diff --git a/apps/ai/src/mcp/lib/infra.ts b/apps/ai/src/mcp/lib/infra.ts new file mode 100644 index 000000000..1f695e245 --- /dev/null +++ b/apps/ai/src/mcp/lib/infra.ts @@ -0,0 +1,571 @@ +/** + * Infrastructure reads for the MCP tools: the queries behind the web app's Infrastructure pages, + * flattened onto one row shape so an agent reads hosts, pods and containers the same way. + */ +import { Array as Arr, DateTime, Effect } from "effect" +import type { OrgId } from "@maple/domain" +import { CH, formatWarehouseDateTime, parseWarehouseDateTime } from "@maple/query-engine" +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 interface InfraWindow { + readonly startTime: string + readonly endTime: string +} + +export interface InfraFilters { + readonly search?: string | undefined + readonly namespace?: string | undefined + readonly cluster?: string | undefined + readonly node?: string | undefined + readonly host?: string | undefined + readonly environment?: string | undefined + readonly workloadKind?: WorkloadKind | undefined + readonly workload?: string | undefined + readonly status?: InfraStatusFilter | undefined + /** Pods that stopped reporting during the window too (rollouts, OOM kills, scale-in). */ + readonly includeEnded?: boolean | undefined +} + +/** `hot`: a peak at 60% or more; `at_limit`: 90% or more; `no_limits`: neither CPU nor memory limited. */ +export type InfraStatusFilter = "hot" | "at_limit" | "no_limits" + +export type InfraSort = "saturation" | "cpu" | "memory" | "name" | "last_seen" + +/** Same `YYYY-MM-DD HH:mm:ss` UTC as the window, so a peak reads against its bounds. */ +export const iso = (value: DateTime.Utc): string => formatWarehouseDateTime(DateTime.toEpochMillis(value)) +const list = (value: string | undefined): ReadonlyArray | undefined => + value === undefined ? undefined : [value] +const nonEmpty = (value: string): string | undefined => (value === "" ? undefined : value) + +const compileParams = (orgId: OrgId, window: InfraWindow) => ({ + orgId, + startTime: window.startTime, + endTime: window.endTime, +}) + +/** Which kinds report telemetry in the window: the web sidebar's visibility gate. */ +export const infraPresence = (window: InfraWindow) => + Effect.gen(function* () { + const executor = yield* WarehouseExecutor + const rows = yield* executor.compiledQuery( + CH.compileUnion(CH.infraPresenceQuery(), compileParams(executor.orgId, window)), + { profile: "discovery", context: "infraPresence" }, + ) + const surfaces = new Set(rows.map((row) => row.surface)) + const kinds: Array = [] + if (surfaces.has("hosts")) kinds.push("hosts") + if (surfaces.has("k8sPods")) kinds.push("pods") + if (surfaces.has("k8sNodes")) kinds.push("nodes") + if (surfaces.has("k8sWorkloads")) kinds.push("workloads") + if (surfaces.has("containers")) kinds.push("containers") + return kinds + }) + +const byNumberDesc = + (f: (a: A) => number | undefined) => + (a: A, b: A): number => + (f(b) ?? 0) - (f(a) ?? 0) + +/** Sorts rows the warehouse could not sort for us (hosts, nodes, merged workload kinds). */ +export const sortRows = (rows: ReadonlyArray, sort: InfraSort): Array => { + const copy = [...rows] + switch (sort) { + case "name": + return copy.sort((a, b) => a.name.localeCompare(b.name)) + case "last_seen": + return copy.sort((a, b) => b.lastSeen.localeCompare(a.lastSeen)) + case "cpu": + return copy.sort(byNumberDesc((r) => r.cpuPeak ?? r.cpu ?? r.cpuCores)) + case "memory": + return copy.sort(byNumberDesc((r) => r.memoryPeak ?? r.memory)) + case "saturation": + return copy.sort( + byNumberDesc( + (r) => + r.saturation ?? + Math.max(r.cpuPeak ?? r.cpu ?? 0, r.memoryPeak ?? r.memory ?? 0, r.disk ?? 0), + ), + ) + } +} + +/** Thresholds the web app tones amber and red at. */ +export const HOT_THRESHOLD = 0.6 +export const LIMIT_THRESHOLD = 0.9 + +/** + * The resource closest to its ceiling over the window: a pod's or container's peak against its + * limit, a host's average CPU and memory and its fullest disk. + */ +export const pressure = (row: InfraEntityRow): { readonly resource: string; readonly level: number } => { + const candidates: ReadonlyArray = [ + ["memory", row.memoryPeak ?? row.memory], + ["cpu", row.cpuPeak ?? row.cpu], + ["disk", row.disk], + ] + return candidates.reduce<{ resource: string; level: number }>( + (best, [resource, level]) => (level !== undefined && level > best.level ? { resource, level } : best), + { resource: "cpu", level: 0 }, + ) +} + +const keep = + (status: InfraStatusFilter | undefined) => + (row: InfraEntityRow): boolean => { + switch (status) { + case undefined: + return true + case "no_limits": + return row.unbounded === true + case "hot": + return pressure(row).level >= HOT_THRESHOLD + case "at_limit": + return pressure(row).level >= LIMIT_THRESHOLD + } + } + +const podSortKey = (sort: InfraSort) => { + switch (sort) { + case "saturation": + return "saturation" as const + case "cpu": + return "cpuUsage" as const + case "memory": + return "memoryLimitPct" as const + case "name": + return "podName" as const + case "last_seen": + return "lastSeen" as const + } +} + +const containerSortKey = (sort: InfraSort) => { + switch (sort) { + case "saturation": + return "saturation" as const + case "cpu": + return "cpuPct" as const + case "memory": + return "memoryPct" as const + case "name": + return "containerName" as const + case "last_seen": + return "lastSeen" as const + } +} + +const podWorkload = (row: { + readonly deploymentName: string + readonly statefulsetName: string + readonly daemonsetName: string + readonly jobName?: string +}): { readonly workload?: string; readonly workloadKind?: string } => { + if (row.deploymentName !== "") return { workload: row.deploymentName, workloadKind: "deployment" } + if (row.statefulsetName !== "") return { workload: row.statefulsetName, workloadKind: "statefulset" } + if (row.daemonsetName !== "") return { workload: row.daemonsetName, workloadKind: "daemonset" } + if (row.jobName !== undefined && row.jobName !== "") return { workload: row.jobName, workloadKind: "job" } + return {} +} + +/** Kubernetes placement fields, each only when the collector reported it. */ +const placement = (row: { + readonly namespace?: string + readonly clusterName?: string + readonly environment?: string + readonly nodeName?: string + readonly hostName?: string +}) => ({ + ...(row.namespace ? { namespace: row.namespace } : undefined), + ...(row.clusterName ? { cluster: row.clusterName } : undefined), + ...(row.environment ? { environment: row.environment } : undefined), + ...(row.nodeName ? { node: row.nodeName } : undefined), + ...(row.hostName ? { host: row.hostName } : undefined), +}) + +const hostRow = (row: CH.ListHostsOutput): InfraEntityRow => ({ + name: row.hostName, + lastSeen: iso(row.lastSeen), + ...(row.osType ? { os: row.osType } : undefined), + cpu: row.cpuPct, + memory: row.memoryPct, + disk: row.diskPct, + load15: row.load15, +}) + +const podRow = (row: CH.ListPodsOutput): InfraEntityRow => ({ + name: row.podName, + lastSeen: iso(row.lastSeen), + ...placement(row), + ...podWorkload(row), + cpuCores: row.cpuUsage, + cpuCoresPeak: row.cpuUsagePeak, + cpu: row.cpuLimitPct, + cpuPeak: row.cpuLimitPctPeak, + memory: row.memoryLimitPct, + memoryPeak: row.memoryLimitPctPeak, + saturation: row.saturation, + ...(row.cpuLimitPct === 0 && row.memoryLimitPct === 0 ? { unbounded: true } : undefined), +}) + +const nodeRow = (row: CH.ListNodesOutput): InfraEntityRow => ({ + name: row.nodeName, + lastSeen: iso(row.lastSeen), + ...placement(row), + cpuCores: row.cpuUsage, + uptimeSeconds: row.uptime, +}) + +const workloadRow = (kind: WorkloadKind, row: CH.ListWorkloadsOutput): InfraEntityRow => ({ + name: row.workloadName, + lastSeen: iso(row.lastSeen), + 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), +}) + +const containerRow = (row: CH.ListContainersOutput): InfraEntityRow => ({ + name: row.containerName, + lastSeen: iso(row.lastSeen), + ...placement(row), + ...(row.imageName ? { image: row.imageName } : undefined), + ...(row.composeService ? { workload: row.composeService, workloadKind: "compose" } : undefined), + cpu: row.cpuPct, + cpuPeak: row.cpuPctPeak, + memory: row.memoryPct, + memoryPeak: row.memoryPctPeak, + saturation: row.saturation, + uptimeSeconds: row.uptimeSeconds, +}) + +/** Container restarts per pod in the window; pods the k8s_cluster receiver does not report are absent. */ +const podRestarts = (podNames: ReadonlyArray, params: ReturnType) => + Effect.gen(function* () { + if (podNames.length === 0) return new Map() + const executor = yield* WarehouseExecutor + const rows = yield* executor.compiledQuery( + CH.compile(CH.podRestartsQuery({ podNames, limit: podNames.length * 4 }), params), + { profile: "aggregation", context: "podRestarts" }, + ) + return new Map( + Object.entries(Arr.groupBy(rows, (row) => row.podName)).map(([pod, containers]) => [ + pod, + containers.reduce((sum, row) => sum + row.restarts, 0), + ]), + ) + }) + +export interface ListKindOptions { + readonly window: InfraWindow + readonly filters: InfraFilters + readonly sort: InfraSort + readonly limit: number +} + +/** + * One page of a kind, ready to render. A status page over-fetches and filters, since those + * thresholds are not scopes every warehouse query takes. + */ +export const listInfraKind = (kind: InfraKind, options: ListKindOptions) => + Effect.gen(function* () { + const { filters, limit } = options + const filtering = filters.status !== undefined + const fetchLimit = filtering ? 500 : limit + 1 + const { rows, summary } = yield* fetchKind(kind, { ...options, limit: fetchLimit }) + const filtered = rows.filter(keep(filters.status)) + const sorted = kind === "pods" || kind === "containers" ? filtered : sortRows(filtered, options.sort) + return { + kind, + ...(summary === undefined ? undefined : { summary }), + rows: sorted.slice(0, limit), + truncated: sorted.length > limit || (filtering && rows.length >= fetchLimit), + } satisfies InfraKindSection + }) + +const fetchKind = (kind: InfraKind, options: ListKindOptions) => + Effect.gen(function* () { + const executor = yield* WarehouseExecutor + const params = compileParams(executor.orgId, options.window) + const { filters, sort, limit } = options + switch (kind) { + case "hosts": { + const rows = yield* executor.compiledQuery( + CH.compile(CH.listHostsQuery({ search: filters.search ?? filters.host, limit }), params), + { profile: "list", context: "listHosts" }, + ) + return { rows: rows.map(hostRow), summary: undefined } + } + case "pods": { + const podFilters = { + search: filters.search, + namespaces: list(filters.namespace), + clusters: list(filters.cluster), + nodeNames: list(filters.node), + environments: list(filters.environment), + ...(filters.includeEnded ? { lifecycle: "all" as const } : undefined), + ...(filters.workload === undefined + ? undefined + : { + workloadKind: filters.workloadKind ?? "deployment", + workloadName: filters.workload, + }), + } + const [rows, counts] = yield* Effect.all( + [ + executor.compiledQuery( + CH.compile( + CH.listPodsQuery({ + ...podFilters, + sortBy: podSortKey(sort), + sortDir: sort === "name" ? "asc" : "desc", + limit, + }), + params, + ), + { profile: "list", context: "listPods" }, + ), + executor.compiledQuery(CH.compile(CH.listPodsSummaryQuery(podFilters), params), { + profile: "aggregation", + context: "listPodsCount", + }), + ], + { concurrency: 2 }, + ) + const count = counts[0] + const restarts = yield* podRestarts( + rows.map((row) => row.podName), + params, + ) + return { + rows: rows.map((row) => { + const pod = podRow(row) + const restarted = restarts.get(row.podName) + return restarted === undefined ? pod : { ...pod, restarts: restarted } + }), + summary: + count === undefined + ? undefined + : { + total: count.livePods, + saturated: count.saturatedPods, + elevated: count.elevatedPods, + unbounded: count.unboundedPods, + ended: count.endedPods, + }, + } + } + case "nodes": { + const rows = yield* executor.compiledQuery( + CH.compile( + CH.listNodesQuery({ + search: filters.search ?? filters.node, + clusters: list(filters.cluster), + environments: list(filters.environment), + limit, + }), + params, + ), + { profile: "list", context: "listNodes" }, + ) + return { rows: rows.map(nodeRow), summary: undefined } + } + case "workloads": { + const kinds = filters.workloadKind === undefined ? WORKLOAD_KINDS : [filters.workloadKind] + const perKind = yield* Effect.forEach( + kinds, + (workloadKind) => + executor + .compiledQuery( + CH.compile( + CH.listWorkloadsQuery({ + kind: workloadKind, + search: filters.search ?? filters.workload, + namespaces: list(filters.namespace), + clusters: list(filters.cluster), + environments: list(filters.environment), + limit, + }), + params, + ), + { profile: "list", context: "listWorkloads" }, + ) + .pipe(Effect.map((rows) => rows.map((row) => workloadRow(workloadKind, row)))), + { concurrency: 3 }, + ) + return { rows: perKind.flat(), summary: undefined } + } + case "containers": { + const containerFilters = { + search: filters.search, + hostNames: list(filters.host), + environments: list(filters.environment), + composeServices: list(filters.workload), + } + const [rows, counts] = yield* Effect.all( + [ + executor.compiledQuery( + CH.compile( + CH.listContainersQuery({ + ...containerFilters, + sortBy: containerSortKey(sort), + sortDir: sort === "name" ? "asc" : "desc", + limit, + }), + params, + ), + { profile: "list", context: "listContainers" }, + ), + executor.compiledQuery( + CH.compile(CH.listContainersSummaryQuery(containerFilters), params), + { + profile: "aggregation", + context: "listContainersCount", + }, + ), + ], + { concurrency: 2 }, + ) + const count = counts[0] + return { + rows: rows.map(containerRow), + summary: + count === undefined + ? undefined + : { + total: count.totalContainers, + saturated: count.saturatedContainers, + elevated: count.elevatedContainers, + stale: count.staleContainers, + }, + } + } + } + }) + +// inspect_infra + +interface SeriesRow { + readonly bucket: DateTime.Utc + readonly attributeValue: string + readonly avgValue: number | null +} + +interface SeriesSpec { + readonly label: string + readonly unit: InfraSeriesStats["unit"] +} + +/** Condenses one bucketed series to the numbers an agent reasons with. Empty in, nothing out. */ +export const seriesStats = ( + spec: SeriesSpec, + points: ReadonlyArray<{ readonly bucket: DateTime.Utc; readonly value: number }>, + group?: string, +): ReadonlyArray => { + const last = points.at(-1) + if (last === undefined) return [] + const peak = points.reduce((best, point) => (point.value > best.value ? point : best), last) + const values = points.map((point) => point.value) + const third = Math.max(1, Math.floor(values.length / 3)) + const mean = (slice: ReadonlyArray) => slice.reduce((sum, value) => sum + value, 0) / slice.length + return [ + { + label: spec.label, + unit: spec.unit, + ...(group === undefined ? undefined : { group }), + min: Math.min(...values), + avg: values.reduce((sum, value) => sum + value, 0) / values.length, + max: peak.value, + last: last.value, + maxAt: iso(peak.bucket), + points: points.length, + change: mean(values.slice(-third)) - mean(values.slice(0, third)), + }, + ] +} + +/** A bucket with no samples averages to null: it is a gap, not a zero. */ +export const ungrouped = (rows: ReadonlyArray) => + rows.flatMap((row) => (row.avgValue === null ? [] : [{ bucket: row.bucket, value: row.avgValue }])) + +/** + * Host CPU arrives per state, each a share of the CPU, so busy is 1 - idle per bucket. + * A collector that drops the idle state leaves the sum of the rest. + */ +export const busyFraction = (rows: ReadonlyArray) => { + const byBucket = Arr.groupBy(ungroupedWithState(rows), (row) => iso(row.bucket)) + return Object.values(byBucket).map((states) => { + const first = states[0] + const idle = states.find((state) => state.state === "idle") + const busy = states.reduce((sum, state) => (state.state === "idle" ? sum : sum + state.value), 0) + return { bucket: first.bucket, value: idle === undefined ? busy : Math.max(0, 1 - idle.value) } + }) +} + +const ungroupedWithState = (rows: ReadonlyArray) => + rows.flatMap((row) => + row.avgValue === null ? [] : [{ bucket: row.bucket, state: row.attributeValue, value: row.avgValue }], + ) + +export const onlyGroup = (rows: ReadonlyArray, value: string) => + ungrouped(rows.filter((row) => row.attributeValue === value)) + +/** About 60 points across the window, on whole minutes. */ +export const bucketSecondsFor = (window: InfraWindow): number => { + const seconds = (parseWarehouseDateTime(window.endTime) - parseWarehouseDateTime(window.startTime)) / 1000 + return Math.max(60, Math.ceil(seconds / 60 / 60) * 60) +} + +// diagnose_service + +const SERVICE_POD_LIMIT = 5 + +/** + * The Kubernetes workload a service's spans name (via the collector's k8sattributes), and its + * pods closest to a limit. Undefined when the service reports no k8s context. + */ +export const serviceInfrastructure = (service: string, window: InfraWindow) => + Effect.gen(function* () { + const executor = yield* WarehouseExecutor + const rows = yield* executor.compiledQuery( + CH.serviceWorkloadsSQL({ services: [service] }, compileParams(executor.orgId, window)), + { profile: "aggregation", context: "serviceWorkloads" }, + ) + const workloads = rows.filter((row) => row.workloadName !== "") + const dominant = workloads.find((row) => row.workloadKind !== "unknown") + if (workloads.length === 0) return undefined + const pods = + dominant === undefined || dominant.workloadKind === "unknown" + ? [] + : (yield* listInfraKind("pods", { + window, + filters: { + workloadKind: dominant.workloadKind, + workload: dominant.workloadName, + ...(dominant.namespace ? { namespace: dominant.namespace } : undefined), + }, + sort: "saturation", + limit: SERVICE_POD_LIMIT, + })).rows + return { + workloads: workloads.map((row) => ({ + kind: row.workloadKind, + name: row.workloadName, + namespace: row.namespace, + ...(row.clusterName ? { cluster: row.clusterName } : undefined), + podCount: row.podCount, + ...(row.avgCpuLimitUtilization === null ? undefined : { cpu: row.avgCpuLimitUtilization }), + ...(row.avgMemoryLimitUtilization === null + ? undefined + : { memory: row.avgMemoryLimitUtilization }), + })), + pods, + } + }) diff --git a/apps/ai/src/mcp/resources/instructions.ts b/apps/ai/src/mcp/resources/instructions.ts index 1833b4d38..dd0ea53e8 100644 --- a/apps/ai/src/mcp/resources/instructions.ts +++ b/apps/ai/src/mcp/resources/instructions.ts @@ -109,6 +109,7 @@ first occurrences ──> (candidate) ──> issue in \`triage\` - Performance analysis: find_slow_traces -> inspect_trace -> get_service_top_operations - Trend analysis: query_data (timeseries or breakdown) - Service discovery: list_services -> diagnose_service +- Infrastructure (hosts, Kubernetes pods/nodes/workloads, Docker containers): list_infra (no \`kind\` = overview of what reports, hottest first) -> inspect_infra (CPU/memory over the window with the peak time, a node's or workload's pods). diagnose_service shows the workload and pods a service runs on. Resource metrics are also chartable: create_dashboard group_by \`resource.host.name\` / \`resource.k8s.pod.name\` on source=metrics - Deploy verification: service_deployments (versions per service, newest vs previous) -> compare_periods around the rollout. Route usage over up to 90 days: route_usage. Silence check before calling an outage: ingest_freshness. Volume per service: ingest_usage. Database query shapes by volume: db_query_volume - Alert management: list_alert_rules / list_alert_destinations -> get_alert_rule -> preview_alert_rule (dry-run before saving; a window with no data is a blind rule, not a healthy one) -> create_alert_rule / update_alert_rule / delete_alert_rule -> list_alert_incidents - AI agent sessions (LLM agent traces — NOT browser replays): list_agent_sessions -> get_agent_session -> inspect_span (decodes an AI span's messages and tool calls) diff --git a/apps/ai/src/mcp/tools/__tests__/infra-tools.test.ts b/apps/ai/src/mcp/tools/__tests__/infra-tools.test.ts new file mode 100644 index 000000000..fe1dfbf26 --- /dev/null +++ b/apps/ai/src/mcp/tools/__tests__/infra-tools.test.ts @@ -0,0 +1,163 @@ +import { afterAll, beforeAll, describe, expect, it } from "vitest" +import { Schema } from "effect" +import { DiagnoseServiceOutput, InspectInfraOutput, ListInfraOutput } from "@maple/domain/mcp-outputs" +import type { McpToolResult } from "../types" +import { installFakeWarehouse, restoreWarehouse, type FixtureRule } from "../../__evals__/fake-warehouse" +import { makeEvalRuntime, markdown, runToolDirect, type EvalRuntime } from "../../__evals__/eval-runtime" +import { INFRA_WORLD, infraFixtureRules } from "../../__evals__/infra-world" + +const WINDOW = { start_time: "2026-10-01 06:00:00", end_time: "2026-10-01 12:00:00" } + +let rt: EvalRuntime +// The fake client is installed once; a case swaps what it answers by replacing the rules in place. +const rules: Array = infraFixtureRules() +beforeAll(() => { + installFakeWarehouse(rules, undefined, "empty") + rt = makeEvalRuntime() +}) +afterAll(async () => { + restoreWarehouse() + await rt.dispose() +}) + +const call = (name: string, params: Record): Promise => + runToolDirect(rt, name, { ...WINDOW, ...params }) + +const decode = >(schema: S, result: McpToolResult) => + Schema.decodeUnknownSync(schema)(result.structuredContent) + +describe("list_infra", () => { + it("gives an overview of every reporting kind, hottest first", async () => { + const result = await call("list_infra", {}) + const output = decode(ListInfraOutput, result) + expect(output.reporting).toEqual(["hosts", "pods", "nodes", "workloads", "containers"]) + const pods = output.sections.find((s) => s.kind === "pods") + expect(pods?.rows[0]?.name).toBe(INFRA_WORLD.oomPod) + expect(pods?.summary?.saturated).toBe(1) + expect(pods?.truncated).toBe(true) + const text = markdown(result) + expect(text).toContain("## Infrastructure overview") + expect(text).toContain("AT LIMIT: memory peaked at 98.00% of limit") + expect(text).toContain(`inspect_infra kind="pod" name="${INFRA_WORLD.oomPod}"`) + }) + + it("narrows pods by namespace and keeps only hot ones", async () => { + const output = decode( + ListInfraOutput, + await call("list_infra", { kind: "pods", namespace: "shop", status: "hot" }), + ) + const names = output.sections[0]?.rows.map((r) => r.name) ?? [] + expect(names[0]).toBe(INFRA_WORLD.oomPod) + expect(names.every((name) => name.startsWith("checkout-"))).toBe(true) + }) + + it("reads a singular kind as its plural", async () => { + const output = decode(ListInfraOutput, await call("list_infra", { kind: "host" })) + expect(output.kind).toBe("hosts") + }) + + it("leaves kinds out of an overview when they cannot apply the filter", async () => { + const output = decode(ListInfraOutput, await call("list_infra", { node: INFRA_WORLD.hotNode })) + expect(output.sections.map((s) => s.kind)).toEqual(["pods", "nodes"]) + expect(output.sections[0]?.rows.every((r) => r.node === INFRA_WORLD.hotNode)).toBe(true) + }) + + it("shows pod restarts in the pod list", async () => { + const output = decode(ListInfraOutput, await call("list_infra", { kind: "pods" })) + expect(output.sections[0]?.rows.find((r) => r.name === INFRA_WORLD.oomPod)?.restarts).toBe(1) + }) + + it("lists only pods without limits", async () => { + const output = decode( + ListInfraOutput, + await call("list_infra", { kind: "pods", status: "no_limits" }), + ) + expect(output.sections[0]?.rows.map((r) => r.name).sort()).toEqual([ + "otel-agent-q4w9e", + "otel-agent-x8k2l", + ]) + }) + + it("flags workloads that set no limits", async () => { + const result = await call("list_infra", { kind: "workloads", workload_kind: "daemonset" }) + const row = decode(ListInfraOutput, result).sections[0]?.rows[0] + expect(row?.name).toBe(INFRA_WORLD.unboundedWorkload) + expect(row?.unbounded).toBe(true) + expect(markdown(result)).toContain("no limits") + }) + + it("says which kinds report when the asked-for kind does not", async () => { + rules.unshift({ match: (sql) => sql.includes(" AS surface"), rows: [{ surface: "hosts" }] }) + const text = markdown(await call("list_infra", { kind: "pods" })) + rules.shift() + expect(text).toContain("No pods report metrics in this window. Reporting: hosts.") + }) +}) + +describe("inspect_infra", () => { + it("finds the time a pod peaked against its memory limit", async () => { + const output = decode( + InspectInfraOutput, + await call("inspect_infra", { kind: "pod", name: INFRA_WORLD.oomPod }), + ) + expect(output.found).toBe(true) + expect(output.entity?.workload).toBe("checkout") + const memory = output.series.find((s) => s.label.startsWith("memory")) + expect(memory?.max).toBeCloseTo(0.98, 2) + expect(memory?.maxAt.startsWith("2026-10-01 10:")).toBe(true) + }) + + it("resolves a workload without its kind and lists its pods", async () => { + const result = await call("inspect_infra", { kind: "workload", name: "checkout" }) + const output = decode(InspectInfraOutput, result) + expect(output.entity?.workloadKind).toBe("deployment") + expect(output.children?.rows.map((r) => r.name)).toContain(INFRA_WORLD.oomPod) + expect(markdown(result)).toContain('attribute_key="k8s.deployment.name"') + }) + + it("reads host CPU as 1 - idle and lists the host's containers", async () => { + const output = decode( + InspectInfraOutput, + await call("inspect_infra", { kind: "host", name: INFRA_WORLD.hotHost }), + ) + const cpu = output.series.find((s) => s.label === "cpu busy") + expect(cpu?.avg).toBeGreaterThan(0.85) + expect(output.children?.rows[0]?.name).toBe(INFRA_WORLD.hotContainer) + expect(output.details?.["disk /var/lib/docker"]).toContain("93.00% full at most") + }) + + it("compares a node with its peers", async () => { + const result = await call("inspect_infra", { kind: "nodes", name: INFRA_WORLD.hotNode }) + const output = decode(InspectInfraOutput, result) + expect(output.kind).toBe("node") + expect(output.details?.["other nodes (CPU cores, avg)"]).toContain("ip-10-0-3-21") + }) + + it("calls out a drop after a memory peak", async () => { + const text = markdown(await call("inspect_infra", { kind: "pod", name: INFRA_WORLD.oomPod })) + expect(text).toContain("then fell to") + expect(text).toContain("OOM kill") + }) + + it("points at list_infra when the name does not exist", async () => { + const result = await call("inspect_infra", { kind: "pod", name: "checkout-typo" }) + expect(decode(InspectInfraOutput, result).found).toBe(false) + expect(markdown(result)).toContain('list_infra kind="pods" search="checkout-typo"') + }) +}) + +describe("diagnose_service", () => { + it("shows the workload and pods a service runs on", async () => { + const result = await call("diagnose_service", { service: "checkout" }) + const output = decode(DiagnoseServiceOutput, result) + expect(output.infrastructure?.workloads[0]?.name).toBe("checkout") + expect(output.infrastructure?.pods[0]?.name).toBe(INFRA_WORLD.oomPod) + expect(markdown(result)).toContain(`inspect_infra kind="pod" name="${INFRA_WORLD.oomPod}"`) + }) + + it("omits infrastructure for a service with no Kubernetes context", async () => { + const output = decode(DiagnoseServiceOutput, await call("diagnose_service", { service: "cron" })) + expect(output.infrastructure).toBeUndefined() + expect(output.infrastructureError).toBeUndefined() + }) +}) diff --git a/apps/ai/src/mcp/tools/create-dashboard.test.ts b/apps/ai/src/mcp/tools/create-dashboard.test.ts index 3a1418495..cd800b9e6 100644 --- a/apps/ai/src/mcp/tools/create-dashboard.test.ts +++ b/apps/ai/src/mcp/tools/create-dashboard.test.ts @@ -12,5 +12,6 @@ describe("normalizeGroupBy", () => { expect(normalizeGroupBy("service.name", "traces")).toBe("service.name") expect(normalizeGroupBy("span_name", "logs")).toBe("span_name") expect(normalizeGroupBy("attr.signal", "metrics")).toBe("attr.signal") + expect(normalizeGroupBy("resource.host.name", "metrics")).toBe("resource.host.name") }) }) diff --git a/apps/ai/src/mcp/tools/create-dashboard.ts b/apps/ai/src/mcp/tools/create-dashboard.ts index 30de56c72..5e5d89eb5 100644 --- a/apps/ai/src/mcp/tools/create-dashboard.ts +++ b/apps/ai/src/mcp/tools/create-dashboard.ts @@ -86,6 +86,9 @@ const VALID_GROUP_BY: Record = { /** Sources whose query builder also groups by an attribute key, `attr.`. */ const ATTR_GROUP_BY_SOURCES: ReadonlySet = new Set(["metrics", "product_events"]) +/** Sources that also group by a resource attribute, `resource.` (host.name, k8s.pod.name). */ +const RESOURCE_GROUP_BY_SOURCES: ReadonlySet = new Set(["metrics"]) + /** query_data's group_by spellings, accepted here too so one vocabulary works in both tools. */ const QUERY_DATA_GROUP_BY_ALIASES: ReadonlyMap = new Map([ ["service", "service.name"], @@ -110,8 +113,14 @@ function validateGroupBy(rawGroupBy: string, source: string, widgetTitle: string if (validOptions.includes(rawGroupBy)) return null if (allowsAttr && rawGroupBy.startsWith("attr.") && rawGroupBy.length > 5) return null - - const optsList = [...validOptions, ...(allowsAttr ? ["attr."] : [])] + const allowsResource = RESOURCE_GROUP_BY_SOURCES.has(source) + if (allowsResource && rawGroupBy.startsWith("resource.") && rawGroupBy.length > 9) return null + + const optsList = [ + ...validOptions, + ...(allowsAttr ? ["attr."] : []), + ...(allowsResource ? ["resource."] : []), + ] return `Widget "${widgetTitle}": invalid group_by "${rawGroupBy}" for source=${source}. Valid: ${optsList.join(", ")}. ${allowsAttr ? "Example: attr.signal" : ""}` } @@ -308,6 +317,7 @@ export function registerCreateDashboardTool(server: McpToolRegistrar) { [ ...(VALID_GROUP_BY[source] ?? []), ...(ATTR_GROUP_BY_SOURCES.has(source) ? ["attr."] : []), + ...(RESOURCE_GROUP_BY_SOURCES.has(source) ? ["resource."] : []), ].join("|") server.define({ @@ -320,6 +330,8 @@ export function registerCreateDashboardTool(server: McpToolRegistrar) { "Create a dashboard one of three ways: a `template`, simplified `widgets` specs, or full `dashboard_json`.\n\n" + "Templates:\n" + templateList + + "\n\nTemplates chart one service or host at a time. To compare entities in one chart (CPU per host, memory per pod), " + + "use `widgets` on source=metrics with group_by `resource.host.name` / `resource.k8s.pod.name`; list_infra names the metrics." + "\n\nCreated widgets are inspected (up to 12) and returned with a verdict " + "(looks_healthy / suspicious / broken). Use inspect_chart_data beyond the cap, or validate=false to skip.", parameters: Schema.Struct({ diff --git a/apps/ai/src/mcp/tools/diagnose-service.ts b/apps/ai/src/mcp/tools/diagnose-service.ts index 953d590bb..18e38d204 100644 --- a/apps/ai/src/mcp/tools/diagnose-service.ts +++ b/apps/ai/src/mcp/tools/diagnose-service.ts @@ -4,10 +4,13 @@ import { formatDurationFromMs, formatPercent, formatNumber, truncate } from "../ import { toMcpQueryError } from "../lib/map-warehouse-error" import * as P from "../lib/params" import { doc, type DocBlock } from "../lib/tool-doc" -import { Effect, Schema } from "effect" +import { Effect, Result, Schema } from "effect" import { DiagnoseServiceOutput } from "@maple/domain/mcp-outputs" import { diagnoseService } from "@maple/query-engine/observability" import { provideWarehouseExecutorFromTenant } from "@maple/backend/services/warehouse/WarehouseQueryService" +import { serviceInfrastructure } from "../lib/infra" +import { kindTable } from "../lib/infra-render" +import { warehouseErrorText } from "../lib/map-warehouse-error" const WINDOW = P.timeWindow({ defaultHours: 6 }) @@ -30,13 +33,23 @@ export function registerDiagnoseServiceTool(server: McpToolRegistrar) { const { st, et } = yield* WINDOW.resolve(params, "diagnose_service") const tenant = yield* CurrentMcpTenant - const result = yield* diagnoseService({ - serviceName: params.service, - timeRange: { startTime: st, endTime: et }, - environment: params.environment, - }).pipe( - provideWarehouseExecutorFromTenant(tenant), - Effect.mapError(toMcpQueryError("service_overview")), + const [result, infrastructure] = yield* Effect.all( + [ + diagnoseService({ + serviceName: params.service, + timeRange: { startTime: st, endTime: et }, + environment: params.environment, + }).pipe( + provideWarehouseExecutorFromTenant(tenant), + Effect.mapError(toMcpQueryError("service_overview")), + ), + // Optional context: a failed lookup is reported beside the diagnosis, never instead of it. + serviceInfrastructure(params.service, { startTime: st, endTime: et }).pipe( + provideWarehouseExecutorFromTenant(tenant), + Effect.result, + ), + ], + { concurrency: 2 }, ) return { @@ -65,6 +78,11 @@ export function registerDiagnoseServiceTool(server: McpToolRegistrar) { spanId: l.spanId, })), ...(params.environment === undefined ? undefined : { environment: params.environment }), + ...(Result.isSuccess(infrastructure) + ? infrastructure.success === undefined + ? undefined + : { infrastructure: infrastructure.success } + : { infrastructureError: warehouseErrorText(infrastructure.failure) }), } }), render: (output) => { @@ -118,6 +136,32 @@ export function registerDiagnoseServiceTool(server: McpToolRegistrar) { ), ) } + const infra = output.infrastructure + if (infra !== undefined) { + blocks.push( + doc.heading("Infrastructure"), + doc.table( + ["Workload", "Namespace", "Pods", "CPU of limit", "Memory of limit"], + infra.workloads.map((w) => [ + `${w.kind}/${w.name}`, + w.namespace, + formatNumber(w.podCount), + w.cpu === undefined ? "no limit" : formatPercent(w.cpu), + w.memory === undefined ? "no limit" : formatPercent(w.memory), + ]), + ), + ) + if (infra.pods.length > 0) { + blocks.push( + doc.text("Pods, closest to a limit first:"), + kindTable({ kind: "pods", rows: infra.pods, truncated: false }), + ) + } + } else if (output.infrastructureError !== undefined) { + blocks.push(doc.text(`Infrastructure lookup failed: ${output.infrastructureError}`)) + } + const workload = infra?.workloads.find((w) => w.kind !== "unknown") + const hotPod = infra?.pods.find((p) => (p.saturation ?? 0) >= 0.9) return { title: `Diagnosis: ${output.serviceName}`, scope: [ @@ -138,6 +182,29 @@ export function registerDiagnoseServiceTool(server: McpToolRegistrar) { .map((t) => doc.next("inspect_trace", { trace_id: t.traceId }, "inspect error trace"), ), + ...(hotPod === undefined + ? [] + : [ + doc.next( + "inspect_infra", + { kind: "pod", name: hotPod.name, namespace: hotPod.namespace }, + `${hotPod.name} is at a CPU or memory limit: see when it peaked`, + ), + ]), + ...(workload === undefined || hotPod !== undefined + ? [] + : [ + doc.next( + "inspect_infra", + { + kind: "workload", + name: workload.name, + namespace: workload.namespace, + workload_kind: workload.kind, + }, + "check whether its pods were short of CPU or memory", + ), + ]), doc.next( "service_map", { service: output.serviceName }, diff --git a/apps/ai/src/mcp/tools/inspect-infra.ts b/apps/ai/src/mcp/tools/inspect-infra.ts new file mode 100644 index 000000000..dc1733c14 --- /dev/null +++ b/apps/ai/src/mcp/tools/inspect-infra.ts @@ -0,0 +1,324 @@ +import { Effect, Schema } from "effect" +import { InspectInfraOutput, type InfraSeriesStats } from "@maple/domain/mcp-outputs" +import type { McpToolRegistrar } from "./types" +import { warehouseToMcpHandlers } from "../lib/map-warehouse-error" +import { withTenantExecutor, CurrentMcpTenant } from "../lib/query-warehouse" +import { MCP_DISCOVERY_MAX_HOURS } from "../lib/time" +import * as P from "../lib/params" +import { doc, type DocBlock, type NextCall } from "../lib/tool-doc" +import { WORKLOAD_KINDS, bucketSecondsFor } from "../lib/infra" +import { ENTITY_KINDS, entityKind } from "../lib/infra-params" +import { inspectInfra } from "../lib/infra-inspect" +import { formatPercent } from "../lib/format" +import { parseWarehouseDateTime } from "@maple/query-engine" +import { + ENTITY_KIND, + KIND_METRICS, + formatUptime, + kindTable, + summaryLine, + windowText, +} from "../lib/infra-render" + +const TOOL = "inspect_infra" +const WINDOW = P.timeWindow({ defaultHours: 1, maxHours: MCP_DISCOVERY_MAX_HOURS }) + +const LIST_KIND = { + host: "hosts", + pod: "pods", + node: "nodes", + workload: "workloads", + container: "containers", +} as const + +const plural = (count: number, noun: string): string => `${count} ${noun}${count === 1 ? "" : "s"}` + +/** The resource attribute that names the entity on its logs and spans. */ +const resourceKeyFor = (kind: (typeof ENTITY_KINDS)[number], workloadKind: string | undefined) => { + switch (kind) { + case "pod": + return "k8s.pod.name" + case "node": + return "k8s.node.name" + case "container": + return "container.name" + case "host": + return "host.name" + case "workload": + return workloadKind === undefined ? undefined : `k8s.${workloadKind}.name` + } +} + +/** Last third against first third: "rising +12.4 pts", "flat". A drop after a peak reads as falling. */ +const trend = (stat: InfraSeriesStats): string => { + const scale = stat.unit === "fraction" ? 100 : 1 + const delta = stat.change * scale + const relative = stat.avg === 0 ? 0 : Math.abs(stat.change) / Math.abs(stat.avg) + if (relative < 0.05) return "flat" + const amount = + stat.unit === "fraction" + ? `${Math.abs(delta).toFixed(1)} pts` + : stat.unit === "bytes" + ? `${(Math.abs(stat.change) / 1024 / 1024).toFixed(0)} MiB` + : Math.abs(delta).toFixed(2) + return `${stat.change > 0 ? "rising +" : "falling -"}${amount}` +} + +const value = (stat: InfraSeriesStats, v: number): string => { + switch (stat.unit) { + case "fraction": + return formatPercent(v) + case "cores": + return `${v.toFixed(3)} cores` + case "load": + return v.toFixed(2) + case "bytes": + return v >= 1024 ** 3 + ? `${(v / 1024 ** 3).toFixed(2)} GiB` + : `${(v / 1024 / 1024).toFixed(0)} MiB` + case "seconds": + return formatUptime(v) + } +} + +export function registerInspectInfraTool(server: McpToolRegistrar) { + server.define({ + name: TOOL, + title: "Inspect Infrastructure", + description: + "One host, Kubernetes pod/node/workload, or Docker container in depth: identity (OS, node, owner workload, image, QoS class; restart count for containers; per-mount disk use for hosts), " + + "CPU and memory over the window as min/avg/max/last with the time of the peak, and its busiest children (a node's or workload's pods, a host's containers). " + + "Use after list_infra, or to check whether a slow or failing service was starved of CPU or memory: compare the peak time with the incident. " + + "Same default window as list_infra (1 hour), so the numbers match; pass start_time to look further back.", + parameters: Schema.Struct({ + kind: entityKind("What `name` refers to"), + name: P.text( + "Exact name: host.name, k8s.pod.name, k8s.node.name, workload name, or container.name", + ), + namespace: P.optionalText( + "Kubernetes namespace, when the same pod or workload name exists in several", + ), + host: P.optionalText("Host of a container, when the same container name runs on several hosts"), + workload_kind: P.optionalOneOf( + WORKLOAD_KINDS, + "Kubernetes workload kind. Omit to look the name up under each kind", + ), + ...WINDOW.fields, + }), + output: InspectInfraOutput, + hints: { readOnly: true }, + phrases: ["Inspecting infrastructure", "Checking resource usage"], + handler: Effect.fn("McpTool.inspectInfra")(function* (params) { + const { st, et } = yield* WINDOW.resolve(params, TOOL) + const tenant = yield* CurrentMcpTenant + yield* Effect.annotateCurrentSpan({ orgId: tenant.orgId, kind: params.kind }) + const result = yield* withTenantExecutor( + inspectInfra( + { + kind: params.kind, + name: params.name, + namespace: params.namespace, + host: params.host, + workloadKind: params.workload_kind, + }, + { startTime: st, endTime: et }, + ), + ).pipe(Effect.catchTags(warehouseToMcpHandlers(TOOL))) + + return { + timeRange: { start: st, end: et }, + kind: params.kind, + name: params.name, + bucketSeconds: bucketSecondsFor({ startTime: st, endTime: et }), + ...result, + } + }), + render: (output) => { + const entity = output.entity + const scope: Array = [ + ["Window", windowText(output.timeRange)], + ] + if (!output.found || entity === undefined) { + return { + title: `${output.kind}: ${output.name}`, + scope, + empty: { + message: `No ${output.kind} named "${output.name}" reported metrics in this window.`, + hints: [ + `Names are exact; list_infra kind="${LIST_KIND[output.kind]}" search="${output.name}" finds near matches.`, + "Widen start_time/end_time if it stopped reporting earlier.", + ], + }, + blocks: [], + next: [ + doc.next( + "list_infra", + { kind: LIST_KIND[output.kind], search: output.name }, + "find the exact name", + ), + ], + } + } + const identity: Array = [ + ["Namespace", entity.namespace], + [ + "Owner", + entity.workload === undefined + ? undefined + : `${entity.workloadKind ?? ""}/${entity.workload}`, + ], + ["Node", entity.node], + ["Host", entity.host], + ["Image", entity.image], + ["OS", entity.os], + ["Pods", entity.podCount === undefined ? undefined : String(entity.podCount)], + [ + "Uptime", + entity.uptimeSeconds === undefined ? undefined : formatUptime(entity.uptimeSeconds), + ], + [ + "First seen", + output.firstSeen === undefined + ? undefined + : parseWarehouseDateTime(output.firstSeen) - + parseWarehouseDateTime(output.timeRange.start) < + output.bucketSeconds * 1000 + ? `${output.firstSeen} (the window start: it was already running)` + : output.firstSeen, + ], + ["Restarts in window", entity.restarts === undefined ? undefined : String(entity.restarts)], + ["Last seen", entity.lastSeen], + ...Object.entries(output.details ?? {}), + ] + const always: Array = [ + ["Fullest disk", entity.disk === undefined ? undefined : formatPercent(entity.disk)], + [ + "Limits", + entity.unbounded + ? "none set: CPU and memory cannot be measured against a limit" + : undefined, + ], + ] + // Window averages from the summary query, shown only when no series covers them. + const averages: Array = [ + ["CPU (avg)", entity.cpu === undefined ? undefined : formatPercent(entity.cpu)], + ["CPU cores (avg)", entity.cpuCores === undefined ? undefined : entity.cpuCores.toFixed(3)], + ["Memory (avg)", entity.memory === undefined ? undefined : formatPercent(entity.memory)], + ["Load 15m (avg)", entity.load15 === undefined ? undefined : entity.load15.toFixed(2)], + ] + const blocks: Array = [ + doc.fields( + [...identity, ...always, ...(output.series.length === 0 ? averages : [])].filter( + ([, v]) => v !== undefined && v !== "", + ), + ), + ] + // A per-pod series names its pod; the entity's own series is the entity. + const restartsOf = (pod: string | undefined) => + pod === undefined + ? entity.restarts + : output.children?.rows.find((row) => row.name === pod)?.restarts + const drops = output.series.filter( + (s) => s.unit === "fraction" && s.max >= 0.9 && s.last <= s.max * 0.6, + ) + // The workload average mirrors its hottest pod's drop; say it once, about the pod. + const notices = ( + drops.some((s) => s.group !== undefined) ? drops.filter((s) => s.group !== undefined) : drops + ).map((s) => { + const restarts = restartsOf(s.group) + const who = s.group === undefined ? s.label : `${s.group}'s ${s.label}` + const cause = + restarts === undefined + ? "most likely a restart (OOM kill or eviction); no restart count is collected for it" + : restarts > 0 + ? `with ${plural(restarts, "container restart")} recorded in the window: consistent with an OOM kill (the termination reason is not collected)` + : "with no restart recorded, so usage fell on its own" + return `${who} reached ${formatPercent(s.max)} at ${s.maxAt}, then fell to ${formatPercent(s.last)} by the end of the window, ${cause}.` + }) + if (output.series.length > 0) { + blocks.push( + doc.heading( + `Over the window (${output.series[0]?.points ?? 0} buckets of ${formatUptime(output.bucketSeconds)}; min/avg/max of bucket averages)`, + ), + doc.table( + ["Metric", "Min", "Avg", "Max", "Peak at", "Last", "Trend"], + output.series.map((s) => [ + s.group === undefined ? s.label : `${s.label} (${s.group})`, + value(s, s.min), + value(s, s.avg), + value(s, s.max), + s.maxAt, + value(s, s.last), + trend(s), + ]), + ), + ) + } else { + blocks.push(doc.text("No time series in this window.")) + } + blocks.push(doc.text(KIND_METRICS[LIST_KIND[output.kind]])) + const children = output.children + if (children !== undefined) { + blocks.push( + doc.heading( + children.kind === "pods" + ? "Pods, closest to a limit first" + : "Containers, busiest first", + ), + doc.text(summaryLine(children)), + kindTable(children), + ) + } + const resourceKey = resourceKeyFor(output.kind, entity.workloadKind) + const peak = output.series.reduce( + (best, s) => (s.unit === "fraction" && (best === undefined || s.max > best.max) ? s : best), + undefined, + ) + const next: Array = [ + ...(children?.rows[0] === undefined + ? [] + : [ + doc.next( + TOOL, + { + kind: ENTITY_KIND[children.kind], + name: children.rows[0].name, + namespace: children.rows[0].namespace, + host: children.kind === "containers" ? children.rows[0].host : undefined, + start_time: output.timeRange.start, + end_time: output.timeRange.end, + }, + `drill into ${children.rows[0].name}`, + ), + ]), + ...(resourceKey === undefined + ? [] + : [ + doc.next( + "query_data", + { + source: "logs", + kind: "timeseries", + group_by: "severity", + attribute_key: resourceKey, + attribute_value: output.name, + attribute_scope: "resource", + start_time: output.timeRange.start, + end_time: output.timeRange.end, + }, + peak !== undefined && peak.max >= 0.9 + ? `log volume by severity around the ${peak.label} peak at ${peak.maxAt}` + : `log volume by severity from this ${output.kind}`, + ), + ]), + ] + return { + title: `${output.kind}: ${entity.name}`, + scope, + blocks, + notices, + next, + } + }, + }) +} diff --git a/apps/ai/src/mcp/tools/list-infra.ts b/apps/ai/src/mcp/tools/list-infra.ts new file mode 100644 index 000000000..42af0fde2 --- /dev/null +++ b/apps/ai/src/mcp/tools/list-infra.ts @@ -0,0 +1,271 @@ +import { Effect, Schema } from "effect" +import { ListInfraOutput } from "@maple/domain/mcp-outputs" +import type { McpToolRegistrar } from "./types" +import { warehouseToMcpHandlers } from "../lib/map-warehouse-error" +import { withTenantExecutor, CurrentMcpTenant } from "../lib/query-warehouse" +import { MCP_DISCOVERY_MAX_HOURS } from "../lib/time" +import * as P from "../lib/params" +import { doc, type DocBlock, type NextCall } from "../lib/tool-doc" +import { infraPresence, listInfraKind, WORKLOAD_KINDS, type InfraFilters } from "../lib/infra" +import { INFRA_KINDS, optionalInfraKind } from "../lib/infra-params" +import { + ENTITY_KIND, + KIND_METRICS, + KIND_NOUN, + hottest, + kindTable, + statusOf, + summaryLine, + windowText, +} from "../lib/infra-render" + +const TOOL = "list_infra" +const WINDOW = P.timeWindow({ defaultHours: 1, maxHours: MCP_DISCOVERY_MAX_HOURS }) + +const SORTS = ["saturation", "cpu", "memory", "name", "last_seen"] as const +const STATUSES = ["hot", "at_limit", "no_limits"] as const +/** Rows per kind in the all-kinds overview: enough to spot the hot ones. */ +const OVERVIEW_ROWS = 5 + +/** The filters each kind can apply; search and status apply to every kind. */ +const KIND_FILTERS = { + hosts: new Set(["host"]), + pods: new Set([ + "namespace", + "cluster", + "node", + "workload", + "workloadKind", + "environment", + ]), + nodes: new Set(["cluster", "node", "environment"]), + workloads: new Set([ + "namespace", + "cluster", + "workload", + "workloadKind", + "environment", + ]), + containers: new Set(["host", "workload", "environment"]), +} satisfies Record<(typeof INFRA_KINDS)[number], ReadonlySet> +const SCOPING_FILTERS: ReadonlyArray = [ + "host", + "namespace", + "cluster", + "node", + "workload", + "workloadKind", + "environment", +] + +/** In an overview, a kind that cannot apply a given filter would list unrelated rows; leave it out. */ +const supports = + (filters: InfraFilters) => + (kind: (typeof INFRA_KINDS)[number]): boolean => + SCOPING_FILTERS.every((key) => filters[key] === undefined || KIND_FILTERS[kind].has(key)) + +export function registerListInfraTool(server: McpToolRegistrar) { + server.define({ + name: TOOL, + title: "List Infrastructure", + description: + "Hosts, Kubernetes pods/nodes/workloads and Docker containers reporting resource metrics, closest to a limit first: CPU, memory and disk use, CPU/memory against limits, pod counts. " + + "Without `kind` it returns an overview of every kind that reports, with fleet counts and the top 5 of each. " + + "Filter with search, namespace, node, host, workload, and status (hot, at_limit, no_limits). " + + "Use it to answer 'is anything running hot?', 'which pods have no limits?', 'what runs on this node?', then inspect_infra for one entity over time. " + + "Status reflects the window peak, not the latest sample.", + parameters: Schema.Struct({ + kind: optionalInfraKind("Which kind to list. Omit for an overview of every kind that reports"), + search: P.optionalText("Case-insensitive substring of the entity name"), + namespace: P.optionalText("Only this Kubernetes namespace (pods, workloads)"), + cluster: P.optionalText("Only this Kubernetes cluster (pods, nodes, workloads)"), + node: P.optionalText("Only pods on this Kubernetes node"), + host: P.optionalText("Only containers on this host"), + workload: P.optionalText( + "Only pods owned by this workload (with workload_kind), or containers of this compose service", + ), + workload_kind: P.optionalOneOf( + WORKLOAD_KINDS, + "Kubernetes workload kind (workloads, and pods with `workload`)", + ), + environment: P.environment( + "Only this deployment environment (pods, nodes, workloads, containers)", + ), + status: P.optionalOneOf( + STATUSES, + "hot: peak at 60% or more of a limit (hosts: average CPU/memory or fullest disk); at_limit: 90% or more; no_limits: pods and workloads with no CPU or memory limit", + ), + include_ended: P.optionalFlag( + "Pods only: also list pods that stopped reporting during the window (rollouts, OOM kills)", + ), + sort: P.optionalOneOf(SORTS, "Sort order (default saturation: closest to a limit first)"), + limit: P.limit({ default: 20, max: 200, noun: "rows per kind" }), + ...WINDOW.fields, + }), + output: ListInfraOutput, + hints: { readOnly: true }, + phrases: ["Listing infrastructure", "Checking hosts, pods and containers"], + handler: Effect.fn("McpTool.listInfra")(function* (params) { + const { st, et } = yield* WINDOW.resolve(params, TOOL) + const tenant = yield* CurrentMcpTenant + const sort = params.sort ?? "saturation" + yield* Effect.annotateCurrentSpan({ orgId: tenant.orgId, kind: params.kind ?? "overview" }) + const window = { startTime: st, endTime: et } + const filters = { + search: params.search, + namespace: params.namespace, + cluster: params.cluster, + node: params.node, + host: params.host, + workload: params.workload, + workloadKind: params.workload_kind, + environment: params.environment, + status: params.status, + includeEnded: params.include_ended, + } + + const { reporting, sections } = yield* withTenantExecutor( + Effect.gen(function* () { + const reporting = yield* infraPresence(window) + const kinds = + params.kind === undefined ? reporting.filter(supports(filters)) : [params.kind] + const limit = + params.kind === undefined ? Math.min(params.limit, OVERVIEW_ROWS) : params.limit + const sections = yield* Effect.forEach( + kinds, + (kind) => listInfraKind(kind, { window, filters, sort, limit }), + { concurrency: 3 }, + ) + return { reporting, sections } + }), + ).pipe(Effect.catchTags(warehouseToMcpHandlers(TOOL))) + yield* Effect.annotateCurrentSpan( + "result.rowCount", + sections.reduce((sum, section) => sum + section.rows.length, 0), + ) + + return { + timeRange: { start: st, end: et }, + reporting, + sections, + ...(params.kind === undefined ? undefined : { kind: params.kind }), + ...(params.search === undefined ? undefined : { search: params.search }), + ...(params.sort === undefined ? undefined : { sort: params.sort }), + ...(params.status === undefined ? undefined : { status: params.status }), + } + }), + render: (output) => { + const notices: Array = [] + if (output.kind !== undefined && !output.reporting.includes(output.kind)) { + notices.push( + output.reporting.length === 0 + ? `No ${KIND_NOUN[output.kind]} report metrics in this window, and neither does any other kind.` + : `No ${KIND_NOUN[output.kind]} report metrics in this window. Reporting: ${output.reporting.join(", ")}.`, + ) + } + const nonEmpty = output.sections.filter((section) => section.rows.length > 0) + const window = { start_time: output.timeRange.start, end_time: output.timeRange.end } + const blocks: Array = nonEmpty.flatMap((section) => { + return [ + doc.heading(KIND_NOUN[section.kind]), + doc.text(summaryLine(section, output.status)), + kindTable(section), + doc.text(KIND_METRICS[section.kind]), + ...(section.truncated && output.kind === undefined + ? [doc.text(`More ${KIND_NOUN[section.kind]}: list_infra kind="${section.kind}".`)] + : []), + ] + }) + // A search that names one entity exactly is a request to look at that entity. + const exact = nonEmpty.flatMap((section) => + section.rows.filter((row) => row.name === output.search).map((row) => ({ section, row })), + ) + const exactNext: Array = exact.map(({ section, row }) => + doc.next( + "inspect_infra", + { + kind: ENTITY_KIND[section.kind], + name: row.name, + namespace: row.namespace, + host: section.kind === "containers" ? row.host : undefined, + ...window, + }, + section.kind === "nodes" + ? `${row.name}: its pods, CPU over the window and peer nodes` + : `${row.name} over the window`, + ), + ) + const next: Array = nonEmpty.flatMap((section) => { + const hot = hottest(section) + const target = hot ?? section.rows[0] + if (target === undefined) return [] + return [ + doc.next( + "inspect_infra", + { + kind: ENTITY_KIND[section.kind], + name: target.name, + namespace: target.namespace, + host: section.kind === "containers" ? target.host : undefined, + workload_kind: + section.kind === "workloads" && target.workloadKind !== undefined + ? target.workloadKind + : undefined, + ...window, + }, + hot === undefined + ? `see ${target.name} over the same window` + : `${target.name} (${statusOf(hot, section.kind)}): see when it peaked`, + ), + ] + }) + const truncatedSingle = output.kind !== undefined ? nonEmpty[0] : undefined + return { + title: + output.kind === undefined + ? "Infrastructure overview" + : `Infrastructure: ${KIND_NOUN[output.kind]}`, + scope: [ + ["Window", `${windowText(output.timeRange)}; start_time widens it`], + ["Reporting", output.reporting.length === 0 ? "nothing" : output.reporting.join(", ")], + ["Search", output.search], + ["Sort", output.sort], + ], + ...(nonEmpty.length === 0 + ? { + empty: { + message: + output.reporting.length === 0 + ? "No infrastructure metrics in this window: no hostmetrics, kubeletstats or docker stats receivers are reporting." + : "Nothing matched these filters.", + hints: + output.reporting.length === 0 + ? [ + "Widen start_time/end_time if the collector was down.", + "Infra pages need an OpenTelemetry Collector with the hostmetrics, kubeletstats or docker_stats receiver; audit_setup checks what is configured.", + ] + : ["Drop search/namespace/status, or pick a kind from Reporting."], + }, + } + : undefined), + blocks, + notices, + ...(truncatedSingle?.truncated === true + ? { + truncation: { + shown: truncatedSingle.rows.length, + noun: KIND_NOUN[truncatedSingle.kind], + }, + } + : undefined), + next: [ + ...exactNext, + ...next.filter((call) => !exactNext.some((e) => e.args.name === call.args.name)), + ...(output.reporting.length === 0 + ? [doc.next("audit_setup", {}, "check which collectors report")] + : []), + ], + } + }, + }) +} diff --git a/apps/ai/src/mcp/tools/registry.ts b/apps/ai/src/mcp/tools/registry.ts index f0b8f7fa7..04f3e71ac 100644 --- a/apps/ai/src/mcp/tools/registry.ts +++ b/apps/ai/src/mcp/tools/registry.ts @@ -26,6 +26,8 @@ import { registerServiceDeploymentsTool } from "./service-deployments" import { registerRouteUsageTool } from "./route-usage" import { registerIngestFreshnessTool } from "./ingest-freshness" import { registerDbQueryVolumeTool } from "./db-query-volume" +import { registerListInfraTool } from "./list-infra" +import { registerInspectInfraTool } from "./inspect-infra" import { registerIngestUsageTool } from "./ingest-usage" import { registerInspectChartDataTool } from "./inspect-chart-data" import { registerInspectTraceTool } from "./inspect-trace" @@ -408,6 +410,8 @@ const collectMapleToolDefinitions = (): ReadonlyArray => { registerRouteUsageTool(registrar) registerIngestFreshnessTool(registrar) registerDbQueryVolumeTool(registrar) + registerListInfraTool(registrar) + registerInspectInfraTool(registrar) registerIngestUsageTool(registrar) registerGetInstrumentationRecommendationsTool(registrar) registerAuditSetupTool(registrar) diff --git a/apps/ai/vitest.playground.config.ts b/apps/ai/vitest.playground.config.ts new file mode 100644 index 000000000..e10b1293c --- /dev/null +++ b/apps/ai/vitest.playground.config.ts @@ -0,0 +1,20 @@ +import { fileURLToPath } from "node:url" +import { defineConfig } from "vitest/config" + +// The tool playground (`scripts/tool-playground.sh`): one tool call per run against a fixture +// world, for agents testing the tools by using them. Vitest only because the eval runtime's +// PGlite snapshot comes from the global setup. +export default defineConfig({ + resolve: { + alias: { + "@": fileURLToPath(new URL("./src", import.meta.url)), + }, + }, + test: { + environment: "node", + include: ["src/evals/playground/*.playground.ts"], + globalSetup: ["../../packages/backend/test/global-setup.ts"], + testTimeout: 60_000, + hookTimeout: 60_000, + }, +}) diff --git a/packages/domain/src/mcp-outputs/catalog.ts b/packages/domain/src/mcp-outputs/catalog.ts index 0b49758ce..7407ca543 100644 --- a/packages/domain/src/mcp-outputs/catalog.ts +++ b/packages/domain/src/mcp-outputs/catalog.ts @@ -3,6 +3,7 @@ * each tool declares exactly this schema) and the chat UI's `StructuredToolOutput` derive from. */ import { SendMapleFeedbackOutput } from "./feedback" +import { InspectInfraOutput, ListInfraOutput } from "./infra" import { CreateAlertRuleOutput, DeleteAlertRuleOutput, @@ -127,6 +128,7 @@ export const McpToolOutputs = { ingest_freshness: IngestFreshnessOutput, ingest_usage: IngestUsageOutput, inspect_chart_data: InspectChartDataOutput, + inspect_infra: InspectInfraOutput, inspect_span: InspectSpanOutput, inspect_trace: InspectTraceOutput, link_pull_request: LinkPullRequestOutput, @@ -140,6 +142,7 @@ export const McpToolOutputs = { list_error_incidents: ListErrorIncidentsOutput, list_error_issue_events: ListErrorIssueEventsOutput, list_error_issues: ListErrorIssuesOutput, + list_infra: ListInfraOutput, list_metrics: ListMetricsOutput, list_product_events: ListProductEventsOutput, list_services: ListServicesOutput, diff --git a/packages/domain/src/mcp-outputs/index.ts b/packages/domain/src/mcp-outputs/index.ts index 1baedc6de..2d9a4633f 100644 --- a/packages/domain/src/mcp-outputs/index.ts +++ b/packages/domain/src/mcp-outputs/index.ts @@ -9,4 +9,5 @@ export * from "./issues" export * from "./setup" export * from "./feedback" export * from "./service-activity" +export * from "./infra" export * from "./catalog" diff --git a/packages/domain/src/mcp-outputs/infra.ts b/packages/domain/src/mcp-outputs/infra.ts new file mode 100644 index 000000000..6a2a4c8ec --- /dev/null +++ b/packages/domain/src/mcp-outputs/infra.ts @@ -0,0 +1,114 @@ +/** Output schemas for the infrastructure MCP tools (hosts, Kubernetes, containers). */ +import { Schema } from "effect" +import { OutputTimeRange } from "./shared" + +export const InfraKind = Schema.Literals(["hosts", "pods", "nodes", "workloads", "containers"]) +export type InfraKind = typeof InfraKind.Type + +export const InfraEntityKind = Schema.Literals(["host", "pod", "node", "workload", "container"]) +export type InfraEntityKind = typeof InfraEntityKind.Type + +/** + * One infrastructure entity, flattened across kinds. Utilization fields are 0..1 fractions; + * a field is absent when the kind does not report it, never zero-filled. + */ +export const InfraEntityRow = Schema.Struct({ + name: Schema.String, + lastSeen: Schema.String, + namespace: Schema.optionalKey(Schema.String), + cluster: Schema.optionalKey(Schema.String), + environment: Schema.optionalKey(Schema.String), + node: Schema.optionalKey(Schema.String), + host: Schema.optionalKey(Schema.String), + /** `deployment/checkout` style owner of a pod. */ + workload: Schema.optionalKey(Schema.String), + workloadKind: Schema.optionalKey(Schema.String), + image: Schema.optionalKey(Schema.String), + os: Schema.optionalKey(Schema.String), + cpuCores: Schema.optionalKey(Schema.Number), + cpuCoresPeak: Schema.optionalKey(Schema.Number), + /** Host/container CPU utilization, or pod CPU as a fraction of its limit. */ + cpu: Schema.optionalKey(Schema.Number), + cpuPeak: Schema.optionalKey(Schema.Number), + /** Host/container memory utilization, or pod memory as a fraction of its limit. */ + memory: Schema.optionalKey(Schema.Number), + memoryPeak: Schema.optionalKey(Schema.Number), + disk: Schema.optionalKey(Schema.Number), + load15: Schema.optionalKey(Schema.Number), + /** Peak of CPU and memory against their limits. 0 when no limit is set. */ + saturation: Schema.optionalKey(Schema.Number), + /** Running with no CPU or memory limit, so saturation cannot be measured. */ + unbounded: Schema.optionalKey(Schema.Boolean), + podCount: Schema.optionalKey(Schema.Number), + uptimeSeconds: Schema.optionalKey(Schema.Number), + /** Container restarts in the window (k8s.container.restarts, or Docker's container.restarts). */ + restarts: Schema.optionalKey(Schema.Number), +}) +export type InfraEntityRow = typeof InfraEntityRow.Type + +/** Fleet counts for one kind, where the warehouse reports them. */ +export const InfraFleetSummary = Schema.Struct({ + total: Schema.Number, + saturated: Schema.optionalKey(Schema.Number), + elevated: Schema.optionalKey(Schema.Number), + unbounded: Schema.optionalKey(Schema.Number), + stale: Schema.optionalKey(Schema.Number), + ended: Schema.optionalKey(Schema.Number), +}) + +export const InfraKindSection = Schema.Struct({ + kind: InfraKind, + summary: Schema.optionalKey(InfraFleetSummary), + rows: Schema.Array(InfraEntityRow), + truncated: Schema.Boolean, +}) +export type InfraKindSection = typeof InfraKindSection.Type + +// list_infra + +export const ListInfraOutput = Schema.Struct({ + timeRange: OutputTimeRange, + /** Kinds reporting telemetry in the window, whatever was asked for. */ + reporting: Schema.Array(InfraKind), + sections: Schema.Array(InfraKindSection), + kind: Schema.optionalKey(InfraKind), + search: Schema.optionalKey(Schema.String), + sort: Schema.optionalKey(Schema.String), + status: Schema.optionalKey(Schema.String), +}) + +// inspect_infra + +export const InfraSeriesStats = Schema.Struct({ + /** What the series measures, e.g. "cpu (fraction of limit)". */ + label: Schema.String, + unit: Schema.Literals(["fraction", "cores", "load", "bytes", "seconds"]), + /** Series value per bucket for grouped metrics (filesystem by mountpoint, pods of a workload). */ + group: Schema.optionalKey(Schema.String), + min: Schema.Number, + avg: Schema.Number, + max: Schema.Number, + last: Schema.Number, + maxAt: Schema.String, + points: Schema.Number, + /** Mean of the last third of the window minus the mean of the first third. */ + change: Schema.Number, +}) +export type InfraSeriesStats = typeof InfraSeriesStats.Type + +export const InspectInfraOutput = Schema.Struct({ + timeRange: OutputTimeRange, + kind: InfraEntityKind, + name: Schema.String, + found: Schema.Boolean, + /** Width of each series bucket; min/avg/max are over bucket averages, not raw samples. */ + bucketSeconds: Schema.Number, + entity: Schema.optionalKey(InfraEntityRow), + firstSeen: Schema.optionalKey(Schema.String), + details: Schema.optionalKey(Schema.Record(Schema.String, Schema.String)), + series: Schema.Array(InfraSeriesStats), + /** Pods of a workload, or containers on a host, busiest first. */ + children: Schema.optionalKey(InfraKindSection), + /** Services whose spans name this workload (k8sattributes). */ + services: Schema.optionalKey(Schema.Array(Schema.String)), +}) diff --git a/packages/domain/src/mcp-outputs/services.ts b/packages/domain/src/mcp-outputs/services.ts index 72168d3ef..6c05031f8 100644 --- a/packages/domain/src/mcp-outputs/services.ts +++ b/packages/domain/src/mcp-outputs/services.ts @@ -1,6 +1,7 @@ /** Output schemas for the services, metrics and query MCP tools. */ import { Schema } from "effect" import { OutputPagination, OutputTimeRange } from "./shared" +import { InfraEntityRow } from "./infra" // list_services @@ -59,6 +60,27 @@ export const DiagnoseServiceOutput = Schema.Struct({ recentTraces: Schema.Array(DiagnoseServiceTraceRow), recentLogs: Schema.Array(DiagnoseServiceLogRow), environment: Schema.optionalKey(Schema.String), + /** Where the service runs, from the Kubernetes workload its spans name. Absent without k8s context. */ + infrastructure: Schema.optionalKey( + Schema.Struct({ + workloads: Schema.Array( + Schema.Struct({ + kind: Schema.String, + name: Schema.String, + namespace: Schema.String, + cluster: Schema.optionalKey(Schema.String), + podCount: Schema.Number, + /** Average CPU / memory against the pods' limits, 0..1. Absent when no limit is set. */ + cpu: Schema.optionalKey(Schema.Number), + memory: Schema.optionalKey(Schema.Number), + }), + ), + /** The busiest pods of the dominant workload, closest to a limit first. */ + pods: Schema.Array(InfraEntityRow), + }), + ), + /** Set when the infrastructure lookup failed; the rest of the diagnosis still stands. */ + infrastructureError: Schema.optionalKey(Schema.String), }) // get_service_top_operations diff --git a/packages/query-engine/src/benchmark/builders.ts b/packages/query-engine/src/benchmark/builders.ts index e13885d50..eafcb6468 100644 --- a/packages/query-engine/src/benchmark/builders.ts +++ b/packages/query-engine/src/benchmark/builders.ts @@ -1226,6 +1226,26 @@ export const builderFixtures: ReadonlyArray = [ }, // Infra gauge timeseries and facet unions. + { + module: "infra", + name: "hostFilesystemsQuery", + label: "default", + compile: () => CH.compileUnsafe(CH.hostFilesystemsQuery({ hostName: "ip-10-0-1-42" }), window), + }, + { + module: "infra", + name: "podRestartsQuery", + label: "default", + compile: () => + CH.compileUnsafe( + CH.podRestartsQuery({ + workloadKind: "deployment", + workloadName: "api", + namespace: "backend", + }), + window, + ), + }, { module: "infra", name: "hostGaugeTimeseriesQuery", diff --git a/packages/query-engine/src/ch/index.ts b/packages/query-engine/src/ch/index.ts index 7ece54c1a..fd8163351 100644 --- a/packages/query-engine/src/ch/index.ts +++ b/packages/query-engine/src/ch/index.ts @@ -590,6 +590,8 @@ export { hostDetailSummaryQuery, hostGaugeTimeseriesQuery, hostNetworkTimeseriesQuery, + hostFilesystemsQuery, + podRestartsQuery, listPodsQuery, listPodsSummaryQuery, podDetailSummaryQuery, @@ -611,6 +613,10 @@ export { type HostGaugeTimeseriesOutput, type HostNetworkTimeseriesOpts, type HostNetworkTimeseriesOutput, + type HostFilesystemsOpts, + type HostFilesystemsOutput, + type PodRestartsOpts, + type PodRestartsOutput, type ListPodsOpts, type ListPodsOutput, type ListPodsSummaryOutput, diff --git a/packages/query-engine/src/ch/queries/infra.ts b/packages/query-engine/src/ch/queries/infra.ts index 15c27ee3f..7c4c9d6d7 100644 --- a/packages/query-engine/src/ch/queries/infra.ts +++ b/packages/query-engine/src/ch/queries/infra.ts @@ -213,6 +213,89 @@ export function hostNetworkTimeseriesQuery(opts: HostNetworkTimeseriesOpts) { .format("JSON") } +// Host filesystems: used fraction per mountpoint. `state` must be pinned to `used`; +// averaging across used/free/reserved reads every disk as about a third full. + +export interface HostFilesystemsOpts { + hostName: string + limit?: number +} + +export interface HostFilesystemsOutput { + readonly mountpoint: string + readonly device: string + readonly usedAvg: number + readonly usedMax: number +} + +export function hostFilesystemsQuery(opts: HostFilesystemsOpts) { + return from(MetricsGauge) + .select(($) => ({ + mountpoint: $.Attributes.get("mountpoint"), + device: CH.any_($.Attributes.get("device")), + usedAvg: avgIfOrZero($.Value, $.Attributes.get("state").eq("used")), + usedMax: maxIfOrZero($.Value, $.Attributes.get("state").eq("used")), + })) + .where(($) => [ + $.OrgId.eq(orgIdParam), + $.TimeUnix.gte(param.dateTime("startTime")), + $.TimeUnix.lte(param.dateTime("endTime")), + $.ResourceAttributes.get("host.name").eq(opts.hostName), + $.MetricName.eq("system.filesystem.utilization"), + $.Attributes.get("state").eq("used"), + ]) + .groupBy("mountpoint") + .orderBy(["usedMax", "desc"]) + .limit(opts.limit ?? 20) + .format("JSON") +} + +// Pod container restarts, from the k8s_cluster receiver's `k8s.container.restarts` +// (a running total per container). The window's restarts are max − min per container. + +export interface PodRestartsOpts { + podNames?: ReadonlyArray + namespace?: string + workloadKind?: "deployment" | "statefulset" | "daemonset" + workloadName?: string + limit?: number +} + +export interface PodRestartsOutput { + readonly podName: string + readonly containerName: string + readonly restarts: number + readonly totalRestarts: number +} + +export function podRestartsQuery(opts: PodRestartsOpts) { + return from(MetricsGauge) + .select(($) => ({ + podName: $.ResourceAttributes.get("k8s.pod.name"), + containerName: $.ResourceAttributes.get("k8s.container.name"), + restarts: CH.ifNotFinite(CH.max_($.Value).sub(CH.min_($.Value)), 0), + totalRestarts: CH.ifNotFinite(CH.max_($.Value), 0), + })) + .where(($) => [ + $.OrgId.eq(orgIdParam), + $.TimeUnix.gte(param.dateTime("startTime")), + $.TimeUnix.lte(param.dateTime("endTime")), + $.MetricName.eq("k8s.container.restarts"), + $.ResourceAttributes.get("k8s.pod.name").neq(""), + CH.when(opts.podNames, (names: ReadonlyArray) => + $.ResourceAttributes.get("k8s.pod.name").in_(...names), + ), + CH.when(opts.namespace, (v: string) => $.ResourceAttributes.get("k8s.namespace.name").eq(v)), + CH.when(opts.workloadName, (v: string) => + $.ResourceAttributes.get(workloadAttrKey(opts.workloadKind ?? "deployment")).eq(v), + ), + ]) + .groupBy("podName", "containerName") + .orderBy(["restarts", "desc"]) + .limit(opts.limit ?? 50) + .format("JSON") +} + // Kubernetes — pod aggregations over k8s.pod.* metrics emitted by the kubelet // stats receiver. Identity carried on ResourceAttributes: // k8s.pod.name, k8s.pod.uid, k8s.namespace.name, k8s.node.name, From 22602486e0ba7e33a01d403eec827b0f13b09e84 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Sun, 11 Oct 2026 01:22:29 +0200 Subject: [PATCH 2/3] test(ai): escape every regex metacharacter in the infra fixture's key matcher --- apps/ai/src/mcp/__evals__/infra-world.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/ai/src/mcp/__evals__/infra-world.ts b/apps/ai/src/mcp/__evals__/infra-world.ts index 438b44262..6b5ba4564 100644 --- a/apps/ai/src/mcp/__evals__/infra-world.ts +++ b/apps/ai/src/mcp/__evals__/infra-world.ts @@ -172,7 +172,7 @@ const firstSeen = (sql: string) => fmt(windowOf(sql).start + 30_000) /** Values for `ResourceAttributes['key'] = 'v'` or `IN ('a', 'b')` in the SQL; undefined when unfiltered. */ const filterValues = (sql: string, key: string): ReadonlyArray | undefined => { - const escaped = key.replace(/\./g, "\\.") + const escaped = key.replace(/[.*+?^${}()|[\]\\]/g, "\\$&") const eq = [...sql.matchAll(new RegExp(`\\['${escaped}'\\] = '([^']*)'`, "g"))].map((m) => m[1] ?? "") const inList = [...sql.matchAll(new RegExp(`\\['${escaped}'\\] IN \\(([^)]*)\\)`, "g"))].flatMap((m) => [...(m[1] ?? "").matchAll(/'([^']*)'/g)].map((v) => v[1] ?? ""), From d8c59875346a7b25870f3e293609a01fc99c4205 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Sun, 11 Oct 2026 01:29:06 +0200 Subject: [PATCH 3/3] fix(ai): key pod restarts by namespace and pod name Pod names repeat across namespaces, so list_infra summed restarts of same-named pods into every one of their rows. --- apps/ai/src/mcp/__evals__/infra-world.ts | 1 + apps/ai/src/mcp/lib/infra.ts | 12 +++++---- .../mcp/tools/__tests__/infra-tools.test.ts | 25 +++++++++++++++++++ packages/query-engine/src/ch/queries/infra.ts | 4 ++- 4 files changed, 36 insertions(+), 6 deletions(-) diff --git a/apps/ai/src/mcp/__evals__/infra-world.ts b/apps/ai/src/mcp/__evals__/infra-world.ts index 6b5ba4564..d18644536 100644 --- a/apps/ai/src/mcp/__evals__/infra-world.ts +++ b/apps/ai/src/mcp/__evals__/infra-world.ts @@ -575,6 +575,7 @@ export const infraFixtureRules = (): FixtureRule[] => [ match: (sql) => has(sql, "totalRestarts"), rows: (sql) => podsFor(sql).map((p) => ({ + namespace: p.namespace, podName: p.podName, containerName: p.owner[1], restarts: RESTARTS.get(p.podName) ?? 0, diff --git a/apps/ai/src/mcp/lib/infra.ts b/apps/ai/src/mcp/lib/infra.ts index 1f695e245..db9297509 100644 --- a/apps/ai/src/mcp/lib/infra.ts +++ b/apps/ai/src/mcp/lib/infra.ts @@ -246,6 +246,9 @@ const containerRow = (row: CH.ListContainersOutput): InfraEntityRow => ({ uptimeSeconds: row.uptimeSeconds, }) +/** Pod names repeat across namespaces (`redis-0` in staging and prod), so restarts key on both. */ +const podKey = (namespace: string | undefined, podName: string) => `${namespace ?? ""}/${podName}` + /** Container restarts per pod in the window; pods the k8s_cluster receiver does not report are absent. */ const podRestarts = (podNames: ReadonlyArray, params: ReturnType) => Effect.gen(function* () { @@ -256,10 +259,9 @@ const podRestarts = (podNames: ReadonlyArray, params: ReturnType row.podName)).map(([pod, containers]) => [ - pod, - containers.reduce((sum, row) => sum + row.restarts, 0), - ]), + Object.entries(Arr.groupBy(rows, (row) => podKey(row.namespace, row.podName))).map( + ([pod, containers]) => [pod, containers.reduce((sum, row) => sum + row.restarts, 0)], + ), ) }) @@ -347,7 +349,7 @@ const fetchKind = (kind: InfraKind, options: ListKindOptions) => return { rows: rows.map((row) => { const pod = podRow(row) - const restarted = restarts.get(row.podName) + const restarted = restarts.get(podKey(row.namespace, row.podName)) return restarted === undefined ? pod : { ...pod, restarts: restarted } }), summary: diff --git a/apps/ai/src/mcp/tools/__tests__/infra-tools.test.ts b/apps/ai/src/mcp/tools/__tests__/infra-tools.test.ts index fe1dfbf26..e8f82319f 100644 --- a/apps/ai/src/mcp/tools/__tests__/infra-tools.test.ts +++ b/apps/ai/src/mcp/tools/__tests__/infra-tools.test.ts @@ -67,6 +67,31 @@ describe("list_infra", () => { expect(output.sections[0]?.rows.find((r) => r.name === INFRA_WORLD.oomPod)?.restarts).toBe(1) }) + it("keys pod restarts by namespace as well as name", async () => { + rules.unshift({ + match: (sql) => sql.includes(" AS totalRestarts"), + rows: [ + { + namespace: "shop", + podName: INFRA_WORLD.oomPod, + containerName: "checkout", + restarts: 1, + totalRestarts: 3, + }, + { + namespace: "staging", + podName: INFRA_WORLD.oomPod, + containerName: "checkout", + restarts: 7, + totalRestarts: 9, + }, + ], + }) + const output = decode(ListInfraOutput, await call("list_infra", { kind: "pods" })) + rules.shift() + expect(output.sections[0]?.rows.find((r) => r.name === INFRA_WORLD.oomPod)?.restarts).toBe(1) + }) + it("lists only pods without limits", async () => { const output = decode( ListInfraOutput, diff --git a/packages/query-engine/src/ch/queries/infra.ts b/packages/query-engine/src/ch/queries/infra.ts index 7c4c9d6d7..0eb106a35 100644 --- a/packages/query-engine/src/ch/queries/infra.ts +++ b/packages/query-engine/src/ch/queries/infra.ts @@ -262,6 +262,7 @@ export interface PodRestartsOpts { } export interface PodRestartsOutput { + readonly namespace: string readonly podName: string readonly containerName: string readonly restarts: number @@ -271,6 +272,7 @@ export interface PodRestartsOutput { export function podRestartsQuery(opts: PodRestartsOpts) { return from(MetricsGauge) .select(($) => ({ + namespace: $.ResourceAttributes.get("k8s.namespace.name"), podName: $.ResourceAttributes.get("k8s.pod.name"), containerName: $.ResourceAttributes.get("k8s.container.name"), restarts: CH.ifNotFinite(CH.max_($.Value).sub(CH.min_($.Value)), 0), @@ -290,7 +292,7 @@ export function podRestartsQuery(opts: PodRestartsOpts) { $.ResourceAttributes.get(workloadAttrKey(opts.workloadKind ?? "deployment")).eq(v), ), ]) - .groupBy("podName", "containerName") + .groupBy("namespace", "podName", "containerName") .orderBy(["restarts", "desc"]) .limit(opts.limit ?? 50) .format("JSON")