[CELEBORN-1651] Support ratio threshold of unhealthy disks for excluding worker

common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala

@@ -780,6 +780,8 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
   def haRatisCustomConfigs: JMap[String, String] = {
     settings.asScala.filter(_._1.startsWith("celeborn.ratis")).toMap.asJava
   }
+  def masterExcludeWorkerUnhealthyDiskRatioThreshold: Double =
+    get(MASTER_EXCLUDE_WORKER_UNHEALTHY_DISK_RATIO_THRESHOLD)
 
   // //////////////////////////////////////////////////////
   //                      Worker                         //
@@ -5830,4 +5832,14 @@ object CelebornConf extends Logging {
         "`prod,high-io` filters workers that have both the `prod` and `high-io` tags.")
       .stringConf
       .createWithDefault("")
+
+  val MASTER_EXCLUDE_WORKER_UNHEALTHY_DISK_RATIO_THRESHOLD: ConfigEntry[Double] =
+    buildConf("celeborn.master.excludeWorker.unhealthyDiskRatioThreshold")
+      .categories("master")
+      .version("0.6.0")
+      .doc("Max ratio of unhealthy disks for excluding worker, when unhealthy disk is larger than max unhealthy count, master will exclude worker." +
+        " If this value is set to 1, master will exclude worker of which disks are all unhealthy.")
+      .doubleConf
+      .checkValue(v => v > 0.0 && v <= 1.0, "Should be in (0.0, 1.0].")
+      .createWithDefault(1)
 }

docs/configuration/master.md

@@ -42,6 +42,7 @@ license: |
 | celeborn.master.estimatedPartitionSize.minSize | 8mb | false | Ignore partition size smaller than this configuration of partition size for estimation. | 0.3.0 | celeborn.shuffle.minPartitionSizeToEstimate | 
 | celeborn.master.estimatedPartitionSize.update.initialDelay | 5min | false | Initial delay time before start updating partition size for estimation. | 0.3.0 | celeborn.shuffle.estimatedPartitionSize.update.initialDelay | 
 | celeborn.master.estimatedPartitionSize.update.interval | 10min | false | Interval of updating partition size for estimation. | 0.3.0 | celeborn.shuffle.estimatedPartitionSize.update.interval | 
+| celeborn.master.excludeWorker.unhealthyDiskRatioThreshold | 1.0 | false | Max ratio of unhealthy disks for excluding worker, when unhealthy disk is larger than max unhealthy count, master will exclude worker. If this value is set to 1, master will exclude worker of which disks are all unhealthy. | 0.6.0 |  | 
 | celeborn.master.hdfs.expireDirs.timeout | 1h | false | The timeout for a expire dirs to be deleted on HDFS. | 0.3.0 |  | 
 | celeborn.master.heartbeat.application.timeout | 300s | false | Application heartbeat timeout. | 0.3.0 | celeborn.application.heartbeat.timeout | 
 | celeborn.master.heartbeat.worker.timeout | 120s | false | Worker heartbeat timeout. | 0.3.0 | celeborn.worker.heartbeat.timeout | 

master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/AbstractMetaManager.java

@@ -213,14 +213,16 @@ public void updateWorkerHeartbeatMeta(
 
     appDiskUsageMetric.update(estimatedAppDiskUsage);
     // If using HDFSONLY mode, workers with empty disks should not be put into excluded worker list.
-    long healthyDiskNum =
-        disks.values().stream().filter(s -> s.status().equals(DiskStatus.HEALTHY)).count();
+    long unhealthyDiskNum =
+        disks.values().stream().filter(s -> !s.status().equals(DiskStatus.HEALTHY)).count();
+    boolean exceed =
+        unhealthyDiskNum * 1.0 / disks.size()
+            >= conf.masterExcludeWorkerUnhealthyDiskRatioThreshold();
     if (!excludedWorkers.contains(worker)
-        && (((disks.isEmpty() || healthyDiskNum <= 0)
-                && (!conf.hasHDFSStorage())
-                && (!conf.hasS3Storage()))
+        && (((disks.isEmpty() || exceed) && !conf.hasHDFSStorage() && !conf.hasS3Storage())
             || highWorkload)) {
-      LOG.debug("Worker: {} num total slots is 0, add to excluded list", worker);
+      LOG.debug(
+          "Worker {} (unhealthy disks num: {}) adds to excluded workers", worker, unhealthyDiskNum);
       excludedWorkers.add(worker);
     } else if ((availableSlots.get() > 0 || conf.hasHDFSStorage() || conf.hasS3Storage())
         && !highWorkload) {