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, + }); + }; +}