This is an automated email from the ASF dual-hosted git repository.
SteNicholas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new 9ebbc6b36 [CELEBORN-2324] Fix JVMQuake threshold and JVMStat timer
unit conversion
9ebbc6b36 is described below
commit 9ebbc6b36ea94b1b665954d699d76e4711c3dd94
Author: pithecuse527 <[email protected]>
AuthorDate: Wed May 13 15:35:43 2026 +0800
[CELEBORN-2324] Fix JVMQuake threshold and JVMStat timer unit conversion
### What changes were proposed in this pull request?
This PR fixes JVMQuake time accounting by preserving threshold config
values as milliseconds and converting JVMStat GC timer tick deltas to
nanoseconds before updating the token bucket.
### Why are the changes needed?
JVMQuake thresholds were parsed as milliseconds but wrapped as
microseconds, making values such as 60s behave like 60ms. JVMStat GC timer
metrics are reported in ticks, so using them directly can misaccount GC time.
### Does this PR resolve a correctness bug?
Yes
### Does this PR introduce _any_ user-facing change?
Yes
### How was this patch tested?
1. UT - Added unit coverage for JVMQuake threshold parsing and JVMStat
tick-to-nanosecond conversion.
2. E2E - Verified the patched image in a Kubernetes spark namespace with
JVMQuake enabled using `dump.threshold=30s`, `kill.threshold=60s`, and
`runtimeWeight=0`. Under repeated GC, the worker stayed Ready with restart
count 0, while the original image terminated early because the configured `60s`
threshold was effectively interpreted as `60ms`.
Verified the kill path with a low-threshold configuration:
`dump.threshold=30ms`, `kill.threshold=60ms`, `runtimeWeight=0`, and
`check.interval=100ms`. Under GC activity, the worker logged bucket: `62995087`
and killThreshold: `60000000`, exited via JVMQuake, and Kubernetes restarted
the pod, increasing the restart count from 0 -> 1.
Closes #3682 from pithecuse527/CELEBORN-2324.
Authored-by: pithecuse527 <[email protected]>
Signed-off-by: SteNicholas <[email protected]>
---
.../org/apache/celeborn/common/CelebornConf.scala | 4 ++--
.../apache/celeborn/common/CelebornConfSuite.scala | 9 ++++++++
.../service/deploy/worker/monitor/JVMQuake.scala | 27 ++++++++++++++++++----
.../deploy/worker/monitor/JVMQuakeSuite.scala | 11 +++++++++
4 files changed, 44 insertions(+), 7 deletions(-)
diff --git
a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
index 796fdf7d4..ed7013ed5 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -896,11 +896,11 @@ class CelebornConf(loadDefaults: Boolean) extends
Cloneable with Logging with Se
def workerJvmQuakeDumpThreshold: Duration =
getTimeAsMs(
WORKER_JVM_QUAKE_DUMP_THRESHOLD.key,
- WORKER_JVM_QUAKE_DUMP_THRESHOLD.defaultValueString).microsecond
+ WORKER_JVM_QUAKE_DUMP_THRESHOLD.defaultValueString).millisecond
def workerJvmQuakeKillThreshold: Duration =
getTimeAsMs(
WORKER_JVM_QUAKE_KILL_THRESHOLD.key,
- WORKER_JVM_QUAKE_KILL_THRESHOLD.defaultValueString).microsecond
+ WORKER_JVM_QUAKE_KILL_THRESHOLD.defaultValueString).millisecond
def workerJvmQuakeExitCode: Int = get(WORKER_JVM_QUAKE_EXIT_CODE)
// //////////////////////////////////////////////////////
diff --git
a/common/src/test/scala/org/apache/celeborn/common/CelebornConfSuite.scala
b/common/src/test/scala/org/apache/celeborn/common/CelebornConfSuite.scala
index ff2584ac6..953255cde 100644
--- a/common/src/test/scala/org/apache/celeborn/common/CelebornConfSuite.scala
+++ b/common/src/test/scala/org/apache/celeborn/common/CelebornConfSuite.scala
@@ -25,6 +25,15 @@ import org.apache.celeborn.common.protocol.StorageInfo
class CelebornConfSuite extends CelebornFunSuite {
+ test("JVMQuake thresholds should preserve configured time units") {
+ val conf = new CelebornConf()
+ .set(WORKER_JVM_QUAKE_DUMP_THRESHOLD.key, "30s")
+ .set(WORKER_JVM_QUAKE_KILL_THRESHOLD.key, "60s")
+
+ assert(conf.workerJvmQuakeDumpThreshold.toMillis == 30000L)
+ assert(conf.workerJvmQuakeKillThreshold.toMillis == 60000L)
+ }
+
test("celeborn.master.endpoints support multi nodes") {
val conf = new CelebornConf()
.set(CelebornConf.MASTER_ENDPOINTS.key,
"localhost1:9097,localhost2:9097")
diff --git
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/monitor/JVMQuake.scala
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/monitor/JVMQuake.scala
index 40b14c6d0..477b1ad2f 100644
---
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/monitor/JVMQuake.scala
+++
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/monitor/JVMQuake.scala
@@ -87,11 +87,15 @@ class JVMQuake(conf: CelebornConf, uniqueId: String =
UUID.randomUUID().toString
private def run(): Unit = {
val currentExitTime = getLastExitTime
val currentGCTime = getLastGCTime
- val gcTime = currentGCTime - lastGCTime
- val runTime = currentExitTime - lastExitTime - gcTime
-
- bucket = Math.max(0, bucket + gcTime - BigDecimal(runTime *
runtimeWeight).toLong)
- logDebug(s"Time: (gc time: ${Utils.msDurationToString(gcTime)}, execution
time: ${Utils.msDurationToString(runTime)})")
+ val gcTimeTicks = currentGCTime - lastGCTime
+ val runTimeTicks = currentExitTime - lastExitTime - gcTimeTicks
+ // JVMStat time monitors are reported in ticks. Convert deltas to nanos
before comparing
+ // them against JVMQuake thresholds, which are stored as nanos.
+ val gcTime = ticksToNanos(gcTimeTicks)
+ val runTime = ticksToNanos(runTimeTicks)
+
+ bucket = Math.max(0, bucket + gcTime - (BigDecimal(runTime) *
BigDecimal(runtimeWeight)).toLong)
+ logDebug(s"Time: (gc time: ${Utils.nanoDurationToString(gcTime)},
execution time: ${Utils.nanoDurationToString(runTime)})")
logDebug(
s"Capacity: (bucket: $bucket, dump threshold: $dumpThreshold, kill
threshold: $killThreshold)")
@@ -161,6 +165,8 @@ class JVMQuake(conf: CelebornConf, uniqueId: String =
UUID.randomUUID().toString
object JVMQuake {
+ private[this] val NANOS_PER_SECOND = 1000000000L
+
private[this] var quake: JVMQuake = _
def create(conf: CelebornConf, uniqueId: String): JVMQuake = {
@@ -187,6 +193,8 @@ object JVMQuake {
monitoredVm.findByName("sun.gc.collector.1.lastExitTime")
private[this] lazy val ygcTimeMonitor: Monitor =
monitoredVm.findByName("sun.gc.collector.0.time")
private[this] lazy val fgcTimeMonitor: Monitor =
monitoredVm.findByName("sun.gc.collector.1.time")
+ private[this] lazy val hrtFrequencyMonitor: Monitor =
+ monitoredVm.findByName("sun.os.hrt.frequency")
private def getLastExitTime: Long = Math.max(
ygcExitTimeMonitor.getValue.asInstanceOf[Long],
@@ -194,4 +202,13 @@ object JVMQuake {
private def getLastGCTime: Long =
ygcTimeMonitor.getValue.asInstanceOf[Long] +
fgcTimeMonitor.getValue.asInstanceOf[Long]
+
+ private def hrtFrequency: Long =
hrtFrequencyMonitor.getValue.asInstanceOf[Long]
+
+ private[monitor] def ticksToNanos(ticks: Long): Long = ticksToNanos(ticks,
hrtFrequency)
+
+ private[monitor] def ticksToNanos(ticks: Long, frequency: Long): Long = {
+ require(frequency > 0, s"Invalid JVMStat high-resolution timer frequency:
$frequency")
+ ((BigInt(ticks) * BigInt(NANOS_PER_SECOND)) / BigInt(frequency)).toLong
+ }
}
diff --git
a/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/monitor/JVMQuakeSuite.scala
b/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/monitor/JVMQuakeSuite.scala
index 7ef49f225..fa3b7f36b 100644
---
a/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/monitor/JVMQuakeSuite.scala
+++
b/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/monitor/JVMQuakeSuite.scala
@@ -37,6 +37,17 @@ class JVMQuakeSuite extends CelebornFunSuite {
System.gc()
}
+ test("Convert JVMStat timer ticks to nanoseconds") {
+ assert(JVMQuake.ticksToNanos(1L, 1000000000L) === 1L)
+ assert(JVMQuake.ticksToNanos(1000L, 1000L) === 1000000000L)
+ assert(JVMQuake.ticksToNanos(1500L, 1000L) === 1500000000L)
+ assert(JVMQuake.ticksToNanos(-1000L, 1000L) === -1000000000L)
+
+ intercept[IllegalArgumentException] {
+ JVMQuake.ticksToNanos(1L, 0L)
+ }
+ }
+
test("[CELEBORN-1092] Introduce JVM monitoring in Celeborn Worker using
JVMQuake") {
val quake = new JVMQuake(new
CelebornConf().set(WORKER_JVM_QUAKE_ENABLED.key, "true")
.set(WORKER_JVM_QUAKE_RUNTIME_WEIGHT.key, "1")
