This is an automated email from the ASF dual-hosted git repository.
ferenc-csaky pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 12c944d5def [FLINK-39566][runtime-web] Add checkpoint duration Gantt
view
12c944d5def is described below
commit 12c944d5def4edebe93e68f942922fca4b0d7b19
Author: Purushottam Sinha <[email protected]>
AuthorDate: Fri Nov 6 19:10:17 2026 +0530
[FLINK-39566][runtime-web] Add checkpoint duration Gantt view
Generated-by: Claude Code (Claude Opus 5.5)
---
.../docs/ops/monitoring/checkpoint_monitoring.md | 16 +-
.../docs/ops/monitoring/checkpoint_monitoring.md | 14 +-
.../src/app/interfaces/job-checkpoint.ts | 2 +-
.../job/checkpoints/job-checkpoints.component.html | 12 +-
.../checkpoints/job-checkpoints.component.spec.ts | 18 +-
.../job/checkpoints/job-checkpoints.component.ts | 4 +-
.../timeline/checkpoint-timeline-rows.spec.ts | 190 +++++
.../timeline/checkpoint-timeline-rows.ts | 267 ++++++
.../timeline/checkpoint-timeline.testing.ts | 70 ++
.../job-checkpoints-timeline.component.html | 202 +++++
.../job-checkpoints-timeline.component.less | 201 +++++
.../job-checkpoints-timeline.component.spec.ts | 672 +++++++++++++++
.../timeline/job-checkpoints-timeline.component.ts | 913 +++++++++++++++++++++
.../job/timeline/job-timeline.component.spec.ts | 20 +-
.../pages/job/timeline/job-timeline.component.ts | 24 +-
.../src/app/services/job.service.spec.ts | 41 +-
.../web-dashboard/src/app/services/job.service.ts | 39 +-
.../web-dashboard/src/app/utils/public-api.ts | 1 +
.../web-dashboard/src/app/utils/time-axis.spec.ts | 43 +
.../web-dashboard/src/app/utils/time-axis.ts | 42 +
20 files changed, 2735 insertions(+), 56 deletions(-)
diff --git a/docs/content.zh/docs/ops/monitoring/checkpoint_monitoring.md
b/docs/content.zh/docs/ops/monitoring/checkpoint_monitoring.md
index 419151b6bcf..dfadfe3aed2 100644
--- a/docs/content.zh/docs/ops/monitoring/checkpoint_monitoring.md
+++ b/docs/content.zh/docs/ops/monitoring/checkpoint_monitoring.md
@@ -31,7 +31,7 @@ under the License.
## 概览（Overview）
-Flink 的 Web 界面提供了`选项卡/标签（tab）`来监视作业的 checkpoint
信息。作业终止后，这些统计信息仍然可用。有四个不同的选项卡可显示有关 checkpoint
的信息：概览（Overview），历史记录（History），摘要信息（Summary）和配置信息（Configuration）。以下各节将依次介绍这些内容。
+Flink 的 Web 界面提供了`选项卡/标签（tab）`来监视作业的 checkpoint
信息。作业终止后，这些统计信息仍然可用。有五个不同的选项卡可显示有关 checkpoint
的信息：概览（Overview），时间线（Timeline），历史记录（History），摘要信息（Summary）和配置信息（Configuration）。以下各节将依次介绍这些内容。
<a name="monitoring"></a>
@@ -56,6 +56,20 @@ Flink 的 Web 界面提供了`选项卡/标签（tab）`来监视作业的 check
- Restore from Checkpoint：从 checkpoint 恢复。
- Restore from Savepoint：从 savepoint 恢复。
+<a name="timeline-tab"></a>
+
+### Timeline Tab
+
+The timeline tab shows where the time of a checkpoint went, subtask by subtask.
+
+- **Recent Checkpoints**: One bar per checkpoint in the
[history](#history-size-configuration), colored by status (completed,
savepoint, in progress, failed). The bar height is the end to end duration.
Hover over a bar for its status, duration and acknowledged subtasks.
+- **Per-subtask timeline**: One row per subtask, in operator then subtask
order, plotted against wall-clock time from the moment the checkpoint was
triggered. Each row stacks the subtask's phases: start delay, alignment, sync
and async (see [History Tab](#history-tab) for what they mean). With unaligned
checkpoints, only the alignment before the switch to unaligned is drawn: the
switch happens at the first barrier, or once
`execution.checkpointing.aligned-checkpoint-timeout` has passed s [...]
+ - Subtasks that started their snapshot but did not finish it before the
checkpoint failed are outlined as *aborted*.
+ - Subtasks that reported no statistics get a hatched bar spanning the
checkpoint: *pending* while the checkpoint is in progress, *no ack* once it has
failed.
+ - Subtasks whose phases all took 0 ms, for example tasks that have
already finished, are marked with a thin line at the trigger time.
+
+By default the per-subtask timeline follows the newest checkpoint. Click a bar
in *Recent Checkpoints* to pin that checkpoint instead, and click it again or
click *Follow newest* to resume following. While the tab is open, it refreshes
on its own at the checkpoint interval, bounded between 1.5 and 15 seconds.
+
<a name="history-tab"></a>
### 历史记录（History）选项卡
diff --git a/docs/content/docs/ops/monitoring/checkpoint_monitoring.md
b/docs/content/docs/ops/monitoring/checkpoint_monitoring.md
index b961ae8ce78..a84e6d2a021 100644
--- a/docs/content/docs/ops/monitoring/checkpoint_monitoring.md
+++ b/docs/content/docs/ops/monitoring/checkpoint_monitoring.md
@@ -29,7 +29,7 @@ under the License.
## Overview
-Flink's web interface provides a tab to monitor the checkpoints of jobs. These
stats are also available after the job has terminated. There are four different
tabs to display information about your checkpoints: Overview, History, Summary,
and Configuration. The following sections will cover all of these in turn.
+Flink's web interface provides a tab to monitor the checkpoints of jobs. These
stats are also available after the job has terminated. There are five different
tabs to display information about your checkpoints: Overview, Timeline,
History, Summary, and Configuration. The following sections will cover all of
these in turn.
## Monitoring
@@ -50,6 +50,18 @@ The overview tabs lists the following statistics. Note that
these statistics don
- Restore from Checkpoint: We restored from a regular periodic
checkpoint.
- Restore from Savepoint: We restored from a savepoint.
+### Timeline Tab
+
+The timeline tab shows where the time of a checkpoint went, subtask by subtask.
+
+- **Recent Checkpoints**: One bar per checkpoint in the
[history](#history-size-configuration), colored by status (completed,
savepoint, in progress, failed). The bar height is the end to end duration.
Hover over a bar for its status, duration and acknowledged subtasks.
+- **Per-subtask timeline**: One row per subtask, in operator then subtask
order, plotted against wall-clock time from the moment the checkpoint was
triggered. Each row stacks the subtask's phases: start delay, alignment, sync
and async (see [History Tab](#history-tab) for what they mean). With unaligned
checkpoints, only the alignment before the switch to unaligned is drawn: the
switch happens at the first barrier, or once
`execution.checkpointing.aligned-checkpoint-timeout` has passed s [...]
+ - Subtasks that started their snapshot but did not finish it before the
checkpoint failed are outlined as *aborted*.
+ - Subtasks that reported no statistics get a hatched bar spanning the
checkpoint: *pending* while the checkpoint is in progress, *no ack* once it has
failed.
+ - Subtasks whose phases all took 0 ms, for example tasks that have
already finished, are marked with a thin line at the trigger time.
+
+By default the per-subtask timeline follows the newest checkpoint. Click a bar
in *Recent Checkpoints* to pin that checkpoint instead, and click it again or
click *Follow newest* to resume following. While the tab is open, it refreshes
on its own at the checkpoint interval, bounded between 1.5 and 15 seconds.
+
### History Tab
The checkpoint history keeps statistics about recently triggered checkpoints,
including those that are currently in progress.
diff --git
a/flink-runtime-web/web-dashboard/src/app/interfaces/job-checkpoint.ts
b/flink-runtime-web/web-dashboard/src/app/interfaces/job-checkpoint.ts
index 5979a7d0cd0..27a7835cf06 100644
--- a/flink-runtime-web/web-dashboard/src/app/interfaces/job-checkpoint.ts
+++ b/flink-runtime-web/web-dashboard/src/app/interfaces/job-checkpoint.ts
@@ -57,8 +57,8 @@ export interface Checkpoint {
is_savepoint: boolean;
external_path: string;
};
- history: CheckpointHistory;
};
+ history: CheckpointHistory[];
}
export interface CheckpointHistory {
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/job-checkpoints.component.html
b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/job-checkpoints.component.html
index 692b9cbff0b..49518fd7f3e 100644
---
a/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/job-checkpoints.component.html
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/job-checkpoints.component.html
@@ -222,11 +222,19 @@
</tbody>
</nz-table>
</nz-tab>
+ <nz-tab nzTitle="Timeline">
+ <flink-job-checkpoints-timeline
+ [jobId]="jobDetail?.jid"
+ [vertices]="jobDetail?.vertices || []"
+ [history]="checkPointStats?.history || []"
+ [config]="checkPointConfig"
+ ></flink-job-checkpoints-timeline>
+ </nz-tab>
<nz-tab nzTitle="History">
<nz-table
class="no-border small"
[nzSize]="'small'"
- [nzData]="checkPointStats['history'] || []"
+ [nzData]="checkPointStats?.history || []"
[nzFrontPagination]="false"
[nzShowPagination]="false"
>
@@ -256,7 +264,7 @@
</tr>
</thead>
<tbody>
- <ng-container *ngFor="let checkpoint of checkPointStats['history'];
trackBy: trackById">
+ <ng-container *ngFor="let checkpoint of checkPointStats?.history;
trackBy: trackById">
<tr>
<td nzShowExpand [(nzExpand)]="checkpoint.expand"></td>
<td>{{ checkpoint['id'] }}</td>
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/job-checkpoints.component.spec.ts
b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/job-checkpoints.component.spec.ts
index 7872f607f33..4d33e2522be 100644
---
a/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/job-checkpoints.component.spec.ts
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/job-checkpoints.component.spec.ts
@@ -16,6 +16,7 @@
* limitations under the License.
*/
+import { Component, Input } from '@angular/core';
import { ComponentFixture, TestBed } from '@angular/core/testing';
import { of, throwError } from 'rxjs';
@@ -26,6 +27,7 @@ import { beforeEach, describe, expect, it, vi } from 'vitest';
import { JobCheckpointsComponent } from './job-checkpoints.component';
import { JobLocalService } from '../job-local.service';
+import { JobCheckpointsTimelineComponent } from
'./timeline/job-checkpoints-timeline.component';
const mockStats = {
counts: { restored: 0, total: 3, in_progress: 0, completed: 3, failed: 0 },
@@ -49,6 +51,15 @@ const mockConfig = {
state_changelog_enabled: false
};
+// The real timeline draws with G2 on a canvas, which jsdom does not provide.
+@Component({ selector: 'flink-job-checkpoints-timeline', template: '' })
+class JobCheckpointsTimelineStubComponent {
+ @Input() jobId?: string;
+ @Input() vertices: unknown[] = [];
+ @Input() history: unknown[] = [];
+ @Input() config?: unknown;
+}
+
describe('JobCheckpointsComponent', () => {
let fixture: ComponentFixture<JobCheckpointsComponent>;
let element: HTMLElement;
@@ -65,7 +76,12 @@ describe('JobCheckpointsComponent', () => {
{ provide: JobService, useValue: { loadCheckpointStats,
loadCheckpointConfig } },
{ provide: JobLocalService, useValue: { jobDetailChanges: () => of({
jid: 'job-1' }) } }
]
- }).compileComponents();
+ })
+ .overrideComponent(JobCheckpointsComponent, {
+ remove: { imports: [JobCheckpointsTimelineComponent] },
+ add: { imports: [JobCheckpointsTimelineStubComponent] }
+ })
+ .compileComponents();
fixture = TestBed.createComponent(JobCheckpointsComponent);
element = fixture.nativeElement as HTMLElement;
});
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/job-checkpoints.component.ts
b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/job-checkpoints.component.ts
index 279f50efa9b..8566a1cc09c 100644
---
a/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/job-checkpoints.component.ts
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/job-checkpoints.component.ts
@@ -26,6 +26,7 @@ import { HumanizeBytesPipe } from
'@flink-runtime-web/components/humanize-bytes.
import { HumanizeDurationPipe } from
'@flink-runtime-web/components/humanize-duration.pipe';
import { CheckpointConfig, CheckpointHistory, Checkpoint, JobDetailCorrect }
from '@flink-runtime-web/interfaces';
import { JobCheckpointsDetailComponent } from
'@flink-runtime-web/pages/job/checkpoints/detail/job-checkpoints-detail.component';
+import { JobCheckpointsTimelineComponent } from
'@flink-runtime-web/pages/job/checkpoints/timeline/job-checkpoints-timeline.component';
import { JobService } from '@flink-runtime-web/services';
import { NzButtonModule } from 'ng-zorro-antd/button';
import { NzCollapseModule } from 'ng-zorro-antd/collapse';
@@ -59,7 +60,8 @@ import { JobLocalService } from '../job-local.service';
NzButtonModule,
NzEmptyModule,
NzCollapseModule,
- NzTooltipModule
+ NzTooltipModule,
+ JobCheckpointsTimelineComponent
]
})
export class JobCheckpointsComponent implements OnInit, OnDestroy {
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/checkpoint-timeline-rows.spec.ts
b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/checkpoint-timeline-rows.spec.ts
new file mode 100644
index 00000000000..352ab73411a
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/checkpoint-timeline-rows.spec.ts
@@ -0,0 +1,190 @@
+/*
+ * 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 { CheckpointConfig, CheckpointHistory, CheckpointSubTask, VerticesItem
} from '@flink-runtime-web/interfaces';
+import { describe, expect, it } from 'vitest';
+
+import { buildTimelineRows, TimelineCheckpoint, TimelineRows, toStripDatum }
from './checkpoint-timeline-rows';
+import { completed, details, noStats, TRIGGERED, vertex } from
'./checkpoint-timeline.testing';
+
+const checkpoint = (rawStatus: string, duration: number, isSavepoint = false):
TimelineCheckpoint => ({
+ duration,
+ rawStatus,
+ isSavepoint,
+ triggered: TRIGGERED
+});
+
+const build = (
+ datum: TimelineCheckpoint,
+ vertices: VerticesItem[],
+ perVertex: Record<string, CheckpointSubTask | null>,
+ config?: CheckpointConfig
+): TimelineRows => buildTimelineRows(datum, vertices, perVertex, config);
+
+const unalignedConfig = (alignedCheckpointTimeout: number, mode =
'exactly_once'): CheckpointConfig =>
+ ({ mode, unaligned_checkpoints: true, aligned_checkpoint_timeout:
alignedCheckpointTimeout }) as CheckpointConfig;
+
+const SUBTASK_PHASES = { start_delay: 2, alignment: 5, sync: 1, async: 8 };
+
+const segments = (rows: TimelineRows['rows'], row: string): Array<[string,
string, [number, number], number]> =>
+ rows.filter(r => r.row === row).map(r => [r.phase, r.phaseLabel, r.range,
r.totalDuration]);
+
+describe('buildTimelineRows', () => {
+ const agg = [vertex('a', 'Agg')];
+
+ it('orders rows by operator, in job graph order, then by subtask index', ()
=> {
+ // Chosen so that every other order differs: the response order, by
operator name, by subtask index alone,
+ // and by duration either way.
+ const { rowOrder } = build(checkpoint('COMPLETED', 60), [vertex('s',
'Source'), vertex('a', 'Agg')], {
+ s: details(completed(1, { async: 1 }), completed(0, { async: 30 })),
+ a: details(completed(2, { async: 5 }), completed(0, { async: 50 }),
completed(1, { async: 20 }))
+ });
+
+ expect(rowOrder).toEqual(['Source #0', 'Source #1', 'Agg #0', 'Agg #1',
'Agg #2']);
+ });
+
+ it('keeps a row with an instant marker for subtasks whose phases all rounded
to 0ms', () => {
+ const { instants, rowOrder } = build(checkpoint('COMPLETED', 15),
[vertex('s', 'Source')], {
+ s: details(completed(0, {}), completed(1, { start_delay: 1 }))
+ });
+
+ expect(rowOrder).toEqual(['Source #0', 'Source #1']);
+ expect(instants).toEqual([
+ expect.objectContaining({ row: 'Source #0', aborted: false, range:
[TRIGGERED, TRIGGERED] })
+ ]);
+ });
+
+ it('draws a no-ack bar spanning a failed checkpoint for subtasks without
stats', () => {
+ const { unacked, rowOrder } = build(checkpoint('FAILED', 10), agg, {
+ a: details(completed(0, { start_delay: 1, async: 3 }, true), noStats(1),
completed(2, { async: 2 }))
+ });
+
+ expect(unacked).toEqual([
+ expect.objectContaining({ row: 'Agg #1', label: 'no ack', range:
[TRIGGERED, TRIGGERED + 10] })
+ ]);
+ expect(rowOrder).toEqual(['Agg #0', 'Agg #1', 'Agg #2']);
+ });
+
+ it('marks subtasks without stats as pending up to the latest known progress
while in progress', () => {
+ const { unacked } = build(checkpoint('IN_PROGRESS', 0), agg, {
+ a: details(completed(0, { start_delay: 1, async: 4 }), noStats(1))
+ });
+
+ expect(unacked).toEqual([expect.objectContaining({ label: 'pending',
range: [TRIGGERED, TRIGGERED + 5] })]);
+ });
+
+ it('uses the raw status for savepoints', () => {
+ const { unacked } = build(checkpoint('FAILED', 10, true), agg, { a:
details(noStats(0)) });
+
+ expect(unacked.map(u => u.label)).toEqual(['no ack']);
+ });
+
+ it('ignores missing stats on completed checkpoints', () => {
+ const { unacked, rowOrder } = build(checkpoint('COMPLETED', 10), agg, {
+ a: details(completed(0, { async: 2 }), noStats(1))
+ });
+
+ expect(unacked).toEqual([]);
+ expect(rowOrder).toEqual(['Agg #0']);
+ });
+
+ it('keeps row keys unique when two operator names shorten to the same
label', () => {
+ const name = `${'x'.repeat(70)}`;
+ const { rowOrder } = build(
+ checkpoint('COMPLETED', 10),
+ [vertex('aaaa1', `${name}1`), vertex('bbbb2', `${name}2`)],
+ {
+ aaaa1: details(completed(0, { async: 1 })),
+ bbbb2: details(completed(0, { async: 1 }))
+ }
+ );
+
+ expect(new Set(rowOrder).size).toBe(2);
+ expect(rowOrder[1]).toContain('[bbbb]');
+ });
+
+ // Long.MAX_VALUE (2 ** 63 once parsed) means no timeout, i.e. unaligned
from the first barrier, like 0.
+ it.each([0, 2 ** 63])('draws no alignment before sync for checkpoints
unaligned from the start (timeout %s)', t => {
+ const { rows } = build(
+ checkpoint('COMPLETED', 20),
+ agg,
+ { a: details(completed(0, SUBTASK_PHASES)) },
+ unalignedConfig(t)
+ );
+
+ expect(segments(rows, 'Agg #0')).toEqual([
+ ['start_delay', 'start_delay', [TRIGGERED, TRIGGERED + 2], 11],
+ ['sync', 'sync', [TRIGGERED + 2, TRIGGERED + 3], 11],
+ ['async', 'async (incl. 5ms alignment)', [TRIGGERED + 3, TRIGGERED +
11], 11]
+ ]);
+ });
+
+ it('draws the alignment up to the aligned checkpoint timeout before sync',
() => {
+ const { rows } = build(
+ checkpoint('COMPLETED', 40_000),
+ agg,
+ {
+ a: details(
+ completed(0, { start_delay: 5_000, alignment: 31_000, sync: 10,
async: 8_000 }),
+ completed(1, SUBTASK_PHASES)
+ )
+ },
+ unalignedConfig(30_000)
+ );
+
+ expect(segments(rows, 'Agg #0')).toEqual([
+ ['start_delay', 'start_delay', [TRIGGERED, TRIGGERED + 5_000], 38_010],
+ ['alignment', 'alignment', [TRIGGERED + 5_000, TRIGGERED + 30_000],
38_010],
+ ['sync', 'sync', [TRIGGERED + 30_000, TRIGGERED + 30_010], 38_010],
+ ['async', 'async (incl. 6000ms alignment)', [TRIGGERED + 30_010,
TRIGGERED + 38_010], 38_010]
+ ]);
+ // Aligned within the timeout, so it never switched.
+ expect(segments(rows, 'Agg #1').map(([, label]) =>
label)).toEqual(['start_delay', 'alignment', 'sync', 'async']);
+ });
+
+ it.each([
+ ['savepoints', unalignedConfig(0), true],
+ ['at-least-once checkpoints', unalignedConfig(0, 'at_least_once'), false],
+ ['jobs without unaligned checkpoints', { ...unalignedConfig(0),
unaligned_checkpoints: false }, false]
+ ])('draws the whole alignment before sync for %s', (_, config, isSavepoint)
=> {
+ const { rows } = build(
+ checkpoint('COMPLETED', 20, isSavepoint),
+ agg,
+ { a: details(completed(0, SUBTASK_PHASES)) },
+ config
+ );
+
+ expect(segments(rows, 'Agg #0')).toEqual([
+ ['start_delay', 'start_delay', [TRIGGERED, TRIGGERED + 2], 16],
+ ['alignment', 'alignment', [TRIGGERED + 2, TRIGGERED + 7], 16],
+ ['sync', 'sync', [TRIGGERED + 7, TRIGGERED + 8], 16],
+ ['async', 'async', [TRIGGERED + 8, TRIGGERED + 16], 16]
+ ]);
+ });
+});
+
+describe('toStripDatum', () => {
+ const savepoint = (status: string): CheckpointHistory =>
+ ({ id: 7, status, is_savepoint: true, num_acknowledged_subtasks: 1,
num_subtasks: 4 }) as CheckpointHistory;
+
+ it('draws a failed savepoint as failed, and an in-progress one as a
savepoint', () => {
+ expect(toStripDatum(savepoint('FAILED')).status).toBe('FAILED');
+ expect(toStripDatum(savepoint('IN_PROGRESS')).status).toBe('SAVEPOINT');
+ expect(toStripDatum(savepoint('IN_PROGRESS')).acked).toBe('1/4');
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/checkpoint-timeline-rows.ts
b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/checkpoint-timeline-rows.ts
new file mode 100644
index 00000000000..0653f393759
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/checkpoint-timeline-rows.ts
@@ -0,0 +1,267 @@
+/*
+ * 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 {
+ CheckpointConfig,
+ CheckpointHistory,
+ CheckpointSubTask,
+ CompletedSubTaskCheckpointStatistics,
+ SubTaskCheckpointStatisticsItem,
+ VerticesItem
+} from '@flink-runtime-web/interfaces';
+
+export type PhaseKey = 'start_delay' | 'alignment' | 'sync' | 'async';
+
+// One bar of the recent-checkpoints strip, and the checkpoint the timeline
below it shows.
+export interface StripDatum {
+ id: number;
+ idLabel: string;
+ duration: number;
+ status: 'COMPLETED' | 'SAVEPOINT' | 'IN_PROGRESS' | 'FAILED';
+ rawStatus: string;
+ isSavepoint: boolean;
+ triggered: number;
+ acked: string;
+ size: number;
+}
+
+export 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.
+export 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.
+export interface InstantDatum {
+ row: string;
+ operatorFull: string;
+ subtask: number;
+ aborted: boolean;
+ stateSize: number;
+ checkpointedSize: number;
+ range: [number, number];
+}
+
+export interface TimelineRows {
+ rows: TimelineDatum[];
+ unacked: UnackedDatum[];
+ instants: InstantDatum[];
+ rowOrder: string[];
+}
+
+// What building the rows reads from a strip bar.
+export type TimelineCheckpoint = Pick<StripDatum, 'duration' | 'rawStatus' |
'isSavepoint' | 'triggered'>;
+
+const OPERATOR_NAME_MAX = 60;
+
+export function toStripDatum(h: CheckpointHistory): StripDatum {
+ // FAILED first, so a failed savepoint shows as failed rather than as a
savepoint.
+ const status: StripDatum['status'] =
+ h.status === 'FAILED'
+ ? 'FAILED'
+ : h.is_savepoint
+ ? 'SAVEPOINT'
+ : h.status === 'IN_PROGRESS'
+ ? 'IN_PROGRESS'
+ : 'COMPLETED';
+ return {
+ id: h.id,
+ idLabel: String(h.id),
+ duration: Math.max(h.end_to_end_duration || 0, 0),
+ status,
+ rawStatus: h.status,
+ isSavepoint: h.is_savepoint,
+ triggered: h.trigger_timestamp,
+ acked: `${h.num_acknowledged_subtasks}/${h.num_subtasks}`,
+ size: h.state_size
+ };
+}
+
+// One row per subtask, in operator then subtask order, with a segment per
checkpoint phase.
+export function buildTimelineRows(
+ datum: TimelineCheckpoint,
+ vertices: VerticesItem[],
+ perVertex: Record<string, CheckpointSubTask | null>,
+ config?: CheckpointConfig
+): TimelineRows {
+ const base = datum.triggered;
+ // A completed checkpoint was acked by every running subtask, so missing
stats there are not a failure.
+ const unackedLabel =
+ datum.rawStatus === 'FAILED' ? 'no ack' : datum.rawStatus ===
'IN_PROGRESS' ? 'pending' : undefined;
+ const rows: TimelineDatum[] = [];
+ const unacked: UnackedDatum[] = [];
+ const instants: InstantDatum[] = [];
+ const rowOrder: string[] = [];
+ let maxTotal = 0;
+ const rowPrefixOwner = new Map<string, string>();
+
+ for (const vertex of 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);
+
+ // By index, rather than in the order the response lists them.
+ for (const s of [...cp.subtasks].sort((x, y) => x.index - y.index)) {
+ const row = `${opName} #${s.index}`;
+ if (!isCompletedSubtask(s)) {
+ // Finished tasks never land here: the JobManager records them as
acknowledged when it triggers.
+ if (unackedLabel) {
+ unacked.push({ row, operatorFull: opNameFull, subtask: s.index,
label: unackedLabel, range: [base, base] });
+ rowOrder.push(row);
+ }
+ continue;
+ }
+ const startDelay = Math.max(s.start_delay || 0, 0);
+ const reportedAlignment = Math.max(s.alignment.duration || 0, 0);
+ const alignment = alignmentBeforeSync(datum, config, startDelay,
reportedAlignment);
+ const alignmentInAsync = reportedAlignment - alignment;
+ const sync = Math.max(s.checkpoint.sync || 0, 0);
+ const asyncMs = Math.max(s.checkpoint.async || 0, 0);
+ const total = startDelay + alignment + sync + asyncMs;
+ rowOrder.push(row);
+ maxTotal = Math.max(maxTotal, total);
+ if (total === 0) {
+ instants.push({
+ row,
+ operatorFull: opNameFull,
+ subtask: s.index,
+ aborted: !!s.aborted,
+ stateSize: s.state_size,
+ checkpointedSize: s.checkpointed_size,
+ range: [base, base]
+ });
+ continue;
+ }
+
+ const segs: Array<[PhaseKey, number]> = [
+ ['start_delay', startDelay],
+ ['alignment', alignment],
+ ['sync', sync],
+ ['async', asyncMs]
+ ];
+ let cursor = base;
+ for (const [phase, ms] of segs) {
+ if (ms > 0) {
+ rows.push({
+ row,
+ operatorFull: opNameFull,
+ subtask: s.index,
+ phase,
+ phaseLabel:
+ phase === 'async' && alignmentInAsync > 0 ? `${phase} (incl.
${alignmentInAsync}ms alignment)` : phase,
+ range: [cursor, cursor + ms],
+ durationMs: ms,
+ aborted: !!s.aborted,
+ stateSize: s.state_size,
+ checkpointedSize: s.checkpointed_size,
+ totalDuration: total
+ });
+ }
+ cursor += ms;
+ }
+ }
+ }
+
+ // Unacked subtasks span the whole checkpoint: until it failed, or the
latest ack so far while in progress.
+ const unackedEnd = base + Math.max(datum.duration, maxTotal, 1);
+ for (const u of unacked) {
+ u.range = [base, unackedEnd];
+ }
+
+ return { rows, unacked, instants, rowOrder };
+}
+
+// How much of a subtask's alignment came before its sync phase. With
unaligned checkpoints, a checkpoint starts
+// aligned and switches to unaligned once aligned_checkpoint_timeout has
passed since the trigger; after that the
+// rest of the alignment runs during the async phase, whose duration includes
it (checkpoint_monitoring.md).
+// Savepoints and at-least-once checkpoints never switch. The per-subtask
unaligned_checkpoint flag can't tell: it
+// only says whether in-flight data was persisted.
+function alignmentBeforeSync(
+ datum: TimelineCheckpoint,
+ config: CheckpointConfig | undefined,
+ startDelay: number,
+ alignment: number
+): number {
+ if (!config?.unaligned_checkpoints || config.mode !== 'exactly_once' ||
datum.isSavepoint) {
+ return alignment;
+ }
+ const timeout = config.aligned_checkpoint_timeout;
+ // 0 and Long.MAX_VALUE both mean unaligned from the first barrier
(CheckpointOptions.forConfig).
+ if (!(timeout > 0) || timeout >= Number.MAX_SAFE_INTEGER) {
+ return 0;
+ }
+ return Math.max(0, Math.min(alignment, timeout - startDelay));
+}
+
+// Rows by when they finished, latest first; ties keep operator then subtask
order (the sort is stable). Subtasks that
+// never finished come first: those without an ack, and aborted ones, whose
reported phases stop short of where they
+// were when the checkpoint failed (they are why it failed).
+export function slowestFirstOrder({ rows, unacked, instants, rowOrder }:
TimelineRows): string[] {
+ const end = new Map<string, number>();
+ for (const d of [...rows, ...instants]) {
+ end.set(d.row, d.aborted ? Infinity : Math.max(end.get(d.row) ??
-Infinity, d.range[1]));
+ }
+ for (const d of unacked) {
+ end.set(d.row, Infinity);
+ }
+ return [...rowOrder].sort((a, b) => {
+ const [ea, eb] = [end.get(a) ?? 0, end.get(b) ?? 0];
+ return ea === eb ? 0 : eb > ea ? 1 : -1;
+ });
+}
+
+// Backend tags completed subtasks "completed" (incl. aborted) and others
"pending_or_failed".
+function isCompletedSubtask(
+ s: SubTaskCheckpointStatisticsItem
+): s is SubTaskCheckpointStatisticsItem & CompletedSubTaskCheckpointStatistics
{
+ return s.status === 'completed';
+}
+
+function shortenOperatorName(name: string): string {
+ if (!name) {
+ return '';
+ }
+ const compact = name.replace(/\s+/g, ' ').trim();
+ return compact.length <= OPERATOR_NAME_MAX ? compact : `${compact.slice(0,
OPERATOR_NAME_MAX - 3)}...`;
+}
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/checkpoint-timeline.testing.ts
b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/checkpoint-timeline.testing.ts
new file mode 100644
index 00000000000..ab92d621718
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/checkpoint-timeline.testing.ts
@@ -0,0 +1,70 @@
+/*
+ * 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.
+ */
+
+// Test data shared by the timeline specs.
+
+import {
+ CheckpointHistory,
+ CheckpointSubTask,
+ SubTaskCheckpointStatisticsItem,
+ VerticesItem
+} from '@flink-runtime-web/interfaces';
+
+export const TRIGGERED = 1_000;
+
+export interface Phases {
+ start_delay?: number;
+ alignment?: number;
+ sync?: number;
+ async?: number;
+}
+
+export const completed = (index: number, phases: Phases, aborted = false):
SubTaskCheckpointStatisticsItem =>
+ ({
+ index,
+ status: 'completed',
+ ack_timestamp: 0,
+ end_to_end_duration: 0,
+ checkpointed_size: 0,
+ state_size: 0,
+ checkpoint: { sync: phases.sync ?? 0, async: phases.async ?? 0 },
+ alignment: { buffer: 0, processed: 0, persisted: 0, duration:
phases.alignment ?? 0 },
+ start_delay: phases.start_delay ?? 0,
+ unaligned_checkpoint: false,
+ aborted
+ }) as SubTaskCheckpointStatisticsItem;
+
+export const noStats = (index: number): SubTaskCheckpointStatisticsItem =>
+ ({ index, status: 'pending_or_failed' }) as SubTaskCheckpointStatisticsItem;
+
+export const details = (...subtasks: SubTaskCheckpointStatisticsItem[]):
CheckpointSubTask =>
+ ({ subtasks }) as unknown as CheckpointSubTask;
+
+export const vertex = (id: string, name: string): VerticesItem => ({ id, name,
status: 'RUNNING' }) as VerticesItem;
+
+export const historyEntry = (id: number, status: string, acked: number,
isSavepoint = false): CheckpointHistory =>
+ ({
+ id,
+ status,
+ is_savepoint: isSavepoint,
+ trigger_timestamp: TRIGGERED,
+ end_to_end_duration: 5,
+ num_acknowledged_subtasks: acked,
+ num_subtasks: 4,
+ state_size: 0
+ }) as CheckpointHistory;
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.html
b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.html
new file mode 100644
index 00000000000..2cab659262e
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.html
@@ -0,0 +1,202 @@
+<!--
+ ~ 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>
+ <nz-alert
+ *ngIf="statsRefreshFailed"
+ class="stale-stats"
+ nzType="warning"
+ nzShowIcon
+ [nzMessage]="staleStatsMessage"
+ ></nz-alert>
+ <ng-template #staleStatsMessage>
+ Could not refresh the checkpoint statistics from the JobManager; showing
the ones loaded
+ {{ statsUpdatedAt ? 'at ' + (statsUpdatedAt | date: 'HH:mm:ss') :
'earlier' }}. Retrying.
+ </ng-template>
+ <div class="strip-canvas" #stripContainer></div>
+ <nz-empty *ngIf="!hasHistory" nzNotFoundContent="No checkpoint history
yet"></nz-empty>
+ </section>
+
+ <section class="timeline-section">
+ <header class="timeline-header">
+ <div class="timeline-title">
+ <strong *ngIf="selected">
+ Checkpoint #{{ selected.id }}
+ <nz-tag
+ *ngIf="userPinned"
+ class="pin-badge"
+ nzColor="gold"
+ nz-tooltip
+ nzTooltipTitle="Pinned: the timeline below will not change until
you click another checkpoint or 'Follow newest'."
+ >
+ pinned
+ </nz-tag>
+ </strong>
+ <strong *ngIf="!selected">Per-subtask Timeline</strong>
+ <span *ngIf="selected" class="hint">
+ {{ selected.rawStatus }}
+ <ng-container *ngIf="selected.isSavepoint">· savepoint</ng-container>
+ · triggered {{ selected.triggered | date: 'yyyy-MM-dd HH:mm:ss.SSS'
}} ·
+ {{ selected.duration | humanizeDuration }} · acked {{ selected.acked
}} ·
+ {{ selected.size | humanizeBytes }}
+ </span>
+ <div class="timeline-actions">
+ <button
+ *ngIf="userPinned"
+ nz-button
+ nzSize="small"
+ type="button"
+ class="follow-newest"
+ (click)="followNewest()"
+ >
+ Follow newest
+ </button>
+ </div>
+ </div>
+ <ul class="legend">
+ <li>
+ <span class="swatch phase-start"></span>
+ start_delay
+ </li>
+ <li>
+ <span class="swatch phase-align"></span>
+ alignment
+ </li>
+ <li>
+ <span class="swatch phase-sync"></span>
+ sync
+ </li>
+ <li>
+ <span class="swatch phase-async"></span>
+ async
+ </li>
+ <li nz-tooltip [nzTooltipTitle]="abortedTip"
nzTooltipPlacement="bottom">
+ <span class="swatch aborted"></span>
+ aborted
+ </li>
+ <li nz-tooltip [nzTooltipTitle]="noAckTip" nzTooltipPlacement="bottom">
+ <span class="swatch no-ack"></span>
+ no ack
+ </li>
+ <ng-template #abortedTip>
+ Started its snapshot, but the checkpoint failed before it finished.
The bar shows the
+ partial work.
+ </ng-template>
+ <ng-template #noAckTip>
+ Reported no statistics for this checkpoint: still pending, or never
acknowledged before
+ the checkpoint failed (for example, the barrier never arrived). The
bar spans the whole
+ checkpoint.
+ </ng-template>
+ </ul>
+ </header>
+
+ <!-- Reserve the line while a checkpoint is selected so the chart doesn't
shift. -->
+ <div *ngIf="selected" class="timeline-note">
+ <span *ngIf="selectedRotatedOut && !timelineEmpty">
+ Checkpoint #{{ selected?.id }} is no longer in the recent-checkpoints
strip above — showing
+ the frozen snapshot from when it was pinned.
+ </span>
+ <span *ngIf="totalRows > 1 && !timelineEmpty" class="timeline-controls">
+ <!-- Text, not whitespace-only nodes, around the select: Angular drops
whitespace-only nodes. -->
+ <span *ngIf="slowestCountOptions.length > 0">
+ {{ shownRows < totalRows ? 'Showing the' : 'Showing all' }}
+ <nz-select
+ class="row-count"
+ nzSize="small"
+ aria-label="Number of slowest subtasks shown"
+ [ngModel]="shownRows"
+ (ngModelChange)="setSlowestCount($event)"
+ >
+ <nz-option
+ *ngFor="let n of slowestCountOptions"
+ [nzValue]="n"
+ [nzLabel]="'' + n"
+ ></nz-option>
+ </nz-select>
+ {{ shownRows < totalRows ? 'slowest of ' + totalRows + ' subtasks' :
'subtasks' }}
+ {{ scrollable ? '· rows ' + (firstRow + 1) + '–' + (firstRow +
visibleRows) : '' }}
+ </span>
+ <label
+ nz-checkbox
+ class="slowest-first"
+ [ngModel]="slowestFirst"
+ (ngModelChange)="setSlowestFirst($event)"
+ >
+ slowest first
+ </label>
+ </span>
+ </div>
+
+ <nz-alert
+ *ngIf="failedVertexCount > 0"
+ class="partial-failure"
+ nzType="warning"
+ nzShowIcon
+ [nzMessage]="partialFailureMessage"
+ ></nz-alert>
+
+ <div class="timeline-canvas-wrapper" [class.loading]="loadingTimeline"
#timelineWrapper>
+ <nz-spin [nzSpinning]="loadingTimeline">
+ <!-- With more rows than are visible, the chart stays in place while
this box scrolls. -->
+ <div
+ class="timeline-scroll"
+ [class.scrollable]="scrollable"
+ [style.height.px]="scrollable ? chartHeight : null"
+ (scroll)="onTimelineScroll()"
+ #timelineScroll
+ >
+ <div [style.height.px]="scrollable ? scrollContentHeight : null">
+ <!-- G2 sets an inline position on its container, so the sticky
goes on this wrapper. -->
+ <div class="timeline-sticky">
+ <div class="timeline-canvas" [class.hidden]="timelineEmpty"
#timelineContainer></div>
+ </div>
+ </div>
+ </div>
+ <nz-empty
+ *ngIf="timelineEmpty && !loadingTimeline"
+ [nzNotFoundContent]="emptyTimelineMessage"
+ ></nz-empty>
+ </nz-spin>
+ </div>
+ </section>
+</div>
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.less
b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.less
new file mode 100644
index 00000000000..9e2cadd53a8
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.less
@@ -0,0 +1,201 @@
+/*
+ * 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 "theme";
+
+// The theme has no neutral palette token for this one; keep in sync with
PHASE_COLOR in the .ts.
+@phase-start-color: #b8c0cc;
+
+.timeline-wrapper {
+ display: flex;
+ flex-direction: column;
+ gap: @margin-lg;
+}
+
+header {
+ display: flex;
+ flex-wrap: wrap;
+ gap: @padding-sm;
+ justify-content: space-between;
+ align-items: center;
+ margin-bottom: @margin-xs;
+}
+
+.hint {
+ margin-left: @margin-xs;
+ color: @text-color-secondary;
+ font-size: @font-size-sm;
+}
+
+.legend {
+ display: flex;
+ gap: @padding-sm;
+ margin: 0;
+ padding: 0;
+ list-style: none;
+ color: @text-color-secondary;
+ font-size: @font-size-sm;
+
+ li {
+ display: inline-flex;
+ align-items: center;
+ }
+
+ .swatch {
+ display: inline-block;
+ width: 10px;
+ height: 10px;
+ margin-right: 4px;
+ border: 1px solid transparent;
+ border-radius: 2px;
+ }
+
+ .completed {
+ background: @green-6;
+ }
+
+ .savepoint {
+ background: @purple-6;
+ }
+
+ .in-progress {
+ background: @gold-6;
+ }
+
+ .failed {
+ background: @red-6;
+ }
+
+ .phase-start {
+ background: @phase-start-color;
+ }
+
+ .phase-align {
+ background: @blue-3;
+ }
+
+ .phase-sync {
+ background: @orange-6;
+ }
+
+ .phase-async {
+ background: @green-6;
+ }
+
+ .aborted {
+ border: 1.5px dashed @red-6;
+ background: transparent;
+ }
+
+ .no-ack {
+ border: 1.5px dashed @red-6;
+ background: repeating-linear-gradient(-45deg, @red-1 0 3px, @red-3 3px
4.5px);
+ }
+}
+
+.stale-stats {
+ margin-bottom: @margin-xs;
+}
+
+.strip-canvas {
+ width: 100%;
+ // The chart's height, reserved before it first renders.
+ min-height: 160px;
+ cursor: pointer;
+}
+
+.pin-badge {
+ vertical-align: middle;
+ margin-left: @margin-xs;
+}
+
+.timeline-note {
+ display: flex;
+ gap: @padding-sm;
+ justify-content: space-between;
+ align-items: center;
+ // Fixed height (a small select's) so the chart doesn't shift when the note
toggles.
+ min-height: 24px;
+ margin: 0 0 @margin-xs;
+ color: @text-color-secondary;
+ font-size: @font-size-sm;
+
+ .timeline-controls {
+ display: inline-flex;
+ gap: @padding-sm;
+ align-items: center;
+ margin-left: auto;
+ white-space: nowrap;
+ }
+
+ .slowest-first {
+ color: @text-color-secondary;
+ font-size: @font-size-sm;
+ }
+
+ .row-count {
+ width: 72px;
+ margin: 0 4px;
+ }
+}
+
+.timeline-title {
+ display: flex;
+ flex: 1 1 auto;
+ flex-wrap: wrap;
+ gap: @margin-xs;
+ align-items: center;
+ min-width: 0;
+}
+
+.timeline-actions {
+ display: inline-flex;
+ gap: @margin-xs;
+ align-items: center;
+ margin-left: auto;
+}
+
+.partial-failure {
+ margin-bottom: @padding-sm;
+}
+
+.timeline-canvas-wrapper {
+ position: relative;
+ min-height: 160px;
+
+ &.loading {
+ opacity: 0.7;
+ }
+}
+
+.timeline-scroll.scrollable {
+ overflow-y: auto;
+}
+
+.timeline-sticky {
+ position: sticky;
+ top: 0;
+}
+
+.timeline-canvas {
+ width: 100%;
+
+ &.hidden {
+ display: none;
+ }
+}
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.spec.ts
b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.spec.ts
new file mode 100644
index 00000000000..c69b6a530d7
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.spec.ts
@@ -0,0 +1,672 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { HttpErrorResponse } from '@angular/common/http';
+import { ComponentFixture, TestBed } from '@angular/core/testing';
+import { NEVER, Observable, of, Subject, throwError } from 'rxjs';
+
+import { Checkpoint, CheckpointConfig, CheckpointHistory, CheckpointSubTask }
from '@flink-runtime-web/interfaces';
+import { JobService } from '@flink-runtime-web/services';
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+
+import { StripDatum } from './checkpoint-timeline-rows';
+import { completed, details, historyEntry, noStats, vertex } from
'./checkpoint-timeline.testing';
+import { JobCheckpointsTimelineComponent } from
'./job-checkpoints-timeline.component';
+
+// jsdom has no canvas, so g2 is swapped for a fake chart that records what
the component asks it to draw.
+const g2 = vi.hoisted(() => {
+ class FakeMark {
+ public readonly dataCalls: unknown[][] = [];
+ public readonly scales: Array<[string, { domain: unknown[] }]> = [];
+ public readonly interactions: Array<[string, unknown]> = [];
+
+ public encode(): this {
+ return this;
+ }
+ public axis(): this {
+ return this;
+ }
+ public legend(): this {
+ return this;
+ }
+ public animate(): this {
+ return this;
+ }
+ public style(): this {
+ return this;
+ }
+ public tooltip(): this {
+ return this;
+ }
+ public data(data: unknown[]): this {
+ this.dataCalls.push(data);
+ return this;
+ }
+ public scale(channel: string, options: { domain: unknown[] }): this {
+ this.scales.push([channel, options]);
+ return this;
+ }
+ public interaction(name: string, options: unknown): this {
+ this.interactions.push([name, options]);
+ return this;
+ }
+ }
+
+ class FakeChart {
+ public static readonly instances: FakeChart[] = [];
+ public readonly marks: FakeMark[] = [];
+ public readonly handlers = new Map<string, (event?: unknown) => void>();
+ public readonly emitted: Array<[string, unknown]> = [];
+ public readonly attrs = new Map<string, unknown>();
+ public readonly interactions: Array<[string, unknown]> = [];
+ public renders = 0;
+ public destroyed = false;
+
+ constructor(public readonly options: Record<string, unknown>) {
+ FakeChart.instances.push(this);
+ }
+
+ public interval(): FakeMark {
+ const mark = new FakeMark();
+ this.marks.push(mark);
+ return mark;
+ }
+ public coordinate(): this {
+ return this;
+ }
+ public interaction(name: string, options: unknown): this {
+ this.interactions.push([name, options]);
+ return this;
+ }
+ public on(event: string, handler: (event?: unknown) => void): this {
+ this.handlers.set(event, handler);
+ return this;
+ }
+ public emit(event: string, payload?: unknown): void {
+ this.emitted.push([event, payload]);
+ }
+ public attr(key: string, value: unknown): this {
+ this.attrs.set(key, value);
+ return this;
+ }
+ public render(): Promise<this> {
+ this.renders++;
+ return Promise.resolve(this);
+ }
+ public forceFit(): Promise<void> {
+ return Promise.resolve();
+ }
+ public destroy(): void {
+ this.destroyed = true;
+ }
+ }
+ return { FakeChart };
+});
+
+vi.mock('@antv/g2', () => ({ Chart: g2.FakeChart }));
+
+type FakeChart = InstanceType<typeof g2.FakeChart>;
+
+describe('JobCheckpointsTimelineComponent', () => {
+ let fixture: ComponentFixture<JobCheckpointsTimelineComponent>;
+ let element: HTMLElement;
+ let loadStats: ReturnType<typeof vi.fn>;
+ let loadDetails: ReturnType<typeof vi.fn>;
+ // The Timeline tab is shown when the strip has a width; jsdom lays nothing
out, so the width is faked.
+ let stripWidth: number;
+ let resizeCallbacks: Array<() => void>;
+ const serverError = new HttpErrorResponse({ status: 500 });
+
+ const strip = (): FakeChart => g2.FakeChart.instances[0];
+ const timeline = (): FakeChart | undefined =>
g2.FakeChart.instances.slice(1).filter(c => !c.destroyed)[0];
+ const timelineCharts = (): FakeChart[] => g2.FakeChart.instances.slice(1);
+ const stripData = (): StripDatum[] => strip().marks[0].dataCalls.at(-1) as
StripDatum[];
+ const stripBar = (id: number): StripDatum => stripData().find(d => d.id ===
id)!;
+ const detailsRequests = (): Array<[string, number]> =>
+ loadDetails.mock.calls.map(([jobId, id]) => [jobId as string, id as
number]);
+ const text = (): string => element.textContent!.replace(/\s+/g, ' ');
+
+ const setHistory = (history: CheckpointHistory[]): void => {
+ fixture.componentRef.setInput('history', history);
+ fixture.detectChanges();
+ };
+ const mount = (history: CheckpointHistory[], vertexIds = ['a'], config?:
CheckpointConfig): void => {
+ fixture.componentRef.setInput('jobId', 'job-1');
+ fixture.componentRef.setInput(
+ 'vertices',
+ vertexIds.map(id => vertex(id, id === 'a' ? 'Agg' : id === 's' ? 'Sink'
: id))
+ );
+ fixture.componentRef.setInput('config', config);
+ setHistory(history);
+ // ngAfterViewInit loads the newest checkpoint and starts polling, after
the first check of the view.
+ fixture.detectChanges();
+ };
+ const clickBar = (id: number): void => {
+ strip().handlers.get('interval:click')!({ data: { data: stripBar(id) } });
+ fixture.detectChanges();
+ };
+ const showTab = (): void => {
+ stripWidth = 800;
+ resizeCallbacks.forEach(callback => callback());
+ fixture.detectChanges();
+ };
+
+ beforeEach(async () => {
+ g2.FakeChart.instances.length = 0;
+ loadStats = vi.fn().mockReturnValue(NEVER);
+ loadDetails = vi.fn().mockReturnValue(of({ a: details(completed(0, {
async: 5 })) }));
+ stripWidth = 800;
+ resizeCallbacks = [];
+ vi.spyOn(HTMLElement.prototype, 'clientWidth',
'get').mockImplementation(() => stripWidth);
+ vi.stubGlobal(
+ 'ResizeObserver',
+ class {
+ constructor(callback: () => void) {
+ resizeCallbacks.push(callback);
+ }
+ public observe(): void {}
+ public disconnect(): void {}
+ }
+ );
+ vi.stubGlobal('requestAnimationFrame', (callback: FrameRequestCallback) =>
{
+ callback(0);
+ return 1;
+ });
+ await TestBed.configureTestingModule({
+ imports: [JobCheckpointsTimelineComponent],
+ providers: [
+ {
+ provide: JobService,
+ useValue: { loadCheckpointStats: loadStats,
loadCheckpointAllSubtaskDetails: loadDetails }
+ }
+ ]
+ }).compileComponents();
+ fixture = TestBed.createComponent(JobCheckpointsTimelineComponent);
+ element = fixture.nativeElement as HTMLElement;
+ });
+
+ afterEach(() => {
+ fixture.destroy();
+ vi.useRealTimers();
+ vi.unstubAllGlobals();
+ vi.restoreAllMocks();
+ });
+
+ describe('strip', () => {
+ it('draws the recent checkpoints oldest first in a 160px chart', () => {
+ mount([historyEntry(9, 'COMPLETED', 4), historyEntry(8, 'FAILED', 2),
historyEntry(7, 'COMPLETED', 4)]);
+
+ expect(strip().options['height']).toBe(160);
+ expect(stripData().map(d => [d.id, d.status])).toEqual([
+ [7, 'COMPLETED'],
+ [8, 'FAILED'],
+ [9, 'COMPLETED']
+ ]);
+ expect(text()).not.toContain('No checkpoint history yet');
+ });
+
+ it('says when there is no checkpoint history', () => {
+ mount([]);
+
+ expect(text()).toContain('No checkpoint history yet');
+ expect(text()).toContain('Per-subtask Timeline');
+ });
+
+ it('redraws only when the bars or the pin change', () => {
+ mount([historyEntry(7, 'COMPLETED', 4)]);
+ const renders = strip().renders;
+
+ setHistory([historyEntry(7, 'COMPLETED', 4)]);
+ expect(strip().renders).toBe(renders);
+ clickBar(7);
+ expect(strip().renders).toBe(renders + 1);
+ setHistory([historyEntry(8, 'IN_PROGRESS', 0), historyEntry(7,
'COMPLETED', 4)]);
+ expect(strip().renders).toBe(renders + 2);
+ // The bars moved, so an open tooltip would point at the wrong one.
+ expect(strip().emitted.at(-1)).toEqual(['tooltip:hide', undefined]);
+ });
+
+ it('shows the tooltip only while the pointer is over a bar', () => {
+ mount([historyEntry(7, 'COMPLETED', 4)]);
+ const pointer = (event: string, payload?: unknown): void =>
strip().handlers.get(event)!(payload);
+
+ expect(strip().marks[0].interactions).toContainEqual(['tooltip', {
disableNative: true, wait: 0 }]);
+ strip().emitted.length = 0;
+ pointer('interval:pointermove', { data: { data: stripBar(7) } });
+ pointer('plot:pointermove', { target: { className: 'element' } });
+ pointer('interval:pointerout');
+ // Over the plot but off the bars, e.g. after a re-render moved them.
+ pointer('plot:pointermove', { target: { className: 'plot' } });
+
+ expect(strip().emitted).toEqual([
+ ['tooltip:show', { data: { data: stripBar(7) } }],
+ ['tooltip:hide', undefined],
+ ['tooltip:hide', undefined]
+ ]);
+ });
+ });
+
+ describe('selection', () => {
+ it('follows the newest checkpoint', () => {
+ mount([historyEntry(8, 'COMPLETED', 4), historyEntry(7, 'COMPLETED',
4)]);
+
+ expect(text()).toContain('Checkpoint #8');
+ expect(element.querySelector('.pin-badge')).toBeNull();
+ expect(detailsRequests()).toEqual([['job-1', 8]]);
+ });
+
+ it('pins a clicked checkpoint until Follow newest is clicked', () => {
+ mount([historyEntry(8, 'COMPLETED', 4), historyEntry(7, 'COMPLETED',
4)]);
+
+ clickBar(7);
+ expect(text()).toContain('Checkpoint #7');
+
expect(element.querySelector('nz-tag.pin-badge')?.textContent?.trim()).toBe('pinned');
+ setHistory([historyEntry(9, 'COMPLETED', 4), historyEntry(8,
'COMPLETED', 4), historyEntry(7, 'COMPLETED', 4)]);
+ expect(text()).toContain('Checkpoint #7');
+
+ const followNewest =
element.querySelector<HTMLButtonElement>('button.follow-newest')!;
+ expect(followNewest.classList).toContain('ant-btn');
+ followNewest.click();
+ fixture.detectChanges();
+
+ expect(text()).toContain('Checkpoint #9');
+ expect(element.querySelector('.pin-badge')).toBeNull();
+ expect(element.querySelector('button.follow-newest')).toBeNull();
+ });
+
+ it('explains when the pinned checkpoint leaves the strip', () => {
+ mount([historyEntry(7, 'COMPLETED', 4)]);
+ clickBar(7);
+
+ setHistory([historyEntry(9, 'COMPLETED', 4), historyEntry(8,
'COMPLETED', 4)]);
+
+ expect(text()).toContain('Checkpoint #7 is no longer in the
recent-checkpoints strip above');
+ });
+
+ it('follows the newest checkpoint again once the history empties while
pinned', () => {
+ mount([historyEntry(7, 'COMPLETED', 4)]);
+ clickBar(7);
+
+ setHistory([]);
+ expect(text()).toContain('Per-subtask Timeline');
+ setHistory([historyEntry(1, 'COMPLETED', 4)]);
+
+ expect(text()).toContain('Checkpoint #1');
+ expect(element.querySelector('.pin-badge')).toBeNull();
+ });
+ });
+
+ describe('reloads', () => {
+ it('reloads the followed checkpoint once it completes, without waiting for
a newer one', () => {
+ mount([historyEntry(7, 'IN_PROGRESS', 2)]);
+ setHistory([historyEntry(7, 'COMPLETED', 4)]);
+
+ expect(detailsRequests()).toEqual([
+ ['job-1', 7],
+ ['job-1', 7]
+ ]);
+ expect(text()).toContain('acked 4/4');
+ });
+
+ it('updates one timeline chart across reloads instead of rebuilding it',
() => {
+ mount([historyEntry(7, 'IN_PROGRESS', 2)]);
+ setHistory([historyEntry(7, 'COMPLETED', 4)]);
+
+ expect(timelineCharts().length).toBe(1);
+ expect([timeline()!.renders, timeline()!.destroyed]).toEqual([2, false]);
+ });
+
+ it('does not reload when nothing changed or while a load is in flight', ()
=> {
+ mount([historyEntry(7, 'IN_PROGRESS', 2)]);
+ setHistory([historyEntry(7, 'IN_PROGRESS', 2)]);
+ loadDetails.mockReturnValue(NEVER);
+ setHistory([historyEntry(7, 'IN_PROGRESS', 3)]);
+ setHistory([historyEntry(7, 'IN_PROGRESS', 4)]);
+
+ expect(detailsRequests().length).toBe(2);
+ });
+
+ it('keeps refreshing a pinned in-progress checkpoint until it finishes',
() => {
+ mount([historyEntry(7, 'IN_PROGRESS', 2)]);
+ clickBar(7);
+ setHistory([historyEntry(8, 'IN_PROGRESS', 0), historyEntry(7,
'IN_PROGRESS', 3)]);
+ setHistory([historyEntry(8, 'IN_PROGRESS', 1), historyEntry(7,
'COMPLETED', 4)]);
+ setHistory([historyEntry(8, 'COMPLETED', 4), historyEntry(7,
'COMPLETED', 4)]);
+
+ expect(detailsRequests().map(([, id]) => id)).toEqual([7, 7, 7, 7]);
+ expect(text()).toContain('Checkpoint #7');
+ expect(text()).toContain('acked 4/4');
+ expect(element.querySelector('.pin-badge')).not.toBeNull();
+ });
+
+ it('re-selects a finished checkpoint from the cache, and fetches an
in-progress savepoint again', () => {
+ mount([historyEntry(8, 'IN_PROGRESS', 2, true), historyEntry(7,
'COMPLETED', 4, true)]);
+ clickBar(7);
+ clickBar(8);
+ clickBar(7);
+ clickBar(8);
+
+ expect(detailsRequests().map(([, id]) => id)).toEqual([8, 7, 8, 8]);
+ });
+
+ it('keeps cached details only for checkpoints still in the strip, plus the
pinned one', () => {
+ // The cache isn't visible in the page, so this one reads it.
+ const cached = (): number[] =>
+ [...(fixture.componentInstance as unknown as { detailsCache:
Map<number, unknown> }).detailsCache.keys()].sort(
+ (a, b) => a - b
+ );
+ mount([historyEntry(7, 'COMPLETED', 4)]);
+ clickBar(7);
+ setHistory([historyEntry(9, 'COMPLETED', 4), historyEntry(8,
'COMPLETED', 4)]);
+ expect(cached()).toEqual([7]);
+
+
element.querySelector<HTMLButtonElement>('button.follow-newest')!.click();
+ setHistory([historyEntry(10, 'COMPLETED', 4), historyEntry(9,
'COMPLETED', 4)]);
+ expect(cached()).toEqual([9, 10]);
+ });
+ });
+
+ describe('loading', () => {
+ it('does not fetch subtask details until the Timeline tab is shown', () =>
{
+ stripWidth = 0;
+ mount([historyEntry(7, 'COMPLETED', 4)]);
+ expect(loadDetails).not.toHaveBeenCalled();
+ expect(loadStats).not.toHaveBeenCalled();
+
+ showTab();
+
+ expect(detailsRequests()).toEqual([['job-1', 7]]);
+ expect(loadStats).toHaveBeenCalledWith('job-1');
+ expect(timeline()).toBeDefined();
+ });
+
+ it('says the JobManager no longer retains the checkpoint when every vertex
returns 404', () => {
+ loadDetails.mockReturnValue(of({ a: null }));
+ mount([historyEntry(7, 'COMPLETED', 4)]);
+
+ expect(text()).toContain('the JobManager no longer retains this
checkpoint');
+ expect(timeline()).toBeUndefined();
+ });
+
+ it('reports other request errors as a failed request, and retries once
rather than on every poll', () => {
+ loadDetails.mockReturnValue(throwError(() => serverError));
+ mount([historyEntry(7, 'COMPLETED', 4)]);
+ expect(text()).toContain('the request to the JobManager failed');
+
+ for (let poll = 0; poll < 3; poll++) {
+ setHistory([historyEntry(7, 'COMPLETED', 4)]);
+ }
+
+ expect(loadDetails).toHaveBeenCalledTimes(2);
+ });
+
+ it('draws a partial load with a warning, and does not cache it', () => {
+ loadDetails.mockReturnValue(of({ a: details(completed(0, { async: 5 })),
b: null }));
+ mount([historyEntry(7, 'COMPLETED', 4)], ['a', 'b']);
+
+ expect(text()).toContain('1 of 2 operators failed to load — the timeline
below is incomplete.');
+ expect(timeline()!.marks[0].dataCalls.at(-1)).toHaveLength(1);
+ clickBar(7);
+ expect(loadDetails).toHaveBeenCalledTimes(2);
+ });
+
+ it('drops a load still in flight when the job changes', () => {
+ const response = new Subject<Record<string, CheckpointSubTask | null>>();
+ loadDetails.mockReturnValue(response);
+ mount([historyEntry(42, 'COMPLETED', 4)]);
+
+ fixture.componentRef.setInput('jobId', 'job-2');
+ fixture.detectChanges();
+ response.next({ a: details(completed(0, { async: 5 })) });
+ expect(timeline()).toBeUndefined();
+
+ loadDetails.mockReturnValue(of({ a: details(completed(0, { async: 5 }))
}));
+ setHistory([historyEntry(42, 'COMPLETED', 4)]);
+ expect(detailsRequests()).toEqual([
+ ['job-1', 42],
+ ['job-2', 42]
+ ]);
+ });
+ });
+
+ describe('polling', () => {
+ const stats = (history: CheckpointHistory[]): Observable<Checkpoint> =>
of({ history } as Checkpoint);
+
+ it('shows polled history before the parent passes any', () => {
+ loadStats.mockReturnValue(stats([historyEntry(7, 'COMPLETED', 4)]));
+ mount([]);
+
+ expect(text()).not.toContain('No checkpoint history yet');
+ expect(text()).toContain('Checkpoint #7');
+ });
+
+ it('drops a stats poll for the previous job that lands after the job
changes', () => {
+ const oldJobStats = new Subject<Checkpoint>();
+ loadStats.mockReturnValue(oldJobStats);
+ mount([]);
+
+ fixture.componentRef.setInput('jobId', 'job-2');
+ fixture.detectChanges();
+ oldJobStats.next({ history: [historyEntry(42, 'COMPLETED', 4)] } as
Checkpoint);
+ fixture.detectChanges();
+
+ expect(loadStats).toHaveBeenCalledWith('job-1');
+ expect(text()).toContain('No checkpoint history yet');
+ expect(loadDetails).not.toHaveBeenCalled();
+ });
+
+ it.each([
+ [undefined, 15_000],
+ [0, 15_000],
+ [100, 1_500],
+ [5_000, 5_000],
+ [60_000, 15_000],
+ // Long.MAX_VALUE: periodic checkpointing disabled.
+ [2 ** 63, 15_000]
+ ])('polls a job with a %s ms checkpoint interval every %s ms', (interval,
delay) => {
+ vi.useFakeTimers();
+ loadStats.mockReturnValue(stats([historyEntry(7, 'COMPLETED', 4)]));
+ mount([], ['a'], { interval } as CheckpointConfig);
+ expect(loadStats).toHaveBeenCalledTimes(1);
+
+ vi.advanceTimersByTime(delay - 1);
+ expect(loadStats).toHaveBeenCalledTimes(1);
+ vi.advanceTimersByTime(1);
+ expect(loadStats).toHaveBeenCalledTimes(2);
+ });
+
+ it('keeps the last stats on a failed poll, marked as stale, and keeps
polling', () => {
+ vi.useFakeTimers();
+ vi.setSystemTime(new Date(2026, 0, 1, 10, 20, 30));
+ loadStats
+ .mockReturnValueOnce(throwError(() => serverError))
+ .mockReturnValue(stats([historyEntry(8, 'COMPLETED', 4),
historyEntry(7, 'COMPLETED', 4)]));
+ mount([historyEntry(7, 'COMPLETED', 4)]);
+
+ expect(text()).toContain('Checkpoint #7');
+
expect(element.querySelector('nz-alert.stale-stats')?.textContent).toContain(
+ 'Could not refresh the checkpoint statistics from the JobManager;
showing the ones loaded at 10:20:30. Retrying.'
+ );
+
+ vi.advanceTimersByTime(15_000);
+ fixture.detectChanges();
+
+ expect(element.querySelector('nz-alert.stale-stats')).toBeNull();
+ expect(text()).toContain('Checkpoint #8');
+ });
+ });
+
+ describe('row window', () => {
+ let nextId = 1;
+ const phases = (): FakeChart['marks'][number] => timeline()!.marks[0];
+ const lastScale = (channel: string): { domain: unknown[] } =>
+ phases()
+ .scales.filter(([c]) => c === channel)
+ .at(-1)![1];
+ const shownRows = (): unknown[] => lastScale('x').domain;
+ const scrollBox = (): HTMLDivElement =>
element.querySelector<HTMLDivElement>('.timeline-scroll')!;
+ // A checkpoint of one operator where a higher subtask index finishes
later: the slowest 25 of 300 are #275–#299.
+ const showSubtasks = (count: number): void => {
+ loadDetails.mockReturnValue(
+ of({ a: details(...Array.from({ length: count }, (_, i) =>
completed(i, { async: 5 + i }))) })
+ );
+ setHistory([historyEntry(nextId++, 'COMPLETED', 4)]);
+ };
+ const scrollTo = (row: number): void => {
+ Object.defineProperty(scrollBox(), 'scrollTop', { value: row * 22,
writable: true, configurable: true });
+ scrollBox().dispatchEvent(new Event('scroll'));
+ fixture.detectChanges();
+ };
+ const toggleSlowestFirst = (): void => {
+ element.querySelector<HTMLLabelElement>('label.slowest-first')!.click();
+ fixture.detectChanges();
+ };
+ const pickSlowest = (count: number): void => {
+ fixture.componentInstance.setSlowestCount(count);
+ fixture.detectChanges();
+ };
+
+ beforeEach(() => mount([]));
+
+ it('shows the 25 slowest subtasks in operator order, and offers up to 200
or the row count', async () => {
+ showSubtasks(300);
+ // ngModel writes the select's value asynchronously.
+ await fixture.whenStable();
+ fixture.detectChanges();
+
+ expect(shownRows()).toEqual(Array.from({ length: 25 }, (_, i) => `Agg
#${275 + i}`));
+ expect(new Set((phases().dataCalls.at(-1) as Array<{ row: string
}>).map(r => r.row)).size).toBe(25);
+ expect(timeline()!.attrs.get('height')).toBe(25 * 22 + 56);
+ expect(text()).toContain('Showing the 25 slowest of 300 subtasks');
+ expect(element.querySelector('nz-select.row-count')).not.toBeNull();
+ expect(scrollBox().classList).not.toContain('scrollable');
+ expect(fixture.componentInstance.slowestCountOptions).toEqual([25, 50,
75, 100, 125, 150, 175, 200]);
+
+ showSubtasks(60);
+ expect(fixture.componentInstance.slowestCountOptions).toEqual([25, 50,
60]);
+
+ showSubtasks(20);
+ expect(element.querySelector('nz-select.row-count')).toBeNull();
+ expect(timeline()!.attrs.get('height')).toBe(20 * 22 + 56);
+ });
+
+ it('scrolls through more of the slowest than fit, without moving the time
scale', () => {
+ showSubtasks(300);
+ const timeDomain = lastScale('y').domain;
+
+ pickSlowest(100);
+ expect(scrollBox().classList).toContain('scrollable');
+ expect(scrollBox().style.height).toBe(`${25 * 22 + 56}px`);
+ expect([shownRows()[0], text().includes('rows 1–25')]).toEqual(['Agg
#200', true]);
+ scrollTo(30);
+
+ expect([shownRows()[0], shownRows().length, text().includes('rows
31–55')]).toEqual(['Agg #230', 25, true]);
+ expect(lastScale('y').domain).toEqual(timeDomain);
+ });
+
+ it('keeps the window within the picked rows, and starts from the top for
another count', () => {
+ showSubtasks(300);
+ pickSlowest(100);
+ scrollTo(90);
+ expect(text()).toContain('rows 76–100');
+
+ pickSlowest(50);
+
+ expect([scrollBox().scrollTop, shownRows()[0]]).toEqual([0, 'Agg #250']);
+ expect(text()).toContain('rows 1–25');
+ });
+
+ it('sorts the picked rows by when they finished, latest first, and starts
again from the top', () => {
+ showSubtasks(300);
+ pickSlowest(50);
+ scrollTo(10);
+
+ toggleSlowestFirst();
+
+ expect(scrollBox().scrollTop).toBe(0);
+ expect(shownRows().slice(0, 3)).toEqual(['Agg #299', 'Agg #298', 'Agg
#297']);
+ toggleSlowestFirst();
+ expect(shownRows()[0]).toBe('Agg #250');
+ });
+
+ it('puts subtasks without an ack first when sorting, keeping operator
order for ties', () => {
+ fixture.componentRef.setInput('vertices', [vertex('a', 'Agg'),
vertex('s', 'Sink')]);
+ loadDetails.mockReturnValue(
+ of({
+ a: details(completed(0, { async: 10 }), noStats(1), completed(2, {
async: 30 })),
+ s: details(noStats(0), completed(1, { async: 30 }))
+ })
+ );
+ setHistory([historyEntry(nextId++, 'FAILED', 3)]);
+
+ toggleSlowestFirst();
+
+ expect(shownRows()).toEqual(['Agg #1', 'Sink #0', 'Agg #2', 'Sink #1',
'Agg #0']);
+ });
+
+ it('counts aborted subtasks as the slowest, however short their reported
phases', () => {
+ // A failed checkpoint: 30 subtasks finished, #30 was aborted after
reporting 1ms, #31 never acked.
+ loadDetails.mockReturnValue(
+ of({
+ a: details(
+ ...Array.from({ length: 30 }, (_, i) => completed(i, { async: 5 +
i })),
+ completed(30, { async: 1 }, true),
+ noStats(31)
+ )
+ })
+ );
+ setHistory([historyEntry(nextId++, 'FAILED', 30)]);
+
+ expect(shownRows()).toContain('Agg #30');
+ expect(shownRows()).toContain('Agg #31');
+ toggleSlowestFirst();
+ expect(shownRows().slice(0, 3)).toEqual(['Agg #30', 'Agg #31', 'Agg
#29']);
+ });
+ });
+
+ it('highlights the hovered row across the timeline', () => {
+ mount([historyEntry(7, 'COMPLETED', 4)]);
+
+ expect(timeline()!.interactions).toContainEqual(['elementHighlightByX',
true]);
+ });
+
+ it('shows a timeline tooltip only over a bar, next to the pointer, and hides
it when the rows are redrawn', () => {
+ mount([historyEntry(7, 'COMPLETED', 4)]);
+ const chart = timeline()!;
+ const pointer = (event: string, payload?: unknown): void =>
chart.handlers.get(event)!(payload);
+ const bar = { phase: 'async' };
+ const plot = { className: 'plot', getBBox: () => ({ x: 50, y: 20 }) };
+
+ expect(chart.interactions).toContainEqual(['tooltip',
expect.objectContaining({ disableNative: true, wait: 0 })]);
+ chart.emitted.length = 0;
+ pointer('interval:pointermove', {
+ data: { data: bar },
+ canvas: { x: 80, y: 30 },
+ target: { className: 'element', parentNode: { className: 'group',
parentNode: plot } }
+ });
+ pointer('plot:pointermove', { target: { className: 'plot' } });
+ clickBar(7);
+
+ expect(chart.emitted).toEqual([
+ // g2 takes the position relative to the plot area.
+ ['tooltip:show', { offsetX: 30, offsetY: 10, data: { data: bar } }],
+ ['tooltip:hide', undefined],
+ ['tooltip:hide', undefined]
+ ]);
+ });
+});
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.ts
b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.ts
new file mode 100644
index 00000000000..7f0ce9e54e3
--- /dev/null
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job/checkpoints/timeline/job-checkpoints-timeline.component.ts
@@ -0,0 +1,913 @@
+/*
+ * 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 { FormsModule } from '@angular/forms';
+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 {
+ Checkpoint,
+ CheckpointConfig,
+ CheckpointHistory,
+ CheckpointSubTask,
+ 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 { NzButtonModule } from 'ng-zorro-antd/button';
+import { NzCheckboxModule } from 'ng-zorro-antd/checkbox';
+import { NzEmptyModule } from 'ng-zorro-antd/empty';
+import { NzSelectModule } from 'ng-zorro-antd/select';
+import { NzSpinModule } from 'ng-zorro-antd/spin';
+import { NzTagModule } from 'ng-zorro-antd/tag';
+import { NzTooltipModule } from 'ng-zorro-antd/tooltip';
+
+import {
+ buildTimelineRows,
+ InstantDatum,
+ PhaseKey,
+ slowestFirstOrder,
+ StripDatum,
+ TimelineDatum,
+ TimelineRows,
+ toStripDatum,
+ UnackedDatum
+} from './checkpoint-timeline-rows';
+
+// '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;
+}
+
+// The legend swatches in the .less file use the theme's palette tokens for
the same colors.
+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: '#91d5ff',
+ 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';
+
+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,
+ FormsModule,
+ HumanizeDurationPipe,
+ HumanizeBytesPipe,
+ NzAlertModule,
+ NzButtonModule,
+ NzCheckboxModule,
+ NzEmptyModule,
+ NzSelectModule,
+ NzSpinModule,
+ NzTagModule,
+ 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;
+ // When the shown checkpoint stats were last fetched, and whether the latest
poll for them failed.
+ public statsUpdatedAt?: number;
+ public statsRefreshFailed = false;
+
+ // Diverges from @Input once polling starts feeding live updates.
+ private currentHistory: CheckpointHistory[] = [];
+ public userPinned = false;
+ private stripChart?: Chart;
+ private stripInterval?: IntervalMark;
+ // What the strip last drew; polls that change nothing skip the redraw,
which would drop an open tooltip.
+ private stripDrawn?: string;
+ 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 STRIP_HEIGHT = 160;
+ 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(toStripDatum(this.currentHistory[0]), false);
+ }
+ this.startPolling();
+ // G2 locks autoFit at construction; re-fit + resume polling once width is
real.
+ this.visible$.next(this.stripContainer.nativeElement.clientWidth > 0);
+ if (typeof ResizeObserver !== 'undefined') {
+ this.resizeObserver = new ResizeObserver(() => {
+ const stripVisible = this.stripContainer.nativeElement.clientWidth > 0;
+ if (this.stripChart && stripVisible) {
+ void this.stripChart.forceFit();
+ }
+ if (this.timeline && this.timelineContainer.nativeElement.clientWidth
> 0) {
+ void this.timeline.chart.forceFit();
+ }
+ this.visible$.next(stripVisible);
+ });
+ this.resizeObserver.observe(this.stripContainer.nativeElement);
+ this.resizeObserver.observe(this.timelineContainer.nativeElement);
+ }
+ }
+
+ public ngOnChanges(changes: SimpleChanges): void {
+ if (!this.stripContainer || !this.stripContainer.nativeElement) {
+ return;
+ }
+ if (changes['jobId'] && !changes['jobId'].firstChange) {
+ this.userPinned = false;
+ this.selected = undefined;
+ this.currentHistory = [];
+ this.statsUpdatedAt = undefined;
+ this.statsRefreshFailed = false;
+ // Otherwise a load still in flight for the old job lands in the cache
under an id the new job reuses.
+ this.loadTimeline$.next(null);
+ this.detailsCache.clear();
+ this.retriedCheckpointId = undefined;
+ this.clearTimeline();
+ this.renderStrip();
+ }
+ if (changes['history']) {
+ this.markStatsFresh();
+ this.applyHistory(this.history || []);
+ }
+ if (changes['vertices'] && !changes['vertices'].firstChange &&
this.selected) {
+ this.loadTimeline(this.selected);
+ }
+ }
+
+ public ngOnDestroy(): void {
+ this.destroy$.next();
+ this.destroy$.complete();
+ this.resizeObserver?.disconnect();
+ if (this.scrollFrame !== undefined) {
+ cancelAnimationFrame(this.scrollFrame);
+ }
+ this.stripChart?.destroy();
+ this.timeline?.chart.destroy();
+ }
+
+ private renderStrip(): void {
+ const data = (this.currentHistory || [])
+ .slice(0, JobCheckpointsTimelineComponent.STRIP_MAX_BARS)
+ .map(toStripDatum)
+ .reverse();
+
+ if (!this.stripChart || !this.stripInterval) {
+ this.stripChart = new Chart({
+ container: this.stripContainer.nativeElement,
+ autoFit: true,
+ height: JobCheckpointsTimelineComponent.STRIP_HEIGHT,
+ paddingTop: 16,
+ paddingRight: 16,
+ paddingBottom: 36,
+ paddingLeft: 56
+ });
+ const chart = this.stripChart;
+ const isPinned = (d: StripDatum): boolean => this.userPinned &&
this.selected?.id === d.id;
+ this.stripInterval = chart
+ .interval()
+ .encode('x', 'idLabel')
+ .encode('y', 'duration')
+ .encode('color', 'status')
+ .scale('y', { nice: true, domainMin: 0 })
+ .scale('color', { domain: STRIP_STATUSES, range: STRIP_STATUSES.map(s
=> STRIP_COLOR[s]) })
+ .axis('x', { title: false, labelAutoHide: true, labelAutoRotate: false
})
+ .axis('y', { title: false })
+ .legend(false)
+ .animate(false)
+ .style('maxWidth', 18)
+ // Keeps 0ms bars (e.g. in progress before the first ack) and very
short ones visible and clickable.
+ .style('minHeight', 3)
+ .style('stroke', (d: StripDatum) => (isPinned(d) ? '#1890ff' :
'transparent'))
+ .style('lineWidth', (d: StripDatum) => (isPinned(d) ? 2.5 : 0))
+ .tooltip({
+ title: '',
+ items: [
+ (d: StripDatum) => ({
+ name: `#${d.idLabel}`,
+ color: STRIP_COLOR[d.status],
+ value: `${d.rawStatus} ·
${this.humanizeDurationPipe.transform(d.duration)} · acked ${d.acked}`
+ })
+ ]
+ })
+ .interaction('tooltip', BARS_ONLY_TOOLTIP);
+ showTooltipOverBarsOnly(chart);
+ chart.on('interval:click', (ev: { data?: { data?: StripDatum } }) => {
+ const datum = ev.data?.data;
+ if (datum) {
+ this.select(datum);
+ }
+ });
+ }
+
+ const drawn = JSON.stringify([this.userPinned ? this.selected?.id : null,
data]);
+ if (drawn === this.stripDrawn) {
+ return;
+ }
+ this.stripDrawn = drawn;
+ // The bars move under the pointer; the next move over one shows its
tooltip again.
+ this.stripChart.emit('tooltip:hide');
+ 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(toStripDatum(this.currentHistory[0]), false);
+ } else {
+ this.cdr.markForCheck();
+ }
+ }
+
+ private selectInternal(datum: StripDatum, pinned: boolean): void {
+ this.selected = datum;
+ if (pinned) {
+ this.userPinned = true;
+ }
+ this.cdr.markForCheck();
+ this.renderStrip();
+ this.loadTimeline(datum);
+ }
+
+ private applyHistory(next: CheckpointHistory[]): void {
+ const prevNewestId = this.currentHistory.length > 0 ?
this.currentHistory[0].id : -1;
+ const nextNewestId = next.length > 0 ? next[0].id : -1;
+ this.currentHistory = next;
+ this.pruneDetailsCache();
+ this.renderStrip();
+ // The empty state and the rotated-out note read currentHistory.
+ this.cdr.markForCheck();
+
+ if (next.length === 0) {
+ // The pinned checkpoint is gone too (e.g. after a JobManager failover),
so follow whatever comes next.
+ this.userPinned = false;
+ this.selected = undefined;
+ this.clearTimeline();
+ return;
+ }
+
+ if (!this.userPinned && nextNewestId !== prevNewestId) {
+ this.selectInternal(toStripDatum(next[0]), false);
+ return;
+ }
+ // Still showing the same checkpoint (the followed newest one, or the
pinned one): reload once it gained acks or
+ // finished, or retry a failed load once (retrying on every poll would
repeat the error notifications). Not while
+ // a load is in flight, so polling doesn't keep cancelling a slow fetch.
+ const selected = this.selected;
+ const latest = this.userPinned ? next.find(h => h.id === selected?.id) :
next[0];
+ if (!selected || !latest || latest.id !== selected.id ||
this.loadingTimeline) {
+ return;
+ }
+ const current = toStripDatum(latest);
+ const progressed = current.rawStatus !== selected.rawStatus ||
current.acked !== selected.acked;
+ const retry =
+ (this.loadError !== undefined || this.failedVertexCount > 0) &&
this.retriedCheckpointId !== current.id;
+ if (retry) {
+ this.retriedCheckpointId = current.id;
+ }
+ if (progressed || retry) {
+ this.selectInternal(current, false);
+ }
+ }
+
+ private pruneDetailsCache(): void {
+ const selectable = new Set(
+ this.currentHistory.slice(0,
JobCheckpointsTimelineComponent.STRIP_MAX_BARS).map(h => h.id)
+ );
+ for (const id of this.detailsCache.keys()) {
+ if (!selectable.has(id) && id !== this.selected?.id) {
+ this.detailsCache.delete(id);
+ }
+ }
+ }
+
+ private startPolling(): void {
+ this.visible$
+ .pipe(
+ distinctUntilChanged(),
+ switchMap(visible =>
+ visible
+ ? defer((): Observable<Checkpoint | 'failed'> => {
+ const jobId = this.jobId;
+ return jobId
+ ? this.jobService.loadCheckpointStats(jobId).pipe(
+ catchError(() => of('failed' as const)),
+ // A response for the previous job can land after a
jobId change has reset the state.
+ filter(() => jobId === this.jobId)
+ )
+ : EMPTY;
+ }).pipe(repeat({ delay: () => timer(this.computePollDelayMs())
}))
+ : EMPTY
+ ),
+ takeUntil(this.destroy$)
+ )
+ .subscribe(stats => {
+ if (stats === 'failed') {
+ // Keep showing the last stats, marked as stale; the next poll tries
again.
+ this.statsRefreshFailed = true;
+ this.cdr.markForCheck();
+ return;
+ }
+ this.markStatsFresh();
+ this.applyHistory(stats.history || []);
+ });
+ }
+
+ private markStatsFresh(): void {
+ this.statsUpdatedAt = Date.now();
+ this.statsRefreshFailed = false;
+ }
+
+ private computePollDelayMs(): number {
+ // Disabled periodic checkpointing reports Long.MAX_VALUE, which the upper
clamp covers.
+ const interval = this.config?.interval ?? 0;
+ return interval > 0
+ ? Math.max(
+ JobCheckpointsTimelineComponent.POLL_MIN_MS,
+ Math.min(interval, JobCheckpointsTimelineComponent.POLL_MAX_MS)
+ )
+ : JobCheckpointsTimelineComponent.POLL_MAX_MS;
+ }
+
+ private loadTimeline(datum: StripDatum): void {
+ // The component also lives in the hidden tab; don't fetch every vertex's
details until it is shown.
+ if (!this.visible$.value) {
+ this.loadDeferred = true;
+ return;
+ }
+ this.loadDeferred = false;
+ if (!this.jobId || this.vertices.length === 0) {
+ this.showEmptyTimeline();
+ return;
+ }
+ this.loadingTimeline = true;
+ this.timelineEmpty = false;
+ this.loadError = undefined;
+ this.failedVertexCount = 0;
+ this.cdr.markForCheck();
+ this.loadTimeline$.next(datum);
+ }
+
+ private cacheDetails(datum: StripDatum, perVertex: Record<string,
CheckpointSubTask | null>): void {
+ // In-progress stats still change; only cache terminal checkpoints.
`status` is 'SAVEPOINT' for an in-progress
+ // savepoint, so check rawStatus.
+ if (datum.rawStatus !== 'IN_PROGRESS') {
+ this.detailsCache.set(datum.id, perVertex);
+ }
+ }
+
+ private subscribeTimelineLoads(): void {
+ this.visible$
+ .pipe(
+ filter(visible => visible && this.loadDeferred),
+ takeUntil(this.destroy$)
+ )
+ .subscribe(() => {
+ if (this.selected) {
+ this.loadTimeline(this.selected);
+ }
+ });
+
+ this.loadTimeline$
+ .pipe(
+ switchMap((datum): Observable<TimelineLoad> => {
+ if (!datum) {
+ return EMPTY;
+ }
+ const requestedCount = this.vertices.length;
+ const cached = this.detailsCache.get(datum.id);
+ if (cached) {
+ return of({ datum, perVertex: cached, failedCount: 0,
requestedCount });
+ }
+ const vertexIds = this.vertices.map(v => v.id);
+ return this.jobService.loadCheckpointAllSubtaskDetails(this.jobId,
datum.id, vertexIds).pipe(
+ map(perVertex => {
+ const failedCount = requestedCount -
Object.values(perVertex).filter(v => v !== null).length;
+ if (failedCount === 0) {
+ this.cacheDetails(datum, perVertex);
+ }
+ const error: LoadError | undefined =
+ requestedCount > 0 && failedCount === requestedCount ?
'not-retained' : undefined;
+ return { datum, perVertex, failedCount, requestedCount, error };
+ }),
+ catchError(() =>
+ of<TimelineLoad>({
+ datum,
+ perVertex: {},
+ failedCount: requestedCount,
+ requestedCount,
+ error: 'request-failed'
+ })
+ )
+ );
+ }),
+ takeUntil(this.destroy$)
+ )
+ .subscribe(({ datum, perVertex, failedCount, requestedCount, error }) =>
{
+ this.loadingTimeline = false;
+ if (error) {
+ this.showEmptyTimeline(error);
+ return;
+ }
+ this.failedVertexCount = failedCount;
+ this.requestedVertexCount = requestedCount;
+ this.renderTimeline(datum, perVertex);
+ });
+ }
+
+ private renderTimeline(datum: StripDatum, perVertex: Record<string,
CheckpointSubTask | null>): void {
+ const built = buildTimelineRows(datum, this.vertices, perVertex,
this.config);
+ if (built.rowOrder.length === 0) {
+ this.showEmptyTimeline();
+ return;
+ }
+ this.timelineEmpty = false;
+ this.timelineRows = { datum, ...built, slowestOrder:
slowestFirstOrder(built) };
+ this.totalRows = built.rowOrder.length;
+ this.rowTitles = new Map();
+ for (const row of [...built.rows, ...built.unacked, ...built.instants]) {
+ if (!this.rowTitles.has(row.row)) {
+ this.rowTitles.set(row.row, `${row.operatorFull} #${row.subtask}`);
+ }
+ }
+ this.drawTimelineWindow();
+ }
+
+ // Draws the rows scrolled to, out of the slowest ones picked. The chart is
at most 25 rows tall and stays in place
+ // while the box around it scrolls, so its canvas stays far below the 65,535
px browsers allow, however big the job.
+ private drawTimelineWindow(): void {
+ const data = this.timelineRows;
+ if (!data) {
+ return;
+ }
+ const { datum, rows, unacked, instants, rowOrder } = data;
+ const C = JobCheckpointsTimelineComponent;
+ const slowest = data.slowestOrder.slice(0, this.shownRows);
+ const picked = new Set(slowest);
+ const order = this.slowestFirst ? slowest : rowOrder.filter(row =>
picked.has(row));
+ const visible = this.visibleRows;
+ this.firstRow = Math.max(0, Math.min(this.firstRow, order.length -
visible));
+ const windowRows = order.slice(this.firstRow, this.firstRow + visible);
+ const inWindow = new Set(windowRows);
+ const shown = <T extends { row: string }>(items: T[]): T[] =>
items.filter(d => inWindow.has(d.row));
+ this.chartHeight = Math.max(visible * C.TIMELINE_ROW_HEIGHT +
C.TIMELINE_HEIGHT_BASE, C.TIMELINE_HEIGHT_MIN);
+
+ // The label padding and the time scale cover all rows, so scrolling or
picking more rows doesn't shift them.
+ const longestLabel = rowOrder.reduce((m, s) => Math.max(m, s.length), 0);
+ const leftPadding = Math.min(
+ C.LABEL_PADDING_MAX,
+ Math.max(C.LABEL_PADDING_MIN, longestLabel * C.LABEL_CHAR_WIDTH +
C.LABEL_PADDING_BUFFER)
+ );
+ const scaleMin = datum.triggered;
+ const scaleMax = Math.max(
+ [...rows, ...unacked].reduce((m, r) => Math.max(m, r.range[1]),
scaleMin),
+ scaleMin + 1
+ );
+ const xScale = { domain: windowRows };
+ const yScale = {
+ domain: [scaleMin, scaleMax],
+ nice: false,
+ tickMethod: () => startAlignedTicks(new Date(scaleMin), new
Date(scaleMax)).map(Number)
+ };
+
+ const timeline = (this.timeline ??= this.createTimelineChart());
+ timeline.chart.attr('height', this.chartHeight);
+ timeline.chart.attr('paddingLeft', leftPadding);
+ timeline.phases.data(shown(rows)).scale('x', xScale).scale('y', yScale);
+ timeline.unacked.data(shown(unacked)).scale('x', xScale).scale('y',
yScale);
+ timeline.instants.data(shown(instants)).scale('x', xScale).scale('y',
yScale);
+ // The rows move under the pointer; the next move over a bar shows its
tooltip again.
+ timeline.chart.emit('tooltip:hide');
+ void timeline.chart.render();
+ this.cdr.markForCheck();
+ }
+
+ // How many rows are drawn: the slowest ones picked, or all rows if there
are fewer.
+ public get shownRows(): number {
+ return Math.min(this.slowestCount, this.totalRows);
+ }
+
+ public get visibleRows(): number {
+ return Math.min(JobCheckpointsTimelineComponent.VISIBLE_ROWS,
this.shownRows);
+ }
+
+ public get scrollable(): boolean {
+ return this.shownRows > this.visibleRows;
+ }
+
+ // The scroll box's content: the chart plus one row height for every shown
row out of view.
+ public get scrollContentHeight(): number {
+ return this.chartHeight + (this.shownRows - this.visibleRows) *
JobCheckpointsTimelineComponent.TIMELINE_ROW_HEIGHT;
+ }
+
+ // 25, 50, ... up to 200 or the number of rows, whichever is smaller; none
when every row is already shown.
+ public get slowestCountOptions(): number[] {
+ const C = JobCheckpointsTimelineComponent;
+ const max = Math.min(C.SHOWN_ROWS_MAX, this.totalRows);
+ if (max <= C.SHOWN_ROWS_DEFAULT) {
+ return [];
+ }
+ const options: number[] = [];
+ for (let n = C.SHOWN_ROWS_DEFAULT; n < max; n += C.SHOWN_ROWS_DEFAULT) {
+ options.push(n);
+ }
+ options.push(max);
+ return options;
+ }
+
+ public setSlowestCount(count: number): void {
+ this.slowestCount = count;
+ this.scrollToTop();
+ }
+
+ public setSlowestFirst(slowestFirst: boolean): void {
+ this.slowestFirst = slowestFirst;
+ this.scrollToTop();
+ }
+
+ // Other rows or another order: start again from the top.
+ private scrollToTop(): void {
+ this.firstRow = 0;
+ this.timelineScroll.nativeElement.scrollTop = 0;
+ this.drawTimelineWindow();
+ }
+
+ public onTimelineScroll(): void {
+ const first = Math.round(
+ this.timelineScroll.nativeElement.scrollTop /
JobCheckpointsTimelineComponent.TIMELINE_ROW_HEIGHT
+ );
+ if (first === this.firstRow) {
+ return;
+ }
+ this.firstRow = first;
+ // One redraw per frame, however many scroll events arrive.
+ if (this.scrollFrame === undefined) {
+ this.scrollFrame = requestAnimationFrame(() => {
+ this.scrollFrame = undefined;
+ this.drawTimelineWindow();
+ });
+ }
+ }
+
+ private createTimelineChart(): TimelineChart {
+ const container = this.timelineContainer.nativeElement;
+ const chart = new Chart({ container, autoFit: true, paddingTop: 32,
paddingRight: 32, paddingBottom: 16 });
+ chart.coordinate({ transform: [{ type: 'transpose' }] });
+ const phases = chart
+ .interval()
+ .encode('x', 'row')
+ .encode('y', 'range')
+ .encode('color', 'phase')
+ .scale('color', { domain: PHASES, range: PHASES.map(p => PHASE_COLOR[p])
})
+ .axis('x', {
+ title: false,
+ labelAutoHide: false,
+ labelAutoRotate: false,
+ labelFill: '#262626',
+ labelOpacity: 1
+ })
+ // On top, so the time axis is in view as soon as the chart is.
+ .axis('y', { title: false, position: 'top', labelFormatter: (v: number)
=> formatTimeAxis(new Date(Number(v))) })
+ .legend(false)
+ .animate(false)
+ .style('stroke', (d: TimelineDatum) => (d.aborted ? NOT_COMPLETED_STROKE
: 'transparent'))
+ .style('lineWidth', (d: TimelineDatum) => (d.aborted ? 1.5 : 0))
+ .style('lineDash', (d: TimelineDatum) => (d.aborted ? [3, 2] : [0, 0]))
+ .tooltip({
+ title: (d: TimelineDatum) => `${d.operatorFull} #${d.subtask}`,
+ items: [
+ (d: TimelineDatum) => ({ name: d.phaseLabel, color:
PHASE_COLOR[d.phase], value: `${d.durationMs}ms` }),
+ (d: TimelineDatum) => ({ name: 'total', color: 'transparent', value:
`${d.totalDuration}ms` }),
+ (d: TimelineDatum) => ({
+ name: 'state',
+ color: 'transparent',
+ value: this.humanizeBytesPipe.transform(d.stateSize)
+ }),
+ (d: TimelineDatum) => ({
+ name: 'checkpointed',
+ color: 'transparent',
+ value: this.humanizeBytesPipe.transform(d.checkpointedSize)
+ })
+ ]
+ });
+ // On the chart rather than on one mark: the other marks' defaults would
override it.
+ chart.interaction('tooltip', {
+ ...BARS_ONLY_TOOLTIP,
+ // Outside the scroll box, which would clip it on the last visible rows.
g2 places a tooltip before measuring
+ // it, so on its first frame it doesn't flip up yet.
+ mount: this.timelineWrapper.nativeElement,
+ // g2 5 ellipsizes the tooltip title at 360px; let full operator names
wrap instead.
+ css: {
+ '.g2-tooltip': { 'max-width': '480px' },
+ '.g2-tooltip-title': { 'white-space': 'normal', 'overflow-wrap':
'anywhere' }
+ }
+ });
+ showTooltipOverBarsOnly(chart);
+ // Returned from a callback: g2 deep-copies constant style objects, which
would turn the canvas into a plain object.
+ const hatchFill = { image: (this.noAckHatch ??= createHatchPattern()),
repetition: 'repeat' };
+ const unacked = chart
+ .interval()
+ .encode('x', 'row')
+ .encode('y', 'range')
+ .animate(false)
+ .style('fill', () => hatchFill)
+ .style('stroke', NOT_COMPLETED_STROKE)
+ .style('lineWidth', 1.5)
+ .style('lineDash', [3, 2])
+ .tooltip({
+ title: (d: UnackedDatum) => `${d.operatorFull} #${d.subtask}`,
+ items: [(d: UnackedDatum) => ({ name: d.label, color:
NOT_COMPLETED_STROKE, value: 'no stats reported' })]
+ });
+ const instants = chart
+ .interval()
+ .encode('x', 'row')
+ .encode('y', 'range')
+ .animate(false)
+ // The range is zero-length; once transposed, minHeight is the on-screen
length along the time axis.
+ .style('minHeight', 2)
+ .style('fill', (d: InstantDatum) => (d.aborted ? NOT_COMPLETED_STROKE :
INSTANT_FILL))
+ .tooltip({
+ title: (d: InstantDatum) => `${d.operatorFull} #${d.subtask}`,
+ // Same rows as the phase bars, so 'total' always means the sum of the
phases.
+ items: [
+ (d: InstantDatum) => ({
+ name: 'all phases',
+ color: d.aborted ? NOT_COMPLETED_STROKE : INSTANT_FILL,
+ value: '0ms'
+ }),
+ () => ({ name: 'total', color: 'transparent', value: '0ms' }),
+ (d: InstantDatum) => ({
+ name: 'state',
+ color: 'transparent',
+ value: this.humanizeBytesPipe.transform(d.stateSize)
+ }),
+ (d: InstantDatum) => ({
+ name: 'checkpointed',
+ color: 'transparent',
+ value: this.humanizeBytesPipe.transform(d.checkpointedSize)
+ })
+ ]
+ });
+
+ // Row labels are truncated; show the full operator name as a native
tooltip while hovering one.
+ chart.on('axis-label-item:pointerover', (ev: { target?: { attributes?: {
text?: unknown } } }) => {
+ container.title = this.rowTitles.get(String(ev.target?.attributes?.text
?? '')) ?? '';
+ });
+ chart.on('axis-label-item:pointerout', () => (container.title = ''));
+ // Fades the other rows while hovering one, to follow a row across the
chart.
+ chart.interaction('elementHighlightByX', true);
+ return { chart, phases, unacked, instants };
+ }
+
+ private clearTimeline(): void {
+ this.timeline?.chart.destroy();
+ this.timeline = undefined;
+ this.timelineRows = undefined;
+ this.totalRows = 0;
+ this.firstRow = 0;
+ this.timelineContainer.nativeElement.title = '';
+ this.loadingTimeline = false;
+ this.timelineEmpty = false;
+ this.loadError = undefined;
+ this.failedVertexCount = 0;
+ this.requestedVertexCount = 0;
+ }
+
+ private showEmptyTimeline(error?: LoadError): void {
+ this.clearTimeline();
+ this.timelineEmpty = true;
+ this.loadError = error;
+ this.cdr.markForCheck();
+ }
+
+ public get emptyTimelineMessage(): string {
+ switch (this.loadError) {
+ case 'not-retained':
+ return 'Could not load subtask data — the JobManager no longer retains
this checkpoint';
+ case 'request-failed':
+ return 'Could not load subtask data — the request to the JobManager
failed';
+ default:
+ return 'No subtask data for this checkpoint';
+ }
+ }
+
+ // The `history` input is the parent's snapshot; polling only updates
currentHistory.
+ public get hasHistory(): boolean {
+ return this.currentHistory.length > 0;
+ }
+
+ public get selectedRotatedOut(): boolean {
+ return this.userPinned && !!this.selected && !this.currentHistory.some(h
=> h.id === this.selected!.id);
+ }
+
+ public get partialFailureMessage(): string {
+ const operators = this.requestedVertexCount === 1 ? 'operator' :
'operators';
+ return `${this.failedVertexCount} of ${this.requestedVertexCount}
${operators} failed to load — the timeline below is incomplete.`;
+ }
+}
+
+// g2 points a bar chart's tooltip at the bar nearest the pointer along the
band axis, so it also shows over the empty
+// space next to short bars. With this, a chart shows it only while the
pointer is over a bar. No throttling, which
+// would drop the show for the bar the pointer ends on.
+const BARS_ONLY_TOOLTIP = { disableNative: true, wait: 0 };
+
+interface G2Node {
+ className?: string;
+ parentNode?: G2Node | null;
+ getBBox?: () => { x: number; y: number };
+}
+
+interface BarPointerEvent {
+ data?: { data?: object };
+ canvas?: { x: number; y: number };
+ target?: G2Node;
+}
+
+function showTooltipOverBarsOnly(chart: Chart): void {
+ const hide = (): void => void chart.emit('tooltip:hide');
+ // On every move, not just on entering a bar: a redraw drops the tooltip.
+ chart.on('interval:pointermove', (ev: BarPointerEvent) => {
+ const datum = ev.data?.data;
+ if (!datum) {
+ return;
+ }
+ // At the pointer, as g2 would place it; g2 takes the position relative to
the plot area.
+ let plot = ev.target;
+ while (plot && plot.className !== 'plot') {
+ plot = plot.parentNode ?? undefined;
+ }
+ const origin = plot?.getBBox?.();
+ const at = origin && ev.canvas ? { offsetX: ev.canvas.x - origin.x,
offsetY: ev.canvas.y - origin.y } : {};
+ chart.emit('tooltip:show', { ...at, data: { data: datum } });
+ });
+ chart.on('interval:pointerout', hide);
+ // A redraw can also move the bars away from a still pointer, with no
pointerout.
+ chart.on('plot:pointermove', (ev: BarPointerEvent) => {
+ if (ev.target?.className !== 'element') {
+ hide();
+ }
+ });
+}
+
+// Diagonal hatch for unacked rows, used by g2 as a repeating pattern fill.
+function createHatchPattern(): HTMLCanvasElement {
+ const canvas = document.createElement('canvas');
+ canvas.width = 8;
+ canvas.height = 8;
+ const ctx = canvas.getContext('2d');
+ if (ctx) {
+ ctx.fillStyle = NO_ACK_FILL;
+ ctx.fillRect(0, 0, 8, 8);
+ ctx.strokeStyle = NO_ACK_STRIPE;
+ ctx.lineWidth = 1.5;
+ ctx.beginPath();
+ // Main diagonal plus the two corner stubs so the stripes stay continuous
when tiled.
+ ctx.moveTo(0, 8);
+ ctx.lineTo(8, 0);
+ ctx.moveTo(-2, 2);
+ ctx.lineTo(2, -2);
+ ctx.moveTo(6, 10);
+ ctx.lineTo(10, 6);
+ ctx.stroke();
+ }
+ return canvas;
+}
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job/timeline/job-timeline.component.spec.ts
b/flink-runtime-web/web-dashboard/src/app/pages/job/timeline/job-timeline.component.spec.ts
index fb801c71748..9e07ec614b4 100644
---
a/flink-runtime-web/web-dashboard/src/app/pages/job/timeline/job-timeline.component.spec.ts
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job/timeline/job-timeline.component.spec.ts
@@ -23,7 +23,7 @@ import type { Chart } from '@antv/g2';
import { JobService, ConfigService } from '@flink-runtime-web/services';
import { beforeEach, describe, expect, it, vi } from 'vitest';
-import { JobTimelineComponent, startAlignedTicks } from
'./job-timeline.component';
+import { JobTimelineComponent } from './job-timeline.component';
import { JobLocalService } from '../job-local.service';
const mockJobDetail = {
@@ -130,21 +130,3 @@ describe('JobTimelineComponent', () => {
expect(fakeSub.render).toHaveBeenCalled();
});
});
-
-describe('startAlignedTicks', () => {
- const DAY_MS = 86_400_000;
-
- it('anchors ticks at the domain minimum', () => {
- const min = new Date(1_000);
- const ticks = startAlignedTicks(min, new Date(1_000 + 60_000));
- expect(ticks[0]).toEqual(min);
- expect(ticks.length).toBeLessThanOrEqual(8);
- });
-
- it('keeps the tick count bounded for ranges beyond the largest step', () => {
- const min = new Date(0);
- const ticks = startAlignedTicks(min, new Date(30 * DAY_MS));
- expect(ticks[0]).toEqual(min);
- expect(ticks.length).toBeLessThanOrEqual(8);
- });
-});
diff --git
a/flink-runtime-web/web-dashboard/src/app/pages/job/timeline/job-timeline.component.ts
b/flink-runtime-web/web-dashboard/src/app/pages/job/timeline/job-timeline.component.ts
index 7fce3648ef1..cbec5e4ba11 100644
---
a/flink-runtime-web/web-dashboard/src/app/pages/job/timeline/job-timeline.component.ts
+++
b/flink-runtime-web/web-dashboard/src/app/pages/job/timeline/job-timeline.component.ts
@@ -32,33 +32,11 @@ import { distinctUntilChanged, filter, takeUntil } from
'rxjs/operators';
import { Chart } from '@antv/g2';
import { JobDetailCorrect, VerticesItemRange } from
'@flink-runtime-web/interfaces';
import { JobService, ColorKey, ConfigService } from
'@flink-runtime-web/services';
-import { timeFormat } from 'd3-time-format';
+import { formatTimeAxis, startAlignedTicks } from '@flink-runtime-web/utils';
import { NzDividerModule } from 'ng-zorro-antd/divider';
import { JobLocalService } from '../job-local.service';
-// g2 5's string labelFormatter uses d3's numeric formatter, so a time scale
needs
-// a function; mirror the previous 'HH:mm:ss.SSS' mask.
-const formatTimeAxis = timeFormat('%H:%M:%S.%L');
-
-const TICK_STEPS_MS = [
- 1, 2, 5, 10, 15, 30, 60, 120, 180, 300, 600, 900, 1800, 3600, 7200, 10800,
21600, 43200, 86400
-].map(s => s * 1000);
-
-const MAX_TICKS = 7;
-
-// Unlike g2's default "nice" ticks, start at the domain minimum so the axis
is anchored to the actual start time.
-// Beyond the largest step, spread the ticks evenly instead so the axis never
gets one tick per day.
-export function startAlignedTicks(min: Date, max: Date): Date[] {
- const range = Number(max) - Number(min);
- const step = TICK_STEPS_MS.find(s => range / s <= MAX_TICKS) ??
Math.ceil(range / MAX_TICKS);
- const ticks: Date[] = [];
- for (let t = Number(min); t <= Number(max) && step > 0; t += step) {
- ticks.push(new Date(t));
- }
- return ticks;
-}
-
@Component({
selector: 'flink-job-timeline',
templateUrl: './job-timeline.component.html',
diff --git
a/flink-runtime-web/web-dashboard/src/app/services/job.service.spec.ts
b/flink-runtime-web/web-dashboard/src/app/services/job.service.spec.ts
index 9da4661ed15..7fc9e5a27e9 100644
--- a/flink-runtime-web/web-dashboard/src/app/services/job.service.spec.ts
+++ b/flink-runtime-web/web-dashboard/src/app/services/job.service.spec.ts
@@ -16,14 +16,15 @@
* limitations under the License.
*/
-import { HttpClient } from '@angular/common/http';
+import { HttpClient, HttpContext, HttpErrorResponse } from
'@angular/common/http';
import { firstValueFrom, of, throwError, toArray } from 'rxjs';
-import { JobDetail, JobOverview } from '@flink-runtime-web/interfaces';
+import { CheckpointSubTask, JobDetail, JobOverview } from
'@flink-runtime-web/interfaces';
import { JobResourceRequirements } from
'@flink-runtime-web/interfaces/job-resource-requirements';
import { beforeEach, describe, expect, it, vi } from 'vitest';
import { ConfigService } from './config.service';
+import { EXPECTED_NOT_FOUND } from './http-context';
import { JobService } from './job.service';
describe('JobService', () => {
@@ -133,6 +134,42 @@ describe('JobService', () => {
});
});
+ describe('loadCheckpointAllSubtaskDetails', () => {
+ const subtasksOf = (vid: string): string =>
`./jobs/job-1/checkpoints/details/7/subtasks/${vid}`;
+
+ it('maps a 404 to null for that vertex and keeps the other vertices',
async () => {
+ const detail = { id: 7 } as unknown as CheckpointSubTask;
+ httpClient.get.mockImplementation((url: string) =>
+ url === subtasksOf('v1') ? of(detail) : throwError(() => new
HttpErrorResponse({ status: 404 }))
+ );
+
+ const result = await
firstValueFrom(service.loadCheckpointAllSubtaskDetails('job-1', 7, ['v1',
'v2']));
+
+ expect(result).toEqual({ v1: detail, v2: null });
+ });
+
+ it('fails the whole load on any other error', async () => {
+ const error = new HttpErrorResponse({ status: 500 });
+ httpClient.get.mockImplementation((url: string) => (url ===
subtasksOf('v1') ? of({}) : throwError(() => error)));
+
+ await
expect(firstValueFrom(service.loadCheckpointAllSubtaskDetails('job-1', 7,
['v1', 'v2']))).rejects.toBe(
+ error
+ );
+ });
+
+ it('marks its requests as expecting a 404, unlike the single-vertex
default', () => {
+ httpClient.get.mockReturnValue(of({}));
+
+ service.loadCheckpointAllSubtaskDetails('job-1', 7, ['v1',
'v2']).subscribe();
+ service.loadCheckpointSubtaskDetails('job-1', 7, 'v3');
+
+ const expectsNotFound = httpClient.get.mock.calls.map(([, options]) =>
+ (options as { context: HttpContext }).context.get(EXPECTED_NOT_FOUND)
+ );
+ expect(expectsNotFound).toEqual([true, true, false]);
+ });
+ });
+
describe('URL construction', () => {
it('cancels a job through the yarn-cancel handler', async () => {
httpClient.get.mockReturnValue(of(undefined));
diff --git a/flink-runtime-web/web-dashboard/src/app/services/job.service.ts
b/flink-runtime-web/web-dashboard/src/app/services/job.service.ts
index 2e2ca5d46ea..0ed64bfb736 100644
--- a/flink-runtime-web/web-dashboard/src/app/services/job.service.ts
+++ b/flink-runtime-web/web-dashboard/src/app/services/job.service.ts
@@ -16,10 +16,10 @@
* limitations under the License.
*/
-import { HttpClient } from '@angular/common/http';
+import { HttpClient, HttpContext, HttpErrorResponse, HttpStatusCode } from
'@angular/common/http';
import { Injectable } from '@angular/core';
-import { EMPTY, forkJoin, mergeMap, Observable } from 'rxjs';
-import { catchError, map } from 'rxjs/operators';
+import { EMPTY, forkJoin, from, mergeMap, Observable, of, throwError } from
'rxjs';
+import { catchError, map, reduce } from 'rxjs/operators';
import {
Checkpoint,
@@ -53,11 +53,14 @@ import {
import { JobResourceRequirements } from
'@flink-runtime-web/interfaces/job-resource-requirements';
import { ConfigService } from './config.service';
+import { EXPECTED_NOT_FOUND } from './http-context';
@Injectable({
providedIn: 'root'
})
export class JobService {
+ private static readonly SUBTASK_DETAILS_CONCURRENCY = 6;
+
constructor(
private readonly httpClient: HttpClient,
private readonly configService: ConfigService
@@ -180,10 +183,36 @@ export class JobService {
public loadCheckpointSubtaskDetails(
jobId: string,
checkPointId: number,
- vertexId: string
+ vertexId: string,
+ expectNotFound = false
): Observable<CheckpointSubTask> {
return this.httpClient.get<CheckpointSubTask>(
-
`${this.configService.BASE_URL}/jobs/${jobId}/checkpoints/details/${checkPointId}/subtasks/${vertexId}`
+
`${this.configService.BASE_URL}/jobs/${jobId}/checkpoints/details/${checkPointId}/subtasks/${vertexId}`,
+ { context: new HttpContext().set(EXPECTED_NOT_FOUND, expectNotFound) }
+ );
+ }
+
+ public loadCheckpointAllSubtaskDetails(
+ jobId: string,
+ checkPointId: number,
+ vertexIds: string[]
+ ): Observable<Record<string, CheckpointSubTask | null>> {
+ // A 404 means the checkpoint has dropped out of the JobManager's stats
history; that vertex maps to null.
+ // Any other error fails the whole load. Bounded so jobs with many
vertices don't send every request at once.
+ return from(vertexIds).pipe(
+ mergeMap(
+ vid =>
+ this.loadCheckpointSubtaskDetails(jobId, checkPointId, vid,
true).pipe(
+ catchError((err: unknown) =>
+ err instanceof HttpErrorResponse && err.status ===
HttpStatusCode.NotFound
+ ? of<CheckpointSubTask | null>(null)
+ : throwError(() => err)
+ ),
+ map(result => ({ [vid]: result }))
+ ),
+ JobService.SUBTASK_DETAILS_CONCURRENCY
+ ),
+ reduce((acc, part) => Object.assign(acc, part), {} as Record<string,
CheckpointSubTask | null>)
);
}
diff --git a/flink-runtime-web/web-dashboard/src/app/utils/public-api.ts
b/flink-runtime-web/web-dashboard/src/app/utils/public-api.ts
index 328d9f1adef..7bcbea555b0 100644
--- a/flink-runtime-web/web-dashboard/src/app/utils/public-api.ts
+++ b/flink-runtime-web/web-dashboard/src/app/utils/public-api.ts
@@ -18,3 +18,4 @@
export * from './is-nil';
export * from './strong-type';
+export * from './time-axis';
diff --git a/flink-runtime-web/web-dashboard/src/app/utils/time-axis.spec.ts
b/flink-runtime-web/web-dashboard/src/app/utils/time-axis.spec.ts
new file mode 100644
index 00000000000..58d65f599cd
--- /dev/null
+++ b/flink-runtime-web/web-dashboard/src/app/utils/time-axis.spec.ts
@@ -0,0 +1,43 @@
+/*
+ * 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 { describe, expect, it } from 'vitest';
+
+import { startAlignedTicks } from './time-axis';
+
+describe('startAlignedTicks', () => {
+ const DAY_MS = 86_400_000;
+
+ it('anchors ticks at the domain minimum', () => {
+ const min = new Date(1_000);
+ const ticks = startAlignedTicks(min, new Date(1_000 + 60_000));
+ expect(ticks[0]).toEqual(min);
+ expect(ticks.length).toBeLessThanOrEqual(8);
+ });
+
+ it('uses sub-second steps for short ranges', () => {
+ expect(startAlignedTicks(new Date(0), new
Date(300)).map(Number)).toEqual([0, 50, 100, 150, 200, 250, 300]);
+ });
+
+ it('keeps the tick count bounded for ranges beyond the largest step', () => {
+ const min = new Date(0);
+ const ticks = startAlignedTicks(min, new Date(30 * DAY_MS));
+ expect(ticks[0]).toEqual(min);
+ expect(ticks.length).toBeLessThanOrEqual(8);
+ });
+});
diff --git a/flink-runtime-web/web-dashboard/src/app/utils/time-axis.ts
b/flink-runtime-web/web-dashboard/src/app/utils/time-axis.ts
new file mode 100644
index 00000000000..8b54cce5faa
--- /dev/null
+++ b/flink-runtime-web/web-dashboard/src/app/utils/time-axis.ts
@@ -0,0 +1,42 @@
+/*
+ * 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 { timeFormat } from 'd3-time-format';
+
+// g2 5's string labelFormatter uses d3's numeric formatter, so a time scale
needs
+// a function; mirror the previous 'HH:mm:ss.SSS' mask.
+export const formatTimeAxis = timeFormat('%H:%M:%S.%L');
+
+const TICK_STEPS_MS = [
+ ...[1, 2, 5, 10, 20, 50, 100, 200, 500],
+ ...[1, 2, 5, 10, 15, 30, 60, 120, 180, 300, 600, 900, 1800, 3600, 7200,
10800, 21600, 43200, 86400].map(s => s * 1000)
+];
+
+const MAX_TICKS = 7;
+
+// Unlike g2's default "nice" ticks, start at the domain minimum so the axis
is anchored to the actual start time.
+// Beyond the largest step, spread the ticks evenly instead so the axis never
gets one tick per day.
+export function startAlignedTicks(min: Date, max: Date): Date[] {
+ const range = Number(max) - Number(min);
+ const step = TICK_STEPS_MS.find(s => range / s <= MAX_TICKS) ??
Math.ceil(range / MAX_TICKS);
+ const ticks: Date[] = [];
+ for (let t = Number(min); t <= Number(max) && step > 0; t += step) {
+ ticks.push(new Date(t));
+ }
+ return ticks;
+}
