[CELEBORN-1092] Introduce JVM monitoring in Celeborn Worker using JVM…

…Quake

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

Introduce JVM monitoring in Celeborn Worker using JVMQuake to enable early detection of memory management issues and facilitate fast failure.

### Why are the changes needed?

When facing out-of-control memory management in Celeborn Worker we typically use JVMkill as a remedy by killing the process and generating a heap dump for post-analysis. However, even with jvmkill protection, we may still encounter issues caused by JVM running out of memory, such as repeated execution of Full GC without performing any useful work during the pause time. Since the JVM does not exhaust 100% of resources, JVMkill will not be triggered. Therefore JVMQuake is introduced to provide more granular monitoring of GC behavior, enabling early detection of memory management issues and facilitating fast failure. Refers to the principle of [jvmquake](https://github.com/Netflix-Skunkworks/jvmquake) which is a JVMTI agent that attaches to your JVM and automatically signals and kills it when the program has become unstable.

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

No.

### How was this patch tested?

`JVMQuakeSuite`

Closes #2061 from SteNicholas/CELEBORN-1092.

Authored-by: SteNicholas <[email protected]>
Signed-off-by: Fu Chen <[email protected]>

LICENSE-binary

@@ -304,6 +304,8 @@ javax.servlet:javax.servlet-api
 
 MIT License
 ------------
+See license/LISENCE-jdktools.txt for details.
+com.github.olivergondza:maven-jdk-tools-wrapper
 See license/LICENSE-slf4j.txt for details.
 org.slf4j:jul-to-slf4j
 org.slf4j:slf4j-api

common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala

@@ -689,6 +689,21 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
   def workerFetchHeartbeatEnabled: Boolean = get(WORKER_FETCH_HEARTBEAT_ENABLED)
   def workerPartitionSplitEnabled: Boolean = get(WORKER_PARTITION_SPLIT_ENABLED)
   def workerActiveConnectionMax: Option[Long] = get(WORKER_ACTIVE_CONNECTION_MAX)
+  def workerJvmQuakeEnabled: Boolean = get(WORKER_JVM_QUAKE_ENABLED)
+  def workerJvmQuakeCheckInterval: Long = get(WORKER_JVM_QUAKE_CHECK_INTERVAL)
+  def workerJvmQuakeRuntimeWeight: Double = get(WORKER_JVM_QUAKE_RUNTIME_WEIGHT)
+  def workerJvmQuakeDumpEnabled: Boolean = get(WORKER_JVM_QUAKE_DUMP_ENABLED)
+  def workerJvmQuakeDumpPath: String = get(WORKER_JVM_QUAKE_DUMP_PATH)
+
+  def workerJvmQuakeDumpThreshold: Duration =
+    getTimeAsMs(
+      WORKER_JVM_QUAKE_DUMP_THRESHOLD.key,
+      WORKER_JVM_QUAKE_DUMP_THRESHOLD.defaultValueString).microsecond
+  def workerJvmQuakeKillThreshold: Duration =
+    getTimeAsMs(
+      WORKER_JVM_QUAKE_KILL_THRESHOLD.key,
+      WORKER_JVM_QUAKE_KILL_THRESHOLD.defaultValueString).microsecond
+  def workerJvmQuakeExitCode: Int = get(WORKER_JVM_QUAKE_EXIT_CODE)
 
   // //////////////////////////////////////////////////////
   //                 Metrics System                      //
@@ -2896,6 +2911,77 @@ object CelebornConf extends Logging {
       .longConf
       .createOptional
 
+  val WORKER_JVM_QUAKE_ENABLED: ConfigEntry[Boolean] =
+    buildConf("celeborn.worker.jvmQuake.enabled")
+      .categories("worker")
+      .version("0.4.0")
+      .doc("When true, Celeborn worker will start the jvm quake to monitor of gc behavior, " +
+        "which enables early detection of memory management issues and facilitates fast failure.")
+      .booleanConf
+      .createWithDefault(false)
+
+  val WORKER_JVM_QUAKE_CHECK_INTERVAL: ConfigEntry[Long] =
+    buildConf("celeborn.worker.jvmQuake.check.interval")
+      .categories("worker")
+      .version("0.4.0")
+      .doc("Interval of gc behavior checking for worker jvm quake.")
+      .timeConf(TimeUnit.MILLISECONDS)
+      .createWithDefaultString("1s")
+
+  val WORKER_JVM_QUAKE_RUNTIME_WEIGHT: ConfigEntry[Double] =
+    buildConf("celeborn.worker.jvmQuake.runtimeWeight")
+      .categories("worker")
+      .version("0.4.0")
+      .doc(
+        "The factor by which to multiply running JVM time, when weighing it against GCing time. " +
+          "'Deficit' is accumulated as `gc_time - runtime * runtime_weight`, and is compared against threshold " +
+          "to determine whether to take action.")
+      .doubleConf
+      .createWithDefault(5)
+
+  val WORKER_JVM_QUAKE_DUMP_THRESHOLD: ConfigEntry[Long] =
+    buildConf("celeborn.worker.jvmQuake.dump.threshold")
+      .categories("worker")
+      .version("0.4.0")
+      .doc("The threshold of heap dump for the maximum GC 'deficit' which can be accumulated before jvmquake takes action. " +
+        "Meanwhile, there is no heap dump generated when dump threshold is greater than kill threshold.")
+      .timeConf(TimeUnit.MILLISECONDS)
+      .createWithDefaultString("30s")
+
+  val WORKER_JVM_QUAKE_KILL_THRESHOLD: ConfigEntry[Long] =
+    buildConf("celeborn.worker.jvmQuake.kill.threshold")
+      .categories("worker")
+      .version("0.4.0")
+      .doc("The threshold of system kill for the maximum GC 'deficit' which can be accumulated before jvmquake takes action.")
+      .timeConf(TimeUnit.MILLISECONDS)
+      .createWithDefaultString("60s")
+
+  val WORKER_JVM_QUAKE_DUMP_ENABLED: ConfigEntry[Boolean] =
+    buildConf("celeborn.worker.jvmQuake.dump.enabled")
+      .categories("worker")
+      .version("0.4.0")
+      .doc("Whether to heap dump for the maximum GC 'deficit' during worker jvm quake.")
+      .booleanConf
+      .createWithDefault(true)
+
+  val WORKER_JVM_QUAKE_DUMP_PATH: ConfigEntry[String] =
+    buildConf("celeborn.worker.jvmQuake.dump.path")
+      .categories("worker")
+      .version("0.4.0")
+      .doc("The path of heap dump for the maximum GC 'deficit' during worker jvm quake.")
+      .stringConf
+      .transform(_.replace("<tmp>", System.getProperty("java.io.tmpdir"))
+        .replace("<pid>", Utils.getProcessId))
+      .createWithDefault(s"<tmp>/jvm-quake/dump/<pid>")
+
+  val WORKER_JVM_QUAKE_EXIT_CODE: ConfigEntry[Int] =
+    buildConf("celeborn.worker.jvmQuake.exitCode")
+      .categories("worker")
+      .version("0.4.0")
+      .doc("The exit code of system kill for the maximum GC 'deficit' during worker jvm quake.")
+      .intConf
+      .createWithDefault(502)
+
   val APPLICATION_HEARTBEAT_INTERVAL: ConfigEntry[Long] =
     buildConf("celeborn.client.application.heartbeatInterval")
       .withAlternative("celeborn.application.heartbeatInterval")

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

@@ -1091,4 +1091,6 @@ object Utils extends Logging {
     }
     labelPart(0).trim -> labelPart(1).trim
   }
+
+  def getProcessId: String = ManagementFactory.getRuntimeMXBean.getName.split("@")(0)
 }

dev/dependencies.sh

@@ -56,9 +56,9 @@ function mvn_build_classpath() {
 }
 
 function sbt_build_client_classpath() {
-  PATTERN="$SBT_PROJECT / Runtime / externalDependencyClasspath"
+  PATTERN="$SBT_PROJECT / Runtime / managedClasspath"
   deps=$(
-    $SBT -P$MODULE "clean; export Runtime/externalDependencyClasspath" | \
+    $SBT -P$MODULE "clean; export Runtime/managedClasspath" | \
       awk -v pat="$PATTERN" '$0 ~ pat { found=1 } found { print }' | \
       awk 'NR==2' | \
       tr ":" "\n"
@@ -96,8 +96,8 @@ function sbt_build_client_classpath() {
 }
 
 function sbt_build_server_classpath() {
-  $SBT "error; clean; export externalDependencyClasspath" | \
-    awk '/externalDependencyClasspath/ { found=1 } found { print }' | \
+  $SBT "error; clean; export managedClasspath" | \
+    awk '/managedClasspath/ { found=1 } found { print }' | \
     awk 'NR % 2 == 0' | \
     # This will skip the last line 
     sed '$d' | \

dev/deps/dependencies-client-flink-1.14

@@ -30,6 +30,7 @@ jsr305/1.3.9//jsr305-1.3.9.jar
 jul-to-slf4j/1.7.36//jul-to-slf4j-1.7.36.jar
 leveldbjni-all/1.8//leveldbjni-all-1.8.jar
 lz4-java/1.8.0//lz4-java-1.8.0.jar
+maven-jdk-tools-wrapper/0.1//maven-jdk-tools-wrapper-0.1.jar
 metrics-core/3.2.6//metrics-core-3.2.6.jar
 metrics-graphite/3.2.6//metrics-graphite-3.2.6.jar
 metrics-jvm/3.2.6//metrics-jvm-3.2.6.jar

dev/deps/dependencies-client-flink-1.15

@@ -30,6 +30,7 @@ jsr305/1.3.9//jsr305-1.3.9.jar
 jul-to-slf4j/1.7.36//jul-to-slf4j-1.7.36.jar
 leveldbjni-all/1.8//leveldbjni-all-1.8.jar
 lz4-java/1.8.0//lz4-java-1.8.0.jar
+maven-jdk-tools-wrapper/0.1//maven-jdk-tools-wrapper-0.1.jar
 metrics-core/3.2.6//metrics-core-3.2.6.jar
 metrics-graphite/3.2.6//metrics-graphite-3.2.6.jar
 metrics-jvm/3.2.6//metrics-jvm-3.2.6.jar

dev/deps/dependencies-client-flink-1.17

@@ -30,6 +30,7 @@ jsr305/1.3.9//jsr305-1.3.9.jar
 jul-to-slf4j/1.7.36//jul-to-slf4j-1.7.36.jar
 leveldbjni-all/1.8//leveldbjni-all-1.8.jar
 lz4-java/1.8.0//lz4-java-1.8.0.jar
+maven-jdk-tools-wrapper/0.1//maven-jdk-tools-wrapper-0.1.jar
 metrics-core/3.2.6//metrics-core-3.2.6.jar
 metrics-graphite/3.2.6//metrics-graphite-3.2.6.jar
 metrics-jvm/3.2.6//metrics-jvm-3.2.6.jar

dev/deps/dependencies-client-flink-1.18

@@ -30,6 +30,7 @@ jsr305/1.3.9//jsr305-1.3.9.jar
 jul-to-slf4j/1.7.36//jul-to-slf4j-1.7.36.jar
 leveldbjni-all/1.8//leveldbjni-all-1.8.jar
 lz4-java/1.8.0//lz4-java-1.8.0.jar
+maven-jdk-tools-wrapper/0.1//maven-jdk-tools-wrapper-0.1.jar
 metrics-core/3.2.6//metrics-core-3.2.6.jar
 metrics-graphite/3.2.6//metrics-graphite-3.2.6.jar
 metrics-jvm/3.2.6//metrics-jvm-3.2.6.jar

dev/deps/dependencies-client-mr

@@ -122,6 +122,7 @@ kerby-util/1.0.1//kerby-util-1.0.1.jar
 kerby-xdr/1.0.1//kerby-xdr-1.0.1.jar
 leveldbjni-all/1.8//leveldbjni-all-1.8.jar
 lz4-java/1.8.0//lz4-java-1.8.0.jar
+maven-jdk-tools-wrapper/0.1//maven-jdk-tools-wrapper-0.1.jar
 metrics-core/3.2.6//metrics-core-3.2.6.jar
 metrics-graphite/3.2.6//metrics-graphite-3.2.6.jar
 metrics-jvm/3.2.6//metrics-jvm-3.2.6.jar

dev/deps/dependencies-client-spark-2.4

@@ -30,6 +30,7 @@ jsr305/1.3.9//jsr305-1.3.9.jar
 jul-to-slf4j/1.7.36//jul-to-slf4j-1.7.36.jar
 leveldbjni-all/1.8//leveldbjni-all-1.8.jar
 lz4-java/1.4.0//lz4-java-1.4.0.jar
+maven-jdk-tools-wrapper/0.1//maven-jdk-tools-wrapper-0.1.jar
 metrics-core/3.2.6//metrics-core-3.2.6.jar
 metrics-graphite/3.2.6//metrics-graphite-3.2.6.jar
 metrics-jvm/3.2.6//metrics-jvm-3.2.6.jar

dev/deps/dependencies-client-spark-3.0

@@ -30,6 +30,7 @@ jsr305/1.3.9//jsr305-1.3.9.jar
 jul-to-slf4j/1.7.36//jul-to-slf4j-1.7.36.jar
 leveldbjni-all/1.8//leveldbjni-all-1.8.jar
 lz4-java/1.7.1//lz4-java-1.7.1.jar
+maven-jdk-tools-wrapper/0.1//maven-jdk-tools-wrapper-0.1.jar
 metrics-core/3.2.6//metrics-core-3.2.6.jar
 metrics-graphite/3.2.6//metrics-graphite-3.2.6.jar
 metrics-jvm/3.2.6//metrics-jvm-3.2.6.jar

dev/deps/dependencies-client-spark-3.1

@@ -30,6 +30,7 @@ jsr305/1.3.9//jsr305-1.3.9.jar
 jul-to-slf4j/1.7.36//jul-to-slf4j-1.7.36.jar
 leveldbjni-all/1.8//leveldbjni-all-1.8.jar
 lz4-java/1.7.1//lz4-java-1.7.1.jar
+maven-jdk-tools-wrapper/0.1//maven-jdk-tools-wrapper-0.1.jar
 metrics-core/3.2.6//metrics-core-3.2.6.jar
 metrics-graphite/3.2.6//metrics-graphite-3.2.6.jar
 metrics-jvm/3.2.6//metrics-jvm-3.2.6.jar

dev/deps/dependencies-client-spark-3.2

@@ -30,6 +30,7 @@ jsr305/1.3.9//jsr305-1.3.9.jar
 jul-to-slf4j/1.7.36//jul-to-slf4j-1.7.36.jar
 leveldbjni-all/1.8//leveldbjni-all-1.8.jar
 lz4-java/1.7.1//lz4-java-1.7.1.jar
+maven-jdk-tools-wrapper/0.1//maven-jdk-tools-wrapper-0.1.jar
 metrics-core/3.2.6//metrics-core-3.2.6.jar
 metrics-graphite/3.2.6//metrics-graphite-3.2.6.jar
 metrics-jvm/3.2.6//metrics-jvm-3.2.6.jar

dev/deps/dependencies-client-spark-3.3

@@ -30,6 +30,7 @@ jsr305/1.3.9//jsr305-1.3.9.jar
 jul-to-slf4j/1.7.36//jul-to-slf4j-1.7.36.jar
 leveldbjni-all/1.8//leveldbjni-all-1.8.jar
 lz4-java/1.8.0//lz4-java-1.8.0.jar
+maven-jdk-tools-wrapper/0.1//maven-jdk-tools-wrapper-0.1.jar
 metrics-core/3.2.6//metrics-core-3.2.6.jar
 metrics-graphite/3.2.6//metrics-graphite-3.2.6.jar
 metrics-jvm/3.2.6//metrics-jvm-3.2.6.jar

dev/deps/dependencies-client-spark-3.4

@@ -30,6 +30,7 @@ jsr305/1.3.9//jsr305-1.3.9.jar
 jul-to-slf4j/1.7.36//jul-to-slf4j-1.7.36.jar
 leveldbjni-all/1.8//leveldbjni-all-1.8.jar
 lz4-java/1.8.0//lz4-java-1.8.0.jar
+maven-jdk-tools-wrapper/0.1//maven-jdk-tools-wrapper-0.1.jar
 metrics-core/3.2.6//metrics-core-3.2.6.jar
 metrics-graphite/3.2.6//metrics-graphite-3.2.6.jar
 metrics-jvm/3.2.6//metrics-jvm-3.2.6.jar

dev/deps/dependencies-client-spark-3.5

@@ -30,6 +30,7 @@ jsr305/1.3.9//jsr305-1.3.9.jar
 jul-to-slf4j/1.7.36//jul-to-slf4j-1.7.36.jar
 leveldbjni-all/1.8//leveldbjni-all-1.8.jar
 lz4-java/1.8.0//lz4-java-1.8.0.jar
+maven-jdk-tools-wrapper/0.1//maven-jdk-tools-wrapper-0.1.jar
 metrics-core/3.2.6//metrics-core-3.2.6.jar
 metrics-graphite/3.2.6//metrics-graphite-3.2.6.jar
 metrics-jvm/3.2.6//metrics-jvm-3.2.6.jar

dev/deps/dependencies-server

@@ -37,6 +37,7 @@ log4j-api/2.17.2//log4j-api-2.17.2.jar
 log4j-core/2.17.2//log4j-core-2.17.2.jar
 log4j-slf4j-impl/2.17.2//log4j-slf4j-impl-2.17.2.jar
 lz4-java/1.8.0//lz4-java-1.8.0.jar
+maven-jdk-tools-wrapper/0.1//maven-jdk-tools-wrapper-0.1.jar
 metrics-core/3.2.6//metrics-core-3.2.6.jar
 metrics-graphite/3.2.6//metrics-graphite-3.2.6.jar
 metrics-jvm/3.2.6//metrics-jvm-3.2.6.jar

docs/configuration/worker.md

@@ -70,6 +70,14 @@ license: |
 | celeborn.worker.graceful.shutdown.timeout | 600s | The worker's graceful shutdown timeout time. | 0.2.0 | 
 | celeborn.worker.http.host | <localhost> | Worker's http host. | 0.4.0 | 
 | celeborn.worker.http.port | 9096 | Worker's http port. | 0.4.0 | 
+| celeborn.worker.jvmQuake.check.interval | 1s | Interval of gc behavior checking for worker jvm quake. | 0.4.0 | 
+| celeborn.worker.jvmQuake.dump.enabled | true | Whether to heap dump for the maximum GC 'deficit' during worker jvm quake. | 0.4.0 | 
+| celeborn.worker.jvmQuake.dump.path | <tmp>/jvm-quake/dump/<pid> | The path of heap dump for the maximum GC 'deficit' during worker jvm quake. | 0.4.0 | 
+| celeborn.worker.jvmQuake.dump.threshold | 30s | The threshold of heap dump for the maximum GC 'deficit' which can be accumulated before jvmquake takes action. Meanwhile, there is no heap dump generated when dump threshold is greater than kill threshold. | 0.4.0 | 
+| celeborn.worker.jvmQuake.enabled | false | When true, Celeborn worker will start the jvm quake to monitor of gc behavior, which enables early detection of memory management issues and facilitates fast failure. | 0.4.0 | 
+| celeborn.worker.jvmQuake.exitCode | 502 | The exit code of system kill for the maximum GC 'deficit' during worker jvm quake. | 0.4.0 | 
+| celeborn.worker.jvmQuake.kill.threshold | 60s | The threshold of system kill for the maximum GC 'deficit' which can be accumulated before jvmquake takes action. | 0.4.0 | 
+| celeborn.worker.jvmQuake.runtimeWeight | 5.0 | The factor by which to multiply running JVM time, when weighing it against GCing time. 'Deficit' is accumulated as `gc_time - runtime * runtime_weight`, and is compared against threshold to determine whether to take action. | 0.4.0 | 
 | celeborn.worker.monitor.disk.check.interval | 30s | Intervals between device monitor to check disk. | 0.3.0 | 
 | celeborn.worker.monitor.disk.check.timeout | 30s | Timeout time for worker check device status. | 0.3.0 | 
 | celeborn.worker.monitor.disk.checklist | readwrite,diskusage | Monitor type for disk, available items are: iohang, readwrite and diskusage. | 0.2.0 | 

licenses-binary/LISENCE-jdktools.txt

@@ -0,0 +1,21 @@
+The MIT License
+
+Copyright (c) 2016 Red Hat, Inc.
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.

pom.xml

@@ -122,6 +122,7 @@
 
     <!-- for JDK-17 test-->
     <extraJavaTestArgs>-XX:+IgnoreUnrecognizedVMOptions
+      --add-exports=jdk.internal.jvmstat/sun.jvmstat.monitor=ALL-UNNAMED
       --add-opens=java.base/java.lang=ALL-UNNAMED
       --add-opens=java.base/java.lang.invoke=ALL-UNNAMED
       --add-opens=java.base/java.lang.reflect=ALL-UNNAMED
@@ -1156,6 +1157,13 @@
       <properties>
         <java.version>8</java.version>
       </properties>
+      <dependencies>
+        <dependency>
+          <groupId>com.github.olivergondza</groupId>
+          <artifactId>maven-jdk-tools-wrapper</artifactId>
+          <version>0.1</version>
+        </dependency>
+      </dependencies>
       <build>
         <plugins>
           <!-- Based on https://github.com/google/error-prone/blob/f8e33bc460be82ab22256a7ef8b979d7a2cacaba/docs/installation.md#jdk-8 -->

project/CelebornBuild.scala

@@ -50,6 +50,7 @@ object Dependencies {
   val junitVersion = "4.13.2"
   val leveldbJniVersion = "1.8"
   val log4j2Version = "2.17.2"
+  val jdkToolsVersion = "0.1"
   val metricsVersion = "3.2.6"
   val mockitoVersion = "4.11.0"
   val nettyVersion = "4.1.93.Final"
@@ -71,6 +72,7 @@ object Dependencies {
   val commonsIo = "commons-io" % "commons-io" % commonsIoVersion
   val commonsLang3 = "org.apache.commons" % "commons-lang3" % commonsLang3Version
   val commonsLogging = "commons-logging" % "commons-logging" % commonsLoggingVersion
+  val jdkTools = "com.github.olivergondza" % "maven-jdk-tools-wrapper" % jdkToolsVersion
   val findbugsJsr305 = "com.google.code.findbugs" % "jsr305" % findbugsVersion
   val guava = "com.google.guava" % "guava" % guavaVersion excludeAll(
     ExclusionRule("org.checkerframework", "checker-qual"),
@@ -167,6 +169,7 @@ object CelebornCommonSettings {
     Test / javaOptions ++= Seq(
       "-Xmx4g",
       "-XX:+IgnoreUnrecognizedVMOptions",
+      "--add-exports=jdk.internal.jvmstat/sun.jvmstat.monitor=ALL-UNNAMED",
       "--add-opens=java.base/java.lang=ALL-UNNAMED",
       "--add-opens=java.base/java.lang.invoke=ALL-UNNAMED",
       "--add-opens=java.base/java.lang.reflect=ALL-UNNAMED",
@@ -322,6 +325,7 @@ object CelebornCommon {
         Dependencies.commonsLang3,
         Dependencies.hadoopClientApi,
         Dependencies.hadoopClientRuntime,
+        Dependencies.jdkTools,
         Dependencies.ratisClient,
         Dependencies.ratisCommon,
         Dependencies.leveldbJniAll,