spuru9 commented on code in PR #28060: URL: https://github.com/apache/flink/pull/28060#discussion_r4485612301
########## flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.ts: ########## @@ -0,0 +1,1047 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +import { DatePipe, NgFor, NgIf } from '@angular/common'; +import { + AfterViewInit, + ChangeDetectionStrategy, + ChangeDetectorRef, + Component, + ElementRef, + Input, + OnChanges, + OnDestroy, + SimpleChanges, + ViewChild +} from '@angular/core'; +import { BehaviorSubject, defer, EMPTY, Observable, of, Subject, timer } from 'rxjs'; +import { catchError, distinctUntilChanged, filter, map, repeat, switchMap, takeUntil } from 'rxjs/operators'; + +import { Chart } from '@antv/g2'; +import { HumanizeBytesPipe } from '@flink-runtime-web/components/humanize-bytes.pipe'; +import { HumanizeDurationPipe } from '@flink-runtime-web/components/humanize-duration.pipe'; +import { + CheckpointConfig, + CheckpointHistory, + CheckpointSubTask, + CompletedSubTaskCheckpointStatistics, + SubTaskCheckpointStatisticsItem, + VerticesItem +} from '@flink-runtime-web/interfaces'; +import { JobService } from '@flink-runtime-web/services'; +import { formatTimeAxis, startAlignedTicks } from '@flink-runtime-web/utils'; +import { NzAlertModule } from 'ng-zorro-antd/alert'; +import { NzEmptyModule } from 'ng-zorro-antd/empty'; +import { NzSpinModule } from 'ng-zorro-antd/spin'; +import { NzTooltipModule } from 'ng-zorro-antd/tooltip'; + +type PhaseKey = 'start_delay' | 'alignment' | 'sync' | 'async'; + +// 'not-retained': every vertex returned 404, i.e. the JobManager no longer keeps this checkpoint's stats. +type LoadError = 'not-retained' | 'request-failed'; + +interface TimelineLoad { + datum: StripDatum; + perVertex: Record<string, CheckpointSubTask | null>; + failedCount: number; + requestedCount: number; + error?: LoadError; +} + +interface StripDatum { + id: number; + idLabel: string; + duration: number; + status: 'COMPLETED' | 'SAVEPOINT' | 'IN_PROGRESS' | 'FAILED'; + rawStatus: string; + isSavepoint: boolean; + triggered: number; + acked: string; + size: number; +} + +interface TimelineDatum { + row: string; + operatorFull: string; + subtask: number; + phase: PhaseKey; + phaseLabel: string; + range: [number, number]; + durationMs: number; + aborted: boolean; + stateSize: number; + checkpointedSize: number; + totalDuration: number; +} + +// A subtask that reported no stats for the checkpoint, drawn as one hatched bar spanning it. +interface UnackedDatum { + row: string; + operatorFull: string; + subtask: number; + label: 'no ack' | 'pending'; + range: [number, number]; +} + +// A subtask whose phases all rounded to 0ms, drawn as a hairline at trigger time so its row isn't empty. +interface InstantDatum { + row: string; + operatorFull: string; + subtask: number; + aborted: boolean; + stateSize: number; + checkpointedSize: number; + range: [number, number]; +} + +interface TimelineRows { + rows: TimelineDatum[]; + unacked: UnackedDatum[]; + instants: InstantDatum[]; + rowOrder: string[]; +} + +const STRIP_COLOR: Record<StripDatum['status'], string> = { + COMPLETED: '#52c41a', + SAVEPOINT: '#722ed1', + IN_PROGRESS: '#faad14', + FAILED: '#f5222d' +}; + +const STRIP_STATUSES = Object.keys(STRIP_COLOR) as Array<StripDatum['status']>; + +const PHASE_COLOR: Record<PhaseKey, string> = { + start_delay: '#b8c0cc', + alignment: '#91caff', + sync: '#fa8c16', + async: '#52c41a' +}; + +const PHASES = Object.keys(PHASE_COLOR) as PhaseKey[]; + +const NOT_COMPLETED_STROKE = '#f5222d'; +const NO_ACK_FILL = '#fff1f0'; +const NO_ACK_STRIPE = '#ffa39e'; +const INSTANT_FILL = '#8c8c8c'; + +const OPERATOR_NAME_MAX = 60; + +type IntervalMark = ReturnType<Chart['interval']>; + +// Built once and updated in place on reloads, so following an in-progress checkpoint doesn't rebuild the canvas. +interface TimelineChart { + chart: Chart; + phases: IntervalMark; + unacked: IntervalMark; + instants: IntervalMark; +} + +@Component({ + selector: 'flink-job-checkpoints-timeline', + templateUrl: './job-checkpoints-timeline.component.html', + styleUrls: ['./job-checkpoints-timeline.component.less'], + changeDetection: ChangeDetectionStrategy.OnPush, + imports: [ + NgFor, + NgIf, + DatePipe, + HumanizeDurationPipe, + HumanizeBytesPipe, + NzAlertModule, + NzEmptyModule, + NzSpinModule, + NzTooltipModule + ] +}) +export class JobCheckpointsTimelineComponent implements AfterViewInit, OnChanges, OnDestroy { + @Input() public jobId: string; + @Input() public vertices: VerticesItem[] = []; + @Input() public history: CheckpointHistory[] = []; + @Input() public config?: CheckpointConfig; + + @ViewChild('stripContainer', { static: true }) private readonly stripContainer: ElementRef<HTMLDivElement>; + @ViewChild('timelineContainer', { static: true }) private readonly timelineContainer: ElementRef<HTMLDivElement>; + @ViewChild('timelineScroll', { static: true }) private readonly timelineScroll: ElementRef<HTMLDivElement>; + @ViewChild('timelineWrapper', { static: true }) private readonly timelineWrapper: ElementRef<HTMLDivElement>; + + public selected?: StripDatum; + public loadingTimeline = false; + public timelineEmpty = false; + public loadError?: LoadError; + public failedVertexCount = 0; + public requestedVertexCount = 0; + // Rows of the shown checkpoint, how many of the slowest are shown, and the first one scrolled to. + public totalRows = 0; + public slowestCount = JobCheckpointsTimelineComponent.SHOWN_ROWS_DEFAULT; + public firstRow = 0; + public chartHeight = 0; + public slowestFirst = false; + + // Diverges from @Input once polling starts feeding live updates. + private currentHistory: CheckpointHistory[] = []; + public userPinned = false; + private stripChart?: Chart; + private stripInterval?: IntervalMark; + private timeline?: TimelineChart; + // Full operator name per row label, for the native tooltip on the truncated labels. + private rowTitles = new Map<string, string>(); + // Kept so scrolling can redraw another window of rows without rebuilding them. + private timelineRows?: TimelineRows & { datum: StripDatum; slowestOrder: string[] }; + private scrollFrame?: number; + private noAckHatch?: HTMLCanvasElement; + private resizeObserver?: ResizeObserver; + private readonly destroy$ = new Subject<void>(); + // null cancels a load in flight. + private readonly loadTimeline$ = new Subject<StripDatum | null>(); + // A load requested while the Timeline tab was hidden; it runs once the tab is shown. + private loadDeferred = false; + // The checkpoint whose failed load was already retried while following, so polls don't retry it again. + private retriedCheckpointId?: number; + // Terminal (completed or failed) checkpoint stats no longer change; memoize by id for instant re-selects. + // Holds only the checkpoints that can be selected again: those in the strip, plus the pinned one. + private readonly detailsCache = new Map<number, Record<string, CheckpointSubTask | null>>(); + private readonly visible$ = new BehaviorSubject<boolean>(false); + private static readonly POLL_MIN_MS = 1500; + private static readonly POLL_MAX_MS = 15000; + private static readonly STRIP_MAX_BARS = 60; + private static readonly TIMELINE_ROW_HEIGHT = 22; + // paddingTop (time axis) + paddingBottom, plus a little slack. + private static readonly TIMELINE_HEIGHT_BASE = 56; + // Rows shown by default (the slowest) and the most that can be picked; the chart is at most VISIBLE_ROWS tall. + private static readonly SHOWN_ROWS_DEFAULT = 25; + private static readonly SHOWN_ROWS_MAX = 200; + private static readonly VISIBLE_ROWS = 25; + private static readonly TIMELINE_HEIGHT_MIN = 160; + private static readonly LABEL_CHAR_WIDTH = 7; + private static readonly LABEL_PADDING_BUFFER = 24; + private static readonly LABEL_PADDING_MIN = 260; + private static readonly LABEL_PADDING_MAX = 520; + + private readonly humanizeBytesPipe = new HumanizeBytesPipe(); + private readonly humanizeDurationPipe = new HumanizeDurationPipe(); + + constructor( + private readonly jobService: JobService, + private readonly cdr: ChangeDetectorRef + ) {} + + public ngAfterViewInit(): void { + this.currentHistory = this.history || []; + this.renderStrip(); + this.subscribeTimelineLoads(); + if (this.currentHistory.length > 0) { + this.selectInternal(this.toStripDatum(this.currentHistory[0]), false); + } + this.startPolling(); Review Comment: Agreed this would be nicer. It changes how the other tabs refresh as well, so I've opened FLINK-41364 to follow up on it separately. Have addressed other comments. -- 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]
