nictownsend commented on code in PR #28060: URL: https://github.com/apache/flink/pull/28060#discussion_r4480064433
########## flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.html: ########## @@ -0,0 +1,174 @@ +<!-- + ~ Licensed to the Apache Software Foundation (ASF) under one + ~ or more contributor license agreements. See the NOTICE file + ~ distributed with this work for additional information + ~ regarding copyright ownership. The ASF licenses this file + ~ to you under the Apache License, Version 2.0 (the + ~ "License"); you may not use this file except in compliance + ~ with the License. You may obtain a copy of the License at + ~ + ~ http://www.apache.org/licenses/LICENSE-2.0 + ~ + ~ Unless required by applicable law or agreed to in writing, software + ~ distributed under the License is distributed on an "AS IS" BASIS, + ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + ~ See the License for the specific language governing permissions and + ~ limitations under the License. + --> + +<div class="timeline-wrapper"> + <section class="strip-section"> + <header class="strip-header"> + <div class="strip-title"> + <strong>Recent Checkpoints</strong> + <span class="hint">Click a bar to load its per-subtask timeline below.</span> + </div> + <ul class="legend"> + <li> + <span class="swatch completed"></span> + Completed + </li> + <li> + <span class="swatch savepoint"></span> + Savepoint + </li> + <li> + <span class="swatch in-progress"></span> + In progress + </li> + <li> + <span class="swatch failed"></span> + Failed + </li> + </ul> + </header> + <div class="strip-canvas" #stripContainer></div> + <nz-empty *ngIf="!hasHistory" nzNotFoundContent="No checkpoint history yet"></nz-empty> + </section> + + <section class="timeline-section"> + <header class="timeline-header"> + <div class="timeline-title"> + <strong *ngIf="selected"> + Checkpoint #{{ selected.id }} + <span + *ngIf="userPinned" + class="pin-badge" + title="Pinned: the timeline below will not change until you click another checkpoint or 'Follow newest'." + > + pinned + </span> + </strong> + <strong *ngIf="!selected">Per-subtask Timeline</strong> + <span *ngIf="selected" class="hint"> + {{ selected.rawStatus }} + <ng-container *ngIf="selected.isSavepoint">· savepoint</ng-container> + · triggered {{ selected.triggered | date: 'yyyy-MM-dd HH:mm:ss.SSS' }} · + {{ selected.duration | humanizeDuration }} · acked {{ selected.acked }} · + {{ selected.size | humanizeBytes }} + </span> + <div class="timeline-actions"> + <button *ngIf="userPinned" type="button" class="follow-newest" (click)="followNewest()"> Review Comment: The rest of the repo uses `nz-button` rather than raw stylling ########## flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.html: ########## @@ -0,0 +1,174 @@ +<!-- + ~ Licensed to the Apache Software Foundation (ASF) under one + ~ or more contributor license agreements. See the NOTICE file + ~ distributed with this work for additional information + ~ regarding copyright ownership. The ASF licenses this file + ~ to you under the Apache License, Version 2.0 (the + ~ "License"); you may not use this file except in compliance + ~ with the License. You may obtain a copy of the License at + ~ + ~ http://www.apache.org/licenses/LICENSE-2.0 + ~ + ~ Unless required by applicable law or agreed to in writing, software + ~ distributed under the License is distributed on an "AS IS" BASIS, + ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + ~ See the License for the specific language governing permissions and + ~ limitations under the License. + --> + +<div class="timeline-wrapper"> + <section class="strip-section"> + <header class="strip-header"> + <div class="strip-title"> + <strong>Recent Checkpoints</strong> + <span class="hint">Click a bar to load its per-subtask timeline below.</span> + </div> + <ul class="legend"> + <li> + <span class="swatch completed"></span> + Completed + </li> + <li> + <span class="swatch savepoint"></span> + Savepoint + </li> + <li> + <span class="swatch in-progress"></span> + In progress + </li> + <li> + <span class="swatch failed"></span> + Failed + </li> + </ul> + </header> + <div class="strip-canvas" #stripContainer></div> + <nz-empty *ngIf="!hasHistory" nzNotFoundContent="No checkpoint history yet"></nz-empty> + </section> + + <section class="timeline-section"> + <header class="timeline-header"> + <div class="timeline-title"> + <strong *ngIf="selected"> + Checkpoint #{{ selected.id }} + <span + *ngIf="userPinned" + class="pin-badge" + title="Pinned: the timeline below will not change until you click another checkpoint or 'Follow newest'." + > + pinned + </span> + </strong> + <strong *ngIf="!selected">Per-subtask Timeline</strong> + <span *ngIf="selected" class="hint"> + {{ selected.rawStatus }} + <ng-container *ngIf="selected.isSavepoint">· savepoint</ng-container> + · triggered {{ selected.triggered | date: 'yyyy-MM-dd HH:mm:ss.SSS' }} · + {{ selected.duration | humanizeDuration }} · acked {{ selected.acked }} · + {{ selected.size | humanizeBytes }} + </span> + <div class="timeline-actions"> + <button *ngIf="userPinned" type="button" class="follow-newest" (click)="followNewest()"> + Follow newest + </button> + </div> + </div> + <ul class="legend"> + <li> + <span class="swatch phase-start"></span> + start_delay + </li> + <li> + <span class="swatch phase-align"></span> + alignment + </li> + <li> + <span class="swatch phase-sync"></span> + sync + </li> + <li> + <span class="swatch phase-async"></span> + async + </li> + <li nz-tooltip [nzTooltipTitle]="abortedTip" nzTooltipPlacement="bottom"> + <span class="swatch aborted"></span> + aborted + </li> + <li nz-tooltip [nzTooltipTitle]="noAckTip" nzTooltipPlacement="bottom"> + <span class="swatch no-ack"></span> + no ack + </li> + <ng-template #abortedTip> + Started its snapshot, but the checkpoint failed before it finished. The bar shows the + partial work. + </ng-template> + <ng-template #noAckTip> + Reported no statistics for this checkpoint: still pending, or never acknowledged before + the checkpoint failed (for example, the barrier never arrived). The bar spans the whole + checkpoint. + </ng-template> + </ul> + </header> + + <!-- Reserve the line while a checkpoint is selected so the chart doesn't shift. --> + <div *ngIf="selected" class="timeline-note"> + <span *ngIf="selectedRotatedOut && !timelineEmpty"> + Checkpoint #{{ selected?.id }} is no longer in the recent-checkpoints strip above — showing + the frozen snapshot from when it was pinned. + </span> + <span *ngIf="totalRows > 1 && !timelineEmpty" class="timeline-controls"> + <!-- Text, not whitespace-only nodes, around the select: Angular drops whitespace-only nodes. --> + <span *ngIf="slowestCountOptions.length > 0"> + {{ shownRows < totalRows ? 'Showing the' : 'Showing all' }} + <select + class="row-count" + aria-label="Number of slowest subtasks shown" + (change)="onSlowestCountChange($event)" + > + <option *ngFor="let n of slowestCountOptions" [value]="n" [selected]="n === shownRows"> + {{ n }} + </option> + </select> + {{ shownRows < totalRows ? 'slowest of ' + totalRows + ' subtasks' : 'subtasks' }} + {{ scrollable ? '· rows ' + (firstRow + 1) + '–' + (firstRow + visibleRows) : '' }} + </span> + <label class="slowest-first"> + <input type="checkbox" [checked]="slowestFirst" (change)="onSlowestFirstChange($event)" /> Review Comment: The rest of the repo uses `nz-checkbox` rather than raw stylling ########## flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.html: ########## @@ -0,0 +1,174 @@ +<!-- + ~ Licensed to the Apache Software Foundation (ASF) under one + ~ or more contributor license agreements. See the NOTICE file + ~ distributed with this work for additional information + ~ regarding copyright ownership. The ASF licenses this file + ~ to you under the Apache License, Version 2.0 (the + ~ "License"); you may not use this file except in compliance + ~ with the License. You may obtain a copy of the License at + ~ + ~ http://www.apache.org/licenses/LICENSE-2.0 + ~ + ~ Unless required by applicable law or agreed to in writing, software + ~ distributed under the License is distributed on an "AS IS" BASIS, + ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + ~ See the License for the specific language governing permissions and + ~ limitations under the License. + --> + +<div class="timeline-wrapper"> + <section class="strip-section"> + <header class="strip-header"> + <div class="strip-title"> + <strong>Recent Checkpoints</strong> + <span class="hint">Click a bar to load its per-subtask timeline below.</span> + </div> + <ul class="legend"> + <li> + <span class="swatch completed"></span> + Completed + </li> + <li> + <span class="swatch savepoint"></span> + Savepoint + </li> + <li> + <span class="swatch in-progress"></span> + In progress + </li> + <li> + <span class="swatch failed"></span> + Failed + </li> + </ul> + </header> + <div class="strip-canvas" #stripContainer></div> + <nz-empty *ngIf="!hasHistory" nzNotFoundContent="No checkpoint history yet"></nz-empty> + </section> + + <section class="timeline-section"> + <header class="timeline-header"> + <div class="timeline-title"> + <strong *ngIf="selected"> + Checkpoint #{{ selected.id }} + <span + *ngIf="userPinned" + class="pin-badge" + title="Pinned: the timeline below will not change until you click another checkpoint or 'Follow newest'." + > + pinned + </span> + </strong> + <strong *ngIf="!selected">Per-subtask Timeline</strong> + <span *ngIf="selected" class="hint"> + {{ selected.rawStatus }} + <ng-container *ngIf="selected.isSavepoint">· savepoint</ng-container> + · triggered {{ selected.triggered | date: 'yyyy-MM-dd HH:mm:ss.SSS' }} · + {{ selected.duration | humanizeDuration }} · acked {{ selected.acked }} · + {{ selected.size | humanizeBytes }} + </span> + <div class="timeline-actions"> + <button *ngIf="userPinned" type="button" class="follow-newest" (click)="followNewest()"> + Follow newest + </button> + </div> + </div> + <ul class="legend"> + <li> + <span class="swatch phase-start"></span> + start_delay + </li> + <li> + <span class="swatch phase-align"></span> + alignment + </li> + <li> + <span class="swatch phase-sync"></span> + sync + </li> + <li> + <span class="swatch phase-async"></span> + async + </li> + <li nz-tooltip [nzTooltipTitle]="abortedTip" nzTooltipPlacement="bottom"> + <span class="swatch aborted"></span> + aborted + </li> + <li nz-tooltip [nzTooltipTitle]="noAckTip" nzTooltipPlacement="bottom"> + <span class="swatch no-ack"></span> + no ack + </li> + <ng-template #abortedTip> + Started its snapshot, but the checkpoint failed before it finished. The bar shows the + partial work. + </ng-template> + <ng-template #noAckTip> + Reported no statistics for this checkpoint: still pending, or never acknowledged before + the checkpoint failed (for example, the barrier never arrived). The bar spans the whole + checkpoint. + </ng-template> + </ul> + </header> + + <!-- Reserve the line while a checkpoint is selected so the chart doesn't shift. --> + <div *ngIf="selected" class="timeline-note"> + <span *ngIf="selectedRotatedOut && !timelineEmpty"> + Checkpoint #{{ selected?.id }} is no longer in the recent-checkpoints strip above — showing + the frozen snapshot from when it was pinned. + </span> + <span *ngIf="totalRows > 1 && !timelineEmpty" class="timeline-controls"> + <!-- Text, not whitespace-only nodes, around the select: Angular drops whitespace-only nodes. --> + <span *ngIf="slowestCountOptions.length > 0"> + {{ shownRows < totalRows ? 'Showing the' : 'Showing all' }} + <select + class="row-count" Review Comment: The rest of the repo uses `nz-select` rather than raw stylling ########## flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.html: ########## @@ -0,0 +1,174 @@ +<!-- + ~ Licensed to the Apache Software Foundation (ASF) under one + ~ or more contributor license agreements. See the NOTICE file + ~ distributed with this work for additional information + ~ regarding copyright ownership. The ASF licenses this file + ~ to you under the Apache License, Version 2.0 (the + ~ "License"); you may not use this file except in compliance + ~ with the License. You may obtain a copy of the License at + ~ + ~ http://www.apache.org/licenses/LICENSE-2.0 + ~ + ~ Unless required by applicable law or agreed to in writing, software + ~ distributed under the License is distributed on an "AS IS" BASIS, + ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + ~ See the License for the specific language governing permissions and + ~ limitations under the License. + --> + +<div class="timeline-wrapper"> + <section class="strip-section"> + <header class="strip-header"> + <div class="strip-title"> + <strong>Recent Checkpoints</strong> + <span class="hint">Click a bar to load its per-subtask timeline below.</span> + </div> + <ul class="legend"> + <li> + <span class="swatch completed"></span> + Completed + </li> + <li> + <span class="swatch savepoint"></span> + Savepoint + </li> + <li> + <span class="swatch in-progress"></span> + In progress + </li> + <li> + <span class="swatch failed"></span> + Failed + </li> + </ul> + </header> + <div class="strip-canvas" #stripContainer></div> + <nz-empty *ngIf="!hasHistory" nzNotFoundContent="No checkpoint history yet"></nz-empty> + </section> + + <section class="timeline-section"> + <header class="timeline-header"> + <div class="timeline-title"> + <strong *ngIf="selected"> + Checkpoint #{{ selected.id }} + <span + *ngIf="userPinned" + class="pin-badge" Review Comment: There's an `ng-badge` and `nz-tag` component available ########## flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.spec.ts: ########## @@ -0,0 +1,672 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { HttpErrorResponse } from '@angular/common/http'; +import { ChangeDetectorRef, ElementRef, SimpleChange } from '@angular/core'; +import { BehaviorSubject, of, Subject, throwError } from 'rxjs'; + +import { + CheckpointConfig, + CheckpointHistory, + CheckpointSubTask, + SubTaskCheckpointStatisticsItem, + VerticesItem +} from '@flink-runtime-web/interfaces'; +import { JobService } from '@flink-runtime-web/services'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { JobCheckpointsTimelineComponent } from './job-checkpoints-timeline.component'; + +const TRIGGERED = 1_000; + +interface Phases { + start_delay?: number; + alignment?: number; + sync?: number; + async?: number; +} + +const completed = (index: number, phases: Phases, aborted = false): SubTaskCheckpointStatisticsItem => + ({ + index, + status: 'completed', + ack_timestamp: 0, + end_to_end_duration: 0, + checkpointed_size: 0, + state_size: 0, + checkpoint: { sync: phases.sync ?? 0, async: phases.async ?? 0 }, + alignment: { buffer: 0, processed: 0, persisted: 0, duration: phases.alignment ?? 0 }, + start_delay: phases.start_delay ?? 0, + unaligned_checkpoint: false, + aborted + }) as SubTaskCheckpointStatisticsItem; + +const noStats = (index: number): SubTaskCheckpointStatisticsItem => + ({ index, status: 'pending_or_failed' }) as SubTaskCheckpointStatisticsItem; + +const details = (...subtasks: SubTaskCheckpointStatisticsItem[]): CheckpointSubTask => + ({ subtasks }) as unknown as CheckpointSubTask; + +const vertex = (id: string, name: string): VerticesItem => ({ id, name, status: 'RUNNING' }) as VerticesItem; + +const checkpoint = ( + rawStatus: string, + duration: number, + isSavepoint = false +): { id: number; duration: number; rawStatus: string; isSavepoint: boolean; triggered: number } => ({ + id: 7, + duration, + rawStatus, + isSavepoint, + triggered: TRIGGERED +}); + +type BuildRows = ( + datum: ReturnType<typeof checkpoint>, + perVertex: Record<string, CheckpointSubTask | null> +) => { + rows: Array<{ row: string; phase: string; phaseLabel: string; range: [number, number]; totalDuration: number }>; + unacked: Array<{ row: string; label: string; range: [number, number] }>; + instants: Array<{ row: string; aborted: boolean; range: [number, number] }>; + rowOrder: string[]; +}; + +const unalignedConfig = (alignedCheckpointTimeout: number, mode = 'exactly_once'): CheckpointConfig => + ({ mode, unaligned_checkpoints: true, aligned_checkpoint_timeout: alignedCheckpointTimeout }) as CheckpointConfig; + +const SUBTASK_PHASES = { start_delay: 2, alignment: 5, sync: 1, async: 8 }; + +describe('JobCheckpointsTimelineComponent rows', () => { + let component: JobCheckpointsTimelineComponent; + let buildRows: BuildRows; + + beforeEach(() => { + component = new JobCheckpointsTimelineComponent( + {} as JobService, + { markForCheck: () => undefined } as unknown as ChangeDetectorRef + ); + buildRows = (component as unknown as { buildTimelineRows: BuildRows }).buildTimelineRows.bind(component); + }); + + it('keeps rows in operator then subtask order instead of sorting by duration', () => { + component.vertices = [vertex('a', 'Agg'), vertex('s', 'Sink')]; + + const { rowOrder } = buildRows(checkpoint('COMPLETED', 60), { + a: details(completed(0, { async: 5 }), completed(1, { async: 50 })), Review Comment: Can we swap the 5 and 50 for `a` - so that we get `agg 1, agg 0, sink 0` - just to avoid any regressions like "sort by index" would break this test. ########## flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.spec.ts: ########## @@ -0,0 +1,672 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { HttpErrorResponse } from '@angular/common/http'; +import { ChangeDetectorRef, ElementRef, SimpleChange } from '@angular/core'; +import { BehaviorSubject, of, Subject, throwError } from 'rxjs'; + +import { + CheckpointConfig, + CheckpointHistory, + CheckpointSubTask, + SubTaskCheckpointStatisticsItem, + VerticesItem +} from '@flink-runtime-web/interfaces'; +import { JobService } from '@flink-runtime-web/services'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { JobCheckpointsTimelineComponent } from './job-checkpoints-timeline.component'; + +const TRIGGERED = 1_000; + +interface Phases { + start_delay?: number; + alignment?: number; + sync?: number; + async?: number; +} + +const completed = (index: number, phases: Phases, aborted = false): SubTaskCheckpointStatisticsItem => + ({ + index, + status: 'completed', + ack_timestamp: 0, + end_to_end_duration: 0, + checkpointed_size: 0, + state_size: 0, + checkpoint: { sync: phases.sync ?? 0, async: phases.async ?? 0 }, + alignment: { buffer: 0, processed: 0, persisted: 0, duration: phases.alignment ?? 0 }, + start_delay: phases.start_delay ?? 0, + unaligned_checkpoint: false, + aborted + }) as SubTaskCheckpointStatisticsItem; + +const noStats = (index: number): SubTaskCheckpointStatisticsItem => + ({ index, status: 'pending_or_failed' }) as SubTaskCheckpointStatisticsItem; + +const details = (...subtasks: SubTaskCheckpointStatisticsItem[]): CheckpointSubTask => + ({ subtasks }) as unknown as CheckpointSubTask; + +const vertex = (id: string, name: string): VerticesItem => ({ id, name, status: 'RUNNING' }) as VerticesItem; + +const checkpoint = ( + rawStatus: string, + duration: number, + isSavepoint = false +): { id: number; duration: number; rawStatus: string; isSavepoint: boolean; triggered: number } => ({ + id: 7, + duration, + rawStatus, + isSavepoint, + triggered: TRIGGERED +}); + +type BuildRows = ( + datum: ReturnType<typeof checkpoint>, + perVertex: Record<string, CheckpointSubTask | null> +) => { + rows: Array<{ row: string; phase: string; phaseLabel: string; range: [number, number]; totalDuration: number }>; + unacked: Array<{ row: string; label: string; range: [number, number] }>; + instants: Array<{ row: string; aborted: boolean; range: [number, number] }>; + rowOrder: string[]; +}; + +const unalignedConfig = (alignedCheckpointTimeout: number, mode = 'exactly_once'): CheckpointConfig => + ({ mode, unaligned_checkpoints: true, aligned_checkpoint_timeout: alignedCheckpointTimeout }) as CheckpointConfig; + +const SUBTASK_PHASES = { start_delay: 2, alignment: 5, sync: 1, async: 8 }; + +describe('JobCheckpointsTimelineComponent rows', () => { + let component: JobCheckpointsTimelineComponent; + let buildRows: BuildRows; + + beforeEach(() => { + component = new JobCheckpointsTimelineComponent( + {} as JobService, + { markForCheck: () => undefined } as unknown as ChangeDetectorRef + ); + buildRows = (component as unknown as { buildTimelineRows: BuildRows }).buildTimelineRows.bind(component); + }); + + it('keeps rows in operator then subtask order instead of sorting by duration', () => { + component.vertices = [vertex('a', 'Agg'), vertex('s', 'Sink')]; + + const { rowOrder } = buildRows(checkpoint('COMPLETED', 60), { + a: details(completed(0, { async: 5 }), completed(1, { async: 50 })), + s: details(completed(0, { start_delay: 1 })) + }); + + expect(rowOrder).toEqual(['Agg #0', 'Agg #1', 'Sink #0']); + }); + + it('keeps a row with an instant marker for subtasks whose phases all rounded to 0ms', () => { + component.vertices = [vertex('s', 'Source')]; + + const { instants, rowOrder } = buildRows(checkpoint('COMPLETED', 15), { + s: details(completed(0, {}), completed(1, { start_delay: 1 })) + }); + + expect(rowOrder).toEqual(['Source #0', 'Source #1']); + expect(instants).toEqual([ + expect.objectContaining({ row: 'Source #0', aborted: false, range: [TRIGGERED, TRIGGERED] }) + ]); + }); + + it('draws a no-ack bar spanning a failed checkpoint for subtasks without stats', () => { + component.vertices = [vertex('a', 'Agg')]; + + const { unacked, rowOrder } = buildRows(checkpoint('FAILED', 10), { + a: details(completed(0, { start_delay: 1, async: 3 }, true), noStats(1), completed(2, { async: 2 })) + }); + + expect(unacked).toEqual([ + expect.objectContaining({ row: 'Agg #1', label: 'no ack', range: [TRIGGERED, TRIGGERED + 10] }) + ]); + expect(rowOrder).toEqual(['Agg #0', 'Agg #1', 'Agg #2']); + }); + + it('marks subtasks without stats as pending up to the latest known progress while in progress', () => { + component.vertices = [vertex('a', 'Agg')]; + + const { unacked } = buildRows(checkpoint('IN_PROGRESS', 0), { + a: details(completed(0, { start_delay: 1, async: 4 }), noStats(1)) + }); + + expect(unacked).toEqual([expect.objectContaining({ label: 'pending', range: [TRIGGERED, TRIGGERED + 5] })]); + }); + + it('uses the raw status for savepoints', () => { + component.vertices = [vertex('a', 'Agg')]; + + const { unacked } = buildRows(checkpoint('FAILED', 10, true), { a: details(noStats(0)) }); + + expect(unacked.map(u => u.label)).toEqual(['no ack']); + }); + + it('ignores missing stats on completed checkpoints', () => { + component.vertices = [vertex('a', 'Agg')]; + + const { unacked, rowOrder } = buildRows(checkpoint('COMPLETED', 10), { + a: details(completed(0, { async: 2 }), noStats(1)) + }); + + expect(unacked).toEqual([]); + expect(rowOrder).toEqual(['Agg #0']); + }); + + const segments = ( + rows: ReturnType<BuildRows>['rows'], + row: string + ): Array<[string, string, [number, number], number]> => + rows.filter(r => r.row === row).map(r => [r.phase, r.phaseLabel, r.range, r.totalDuration]); + + // Long.MAX_VALUE (2 ** 63 once parsed) means no timeout, i.e. unaligned from the first barrier, like 0. + it.each([0, 2 ** 63])('draws no alignment before sync for checkpoints unaligned from the start (timeout %s)', t => { + component.vertices = [vertex('a', 'Agg')]; + component.config = unalignedConfig(t); + + const { rows } = buildRows(checkpoint('COMPLETED', 20), { a: details(completed(0, SUBTASK_PHASES)) }); + + expect(segments(rows, 'Agg #0')).toEqual([ + ['start_delay', 'start_delay', [TRIGGERED, TRIGGERED + 2], 11], + ['sync', 'sync', [TRIGGERED + 2, TRIGGERED + 3], 11], + ['async', 'async (incl. 5ms alignment)', [TRIGGERED + 3, TRIGGERED + 11], 11] + ]); + }); + + it('draws the alignment up to the aligned checkpoint timeout before sync', () => { + component.vertices = [vertex('a', 'Agg')]; + component.config = unalignedConfig(30_000); + + const { rows } = buildRows(checkpoint('COMPLETED', 40_000), { + a: details( + completed(0, { start_delay: 5_000, alignment: 31_000, sync: 10, async: 8_000 }), + completed(1, SUBTASK_PHASES) + ) + }); + + expect(segments(rows, 'Agg #0')).toEqual([ + ['start_delay', 'start_delay', [TRIGGERED, TRIGGERED + 5_000], 38_010], + ['alignment', 'alignment', [TRIGGERED + 5_000, TRIGGERED + 30_000], 38_010], + ['sync', 'sync', [TRIGGERED + 30_000, TRIGGERED + 30_010], 38_010], + ['async', 'async (incl. 6000ms alignment)', [TRIGGERED + 30_010, TRIGGERED + 38_010], 38_010] + ]); + // Aligned within the timeout, so it never switched. + expect(segments(rows, 'Agg #1').map(([, label]) => label)).toEqual(['start_delay', 'alignment', 'sync', 'async']); + }); + + it.each([ + ['savepoints', unalignedConfig(0), true], + ['at-least-once checkpoints', unalignedConfig(0, 'at_least_once'), false], + ['jobs without unaligned checkpoints', { ...unalignedConfig(0), unaligned_checkpoints: false }, false] + ])('draws the whole alignment before sync for %s', (_, config, isSavepoint) => { + component.vertices = [vertex('a', 'Agg')]; + component.config = config; + + const { rows } = buildRows(checkpoint('COMPLETED', 20, isSavepoint), { a: details(completed(0, SUBTASK_PHASES)) }); + + expect(segments(rows, 'Agg #0')).toEqual([ + ['start_delay', 'start_delay', [TRIGGERED, TRIGGERED + 2], 16], + ['alignment', 'alignment', [TRIGGERED + 2, TRIGGERED + 7], 16], + ['sync', 'sync', [TRIGGERED + 7, TRIGGERED + 8], 16], + ['async', 'async', [TRIGGERED + 8, TRIGGERED + 16], 16] + ]); + }); + + it('updates one timeline chart across reloads instead of rebuilding it', () => { + const mark = (): { data: ReturnType<typeof vi.fn>; scale: ReturnType<typeof vi.fn> } => { + const m = { data: vi.fn(), scale: vi.fn() }; + m.data.mockReturnValue(m); + m.scale.mockReturnValue(m); + return m; + }; + const chart = { attr: vi.fn(), render: vi.fn(), destroy: vi.fn() }; + const create = vi.fn(() => ({ chart, phases: mark(), unacked: mark(), instants: mark() })); + const internals = component as unknown as { + createTimelineChart: typeof create; + renderTimeline: (datum: ReturnType<typeof checkpoint>, perVertex: Record<string, CheckpointSubTask>) => void; + }; + internals.createTimelineChart = create; Review Comment: This is patching internals, could we instead perhaps use TestBed with a mock at the g2 level with a fake Chart that tracks/counts these calls? That way we can assert changes based on public inputs to the component, not `renderTimeline` etc. ``` fixture = TestBed.createComponent(JobCheckpointsTimelineComponent); fixture.componentRef.setInput('jobId', 'job-1'); fixture.componentRef.setInput('history', [historyEntry(7, 'COMPLETED', 4)]); fixture.detectChanges(); // first render: ngOnChanges (firstChange), ngOnInit, ngAfterViewInit fixture.componentRef.setInput('history', [historyEntry(8, 'COMPLETED', 4), ...]); fixture.detectChanges(); // ngOnChanges again, firstChange: false ``` ########## flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.ts: ########## @@ -0,0 +1,1047 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { DatePipe, NgFor, NgIf } from '@angular/common'; +import { + AfterViewInit, + ChangeDetectionStrategy, + ChangeDetectorRef, + Component, + ElementRef, + Input, + OnChanges, + OnDestroy, + SimpleChanges, + ViewChild +} from '@angular/core'; +import { BehaviorSubject, defer, EMPTY, Observable, of, Subject, timer } from 'rxjs'; +import { catchError, distinctUntilChanged, filter, map, repeat, switchMap, takeUntil } from 'rxjs/operators'; + +import { Chart } from '@antv/g2'; +import { HumanizeBytesPipe } from '@flink-runtime-web/components/humanize-bytes.pipe'; +import { HumanizeDurationPipe } from '@flink-runtime-web/components/humanize-duration.pipe'; +import { + CheckpointConfig, + CheckpointHistory, + CheckpointSubTask, + CompletedSubTaskCheckpointStatistics, + SubTaskCheckpointStatisticsItem, + VerticesItem +} from '@flink-runtime-web/interfaces'; +import { JobService } from '@flink-runtime-web/services'; +import { formatTimeAxis, startAlignedTicks } from '@flink-runtime-web/utils'; +import { NzAlertModule } from 'ng-zorro-antd/alert'; +import { NzEmptyModule } from 'ng-zorro-antd/empty'; +import { NzSpinModule } from 'ng-zorro-antd/spin'; +import { NzTooltipModule } from 'ng-zorro-antd/tooltip'; + +type PhaseKey = 'start_delay' | 'alignment' | 'sync' | 'async'; + +// 'not-retained': every vertex returned 404, i.e. the JobManager no longer keeps this checkpoint's stats. +type LoadError = 'not-retained' | 'request-failed'; + +interface TimelineLoad { + datum: StripDatum; + perVertex: Record<string, CheckpointSubTask | null>; + failedCount: number; + requestedCount: number; + error?: LoadError; +} + +interface StripDatum { + id: number; + idLabel: string; + duration: number; + status: 'COMPLETED' | 'SAVEPOINT' | 'IN_PROGRESS' | 'FAILED'; + rawStatus: string; + isSavepoint: boolean; + triggered: number; + acked: string; + size: number; +} + +interface TimelineDatum { + row: string; + operatorFull: string; + subtask: number; + phase: PhaseKey; + phaseLabel: string; + range: [number, number]; + durationMs: number; + aborted: boolean; + stateSize: number; + checkpointedSize: number; + totalDuration: number; +} + +// A subtask that reported no stats for the checkpoint, drawn as one hatched bar spanning it. +interface UnackedDatum { + row: string; + operatorFull: string; + subtask: number; + label: 'no ack' | 'pending'; + range: [number, number]; +} + +// A subtask whose phases all rounded to 0ms, drawn as a hairline at trigger time so its row isn't empty. +interface InstantDatum { + row: string; + operatorFull: string; + subtask: number; + aborted: boolean; + stateSize: number; + checkpointedSize: number; + range: [number, number]; +} + +interface TimelineRows { + rows: TimelineDatum[]; + unacked: UnackedDatum[]; + instants: InstantDatum[]; + rowOrder: string[]; +} + +const STRIP_COLOR: Record<StripDatum['status'], string> = { + COMPLETED: '#52c41a', + SAVEPOINT: '#722ed1', + IN_PROGRESS: '#faad14', + FAILED: '#f5222d' +}; + +const STRIP_STATUSES = Object.keys(STRIP_COLOR) as Array<StripDatum['status']>; + +const PHASE_COLOR: Record<PhaseKey, string> = { + start_delay: '#b8c0cc', + alignment: '#91caff', + sync: '#fa8c16', + async: '#52c41a' +}; + +const PHASES = Object.keys(PHASE_COLOR) as PhaseKey[]; + +const NOT_COMPLETED_STROKE = '#f5222d'; +const NO_ACK_FILL = '#fff1f0'; +const NO_ACK_STRIPE = '#ffa39e'; +const INSTANT_FILL = '#8c8c8c'; + +const OPERATOR_NAME_MAX = 60; + +type IntervalMark = ReturnType<Chart['interval']>; + +// Built once and updated in place on reloads, so following an in-progress checkpoint doesn't rebuild the canvas. +interface TimelineChart { + chart: Chart; + phases: IntervalMark; + unacked: IntervalMark; + instants: IntervalMark; +} + +@Component({ + selector: 'flink-job-checkpoints-timeline', + templateUrl: './job-checkpoints-timeline.component.html', + styleUrls: ['./job-checkpoints-timeline.component.less'], + changeDetection: ChangeDetectionStrategy.OnPush, + imports: [ + NgFor, + NgIf, + DatePipe, + HumanizeDurationPipe, + HumanizeBytesPipe, + NzAlertModule, + NzEmptyModule, + NzSpinModule, + NzTooltipModule + ] +}) +export class JobCheckpointsTimelineComponent implements AfterViewInit, OnChanges, OnDestroy { + @Input() public jobId: string; + @Input() public vertices: VerticesItem[] = []; + @Input() public history: CheckpointHistory[] = []; + @Input() public config?: CheckpointConfig; + + @ViewChild('stripContainer', { static: true }) private readonly stripContainer: ElementRef<HTMLDivElement>; + @ViewChild('timelineContainer', { static: true }) private readonly timelineContainer: ElementRef<HTMLDivElement>; + @ViewChild('timelineScroll', { static: true }) private readonly timelineScroll: ElementRef<HTMLDivElement>; + @ViewChild('timelineWrapper', { static: true }) private readonly timelineWrapper: ElementRef<HTMLDivElement>; + + public selected?: StripDatum; + public loadingTimeline = false; + public timelineEmpty = false; + public loadError?: LoadError; + public failedVertexCount = 0; + public requestedVertexCount = 0; + // Rows of the shown checkpoint, how many of the slowest are shown, and the first one scrolled to. + public totalRows = 0; + public slowestCount = JobCheckpointsTimelineComponent.SHOWN_ROWS_DEFAULT; + public firstRow = 0; + public chartHeight = 0; + public slowestFirst = false; + + // Diverges from @Input once polling starts feeding live updates. + private currentHistory: CheckpointHistory[] = []; + public userPinned = false; + private stripChart?: Chart; + private stripInterval?: IntervalMark; + private timeline?: TimelineChart; + // Full operator name per row label, for the native tooltip on the truncated labels. + private rowTitles = new Map<string, string>(); + // Kept so scrolling can redraw another window of rows without rebuilding them. + private timelineRows?: TimelineRows & { datum: StripDatum; slowestOrder: string[] }; + private scrollFrame?: number; + private noAckHatch?: HTMLCanvasElement; + private resizeObserver?: ResizeObserver; + private readonly destroy$ = new Subject<void>(); + // null cancels a load in flight. + private readonly loadTimeline$ = new Subject<StripDatum | null>(); + // A load requested while the Timeline tab was hidden; it runs once the tab is shown. + private loadDeferred = false; + // The checkpoint whose failed load was already retried while following, so polls don't retry it again. + private retriedCheckpointId?: number; + // Terminal (completed or failed) checkpoint stats no longer change; memoize by id for instant re-selects. + // Holds only the checkpoints that can be selected again: those in the strip, plus the pinned one. + private readonly detailsCache = new Map<number, Record<string, CheckpointSubTask | null>>(); + private readonly visible$ = new BehaviorSubject<boolean>(false); + private static readonly POLL_MIN_MS = 1500; + private static readonly POLL_MAX_MS = 15000; + private static readonly STRIP_MAX_BARS = 60; + private static readonly TIMELINE_ROW_HEIGHT = 22; + // paddingTop (time axis) + paddingBottom, plus a little slack. + private static readonly TIMELINE_HEIGHT_BASE = 56; + // Rows shown by default (the slowest) and the most that can be picked; the chart is at most VISIBLE_ROWS tall. + private static readonly SHOWN_ROWS_DEFAULT = 25; + private static readonly SHOWN_ROWS_MAX = 200; + private static readonly VISIBLE_ROWS = 25; + private static readonly TIMELINE_HEIGHT_MIN = 160; + private static readonly LABEL_CHAR_WIDTH = 7; + private static readonly LABEL_PADDING_BUFFER = 24; + private static readonly LABEL_PADDING_MIN = 260; + private static readonly LABEL_PADDING_MAX = 520; + + private readonly humanizeBytesPipe = new HumanizeBytesPipe(); + private readonly humanizeDurationPipe = new HumanizeDurationPipe(); + + constructor( + private readonly jobService: JobService, + private readonly cdr: ChangeDetectorRef + ) {} + + public ngAfterViewInit(): void { + this.currentHistory = this.history || []; + this.renderStrip(); + this.subscribeTimelineLoads(); + if (this.currentHistory.length > 0) { + this.selectInternal(this.toStripDatum(this.currentHistory[0]), false); + } + this.startPolling(); Review Comment: What was the reason for polling inside the component? If the parent polled, would this give a better experience on sibling pages as well as this one? ########## flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.spec.ts: ########## @@ -0,0 +1,672 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { HttpErrorResponse } from '@angular/common/http'; +import { ChangeDetectorRef, ElementRef, SimpleChange } from '@angular/core'; +import { BehaviorSubject, of, Subject, throwError } from 'rxjs'; + +import { + CheckpointConfig, + CheckpointHistory, + CheckpointSubTask, + SubTaskCheckpointStatisticsItem, + VerticesItem +} from '@flink-runtime-web/interfaces'; +import { JobService } from '@flink-runtime-web/services'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { JobCheckpointsTimelineComponent } from './job-checkpoints-timeline.component'; + +const TRIGGERED = 1_000; + +interface Phases { + start_delay?: number; + alignment?: number; + sync?: number; + async?: number; +} + +const completed = (index: number, phases: Phases, aborted = false): SubTaskCheckpointStatisticsItem => + ({ + index, + status: 'completed', + ack_timestamp: 0, + end_to_end_duration: 0, + checkpointed_size: 0, + state_size: 0, + checkpoint: { sync: phases.sync ?? 0, async: phases.async ?? 0 }, + alignment: { buffer: 0, processed: 0, persisted: 0, duration: phases.alignment ?? 0 }, + start_delay: phases.start_delay ?? 0, + unaligned_checkpoint: false, + aborted + }) as SubTaskCheckpointStatisticsItem; + +const noStats = (index: number): SubTaskCheckpointStatisticsItem => + ({ index, status: 'pending_or_failed' }) as SubTaskCheckpointStatisticsItem; + +const details = (...subtasks: SubTaskCheckpointStatisticsItem[]): CheckpointSubTask => + ({ subtasks }) as unknown as CheckpointSubTask; + +const vertex = (id: string, name: string): VerticesItem => ({ id, name, status: 'RUNNING' }) as VerticesItem; + +const checkpoint = ( + rawStatus: string, + duration: number, + isSavepoint = false +): { id: number; duration: number; rawStatus: string; isSavepoint: boolean; triggered: number } => ({ + id: 7, + duration, + rawStatus, + isSavepoint, + triggered: TRIGGERED +}); + +type BuildRows = ( + datum: ReturnType<typeof checkpoint>, + perVertex: Record<string, CheckpointSubTask | null> +) => { + rows: Array<{ row: string; phase: string; phaseLabel: string; range: [number, number]; totalDuration: number }>; + unacked: Array<{ row: string; label: string; range: [number, number] }>; + instants: Array<{ row: string; aborted: boolean; range: [number, number] }>; + rowOrder: string[]; +}; + +const unalignedConfig = (alignedCheckpointTimeout: number, mode = 'exactly_once'): CheckpointConfig => + ({ mode, unaligned_checkpoints: true, aligned_checkpoint_timeout: alignedCheckpointTimeout }) as CheckpointConfig; + +const SUBTASK_PHASES = { start_delay: 2, alignment: 5, sync: 1, async: 8 }; + +describe('JobCheckpointsTimelineComponent rows', () => { Review Comment: Wondering if the row building could be a pure function imported by this component? Makes the component smaller, and we can then test it without driving component internals. ########## flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.ts: ########## @@ -0,0 +1,1047 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { DatePipe, NgFor, NgIf } from '@angular/common'; +import { + AfterViewInit, + ChangeDetectionStrategy, + ChangeDetectorRef, + Component, + ElementRef, + Input, + OnChanges, + OnDestroy, + SimpleChanges, + ViewChild +} from '@angular/core'; +import { BehaviorSubject, defer, EMPTY, Observable, of, Subject, timer } from 'rxjs'; +import { catchError, distinctUntilChanged, filter, map, repeat, switchMap, takeUntil } from 'rxjs/operators'; + +import { Chart } from '@antv/g2'; +import { HumanizeBytesPipe } from '@flink-runtime-web/components/humanize-bytes.pipe'; +import { HumanizeDurationPipe } from '@flink-runtime-web/components/humanize-duration.pipe'; +import { + CheckpointConfig, + CheckpointHistory, + CheckpointSubTask, + CompletedSubTaskCheckpointStatistics, + SubTaskCheckpointStatisticsItem, + VerticesItem +} from '@flink-runtime-web/interfaces'; +import { JobService } from '@flink-runtime-web/services'; +import { formatTimeAxis, startAlignedTicks } from '@flink-runtime-web/utils'; +import { NzAlertModule } from 'ng-zorro-antd/alert'; +import { NzEmptyModule } from 'ng-zorro-antd/empty'; +import { NzSpinModule } from 'ng-zorro-antd/spin'; +import { NzTooltipModule } from 'ng-zorro-antd/tooltip'; + +type PhaseKey = 'start_delay' | 'alignment' | 'sync' | 'async'; + +// 'not-retained': every vertex returned 404, i.e. the JobManager no longer keeps this checkpoint's stats. +type LoadError = 'not-retained' | 'request-failed'; + +interface TimelineLoad { + datum: StripDatum; + perVertex: Record<string, CheckpointSubTask | null>; + failedCount: number; + requestedCount: number; + error?: LoadError; +} + +interface StripDatum { + id: number; + idLabel: string; + duration: number; + status: 'COMPLETED' | 'SAVEPOINT' | 'IN_PROGRESS' | 'FAILED'; + rawStatus: string; + isSavepoint: boolean; + triggered: number; + acked: string; + size: number; +} + +interface TimelineDatum { + row: string; + operatorFull: string; + subtask: number; + phase: PhaseKey; + phaseLabel: string; + range: [number, number]; + durationMs: number; + aborted: boolean; + stateSize: number; + checkpointedSize: number; + totalDuration: number; +} + +// A subtask that reported no stats for the checkpoint, drawn as one hatched bar spanning it. +interface UnackedDatum { + row: string; + operatorFull: string; + subtask: number; + label: 'no ack' | 'pending'; + range: [number, number]; +} + +// A subtask whose phases all rounded to 0ms, drawn as a hairline at trigger time so its row isn't empty. +interface InstantDatum { + row: string; + operatorFull: string; + subtask: number; + aborted: boolean; + stateSize: number; + checkpointedSize: number; + range: [number, number]; +} + +interface TimelineRows { + rows: TimelineDatum[]; + unacked: UnackedDatum[]; + instants: InstantDatum[]; + rowOrder: string[]; +} + +const STRIP_COLOR: Record<StripDatum['status'], string> = { + COMPLETED: '#52c41a', + SAVEPOINT: '#722ed1', + IN_PROGRESS: '#faad14', + FAILED: '#f5222d' +}; + +const STRIP_STATUSES = Object.keys(STRIP_COLOR) as Array<StripDatum['status']>; + +const PHASE_COLOR: Record<PhaseKey, string> = { + start_delay: '#b8c0cc', + alignment: '#91caff', + sync: '#fa8c16', + async: '#52c41a' +}; + +const PHASES = Object.keys(PHASE_COLOR) as PhaseKey[]; + +const NOT_COMPLETED_STROKE = '#f5222d'; +const NO_ACK_FILL = '#fff1f0'; +const NO_ACK_STRIPE = '#ffa39e'; +const INSTANT_FILL = '#8c8c8c'; + +const OPERATOR_NAME_MAX = 60; + +type IntervalMark = ReturnType<Chart['interval']>; + +// Built once and updated in place on reloads, so following an in-progress checkpoint doesn't rebuild the canvas. +interface TimelineChart { + chart: Chart; + phases: IntervalMark; + unacked: IntervalMark; + instants: IntervalMark; +} + +@Component({ + selector: 'flink-job-checkpoints-timeline', + templateUrl: './job-checkpoints-timeline.component.html', + styleUrls: ['./job-checkpoints-timeline.component.less'], + changeDetection: ChangeDetectionStrategy.OnPush, + imports: [ + NgFor, + NgIf, + DatePipe, + HumanizeDurationPipe, + HumanizeBytesPipe, + NzAlertModule, + NzEmptyModule, + NzSpinModule, + NzTooltipModule + ] +}) +export class JobCheckpointsTimelineComponent implements AfterViewInit, OnChanges, OnDestroy { + @Input() public jobId: string; + @Input() public vertices: VerticesItem[] = []; + @Input() public history: CheckpointHistory[] = []; + @Input() public config?: CheckpointConfig; + + @ViewChild('stripContainer', { static: true }) private readonly stripContainer: ElementRef<HTMLDivElement>; + @ViewChild('timelineContainer', { static: true }) private readonly timelineContainer: ElementRef<HTMLDivElement>; + @ViewChild('timelineScroll', { static: true }) private readonly timelineScroll: ElementRef<HTMLDivElement>; + @ViewChild('timelineWrapper', { static: true }) private readonly timelineWrapper: ElementRef<HTMLDivElement>; + + public selected?: StripDatum; + public loadingTimeline = false; + public timelineEmpty = false; + public loadError?: LoadError; + public failedVertexCount = 0; + public requestedVertexCount = 0; + // Rows of the shown checkpoint, how many of the slowest are shown, and the first one scrolled to. + public totalRows = 0; + public slowestCount = JobCheckpointsTimelineComponent.SHOWN_ROWS_DEFAULT; + public firstRow = 0; + public chartHeight = 0; + public slowestFirst = false; + + // Diverges from @Input once polling starts feeding live updates. + private currentHistory: CheckpointHistory[] = []; + public userPinned = false; + private stripChart?: Chart; + private stripInterval?: IntervalMark; + private timeline?: TimelineChart; + // Full operator name per row label, for the native tooltip on the truncated labels. + private rowTitles = new Map<string, string>(); + // Kept so scrolling can redraw another window of rows without rebuilding them. + private timelineRows?: TimelineRows & { datum: StripDatum; slowestOrder: string[] }; + private scrollFrame?: number; + private noAckHatch?: HTMLCanvasElement; + private resizeObserver?: ResizeObserver; + private readonly destroy$ = new Subject<void>(); + // null cancels a load in flight. + private readonly loadTimeline$ = new Subject<StripDatum | null>(); + // A load requested while the Timeline tab was hidden; it runs once the tab is shown. + private loadDeferred = false; + // The checkpoint whose failed load was already retried while following, so polls don't retry it again. + private retriedCheckpointId?: number; + // Terminal (completed or failed) checkpoint stats no longer change; memoize by id for instant re-selects. + // Holds only the checkpoints that can be selected again: those in the strip, plus the pinned one. + private readonly detailsCache = new Map<number, Record<string, CheckpointSubTask | null>>(); + private readonly visible$ = new BehaviorSubject<boolean>(false); + private static readonly POLL_MIN_MS = 1500; + private static readonly POLL_MAX_MS = 15000; + private static readonly STRIP_MAX_BARS = 60; + private static readonly TIMELINE_ROW_HEIGHT = 22; + // paddingTop (time axis) + paddingBottom, plus a little slack. + private static readonly TIMELINE_HEIGHT_BASE = 56; + // Rows shown by default (the slowest) and the most that can be picked; the chart is at most VISIBLE_ROWS tall. + private static readonly SHOWN_ROWS_DEFAULT = 25; + private static readonly SHOWN_ROWS_MAX = 200; + private static readonly VISIBLE_ROWS = 25; + private static readonly TIMELINE_HEIGHT_MIN = 160; + private static readonly LABEL_CHAR_WIDTH = 7; + private static readonly LABEL_PADDING_BUFFER = 24; + private static readonly LABEL_PADDING_MIN = 260; + private static readonly LABEL_PADDING_MAX = 520; + + private readonly humanizeBytesPipe = new HumanizeBytesPipe(); + private readonly humanizeDurationPipe = new HumanizeDurationPipe(); + + constructor( + private readonly jobService: JobService, + private readonly cdr: ChangeDetectorRef + ) {} + + public ngAfterViewInit(): void { + this.currentHistory = this.history || []; + this.renderStrip(); + this.subscribeTimelineLoads(); + if (this.currentHistory.length > 0) { + this.selectInternal(this.toStripDatum(this.currentHistory[0]), false); + } + this.startPolling(); + // G2 locks autoFit at construction; re-fit + resume polling once width is real. + this.visible$.next(this.stripContainer.nativeElement.clientWidth > 0); + if (typeof ResizeObserver !== 'undefined') { + this.resizeObserver = new ResizeObserver(() => { + const stripVisible = this.stripContainer.nativeElement.clientWidth > 0; + if (this.stripChart && stripVisible) { + void this.stripChart.forceFit(); + } + if (this.timeline && this.timelineContainer.nativeElement.clientWidth > 0) { + void this.timeline.chart.forceFit(); + } + this.visible$.next(stripVisible); + }); + this.resizeObserver.observe(this.stripContainer.nativeElement); + this.resizeObserver.observe(this.timelineContainer.nativeElement); + } + } + + public ngOnChanges(changes: SimpleChanges): void { + if (!this.stripContainer || !this.stripContainer.nativeElement) { + return; + } + if (changes['jobId'] && !changes['jobId'].firstChange) { + this.userPinned = false; + this.selected = undefined; + this.currentHistory = []; + // Otherwise a load still in flight for the old job lands in the cache under an id the new job reuses. + this.loadTimeline$.next(null); + this.detailsCache.clear(); + this.retriedCheckpointId = undefined; + this.clearTimeline(); + this.renderStrip(); + } + if (changes['history']) { + this.applyHistory(this.history || []); + } + if (changes['vertices'] && !changes['vertices'].firstChange && this.selected) { + this.loadTimeline(this.selected); + } + } + + public ngOnDestroy(): void { + this.destroy$.next(); + this.destroy$.complete(); + this.resizeObserver?.disconnect(); + if (this.scrollFrame !== undefined) { + cancelAnimationFrame(this.scrollFrame); + } + this.stripChart?.destroy(); + this.timeline?.chart.destroy(); + } + + private toStripDatum(h: CheckpointHistory): StripDatum { + // FAILED first, so a failed savepoint shows as failed rather than as a savepoint. + const status: StripDatum['status'] = + h.status === 'FAILED' + ? 'FAILED' + : h.is_savepoint + ? 'SAVEPOINT' + : h.status === 'IN_PROGRESS' + ? 'IN_PROGRESS' + : 'COMPLETED'; + return { + id: h.id, + idLabel: String(h.id), + duration: Math.max(h.end_to_end_duration || 0, 0), + status, + rawStatus: h.status, + isSavepoint: h.is_savepoint, + triggered: h.trigger_timestamp, + acked: `${h.num_acknowledged_subtasks}/${h.num_subtasks}`, + size: h.state_size + }; + } + + private renderStrip(): void { + const data = (this.currentHistory || []) + .slice(0, JobCheckpointsTimelineComponent.STRIP_MAX_BARS) + .map(h => this.toStripDatum(h)) + .reverse(); + + if (!this.stripChart || !this.stripInterval) { + this.stripChart = new Chart({ + container: this.stripContainer.nativeElement, + autoFit: true, + height: 110, + paddingTop: 16, + paddingRight: 16, + paddingBottom: 36, + paddingLeft: 56 + }); + const isPinned = (d: StripDatum): boolean => this.userPinned && this.selected?.id === d.id; + this.stripInterval = this.stripChart + .interval() + .encode('x', 'idLabel') + .encode('y', 'duration') + .encode('color', 'status') + .scale('y', { nice: true, domainMin: 0 }) + .scale('color', { domain: STRIP_STATUSES, range: STRIP_STATUSES.map(s => STRIP_COLOR[s]) }) + .axis('x', { title: false, labelAutoHide: true, labelAutoRotate: false }) + .axis('y', { title: false }) + .legend(false) + .animate(false) + .style('maxWidth', 18) + // Keeps 0ms bars (e.g. in progress before the first ack) and very short ones visible and clickable. + .style('minHeight', 3) + .style('stroke', (d: StripDatum) => (isPinned(d) ? '#1890ff' : 'transparent')) + .style('lineWidth', (d: StripDatum) => (isPinned(d) ? 2.5 : 0)) + .tooltip({ + title: '', + items: [ + (d: StripDatum) => ({ + name: `#${d.idLabel}`, + color: STRIP_COLOR[d.status], + value: `${d.rawStatus} · ${this.humanizeDurationPipe.transform(d.duration)} · acked ${d.acked}` + }) + ] + }); + this.stripChart.on('interval:click', (ev: { data?: { data?: StripDatum } }) => { + const datum = ev.data?.data; + if (datum) { + this.select(datum); + } + }); + } + + this.stripInterval.scale('x', { domain: data.map(d => d.idLabel) }); + this.stripInterval.data(data); + void this.stripChart.render(); + } + + public select(datum: StripDatum): void { + // Clicking the already-pinned bar resumes auto-follow. + if (this.userPinned && this.selected && this.selected.id === datum.id) { + this.followNewest(); + return; + } + this.selectInternal(datum, true); + } + + public followNewest(): void { + this.userPinned = false; + if (this.currentHistory.length > 0) { + this.selectInternal(this.toStripDatum(this.currentHistory[0]), false); + } else { + this.cdr.markForCheck(); + } + } + + private selectInternal(datum: StripDatum, pinned: boolean): void { + this.selected = datum; + if (pinned) { + this.userPinned = true; + } + this.cdr.markForCheck(); + this.renderStrip(); + this.loadTimeline(datum); + } + + private applyHistory(next: CheckpointHistory[]): void { + const prevNewestId = this.currentHistory.length > 0 ? this.currentHistory[0].id : -1; + const nextNewestId = next.length > 0 ? next[0].id : -1; + this.currentHistory = next; + this.pruneDetailsCache(); + this.renderStrip(); + // The empty state and the rotated-out note read currentHistory. + this.cdr.markForCheck(); + + if (next.length === 0) { + // The pinned checkpoint is gone too (e.g. after a JobManager failover), so follow whatever comes next. + this.userPinned = false; + this.selected = undefined; + this.clearTimeline(); + return; + } + + if (!this.userPinned && nextNewestId !== prevNewestId) { + this.selectInternal(this.toStripDatum(next[0]), false); + return; + } + // Still showing the same checkpoint (the followed newest one, or the pinned one): reload once it gained acks or + // finished, or retry a failed load once (retrying on every poll would repeat the error notifications). Not while + // a load is in flight, so polling doesn't keep cancelling a slow fetch. + const selected = this.selected; + const latest = this.userPinned ? next.find(h => h.id === selected?.id) : next[0]; + if (!selected || !latest || latest.id !== selected.id || this.loadingTimeline) { + return; + } + const current = this.toStripDatum(latest); + const progressed = current.rawStatus !== selected.rawStatus || current.acked !== selected.acked; + const retry = + (this.loadError !== undefined || this.failedVertexCount > 0) && this.retriedCheckpointId !== current.id; + if (retry) { + this.retriedCheckpointId = current.id; + } + if (progressed || retry) { + this.selectInternal(current, false); + } + } + + private pruneDetailsCache(): void { + const selectable = new Set( + this.currentHistory.slice(0, JobCheckpointsTimelineComponent.STRIP_MAX_BARS).map(h => h.id) + ); + for (const id of this.detailsCache.keys()) { + if (!selectable.has(id) && id !== this.selected?.id) { + this.detailsCache.delete(id); + } + } + } + + private startPolling(): void { + this.visible$ + .pipe( + distinctUntilChanged(), + switchMap(visible => + visible + ? defer(() => { + const jobId = this.jobId; + return jobId + ? this.jobService.loadCheckpointStats(jobId).pipe( + catchError(() => of(undefined)), + // A response for the previous job can land after a jobId change has reset the state. + filter(() => jobId === this.jobId) + ) + : of(undefined); + }).pipe(repeat({ delay: () => timer(this.computePollDelayMs()) })) + : EMPTY + ), + takeUntil(this.destroy$) + ) + .subscribe(stats => { + if (!stats) { + return; + } + this.applyHistory(stats.history || []); + }); + } + + private computePollDelayMs(): number { + // Disabled periodic checkpointing reports Long.MAX_VALUE, which the upper clamp covers. + const interval = this.config?.interval ?? 0; + return interval > 0 + ? Math.max( + JobCheckpointsTimelineComponent.POLL_MIN_MS, + Math.min(interval, JobCheckpointsTimelineComponent.POLL_MAX_MS) + ) + : JobCheckpointsTimelineComponent.POLL_MAX_MS; + } + + private loadTimeline(datum: StripDatum): void { + // The component also lives in the hidden tab; don't fetch every vertex's details until it is shown. + if (!this.visible$.value) { + this.loadDeferred = true; + return; + } + this.loadDeferred = false; + if (!this.jobId || this.vertices.length === 0) { + this.showEmptyTimeline(); + return; + } + this.loadingTimeline = true; + this.timelineEmpty = false; + this.loadError = undefined; + this.failedVertexCount = 0; + this.cdr.markForCheck(); + this.loadTimeline$.next(datum); + } + + private cacheDetails(datum: StripDatum, perVertex: Record<string, CheckpointSubTask | null>): void { + // In-progress stats still change; only cache terminal checkpoints. `status` is 'SAVEPOINT' for an in-progress + // savepoint, so check rawStatus. + if (datum.rawStatus !== 'IN_PROGRESS') { + this.detailsCache.set(datum.id, perVertex); + } + } + + private subscribeTimelineLoads(): void { + this.visible$ + .pipe( + filter(visible => visible && this.loadDeferred), + takeUntil(this.destroy$) + ) + .subscribe(() => { + if (this.selected) { + this.loadTimeline(this.selected); + } + }); + + this.loadTimeline$ + .pipe( + switchMap((datum): Observable<TimelineLoad> => { + if (!datum) { + return EMPTY; + } + const requestedCount = this.vertices.length; + const cached = this.detailsCache.get(datum.id); + if (cached) { + return of({ datum, perVertex: cached, failedCount: 0, requestedCount }); + } + const vertexIds = this.vertices.map(v => v.id); + return this.jobService.loadCheckpointAllSubtaskDetails(this.jobId, datum.id, vertexIds).pipe( + map(perVertex => { + const failedCount = requestedCount - Object.values(perVertex).filter(v => v !== null).length; + if (failedCount === 0) { + this.cacheDetails(datum, perVertex); + } + const error: LoadError | undefined = + requestedCount > 0 && failedCount === requestedCount ? 'not-retained' : undefined; + return { datum, perVertex, failedCount, requestedCount, error }; + }), + catchError(() => + of<TimelineLoad>({ + datum, + perVertex: {}, + failedCount: requestedCount, + requestedCount, + error: 'request-failed' + }) + ) + ); + }), + takeUntil(this.destroy$) + ) + .subscribe(({ datum, perVertex, failedCount, requestedCount, error }) => { + this.loadingTimeline = false; + if (error) { + this.showEmptyTimeline(error); + return; + } + this.failedVertexCount = failedCount; + this.requestedVertexCount = requestedCount; + this.renderTimeline(datum, perVertex); + }); + } + + private buildTimelineRows(datum: StripDatum, perVertex: Record<string, CheckpointSubTask | null>): TimelineRows { + const base = datum.triggered; + // A completed checkpoint was acked by every running subtask, so missing stats there are not a failure. + const unackedLabel = + datum.rawStatus === 'FAILED' ? 'no ack' : datum.rawStatus === 'IN_PROGRESS' ? 'pending' : undefined; + const rows: TimelineDatum[] = []; + const unacked: UnackedDatum[] = []; + const instants: InstantDatum[] = []; + const rowOrder: string[] = []; + let maxTotal = 0; + const rowPrefixOwner = new Map<string, string>(); + + for (const vertex of this.vertices) { + const cp = perVertex[vertex.id]; + if (!cp || !cp.subtasks) { + continue; + } + const opNameFull = (vertex.name || '').replace(/\s+/g, ' ').trim(); + let opName = shortenOperatorName(vertex.name); + // shortenOperatorName truncates, so distinct vertices can collide; the row key must stay unique. + const owner = rowPrefixOwner.get(opName); + if (owner !== undefined && owner !== vertex.id) { + opName = `${opName} [${vertex.id.slice(0, 4)}]`; + } + rowPrefixOwner.set(opName, vertex.id); + + for (const s of cp.subtasks) { + const row = `${opName} #${s.index}`; + if (!isCompletedSubtask(s)) { + // Finished tasks never land here: the JobManager records them as acknowledged when it triggers. + if (unackedLabel) { + unacked.push({ row, operatorFull: opNameFull, subtask: s.index, label: unackedLabel, range: [base, base] }); + rowOrder.push(row); + } + continue; + } + const startDelay = Math.max(s.start_delay || 0, 0); + const reportedAlignment = Math.max(s.alignment.duration || 0, 0); + const alignment = this.alignmentBeforeSync(datum, startDelay, reportedAlignment); + const alignmentInAsync = reportedAlignment - alignment; + const sync = Math.max(s.checkpoint.sync || 0, 0); + const asyncMs = Math.max(s.checkpoint.async || 0, 0); + const total = startDelay + alignment + sync + asyncMs; + rowOrder.push(row); + maxTotal = Math.max(maxTotal, total); + if (total === 0) { + instants.push({ + row, + operatorFull: opNameFull, + subtask: s.index, + aborted: !!s.aborted, + stateSize: s.state_size, + checkpointedSize: s.checkpointed_size, + range: [base, base] + }); + continue; + } + + const segs: Array<[PhaseKey, number]> = [ + ['start_delay', startDelay], + ['alignment', alignment], + ['sync', sync], + ['async', asyncMs] + ]; + let cursor = base; + for (const [phase, ms] of segs) { + if (ms > 0) { + rows.push({ + row, + operatorFull: opNameFull, + subtask: s.index, + phase, + phaseLabel: + phase === 'async' && alignmentInAsync > 0 ? `${phase} (incl. ${alignmentInAsync}ms alignment)` : phase, + range: [cursor, cursor + ms], + durationMs: ms, + aborted: !!s.aborted, + stateSize: s.state_size, + checkpointedSize: s.checkpointed_size, + totalDuration: total + }); + } + cursor += ms; + } + } + } + + // Unacked subtasks span the whole checkpoint: until it failed, or the latest ack so far while in progress. + const unackedEnd = base + Math.max(datum.duration, maxTotal, 1); + for (const u of unacked) { + u.range = [base, unackedEnd]; + } + + return { rows, unacked, instants, rowOrder }; + } + + // How much of a subtask's alignment came before its sync phase. With unaligned checkpoints, a checkpoint starts + // aligned and switches to unaligned once aligned_checkpoint_timeout has passed since the trigger; after that the + // rest of the alignment runs during the async phase, whose duration includes it (checkpoint_monitoring.md). + // Savepoints and at-least-once checkpoints never switch. The per-subtask unaligned_checkpoint flag can't tell: it + // only says whether in-flight data was persisted. + private alignmentBeforeSync(datum: StripDatum, startDelay: number, alignment: number): number { + const config = this.config; + if (!config?.unaligned_checkpoints || config.mode !== 'exactly_once' || datum.isSavepoint) { + return alignment; + } + const timeout = config.aligned_checkpoint_timeout; + // 0 and Long.MAX_VALUE both mean unaligned from the first barrier (CheckpointOptions.forConfig). + if (!(timeout > 0) || timeout >= Number.MAX_SAFE_INTEGER) { + return 0; + } + return Math.max(0, Math.min(alignment, timeout - startDelay)); + } + + private renderTimeline(datum: StripDatum, perVertex: Record<string, CheckpointSubTask | null>): void { + const built = this.buildTimelineRows(datum, perVertex); + if (built.rowOrder.length === 0) { + this.showEmptyTimeline(); + return; + } + this.timelineEmpty = false; + this.timelineRows = { datum, ...built, slowestOrder: slowestFirstOrder(built) }; + this.totalRows = built.rowOrder.length; + this.rowTitles = new Map(); + for (const row of [...built.rows, ...built.unacked, ...built.instants]) { + if (!this.rowTitles.has(row.row)) { + this.rowTitles.set(row.row, `${row.operatorFull} #${row.subtask}`); + } + } + this.drawTimelineWindow(); + } + + // Draws the rows scrolled to, out of the slowest ones picked. The chart is at most 25 rows tall and stays in place + // while the box around it scrolls, so its canvas stays far below the 65,535 px browsers allow, however big the job. + private drawTimelineWindow(): void { + const data = this.timelineRows; + if (!data) { + return; + } + const { datum, rows, unacked, instants, rowOrder } = data; + const C = JobCheckpointsTimelineComponent; + const slowest = data.slowestOrder.slice(0, this.shownRows); + const picked = new Set(slowest); + const order = this.slowestFirst ? slowest : rowOrder.filter(row => picked.has(row)); + const visible = this.visibleRows; + this.firstRow = Math.max(0, Math.min(this.firstRow, order.length - visible)); + const windowRows = order.slice(this.firstRow, this.firstRow + visible); + const inWindow = new Set(windowRows); + const shown = <T extends { row: string }>(items: T[]): T[] => items.filter(d => inWindow.has(d.row)); + this.chartHeight = Math.max(visible * C.TIMELINE_ROW_HEIGHT + C.TIMELINE_HEIGHT_BASE, C.TIMELINE_HEIGHT_MIN); + + // The label padding and the time scale cover all rows, so scrolling or picking more rows doesn't shift them. + const longestLabel = rowOrder.reduce((m, s) => Math.max(m, s.length), 0); + const leftPadding = Math.min( + C.LABEL_PADDING_MAX, + Math.max(C.LABEL_PADDING_MIN, longestLabel * C.LABEL_CHAR_WIDTH + C.LABEL_PADDING_BUFFER) + ); + const scaleMin = datum.triggered; + const scaleMax = Math.max( + [...rows, ...unacked].reduce((m, r) => Math.max(m, r.range[1]), scaleMin), + scaleMin + 1 + ); + const xScale = { domain: windowRows }; + const yScale = { + domain: [scaleMin, scaleMax], + nice: false, + tickMethod: () => startAlignedTicks(new Date(scaleMin), new Date(scaleMax)).map(Number) + }; + + const timeline = (this.timeline ??= this.createTimelineChart()); + timeline.chart.attr('height', this.chartHeight); + timeline.chart.attr('paddingLeft', leftPadding); + timeline.phases.data(shown(rows)).scale('x', xScale).scale('y', yScale); + timeline.unacked.data(shown(unacked)).scale('x', xScale).scale('y', yScale); + timeline.instants.data(shown(instants)).scale('x', xScale).scale('y', yScale); + void timeline.chart.render(); + this.cdr.markForCheck(); + } + + // How many rows are drawn: the slowest ones picked, or all rows if there are fewer. + public get shownRows(): number { + return Math.min(this.slowestCount, this.totalRows); + } + + public get visibleRows(): number { + return Math.min(JobCheckpointsTimelineComponent.VISIBLE_ROWS, this.shownRows); + } + + public get scrollable(): boolean { + return this.shownRows > this.visibleRows; + } + + // The scroll box's content: the chart plus one row height for every shown row out of view. + public get scrollContentHeight(): number { + return this.chartHeight + (this.shownRows - this.visibleRows) * JobCheckpointsTimelineComponent.TIMELINE_ROW_HEIGHT; + } + + // 25, 50, ... up to 200 or the number of rows, whichever is smaller; none when every row is already shown. + public get slowestCountOptions(): number[] { + const C = JobCheckpointsTimelineComponent; + const max = Math.min(C.SHOWN_ROWS_MAX, this.totalRows); + if (max <= C.SHOWN_ROWS_DEFAULT) { + return []; + } + const options: number[] = []; + for (let n = C.SHOWN_ROWS_DEFAULT; n < max; n += C.SHOWN_ROWS_DEFAULT) { + options.push(n); + } + options.push(max); + return options; + } + + public setSlowestCount(count: number): void { + this.slowestCount = count; + this.scrollToTop(); + } + + public setSlowestFirst(slowestFirst: boolean): void { + this.slowestFirst = slowestFirst; + this.scrollToTop(); + } + + // Other rows or another order: start again from the top. + private scrollToTop(): void { + this.firstRow = 0; + this.timelineScroll.nativeElement.scrollTop = 0; + this.drawTimelineWindow(); + } + + public onSlowestFirstChange(event: Event): void { + this.setSlowestFirst((event.target as HTMLInputElement).checked); + } + + public onSlowestCountChange(event: Event): void { + this.setSlowestCount(Number((event.target as HTMLSelectElement).value)); + } + + public onTimelineScroll(): void { + const first = Math.round( + this.timelineScroll.nativeElement.scrollTop / JobCheckpointsTimelineComponent.TIMELINE_ROW_HEIGHT + ); + if (first === this.firstRow) { + return; + } + this.firstRow = first; + // One redraw per frame, however many scroll events arrive. + if (this.scrollFrame === undefined) { + this.scrollFrame = requestAnimationFrame(() => { + this.scrollFrame = undefined; + this.drawTimelineWindow(); + }); + } + } + + private createTimelineChart(): TimelineChart { + const container = this.timelineContainer.nativeElement; + const chart = new Chart({ container, autoFit: true, paddingTop: 32, paddingRight: 32, paddingBottom: 16 }); + chart.coordinate({ transform: [{ type: 'transpose' }] }); + const phases = chart + .interval() + .encode('x', 'row') + .encode('y', 'range') + .encode('color', 'phase') + .scale('color', { domain: PHASES, range: PHASES.map(p => PHASE_COLOR[p]) }) + .axis('x', { + title: false, + labelAutoHide: false, + labelAutoRotate: false, + labelFill: '#262626', + labelOpacity: 1 + }) + // On top, so the time axis is in view as soon as the chart is. + .axis('y', { title: false, position: 'top', labelFormatter: (v: number) => formatTimeAxis(new Date(Number(v))) }) + .legend(false) + .animate(false) + .style('stroke', (d: TimelineDatum) => (d.aborted ? NOT_COMPLETED_STROKE : 'transparent')) + .style('lineWidth', (d: TimelineDatum) => (d.aborted ? 1.5 : 0)) + .style('lineDash', (d: TimelineDatum) => (d.aborted ? [3, 2] : [0, 0])) + .tooltip({ + title: (d: TimelineDatum) => `${d.operatorFull} #${d.subtask}`, + items: [ + (d: TimelineDatum) => ({ name: d.phaseLabel, color: PHASE_COLOR[d.phase], value: `${d.durationMs}ms` }), + (d: TimelineDatum) => ({ name: 'total', color: 'transparent', value: `${d.totalDuration}ms` }), + (d: TimelineDatum) => ({ + name: 'state', + color: 'transparent', + value: this.humanizeBytesPipe.transform(d.stateSize) + }), + (d: TimelineDatum) => ({ + name: 'checkpointed', + color: 'transparent', + value: this.humanizeBytesPipe.transform(d.checkpointedSize) + }) + ] + }) + .interaction('tooltip', { + // Outside the scroll box, which would clip it on the last visible rows. g2 places a tooltip before measuring + // it, so on its first frame it doesn't flip up yet. + mount: this.timelineWrapper.nativeElement, + // g2 5 ellipsizes the tooltip title at 360px; let full operator names wrap instead. + css: { + '.g2-tooltip': { 'max-width': '480px' }, + '.g2-tooltip-title': { 'white-space': 'normal', 'overflow-wrap': 'anywhere' } + } + }); + // Returned from a callback: g2 deep-copies constant style objects, which would turn the canvas into a plain object. + const hatchFill = { image: (this.noAckHatch ??= createHatchPattern()), repetition: 'repeat' }; + const unacked = chart + .interval() + .encode('x', 'row') + .encode('y', 'range') + .animate(false) + .style('fill', () => hatchFill) + .style('stroke', NOT_COMPLETED_STROKE) + .style('lineWidth', 1.5) + .style('lineDash', [3, 2]) + .tooltip({ + title: (d: UnackedDatum) => `${d.operatorFull} #${d.subtask}`, + items: [(d: UnackedDatum) => ({ name: d.label, color: NOT_COMPLETED_STROKE, value: 'no stats reported' })] + }); + const instants = chart + .interval() + .encode('x', 'row') + .encode('y', 'range') + .animate(false) + // The range is zero-length; once transposed, minHeight is the on-screen length along the time axis. + .style('minHeight', 2) + .style('fill', (d: InstantDatum) => (d.aborted ? NOT_COMPLETED_STROKE : INSTANT_FILL)) + .tooltip({ + title: (d: InstantDatum) => `${d.operatorFull} #${d.subtask}`, + // Same rows as the phase bars, so 'total' always means the sum of the phases. + items: [ + (d: InstantDatum) => ({ + name: 'all phases', + color: d.aborted ? NOT_COMPLETED_STROKE : INSTANT_FILL, + value: '0ms' + }), + () => ({ name: 'total', color: 'transparent', value: '0ms' }), + (d: InstantDatum) => ({ + name: 'state', + color: 'transparent', + value: this.humanizeBytesPipe.transform(d.stateSize) + }), + (d: InstantDatum) => ({ + name: 'checkpointed', + color: 'transparent', + value: this.humanizeBytesPipe.transform(d.checkpointedSize) + }) + ] + }); + + // Row labels are truncated; show the full operator name as a native tooltip while hovering one. + chart.on('axis-label-item:pointerover', (ev: { target?: { attributes?: { text?: unknown } } }) => { + container.title = this.rowTitles.get(String(ev.target?.attributes?.text ?? '')) ?? ''; + }); + chart.on('axis-label-item:pointerout', () => (container.title = '')); Review Comment: ```suggestion chart.on('axis-label-item:pointerout', () => (container.title = '')); chart.interaction('elementHighlightByX', true); ``` Makes it easy to see which row when there are lots of subtasks. ########## flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.html: ########## @@ -0,0 +1,174 @@ +<!-- + ~ Licensed to the Apache Software Foundation (ASF) under one + ~ or more contributor license agreements. See the NOTICE file + ~ distributed with this work for additional information + ~ regarding copyright ownership. The ASF licenses this file + ~ to you under the Apache License, Version 2.0 (the + ~ "License"); you may not use this file except in compliance + ~ with the License. You may obtain a copy of the License at + ~ + ~ http://www.apache.org/licenses/LICENSE-2.0 + ~ + ~ Unless required by applicable law or agreed to in writing, software + ~ distributed under the License is distributed on an "AS IS" BASIS, + ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + ~ See the License for the specific language governing permissions and + ~ limitations under the License. + --> + +<div class="timeline-wrapper"> + <section class="strip-section"> + <header class="strip-header"> + <div class="strip-title"> + <strong>Recent Checkpoints</strong> + <span class="hint">Click a bar to load its per-subtask timeline below.</span> + </div> + <ul class="legend"> + <li> + <span class="swatch completed"></span> + Completed + </li> + <li> + <span class="swatch savepoint"></span> + Savepoint + </li> + <li> + <span class="swatch in-progress"></span> + In progress + </li> + <li> + <span class="swatch failed"></span> + Failed + </li> + </ul> + </header> + <div class="strip-canvas" #stripContainer></div> + <nz-empty *ngIf="!hasHistory" nzNotFoundContent="No checkpoint history yet"></nz-empty> + </section> + + <section class="timeline-section"> + <header class="timeline-header"> + <div class="timeline-title"> + <strong *ngIf="selected"> + Checkpoint #{{ selected.id }} + <span + *ngIf="userPinned" Review Comment: Can we use a tooltip like with the legend, for consistency? ########## flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.less: ########## @@ -0,0 +1,218 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +.timeline-wrapper { + display: flex; + flex-direction: column; + gap: 24px; +} + +header { + display: flex; + flex-wrap: wrap; + gap: 12px; + align-items: center; + justify-content: space-between; + margin-bottom: 8px; +} + +.hint { + margin-left: 8px; + color: #8c8c8c; Review Comment: We should try to use theme imports - like in `flink-runtime-web/web-dashboard/src/app/components/blocked-badge/blocked-badge.component.less` and reference color tokens ########## flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.spec.ts: ########## @@ -0,0 +1,672 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { HttpErrorResponse } from '@angular/common/http'; Review Comment: We never actually render the component - can we have tests with TestBed (e.g `pages/job-manager/metrics/job-manager-metrics.component.spec.ts`) so that we can assert on render behaviour (e.g ngIf and bindings) ########## flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.ts: ########## @@ -0,0 +1,1047 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { DatePipe, NgFor, NgIf } from '@angular/common'; +import { + AfterViewInit, + ChangeDetectionStrategy, + ChangeDetectorRef, + Component, + ElementRef, + Input, + OnChanges, + OnDestroy, + SimpleChanges, + ViewChild +} from '@angular/core'; +import { BehaviorSubject, defer, EMPTY, Observable, of, Subject, timer } from 'rxjs'; +import { catchError, distinctUntilChanged, filter, map, repeat, switchMap, takeUntil } from 'rxjs/operators'; + +import { Chart } from '@antv/g2'; +import { HumanizeBytesPipe } from '@flink-runtime-web/components/humanize-bytes.pipe'; +import { HumanizeDurationPipe } from '@flink-runtime-web/components/humanize-duration.pipe'; +import { + CheckpointConfig, + CheckpointHistory, + CheckpointSubTask, + CompletedSubTaskCheckpointStatistics, + SubTaskCheckpointStatisticsItem, + VerticesItem +} from '@flink-runtime-web/interfaces'; +import { JobService } from '@flink-runtime-web/services'; +import { formatTimeAxis, startAlignedTicks } from '@flink-runtime-web/utils'; +import { NzAlertModule } from 'ng-zorro-antd/alert'; +import { NzEmptyModule } from 'ng-zorro-antd/empty'; +import { NzSpinModule } from 'ng-zorro-antd/spin'; +import { NzTooltipModule } from 'ng-zorro-antd/tooltip'; + +type PhaseKey = 'start_delay' | 'alignment' | 'sync' | 'async'; + +// 'not-retained': every vertex returned 404, i.e. the JobManager no longer keeps this checkpoint's stats. +type LoadError = 'not-retained' | 'request-failed'; + +interface TimelineLoad { + datum: StripDatum; + perVertex: Record<string, CheckpointSubTask | null>; + failedCount: number; + requestedCount: number; + error?: LoadError; +} + +interface StripDatum { + id: number; + idLabel: string; + duration: number; + status: 'COMPLETED' | 'SAVEPOINT' | 'IN_PROGRESS' | 'FAILED'; + rawStatus: string; + isSavepoint: boolean; + triggered: number; + acked: string; + size: number; +} + +interface TimelineDatum { + row: string; + operatorFull: string; + subtask: number; + phase: PhaseKey; + phaseLabel: string; + range: [number, number]; + durationMs: number; + aborted: boolean; + stateSize: number; + checkpointedSize: number; + totalDuration: number; +} + +// A subtask that reported no stats for the checkpoint, drawn as one hatched bar spanning it. +interface UnackedDatum { + row: string; + operatorFull: string; + subtask: number; + label: 'no ack' | 'pending'; + range: [number, number]; +} + +// A subtask whose phases all rounded to 0ms, drawn as a hairline at trigger time so its row isn't empty. +interface InstantDatum { + row: string; + operatorFull: string; + subtask: number; + aborted: boolean; + stateSize: number; + checkpointedSize: number; + range: [number, number]; +} + +interface TimelineRows { + rows: TimelineDatum[]; + unacked: UnackedDatum[]; + instants: InstantDatum[]; + rowOrder: string[]; +} + +const STRIP_COLOR: Record<StripDatum['status'], string> = { + COMPLETED: '#52c41a', + SAVEPOINT: '#722ed1', + IN_PROGRESS: '#faad14', + FAILED: '#f5222d' +}; + +const STRIP_STATUSES = Object.keys(STRIP_COLOR) as Array<StripDatum['status']>; + +const PHASE_COLOR: Record<PhaseKey, string> = { + start_delay: '#b8c0cc', + alignment: '#91caff', + sync: '#fa8c16', + async: '#52c41a' +}; + +const PHASES = Object.keys(PHASE_COLOR) as PhaseKey[]; + +const NOT_COMPLETED_STROKE = '#f5222d'; +const NO_ACK_FILL = '#fff1f0'; +const NO_ACK_STRIPE = '#ffa39e'; +const INSTANT_FILL = '#8c8c8c'; + +const OPERATOR_NAME_MAX = 60; + +type IntervalMark = ReturnType<Chart['interval']>; + +// Built once and updated in place on reloads, so following an in-progress checkpoint doesn't rebuild the canvas. +interface TimelineChart { + chart: Chart; + phases: IntervalMark; + unacked: IntervalMark; + instants: IntervalMark; +} + +@Component({ + selector: 'flink-job-checkpoints-timeline', + templateUrl: './job-checkpoints-timeline.component.html', + styleUrls: ['./job-checkpoints-timeline.component.less'], + changeDetection: ChangeDetectionStrategy.OnPush, + imports: [ + NgFor, + NgIf, + DatePipe, + HumanizeDurationPipe, + HumanizeBytesPipe, + NzAlertModule, + NzEmptyModule, + NzSpinModule, + NzTooltipModule + ] +}) +export class JobCheckpointsTimelineComponent implements AfterViewInit, OnChanges, OnDestroy { + @Input() public jobId: string; + @Input() public vertices: VerticesItem[] = []; + @Input() public history: CheckpointHistory[] = []; + @Input() public config?: CheckpointConfig; + + @ViewChild('stripContainer', { static: true }) private readonly stripContainer: ElementRef<HTMLDivElement>; + @ViewChild('timelineContainer', { static: true }) private readonly timelineContainer: ElementRef<HTMLDivElement>; + @ViewChild('timelineScroll', { static: true }) private readonly timelineScroll: ElementRef<HTMLDivElement>; + @ViewChild('timelineWrapper', { static: true }) private readonly timelineWrapper: ElementRef<HTMLDivElement>; + + public selected?: StripDatum; + public loadingTimeline = false; + public timelineEmpty = false; + public loadError?: LoadError; + public failedVertexCount = 0; + public requestedVertexCount = 0; + // Rows of the shown checkpoint, how many of the slowest are shown, and the first one scrolled to. + public totalRows = 0; + public slowestCount = JobCheckpointsTimelineComponent.SHOWN_ROWS_DEFAULT; + public firstRow = 0; + public chartHeight = 0; + public slowestFirst = false; + + // Diverges from @Input once polling starts feeding live updates. + private currentHistory: CheckpointHistory[] = []; + public userPinned = false; + private stripChart?: Chart; + private stripInterval?: IntervalMark; + private timeline?: TimelineChart; + // Full operator name per row label, for the native tooltip on the truncated labels. + private rowTitles = new Map<string, string>(); + // Kept so scrolling can redraw another window of rows without rebuilding them. + private timelineRows?: TimelineRows & { datum: StripDatum; slowestOrder: string[] }; + private scrollFrame?: number; + private noAckHatch?: HTMLCanvasElement; + private resizeObserver?: ResizeObserver; + private readonly destroy$ = new Subject<void>(); + // null cancels a load in flight. + private readonly loadTimeline$ = new Subject<StripDatum | null>(); + // A load requested while the Timeline tab was hidden; it runs once the tab is shown. + private loadDeferred = false; + // The checkpoint whose failed load was already retried while following, so polls don't retry it again. + private retriedCheckpointId?: number; + // Terminal (completed or failed) checkpoint stats no longer change; memoize by id for instant re-selects. + // Holds only the checkpoints that can be selected again: those in the strip, plus the pinned one. + private readonly detailsCache = new Map<number, Record<string, CheckpointSubTask | null>>(); + private readonly visible$ = new BehaviorSubject<boolean>(false); + private static readonly POLL_MIN_MS = 1500; + private static readonly POLL_MAX_MS = 15000; + private static readonly STRIP_MAX_BARS = 60; + private static readonly TIMELINE_ROW_HEIGHT = 22; + // paddingTop (time axis) + paddingBottom, plus a little slack. + private static readonly TIMELINE_HEIGHT_BASE = 56; + // Rows shown by default (the slowest) and the most that can be picked; the chart is at most VISIBLE_ROWS tall. + private static readonly SHOWN_ROWS_DEFAULT = 25; + private static readonly SHOWN_ROWS_MAX = 200; + private static readonly VISIBLE_ROWS = 25; + private static readonly TIMELINE_HEIGHT_MIN = 160; + private static readonly LABEL_CHAR_WIDTH = 7; + private static readonly LABEL_PADDING_BUFFER = 24; + private static readonly LABEL_PADDING_MIN = 260; + private static readonly LABEL_PADDING_MAX = 520; + + private readonly humanizeBytesPipe = new HumanizeBytesPipe(); + private readonly humanizeDurationPipe = new HumanizeDurationPipe(); + + constructor( + private readonly jobService: JobService, + private readonly cdr: ChangeDetectorRef + ) {} + + public ngAfterViewInit(): void { + this.currentHistory = this.history || []; + this.renderStrip(); + this.subscribeTimelineLoads(); + if (this.currentHistory.length > 0) { + this.selectInternal(this.toStripDatum(this.currentHistory[0]), false); + } + this.startPolling(); + // G2 locks autoFit at construction; re-fit + resume polling once width is real. + this.visible$.next(this.stripContainer.nativeElement.clientWidth > 0); + if (typeof ResizeObserver !== 'undefined') { + this.resizeObserver = new ResizeObserver(() => { + const stripVisible = this.stripContainer.nativeElement.clientWidth > 0; + if (this.stripChart && stripVisible) { + void this.stripChart.forceFit(); + } + if (this.timeline && this.timelineContainer.nativeElement.clientWidth > 0) { + void this.timeline.chart.forceFit(); + } + this.visible$.next(stripVisible); + }); + this.resizeObserver.observe(this.stripContainer.nativeElement); + this.resizeObserver.observe(this.timelineContainer.nativeElement); + } + } + + public ngOnChanges(changes: SimpleChanges): void { + if (!this.stripContainer || !this.stripContainer.nativeElement) { + return; + } + if (changes['jobId'] && !changes['jobId'].firstChange) { + this.userPinned = false; + this.selected = undefined; + this.currentHistory = []; + // Otherwise a load still in flight for the old job lands in the cache under an id the new job reuses. + this.loadTimeline$.next(null); + this.detailsCache.clear(); + this.retriedCheckpointId = undefined; + this.clearTimeline(); + this.renderStrip(); + } + if (changes['history']) { + this.applyHistory(this.history || []); + } + if (changes['vertices'] && !changes['vertices'].firstChange && this.selected) { + this.loadTimeline(this.selected); + } + } + + public ngOnDestroy(): void { + this.destroy$.next(); + this.destroy$.complete(); + this.resizeObserver?.disconnect(); + if (this.scrollFrame !== undefined) { + cancelAnimationFrame(this.scrollFrame); + } + this.stripChart?.destroy(); + this.timeline?.chart.destroy(); + } + + private toStripDatum(h: CheckpointHistory): StripDatum { + // FAILED first, so a failed savepoint shows as failed rather than as a savepoint. + const status: StripDatum['status'] = + h.status === 'FAILED' + ? 'FAILED' + : h.is_savepoint + ? 'SAVEPOINT' + : h.status === 'IN_PROGRESS' + ? 'IN_PROGRESS' + : 'COMPLETED'; + return { + id: h.id, + idLabel: String(h.id), + duration: Math.max(h.end_to_end_duration || 0, 0), + status, + rawStatus: h.status, + isSavepoint: h.is_savepoint, + triggered: h.trigger_timestamp, + acked: `${h.num_acknowledged_subtasks}/${h.num_subtasks}`, + size: h.state_size + }; + } + + private renderStrip(): void { + const data = (this.currentHistory || []) + .slice(0, JobCheckpointsTimelineComponent.STRIP_MAX_BARS) + .map(h => this.toStripDatum(h)) + .reverse(); + + if (!this.stripChart || !this.stripInterval) { + this.stripChart = new Chart({ + container: this.stripContainer.nativeElement, + autoFit: true, + height: 110, + paddingTop: 16, + paddingRight: 16, + paddingBottom: 36, + paddingLeft: 56 + }); + const isPinned = (d: StripDatum): boolean => this.userPinned && this.selected?.id === d.id; + this.stripInterval = this.stripChart + .interval() + .encode('x', 'idLabel') + .encode('y', 'duration') + .encode('color', 'status') + .scale('y', { nice: true, domainMin: 0 }) + .scale('color', { domain: STRIP_STATUSES, range: STRIP_STATUSES.map(s => STRIP_COLOR[s]) }) + .axis('x', { title: false, labelAutoHide: true, labelAutoRotate: false }) + .axis('y', { title: false }) + .legend(false) + .animate(false) + .style('maxWidth', 18) + // Keeps 0ms bars (e.g. in progress before the first ack) and very short ones visible and clickable. + .style('minHeight', 3) + .style('stroke', (d: StripDatum) => (isPinned(d) ? '#1890ff' : 'transparent')) + .style('lineWidth', (d: StripDatum) => (isPinned(d) ? 2.5 : 0)) Review Comment: ``` const chart = new Chart(..); this.stripChart = chart; ... .interaction('tooltip', { disableNative: true }) .interaction('elementHighlight', true) ... chart.on('interval:pointerover', (ev: { data?: { data?: StripDatum } }) => { const datum = ev.data?.data; if (datum) { chart.emit('tooltip:show', { data: { data: datum } }); } }); chart.on('interval:pointerout', () => chart.emit('tooltip:hide')); ``` This will allow the tooltips to only show when you hover over the actual bar in the recent checkpoints. Currently it's popping up whenever you hover, which for small bars with large whitespace is making it hard to see which is hovered over. ########## flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.ts: ########## @@ -0,0 +1,1047 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { DatePipe, NgFor, NgIf } from '@angular/common'; +import { + AfterViewInit, + ChangeDetectionStrategy, + ChangeDetectorRef, + Component, + ElementRef, + Input, + OnChanges, + OnDestroy, + SimpleChanges, + ViewChild +} from '@angular/core'; +import { BehaviorSubject, defer, EMPTY, Observable, of, Subject, timer } from 'rxjs'; +import { catchError, distinctUntilChanged, filter, map, repeat, switchMap, takeUntil } from 'rxjs/operators'; + +import { Chart } from '@antv/g2'; +import { HumanizeBytesPipe } from '@flink-runtime-web/components/humanize-bytes.pipe'; +import { HumanizeDurationPipe } from '@flink-runtime-web/components/humanize-duration.pipe'; +import { + CheckpointConfig, + CheckpointHistory, + CheckpointSubTask, + CompletedSubTaskCheckpointStatistics, + SubTaskCheckpointStatisticsItem, + VerticesItem +} from '@flink-runtime-web/interfaces'; +import { JobService } from '@flink-runtime-web/services'; +import { formatTimeAxis, startAlignedTicks } from '@flink-runtime-web/utils'; +import { NzAlertModule } from 'ng-zorro-antd/alert'; +import { NzEmptyModule } from 'ng-zorro-antd/empty'; +import { NzSpinModule } from 'ng-zorro-antd/spin'; +import { NzTooltipModule } from 'ng-zorro-antd/tooltip'; + +type PhaseKey = 'start_delay' | 'alignment' | 'sync' | 'async'; + +// 'not-retained': every vertex returned 404, i.e. the JobManager no longer keeps this checkpoint's stats. +type LoadError = 'not-retained' | 'request-failed'; + +interface TimelineLoad { + datum: StripDatum; + perVertex: Record<string, CheckpointSubTask | null>; + failedCount: number; + requestedCount: number; + error?: LoadError; +} + +interface StripDatum { + id: number; + idLabel: string; + duration: number; + status: 'COMPLETED' | 'SAVEPOINT' | 'IN_PROGRESS' | 'FAILED'; + rawStatus: string; + isSavepoint: boolean; + triggered: number; + acked: string; + size: number; +} + +interface TimelineDatum { + row: string; + operatorFull: string; + subtask: number; + phase: PhaseKey; + phaseLabel: string; + range: [number, number]; + durationMs: number; + aborted: boolean; + stateSize: number; + checkpointedSize: number; + totalDuration: number; +} + +// A subtask that reported no stats for the checkpoint, drawn as one hatched bar spanning it. +interface UnackedDatum { + row: string; + operatorFull: string; + subtask: number; + label: 'no ack' | 'pending'; + range: [number, number]; +} + +// A subtask whose phases all rounded to 0ms, drawn as a hairline at trigger time so its row isn't empty. +interface InstantDatum { + row: string; + operatorFull: string; + subtask: number; + aborted: boolean; + stateSize: number; + checkpointedSize: number; + range: [number, number]; +} + +interface TimelineRows { + rows: TimelineDatum[]; + unacked: UnackedDatum[]; + instants: InstantDatum[]; + rowOrder: string[]; +} + +const STRIP_COLOR: Record<StripDatum['status'], string> = { + COMPLETED: '#52c41a', + SAVEPOINT: '#722ed1', + IN_PROGRESS: '#faad14', + FAILED: '#f5222d' +}; + +const STRIP_STATUSES = Object.keys(STRIP_COLOR) as Array<StripDatum['status']>; + +const PHASE_COLOR: Record<PhaseKey, string> = { + start_delay: '#b8c0cc', + alignment: '#91caff', + sync: '#fa8c16', + async: '#52c41a' +}; + +const PHASES = Object.keys(PHASE_COLOR) as PhaseKey[]; + +const NOT_COMPLETED_STROKE = '#f5222d'; +const NO_ACK_FILL = '#fff1f0'; +const NO_ACK_STRIPE = '#ffa39e'; +const INSTANT_FILL = '#8c8c8c'; + +const OPERATOR_NAME_MAX = 60; + +type IntervalMark = ReturnType<Chart['interval']>; + +// Built once and updated in place on reloads, so following an in-progress checkpoint doesn't rebuild the canvas. +interface TimelineChart { + chart: Chart; + phases: IntervalMark; + unacked: IntervalMark; + instants: IntervalMark; +} + +@Component({ + selector: 'flink-job-checkpoints-timeline', + templateUrl: './job-checkpoints-timeline.component.html', + styleUrls: ['./job-checkpoints-timeline.component.less'], + changeDetection: ChangeDetectionStrategy.OnPush, + imports: [ + NgFor, + NgIf, + DatePipe, + HumanizeDurationPipe, + HumanizeBytesPipe, + NzAlertModule, + NzEmptyModule, + NzSpinModule, + NzTooltipModule + ] +}) +export class JobCheckpointsTimelineComponent implements AfterViewInit, OnChanges, OnDestroy { + @Input() public jobId: string; + @Input() public vertices: VerticesItem[] = []; + @Input() public history: CheckpointHistory[] = []; + @Input() public config?: CheckpointConfig; + + @ViewChild('stripContainer', { static: true }) private readonly stripContainer: ElementRef<HTMLDivElement>; + @ViewChild('timelineContainer', { static: true }) private readonly timelineContainer: ElementRef<HTMLDivElement>; + @ViewChild('timelineScroll', { static: true }) private readonly timelineScroll: ElementRef<HTMLDivElement>; + @ViewChild('timelineWrapper', { static: true }) private readonly timelineWrapper: ElementRef<HTMLDivElement>; + + public selected?: StripDatum; + public loadingTimeline = false; + public timelineEmpty = false; + public loadError?: LoadError; + public failedVertexCount = 0; + public requestedVertexCount = 0; + // Rows of the shown checkpoint, how many of the slowest are shown, and the first one scrolled to. + public totalRows = 0; + public slowestCount = JobCheckpointsTimelineComponent.SHOWN_ROWS_DEFAULT; + public firstRow = 0; + public chartHeight = 0; + public slowestFirst = false; + + // Diverges from @Input once polling starts feeding live updates. + private currentHistory: CheckpointHistory[] = []; + public userPinned = false; + private stripChart?: Chart; + private stripInterval?: IntervalMark; + private timeline?: TimelineChart; + // Full operator name per row label, for the native tooltip on the truncated labels. + private rowTitles = new Map<string, string>(); + // Kept so scrolling can redraw another window of rows without rebuilding them. + private timelineRows?: TimelineRows & { datum: StripDatum; slowestOrder: string[] }; + private scrollFrame?: number; + private noAckHatch?: HTMLCanvasElement; + private resizeObserver?: ResizeObserver; + private readonly destroy$ = new Subject<void>(); + // null cancels a load in flight. + private readonly loadTimeline$ = new Subject<StripDatum | null>(); + // A load requested while the Timeline tab was hidden; it runs once the tab is shown. + private loadDeferred = false; + // The checkpoint whose failed load was already retried while following, so polls don't retry it again. + private retriedCheckpointId?: number; + // Terminal (completed or failed) checkpoint stats no longer change; memoize by id for instant re-selects. + // Holds only the checkpoints that can be selected again: those in the strip, plus the pinned one. + private readonly detailsCache = new Map<number, Record<string, CheckpointSubTask | null>>(); + private readonly visible$ = new BehaviorSubject<boolean>(false); + private static readonly POLL_MIN_MS = 1500; + private static readonly POLL_MAX_MS = 15000; + private static readonly STRIP_MAX_BARS = 60; + private static readonly TIMELINE_ROW_HEIGHT = 22; + // paddingTop (time axis) + paddingBottom, plus a little slack. + private static readonly TIMELINE_HEIGHT_BASE = 56; + // Rows shown by default (the slowest) and the most that can be picked; the chart is at most VISIBLE_ROWS tall. + private static readonly SHOWN_ROWS_DEFAULT = 25; + private static readonly SHOWN_ROWS_MAX = 200; + private static readonly VISIBLE_ROWS = 25; + private static readonly TIMELINE_HEIGHT_MIN = 160; + private static readonly LABEL_CHAR_WIDTH = 7; + private static readonly LABEL_PADDING_BUFFER = 24; + private static readonly LABEL_PADDING_MIN = 260; + private static readonly LABEL_PADDING_MAX = 520; + + private readonly humanizeBytesPipe = new HumanizeBytesPipe(); + private readonly humanizeDurationPipe = new HumanizeDurationPipe(); + + constructor( + private readonly jobService: JobService, + private readonly cdr: ChangeDetectorRef + ) {} + + public ngAfterViewInit(): void { + this.currentHistory = this.history || []; + this.renderStrip(); + this.subscribeTimelineLoads(); + if (this.currentHistory.length > 0) { + this.selectInternal(this.toStripDatum(this.currentHistory[0]), false); + } + this.startPolling(); + // G2 locks autoFit at construction; re-fit + resume polling once width is real. + this.visible$.next(this.stripContainer.nativeElement.clientWidth > 0); + if (typeof ResizeObserver !== 'undefined') { + this.resizeObserver = new ResizeObserver(() => { + const stripVisible = this.stripContainer.nativeElement.clientWidth > 0; + if (this.stripChart && stripVisible) { + void this.stripChart.forceFit(); + } + if (this.timeline && this.timelineContainer.nativeElement.clientWidth > 0) { + void this.timeline.chart.forceFit(); + } + this.visible$.next(stripVisible); + }); + this.resizeObserver.observe(this.stripContainer.nativeElement); + this.resizeObserver.observe(this.timelineContainer.nativeElement); + } + } + + public ngOnChanges(changes: SimpleChanges): void { + if (!this.stripContainer || !this.stripContainer.nativeElement) { + return; + } + if (changes['jobId'] && !changes['jobId'].firstChange) { + this.userPinned = false; + this.selected = undefined; + this.currentHistory = []; + // Otherwise a load still in flight for the old job lands in the cache under an id the new job reuses. + this.loadTimeline$.next(null); + this.detailsCache.clear(); + this.retriedCheckpointId = undefined; + this.clearTimeline(); + this.renderStrip(); + } + if (changes['history']) { + this.applyHistory(this.history || []); + } + if (changes['vertices'] && !changes['vertices'].firstChange && this.selected) { + this.loadTimeline(this.selected); + } + } + + public ngOnDestroy(): void { + this.destroy$.next(); + this.destroy$.complete(); + this.resizeObserver?.disconnect(); + if (this.scrollFrame !== undefined) { + cancelAnimationFrame(this.scrollFrame); + } + this.stripChart?.destroy(); + this.timeline?.chart.destroy(); + } + + private toStripDatum(h: CheckpointHistory): StripDatum { + // FAILED first, so a failed savepoint shows as failed rather than as a savepoint. + const status: StripDatum['status'] = + h.status === 'FAILED' + ? 'FAILED' + : h.is_savepoint + ? 'SAVEPOINT' + : h.status === 'IN_PROGRESS' + ? 'IN_PROGRESS' + : 'COMPLETED'; + return { + id: h.id, + idLabel: String(h.id), + duration: Math.max(h.end_to_end_duration || 0, 0), + status, + rawStatus: h.status, + isSavepoint: h.is_savepoint, + triggered: h.trigger_timestamp, + acked: `${h.num_acknowledged_subtasks}/${h.num_subtasks}`, + size: h.state_size + }; + } + + private renderStrip(): void { + const data = (this.currentHistory || []) + .slice(0, JobCheckpointsTimelineComponent.STRIP_MAX_BARS) + .map(h => this.toStripDatum(h)) + .reverse(); + + if (!this.stripChart || !this.stripInterval) { + this.stripChart = new Chart({ + container: this.stripContainer.nativeElement, + autoFit: true, + height: 110, + paddingTop: 16, + paddingRight: 16, + paddingBottom: 36, + paddingLeft: 56 + }); + const isPinned = (d: StripDatum): boolean => this.userPinned && this.selected?.id === d.id; + this.stripInterval = this.stripChart + .interval() + .encode('x', 'idLabel') + .encode('y', 'duration') + .encode('color', 'status') + .scale('y', { nice: true, domainMin: 0 }) + .scale('color', { domain: STRIP_STATUSES, range: STRIP_STATUSES.map(s => STRIP_COLOR[s]) }) + .axis('x', { title: false, labelAutoHide: true, labelAutoRotate: false }) + .axis('y', { title: false }) + .legend(false) + .animate(false) + .style('maxWidth', 18) + // Keeps 0ms bars (e.g. in progress before the first ack) and very short ones visible and clickable. + .style('minHeight', 3) + .style('stroke', (d: StripDatum) => (isPinned(d) ? '#1890ff' : 'transparent')) + .style('lineWidth', (d: StripDatum) => (isPinned(d) ? 2.5 : 0)) + .tooltip({ + title: '', + items: [ + (d: StripDatum) => ({ + name: `#${d.idLabel}`, + color: STRIP_COLOR[d.status], + value: `${d.rawStatus} · ${this.humanizeDurationPipe.transform(d.duration)} · acked ${d.acked}` + }) + ] + }); + this.stripChart.on('interval:click', (ev: { data?: { data?: StripDatum } }) => { + const datum = ev.data?.data; + if (datum) { + this.select(datum); + } + }); + } + + this.stripInterval.scale('x', { domain: data.map(d => d.idLabel) }); + this.stripInterval.data(data); + void this.stripChart.render(); + } + + public select(datum: StripDatum): void { + // Clicking the already-pinned bar resumes auto-follow. + if (this.userPinned && this.selected && this.selected.id === datum.id) { + this.followNewest(); + return; + } + this.selectInternal(datum, true); + } + + public followNewest(): void { + this.userPinned = false; + if (this.currentHistory.length > 0) { + this.selectInternal(this.toStripDatum(this.currentHistory[0]), false); + } else { + this.cdr.markForCheck(); + } + } + + private selectInternal(datum: StripDatum, pinned: boolean): void { + this.selected = datum; + if (pinned) { + this.userPinned = true; + } + this.cdr.markForCheck(); + this.renderStrip(); + this.loadTimeline(datum); + } + + private applyHistory(next: CheckpointHistory[]): void { + const prevNewestId = this.currentHistory.length > 0 ? this.currentHistory[0].id : -1; + const nextNewestId = next.length > 0 ? next[0].id : -1; + this.currentHistory = next; + this.pruneDetailsCache(); + this.renderStrip(); + // The empty state and the rotated-out note read currentHistory. + this.cdr.markForCheck(); + + if (next.length === 0) { + // The pinned checkpoint is gone too (e.g. after a JobManager failover), so follow whatever comes next. + this.userPinned = false; + this.selected = undefined; + this.clearTimeline(); + return; + } + + if (!this.userPinned && nextNewestId !== prevNewestId) { + this.selectInternal(this.toStripDatum(next[0]), false); + return; + } + // Still showing the same checkpoint (the followed newest one, or the pinned one): reload once it gained acks or + // finished, or retry a failed load once (retrying on every poll would repeat the error notifications). Not while + // a load is in flight, so polling doesn't keep cancelling a slow fetch. + const selected = this.selected; + const latest = this.userPinned ? next.find(h => h.id === selected?.id) : next[0]; + if (!selected || !latest || latest.id !== selected.id || this.loadingTimeline) { + return; + } + const current = this.toStripDatum(latest); + const progressed = current.rawStatus !== selected.rawStatus || current.acked !== selected.acked; + const retry = + (this.loadError !== undefined || this.failedVertexCount > 0) && this.retriedCheckpointId !== current.id; + if (retry) { + this.retriedCheckpointId = current.id; + } + if (progressed || retry) { + this.selectInternal(current, false); + } + } + + private pruneDetailsCache(): void { + const selectable = new Set( + this.currentHistory.slice(0, JobCheckpointsTimelineComponent.STRIP_MAX_BARS).map(h => h.id) + ); + for (const id of this.detailsCache.keys()) { + if (!selectable.has(id) && id !== this.selected?.id) { + this.detailsCache.delete(id); + } + } + } + + private startPolling(): void { + this.visible$ + .pipe( + distinctUntilChanged(), + switchMap(visible => + visible + ? defer(() => { + const jobId = this.jobId; + return jobId + ? this.jobService.loadCheckpointStats(jobId).pipe( + catchError(() => of(undefined)), Review Comment: This causes silent failures of the timeline page, making it look like checkpoints have stopped. It would be better to show an error or at least an indicator that the data is stale. <img width="1542" height="997" alt="Image" src="https://github.com/user-attachments/assets/32e723f6-bfdb-4e6c-a95a-b76382ab5463"; /> ########## flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.ts: ########## @@ -0,0 +1,1047 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { DatePipe, NgFor, NgIf } from '@angular/common'; +import { + AfterViewInit, + ChangeDetectionStrategy, + ChangeDetectorRef, + Component, + ElementRef, + Input, + OnChanges, + OnDestroy, + SimpleChanges, + ViewChild +} from '@angular/core'; +import { BehaviorSubject, defer, EMPTY, Observable, of, Subject, timer } from 'rxjs'; +import { catchError, distinctUntilChanged, filter, map, repeat, switchMap, takeUntil } from 'rxjs/operators'; + +import { Chart } from '@antv/g2'; +import { HumanizeBytesPipe } from '@flink-runtime-web/components/humanize-bytes.pipe'; +import { HumanizeDurationPipe } from '@flink-runtime-web/components/humanize-duration.pipe'; +import { + CheckpointConfig, + CheckpointHistory, + CheckpointSubTask, + CompletedSubTaskCheckpointStatistics, + SubTaskCheckpointStatisticsItem, + VerticesItem +} from '@flink-runtime-web/interfaces'; +import { JobService } from '@flink-runtime-web/services'; +import { formatTimeAxis, startAlignedTicks } from '@flink-runtime-web/utils'; +import { NzAlertModule } from 'ng-zorro-antd/alert'; +import { NzEmptyModule } from 'ng-zorro-antd/empty'; +import { NzSpinModule } from 'ng-zorro-antd/spin'; +import { NzTooltipModule } from 'ng-zorro-antd/tooltip'; + +type PhaseKey = 'start_delay' | 'alignment' | 'sync' | 'async'; + +// 'not-retained': every vertex returned 404, i.e. the JobManager no longer keeps this checkpoint's stats. +type LoadError = 'not-retained' | 'request-failed'; + +interface TimelineLoad { + datum: StripDatum; + perVertex: Record<string, CheckpointSubTask | null>; + failedCount: number; + requestedCount: number; + error?: LoadError; +} + +interface StripDatum { + id: number; + idLabel: string; + duration: number; + status: 'COMPLETED' | 'SAVEPOINT' | 'IN_PROGRESS' | 'FAILED'; + rawStatus: string; + isSavepoint: boolean; + triggered: number; + acked: string; + size: number; +} + +interface TimelineDatum { + row: string; + operatorFull: string; + subtask: number; + phase: PhaseKey; + phaseLabel: string; + range: [number, number]; + durationMs: number; + aborted: boolean; + stateSize: number; + checkpointedSize: number; + totalDuration: number; +} + +// A subtask that reported no stats for the checkpoint, drawn as one hatched bar spanning it. +interface UnackedDatum { + row: string; + operatorFull: string; + subtask: number; + label: 'no ack' | 'pending'; + range: [number, number]; +} + +// A subtask whose phases all rounded to 0ms, drawn as a hairline at trigger time so its row isn't empty. +interface InstantDatum { + row: string; + operatorFull: string; + subtask: number; + aborted: boolean; + stateSize: number; + checkpointedSize: number; + range: [number, number]; +} + +interface TimelineRows { + rows: TimelineDatum[]; + unacked: UnackedDatum[]; + instants: InstantDatum[]; + rowOrder: string[]; +} + +const STRIP_COLOR: Record<StripDatum['status'], string> = { + COMPLETED: '#52c41a', + SAVEPOINT: '#722ed1', + IN_PROGRESS: '#faad14', + FAILED: '#f5222d' +}; + +const STRIP_STATUSES = Object.keys(STRIP_COLOR) as Array<StripDatum['status']>; + +const PHASE_COLOR: Record<PhaseKey, string> = { + start_delay: '#b8c0cc', + alignment: '#91caff', + sync: '#fa8c16', + async: '#52c41a' +}; + +const PHASES = Object.keys(PHASE_COLOR) as PhaseKey[]; + +const NOT_COMPLETED_STROKE = '#f5222d'; +const NO_ACK_FILL = '#fff1f0'; +const NO_ACK_STRIPE = '#ffa39e'; +const INSTANT_FILL = '#8c8c8c'; + +const OPERATOR_NAME_MAX = 60; + +type IntervalMark = ReturnType<Chart['interval']>; + +// Built once and updated in place on reloads, so following an in-progress checkpoint doesn't rebuild the canvas. +interface TimelineChart { + chart: Chart; + phases: IntervalMark; + unacked: IntervalMark; + instants: IntervalMark; +} + +@Component({ + selector: 'flink-job-checkpoints-timeline', + templateUrl: './job-checkpoints-timeline.component.html', + styleUrls: ['./job-checkpoints-timeline.component.less'], + changeDetection: ChangeDetectionStrategy.OnPush, + imports: [ + NgFor, + NgIf, + DatePipe, + HumanizeDurationPipe, + HumanizeBytesPipe, + NzAlertModule, + NzEmptyModule, + NzSpinModule, + NzTooltipModule + ] +}) +export class JobCheckpointsTimelineComponent implements AfterViewInit, OnChanges, OnDestroy { + @Input() public jobId: string; + @Input() public vertices: VerticesItem[] = []; + @Input() public history: CheckpointHistory[] = []; + @Input() public config?: CheckpointConfig; + + @ViewChild('stripContainer', { static: true }) private readonly stripContainer: ElementRef<HTMLDivElement>; + @ViewChild('timelineContainer', { static: true }) private readonly timelineContainer: ElementRef<HTMLDivElement>; + @ViewChild('timelineScroll', { static: true }) private readonly timelineScroll: ElementRef<HTMLDivElement>; + @ViewChild('timelineWrapper', { static: true }) private readonly timelineWrapper: ElementRef<HTMLDivElement>; + + public selected?: StripDatum; + public loadingTimeline = false; + public timelineEmpty = false; + public loadError?: LoadError; + public failedVertexCount = 0; + public requestedVertexCount = 0; + // Rows of the shown checkpoint, how many of the slowest are shown, and the first one scrolled to. + public totalRows = 0; + public slowestCount = JobCheckpointsTimelineComponent.SHOWN_ROWS_DEFAULT; + public firstRow = 0; + public chartHeight = 0; + public slowestFirst = false; + + // Diverges from @Input once polling starts feeding live updates. + private currentHistory: CheckpointHistory[] = []; + public userPinned = false; + private stripChart?: Chart; + private stripInterval?: IntervalMark; + private timeline?: TimelineChart; + // Full operator name per row label, for the native tooltip on the truncated labels. + private rowTitles = new Map<string, string>(); + // Kept so scrolling can redraw another window of rows without rebuilding them. + private timelineRows?: TimelineRows & { datum: StripDatum; slowestOrder: string[] }; + private scrollFrame?: number; + private noAckHatch?: HTMLCanvasElement; + private resizeObserver?: ResizeObserver; + private readonly destroy$ = new Subject<void>(); + // null cancels a load in flight. + private readonly loadTimeline$ = new Subject<StripDatum | null>(); + // A load requested while the Timeline tab was hidden; it runs once the tab is shown. + private loadDeferred = false; + // The checkpoint whose failed load was already retried while following, so polls don't retry it again. + private retriedCheckpointId?: number; + // Terminal (completed or failed) checkpoint stats no longer change; memoize by id for instant re-selects. + // Holds only the checkpoints that can be selected again: those in the strip, plus the pinned one. + private readonly detailsCache = new Map<number, Record<string, CheckpointSubTask | null>>(); + private readonly visible$ = new BehaviorSubject<boolean>(false); + private static readonly POLL_MIN_MS = 1500; + private static readonly POLL_MAX_MS = 15000; + private static readonly STRIP_MAX_BARS = 60; + private static readonly TIMELINE_ROW_HEIGHT = 22; + // paddingTop (time axis) + paddingBottom, plus a little slack. + private static readonly TIMELINE_HEIGHT_BASE = 56; + // Rows shown by default (the slowest) and the most that can be picked; the chart is at most VISIBLE_ROWS tall. + private static readonly SHOWN_ROWS_DEFAULT = 25; + private static readonly SHOWN_ROWS_MAX = 200; + private static readonly VISIBLE_ROWS = 25; + private static readonly TIMELINE_HEIGHT_MIN = 160; + private static readonly LABEL_CHAR_WIDTH = 7; + private static readonly LABEL_PADDING_BUFFER = 24; + private static readonly LABEL_PADDING_MIN = 260; + private static readonly LABEL_PADDING_MAX = 520; + + private readonly humanizeBytesPipe = new HumanizeBytesPipe(); + private readonly humanizeDurationPipe = new HumanizeDurationPipe(); + + constructor( + private readonly jobService: JobService, + private readonly cdr: ChangeDetectorRef + ) {} + + public ngAfterViewInit(): void { + this.currentHistory = this.history || []; + this.renderStrip(); + this.subscribeTimelineLoads(); + if (this.currentHistory.length > 0) { + this.selectInternal(this.toStripDatum(this.currentHistory[0]), false); + } + this.startPolling(); Review Comment: As in, the parent could own the polling and pass a live history to the timeline, leaving the other tabs on manual refresh. Then the refresh button could say "new data available" or something to indicate that it has new data, so the refresh button is more of a "load latest state". ########## flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.ts: ########## @@ -0,0 +1,1047 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { DatePipe, NgFor, NgIf } from '@angular/common'; +import { + AfterViewInit, + ChangeDetectionStrategy, + ChangeDetectorRef, + Component, + ElementRef, + Input, + OnChanges, + OnDestroy, + SimpleChanges, + ViewChild +} from '@angular/core'; +import { BehaviorSubject, defer, EMPTY, Observable, of, Subject, timer } from 'rxjs'; +import { catchError, distinctUntilChanged, filter, map, repeat, switchMap, takeUntil } from 'rxjs/operators'; + +import { Chart } from '@antv/g2'; +import { HumanizeBytesPipe } from '@flink-runtime-web/components/humanize-bytes.pipe'; +import { HumanizeDurationPipe } from '@flink-runtime-web/components/humanize-duration.pipe'; +import { + CheckpointConfig, + CheckpointHistory, + CheckpointSubTask, + CompletedSubTaskCheckpointStatistics, + SubTaskCheckpointStatisticsItem, + VerticesItem +} from '@flink-runtime-web/interfaces'; +import { JobService } from '@flink-runtime-web/services'; +import { formatTimeAxis, startAlignedTicks } from '@flink-runtime-web/utils'; +import { NzAlertModule } from 'ng-zorro-antd/alert'; +import { NzEmptyModule } from 'ng-zorro-antd/empty'; +import { NzSpinModule } from 'ng-zorro-antd/spin'; +import { NzTooltipModule } from 'ng-zorro-antd/tooltip'; + +type PhaseKey = 'start_delay' | 'alignment' | 'sync' | 'async'; + +// 'not-retained': every vertex returned 404, i.e. the JobManager no longer keeps this checkpoint's stats. +type LoadError = 'not-retained' | 'request-failed'; + +interface TimelineLoad { + datum: StripDatum; + perVertex: Record<string, CheckpointSubTask | null>; + failedCount: number; + requestedCount: number; + error?: LoadError; +} + +interface StripDatum { + id: number; + idLabel: string; + duration: number; + status: 'COMPLETED' | 'SAVEPOINT' | 'IN_PROGRESS' | 'FAILED'; + rawStatus: string; + isSavepoint: boolean; + triggered: number; + acked: string; + size: number; +} + +interface TimelineDatum { + row: string; + operatorFull: string; + subtask: number; + phase: PhaseKey; + phaseLabel: string; + range: [number, number]; + durationMs: number; + aborted: boolean; + stateSize: number; + checkpointedSize: number; + totalDuration: number; +} + +// A subtask that reported no stats for the checkpoint, drawn as one hatched bar spanning it. +interface UnackedDatum { + row: string; + operatorFull: string; + subtask: number; + label: 'no ack' | 'pending'; + range: [number, number]; +} + +// A subtask whose phases all rounded to 0ms, drawn as a hairline at trigger time so its row isn't empty. +interface InstantDatum { + row: string; + operatorFull: string; + subtask: number; + aborted: boolean; + stateSize: number; + checkpointedSize: number; + range: [number, number]; +} + +interface TimelineRows { + rows: TimelineDatum[]; + unacked: UnackedDatum[]; + instants: InstantDatum[]; + rowOrder: string[]; +} + +const STRIP_COLOR: Record<StripDatum['status'], string> = { + COMPLETED: '#52c41a', + SAVEPOINT: '#722ed1', + IN_PROGRESS: '#faad14', + FAILED: '#f5222d' +}; + +const STRIP_STATUSES = Object.keys(STRIP_COLOR) as Array<StripDatum['status']>; + +const PHASE_COLOR: Record<PhaseKey, string> = { + start_delay: '#b8c0cc', + alignment: '#91caff', + sync: '#fa8c16', + async: '#52c41a' +}; + +const PHASES = Object.keys(PHASE_COLOR) as PhaseKey[]; + +const NOT_COMPLETED_STROKE = '#f5222d'; +const NO_ACK_FILL = '#fff1f0'; +const NO_ACK_STRIPE = '#ffa39e'; +const INSTANT_FILL = '#8c8c8c'; + +const OPERATOR_NAME_MAX = 60; + +type IntervalMark = ReturnType<Chart['interval']>; + +// Built once and updated in place on reloads, so following an in-progress checkpoint doesn't rebuild the canvas. +interface TimelineChart { + chart: Chart; + phases: IntervalMark; + unacked: IntervalMark; + instants: IntervalMark; +} + +@Component({ + selector: 'flink-job-checkpoints-timeline', + templateUrl: './job-checkpoints-timeline.component.html', + styleUrls: ['./job-checkpoints-timeline.component.less'], + changeDetection: ChangeDetectionStrategy.OnPush, + imports: [ + NgFor, + NgIf, + DatePipe, + HumanizeDurationPipe, + HumanizeBytesPipe, + NzAlertModule, + NzEmptyModule, + NzSpinModule, + NzTooltipModule + ] +}) +export class JobCheckpointsTimelineComponent implements AfterViewInit, OnChanges, OnDestroy { + @Input() public jobId: string; + @Input() public vertices: VerticesItem[] = []; + @Input() public history: CheckpointHistory[] = []; + @Input() public config?: CheckpointConfig; + + @ViewChild('stripContainer', { static: true }) private readonly stripContainer: ElementRef<HTMLDivElement>; + @ViewChild('timelineContainer', { static: true }) private readonly timelineContainer: ElementRef<HTMLDivElement>; + @ViewChild('timelineScroll', { static: true }) private readonly timelineScroll: ElementRef<HTMLDivElement>; + @ViewChild('timelineWrapper', { static: true }) private readonly timelineWrapper: ElementRef<HTMLDivElement>; + + public selected?: StripDatum; + public loadingTimeline = false; + public timelineEmpty = false; + public loadError?: LoadError; + public failedVertexCount = 0; + public requestedVertexCount = 0; + // Rows of the shown checkpoint, how many of the slowest are shown, and the first one scrolled to. + public totalRows = 0; + public slowestCount = JobCheckpointsTimelineComponent.SHOWN_ROWS_DEFAULT; + public firstRow = 0; + public chartHeight = 0; + public slowestFirst = false; + + // Diverges from @Input once polling starts feeding live updates. + private currentHistory: CheckpointHistory[] = []; + public userPinned = false; + private stripChart?: Chart; + private stripInterval?: IntervalMark; + private timeline?: TimelineChart; + // Full operator name per row label, for the native tooltip on the truncated labels. + private rowTitles = new Map<string, string>(); + // Kept so scrolling can redraw another window of rows without rebuilding them. + private timelineRows?: TimelineRows & { datum: StripDatum; slowestOrder: string[] }; + private scrollFrame?: number; + private noAckHatch?: HTMLCanvasElement; + private resizeObserver?: ResizeObserver; + private readonly destroy$ = new Subject<void>(); + // null cancels a load in flight. + private readonly loadTimeline$ = new Subject<StripDatum | null>(); + // A load requested while the Timeline tab was hidden; it runs once the tab is shown. + private loadDeferred = false; + // The checkpoint whose failed load was already retried while following, so polls don't retry it again. + private retriedCheckpointId?: number; + // Terminal (completed or failed) checkpoint stats no longer change; memoize by id for instant re-selects. + // Holds only the checkpoints that can be selected again: those in the strip, plus the pinned one. + private readonly detailsCache = new Map<number, Record<string, CheckpointSubTask | null>>(); + private readonly visible$ = new BehaviorSubject<boolean>(false); + private static readonly POLL_MIN_MS = 1500; + private static readonly POLL_MAX_MS = 15000; + private static readonly STRIP_MAX_BARS = 60; + private static readonly TIMELINE_ROW_HEIGHT = 22; + // paddingTop (time axis) + paddingBottom, plus a little slack. + private static readonly TIMELINE_HEIGHT_BASE = 56; + // Rows shown by default (the slowest) and the most that can be picked; the chart is at most VISIBLE_ROWS tall. + private static readonly SHOWN_ROWS_DEFAULT = 25; + private static readonly SHOWN_ROWS_MAX = 200; + private static readonly VISIBLE_ROWS = 25; + private static readonly TIMELINE_HEIGHT_MIN = 160; + private static readonly LABEL_CHAR_WIDTH = 7; + private static readonly LABEL_PADDING_BUFFER = 24; + private static readonly LABEL_PADDING_MIN = 260; + private static readonly LABEL_PADDING_MAX = 520; + + private readonly humanizeBytesPipe = new HumanizeBytesPipe(); + private readonly humanizeDurationPipe = new HumanizeDurationPipe(); + + constructor( + private readonly jobService: JobService, + private readonly cdr: ChangeDetectorRef + ) {} + + public ngAfterViewInit(): void { + this.currentHistory = this.history || []; + this.renderStrip(); + this.subscribeTimelineLoads(); + if (this.currentHistory.length > 0) { + this.selectInternal(this.toStripDatum(this.currentHistory[0]), false); + } + this.startPolling(); + // G2 locks autoFit at construction; re-fit + resume polling once width is real. + this.visible$.next(this.stripContainer.nativeElement.clientWidth > 0); + if (typeof ResizeObserver !== 'undefined') { + this.resizeObserver = new ResizeObserver(() => { + const stripVisible = this.stripContainer.nativeElement.clientWidth > 0; + if (this.stripChart && stripVisible) { + void this.stripChart.forceFit(); + } + if (this.timeline && this.timelineContainer.nativeElement.clientWidth > 0) { + void this.timeline.chart.forceFit(); + } + this.visible$.next(stripVisible); + }); + this.resizeObserver.observe(this.stripContainer.nativeElement); + this.resizeObserver.observe(this.timelineContainer.nativeElement); + } + } + + public ngOnChanges(changes: SimpleChanges): void { + if (!this.stripContainer || !this.stripContainer.nativeElement) { + return; + } + if (changes['jobId'] && !changes['jobId'].firstChange) { + this.userPinned = false; + this.selected = undefined; + this.currentHistory = []; + // Otherwise a load still in flight for the old job lands in the cache under an id the new job reuses. + this.loadTimeline$.next(null); + this.detailsCache.clear(); + this.retriedCheckpointId = undefined; + this.clearTimeline(); + this.renderStrip(); + } + if (changes['history']) { + this.applyHistory(this.history || []); + } + if (changes['vertices'] && !changes['vertices'].firstChange && this.selected) { + this.loadTimeline(this.selected); + } + } + + public ngOnDestroy(): void { + this.destroy$.next(); + this.destroy$.complete(); + this.resizeObserver?.disconnect(); + if (this.scrollFrame !== undefined) { + cancelAnimationFrame(this.scrollFrame); + } + this.stripChart?.destroy(); + this.timeline?.chart.destroy(); + } + + private toStripDatum(h: CheckpointHistory): StripDatum { + // FAILED first, so a failed savepoint shows as failed rather than as a savepoint. + const status: StripDatum['status'] = + h.status === 'FAILED' + ? 'FAILED' + : h.is_savepoint + ? 'SAVEPOINT' + : h.status === 'IN_PROGRESS' + ? 'IN_PROGRESS' + : 'COMPLETED'; + return { + id: h.id, + idLabel: String(h.id), + duration: Math.max(h.end_to_end_duration || 0, 0), + status, + rawStatus: h.status, + isSavepoint: h.is_savepoint, + triggered: h.trigger_timestamp, + acked: `${h.num_acknowledged_subtasks}/${h.num_subtasks}`, + size: h.state_size + }; + } + + private renderStrip(): void { + const data = (this.currentHistory || []) + .slice(0, JobCheckpointsTimelineComponent.STRIP_MAX_BARS) + .map(h => this.toStripDatum(h)) + .reverse(); + + if (!this.stripChart || !this.stripInterval) { + this.stripChart = new Chart({ + container: this.stripContainer.nativeElement, + autoFit: true, + height: 110, Review Comment: This is quite small, particularly when one checkpoint is longer than most, the small checkpoints are hard to see in the chart. Maybe 160? Also [job-checkpoints-timeline.component.less](https://github.com/apache/flink/pull/28060/changes#diff-8cbb2bf30835ffadb766c5014cadfeaac0bbe2e6f3dacca8bb71219bc60df449) sets height of 80, it should match the 160. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
