[incubator-celeborn] branch branch-0.2 updated: [CELEBORN-130] Correct CommitFilesTime metric (#1073) (#1076)

[zhouky]

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

zhouky pushed a commit to branch branch-0.2
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git


The following commit(s) were added to refs/heads/branch-0.2 by this push:
     new 3ab49ef8 [CELEBORN-130] Correct CommitFilesTime metric (#1073) (#1076)
3ab49ef8 is described below

commit 3ab49ef8fb4a392842bcd6d7d1aed9fa5ef04a85
Author: Keyong Zhou <zho...@apache.org>
AuthorDate: Tue Dec 13 20:08:26 2022 +0800

    [CELEBORN-130] Correct CommitFilesTime metric (#1073) (#1076)
---
 .../celeborn/service/deploy/worker/Controller.scala   | 19 ++++++++++---------
 1 file changed, 10 insertions(+), 9 deletions(-)

diff --git 
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala
 
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala
index 4290db04..8e2faffe 100644
--- 
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala
+++ 
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Controller.scala
@@ -108,15 +108,13 @@ private[deploy] class Controller(
 
     case CommitFiles(applicationId, shuffleId, masterIds, slaveIds, 
mapAttempts, epoch) =>
       val shuffleKey = Utils.makeShuffleKey(applicationId, shuffleId)
-      workerSource.sample(WorkerSource.CommitFilesTime, shuffleKey) {
-        logDebug(s"Received CommitFiles request, $shuffleKey, master files" +
-          s" ${masterIds.asScala.mkString(",")}; slave files 
${slaveIds.asScala.mkString(",")}.")
-        val commitFilesTimeMs = Utils.timeIt({
-          handleCommitFiles(context, shuffleKey, masterIds, slaveIds, 
mapAttempts, epoch)
-        })
-        logDebug(s"Done processed CommitFiles request with shuffleKey 
$shuffleKey, in " +
-          s"$commitFilesTimeMs ms.")
-      }
+      logDebug(s"Received CommitFiles request, $shuffleKey, master files" +
+        s" ${masterIds.asScala.mkString(",")}; slave files 
${slaveIds.asScala.mkString(",")}.")
+      val commitFilesTimeMs = Utils.timeIt({
+        handleCommitFiles(context, shuffleKey, masterIds, slaveIds, 
mapAttempts, epoch)
+      })
+      logDebug(s"Done processed CommitFiles request with shuffleKey 
$shuffleKey, in " +
+        s"$commitFilesTimeMs ms.")
 
     case GetWorkerInfos =>
       handleGetWorkerInfos(context)
@@ -380,6 +378,7 @@ private[deploy] class Controller(
       } else {
         logInfo(s"Start commitFiles for ${shuffleKey}")
         commitInfo.status = CommitInfo.COMMIT_INPROCESS
+        workerSource.startTimer(WorkerSource.CommitFilesTime, shuffleKey)
       }
     }
 
@@ -498,6 +497,8 @@ private[deploy] class Controller(
         commitInfo.status = CommitInfo.COMMIT_FINISHED
       }
       context.reply(response)
+
+      workerSource.stopTimer(WorkerSource.CommitFilesTime, shuffleKey)
     }
 
     if (future != null) {