[
https://issues.apache.org/jira/browse/FLINK-39489?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
featzhang updated FLINK-39489:
------------------------------
Description:
h2. Summary
Add a "Top N" tab to the job detail page of the Flink Web UI, backed by a
new REST endpoint {{GET /jobs/:jobid/metrics/top-n}}. The tab surfaces, for
the currently open job, the worst offenders across five operator/JVM health
dimensions in a single view, so operators can triage a misbehaving job
without clicking through every vertex.
This ticket is the implementation umbrella for FLIP-XXX. A prototype PR is
already open for early feedback; it is *not* requesting merge before the
FLIP is accepted.
h2. Motivation
When a Flink job misbehaves (lag, slow checkpoints, instability), the first
question an operator asks is _"which subtask / which TaskManager should I
be looking at?"_. Today the Web UI answers this only indirectly:
* per-subtask backpressure and busy time are on the vertex page,
* FLIP-33 source-lag metrics are on the source-vertex page,
* JVM CPU and GC are on the TaskManager page.
Each of these is already exposed by the runtime - the gap is that they are
not surfaced in one place ranked by severity. For jobs with many vertices
and high parallelism, operators today click through the UI vertex by
vertex, which is slow and error-prone.
h2. Goal
Expose one new REST endpoint and one new UI tab that, for the current job,
show top-N ranked lists across five dimensions grouped into two tiers.
h3. Primary triage (per-subtask / per-vertex, directly actionable)
|| # || Dimension || Source metric(s) || Notes ||
| 1 | Top Backpressured Subtasks | {{backPressuredTimeMsPerSecond}} | exposed
as a ratio in {{[0, 1]}} |
| 2 | Top Busy Subtasks | {{busyTimeMsPerSecond}} | exposed as a ratio in {{[0,
1]}} |
| 3 | Top Lagging Sources | {{pendingRecords}}, {{currentFetchEventTimeLag}},
{{currentEmitEventTimeLag}} (FLIP-33) | backlog summed across subtasks; lag
metrics taken as max |
h3. JVM diagnostics (TaskManager-scoped, coarser)
|| # || Dimension || Source metric(s) || Notes ||
| 4 | Top CPU Consumers | {{Status.JVM.CPU.Load}} | TaskManager-scoped |
| 5 | Top GC-intensive TaskManagers | {{Status.JVM.GarbageCollector.*.Time}} |
summed across collectors per TM |
Dimensions 4-5 are *TaskManager-scoped*, not per-job, because CPU and GC
are JVM-process metrics and per-job attribution requires a slot-to-TM
mapping that the metric store does not currently expose. This limitation
is surfaced in the UI (the JVM tier is visually demoted and collapsed by
default).
h2. Non-goals
* (x) No new {{ConfigOption}}.
* (x) No new metric emitted by the runtime - everything is derived from
metrics that already exist.
* (x) No per-job attribution of CPU / GC (future work, tracked separately).
* (x) No Top N for checkpoints in this ticket; its data source is
{{CheckpointStatsSnapshot}}, not {{MetricStore}}, and it fits better
as a follow-up.
* (x) No persistence of "top-N over time" - the endpoint is a live
snapshot.
* (x) No changes to any existing REST endpoint, response body, or metric.
All wire additions are purely additive.
h2. Proposed changes
h3. Backend ({{flink-runtime}})
* New handler {{TopNMetricsHandler}} on
{{GET /jobs/:jobid/metrics/top-n?top=N}} (default {{top=5}}, bounded to
a reasonable cap). Reads from the existing {{MetricStore}}.
* New response body {{TopNMetricsResponseBody}} with five *additive*
lists: {{topBackpressuredSubtasks}}, {{topBusySubtasks}},
{{topLaggingSources}}, {{topCpuConsumers}},
{{topGcIntensiveTaskManagers}}. All lists are always present; an empty
list means "no data".
* {{backPressuredTimeMsPerSecond}} and {{busyTimeMsPerSecond}} are
normalized to {{[0, 1]}}.
* Source lag aggregates {{pendingRecords}} as a sum and the two FLIP-33
lag metrics as max across subtasks of the same vertex.
* GC time is summed across collectors per TaskManager.
* {{-1}} is used as a numeric sentinel for a missing source-lag
dimension so the JSON stays flat numeric.
* REST API snapshot and docs are regenerated; the new {{topN}} query
parameter is reflected there.
h3. Frontend ({{flink-runtime-web}})
* New "Top N" tab on the job detail page, wired to the real endpoint
(no mock data left behind).
* Two-tier layout: the per-subtask tier (backpressure / busy / source
lag) is expanded by default; the JVM tier (CPU / GC) is collapsed by
default.
* Each row links back to the corresponding vertex / subtask /
TaskManager page so triage stays one click away from the existing UI.
* {{"n/a"}} is rendered for missing source-lag dimensions.
* The earlier empty "Top N" placeholder on the Overview page has been
removed; only the job-scoped tab remains.
h3. Tests
* {{TopNMetricsHandlerTest}} covering: ranking, {{top}} clamping,
operator-scoped metric discovery (suffix match), source-lag
aggregation (sum vs. max), missing-value sentinels, empty metric
store, and the FLIP-33 source-vertex heuristic.
* Angular component / service unit tests for the new tab and for the
collapse/expand state of each tier.
* Manual end-to-end verification against a local cluster running a job
that intentionally exhibits backpressure, a busy map, a lagging Kafka
source, a hot TaskManager, and a GC-heavy TaskManager.
h2. Sample response
{code:json}
GET /jobs/3a1b.../metrics/top-n?top=3
{
"topBackpressuredSubtasks": [
{ "vertexId": "abc123", "vertexName": "Sink: kafka", "subtaskIndex": 2,
"backpressureRatio": 0.87 }
],
"topBusySubtasks": [
{ "vertexId": "def456", "vertexName": "Map", "subtaskIndex": 0,
"busyRatio": 0.94 }
],
"topLaggingSources": [
{
"vertexId": "src789",
"vertexName": "Source: kafka-input",
"pendingRecords": 152340,
"fetchLagMs": 2100,
"emitLagMs": -1
}
],
"topCpuConsumers": [
{ "taskManagerId": "tm-4", "cpuLoad": 0.91 }
],
"topGcIntensiveTaskManagers": [
{ "taskManagerId": "tm-2", "gcTimeMs": 13450 }
]
}
{code}
h2. Implementation notes
* The handler is stateless and reads a snapshot of {{MetricStore}} per
request; no caching, no background thread.
* Backpressure and busy time are task-scoped, read via
{{SubtaskMetricStore#getMetric(name)}}.
* FLIP-33 source metrics are operator-scoped and stored as
{{<operatorName>.<metricName>}} inside {{SubtaskMetricStore.metrics}};
the handler discovers them by suffix match. Chained source operators
in the same subtask are summed (for {{pendingRecords}}) or
max-reduced (for lag).
* "Source vertex" is detected *structurally* (the vertex exposes at
least one FLIP-33 metric), not topologically.
{{AccessExecutionJobVertex}} does not expose {{getInputs()}}, and
threading the job graph through the handler was judged not worth the
coupling. This also means every FLIP-33-compliant connector is
covered automatically.
* {{TopNMetricsResponseBody}} is {{@Internal}}; a 3-arg convenience
constructor is retained for backwards compatibility with older call
sites, delegating to the canonical 5-arg constructor.
h2. How to try the prototype
{noformat}
git fetch https://github.com/apache/flink refs/pull/27774/head:flink-topn
git checkout flink-topn
./mvnw -pl flink-runtime-web -Pskip-webui-build=false -DskipTests clean install
# start any job, then either open:
# http://<jm-host>:8081/#/job/<jobid>/top-n
# or hit the REST API directly:
# curl 'http://<jm-host>:8081/jobs/<jobid>/metrics/top-n?top=10'
{noformat}
h2. Status
* (/) Backend handler, response body, and REST wiring implemented.
* (/) REST API snapshot and docs regenerated.
* (/) Unit tests for {{TopNMetricsHandler}} added.
* (/) Frontend tab wired to real data; mock data removed; empty
Overview placeholder removed.
* (!) Awaiting FLIP discussion outcome on dev@ before requesting merge.
h2. Out of scope (follow-up tickets)
* Top N slowest checkpoints (separate FLIP / JIRA, driven from
{{CheckpointStatsSnapshot}}).
* Per-job attribution of CPU / GC (requires a slot-to-TM mapping not
currently exposed by the metric store).
* Sparkline history per row (the endpoint is a live snapshot today).
h2. Links
* FLIP-XXX wiki page: <to be added once the FLIP page is created>
* {{[DISCUSS]}} mail thread on dev@: <to be added after sending>
* Pull request: https://github.com/apache/flink/pull/27774
was:
h2. Background
Operators troubleshooting a running Flink job today have to navigate
through many subtask-level pages in the Web UI to locate resource
hotspots (high CPU, backpressure, GC). There is no single place that
ranks tasks/operators by these signals.
h2. Goal
Provide a *Top N Metrics Dashboard* in the Flink Web UI that, for a
given job, lists the most resource-intensive components across three
dimensions:
* Top N CPU consumers — tasks with the highest CPU usage
* Top N backpressured operators — operators with the highest
backpressure ratio
* Top N GC-intensive tasks — tasks with the highest GC overhead
h2. Proposed Solution
* Add a new REST endpoint {{GET /jobs/:jobid/metrics/top-n}} served by a
new {{TopNMetricsHandler}} (extends {{{}AbstractRestHandler{}}}), wired
into {{{}WebMonitorEndpoint{}}}.
* Aggregate metrics via the *public* {{MetricStore}} API
({{{}getRepresentativeAttempts(){}}},
{{{}getAllSubtaskMetricStores(){}}}) — no reliance on package-private
state.
* Response body lives under
{{org.apache.flink.runtime.rest.messages.job.metrics}} and returns
three ranked sections (CPU / Backpressure / GC), default N = 5.
* Add an Angular {{TopNMetricsComponent}} + {{TopNMetricsService}} in
{{flink-runtime-web}} that renders the three Top N lists and an
optional {{/overview-demo}} entry point.
h2. Non-goals
* No changes to how metrics are collected or cached.
* No changes to existing REST endpoints or public APIs.
* Historical/time-series ranking is out of scope for this issue.
h2. Verification
# Build: {{mvn clean install -DskipTests}} and
{{npm run build}} in {{{}flink-runtime-web/web-dashboard{}}}.
# Integration: run a sample job, call
{\{GET /jobs/ {jobId}
/metrics/top-n}} and assert the three sections
are populated.
# UI: load the dashboard page in the Web UI and visually verify
the three Top N lists render with live data.
# Add unit tests for {{TopNMetricsHandler}} aggregation logic.
h2. Notes
Previous attempts: [PR
#27771|https://github.com/apache/flink/pull/27771](dropped — architectural
issues) and
[PR #27773|https://github.com/apache/flink/pull/27773] (dropped).
The replacement PR is[ PR #27774|https://github.com/apache/flink/pull/27774],
which currently references an incorrect JIRA id ({{{}FLINK-27773{}}}, which
belongs to "Introduce the E2E tests for SQL Gateway") and must be re-linked to
this new ticket.
> [web dashboard]Add Top N Metrics Dashboard to Flink Web UI
> ----------------------------------------------------------
>
> Key: FLINK-39489
> URL: https://issues.apache.org/jira/browse/FLINK-39489
> Project: Flink
> Issue Type: New Feature
> Components: Runtime / Web Frontend
> Reporter: featzhang
> Priority: Minor
> Labels: pull-request-available
>
> h2. Summary
> Add a "Top N" tab to the job detail page of the Flink Web UI, backed by a
> new REST endpoint {{GET /jobs/:jobid/metrics/top-n}}. The tab surfaces, for
> the currently open job, the worst offenders across five operator/JVM health
> dimensions in a single view, so operators can triage a misbehaving job
> without clicking through every vertex.
> This ticket is the implementation umbrella for FLIP-XXX. A prototype PR is
> already open for early feedback; it is *not* requesting merge before the
> FLIP is accepted.
> h2. Motivation
> When a Flink job misbehaves (lag, slow checkpoints, instability), the first
> question an operator asks is _"which subtask / which TaskManager should I
> be looking at?"_. Today the Web UI answers this only indirectly:
> * per-subtask backpressure and busy time are on the vertex page,
> * FLIP-33 source-lag metrics are on the source-vertex page,
> * JVM CPU and GC are on the TaskManager page.
> Each of these is already exposed by the runtime - the gap is that they are
> not surfaced in one place ranked by severity. For jobs with many vertices
> and high parallelism, operators today click through the UI vertex by
> vertex, which is slow and error-prone.
> h2. Goal
> Expose one new REST endpoint and one new UI tab that, for the current job,
> show top-N ranked lists across five dimensions grouped into two tiers.
> h3. Primary triage (per-subtask / per-vertex, directly actionable)
> || # || Dimension || Source metric(s) || Notes ||
> | 1 | Top Backpressured Subtasks | {{backPressuredTimeMsPerSecond}} | exposed
> as a ratio in {{[0, 1]}} |
> | 2 | Top Busy Subtasks | {{busyTimeMsPerSecond}} | exposed as a ratio in
> {{[0, 1]}} |
> | 3 | Top Lagging Sources | {{pendingRecords}}, {{currentFetchEventTimeLag}},
> {{currentEmitEventTimeLag}} (FLIP-33) | backlog summed across subtasks; lag
> metrics taken as max |
> h3. JVM diagnostics (TaskManager-scoped, coarser)
> || # || Dimension || Source metric(s) || Notes ||
> | 4 | Top CPU Consumers | {{Status.JVM.CPU.Load}} | TaskManager-scoped |
> | 5 | Top GC-intensive TaskManagers | {{Status.JVM.GarbageCollector.*.Time}}
> | summed across collectors per TM |
> Dimensions 4-5 are *TaskManager-scoped*, not per-job, because CPU and GC
> are JVM-process metrics and per-job attribution requires a slot-to-TM
> mapping that the metric store does not currently expose. This limitation
> is surfaced in the UI (the JVM tier is visually demoted and collapsed by
> default).
> h2. Non-goals
> * (x) No new {{ConfigOption}}.
> * (x) No new metric emitted by the runtime - everything is derived from
> metrics that already exist.
> * (x) No per-job attribution of CPU / GC (future work, tracked separately).
> * (x) No Top N for checkpoints in this ticket; its data source is
> {{CheckpointStatsSnapshot}}, not {{MetricStore}}, and it fits better
> as a follow-up.
> * (x) No persistence of "top-N over time" - the endpoint is a live
> snapshot.
> * (x) No changes to any existing REST endpoint, response body, or metric.
> All wire additions are purely additive.
> h2. Proposed changes
> h3. Backend ({{flink-runtime}})
> * New handler {{TopNMetricsHandler}} on
> {{GET /jobs/:jobid/metrics/top-n?top=N}} (default {{top=5}}, bounded to
> a reasonable cap). Reads from the existing {{MetricStore}}.
> * New response body {{TopNMetricsResponseBody}} with five *additive*
> lists: {{topBackpressuredSubtasks}}, {{topBusySubtasks}},
> {{topLaggingSources}}, {{topCpuConsumers}},
> {{topGcIntensiveTaskManagers}}. All lists are always present; an empty
> list means "no data".
> * {{backPressuredTimeMsPerSecond}} and {{busyTimeMsPerSecond}} are
> normalized to {{[0, 1]}}.
> * Source lag aggregates {{pendingRecords}} as a sum and the two FLIP-33
> lag metrics as max across subtasks of the same vertex.
> * GC time is summed across collectors per TaskManager.
> * {{-1}} is used as a numeric sentinel for a missing source-lag
> dimension so the JSON stays flat numeric.
> * REST API snapshot and docs are regenerated; the new {{topN}} query
> parameter is reflected there.
> h3. Frontend ({{flink-runtime-web}})
> * New "Top N" tab on the job detail page, wired to the real endpoint
> (no mock data left behind).
> * Two-tier layout: the per-subtask tier (backpressure / busy / source
> lag) is expanded by default; the JVM tier (CPU / GC) is collapsed by
> default.
> * Each row links back to the corresponding vertex / subtask /
> TaskManager page so triage stays one click away from the existing UI.
> * {{"n/a"}} is rendered for missing source-lag dimensions.
> * The earlier empty "Top N" placeholder on the Overview page has been
> removed; only the job-scoped tab remains.
> h3. Tests
> * {{TopNMetricsHandlerTest}} covering: ranking, {{top}} clamping,
> operator-scoped metric discovery (suffix match), source-lag
> aggregation (sum vs. max), missing-value sentinels, empty metric
> store, and the FLIP-33 source-vertex heuristic.
> * Angular component / service unit tests for the new tab and for the
> collapse/expand state of each tier.
> * Manual end-to-end verification against a local cluster running a job
> that intentionally exhibits backpressure, a busy map, a lagging Kafka
> source, a hot TaskManager, and a GC-heavy TaskManager.
> h2. Sample response
> {code:json}
> GET /jobs/3a1b.../metrics/top-n?top=3
> {
> "topBackpressuredSubtasks": [
> { "vertexId": "abc123", "vertexName": "Sink: kafka", "subtaskIndex": 2,
> "backpressureRatio": 0.87 }
> ],
> "topBusySubtasks": [
> { "vertexId": "def456", "vertexName": "Map", "subtaskIndex": 0,
> "busyRatio": 0.94 }
> ],
> "topLaggingSources": [
> {
> "vertexId": "src789",
> "vertexName": "Source: kafka-input",
> "pendingRecords": 152340,
> "fetchLagMs": 2100,
> "emitLagMs": -1
> }
> ],
> "topCpuConsumers": [
> { "taskManagerId": "tm-4", "cpuLoad": 0.91 }
> ],
> "topGcIntensiveTaskManagers": [
> { "taskManagerId": "tm-2", "gcTimeMs": 13450 }
> ]
> }
> {code}
> h2. Implementation notes
> * The handler is stateless and reads a snapshot of {{MetricStore}} per
> request; no caching, no background thread.
> * Backpressure and busy time are task-scoped, read via
> {{SubtaskMetricStore#getMetric(name)}}.
> * FLIP-33 source metrics are operator-scoped and stored as
> {{<operatorName>.<metricName>}} inside {{SubtaskMetricStore.metrics}};
> the handler discovers them by suffix match. Chained source operators
> in the same subtask are summed (for {{pendingRecords}}) or
> max-reduced (for lag).
> * "Source vertex" is detected *structurally* (the vertex exposes at
> least one FLIP-33 metric), not topologically.
> {{AccessExecutionJobVertex}} does not expose {{getInputs()}}, and
> threading the job graph through the handler was judged not worth the
> coupling. This also means every FLIP-33-compliant connector is
> covered automatically.
> * {{TopNMetricsResponseBody}} is {{@Internal}}; a 3-arg convenience
> constructor is retained for backwards compatibility with older call
> sites, delegating to the canonical 5-arg constructor.
> h2. How to try the prototype
> {noformat}
> git fetch https://github.com/apache/flink refs/pull/27774/head:flink-topn
> git checkout flink-topn
> ./mvnw -pl flink-runtime-web -Pskip-webui-build=false -DskipTests clean
> install
> # start any job, then either open:
> # http://<jm-host>:8081/#/job/<jobid>/top-n
> # or hit the REST API directly:
> # curl 'http://<jm-host>:8081/jobs/<jobid>/metrics/top-n?top=10'
> {noformat}
> h2. Status
> * (/) Backend handler, response body, and REST wiring implemented.
> * (/) REST API snapshot and docs regenerated.
> * (/) Unit tests for {{TopNMetricsHandler}} added.
> * (/) Frontend tab wired to real data; mock data removed; empty
> Overview placeholder removed.
> * (!) Awaiting FLIP discussion outcome on dev@ before requesting merge.
> h2. Out of scope (follow-up tickets)
> * Top N slowest checkpoints (separate FLIP / JIRA, driven from
> {{CheckpointStatsSnapshot}}).
> * Per-job attribution of CPU / GC (requires a slot-to-TM mapping not
> currently exposed by the metric store).
> * Sparkline history per row (the endpoint is a live snapshot today).
> h2. Links
> * FLIP-XXX wiki page: <to be added once the FLIP page is created>
> * {{[DISCUSS]}} mail thread on dev@: <to be added after sending>
> * Pull request: https://github.com/apache/flink/pull/27774
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
