You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by an...@apache.org on 2024/01/30 10:44:15 UTC
(incubator-celeborn) branch main updated: [CELEBORN-1241] Introduce hot load for CelebornRackResolver
This is an automated email from the ASF dual-hosted git repository.
angerszhuuuu pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new 62c8ac9c5 [CELEBORN-1241] Introduce hot load for CelebornRackResolver
62c8ac9c5 is described below
commit 62c8ac9c5231662774fd5ffa366d2d95ba44a0d3
Author: Xianming Lei <je...@apache.org>
AuthorDate: Tue Jan 30 18:44:02 2024 +0800
[CELEBORN-1241] Introduce hot load for CelebornRackResolver
### What changes were proposed in this pull request?
Introduce hot load for CelebornRackResolver.
### Why are the changes needed?
In production environment, we often expand the machine, so the rack configuration also needs to be updated in time.
### Does this PR introduce _any_ user-facing change?
master.md
### How was this patch tested?
UTs.
Closes #2246 from leixm/issue_1241.
Lead-authored-by: Xianming Lei <je...@apache.org>
Co-authored-by: Xianming Lei <31...@users.noreply.github.com>
Signed-off-by: Angerszhuuuu <an...@gmail.com>
---
.../org/apache/celeborn/common/CelebornConf.scala | 12 ++++
docs/configuration/master.md | 1 +
.../clustermeta/SingleMasterMetaManager.java | 5 ++
.../master/clustermeta/ha/HAMasterMetaManager.java | 6 +-
.../celeborn/service/deploy/master/Master.scala | 7 ++-
.../master/network/CelebornRackResolver.scala | 64 ++++++++++++++++++++--
.../master/network/CelebornRackResolverSuite.scala | 36 ++++++++++++
7 files changed, 124 insertions(+), 7 deletions(-)
diff --git a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
index 713bcc1d8..240fb24f9 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -1123,6 +1123,11 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
// //////////////////////////////////////////////////////
def internalPortEnabled: Boolean = get(INTERNAL_PORT_ENABLED)
+ // //////////////////////////////////////////////////////
+ // Rack Resolver //
+ // //////////////////////////////////////////////////////
+ def rackResolverRefreshInterval = get(RACKRESOLVER_REFRESH_INTERVAL)
+
def haMasterNodeInternalPort(nodeId: String): Int = {
val key = HA_MASTER_NODE_INTERNAL_PORT.key.replace("<id>", nodeId)
val legacyKey = HA_MASTER_NODE_INTERNAL_PORT.alternatives.head._1.replace("<id>", nodeId)
@@ -4426,4 +4431,11 @@ object CelebornConf extends Logging {
.checkValue(p => p >= 1024 && p < 65535, "Invalid port")
.createWithDefault(8097)
+ val RACKRESOLVER_REFRESH_INTERVAL: ConfigEntry[Long] =
+ buildConf("celeborn.master.rackResolver.refresh.interval")
+ .categories("master")
+ .version("0.5.0")
+ .doc("Interval for refreshing the node rack information periodically.")
+ .timeConf(TimeUnit.MILLISECONDS)
+ .createWithDefaultString("30s")
}
diff --git a/docs/configuration/master.md b/docs/configuration/master.md
index 416270f6a..65d9079e3 100644
--- a/docs/configuration/master.md
+++ b/docs/configuration/master.md
@@ -33,6 +33,7 @@ license: |
| celeborn.master.http.port | 9098 | Master's http port. | 0.4.0 | celeborn.metrics.master.prometheus.port,celeborn.master.metrics.prometheus.port |
| celeborn.master.internal.port | 8097 | Internal port on the master where both workers and other master nodes connect. | 0.5.0 | |
| celeborn.master.port | 9097 | Port for master to bind. | 0.2.0 | |
+| celeborn.master.rackResolver.refresh.interval | 30s | Interval for refreshing the node rack information periodically. | 0.5.0 | |
| celeborn.master.slot.assign.extraSlots | 2 | Extra slots number when master assign slots. | 0.3.0 | celeborn.slots.assign.extraSlots |
| celeborn.master.slot.assign.loadAware.diskGroupGradient | 0.1 | This value means how many more workload will be placed into a faster disk group than a slower group. | 0.3.0 | celeborn.slots.assign.loadAware.diskGroupGradient |
| celeborn.master.slot.assign.loadAware.fetchTimeWeight | 1.0 | Weight of average fetch time when calculating ordering in load-aware assignment strategy | 0.3.0 | celeborn.slots.assign.loadAware.fetchTimeWeight |
diff --git a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/SingleMasterMetaManager.java b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/SingleMasterMetaManager.java
index 2ddd1ec06..1aea3841e 100644
--- a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/SingleMasterMetaManager.java
+++ b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/SingleMasterMetaManager.java
@@ -36,6 +36,11 @@ public class SingleMasterMetaManager extends AbstractMetaManager {
private static final Logger LOG = LoggerFactory.getLogger(SingleMasterMetaManager.class);
public SingleMasterMetaManager(RpcEnv rpcEnv, CelebornConf conf) {
+ this(rpcEnv, conf, new CelebornRackResolver(conf));
+ }
+
+ public SingleMasterMetaManager(
+ RpcEnv rpcEnv, CelebornConf conf, CelebornRackResolver rackResolver) {
this.rpcEnv = rpcEnv;
this.conf = conf;
this.initialEstimatedPartitionSize = conf.initialEstimatedPartitionSize();
diff --git a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/HAMasterMetaManager.java b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/HAMasterMetaManager.java
index 087ce0374..16fd84b9b 100644
--- a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/HAMasterMetaManager.java
+++ b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/HAMasterMetaManager.java
@@ -46,12 +46,16 @@ public class HAMasterMetaManager extends AbstractMetaManager {
protected HARaftServer ratisServer;
public HAMasterMetaManager(RpcEnv rpcEnv, CelebornConf conf) {
+ this(rpcEnv, conf, new CelebornRackResolver(conf));
+ }
+
+ public HAMasterMetaManager(RpcEnv rpcEnv, CelebornConf conf, CelebornRackResolver rackResolver) {
this.rpcEnv = rpcEnv;
this.conf = conf;
this.initialEstimatedPartitionSize = conf.initialEstimatedPartitionSize();
this.estimatedPartitionSize = initialEstimatedPartitionSize;
this.appDiskUsageMetric = new AppDiskUsageMetric(conf);
- this.rackResolver = new CelebornRackResolver(conf);
+ this.rackResolver = rackResolver;
}
public HARaftServer getRatisServer() {
diff --git a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala
index b193518fb..bcc2b0eb8 100644
--- a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala
+++ b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala
@@ -47,6 +47,7 @@ import org.apache.celeborn.common.util.{CelebornHadoopUtils, JavaUtils, PbSerDeU
import org.apache.celeborn.server.common.{HttpService, Service}
import org.apache.celeborn.service.deploy.master.clustermeta.SingleMasterMetaManager
import org.apache.celeborn.service.deploy.master.clustermeta.ha.{HAHelper, HAMasterMetaManager, MetaHandler}
+import org.apache.celeborn.service.deploy.master.network.CelebornRackResolver
private[celeborn] class Master(
override val conf: CelebornConf,
@@ -95,9 +96,10 @@ private[celeborn] class Master(
internalRpcEnvInUse = internalRpcEnv
}
+ private val rackResolver = new CelebornRackResolver(conf)
private val statusSystem =
if (conf.haEnabled) {
- val sys = new HAMasterMetaManager(internalRpcEnvInUse, conf)
+ val sys = new HAMasterMetaManager(internalRpcEnvInUse, conf, rackResolver)
val handler = new MetaHandler(sys)
try {
handler.setUpMasterRatisServer(conf, masterArgs.masterClusterInfo.get)
@@ -115,7 +117,7 @@ private[celeborn] class Master(
}
sys
} else {
- new SingleMasterMetaManager(internalRpcEnvInUse, conf)
+ new SingleMasterMetaManager(internalRpcEnvInUse, conf, rackResolver)
}
// Threads
@@ -293,6 +295,7 @@ private[celeborn] class Master(
checkForHDFSRemnantDirsTimeOutTask.cancel(true)
}
forwardMessageThread.shutdownNow()
+ rackResolver.stop()
logInfo("Celeborn Master is stopped.")
}
diff --git a/master/src/main/scala/org/apache/celeborn/service/deploy/master/network/CelebornRackResolver.scala b/master/src/main/scala/org/apache/celeborn/service/deploy/master/network/CelebornRackResolver.scala
index 7e376f8a7..bd1af1526 100644
--- a/master/src/main/scala/org/apache/celeborn/service/deploy/master/network/CelebornRackResolver.scala
+++ b/master/src/main/scala/org/apache/celeborn/service/deploy/master/network/CelebornRackResolver.scala
@@ -17,21 +17,28 @@
package org.apache.celeborn.service.deploy.master.network
+import java.io.File
+import java.util.concurrent.{ScheduledExecutorService, TimeUnit}
+
import scala.collection.JavaConverters._
import scala.collection.mutable.ArrayBuffer
import com.google.common.base.Strings
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.CommonConfigurationKeysPublic
-import org.apache.hadoop.net.{CachedDNSToSwitchMapping, DNSToSwitchMapping, NetworkTopology, Node, NodeBase, ScriptBasedMapping}
+import org.apache.hadoop.net.{CachedDNSToSwitchMapping, DNSToSwitchMapping, NetworkTopology, Node, NodeBase, ScriptBasedMapping, TableMapping}
import org.apache.hadoop.util.ReflectionUtils
import org.apache.celeborn.common.CelebornConf
import org.apache.celeborn.common.internal.Logging
-import org.apache.celeborn.common.util.CelebornHadoopUtils
+import org.apache.celeborn.common.util.{CelebornHadoopUtils, ThreadUtils}
class CelebornRackResolver(celebornConf: CelebornConf) extends Logging {
+ private var rackResolveRefreshThreadPool: ScheduledExecutorService = _
+
+ private var rackResolveLastModifiedTime = 0L
+
private val dnsToSwitchMapping: DNSToSwitchMapping = {
val conf: Configuration = CelebornHadoopUtils.newConfiguration(celebornConf)
val dnsToSwitchMappingClass =
@@ -39,11 +46,60 @@ class CelebornRackResolver(celebornConf: CelebornConf) extends Logging {
CommonConfigurationKeysPublic.NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
classOf[ScriptBasedMapping],
classOf[DNSToSwitchMapping])
- ReflectionUtils.newInstance(dnsToSwitchMappingClass, conf)
- .asInstanceOf[DNSToSwitchMapping] match {
+ val switchMapping = ReflectionUtils.newInstance(dnsToSwitchMappingClass, conf)
+ .asInstanceOf[DNSToSwitchMapping]
+ val mapping = switchMapping match {
case c: CachedDNSToSwitchMapping => c
case o => new CachedDNSToSwitchMapping(o)
}
+ val refreshInterval = celebornConf.rackResolverRefreshInterval
+ rackResolveRefreshThreadPool =
+ ThreadUtils.newDaemonSingleThreadScheduledExecutor("master-rack-resolver-refresher")
+ val fileName = switchMapping match {
+ case _: ScriptBasedMapping =>
+ conf.get(CommonConfigurationKeysPublic.NET_TOPOLOGY_SCRIPT_FILE_NAME_KEY)
+ case _: TableMapping =>
+ conf.get(CommonConfigurationKeysPublic.NET_TOPOLOGY_TABLE_MAPPING_FILE_KEY)
+ case _ => null
+ }
+ if (fileName != null) {
+ var scriptFile = new File(fileName)
+ rackResolveLastModifiedTime = scriptFile.lastModified()
+ rackResolveRefreshThreadPool.scheduleWithFixedDelay(
+ new Runnable {
+ override def run(): Unit = {
+ scriptFile = new File(fileName)
+ if (scriptFile.canRead) {
+ val currentLastModifiedTime = scriptFile.lastModified()
+ if (currentLastModifiedTime != rackResolveLastModifiedTime) {
+ rackResolveLastModifiedTime = currentLastModifiedTime
+ mapping.reloadCachedMappings()
+ }
+ } else {
+ logWarning(s"Script file $fileName is not readable, reload cache directly")
+ mapping.reloadCachedMappings()
+ }
+ }
+ },
+ refreshInterval,
+ refreshInterval,
+ TimeUnit.MILLISECONDS)
+ } else {
+ rackResolveRefreshThreadPool.scheduleWithFixedDelay(
+ new Runnable {
+ override def run(): Unit = mapping.reloadCachedMappings()
+ },
+ refreshInterval,
+ refreshInterval,
+ TimeUnit.MILLISECONDS)
+ }
+ mapping
+ }
+
+ def stop(): Unit = {
+ if (null != rackResolveRefreshThreadPool) {
+ rackResolveRefreshThreadPool.shutdownNow()
+ }
}
def resolve(hostName: String): Node = {
diff --git a/master/src/test/scala/org/apache/celeborn/service/deploy/master/network/CelebornRackResolverSuite.scala b/master/src/test/scala/org/apache/celeborn/service/deploy/master/network/CelebornRackResolverSuite.scala
index 71337e6ae..3bf3da56c 100644
--- a/master/src/test/scala/org/apache/celeborn/service/deploy/master/network/CelebornRackResolverSuite.scala
+++ b/master/src/test/scala/org/apache/celeborn/service/deploy/master/network/CelebornRackResolverSuite.scala
@@ -20,8 +20,10 @@ package org.apache.celeborn.service.deploy.master.network
import java.io.{File, FileWriter}
import java.nio.charset.StandardCharsets
import java.util
+import java.util.concurrent.TimeUnit
import com.google.common.io.Files
+import com.google.common.util.concurrent.Uninterruptibles
import org.apache.hadoop.fs.CommonConfigurationKeysPublic.{NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY, NET_TOPOLOGY_TABLE_MAPPING_FILE_KEY}
import org.apache.hadoop.net.{Node, TableMapping}
import org.junit.Assert.assertEquals
@@ -105,4 +107,38 @@ class CelebornRackResolverSuite extends AnyFunSuite {
// check both side don't have rack info
assertEquals(true, resolver.isOnSameRack(hostName5, hostName6))
}
+
+ test("test TableMapping refresh") {
+ val hostName1 = "1.2.3.4"
+ val hostName2 = "1.2.3.5"
+ val hostName3 = "1.2.3.6"
+ val mapFile: File = File.createTempFile(getClass.getSimpleName + ".testResolve2", ".txt")
+ Files.asCharSink(mapFile, StandardCharsets.UTF_8).write(
+ s"""
+ |$hostName1 /default/rack1
+ |$hostName2 /default/rack1
+ |""".stripMargin)
+ val conf = new CelebornConf()
+ conf.set(
+ "celeborn.hadoop." + NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
+ classOf[TableMapping].getName)
+ conf.set("celeborn.hadoop." + NET_TOPOLOGY_TABLE_MAPPING_FILE_KEY, mapFile.getCanonicalPath)
+ conf.set(CelebornConf.RACKRESOLVER_REFRESH_INTERVAL.key, "50")
+ val resolver = new CelebornRackResolver(conf)
+ assertEquals("/default/rack1", resolver.resolve(hostName1).getNetworkLocation)
+ assertEquals("/default/rack1", resolver.resolve(hostName2).getNetworkLocation)
+ assertEquals(true, resolver.isOnSameRack(hostName1, hostName2))
+
+ Files.asCharSink(mapFile, StandardCharsets.UTF_8).write(
+ s"""
+ |$hostName1 /default/rack1
+ |$hostName2 /default/rack2
+ |$hostName3 /default/rack1
+ |""".stripMargin)
+
+ Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS)
+ assertEquals("/default/rack1", resolver.resolve(hostName3).getNetworkLocation)
+ assertEquals(false, resolver.isOnSameRack(hostName1, hostName2))
+ assertEquals(true, resolver.isOnSameRack(hostName1, hostName3))
+ }
}