From eaa34e367e95fc88148474df62c83a194d2df1d3 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. --- .../cluster/largestClusterReplica.ts | 1 + .../cluster/largestMaintainedQueries.ts | 24 ++ .../cluster/replicaUtilizationBinning.ts | 14 ++ .../materialize/subscribeCollection.test.ts | 32 +++ .../api/materialize/subscribeCollection.ts | 75 +++++- .../src/api/materialize/subscribeSession.ts | 8 + .../src/api/materialize/useSubscribe.test.tsx | 28 +++ .../ClusterPage/MemoryByObject.test.tsx | 143 ++++++++++- .../clusters/ClusterPage/MemoryByObject.tsx | 221 +++++++++++------ .../ResourceUsage/ResourceUsage.test.tsx | 55 +++++ .../useLiveClusterUtilization.test.ts | 228 ++++++++++++++++++ .../useLiveClusterUtilization.ts | 210 ++++++++++++++++ .../ResourceUsage/useResourceUsageData.ts | 32 ++- .../ClusterPage/memoryByObjectModel.ts | 15 ++ .../ClusterPage/useLiveLargestObjects.test.ts | 97 ++++++++ .../ClusterPage/useLiveLargestObjects.ts | 124 ++++++++++ .../clusters/LargestMaintainedQueries.tsx | 2 + console/src/platform/clusters/queries.ts | 98 +++++--- 18 files changed, 1280 insertions(+), 127 deletions(-) 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/src/api/materialize/cluster/largestClusterReplica.ts b/console/src/api/materialize/cluster/largestClusterReplica.ts index 91bb9f95aa732..b47c8b80d15d1 100644 --- a/console/src/api/materialize/cluster/largestClusterReplica.ts +++ b/console/src/api/materialize/cluster/largestClusterReplica.ts @@ -38,6 +38,7 @@ export function buildLargestClusterReplicaQuery(clusterId: string) { ) .leftJoin("mz_cluster_replica_sizes as crs", "crs.size", "cr.size") .select([ + "cr.id", "cr.name", "cr.size", sql`${replicaHeapLimit}::text`.as("heapLimit"), diff --git a/console/src/api/materialize/cluster/largestMaintainedQueries.ts b/console/src/api/materialize/cluster/largestMaintainedQueries.ts index 16a646bd7a633..a06bcc728e3b7 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; @@ -162,6 +163,29 @@ 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", + 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/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..e5c96b8c5ff86 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,22 @@ 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("surfaces error and snapshotComplete on the status atom", () => { const { applySnapshot, statusAtom } = freshCollection(); @@ -302,3 +319,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..526251b00c17e 100644 --- a/console/src/api/materialize/subscribeCollection.ts +++ b/console/src/api/materialize/subscribeCollection.ts @@ -57,6 +57,9 @@ 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; } export interface SubscribeCollection { @@ -106,6 +109,7 @@ export function createSubscribeCollection(options: { const statusAtom = atom({ error: undefined, snapshotComplete: false, + isRefreshing: false, }); // Captured when the collection starts syncing; cleared when it pauses. @@ -214,7 +218,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 +232,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 +249,7 @@ export function createSubscribeCollection(options: { writeStatus({ error: state.error, snapshotComplete: state.snapshotComplete, + isRefreshing: false, }); }; @@ -296,7 +306,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 +327,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 +351,51 @@ export function createSubscribeCollection(options: { suspendPersistence, }; } + +export interface ScopedSubscribeCollections { + /** The collection for `key` under `scope`, created on first use. */ + get: (scope: string, key: string) => SubscribeCollection; + /** A session target while the scope resolves. It is shared by every key, so + * nothing may subscribe into it. */ + placeholder: SubscribeCollection; +} + +/** + * 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; + }, + placeholder: createSubscribeCollection({ + id: `${id}|placeholder`, + 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/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/platform/clusters/ClusterPage/MemoryByObject.test.tsx b/console/src/platform/clusters/ClusterPage/MemoryByObject.test.tsx index d7c78fabada67..97a82211a0c62 100644 --- a/console/src/platform/clusters/ClusterPage/MemoryByObject.test.tsx +++ b/console/src/platform/clusters/ClusterPage/MemoryByObject.test.tsx @@ -10,18 +10,31 @@ import { screen, waitFor, within } from "@testing-library/react"; import React from "react"; +import { DatabaseObject } from "~/api/materialize/objects"; import { ErrorCode, MzDataType } from "~/api/materialize/types"; import { + buildColumn, buildColumns, buildSqlQueryHandlerV2, mapKyselyToTabular, } from "~/api/mocks/buildSqlQueryHandler"; import server from "~/api/mocks/server"; +import { getStore } from "~/jotai"; import { clusterQueryKeys } from "~/platform/clusters/queries"; -import { renderComponent } from "~/test/utils"; +import { allObjectsCollection } from "~/store/allObjectsCollection"; +import { syncEngineCacheScopeAtom } from "~/store/syncEngineCache"; +import { mockSubscribeSocket } from "~/test/mockSubscribeSocket"; +import { + healthyEnvironment, + renderComponent, + setFakeEnvironment, +} from "~/test/utils"; +import { assert } from "~/util"; +import { parseDbVersion } from "~/version/api"; import { MemoryByObject } from "./MemoryByObject"; import { segmentWidths } from "./memoryByObjectModel"; +import { sizesByReplica } from "./useLiveLargestObjects"; const flags = vi.hoisted(() => ({ values: {} as Record })); @@ -29,11 +42,12 @@ vi.mock("~/hooks/useFlags", () => ({ useFlags: () => flags.values })); const HEAP_LIMIT = 4069523456; -const largestReplicaHandler = (rows: Array>) => +const largestReplicaHandler = (rows: Array>) => buildSqlQueryHandlerV2({ queryKey: clusterQueryKeys.largestClusterReplica({ clusterId: "u1" }), results: mapKyselyToTabular({ columns: buildColumns([ + "id", "name", "size", { type_oid: MzDataType.numeric, name: "heapLimit" }, @@ -188,6 +202,131 @@ describe("MemoryByObject", () => { }); }); +// 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", +}); + +// Query keys embed the environment version, so set the environment before +// building the handlers that match on them. Each test streams a replica of its +// own: the sizes collections outlive a test. +const renderLiveMemoryByObject = async ({ + replicaId, + heapLimit = String(HEAP_LIMIT), +}: { + replicaId: string; + heapLimit?: string | null; +}) => { + await setFakeEnvironment( + getStore().set, + "aws/us-east-1", + LIVE_SIZES_ENVIRONMENT, + ); + server.use( + largestReplicaHandler([ + { id: replicaId, name: "r1", size: "25cc", heapLimit }, + ]), + mockSubscribeSocket((sql, params) => + sql.includes("mz_object_arrangement_sizes") && params.includes(replicaId) + ? { + columns: [ + buildColumn({ name: "objectId" }), + buildColumn({ name: "size", type_oid: MzDataType.float8 }), + ], + rows: [ + ["u188", String(2 * GIB)], + ["u404", String(GIB)], + ], + } + : undefined, + ), + ); + return renderComponent( + , + { + initializeState: ({ set }) => + setFakeEnvironment(set, "aws/us-east-1", LIVE_SIZES_ENVIRONMENT), + }, + ); +}; + +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(); + // 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( + 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(); + }); +}); + describe("segmentWidths", () => { it("stops the bar at the heap limit", () => { expect(segmentWidths([60, 30, 25, null])).toEqual([60, 30, 10, 0]); diff --git a/console/src/platform/clusters/ClusterPage/MemoryByObject.tsx b/console/src/platform/clusters/ClusterPage/MemoryByObject.tsx index 30749272da02c..c07ee7067163a 100644 --- a/console/src/platform/clusters/ClusterPage/MemoryByObject.tsx +++ b/console/src/platform/clusters/ClusterPage/MemoryByObject.tsx @@ -20,22 +20,32 @@ import { import React from "react"; import { Link as RouterLink } from "react-router-dom"; +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 { + UNIFIED_OBJECT_SIZES_MIN_VERSION, + useLargestMaintainedQueries, +} 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, + segmentWidths, + TOP_OBJECT_COUNT, +} from "./memoryByObjectModel"; import { PERCENT_TICKS } from "./ResourceUsage/resourceUsageStyles"; +import { useLiveLargestObjects } from "./useLiveLargestObjects"; export interface MemoryByObjectProps { clusterId: string; @@ -46,32 +56,37 @@ export interface MemoryByObjectProps { export const MemoryByObject = ({ clusterId, clusterName, -}: MemoryByObjectProps) => ( - - {(replica) => ( - - )} - -); +}: MemoryByObjectProps) => { + const hasLiveSizes = + useEnvironmentGate(UNIFIED_OBJECT_SIZES_MIN_VERSION) === true; + return ( + + {(replica) => + hasLiveSizes ? ( + + ) : ( + + ) + } + + ); +}; -const MemoryByObjectCard = ({ +const PolledMemoryByObject = ({ clusterId, clusterName, replicaName, replicaHeapLimit, }: MemoryByObjectProps & { replicaName: string; replicaHeapLimit: number }) => { - const { colors } = useTheme(); - // The maintained objects page only has routes while its flag is on. - const hasMaintainedObjectsPage = Boolean( - useFlags()["maintained-objects-ui-50"], - ); - const regionSlug = useRegionSlug(); - const workflowGraphPath = useBuildWorkflowGraphPath(); const { data: objects } = useLargestMaintainedQueries({ clusterId, clusterName, @@ -79,8 +94,60 @@ const MemoryByObjectCard = ({ replicaName, replicaHeapLimit, }); + return ( + 0} + /> + ); +}; - if (!objects || objects.length === 0) return null; +const LiveMemoryByObject = ({ + replicaId, + replicaName, + replicaHeapLimit, +}: { + replicaId: string; + replicaName: string; + replicaHeapLimit: number; +}) => { + const { objects, snapshotComplete, isError } = useLiveLargestObjects({ + replicaId, + heapLimit: replicaHeapLimit, + }); + if (isError) { + return ; + } + if (!snapshotComplete) return ; + return ( + 0} + /> + ); +}; + +const MemoryByObjectCard = ({ + objects, + replicaName, + hasHeapLimit, +}: { + objects: LargestObject[]; + replicaName: string; + /** Without a heap limit there are no shares, so the card lists sizes alone. */ + hasHeapLimit: boolean; +}) => { + const { colors } = useTheme(); + // The maintained objects page only has routes while its flag is on. + const hasMaintainedObjectsPage = Boolean( + useFlags()["maintained-objects-ui-50"], + ); + const regionSlug = useRegionSlug(); + const workflowGraphPath = useBuildWorkflowGraphPath(); + + if (objects.length === 0) return null; // Colored by rank like the Freshness page's lines, so the largest objects // are furthest apart in hue. @@ -89,21 +156,21 @@ const MemoryByObjectCard = ({ objects.map((object) => object.memoryPercentage), ); const segments = objects.map((object, index) => { - const { id, name, databaseName, schemaName } = object; + const { id, name, databaseName, schemaName, type } = object; return { key: id ?? object.dataflowId ?? String(index), label: name ?? object.dataflowName ?? "-", - usage: formatMemoryUsage(object), + usage: formatMemoryUsage(hasHeapLimit ? object : { size: object.size }), 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, }), }; }); @@ -124,56 +191,60 @@ const MemoryByObjectCard = ({ : `Top ${TOP_OBJECT_COUNT} Objects by Memory`} - % of heap limit on {replicaName} + {hasHeapLimit + ? `% of heap limit on ${replicaName}` + : `Size on ${replicaName}`} - - - {segments.map( - (segment) => - segment.widthPercent > 0 && ( - - - - ), - )} - - - {PERCENT_TICKS.map((tick) => ( - - {tick}% - - ))} + {hasHeapLimit && ( + + + {segments.map( + (segment) => + segment.widthPercent > 0 && ( + + + + ), + )} + + + {PERCENT_TICKS.map((tick) => ( + + {tick}% + + ))} + - + )} { ).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..dd9bf0c52f4a2 --- /dev/null +++ b/console/src/platform/clusters/ClusterPage/ResourceUsage/useLiveClusterUtilization.test.ts @@ -0,0 +1,228 @@ +// 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 } 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 } 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; + + // 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 = new Date(Date.now() - 70 * millisecondsInMinute); + server.use( + buildSqlQueryHandlerV2({ + queryKey: clusterQueryKeys.deploymentLineage({ + clusterIdsKey: clusterId, + }), + results: mapKyselyToTabular({ + columns: buildColumns([ + "clusterId", + "currentDeploymentClusterId", + "clusterName", + ]), + rows: [], + }), + }), + mockSubscribeSocket((sql) => { + if (!sql.includes("utilization_overview_3h")) return undefined; + subscribe.requests += 1; + return subscribe.serve + ? { + columns: [ + buildColumn({ name: "replicaId" }), + buildColumn({ name: "occurredAt" }), + buildColumn({ + name: "cpuPercent", + type_oid: MzDataType.float8, + }), + ], + rows: [["u1", occurredAt.toISOString(), "0.5"]], + } + : undefined; + }), + ); + return subscribe; + }; + + const renderOneHourWindow = ( + clusterId: string, + wrapper: React.ComponentType, + ) => { + const renders: LiveUtilization[] = []; + const rendered = renderHook( + () => { + const result = useLiveClusterUtilization({ + clusterId, + timePeriodMinutes: 60, + bucketSizeMs: millisecondsInMinute, + tier: "unbinned3h", + includeMemoryBreakdown: false, + }); + renders.push(result); + return result; + }, + { wrapper }, + ); + return { ...rendered, renders }; + }; + + // 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); + }); +}); 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..dbbcc8fe1d2c6 --- /dev/null +++ b/console/src/platform/clusters/ClusterPage/ResourceUsage/useLiveClusterUtilization.ts @@ -0,0 +1,210 @@ +// Copyright Materialize, Inc. and contributors. All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +import { useLiveQuery } from "@tanstack/react-db"; +import { subMinutes } from "date-fns"; +import { useAtomValue } from "jotai"; +import React from "react"; + +import { + 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 { SubscribeRow } from "~/api/materialize/SubscribeManager"; +import { useGlobalSubscribeCollection } from "~/api/materialize/useSubscribe"; +import { + LIVE_TIER_RETENTION_MINUTES, + useClusterLineageIds, +} from "~/platform/clusters/queries"; +import { syncEngineCacheScopeLoadableAtom } from "~/store/syncEngineCache"; + +/** 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 scope = useAtomValue(syncEngineCacheScopeLoadableAtom); + const scopeId = scope.state === "hasData" ? scope.data : undefined; + const samples = scopeId + ? samplesByCluster.get(scopeId, clusterId) + : undefined; + const buckets = scopeId + ? bucketsByCluster.get(scopeId, clusterId) + : undefined; + const lineage = useClusterLineageIds([clusterId], tier !== undefined); + const lineageIdsKey = lineage.data?.join(",") ?? ""; + + const samplesOptions = React.useMemo(() => { + const clusterIds = lineageIdsKey ? lineageIdsKey.split(",") : []; + return { + target: samples ?? samplesByCluster.placeholder, + subscribe: + samples && tier === "unbinned3h" && clusterIds.length > 0 + ? buildConsoleClusterUtilizationUnbinned3hSubscribe( + clusterIds, + subMinutes(new Date(), LIVE_TIER_RETENTION_MINUTES.unbinned3h), + includeMemoryBreakdown, + ) + : undefined, + select: (row: SubscribeRow) => row.data, + upsertKey: (row: SubscribeRow) => sampleKey(row.data), + }; + }, [samples, tier, lineageIdsKey, includeMemoryBreakdown]); + useGlobalSubscribeCollection(samplesOptions); + + const bucketsOptions = React.useMemo(() => { + const clusterIds = lineageIdsKey ? lineageIdsKey.split(",") : []; + return { + target: buckets ?? bucketsByCluster.placeholder, + subscribe: + buckets && tier === "binned24h" && clusterIds.length > 0 + ? buildConsoleClusterUtilizationOverview24hSubscribe( + clusterIds, + subMinutes(new Date(), LIVE_TIER_RETENTION_MINUTES.binned24h), + includeMemoryBreakdown, + ) + : undefined, + select: (row: SubscribeRow) => row.data, + upsertKey: (row: SubscribeRow) => bucketKey(row.data), + }; + }, [buckets, tier, lineageIdsKey, includeMemoryBreakdown]); + useGlobalSubscribeCollection(bucketsOptions); + + const active = + tier === "unbinned3h" + ? samples + : tier === "binned24h" + ? buckets + : undefined; + const { data: rows } = useLiveQuery( + (q) => (active ? q.from({ row: active.collection }) : undefined), + [active], + ); + const status = useAtomValue( + (active ?? samplesByCluster.placeholder).statusAtom, + ); + + const data = React.useMemo(() => { + if (!active || !status.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, + }); + }, [ + active, + status.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 = status.isRefreshing && data?.graphData.length === 0; + return { + data: isAwaitingRows ? undefined : data, + isLoading: + active !== undefined && (!status.snapshotComplete || isAwaitingRows), + isError: + Boolean(status.error) || + lineage.isLoadingError || + scope.state === "hasError", + isRefreshing: status.isRefreshing, + }; +} diff --git a/console/src/platform/clusters/ClusterPage/ResourceUsage/useResourceUsageData.ts b/console/src/platform/clusters/ClusterPage/ResourceUsage/useResourceUsageData.ts index b3b6c5c720798..e3e5a78411b6a 100644 --- a/console/src/platform/clusters/ClusterPage/ResourceUsage/useResourceUsageData.ts +++ b/console/src/platform/clusters/ClusterPage/ResourceUsage/useResourceUsageData.ts @@ -19,11 +19,12 @@ import { OVERVIEW_MAX_MINUTES, REPLICA_MEMORY_BREAKDOWN_MIN_VERSION, useObjectCreationTimes, + usePolledReplicaUtilizationHistory, useReplicaHeapLimits, useReplicaHydrationEpisodes, useReplicaStatusHistory, - useReplicaUtilizationHistory, useUnhydratedObjectCounts, + useUtilizationTier, } from "~/platform/clusters/queries"; import { useClusterObjectsLive } from "~/store/allObjectsCollection"; import { useEnvironmentGate } from "~/store/environments"; @@ -40,6 +41,7 @@ import { transformDdlEvents, transformReplicaTimeline, } from "./resourceUsageModel"; +import { useLiveClusterUtilization } from "./useLiveClusterUtilization"; const DDL_OBJECT_TYPES = ["index", "materialized-view"]; const EMPTY_SERIES: ReplicaData[] = []; @@ -90,8 +92,18 @@ 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), timePeriodMinutes, @@ -99,8 +111,17 @@ export function useResourceUsageData({ 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 @@ -187,8 +208,7 @@ export function useResourceUsageData({ // allocation instead, and no RAM limit share. Only the views' rows on 26.44 // and later carry that share, so other windows keep the heap labels. const rowsCarryHeapLimit = - useEnvironmentGate(REPLICA_MEMORY_BREAKDOWN_MIN_VERSION) === true && - timePeriodMinutes <= OVERVIEW_MAX_MINUTES; + hasMemoryBreakdown && timePeriodMinutes <= OVERVIEW_MAX_MINUTES; const hasHeapLimit = !rowsCarryHeapLimit || chartData.some(({ data }) => diff --git a/console/src/platform/clusters/ClusterPage/memoryByObjectModel.ts b/console/src/platform/clusters/ClusterPage/memoryByObjectModel.ts index 7e178aa276e00..5710a1a81847f 100644 --- a/console/src/platform/clusters/ClusterPage/memoryByObjectModel.ts +++ b/console/src/platform/clusters/ClusterPage/memoryByObjectModel.ts @@ -12,6 +12,21 @@ 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; +} + /** * 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..516a9705278f9 --- /dev/null +++ b/console/src/platform/clusters/ClusterPage/useLiveLargestObjects.ts @@ -0,0 +1,124 @@ +// 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, + SubscribeCollection, +} from "~/api/materialize/subscribeCollection"; +import { SubscribeRow } from "~/api/materialize/SubscribeManager"; +import { useGlobalSubscribeCollection } from "~/api/materialize/useSubscribe"; +import { allObjectsCollection } from "~/store/allObjectsCollection"; +import { syncEngineCacheScopeLoadableAtom } from "~/store/syncEngineCache"; + +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, +) { + 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 ?? sizesByReplica.placeholder).statusAtom); + const objectsStatus = useAtomValue(allObjectsCollection.statusAtom); + const objects = React.useMemo( + () => + (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, + })), + [rows, heapLimit], + ); + return { + objects, + // Names come from the catalog, so without it every object would read as + // an orphaned dataflow. + snapshotComplete: status.snapshotComplete && objectsStatus.snapshotComplete, + 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; +}) { + const scope = useAtomValue(syncEngineCacheScopeLoadableAtom); + const sizes = + scope.state === "hasData" && scope.data + ? sizesByReplica.get(scope.data, replicaId) + : undefined; + const options = React.useMemo( + () => ({ + target: sizes ?? sizesByReplica.placeholder, + subscribe: sizes + ? buildReplicaObjectSizesSubscribe(replicaId) + : undefined, + select: (row: SubscribeRow) => row.data, + upsertKey: (row: SubscribeRow) => row.data.objectId, + }), + [sizes, replicaId], + ); + useGlobalSubscribeCollection(options); + const largest = useLargestObjects(sizes, heapLimit); + return { + ...largest, + isError: largest.isError || scope.state === "hasError", + }; +} diff --git a/console/src/platform/clusters/LargestMaintainedQueries.tsx b/console/src/platform/clusters/LargestMaintainedQueries.tsx index d89711504f10c..a271a94e22c8f 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; heapLimit: number; } @@ -173,6 +174,7 @@ const LargestReplicaLoader = ({ )} > {children({ + id: largestReplica.id, name: largestReplica.name, heapLimit: Number(largestReplica.heapLimit), })} diff --git a/console/src/platform/clusters/queries.ts b/console/src/platform/clusters/queries.ts index 750787a85bf36..0743dada14320 100644 --- a/console/src/platform/clusters/queries.ts +++ b/console/src/platform/clusters/queries.ts @@ -93,6 +93,7 @@ import { attachOfflineEvents, BinnedSubscribeRow, bucketRowsToBucketsByReplicaId, + clipBinnedRowsToWindow, parseBinnedSubscribeRow, rebucketUtilizationSamples, toReplicaUtilizationGraphData, @@ -478,6 +479,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; @@ -490,7 +493,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 @@ -736,6 +740,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. @@ -854,7 +863,6 @@ export function useUtilizationTier(timePeriodMinutes: number) { function useReplicaUtilizationHistorySubscribe( params: ReplicaUtilizationHistoryFilters, enabled: boolean, - includeOfflineEvents: boolean, ) { const { replicaId, timePeriodMinutes, bucketSizeMs, includeMemoryBreakdown } = params; @@ -891,7 +899,7 @@ function useReplicaUtilizationHistorySubscribe( const offlineEvents = useReplicaOfflineEvents( data, timePeriodMinutes, - enabled && includeOfflineEvents, + enabled, ); const result = useMemo(() => { @@ -959,15 +967,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), @@ -984,45 +987,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; @@ -1046,6 +1035,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