[CELEBORN-747][FOLLOWUP] avgFlushTime and avgFetchTime are in nano se…

…conds

<!--
Thanks for sending a pull request!  Here are some tips for you:
  - Make sure the PR title start w/ a JIRA ticket, e.g. '[CELEBORN-XXXX] Your PR title ...'.
  - Be sure to keep the PR description updated to reflect all changes.
  - Please write your PR title to summarize what this PR proposes.
  - If possible, provide a concise example to reproduce the issue for a faster review.
-->

### What changes were proposed in this pull request?

`avgFlushTime` and `avgFetchTime` are in nano seconds, it was accidentally formatted by `msDurationToString` and caused unreasonable logs.

```
usableSpace: 1602.1 GiB, avgFlushTime: 1.35 h, avgFetchTime: 1187.66 h
```

### Why are the changes needed?

Fix logs.

### Does this PR introduce _any_ user-facing change?

No.

### How was this patch tested?

UT is updated

Closes apache#1687 from pan3793/CELEBORN-747-followup.

Authored-by: Cheng Pan <[email protected]>
Signed-off-by: Cheng Pan <[email protected]>

common/src/main/scala/org/apache/celeborn/common/meta/DeviceInfo.scala

@@ -34,9 +34,8 @@ import org.apache.celeborn.common.util.Utils.runCommand
 class DiskInfo(
     val mountPoint: String,
     var actualUsableSpace: Long,
-    // avgFlushTime is nano seconds
-    var avgFlushTime: Long,
-    var avgFetchTime: Long,
+    var avgFlushTime: Long, // in nano seconds
+    var avgFetchTime: Long, // in nano seconds
     var activeSlots: Long,
     val dirs: List[File],
     val deviceInfo: DeviceInfo) extends Serializable with Logging {
@@ -135,8 +134,8 @@ class DiskInfo(
       s" shuffleAllocations: $nonEmptyShuffles," +
       s" mountPoint: $mountPoint," +
       s" usableSpace: ${Utils.bytesToString(actualUsableSpace)}," +
-      s" avgFlushTime: ${Utils.msDurationToString(avgFlushTime)}," +
-      s" avgFetchTime: ${Utils.msDurationToString(avgFetchTime)}," +
+      s" avgFlushTime: ${Utils.nanoDurationToString(avgFlushTime)}," +
+      s" avgFetchTime: ${Utils.nanoDurationToString(avgFetchTime)}," +
       s" activeSlots: $activeSlots)" +
       s" status: $status" +
       s" dirs ${dirs.mkString("\t")}"

common/src/main/scala/org/apache/celeborn/common/util/Utils.scala

@@ -145,6 +145,27 @@ object Utils extends Logging {
     }
   }
 
+  def nanoDurationToString(ns: Long): String = {
+    val ms = 1000 * 1000
+    val second = 1000 * ms
+    val minute = 60 * second
+    val hour = 60 * minute
+    val locale = Locale.US
+
+    ns match {
+      case t if t < ms =>
+        "%d ns".formatLocal(locale, t)
+      case t if t < second =>
+        "%d ms".formatLocal(locale, t)
+      case t if t < minute =>
+        "%.1f s".formatLocal(locale, t.toFloat / second)
+      case t if t < hour =>
+        "%.1f m".formatLocal(locale, t.toFloat / minute)
+      case t =>
+        "%.2f h".formatLocal(locale, t.toFloat / hour)
+    }
+  }
+
   @throws(classOf[CelebornException])
   def extractHostPortFromCelebornUrl(celebornUrl: String): (String, Int) = {
     try {

common/src/test/scala/org/apache/celeborn/common/meta/WorkerInfoSuite.scala

@@ -283,9 +283,9 @@ class WorkerInfoSuite extends CelebornFunSuite {
          |SlotsUsed: 60
          |LastHeartbeat: 0
          |Disks: $placeholder
-         |  DiskInfo0: DiskInfo(maxSlots: 0, committed shuffles 0 shuffleAllocations: Map(), mountPoint: disk3, usableSpace: 2048.0 MiB, avgFlushTime: 3 ms, avgFetchTime: 3 ms, activeSlots: 30) status: HEALTHY dirs $placeholder
-         |  DiskInfo1: DiskInfo(maxSlots: 0, committed shuffles 0 shuffleAllocations: Map(), mountPoint: disk1, usableSpace: 2048.0 MiB, avgFlushTime: 1 ms, avgFetchTime: 1 ms, activeSlots: 10) status: HEALTHY dirs $placeholder
-         |  DiskInfo2: DiskInfo(maxSlots: 0, committed shuffles 0 shuffleAllocations: Map(), mountPoint: disk2, usableSpace: 2048.0 MiB, avgFlushTime: 2 ms, avgFetchTime: 2 ms, activeSlots: 20) status: HEALTHY dirs $placeholder
+         |  DiskInfo0: DiskInfo(maxSlots: 0, committed shuffles 0 shuffleAllocations: Map(), mountPoint: disk3, usableSpace: 2048.0 MiB, avgFlushTime: 3 ns, avgFetchTime: 3 ns, activeSlots: 30) status: HEALTHY dirs $placeholder
+         |  DiskInfo1: DiskInfo(maxSlots: 0, committed shuffles 0 shuffleAllocations: Map(), mountPoint: disk1, usableSpace: 2048.0 MiB, avgFlushTime: 1 ns, avgFetchTime: 1 ns, activeSlots: 10) status: HEALTHY dirs $placeholder
+         |  DiskInfo2: DiskInfo(maxSlots: 0, committed shuffles 0 shuffleAllocations: Map(), mountPoint: disk2, usableSpace: 2048.0 MiB, avgFlushTime: 2 ns, avgFetchTime: 2 ns, activeSlots: 20) status: HEALTHY dirs $placeholder
          |UserResourceConsumption: $placeholder
          |  UserIdentifier: `tenant1`.`name1`, ResourceConsumption: ResourceConsumption(diskBytesWritten: 20.0 MiB, diskFileCount: 1, hdfsBytesWritten: 50.0 MiB, hdfsFileCount: 1)
          |WorkerRef: null