[#2211] feat(dashboard) Display the write information of application in dashboard

coordinator/src/main/java/org/apache/uniffle/coordinator/CoordinatorGrpcService.java

@@ -537,7 +537,8 @@ private ServerNode toServerNode(ShuffleServerHeartBeatRequest request) {
         request.getServerId().getJettyPort(),
         request.getStartTimeMs(),
         request.getVersion(),
-        request.getGitCommitId());
+        request.getGitCommitId(),
+        request.getApplicationInfoList());
   }
 
   /**

coordinator/src/main/java/org/apache/uniffle/coordinator/ServerNode.java

@@ -17,14 +17,18 @@
 
 package org.apache.uniffle.coordinator;
 
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
 
 import org.apache.uniffle.common.ServerStatus;
 import org.apache.uniffle.common.storage.StorageInfo;
+import org.apache.uniffle.proto.RssProtos;
 import org.apache.uniffle.proto.RssProtos.ShuffleServerId;
 
 public class ServerNode implements Comparable<ServerNode> {
@@ -46,6 +50,7 @@ public class ServerNode implements Comparable<ServerNode> {
   private long startTime = -1;
   private String version;
   private String gitCommitId;
+  Map<String, RssProtos.ApplicationInfo> appIdToInfos;
 
   public ServerNode(String id) {
     this(id, "", 0, 0, 0, 0, 0, Sets.newHashSet(), ServerStatus.EXCLUDED);
@@ -181,7 +186,8 @@ public ServerNode(
         jettyPort,
         startTime,
         "",
-        "");
+        "",
+        Collections.EMPTY_LIST);
   }
 
   public ServerNode(
@@ -199,7 +205,8 @@ public ServerNode(
       int jettyPort,
       long startTime,
       String version,
-      String gitCommitId) {
+      String gitCommitId,
+      List<RssProtos.ApplicationInfo> appInfos) {
     this.id = id;
     this.ip = ip;
     this.grpcPort = grpcPort;
@@ -221,6 +228,8 @@ public ServerNode(
     this.startTime = startTime;
     this.version = version;
     this.gitCommitId = gitCommitId;
+    this.appIdToInfos = new ConcurrentHashMap<>();
+    appInfos.forEach(appInfo -> appIdToInfos.put(appInfo.getAppId(), appInfo));
   }
 
   public ShuffleServerId convertToGrpcProto() {
@@ -375,4 +384,8 @@ public String getVersion() {
   public String getGitCommitId() {
     return gitCommitId;
   }
+
+  public Map<String, RssProtos.ApplicationInfo> getAppIdToInfos() {
+    return appIdToInfos;
+  }
 }

coordinator/src/main/java/org/apache/uniffle/coordinator/web/resource/ApplicationResource.java

@@ -36,9 +36,12 @@
 import org.apache.uniffle.common.web.resource.Response;
 import org.apache.uniffle.coordinator.AppInfo;
 import org.apache.uniffle.coordinator.ApplicationManager;
+import org.apache.uniffle.coordinator.CoordinatorServer;
+import org.apache.uniffle.coordinator.ServerNode;
 import org.apache.uniffle.coordinator.metric.CoordinatorMetrics;
 import org.apache.uniffle.coordinator.web.vo.AppInfoVO;
 import org.apache.uniffle.coordinator.web.vo.UserAppNumVO;
+import org.apache.uniffle.proto.RssProtos;
 
 @Produces({MediaType.APPLICATION_JSON})
 public class ApplicationResource extends BaseResource {
@@ -84,19 +87,44 @@ public Response<List<AppInfoVO>> getAppInfoList() {
           List<AppInfoVO> userToAppList = new ArrayList<>();
           Map<String, Map<String, AppInfo>> currentUserAndApp =
               getApplicationManager().getCurrentUserAndApp();
+          List<ServerNode> serverNodes = getCoordinatorServer().getClusterManager().list();
           for (Map.Entry<String, Map<String, AppInfo>> userAppIdTimestampMap :
               currentUserAndApp.entrySet()) {
             for (Map.Entry<String, AppInfo> appIdTimestampMap :
                 userAppIdTimestampMap.getValue().entrySet()) {
+              String user = appIdTimestampMap.getKey();
               AppInfo appInfo = appIdTimestampMap.getValue();
-              userToAppList.add(
+              AppInfoVO appInfoVO =
                   new AppInfoVO(
-                      userAppIdTimestampMap.getKey(),
+                      user,
                       appInfo.getAppId(),
                       appInfo.getUpdateTime(),
                       appInfo.getRegistrationTime(),
                       appInfo.getVersion(),
-                      appInfo.getGitCommitId()));
+                      appInfo.getGitCommitId(),
+                      0,
+                      0,
+                      0,
+                      0,
+                      0,
+                      0,
+                      0);
+              for (ServerNode server : serverNodes) {
+                Map<String, RssProtos.ApplicationInfo> appIdToInfos = server.getAppIdToInfos();
+                if (appIdToInfos.containsKey(appInfoVO.getAppId())) {
+                  RssProtos.ApplicationInfo app = appIdToInfos.get(appInfoVO.getAppId());
+                  appInfoVO.setPartitionNum(appInfoVO.getPartitionNum() + app.getPartitionNum());
+                  appInfoVO.setMemorySize(appInfoVO.getMemorySize() + app.getMemorySize());
+                  appInfoVO.setLocalFileNum(appInfoVO.getLocalFileNum() + app.getLocalFileNum());
+                  appInfoVO.setLocalTotalSize(
+                      appInfoVO.getLocalTotalSize() + app.getLocalTotalSize());
+                  appInfoVO.setHadoopFileNum(appInfoVO.getHadoopFileNum() + app.getHadoopFileNum());
+                  appInfoVO.setHadoopTotalSize(
+                      appInfoVO.getHadoopTotalSize() + app.getHadoopTotalSize());
+                  appInfoVO.setTotalSize(appInfoVO.getTotalSize() + app.getTotalSize());
+                }
+              }
+              userToAppList.add(appInfoVO);
             }
           }
           // Display is inverted by the submission time of the application.
@@ -109,4 +137,9 @@ private ApplicationManager getApplicationManager() {
     return (ApplicationManager)
         servletContext.getAttribute(ApplicationManager.class.getCanonicalName());
   }
+
+  private CoordinatorServer getCoordinatorServer() {
+    return (CoordinatorServer)
+        servletContext.getAttribute(CoordinatorServer.class.getCanonicalName());
+  }
 }

coordinator/src/main/java/org/apache/uniffle/coordinator/web/vo/AppInfoVO.java

@@ -31,6 +31,13 @@ public class AppInfoVO implements Comparable<AppInfoVO> {
   private long registrationTime;
   private String version;
   private String gitCommitId;
+  private long partitionNum;
+  private long memorySize;
+  private long localFileNum;
+  private long localTotalSize;
+  private long hadoopFileNum;
+  private long hadoopTotalSize;
+  private long totalSize;
 
   @Override
   public int compareTo(AppInfoVO appInfoVO) {

dashboard/src/main/webapp/src/pages/ApplicationPage.vue

@@ -89,6 +89,37 @@
           label="GitCommitId"
           min-width="180"
         />
+        <el-table-column prop="partitionNum" label="PartitionNum" min-width="180" />
+        <el-table-column
+            prop="memorySize"
+            label="MemorySize"
+            min-width="180"
+            :formatter="memFormatter"
+        />
+        <el-table-column label="LocalFile" min-width="180">
+          <template v-slot="{ row }">
+            <div class="mb-4">
+              {{ row.localFileNum }}({{
+                memFormatter(null, null, row.localTotalSize)
+              }})
+            </div>
+          </template>
+        </el-table-column>
+        <el-table-column label="HadoopFile" min-width="180">
+          <template v-slot="{ row }">
+            <div class="mb-4">
+              {{ row.hadoopFileNum }}({{
+                memFormatter(null, null, row.hadoopTotalSize)
+              }})
+            </div>
+          </template>
+        </el-table-column>
+        <el-table-column
+            prop="totalSize"
+            label="TotalSize"
+            min-width="180"
+            :formatter="memFormatter"
+        />
       </el-table>
     </div>
   </div>
@@ -97,7 +128,7 @@
 <script>
 import { getApplicationInfoList, getAppTotal, getTotalForUser } from '@/api/api'
 import { onMounted, reactive } from 'vue'
-import { dateFormatter } from '@/utils/common'
+import { dateFormatter, memFormatter } from '@/utils/common'
 import { useCurrentServerStore } from '@/store/useCurrentServerStore'
 
 export default {
@@ -164,7 +195,8 @@ export default {
       sortAppCollectChangeEvent,
       sortApp,
       sortAppChangeEvent,
-      dateFormatter
+      dateFormatter,
+      memFormatter
     }
   }
 }

internal-client/src/main/java/org/apache/uniffle/client/impl/grpc/CoordinatorGrpcClient.java

@@ -127,7 +127,8 @@ public ShuffleServerHeartBeatResponse doSendHeartBeat(
       Map<String, StorageInfo> storageInfo,
       int nettyPort,
       int jettyPort,
-      long startTimeMs) {
+      long startTimeMs,
+      List<RssProtos.ApplicationInfo> appInfos) {
     ShuffleServerId serverId =
         ShuffleServerId.newBuilder()
             .setId(id)
@@ -149,6 +150,7 @@ public ShuffleServerHeartBeatResponse doSendHeartBeat(
             .setStartTimeMs(startTimeMs)
             .setVersion(Constants.VERSION)
             .setGitCommitId(Constants.REVISION_SHORT)
+            .addAllApplicationInfo(appInfos)
             .build();
 
     RssProtos.StatusCode status;
@@ -225,7 +227,8 @@ public RssSendHeartBeatResponse sendHeartBeat(RssSendHeartBeatRequest request) {
             request.getStorageInfo(),
             request.getNettyPort(),
             request.getJettyPort(),
-            request.getStartTimeMs());
+            request.getStartTimeMs(),
+            request.getAppInfos());
 
     RssSendHeartBeatResponse response;
     RssProtos.StatusCode statusCode = rpcResponse.getStatus();

internal-client/src/main/java/org/apache/uniffle/client/request/RssSendHeartBeatRequest.java

@@ -17,11 +17,13 @@
 
 package org.apache.uniffle.client.request;
 
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
 import org.apache.uniffle.common.ServerStatus;
 import org.apache.uniffle.common.storage.StorageInfo;
+import org.apache.uniffle.proto.RssProtos;
 
 public class RssSendHeartBeatRequest {
 
@@ -39,6 +41,7 @@ public class RssSendHeartBeatRequest {
   private final int nettyPort;
   private final int jettyPort;
   private final long startTimeMs;
+  private final List<RssProtos.ApplicationInfo> appInfos;
 
   public RssSendHeartBeatRequest(
       String shuffleServerId,
@@ -54,7 +57,8 @@ public RssSendHeartBeatRequest(
       Map<String, StorageInfo> storageInfo,
       int nettyPort,
       int jettyPort,
-      long startTimeMs) {
+      long startTimeMs,
+      List<RssProtos.ApplicationInfo> appInfos) {
     this.shuffleServerId = shuffleServerId;
     this.shuffleServerIp = shuffleServerIp;
     this.shuffleServerPort = shuffleServerPort;
@@ -69,6 +73,7 @@ public RssSendHeartBeatRequest(
     this.nettyPort = nettyPort;
     this.jettyPort = jettyPort;
     this.startTimeMs = startTimeMs;
+    this.appInfos = appInfos;
   }
 
   public String getShuffleServerId() {
@@ -126,4 +131,8 @@ public int getJettyPort() {
   public long getStartTimeMs() {
     return startTimeMs;
   }
+
+  public List<RssProtos.ApplicationInfo> getAppInfos() {
+    return appInfos;
+  }
 }

proto/src/main/proto/Rss.proto

@@ -273,6 +273,17 @@ enum ServerStatus {
   // todo: more status, such as UPGRADING
 }
 
+message ApplicationInfo {
+  string appId = 1;
+  int64 partitionNum = 2;
+  int64 memorySize = 3;
+  int64 localFileNum = 4;
+  int64 localTotalSize = 5;
+  int64 hadoopFileNum = 6;
+  int64 hadoopTotalSize = 7;
+  int64 totalSize = 8;
+}
+
 message ShuffleServerHeartBeatRequest {
   ShuffleServerId serverId = 1;
   int64 usedMemory = 2;
@@ -286,6 +297,7 @@ message ShuffleServerHeartBeatRequest {
   optional string version = 22;
   optional string gitCommitId = 23;
   optional int64 startTimeMs = 24;
+  repeated ApplicationInfo applicationInfo = 25;
 }
 
 message ShuffleServerHeartBeatResponse {

server/src/main/java/org/apache/uniffle/server/RegisterHeartBeat.java

@@ -17,6 +17,7 @@
 
 package org.apache.uniffle.server;
 
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ScheduledExecutorService;
@@ -33,6 +34,7 @@
 import org.apache.uniffle.common.rpc.StatusCode;
 import org.apache.uniffle.common.storage.StorageInfo;
 import org.apache.uniffle.common.util.ThreadUtils;
+import org.apache.uniffle.proto.RssProtos;
 
 public class RegisterHeartBeat {
 
@@ -84,7 +86,8 @@ public void startHeartBeat() {
                 shuffleServer.getStorageManager().getStorageInfo(),
                 shuffleServer.getNettyPort(),
                 shuffleServer.getJettyPort(),
-                shuffleServer.getStartTimeMs());
+                shuffleServer.getStartTimeMs(),
+                shuffleServer.getAppInfos());
           } catch (Exception e) {
             LOG.warn("Error happened when send heart beat to coordinator");
           }
@@ -107,7 +110,8 @@ public boolean sendHeartBeat(
       Map<String, StorageInfo> localStorageInfo,
       int nettyPort,
       int jettyPort,
-      long startTimeMs) {
+      long startTimeMs,
+      List<RssProtos.ApplicationInfo> appInfos) {
     // use `rss.server.heartbeat.interval` as the timeout option
     RssSendHeartBeatRequest request =
         new RssSendHeartBeatRequest(
@@ -124,7 +128,8 @@ public boolean sendHeartBeat(
             localStorageInfo,
             nettyPort,
             jettyPort,
-            startTimeMs);
+            startTimeMs,
+            appInfos);
 
     if (coordinatorClient.sendHeartBeat(request).getStatusCode() == StatusCode.SUCCESS) {
       return true;