From 1e047f7e15d8dc0d7ed0ee08a612a43b7f1889cd Mon Sep 17 00:00:00 2001 From: Qindeel Ishtiaq Date: Fri, 16 Oct 2026 00:12:39 -0400 Subject: [PATCH] console: show how a replica's workers share its CPU on the redesigned cluster page Adds a Workers button to the redesigned cluster page's CPU chart. It opens a drawer that reads the replica's per-worker scheduling counters over one WebSocket session while open, and leads with a finding in the CPU troubleshooting guide's terms: a saturated or uneven worker, every worker busy, high CPU inside or outside running dataflows, idle, or balanced. Below it are the replica's CPU against its time running dataflows, the busiest and average worker over time, the busiest workers, and a table of objects with one bar per worker against an even share. --- .../materialize/WebsocketQueryClient.test.ts | 180 ++++++ .../api/materialize/WebsocketQueryClient.ts | 137 +++++ .../cluster/replicaProcessMetrics.ts | 3 + .../src/api/materialize/cluster/workerCpu.ts | 64 ++ console/src/api/materialize/executeSqlV2.ts | 2 +- .../api/materialize/subscribeCollection.ts | 4 + console/src/api/materialize/types.ts | 1 + console/src/mz-doc-urls.json | 1 + .../clusters/ClusterPage/MemoryByObject.tsx | 30 +- .../ResourceUsage/ResourceUsage.tsx | 63 +- .../ResourceUsage/ResourceUsageCharts.tsx | 8 +- .../processReadingsModel.test.ts | 1 + .../ClusterPage/WorkerCpu/BusiestWorkers.tsx | 212 +++++++ .../WorkerCpu/ObjectCpuTable.test.tsx | 106 ++++ .../ClusterPage/WorkerCpu/ObjectCpuTable.tsx | 522 ++++++++++++++++ .../ClusterPage/WorkerCpu/WorkerCpuCharts.tsx | 495 +++++++++++++++ .../ClusterPage/WorkerCpu/WorkerCpuDrawer.tsx | 441 ++++++++++++++ .../WorkerCpu/WorkerCpuVerdict.test.tsx | 221 +++++++ .../WorkerCpu/WorkerCpuVerdict.tsx | 377 ++++++++++++ .../ClusterPage/WorkerCpu/useReplicaCpu.ts | 77 +++ .../WorkerCpu/workerCpuModel.test.ts | 463 ++++++++++++++ .../ClusterPage/WorkerCpu/workerCpuModel.ts | 569 ++++++++++++++++++ .../WorkerCpu/workerCpuSampler.test.ts | 382 ++++++++++++ .../ClusterPage/WorkerCpu/workerCpuSampler.ts | 395 ++++++++++++ .../ClusterPage/WorkerCpu/workerCpuStyles.ts | 48 ++ .../clusters/ClusterPage/useObjectLink.ts | 46 ++ 26 files changed, 4813 insertions(+), 35 deletions(-) create mode 100644 console/src/api/materialize/WebsocketQueryClient.test.ts create mode 100644 console/src/api/materialize/WebsocketQueryClient.ts create mode 100644 console/src/api/materialize/cluster/workerCpu.ts create mode 100644 console/src/platform/clusters/ClusterPage/WorkerCpu/BusiestWorkers.tsx create mode 100644 console/src/platform/clusters/ClusterPage/WorkerCpu/ObjectCpuTable.test.tsx create mode 100644 console/src/platform/clusters/ClusterPage/WorkerCpu/ObjectCpuTable.tsx create mode 100644 console/src/platform/clusters/ClusterPage/WorkerCpu/WorkerCpuCharts.tsx create mode 100644 console/src/platform/clusters/ClusterPage/WorkerCpu/WorkerCpuDrawer.tsx create mode 100644 console/src/platform/clusters/ClusterPage/WorkerCpu/WorkerCpuVerdict.test.tsx create mode 100644 console/src/platform/clusters/ClusterPage/WorkerCpu/WorkerCpuVerdict.tsx create mode 100644 console/src/platform/clusters/ClusterPage/WorkerCpu/useReplicaCpu.ts create mode 100644 console/src/platform/clusters/ClusterPage/WorkerCpu/workerCpuModel.test.ts create mode 100644 console/src/platform/clusters/ClusterPage/WorkerCpu/workerCpuModel.ts create mode 100644 console/src/platform/clusters/ClusterPage/WorkerCpu/workerCpuSampler.test.ts create mode 100644 console/src/platform/clusters/ClusterPage/WorkerCpu/workerCpuSampler.ts create mode 100644 console/src/platform/clusters/ClusterPage/WorkerCpu/workerCpuStyles.ts create mode 100644 console/src/platform/clusters/ClusterPage/useObjectLink.ts diff --git a/console/src/api/materialize/WebsocketQueryClient.test.ts b/console/src/api/materialize/WebsocketQueryClient.test.ts new file mode 100644 index 0000000000000..4a3e7346f3dbf --- /dev/null +++ b/console/src/api/materialize/WebsocketQueryClient.test.ts @@ -0,0 +1,180 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +import { CompiledQuery } from "kysely"; +import { ws } from "msw"; + +import { Column, MzDataType } from "~/api/materialize/types"; +import { buildColumn } from "~/api/mocks/buildSqlQueryHandler"; +import server from "~/api/mocks/server"; + +import { + createWebsocketQueryClient, + WebsocketQueryClient, +} from "./WebsocketQueryClient"; + +interface SentQuery { + query: string; + params: unknown[]; +} + +type Answer = { columns: Column[]; rows: unknown[][] } | { error: string }; + +const READY = { type: "ReadyForQuery", payload: "I" }; + +const WORKER_COLUMNS = [ + buildColumn({ name: "workerId", type_oid: MzDataType.int4 }), + buildColumn({ name: "elapsedNs", type_oid: MzDataType.float8 }), +]; + +/** + * Serves the WebSocket SQL protocol. Each query is answered on a later task, + * so a query sent without waiting for the previous one's `ReadyForQuery` + * arrives while that one is still open and counts as overlapping. Returning + * "close" from `answer` closes the socket instead of answering. + */ +function serveQueries(answer: (query: SentQuery) => Answer | "close") { + const received: SentQuery[] = []; + const socket = { received, overlapped: false }; + let isAnswering = false; + server.use( + ws.link(/^wss?:\/\//).addEventListener("connection", ({ client }) => { + let hasOptions = false; + const send = (message: unknown) => client.send(JSON.stringify(message)); + client.addEventListener("message", (event) => { + // The first message carries the session options. + if (!hasOptions) { + hasOptions = true; + send(READY); + return; + } + const { + queries: [query], + } = JSON.parse(String(event.data)) as { queries: SentQuery[] }; + received.push(query); + if (isAnswering) socket.overlapped = true; + isAnswering = true; + const reply = answer(query); + setTimeout(() => { + isAnswering = false; + if (reply === "close") { + client.close(); + return; + } + if ("error" in reply) { + send({ + type: "Error", + payload: { message: reply.error, code: "XX000" }, + }); + } else { + send({ type: "Rows", payload: { columns: reply.columns } }); + for (const row of reply.rows) send({ type: "Row", payload: row }); + send({ type: "CommandComplete", payload: "SELECT" }); + } + send(READY); + }, 10); + }); + }), + ); + return socket; +} + +const workerRows = (rows: unknown[][]): Answer => ({ + columns: WORKER_COLUMNS, + rows, +}); + +const clients: WebsocketQueryClient[] = []; + +const createClient = () => { + const client = createWebsocketQueryClient({ + httpAddress: "localhost:6876", + }); + clients.push(client); + return client; +}; + +const query = (sql: string, parameters: unknown[] = []) => + CompiledQuery.raw(sql, parameters) as CompiledQuery>; + +describe("createWebsocketQueryClient", () => { + afterEach(() => { + for (const client of clients.splice(0)) client.close(); + }); + + it("runs queued queries one at a time and resolves each with its own rows keyed by column", async () => { + const socket = serveQueries(({ query: sql }) => + sql === "SELECT 1" + ? workerRows([["0", "1.5"]]) + : workerRows([ + ["0", "2"], + ["1", "3"], + ]), + ); + const client = createClient(); + + const results = await Promise.all([ + client.query(query("SELECT 1")), + client.query(query("SELECT 2")), + ]); + + expect(results).toEqual([ + [{ workerId: 0, elapsedNs: 1.5 }], + [ + { workerId: 0, elapsedNs: 2 }, + { workerId: 1, elapsedNs: 3 }, + ], + ]); + expect(socket.received.map(({ query: sql }) => sql)).toEqual([ + "SELECT 1", + "SELECT 2", + ]); + expect(socket.overlapped).toBe(false); + }); + + it("rejects only the query that errored and answers the next one", async () => { + serveQueries(({ query: sql }) => + sql === "SELECT 1/0" + ? { error: "division by zero" } + : workerRows([["0", "1"]]), + ); + const client = createClient(); + + const failed = client.query(query("SELECT 1/0")); + const next = client.query(query("SELECT 1")); + + await expect(failed).rejects.toThrow("division by zero"); + await expect(next).resolves.toEqual([{ workerId: 0, elapsedNs: 1 }]); + }); + + it("rejects the pending query and every later one once the socket closes", async () => { + serveQueries(() => "close"); + const client = createClient(); + + await expect(client.query(query("SELECT 1"))).rejects.toThrow( + "Connection closed", + ); + await expect(client.query(query("SELECT 2"))).rejects.toThrow( + "Connection closed", + ); + }); + + it("sends numeric parameters as strings", async () => { + const socket = serveQueries(() => workerRows([])); + const client = createClient(); + + await client.query( + query("SELECT $1::int4, $2::float8, $3", [42, 1.5, "a"]), + ); + + expect(socket.received).toEqual([ + { query: "SELECT $1::int4, $2::float8, $3", params: ["42", "1.5", "a"] }, + ]); + }); +}); diff --git a/console/src/api/materialize/WebsocketQueryClient.ts b/console/src/api/materialize/WebsocketQueryClient.ts new file mode 100644 index 0000000000000..4363f5b54c8e9 --- /dev/null +++ b/console/src/api/materialize/WebsocketQueryClient.ts @@ -0,0 +1,137 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +import { CompiledQuery } from "kysely"; + +import { + buildSessionVariables, + mapColumnToColumnMetadata, + mapRowToObject, +} from "."; +import { buildExtendedRequestQuery } from "./executeSqlV2"; +import { MaterializeWebsocket, WebSocketResult } from "./MaterializeWebsocket"; +import { ColumnMetadata, SessionVariables } from "./types"; + +class WebsocketQueryError extends Error {} + +/** + * Runs queries one at a time over a single WebSocket session. + * + * Every HTTP request opens a new session, and each session start costs + * `mz_catalog_server` about 35 ms of CPU (measured on an M.1-nano), so a + * caller that repeats a query every few seconds should hold one of these + * instead. Rows come back keyed and converted as `executeSqlV2` returns them. + */ +export interface WebsocketQueryClient { + query: (query: CompiledQuery) => Promise; + close: () => void; +} + +interface PendingQuery { + resolve: (rows: unknown[]) => void; + reject: (error: Error) => void; + columns: ColumnMetadata[]; + rows: unknown[]; + error?: Error; +} + +export function createWebsocketQueryClient({ + httpAddress, + sessionVariables, +}: { + httpAddress: string; + sessionVariables?: SessionVariables; +}): WebsocketQueryClient { + let closedError: Error | undefined; + let pending: PendingQuery | undefined; + let markReady: () => void = () => {}; + let failReady: (error: Error) => void = () => {}; + const ready = new Promise((resolve, reject) => { + markReady = resolve; + failReady = reject; + }); + // Swallows the rejection for callers that never query. + ready.catch(() => {}); + // Chains queries so only one is in flight on the session. + let queue: Promise = Promise.resolve(); + + const fail = (error: Error) => { + closedError ??= error; + failReady(closedError); + pending?.reject(closedError); + pending = undefined; + }; + + const handleMessage = (message: WebSocketResult) => { + switch (message.type) { + case "ReadyForQuery": { + markReady(); + if (!pending) return; + const { resolve, reject, rows, error } = pending; + pending = undefined; + if (error) reject(error); + else resolve(rows); + return; + } + case "Rows": + if (pending) { + pending.columns = message.payload.columns.map( + mapColumnToColumnMetadata, + ); + } + return; + case "Row": + pending?.rows.push(mapRowToObject(message.payload, pending.columns)); + return; + case "Error": + if (pending) { + pending.error = new WebsocketQueryError(message.payload.message); + } else { + fail(new WebsocketQueryError(message.payload.message)); + } + return; + default: + return; + } + }; + + const socket = new MaterializeWebsocket({ + httpAddress, + sessionVariables: buildSessionVariables(sessionVariables), + onMessage: handleMessage, + onClose: () => fail(new WebsocketQueryError("Connection closed")), + }); + socket.connect(); + + const run = async (query: CompiledQuery): Promise => { + await ready; + if (closedError) throw closedError; + return new Promise((resolve, reject) => { + pending = { + resolve: resolve as (rows: unknown[]) => void, + reject, + columns: [], + rows: [], + }; + socket.send({ queries: [buildExtendedRequestQuery(query)] }); + }); + }; + + return { + query: (query: CompiledQuery) => { + const result = queue.then(() => run(query)); + queue = result.catch(() => {}); + return result; + }, + close: () => { + fail(new WebsocketQueryError("Client closed")); + socket.disconnect(); + }, + }; +} diff --git a/console/src/api/materialize/cluster/replicaProcessMetrics.ts b/console/src/api/materialize/cluster/replicaProcessMetrics.ts index 784a1e133ce8d..ce25a345ef820 100644 --- a/console/src/api/materialize/cluster/replicaProcessMetrics.ts +++ b/console/src/api/materialize/cluster/replicaProcessMetrics.ts @@ -32,6 +32,8 @@ export interface ProcessMetricsSample { /** A size's process count and each process's CPU, memory and disk allocation. */ export interface SizeAllocation { processes: number; + /** Timely workers per process. */ + workers: number; cpuNanoCores: number; memoryBytes: number; diskBytes: number; @@ -96,6 +98,7 @@ export function buildReplicaSizeAllocationsQuery({ .select([ "size", sql`processes::float8`.as("processes"), + sql`workers::float8`.as("workers"), sql`cpu_nano_cores::float8`.as("cpuNanoCores"), sql`memory_bytes::float8`.as("memoryBytes"), sql`disk_bytes::float8`.as("diskBytes"), diff --git a/console/src/api/materialize/cluster/workerCpu.ts b/console/src/api/materialize/cluster/workerCpu.ts new file mode 100644 index 0000000000000..d2408f306a4cc --- /dev/null +++ b/console/src/api/materialize/cluster/workerCpu.ts @@ -0,0 +1,64 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +import { sql } from "kysely"; + +import { queryBuilder } from "~/api/materialize"; + +// These read a replica's introspection relations, so they must run in a +// session pinned to that replica (`cluster` and `cluster_replica`). They then +// plan and run on that cluster, not on `mz_catalog_server`. +// +// A dataflow's root operator (an address of length one) accounts for all the +// time spent in the regions and operators it contains, so summing roots counts +// each nanosecond once. Summing every operator would count it once per level +// of nesting. + +/** Each worker's cumulative time running dataflows, in nanoseconds. */ +export function buildWorkerCpuQuery() { + return queryBuilder + .selectFrom("mz_scheduling_elapsed_per_worker as e") + .innerJoin("mz_dataflow_addresses as a", "a.id", "e.id") + .where(sql`list_length(${sql.ref("a.address")}) = 1`) + .select([ + sql`${sql.ref("e.worker_id")}::int4`.as("workerId"), + sql`sum(${sql.ref("e.elapsed_ns")})::float8`.as("elapsedNs"), + ]) + .groupBy("e.worker_id"); +} + +/** Each dataflow's cumulative time on each worker, in nanoseconds. */ +export function buildDataflowWorkerCpuQuery() { + return queryBuilder + .selectFrom("mz_scheduling_elapsed_per_worker as e") + .innerJoin("mz_dataflow_addresses as a", "a.id", "e.id") + .where(sql`list_length(${sql.ref("a.address")}) = 1`) + .select([ + sql`${sql.ref("a.address")}[1]::text`.as("dataflowId"), + sql`${sql.ref("e.worker_id")}::int4`.as("workerId"), + sql`${sql.ref("e.elapsed_ns")}::float8`.as("elapsedNs"), + ]); +} + +/** + * What each dataflow maintains: user objects (`u` ids) and the temporary + * dataflows built for ad-hoc queries and subscribes (`t` ids). System + * dataflows still count toward each worker's total, but have nothing to name. + */ +export function buildDataflowExportsQuery() { + return queryBuilder + .selectFrom("mz_compute_exports") + .select([ + "export_id as objectId", + sql`dataflow_id::text`.as("dataflowId"), + ]) + .where((eb) => + eb.or([eb("export_id", "like", "u%"), eb("export_id", "like", "t%")]), + ); +} diff --git a/console/src/api/materialize/executeSqlV2.ts b/console/src/api/materialize/executeSqlV2.ts index 6c0152c0d5a36..627ad41136921 100644 --- a/console/src/api/materialize/executeSqlV2.ts +++ b/console/src/api/materialize/executeSqlV2.ts @@ -59,7 +59,7 @@ function isCompiledQueryList(queries: unknown): queries is CompiledQueryList { return Array.isArray(queries) && queries.every(isCompiledQuery); } -function buildExtendedRequestQuery(query: CompiledQuery) { +export function buildExtendedRequestQuery(query: CompiledQuery) { const params = query.parameters.map((param) => { return typeof param !== "string" ? JSON.stringify(param) : param; }); diff --git a/console/src/api/materialize/subscribeCollection.ts b/console/src/api/materialize/subscribeCollection.ts index 526251b00c17e..93bad5cd75e5f 100644 --- a/console/src/api/materialize/subscribeCollection.ts +++ b/console/src/api/materialize/subscribeCollection.ts @@ -69,6 +69,9 @@ export interface SubscribeCollection { * full current row set is diffed against what's already synced and only the * delta is written into the collection. Safe to call before sync starts. */ applySnapshot: (state: SubscribeState) => void; + /** The latest applied rows, including while the collection is paused + * between subscribers. */ + getRows: () => T[]; /** Seed the collection from the localStorage cache scoped to `scope` * (`organizationId|regionId`) and route future writes to that scoped key. * Deferred and scoped on purpose: a constant module-level key would seed one @@ -346,6 +349,7 @@ export function createSubscribeCollection(options: { collection, statusAtom, applySnapshot, + getRows: () => Array.from(desired.values()), hydrate, reset, suspendPersistence, diff --git a/console/src/api/materialize/types.ts b/console/src/api/materialize/types.ts index 48e93f96a8d2c..85f57caed89a9 100644 --- a/console/src/api/materialize/types.ts +++ b/console/src/api/materialize/types.ts @@ -19,6 +19,7 @@ export type SessionVariables = { cluster_replica?: string | undefined; database?: string | undefined; search_path?: string | undefined; + statement_timeout?: string | undefined; transaction_isolation?: string | undefined; }; diff --git a/console/src/mz-doc-urls.json b/console/src/mz-doc-urls.json index eeae4c3e003c8..df953eb364582 100644 --- a/console/src/mz-doc-urls.json +++ b/console/src/mz-doc-urls.json @@ -435,6 +435,7 @@ "/docs/sql/functions/substring/": "https://materialize.com/docs/sql/functions/substring/", "/docs/support/": "https://materialize.com/docs/support/", "/docs/clusters/troubleshoot-clusters/memory-spike/": "https://materialize.com/docs/clusters/troubleshoot-clusters/memory-spike/", + "/docs/clusters/troubleshoot-clusters/cpu-troubleshooting/": "https://materialize.com/docs/clusters/troubleshoot-clusters/cpu-troubleshooting/", "/docs/clusters/troubleshoot-clusters/hydration-failures/": "https://materialize.com/docs/clusters/troubleshoot-clusters/hydration-failures/", "/docs/sql/system-catalog/": "https://materialize.com/docs/sql/system-catalog/", "/docs/sql/system-clusters/": "https://materialize.com/docs/sql/system-clusters/", diff --git a/console/src/platform/clusters/ClusterPage/MemoryByObject.tsx b/console/src/platform/clusters/ClusterPage/MemoryByObject.tsx index c07ee7067163a..1476d190ded17 100644 --- a/console/src/platform/clusters/ClusterPage/MemoryByObject.tsx +++ b/console/src/platform/clusters/ClusterPage/MemoryByObject.tsx @@ -25,17 +25,12 @@ import { TooltipColorSwatch } from "~/components/graphComponents"; import { LoadingContainer } from "~/components/LoadingContainer"; import TextLink from "~/components/TextLink"; import { generateRainbowPalette } from "~/components/ThresholdLineGraph/thresholdLineGraphHelpers"; -import { useFlags } from "~/hooks/useFlags"; import { LargestReplicaBoundary } from "~/platform/clusters/LargestMaintainedQueries"; import { UNIFIED_OBJECT_SIZES_MIN_VERSION, useLargestMaintainedQueries, } from "~/platform/clusters/queries"; -import { - absoluteMaintainedObjectPath, - useBuildWorkflowGraphPath, -} from "~/platform/routeHelpers"; -import { useEnvironmentGate, useRegionSlug } from "~/store/environments"; +import { useEnvironmentGate } from "~/store/environments"; import { MaterializeTheme } from "~/theme"; import { formatMemoryUsage } from "~/utils/format"; @@ -46,6 +41,7 @@ import { } from "./memoryByObjectModel"; import { PERCENT_TICKS } from "./ResourceUsage/resourceUsageStyles"; import { useLiveLargestObjects } from "./useLiveLargestObjects"; +import { useObjectLink } from "./useObjectLink"; export interface MemoryByObjectProps { clusterId: string; @@ -140,12 +136,7 @@ const MemoryByObjectCard = ({ hasHeapLimit: boolean; }) => { const { colors } = useTheme(); - // The maintained objects page only has routes while its flag is on. - const hasMaintainedObjectsPage = Boolean( - useFlags()["maintained-objects-ui-50"], - ); - const regionSlug = useRegionSlug(); - const workflowGraphPath = useBuildWorkflowGraphPath(); + const objectLink = useObjectLink(); if (objects.length === 0) return null; @@ -156,22 +147,13 @@ const MemoryByObjectCard = ({ objects.map((object) => object.memoryPercentage), ); const segments = objects.map((object, index) => { - const { id, name, databaseName, schemaName, type } = object; return { - key: id ?? object.dataflowId ?? String(index), - label: name ?? object.dataflowName ?? "-", + key: object.id ?? object.dataflowId ?? String(index), + label: object.name ?? object.dataflowName ?? "-", usage: formatMemoryUsage(hasHeapLimit ? object : { size: object.size }), color: palette[index], widthPercent: widths[index], - path: - object.isOrphanedDataflow || !id || !name || !schemaName || !type - ? undefined - : hasMaintainedObjectsPage - ? absoluteMaintainedObjectPath(regionSlug, id) - : workflowGraphPath({ - databaseObject: { id, name, databaseName, schemaName }, - type, - }), + path: object.isOrphanedDataflow ? undefined : objectLink(object), }; }); diff --git a/console/src/platform/clusters/ClusterPage/ResourceUsage/ResourceUsage.tsx b/console/src/platform/clusters/ClusterPage/ResourceUsage/ResourceUsage.tsx index 57794a44f97c0..edc02f52132b8 100644 --- a/console/src/platform/clusters/ClusterPage/ResourceUsage/ResourceUsage.tsx +++ b/console/src/platform/clusters/ClusterPage/ResourceUsage/ResourceUsage.tsx @@ -49,6 +49,8 @@ import { capitalizeSentence } from "~/util"; import { formatPercentage } from "~/utils/format"; import { buildXYGraphLayoutProps } from "~/utils/graph"; +import { WorkerCpuDrawer } from "../WorkerCpu/WorkerCpuDrawer"; +import { WorkerReplica } from "../WorkerCpu/workerCpuModel"; import { ProcessesDrawer, ProcessReplica } from "./ProcessesDrawer"; import { CpuChart, @@ -595,6 +597,30 @@ export const ResourceUsage = ({ cluster }: ResourceUsageProps) => { [cluster.replicas, allocationsBySize, colorMap, colors.lineGraph], ); + const [isWorkersOpen, setIsWorkersOpen] = React.useState(false); + // Skew needs more than one worker, so single-worker sizes get no button. + // Memoized because the drawer's object table is memoized on it. + const workerReplicas = React.useMemo( + () => + cluster.replicas.flatMap(({ id, name, size }): WorkerReplica[] => { + const allocation = size ? allocationsBySize?.get(size) : undefined; + if (!size || !allocation) return []; + return allocation.processes * allocation.workers > 1 + ? [ + { + id, + name, + size, + processes: allocation.processes, + workersPerProcess: allocation.workers, + cpuNanoCoresPerProcess: allocation.cpuNanoCores, + }, + ] + : []; + }), + [cluster.replicas, allocationsBySize], + ); + let body: React.ReactNode; if (history.isError) { body = ; @@ -635,14 +661,27 @@ export const ResourceUsage = ({ cluster }: ResourceUsageProps) => { isStatusError={isStatusError} isHydrationError={isHydrationError} cpuHeaderAction={ - processReplicas.length > 0 && ( - + (processReplicas.length > 0 || workerReplicas.length > 0) && ( + + {processReplicas.length > 0 && ( + + )} + {workerReplicas.length > 0 && ( + + )} + ) } /> @@ -719,6 +758,14 @@ export const ResourceUsage = ({ cluster }: ResourceUsageProps) => { replicas={processReplicas} timePeriodMinutes={timePeriodMinutes} /> + setIsWorkersOpen(false)} + clusterId={cluster.id} + clusterName={cluster.name} + replicas={workerReplicas} + unhydratedObjectCounts={unhydratedObjectCounts} + /> ); }; diff --git a/console/src/platform/clusters/ClusterPage/ResourceUsage/ResourceUsageCharts.tsx b/console/src/platform/clusters/ClusterPage/ResourceUsage/ResourceUsageCharts.tsx index da88ffe6e3c5a..4658037820cd5 100644 --- a/console/src/platform/clusters/ClusterPage/ResourceUsage/ResourceUsageCharts.tsx +++ b/console/src/platform/clusters/ClusterPage/ResourceUsage/ResourceUsageCharts.tsx @@ -781,9 +781,11 @@ export const MemoryChart = ({ export interface TimeAxisProps { width: number; xScale: TimeScale; + /** Labels each tick. Defaults to the time before the window's end. */ + tickFormat?: (ms: number) => string; } -export const TimeAxis = ({ width, xScale }: TimeAxisProps) => { +export const TimeAxis = ({ width, xScale, tickFormat }: TimeAxisProps) => { const { colors, fonts } = useTheme(); const [startDate, endDate] = xScale.domain(); // Evenly spaced and labeled by time before now, matching `UtilizationGraph`. @@ -805,7 +807,9 @@ export const TimeAxis = ({ width, xScale }: TimeAxisProps) => { stroke={colors.border.secondary} strokeWidth={2} tickFormat={(value) => - relativeTimeTickFormat(value.valueOf(), startDate, endDate) + tickFormat + ? tickFormat(value.valueOf()) + : relativeTimeTickFormat(value.valueOf(), startDate, endDate) } tickLabelProps={() => ({ fill: colors.foreground.secondary, diff --git a/console/src/platform/clusters/ClusterPage/ResourceUsage/processReadingsModel.test.ts b/console/src/platform/clusters/ClusterPage/ResourceUsage/processReadingsModel.test.ts index 5f0b2124e16b8..ee1b0154454ed 100644 --- a/console/src/platform/clusters/ClusterPage/ResourceUsage/processReadingsModel.test.ts +++ b/console/src/platform/clusters/ClusterPage/ResourceUsage/processReadingsModel.test.ts @@ -20,6 +20,7 @@ import { const MINUTE = 60_000; const ALLOCATION = { processes: 3, + workers: 1, cpuNanoCores: 1e9, memoryBytes: 1000, diskBytes: 0, diff --git a/console/src/platform/clusters/ClusterPage/WorkerCpu/BusiestWorkers.tsx b/console/src/platform/clusters/ClusterPage/WorkerCpu/BusiestWorkers.tsx new file mode 100644 index 0000000000000..0befda9280ce5 --- /dev/null +++ b/console/src/platform/clusters/ClusterPage/WorkerCpu/BusiestWorkers.tsx @@ -0,0 +1,212 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +import { Box, Grid, HStack, Text, useTheme, VStack } from "@chakra-ui/react"; +import React from "react"; + +import { MaterializeTheme } from "~/theme"; +import { formatDate, TIME_FORMAT } from "~/utils/dateFormat"; +import { formatPercentage } from "~/utils/format"; + +import { + objectCpuRows, + processOfWorker, + topObjectOnWorker, + WorkerBusy, + WorkerReplica, +} from "./workerCpuModel"; + +const TOP_WORKER_COUNT = 5; + +/** A filled span of a track, from `from` to `to`, each a fraction of it. */ +export const TrackBar = ({ + from = 0, + to, + color, + height = "10px", +}: { + from?: number; + to: number; + color: string; + height?: string; +}) => { + const { colors } = useTheme(); + return ( + + + + ); +}; + +const WorkerRow = ({ + label, + bar, + value, + note, + hasBorder, +}: { + label: React.ReactNode; + bar: React.ReactNode; + value: string; + note?: string; + hasBorder: boolean; +}) => { + const { colors } = useTheme(); + return ( + + {label} + {bar} + + {value} + + + {note} + + + ); +}; + +export interface BusiestWorkersProps { + /** Busiest first. */ + workers: WorkerBusy[]; + sinceMs: number; + /** Without skew, a ranked list only reorders noise, so one range shows. */ + isSkewed: boolean; + objects: ReturnType | undefined; + objectName: (objectId: string) => string; + replica: WorkerReplica; +} + +export const BusiestWorkers = ({ + workers, + sinceMs, + isSkewed, + objects, + objectName, + replica, +}: BusiestWorkersProps) => { + const { colors } = useTheme(); + const top = isSkewed ? workers.slice(0, TOP_WORKER_COUNT) : []; + const rest = workers.slice(top.length); + const restLowest = rest.at(-1)?.busyFraction; + const restHighest = rest.at(0)?.busyFraction; + return ( + + + + {isSkewed ? "Busiest workers" : "Workers"} + + + % of one core spent running dataflows since{" "} + {formatDate(new Date(sinceMs), TIME_FORMAT)} + + + + {top.map((worker, index) => { + const driver = + objects && + topObjectOnWorker( + objects.rows, + objects.workerTotalsNs, + worker.workerId, + ); + return ( + + + w{worker.workerId} + + {replica.processes > 1 && ( + + process{" "} + {processOfWorker( + worker.workerId, + replica.workersPerProcess, + )} + + )} + + } + bar={ + + } + value={formatPercentage(worker.busyFraction, 0)} + note={ + driver && + `mostly ${objectName(driver.objectId)} · ${formatPercentage(driver.share, 0)} of its time` + } + /> + ); + })} + {restLowest !== undefined && restHighest !== undefined && ( + + {top.length > 0 + ? `Other ${rest.length} workers` + : `All ${rest.length} workers`} + + } + bar={ + + } + value={`${formatPercentage(restLowest, 0)}–${formatPercentage(restHighest, 0)}`} + /> + )} + + + ); +}; diff --git a/console/src/platform/clusters/ClusterPage/WorkerCpu/ObjectCpuTable.test.tsx b/console/src/platform/clusters/ClusterPage/WorkerCpu/ObjectCpuTable.test.tsx new file mode 100644 index 0000000000000..d187cd3013c3b --- /dev/null +++ b/console/src/platform/clusters/ClusterPage/WorkerCpu/ObjectCpuTable.test.tsx @@ -0,0 +1,106 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +import { screen } from "@testing-library/react"; +import React from "react"; + +import { renderComponent } from "~/test/utils"; + +import { ObjectCpuTable, ObjectCpuTableProps } from "./ObjectCpuTable"; +import { objectCpuRows, WorkerReplica } from "./workerCpuModel"; + +const SECOND_NS = 1e9; + +const REPLICA: WorkerReplica = { + id: "u2", + name: "r1", + size: "400cc", + processes: 1, + workersPerProcess: 4, + cpuNanoCoresPerProcess: 4e9, +}; + +const perWorker = (dataflowId: string, seconds: number[]) => + seconds.map((s, workerId) => ({ + dataflowId, + workerId, + elapsedNs: s * SECOND_NS, + })); + +const renderTable = ( + latest: ReturnType, + props: Partial = {}, +) => { + const { rows, workerTotalsNs } = objectCpuRows({ + latest, + exports: [ + { objectId: "u5", dataflowId: "1" }, + { objectId: "u6", dataflowId: "2" }, + ], + workerCount: 4, + }); + return renderComponent( + {}} + dataflowTotalNs={workerTotalsNs.reduce((sum, ns) => sum + ns, 0)} + allocation={undefined} + replica={REPLICA} + objectName={(objectId) => objectId} + objectsById={new Map()} + {...props} + />, + ); +}; + +describe("ObjectCpuTable", () => { + it("measures a window against the replica's allocation and says what ran outside dataflows", async () => { + // 60s on 4 cores is 240s of allocation, of which dataflows used 48s. + await renderTable(perWorker("1", [12, 12, 12, 12]), { + allocation: { ns: 240 * SECOND_NS, replicaCpuFraction: 0.5 }, + }); + + expect(screen.getByText("% of replica CPU")).toBeVisible(); + expect(screen.queryByText("Share of dataflow CPU")).toBeNull(); + expect(screen.getByText("20%")).toBeVisible(); + expect( + screen.getByText(/Work outside running dataflows took 30%/), + ).toBeVisible(); + }); + + it("measures a lifetime against all dataflow time", async () => { + await renderTable(perWorker("1", [12, 12, 12, 12]), { mode: "lifetime" }); + + expect(screen.getByText("Share of dataflow CPU")).toBeVisible(); + expect(screen.queryByText("% of replica CPU")).toBeNull(); + }); + + it("doesn't flag skew in an object too small to matter", async () => { + // u6 sits on one worker but has well under 1% of the dataflow time. + await renderTable([ + ...perWorker("1", [1000, 1000, 1000, 1000]), + ...perWorker("2", [1, 0, 0, 0]), + ]); + + expect(screen.queryByText("Extra time")).toBeNull(); + }); + + it("flags skew in an object with a real share of the time", async () => { + await renderTable([ + ...perWorker("1", [1000, 1000, 1000, 1000]), + ...perWorker("2", [400, 0, 0, 0]), + ]); + + expect(screen.getByText("Extra time")).toBeVisible(); + }); +}); diff --git a/console/src/platform/clusters/ClusterPage/WorkerCpu/ObjectCpuTable.tsx b/console/src/platform/clusters/ClusterPage/WorkerCpu/ObjectCpuTable.tsx new file mode 100644 index 0000000000000..3194ec5f655f2 --- /dev/null +++ b/console/src/platform/clusters/ClusterPage/WorkerCpu/ObjectCpuTable.tsx @@ -0,0 +1,522 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +import { + Box, + Button, + ButtonGroup, + Flex, + HStack, + Text, + useTheme, + VStack, +} from "@chakra-ui/react"; +import { + createColumnHelper, + functionalUpdate, + SortingState, + Table, +} from "@tanstack/react-table"; +import React from "react"; +import { Link as RouterLink } from "react-router-dom"; + +import { DatabaseObject } from "~/api/materialize/objects"; +import { TablePagination } from "~/components/Table/TablePagination"; +import { UniversalTable } from "~/components/Table/UniversalTable"; +import { useUniversalTable } from "~/components/Table/useUniversalTable"; +import TextLink from "~/components/TextLink"; +import { MaterializeTheme } from "~/theme"; +import { pluralize } from "~/util"; +import { formatDate, TIME_FORMAT } from "~/utils/dateFormat"; +import { formatDurationExact, formatPercentage } from "~/utils/format"; + +import { useObjectLink } from "../useObjectLink"; +import { TrackBar } from "./BusiestWorkers"; +import { + ADHOC_ROW_ID, + EVEN_RATIO, + NOTABLE_SHARE, + ObjectCpuRow, + SKEW_RATIO, + WorkerReplica, +} from "./workerCpuModel"; +import { BAR_CEILING_RATIO, useWorkerBarColor } from "./workerCpuStyles"; + +const OBJECT_PAGE_SIZE = 10; +const WORKER_BARS_WIDTH = 240; +const WORKER_BARS_HEIGHT = 28; +const PROCESS_GAP = 4; + +/** Which span the table measures. "auto" picks the window once it exists. */ +export type ObjectMode = "auto" | "window" | "lifetime"; + +interface ObjectTableRow { + cpu: ObjectCpuRow; + name: string; + path: string | undefined; + /** Of the replica's allocation, or of all dataflow time without one. */ + share: number; + /** Too small a share of the dataflow time to flag its skew. */ + isMinor: boolean; +} + +interface ObjectTableMeta { + replica: WorkerReplica; + workerCount: number; + /** Dataflows that export no user object, so the shares add up. */ + other: { share: number; totalNs: number }; +} + +const tableMeta = (table: Table) => + table.options.meta as ObjectTableMeta; + +const EvenShareKey = () => { + const { colors } = useTheme(); + return ( + + + + ); +}; + +/** + * One bar per worker for its time on the object, against a dashed line at an + * even share, so even work reads as a flat row at the line. + */ +const WorkerBars = ({ + row, + workerCount, + workersPerProcess, + isMinor, +}: { + row: ObjectCpuRow; + workerCount: number; + workersPerProcess: number; + isMinor: boolean; +}) => { + const { colors } = useTheme(); + const skewColor = useWorkerBarColor(); + const barColor = isMinor ? () => colors.foreground.tertiary : skewColor; + const perProcess = Math.max(1, workersPerProcess); + const processCount = Math.ceil(workerCount / perProcess); + const slot = + (WORKER_BARS_WIDTH - PROCESS_GAP * (processCount - 1)) / workerCount; + // Touching bars of one height merge into a block, so wide slots keep a gap. + const barWidth = slot >= 4 ? slot - 1 : slot; + const yOf = (ratio: number) => + WORKER_BARS_HEIGHT * + (1 - Math.min(ratio, BAR_CEILING_RATIO) / BAR_CEILING_RATIO); + const evenY = yOf(1); + return ( + + + {row.perWorkerNs.map((ns, workerId) => { + const ratio = row.averageNs > 0 ? ns / row.averageNs : 0; + const y = yOf(ratio); + return ( + + + {`w${workerId}: ${formatDurationExact(ns / 1e6)}, ${ratio.toFixed(2)}× an even share`} + + + ); + })} + + + ); +}; + +const ObjectNameCell = ({ row }: { row: ObjectTableRow }) => { + const { colors } = useTheme(); + if (row.path) { + return ( + + {row.name} + + ); + } + const { dataflowCount } = row.cpu; + return ( + + {row.name} + {row.cpu.objectId === ADHOC_ROW_ID && ( + + {" "} + · {dataflowCount} temporary{" "} + {pluralize(dataflowCount, "dataflow", "dataflows")} + + )} + + ); +}; + +const OtherDataflowsName = () => { + const { colors } = useTheme(); + return ( + + Other dataflows + + {" "} + · system and introspection + + + ); +}; + +const HottestWorkerCell = ({ + row, + isMinor, +}: { + row: ObjectCpuRow; + isMinor: boolean; +}) => { + const { colors } = useTheme(); + const isEven = row.ratio < EVEN_RATIO; + return ( + = SKEW_RATIO + ? colors.accent.red + : undefined + } + > + {isEven + ? `even, max ${row.ratio.toFixed(1)}×` + : `w${row.hottestWorkerId} · ${row.ratio.toFixed(1)}×`} + + ); +}; + +const ShareBar = ({ share, isOther }: { share: number; isOther?: boolean }) => { + const { colors } = useTheme(); + return ( + + + + {formatPercentage(share, 0)} + + + ); +}; + +const columnHelper = createColumnHelper(); + +const columns = [ + columnHelper.accessor("name", { + id: "object", + header: "Object", + sortingFn: "alphanumeric", + cell: (info) => , + footer: () => , + meta: { cellProps: { maxWidth: "220px" } }, + }), + columnHelper.display({ + id: "workers", + header: ({ table }) => + tableMeta(table).replica.processes > 1 + ? "Time per worker, by process" + : "Time per worker", + cell: ({ row, table }) => ( + + ), + enableSorting: false, + }), + columnHelper.accessor((row) => row.cpu.ratio, { + id: "hottest", + header: "Hottest worker", + cell: (info) => ( + + ), + meta: { + tooltip: + "The worker that spent the longest on the object, as a multiple of an even share. 1× is even, and 2× is twice an even share.", + cellProps: { whiteSpace: "nowrap" }, + }, + }), + columnHelper.accessor((row) => row.cpu.extraNs, { + id: "extra", + header: "Extra time", + cell: (info) => + info.row.original.cpu.ratio >= EVEN_RATIO + ? `+${formatDurationExact(info.getValue() / 1e6)}` + : null, + meta: { + tooltip: + "How much longer the hottest worker spent on the object than an even share. The object's results wait for that worker.", + isNumeric: true, + cellProps: { whiteSpace: "nowrap" }, + }, + }), + // Two headers over one value: a window has an allocation to measure + // against, and a lifetime, whose dataflows started at different times, + // doesn't. + columnHelper.accessor("share", { + id: "cpuShare", + header: "% of replica CPU", + cell: (info) => , + footer: ({ table }) => ( + + ), + meta: { + tooltip: + "The object's CPU time as a share of the replica's CPU allocation over the same span, the unit of the CPU chart. Objects, other dataflows and work outside running dataflows add up to the replica's CPU.", + minWidth: "160px", + }, + }), + columnHelper.accessor("share", { + id: "dataflowShare", + header: "Share of dataflow CPU", + cell: (info) => , + footer: ({ table }) => ( + + ), + meta: { + tooltip: + "The object's part of the CPU time workers spent running dataflows, including system dataflows and ad-hoc queries. CPU used outside dataflows, such as scheduling and networking, isn't counted, so this isn't a share of the replica's CPU.", + minWidth: "160px", + }, + }), + columnHelper.accessor((row) => row.cpu.totalNs, { + id: "cpuTime", + header: "CPU time", + cell: (info) => formatDurationExact(info.getValue() / 1e6), + footer: ({ table }) => + formatDurationExact(tableMeta(table).other.totalNs / 1e6), + meta: { + tooltip: + "How long workers spent running the object's dataflow, added up across all workers.", + isNumeric: true, + cellProps: { whiteSpace: "nowrap" }, + }, + }), +]; + +export interface ObjectCpuTableProps { + rows: ObjectCpuRow[]; + /** The order until someone sorts by a column. */ + sortBy: "extra" | "total"; + mode: "window" | "lifetime"; + canMeasureWindow: boolean; + sinceMs: number | undefined; + onModeChange: (mode: ObjectMode) => void; + /** All dataflow time in the same span, system and ad-hoc included. */ + dataflowTotalNs: number; + /** + * The replica's CPU allocation over a window, and its measured CPU then. + * With it, shares are of the allocation. Without it, of `dataflowTotalNs`. + */ + allocation: + { ns: number; replicaCpuFraction: number | undefined } | undefined; + replica: WorkerReplica; + objectName: (objectId: string) => string; + objectsById: Map; +} + +// Memoized because its rows change only with the per-object read, while the +// panel renders on every worker reading. +export const ObjectCpuTable = React.memo( + ({ + rows, + sortBy, + mode, + canMeasureWindow, + sinceMs, + onModeChange, + dataflowTotalNs, + allocation, + replica, + objectName, + objectsById, + }: ObjectCpuTableProps) => { + const { colors } = useTheme(); + const objectLink = useObjectLink(); + const workerCount = replica.processes * replica.workersPerProcess; + const basisNs = allocation?.ns ?? dataflowTotalNs; + const data = React.useMemo( + () => + rows.map((cpu): ObjectTableRow => { + const object = objectsById.get(cpu.objectId); + return { + cpu, + name: objectName(cpu.objectId), + path: + object && + objectLink({ + id: object.id, + name: object.name, + databaseName: object.databaseName, + schemaName: object.schemaName, + type: object.objectType, + }), + share: basisNs > 0 ? cpu.totalNs / basisNs : 0, + isMinor: + dataflowTotalNs <= 0 || + cpu.totalNs / dataflowTotalNs < NOTABLE_SHARE, + }; + }), + [rows, objectsById, objectName, objectLink, basisNs, dataflowTotalNs], + ); + const otherNs = Math.max( + 0, + dataflowTotalNs - rows.reduce((sum, row) => sum + row.totalNs, 0), + ); + const [sorting, setSorting] = React.useState(); + const effectiveSorting = React.useMemo( + () => + sorting ?? [ + { id: sortBy === "extra" ? "extra" : "cpuTime", desc: true }, + ], + [sorting, sortBy], + ); + const table = useUniversalTable({ + data, + columns, + getRowId: (row) => row.cpu.objectId, + pageSize: OBJECT_PAGE_SIZE, + state: { + sorting: effectiveSorting, + columnVisibility: { + extra: data.some( + (row) => !row.isMinor && row.cpu.ratio >= EVEN_RATIO, + ), + cpuShare: Boolean(allocation), + dataflowShare: !allocation, + }, + }, + onSortingChange: (updater) => + setSorting(functionalUpdate(updater, effectiveSorting)), + meta: { + replica, + workerCount, + other: { + share: basisNs > 0 ? otherNs / basisNs : 0, + totalNs: otherNs, + }, + } satisfies ObjectTableMeta, + }); + return ( + + + Where the CPU went + + + + + + + + + {allocation?.replicaCpuFraction !== undefined && ( + + Work outside running dataflows took{" "} + {formatPercentage( + Math.max( + 0, + allocation.replicaCpuFraction - dataflowTotalNs / allocation.ns, + ), + 0, + )}{" "} + of the allocation over the same span, for{" "} + {formatPercentage(allocation.replicaCpuFraction, 0)} of the + replica's CPU in all. It includes dataflows that finished + between readings, such as those for ad-hoc queries. + + )} + + + + + An even share, 1×: the object's CPU time split across its{" "} + {workerCount} workers. Each bar is one worker, and bars stop at{" "} + {BAR_CEILING_RATIO}×. + + + {rows.length > OBJECT_PAGE_SIZE && ( + + )} + + + ); + }, +); diff --git a/console/src/platform/clusters/ClusterPage/WorkerCpu/WorkerCpuCharts.tsx b/console/src/platform/clusters/ClusterPage/WorkerCpu/WorkerCpuCharts.tsx new file mode 100644 index 0000000000000..8846894ba9e07 --- /dev/null +++ b/console/src/platform/clusters/ClusterPage/WorkerCpu/WorkerCpuCharts.tsx @@ -0,0 +1,495 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +import { Box, HStack, Text, useTheme, VStack } from "@chakra-ui/react"; +import { scaleLinear, scaleTime } from "@visx/scale"; +import { Area, Line, LinePath } from "@visx/shape"; +import { useTooltip, useTooltipInPortal } from "@visx/tooltip"; +import React from "react"; + +import { GraphLineCursor, GraphTooltip } from "~/components/graphComponents"; +import { MaterializeTheme } from "~/theme"; +import { + formatDate, + TIME_FORMAT, + TIME_FORMAT_NO_SECONDS, +} from "~/utils/dateFormat"; +import { formatPercentage } from "~/utils/format"; +import { buildXYGraphLayoutProps } from "~/utils/graph"; + +import { + ChartHeader, + HoverOverlay, + Layout, + PercentAxis, + PercentScale, + TimeAxis, + TimeScale, +} from "../ResourceUsage/ResourceUsageCharts"; +import { + CHART_MARGIN, + PERCENT_TICKS, +} from "../ResourceUsage/resourceUsageStyles"; +import { + CpuPoint, + SMOOTHING_WINDOW_MS, + WorkerBusyPoint, + workerLabel, + WorkerReplica, + workersPerCore, +} from "./workerCpuModel"; +import { gapFill } from "./workerCpuStyles"; + +const PLOT_HEIGHT_PX = 132; +const SVG_HEIGHT_PX = CHART_MARGIN.top + PLOT_HEIGHT_PX + CHART_MARGIN.bottom; + +export interface WorkerCpuChartsProps { + width: number; + startMs: number; + endMs: number; + openedAtMs: number; + replicaCpu: CpuPoint[]; + series: WorkerBusyPoint[]; + replica: WorkerReplica; +} + +const nearest = (points: T[], atMs: number) => { + let best: T | undefined; + for (const point of points) { + if (!best || Math.abs(point.atMs - atMs) < Math.abs(best.atMs - atMs)) { + best = point; + } + } + return best; +}; + +const BeforeOpen = ({ + layout, + x, + label, +}: { + layout: Layout; + x: number; + label?: string; +}) => { + const { colors, fonts } = useTheme(); + const [left] = layout.xScaleRange; + const { graphTop, graphBottom } = layout.graphLineCursorProps; + const width = Math.max(0, x - left); + return ( + + + {label && width > 240 && ( + + {label} + + )} + + + ); +}; + +// Clock times, since the window runs ahead of now while readings fill it and +// times relative to its end would read wrong. +const clockTick = (ms: number) => + formatDate(new Date(ms), TIME_FORMAT_NO_SECONDS); + +type ChartPoint = WorkerBusyPoint & { + /** Time running dataflows, as a percent of the replica's allocation. */ + dataflowPercent: number; + replicaPercent: number | undefined; +}; + +interface MarksProps { + layout: Layout; + xScale: TimeScale; + yScale: PercentScale; + openedX: number; + /** Runs of points between pauses, so no line bridges a pause. */ + segments: ChartPoint[][]; + clipId: string; +} + +// The marks are memoized so pointer moves, which only change the cursor and +// tooltip, skip redrawing them. +const CpuMarks = React.memo( + ({ + layout, + xScale, + yScale, + openedX, + segments, + clipId, + replicaCpu, + }: MarksProps & { replicaCpu: CpuPoint[] }) => { + const { colors } = useTheme(); + const outsideFill = gapFill(colors.foreground.secondary); + const { graphTop, graphBottom } = layout.graphLineCursorProps; + const [plotLeft, plotRight] = layout.xScaleRange; + const x = (point: { atMs: number }) => xScale(new Date(point.atMs)); + return ( + <> + + + {/* Readings and metrics outside the window would otherwise draw over + the axis labels. Both charts clip to this. */} + + + + + + + {segments.map((segment) => ( + yScale(point.dataflowPercent)} + y1={(point) => + yScale( + Math.max(point.dataflowPercent, point.replicaPercent ?? 0), + ) + } + fill={outsideFill} + pointerEvents="none" + /> + ))} + yScale(point.percent)} + stroke={colors.accent.purple} + strokeWidth={2} + pointerEvents="none" + /> + {segments.map((segment) => ( + yScale(point.dataflowPercent)} + stroke={colors.foreground.primary} + strokeWidth={2} + pointerEvents="none" + /> + ))} + + + ); + }, +); + +const WorkerMarks = React.memo( + ({ layout, xScale, yScale, openedX, segments, clipId }: MarksProps) => { + const { colors } = useTheme(); + const skewFill = gapFill(colors.accent.orange); + const x = (point: { atMs: number }) => xScale(new Date(point.atMs)); + return ( + <> + + + + {segments.map((segment) => ( + + yScale(point.averageBusyFraction * 100)} + y1={(point) => yScale(point.busiest.busyFraction * 100)} + fill={skewFill} + pointerEvents="none" + /> + yScale(point.averageBusyFraction * 100)} + stroke={colors.foreground.secondary} + strokeWidth={2} + strokeDasharray="4 3" + pointerEvents="none" + /> + yScale(point.busiest.busyFraction * 100)} + stroke={colors.accent.orange} + strokeWidth={2} + pointerEvents="none" + /> + + ))} + + + ); + }, +); + +/** + * Two charts on one time axis, each in one unit: the replica's CPU against + * the time its workers spent running dataflows, as shares of its allocation, + * and the busiest worker against the average one, as shares of one core. + */ +export const WorkerCpuCharts = ({ + width, + startMs, + endMs, + openedAtMs, + replicaCpu, + series, + replica, +}: WorkerCpuChartsProps) => { + const { colors } = useTheme(); + const { tooltipData, tooltipLeft, tooltipTop, showTooltip, hideTooltip } = + useTooltip(); + const { containerRef, containerBounds, TooltipInPortal } = useTooltipInPortal( + { detectBounds: true, scroll: true }, + ); + const layout = React.useMemo( + () => + buildXYGraphLayoutProps({ + width, + height: SVG_HEIGHT_PX, + margin: CHART_MARGIN, + }), + [width], + ); + const xScale = React.useMemo( + () => + scaleTime({ + domain: [new Date(startMs), new Date(endMs)], + range: layout.xScaleRange, + }), + [startMs, endMs, layout], + ); + const yScale = React.useMemo( + () => + scaleLinear({ + domain: [0, 100], + range: layout.yScaleRange, + clamp: true, + }), + [layout], + ); + const perCore = workersPerCore( + replica.workersPerProcess, + replica.cpuNanoCoresPerProcess, + ); + const segments = React.useMemo(() => { + const result: ChartPoint[][] = []; + for (const point of series) { + if (point.afterGap || result.length === 0) result.push([]); + result[result.length - 1].push({ + ...point, + dataflowPercent: point.averageBusyFraction * perCore * 100, + replicaPercent: nearest(replicaCpu, point.atMs)?.percent, + }); + } + return result; + }, [series, replicaCpu, perCore]); + const openedX = xScale(new Date(openedAtMs)); + const { graphTop, graphBottom } = layout.graphLineCursorProps; + const clipId = `worker-cpu-clip-${React.useId().replace(/[^a-zA-Z0-9]/g, "")}`; + const marksProps = { layout, xScale, yScale, openedX, segments, clipId }; + + const hoverPoint = + tooltipData === undefined + ? undefined + : nearest(segments.flat(), tooltipData); + const hoverReplica = + tooltipData === undefined + ? undefined + : nearest(replicaCpu, tooltipData)?.percent; + const hover = ( + <> + {tooltipData !== undefined && ( + + )} + + showTooltip({ + tooltipData: timeMs, + tooltipLeft: event.clientX - containerBounds.left, + tooltipTop: event.clientY - containerBounds.top, + }) + } + onPointerLeave={hideTooltip} + /> + + ); + + return ( + + + + + + + {hover} + + + + + + + + {hover} + + + + + + {tooltipData !== undefined && + tooltipLeft !== undefined && + tooltipTop !== undefined && ( + + + + {formatDate( + new Date(hoverPoint?.atMs ?? tooltipData), + TIME_FORMAT, + )} + + {hoverReplica !== undefined && ( + + )} + {hoverPoint ? ( + <> + + + + + ) : ( + + No worker reading at this time. + + )} + + + )} + + ); +}; + +const TooltipRow = ({ + label, + value, + isStrong, +}: { + label: string; + value: string; + isStrong?: boolean; +}) => ( + + {label} + + {value} + + +); diff --git a/console/src/platform/clusters/ClusterPage/WorkerCpu/WorkerCpuDrawer.tsx b/console/src/platform/clusters/ClusterPage/WorkerCpu/WorkerCpuDrawer.tsx new file mode 100644 index 0000000000000..e1dd5c9808b73 --- /dev/null +++ b/console/src/platform/clusters/ClusterPage/WorkerCpu/WorkerCpuDrawer.tsx @@ -0,0 +1,441 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +import { + Box, + Button, + Flex, + Tab, + TabList, + TabPanel, + TabPanels, + Tabs, + Text, + useTheme, + VStack, +} from "@chakra-ui/react"; +import { useParentSize } from "@visx/responsive"; +import { mean } from "d3"; +import { millisecondsInMinute } from "date-fns/constants"; +import React from "react"; + +import Alert from "~/components/Alert"; +import SideDrawer from "~/components/SideDrawer"; +import TextLink from "~/components/TextLink"; +import { useClusterObjectsLive } from "~/store/allObjectsCollection"; +import { MaterializeTheme } from "~/theme"; +import { pluralize } from "~/util"; +import { formatDate, TIME_FORMAT } from "~/utils/dateFormat"; +import { formatDurationExact } from "~/utils/format"; + +import { BusiestWorkers } from "./BusiestWorkers"; +import { ObjectCpuTable, ObjectMode } from "./ObjectCpuTable"; +import { useReplicaCpu } from "./useReplicaCpu"; +import { WorkerCpuCharts } from "./WorkerCpuCharts"; +import { + ADHOC_ROW_ID, + CPU_DOCS, + cpuVerdict, + isSkewVerdict, + objectCpuRows, + SMOOTHING_WINDOW_MS, + verdictWindowStart, + workerBusyBetween, + workerBusySeries, + WorkerReplica, + workersPerCore, +} from "./workerCpuModel"; +import { + READ_INTERVAL_MS, + SAMPLING_LIMIT_MS, + useWorkerCpuSampler, + WorkerCpuState, +} from "./workerCpuSampler"; +import { FirstLook, Verdict } from "./WorkerCpuVerdict"; + +const CHART_WINDOW_MS = 15 * millisecondsInMinute; +// Room on either side of the readings: the replica's CPU just before opening, +// and space the readings fill in, so the first minutes aren't a sliver. +const CHART_LEAD_MS = 2 * millisecondsInMinute; +const CHART_RUNWAY_MS = 3 * millisecondsInMinute; +// The verdict judges recent readings, so a reopened drawer reports the present +// rather than averaging in readings from before it closed. +const VERDICT_WINDOW_MS = 5 * millisecondsInMinute; +// Objects whose dataflows run on a cluster's replicas. +const EXPORT_OBJECT_TYPES = ["index", "materialized-view"] as const; + +export interface WorkerCpuDrawerProps { + isOpen: boolean; + onClose: () => void; + clusterId: string; + clusterName: string; + /** Replicas with more than one worker, the only ones skew applies to. */ + replicas: WorkerReplica[]; + unhydratedObjectCounts: Map | undefined; +} + +/** + * How a cluster's replicas spend their CPU across workers, read live from the + * replica while open. + */ +export const WorkerCpuDrawer = ({ + isOpen, + onClose, + clusterId, + clusterName, + replicas, + unhydratedObjectCounts, +}: WorkerCpuDrawerProps) => ( + + + + {replicas.map((replica) => ( + {replica.name} + ))} + + + {replicas.map((replica) => ( + + + + ))} + + + +); + +const chartWindow = ( + firstMs: number | undefined, + lastMs: number | undefined, +) => { + const openedAtMs = firstMs ?? Date.now(); + const endMs = Math.max(lastMs ?? openedAtMs, openedAtMs + CHART_RUNWAY_MS); + return { + openedAtMs, + startMs: Math.max(openedAtMs - CHART_LEAD_MS, endMs - CHART_WINDOW_MS), + endMs, + }; +}; + +const WorkerCpuPanel = ({ + clusterId, + clusterName, + replica, + unhydratedCount, +}: { + clusterId: string; + clusterName: string; + replica: WorkerReplica; + unhydratedCount: number | undefined; +}) => { + const { colors } = useTheme(); + const { parentRef, width } = useParentSize({ debounceTime: 10 }); + const { state, pause, resume } = useWorkerCpuSampler({ + clusterName, + replicaId: replica.id, + replicaName: replica.name, + }); + const replicaCpu = useReplicaCpu({ + replicaId: replica.id, + cpuNanoCoresPerProcess: replica.cpuNanoCoresPerProcess, + }); + const [objectMode, setObjectMode] = React.useState("auto"); + const { data: clusterObjects } = useClusterObjectsLive( + clusterId, + EXPORT_OBJECT_TYPES, + ); + const objectsById = React.useMemo( + () => new Map(clusterObjects.map((object) => [object.id, object])), + [clusterObjects], + ); + const objectName = React.useCallback( + (objectId: string) => + objectId === ADHOC_ROW_ID + ? "Ad-hoc queries and subscribes" + : (objectsById.get(objectId)?.name ?? objectId), + [objectsById], + ); + const workerCount = replica.processes * replica.workersPerProcess; + + const { readings } = state; + const first = readings.at(0); + const last = readings.at(-1); + const series = React.useMemo(() => workerBusySeries(readings), [readings]); + const verdictStart = verdictWindowStart(readings, VERDICT_WINDOW_MS); + const windowMs = verdictStart && last ? last.atMs - verdictStart.atMs : 0; + const windowWorkers = + verdictStart && last && windowMs > 0 + ? workerBusyBetween(verdictStart, last) + : []; + // Metrics land about once a minute, so include the one before the window. + const windowCpu = mean( + verdictStart + ? replicaCpu.filter( + (point) => point.atMs >= verdictStart.atMs - millisecondsInMinute, + ) + : [], + (point) => point.percent / 100, + ); + // Shorter windows swing with the counters' one-second ticks. + const verdict = + windowMs >= SMOOTHING_WINDOW_MS && windowWorkers.length > 0 + ? cpuVerdict( + windowWorkers, + windowCpu, + workersPerCore( + replica.workersPerProcess, + replica.cpuNanoCoresPerProcess, + ), + ) + : undefined; + const isSkewed = isSkewVerdict(verdict); + + const hasWindowObjects = Boolean( + state.firstObjects && + state.latestObjects && + state.latestObjects.atMs > state.firstObjects.atMs, + ); + const lifetimeObjects = React.useMemo( + () => + state.latestObjects && + objectCpuRows({ + latest: state.latestObjects.rows, + exports: state.exports, + workerCount, + }), + [state.latestObjects, state.exports, workerCount], + ); + const windowObjects = React.useMemo( + () => + hasWindowObjects && state.firstObjects && state.latestObjects + ? objectCpuRows({ + baseline: state.firstObjects.rows, + latest: state.latestObjects.rows, + exports: state.exports, + workerCount, + }) + : undefined, + [ + hasWindowObjects, + state.firstObjects, + state.latestObjects, + state.exports, + workerCount, + ], + ); + const tableMode = + objectMode === "auto" + ? windowObjects + ? "window" + : "lifetime" + : objectMode; + const tableObjects = + tableMode === "window" + ? (windowObjects ?? lifetimeObjects) + : lifetimeObjects; + // A window's objects measure against the allocation over it, the CPU + // chart's unit. + const { firstObjects, latestObjects } = state; + const allocation = React.useMemo(() => { + if (tableMode !== "window" || !windowObjects) return undefined; + if (!firstObjects || !latestObjects) return undefined; + const cores = (replica.processes * replica.cpuNanoCoresPerProcess) / 1e9; + const ns = (latestObjects.atMs - firstObjects.atMs) * 1e6 * cores; + if (ns <= 0) return undefined; + return { + ns, + // Metrics land about once a minute, so include the one before. + replicaCpuFraction: mean( + replicaCpu.filter( + (point) => + point.atMs >= firstObjects.atMs - millisecondsInMinute && + point.atMs <= latestObjects.atMs, + ), + (point) => point.percent / 100, + ), + }; + }, [ + tableMode, + windowObjects, + firstObjects, + latestObjects, + replica, + replicaCpu, + ]); + + return ( + + + {clusterName} · {replica.name} · {replica.size} · {workerCount} workers + {replica.processes > 1 ? ` on ${replica.processes} processes` : ""} + + {unhydratedCount !== undefined && unhydratedCount > 0 && ( + + Hydration reprocesses each object's inputs from scratch, so + busy workers are expected until it finishes.{" "} + + Troubleshoot hydration + + + } + /> + )} + {state.status === "error" && readings.length === 0 ? ( + + ) : verdict ? ( + + ) : ( + + )} + + {width > 0 && ( + + )} + + {verdict && verdictStart && ( + + )} + {tableObjects && ( + sum + ns, + 0, + )} + allocation={allocation} + replica={replica} + objectName={objectName} + objectsById={objectsById} + /> + )} + + + ); +}; + +const SamplingStatus = ({ + replicaName, + state, + sinceMs, + onPause, + onResume, +}: { + replicaName: string; + state: WorkerCpuState; + sinceMs: number | undefined; + onPause: (() => void) | undefined; + onResume: (() => void) | undefined; +}) => { + const { colors } = useTheme(); + let message: string; + if (state.status === "error") { + message = `Couldn't read worker CPU on ${replicaName}: ${state.error ?? "unknown error"}. Readings so far are kept.`; + } else if (state.status === "paused") { + message = state.pausedForLimit + ? `Paused after ${SAMPLING_LIMIT_MS / millisecondsInMinute} minutes so the readings stop adding load to ${replicaName}.` + : "Paused."; + } else if (state.intervalMs > READ_INTERVAL_MS) { + message = `${replicaName} answered slowly, so readings now run every ${formatDurationExact(state.intervalMs)} to keep load off it.`; + } else { + message = `Reading worker CPU on ${replicaName} every ${READ_INTERVAL_MS / 1000}s${ + sinceMs !== undefined + ? ` since ${formatDate(new Date(sinceMs), TIME_FORMAT)}` + : "" + }. Stops when you close this, pauses after ${SAMPLING_LIMIT_MS / millisecondsInMinute} minutes.`; + } + const isReading = state.status === "reading"; + return ( + + + {message} + + + + ); +}; diff --git a/console/src/platform/clusters/ClusterPage/WorkerCpu/WorkerCpuVerdict.test.tsx b/console/src/platform/clusters/ClusterPage/WorkerCpu/WorkerCpuVerdict.test.tsx new file mode 100644 index 0000000000000..4262e9797fd09 --- /dev/null +++ b/console/src/platform/clusters/ClusterPage/WorkerCpu/WorkerCpuVerdict.test.tsx @@ -0,0 +1,221 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +import { screen } from "@testing-library/react"; +import { millisecondsInMinute } from "date-fns/constants"; +import React from "react"; + +import { DatabaseObject } from "~/api/materialize/objects"; +import { renderComponent } from "~/test/utils"; + +import { + CPU_DOCS, + CpuVerdict, + objectCpuRows, + WorkerReplica, +} from "./workerCpuModel"; +import { + FirstLook, + FirstLookProps, + Verdict, + VerdictProps, +} from "./WorkerCpuVerdict"; + +const SECOND_NS = 1e9; + +const REPLICA: WorkerReplica = { + id: "u2", + name: "r1", + size: "400cc", + processes: 1, + workersPerProcess: 4, + cpuNanoCoresPerProcess: 4e9, +}; + +const ORDERS_INDEX: DatabaseObject = { + id: "u5", + name: "orders_idx", + objectType: "index", + schemaId: "u3", + schemaName: "public", + databaseId: "u1", + databaseName: "materialize", + sourceType: null, + isWebhookTable: null, + clusterId: "u1", + clusterName: "quickstart", +}; + +const OBJECTS_BY_ID = new Map([[ORDERS_INDEX.id, ORDERS_INDEX]]); + +const objectName = (objectId: string) => + OBJECTS_BY_ID.get(objectId)?.name ?? objectId; + +// The index's dataflow took most of worker 0's time. +const INDEX_ON_WORKER_0 = objectCpuRows({ + latest: [50, 5, 5, 5].map((seconds, workerId) => ({ + dataflowId: "7", + workerId, + elapsedNs: seconds * SECOND_NS, + })), + exports: [{ objectId: ORDERS_INDEX.id, dataflowId: "7" }], + workerCount: 4, +}); + +const saturated = (): CpuVerdict => ({ + kind: "saturated", + busiest: { workerId: 0, busyFraction: 0.95 }, + averageBusyFraction: 0.31, + othersBusyFraction: 0.1, +}); + +const renderVerdict = ( + props: Pick & Partial, +) => + renderComponent( + , + ); + +const renderFirstLook = (props: Partial) => + renderComponent( + , + ); + +describe("Verdict", () => { + it("offers a copyable EXPLAIN ANALYZE, pinned to the replica, for the index on the saturated worker", async () => { + await renderVerdict({ + verdict: saturated(), + replicaCpuFraction: 0.9, + objects: INDEX_ON_WORKER_0, + }); + + expect( + screen.getByText(/EXPLAIN ANALYZE CPU WITH SKEW FOR INDEX/).textContent, + ).toBe( + [ + 'SET cluster = "quickstart";', + 'SET cluster_replica = "r1";', + 'EXPLAIN ANALYZE CPU WITH SKEW FOR INDEX "materialize"."public"."orders_idx";', + ].join("\n"), + ); + expect(screen.getByRole("button", { name: "Copy text" })).toBeVisible(); + }); + + it("attributes high CPU mostly outside running dataflows and links to ad-hoc query troubleshooting", async () => { + await renderVerdict({ + verdict: { + kind: "highCpu", + replicaCpuFraction: 0.9, + dataflowFraction: 0.2, + outsideFraction: 0.7, + busiest: { workerId: 1, busyFraction: 0.25 }, + }, + replicaCpuFraction: 0.9, + }); + + expect( + screen.getByText( + "r1 is using 90% of its CPU, mostly outside its running dataflows.", + ), + ).toBeVisible(); + expect( + screen.getByRole("link", { name: "Troubleshoot ad-hoc queries" }), + ).toHaveAttribute("href", `${CPU_DOCS}#ad-hoc-query-load`); + }); + + it("does not suggest splitting clusters when dataflows use most of the high CPU", async () => { + await renderVerdict({ + verdict: { + kind: "highCpu", + replicaCpuFraction: 0.9, + dataflowFraction: 0.8, + outsideFraction: 0.1, + busiest: { workerId: 1, busyFraction: 0.85 }, + }, + replicaCpuFraction: 0.9, + }); + + expect(screen.getByText(/mostly running dataflows/)).toBeVisible(); + expect(screen.queryByText(/smaller clusters/)).not.toBeInTheDocument(); + }); + + it("does not call the replica's CPU low for a saturated worker on a replica at 90%", async () => { + await renderVerdict({ verdict: saturated(), replicaCpuFraction: 0.9 }); + + expect(screen.getByText(/Work routed to it waits\./)).toBeVisible(); + expect(screen.queryByText(/only at/)).not.toBeInTheDocument(); + }); + + it("points out a saturated worker behind a replica at 30% of its CPU", async () => { + await renderVerdict({ verdict: saturated(), replicaCpuFraction: 0.3 }); + + expect( + screen.getByText( + /Work routed to it waits, even though r1 is only at 30% of its CPU\./, + ), + ).toBeVisible(); + }); +}); + +describe("FirstLook", () => { + it("names the worker that has done more than twice the others' average over its dataflows' lifetimes", async () => { + await renderFirstLook({ + lastReading: { + atMs: Date.now(), + elapsedNsByWorker: new Map( + [300, 50, 50, 0].map((seconds, workerId) => [ + workerId, + seconds * SECOND_NS, + ]), + ), + }, + }); + + expect( + screen.getByText( + /worker 0 has done 9\.0× as much as the other workers on average/, + ), + ).toBeVisible(); + }); + + it("leaves out a replica CPU metric older than 3 minutes", async () => { + await renderFirstLook({ + latestCpu: { atMs: Date.now() - 5 * millisecondsInMinute, percent: 80 }, + }); + + expect(screen.queryByText(/of its CPU/)).not.toBeInTheDocument(); + }); + + it("states a replica CPU metric from the last 3 minutes", async () => { + await renderFirstLook({ + latestCpu: { atMs: Date.now() - millisecondsInMinute, percent: 80 }, + }); + + expect(screen.getByText("r1 is at 80% of its CPU.")).toBeVisible(); + }); +}); diff --git a/console/src/platform/clusters/ClusterPage/WorkerCpu/WorkerCpuVerdict.tsx b/console/src/platform/clusters/ClusterPage/WorkerCpu/WorkerCpuVerdict.tsx new file mode 100644 index 0000000000000..db3caf74af7f2 --- /dev/null +++ b/console/src/platform/clusters/ClusterPage/WorkerCpu/WorkerCpuVerdict.tsx @@ -0,0 +1,377 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +import { HStack, Spinner, Text, useTheme, VStack } from "@chakra-ui/react"; +import { millisecondsInMinute } from "date-fns/constants"; +import React from "react"; + +import { + formatFullyQualifiedObjectName, + quoteIdentifier, +} from "~/api/materialize"; +import { DatabaseObject } from "~/api/materialize/objects"; +import Alert from "~/components/Alert"; +import { CopyableBox } from "~/components/copyableComponents"; +import TextLink from "~/components/TextLink"; +import { MaterializeTheme } from "~/theme"; +import { kebabToScreamingSpaceCase } from "~/util"; +import { formatDate, TIME_FORMAT } from "~/utils/dateFormat"; +import { formatDurationExact, formatPercentage } from "~/utils/format"; + +import { + ALL_BUSY_FRACTION, + CPU_DOCS, + CpuPoint, + CpuVerdict, + dominantObject, + largestShare, + lifetimeWorkerSkew, + objectCpuRows, + SKEW_RATIO, + SMOOTHING_WINDOW_MS, + topObjectOnWorker, + WorkerCpuReading, + workerLabel, + WorkerReplica, +} from "./workerCpuModel"; + +// Below this, a saturated worker hides behind a replica CPU that reads fine. +const LOW_CPU_FRACTION = 0.5; +// Less work than this outside dataflows isn't worth a sentence. +const OUTSIDE_NOTE_FRACTION = 0.05; +const EXPLAINABLE_TYPES = new Set(["index", "materialized-view"]); + +type ObjectCpu = ReturnType; + +// EXPLAIN ANALYZE reads the session's cluster and replica, so the statement +// pins both to the replica the drawer measured. +const explainStatement = ( + object: DatabaseObject, + clusterName: string, + replicaName: string, +) => + EXPLAINABLE_TYPES.has(object.objectType) + ? [ + `SET cluster = ${quoteIdentifier(clusterName)};`, + `SET cluster_replica = ${quoteIdentifier(replicaName)};`, + `EXPLAIN ANALYZE CPU WITH SKEW FOR ${kebabToScreamingSpaceCase(object.objectType)} ${formatFullyQualifiedObjectName(object, { quoteIdentifiers: true })};`, + ].join("\n") + : undefined; + +const Note = ({ children }: React.PropsWithChildren) => { + const { colors } = useTheme(); + return ( + + {children} + + ); +}; + +const DocLink = ({ + anchor, + children, +}: React.PropsWithChildren<{ anchor: string }>) => ( + + {children} + +); + +export interface FirstLookProps { + replica: WorkerReplica; + workerCount: number; + measuredMs: number; + latestCpu: CpuPoint | undefined; + lastReading: WorkerCpuReading | undefined; + lifetimeObjects: ObjectCpu | undefined; + objectName: (objectId: string) => string; +} + +/** What's known before the live window is long enough to judge. */ +export const FirstLook = ({ + replica, + workerCount, + measuredMs, + latestCpu, + lastReading, + lifetimeObjects, + objectName, +}: FirstLookProps) => { + // Metrics land about once a minute, so an older point is stale. + const isCpuCurrent = + latestCpu && Date.now() - latestCpu.atMs < 3 * millisecondsInMinute; + const lifetimeSkew = lastReading && lifetimeWorkerSkew(lastReading); + const topObject = + lifetimeObjects && + largestShare(lifetimeObjects.rows, lifetimeObjects.workerTotalsNs); + return ( + + + + Measuring {replica.name}'s {workerCount} workers ( + {formatDurationExact(Math.min(measuredMs, SMOOTHING_WINDOW_MS))} of{" "} + {SMOOTHING_WINDOW_MS / 1000}s) + + + } + message={ + + {isCpuCurrent && ( + + {replica.name} is at{" "} + {formatPercentage(latestCpu.percent / 100, 0)} of its CPU. + + )} + {(lifetimeSkew || topObject) && ( + + Over its dataflows' lifetimes,{" "} + {lifetimeSkew + ? lifetimeSkew.ratio >= SKEW_RATIO + ? Number.isFinite(lifetimeSkew.ratio) + ? `worker ${lifetimeSkew.workerId} has done ${lifetimeSkew.ratio.toFixed(1)}× as much as the other workers on average` + : `worker ${lifetimeSkew.workerId} has done all of the work` + : `the work is spread evenly across workers (the busiest has done ${lifetimeSkew.ratio.toFixed(2)}× as much as the others on average)` + : "the per-worker split isn't read yet"} + {topObject && + `, and ${objectName(topObject.objectId)} accounts for ${formatPercentage(topObject.share, 0)} of the dataflow CPU`} + . + + )} + + } + /> + ); +}; + +export interface VerdictProps { + clusterName: string; + replica: WorkerReplica; + verdict: CpuVerdict; + windowMs: number; + /** The replica's CPU over the verdict's window, from its metrics. */ + replicaCpuFraction: number | undefined; + objects: ObjectCpu | undefined; + /** When `objects` starts, or undefined for each dataflow's lifetime. */ + objectsSinceMs: number | undefined; + objectName: (objectId: string) => string; + objectsById: Map; +} + +/** The plain-language finding for the last few minutes of readings. */ +export const Verdict = ({ + clusterName, + replica, + verdict, + windowMs, + replicaCpuFraction, + objects, + objectsSinceMs, + objectName, + objectsById, +}: VerdictProps) => { + const windowLength = formatDurationExact(windowMs); + const objectsSpan = + objectsSinceMs !== undefined + ? `since ${formatDate(new Date(objectsSinceMs), TIME_FORMAT)}` + : "over its dataflows' lifetimes"; + const dominant = + objects && dominantObject(objects.rows, objects.workerTotalsNs); + const dominantSentence = + dominant && + `${objectName(dominant.objectId)} used ${formatPercentage(dominant.share, 0)} of the dataflow CPU ${objectsSpan}, so start there.`; + + if (verdict.kind === "idle") { + return ( + + ); + } + + if (verdict.kind === "highCpu") { + const { dataflowFraction, outsideFraction, busiest, uneven } = verdict; + const cpu = formatPercentage(verdict.replicaCpuFraction, 0); + const dataflowSentence = `Over the last ${windowLength} its workers spent ${formatPercentage(dataflowFraction, 0)} of the allocation running dataflows, and the busiest worker ${formatPercentage(busiest.busyFraction, 0)} of its core${ + uneven + ? `. Work is uneven, ${workerLabel(uneven.busiest.workerId, replica)} at ${formatPercentage(uneven.busiest.busyFraction, 0)} of its core against ${formatPercentage(uneven.othersBusyFraction, 0)} for the other workers, but no worker is at its limit, so skew isn't what uses the CPU.` + : ", so this isn't worker skew." + }`; + // Sizing up helps only when dataflows use the CPU. Work outside them + // grows with the operator count on every worker, which a bigger size + // raises. + if (outsideFraction > dataflowFraction) { + return ( + + + {dataflowSentence} The other{" "} + {formatPercentage(outsideFraction, 0)} went to work that + isn't counted against a running dataflow: answering + queries, dataflows that finished between readings, such as those + built for ad-hoc queries, and scheduling, progress tracking and + networking. + + + {dominantSentence ? `${dominantSentence} ` : ""} + Work outside dataflows grows with the number of operators on + every worker, so sizing up doesn't always lower it. + Splitting objects across smaller clusters can. + + + + Troubleshoot ad-hoc queries + + + Troubleshoot an undersized cluster + + + + } + /> + ); + } + return ( + + + {dataflowSentence} + {dataflowFraction >= ALL_BUSY_FRACTION && + " Less than 10% of the allocation is left idle."} + {outsideFraction >= OUTSIDE_NOTE_FRACTION && + ` Another ${formatPercentage(outsideFraction, 0)} went to work outside running dataflows.`} + + + {dominantSentence ?? + "Where the CPU went ranks objects by their CPU time."} + + {dominant ? ( + + Troubleshoot an expensive object + + ) : ( + + Troubleshoot an undersized cluster + + )} + + } + /> + ); + } + + const { busiest, averageBusyFraction, othersBusyFraction } = verdict; + const busiestName = workerLabel(busiest.workerId, replica); + + if (verdict.kind === "saturated" || verdict.kind === "uneven") { + const top = + objects && + topObjectOnWorker(objects.rows, objects.workerTotalsNs, busiest.workerId); + const topObject = top && objectsById.get(top.objectId); + const statement = + topObject && explainStatement(topObject, clusterName, replica.name); + const isCpuLow = + replicaCpuFraction !== undefined && replicaCpuFraction < LOW_CPU_FRACTION; + return ( + + + Over the last {windowLength} {busiestName} ran at{" "} + {formatPercentage(busiest.busyFraction, 0)} of its core, while the + other workers averaged {formatPercentage(othersBusyFraction, 0)}. + {verdict.kind === "saturated" + ? ` Work routed to it waits${ + isCpuLow + ? `, even though ${replica.name} is only at ${formatPercentage(replicaCpuFraction, 0)} of its CPU` + : "" + }.` + : " No worker is at its limit yet."} + + {top && ( + + {objectName(top.objectId)} took {formatPercentage(top.share, 0)}{" "} + of its time {objectsSpan}. + {statement && + " To find the operator, and the key it hashes on, run:"} + + )} + {statement && ( + + )} + Troubleshoot worker skew + + } + /> + ); + } + + if (verdict.kind === "allBusy") { + return ( + + + Over the last {windowLength} the average worker ran at{" "} + {formatPercentage(averageBusyFraction, 0)} of its core and{" "} + {busiestName}, the busiest, at{" "} + {formatPercentage(busiest.busyFraction, 0)}, so the work is spread + evenly and {replica.name} has less than 10% headroom. + {dominantSentence && ` ${dominantSentence}`} + + {dominant ? ( + + Troubleshoot an expensive object + + ) : ( + + Troubleshoot an undersized cluster + + )} + + } + /> + ); + } + + if (verdict.kind === "light") { + return ( + + ); + } + + return ( + + ); +}; diff --git a/console/src/platform/clusters/ClusterPage/WorkerCpu/useReplicaCpu.ts b/console/src/platform/clusters/ClusterPage/WorkerCpu/useReplicaCpu.ts new file mode 100644 index 0000000000000..2b56bf309bc66 --- /dev/null +++ b/console/src/platform/clusters/ClusterPage/WorkerCpu/useReplicaCpu.ts @@ -0,0 +1,77 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +import { useLiveQuery } from "@tanstack/react-db"; +import { subMinutes } from "date-fns"; +import { useAtomValue } from "jotai"; +import React from "react"; + +import { + buildReplicaProcessMetricsSubscribe, + ProcessMetricsSample, + processSampleKey, +} from "~/api/materialize/cluster/replicaProcessMetrics"; +import { createScopedSubscribeCollections } from "~/api/materialize/subscribeCollection"; +import { SubscribeRow } from "~/api/materialize/SubscribeManager"; +import { useGlobalSubscribeCollection } from "~/api/materialize/useSubscribe"; +import { syncEngineCacheScopeLoadableAtom } from "~/store/syncEngineCache"; + +import { replicaCpuPoints } from "./workerCpuModel"; + +// Covers the drawer's chart window and the lead before its first reading. +const RETENTION_MINUTES = 20; + +const metricsByReplica = createScopedSubscribeCollections( + { id: "worker-cpu-replica-metrics", getKey: processSampleKey }, +); + +/** + * A replica's CPU since `RETENTION_MINUTES` before the calling component + * mounted, from its raw per-minute metrics, streamed while it stays mounted. The resource + * usage card's series can't serve the drawer: past a 1h window they are + * per-bucket peaks, and past 24h they stop at the last poll. + */ +export function useReplicaCpu({ + replicaId, + cpuNanoCoresPerProcess, +}: { + replicaId: string; + cpuNanoCoresPerProcess: number; +}) { + const scope = useAtomValue(syncEngineCacheScopeLoadableAtom); + const samples = + scope.state === "hasData" && scope.data + ? metricsByReplica.get(scope.data, replicaId) + : undefined; + const options = React.useMemo( + () => ({ + target: samples ?? metricsByReplica.placeholder, + subscribe: samples + ? buildReplicaProcessMetricsSubscribe({ + replicaIds: [replicaId], + since: subMinutes(new Date(), RETENTION_MINUTES), + includeSwap: false, + }) + : undefined, + select: (row: SubscribeRow) => row.data, + upsertKey: (row: SubscribeRow) => + processSampleKey(row.data), + }), + [samples, replicaId], + ); + useGlobalSubscribeCollection(options); + const { data: rows } = useLiveQuery( + (q) => (samples ? q.from({ sample: samples.collection }) : undefined), + [samples], + ); + return React.useMemo( + () => replicaCpuPoints(rows ?? [], cpuNanoCoresPerProcess), + [rows, cpuNanoCoresPerProcess], + ); +} diff --git a/console/src/platform/clusters/ClusterPage/WorkerCpu/workerCpuModel.test.ts b/console/src/platform/clusters/ClusterPage/WorkerCpu/workerCpuModel.test.ts new file mode 100644 index 0000000000000..aef7da5aad0c4 --- /dev/null +++ b/console/src/platform/clusters/ClusterPage/WorkerCpu/workerCpuModel.test.ts @@ -0,0 +1,463 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +import { + ADHOC_ROW_ID, + cpuVerdict, + DataflowWorkerCpu, + dominantObject, + hasCounterReset, + hasDataflowCounterReset, + lifetimeWorkerSkew, + objectCpuRows, + processOfWorker, + replicaCpuPoints, + topObjectOnWorker, + verdictWindowStart, + workerBusyBetween, + workerBusySeries, + WorkerCpuReading, + workerLabel, + WorkerReplica, + workersPerCore, + workerVerdict, +} from "./workerCpuModel"; + +const SECOND_NS = 1e9; + +const reading = (atS: number, elapsedS: number[]): WorkerCpuReading => ({ + atMs: atS * 1000, + elapsedNsByWorker: new Map(elapsedS.map((s, w) => [w, s * SECOND_NS])), +}); + +const busy = (fractions: number[]) => + workerBusyBetween( + reading( + 0, + fractions.map(() => 0), + ), + reading( + 10, + fractions.map((f) => f * 10), + ), + ); + +describe("workerBusyBetween", () => { + it("gives each worker's share of one core between two readings, busiest first", () => { + expect( + workerBusyBetween(reading(0, [100, 50, 7]), reading(10, [109, 51, 8])), + ).toEqual([ + { workerId: 0, busyFraction: 0.9 }, + { workerId: 1, busyFraction: 0.1 }, + { workerId: 2, busyFraction: 0.1 }, + ]); + }); + + it("counts a counter that went backwards, as after a restart, as idle", () => { + expect( + workerBusyBetween(reading(0, [500]), reading(10, [3]))[0].busyFraction, + ).toEqual(0); + }); + + it("caps a worker at one core when reads land unevenly", () => { + expect( + workerBusyBetween(reading(0, [0]), reading(10, [10.4]))[0].busyFraction, + ).toEqual(1); + }); +}); + +describe("workerBusySeries", () => { + it("names the busiest worker and the average between each pair of readings", () => { + expect( + workerBusySeries( + [reading(0, [0, 0]), reading(5, [4, 1]), reading(10, [5, 5])], + 0, + ), + ).toEqual([ + { + atMs: 5000, + busiest: { workerId: 0, busyFraction: 0.8 }, + averageBusyFraction: 0.5, + }, + { + atMs: 10000, + busiest: { workerId: 1, busyFraction: 0.8 }, + averageBusyFraction: 0.5, + }, + ]); + }); + + it("leaves out the span of a pause and marks where the line resumes", () => { + expect( + workerBusySeries( + [ + reading(0, [0]), + reading(5, [4]), + { ...reading(600, [100]), afterGap: true }, + reading(605, [101]), + ], + 0, + ).map(({ atMs, afterGap }) => [atMs, Boolean(afterGap)]), + ).toEqual([ + [5000, false], + [605000, true], + ]); + }); + + it("averages each point over the trailing window, smoothing one-off ticks", () => { + // Worker 0 is steadily at 50%, but one 5s read caught an extra tick. + const readings = [0, 5, 10, 15, 20, 25, 30, 35].map((atS) => + reading(atS, [atS * 0.5 + (atS >= 20 ? 1 : 0)]), + ); + + expect( + workerBusySeries(readings, 30_000).map(({ atMs, busiest }) => [ + atMs / 1000, + Math.round(busiest.busyFraction * 100), + ]), + ).toEqual([ + [10, 50], + [15, 50], + [20, 55], + [25, 54], + [30, 53], + [35, 53], + ]); + }); +}); + +describe("counter resets", () => { + it("flags a reading where any worker's counter fell, as when a dataflow is dropped", () => { + expect( + hasCounterReset( + reading(0, [100, 100]).elapsedNsByWorker, + reading(5, [104, 40]).elapsedNsByWorker, + ), + ).toBe(true); + expect( + hasCounterReset( + reading(0, [100, 100]).elapsedNsByWorker, + reading(5, [104, 103]).elapsedNsByWorker, + ), + ).toBe(false); + }); + + it("flags a per-dataflow read where any dataflow's counter fell, as after a restart", () => { + const row = (dataflowId: string, elapsedNs: number) => ({ + dataflowId, + workerId: 0, + elapsedNs, + }); + expect(hasDataflowCounterReset([row("5", 900)], [row("5", 20)])).toBe(true); + expect( + hasDataflowCounterReset([row("5", 900)], [row("5", 950), row("6", 1)]), + ).toBe(false); + }); + + it("starts a new line after a reset", () => { + expect( + workerBusySeries( + [ + reading(0, [0]), + reading(5, [4]), + { ...reading(10, [1]), afterReset: true }, + reading(15, [5]), + ], + 0, + ).map(({ atMs, afterGap }) => [atMs, Boolean(afterGap)]), + ).toEqual([ + [5000, false], + [15000, true], + ]); + }); +}); + +describe("verdictWindowStart", () => { + const readings = [0, 60, 120, 180, 240, 300, 360].map((atS) => + reading(atS, [atS]), + ); + + it("reaches back as far as the window", () => { + expect(verdictWindowStart(readings, 200_000)?.atMs).toEqual(180_000); + }); + + it("doesn't reach back past a reset, since counters before it don't compare", () => { + const withReset = readings.map((r) => + r.atMs === 300_000 ? { ...r, afterReset: true } : r, + ); + expect(verdictWindowStart(withReset, 200_000)?.atMs).toEqual(300_000); + }); + + it("spans a pause, since counters keep counting through it", () => { + const withGap = readings.map((r) => + r.atMs === 300_000 ? { ...r, afterGap: true } : r, + ); + expect(verdictWindowStart(withGap, 200_000)?.atMs).toEqual(180_000); + }); +}); + +describe("workerVerdict", () => { + it("calls one worker at its core's limit saturated when the rest have room", () => { + expect(workerVerdict(busy([0.97, 0.14, 0.12, 0.13])).kind).toEqual( + "saturated", + ); + }); + + it("catches a pegged worker on a two-worker replica", () => { + expect(workerVerdict(busy([1, 0.2])).kind).toEqual("saturated"); + }); + + it("catches a pegged worker whose peers each run well behind it", () => { + expect(workerVerdict(busy([1, 0.4, 0.4, 0.4])).kind).toEqual("saturated"); + }); + + it("calls an evenly busy replica short on CPU, not skewed", () => { + expect(workerVerdict(busy([0.98, 0.9, 0.88, 0.92])).kind).toEqual( + "allBusy", + ); + }); + + it("calls a worker at twice the average uneven when it still has room", () => { + expect(workerVerdict(busy([0.5, 0.1, 0.1, 0.1])).kind).toEqual("uneven"); + }); + + it("calls a lead over the average light while no worker nears its limit", () => { + expect(workerVerdict(busy([0.08, 0.02, 0.02, 0.02])).kind).toEqual("light"); + }); + + it("calls workers within twice the average balanced", () => { + expect(workerVerdict(busy([0.21, 0.17, 0.16, 0.18])).kind).toEqual( + "balanced", + ); + }); + + it("calls a replica with almost no dataflow work idle, whatever the ratios", () => { + expect(workerVerdict(busy([0.02, 0, 0, 0])).kind).toEqual("idle"); + }); +}); + +describe("cpuVerdict", () => { + it("leads with a busy replica's CPU when its workers share the work evenly", () => { + const verdict = cpuVerdict(busy([0.58, 0.55, 0.54, 0.53]), 0.89, 1); + expect(verdict).toMatchObject({ + kind: "highCpu", + replicaCpuFraction: 0.89, + dataflowFraction: 0.55, + }); + expect(verdict.kind === "highCpu" && verdict.outsideFraction).toBeCloseTo( + 0.34, + ); + }); + + it("measures dataflow time against cores when a size runs fewer workers than cores", () => { + expect( + cpuVerdict(busy([0.5, 0.5, 0.5, 0.5]), 0.8, workersPerCore(4, 16e9)), + ).toMatchObject({ kind: "highCpu", dataflowFraction: 0.125 }); + }); + + it("explains a busy replica's CPU even when its dataflows are idle", () => { + expect(cpuVerdict(busy([0.004, 0, 0, 0]), 0.8, 1)).toMatchObject({ + kind: "highCpu", + dataflowFraction: 0.001, + outsideFraction: 0.799, + }); + }); + + it("leads with a busy replica's CPU over uneven work, keeping the uneven worker", () => { + expect(cpuVerdict(busy([0.7, 0.2, 0.2, 0.2]), 0.9, 1)).toMatchObject({ + kind: "highCpu", + uneven: { busiest: { workerId: 0, busyFraction: 0.7 } }, + }); + }); + + it("reports a pegged worker on a busy two-worker replica as saturated, not high CPU", () => { + expect(cpuVerdict(busy([1, 0.2]), 0.8, 1).kind).toEqual("saturated"); + }); + + it("reports a saturated worker even on a busy replica", () => { + expect(cpuVerdict(busy([0.97, 0.14, 0.12, 0.13]), 0.9, 1).kind).toEqual( + "saturated", + ); + }); + + it("judges the workers alone when replica CPU is low or unknown", () => { + expect(cpuVerdict(busy([0.21, 0.17, 0.16, 0.18]), 0.3, 1).kind).toEqual( + "balanced", + ); + expect( + cpuVerdict(busy([0.21, 0.17, 0.16, 0.18]), undefined, 1).kind, + ).toEqual("balanced"); + }); +}); + +describe("lifetimeWorkerSkew", () => { + it("compares the busiest worker's lifetime time to the other workers' average", () => { + expect(lifetimeWorkerSkew(reading(0, [70, 10, 10, 10]))).toEqual({ + workerId: 0, + ratio: 7, + }); + expect(lifetimeWorkerSkew(reading(0, [100, 20]))?.ratio).toEqual(5); + }); +}); + +describe("objectCpuRows", () => { + const exports = [ + { dataflowId: "1", objectId: "u10" }, + { dataflowId: "2", objectId: "u20" }, + ]; + const cpu = (dataflowId: string, perWorkerS: number[]): DataflowWorkerCpu[] => + perWorkerS.map((s, workerId) => ({ + dataflowId, + workerId, + elapsedNs: s * SECOND_NS, + })); + + it("measures each object's hottest worker against an even share", () => { + const { rows } = objectCpuRows({ + latest: [ + ...cpu("1", [70, 10, 10, 10]), + ...cpu("2", [300, 300, 300, 300]), + ], + exports, + workerCount: 4, + }); + + expect( + [...rows] + .sort((a, b) => a.objectId.localeCompare(b.objectId)) + .map(({ objectId, hottestWorkerId, ratio, extraNs }) => [ + objectId, + hottestWorkerId, + ratio, + extraNs / SECOND_NS, + ]), + ).toEqual([ + ["u10", 0, 2.8, 45], + ["u20", 0, 1, 0], + ]); + }); + + it("measures only the time since the baseline read", () => { + const { rows } = objectCpuRows({ + baseline: cpu("1", [1000, 1000]), + latest: cpu("1", [1030, 1010]), + exports, + workerCount: 2, + }); + + expect(rows[0].perWorkerNs.map((ns) => ns / SECOND_NS)).toEqual([30, 10]); + }); + + it("pools temporary dataflows for ad-hoc queries into one row", () => { + const { rows } = objectCpuRows({ + latest: [...cpu("7", [3, 1]), ...cpu("8", [1, 1])], + exports: [ + { dataflowId: "7", objectId: "t41" }, + { dataflowId: "8", objectId: "t42" }, + ], + workerCount: 2, + }); + + expect( + rows.map(({ objectId, dataflowCount, perWorkerNs }) => [ + objectId, + dataflowCount, + perWorkerNs.map((ns) => ns / SECOND_NS), + ]), + ).toEqual([[ADHOC_ROW_ID, 2, [4, 2]]]); + }); + + it("names an object that takes most of the replica's dataflow time", () => { + const heavy = objectCpuRows({ + latest: [ + ...cpu("1", [60, 60]), + ...cpu("2", [20, 20]), + ...cpu("99", [20, 20]), + ], + exports, + workerCount: 2, + }); + const spread = objectCpuRows({ + latest: [ + ...cpu("1", [40, 40]), + ...cpu("2", [35, 35]), + ...cpu("99", [25, 25]), + ], + exports, + workerCount: 2, + }); + + expect(dominantObject(heavy.rows, heavy.workerTotalsNs)).toEqual({ + objectId: "u10", + share: 0.6, + }); + expect(dominantObject(spread.rows, spread.workerTotalsNs)).toBeUndefined(); + }); + + it("counts system and transient dataflows in each worker's total without a row", () => { + const { rows, workerTotalsNs } = objectCpuRows({ + latest: [...cpu("1", [6, 2]), ...cpu("99", [4, 0])], + exports, + workerCount: 2, + }); + + expect(rows.map((row) => row.objectId)).toEqual(["u10"]); + expect(workerTotalsNs.map((ns) => ns / SECOND_NS)).toEqual([10, 2]); + expect(topObjectOnWorker(rows, workerTotalsNs, 0)).toEqual({ + objectId: "u10", + share: 0.6, + }); + }); +}); + +describe("processOfWorker", () => { + it("numbers workers process by process", () => { + expect([0, 61, 62, 123].map((w) => processOfWorker(w, 62))).toEqual([ + 0, 0, 1, 1, + ]); + }); +}); + +describe("replicaCpuPoints", () => { + const sample = ( + atS: number, + cores: number | null, + ): Parameters[0][number] => ({ + occurredAt: new Date(atS * 1000), + cpuNanoCores: cores === null ? null : cores * 1e9, + }); + + it("averages one scrape's processes as a percent of a process's allocation", () => { + expect( + replicaCpuPoints( + [sample(60, 3), sample(61, 1), sample(120, 2), sample(120, null)], + 4e9, + ), + ).toEqual([ + { atMs: 60_000, percent: 50 }, + { atMs: 120_000, percent: 50 }, + ]); + }); +}); + +describe("workerLabel", () => { + const replica = (processes: number): WorkerReplica => ({ + id: "u1", + name: "r1", + size: "large", + processes, + workersPerProcess: 31, + cpuNanoCoresPerProcess: 31e9, + }); + + it("names a worker's process only when the replica runs more than one", () => { + expect(workerLabel(67, replica(4))).toEqual("w67 (process 2)"); + expect(workerLabel(7, replica(1))).toEqual("w7"); + }); +}); diff --git a/console/src/platform/clusters/ClusterPage/WorkerCpu/workerCpuModel.ts b/console/src/platform/clusters/ClusterPage/WorkerCpu/workerCpuModel.ts new file mode 100644 index 0000000000000..caf57bfa78f61 --- /dev/null +++ b/console/src/platform/clusters/ClusterPage/WorkerCpu/workerCpuModel.ts @@ -0,0 +1,569 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +import { millisecondsInMinute } from "date-fns/constants"; + +import { ProcessMetricsSample } from "~/api/materialize/cluster/replicaProcessMetrics"; +import docUrls from "~/mz-doc-urls.json"; + +/** + * @module + * Turns cumulative per-worker CPU counters into what the workers drawer shows. + * + * A worker's counter is the time it spent in every running dataflow, each + * from when the dataflow started, so one reading is a lifetime total. The + * difference between two readings is the time spent in between, and dividing + * it by the wall-clock time between them gives the share of one core a worker + * used. A worker is one thread, so that share is at most 1. + * + * NOTE: a dataflow's time leaves the counters when it shuts down, so the + * counters fall when a dataflow is dropped or the replica restarts, and work + * in dataflows that finished between two readings is never seen. + */ + +export const CPU_DOCS = + docUrls["/docs/clusters/troubleshoot-clusters/cpu-troubleshooting/"]; + +/** One read of every worker's cumulative dataflow time on a replica. */ +export interface WorkerCpuReading { + atMs: number; + elapsedNsByWorker: Map; + /** The first reading after a pause, so no point spans the pause. */ + afterGap?: boolean; + /** + * A counter fell since the previous reading, so the two don't compare and + * nothing is measured across them. + */ + afterReset?: boolean; +} + +export interface WorkerBusy { + workerId: number; + /** Share of one core spent running dataflows. */ + busyFraction: number; +} + +export interface WorkerBusyPoint { + atMs: number; + busiest: WorkerBusy; + averageBusyFraction: number; + /** The first point after a gap, so a line doesn't bridge it. */ + afterGap?: boolean; +} + +// A counter that fell lost a dataflow's time, so its time since the earlier +// reading is unknown. Treating it as zero avoids negative work. +const delta = (before: number | undefined, after: number) => + before === undefined ? after : Math.max(0, after - before); + +/** Whether any worker's counter fell between two readings. */ +export function hasCounterReset( + before: Map, + after: Map, +) { + for (const [workerId, ns] of after) { + if (ns < (before.get(workerId) ?? 0)) return true; + } + return false; +} + +const isSegmentStart = (reading: WorkerCpuReading) => + Boolean(reading.afterGap || reading.afterReset); + +/** Each worker's busy share between two readings, busiest first. */ +export function workerBusyBetween( + previous: WorkerCpuReading, + next: WorkerCpuReading, +): WorkerBusy[] { + const elapsedNs = (next.atMs - previous.atMs) * 1e6; + if (elapsedNs <= 0) return []; + return [...next.elapsedNsByWorker] + .map(([workerId, ns]) => ({ + workerId, + busyFraction: Math.min( + 1, + delta(previous.elapsedNsByWorker.get(workerId), ns) / elapsedNs, + ), + })) + .sort((a, b) => b.busyFraction - a.busyFraction || a.workerId - b.workerId); +} + +const averageBusy = (workers: WorkerBusy[]) => + workers.length === 0 + ? 0 + : workers.reduce((sum, worker) => sum + worker.busyFraction, 0) / + workers.length; + +/** + * How much time each plotted point averages over. Replicas update the + * counters once per introspection interval (1s by default), so a short window + * can catch one tick more or fewer than the next: 20% of a 5s window, 3% of + * 30s. + */ +export const SMOOTHING_WINDOW_MS = 30_000; +// A point needs at least this much time behind it to be worth drawing. +const MIN_POINT_WINDOW_MS = 10_000; + +/** + * The busiest and average worker at each reading, each over the trailing + * `windowMs`, or less early on or right after a pause. + */ +export function workerBusySeries( + readings: WorkerCpuReading[], + windowMs = SMOOTHING_WINDOW_MS, +): WorkerBusyPoint[] { + const points: WorkerBusyPoint[] = []; + let segmentStart = 0; + let isNewSegment = false; + for (let i = 1; i < readings.length; i++) { + if (isSegmentStart(readings[i])) { + segmentStart = i; + isNewSegment = points.length > 0; + continue; + } + const atMs = readings[i].atMs; + let from = segmentStart; + while (from + 1 < i && atMs - readings[from + 1].atMs >= windowMs) { + from++; + } + if (atMs - readings[from].atMs < Math.min(windowMs, MIN_POINT_WINDOW_MS)) { + continue; + } + const workers = workerBusyBetween(readings[from], readings[i]); + if (workers.length === 0) continue; + points.push({ + atMs, + busiest: workers[0], + averageBusyFraction: averageBusy(workers), + ...(isNewSegment && { afterGap: true }), + }); + isNewSegment = false; + } + return points; +} + +/** + * The earliest reading the verdict compares against: within `windowMs` of the + * last one, and not before a counter reset. + */ +export function verdictWindowStart( + readings: WorkerCpuReading[], + windowMs: number, +) { + const last = readings.at(-1); + if (!last) return undefined; + let start = readings.length - 1; + while ( + start > 0 && + !readings[start].afterReset && + last.atMs - readings[start - 1].atMs <= windowMs + ) { + start--; + } + return readings[start]; +} + +/** + * A worker past this many times the other workers' average carries skewed + * work. The dataflow troubleshooting docs use 2 as their example threshold. + */ +export const SKEW_RATIO = 2; +/** A worker this busy is at its core's limit, so work routed to it waits. */ +const SATURATED_FRACTION = 0.9; +/** + * Skew only costs anything as a worker nears its core's limit, so below this + * the busiest worker's lead over the average isn't worth a warning. + */ +const UNEVEN_FRACTION = 0.5; +/** + * Workers this busy on average leave the replica under 10% idle, the CPU + * troubleshooting guide's rule of thumb for having no headroom. + */ +export const ALL_BUSY_FRACTION = 0.9; +/** An object with this share of all dataflow time dominates the replica. */ +const DOMINANT_SHARE = 0.5; +// Below these the replica did almost no dataflow work, so ratios are noise. +const IDLE_AVERAGE_FRACTION = 0.01; +const IDLE_BUSIEST_FRACTION = 0.05; + +export type WorkerVerdict = + | { kind: "idle" } + | { + /** `light`: uneven, but no worker is near its core's limit. */ + kind: "saturated" | "allBusy" | "uneven" | "light" | "balanced"; + busiest: WorkerBusy; + averageBusyFraction: number; + /** Every worker but the busiest, the average skew is judged against. */ + othersBusyFraction: number; + }; + +/** + * What a window's worker shares say, from the most to the least urgent. + * + * NOTE: skew compares the busiest worker with the others. An average that + * includes the busiest worker can't pass `SKEW_RATIO` on two workers unless + * the other is fully idle. + */ +export function workerVerdict(workers: WorkerBusy[]): WorkerVerdict { + const busiest = workers[0]; + const averageBusyFraction = averageBusy(workers); + const othersBusyFraction = averageBusy(workers.slice(1)); + if ( + !busiest || + (averageBusyFraction < IDLE_AVERAGE_FRACTION && + busiest.busyFraction < IDLE_BUSIEST_FRACTION) + ) { + return { kind: "idle" }; + } + const result = { busiest, averageBusyFraction, othersBusyFraction }; + if (averageBusyFraction >= ALL_BUSY_FRACTION) { + return { kind: "allBusy", ...result }; + } + const isSkewed = + workers.length > 1 && + busiest.busyFraction >= SKEW_RATIO * othersBusyFraction; + if (isSkewed && busiest.busyFraction >= SATURATED_FRACTION) { + return { kind: "saturated", ...result }; + } + if (isSkewed && busiest.busyFraction >= UNEVEN_FRACTION) { + return { kind: "uneven", ...result }; + } + if (isSkewed) return { kind: "light", ...result }; + return { kind: "balanced", ...result }; +} + +/** Replica CPU this high is worth explaining, whatever the workers show. */ +const HIGH_CPU_FRACTION = 0.75; + +export type CpuVerdict = + | WorkerVerdict + | { + kind: "highCpu"; + replicaCpuFraction: number; + /** The share of the replica's allocation spent running dataflows. */ + dataflowFraction: number; + /** + * The rest of the replica's CPU. It includes dataflows that finished + * between readings, since a dataflow's time leaves the counters when it + * shuts down. + */ + outsideFraction: number; + busiest: WorkerBusy; + /** Set when the work is also uneven, short of saturating a worker. */ + uneven?: { busiest: WorkerBusy; othersBusyFraction: number }; + }; + +/** Whether a verdict says one worker carries more than its share. */ +export const isSkewVerdict = (verdict: CpuVerdict | undefined) => + verdict?.kind === "saturated" || verdict?.kind === "uneven"; + +/** + * A saturated worker is the most specific finding, so it wins. Otherwise a + * replica near its allocation leads with that, even with idle workers, since + * its CPU then goes to work outside the running dataflows, and uneven work + * short of a saturated worker becomes a note on it. + * `replicaCpuFraction` is the replica's CPU as a share of its allocation over + * the same window, from its metrics. + */ +export function cpuVerdict( + workers: WorkerBusy[], + replicaCpuFraction: number | undefined, + workersPerCore: number, +): CpuVerdict { + const verdict = workerVerdict(workers); + if ( + workers.length > 0 && + replicaCpuFraction !== undefined && + replicaCpuFraction >= HIGH_CPU_FRACTION && + verdict.kind !== "saturated" + ) { + const dataflowFraction = averageBusy(workers) * workersPerCore; + return { + kind: "highCpu", + replicaCpuFraction, + dataflowFraction, + outsideFraction: Math.max(0, replicaCpuFraction - dataflowFraction), + busiest: workers[0], + ...(verdict.kind === "uneven" && { + uneven: { + busiest: verdict.busiest, + othersBusyFraction: verdict.othersBusyFraction, + }, + }), + }; + } + return verdict; +} + +/** + * Workers per core of the replica's CPU allocation, which turns the average + * worker's share of a core into a share of the allocation. Production sizes + * run one worker per core, and some test sizes run fewer. + */ +export const workersPerCore = ( + workersPerProcess: number, + cpuNanoCoresPerProcess: number, +) => + cpuNanoCoresPerProcess > 0 + ? workersPerProcess / (cpuNanoCoresPerProcess / 1e9) + : 1; + +/** Below this ratio an object's work counts as even across its workers. */ +export const EVEN_RATIO = 1.25; +/** An object with less of the dataflow time than this is too small to flag. */ +export const NOTABLE_SHARE = 0.01; + +/** + * The busiest worker's cumulative time over the other workers' average, from + * one reading. Counters start with each dataflow, so this is a lifetime skew, + * available before a second reading can give a rate. `ratio` is infinite when + * the others have done nothing. + */ +export function lifetimeWorkerSkew(reading: WorkerCpuReading) { + let busiest: { workerId: number; elapsedNs: number } | undefined; + let totalNs = 0; + for (const [workerId, elapsedNs] of reading.elapsedNsByWorker) { + totalNs += elapsedNs; + if (!busiest || elapsedNs > busiest.elapsedNs) { + busiest = { workerId, elapsedNs }; + } + } + const workerCount = reading.elapsedNsByWorker.size; + if (!busiest || totalNs === 0 || workerCount < 2) return undefined; + const othersAverageNs = (totalNs - busiest.elapsedNs) / (workerCount - 1); + return { + workerId: busiest.workerId, + ratio: othersAverageNs > 0 ? busiest.elapsedNs / othersAverageNs : Infinity, + }; +} + +/** One dataflow's cumulative time on one worker. */ +export interface DataflowWorkerCpu { + dataflowId: string; + workerId: number; + elapsedNs: number; +} + +export interface DataflowExport { + objectId: string; + dataflowId: string; +} + +/** The row that pools the temporary dataflows built for ad-hoc queries. */ +export const ADHOC_ROW_ID = "adhoc"; + +export interface ObjectCpuRow { + /** A user object's id, or `ADHOC_ROW_ID`. */ + objectId: string; + /** How many dataflows the row pools: 1 for an object. */ + dataflowCount: number; + /** Indexed by worker id. */ + perWorkerNs: number[]; + totalNs: number; + averageNs: number; + hottestWorkerId: number; + /** The hottest worker's time over the average worker's. */ + ratio: number; + /** Time the hottest worker spent beyond the average worker. */ + extraNs: number; +} + +const key = (row: { dataflowId: string; workerId: number }) => + `${row.dataflowId}|${row.workerId}`; + +/** + * Whether any dataflow's counter on any worker fell between two per-dataflow + * reads. Dataflow ids restart with the replica, so a restart reads as a fall. + */ +export function hasDataflowCounterReset( + baseline: DataflowWorkerCpu[], + latest: DataflowWorkerCpu[], +) { + const before = new Map(baseline.map((row) => [key(row), row.elapsedNs])); + return latest.some((row) => row.elapsedNs < (before.get(key(row)) ?? 0)); +} + +/** + * Each user object's time per worker between two reads, or over its + * lifetime without `baseline`, plus every worker's total across all + * dataflows, system and transient ones included. + */ +export function objectCpuRows({ + baseline, + latest, + exports, + workerCount, +}: { + baseline?: DataflowWorkerCpu[]; + latest: DataflowWorkerCpu[]; + exports: DataflowExport[]; + workerCount: number; +}) { + const before = new Map((baseline ?? []).map((row) => [key(row), row])); + // A dataflow can export more than one object. The row is named after the + // first by id. + const objectByDataflow = new Map(); + for (const { dataflowId, objectId } of [...exports].sort((a, b) => + a.objectId.localeCompare(b.objectId), + )) { + if (!objectByDataflow.has(dataflowId)) { + objectByDataflow.set(dataflowId, objectId); + } + } + const workerTotalsNs = new Array(workerCount).fill(0); + const perWorkerByObject = new Map(); + const dataflowsByObject = new Map>(); + for (const row of latest) { + if (row.workerId >= workerCount) continue; + const ns = delta(before.get(key(row))?.elapsedNs, row.elapsedNs); + workerTotalsNs[row.workerId] += ns; + const exported = objectByDataflow.get(row.dataflowId); + if (!exported) continue; + const objectId = exported.startsWith("t") ? ADHOC_ROW_ID : exported; + let perWorker = perWorkerByObject.get(objectId); + if (!perWorker) { + perWorker = new Array(workerCount).fill(0); + perWorkerByObject.set(objectId, perWorker); + dataflowsByObject.set(objectId, new Set()); + } + perWorker[row.workerId] += ns; + if (ns > 0) dataflowsByObject.get(objectId)?.add(row.dataflowId); + } + const rows: ObjectCpuRow[] = []; + for (const [objectId, perWorkerNs] of perWorkerByObject) { + const totalNs = perWorkerNs.reduce((sum, ns) => sum + ns, 0); + if (totalNs === 0) continue; + const averageNs = totalNs / workerCount; + let hottestWorkerId = 0; + perWorkerNs.forEach((ns, workerId) => { + if (ns > perWorkerNs[hottestWorkerId]) hottestWorkerId = workerId; + }); + const hottestNs = perWorkerNs[hottestWorkerId]; + rows.push({ + objectId, + dataflowCount: dataflowsByObject.get(objectId)?.size ?? 0, + perWorkerNs, + totalNs, + averageNs, + hottestWorkerId, + ratio: hottestNs / averageNs, + extraNs: hottestNs - averageNs, + }); + } + return { rows, workerTotalsNs }; +} + +/** The object with the most dataflow time, with its share of all of it. */ +export function largestShare(rows: ObjectCpuRow[], workerTotalsNs: number[]) { + const totalNs = workerTotalsNs.reduce((sum, ns) => sum + ns, 0); + const top = rows + .filter((row) => row.objectId !== ADHOC_ROW_ID) + .reduce( + (best, row) => (!best || row.totalNs > best.totalNs ? row : best), + undefined, + ); + if (!top || totalNs === 0) return undefined; + return { objectId: top.objectId, share: top.totalNs / totalNs }; +} + +/** + * The object that took at least `DOMINANT_SHARE` of all dataflow time, with + * its share, if one did. + */ +export function dominantObject(rows: ObjectCpuRow[], workerTotalsNs: number[]) { + const largest = largestShare(rows, workerTotalsNs); + return largest && largest.share >= DOMINANT_SHARE ? largest : undefined; +} + +/** The object that took most of a worker's time, with its share of it. */ +export function topObjectOnWorker( + rows: ObjectCpuRow[], + workerTotalsNs: number[], + workerId: number, +) { + const totalNs = workerTotalsNs[workerId] ?? 0; + if (totalNs === 0) return undefined; + let top: ObjectCpuRow | undefined; + for (const row of rows) { + if (!top || row.perWorkerNs[workerId] > top.perWorkerNs[workerId]) { + top = row; + } + } + if (!top || top.perWorkerNs[workerId] === 0) return undefined; + return { + objectId: top.objectId, + share: top.perWorkerNs[workerId] / totalNs, + }; +} + +/** + * The process a worker runs in. Timely numbers workers process by process, + * `my_process * threads + index` in its zero-copy allocator. + */ +export const processOfWorker = (workerId: number, workersPerProcess: number) => + Math.floor(workerId / Math.max(1, workersPerProcess)); + +/** A replica the workers drawer reads, with its size's shape. */ +export interface WorkerReplica { + id: string; + name: string; + size: string; + processes: number; + workersPerProcess: number; + cpuNanoCoresPerProcess: number; +} + +/** A worker's name, with its process when the replica runs more than one. */ +export const workerLabel = ( + workerId: number, + { workersPerProcess, processes }: WorkerReplica, +) => + processes > 1 + ? `w${workerId} (process ${processOfWorker(workerId, workersPerProcess)})` + : `w${workerId}`; + +/** The replica's CPU at one time, as a percent of its allocation. */ +export interface CpuPoint { + atMs: number; + percent: number; +} + +/** + * A replica's CPU per minute, as a percent of its allocation. A multi-process + * replica averages the processes that reported, so a process whose metrics + * fetch failed doesn't halve the line. + */ +export function replicaCpuPoints( + samples: Pick[], + cpuNanoCoresPerProcess: number, +): CpuPoint[] { + if (cpuNanoCoresPerProcess <= 0) return []; + const byMinute = new Map(); + for (const { occurredAt, cpuNanoCores } of samples) { + if (cpuNanoCores === null) continue; + // Grouped by minute, the scrape interval, so one scrape's processes + // average together whether or not their timestamps match exactly. + const atMs = + Math.floor(occurredAt.getTime() / millisecondsInMinute) * + millisecondsInMinute; + const values = byMinute.get(atMs) ?? []; + values.push(cpuNanoCores); + byMinute.set(atMs, values); + } + return [...byMinute] + .sort(([a], [b]) => a - b) + .map(([atMs, values]) => ({ + atMs, + percent: + (values.reduce((sum, value) => sum + value, 0) / + values.length / + cpuNanoCoresPerProcess) * + 100, + })); +} diff --git a/console/src/platform/clusters/ClusterPage/WorkerCpu/workerCpuSampler.test.ts b/console/src/platform/clusters/ClusterPage/WorkerCpu/workerCpuSampler.test.ts new file mode 100644 index 0000000000000..fffda869e1362 --- /dev/null +++ b/console/src/platform/clusters/ClusterPage/WorkerCpu/workerCpuSampler.test.ts @@ -0,0 +1,382 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +import { CompiledQuery } from "kysely"; + +import { + buildDataflowExportsQuery, + buildDataflowWorkerCpuQuery, + buildWorkerCpuQuery, +} from "~/api/materialize/cluster/workerCpu"; +import { createSubscribeCollection } from "~/api/materialize/subscribeCollection"; +import { WebsocketQueryClient } from "~/api/materialize/WebsocketQueryClient"; + +import { DataflowExport, DataflowWorkerCpu } from "./workerCpuModel"; +import { + createWorkerCpuSampler, + SAMPLING_LIMIT_MS, + StoredObjectRead, + StoredWorkerReading, +} from "./workerCpuSampler"; + +const START_MS = Date.UTC(2026, 9, 16, 12); +const SECOND_NS = 1e9; + +type QueryKind = "workers" | "dataflows" | "exports"; + +interface Gate { + release: () => void; +} + +/** + * Answers the sampler's three queries from mutable counters. A query is + * answered with the counters as they are when it completes, so a test can move + * them while a held query is in flight. + */ +function createFakeReplica() { + const kindBySql = new Map([ + [buildWorkerCpuQuery().compile().sql, "workers"], + [buildDataflowWorkerCpuQuery().compile().sql, "dataflows"], + [buildDataflowExportsQuery().compile().sql, "exports"], + ]); + const gates = new Map>(); + const replica = { + /** Each worker's cumulative dataflow time, indexed by worker id. */ + workerNs: [10 * SECOND_NS, 10 * SECOND_NS], + dataflows: [] as DataflowWorkerCpu[], + exports: [] as DataflowExport[], + queries: [] as QueryKind[], + clients: [] as Array<{ closed: boolean; dropped: boolean }>, + /** Holds the next query of `kind` until the returned gate is released. */ + holdNext: (kind: QueryKind): Gate => { + let release = () => {}; + gates.set( + kind, + new Promise((resolve) => { + release = resolve; + }), + ); + return { release: () => release() }; + }, + /** Every open session fails its later queries, like a socket that closed. */ + dropConnections: () => { + for (const client of replica.clients) client.dropped = true; + }, + createClient: (): WebsocketQueryClient => { + const client = { closed: false, dropped: false }; + replica.clients.push(client); + return { + query: async (query: CompiledQuery): Promise => { + const kind = kindBySql.get(query.sql); + if (!kind) throw new Error(`Unexpected query: ${query.sql}`); + replica.queries.push(kind); + if (client.closed) throw new Error("Client closed"); + if (client.dropped) throw new Error("Connection closed"); + const gate = gates.get(kind); + gates.delete(kind); + await gate; + const rows = + kind === "workers" + ? replica.workerNs.map((elapsedNs, workerId) => ({ + workerId, + elapsedNs, + })) + : kind === "dataflows" + ? replica.dataflows + : replica.exports; + return rows as T[]; + }, + close: () => { + client.closed = true; + }, + }; + }, + }; + return replica; +} + +type FakeReplica = ReturnType; + +let storeCount = 0; +const createStores = () => { + storeCount += 1; + return { + readingsStore: createSubscribeCollection({ + id: `worker-cpu-sampler-test-readings-${storeCount}`, + getKey: (reading) => String(reading.atMs), + }), + objectsStore: createSubscribeCollection({ + id: `worker-cpu-sampler-test-objects-${storeCount}`, + getKey: (read) => read.role, + }), + }; +}; + +type Stores = ReturnType; + +const samplers: Array> = []; + +const createSampler = (replica: FakeReplica, stores: Stores) => { + const sampler = createWorkerCpuSampler({ + createClient: replica.createClient, + ...stores, + now: () => Date.now(), + }); + samplers.push(sampler); + return sampler; +}; + +const startSampler = (replica: FakeReplica, stores = createStores()) => { + const sampler = createSampler(replica, stores); + sampler.start(); + return sampler; +}; + +/** Moves the clock, running due reads and every read's queries. */ +const advance = (ms: number) => vi.advanceTimersByTimeAsync(ms); + +const countQueries = (replica: FakeReplica, kind: QueryKind) => + replica.queries.filter((queried) => queried === kind).length; + +const sinceStart = (readings: Array<{ atMs: number }>) => + readings.map(({ atMs }) => atMs - START_MS); + +describe("createWorkerCpuSampler", () => { + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(START_MS); + }); + + afterEach(() => { + for (const sampler of samplers.splice(0)) sampler.stop(); + }); + + it("reads every worker's counters every 5 seconds", async () => { + const sampler = startSampler(createFakeReplica()); + + await advance(30_000); + + expect(sinceStart(sampler.getState().readings)).toEqual([ + 0, 5_000, 10_000, 15_000, 20_000, 25_000, 30_000, + ]); + }); + + it("reads per-dataflow counters at most every 30 seconds", async () => { + const replica = createFakeReplica(); + const sampler = startSampler(replica); + + await advance(59_000); + + expect(countQueries(replica, "dataflows")).toBe(2); + expect(sampler.getState().latestObjects?.atMs).toBe(START_MS + 30_000); + }); + + it("waits ten times as long as a slow read took before reading again", async () => { + const replica = createFakeReplica(); + const slowRead = replica.holdNext("workers"); + const sampler = startSampler(replica); + await advance(2_000); + slowRead.release(); + await advance(0); + + expect(sampler.getState().intervalMs).toBe(20_000); + await advance(19_999); + expect(sampler.getState().readings).toHaveLength(1); + await advance(1); + expect(sampler.getState().readings).toHaveLength(2); + }); + + it("keeps the 5 second cadence after a read under half a second", async () => { + const replica = createFakeReplica(); + const read = replica.holdNext("workers"); + const sampler = startSampler(replica); + await advance(300); + read.release(); + await advance(0); + + expect(sampler.getState().intervalMs).toBe(5_000); + }); + + it("pauses itself after 10 minutes of reading and closes its session", async () => { + const replica = createFakeReplica(); + const sampler = startSampler(replica); + + await advance(SAMPLING_LIMIT_MS - 1); + expect(sampler.getState().status).toBe("reading"); + await advance(1); + + expect(sampler.getState()).toMatchObject({ + status: "paused", + pausedForLimit: true, + }); + expect(replica.clients.map(({ closed }) => closed)).toEqual([true]); + const readingCount = sampler.getState().readings.length; + await advance(60_000); + expect(sampler.getState().readings).toHaveLength(readingCount); + }); + + it("keeps one read cadence when paused and resumed while a read is in flight", async () => { + const replica = createFakeReplica(); + const sampler = startSampler(replica); + await advance(0); + const inFlight = replica.holdNext("workers"); + await advance(5_000); + + sampler.pause(); + sampler.resume(); + inFlight.release(); + const workerReadsBefore = countQueries(replica, "workers"); + await advance(30_000); + + expect(countQueries(replica, "workers") - workerReadsBefore).toBe(6); + }); + + it("does not mark a read in flight across a pause and resume as after a gap", async () => { + const replica = createFakeReplica(); + const sampler = startSampler(replica); + await advance(0); + const inFlight = replica.holdNext("workers"); + await advance(5_000); + + sampler.pause(); + sampler.resume(); + inFlight.release(); + await advance(0); + + expect(sampler.getState().readings.at(-1)).toMatchObject({ + atMs: START_MS + 5_000, + afterGap: false, + }); + + await advance(5_000); + expect(sampler.getState().readings.at(-1)).toMatchObject({ + atMs: START_MS + 10_000, + afterGap: false, + }); + }); + + it("reads again on a new session when resumed after a read failed", async () => { + const replica = createFakeReplica(); + const sampler = startSampler(replica); + await advance(0); + replica.dropConnections(); + await advance(5_000); + expect(sampler.getState()).toMatchObject({ + status: "error", + error: "Connection closed", + }); + + sampler.resume(); + await advance(0); + + expect(replica.clients).toHaveLength(2); + expect(sampler.getState()).toMatchObject({ + status: "reading", + error: undefined, + }); + expect(sinceStart(sampler.getState().readings)).toEqual([0, 5_000]); + }); + + it("marks the reading after a worker's counter fell as after a reset", async () => { + const replica = createFakeReplica(); + replica.workerNs = [100 * SECOND_NS, 100 * SECOND_NS]; + const sampler = startSampler(replica); + await advance(0); + + replica.workerNs = [40 * SECOND_NS, 101 * SECOND_NS]; + await advance(5_000); + replica.workerNs = [41 * SECOND_NS, 102 * SECOND_NS]; + await advance(5_000); + + expect( + sampler.getState().readings.map(({ afterReset }) => afterReset), + ).toEqual([false, true, false]); + }); + + it("keeps the first per-dataflow read as the baseline while counters grow", async () => { + const replica = createFakeReplica(); + replica.dataflows = [ + { dataflowId: "7", workerId: 0, elapsedNs: 100 * SECOND_NS }, + ]; + const stores = createStores(); + const sampler = startSampler(replica, stores); + await advance(0); + + replica.dataflows = [ + { dataflowId: "7", workerId: 0, elapsedNs: 130 * SECOND_NS }, + ]; + await advance(30_000); + + expect(sampler.getState().firstObjects?.atMs).toBe(START_MS); + expect( + stores.objectsStore.getRows().find(({ role }) => role === "first")?.atMs, + ).toBe(START_MS); + }); + + it("rebases the first per-dataflow read when a dataflow's counter falls", async () => { + const replica = createFakeReplica(); + replica.dataflows = [ + { dataflowId: "7", workerId: 0, elapsedNs: 100 * SECOND_NS }, + ]; + const stores = createStores(); + const sampler = startSampler(replica, stores); + await advance(0); + + const afterRestart = [ + { dataflowId: "7", workerId: 0, elapsedNs: 40 * SECOND_NS }, + ]; + replica.dataflows = afterRestart; + await advance(30_000); + + expect(sampler.getState().firstObjects).toEqual({ + atMs: START_MS + 30_000, + rows: afterRestart, + }); + expect( + stores.objectsStore.getRows().find(({ role }) => role === "first")?.atMs, + ).toBe(START_MS + 30_000); + }); + + it("continues from an earlier sampler's readings and marks its first reading as after a gap", async () => { + const replica = createFakeReplica(); + const stores = createStores(); + const earlier = startSampler(replica, stores); + await advance(10_000); + earlier.stop(); + await advance(60_000); + + const later = createSampler(replica, stores); + expect(sinceStart(later.getState().readings)).toEqual([0, 5_000, 10_000]); + later.start(); + await advance(0); + + expect(later.getState().readings.at(-1)).toMatchObject({ + atMs: START_MS + 70_000, + afterGap: true, + }); + }); + + it("ignores a read that lands after stop and closes its session", async () => { + const replica = createFakeReplica(); + const stores = createStores(); + const sampler = startSampler(replica, stores); + await advance(0); + const late = replica.holdNext("workers"); + await advance(5_000); + const stateAtStop = sampler.getState(); + + sampler.stop(); + late.release(); + await advance(60_000); + + expect(sampler.getState()).toBe(stateAtStop); + expect(sinceStart(stores.readingsStore.getRows())).toEqual([0]); + expect(replica.clients.map(({ closed }) => closed)).toEqual([true]); + }); +}); diff --git a/console/src/platform/clusters/ClusterPage/WorkerCpu/workerCpuSampler.ts b/console/src/platform/clusters/ClusterPage/WorkerCpu/workerCpuSampler.ts new file mode 100644 index 0000000000000..289698590fb20 --- /dev/null +++ b/console/src/platform/clusters/ClusterPage/WorkerCpu/workerCpuSampler.ts @@ -0,0 +1,395 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +import { millisecondsInMinute } from "date-fns/constants"; +import { useAtomValue } from "jotai"; +import React from "react"; + +import { + buildDataflowExportsQuery, + buildDataflowWorkerCpuQuery, + buildWorkerCpuQuery, +} from "~/api/materialize/cluster/workerCpu"; +import { + createScopedSubscribeCollections, + SubscribeCollection, +} from "~/api/materialize/subscribeCollection"; +import { + createWebsocketQueryClient, + WebsocketQueryClient, +} from "~/api/materialize/WebsocketQueryClient"; +import { useCurrentEnvironmentHttpAddress } from "~/store/environments"; +import { syncEngineCacheScopeLoadableAtom } from "~/store/syncEngineCache"; + +import { + DataflowExport, + DataflowWorkerCpu, + hasCounterReset, + hasDataflowCounterReset, + WorkerCpuReading, +} from "./workerCpuModel"; + +export const READ_INTERVAL_MS = 5_000; +const OBJECT_READ_INTERVAL_MS = 30_000; +export const SAMPLING_LIMIT_MS = 10 * millisecondsInMinute; +// The next read waits at least this many times as long as the last one took. +// A replica that answers slowly is usually a busy one, and these reads run on +// it. +const READ_BACKOFF_FACTOR = 10; +const READ_TIMEOUT = "10s"; +/** How long readings stay, so reopening the drawer continues from them. */ +const READINGS_RETENTION_MS = 30 * millisecondsInMinute; + +/** A reading as the collection holds it, in JSON-friendly form. */ +export interface StoredWorkerReading { + atMs: number; + afterGap: boolean; + afterReset: boolean; + elapsedNs: Array<[number, number]>; +} + +export interface StoredObjectRead { + role: "first" | "latest"; + atMs: number; + rows: DataflowWorkerCpu[]; + exports: DataflowExport[]; +} + +// Per tenant scope and replica, in memory only: an organization or region +// switch drops them. +const readingsByReplica = createScopedSubscribeCollections( + { id: "worker-cpu-readings", getKey: (reading) => String(reading.atMs) }, +); +const objectReadsByReplica = createScopedSubscribeCollections( + { + id: "worker-cpu-object-reads", + getKey: (read) => read.role, + }, +); + +const toReading = ({ + atMs, + afterGap, + afterReset, + elapsedNs, +}: StoredWorkerReading): WorkerCpuReading => ({ + atMs, + afterGap, + afterReset, + elapsedNsByWorker: new Map(elapsedNs), +}); + +const toStored = ({ + atMs, + afterGap, + afterReset, + elapsedNsByWorker, +}: WorkerCpuReading): StoredWorkerReading => ({ + atMs, + afterGap: Boolean(afterGap), + afterReset: Boolean(afterReset), + elapsedNs: [...elapsedNsByWorker], +}); + +const applyRows = ( + collection: SubscribeCollection, + data: T[], +) => + collection.applySnapshot({ data, snapshotComplete: true, error: undefined }); + +interface ObjectCpuRead { + atMs: number; + rows: DataflowWorkerCpu[]; +} + +export interface WorkerCpuState { + status: "reading" | "paused" | "error"; + /** Paused by `SAMPLING_LIMIT_MS` rather than by the user. */ + pausedForLimit: boolean; + readings: WorkerCpuReading[]; + /** The per-object read when sampling started, and the latest one. */ + firstObjects?: ObjectCpuRead; + latestObjects?: ObjectCpuRead; + exports: DataflowExport[]; + /** How long the next read waits. */ + intervalMs: number; + error?: string; +} + +const INITIAL_WORKER_CPU_STATE: WorkerCpuState = { + status: "reading", + pausedForLimit: false, + readings: [], + exports: [], + intervalMs: READ_INTERVAL_MS, +}; + +interface WorkerCpuSampler { + getState: () => WorkerCpuState; + subscribe: (listener: () => void) => () => void; + start: () => void; + pause: () => void; + resume: () => void; + stop: () => void; +} + +/** + * Reads a replica's per-worker CPU counters on a timer, and its per-object + * counters on a slower one, until paused or stopped. It starts from, and keeps + * writing to, the readings `readingsStore` and `objectsStore` hold from an + * earlier session. It holds a session from `createClient` only while reading. + */ +export function createWorkerCpuSampler({ + createClient, + readingsStore, + objectsStore, + now = Date.now, +}: { + createClient: () => WebsocketQueryClient; + readingsStore: SubscribeCollection; + objectsStore: SubscribeCollection; + now?: () => number; +}): WorkerCpuSampler { + const isRetained = (atMs: number) => now() - atMs <= READINGS_RETENTION_MS; + const priorReadings = readingsStore + .getRows() + .filter(({ atMs }) => isRetained(atMs)) + .sort((a, b) => a.atMs - b.atMs) + .map(toReading); + const priorObjects = objectsStore + .getRows() + .filter(({ atMs }) => isRetained(atMs)); + const priorFirst = priorObjects.find(({ role }) => role === "first"); + const priorLatest = priorObjects.find(({ role }) => role === "latest"); + let state: WorkerCpuState = { + ...INITIAL_WORKER_CPU_STATE, + readings: priorReadings, + firstObjects: priorFirst && { + atMs: priorFirst.atMs, + rows: priorFirst.rows, + }, + latestObjects: priorLatest && { + atMs: priorLatest.atMs, + rows: priorLatest.rows, + }, + exports: priorLatest?.exports ?? [], + }; + const listeners = new Set<() => void>(); + let client: WebsocketQueryClient | undefined; + let timer: ReturnType | undefined; + let stopped = false; + // At most one read runs, so resuming during a read doesn't start a second + // timer chain. + let isReading = false; + let limitStartMs = 0; + let afterGap = priorReadings.length > 0; + + const setState = (update: Partial) => { + state = { ...state, ...update }; + for (const listener of listeners) listener(); + }; + + // A closed session can't recover, so each stretch of reading opens its own. + const releaseClient = () => { + client?.close(); + client = undefined; + }; + + const readObjects = async ( + session: WebsocketQueryClient, + ): Promise> => { + const [rows, exports] = await Promise.all([ + session.query(buildDataflowWorkerCpuQuery().compile()), + session.query(buildDataflowExportsQuery().compile()), + ]); + if (stopped) return {}; + const read = { atMs: now(), rows }; + const first = + state.firstObjects && + !hasDataflowCounterReset(state.firstObjects.rows, rows) + ? state.firstObjects + : read; + applyRows(objectsStore, [ + { role: "first", ...first, exports }, + { role: "latest", ...read, exports }, + ]); + return { firstObjects: first, latestObjects: read, exports }; + }; + + const addReading = (reading: WorkerCpuReading) => { + const readings = [...state.readings, reading].filter(({ atMs }) => + isRetained(atMs), + ); + applyRows(readingsStore, readings.map(toStored)); + return readings; + }; + + const tick = async () => { + timer = undefined; + if (stopped || isReading || state.status !== "reading") return; + isReading = true; + const isAfterGap = afterGap; + afterGap = false; + // Applied with the worker reading, so one tick renders once. + let objects: Partial = {}; + try { + const session = (client ??= createClient()); + const startedMs = now(); + if ( + !state.latestObjects || + startedMs - state.latestObjects.atMs >= OBJECT_READ_INTERVAL_MS + ) { + objects = await readObjects(session); + } + if (stopped) return; + const workers = await session.query(buildWorkerCpuQuery().compile()); + if (stopped) return; + const atMs = now(); + const previous = state.readings.at(-1); + const elapsedNsByWorker = new Map( + workers.map(({ workerId, elapsedNs }) => [workerId, elapsedNs]), + ); + const reading: WorkerCpuReading = { + atMs, + elapsedNsByWorker, + afterGap: isAfterGap, + afterReset: + previous !== undefined && + hasCounterReset(previous.elapsedNsByWorker, elapsedNsByWorker), + }; + const intervalMs = Math.max( + READ_INTERVAL_MS, + READ_BACKOFF_FACTOR * (atMs - startedMs), + ); + const readings = addReading(reading); + if (state.status !== "reading") { + // Paused while this read ran. + setState({ ...objects, readings, intervalMs }); + releaseClient(); + return; + } + if (atMs - limitStartMs >= SAMPLING_LIMIT_MS) { + afterGap = true; + setState({ + ...objects, + readings, + intervalMs, + status: "paused", + pausedForLimit: true, + }); + releaseClient(); + return; + } + setState({ ...objects, readings, intervalMs }); + timer = setTimeout(() => void tick(), intervalMs); + } catch (error) { + if (stopped) return; + afterGap = true; + releaseClient(); + setState({ + ...objects, + status: "error", + error: error instanceof Error ? error.message : String(error), + }); + } finally { + isReading = false; + } + }; + + return { + getState: () => state, + subscribe: (listener) => { + listeners.add(listener); + return () => listeners.delete(listener); + }, + start: () => { + limitStartMs = now(); + void tick(); + }, + pause: () => { + if (state.status !== "reading") return; + clearTimeout(timer); + timer = undefined; + afterGap = true; + setState({ status: "paused", pausedForLimit: false }); + // A read in flight releases the session when it lands. + if (!isReading) releaseClient(); + }, + resume: () => { + if (stopped || state.status === "reading") return; + // A read still in flight carries on the chain, so nothing was missed. + if (isReading) afterGap = false; + limitStartMs = now(); + setState({ status: "reading", pausedForLimit: false, error: undefined }); + void tick(); + }, + stop: () => { + stopped = true; + clearTimeout(timer); + releaseClient(); + }, + }; +} + +const noopSubscribe = () => () => {}; + +/** + * Samples one replica's worker CPU while the calling component is mounted, + * continuing from the readings of an earlier session in the last + * `READINGS_RETENTION_MS`. + */ +export function useWorkerCpuSampler({ + clusterName, + replicaId, + replicaName, +}: { + clusterName: string; + replicaId: string; + replicaName: string; +}) { + const httpAddress = useCurrentEnvironmentHttpAddress(); + const scope = useAtomValue(syncEngineCacheScopeLoadableAtom); + const scopeKey = + scope.state === "hasData" && scope.data ? scope.data : undefined; + const [sampler, setSampler] = React.useState(); + React.useEffect(() => { + if (!httpAddress || !scopeKey) return; + const next = createWorkerCpuSampler({ + createClient: () => + createWebsocketQueryClient({ + httpAddress, + sessionVariables: { + cluster: clusterName, + cluster_replica: replicaName, + statement_timeout: READ_TIMEOUT, + }, + }), + readingsStore: readingsByReplica.get(scopeKey, replicaId), + objectsStore: objectReadsByReplica.get(scopeKey, replicaId), + }); + setSampler(next); + next.start(); + return () => { + next.stop(); + setSampler(undefined); + }; + }, [httpAddress, scopeKey, clusterName, replicaId, replicaName]); + const state = React.useSyncExternalStore( + sampler?.subscribe ?? noopSubscribe, + () => sampler?.getState() ?? INITIAL_WORKER_CPU_STATE, + ); + return { + state: + scope.state === "hasError" + ? { ...state, status: "error" as const, error: "Couldn't load." } + : state, + pause: sampler?.pause, + resume: sampler?.resume, + }; +} diff --git a/console/src/platform/clusters/ClusterPage/WorkerCpu/workerCpuStyles.ts b/console/src/platform/clusters/ClusterPage/WorkerCpu/workerCpuStyles.ts new file mode 100644 index 0000000000000..60ffbb374d8d1 --- /dev/null +++ b/console/src/platform/clusters/ClusterPage/WorkerCpu/workerCpuStyles.ts @@ -0,0 +1,48 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +import { useTheme } from "@chakra-ui/react"; +import { color } from "d3"; +import React from "react"; + +import { MaterializeTheme } from "~/theme"; + +import { EVEN_RATIO, SKEW_RATIO } from "./workerCpuModel"; + +const withOpacity = (value: string, opacity: number) => { + const parsed = color(value); + if (!parsed) return value; + parsed.opacity = opacity; + return parsed.formatRgb(); +}; + +/** Worker bars stop at this many times an even share. */ +export const BAR_CEILING_RATIO = 3; + +/** + * A worker bar's color by its time on an object against an even share: + * neutral up to about even, orange once uneven, red at the bars' ceiling. + */ +export function useWorkerBarColor() { + const { colors } = useTheme(); + return React.useMemo(() => { + const uneven = withOpacity(colors.accent.orange, 0.5); + return (ratio: number) => + ratio >= BAR_CEILING_RATIO + ? colors.accent.red + : ratio >= SKEW_RATIO + ? colors.accent.orange + : ratio >= EVEN_RATIO + ? uneven + : colors.foreground.tertiary; + }, [colors]); +} + +/** The shade between two lines, such as the busiest and average worker. */ +export const gapFill = (base: string) => withOpacity(base, 0.16); diff --git a/console/src/platform/clusters/ClusterPage/useObjectLink.ts b/console/src/platform/clusters/ClusterPage/useObjectLink.ts new file mode 100644 index 0000000000000..69c9de89a63b0 --- /dev/null +++ b/console/src/platform/clusters/ClusterPage/useObjectLink.ts @@ -0,0 +1,46 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +import { useFlags } from "~/hooks/useFlags"; +import { + absoluteMaintainedObjectPath, + useBuildWorkflowGraphPath, +} from "~/platform/routeHelpers"; +import { useRegionSlug } from "~/store/environments"; + +export interface LinkableObject { + id: string | null; + name: string | null; + databaseName: string | null; + schemaName: string | null; + type: string | null; +} + +/** + * Builds an object's link: its Maintained Objects drawer, or its workflow + * graph while that page is off. Undefined for an object missing a catalog + * field, such as a dropped object's dataflow. + */ +export function useObjectLink() { + // The maintained objects page only has routes while its flag is on. + const hasMaintainedObjectsPage = Boolean( + useFlags()["maintained-objects-ui-50"], + ); + const regionSlug = useRegionSlug(); + const workflowGraphPath = useBuildWorkflowGraphPath(); + return ({ id, name, databaseName, schemaName, type }: LinkableObject) => { + if (!id || !name || !schemaName || !type) return undefined; + return hasMaintainedObjectsPage + ? absoluteMaintainedObjectPath(regionSlug, id) + : workflowGraphPath({ + databaseObject: { id, name, databaseName, schemaName }, + type, + }); + }; +}