From 2ca09090931260537b1b2bbe61cdf7af79e70085 Mon Sep 17 00:00:00 2001 From: Qindeel Ishtiaq Date: Tue, 6 Oct 2026 17:45:35 -0400 Subject: [PATCH] console: keep the redesigned cluster page's live data in TanStack DB collections Fifth in the cluster details redesign stack. The resource usage charts and the memory by object card read their live data from TanStack DB collections instead of per-hook subscribe state. Charts: each cluster gets a collection fed by the 3h or 24h utilization SUBSCRIBE, and each subscribe covers its view's full retention. Switching windows within a tier reads the collection instead of resubscribing. Leaving the Metrics tab closes the subscribe; coming back paints the held rows at once and marks the card as refreshing until the new snapshot lands. Held rows that have all left the window show as loading, not as a window with no usage. Windows past 24h still poll, through a poll-only hook that mounts no subscribe hooks and so opens no idle WebSocket. The page no longer uses useReplicaUtilizationHistory, so its includeOfflineEvents option goes away. Memory by object: each replica gets a collection fed by a SUBSCRIBE to mz_object_arrangement_sizes, an index lookup on replica_id. A live query joins it to the all objects collection for names and keeps the ten largest. The card waits for that collection, so objects never show as raw ids, and a replica without a heap limit shows sizes without shares. Sizes update live, and the card no longer polls sizes every 60s or object names every 5 minutes. Environments before 26.35 keep the polled card. Collections are keyed by environment and id, since cluster and replica ids repeat across environments, through one createScopedSubscribeCollections helper that drops an earlier environment's collections when the scope changes. They are not persisted. A subscribe collection's status gains isRefreshing for the held-rows case, and ending a collection's session marks its rows as held, so a return before the next subscribe starts shows loading rather than an empty window. --- console/doc/design/20260902_sync_engine.md | 47 +- .../cluster/largestClusterReplica.test.sql.ts | 14 - .../cluster/largestClusterReplica.ts | 90 ++- .../cluster/largestMaintainedQueries.ts | 26 + .../cluster/replicaHeapLimits.test.sql.ts | 90 +++ .../materialize/cluster/replicaHeapLimits.ts | 7 +- .../cluster/replicaUtilizationBinning.ts | 14 + .../materialize/subscribeCollection.test.ts | 49 ++ .../api/materialize/subscribeCollection.ts | 84 ++- .../src/api/materialize/subscribeSession.ts | 8 + .../useScopedSubscribeCollection.ts | 58 ++ .../src/api/materialize/useSubscribe.test.tsx | 28 + console/src/api/materialize/useSubscribe.ts | 14 +- .../clusters/ClusterPage/ClusterMetrics.tsx | 6 +- .../ClusterPage/MemoryByObject.test.tsx | 558 ++++++++++++++---- .../clusters/ClusterPage/MemoryByObject.tsx | 323 ++++++++-- .../ResourceUsage/ResourceUsage.test.tsx | 55 ++ .../useLiveClusterUtilization.test.ts | 337 +++++++++++ .../useLiveClusterUtilization.ts | 195 ++++++ .../ResourceUsage/useResourceUsageData.ts | 35 +- .../ClusterPage/memoryByObjectModel.ts | 33 ++ .../ClusterPage/useLiveLargestObjects.test.ts | 97 +++ .../ClusterPage/useLiveLargestObjects.ts | 107 ++++ .../clusters/LargestMaintainedQueries.tsx | 68 ++- console/src/platform/clusters/queries.test.ts | 90 +++ console/src/platform/clusters/queries.ts | 164 +++-- console/src/test/mockSubscribeSocket.ts | 14 +- 27 files changed, 2267 insertions(+), 344 deletions(-) create mode 100644 console/src/api/materialize/cluster/replicaHeapLimits.test.sql.ts create mode 100644 console/src/api/materialize/useScopedSubscribeCollection.ts create mode 100644 console/src/platform/clusters/ClusterPage/ResourceUsage/useLiveClusterUtilization.test.ts create mode 100644 console/src/platform/clusters/ClusterPage/ResourceUsage/useLiveClusterUtilization.ts create mode 100644 console/src/platform/clusters/ClusterPage/useLiveLargestObjects.test.ts create mode 100644 console/src/platform/clusters/ClusterPage/useLiveLargestObjects.ts diff --git a/console/doc/design/20260902_sync_engine.md b/console/doc/design/20260902_sync_engine.md index 88213bba869b0..3188fe703be68 100644 --- a/console/doc/design/20260902_sync_engine.md +++ b/console/doc/design/20260902_sync_engine.md @@ -69,7 +69,7 @@ count stays the same, and a collection can even be fed from an existing atom | `SubscribeManager` | `api/materialize/SubscribeManager.ts` | The SUBSCRIBE protocol: buffers rows per timestamp, folds the upsert stream into the current row set, exposes snapshots via `onChange`/`getSnapshot`. Holds its last snapshot through a resubscribe so the UI does not flash empty. | | `WebsocketConnectionManager` | `api/materialize/WebsocketConnectionManager.ts` | Connection lifecycle: subscribes to environment health and the current region on the jotai store, connects when healthy, reconnects with backoff, reconnects when the region's address changes, and tears down a socket left pointing at a departed region. | | Subscribe sessions | `api/materialize/subscribeSession.ts` | Composition: one constructor wires the manager, the connection manager, the sink, the region reset, and cache hydration, in one defined order. | -| Hooks | `api/materialize/useSubscribe.ts` | Lifecycle adapters: a single effect creates the session on mount and destroys it on unmount. | +| Hooks | `api/materialize/useSubscribe.ts`, `useScopedSubscribeCollection.ts` | Lifecycle adapters: a single effect creates the session on mount and destroys it on unmount. | | Collection bridge | `api/materialize/subscribeCollection.ts` | `createSubscribeCollection`: turns manager snapshots into incremental insert/update/delete writes on a TanStack DB collection, owns the scoped localStorage cache, and mirrors status (error, snapshotComplete) into a companion atom. | | Cache scope | `store/syncEngineCache.ts` | Derives the cache scope `organizationId\|regionId` as an atom, so hydration can be driven from the store. | @@ -176,11 +176,13 @@ What the engine is not: sinks are transport-agnostic, which is the seam a shared upstream (cross-tab or cross-client dedup) can plug into without touching surfaces, but the engine alone does not provide that. -- **Not for polled telemetry.** Expensive introspection queries - (utilization history, arrangement sizes) are react-query territory, with - polling intervals and its persistence tooling. The engine serves - catalog-shaped data that a SUBSCRIBE can maintain. -- **Not for page-scoped, request-driven subscribes.** Those use +- **Not for polled telemetry.** Introspection data no SUBSCRIBE maintains, + such as utilization beyond 24h or on environments that predate the indexed + views, is react-query territory, with polling intervals and its persistence + tooling. The engine serves data that a SUBSCRIBE can maintain, including + per-key page data (see per-key scoped collections below). +- **Not for one-off, request-driven subscribes.** Subscribes whose rows + nothing reads after the component unmounts use `useSubscribe`/`useSubscribeManager`, which tie a manager to a component and expose state via `useSyncExternalStore`. @@ -194,9 +196,9 @@ queries over the objects collection (fed from the app-wide `allObjects` atom) and the namespaces collection (its own subscribe), and renders from cache on revisit before the live snapshot lands. -Declare configuration once at module scope. The hooks require a referentially -stable options object, and a new object identity restarts the session, which -makes accidental instability visible instead of masked. +Declare configuration once at module scope, or memoize it. The hooks require +a referentially stable options object, and a new object identity restarts the +session, which makes accidental instability visible instead of masked. Atom-backed data set: @@ -278,6 +280,33 @@ component keyed by a path pattern survives param-only navigation, including region switches, which is why region handling lives in the session rather than in mount/unmount. +## Per-key scoped collections + +Page data keyed by an id, such as a cluster's utilization or a replica's +arrangement sizes, lives in `createScopedSubscribeCollections`: one in-memory +collection per tenant scope and key, dropped when the scope changes and never +persisted. `useScopedSubscribeCollection` resolves the scope, picks the key's +collection and runs the subscribe into it while mounted. The rows outlive the +component, so a return to the page serves them at once, with `isRefreshing` +set until the new subscribe's snapshot lands, and a view derived from them, +such as a chart window, can change without resubscribing. + +```ts +const samplesByCluster = createScopedSubscribeCollections({ + id: "samples", + getKey: (sample) => sample.id, +}); + +export function useLiveSamples(clusterId: string) { + // A new builder restarts the subscribe, so memoize it. + const subscribe = React.useMemo( + () => buildSubscribeQuery(buildSamplesQuery(clusterId), { upsertKey: "id" }), + [clusterId], + ); + return useScopedSubscribeCollection(samplesByCluster, clusterId, subscribe); +} +``` + ## Testing Sessions and the collection bridge are plain objects, so their behavior is diff --git a/console/src/api/materialize/cluster/largestClusterReplica.test.sql.ts b/console/src/api/materialize/cluster/largestClusterReplica.test.sql.ts index ddf182a0b88fe..750379f4938ea 100644 --- a/console/src/api/materialize/cluster/largestClusterReplica.test.sql.ts +++ b/console/src/api/materialize/cluster/largestClusterReplica.test.sql.ts @@ -89,20 +89,6 @@ describe("buildLargestClusterReplicaQuery", () => { expect(result.size).toBe("scale=1,workers=1,mem=8GiB"); expect(result.heapLimit).toBe("8589934592"); - const { - rows: [named], - } = await executeSqlHttp( - buildLargestClusterReplicaQuery(cluster.id, "r1").compile(), - { - sessionVariables: { - search_path: mockedSearchPath, - cluster: QUICKSTART_CLUSTER, - }, - }, - ); - expect(named.name).toBe("r1"); - expect(named.heapLimit).toBe("4294967296"); - // Object sizes cover every process of a replica, so its heap limit does // too. Four 4GiB processes outrank one 8GiB process. await client.query(` diff --git a/console/src/api/materialize/cluster/largestClusterReplica.ts b/console/src/api/materialize/cluster/largestClusterReplica.ts index 17e398fd9cd22..b47c8b80d15d1 100644 --- a/console/src/api/materialize/cluster/largestClusterReplica.ts +++ b/console/src/api/materialize/cluster/largestClusterReplica.ts @@ -17,57 +17,52 @@ import { buildClusterReplicaHeapMetricsTable } from "~/api/materialize/expressio // process of the replica. const replicaHeapLimit = sql`(crhm.heap_limit * COALESCE(crs.processes, 1))`; -export function buildLargestClusterReplicaQuery( - clusterId: string, - replicaName?: string, -) { - const query = queryBuilder - .selectFrom("mz_cluster_replicas as cr") - .leftJoin( - buildClusterReplicaHeapMetricsTable().as("crhm"), - "crhm.replica_id", - "cr.id", - ) - // Prefer a fully-hydrated replica, whose reported sizes are stable. - // Joining mz_compute_hydration_times before aggregating lets its - // replica_id index bound the scan to this cluster's replicas, unlike an - // aggregation over the whole environment's mz_hydration_statuses. - .leftJoin("mz_compute_hydration_times as ht", (join) => - join - .onRef("ht.replica_id", "=", "cr.id") - // System introspection indexes may never report hydrated. - .on("ht.object_id", "not like", "s%"), - ) - .leftJoin("mz_cluster_replica_sizes as crs", "crs.size", "cr.size") - .select([ - "cr.name", - "cr.size", - sql`${replicaHeapLimit}::text`.as("heapLimit"), - ]) - .where("cr.cluster_id", "=", clusterId) - .groupBy([ - "cr.id", - "cr.name", - "cr.size", - "crhm.heap_limit", - "crs.processes", - ]) - // Prefer a replica whose compute objects are all hydrated, so its - // reported sizes are stable. time_ns is NULL until an object hydrates. A - // replica with no hydration rows, such as one just added, has bool_and - // over an all-NULL row and so sorts last. Then order by heap limit. - .orderBy(sql`bool_and(ht.time_ns IS NOT NULL) DESC`) - .orderBy(sql`${replicaHeapLimit} DESC NULLS LAST`) - .limit(1); - return replicaName === undefined - ? query - : query.where("cr.name", "=", replicaName); +export function buildLargestClusterReplicaQuery(clusterId: string) { + return ( + queryBuilder + .selectFrom("mz_cluster_replicas as cr") + .leftJoin( + buildClusterReplicaHeapMetricsTable().as("crhm"), + "crhm.replica_id", + "cr.id", + ) + // Prefer a fully-hydrated replica, whose reported sizes are stable. + // Joining mz_compute_hydration_times before aggregating lets its + // replica_id index bound the scan to this cluster's replicas, unlike an + // aggregation over the whole environment's mz_hydration_statuses. + .leftJoin("mz_compute_hydration_times as ht", (join) => + join + .onRef("ht.replica_id", "=", "cr.id") + // System introspection indexes may never report hydrated. + .on("ht.object_id", "not like", "s%"), + ) + .leftJoin("mz_cluster_replica_sizes as crs", "crs.size", "cr.size") + .select([ + "cr.id", + "cr.name", + "cr.size", + sql`${replicaHeapLimit}::text`.as("heapLimit"), + ]) + .where("cr.cluster_id", "=", clusterId) + .groupBy([ + "cr.id", + "cr.name", + "cr.size", + "crhm.heap_limit", + "crs.processes", + ]) + // Prefer a replica whose compute objects are all hydrated, so its + // reported sizes are stable. time_ns is NULL until an object hydrates. A + // replica with no hydration rows, such as one just added, has bool_and + // over an all-NULL row and so sorts last. Then order by heap limit. + .orderBy(sql`bool_and(ht.time_ns IS NOT NULL) DESC`) + .orderBy(sql`${replicaHeapLimit} DESC NULLS LAST`) + .limit(1) + ); } export type LargestClusterReplicaParams = { clusterId: string; - /** Reads this replica, with the same heap limit, instead of the largest. */ - replicaName?: string; }; /** @@ -84,7 +79,6 @@ export async function fetchLargestClusterReplica({ }) { const compiledQuery = buildLargestClusterReplicaQuery( params.clusterId, - params.replicaName, ).compile(); return executeSqlV2({ queries: compiledQuery, diff --git a/console/src/api/materialize/cluster/largestMaintainedQueries.ts b/console/src/api/materialize/cluster/largestMaintainedQueries.ts index 81d2ae3cd0db8..7af2fff0661c6 100644 --- a/console/src/api/materialize/cluster/largestMaintainedQueries.ts +++ b/console/src/api/materialize/cluster/largestMaintainedQueries.ts @@ -15,6 +15,7 @@ import { executeSqlV2, queryBuilder, } from "~/api/materialize"; +import { buildSubscribeQuery } from "~/api/materialize/buildSubscribeQuery"; export type LargestMaintainedQueriesParams = { replicaHeapLimit: number | null; @@ -164,6 +165,31 @@ export function buildLargestMaintainedObjectSizesQuery({ ); } +/** One object's arrangement size on a replica, in bytes. */ +export interface ReplicaObjectSize { + objectId: string; + size: number | null; +} + +/** + * Live arrangement sizes of every object on a replica. Filtering on the + * literal replica id makes this a lookup on `mz_object_arrangement_sizes_ind`. + */ +export function buildReplicaObjectSizesSubscribe(replicaId: string) { + return buildSubscribeQuery( + queryBuilder + .selectFrom("mz_object_arrangement_sizes") + .select([ + "object_id as objectId", + // NOTE: The cast is required. An int8 decodes to a BigInt, which the + // collection bridge's JSON.stringify row signature throws on. + sql`size::float8`.as("size"), + ]) + .where("replica_id", "=", replicaId), + { upsertKey: "objectId" }, + ); +} + export async function fetchLargestMaintainedObjectSizes({ params, queryKey, diff --git a/console/src/api/materialize/cluster/replicaHeapLimits.test.sql.ts b/console/src/api/materialize/cluster/replicaHeapLimits.test.sql.ts new file mode 100644 index 0000000000000..0e3da82f21a6c --- /dev/null +++ b/console/src/api/materialize/cluster/replicaHeapLimits.test.sql.ts @@ -0,0 +1,90 @@ +// 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 { CATALOG_SERVER_CLUSTER, SEARCH_PATH } from "~/api/materialize"; +import { + executeSqlHttp, + getMaterializeClient, + QUICKSTART_CLUSTER, +} from "~/test/sql/materializeSqlClient"; +import { testdrive } from "~/test/sql/mzcompose"; + +import { buildReplicaHeapLimitsQuery } from "./replicaHeapLimits"; + +const MOCKED_SEARCH_PATH = "internal_test, " + SEARCH_PATH; + +describe("buildReplicaHeapLimitsQuery", () => { + it("reads each process's heap limit and use for the given replicas", async () => { + await testdrive(` + > CREATE SCHEMA internal_test; + > SET schema = internal_test; + > CREATE TABLE mz_cluster_replica_metrics ( + replica_id text NOT NULL, + process_id uint8 NOT NULL, + cpu_nano_cores uint8, + memory_bytes uint8, + disk_bytes uint8, + heap_bytes uint8, + heap_limit uint8 + ); + > INSERT INTO mz_cluster_replica_metrics VALUES + ('u1', 0, 100, 1000, 0, 400, 2000), + ('u1', 1, NULL, NULL, NULL, NULL, NULL), + ('u2', 0, 100, 1000, 0, 500, 2000); + `); + + const { rows } = await executeSqlHttp( + buildReplicaHeapLimitsQuery({ replicaIds: ["u1"] }).compile(), + { + sessionVariables: { + search_path: MOCKED_SEARCH_PATH, + cluster: QUICKSTART_CLUSTER, + }, + }, + ); + + // A process whose metrics couldn't be read keeps its row, with nulls. + expect( + [...rows].sort((a, b) => Number(a.processId) - Number(b.processId)), + ).toEqual([ + { + replicaId: "u1", + processId: BigInt(0), + heapLimit: BigInt(2000), + memoryBytes: BigInt(1000), + heapBytes: BigInt(400), + }, + { + replicaId: "u1", + processId: BigInt(1), + heapLimit: null, + memoryBytes: null, + heapBytes: null, + }, + ]); + }); + + it("is a fast-path index lookup on the catalog server", async () => { + const client = await getMaterializeClient(); + await client.query(`SET cluster = ${CATALOG_SERVER_CLUSTER}`); + const { sql, parameters } = buildReplicaHeapLimitsQuery({ + replicaIds: ["u1", "u2"], + }).compile(); + + const { + rows: [plan], + } = await client.query(`EXPLAIN ${sql}`, [...parameters]); + + const explained = String(Object.values(plan)[0]); + expect(explained).toContain("Explained Query (fast path)"); + expect(explained).toContain( + "Index Lookup on mz_internal.mz_cluster_replica_metrics", + ); + }); +}); diff --git a/console/src/api/materialize/cluster/replicaHeapLimits.ts b/console/src/api/materialize/cluster/replicaHeapLimits.ts index 0ead185194e3d..44d4de554891e 100644 --- a/console/src/api/materialize/cluster/replicaHeapLimits.ts +++ b/console/src/api/materialize/cluster/replicaHeapLimits.ts @@ -12,9 +12,9 @@ import { QueryKey } from "@tanstack/react-query"; import { executeSqlV2, queryBuilder } from "~/api/materialize"; /** - * Each process's heap limit (RAM plus swap) for the given replicas, from their - * latest metrics. The literal id list makes this a lookup on - * `mz_cluster_replica_metrics_ind`. + * Each process's heap limit (RAM plus swap) and heap use for the given + * replicas, from their latest metrics. The literal id list makes this a lookup + * on `mz_cluster_replica_metrics_ind`. * * NOTE: callers fold the per-process rows, because a GROUP BY here turns the * fast-path lookup into a dataflow. @@ -31,6 +31,7 @@ export function buildReplicaHeapLimitsQuery({ "process_id as processId", "heap_limit as heapLimit", "memory_bytes as memoryBytes", + "heap_bytes as heapBytes", ]) .where("replica_id", "in", replicaIds); } diff --git a/console/src/api/materialize/cluster/replicaUtilizationBinning.ts b/console/src/api/materialize/cluster/replicaUtilizationBinning.ts index 18332809610de..c373852aa8c5a 100644 --- a/console/src/api/materialize/cluster/replicaUtilizationBinning.ts +++ b/console/src/api/materialize/cluster/replicaUtilizationBinning.ts @@ -477,3 +477,17 @@ export const hasUtilizationReading = (point: DataPoint) => export const hasNoUtilizationMetrics = (graphData: ReplicaData[]) => graphData.length > 0 && graphData.every(({ data }) => !data.some(hasUtilizationReading)); + +/** + * Server-binned rows inside a window that starts at `startDate`, in time order. + * Held rows from a wider window would otherwise stretch the chart past the + * selected range, and ENVELOPE UPSERT yields them unordered. + */ +export function clipBinnedRowsToWindow( + rows: T[], + startDate: Date, +): T[] { + return rows + .filter((row) => row.bucketStart.getTime() >= startDate.getTime()) + .sort((a, b) => a.bucketStart.getTime() - b.bucketStart.getTime()); +} diff --git a/console/src/api/materialize/subscribeCollection.test.ts b/console/src/api/materialize/subscribeCollection.test.ts index 0372c7894146d..d5f24c160140a 100644 --- a/console/src/api/materialize/subscribeCollection.test.ts +++ b/console/src/api/materialize/subscribeCollection.test.ts @@ -10,6 +10,7 @@ import { getStore } from "~/jotai"; import { + createScopedSubscribeCollections, createSubscribeCollection, syncEngineCacheKey, } from "./subscribeCollection"; @@ -257,6 +258,39 @@ describe("createSubscribeCollection", () => { expect(getStore().get(statusAtom).snapshotComplete).toBe(false); }); + it("marks held rows as refreshing until a new subscribe's snapshot lands", () => { + const { applySnapshot, statusAtom } = freshCollection(); + applySnapshot(state([{ id: "1", name: "a" }])); + expect(getStore().get(statusAtom).isRefreshing).toBe(false); + + // A new subscribe opens with an empty pre-snapshot. + applySnapshot(state([], false)); + expect(getStore().get(statusAtom)).toMatchObject({ + snapshotComplete: true, + isRefreshing: true, + }); + + applySnapshot(state([{ id: "1", name: "b" }])); + expect(getStore().get(statusAtom).isRefreshing).toBe(false); + }); + + it("marks the rows a reopened socket holds as refreshing", async () => { + const { collection, applySnapshot, statusAtom } = freshCollection(); + applySnapshot(state([{ id: "1", name: "a" }])); + + // A reopened socket re-emits its held rows until the new snapshot lands. + applySnapshot({ ...state([{ id: "1", name: "a" }]), resubscribing: true }); + await flush(); + expect(collection.get("1")?.name).toBe("a"); + expect(getStore().get(statusAtom)).toMatchObject({ + snapshotComplete: true, + isRefreshing: true, + }); + + applySnapshot(state([{ id: "1", name: "b" }])); + expect(getStore().get(statusAtom).isRefreshing).toBe(false); + }); + it("surfaces error and snapshotComplete on the status atom", () => { const { applySnapshot, statusAtom } = freshCollection(); @@ -302,3 +336,18 @@ describe("createSubscribeCollection", () => { } }); }); + +describe("createScopedSubscribeCollections", () => { + it("keeps one collection per scope and key, dropping other scopes' on a scope change", () => { + const collections = createScopedSubscribeCollections({ + id: "scoped-test", + getKey: (row) => row.id, + }); + const first = collections.get("org|regionA", "u1"); + expect(collections.get("org|regionA", "u1")).toBe(first); + expect(collections.get("org|regionA", "u2")).not.toBe(first); + + expect(collections.get("org|regionB", "u1")).not.toBe(first); + expect(collections.get("org|regionA", "u1")).not.toBe(first); + }); +}); diff --git a/console/src/api/materialize/subscribeCollection.ts b/console/src/api/materialize/subscribeCollection.ts index c8c79691cfe24..a54816b99077c 100644 --- a/console/src/api/materialize/subscribeCollection.ts +++ b/console/src/api/materialize/subscribeCollection.ts @@ -57,8 +57,21 @@ function writePersistedRows(key: string, rows: T[]) { export interface SubscribeCollectionStatus { error: SubscribeError | undefined; snapshotComplete: boolean; + /** Rows from an earlier snapshot are served, and no running subscribe has + * delivered a newer one yet. */ + isRefreshing: boolean; } +const IDLE_STATUS: SubscribeCollectionStatus = { + error: undefined, + snapshotComplete: false, + isRefreshing: false, +}; + +/** The status of a collection that nothing has fed, for a consumer whose + * collection is not known yet. */ +export const idleCollectionStatusAtom = atom(() => IDLE_STATUS); + export interface SubscribeCollection { collection: Collection; statusAtom: PrimitiveAtom; @@ -103,10 +116,7 @@ export function createSubscribeCollection(options: { // `hydrate`, then kept current by `applySnapshot`. let desired = new Map(); - const statusAtom = atom({ - error: undefined, - snapshotComplete: false, - }); + const statusAtom = atom(IDLE_STATUS); // Captured when the collection starts syncing; cleared when it pauses. type WriteApi = Parameters< @@ -214,7 +224,8 @@ export function createSubscribeCollection(options: { const current = store.get(statusAtom); if ( current.error !== next.error || - current.snapshotComplete !== next.snapshotComplete + current.snapshotComplete !== next.snapshotComplete || + current.isRefreshing !== next.isRefreshing ) { store.set(statusAtom, next); } @@ -227,7 +238,11 @@ export function createSubscribeCollection(options: { const isEmptyPreload = !state.snapshotComplete && state.data.length === 0 && !state.error; if (isEmptyPreload) { - writeStatus({ error: undefined, snapshotComplete: lastSnapshotComplete }); + writeStatus({ + error: undefined, + snapshotComplete: lastSnapshotComplete, + isRefreshing: lastSnapshotComplete, + }); return; } @@ -240,6 +255,7 @@ export function createSubscribeCollection(options: { writeStatus({ error: state.error, snapshotComplete: state.snapshotComplete, + isRefreshing: Boolean(state.resubscribing), }); }; @@ -296,7 +312,12 @@ export function createSubscribeCollection(options: { // On a scope change this also flushes the emptied set, so the collection // stops serving the previous scope's rows while the new snapshot loads. materialize(); - writeStatus({ error: undefined, snapshotComplete: lastSnapshotComplete }); + // Cached rows stand in until the live snapshot arrives. + writeStatus({ + error: undefined, + snapshotComplete: lastSnapshotComplete, + isRefreshing: lastSnapshotComplete, + }); }; const clearPendingPersist = () => { @@ -312,7 +333,11 @@ export function createSubscribeCollection(options: { lastSnapshotComplete = false; liveSnapshotApplied = false; materialize(); - writeStatus({ error: undefined, snapshotComplete: false }); + writeStatus({ + error: undefined, + snapshotComplete: false, + isRefreshing: false, + }); }; const suspendPersistence = () => { @@ -332,3 +357,46 @@ export function createSubscribeCollection(options: { suspendPersistence, }; } + +export interface ScopedSubscribeCollections { + /** The collection for `key` under `scope`, created on first use. */ + get: (scope: string, key: string) => SubscribeCollection; + getKey: (row: T) => string; +} + +/** + * In-memory collections, one per tenant scope (`organizationId|regionId`) and + * key, such as a cluster id. Keys repeat across environments, so a collection + * never serves another scope's rows. Rows stay in memory until the scope + * changes, which makes returning to a page instant. They are not persisted, + * since an entry per key would have no bound in localStorage. + */ +export function createScopedSubscribeCollections(options: { + /** Prefix of each collection's id. */ + id: string; + getKey: (row: T) => string; +}): ScopedSubscribeCollections { + const { id, getKey } = options; + const byKey = new Map>(); + let currentScope: string | undefined; + return { + get: (scope, key) => { + if (scope !== currentScope) { + // A return to an earlier scope gets fresh collections, so the dropped + // ones are never served again. + byKey.clear(); + currentScope = scope; + } + let collection = byKey.get(key); + if (!collection) { + collection = createSubscribeCollection({ + id: `${id}|${scope}|${key}`, + getKey, + }); + byKey.set(key, collection); + } + return collection; + }, + getKey, + }; +} diff --git a/console/src/api/materialize/subscribeSession.ts b/console/src/api/materialize/subscribeSession.ts index 6481746a2ebc6..ef790d9d6b34f 100644 --- a/console/src/api/materialize/subscribeSession.ts +++ b/console/src/api/materialize/subscribeSession.ts @@ -185,6 +185,14 @@ export function createCollectionSubscribeSession< unsubscribeScope?.(); keepAlive.unsubscribe(); session.destroy(); + // No subscribe backs the held rows any more, so they read as refreshing + // until a later session's snapshot lands, as behind a new subscribe's + // empty pre-snapshot. + target.applySnapshot({ + data: [], + snapshotComplete: false, + error: undefined, + }); }, }; } diff --git a/console/src/api/materialize/useScopedSubscribeCollection.ts b/console/src/api/materialize/useScopedSubscribeCollection.ts new file mode 100644 index 0000000000000..e05d05d313a12 --- /dev/null +++ b/console/src/api/materialize/useScopedSubscribeCollection.ts @@ -0,0 +1,58 @@ +// 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 { useAtomValue } from "jotai"; +import { RawBuilder } from "kysely"; +import React from "react"; + +import { syncEngineCacheScopeLoadableAtom } from "~/store/syncEngineCache"; + +import { + idleCollectionStatusAtom, + ScopedSubscribeCollections, +} from "./subscribeCollection"; +import { SubscribeRow } from "./SubscribeManager"; +import { useGlobalSubscribeCollection } from "./useSubscribe"; + +// NOTE: kept out of useSubscribe.ts. Importing ~/store/syncEngineCache there +// closes an import cycle, observed to break UsageRoutes.test.tsx. +/** + * Runs `subscribe` into the collection for `key` under the current tenant + * scope, for as long as the calling component is mounted. While the scope + * resolves, `collection` is undefined and no subscribe runs. + * + * `subscribe` must be referentially stable, since a new builder restarts the + * subscribe session. Its upsert key must identify a row as + * `collections.getKey` does. + */ +export function useScopedSubscribeCollection( + collections: ScopedSubscribeCollections, + key: string, + subscribe: RawBuilder | undefined, +) { + const scope = useAtomValue(syncEngineCacheScopeLoadableAtom); + const collection = + scope.state === "hasData" && scope.data + ? collections.get(scope.data, key) + : undefined; + const options = React.useMemo( + () => ({ + target: collection, + subscribe, + select: (row: SubscribeRow) => row.data, + upsertKey: (row: SubscribeRow) => collections.getKey(row.data), + }), + [collections, collection, subscribe], + ); + useGlobalSubscribeCollection(options); + const status = useAtomValue( + collection?.statusAtom ?? idleCollectionStatusAtom, + ); + return { collection, status, isScopeError: scope.state === "hasError" }; +} diff --git a/console/src/api/materialize/useSubscribe.test.tsx b/console/src/api/materialize/useSubscribe.test.tsx index cbd432a011e02..e3c9eb0c4ab84 100644 --- a/console/src/api/materialize/useSubscribe.test.tsx +++ b/console/src/api/materialize/useSubscribe.test.tsx @@ -159,6 +159,34 @@ describe("subscribeSession", () => { } }); + it("marks a collection's held rows as refreshing once its session ends", async () => { + const store = setTwoRegions(); + const target = createSubscribeCollection({ + id: "session-test-destroy", + getKey: (row) => row.id, + }); + const session = createCollectionSubscribeSession({ + store, + request, + upsertKey, + select: (row) => row.data, + target, + }); + target.applySnapshot({ + data: [{ id: "1" }], + snapshotComplete: true, + error: undefined, + }); + session.destroy(); + + await flush(); + expect(target.collection.size).toBe(1); + expect(store.get(target.statusAtom)).toMatchObject({ + snapshotComplete: true, + isRefreshing: true, + }); + }); + it("clears an atom-fed collection when the region switches", async () => { const store = setTwoRegions(); const sourceAtom = atom>(loadingState); diff --git a/console/src/api/materialize/useSubscribe.ts b/console/src/api/materialize/useSubscribe.ts index 50f87b319d92f..a766e9e0a1554 100644 --- a/console/src/api/materialize/useSubscribe.ts +++ b/console/src/api/materialize/useSubscribe.ts @@ -141,10 +141,11 @@ export function useGlobalUpsertSubscribe>( /** * Runs an app-session subscribe that feeds a TanStack DB collection, for as * long as the calling component is mounted. Mirrors useGlobalUpsertSubscribe - * with createCollectionSubscribeSession owning the wiring. + * with createCollectionSubscribeSession owning the wiring. No subscribe runs + * while `subscribe` or `target` is undefined. * - * `options` must be referentially stable (declare it at module scope); a new - * object restarts the subscribe session. + * `options` must be referentially stable (declare it at module scope, or + * memoize it); a new object restarts the subscribe session. * * Note that the subscribe statement must have WITH (PROGRESS) and ENVELOPE UPSERT. */ @@ -154,18 +155,19 @@ export function useGlobalSubscribeCollection< >( options: UseSubscribeOptions & { upsertKey: UpsertKeyFunction; - target: SubscribeCollection; + target: SubscribeCollection | undefined; scopeAtom?: ScopeLoadableAtom; }, ) { const store = useStore(); const request = useSubscribeRequest(options.subscribe); React.useEffect(() => { - if (!request) return; + const { target } = options; + if (!request || !target) return; const session = createCollectionSubscribeSession({ store, request, - target: options.target, + target, scopeAtom: options.scopeAtom, upsertKey: options.upsertKey, select: options.select, diff --git a/console/src/platform/clusters/ClusterPage/ClusterMetrics.tsx b/console/src/platform/clusters/ClusterPage/ClusterMetrics.tsx index cc4d8204da1a2..b707111fdda03 100644 --- a/console/src/platform/clusters/ClusterPage/ClusterMetrics.tsx +++ b/console/src/platform/clusters/ClusterPage/ClusterMetrics.tsx @@ -78,11 +78,7 @@ const ClusterMetricsContent = ({ cluster }: { cluster: Cluster }) => { timePeriodMinutes={timePeriodMinutes} selectedReplicaId={selectedReplica?.id} /> - + ); diff --git a/console/src/platform/clusters/ClusterPage/MemoryByObject.test.tsx b/console/src/platform/clusters/ClusterPage/MemoryByObject.test.tsx index 91fa33d36a4e1..ab486a5cac537 100644 --- a/console/src/platform/clusters/ClusterPage/MemoryByObject.test.tsx +++ b/console/src/platform/clusters/ClusterPage/MemoryByObject.test.tsx @@ -7,11 +7,15 @@ // the Business Source License, use of this software will be governed // by the Apache License, Version 2.0. -import { screen, waitFor, within } from "@testing-library/react"; +import { fireEvent, screen, waitFor, within } from "@testing-library/react"; +import { http } from "msw"; import React from "react"; +import { Cluster, Replica } from "~/api/materialize/cluster/clusterList"; +import { DatabaseObject } from "~/api/materialize/objects"; import { ErrorCode, MzDataType } from "~/api/materialize/types"; import { + buildColumn, buildColumns, buildSqlQueryHandlerV2, mapKyselyToTabular, @@ -19,17 +23,22 @@ import { } from "~/api/mocks/buildSqlQueryHandler"; import server from "~/api/mocks/server"; import { getStore } from "~/jotai"; +import { buildClusterServerResponse } from "~/platform/clusters/clustersTestUtils"; import { clusterQueryKeys } from "~/platform/clusters/queries"; +import { allObjectsCollection } from "~/store/allObjectsCollection"; +import { syncEngineCacheScopeAtom } from "~/store/syncEngineCache"; +import { mockSubscribeSocket } from "~/test/mockSubscribeSocket"; import { - defaultRegionId, healthyEnvironment, renderComponent, setFakeEnvironment, } from "~/test/utils"; +import { assert } from "~/util"; import { parseDbVersion } from "~/version/api"; import { MemoryByObject } from "./MemoryByObject"; -import { segmentWidths } from "./memoryByObjectModel"; +import { otherMemory, segmentWidths } from "./memoryByObjectModel"; +import { sizesByReplica } from "./useLiveLargestObjects"; const flags = vi.hoisted(() => ({ values: {} as Record })); @@ -37,19 +46,142 @@ vi.mock("~/hooks/useFlags", () => ({ useFlags: () => flags.values })); const HEAP_LIMIT = 4069523456; +const buildReplica = (id: string, name: string): Replica => ({ + id, + name, + size: "25cc", + disk: true, + statuses: [], +}); + +const buildCluster = (replicas: Replica[]) => + buildClusterServerResponse({ id: "u1", name: "quickstart" }, replicas); + +/** + * Answers the page's heap reads: each process's latest heap limit and use, by + * replica id, and the size's process count. Build after setting the + * environment, since the query keys embed its version. + */ +const heapReadsHandlers = ( + cluster: Cluster, + { + processes = 1, + heapLimit = String(HEAP_LIMIT), + heapBytes = {}, + }: { + processes?: number; + heapLimit?: string | null; + /** Each process's heap use, by replica id. Unlisted replicas report none. */ + heapBytes?: Record; + } = {}, +) => [ + buildSqlQueryHandlerV2({ + queryKey: clusterQueryKeys.replicaHeapLimits({ + replicaIds: cluster.replicas.map(({ id }) => id).sort(), + }), + results: mapKyselyToTabular({ + columns: buildColumns([ + "replicaId", + "processId", + "heapLimit", + "memoryBytes", + "heapBytes", + ]), + rows: cluster.replicas.flatMap(({ id }) => + Array.from({ length: processes }, (_, processId) => ({ + replicaId: id, + processId: String(processId), + heapLimit, + memoryBytes: "1", + heapBytes: heapBytes[id] ?? null, + })), + ), + }), + }), + buildSqlQueryHandlerV2({ + queryKey: clusterQueryKeys.replicaSizeAllocations({ sizes: ["25cc"] }), + results: mapKyselyToTabular({ + columns: [ + buildColumn({ name: "size" }), + ...["processes", "cpuNanoCores", "memoryBytes", "diskBytes"].map( + (name) => buildColumn({ name, type_oid: MzDataType.float8 }), + ), + ], + rows: [ + { + size: "25cc", + processes: String(processes), + cpuNanoCores: "1", + memoryBytes: "1", + diskBytes: "1", + }, + ], + }), + }), +]; + +/** Counts largest replica queries, leaving their answers to other handlers. */ +const countLargestReplicaQueries = () => { + const requests = { count: 0 }; + const handler = http.post("*/api/sql", ({ request }) => { + const queryKey = JSON.parse( + new URL(request.url).searchParams.get("query_key") ?? "[]", + ); + if (queryKey[1]?.scope === "largestClusterReplica") requests.count += 1; + return undefined; + }); + return { requests, handler }; +}; + +/** The card under a replica select, as the Metrics page lays them out. */ +const ReplicaSwitcher = ({ + cluster, + initialReplicaName, +}: { + cluster: Cluster; + initialReplicaName?: string; +}) => { + const [replicaName, setReplicaName] = React.useState(initialReplicaName); + return ( + <> + + name === replicaName)} + /> + + ); +}; + +/** The select's value for all replicas. */ +const ALL_REPLICAS = ""; + +const selectReplica = (name: string) => + fireEvent.change(screen.getByRole("combobox", { name: "Replica" }), { + target: { value: name }, + }); + const largestReplicaHandler = ( rows: Array>, options?: SqlHandlerOptions, - replicaName?: string, ) => buildSqlQueryHandlerV2( { - queryKey: clusterQueryKeys.largestClusterReplica({ - clusterId: "u1", - replicaName, - }), + queryKey: clusterQueryKeys.largestClusterReplica({ clusterId: "u1" }), results: mapKyselyToTabular({ columns: buildColumns([ + "id", "name", "size", { type_oid: MzDataType.numeric, name: "heapLimit" }, @@ -117,8 +249,12 @@ const largestObjectsHandler = ( }), }); -const renderMemoryByObject = () => - renderComponent(); +const R1 = buildReplica("u678", "r1"); + +const renderMemoryByObject = (cluster = buildCluster([R1])) => { + server.use(...heapReadsHandlers(cluster)); + return renderComponent(); +}; describe("MemoryByObject", () => { beforeEach(() => { @@ -148,25 +284,15 @@ describe("MemoryByObject", () => { expect(within(objects).getByText("5.09 GB (31.8%)")).toBeVisible(); }); - it("shows the named replica instead of the largest", async () => { - server.use( - largestReplicaHandler( - [{ name: "r1", size: "25cc", heapLimit: String(HEAP_LIMIT) }], - undefined, - "r1", - ), - largestObjectsHandler(), - ); - await renderComponent( - , - ); + it("reads a named replica from the page instead of querying the largest", async () => { + const cluster = buildCluster([R1]); + const { requests, handler } = countLargestReplicaQueries(); + server.use(handler, ...heapReadsHandlers(cluster), largestObjectsHandler()); + await renderComponent(); expect(await screen.findByText("% of heap limit on r1")).toBeVisible(); expect(screen.getByText("5.09 GB (31.8%)")).toBeVisible(); + expect(requests.count).toEqual(0); }); it("links to the workflow graph while the maintained objects page is off", async () => { @@ -259,7 +385,7 @@ describe("MemoryByObject", () => { it("shows nothing for a cluster without replicas", async () => { server.use(largestReplicaHandler([])); - await renderMemoryByObject(); + await renderMemoryByObject(buildCluster([])); await waitFor(() => expect(screen.queryByTestId("loading-spinner")).not.toBeInTheDocument(), @@ -292,91 +418,315 @@ describe("MemoryByObject", () => { screen.getByRole("heading", { name: "Objects by Memory" }), ).toBeVisible(); }); +}); - it("reads sizes from mz_object_arrangement_sizes on Materialize >= v26.35", async () => { - const environment = { - ...healthyEnvironment, - status: { - ...healthyEnvironment.status, - version: parseDbVersion("v26.35.0 (ea0d129f)"), - }, - }; - // Query keys embed the environment version, so set it before building - // the handlers that match on them. - await setFakeEnvironment(getStore().set, defaultRegionId, environment); - server.use( - largestReplicaHandler([ - { name: "r1", size: "25cc", heapLimit: String(HEAP_LIMIT) }, - ]), - buildSqlQueryHandlerV2({ - queryKey: clusterQueryKeys.largestMaintainedQueries({ - clusterId: "u1", - clusterName: "quickstart", - limit: 10, - replicaName: "r1", - replicaHeapLimit: HEAP_LIMIT, - unifiedSizes: true, - }), - results: mapKyselyToTabular({ - columns: buildColumns([ - "object_id", - "size", - { type_oid: MzDataType.numeric, name: "memoryPercentage" }, - ]), - rows: [ - { - object_id: "u188", - size: "5469140917", - memoryPercentage: "31.8345902256", - }, - // Missing from the names below: a dropped object's dataflow. - { - object_id: "u999", - size: "424919434", - memoryPercentage: "11.2686157226", - }, - ], - }), - }), - buildSqlQueryHandlerV2({ - queryKey: clusterQueryKeys.maintainedObjectNames(["u188", "u999"]), - results: mapKyselyToTabular({ - columns: buildColumns([ - "id", - "name", - "type", - "schemaName", - "databaseName", - ]), - rows: [ - { - id: "u188", - name: "customer_view", - type: "materialized-view", - schemaName: "public", - databaseName: "materialize", - }, - ], - }), - }), - ); - await renderComponent( - , - { - initializeState: ({ set }) => - setFakeEnvironment(set, defaultRegionId, environment), - }, - ); +// Sizes stream from mz_object_arrangement_sizes on 26.35 and later. +const LIVE_SIZES_ENVIRONMENT = { + ...healthyEnvironment, + status: { + ...healthyEnvironment.status, + version: parseDbVersion("v26.45.0 (0000000000)"), + }, +}; +const GIB = 2 ** 30; + +const catalogObject = (id: string, name: string): DatabaseObject => ({ + databaseName: "materialize", + databaseId: "u1", + name, + schemaName: "public", + schemaId: "u2", + id, + objectType: "materialized-view", + sourceType: null, + isWebhookTable: null, + clusterId: "u1", + clusterName: "quickstart", +}); + +const sizesSnapshot = (rows: Array<[string, number]>) => ({ + columns: [ + buildColumn({ name: "objectId" }), + buildColumn({ name: "size", type_oid: MzDataType.float8 }), + ], + rows: rows.map(([objectId, size]) => [objectId, String(size)]), +}); + +const setLiveSizesEnvironment = () => + setFakeEnvironment(getStore().set, "aws/us-east-1", LIVE_SIZES_ENVIRONMENT); + +const renderLive = (element: React.ReactElement) => + renderComponent(element, { + initializeState: ({ set }) => + setFakeEnvironment(set, "aws/us-east-1", LIVE_SIZES_ENVIRONMENT), + }); + +// Query keys embed the environment version, so set the environment before +// building the handlers that match on them. Each test streams replicas of its +// own: the sizes collections outlive a test. +const renderLiveMemoryByObject = async ({ + replicaId, + heapLimit = String(HEAP_LIMIT), +}: { + replicaId: string; + heapLimit?: string | null; +}) => { + await setLiveSizesEnvironment(); + const cluster = buildCluster([buildReplica(replicaId, "r1")]); + server.use( + ...heapReadsHandlers(cluster), + largestReplicaHandler([ + { id: replicaId, name: "r1", size: "25cc", heapLimit }, + ]), + mockSubscribeSocket((sql, params) => + sql.includes("mz_object_arrangement_sizes") && params.includes(replicaId) + ? sizesSnapshot([ + ["u188", 2 * GIB], + ["u404", GIB], + ]) + : undefined, + ), + ); + return renderLive(); +}; + +describe("MemoryByObject with streamed sizes", () => { + beforeEach(() => { + flags.values = {}; + allObjectsCollection.applySnapshot({ + data: [catalogObject("u188", "customer_view")], + snapshotComplete: true, + error: undefined, + }); + }); + + it("names streamed objects from the catalog", async () => { + await renderLiveMemoryByObject({ replicaId: "u5" }); const objects = await screen.findByRole("list", { name: "Objects" }); expect( within(objects).getByRole("link", { name: "customer_view" }), ).toBeVisible(); - expect(within(objects).getByText("5.09 GB (31.8%)")).toBeVisible(); - expect(within(objects).getByText("u999")).toBeVisible(); + // Not in the catalog, so a dropped object's dataflow. + expect(within(objects).getByText("u404")).toBeVisible(); + expect( + within(objects).queryByRole("link", { name: "u404" }), + ).not.toBeInTheDocument(); + }); + + it("lists sizes without a bar when the replica reports no heap limit", async () => { + await renderLiveMemoryByObject({ replicaId: "u6", heapLimit: null }); + + const objects = await screen.findByRole("list", { name: "Objects" }); + expect(within(objects).getByText("customer_view")).toBeVisible(); + expect(within(objects).getByText("2.00 GB")).toBeVisible(); + expect(screen.getByText("Size on r1")).toBeVisible(); expect( - within(objects).queryByRole("link", { name: "u999" }), + screen.queryByRole("img", { name: "Memory by object" }), ).not.toBeInTheDocument(); + expect(screen.queryByText(/Infinity|NaN/)).not.toBeInTheDocument(); + }); + + it("waits for the catalog before naming objects", async () => { + allObjectsCollection.reset(); + await renderLiveMemoryByObject({ replicaId: "u7" }); + + const scope = await getStore().get(syncEngineCacheScopeAtom); + assert(scope); + const sizes = sizesByReplica.get(scope, "u7"); + await waitFor(() => + expect(getStore().get(sizes.statusAtom).snapshotComplete).toBe(true), + ); + expect(screen.getByTestId("loading-spinner")).toBeVisible(); + expect(screen.queryByText("u188")).not.toBeInTheDocument(); + allObjectsCollection.applySnapshot({ + data: [catalogObject("u188", "customer_view")], + snapshotComplete: true, + error: undefined, + }); + expect(await screen.findByText("customer_view")).toBeVisible(); + }); + + it("takes a named replica's heap limit across its processes", async () => { + await setLiveSizesEnvironment(); + const replica = buildReplica("u20", "r1"); + const cluster = buildCluster([replica]); + const { requests, handler } = countLargestReplicaQueries(); + server.use( + handler, + ...heapReadsHandlers(cluster, { + processes: 2, + heapLimit: String(4 * GIB), + }), + mockSubscribeSocket((sql, params) => + params.includes("u20") ? sizesSnapshot([["u188", 2 * GIB]]) : undefined, + ), + ); + await renderLive(); + + const objects = await screen.findByRole("list", { name: "Objects" }); + expect(within(objects).getByText("2.00 GB (25.0%)")).toBeVisible(); + expect(requests.count).toEqual(0); + }); + + it("keeps the previous replica's objects on screen until the next replica's snapshot", async () => { + await setLiveSizesEnvironment(); + const cluster = buildCluster([ + buildReplica("u21", "r1"), + buildReplica("u22", "r2"), + ]); + const { requests, handler } = countLargestReplicaQueries(); + let releaseR2 = () => {}; + const r2Released = new Promise((resolve) => { + releaseR2 = resolve; + }); + server.use( + handler, + ...heapReadsHandlers(cluster), + mockSubscribeSocket(async (sql, params) => { + if (params.includes("u21")) return sizesSnapshot([["u188", GIB]]); + if (!params.includes("u22")) return undefined; + await r2Released; + return sizesSnapshot([["u404", 2 * GIB]]); + }), + ); + await renderLive( + , + ); + expect(await screen.findByText("customer_view")).toBeVisible(); + + selectReplica("r2"); + expect(await screen.findByTestId("refreshing-spinner")).toBeVisible(); + expect(screen.getByText("% of heap limit on r1")).toBeVisible(); + expect(screen.getByText("customer_view")).toBeVisible(); + expect(screen.queryByTestId("loading-spinner")).not.toBeInTheDocument(); + + releaseR2(); + expect(await screen.findByText("% of heap limit on r2")).toBeVisible(); + expect(screen.getByText("u404")).toBeVisible(); + expect(screen.queryByText("customer_view")).not.toBeInTheDocument(); + expect(screen.queryByTestId("refreshing-spinner")).not.toBeInTheDocument(); + expect(requests.count).toEqual(0); + }); + + it("queries the largest replica once all replicas are selected", async () => { + await setLiveSizesEnvironment(); + const cluster = buildCluster([ + buildReplica("u23", "r1"), + buildReplica("u24", "r2"), + ]); + const { requests, handler } = countLargestReplicaQueries(); + server.use( + handler, + ...heapReadsHandlers(cluster), + largestReplicaHandler([ + { id: "u24", name: "r2", size: "25cc", heapLimit: String(HEAP_LIMIT) }, + ]), + mockSubscribeSocket((sql, params) => + params.includes("u23") || params.includes("u24") + ? sizesSnapshot([["u188", GIB]]) + : undefined, + ), + ); + await renderLive( + , + ); + expect(await screen.findByText("% of heap limit on r1")).toBeVisible(); + expect(requests.count).toEqual(0); + + selectReplica(ALL_REPLICAS); + expect(await screen.findByText("% of heap limit on r2")).toBeVisible(); + expect(requests.count).toEqual(1); + }); + + describe("other memory", () => { + const renderOtherMemory = async ({ + replicaId, + heapLimit = String(4 * GIB), + heapBytes, + }: { + replicaId: string; + heapLimit?: string | null; + heapBytes?: string; + }) => { + await setLiveSizesEnvironment(); + const replica = buildReplica(replicaId, "r1"); + const cluster = buildCluster([replica]); + server.use( + ...heapReadsHandlers(cluster, { + processes: 2, + heapLimit, + heapBytes: heapBytes ? { [replicaId]: heapBytes } : {}, + }), + mockSubscribeSocket((sql, params) => + params.includes(replicaId) + ? sizesSnapshot([ + ["u188", 2 * GIB], + ["u404", GIB], + ]) + : undefined, + ), + ); + await renderLive(); + return screen.findByRole("list", { name: "Objects" }); + }; + + it("sizes it by the heap in use beyond the listed objects", async () => { + // 6 GiB in use over two processes, 3 GiB of it listed, of 8 GiB. + const objects = await renderOtherMemory({ + replicaId: "u30", + heapBytes: String(3 * GIB), + }); + + const other = within(objects).getByText("Other memory").closest("li"); + assert(other); + expect(within(other).getByText("3.00 GB (37.5%)")).toBeVisible(); + expect( + within(other).getByLabelText( + "Heap in use that isn't attributed to the objects listed. Object sizes come from arrangement sizes, which undercount.", + ), + ).toBeInTheDocument(); + }); + + it("leaves it out while the replica reports no heap use", async () => { + const objects = await renderOtherMemory({ replicaId: "u31" }); + + expect(within(objects).getByText("customer_view")).toBeVisible(); + expect( + within(objects).queryByText("Other memory"), + ).not.toBeInTheDocument(); + }); + + it("leaves it out while the replica reports no heap limit", async () => { + const objects = await renderOtherMemory({ + replicaId: "u32", + heapLimit: null, + heapBytes: String(3 * GIB), + }); + + expect(within(objects).getByText("2.00 GB")).toBeVisible(); + expect( + within(objects).queryByText("Other memory"), + ).not.toBeInTheDocument(); + }); + }); +}); + +describe("otherMemory", () => { + it("is the heap in use beyond the listed sizes, as a share of the limit", () => { + expect( + otherMemory(600, 1000, [ + { size: 200 }, + { size: BigInt(100) }, + { size: null }, + ]), + ).toEqual({ size: 300, memoryPercentage: 30 }); + }); + + it("is undefined when the listed sizes cover the heap in use", () => { + expect(otherMemory(300, 1000, [{ size: 200 }, { size: 100 }])).toBe( + undefined, + ); }); }); diff --git a/console/src/platform/clusters/ClusterPage/MemoryByObject.tsx b/console/src/platform/clusters/ClusterPage/MemoryByObject.tsx index dcfdded570930..aa6891b2dc7aa 100644 --- a/console/src/platform/clusters/ClusterPage/MemoryByObject.tsx +++ b/console/src/platform/clusters/ClusterPage/MemoryByObject.tsx @@ -12,6 +12,7 @@ import { Card, Flex, HStack, + Spinner, Text, Tooltip, useTheme, @@ -20,80 +21,214 @@ import { import React from "react"; import { Link as RouterLink } from "react-router-dom"; +import { Cluster, Replica } from "~/api/materialize/cluster/clusterList"; +import Alert from "~/components/Alert"; 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 { useLargestMaintainedQueries } from "~/platform/clusters/queries"; +import { InfoIcon } from "~/icons"; +import { + LargestReplica, + LargestReplicaBoundary, +} from "~/platform/clusters/LargestMaintainedQueries"; +import { + UNIFIED_OBJECT_SIZES_MIN_VERSION, + useLargestMaintainedQueries, + useReplicaHeapBytes, + useReplicaHeapLimits, + useReplicaSizeAllocations, +} from "~/platform/clusters/queries"; import { absoluteMaintainedObjectPath, useBuildWorkflowGraphPath, } from "~/platform/routeHelpers"; -import { useRegionSlug } from "~/store/environments"; +import { useEnvironmentGate, useRegionSlug } from "~/store/environments"; import { MaterializeTheme } from "~/theme"; import { formatMemoryUsage } from "~/utils/format"; -import { segmentWidths, TOP_OBJECT_COUNT } from "./memoryByObjectModel"; +import { + LargestObject, + otherMemory, + segmentWidths, + TOP_OBJECT_COUNT, +} from "./memoryByObjectModel"; import { PERCENT_TICKS } from "./ResourceUsage/resourceUsageStyles"; +import { useLiveLargestObjects } from "./useLiveLargestObjects"; export interface MemoryByObjectProps { - clusterId: string; - clusterName: string; + cluster: Cluster; + /** The replica to show, else the cluster's largest. */ + replica?: Replica; } -type LargestObject = NonNullable< - ReturnType["data"] ->[number]; +const OTHER_MEMORY_DESCRIPTION = + "Heap in use that isn't attributed to the objects listed. Object sizes come from arrangement sizes, which undercount."; + +/** + * Heap limits and use of the cluster's current replicas, from the reads the + * resource usage card makes with the same keys. + */ +const useReplicaHeaps = (cluster: Cluster) => { + const replicaIds = cluster.replicas.map(({ id }) => id).sort(); + const sizes = [ + ...new Set(cluster.replicas.flatMap(({ size }) => (size ? [size] : []))), + ].sort(); + const heapLimits = useReplicaHeapLimits(replicaIds); + const allocations = useReplicaSizeAllocations(sizes); + const heapBytes = useReplicaHeapBytes(replicaIds); + return { + isLoading: + heapLimits.isLoading || allocations.isLoading || heapBytes.isLoading, + /** + * The replica's heap limit across its processes, as + * `buildLargestClusterReplicaQuery` computes it. Null if the process counts + * failed to load, since a per-process limit would inflate every share. + */ + heapLimitOf: ({ id, size }: Replica) => { + const perProcess = heapLimits.data?.get(id); + if (perProcess === undefined || allocations.isError) return null; + const processes = size + ? allocations.data?.get(size)?.processes + : undefined; + return perProcess * (processes ?? 1); + }, + heapBytesOf: (replicaId: string) => heapBytes.data?.get(replicaId), + }; +}; /** * The largest objects on a replica, as shares of its heap limit: the named - * replica, else the cluster's largest. + * replica, else the cluster's largest. A named replica is read from the page's + * clusters and heap reads, so switching to one waits only on its sizes. */ -export const MemoryByObject = ({ - clusterId, - clusterName, - replicaName, -}: MemoryByObjectProps & { replicaName?: string }) => ( - ( - {fallback} - )} - > - {(replica) => ( - - )} - -); +export const MemoryByObject = ({ cluster, replica }: MemoryByObjectProps) => { + const hasLiveSizes = + useEnvironmentGate(UNIFIED_OBJECT_SIZES_MIN_VERSION) === true; + const heaps = useReplicaHeaps(cluster); + return ( + ( + {fallback} + )} + > + {(shown) => + hasLiveSizes ? ( + + ) : heaps.isLoading ? ( + // A named replica's polled sizes are keyed on its heap limit, so + // loading them before it would load them twice. + + + + ) : ( + + ) + } + + ); +}; const PolledMemoryByObject = ({ clusterId, clusterName, - replicaName, - replicaHeapLimit, -}: MemoryByObjectProps & { - replicaName: string; - replicaHeapLimit: number | null; + replica, + heapUsedBytes, +}: { + clusterId: string; + clusterName: string; + replica: LargestReplica; + heapUsedBytes: number | undefined; }) => { const { data: objects } = useLargestMaintainedQueries({ clusterId, clusterName, limit: TOP_OBJECT_COUNT, - replicaName, - replicaHeapLimit, + replicaName: replica.name, + replicaHeapLimit: replica.heapLimit, }); return ( + ); +}; + +/** `value`, or while it's undefined, the last value it had in a commit. */ +function useLastDefined(value: T | undefined) { + const last = React.useRef(value); + React.useEffect(() => { + if (value !== undefined) last.current = value; + }); + return value ?? last.current; +} + +const LiveMemoryByObject = ({ + replica, + heapUsedBytes, + isHeapLoading, +}: { + replica: LargestReplica; + heapUsedBytes: number | undefined; + isHeapLoading: boolean; +}) => { + const { objects, snapshotComplete, isRefreshing, isError } = + useLiveLargestObjects({ + replicaId: replica.id, + heapLimit: replica.heapLimit, + }); + const content: MemoryByObjectContent | undefined = + snapshotComplete && !isHeapLoading + ? { + objects, + replicaName: replica.name, + heapLimit: replica.heapLimit, + heapUsedBytes, + } + : undefined; + // A switch to another replica keeps the previous one's content on screen + // until the new one's first snapshot and heap reads are in. + const shown = useLastDefined(content); + if (isError) { + return ( + + + + ); + } + if (!shown) { + return ( + + + + ); + } + return ( + ); }; @@ -102,10 +237,15 @@ const PolledMemoryByObject = ({ const MemoryByObjectFrame = ({ title = "Objects by Memory", subtitle, + isRefreshing = false, + isStale = false, children, }: { title?: string; subtitle?: string; + isRefreshing?: boolean; + /** Dims content kept on screen while the selected replica's loads. */ + isStale?: boolean; children: React.ReactNode; }) => { const { colors } = useTheme(); @@ -116,6 +256,7 @@ const MemoryByObjectFrame = ({ borderRadius="md" borderWidth="1px" borderColor={colors.border.primary} + aria-busy={isRefreshing} > @@ -123,26 +264,64 @@ const MemoryByObjectFrame = ({ {title} {subtitle && ( - + {subtitle} )} + {isRefreshing && ( + + )} - {children} + + {children} + ); }; -const MemoryByObjectCard = ({ - objects, - replicaName, - hasHeapLimit, -}: { +interface Segment { + key: string; + label: string; + usage: string; + color: string; + widthPercent: number; + path?: string; + description?: string; +} + +interface MemoryByObjectContent { objects: LargestObject[]; replicaName: string; /** Without a heap limit there are no shares, so the card lists sizes alone. */ - hasHeapLimit: boolean; + heapLimit: number | null; + /** The replica's heap in use, which sizes the other memory segment. */ + heapUsedBytes: number | undefined; +} + +const MemoryByObjectCard = ({ + objects, + replicaName, + heapLimit, + heapUsedBytes, + isRefreshing, + isStale, +}: MemoryByObjectContent & { + isRefreshing?: boolean; + isStale?: boolean; }) => { const { colors } = useTheme(); // The maintained objects page only has routes while its flag is on. @@ -152,9 +331,10 @@ const MemoryByObjectCard = ({ const regionSlug = useRegionSlug(); const workflowGraphPath = useBuildWorkflowGraphPath(); + const hasHeapLimit = heapLimit !== null; if (objects.length === 0) { return ( - + No objects are using memory on {replicaName}. @@ -165,11 +345,18 @@ const MemoryByObjectCard = ({ // Colored by rank like the Freshness page's lines, so the largest objects // are furthest apart in hue. const palette = generateRainbowPalette(objects.length); + const other = + heapLimit && heapUsedBytes !== undefined + ? otherMemory(heapUsedBytes, heapLimit, objects) + : undefined; + // Other memory goes last, so the end of the bar clips it before any object. const widths = segmentWidths( - objects.map((object) => object.memoryPercentage), + [...objects, ...(other ? [other] : [])].map( + ({ memoryPercentage }) => memoryPercentage, + ), ); - const segments = objects.map((object, index) => { - const { id, name, databaseName, schemaName } = object; + const segments: Segment[] = objects.map((object, index) => { + const { id, name, databaseName, schemaName, type } = object; return { key: id ?? object.dataflowId ?? String(index), label: name ?? object.dataflowName ?? "-", @@ -177,16 +364,26 @@ const MemoryByObjectCard = ({ color: palette[index], widthPercent: widths[index], path: - object.isOrphanedDataflow || !id || !name || !schemaName + object.isOrphanedDataflow || !id || !name || !schemaName || !type ? undefined : hasMaintainedObjectsPage ? absoluteMaintainedObjectPath(regionSlug, id) : workflowGraphPath({ databaseObject: { id, name, databaseName, schemaName }, - type: object.type, + type, }), }; }); + if (other) { + segments.push({ + key: "other-memory", + label: "Other memory", + usage: formatMemoryUsage(other), + color: colors.foreground.tertiary, + widthPercent: widths[objects.length], + description: OTHER_MEMORY_DESCRIPTION, + }); + } return ( {hasHeapLimit && ( @@ -216,7 +415,14 @@ const MemoryByObjectCard = ({ segment.widthPercent > 0 && ( + {`${segment.label} · ${segment.usage}`} + {segment.description && ( + {segment.description} + )} + + } > {segment.usage} + {segment.description && ( + + + + )} ))} diff --git a/console/src/platform/clusters/ClusterPage/ResourceUsage/ResourceUsage.test.tsx b/console/src/platform/clusters/ClusterPage/ResourceUsage/ResourceUsage.test.tsx index 884e5c23c9e2d..87942cb419b81 100644 --- a/console/src/platform/clusters/ClusterPage/ResourceUsage/ResourceUsage.test.tsx +++ b/console/src/platform/clusters/ClusterPage/ResourceUsage/ResourceUsage.test.tsx @@ -923,6 +923,61 @@ describe("ResourceUsage", () => { ).not.toBeInTheDocument(); }); + it("draws a window of 3h or less from the live subscribe", async () => { + window.localStorage.setItem("mz-cluster-graph-time-period", "60"); + // The lineage key embeds the environment version. + await setFakeEnvironment(getStore().set, "aws/us-east-1", NEW_ENVIRONMENT); + server.use( + buildSqlQueryHandlerV2({ + queryKey: clusterQueryKeys.deploymentLineage({ clusterIdsKey: "u9" }), + results: mapKyselyToTabular({ + columns: buildColumns([ + "clusterId", + "currentDeploymentClusterId", + "clusterName", + ]), + rows: [], + }), + }), + mockSubscribeSocket((sql) => + sql.includes("utilization_overview_3h") + ? { + columns: [ + buildColumn({ name: "replicaId" }), + buildColumn({ name: "name" }), + buildColumn({ name: "occurredAt" }), + buildColumn({ + name: "cpuPercent", + type_oid: MzDataType.float8, + }), + buildColumn({ + name: "heapPercent", + type_oid: MzDataType.float8, + }), + ], + rows: [ + [ + "u1", + "r1", + new Date(NOW_MS - 30 * 60_000).toISOString(), + "0.45", + "0.6", + ], + ], + } + : undefined, + ), + ); + await renderResourceUsage({ withSwap: true }); + + expect( + await screen.findByRole("img", { name: "CPU usage" }), + ).toBeInTheDocument(); + expect( + screen.queryByText("No resource usage in this time period."), + ).not.toBeInTheDocument(); + }); + it("says replicas ran unmeasured when the window has no readings", async () => { await renderResourceUsage({ withSwap: true, utilizationRowCount: 0 }); diff --git a/console/src/platform/clusters/ClusterPage/ResourceUsage/useLiveClusterUtilization.test.ts b/console/src/platform/clusters/ClusterPage/ResourceUsage/useLiveClusterUtilization.test.ts new file mode 100644 index 0000000000000..28a4b78ad1d43 --- /dev/null +++ b/console/src/platform/clusters/ClusterPage/ResourceUsage/useLiveClusterUtilization.test.ts @@ -0,0 +1,337 @@ +// 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 { renderHook, waitFor } from "@testing-library/react"; +import { millisecondsInMinute } from "date-fns/constants"; +import { delay, http, ws } from "msw"; +import React from "react"; + +import { BinnedSubscribeRow } from "~/api/materialize/cluster/replicaUtilizationBinning"; +import { MzDataType } from "~/api/materialize/types"; +import { + buildColumn, + buildColumns, + buildSqlQueryHandlerV2, + mapKyselyToTabular, +} from "~/api/mocks/buildSqlQueryHandler"; +import server from "~/api/mocks/server"; +import { clusterQueryKeys } from "~/platform/clusters/queries"; +import { buildQueryClient } from "~/queryClient"; +import { + mockSubscribeSocket, + SubscribeSnapshot, +} from "~/test/mockSubscribeSocket"; +import { createProviderWrapper } from "~/test/utils"; + +import { + StoredSample, + transformLiveUtilization, + useLiveClusterUtilization, +} from "./useLiveClusterUtilization"; +const END_MS = Date.UTC(2026, 9, 2, 12, 0); + +const sample = (minutesAgo: number, cpuPercent: number): StoredSample => ({ + replicaId: "u1", + clusterId: "u9", + size: "100cc", + name: "r1", + occurredAt: new Date( + END_MS - minutesAgo * millisecondsInMinute, + ).toISOString(), + cpuPercent, + memoryPercent: 0.4, + diskPercent: null, + heapPercent: 0.5, + memoryAndDiskPercent: null, +}); + +const bucket = (minutesAgo: number, cpuPercent: number): BinnedSubscribeRow => { + const start = new Date( + END_MS - minutesAgo * millisecondsInMinute, + ).toISOString(); + return { + bucketStart: start, + bucketEnd: new Date( + END_MS - (minutesAgo - 5) * millisecondsInMinute, + ).toISOString(), + replicaId: "u1", + clusterId: "u9", + size: "100cc", + name: "r1", + maxMemoryPercent: 0.4, + maxMemoryAt: start, + maxDiskPercent: null, + maxDiskAt: start, + maxCpuPercent: cpuPercent, + maxCpuAt: start, + maxHeapPercent: 0.5, + maxHeapAt: start, + maxMemoryAndDiskPercent: null, + maxMemoryAndDiskMemoryPercent: null, + maxMemoryAndDiskDiskPercent: null, + maxMemoryAndDiskAt: start, + offlineEvents: null, + }; +}; + +const cpuSeries = (result: ReturnType) => + result.graphData[0].data.map((point) => point.cpuPercent); + +describe("transformLiveUtilization", () => { + it("bins 3h samples into the window's buckets, keeping each bucket's peak", () => { + const result = transformLiveUtilization({ + rows: { + tier: "unbinned3h", + // The 3h subscribe holds samples from before a 1h window too. + samples: [sample(90, 0.9), sample(30, 0.2), sample(29.5, 0.6)], + }, + endDate: new Date(END_MS), + timePeriodMinutes: 60, + bucketSizeMs: millisecondsInMinute, + }); + expect(cpuSeries(result)).toEqual([60]); + }); + + it("clips 24h buckets to the window and puts them in time order", () => { + const result = transformLiveUtilization({ + rows: { + tier: "binned24h", + buckets: [bucket(60, 0.3), bucket(600, 0.9), bucket(120, 0.1)], + }, + endDate: new Date(END_MS), + timePeriodMinutes: 6 * 60, + bucketSizeMs: 5 * millisecondsInMinute, + }); + expect(cpuSeries(result)).toEqual([10, 30]); + }); +}); + +describe("useLiveClusterUtilization", () => { + type LiveUtilization = ReturnType; + + const lineageHandler = (clusterId: string) => + buildSqlQueryHandlerV2({ + queryKey: clusterQueryKeys.deploymentLineage({ + clusterIdsKey: clusterId, + }), + results: mapKyselyToTabular({ + columns: buildColumns([ + "clusterId", + "currentDeploymentClusterId", + "clusterName", + ]), + rows: [], + }), + }); + + const minutesAgo = (minutes: number) => + new Date(Date.now() - minutes * millisecondsInMinute); + + const sampleSnapshot = (occurredAt: Date): SubscribeSnapshot => ({ + columns: [ + buildColumn({ name: "replicaId" }), + buildColumn({ name: "occurredAt" }), + buildColumn({ + name: "cpuPercent", + type_oid: MzDataType.float8, + }), + ], + rows: [["u1", occurredAt.toISOString(), "0.5"]], + }); + + // Serves a 3h snapshot of one sample from 70 minutes ago, outside a 1h + // window, while `subscribe.serve` is set. + const serveHeldSample = (clusterId: string) => { + const subscribe = { serve: true, requests: 0 }; + const occurredAt = minutesAgo(70); + server.use( + lineageHandler(clusterId), + mockSubscribeSocket((sql) => { + if (!sql.includes("utilization_overview_3h")) return undefined; + subscribe.requests += 1; + return subscribe.serve ? sampleSnapshot(occurredAt) : undefined; + }), + ); + return subscribe; + }; + + // Answers each cluster's 3h subscribe with `snapshotFor(clusterId)`, and + // returns the clusters subscribed to, in order. + const serveSubscribes = ( + clusterIds: string[], + snapshotFor: (clusterId: string) => SubscribeSnapshot | undefined, + ) => { + const requests: string[] = []; + server.use( + ...clusterIds.map(lineageHandler), + mockSubscribeSocket((sql, params) => { + const clusterId = clusterIds.find((id) => params.includes(id)); + if (!sql.includes("utilization_overview_3h") || !clusterId) { + return undefined; + } + requests.push(clusterId); + return snapshotFor(clusterId); + }), + ); + return requests; + }; + + type WindowProps = { clusterId: string; timePeriodMinutes: number }; + + const renderWindow = ( + initialProps: WindowProps, + wrapper: React.ComponentType, + ) => { + const renders: LiveUtilization[] = []; + const rendered = renderHook( + ({ clusterId, timePeriodMinutes }: WindowProps) => { + const result = useLiveClusterUtilization({ + clusterId, + timePeriodMinutes, + bucketSizeMs: millisecondsInMinute, + tier: "unbinned3h", + includeMemoryBreakdown: false, + }); + renders.push(result); + return result; + }, + { wrapper, initialProps }, + ); + return { ...rendered, renders }; + }; + + const renderOneHourWindow = ( + clusterId: string, + wrapper: React.ComponentType, + ) => renderWindow({ clusterId, timePeriodMinutes: 60 }, wrapper); + + // Held rows outside the window must never read as a window with no usage. + const expectEveryRenderAwaitingRows = (renders: LiveUtilization[]) => { + expect(renders.length).toBeGreaterThan(0); + for (const render of renders) { + expect(render).toMatchObject({ + data: undefined, + isLoading: true, + isRefreshing: true, + }); + } + }; + + it("waits for a new snapshot when held rows have left the window", async () => { + const subscribe = serveHeldSample("u9"); + const ProviderWrapper = await createProviderWrapper(); + + const first = renderOneHourWindow("u9", ProviderWrapper); + await waitFor(() => expect(first.result.current.data).toBeDefined()); + expect(first.result.current.data?.graphData).toEqual([]); + first.unmount(); + + // Coming back, the held sample is still outside the window and the new + // subscribe hasn't delivered yet. + subscribe.serve = false; + const second = renderOneHourWindow("u9", ProviderWrapper); + await waitFor(() => expect(subscribe.requests).toEqual(2)); + expectEveryRenderAwaitingRows(second.renders); + }); + + it("holds rows as refreshing while an evicted lineage is refetched", async () => { + serveHeldSample("u10"); + const first = renderOneHourWindow("u10", await createProviderWrapper()); + await waitFor(() => expect(first.result.current.data).toBeDefined()); + first.unmount(); + + // A fresh query cache has no lineage, and its refetch never returns, so + // no subscribe starts. + let lineageRequested = false; + server.use( + http.post("*/api/sql", async () => { + lineageRequested = true; + await delay("infinite"); + }), + ); + const second = renderOneHourWindow( + "u10", + await createProviderWrapper({ queryClient: buildQueryClient() }), + ); + await waitFor(() => expect(lineageRequested).toBe(true)); + expectEveryRenderAwaitingRows(second.renders); + }); + + it("keeps the subscribe when the window changes within a tier", async () => { + const requests = serveSubscribes(["u11"], () => + sampleSnapshot(minutesAgo(30)), + ); + const rendered = renderWindow( + { clusterId: "u11", timePeriodMinutes: 15 }, + await createProviderWrapper(), + ); + await waitFor(() => expect(rendered.result.current.data).toBeDefined()); + expect(rendered.result.current.data?.graphData).toEqual([]); + + const rendersBefore = rendered.renders.length; + rendered.rerender({ clusterId: "u11", timePeriodMinutes: 60 }); + expect(rendered.result.current.data?.graphData).toHaveLength(1); + // Room for a restarted session to send its subscribe. + await new Promise((resolve) => setTimeout(resolve, 250)); + expect(requests).toEqual(["u11"]); + for (const render of rendered.renders.slice(rendersBefore)) { + expect(render.isRefreshing).toBe(false); + } + }); + + it("serves no rows of the previous cluster after a cluster switch", async () => { + const requests = serveSubscribes(["u12", "u13"], (clusterId) => + clusterId === "u12" ? sampleSnapshot(minutesAgo(30)) : undefined, + ); + const rendered = renderOneHourWindow("u12", await createProviderWrapper()); + await waitFor(() => + expect(rendered.result.current.data?.graphData).toHaveLength(1), + ); + + const rendersBefore = rendered.renders.length; + rendered.rerender({ clusterId: "u13", timePeriodMinutes: 60 }); + await waitFor(() => expect(requests).toEqual(["u12", "u13"])); + const afterSwitch = rendered.renders.slice(rendersBefore); + expect(afterSwitch.length).toBeGreaterThan(0); + for (const render of afterSwitch) { + expect(render).toMatchObject({ data: undefined, isLoading: true }); + } + }); + + it("holds rows as refreshing while a dropped socket resubscribes", async () => { + let serve = true; + const requests = serveSubscribes(["u14"], () => + serve ? sampleSnapshot(minutesAgo(30)) : undefined, + ); + const sockets: Array<{ close: () => void }> = []; + server.use( + ws + .link(/^wss?:\/\//) + .addEventListener("connection", ({ client }) => sockets.push(client)), + ); + const rendered = renderOneHourWindow("u14", await createProviderWrapper()); + await waitFor(() => + expect(rendered.result.current.data?.graphData).toHaveLength(1), + ); + expect(rendered.result.current.isRefreshing).toBe(false); + + serve = false; + sockets.forEach((socket) => socket.close()); + // The socket reopens after the connection manager's backoff. + await waitFor(() => { + expect(requests).toEqual(["u14", "u14"]); + expect(rendered.result.current.isRefreshing).toBe(true); + }); + expect(rendered.result.current).toMatchObject({ + isLoading: false, + isError: false, + }); + expect(rendered.result.current.data?.graphData).toHaveLength(1); + }); +}); diff --git a/console/src/platform/clusters/ClusterPage/ResourceUsage/useLiveClusterUtilization.ts b/console/src/platform/clusters/ClusterPage/ResourceUsage/useLiveClusterUtilization.ts new file mode 100644 index 0000000000000..f5d73b79649e4 --- /dev/null +++ b/console/src/platform/clusters/ClusterPage/ResourceUsage/useLiveClusterUtilization.ts @@ -0,0 +1,195 @@ +// 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 React from "react"; + +import { + BinnedSubscribeRow, + bucketRowsToBucketsByReplicaId, + clipBinnedRowsToWindow, + parseBinnedSubscribeRow, + rebucketUtilizationSamples, + toReplicaUtilizationGraphData, + UtilizationSample, +} from "~/api/materialize/cluster/replicaUtilizationBinning"; +import { + buildConsoleClusterUtilizationOverview24hSubscribe, + buildConsoleClusterUtilizationUnbinned3hSubscribe, +} from "~/api/materialize/cluster/replicaUtilizationHistory"; +import { createScopedSubscribeCollections } from "~/api/materialize/subscribeCollection"; +import { useScopedSubscribeCollection } from "~/api/materialize/useScopedSubscribeCollection"; +import { + LIVE_TIER_RETENTION_MINUTES, + useClusterLineageIds, +} from "~/platform/clusters/queries"; + +/** A 3h-view sample as streamed, with its timestamp still a string. */ +export type StoredSample = Omit & { + occurredAt: string; +}; + +/** + * Shapes a live collection's rows for a chart window ending at `endDate`. The + * 3h tier holds raw samples, binned here. The 24h tier holds the whole day's + * 5-minute buckets, clipped here to the window. + */ +export function transformLiveUtilization({ + rows, + endDate, + timePeriodMinutes, + bucketSizeMs, +}: { + rows: + | { tier: "unbinned3h"; samples: StoredSample[] } + | { tier: "binned24h"; buckets: BinnedSubscribeRow[] }; + endDate: Date; + timePeriodMinutes: number; + bucketSizeMs: number; +}) { + const startDate = subMinutes(endDate, timePeriodMinutes); + const bucketRows = + rows.tier === "unbinned3h" + ? rebucketUtilizationSamples( + rows.samples.map((sample) => ({ + ...sample, + occurredAt: new Date(sample.occurredAt), + })), + bucketSizeMs, + startDate.getTime(), + ) + : clipBinnedRowsToWindow( + rows.buckets.map(parseBinnedSubscribeRow), + startDate, + ); + return toReplicaUtilizationGraphData( + bucketRowsToBucketsByReplicaId(bucketRows), + startDate, + endDate, + ); +} + +const sampleKey = (sample: StoredSample) => + `${sample.replicaId}|${sample.occurredAt}`; +const bucketKey = (row: BinnedSubscribeRow) => + `${row.replicaId}|${row.bucketStart}`; + +const samplesByCluster = createScopedSubscribeCollections({ + id: "cluster-utilization-3h", + getKey: sampleKey, +}); +const bucketsByCluster = createScopedSubscribeCollections({ + id: "cluster-utilization-24h", + getKey: bucketKey, +}); + +/** + * Live utilization for a cluster over windows of 24h or less, streamed from the + * indexed 3h or 24h view into a per-cluster TanStack DB collection. Each + * subscribe covers its view's whole retention, so switching between windows + * that one view serves reuses the rows instead of resubscribing. Returns the + * same shape as `useReplicaUtilizationHistory`. + */ +export function useLiveClusterUtilization({ + clusterId, + timePeriodMinutes, + bucketSizeMs, + tier, + includeMemoryBreakdown, +}: { + clusterId: string; + timePeriodMinutes: number; + bucketSizeMs: number; + tier: "unbinned3h" | "binned24h" | undefined; + includeMemoryBreakdown: boolean; +}) { + const lineage = useClusterLineageIds([clusterId], tier !== undefined); + const lineageIdsKey = lineage.data?.join(",") ?? ""; + + const samplesSubscribe = React.useMemo(() => { + const clusterIds = lineageIdsKey ? lineageIdsKey.split(",") : []; + return tier === "unbinned3h" && clusterIds.length > 0 + ? buildConsoleClusterUtilizationUnbinned3hSubscribe( + clusterIds, + subMinutes(new Date(), LIVE_TIER_RETENTION_MINUTES.unbinned3h), + includeMemoryBreakdown, + ) + : undefined; + }, [tier, lineageIdsKey, includeMemoryBreakdown]); + const samples = useScopedSubscribeCollection( + samplesByCluster, + clusterId, + samplesSubscribe, + ); + + const bucketsSubscribe = React.useMemo(() => { + const clusterIds = lineageIdsKey ? lineageIdsKey.split(",") : []; + return tier === "binned24h" && clusterIds.length > 0 + ? buildConsoleClusterUtilizationOverview24hSubscribe( + clusterIds, + subMinutes(new Date(), LIVE_TIER_RETENTION_MINUTES.binned24h), + includeMemoryBreakdown, + ) + : undefined; + }, [tier, lineageIdsKey, includeMemoryBreakdown]); + const buckets = useScopedSubscribeCollection( + bucketsByCluster, + clusterId, + bucketsSubscribe, + ); + + const active = + tier === "unbinned3h" + ? samples + : tier === "binned24h" + ? buckets + : undefined; + const collection = active?.collection; + const { data: rows } = useLiveQuery( + (q) => (collection ? q.from({ row: collection.collection }) : undefined), + [collection], + ); + const snapshotComplete = active?.status.snapshotComplete ?? false; + const isRefreshing = active?.status.isRefreshing ?? false; + + const data = React.useMemo(() => { + if (!collection || !snapshotComplete || !rows) return undefined; + return transformLiveUtilization({ + rows: + tier === "unbinned3h" + ? { tier, samples: rows as StoredSample[] } + : { tier: "binned24h", buckets: rows as BinnedSubscribeRow[] }, + endDate: new Date(), + timePeriodMinutes, + bucketSizeMs, + }); + }, [ + collection, + snapshotComplete, + rows, + tier, + timePeriodMinutes, + bucketSizeMs, + ]); + + // Held rows can have aged out of the window entirely. Until the new + // snapshot lands that would read as a window with no usage. + const isAwaitingRows = isRefreshing && data?.graphData.length === 0; + return { + data: isAwaitingRows ? undefined : data, + isLoading: + collection !== undefined && (!snapshotComplete || isAwaitingRows), + isError: + Boolean(active?.status.error) || + lineage.isLoadingError || + samples.isScopeError, + isRefreshing, + }; +} diff --git a/console/src/platform/clusters/ClusterPage/ResourceUsage/useResourceUsageData.ts b/console/src/platform/clusters/ClusterPage/ResourceUsage/useResourceUsageData.ts index 20a963c5dac54..050bf7b9f9e92 100644 --- a/console/src/platform/clusters/ClusterPage/ResourceUsage/useResourceUsageData.ts +++ b/console/src/platform/clusters/ClusterPage/ResourceUsage/useResourceUsageData.ts @@ -20,11 +20,12 @@ import { replicaStatusesUpdatedAt, useObjectCreationTimes, useObjectHydrationStretches, + usePolledReplicaUtilizationHistory, useReplicaHeapLimits, useReplicaHydrationEpisodes, useReplicaStatusHistory, - useReplicaUtilizationHistory, useUnhydratedObjectCounts, + useUtilizationTier, } from "~/platform/clusters/queries"; import { useClusterObjectsLive } from "~/store/allObjectsCollection"; import { useEnvironmentGate } from "~/store/environments"; @@ -42,6 +43,7 @@ import { transformDdlEvents, transformReplicaTimeline, } from "./resourceUsageModel"; +import { useLiveClusterUtilization } from "./useLiveClusterUtilization"; const DDL_OBJECT_TYPES = ["index", "materialized-view"]; const EMPTY_SERIES: ReplicaData[] = []; @@ -92,17 +94,36 @@ export function useResourceUsageData({ timePeriodMinutes: number; }) { const { colors } = useTheme(); - // The status timeline reads offline stretches from status history. - const history = useReplicaUtilizationHistory( + const bucketSizeMs = Math.max(timePeriodMinutes * 1000, MIN_BUCKET_SIZE_MS); + const tier = useUtilizationTier(timePeriodMinutes); + const hasMemoryBreakdown = + useEnvironmentGate(REPLICA_MEMORY_BREAKDOWN_MIN_VERSION) === true; + const live = useLiveClusterUtilization({ + clusterId: cluster.id, + timePeriodMinutes, + bucketSizeMs, + tier: tier === "poll" ? undefined : tier, + includeMemoryBreakdown: hasMemoryBreakdown, + }); + const polled = usePolledReplicaUtilizationHistory( { - bucketSizeMs: Math.max(timePeriodMinutes * 1000, MIN_BUCKET_SIZE_MS), + bucketSizeMs, timePeriodMinutes, clusterIds: [cluster.id], replicaId: undefined, includeMemoryBreakdown: true, }, - { includeOfflineEvents: false }, + { enabled: tier === "poll" }, ); + const history = + tier === "poll" + ? { + data: polled.data, + isLoading: polled.isLoading, + isError: polled.isError, + isRefreshing: false, + } + : live; const graphData = history.data?.graphData; // Replicas that share a name, such as a scheduled cluster's refreshes, share @@ -199,10 +220,8 @@ export function useResourceUsageData({ // A sample without a heap limit has `heap_percent` as a share of the memory // allocation instead, and no RAM limit share. Only rows on 26.44 and later // carry that share, so older environments keep the heap labels. - const rowsCarryHeapLimit = - useEnvironmentGate(REPLICA_MEMORY_BREAKDOWN_MIN_VERSION) === true; const hasHeapLimit = - !rowsCarryHeapLimit || + !hasMemoryBreakdown || chartData.some(({ data }) => data.some((point) => point.ramLimitPercent !== null), ); diff --git a/console/src/platform/clusters/ClusterPage/memoryByObjectModel.ts b/console/src/platform/clusters/ClusterPage/memoryByObjectModel.ts index 7e178aa276e00..0827b515d7e01 100644 --- a/console/src/platform/clusters/ClusterPage/memoryByObjectModel.ts +++ b/console/src/platform/clusters/ClusterPage/memoryByObjectModel.ts @@ -12,6 +12,39 @@ import { clamp } from "~/util"; /** How many of the largest objects the card shows. */ export const TOP_OBJECT_COUNT = 10; +/** One object on the card, from the polled or the streamed sizes. */ +export interface LargestObject { + id: string | null; + name: string | null; + /** Bytes: a bigint when polled, a number when streamed. */ + size: bigint | number | null; + memoryPercentage: number | null; + type: string | null; + schemaName: string | null; + databaseName: string | null; + dataflowId: string | null; + dataflowName: string | null; + isOrphanedDataflow: boolean; +} + +/** + * Heap in use beyond the listed objects' sizes, with its share of the heap + * limit. Undefined if their sizes add up to at least `heapUsedBytes`. + */ +export const otherMemory = ( + heapUsedBytes: number, + heapLimit: number, + objects: Array>, +) => { + const size = objects.reduce( + (rest, object) => rest - Number(object.size ?? 0), + heapUsedBytes, + ); + return size > 0 + ? { size, memoryPercentage: (size / heapLimit) * 100 } + : undefined; +}; + /** * Bar segment widths for objects' shares of the heap limit, in order. The bar * stops at the heap limit even if the shares add up to more. diff --git a/console/src/platform/clusters/ClusterPage/useLiveLargestObjects.test.ts b/console/src/platform/clusters/ClusterPage/useLiveLargestObjects.test.ts new file mode 100644 index 0000000000000..4a8379a229985 --- /dev/null +++ b/console/src/platform/clusters/ClusterPage/useLiveLargestObjects.test.ts @@ -0,0 +1,97 @@ +// 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 { renderHook, waitFor } from "@testing-library/react"; + +import { ReplicaObjectSize } from "~/api/materialize/cluster/largestMaintainedQueries"; +import { DatabaseObject } from "~/api/materialize/objects"; +import { createSubscribeCollection } from "~/api/materialize/subscribeCollection"; +import { allObjectsCollection } from "~/store/allObjectsCollection"; +import { createProviderWrapper } from "~/test/utils"; + +import { useLargestObjects } from "./useLiveLargestObjects"; + +const MIB = 2 ** 20; +const HEAP_LIMIT = 1000 * MIB; + +const catalogObject = (id: string, name: string): DatabaseObject => ({ + databaseName: "materialize", + databaseId: "u1", + name, + schemaName: "public", + schemaId: "u2", + id, + objectType: "index", + sourceType: null, + isWebhookTable: null, + clusterId: "u9", + clusterName: "prod", +}); + +const renderLargestObjects = async (sizes: ReplicaObjectSize[]) => { + const collection = createSubscribeCollection({ + id: `test-sizes-${Math.random()}`, + getKey: (row) => row.objectId, + }); + collection.applySnapshot({ + data: sizes, + snapshotComplete: true, + error: undefined, + }); + const ProviderWrapper = await createProviderWrapper(); + return renderHook(() => useLargestObjects(collection, HEAP_LIMIT), { + wrapper: ProviderWrapper, + }); +}; + +describe("useLargestObjects", () => { + beforeEach(() => { + allObjectsCollection.applySnapshot({ + data: [ + catalogObject("u1", "orders_idx"), + catalogObject("u2", "items_idx"), + ], + snapshotComplete: true, + error: undefined, + }); + }); + + it("names the largest objects from the catalog, largest first", async () => { + const { result } = await renderLargestObjects([ + { objectId: "u1", size: 100 * MIB }, + { objectId: "u2", size: 300 * MIB }, + { objectId: "u3", size: null }, + ]); + + await waitFor(() => + expect( + result.current.objects.map(({ name, memoryPercentage }) => [ + name, + memoryPercentage, + ]), + ).toEqual([ + ["items_idx", 30], + ["orders_idx", 10], + ["u3", null], + ]), + ); + }); + + it("keeps only the ten largest", async () => { + const { result } = await renderLargestObjects( + Array.from({ length: 12 }, (_, index) => ({ + objectId: `u${100 + index}`, + size: (index + 1) * MIB, + })), + ); + + await waitFor(() => expect(result.current.objects).toHaveLength(10)); + expect(result.current.objects.at(-1)?.id).toEqual("u102"); + }); +}); diff --git a/console/src/platform/clusters/ClusterPage/useLiveLargestObjects.ts b/console/src/platform/clusters/ClusterPage/useLiveLargestObjects.ts new file mode 100644 index 0000000000000..ad8a31e61543b --- /dev/null +++ b/console/src/platform/clusters/ClusterPage/useLiveLargestObjects.ts @@ -0,0 +1,107 @@ +// 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 { eq, useLiveQuery } from "@tanstack/react-db"; +import { useAtomValue } from "jotai"; +import React from "react"; + +import { + buildReplicaObjectSizesSubscribe, + ReplicaObjectSize, +} from "~/api/materialize/cluster/largestMaintainedQueries"; +import { + createScopedSubscribeCollections, + idleCollectionStatusAtom, + SubscribeCollection, +} from "~/api/materialize/subscribeCollection"; +import { useScopedSubscribeCollection } from "~/api/materialize/useScopedSubscribeCollection"; +import { allObjectsCollection } from "~/store/allObjectsCollection"; + +import { LargestObject, TOP_OBJECT_COUNT } from "./memoryByObjectModel"; + +/** Each replica's streamed object sizes. */ +export const sizesByReplica = + createScopedSubscribeCollections({ + id: "replica-object-sizes", + getKey: (row) => row.objectId, + }); + +/** + * The largest objects in `sizes`, named from `allObjectsCollection`. An object + * missing from the catalog is an orphaned dataflow, named by its id. + */ +export function useLargestObjects( + sizes: SubscribeCollection | undefined, + heapLimit: number | null, +) { + const { data: rows } = useLiveQuery( + (q) => + sizes + ? q + .from({ size: sizes.collection }) + .leftJoin( + { object: allObjectsCollection.collection }, + ({ size, object }) => eq(size.objectId, object.id), + ) + .orderBy(({ size }) => size.size, { + direction: "desc", + nulls: "last", + }) + // Sizes are 10 MiB-quantized, so ties are common. + .orderBy(({ size }) => size.objectId) + .limit(TOP_OBJECT_COUNT) + : undefined, + [sizes], + ); + const status = useAtomValue(sizes?.statusAtom ?? idleCollectionStatusAtom); + const objectsStatus = useAtomValue(allObjectsCollection.statusAtom); + const objects = (rows ?? []).map(({ size, object }): LargestObject => ({ + id: size.objectId, + name: object?.name ?? size.objectId, + size: size.size, + // A replica that reports no heap limit has no share to show. + memoryPercentage: + size.size === null || !heapLimit ? null : (size.size / heapLimit) * 100, + type: object?.objectType ?? null, + schemaName: object?.schemaName ?? null, + databaseName: object?.databaseName ?? null, + dataflowId: null, + dataflowName: null, + isOrphanedDataflow: !object, + })); + return { + objects, + // Names come from the catalog, so without it every object would read as + // an orphaned dataflow. + snapshotComplete: status.snapshotComplete && objectsStatus.snapshotComplete, + isRefreshing: status.isRefreshing, + isError: Boolean(status.error) || Boolean(objectsStatus.error), + }; +} + +/** The largest objects on a replica, streamed into a per-replica collection. */ +export function useLiveLargestObjects({ + replicaId, + heapLimit, +}: { + replicaId: string; + heapLimit: number | null; +}) { + const subscribe = React.useMemo( + () => buildReplicaObjectSizesSubscribe(replicaId), + [replicaId], + ); + const { collection, isScopeError } = useScopedSubscribeCollection( + sizesByReplica, + replicaId, + subscribe, + ); + const largest = useLargestObjects(collection, heapLimit); + return { ...largest, isError: largest.isError || isScopeError }; +} diff --git a/console/src/platform/clusters/LargestMaintainedQueries.tsx b/console/src/platform/clusters/LargestMaintainedQueries.tsx index 993e67ad1486f..ee7d30811b94b 100644 --- a/console/src/platform/clusters/LargestMaintainedQueries.tsx +++ b/console/src/platform/clusters/LargestMaintainedQueries.tsx @@ -119,6 +119,7 @@ const LargestMaintainedQueriesError = ({ }; export interface LargestReplica { + id: string; name: string; /** Null while the replica reports none, as always under the process orchestrator. */ heapLimit: number | null; @@ -126,8 +127,8 @@ export interface LargestReplica { export interface LargestReplicaBoundaryProps { clusterId: string; - /** The replica to load instead of the cluster's largest. */ - replicaName?: string; + /** A replica to show instead of the cluster's largest, which then isn't queried. */ + replica?: LargestReplica; children: (replica: LargestReplica) => React.ReactNode; /** Wraps the loading and error states, so a caller can draw them in its card. */ wrapFallback?: (fallback: React.ReactNode) => React.ReactNode; @@ -136,13 +137,13 @@ export interface LargestReplicaBoundaryProps { const identity = (fallback: React.ReactNode) => fallback; /** - * Loads the cluster's largest replica and renders `children` with it, behind - * the loading and error states of the per-object memory insights. Renders - * nothing for a cluster without replicas. + * Loads the cluster's largest replica, unless given one, and renders + * `children` with it, behind the loading and error states of the per-object + * memory insights. Renders nothing for a cluster without replicas. */ export const LargestReplicaBoundary = ({ clusterId, - replicaName, + replica, children, wrapFallback = identity, }: LargestReplicaBoundaryProps) => { @@ -155,7 +156,7 @@ export const LargestReplicaBoundary = ({ )}> {children} @@ -167,37 +168,52 @@ export const LargestReplicaBoundary = ({ const LargestReplicaLoader = ({ clusterId, - replicaName, + replica, children, wrapFallback, -}: Omit, "replicaName"> & - Pick) => { - const { data: largestReplica } = useLargestClusterReplica({ - clusterId, - replicaName, - }); +}: Omit, "replica"> & + Pick) => { + const { data: largestReplica, error } = useLargestClusterReplica( + { clusterId }, + { enabled: replica === undefined }, + ); + const shown = + replica ?? + (largestReplica && { + id: largestReplica.id, + name: largestReplica.name, + heapLimit: + largestReplica.heapLimit === null + ? null + : Number(largestReplica.heapLimit), + }); - // Don't show anything if the cluster has no replicas - if (!largestReplica) return null; + if (!shown) { + // Don't show anything if the cluster has no replicas + if (largestReplica === null) return null; + return wrapFallback( + error ? ( + + ) : ( + + ), + ); + } + // NOTE: `children` renders at the same place whether or not `replica` is + // given, so a switch between a given replica and the largest keeps its state. return ( + renderFallback={({ error: childError }) => wrapFallback( , ) } > - {children({ - name: largestReplica.name, - heapLimit: - largestReplica.heapLimit === null - ? null - : Number(largestReplica.heapLimit), - })} + {children(shown)} ); }; diff --git a/console/src/platform/clusters/queries.test.ts b/console/src/platform/clusters/queries.test.ts index 8f6e6fd39870d..7893250957d2e 100644 --- a/console/src/platform/clusters/queries.test.ts +++ b/console/src/platform/clusters/queries.test.ts @@ -44,6 +44,7 @@ import { useFreshnessObjects, useObjectHydrationStretches, useOwners, + useReplicaHeapBytes, useReplicaHeapLimits, useReplicaOfflineEvents, useReplicaStatusHistory, @@ -396,6 +397,95 @@ describe("useReplicaHeapLimits", () => { }); }); +describe("useReplicaHeapBytes", () => { + let requests: number; + + const answerHeapMetrics = ( + rows: Array<{ + replicaId: string; + processId: string; + heapBytes: string | null; + }>, + ) => { + requests = 0; + server.use( + http.post("*/api/sql", ({ request }) => { + const queryKey = JSON.parse( + new URL(request.url).searchParams.get("query_key") ?? "[]", + ); + if (queryKey[1]?.scope !== "replicaHeapLimits") return undefined; + requests += 1; + return HttpResponse.json({ + results: [ + mapKyselyToTabular({ + rows: rows.map((row) => ({ + ...row, + heapLimit: "1073741824", + memoryBytes: "1073741824", + })), + columns: buildColumns([ + "replicaId", + "processId", + "heapLimit", + "memoryBytes", + "heapBytes", + ]), + }), + ], + }); + }), + ); + }; + + afterEach(() => { + vi.useRealTimers(); + }); + + it("sums each replica's processes, leaving out a replica with a process that reports none", async () => { + answerHeapMetrics([ + { replicaId: "u1", processId: "0", heapBytes: "400" }, + { replicaId: "u1", processId: "1", heapBytes: null }, + { replicaId: "u2", processId: "0", heapBytes: "500" }, + { replicaId: "u2", processId: "1", heapBytes: "600" }, + ]); + const ProviderWrapper = await createProviderWrapper(); + const { result } = renderHook(() => useReplicaHeapBytes(["u1", "u2"]), { + wrapper: ProviderWrapper, + }); + + await waitFor(() => + expect(result.current.data).toEqual(new Map([["u2", 1100]])), + ); + }); + + it("keeps polling once every heap limit is in", async () => { + answerHeapMetrics([{ replicaId: "u1", processId: "0", heapBytes: "400" }]); + vi.useFakeTimers({ + toFake: [ + "Date", + "setTimeout", + "clearTimeout", + "setInterval", + "clearInterval", + ], + }); + const ProviderWrapper = await createProviderWrapper(); + renderHook( + () => { + useReplicaHeapLimits(["u1"]); + useReplicaHeapBytes(["u1"]); + }, + { wrapper: ProviderWrapper }, + ); + await vi.advanceTimersByTimeAsync(1_000); + expect(requests).toBe(1); + + await vi.advanceTimersByTimeAsync(5 * millisecondsInMinute); + + expect(requests).toBeGreaterThan(1); + }); +}); + describe("replicaStatusesUpdatedAt", () => { it("reads the latest change of any replica's process, system replicas included", () => { const status = ( diff --git a/console/src/platform/clusters/queries.ts b/console/src/platform/clusters/queries.ts index 8a42dfab946a4..0887557b3e791 100644 --- a/console/src/platform/clusters/queries.ts +++ b/console/src/platform/clusters/queries.ts @@ -96,6 +96,7 @@ import { attachOfflineEvents, BinnedSubscribeRow, bucketRowsToBucketsByReplicaId, + clipBinnedRowsToWindow, parseBinnedSubscribeRow, rebucketUtilizationSamples, toReplicaUtilizationGraphData, @@ -474,8 +475,12 @@ export function useIndexesList(filters: ListFilters) { }); } -export function useLargestClusterReplica(params: LargestClusterReplicaParams) { - return useSuspenseQuery({ +export function useLargestClusterReplica( + params: LargestClusterReplicaParams, + { enabled = true }: { enabled?: boolean } = {}, +) { + return useQuery({ + enabled, refetchInterval: 60_000, queryKey: clusterQueryKeys.largestClusterReplica(params), queryFn: async ({ queryKey, signal }) => { @@ -493,6 +498,8 @@ export function useLargestClusterReplica(params: LargestClusterReplicaParams) { const STRIP_DATAFLOW_PREFIX = /^Dataflow: /; +export const UNIFIED_OBJECT_SIZES_MIN_VERSION = "26.35.0-dev"; + export type UseLargestMaintainedQueriesParams = { clusterId: string; clusterName: string; @@ -506,7 +513,8 @@ export function useLargestMaintainedQueries( ) { // mz_object_arrangement_sizes reports sizes reliably from v26.35, where // they no longer go stale across replica restarts. - const unifiedSizes = useEnvironmentGate("26.35.0-dev") ?? false; + const unifiedSizes = + useEnvironmentGate(UNIFIED_OBJECT_SIZES_MIN_VERSION) ?? false; const queryClient = useQueryClient(); // queryClient is a stable singleton, not query input. // eslint-disable-next-line @tanstack/query/exhaustive-deps @@ -746,6 +754,11 @@ export function useMaterializationLag(params: MaterializationLagParams) { const SUBSCRIBE_UNBINNED_MAX_MINUTES = 180; // 3h: live un-binned base, client-binned const SUBSCRIBE_BINNED_MAX_MINUTES = 1440; // 24h: live 5-min binned view export const OVERVIEW_MAX_MINUTES = 20160; // 14d: polled overview view; beyond, ad-hoc +/** How far back each live tier's subscribe reaches. */ +export const LIVE_TIER_RETENTION_MINUTES = { + unbinned3h: SUBSCRIBE_UNBINNED_MAX_MINUTES, + binned24h: SUBSCRIBE_BINNED_MAX_MINUTES, +}; // A blue-green cutover changes a cluster's lineage. Refetching on this interval // bounds how long that takes to reach an open chart. @@ -864,7 +877,6 @@ export function useUtilizationTier(timePeriodMinutes: number) { function useReplicaUtilizationHistorySubscribe( params: ReplicaUtilizationHistoryFilters, enabled: boolean, - includeOfflineEvents: boolean, ) { const { replicaId, timePeriodMinutes, bucketSizeMs, includeMemoryBreakdown } = params; @@ -901,7 +913,7 @@ function useReplicaUtilizationHistorySubscribe( const offlineEvents = useReplicaOfflineEvents( data, timePeriodMinutes, - enabled && includeOfflineEvents, + enabled, ); const result = useMemo(() => { @@ -969,15 +981,10 @@ function useReplicaUtilizationHistoryBinnedSubscribe( const endDate = new Date(); const startDate = subMinutes(endDate, timePeriodMinutes); - // Clip to the window like the SQL does. Held rows from a previous wider - // window would otherwise stretch the chart domain past the selected range. - const rows = data.filter( - (row) => - row.bucketStart.getTime() >= startDate.getTime() && - (!replicaId || row.replicaId === replicaId), + const rows = clipBinnedRowsToWindow( + replicaId ? data.filter((row) => row.replicaId === replicaId) : data, + startDate, ); - // ENVELOPE UPSERT yields an unordered keyed set; the chart needs time order. - rows.sort((a, b) => a.bucketStart.getTime() - b.bucketStart.getTime()); return toReplicaUtilizationGraphData( bucketRowsToBucketsByReplicaId(rows), @@ -994,45 +1001,31 @@ function useReplicaUtilizationHistoryBinnedSubscribe( }; } -export function useReplicaUtilizationHistory( - params: ReplicaUtilizationHistoryFilters, - queryOptions?: { - enabled?: boolean; - // `false` skips the <=3h windows' offline-events poll, for callers that - // read replica status from elsewhere. Their buckets carry no events then. - includeOfflineEvents?: boolean; - }, -) { - const enabled = queryOptions?.enabled ?? true; - const tier = useUtilizationTier(params.timePeriodMinutes); +function useUtilizationFilters(params: ReplicaUtilizationHistoryFilters) { const hasMemoryBreakdown = useEnvironmentGate(REPLICA_MEMORY_BREAKDOWN_MIN_VERSION) === true; - const filters = { + return { ...params, // `false` would change the query key for callers that don't pass this. includeMemoryBreakdown: (params.includeMemoryBreakdown && hasMemoryBreakdown) || undefined, }; +} - const useUnbinnedSubscribe = tier === "unbinned3h"; - const useBinnedSubscribe = tier === "binned24h"; - - const unbinnedResult = useReplicaUtilizationHistorySubscribe( - filters, - enabled && useUnbinnedSubscribe, - queryOptions?.includeOfflineEvents ?? true, - ); - const binnedResult = useReplicaUtilizationHistoryBinnedSubscribe( - filters, - enabled && useBinnedSubscribe, - ); - - const queryResult = useQuery({ +/** + * Polled utilization, for windows no subscribe serves. Unlike + * `useReplicaUtilizationHistory` it mounts no subscribe hooks, so it opens no + * WebSocket (see the NOTE on `useReplicaUtilizationHistorySubscribe`). + */ +export function usePolledReplicaUtilizationHistory( + params: ReplicaUtilizationHistoryFilters, + { enabled }: { enabled: boolean }, +) { + const filters = useUtilizationFilters(params); + return useQuery({ queryKey: clusterQueryKeys.replicaUtilizationHistory(filters), refetchInterval: 20_000, - // Poll whatever a subscribe isn't serving: >24h on new mz, and every window - // on old mz (where the subscribes are gated off). - enabled: enabled && !useUnbinnedSubscribe && !useBinnedSubscribe, + enabled, queryFn: async ({ queryKey, signal }) => { const [, queryKeyParams] = queryKey; @@ -1056,6 +1049,35 @@ export function useReplicaUtilizationHistory( return toReplicaUtilizationGraphData(data, startDate, endDate); }, }); +} + +export function useReplicaUtilizationHistory( + params: ReplicaUtilizationHistoryFilters, + queryOptions?: { + enabled?: boolean; + }, +) { + const enabled = queryOptions?.enabled ?? true; + const tier = useUtilizationTier(params.timePeriodMinutes); + const filters = useUtilizationFilters(params); + + const useUnbinnedSubscribe = tier === "unbinned3h"; + const useBinnedSubscribe = tier === "binned24h"; + + const unbinnedResult = useReplicaUtilizationHistorySubscribe( + filters, + enabled && useUnbinnedSubscribe, + ); + const binnedResult = useReplicaUtilizationHistoryBinnedSubscribe( + filters, + enabled && useBinnedSubscribe, + ); + + const queryResult = usePolledReplicaUtilizationHistory(params, { + // Poll whatever a subscribe isn't serving: >24h on new mz, and every window + // on old mz (where the subscribes are gated off). + enabled: enabled && !useUnbinnedSubscribe && !useBinnedSubscribe, + }); const active = useUnbinnedSubscribe ? unbinnedResult @@ -1353,14 +1375,32 @@ export function useRollingWindowStartMs(windowMinutes: number | undefined) { return start?.windowMs === windowMs ? start?.startMs : undefined; } +/** The replicas' latest per-process heap metrics, one read for heap limits and use. */ +const replicaHeapMetricsQuery = (replicaIds: string[]) => ({ + queryKey: clusterQueryKeys.replicaHeapLimits({ replicaIds }), + enabled: replicaIds.length > 0, + queryFn: ({ + queryKey, + signal, + }: QueryFunctionContext< + ReturnType + >) => { + const [, queryKeyParams] = queryKey; + return fetchReplicaHeapLimits({ + params: queryKeyParams, + queryKey, + requestOptions: { signal }, + }); + }, +}); + /** * Each replica's per-process heap limit in bytes, for replicas reporting one. * Idle while `replicaIds` is empty. */ export function useReplicaHeapLimits(replicaIds: string[]) { return useQuery({ - queryKey: clusterQueryKeys.replicaHeapLimits({ replicaIds }), - enabled: replicaIds.length > 0, + ...replicaHeapMetricsQuery(replicaIds), // A replica's heap limit is fixed by its size, so polling stops once each // replica has a row with one, or with memory but none (an orchestrator // without heap limits, or a failed usage fetch on Kubernetes). A failed @@ -1374,18 +1414,38 @@ export function useReplicaHeapLimits(replicaIds: string[]) { ); return replicaIds.every((id) => settled.has(id)) ? false : 60_000; }, - queryFn: ({ queryKey, signal }) => { - const [, queryKeyParams] = queryKey; - return fetchReplicaHeapLimits({ - params: queryKeyParams, - queryKey, - requestOptions: { signal }, - }); - }, select: toHeapLimitBytesByReplicaId, }); } +const toHeapBytesByReplicaId = ( + rows: Awaited>, +) => { + const bytes = new Map(); + const unreported = new Set(); + for (const { replicaId, heapBytes } of rows) { + if (heapBytes === null) unreported.add(replicaId); + else bytes.set(replicaId, (bytes.get(replicaId) ?? 0) + Number(heapBytes)); + } + for (const replicaId of unreported) bytes.delete(replicaId); + return bytes; +}; + +/** + * Each replica's heap use in bytes, summed over its processes, for replicas + * whose every process reports one. Idle while `replicaIds` is empty. + * + * NOTE: shares its read with `useReplicaHeapLimits`, so while this is mounted + * that read keeps polling after every heap limit is in. + */ +export function useReplicaHeapBytes(replicaIds: string[]) { + return useQuery({ + ...replicaHeapMetricsQuery(replicaIds), + refetchInterval: 60_000, + select: toHeapBytesByReplicaId, + }); +} + /** Creation times of the given objects over the window. Idle while `objectIds` is empty. */ export function useObjectCreationTimes(params: { objectIds: string[]; diff --git a/console/src/test/mockSubscribeSocket.ts b/console/src/test/mockSubscribeSocket.ts index 2ac4a1646f724..71be64c0668bd 100644 --- a/console/src/test/mockSubscribeSocket.ts +++ b/console/src/test/mockSubscribeSocket.ts @@ -27,19 +27,20 @@ const METADATA_COLUMNS = [ /** * Answers each SUBSCRIBE sent over the WebSocket with one snapshot of `upsert` * rows and the progress row that completes it, then stays silent. - * `snapshotFor` picks a query's rows from its SQL and parameters. A query it - * returns undefined for never completes, like a subscribe still loading. + * `snapshotFor` picks a query's rows from its SQL and parameters, or returns a + * promise of them to hold the snapshot back. A query it returns undefined for + * never completes, like a subscribe still loading. */ export function mockSubscribeSocket( snapshotFor: ( sql: string, params: unknown[], - ) => SubscribeSnapshot | undefined, + ) => SubscribeSnapshot | undefined | Promise, ) { return ws.link(/^wss?:\/\//).addEventListener("connection", ({ client }) => { let hasOptions = false; const send = (message: unknown) => client.send(JSON.stringify(message)); - client.addEventListener("message", (event) => { + client.addEventListener("message", async (event) => { // The first message carries the session options. if (!hasOptions) { hasOptions = true; @@ -50,7 +51,10 @@ export function mockSubscribeSocket( queries: Array<{ query: string; params?: unknown[] }>; }; const [query] = request.queries; - const snapshot = snapshotFor(query?.query ?? "", query?.params ?? []); + const snapshot = await snapshotFor( + query?.query ?? "", + query?.params ?? [], + ); if (!snapshot) return; const progressRow = (timestamp: string) => [ timestamp,