spuru9 commented on code in PR #28060: URL: https://github.com/apache/flink/pull/28060#discussion_r4434877200
########## 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: Out of scope for this PR: the dashboard's other charts are pointer-only as well, and the History tab already lists every checkpoint in a keyboard-accessible table. ########## 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: Fixed: the empty state now reads the live polled history. ########## 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: Fixed: strip bars have a 3px minimum height; the tooltip keeps the real duration. ########## 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: Fixed: while following, the newest checkpoint reloads when its status or ack count changes or its last load failed, skipping polls while a load is in flight; pinned checkpoints stay frozen. ########## 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: Fixed: the cache check now uses `rawStatus`. ########## 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: Fixed: alignment is no longer stacked for unaligned subtasks; the async tooltip shows it as included and 'total' excludes it. -- 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]
