This is an automated email from the ASF dual-hosted git repository. zhouky pushed a commit to branch branch-0.3 in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
commit 26ad470a22c498d3a4e9093d082c85d4e4a09262 Author: caojiaqing <[email protected]> AuthorDate: Tue Aug 1 21:21:13 2023 +0800 [CELEBORN-852] Adding new metrics to record the number of registered … Adding new metrics to record the number of registered connections Monitor the number of active connections on worker nodes no no Closes #1773 from JQ-Cao/852. Authored-by: caojiaqing <[email protected]> Signed-off-by: zky.zhoukeyong <[email protected]> --- .../celeborn/service/deploy/worker/FetchHandler.scala | 2 ++ .../service/deploy/worker/PushDataHandler.scala | 17 +++++++++++++++++ .../celeborn/service/deploy/worker/WorkerSource.scala | 3 +++ 3 files changed, 22 insertions(+) diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/FetchHandler.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/FetchHandler.scala index 389bf5bb5..624c44655 100644 --- a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/FetchHandler.scala +++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/FetchHandler.scala @@ -349,10 +349,12 @@ class FetchHandler(val conf: CelebornConf, val transportConf: TransportConf) /** Invoked when the channel associated with the given client is active. */ override def channelActive(client: TransportClient): Unit = { logDebug(s"channel active ${client.getSocketAddress}") + workerSource.incCounter(WorkerSource.ACTIVE_CONNECTION_COUNT) super.channelActive(client) } override def channelInactive(client: TransportClient): Unit = { + workerSource.incCounter(WorkerSource.ACTIVE_CONNECTION_COUNT, -1) creditStreamManager.connectionTerminated(client.getChannel) logDebug(s"channel inactive ${client.getSocketAddress}") } diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala index 290fcb3a6..32da0ed6a 100644 --- a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala +++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/PushDataHandler.scala @@ -1128,6 +1128,23 @@ class PushDataHandler extends BaseMessageHandler with Logging { pushClientFactory.createClient(host, port, partitionId) } } + + /** + * Invoked when the channel associated with the given client is active. + */ + override def channelActive(client: TransportClient): Unit = { + workerSource.incCounter(WorkerSource.ACTIVE_CONNECTION_COUNT) + super.channelActive(client) + } + + /** + * Invoked when the channel associated with the given client is inactive. + * No further requests will come from this client. + */ + override def channelInactive(client: TransportClient): Unit = { + workerSource.incCounter(WorkerSource.ACTIVE_CONNECTION_COUNT, -1) + super.channelInactive(client) + } } object PushDataHandler { diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/WorkerSource.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/WorkerSource.scala index 9f3d02d5f..e1f247a08 100644 --- a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/WorkerSource.scala +++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/WorkerSource.scala @@ -36,6 +36,7 @@ class WorkerSource(conf: CelebornConf) extends AbstractSource(conf, MetricsSyste addCounter(PUSH_DATA_HANDSHAKE_FAIL_COUNT) addCounter(REGION_START_FAIL_COUNT) addCounter(REGION_FINISH_FAIL_COUNT) + addCounter(ACTIVE_CONNECTION_COUNT) // add Timers addTimer(COMMIT_FILES_TIME) @@ -98,6 +99,8 @@ object WorkerSource { // slots val SLOTS_ALLOCATED = "SlotsAllocated" + val ACTIVE_CONNECTION_COUNT = "ActiveConnectionCount" + // memory val NETTY_MEMORY = "NettyMemory" val SORT_TIME = "SortTime"
