Copilot commented on code in PR #28060: URL: https://github.com/apache/flink/pull/28060#discussion_r4434571127
########## flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.html: ########## @@ -0,0 +1,146 @@ +<!-- + ~ 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> Review Comment: Checkpoint selection is only available through the chart's `interval:click` handler. This container has no focusable checkpoint controls or keyboard handling, so keyboard and screen-reader users cannot choose or pin a checkpoint; the Follow newest button only appears after a pointer selection. Please provide a labeled native checkpoint selector or focusable list wired to the same selection action, with an accessible equivalent for the chart values. ########## flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.html: ########## @@ -0,0 +1,146 @@ +<!-- + ~ 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="!history || history.length === 0" Review Comment: Polling updates `currentHistory`, not this `history` input, and the parent only refreshes its snapshot on initial load or a manual refresh. If the page is opened before the first checkpoint, polls can populate the strip and timeline while this still displays “No checkpoint history yet.” Please bind the empty state to the same live history used by the strip, exposed through a public field or getter. ########## flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.ts: ########## @@ -0,0 +1,855 @@ +/* + * 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, NgIf } from '@angular/common'; +import { + AfterViewInit, + ChangeDetectionStrategy, + ChangeDetectorRef, + Component, + ElementRef, + Input, + OnChanges, + OnDestroy, + SimpleChanges, + ViewChild +} from '@angular/core'; +import { BehaviorSubject, defer, EMPTY, of, Subject, timer } from 'rxjs'; +import { catchError, distinctUntilChanged, 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 { 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'; + +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; + unaligned: boolean; + 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]; +} + +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 PHASE_LABEL: Record<PhaseKey, string> = { + start_delay: 'start_delay', + alignment: 'alignment', + sync: 'sync', + async: 'async' +}; + +const OPERATOR_NAME_MAX = 60; + +// Local HH:MM:SS.mmm — matches the header's DatePipe default (browser timezone). +function formatWallClock(epochMs: number): string { + const d = new Date(epochMs); + const pad = (n: number, len = 2): string => String(n).padStart(len, '0'); + return `${pad(d.getHours())}:${pad(d.getMinutes())}:${pad(d.getSeconds())}.${pad(d.getMilliseconds(), 3)}`; +} + +function buildTimelineTicks(min: number, max: number, target = 5): number[] { + const span = Math.max(max - min, 1); + const niceSteps = [1, 2, 5, 10, 20, 50, 100, 200, 500, 1000, 2000, 5000, 10000, 30000, 60000]; + const desired = span / target; + const step = niceSteps.find(s => s >= desired) ?? span; + const ticks: number[] = []; + for (let t = min; t <= max + 0.5; t += step) { + ticks.push(t); + } + if (ticks[ticks.length - 1] < max) { + ticks.push(max); + } + return ticks; +} + +@Component({ + selector: 'flink-job-checkpoints-timeline', + templateUrl: './job-checkpoints-timeline.component.html', + styleUrls: ['./job-checkpoints-timeline.component.less'], + changeDetection: ChangeDetectionStrategy.OnPush, + imports: [ + 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>; + + public selected?: StripDatum; + public loadingTimeline = false; + public timelineEmpty = false; + public loadFailed = false; + public failedVertexCount = 0; + public requestedVertexCount = 0; + + // Diverges from @Input once polling starts feeding live updates. + private currentHistory: CheckpointHistory[] = []; + public userPinned = false; + private stripChart?: Chart; + private stripInterval?: ReturnType<Chart['interval']>; + private timelineChart?: Chart; + private noAckHatch?: HTMLCanvasElement; + private resizeObserver?: ResizeObserver; + private readonly destroy$ = new Subject<void>(); + private readonly loadTimeline$ = new Subject<StripDatum>(); + // Completed-checkpoint stats are immutable; memoize by id for instant re-selects. + 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 DETAILS_CACHE_MAX = 100; + // Flink sends Long.MAX_VALUE (well above MAX_SAFE_INTEGER) when periodic checkpointing is disabled. + private static readonly DISABLED_INTERVAL_THRESHOLD = Number.MAX_SAFE_INTEGER; + private static readonly TIMELINE_ROW_HEIGHT = 22; + private static readonly TIMELINE_HEIGHT_BASE = 80; + 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.timelineChart && this.timelineContainer.nativeElement.clientWidth > 0) { + void this.timelineChart.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 = []; + this.detailsCache.clear(); + 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(); + this.stripChart?.destroy(); + this.timelineChart?.destroy(); + } + + private toStripDatum(h: CheckpointHistory): StripDatum { + const status: StripDatum['status'] = h.is_savepoint + ? 'SAVEPOINT' + : h.status === 'FAILED' + ? 'FAILED' + : 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) + .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.renderStrip(); + + if (next.length === 0) { + this.selected = undefined; + this.clearTimeline(); + this.cdr.markForCheck(); + return; + } + + // Frozen while pinned; still markForCheck so the rotated-out note updates on poll. + if (this.userPinned) { + this.cdr.markForCheck(); + return; + } + + if (nextNewestId !== prevNewestId) { + this.selectInternal(this.toStripDatum(next[0]), false); + } Review Comment: When polling returns the same newest checkpoint ID, only the strip updates; the followed timeline keeps stale status, acknowledgement counts, and subtask details even after the checkpoint finishes. Refresh same-ID selections and retry unavailable details in follow mode, while preserving the intentional freeze for pinned selections. Coordinate reloads with in-flight requests so polling does not repeatedly cancel a slow fetch. ########## flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.ts: ########## @@ -0,0 +1,855 @@ +/* + * 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, NgIf } from '@angular/common'; +import { + AfterViewInit, + ChangeDetectionStrategy, + ChangeDetectorRef, + Component, + ElementRef, + Input, + OnChanges, + OnDestroy, + SimpleChanges, + ViewChild +} from '@angular/core'; +import { BehaviorSubject, defer, EMPTY, of, Subject, timer } from 'rxjs'; +import { catchError, distinctUntilChanged, 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 { 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'; + +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; + unaligned: boolean; + 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]; +} + +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 PHASE_LABEL: Record<PhaseKey, string> = { + start_delay: 'start_delay', + alignment: 'alignment', + sync: 'sync', + async: 'async' +}; + +const OPERATOR_NAME_MAX = 60; + +// Local HH:MM:SS.mmm — matches the header's DatePipe default (browser timezone). +function formatWallClock(epochMs: number): string { + const d = new Date(epochMs); + const pad = (n: number, len = 2): string => String(n).padStart(len, '0'); + return `${pad(d.getHours())}:${pad(d.getMinutes())}:${pad(d.getSeconds())}.${pad(d.getMilliseconds(), 3)}`; +} + +function buildTimelineTicks(min: number, max: number, target = 5): number[] { + const span = Math.max(max - min, 1); + const niceSteps = [1, 2, 5, 10, 20, 50, 100, 200, 500, 1000, 2000, 5000, 10000, 30000, 60000]; + const desired = span / target; + const step = niceSteps.find(s => s >= desired) ?? span; + const ticks: number[] = []; + for (let t = min; t <= max + 0.5; t += step) { + ticks.push(t); + } + if (ticks[ticks.length - 1] < max) { + ticks.push(max); + } + return ticks; +} + +@Component({ + selector: 'flink-job-checkpoints-timeline', + templateUrl: './job-checkpoints-timeline.component.html', + styleUrls: ['./job-checkpoints-timeline.component.less'], + changeDetection: ChangeDetectionStrategy.OnPush, + imports: [ + 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>; + + public selected?: StripDatum; + public loadingTimeline = false; + public timelineEmpty = false; + public loadFailed = false; + public failedVertexCount = 0; + public requestedVertexCount = 0; + + // Diverges from @Input once polling starts feeding live updates. + private currentHistory: CheckpointHistory[] = []; + public userPinned = false; + private stripChart?: Chart; + private stripInterval?: ReturnType<Chart['interval']>; + private timelineChart?: Chart; + private noAckHatch?: HTMLCanvasElement; + private resizeObserver?: ResizeObserver; + private readonly destroy$ = new Subject<void>(); + private readonly loadTimeline$ = new Subject<StripDatum>(); + // Completed-checkpoint stats are immutable; memoize by id for instant re-selects. + 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 DETAILS_CACHE_MAX = 100; + // Flink sends Long.MAX_VALUE (well above MAX_SAFE_INTEGER) when periodic checkpointing is disabled. + private static readonly DISABLED_INTERVAL_THRESHOLD = Number.MAX_SAFE_INTEGER; + private static readonly TIMELINE_ROW_HEIGHT = 22; + private static readonly TIMELINE_HEIGHT_BASE = 80; + 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.timelineChart && this.timelineContainer.nativeElement.clientWidth > 0) { + void this.timelineChart.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 = []; + this.detailsCache.clear(); + 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(); + this.stripChart?.destroy(); + this.timelineChart?.destroy(); + } + + private toStripDatum(h: CheckpointHistory): StripDatum { + const status: StripDatum['status'] = h.is_savepoint + ? 'SAVEPOINT' + : h.status === 'FAILED' + ? 'FAILED' + : 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) Review Comment: The strip has no minimum bar height, so checkpoints with a zero duration produce invisible, unclickable bars. This also affects an IN_PROGRESS checkpoint before its first acknowledgement: the backend returns -1, which `toStripDatum` clamps to zero. Give these bars a minimum visual height, as the timeline already does for instant rows, while keeping the actual duration in the tooltip. ########## flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.ts: ########## @@ -0,0 +1,855 @@ +/* + * 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, NgIf } from '@angular/common'; +import { + AfterViewInit, + ChangeDetectionStrategy, + ChangeDetectorRef, + Component, + ElementRef, + Input, + OnChanges, + OnDestroy, + SimpleChanges, + ViewChild +} from '@angular/core'; +import { BehaviorSubject, defer, EMPTY, of, Subject, timer } from 'rxjs'; +import { catchError, distinctUntilChanged, 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 { 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'; + +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; + unaligned: boolean; + 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]; +} + +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 PHASE_LABEL: Record<PhaseKey, string> = { + start_delay: 'start_delay', + alignment: 'alignment', + sync: 'sync', + async: 'async' +}; + +const OPERATOR_NAME_MAX = 60; + +// Local HH:MM:SS.mmm — matches the header's DatePipe default (browser timezone). +function formatWallClock(epochMs: number): string { + const d = new Date(epochMs); + const pad = (n: number, len = 2): string => String(n).padStart(len, '0'); + return `${pad(d.getHours())}:${pad(d.getMinutes())}:${pad(d.getSeconds())}.${pad(d.getMilliseconds(), 3)}`; +} + +function buildTimelineTicks(min: number, max: number, target = 5): number[] { + const span = Math.max(max - min, 1); + const niceSteps = [1, 2, 5, 10, 20, 50, 100, 200, 500, 1000, 2000, 5000, 10000, 30000, 60000]; + const desired = span / target; + const step = niceSteps.find(s => s >= desired) ?? span; + const ticks: number[] = []; + for (let t = min; t <= max + 0.5; t += step) { + ticks.push(t); + } + if (ticks[ticks.length - 1] < max) { + ticks.push(max); + } + return ticks; +} + +@Component({ + selector: 'flink-job-checkpoints-timeline', + templateUrl: './job-checkpoints-timeline.component.html', + styleUrls: ['./job-checkpoints-timeline.component.less'], + changeDetection: ChangeDetectionStrategy.OnPush, + imports: [ + 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>; + + public selected?: StripDatum; + public loadingTimeline = false; + public timelineEmpty = false; + public loadFailed = false; + public failedVertexCount = 0; + public requestedVertexCount = 0; + + // Diverges from @Input once polling starts feeding live updates. + private currentHistory: CheckpointHistory[] = []; + public userPinned = false; + private stripChart?: Chart; + private stripInterval?: ReturnType<Chart['interval']>; + private timelineChart?: Chart; + private noAckHatch?: HTMLCanvasElement; + private resizeObserver?: ResizeObserver; + private readonly destroy$ = new Subject<void>(); + private readonly loadTimeline$ = new Subject<StripDatum>(); + // Completed-checkpoint stats are immutable; memoize by id for instant re-selects. + 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 DETAILS_CACHE_MAX = 100; + // Flink sends Long.MAX_VALUE (well above MAX_SAFE_INTEGER) when periodic checkpointing is disabled. + private static readonly DISABLED_INTERVAL_THRESHOLD = Number.MAX_SAFE_INTEGER; + private static readonly TIMELINE_ROW_HEIGHT = 22; + private static readonly TIMELINE_HEIGHT_BASE = 80; + 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.timelineChart && this.timelineContainer.nativeElement.clientWidth > 0) { + void this.timelineChart.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 = []; + this.detailsCache.clear(); + 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(); + this.stripChart?.destroy(); + this.timelineChart?.destroy(); + } + + private toStripDatum(h: CheckpointHistory): StripDatum { + const status: StripDatum['status'] = h.is_savepoint + ? 'SAVEPOINT' + : h.status === 'FAILED' + ? 'FAILED' + : 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) + .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.renderStrip(); + + if (next.length === 0) { + this.selected = undefined; + this.clearTimeline(); + this.cdr.markForCheck(); + return; + } + + // Frozen while pinned; still markForCheck so the rotated-out note updates on poll. + if (this.userPinned) { + this.cdr.markForCheck(); + return; + } + + if (nextNewestId !== prevNewestId) { + this.selectInternal(this.toStripDatum(next[0]), false); + } + } + + private startPolling(): void { + this.visible$ + .pipe( + distinctUntilChanged(), + switchMap(visible => + visible + ? defer(() => + this.jobId + ? this.jobService.loadCheckpointStats(this.jobId).pipe(catchError(() => of(undefined))) + : of(undefined) + ).pipe(repeat({ delay: () => timer(this.computePollDelayMs()) })) + : EMPTY + ), + takeUntil(this.destroy$) + ) + .subscribe(stats => { + if (!stats) { + return; + } + this.applyHistory(stats.history || []); + }); + } + + private computePollDelayMs(): number { + const interval = this.config?.interval; + if ( + interval !== undefined && + interval > 0 && + interval < JobCheckpointsTimelineComponent.DISABLED_INTERVAL_THRESHOLD + ) { + return Math.max( + JobCheckpointsTimelineComponent.POLL_MIN_MS, + Math.min(interval, JobCheckpointsTimelineComponent.POLL_MAX_MS) + ); + } + return JobCheckpointsTimelineComponent.POLL_MAX_MS; + } + + private loadTimeline(datum: StripDatum): void { + if (!this.jobId || this.vertices.length === 0) { + this.showEmptyTimeline(false); + return; + } + this.loadingTimeline = true; + this.timelineEmpty = false; + this.loadFailed = false; + this.failedVertexCount = 0; + this.cdr.markForCheck(); + this.loadTimeline$.next(datum); + } + + private cacheDetails(datum: StripDatum, perVertex: Record<string, CheckpointSubTask | null>): void { + if (datum.status === 'IN_PROGRESS') { Review Comment: `status` is the strip's color category, so an in-progress savepoint has `status === 'SAVEPOINT'` and passes this guard. Its still-changing subtask data is then cached and reused even after the savepoint completes, including when the user reselects it or clicks Follow newest. Check `rawStatus` here so in-progress savepoints remain uncached just like ordinary checkpoints. ########## flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.ts: ########## @@ -0,0 +1,855 @@ +/* + * 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, NgIf } from '@angular/common'; +import { + AfterViewInit, + ChangeDetectionStrategy, + ChangeDetectorRef, + Component, + ElementRef, + Input, + OnChanges, + OnDestroy, + SimpleChanges, + ViewChild +} from '@angular/core'; +import { BehaviorSubject, defer, EMPTY, of, Subject, timer } from 'rxjs'; +import { catchError, distinctUntilChanged, 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 { 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'; + +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; + unaligned: boolean; + 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]; +} + +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 PHASE_LABEL: Record<PhaseKey, string> = { + start_delay: 'start_delay', + alignment: 'alignment', + sync: 'sync', + async: 'async' +}; + +const OPERATOR_NAME_MAX = 60; + +// Local HH:MM:SS.mmm — matches the header's DatePipe default (browser timezone). +function formatWallClock(epochMs: number): string { + const d = new Date(epochMs); + const pad = (n: number, len = 2): string => String(n).padStart(len, '0'); + return `${pad(d.getHours())}:${pad(d.getMinutes())}:${pad(d.getSeconds())}.${pad(d.getMilliseconds(), 3)}`; +} + +function buildTimelineTicks(min: number, max: number, target = 5): number[] { + const span = Math.max(max - min, 1); + const niceSteps = [1, 2, 5, 10, 20, 50, 100, 200, 500, 1000, 2000, 5000, 10000, 30000, 60000]; + const desired = span / target; + const step = niceSteps.find(s => s >= desired) ?? span; + const ticks: number[] = []; + for (let t = min; t <= max + 0.5; t += step) { + ticks.push(t); + } + if (ticks[ticks.length - 1] < max) { + ticks.push(max); + } + return ticks; +} + +@Component({ + selector: 'flink-job-checkpoints-timeline', + templateUrl: './job-checkpoints-timeline.component.html', + styleUrls: ['./job-checkpoints-timeline.component.less'], + changeDetection: ChangeDetectionStrategy.OnPush, + imports: [ + 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>; + + public selected?: StripDatum; + public loadingTimeline = false; + public timelineEmpty = false; + public loadFailed = false; + public failedVertexCount = 0; + public requestedVertexCount = 0; + + // Diverges from @Input once polling starts feeding live updates. + private currentHistory: CheckpointHistory[] = []; + public userPinned = false; + private stripChart?: Chart; + private stripInterval?: ReturnType<Chart['interval']>; + private timelineChart?: Chart; + private noAckHatch?: HTMLCanvasElement; + private resizeObserver?: ResizeObserver; + private readonly destroy$ = new Subject<void>(); + private readonly loadTimeline$ = new Subject<StripDatum>(); + // Completed-checkpoint stats are immutable; memoize by id for instant re-selects. + 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 DETAILS_CACHE_MAX = 100; + // Flink sends Long.MAX_VALUE (well above MAX_SAFE_INTEGER) when periodic checkpointing is disabled. + private static readonly DISABLED_INTERVAL_THRESHOLD = Number.MAX_SAFE_INTEGER; + private static readonly TIMELINE_ROW_HEIGHT = 22; + private static readonly TIMELINE_HEIGHT_BASE = 80; + 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.timelineChart && this.timelineContainer.nativeElement.clientWidth > 0) { + void this.timelineChart.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 = []; + this.detailsCache.clear(); + 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(); + this.stripChart?.destroy(); + this.timelineChart?.destroy(); + } + + private toStripDatum(h: CheckpointHistory): StripDatum { + const status: StripDatum['status'] = h.is_savepoint + ? 'SAVEPOINT' + : h.status === 'FAILED' + ? 'FAILED' + : 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) + .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.renderStrip(); + + if (next.length === 0) { + this.selected = undefined; + this.clearTimeline(); + this.cdr.markForCheck(); + return; + } + + // Frozen while pinned; still markForCheck so the rotated-out note updates on poll. + if (this.userPinned) { + this.cdr.markForCheck(); + return; + } + + if (nextNewestId !== prevNewestId) { + this.selectInternal(this.toStripDatum(next[0]), false); + } + } + + private startPolling(): void { + this.visible$ + .pipe( + distinctUntilChanged(), + switchMap(visible => + visible + ? defer(() => + this.jobId + ? this.jobService.loadCheckpointStats(this.jobId).pipe(catchError(() => of(undefined))) + : of(undefined) + ).pipe(repeat({ delay: () => timer(this.computePollDelayMs()) })) + : EMPTY + ), + takeUntil(this.destroy$) + ) + .subscribe(stats => { + if (!stats) { + return; + } + this.applyHistory(stats.history || []); + }); + } + + private computePollDelayMs(): number { + const interval = this.config?.interval; + if ( + interval !== undefined && + interval > 0 && + interval < JobCheckpointsTimelineComponent.DISABLED_INTERVAL_THRESHOLD + ) { + return Math.max( + JobCheckpointsTimelineComponent.POLL_MIN_MS, + Math.min(interval, JobCheckpointsTimelineComponent.POLL_MAX_MS) + ); + } + return JobCheckpointsTimelineComponent.POLL_MAX_MS; + } + + private loadTimeline(datum: StripDatum): void { + if (!this.jobId || this.vertices.length === 0) { + this.showEmptyTimeline(false); + return; + } + this.loadingTimeline = true; + this.timelineEmpty = false; + this.loadFailed = false; + this.failedVertexCount = 0; + this.cdr.markForCheck(); + this.loadTimeline$.next(datum); + } + + private cacheDetails(datum: StripDatum, perVertex: Record<string, CheckpointSubTask | null>): void { + if (datum.status === 'IN_PROGRESS') { + return; // in-progress stats still change; only cache terminal checkpoints + } + this.detailsCache.set(datum.id, perVertex); + // Evict oldest (Map keeps insertion order). + while (this.detailsCache.size > JobCheckpointsTimelineComponent.DETAILS_CACHE_MAX) { + const oldest = this.detailsCache.keys().next().value; + if (oldest === undefined) { + break; + } + this.detailsCache.delete(oldest); + } + } + + private subscribeTimelineLoads(): void { + this.loadTimeline$ + .pipe( + switchMap(datum => { + const cached = this.detailsCache.get(datum.id); + if (cached) { + return of({ + datum, + perVertex: cached, + failed: false, + failedCount: 0, + requestedCount: this.vertices.length + }); + } + const vertexIds = this.vertices.map(v => v.id); + const requestedCount = vertexIds.length; + return this.jobService.loadCheckpointAllSubtaskDetails(this.jobId, datum.id, vertexIds).pipe( + map(perVertex => { + const succeeded = Object.values(perVertex).filter(v => v !== null).length; + const failedCount = requestedCount - succeeded; + if (failedCount === 0) { + this.cacheDetails(datum, perVertex); + } + return { + datum, + perVertex, + failed: requestedCount > 0 && succeeded === 0, + failedCount, + requestedCount + }; + }), + catchError(() => + of({ + datum, + perVertex: {} as Record<string, CheckpointSubTask | null>, + failed: true, + failedCount: requestedCount, + requestedCount + }) + ) + ); + }), + takeUntil(this.destroy$) + ) + .subscribe(({ datum, perVertex, failed, failedCount, requestedCount }) => { + this.loadingTimeline = false; + if (failed) { + this.showEmptyTimeline(true); + return; + } + this.loadFailed = false; + this.failedVertexCount = failedCount; + this.requestedVertexCount = requestedCount; + this.renderTimeline(datum, perVertex); + }); + } + + private buildTimelineRows( + datum: StripDatum, + perVertex: Record<string, CheckpointSubTask | null> + ): { rows: TimelineDatum[]; unacked: UnackedDatum[]; instants: InstantDatum[]; rowOrder: string[] } { + 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 completed = s; + const startDelay = Math.max(completed.start_delay || 0, 0); + const alignment = Math.max(completed.alignment.duration || 0, 0); + const sync = Math.max(completed.checkpoint.sync || 0, 0); + const asyncMs = Math.max(completed.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: completed.index, + aborted: !!completed.aborted, + stateSize: completed.state_size, + checkpointedSize: completed.checkpointed_size, + range: [base, base] + }); + continue; + } + + const segs: Array<[PhaseKey, number]> = [ + ['start_delay', startDelay], + ['alignment', alignment], + ['sync', sync], + ['async', asyncMs] Review Comment: For unaligned checkpoints, alignment overlaps snapshot work: `docs/content/docs/ops/monitoring/checkpoint_monitoring.md` documents that async duration includes waiting for the remaining barriers. Stacking alignment before sync/async therefore shifts the displayed wall-clock times and double-counts elapsed time. Represent overlapping phases, or show durations where exact boundaries cannot be inferred, and use the reported end-to-end duration for elapsed time. -- 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]
