(incubator-celeborn) branch main updated: [CELEBORN-1266] Improve log of current failed workers for WorkerStatusTracker

[ni...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

nicholasjiang 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 3f5b1de75 [CELEBORN-1266] Improve log of current failed workers for WorkerStatusTracker
3f5b1de75 is described below

commit 3f5b1de759ff0784c580029ff537a9e2057fdcf3
Author: SteNicholas <pr...@163.com>
AuthorDate: Sun Feb 18 18:58:20 2024 +0800

    [CELEBORN-1266] Improve log of current failed workers for WorkerStatusTracker
    
    ### What changes were proposed in this pull request?
    
    Improve log of current failed workers for `WorkerStatusTracker#recordWorkerFailure` and `WorkerStatusTracker#handleHeartbeatResponse`.
    
    ### Why are the changes needed?
    
    It's recommended to improve the log of current failed workers in `recordWorkerFailure` and `handleHeartbeatResponse` of `WorkerStatusTracker`. Meanwhile the log level of current failed workers could be warn.
    
    ### Does this PR introduce _any_ user-facing change?
    
    No.
    
    ### How was this patch tested?
    
    No.
    
    Closes #2290 from SteNicholas/CELEBORN-1266.
    
    Authored-by: SteNicholas <pr...@163.com>
    Signed-off-by: SteNicholas <pr...@163.com>
---
 .../celeborn/client/WorkerStatusTracker.scala      | 63 ++++++++++++++--------
 1 file changed, 40 insertions(+), 23 deletions(-)

diff --git a/client/src/main/scala/org/apache/celeborn/client/WorkerStatusTracker.scala b/client/src/main/scala/org/apache/celeborn/client/WorkerStatusTracker.scala
index 14a0df092..d7ccf0fe8 100644
--- a/client/src/main/scala/org/apache/celeborn/client/WorkerStatusTracker.scala
+++ b/client/src/main/scala/org/apache/celeborn/client/WorkerStatusTracker.scala
@@ -120,23 +120,15 @@ class WorkerStatusTracker(
 
   def recordWorkerFailure(failures: ShuffleFailedWorkers): Unit = {
     if (!failures.isEmpty) {
-      val failedWorker = new ShuffleFailedWorkers(failures)
-      val failedWorkerMsg = failedWorker.asScala.map { case (worker, (status, time)) =>
+      val failedWorkers = new ShuffleFailedWorkers(failures)
+      val failedWorkersMsg = failedWorkers.asScala.map { case (worker, (status, time)) =>
         s"${worker.readableAddress()}   ${status.name()}   ${Utils.formatTimestamp(time)}"
       }.mkString("\n")
-      val excludedWorkerMsg = excludedWorkers.asScala.map { case (worker, (status, time)) =>
-        s"${worker.readableAddress()}   ${status.name()}   ${Utils.formatTimestamp(time)}"
-      }.mkString("\n")
-      val shuttingDownMsg = shuttingWorkers.asScala.map(_.readableAddress()).mkString("\n")
-      logInfo(
+      logWarning(
         s"""
-           |Reporting failed worker:
-           |$failedWorkerMsg
-           |Current excluded worker:
-           |$excludedWorkerMsg
-           |Current shutting down worker:
-           |$shuttingDownMsg""".stripMargin)
-      failedWorker.asScala.foreach {
+           |Reporting failed workers:
+           |$failedWorkersMsg$currentFailedWorkers""".stripMargin)
+      failedWorkers.asScala.foreach {
         case (worker, (StatusCode.WORKER_SHUTDOWN, _)) =>
           shuttingWorkers.add(worker)
         case (worker, (statusCode, registerTime)) if !excludedWorkers.containsKey(worker) =>
@@ -213,16 +205,41 @@ class WorkerStatusTracker(
         }
       }
       if (statusChanged) {
-        logWarning("Worker status changed from application heartbeat response")
-        logInfo(
-          s"""
-             |Current excluded workers:
-             |${excludedWorkers.asScala.mkString("\n")}
-             |
-             |Current shutting down workers:
-             |${shuttingWorkers.asScala.map(_.readableAddress()).mkString("\n")}
-             |""".stripMargin)
+        logWarning(
+          s"Worker status changed from application heartbeat response.$currentFailedWorkers")
+      }
+    }
+  }
+
+  private def currentFailedWorkers: String = {
+    val excludedWorkersMsg =
+      excludedWorkers.asScala.groupBy(_._2._1).map { case (status, workers) =>
+        (
+          status,
+          workers.map { case (worker, (_, time)) =>
+            s"${worker.readableAddress()}   ${Utils.formatTimestamp(time)}"
+          }.mkString("\n"))
       }
+    val shutdownWorkersMsg = shuttingWorkers.asScala.map(_.readableAddress()).mkString("\n")
+    var failedWorkersMsg = ""
+    if (excludedWorkersMsg.contains(StatusCode.WORKER_EXCLUDED)) {
+      failedWorkersMsg +=
+        s"""
+           |Current excluded workers:
+           |${excludedWorkersMsg(StatusCode.WORKER_EXCLUDED)}""".stripMargin
+    }
+    if (excludedWorkersMsg.contains(StatusCode.WORKER_UNKNOWN)) {
+      failedWorkersMsg +=
+        s"""
+           |Current unknown workers:
+           |${excludedWorkersMsg(StatusCode.WORKER_UNKNOWN)}""".stripMargin
+    }
+    if (shutdownWorkersMsg.nonEmpty) {
+      failedWorkersMsg +=
+        s"""
+           |Current shutdown workers:
+           |$shutdownWorkersMsg""".stripMargin
     }
+    failedWorkersMsg
   }
 }