spuru9 commented on code in PR #28060: URL: https://github.com/apache/flink/pull/28060#discussion_r4267427803
########## flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.ts: ########## @@ -0,0 +1,764 @@ +/* + * 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 * as G2 from '@antv/g2'; +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; + outlier: boolean; + aborted: boolean; + stateSize: number; + checkpointedSize: number; + totalDuration: number; +} + +const STRIP_COLOR: Record<StripDatum['status'], string> = { + COMPLETED: '#52c41a', + SAVEPOINT: '#722ed1', + IN_PROGRESS: '#faad14', + FAILED: '#f5222d' +}; + +const PHASE_COLOR: Record<PhaseKey, string> = { + start_delay: '#b8c0cc', + alignment: '#91caff', + sync: '#fa8c16', + async: '#52c41a' +}; + +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 timelineChart?: Chart; + 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) { + this.stripChart.forceFit(); + } + if (this.timelineChart && this.timelineContainer.nativeElement.clientWidth > 0) { + 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.clearTimeline(); + this.renderStrip(); + } Review Comment: Fixed: `detailsCache` is now cleared when `jobId` changes. ########## flink-runtime-web/web-dashboard/src/app/services/job.service.ts: ########## @@ -183,6 +183,23 @@ export class JobService { ); } + public loadCheckpointAllSubtaskDetails( + jobId: string, + checkPointId: number, + vertexIds: string[] + ): Observable<Record<string, CheckpointSubTask | null>> { + if (vertexIds.length === 0) { + return of({} as Record<string, CheckpointSubTask | null>); + } + const requests = vertexIds.map(vid => + this.loadCheckpointSubtaskDetails(jobId, checkPointId, vid).pipe( + catchError(() => of<CheckpointSubTask | null>(null)), + map(result => ({ [vid]: result })) + ) + ); + return forkJoin(requests).pipe(map(parts => Object.assign({}, ...parts))); + } Review Comment: Fixed: requests are now bounded to 6 concurrent via `mergeMap`, still returning one merged map. ########## flink-runtime-web/web-dashboard/src/app/app.interceptor.ts: ########## @@ -59,9 +59,18 @@ export class AppInterceptor implements HttpInterceptor { window.location.href = String(res.headers.get('Location')); } + // A per-subtask checkpoint fetch can race the JM reaping that checkpoint (or hit it + // before it acks), yielding a benign 404. Suppress only that case; real 5xx/auth on + // the same path still surface. + const isExpectedSubtaskNotFound = + res instanceof HttpResponseBase && + res.status === HttpStatusCode.NotFound && + /\/checkpoints\/details\/\d+\/subtasks\//.test(res.url || ''); + Review Comment: Fixed: suppression now uses the `EXPECTED_NOT_FOUND` request context and is set only by the timeline fetches, so other pages still show the toast. -- 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]
