This is an automated email from the ASF dual-hosted git repository.
zhouky 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 5d7892988 [CELEBORN-1012] Add a dedicated internal port in Master to
talk to Workers and other Masters
5d7892988 is described below
commit 5d7892988a0e232a76e8df0ab7392248ab6e1f21
Author: Chandni Singh <[email protected]>
AuthorDate: Tue Jan 30 09:53:35 2024 +0800
[CELEBORN-1012] Add a dedicated internal port in Master to talk to Workers
and other Masters
### What changes were proposed in this pull request?
With authentication
([CELEBORN-1011](https://issues.apache.org/jira/browse/CELEBORN-1011)), the
handling of messages from Client by Celeborn services (Master/workers) will go
through a SASL handshake. However, messages exchanged between Masters and
Workers will not.
With a single netty server on the Master/Workers handling both client and
master/workers messages, differentiating between the two types of connection is
a challenge. It will be better if Master/Workers have a separate designated
port for Clients and a separate one for internal components (workers and other
Masters).
In this change, we propose
- the config that enables creating dedicated internal ports on
Masters/Workers.
- creation of the dedicated internal port in just the Master. A subsequent
PR will add that creation of the dedicated internal port in Workers.
### Why are the changes needed?
This change is required for adding authentication.
([CELEBORN-1011](https://issues.apache.org/jira/browse/CELEBORN-1011)).
### Does this PR introduce _any_ user-facing change?
Yes, there are new configurations added.
### How was this patch tested?
Added a UT
Closes #2265 from otterc/CELEBORN-1012.
Authored-by: Chandni Singh <[email protected]>
Signed-off-by: zky.zhoukeyong <[email protected]>
---
.../celeborn/common/protocol/RpcNameConstants.java | 2 +
.../org/apache/celeborn/common/CelebornConf.scala | 45 ++++++++++++++++++
docs/configuration/ha.md | 1 +
docs/configuration/master.md | 2 +
docs/configuration/worker.md | 1 +
.../deploy/master/InternalRpcEndpoint.scala | 54 ++++++++++++++++++++++
.../celeborn/service/deploy/master/Master.scala | 44 ++++++++++++++++--
.../service/deploy/master/MasterArguments.scala | 10 ++++
.../service/deploy/master/MasterSuite.scala | 36 +++++++++++++++
9 files changed, 192 insertions(+), 3 deletions(-)
diff --git
a/common/src/main/java/org/apache/celeborn/common/protocol/RpcNameConstants.java
b/common/src/main/java/org/apache/celeborn/common/protocol/RpcNameConstants.java
index c47f04f98..6c3898fdc 100644
---
a/common/src/main/java/org/apache/celeborn/common/protocol/RpcNameConstants.java
+++
b/common/src/main/java/org/apache/celeborn/common/protocol/RpcNameConstants.java
@@ -20,8 +20,10 @@ package org.apache.celeborn.common.protocol;
public class RpcNameConstants {
// For Master
public static String MASTER_SYS = "Master";
+ public static String MASTER_INTERNAL_SYS = "MasterInternal";
// Master Endpoint Name
public static String MASTER_EP = "MasterEndpoint";
+ public static String MASTER_INTERNAL_EP = "MasterInternalEndpoint";
// For Worker
public static String WORKER_SYS = "Worker";
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 887be0830..713bcc1d8 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -1117,6 +1117,19 @@ class CelebornConf(loadDefaults: Boolean) extends
Cloneable with Logging with Se
// Authentication //
// //////////////////////////////////////////////////////
def authEnabled: Boolean = get(AUTH_ENABLED)
+
+ // //////////////////////////////////////////////////////
+ // Internal Port //
+ // //////////////////////////////////////////////////////
+ def internalPortEnabled: Boolean = get(INTERNAL_PORT_ENABLED)
+
+ 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)
+ getInt(key, getInt(legacyKey,
HA_MASTER_NODE_INTERNAL_PORT.defaultValue.get))
+ }
+
+ def masterInternalPort: Int = get(MASTER_INTERNAL_PORT)
}
object CelebornConf extends Logging {
@@ -4381,4 +4394,36 @@ object CelebornConf extends Logging {
.doc("Whether to enable authentication.")
.booleanConf
.createWithDefault(false)
+
+ val INTERNAL_PORT_ENABLED: ConfigEntry[Boolean] =
+ buildConf("celeborn.internal.port.enabled")
+ .categories("master", "worker")
+ .version("0.5.0")
+ .doc("Whether to create a internal port on Masters/Workers for " +
+ "inter-Masters/Workers communication. This is beneficial when SASL
authentication " +
+ "is enforced for all interactions between clients and Celeborn
Services, but the services " +
+ "can exchange messages without being subject to SASL authentication.")
+ .booleanConf
+ .createWithDefault(false)
+
+ val MASTER_INTERNAL_PORT: ConfigEntry[Int] =
+ buildConf("celeborn.master.internal.port")
+ .categories("master")
+ .version("0.5.0")
+ .doc(
+ "Internal port on the master where both workers and other master nodes
connect.")
+ .intConf
+ .checkValue(p => p >= 1024 && p < 65535, "Invalid port")
+ .createWithDefault(8097)
+
+ val HA_MASTER_NODE_INTERNAL_PORT: ConfigEntry[Int] =
+ buildConf("celeborn.master.ha.node.<id>.internal.port")
+ .categories("ha")
+ .doc(
+ "Internal port for the workers and other masters to bind to a master
node <id> in HA mode.")
+ .version("0.5.0")
+ .intConf
+ .checkValue(p => p >= 1024 && p < 65535, "Invalid port")
+ .createWithDefault(8097)
+
}
diff --git a/docs/configuration/ha.md b/docs/configuration/ha.md
index a6b2ea53f..faef0916e 100644
--- a/docs/configuration/ha.md
+++ b/docs/configuration/ha.md
@@ -21,6 +21,7 @@ license: |
| --- | ------- | ----------- | ----- | ---------- |
| celeborn.master.ha.enabled | false | When true, master nodes run as Raft
cluster mode. | 0.3.0 | celeborn.ha.enabled |
| celeborn.master.ha.node.<id>.host | <required> | Host to bind of
master node <id> in HA mode. | 0.3.0 | celeborn.ha.master.node.<id>.host
|
+| celeborn.master.ha.node.<id>.internal.port | 8097 | Internal port for
the workers and other masters to bind to a master node <id> in HA mode. | 0.5.0
| |
| celeborn.master.ha.node.<id>.port | 9097 | Port to bind of master node
<id> in HA mode. | 0.3.0 | celeborn.ha.master.node.<id>.port |
| celeborn.master.ha.node.<id>.ratis.port | 9872 | Ratis port to bind of
master node <id> in HA mode. | 0.3.0 |
celeborn.ha.master.node.<id>.ratis.port |
| celeborn.master.ha.ratis.raft.rpc.type | netty | RPC type for Ratis,
available options: netty, grpc. | 0.3.0 |
celeborn.ha.master.ratis.raft.rpc.type |
diff --git a/docs/configuration/master.md b/docs/configuration/master.md
index c80d324c3..416270f6a 100644
--- a/docs/configuration/master.md
+++ b/docs/configuration/master.md
@@ -21,6 +21,7 @@ license: |
| --- | ------- | ----------- | ----- | ---------- |
| celeborn.dynamicConfig.refresh.interval | 120s | Interval for refreshing the
corresponding dynamic config periodically. | 0.4.0 | |
| celeborn.dynamicConfig.store.backend | NONE | Store backend for dynamic
config. Available options: NONE, FS. Note: NONE means disabling dynamic config
store. | 0.4.0 | |
+| celeborn.internal.port.enabled | false | Whether to create a internal port
on Masters/Workers for inter-Masters/Workers communication. This is beneficial
when SASL authentication is enforced for all interactions between clients and
Celeborn Services, but the services can exchange messages without being subject
to SASL authentication. | 0.5.0 | |
| celeborn.master.estimatedPartitionSize.initialSize | 64mb | Initial
partition size for estimation, it will change according to runtime stats. |
0.3.0 | celeborn.shuffle.initialEstimatedPartitionSize |
| celeborn.master.estimatedPartitionSize.update.initialDelay | 5min | Initial
delay time before start updating partition size for estimation. | 0.3.0 |
celeborn.shuffle.estimatedPartitionSize.update.initialDelay |
| celeborn.master.estimatedPartitionSize.update.interval | 10min | Interval of
updating partition size for estimation. | 0.3.0 |
celeborn.shuffle.estimatedPartitionSize.update.interval |
@@ -30,6 +31,7 @@ license: |
| celeborn.master.host | <localhost> | Hostname for master to bind. |
0.2.0 | |
| celeborn.master.http.host | <localhost> | Master's http host. | 0.4.0
|
celeborn.metrics.master.prometheus.host,celeborn.master.metrics.prometheus.host
|
| 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.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 |
diff --git a/docs/configuration/worker.md b/docs/configuration/worker.md
index a6773fbfd..7cef6bdbe 100644
--- a/docs/configuration/worker.md
+++ b/docs/configuration/worker.md
@@ -21,6 +21,7 @@ license: |
| --- | ------- | ----------- | ----- | ---------- |
| celeborn.dynamicConfig.refresh.interval | 120s | Interval for refreshing the
corresponding dynamic config periodically. | 0.4.0 | |
| celeborn.dynamicConfig.store.backend | NONE | Store backend for dynamic
config. Available options: NONE, FS. Note: NONE means disabling dynamic config
store. | 0.4.0 | |
+| celeborn.internal.port.enabled | false | Whether to create a internal port
on Masters/Workers for inter-Masters/Workers communication. This is beneficial
when SASL authentication is enforced for all interactions between clients and
Celeborn Services, but the services can exchange messages without being subject
to SASL authentication. | 0.5.0 | |
| celeborn.master.endpoints | <localhost>:9097 | Endpoints of master
nodes for celeborn client to connect, allowed pattern is:
`<host1>:<port1>[,<host2>:<port2>]*`, e.g. `clb1:9097,clb2:9098,clb3:9099`. If
the port is omitted, 9097 will be used. | 0.2.0 | |
| celeborn.master.estimatedPartitionSize.minSize | 8mb | Ignore partition size
smaller than this configuration of partition size for estimation. | 0.3.0 |
celeborn.shuffle.minPartitionSizeToEstimate |
| celeborn.shuffle.chunk.size | 8m | Max chunk size of reducer's merged
shuffle data. For example, if a reducer's shuffle data is 128M and the data
will need 16 fetch chunk requests to fetch. | 0.2.0 | |
diff --git
a/master/src/main/scala/org/apache/celeborn/service/deploy/master/InternalRpcEndpoint.scala
b/master/src/main/scala/org/apache/celeborn/service/deploy/master/InternalRpcEndpoint.scala
new file mode 100644
index 000000000..92ece136d
--- /dev/null
+++
b/master/src/main/scala/org/apache/celeborn/service/deploy/master/InternalRpcEndpoint.scala
@@ -0,0 +1,54 @@
+/*
+ * 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.celeborn.service.deploy.master
+
+import org.apache.celeborn.common.CelebornConf
+import org.apache.celeborn.common.internal.Logging
+import org.apache.celeborn.common.rpc._
+
+/**
+ * Internal RPC endpoint used by the Master to communicate with the Workers
and other Masters when using a dedicated
+ * internal port.
+ */
+private[celeborn] class InternalRpcEndpoint(
+ val master: Master,
+ override val rpcEnv: RpcEnv,
+ val conf: CelebornConf)
+ extends RpcEndpoint with Logging {
+
+ // start threads to check timeout for workers and applications
+ override def onStart(): Unit = {
+ master.onStart()
+ }
+
+ override def onStop(): Unit = {
+ master.onStop()
+ }
+
+ override def onDisconnected(address: RpcAddress): Unit = {
+ logDebug(s"Client $address got disconnected.")
+ }
+
+ override def receive: PartialFunction[Any, Unit] = {
+ master.receive
+ }
+
+ override def receiveAndReply(context: RpcCallContext): PartialFunction[Any,
Unit] = {
+ master.receiveAndReply(context)
+ }
+}
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 4a0df0a1b..b193518fb 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
@@ -21,6 +21,7 @@ import java.io.IOException
import java.net.BindException
import java.util
import java.util.concurrent.{ConcurrentHashMap, ScheduledFuture, TimeUnit}
+import java.util.concurrent.atomic.AtomicBoolean
import java.util.function.ToLongFunction
import scala.collection.JavaConverters._
@@ -42,7 +43,6 @@ import
org.apache.celeborn.common.protocol.message.{ControlMessages, StatusCode}
import org.apache.celeborn.common.protocol.message.ControlMessages._
import org.apache.celeborn.common.quota.{QuotaManager, ResourceConsumption}
import org.apache.celeborn.common.rpc._
-import org.apache.celeborn.common.rpc.netty.NettyRpcEnv
import org.apache.celeborn.common.util.{CelebornHadoopUtils, JavaUtils,
PbSerDeUtils, ThreadUtils, Utils}
import org.apache.celeborn.server.common.{HttpService, Service}
import
org.apache.celeborn.service.deploy.master.clustermeta.SingleMasterMetaManager
@@ -79,9 +79,25 @@ private[celeborn] class Master(
conf,
Math.max(64, Runtime.getRuntime.availableProcessors()))
+ // Visible for testing
+ private[master] var internalRpcEnvInUse = rpcEnv
+
+ if (conf.internalPortEnabled) {
+ val internalRpcEnv: RpcEnv = RpcEnv.create(
+ RpcNameConstants.MASTER_INTERNAL_SYS,
+ masterArgs.host,
+ masterArgs.host,
+ masterArgs.internalPort,
+ conf,
+ Math.max(64, Runtime.getRuntime.availableProcessors()))
+ logInfo(
+ s"Internal port enabled, using internal port ${masterArgs.internalPort}
for internal RPC.")
+ internalRpcEnvInUse = internalRpcEnv
+ }
+
private val statusSystem =
if (conf.haEnabled) {
- val sys = new HAMasterMetaManager(rpcEnv, conf)
+ val sys = new HAMasterMetaManager(internalRpcEnvInUse, conf)
val handler = new MetaHandler(sys)
try {
handler.setUpMasterRatisServer(conf, masterArgs.masterClusterInfo.get)
@@ -99,7 +115,7 @@ private[celeborn] class Master(
}
sys
} else {
- new SingleMasterMetaManager(rpcEnv, conf)
+ new SingleMasterMetaManager(internalRpcEnvInUse, conf)
}
// Threads
@@ -198,10 +214,23 @@ private[celeborn] class Master(
}
masterSource.addGauge(MasterSource.IS_ACTIVE_MASTER) { () => isMasterActive }
+ private val threadsStarted: AtomicBoolean = new AtomicBoolean(false)
rpcEnv.setupEndpoint(RpcNameConstants.MASTER_EP, this)
+ // Visible for testing
+ private[master] var internalRpcEndpoint: RpcEndpoint = _
+ private var internalRpcEndpointRef: RpcEndpointRef = _
+ if (conf.internalPortEnabled) {
+ internalRpcEndpoint = new InternalRpcEndpoint(this, internalRpcEnvInUse,
conf)
+ internalRpcEndpointRef = internalRpcEnvInUse.setupEndpoint(
+ RpcNameConstants.MASTER_INTERNAL_EP,
+ internalRpcEndpoint)
+ }
// start threads to check timeout for workers and applications
override def onStart(): Unit = {
+ if (!threadsStarted.compareAndSet(false, true)) {
+ return
+ }
checkForWorkerTimeOutTask = forwardMessageThread.scheduleWithFixedDelay(
new Runnable {
override def run(): Unit = Utils.tryLogNonFatalError {
@@ -247,6 +276,9 @@ private[celeborn] class Master(
}
override def onStop(): Unit = {
+ if (!threadsStarted.compareAndSet(true, false)) {
+ return
+ }
logInfo("Stopping Celeborn Master.")
if (checkForWorkerTimeOutTask != null) {
checkForWorkerTimeOutTask.cancel(true)
@@ -1084,12 +1116,18 @@ private[celeborn] class Master(
super.initialize()
logInfo("Master started.")
rpcEnv.awaitTermination()
+ if (conf.internalPortEnabled) {
+ internalRpcEnvInUse.awaitTermination()
+ }
}
override def stop(exitKind: Int): Unit = synchronized {
if (!stopped) {
logInfo("Stopping Master")
rpcEnv.stop(self)
+ if (conf.internalPortEnabled) {
+ internalRpcEnvInUse.stop(internalRpcEndpointRef)
+ }
super.stop(exitKind)
logInfo("Master stopped.")
stopped = true
diff --git
a/master/src/main/scala/org/apache/celeborn/service/deploy/master/MasterArguments.scala
b/master/src/main/scala/org/apache/celeborn/service/deploy/master/MasterArguments.scala
index 195647463..738d582e4 100644
---
a/master/src/main/scala/org/apache/celeborn/service/deploy/master/MasterArguments.scala
+++
b/master/src/main/scala/org/apache/celeborn/service/deploy/master/MasterArguments.scala
@@ -27,6 +27,7 @@ class MasterArguments(args: Array[String], conf:
CelebornConf) {
private var _host: Option[String] = None
private var _port: Option[Int] = None
+ private var _internalPort: Option[Int] = None
private var _propertiesFile: Option[String] = None
private var _masterClusterInfo: Option[MasterClusterInfo] = None
@@ -43,16 +44,20 @@ class MasterArguments(args: Array[String], conf:
CelebornConf) {
val localNode = clusterInfo.localNode
_host = _host.orElse(Some(conf.haMasterNodeHost(localNode.nodeId)))
_port = _port.orElse(Some(conf.haMasterNodePort(localNode.nodeId)))
+ _internalPort =
_internalPort.orElse(Some(conf.haMasterNodeInternalPort(localNode.nodeId)))
_masterClusterInfo = Some(clusterInfo)
} else {
_host = _host.orElse(Some(conf.masterHost))
_port = _port.orElse(Some(conf.masterPort))
+ _internalPort = _internalPort.orElse(Some(conf.masterInternalPort))
}
def host: String = _host.get
def port: Int = _port.get
+ def internalPort: Int = _internalPort.get
+
def masterClusterInfo: Option[MasterClusterInfo] = _masterClusterInfo
@tailrec
@@ -66,6 +71,10 @@ class MasterArguments(args: Array[String], conf:
CelebornConf) {
_port = Some(value)
parse(tail)
+ case ("--internal-port") :: IntParam(value) :: tail =>
+ _internalPort = Some(value)
+ parse(tail)
+
case "--properties-file" :: value :: tail =>
_propertiesFile = Some(value)
parse(tail)
@@ -90,6 +99,7 @@ class MasterArguments(args: Array[String], conf:
CelebornConf) {
|Options:
| -h HOST, --host HOST Hostname to listen on
| -p PORT, --port PORT Port to listen on (default: 9097)
+ | --internal-port PORT Internal port for internal communication
(default: 8097)
| --properties-file FILE Path to a custom Celeborn properties file,
| default is conf/celeborn-defaults.conf.
|""".stripMargin)
diff --git
a/master/src/test/scala/org/apache/celeborn/service/deploy/master/MasterSuite.scala
b/master/src/test/scala/org/apache/celeborn/service/deploy/master/MasterSuite.scala
index 1031dad1d..2d3e66912 100644
---
a/master/src/test/scala/org/apache/celeborn/service/deploy/master/MasterSuite.scala
+++
b/master/src/test/scala/org/apache/celeborn/service/deploy/master/MasterSuite.scala
@@ -18,11 +18,13 @@
package org.apache.celeborn.service.deploy.master
import com.google.common.io.Files
+import org.mockito.Mockito.mock
import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
import org.scalatest.funsuite.AnyFunSuite
import org.apache.celeborn.common.CelebornConf
import org.apache.celeborn.common.internal.Logging
+import org.apache.celeborn.common.protocol.{PbCheckForWorkerTimeout,
PbRegisterWorker}
import org.apache.celeborn.common.util.{CelebornExitKind, Utils}
class MasterSuite extends AnyFunSuite
@@ -60,4 +62,38 @@ class MasterSuite extends AnyFunSuite
master.stop(CelebornExitKind.EXIT_IMMEDIATELY)
master.rpcEnv.shutdown()
}
+
+ test("test dedicated internal port receives") {
+ val conf = new CelebornConf()
+ conf.set(CelebornConf.HA_ENABLED.key, "false")
+ conf.set(CelebornConf.HA_MASTER_RATIS_STORAGE_DIR.key, getTmpDir())
+ conf.set(CelebornConf.WORKER_STORAGE_DIRS.key, getTmpDir())
+ conf.set(CelebornConf.METRICS_ENABLED.key, "true")
+ conf.set(CelebornConf.INTERNAL_PORT_ENABLED.key, "true")
+
+ val args = Array("-h", "localhost", "-p", "9097", "--internal-port",
"8097")
+
+ val masterArgs = new MasterArguments(args, conf)
+ val master = new Master(conf, masterArgs)
+ new Thread() {
+ override def run(): Unit = {
+ master.initialize()
+ }
+ }.start()
+ Thread.sleep(5000L)
+ master.receive.applyOrElse(
+ PbCheckForWorkerTimeout.newBuilder().build(),
+ (_: Any) => fail("Unexpected message"))
+ master.internalRpcEndpoint.receive.applyOrElse(
+ PbCheckForWorkerTimeout.newBuilder().build(),
+ (_: Any) => fail("Unexpected message"))
+
+ master.internalRpcEndpoint.receiveAndReply(
+
mock(classOf[org.apache.celeborn.common.rpc.RpcCallContext])).applyOrElse(
+ PbRegisterWorker.newBuilder().build(),
+ (_: Any) => fail("Unexpected message"))
+ master.stop(CelebornExitKind.EXIT_IMMEDIATELY)
+ master.rpcEnv.shutdown()
+ master.internalRpcEnvInUse.shutdown()
+ }
}
