You are viewing a plain text version of this content. The canonical link for it is here.

Posted to issues@celeborn.apache.org by "Kalvin2077 (via GitHub)" <gi...@apache.org> on 2025/07/04 09:19:44 UTC

[PR] refactor remote storage configration usage. [celeborn]

Kalvin2077 opened a new pull request, #3353:
URL: https://github.com/apache/celeborn/pull/3353

   <!--
   Thanks for sending a pull request!  Here are some tips for you:
     - Make sure the PR title start w/ a JIRA ticket, e.g. '[CELEBORN-XXXX] Your PR title ...'.
     - Be sure to keep the PR description updated to reflect all changes.
     - Please write your PR title to summarize what this PR proposes.
     - If possible, provide a concise example to reproduce the issue for a faster review.
   -->
   
   ### What changes were proposed in this pull request?
   
   Refactoring similar code about configuration usage.
   
   ### Why are the changes needed?
   
   Improve scalability for possible new remote storage in the future.
   
   ### Does this PR introduce _any_ user-facing change?
   
   No.
   
   ### How was this patch tested?
   
   Unit tests.
   
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

Re: [PR] [CELEBORN-2053] Refactor remote storage configration usage. [celeborn]

Posted by "RexXiong (via GitHub)" <gi...@apache.org>.

RexXiong commented on code in PR #3353:
URL: https://github.com/apache/celeborn/pull/3353#discussion_r2192760175


##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -1241,6 +1241,30 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
     }.getOrElse("")
   }
 
+  def remoteStorageDirs: Option[Set[(StorageInfo.Type, String)]] = {
+    val supported = Seq(
+      (StorageInfo.Type.HDFS, HDFS_DIR, Utils.isHdfsPath _),
+      (StorageInfo.Type.S3, S3_DIR, Utils.isS3Path _),
+      (StorageInfo.Type.OSS, OSS_DIR, Utils.isOssPath _))
+
+    val activeNames = get(ACTIVE_STORAGE_TYPES)
+
+    val validDirs = supported.flatMap { case (ty, e, checker) =>
+      if (!activeNames.contains(ty.name)) None

Review Comment:
   Split activeNames to List then we can contain the exactly storage type



##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala:
##########
@@ -86,27 +85,20 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs
         (new File(workdir, conf.workerWorkingDir), maxSpace, flusherThread, storageType)
       }
 
-    if (workingDirInfos.size <= 0 && !hasHDFSStorage && !hasS3Storage && !hasOssStorage) {
+    if (workingDirInfos.size <= 0 && remoteStorageDirs.isEmpty) {
       throw new IOException("Empty working directory configuration!")
     }
 
     DeviceInfo.getDeviceAndDiskInfos(workingDirInfos, conf)
   }
   val mountPoints = new util.HashSet[String](diskInfos.keySet())
-  val hdfsDiskInfo =
-    if (conf.hasHDFSStorage)
-      Option(new DiskInfo("HDFS", Long.MaxValue, 999999, 999999, 0, StorageInfo.Type.HDFS))
-    else None
-
-  val s3DiskInfo =
-    if (conf.hasS3Storage)
-      Option(new DiskInfo("S3", Long.MaxValue, 999999, 999999, 0, StorageInfo.Type.S3))
-    else None
 
-  val ossDiskInfo =
-    if (conf.hasOssStorage)
-      Option(new DiskInfo("OSS", Long.MaxValue, 999999, 999999, 0, StorageInfo.Type.OSS))
-    else None
+  val remoteDiskInfos: Option[Set[DiskInfo]] = remoteStorageDirs.flatMap { dirs =>
+    val diskInfoSet = dirs.flatMap { case (storageInfoType, _) =>

Review Comment:
   dirs.map is better?



##########
master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/AbstractMetaManager.java:
##########
@@ -287,20 +288,14 @@ public void updateWorkerHeartbeatMeta(
     long unhealthyDiskNum =
         disks.values().stream().filter(s -> !s.status().equals(DiskStatus.HEALTHY)).count();
     boolean exceed = unhealthyDiskNum * 1.0 / disks.size() >= unhealthyDiskRatioThreshold;
+    Option<scala.collection.immutable.Set<Tuple2<StorageInfo.Type, String>>> remoteStorageDirs =

Review Comment:
   Let's introduce a final variable for `conf.remoteStorageDirs().isDefined()`



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org