From 6da99429be3a45bcf83f82e4e0bc8e8e707d2811 Mon Sep 17 00:00:00 2001 From: Karakatiza666 Date: Tue, 21 Jul 2026 14:33:17 +0000 Subject: [PATCH] [web-console] Keep showing the pipeline metrics while the pipeline is stopping Signed-off-by: Karakatiza666 --- .../pipelines/editor/TabPerformance.svelte | 30 +- .../pipelines/editor/WarningBanner.svelte | 3 +- .../compositions/usePipelineManager.svelte.ts | 329 ++++++++++-------- .../src/lib/functions/pipelines/status.ts | 2 +- 4 files changed, 204 insertions(+), 160 deletions(-) diff --git a/js-packages/web-console/src/lib/components/pipelines/editor/TabPerformance.svelte b/js-packages/web-console/src/lib/components/pipelines/editor/TabPerformance.svelte index 5ab11db98bc..e3061498ae0 100644 --- a/js-packages/web-console/src/lib/components/pipelines/editor/TabPerformance.svelte +++ b/js-packages/web-console/src/lib/components/pipelines/editor/TabPerformance.svelte @@ -49,10 +49,18 @@ const { formatElapsedTime } = useElapsedTime() let timeSeries: TimeSeriesEntry[] = $state([]) + // Metrics stream lifecycle. 'connecting' is the initial sate and never happens again, + // stream can be 'interrupted' every time the 'live' conection drops + let metricsStreamState = $state<'connecting' | 'live' | 'interrupted'>('connecting') let statusTab: 'age' | 'updated' = $state('age') const isXl = useIsScreenXl() const api = usePipelineManager() + // Background polling races pipeline shutdown/startup, so don't show toast popups + // when the expected "pipeline not running" errors happen while reconnecting. + const pollingApi = api.silence((e) => + ['PipelineInteractionNotDeployed', 'PipelineUnavailable'].includes(e?.error_code) + ) type DrawerState = | { @@ -81,6 +89,8 @@ const metricsAvailable = $derived(metricsStatus === 'yes') // When metrics are temporarily unavailable ('missing'), freeze graphs and stats until the pipeline is reachable again. const metricsDesired = $derived(metricsStatus === 'yes' || metricsStatus === 'missing') + // Show the reconnect banner only after a live stream has dropped, never during the first connect. + const metricsStreamInterrupted = $derived(metricsAvailable && metricsStreamState === 'interrupted') // Keep reconnecting to time_series_stream for as long as the tab is mounted and metrics are desired // Reconnect on end-of-stream immediately, or with 1s backoff on mid-stream or stream-open errors @@ -103,6 +113,7 @@ // Start each session with empty stats so a previous pipeline's samples // never bleed into the newly selected one. timeSeries = [] + metricsStreamState = 'connecting' let cancelled = false let cancelActive: (() => void) | undefined @@ -118,7 +129,7 @@ await sleep(RECONNECT_BACKOFF_MS) continue } - const result = await api.pipelineTimeSeriesStream(targetPipelineName) + const result = await pollingApi.pipelineTimeSeriesStream(targetPipelineName) if (cancelled) { if (!(result instanceof Error)) { result.cancel() @@ -135,6 +146,7 @@ abortCtrl.abort() result.cancel() } + metricsStreamState = 'live' // Subscribe to the metrics stream, overwrite the previous data only when the first sample is received. try { @@ -161,6 +173,7 @@ } } cancelActive = undefined + metricsStreamState = 'interrupted' } } runMetricsStream() @@ -184,10 +197,10 @@ } // Poll checkpoint-related endpoints so the UI stays current with ongoing checkpoint activity. const fetchCheckpoints = () => { - api.getPipelineCheckpoints(pipelineName).then((v) => { + pollingApi.getPipelineCheckpoints(pipelineName).then((v) => { checkpoints = v }) - api.getCheckpointStatus(pipelineName).then((v) => { + pollingApi.getCheckpointStatus(pipelineName).then((v) => { checkpointStatus = v }) } @@ -225,9 +238,14 @@ Pipeline has been unavailable for {formatElapsedTime( new Date(pipeline.current.deploymentStatusSince) - )} since {Dayjs(pipeline.current.deploymentStatusSince).format('MMM D, YYYY h:mm A')}. - Showing the last known metrics while reconnecting. You can attempt to suspend or shut it - down. + )} since {Dayjs(pipeline.current.deploymentStatusSince).format( + 'MMM D, YYYY h:mm A' + )}. Showing the last known metrics while reconnecting. You can attempt to suspend or + shut it down. + + {:else if metricsStreamInterrupted} + + Not receiving live metrics. Attempting to reconnect... {/if}
diff --git a/js-packages/web-console/src/lib/components/pipelines/editor/WarningBanner.svelte b/js-packages/web-console/src/lib/components/pipelines/editor/WarningBanner.svelte index d4bbf79a38f..7e2087eb8d4 100644 --- a/js-packages/web-console/src/lib/components/pipelines/editor/WarningBanner.svelte +++ b/js-packages/web-console/src/lib/components/pipelines/editor/WarningBanner.svelte @@ -1,4 +1,5 @@ -
+
{@render children()} diff --git a/js-packages/web-console/src/lib/compositions/usePipelineManager.svelte.ts b/js-packages/web-console/src/lib/compositions/usePipelineManager.svelte.ts index 548663cbfc6..e39fbbc19ca 100644 --- a/js-packages/web-console/src/lib/compositions/usePipelineManager.svelte.ts +++ b/js-packages/web-console/src/lib/compositions/usePipelineManager.svelte.ts @@ -121,59 +121,180 @@ export const usePipelineManager = (options?: FetchOptions) => { return false } - const reportError = _trackHealth(options, toastError('API request'), doNotReportIfCancelled) const trackHealth = _trackHealth(options) - const getPipelineSupportBundle = ( - ...[pipelineName, options, onProgress]: Parameters - ) => { - const result = _getPipelineSupportBundle(pipelineName, options, onProgress) - return { - cancel: result.cancel, + // Build the API surface against a given error reporter. `.silence()` reuses this to produce a + // variant whose toasts are suppressed for errors matching a predicate + const build = (reportError: ReturnType) => { + const getPipelineSupportBundle = ( + ...[pipelineName, options, onProgress]: Parameters + ) => { + const result = _getPipelineSupportBundle(pipelineName, options, onProgress) + return { + cancel: result.cancel, - dataPromise: reportError( - () => { - const download = result.downloadPromise.then(async (download) => ({ - filename: download.filename, - data: await download.dataPromise - })) - return download - }, - () => `Failed to download support bundle of pipeline ${pipelineName}` - )() + dataPromise: reportError( + () => { + const download = result.downloadPromise.then(async (download) => ({ + filename: download.filename, + data: await download.dataPromise + })) + return download + }, + () => `Failed to download support bundle of pipeline ${pipelineName}` + )() + } } - } - const downloadPipelineSupportBundle = ( - pipelineName: string, - options: Partial, - onProgress?: (bytesDownloaded: number, bytesTotal: number) => void - ) => { - const { dataPromise, cancel } = getPipelineSupportBundle(pipelineName, options, onProgress) - dataPromise.then((download) => triggerFileDownload(download.filename, download.data)) - return { - cancel, - dataPromise + const downloadPipelineSupportBundle = ( + pipelineName: string, + options: Partial, + onProgress?: (bytesDownloaded: number, bytesTotal: number) => void + ) => { + const { dataPromise, cancel } = getPipelineSupportBundle(pipelineName, options, onProgress) + dataPromise.then((download) => triggerFileDownload(download.filename, download.data)) + return { + cancel, + dataPromise + } + } + + const downloadSamplyProfile = async ( + pipelineName: string, + latest: boolean, + onProgress?: (bytesDownloaded: number, bytesTotal: number) => void + ) => { + const result = await getSamplyProfile(pipelineName, latest, onProgress) + if ('expectedInSeconds' in result) { + return result + } + result.downloadPromise.then((download) => + download.dataPromise.then((data) => { + triggerFileDownload(download.filename, data) + }) + ) + return { cancel: result.cancel } } - } - const downloadSamplyProfile = async ( - pipelineName: string, - latest: boolean, - onProgress?: (bytesDownloaded: number, bytesTotal: number) => void - ) => { - const result = await getSamplyProfile(pipelineName, latest, onProgress) - if ('expectedInSeconds' in result) { - return result + return { + getExtendedPipeline: reportError( + getExtendedPipeline, + (pipelineName) => `Failed to fetch ${pipelineName} pipeline` + ), + postPipeline: reportError( + postPipeline, + (pipelineName) => `Failed to create ${pipelineName} pipeline` + ), + putPipeline: reportError( + putPipeline, + (pipelineName) => `Failed to update ${pipelineName} pipeline` + ), + patchPipeline: reportError( + patchPipeline, + (pipelineName) => `Failed to update ${pipelineName} pipeline` + ), + getPipelines: trackHealth(getPipelines), + getPipelineThumb: reportError( + getPipelineThumb, + (pipelineName) => `Failed to get ${pipelineName} pipeline's status` + ), + getPipelineStats: getPipelineStats, + deletePipeline: reportError( + deletePipeline, + (pipelineName) => `Failed to delete ${pipelineName} pipeline` + ), + postPipelineAction: reportError( + postPipelineAction, + (pipelineName, action) => `Failed to ${action} ${pipelineName} pipeline` + ), + postUpdateRuntime: reportError( + postUpdateRuntime, + (pipelineName) => `Failed to update runtime for ${pipelineName} pipeline` + ), + getAuthConfig: getAuthConfig, + getConfig: getConfig, + getConfigSession: reportError( + getConfigSession, + () => 'Failed to fetch session configuration' + ), + getApiKeys: reportError(getApiKeys, () => 'Failed to fetch API keys'), + postApiKey: async (name: string, options?: FetchOptions | undefined) => { + const x = await reportError(postApiKey, (keyName) => `Failed to create ${keyName} API key`)( + name, + options + ) + return x + }, + deleteApiKey: reportError(deleteApiKey, (keyName) => `Failed to delete ${keyName} API key`), + dismissDeploymentError: reportError( + dismissDeploymentError, + (pipelineName) => `Failed to dismiss deployment error for ${pipelineName} pipeline` + ), + getClusterEvents: reportError(getClusterEvents), + getClusterEvent: reportError(getClusterEvent), + getPipelineEvents: reportError( + getPipelineEvents, + (pipelineName) => `Failed to fetch ${pipelineName} pipeline events` + ), + getPipelineEvent: reportError( + getPipelineEvent, + (pipelineName) => `Failed to fetch ${pipelineName} pipeline event` + ), + getSamplyProfile: reportError(getSamplyProfile), + downloadSamplyProfile: reportError( + downloadSamplyProfile, + (pipelineName) => `Failed to download samply profile for ${pipelineName} pipeline` + ), + collectSamplyProfile: reportError(collectSamplyProfile), + relationEgressStream: reportError( + relationEgressStream, + (_, relationName) => `Failed to connect to the egress stream of relation ${relationName}` + ), + pipelineLogsStream: reportError( + pipelineLogsStream, + () => `Failed to connect to the log stream` + ), + adHocQuery: reportError(adHocQuery, () => `Failed to invoke an ad-hoc query`), + pipelineTimeSeriesStream: reportError( + pipelineTimeSeriesStream, + () => `Failed to connect to the time series stream` + ), + relationIngress: reportError( + relationIngress, + (_, tableName) => `Failed to push data to the ${tableName} table` + ), + getDemos: reportError(getDemos, () => `Failed to fetch available demos`), + downloadPipelineSupportBundle, + getPipelineCheckpoints: reportError( + getPipelineCheckpoints, + (pipelineName) => `Failed to fetch checkpoints for ${pipelineName}` + ), + checkpointPipeline: reportError( + checkpointPipeline, + (pipelineName) => `Failed to initiate checkpoint for ${pipelineName}` + ), + getCheckpointStatus: reportError( + getCheckpointStatus, + (pipelineName) => `Failed to fetch checkpoint status for ${pipelineName}` + ), + syncCheckpoint: reportError( + syncCheckpoint, + (pipelineName) => `Failed to sync checkpoint for ${pipelineName}` + ), + getCheckpointSyncStatus: reportError( + getCheckpointSyncStatus, + (pipelineName) => `Failed to fetch checkpoint sync status for ${pipelineName}` + ), + getPipelineDataflowGraph: reportError( + getPipelineDataflowGraph, + (pipelineName) => `Failed to load dataflow graph of pipeline ${pipelineName}` + ), + getPipelineSupportBundle } - result.downloadPromise.then((download) => - download.dataPromise.then((data) => { - triggerFileDownload(download.filename, data) - }) - ) - return { cancel: result.cancel } } + const reportError = _trackHealth(options, toastError('API request'), doNotReportIfCancelled) + return { get isNetworkHealthy() { return isNetworkHealthy @@ -181,115 +302,19 @@ export const usePipelineManager = (options?: FetchOptions) => { get isAuthHealthy() { return isAuthHealthy }, - getExtendedPipeline: reportError( - getExtendedPipeline, - (pipelineName) => `Failed to fetch ${pipelineName} pipeline` - ), - postPipeline: reportError( - postPipeline, - (pipelineName) => `Failed to create ${pipelineName} pipeline` - ), - putPipeline: reportError( - putPipeline, - (pipelineName) => `Failed to update ${pipelineName} pipeline` - ), - patchPipeline: reportError( - patchPipeline, - (pipelineName) => `Failed to update ${pipelineName} pipeline` - ), - getPipelines: trackHealth(getPipelines), - getPipelineThumb: reportError( - getPipelineThumb, - (pipelineName) => `Failed to get ${pipelineName} pipeline's status` - ), - getPipelineStats: getPipelineStats, - deletePipeline: reportError( - deletePipeline, - (pipelineName) => `Failed to delete ${pipelineName} pipeline` - ), - postPipelineAction: reportError( - postPipelineAction, - (pipelineName, action) => `Failed to ${action} ${pipelineName} pipeline` - ), - postUpdateRuntime: reportError( - postUpdateRuntime, - (pipelineName) => `Failed to update runtime for ${pipelineName} pipeline` - ), - getAuthConfig: getAuthConfig, - getConfig: getConfig, - getConfigSession: reportError(getConfigSession, () => 'Failed to fetch session configuration'), - getApiKeys: reportError(getApiKeys, () => 'Failed to fetch API keys'), - postApiKey: async (name: string, options?: FetchOptions | undefined) => { - const x = await reportError(postApiKey, (keyName) => `Failed to create ${keyName} API key`)( - name, - options + ...build(reportError), + /** + * Returns an API variant whose error toasts are suppressed for errors matching `predicate`. + * The predicate receives the API error body (carrying `error_code`) + * @example api.silence((e) => e?.error_code === 'PipelineInteractionNotDeployed').getCheckpointStatus(name) + */ + silence: (predicate: (error: any) => boolean) => + build( + _trackHealth( + options, + toastError('API request'), + (error) => doNotReportIfCancelled(error) || predicate((error as any)?.cause ?? error) + ) ) - return x - }, - deleteApiKey: reportError(deleteApiKey, (keyName) => `Failed to delete ${keyName} API key`), - dismissDeploymentError: reportError( - dismissDeploymentError, - (pipelineName) => `Failed to dismiss deployment error for ${pipelineName} pipeline` - ), - getClusterEvents: reportError(getClusterEvents), - getClusterEvent: reportError(getClusterEvent), - getPipelineEvents: reportError( - getPipelineEvents, - (pipelineName) => `Failed to fetch ${pipelineName} pipeline events` - ), - getPipelineEvent: reportError( - getPipelineEvent, - (pipelineName) => `Failed to fetch ${pipelineName} pipeline event` - ), - getSamplyProfile: reportError(getSamplyProfile), - downloadSamplyProfile: reportError( - downloadSamplyProfile, - (pipelineName) => `Failed to download samply profile for ${pipelineName} pipeline` - ), - collectSamplyProfile: reportError(collectSamplyProfile), - relationEgressStream: reportError( - relationEgressStream, - (_, relationName) => `Failed to connect to the egress stream of relation ${relationName}` - ), - pipelineLogsStream: reportError( - pipelineLogsStream, - () => `Failed to connect to the log stream` - ), - adHocQuery: reportError(adHocQuery, () => `Failed to invoke an ad-hoc query`), - pipelineTimeSeriesStream: reportError( - pipelineTimeSeriesStream, - () => `Failed to connect to the time series stream` - ), - relationIngress: reportError( - relationIngress, - (_, tableName) => `Failed to push data to the ${tableName} table` - ), - getDemos: reportError(getDemos, () => `Failed to fetch available demos`), - downloadPipelineSupportBundle, - getPipelineCheckpoints: reportError( - getPipelineCheckpoints, - (pipelineName) => `Failed to fetch checkpoints for ${pipelineName}` - ), - checkpointPipeline: reportError( - checkpointPipeline, - (pipelineName) => `Failed to initiate checkpoint for ${pipelineName}` - ), - getCheckpointStatus: reportError( - getCheckpointStatus, - (pipelineName) => `Failed to fetch checkpoint status for ${pipelineName}` - ), - syncCheckpoint: reportError( - syncCheckpoint, - (pipelineName) => `Failed to sync checkpoint for ${pipelineName}` - ), - getCheckpointSyncStatus: reportError( - getCheckpointSyncStatus, - (pipelineName) => `Failed to fetch checkpoint sync status for ${pipelineName}` - ), - getPipelineDataflowGraph: reportError( - getPipelineDataflowGraph, - (pipelineName) => `Failed to load dataflow graph of pipeline ${pipelineName}` - ), - getPipelineSupportBundle } } diff --git a/js-packages/web-console/src/lib/functions/pipelines/status.ts b/js-packages/web-console/src/lib/functions/pipelines/status.ts index 72b6b9d8106..fe86878586d 100644 --- a/js-packages/web-console/src/lib/functions/pipelines/status.ts +++ b/js-packages/web-console/src/lib/functions/pipelines/status.ts @@ -245,7 +245,7 @@ export const isMetricsAvailable = (status: PipelineStatus) => { .with('Running', () => 'yes') .with('Pausing', () => 'yes') .with('Resuming', () => 'yes') - .with('Stopping', () => 'no') + .with('Stopping', () => 'yes') .with( { Queued: P.any }, { CompilingSql: P.any },