[FLINK-38719][runtime] Display the number of tasks on the TaskManagers page.

Author: RocMarshal

docs/layouts/shortcodes/generated/rest_v1_dispatcher.html

@@ -5240,6 +5240,9 @@
           "freeSlots" : {
             "type" : "integer"
           },
+          "numberOfTasks" : {
+            "type" : "integer"
+          },
           "hardware" : {
             "type" : "object",
             "id" : "urn:jsonschema:org:apache:flink:runtime:instance:HardwareDescription",
@@ -5464,6 +5467,9 @@
     "freeSlots" : {
       "type" : "integer"
     },
+    "numberOfTasks" : {
+      "type" : "integer"
+    },
     "hardware" : {
       "type" : "object",
       "id" : "urn:jsonschema:org:apache:flink:runtime:instance:HardwareDescription",

flink-runtime-web/src/test/resources/rest_api_v1.snapshot

@@ -3931,6 +3931,9 @@
               "freeSlots" : {
                 "type" : "integer"
               },
+              "numberOfTasks" : {
+                "type" : "integer"
+              },
               "totalResource" : {
                 "type" : "object",
                 "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:ResourceProfileInfo",
@@ -4093,6 +4096,9 @@
         "freeSlots" : {
           "type" : "integer"
         },
+        "numberOfTasks" : {
+          "type" : "integer"
+        },
         "totalResource" : {
           "type" : "object",
           "id" : "urn:jsonschema:org:apache:flink:runtime:rest:messages:ResourceProfileInfo",

flink-runtime-web/web-dashboard/src/app/interfaces/task-manager.ts

@@ -27,6 +27,7 @@ export interface TaskManagerDetail {
   timeSinceLastHeartbeat: number;
   slotsNumber: number;
   freeSlots: number;
+  numberOfTasks: number;
   hardware: Hardware;
   metrics: Metrics;
   memoryConfiguration: MemoryConfiguration;
@@ -68,6 +69,7 @@ export interface TaskManagersItem {
   timeSinceLastHeartbeat: number;
   slotsNumber: number;
   freeSlots: number;
+  numberOfTasks: number;
   hardware: Hardware;
   blocked?: boolean;
 }

flink-runtime-web/web-dashboard/src/app/pages/task-manager/list/task-manager-list.component.html

@@ -32,6 +32,7 @@
         <th [nzSortFn]="sortHeartBeatFn" [nzWidth]="'160px'">Last Heartbeat</th>
         <th [nzSortFn]="sortSlotsNumberFn" [nzWidth]="'90px'">All Slots</th>
         <th [nzSortFn]="sortFreeSlotsFn" [nzWidth]="'100px'">Free Slots</th>
+        <th [nzSortFn]="sortTasksFn" [nzWidth]="'100px'">Tasks</th>
         <th [nzSortFn]="sortCpuCoresFn" [nzWidth]="'110px'">CPU Cores</th>
         <th [nzSortFn]="sortPhysicalMemoryFn" [nzWidth]="'120px'">Physical MEM</th>
         <th [nzSortFn]="sortFreeMemoryFn" [nzWidth]="'130px'">JVM Heap Size</th>
@@ -53,6 +54,7 @@
           <td>{{ manager.timeSinceLastHeartbeat | date: 'yyyy-MM-dd HH:mm:ss' }}</td>
           <td>{{ manager.slotsNumber }}</td>
           <td>{{ manager.freeSlots }}</td>
+          <td>{{ manager.numberOfTasks }}</td>
           <td>{{ manager.hardware.cpuCores }}</td>
           <td [attr.title]="manager.hardware.physicalMemory + ' bytes'">
             {{ manager.hardware.physicalMemory | humanizeBytes }}

flink-runtime-web/web-dashboard/src/app/pages/task-manager/list/task-manager-list.component.ts

@@ -49,6 +49,7 @@ export class TaskManagerListComponent implements OnInit, OnDestroy {
   public readonly sortHeartBeatFn = createSortFn(item => item.timeSinceLastHeartbeat);
   public readonly sortSlotsNumberFn = createSortFn(item => item.slotsNumber);
   public readonly sortFreeSlotsFn = createSortFn(item => item.freeSlots);
+  public readonly sortTasksFn = createSortFn(item => item.numberOfTasks);
   public readonly sortCpuCoresFn = createSortFn(item => item.hardware?.cpuCores);
   public readonly sortPhysicalMemoryFn = createSortFn(item => item.hardware?.physicalMemory);
   public readonly sortFreeMemoryFn = createSortFn(item => item.hardware?.freeMemory);

flink-runtime-web/web-dashboard/src/app/pages/task-manager/status/task-manager-status.component.html

@@ -22,12 +22,15 @@
     <flink-blocked-badge *ngIf="taskManagerDetail?.blocked"></flink-blocked-badge>
   </div>
   <nz-descriptions *ngIf="taskManagerDetail" nzBordered nzSize="small">
-    <nz-descriptions-item [nzSpan]="2" nzTitle="Path">
+    <nz-descriptions-item [nzSpan]="1" nzTitle="Path">
       {{ taskManagerDetail.path }}
     </nz-descriptions-item>
     <nz-descriptions-item [nzSpan]="1" nzTitle="Free/All Slots">
       {{ taskManagerDetail.freeSlots }} / {{ taskManagerDetail.slotsNumber }}
     </nz-descriptions-item>
+    <nz-descriptions-item [nzSpan]="1" nzTitle="Tasks">
+      {{ taskManagerDetail.numberOfTasks }}
+    </nz-descriptions-item>
     <nz-descriptions-item [nzSpan]="1" nzTitle="Last Heartbeat">
       {{ taskManagerDetail.timeSinceLastHeartbeat | date: 'yyyy-MM-dd HH:mm:ss' }}
     </nz-descriptions-item>

flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java

@@ -687,6 +687,10 @@ public CompletableFuture<Collection<TaskManagerInfo>> requestTaskManagerInfo(Dur
                             taskManagerHeartbeatManager.getLastHeartbeatFrom(resourceId),
                             slotManager.getNumberRegisteredSlotsOf(taskExecutor.getInstanceID()),
                             slotManager.getNumberFreeSlotsOf(taskExecutor.getInstanceID()),
+                            (int)
+                                    slotManager
+                                            .getLoadingWeightOf(taskExecutor.getInstanceID())
+                                            .getLoading(),
                             slotManager.getRegisteredResourceOf(taskExecutor.getInstanceID()),
                             slotManager.getFreeResourceOf(taskExecutor.getInstanceID()),
                             taskExecutor.getHardwareDescription(),
@@ -717,6 +721,7 @@ public CompletableFuture<TaskManagerInfoWithSlots> requestTaskManagerDetailsInfo
                                     taskManagerHeartbeatManager.getLastHeartbeatFrom(resourceId),
                                     slotManager.getNumberRegisteredSlotsOf(instanceId),
                                     slotManager.getNumberFreeSlotsOf(instanceId),
+                                    (int) slotManager.getLoadingWeightOf(instanceId).getLoading(),
                                     slotManager.getRegisteredResourceOf(instanceId),
                                     slotManager.getFreeResourceOf(instanceId),
                                     taskExecutor.getHardwareDescription(),

flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/ClusterResourceStatisticsProvider.java

@@ -20,10 +20,14 @@
 
 import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
 import org.apache.flink.runtime.instance.InstanceID;
+import org.apache.flink.runtime.scheduler.loading.LoadingWeight;
 
 /** Provides statistics of cluster resources. */
 public interface ClusterResourceStatisticsProvider {
 
+    /** Get total loading weight of the current instance. */
+    LoadingWeight getLoadingWeightOf(InstanceID instanceId);
+
     /** Get total number of registered slots. */
     int getNumberRegisteredSlots();
 

flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManager.java

@@ -34,6 +34,7 @@
 import org.apache.flink.runtime.resourcemanager.WorkerResourceSpec;
 import org.apache.flink.runtime.resourcemanager.registration.TaskExecutorConnection;
 import org.apache.flink.runtime.rest.messages.taskmanager.SlotInfo;
+import org.apache.flink.runtime.scheduler.loading.LoadingWeight;
 import org.apache.flink.runtime.slots.ResourceRequirement;
 import org.apache.flink.runtime.slots.ResourceRequirements;
 import org.apache.flink.runtime.taskexecutor.SlotReport;
@@ -755,6 +756,11 @@ private Set<PendingTaskManagerId> allocateTaskManagersAccordingTo(
     // Legacy APIs
     // ---------------------------------------------------------------------------------------------
 
+    @Override
+    public LoadingWeight getLoadingWeightOf(InstanceID instanceId) {
+        return taskManagerTracker.getLoadingWeightOf(instanceId);
+    }
+
     @Override
     public int getNumberRegisteredSlots() {
         return taskManagerTracker.getNumberRegisteredSlots();

flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedTaskManagerRegistration.java

@@ -22,8 +22,12 @@
 import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
 import org.apache.flink.runtime.instance.InstanceID;
 import org.apache.flink.runtime.resourcemanager.registration.TaskExecutorConnection;
+import org.apache.flink.runtime.scheduler.loading.DefaultLoadingWeight;
+import org.apache.flink.runtime.scheduler.loading.LoadingWeight;
 import org.apache.flink.util.Preconditions;
 
+import javax.annotation.Nonnull;
+
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
@@ -163,4 +167,12 @@ public void notifyAllocation(
         slots.put(allocationId, taskManagerSlot);
         idleSince = Long.MAX_VALUE;
     }
+
+    @Nonnull
+    @Override
+    public LoadingWeight getLoading() {
+        return slots.values().stream()
+                .map(TaskManagerSlotInformation::getLoading)
+                .reduce(DefaultLoadingWeight.EMPTY, LoadingWeight::merge);
+    }
 }