[#1074] feat: Introduce the metric of local_storage_uniffle_used_space

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

@@ -79,6 +79,7 @@ public boolean checkIsHealthy() {
     AtomicInteger num = new AtomicInteger(0);
     AtomicLong totalSpace = new AtomicLong(0L);
     AtomicLong usedSpace = new AtomicLong(0L);
+    AtomicLong uniffleUsedSpace = new AtomicLong(0L);
     AtomicInteger corruptedDirs = new AtomicInteger(0);
     CountDownLatch cdl = new CountDownLatch(storageInfos.size());
     storageInfos
@@ -94,6 +95,7 @@ public boolean checkIsHealthy() {
 
               totalSpace.addAndGet(getTotalSpace(storageInfo.storageDir));
               usedSpace.addAndGet(getUsedSpace(storageInfo.storageDir));
+              uniffleUsedSpace.addAndGet(getUniffleUsedSpace(storageInfo.storageDir));
 
               if (storageInfo.checkIsSpaceEnough()) {
                 num.incrementAndGet();
@@ -107,6 +109,7 @@ public boolean checkIsHealthy() {
     }
     ShuffleServerMetrics.gaugeLocalStorageTotalSpace.set(totalSpace.get());
     ShuffleServerMetrics.gaugeLocalStorageUsedSpace.set(usedSpace.get());
+    ShuffleServerMetrics.gaugeLocalStorageUniffleUsedSpace.set(uniffleUsedSpace.get());
     ShuffleServerMetrics.gaugeLocalStorageTotalDirsNum.set(storageInfos.size());
     ShuffleServerMetrics.gaugeLocalStorageCorruptedDirsNum.set(corruptedDirs.get());
     ShuffleServerMetrics.gaugeLocalStorageUsedSpaceRatio.set(
@@ -147,6 +150,32 @@ long getUsedSpace(File file) {
     return file.getTotalSpace() - file.getUsableSpace();
   }
 
+  protected static long getUniffleUsedSpace(File storageDir) {
+    if (storageDir == null || !storageDir.exists()) {
+      return 0;
+    }
+
+    if (storageDir.isFile()) {
+      return storageDir.length();
+    }
+
+    File[] files = storageDir.listFiles();
+    if (files == null) {
+      return 0;
+    }
+
+    long totalUsage = 0;
+    for (File file : files) {
+      if (file.isFile()) {
+        totalUsage += file.length();
+      } else {
+        totalUsage += getUniffleUsedSpace(file);
+      }
+    }
+
+    return totalUsage;
+  }
+
   // todo: This function will be integrated to MultiStorageManager, currently we only support disk
   // check.
   class StorageInfo {

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

@@ -62,6 +62,7 @@ public class ShuffleServerMetrics {
   private static final String LOCAL_STORAGE_CORRUPTED_DIRS_NUM = "local_storage_corrupted_dirs_num";
   private static final String LOCAL_STORAGE_TOTAL_SPACE = "local_storage_total_space";
   private static final String LOCAL_STORAGE_USED_SPACE = "local_storage_used_space";
+  private static final String LOCAL_STORAGE_UNIFFLE_USED_SPACE = "local_storage_uniffle_used_space";
   private static final String LOCAL_STORAGE_USED_SPACE_RATIO = "local_storage_used_space_ratio";
 
   private static final String IS_HEALTHY = "is_healthy";
@@ -144,6 +145,7 @@ public class ShuffleServerMetrics {
   public static Gauge.Child gaugeLocalStorageCorruptedDirsNum;
   public static Gauge.Child gaugeLocalStorageTotalSpace;
   public static Gauge.Child gaugeLocalStorageUsedSpace;
+  public static Gauge.Child gaugeLocalStorageUniffleUsedSpace;
   public static Gauge.Child gaugeLocalStorageUsedSpaceRatio;
 
   public static Gauge.Child gaugeIsHealthy;
@@ -291,6 +293,8 @@ private static void setUpMetrics() {
         metricsManager.addLabeledGauge(LOCAL_STORAGE_CORRUPTED_DIRS_NUM);
     gaugeLocalStorageTotalSpace = metricsManager.addLabeledGauge(LOCAL_STORAGE_TOTAL_SPACE);
     gaugeLocalStorageUsedSpace = metricsManager.addLabeledGauge(LOCAL_STORAGE_USED_SPACE);
+    gaugeLocalStorageUniffleUsedSpace =
+        metricsManager.addLabeledGauge(LOCAL_STORAGE_UNIFFLE_USED_SPACE);
     gaugeLocalStorageUsedSpaceRatio =
         metricsManager.addLabeledGauge(LOCAL_STORAGE_USED_SPACE_RATIO);
 

server/src/test/java/org/apache/uniffle/server/LocalStorageCheckerTest.java

@@ -0,0 +1,60 @@
+/*
+ * 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.
+ */
+
+package org.apache.uniffle.server;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+public class LocalStorageCheckerTest {
+
+  @Test
+  public void testGetUniffleUsedSpace(@TempDir File tempDir) throws IOException {
+    File file1 = createTempFile(tempDir, "file1.txt", 1000);
+    File file2 = createTempFile(tempDir, "file2.txt", 2000);
+    File subdir1 = createTempSubDirectory(tempDir, "subdir1");
+    File file3 = createTempFile(subdir1, "file3.txt", 500);
+    File subdir2 = createTempSubDirectory(subdir1, "subdir2");
+    File file4 = createTempFile(subdir2, "file4.txt", 1500);
+
+    // Call the method to calculate disk usage
+    long calculatedUsage = LocalStorageChecker.getUniffleUsedSpace(tempDir);
+
+    // The expected total usage should be the sum of file1 + file2 + file3 + file4
+    long expectedUsage = file1.length() + file2.length() + file3.length() + file4.length();
+
+    // Assert that the calculated result matches the expected value
+    Assertions.assertEquals(expectedUsage, calculatedUsage);
+  }
+
+  private File createTempFile(File directory, String fileName, long fileSize) throws IOException {
+    File file = new File(directory, fileName);
+    Files.write(file.toPath(), new byte[(int) fileSize]);
+    return file;
+  }
+
+  private File createTempSubDirectory(File parentDirectory, String directoryName) {
+    File subDir = new File(parentDirectory, directoryName);
+    subDir.mkdirs();
+    return subDir;
+  }
+}