[FLINK-39079][Web UI] Add Top N Metrics Dashboard for performance diagnostics

flink-runtime-web/web-dashboard/src/app/interfaces/top-n-metrics.ts

@@ -0,0 +1,44 @@
+/*
+ * 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.
+ */
+
+export interface TopNMetrics {
+  topCpuConsumers: CpuConsumer[];
+  topBackpressureOperators: BackpressureOperator[];
+  topGcIntensiveTasks: GcIntensiveTask[];
+}
+
+export interface CpuConsumer {
+  subtaskId: number;
+  taskName: string;
+  cpuPercentage: number;
+  taskManagerId: string | null;
+}
+
+export interface BackpressureOperator {
+  operatorId: string;
+  operatorName: string;
+  backpressureRatio: number;
+  subtaskId: number;
+}
+
+export interface GcIntensiveTask {
+  taskName: string;
+  gcTimePercentage: number;
+  taskManagerId: string;
+}

flink-runtime-web/web-dashboard/src/app/pages/job/overview/job-overview.component.html

@@ -22,6 +22,12 @@
   nzType="info"
   nzMessage="Job is not running yet."
 ></nz-alert>
+<flink-top-n-metrics
+  *ngIf="nodes.length > 0"
+  [topCpuConsumers]="topNMetrics.topCpuConsumers"
+  [topBackpressureOperators]="topNMetrics.topBackpressureOperators"
+  [topGcIntensiveTasks]="topNMetrics.topGcIntensiveTasks"
+></flink-top-n-metrics>
 <div class="container" [style.height.px]="top">
   <flink-dagre
     (nodeClick)="onNodeClick($event)"

flink-runtime-web/web-dashboard/src/app/pages/job/overview/job-overview.component.ts

@@ -33,8 +33,11 @@ import { catchError, filter, map, mergeMap, takeUntil } from 'rxjs/operators';
 import { DagreComponent } from '@flink-runtime-web/components/dagre/dagre.component';
 import { ResizeComponent } from '@flink-runtime-web/components/resize/resize.component';
 import { NodesItemCorrect, NodesItemLink } from '@flink-runtime-web/interfaces';
+import { TopNMetrics } from '@flink-runtime-web/interfaces/top-n-metrics';
 import { JobOverviewListComponent } from '@flink-runtime-web/pages/job/overview/list/job-overview-list.component';
+import { TopNMetricsComponent } from '@flink-runtime-web/pages/job/overview/top-n-metrics/top-n-metrics.component';
 import { JobService, MetricsService } from '@flink-runtime-web/services';
+import { TopNMetricsService } from '@flink-runtime-web/services/top-n-metrics.service';
 import { NzAlertModule } from 'ng-zorro-antd/alert';
 import { NzNotificationService } from 'ng-zorro-antd/notification';
 
@@ -45,7 +48,7 @@ import { JobLocalService } from '../job-local.service';
   templateUrl: './job-overview.component.html',
   styleUrls: ['./job-overview.component.less'],
   changeDetection: ChangeDetectionStrategy.OnPush,
-  imports: [NzAlertModule, NgIf, DagreComponent, RouterOutlet, JobOverviewListComponent, ResizeComponent]
+  imports: [NzAlertModule, NgIf, DagreComponent, RouterOutlet, JobOverviewListComponent, ResizeComponent, TopNMetricsComponent]
 })
 export class JobOverviewComponent implements OnInit, OnDestroy {
   public nodes: NodesItemCorrect[] = [];
@@ -55,6 +58,11 @@ export class JobOverviewComponent implements OnInit, OnDestroy {
   public pendingNodes: NodesItemCorrect[] = [];
   public pendingLinks: NodesItemLink[] = [];
   public selectedNode: NodesItemCorrect | null;
+  public topNMetrics: TopNMetrics = {
+    topCpuConsumers: [],
+    topBackpressureOperators: [],
+    topGcIntensiveTasks: []
+  };
   public top = 500;
   public jobId: string;
   public timeoutId: number;
@@ -71,7 +79,8 @@ export class JobOverviewComponent implements OnInit, OnDestroy {
     private readonly jobLocalService: JobLocalService,
     private readonly jobService: JobService,
     private readonly notificationService: NzNotificationService,
-    private readonly cdr: ChangeDetectorRef
+    private readonly cdr: ChangeDetectorRef,
+    private readonly topNMetricsService: TopNMetricsService
   ) {}
 
   public ngOnInit(): void {
@@ -91,6 +100,7 @@ export class JobOverviewComponent implements OnInit, OnDestroy {
           this.updatePendingInfo();
           this.refreshGraph(this.dagreComponent.showPendingOperators);
           this.refreshNodesWithMetrics();
+          this.loadTopNMetrics();
         } else {
           this.nodes = data.plan.nodes;
           this.refreshNodesWithMetrics();
@@ -225,4 +235,23 @@ export class JobOverviewComponent implements OnInit, OnDestroy {
       });
     return pendingLinks;
   }
+
+  private loadTopNMetrics(): void {
+    this.topNMetricsService
+      .loadTopNMetrics(this.jobId)
+      .pipe(
+        catchError(() => {
+          return of({
+            topCpuConsumers: [],
+            topBackpressureOperators: [],
+            topGcIntensiveTasks: []
+          });
+        }),
+        takeUntil(this.destroy$)
+      )
+      .subscribe(metrics => {
+        this.topNMetrics = metrics;
+        this.cdr.markForCheck();
+      });
+  }
 }

flink-runtime-web/web-dashboard/src/app/pages/job/overview/top-n-metrics/top-n-metrics.component.ts

@@ -0,0 +1,87 @@
+/*
+ * 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 { Component, Input } from '@angular/core';
+import { CommonModule } from '@angular/common';
+
+import { CpuConsumer, BackpressureOperator, GcIntensiveTask } from '@flink-runtime-web/interfaces/top-n-metrics';
+
+@Component({
+  selector: 'flink-top-n-metrics',
+  standalone: true,
+  template: `
+    <div class="top-n-metrics" *ngIf="topCpuConsumers?.length || topBackpressureOperators?.length || topGcIntensiveTasks?.length">
+      <h3>Top N Metrics</h3>
+
+      <div class="metric-section" *ngIf="topCpuConsumers?.length">
+        <h4>Top {{ topCpuConsumers.length }} CPU Consumers</h4>
+        <ul>
+          <li *ngFor="let cpu of topCpuConsumers">
+            {{ cpu.taskName }} (Subtask {{ cpu.subtaskId }}): {{ cpu.cpuPercentage | number:'1.2' }}% CPU
+          </li>
+        </ul>
+      </div>
+
+      <div class="metric-section" *ngIf="topBackpressureOperators?.length">
+        <h4>Top {{ topBackpressureOperators.length }} Backpressure Operators</h4>
+        <ul>
+          <li *ngFor="let bp of topBackpressureOperators">
+            {{ bp.operatorName }} (Subtask {{ bp.subtaskId }}): {{ bp.backpressureRatio | percent }}
+          </li>
+        </ul>
+      </div>
+
+      <div class="metric-section" *ngIf="topGcIntensiveTasks?.length">
+        <h4>Top {{ topGcIntensiveTasks.length }} GC Intensive Tasks</h4>
+        <ul>
+          <li *ngFor="let gc of topGcIntensiveTasks">
+            {{ gc.taskName }}: {{ gc.gcTimePercentage | number:'1.2' }}% GC Time
+          </li>
+        </ul>
+      </div>
+    </div>
+  `,
+  styles: [`
+    .top-n-metrics {
+      padding: 16px;
+      margin: 16px 0;
+      background-color: #f5f5f5;
+      border-radius: 4px;
+    }
+    .metric-section {
+      margin: 12px 0;
+    }
+    h3 {
+      margin: 0 0 12px 0;
+    }
+    h4 {
+      margin: 8px 0;
+    }
+    ul {
+      margin: 4px 0;
+      padding-left: 20px;
+    }
+  `],
+  imports: [CommonModule]
+})
+export class TopNMetricsComponent {
+  @Input() topCpuConsumers: CpuConsumer[] = [];
+  @Input() topBackpressureOperators: BackpressureOperator[] = [];
+  @Input() topGcIntensiveTasks: GcIntensiveTask[] = [];
+}

flink-runtime-web/web-dashboard/src/app/services/top-n-metrics.service.ts

@@ -0,0 +1,36 @@
+/*
+ * 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 { Injectable } from '@angular/core';
+import { Observable } from 'rxjs';
+
+import { HttpClient } from '@angular/common/http';
+
+import { TopNMetrics } from '@flink-runtime-web/interfaces/top-n-metrics';
+
+@Injectable({
+  providedIn: 'root'
+})
+export class TopNMetricsService {
+  constructor(private readonly http: HttpClient) {}
+
+  public loadTopNMetrics(jobId: string): Observable<TopNMetrics> {
+    return this.http.get<TopNMetrics>(`/jobs/${jobId}/metrics/top-n`);
+  }
+}