[incubator-celeborn] branch main updated: [CELEBORN-420] Add unavailablePeers to http request (#1345)

[an...@apache.org]

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 7d7279a9b [CELEBORN-420] Add unavailablePeers to http request (#1345)
7d7279a9b is described below

commit 7d7279a9bc8dd03f56f7da0f7756fc825b2fbbd1
Author: Angerszhuuuu <an...@gmail.com>
AuthorDate: Tue Mar 14 17:23:45 2023 +0800

    [CELEBORN-420] Add unavailablePeers to http request (#1345)
    
    * [CELEBORN-420] Add unavailablePeers to http request
---
 .../scala/org/apache/celeborn/service/deploy/master/Master.scala | 2 ++
 .../scala/org/apache/celeborn/server/common/HttpService.scala    | 2 ++
 .../apache/celeborn/server/common/http/HttpRequestHandler.scala  | 2 ++
 .../scala/org/apache/celeborn/service/deploy/worker/Worker.scala | 9 +++++++++
 4 files changed, 15 insertions(+)

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 5c0913cc3..1fc2a817f 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
@@ -775,6 +775,8 @@ private[celeborn] class Master(
 
   override def listPartitionLocationInfo: String = throw new UnsupportedOperationException()
 
+  override def getUnavailablePeers: String = throw new UnsupportedOperationException()
+
   private def requestGetWorkerInfos(endpoint: RpcEndpointRef): GetWorkerInfosResponse = {
     try {
       if (endpoint != null) {
diff --git a/service/src/main/scala/org/apache/celeborn/server/common/HttpService.scala b/service/src/main/scala/org/apache/celeborn/server/common/HttpService.scala
index 93f9b52be..b9adc578c 100644
--- a/service/src/main/scala/org/apache/celeborn/server/common/HttpService.scala
+++ b/service/src/main/scala/org/apache/celeborn/server/common/HttpService.scala
@@ -53,6 +53,8 @@ abstract class HttpService extends Service with Logging {
 
   def listPartitionLocationInfo: String
 
+  def getUnavailablePeers: String
+
   def startHttpServer(): Unit = {
     val handlers =
       if (metricsSystem.running) {
diff --git a/service/src/main/scala/org/apache/celeborn/server/common/http/HttpRequestHandler.scala b/service/src/main/scala/org/apache/celeborn/server/common/http/HttpRequestHandler.scala
index ac853d26d..3b55368f4 100644
--- a/service/src/main/scala/org/apache/celeborn/server/common/http/HttpRequestHandler.scala
+++ b/service/src/main/scala/org/apache/celeborn/server/common/http/HttpRequestHandler.scala
@@ -82,6 +82,8 @@ class HttpRequestHandler(
         service.listTopDiskUseApps
       case "/listPartitionLocationInfo" if service.serviceName == Service.WORKER =>
         service.listPartitionLocationInfo
+      case "/unavailablePeers" if service.serviceName == Service.WORKER =>
+        service.getUnavailablePeers
       case _ => INVALID
     }
   }
diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala
index 7f0e8de49..ff1208e62 100644
--- a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala
+++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala
@@ -475,6 +475,15 @@ private[celeborn] class Worker(
     partitionLocationInfo.toString
   }
 
+  override def getUnavailablePeers: String = {
+    val sb = new StringBuilder
+    sb.append("==========Unavailable Peers of Worker==========\n")
+    unavailablePeers.asScala.foreach { case (peer, time) =>
+      sb.append(s"${peer.toUniqueId().padTo(55, "")}$time\n");
+    }
+    sb.toString()
+  }
+
   @VisibleForTesting
   def isRegistered(): Boolean = {
     registered.get()