You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by et...@apache.org on 2024/01/25 12:28:41 UTC
(incubator-celeborn) branch main updated: [CELEBORN-1252] Fix resource consumption of worker does not update when update interval is greater than heartbeat interval
This is an automated email from the ASF dual-hosted git repository.
ethanfeng 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 5c54388bc [CELEBORN-1252] Fix resource consumption of worker does not update when update interval is greater than heartbeat interval
5c54388bc is described below
commit 5c54388bc2cfee712b82c91aaf0be933358dfed0
Author: Angerszhuuuu <an...@gmail.com>
AuthorDate: Thu Jan 25 20:28:19 2024 +0800
[CELEBORN-1252] Fix resource consumption of worker does not update when update interval is greater than heartbeat interval
### What changes were proposed in this pull request?
Resource consumption of worker does not update when update interval of resource consumpution is greater than heartbeat interval.
<img width="1741" alt="截屏2024-01-24 14 49 50" src="https://github.com/apache/incubator-celeborn/assets/46485123/21cfd412-c69e-4955-8bc8-155ee470697d">
This pull request introduces below changes:
1. Avoid master repeat add gauge for same user
2. For worker, user resource consumption can directly get from worker's snapshot, didn't need update interval
### Why are the changes needed?
No.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
No.
Closes #2260 from AngersZhuuuu/CELEBORN-1252.
Authored-by: Angerszhuuuu <an...@gmail.com>
Signed-off-by: mingji <fe...@alibaba-inc.com>
---
.../org/apache/celeborn/common/CelebornConf.scala | 9 ---
.../apache/celeborn/common/meta/WorkerInfo.scala | 9 ++-
docs/configuration/worker.md | 1 -
docs/migration.md | 6 +-
.../celeborn/service/deploy/worker/Worker.scala | 69 +++++++---------------
5 files changed, 31 insertions(+), 63 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 76f37e241..ca2f8f132 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -543,7 +543,6 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
def estimatedPartitionSizeForEstimationUpdateInterval: Long =
get(ESTIMATED_PARTITION_SIZE_UPDATE_INTERVAL)
def masterResourceConsumptionInterval: Long = get(MASTER_RESOURCE_CONSUMPTION_INTERVAL)
- def workerResourceConsumptionInterval: Long = get(WORKER_RESOURCE_CONSUMPTION_INTERVAL)
// //////////////////////////////////////////////////////
// Address && HA && RATIS //
@@ -2186,14 +2185,6 @@ object CelebornConf extends Logging {
.timeConf(TimeUnit.MILLISECONDS)
.createWithDefaultString("30s")
- val WORKER_RESOURCE_CONSUMPTION_INTERVAL: ConfigEntry[Long] =
- buildConf("celeborn.worker.userResourceConsumption.update.interval")
- .categories("worker")
- .doc("Time length for a window about compute user resource consumption.")
- .version("0.3.2")
- .timeConf(TimeUnit.MILLISECONDS)
- .createWithDefaultString("30s")
-
val SHUFFLE_CHUNK_SIZE: ConfigEntry[Long] =
buildConf("celeborn.shuffle.chunk.size")
.categories("worker")
diff --git a/common/src/main/scala/org/apache/celeborn/common/meta/WorkerInfo.scala b/common/src/main/scala/org/apache/celeborn/common/meta/WorkerInfo.scala
index 20561eebf..683a19557 100644
--- a/common/src/main/scala/org/apache/celeborn/common/meta/WorkerInfo.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/meta/WorkerInfo.scala
@@ -188,11 +188,14 @@ class WorkerInfo(
JavaUtils.newConcurrentHashMap[String, DiskInfo](diskInfos)
}
- def updateThenGetUserResourceConsumption(consumption: util.Map[
+ def updateThenGetUserResourceConsumption(resourceConsumptions: util.Map[
UserIdentifier,
ResourceConsumption]): util.Map[UserIdentifier, ResourceConsumption] = {
- userResourceConsumption.clear()
- userResourceConsumption.putAll(consumption)
+ userResourceConsumption.keys().asScala.filterNot(resourceConsumptions.containsKey).foreach {
+ identifier =>
+ userResourceConsumption.put(identifier, ResourceConsumption(0, 0, 0, 0))
+ }
+ userResourceConsumption.putAll(resourceConsumptions)
userResourceConsumption
}
diff --git a/docs/configuration/worker.md b/docs/configuration/worker.md
index 24340be0b..1f7f48a24 100644
--- a/docs/configuration/worker.md
+++ b/docs/configuration/worker.md
@@ -122,7 +122,6 @@ license: |
| celeborn.worker.storage.disk.reserve.size | 5G | Celeborn worker reserved space for each disk. | 0.3.0 | celeborn.worker.disk.reserve.size |
| celeborn.worker.storage.expireDirs.timeout | 1h | The timeout for a expire dirs to be deleted on disk. | 0.3.2 | |
| celeborn.worker.storage.workingDir | celeborn-worker/shuffle_data | Worker's working dir path name. | 0.3.0 | celeborn.worker.workingDir |
-| celeborn.worker.userResourceConsumption.update.interval | 30s | Time length for a window about compute user resource consumption. | 0.3.2 | |
| celeborn.worker.writer.close.timeout | 120s | Timeout for a file writer to close | 0.2.0 | |
| celeborn.worker.writer.create.maxAttempts | 3 | Retry count for a file writer to create if its creation was failed. | 0.2.0 | |
<!--end-include-->
diff --git a/docs/migration.md b/docs/migration.md
index 81a28d406..40311fa83 100644
--- a/docs/migration.md
+++ b/docs/migration.md
@@ -23,9 +23,9 @@ license: |
## Upgrading from 0.4 to 0.5
-- Since 0.5.0 Celeborn worker metrics `ChunkStreamCount` is renamed as `ActiveChunkStreamCount`.
+- Since 0.5.0, Celeborn worker metrics `ChunkStreamCount` is renamed as `ActiveChunkStreamCount`.
-- Since 0.5.0 Celeborn worker metrics `CreditStreamCount` is renamed as `ActiveCreditStreamCount`.
+- Since 0.5.0, Celeborn worker metrics `CreditStreamCount` is renamed as `ActiveCreditStreamCount`.
## Upgrading from 0.3 to 0.4
@@ -50,6 +50,8 @@ license: |
- Since 0.4.0, Celeborn deprecate `celeborn.storage.activeTypes`. Please use `celeborn.storage.availableTypes` instead.
+- Since 0.4.0, Celeborn worker removes configuration `celeborn.worker.userResourceConsumption.update.interval`.
+
- Since 0.4.0, Celeborn master metrics `PartitionWritten` is renamed as `ActiveShuffleSize`.
- Since 0.4.0, Celeborn master metrics `PartitionFileCount` is renamed as `ActiveShuffleFileCount`.
diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala
index c438ff8fe..b237f21ee 100644
--- a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala
+++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala
@@ -213,11 +213,6 @@ private[celeborn] class Worker(
diskInfos.put(diskInfo.mountPoint, diskInfo)
}
- // need to ensure storageManager has recovered fileinfos data if enable graceful shutdown before retrieve consumption
- val userResourceConsumption: ConcurrentHashMap[UserIdentifier, ResourceConsumption] =
- JavaUtils.newConcurrentHashMap[UserIdentifier, ResourceConsumption](
- storageManager.userResourceConsumptionSnapshot().asJava)
-
val workerInfo =
new WorkerInfo(
host,
@@ -226,7 +221,7 @@ private[celeborn] class Worker(
fetchPort,
replicatePort,
diskInfos,
- userResourceConsumption)
+ JavaUtils.newConcurrentHashMap[UserIdentifier, ResourceConsumption])
// whether this Worker registered to Master successfully
val registered = new AtomicBoolean(false)
@@ -273,10 +268,6 @@ private[celeborn] class Worker(
var cleaner: ExecutorService =
ThreadUtils.newDaemonSingleThreadExecutor("worker-expired-shuffle-cleaner")
- private val workerResourceConsumptionInterval = conf.workerResourceConsumptionInterval
- private val userResourceConsumptions =
- JavaUtils.newConcurrentHashMap[UserIdentifier, (ResourceConsumption, Long)]()
-
private var jvmQuake: JVMQuake = _
if (conf.workerJvmQuakeEnabled) {
jvmQuake = JVMQuake.create(conf, workerInfo.toUniqueId().replace(":", "-"))
@@ -534,47 +525,29 @@ private[celeborn] class Worker(
private def handleResourceConsumption(): util.Map[UserIdentifier, ResourceConsumption] = {
val resourceConsumptionSnapshot = storageManager.userResourceConsumptionSnapshot()
- resourceConsumptionSnapshot.foreach { resourceConsumption =>
- {
- resourceConsumptionSource.addGauge(
- ResourceConsumptionSource.DISK_FILE_COUNT,
- resourceConsumption._1.toMap) { () =>
- computeUserResourceConsumption(resourceConsumption).diskFileCount
- }
- resourceConsumptionSource.addGauge(
- ResourceConsumptionSource.DISK_BYTES_WRITTEN,
- resourceConsumption._1.toMap) { () =>
- computeUserResourceConsumption(resourceConsumption).diskBytesWritten
- }
- resourceConsumptionSource.addGauge(
- ResourceConsumptionSource.HDFS_FILE_COUNT,
- resourceConsumption._1.toMap) { () =>
- computeUserResourceConsumption(resourceConsumption).hdfsFileCount
- }
- resourceConsumptionSource.addGauge(
- ResourceConsumptionSource.HDFS_BYTES_WRITTEN,
- resourceConsumption._1.toMap) { () =>
- computeUserResourceConsumption(resourceConsumption).hdfsBytesWritten
- }
+ resourceConsumptionSnapshot.foreach { case (userIdentifier, _) =>
+ resourceConsumptionSource.addGauge(
+ ResourceConsumptionSource.DISK_FILE_COUNT,
+ userIdentifier.toMap) { () =>
+ workerInfo.userResourceConsumption.get(userIdentifier).diskFileCount
}
- }
- workerInfo.updateThenGetUserResourceConsumption(resourceConsumptionSnapshot.asJava)
- }
-
- private def computeUserResourceConsumption(userResourceConsumption: (
- UserIdentifier,
- ResourceConsumption)): ResourceConsumption = {
- val userIdentifier = userResourceConsumption._1
- val resourceConsumption = userResourceConsumption._2
- val current = System.currentTimeMillis()
- if (userResourceConsumptions.containsKey(userIdentifier)) {
- val resourceConsumptionAndUpdateTime = userResourceConsumptions.get(userIdentifier)
- if (current - resourceConsumptionAndUpdateTime._2 <= workerResourceConsumptionInterval) {
- return resourceConsumptionAndUpdateTime._1
+ resourceConsumptionSource.addGauge(
+ ResourceConsumptionSource.DISK_BYTES_WRITTEN,
+ userIdentifier.toMap) { () =>
+ workerInfo.userResourceConsumption.get(userIdentifier).diskBytesWritten
+ }
+ resourceConsumptionSource.addGauge(
+ ResourceConsumptionSource.HDFS_FILE_COUNT,
+ userIdentifier.toMap) { () =>
+ workerInfo.userResourceConsumption.get(userIdentifier).hdfsFileCount
+ }
+ resourceConsumptionSource.addGauge(
+ ResourceConsumptionSource.HDFS_BYTES_WRITTEN,
+ userIdentifier.toMap) { () =>
+ workerInfo.userResourceConsumption.get(userIdentifier).hdfsBytesWritten
}
}
- userResourceConsumptions.put(userIdentifier, (resourceConsumption, current))
- resourceConsumption
+ workerInfo.updateThenGetUserResourceConsumption(resourceConsumptionSnapshot.asJava)
}
@VisibleForTesting