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