diff --git a/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala b/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala index e91e236b6c4..3ac3920bc81 100644 --- a/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala +++ b/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala @@ -23,6 +23,7 @@ import java.util.function.Consumer import scala.collection.JavaConverters._ +import com.google.common.annotations.VisibleForTesting import org.apache.commons.lang3.StringUtils import org.apache.celeborn.common.CelebornConf @@ -47,6 +48,13 @@ class ApplicationHeartbeater( private var stopped = false private val reviseLostShuffles = conf.reviseLostShufflesEnabled + private val gcOnOverloadEnabled = conf.clientGcOnOverloadEnabled + private val gcOnOverloadMinIntervalMs = conf.clientGcOnOverloadMinIntervalMs + private val gcOnOverloadMinIntervalNs = TimeUnit.MILLISECONDS.toNanos(gcOnOverloadMinIntervalMs) + + @VisibleForTesting + @volatile private[client] var lastGcTriggerTimeNs = 0L + // Use independent app heartbeat threads to avoid being blocked by other operations. private val appHeartbeatIntervalMs = conf.appHeartbeatIntervalMs private val applicationUnregisterEnabled = conf.applicationUnregisterEnabled @@ -91,6 +99,7 @@ class ApplicationHeartbeater( logDebug("Successfully send app heartbeat.") workerStatusTracker.handleHeartbeatResponse(response) checkQuotaExceeds(response.checkQuotaResponse) + handleGcSignal(response.shouldTriggerGc) // revise shuffle id if there are lost shuffles if (reviseLostShuffles) { val masterRecordedShuffleIds = response.registeredShuffles @@ -170,6 +179,21 @@ class ApplicationHeartbeater( } } + private[client] def handleGcSignal(shouldTriggerGc: Boolean): Unit = { + if (!gcOnOverloadEnabled || !shouldTriggerGc) return + val nowNs = System.nanoTime() + if (nowNs - lastGcTriggerTimeNs >= gcOnOverloadMinIntervalNs) { + logInfo( + "Cluster is overloaded; triggering System.gc() to release stale shuffle dependencies.") + lastGcTriggerTimeNs = nowNs + System.gc() + } else { + logDebug( + s"Cluster overload GC signal received but skipped: last GC was triggered " + + s"${(nowNs - lastGcTriggerTimeNs) / 1000000L}ms ago (min interval: ${gcOnOverloadMinIntervalMs}ms).") + } + } + def stop(): Unit = { this.synchronized { if (!stopped) { diff --git a/client/src/test/scala/org/apache/celeborn/client/ApplicationHeartbeaterSuite.scala b/client/src/test/scala/org/apache/celeborn/client/ApplicationHeartbeaterSuite.scala new file mode 100644 index 00000000000..69a1e47dd4d --- /dev/null +++ b/client/src/test/scala/org/apache/celeborn/client/ApplicationHeartbeaterSuite.scala @@ -0,0 +1,112 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.celeborn.client + +import java.util.concurrent.ConcurrentHashMap + +import org.apache.celeborn.CelebornFunSuite +import org.apache.celeborn.common.CelebornConf + +class ApplicationHeartbeaterSuite extends CelebornFunSuite { + + private def makeHeartbeater( + gcEnabled: Boolean = true, + minIntervalMs: Long = 60000L): ApplicationHeartbeater = { + val conf = new CelebornConf() + conf.set(CelebornConf.CLIENT_GC_ON_CLUSTER_OVERLOAD_ENABLED, gcEnabled) + conf.set(CelebornConf.CLIENT_GC_ON_CLUSTER_OVERLOAD_MIN_INTERVAL, minIntervalMs) + + val registeredShuffles = ConcurrentHashMap.newKeySet[Int]() + .asInstanceOf[ConcurrentHashMap.KeySetView[Int, java.lang.Boolean]] + + new ApplicationHeartbeater( + "test-app", + conf, + null, // MasterClient not needed for handleGcSignal unit tests + () => (0L, 0L) -> (0L, 0L, Map.empty, Map.empty), + null, // WorkerStatusTracker not needed + registeredShuffles, + _ => ()) + } + + test("GC is not triggered when feature is disabled") { + val hb = makeHeartbeater(gcEnabled = false) + hb.handleGcSignal(shouldTriggerGc = true) + assert(hb.lastGcTriggerTimeNs == 0L) + } + + test("GC is not triggered when signal is false") { + val hb = makeHeartbeater(gcEnabled = true) + hb.handleGcSignal(shouldTriggerGc = false) + assert(hb.lastGcTriggerTimeNs == 0L) + } + + test("GC is triggered on first signal when enabled") { + val hb = makeHeartbeater(gcEnabled = true, minIntervalMs = 0L) + val before = System.nanoTime() + hb.handleGcSignal(shouldTriggerGc = true) + assert(hb.lastGcTriggerTimeNs >= before) + } + + test("GC is skipped when called again within the cooldown interval") { + val hb = makeHeartbeater(gcEnabled = true, minIntervalMs = 60000L) + + hb.handleGcSignal(shouldTriggerGc = true) + val firstTrigger = hb.lastGcTriggerTimeNs + assert(firstTrigger > 0L) + + // Second call immediately — well within 60s cooldown + hb.handleGcSignal(shouldTriggerGc = true) + assert( + hb.lastGcTriggerTimeNs == firstTrigger, + "lastGcTriggerTimeNs should not change on second call") + } + + test("GC fires again after cooldown interval has elapsed") { + // Use 0ms cooldown so any elapsed time satisfies it. + val hb = makeHeartbeater(gcEnabled = true, minIntervalMs = 0L) + + hb.handleGcSignal(shouldTriggerGc = true) + val firstTrigger = hb.lastGcTriggerTimeNs + + // With 0ms interval the next call should always be allowed. + hb.handleGcSignal(shouldTriggerGc = true) + assert(hb.lastGcTriggerTimeNs >= firstTrigger) + } + + test("no GC when both signal is false and feature is disabled") { + val hb = makeHeartbeater(gcEnabled = false) + hb.handleGcSignal(shouldTriggerGc = false) + assert(hb.lastGcTriggerTimeNs == 0L) + } + + test("cooldown is measured from the last successful GC trigger") { + val hb = makeHeartbeater(gcEnabled = true, minIntervalMs = 60000L) + + // First trigger — sets lastGcTriggerTimeNs + hb.handleGcSignal(shouldTriggerGc = true) + val firstTrigger = hb.lastGcTriggerTimeNs + assert(firstTrigger > 0L) + + // Three more calls within cooldown — none should update lastGcTriggerTimeNs + hb.handleGcSignal(shouldTriggerGc = true) + hb.handleGcSignal(shouldTriggerGc = true) + hb.handleGcSignal(shouldTriggerGc = true) + assert(hb.lastGcTriggerTimeNs == firstTrigger) + } +} diff --git a/common/src/main/proto/TransportMessages.proto b/common/src/main/proto/TransportMessages.proto index a813a9e5015..223f981b7ca 100644 --- a/common/src/main/proto/TransportMessages.proto +++ b/common/src/main/proto/TransportMessages.proto @@ -510,6 +510,7 @@ message PbHeartbeatFromApplicationResponse { repeated PbWorkerInfo shuttingWorkers = 4; repeated int32 registeredShuffles = 5; PbCheckQuotaResponse checkQuotaResponse = 6; + bool shouldTriggerGc = 7; } message PbCheckQuota { 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 668fbcf79e4..fe04f82dfb8 100644 --- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala +++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala @@ -936,6 +936,10 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se def userQuotaEnabled: Boolean = get(USER_QUOTA_ENABLED) def quotaInterruptShuffleEnabled: Boolean = get(QUOTA_INTERRUPT_SHUFFLE_ENABLED) + def clusterOverloadGcEnabled: Boolean = get(MASTER_CLUSTER_OVERLOAD_GC_ENABLED) + def clientGcOnOverloadEnabled: Boolean = get(CLIENT_GC_ON_CLUSTER_OVERLOAD_ENABLED) + def clientGcOnOverloadMinIntervalMs: Long = get(CLIENT_GC_ON_CLUSTER_OVERLOAD_MIN_INTERVAL) + // ////////////////////////////////////////////////////// // Identity // // ////////////////////////////////////////////////////// @@ -2517,6 +2521,15 @@ object CelebornConf extends Logging { .timeConf(TimeUnit.MILLISECONDS) .createWithDefaultString("300s") + val MASTER_CLUSTER_OVERLOAD_GC_ENABLED: ConfigEntry[Boolean] = + buildConf("celeborn.master.clusterOverload.gc.enabled") + .categories("master") + .version("0.7.0") + .doc("Whether to enable the master signaling clients to trigger GC when the cluster " + + "storage is overloaded (any cluster quota dimension exceeds the overload threshold).") + .booleanConf + .createWithDefault(false) + val DFS_EXPIRE_DIRS_TIMEOUT: ConfigEntry[Long] = buildConf("celeborn.master.dfs.expireDirs.timeout") .categories("master") @@ -4776,6 +4789,26 @@ object CelebornConf extends Logging { .booleanConf .createWithDefault(true) + val CLIENT_GC_ON_CLUSTER_OVERLOAD_ENABLED: ConfigEntry[Boolean] = + buildConf("celeborn.client.clusterOverload.gc.enabled") + .categories("client") + .version("0.7.0") + .doc("When true, the client will trigger System.gc() upon receiving a GC signal from " + + "the master indicating cluster storage is overloaded. Disable to ignore the signal.") + .booleanConf + .createWithDefault(true) + + val CLIENT_GC_ON_CLUSTER_OVERLOAD_MIN_INTERVAL: ConfigEntry[Long] = + buildConf("celeborn.client.clusterOverload.gc.minInterval") + .categories("client") + .version("0.7.0") + .doc("Minimum time that must elapse between consecutive GC triggers on the client side " + + "in response to master GC signals. This prevents excessive GC pressure when the " + + "cluster remains overloaded across multiple heartbeat intervals.") + .timeConf(TimeUnit.MILLISECONDS) + .checkValue(_ >= 0, "Should be >= 0.") + .createWithDefaultString("5m") + val CLIENT_EXCLUDE_PEER_WORKER_ON_FAILURE_ENABLED: ConfigEntry[Boolean] = buildConf("celeborn.client.excludePeerWorkerOnFailure.enabled") .categories("client") @@ -6895,6 +6928,18 @@ object CelebornConf extends Logging { .checkValue(v => v >= 0.0 && v < 1.0, "Should be in [0.0, 1).") .createWithDefault(0.3) + val QUOTA_CLUSTER_OVERLOAD_FACTOR: ConfigEntry[Double] = + buildConf("celeborn.quota.overload.factor") + .categories("quota") + .dynamic + .doc("This config decides the quota * factor at which to consider the cluster 'overloaded'." + + " When the cluster is overloaded and `celeborn.master.clusterOverload.gc.enabled` is true," + + " application heartbeat responses contain a signal to trigger a GC to clean up dangling shuffle dependencies") + .version("0.7.0") + .doubleConf + .checkValue(v => v > 0.0 && v <= 1.0, "Should be in (0.0, 1.0].") + .createWithDefault(0.8) + val QUOTA_CLUSTER_DISK_BYTES_WRITTEN: ConfigEntry[Long] = buildConf("celeborn.quota.cluster.diskBytesWritten") .categories("quota") diff --git a/common/src/main/scala/org/apache/celeborn/common/protocol/message/ControlMessages.scala b/common/src/main/scala/org/apache/celeborn/common/protocol/message/ControlMessages.scala index e12d4b697f7..31c4c1fb0d7 100644 --- a/common/src/main/scala/org/apache/celeborn/common/protocol/message/ControlMessages.scala +++ b/common/src/main/scala/org/apache/celeborn/common/protocol/message/ControlMessages.scala @@ -397,7 +397,8 @@ object ControlMessages extends Logging { unknownWorkers: util.List[WorkerInfo], shuttingWorkers: util.List[WorkerInfo], registeredShuffles: util.List[Integer], - checkQuotaResponse: CheckQuotaResponse) extends Message + checkQuotaResponse: CheckQuotaResponse, + shouldTriggerGc: Boolean = false) extends Message case class CheckQuota(userIdentifier: UserIdentifier) extends Message @@ -890,7 +891,8 @@ object ControlMessages extends Logging { unknownWorkers, shuttingWorkers, registeredShuffles, - checkQuotaResponse) => + checkQuotaResponse, + shouldTriggerGc) => val pbCheckQuotaResponse = PbCheckQuotaResponse.newBuilder().setAvailable( checkQuotaResponse.isAvailable).setReason(checkQuotaResponse.reason) val payload = PbHeartbeatFromApplicationResponse.newBuilder() @@ -903,6 +905,7 @@ object ControlMessages extends Logging { shuttingWorkers.asScala.map(PbSerDeUtils.toPbWorkerInfo(_, true, true)).toList.asJava) .addAllRegisteredShuffles(registeredShuffles) .setCheckQuotaResponse(pbCheckQuotaResponse) + .setShouldTriggerGc(shouldTriggerGc) .build().toByteArray new TransportMessage(MessageType.HEARTBEAT_FROM_APPLICATION_RESPONSE, payload) @@ -1380,7 +1383,8 @@ object ControlMessages extends Logging { pbHeartbeatFromApplicationResponse.getShuttingWorkersList.asScala .map(PbSerDeUtils.fromPbWorkerInfo).toList.asJava, pbHeartbeatFromApplicationResponse.getRegisteredShufflesList, - checkQuotaResponse) + checkQuotaResponse, + pbHeartbeatFromApplicationResponse.getShouldTriggerGc) case CHECK_QUOTA_VALUE => val pbCheckAvailable = PbCheckQuota.parseFrom(message.getPayload) diff --git a/common/src/main/scala/org/apache/celeborn/common/quota/StorageQuota.scala b/common/src/main/scala/org/apache/celeborn/common/quota/StorageQuota.scala index 1a7a8e52abf..d99bcaf80b6 100644 --- a/common/src/main/scala/org/apache/celeborn/common/quota/StorageQuota.scala +++ b/common/src/main/scala/org/apache/celeborn/common/quota/StorageQuota.scala @@ -37,4 +37,5 @@ case class StorageQuota( object StorageQuota { val DEFAULT_QUOTA = StorageQuota(Long.MaxValue, Long.MaxValue, Long.MaxValue, Long.MaxValue) + val CLUSTER_OVERLOAD_LIMIT_DEFAULT_MULTIPLIER: Double = 0.8 } diff --git a/common/src/test/scala/org/apache/celeborn/common/util/PbSerDeUtilsTest.scala b/common/src/test/scala/org/apache/celeborn/common/util/PbSerDeUtilsTest.scala index 15c4df0fc71..a70413d1932 100644 --- a/common/src/test/scala/org/apache/celeborn/common/util/PbSerDeUtilsTest.scala +++ b/common/src/test/scala/org/apache/celeborn/common/util/PbSerDeUtilsTest.scala @@ -814,7 +814,8 @@ class PbSerDeUtilsTest extends CelebornFunSuite { mockWorkers("host1").toList.asJava, mockWorkers("host2").toList.asJava, Array(Integer.valueOf(1)).toList.asJava, - CheckQuotaResponse(isAvailable = false, "test_reason")) + CheckQuotaResponse(isAvailable = false, "test_reason"), + shouldTriggerGc = true) val toTransportHeartbeatFromApplicationResponse = ControlMessages.toTransportMessage(heartbeatFromApplicationResponse) val fromTransportHeartbeatFromApplicationResponse = @@ -822,6 +823,7 @@ class PbSerDeUtilsTest extends CelebornFunSuite { .asInstanceOf[HeartbeatFromApplicationResponse] assert(fromTransportHeartbeatFromApplicationResponse.equals(heartbeatFromApplicationResponse)) + assert(fromTransportHeartbeatFromApplicationResponse.shouldTriggerGc) } test("HeartbeatFromApplicationResponse backward compatibility without checkQuotaResponse") { diff --git a/docs/configuration/client.md b/docs/configuration/client.md index 5d15c6859fb..39ea80c035f 100644 --- a/docs/configuration/client.md +++ b/docs/configuration/client.md @@ -27,6 +27,8 @@ license: | | celeborn.client.application.uuidSuffix.enabled | false | false | Whether to add UUID suffix for application id for unique. When `true`, add UUID suffix for unique application id. Currently, this only applies to Spark and MR. | 0.6.0 | | | celeborn.client.chunk.prefetch.enabled | false | false | Whether to enable chunk prefetch when creating CelebornInputStream. | 0.5.1 | | | celeborn.client.closeIdleConnections | true | false | Whether client will close idle connections. | 0.3.0 | | +| celeborn.client.clusterOverload.gc.enabled | true | false | When true, the client will trigger System.gc() upon receiving a GC signal from the master indicating cluster storage is overloaded. Disable to ignore the signal. | 0.7.0 | | +| celeborn.client.clusterOverload.gc.minInterval | 5m | false | Minimum time that must elapse between consecutive GC triggers on the client side in response to master GC signals. This prevents excessive GC pressure when the cluster remains overloaded across multiple heartbeat intervals. | 0.7.0 | | | celeborn.client.commitFiles.ignoreExcludedWorker | false | false | When true, LifecycleManager will skip workers which are in the excluded list. | 0.3.0 | | | celeborn.client.eagerlyCreateInputStream.threads | 32 | false | Threads count for streamCreatorPool in CelebornShuffleReader. | 0.3.1 | | | celeborn.client.excludePeerWorkerOnFailure.enabled | true | false | When true, Celeborn will exclude partition's peer worker on failure when push data to replica failed. | 0.3.0 | | diff --git a/docs/configuration/master.md b/docs/configuration/master.md index 1f1c900c946..302fb32617f 100644 --- a/docs/configuration/master.md +++ b/docs/configuration/master.md @@ -37,6 +37,7 @@ license: | | celeborn.internal.port.enabled | false | false | Whether to create a internal port on Masters/Workers for inter-Masters/Workers communication. This is beneficial when SASL authentication is enforced for all interactions between clients and Celeborn Services, but the services can exchange messages without being subject to SASL authentication. | 0.5.0 | | | celeborn.logConf.enabled | false | false | When `true`, log the CelebornConf for debugging purposes. | 0.5.0 | | | celeborn.master.allowWorkerHostPattern | <undefined> | false | Pattern of worker host that allowed to register with the master. If not set, all workers are allowed to register. | 0.6.0 | | +| celeborn.master.clusterOverload.gc.enabled | false | false | Whether to enable the master signaling clients to trigger GC when the cluster storage is overloaded (any cluster quota dimension exceeds the overload threshold). | 0.7.0 | | | celeborn.master.denyWorkerHostPattern | <undefined> | false | Pattern of worker host that denied to register with the master. If not set, no workers are denied to register. | 0.6.0 | | | celeborn.master.dfs.expireDirs.timeout | 1h | false | The timeout for an expired dirs to be deleted on dfs like HDFS, S3, OSS. | 0.6.0 | | | celeborn.master.estimatedPartitionSize.initialSize | 64mb | false | Initial partition size for estimation, it will change according to runtime stats. | 0.3.0 | celeborn.shuffle.initialEstimatedPartitionSize | diff --git a/docs/configuration/quota.md b/docs/configuration/quota.md index 4174940e2a8..111492e6ec5 100644 --- a/docs/configuration/quota.md +++ b/docs/configuration/quota.md @@ -26,6 +26,7 @@ license: | | celeborn.quota.cluster.hdfsFileCount | 9223372036854775807 | true | Cluster level quota dynamic configuration for written hdfs file count. | 0.6.0 | | | celeborn.quota.enabled | true | false | When Master side sets to true, the master will enable to check the quota via QuotaManager. When Client side sets to true, LifecycleManager will request Master side to check whether the current user has enough quota before registration of shuffle. Fallback to the default shuffle service when Master side checks that there is no enough quota for current user. | 0.2.0 | | | celeborn.quota.interruptShuffle.enabled | false | false | Whether to enable interrupt shuffle when quota exceeds. | 0.6.0 | | +| celeborn.quota.overload.factor | 0.8 | true | This config decides the quota * factor at which to consider the cluster 'overloaded'. When the cluster is overloaded and `celeborn.master.clusterOverload.gc.enabled` is true, application heartbeat responses contain a signal to trigger a GC to clean up dangling shuffle dependencies | 0.7.0 | | | celeborn.quota.tenant.diskBytesWritten | 9223372036854775807b | true | Tenant level quota dynamic configuration for written disk bytes. | 0.5.0 | | | celeborn.quota.tenant.diskFileCount | 9223372036854775807 | true | Tenant level quota dynamic configuration for written disk file count. | 0.5.0 | | | celeborn.quota.tenant.enabled | true | false | Whether to enable tenant-level quota. | 0.6.0 | | diff --git a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala index 6b25552060a..bf14e73b929 100644 --- a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala +++ b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala @@ -1255,7 +1255,8 @@ private[celeborn] class Master( new util.ArrayList[WorkerInfo]( (statusSystem.shutdownWorkers.asScala ++ statusSystem.decommissionWorkers.asScala).asJava), new util.ArrayList(appRelatedShuffles), - quotaManager.checkApplicationQuotaStatus(appId))) + quotaManager.checkApplicationQuotaStatus(appId), + if (conf.clusterOverloadGcEnabled) quotaManager.isClusterOverloaded else false)) } else { context.reply(OneWayMessageResponse) } diff --git a/master/src/main/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManager.scala b/master/src/main/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManager.scala index 9b2f4d8818d..b03ffcb0377 100644 --- a/master/src/main/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManager.scala +++ b/master/src/main/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManager.scala @@ -49,6 +49,8 @@ class QuotaManager( val resourceConsumptionMetricsEnabled = celebornConf.masterResourceConsumptionMetricsEnabled @volatile var clusterQuotaStatus: QuotaStatus = new QuotaStatus() + @volatile + var clusterOverloaded: Boolean = false val appQuotaStatus: JMap[String, QuotaStatus] = JavaUtils.newConcurrentHashMap() val userResourceConsumptionCache: JMap[UserIdentifier, ResourceConsumption] = JavaUtils.newConcurrentHashMap() @@ -99,6 +101,10 @@ class QuotaManager( CheckQuotaResponse(!status.exceed, status.exceedReason) } + def isClusterOverloaded: Boolean = { + clusterOverloaded + } + def getUserStorageQuota(user: UserIdentifier): StorageQuota = { Option(configService) .map(_.getTenantUserConfigFromCache(user.tenantId, user.name).getUserStorageQuota) @@ -111,6 +117,12 @@ class QuotaManager( .getOrElse(StorageQuota.DEFAULT_QUOTA) } + def getClusterOverloadLimitFactor: Double = { + Option(configService) + .map(_.getSystemConfigFromCache.getClusterOverloadQuotaFactor.doubleValue()) + .getOrElse(StorageQuota.CLUSTER_OVERLOAD_LIMIT_DEFAULT_MULTIPLIER) + } + def getClusterStorageQuota: StorageQuota = { Option(configService) .map(_.getSystemConfigFromCache.getClusterStorageQuota) @@ -141,6 +153,28 @@ class QuotaManager( checkQuotaSpace(CLUSTER_EXHAUSTED, consumption, getClusterStorageQuota) } + // Calling the cluster overloaded (leads to gc triggers on app side to relieve storage) + private def checkClusterOverloaded(consumption: ResourceConsumption): Boolean = { + val overloadQuota = getClusterStorageQuota + val factor = getClusterOverloadLimitFactor + def scale(q: Long): Long = { + if (q <= 0L || q == Long.MaxValue) q + else math.max(1L, math.ceil(factor * q.toDouble).toLong) + } + + val threshold = StorageQuota( + diskBytesWritten = scale(overloadQuota.diskBytesWritten), + diskFileCount = scale(overloadQuota.diskFileCount), + hdfsBytesWritten = scale(overloadQuota.hdfsBytesWritten), + hdfsFileCount = scale(overloadQuota.hdfsFileCount)) + def exceeded(used: Long, limit: Long): Boolean = limit > 0L && used >= limit + + exceeded(consumption.diskBytesWritten, threshold.diskBytesWritten) || + exceeded(consumption.diskFileCount, threshold.diskFileCount) || + exceeded(consumption.hdfsBytesWritten, threshold.hdfsBytesWritten) || + exceeded(consumption.hdfsFileCount, threshold.hdfsFileCount) + } + private def checkQuotaSpace( reason: String, consumption: ResourceConsumption, @@ -261,6 +295,7 @@ class QuotaManager( // Expire cluster level exceeded app except already expired app clusterQuotaStatus = checkClusterQuotaSpace(clusterResourceConsumption) + clusterOverloaded = checkClusterOverloaded(clusterResourceConsumption) if (interruptShuffleEnabled && clusterQuotaStatus.exceed) { checkClusterResourceConsumption( tenantResourceConsumptions, diff --git a/master/src/test/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManagerSuite.scala b/master/src/test/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManagerSuite.scala index 425a3404b4f..ba4030b1f74 100644 --- a/master/src/test/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManagerSuite.scala +++ b/master/src/test/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManagerSuite.scala @@ -681,6 +681,136 @@ class QuotaManagerSuite extends CelebornFunSuite clearUserConsumption() } + test("test cluster overload - below threshold is not overloaded") { + val user = UserIdentifier("tenant_01", "Jerry") + // Cluster quota = 100GB disk, factor = 0.8 → overload threshold = 80GB + conf.set("celeborn.quota.cluster.diskBytesWritten", "100gb") + conf.set("celeborn.quota.overload.factor", "0.8") + configService.refreshCache() + + // 70GB is below the 80GB threshold + addUserConsumption(user, ResourceConsumption(Utils.byteStringAsBytes("70g"), 0, 0, 0)) + quotaManager.updateResourceConsumption() + + assertFalse(quotaManager.isClusterOverloaded) + clearUserConsumption() + } + + test("test cluster overload - above threshold is overloaded") { + val user = UserIdentifier("tenant_01", "Jerry") + // Cluster quota = 100GB disk, factor = 0.8 → overload threshold = 80GB + conf.set("celeborn.quota.cluster.diskBytesWritten", "100gb") + conf.set("celeborn.quota.overload.factor", "0.8") + configService.refreshCache() + + // 85GB is above the 80GB threshold + addUserConsumption(user, ResourceConsumption(Utils.byteStringAsBytes("85g"), 0, 0, 0)) + quotaManager.updateResourceConsumption() + + assertTrue(quotaManager.isClusterOverloaded) + clearUserConsumption() + } + + test("test cluster overload - at exact threshold boundary is overloaded") { + val user = UserIdentifier("tenant_01", "Jerry") + // Cluster quota = 100GB disk, factor = 0.8 → overload threshold = 80GB + conf.set("celeborn.quota.cluster.diskBytesWritten", "100gb") + conf.set("celeborn.quota.overload.factor", "0.8") + configService.refreshCache() + + // Consumption exactly at threshold (80GB): checkQuota uses value >= quota so this triggers overload + val thresholdBytes = (0.8 * Utils.byteStringAsBytes("100g")).toLong + addUserConsumption(user, ResourceConsumption(thresholdBytes, 0, 0, 0)) + quotaManager.updateResourceConsumption() + + assertTrue(quotaManager.isClusterOverloaded) + clearUserConsumption() + } + + test("test cluster overload - status resets when consumption drops below threshold") { + val user = UserIdentifier("tenant_01", "Jerry") + conf.set("celeborn.quota.cluster.diskBytesWritten", "100gb") + conf.set("celeborn.quota.overload.factor", "0.8") + configService.refreshCache() + + // First: consumption above threshold + addUserConsumption(user, ResourceConsumption(Utils.byteStringAsBytes("85g"), 0, 0, 0)) + quotaManager.updateResourceConsumption() + assertTrue(quotaManager.isClusterOverloaded) + + // Then: consumption drops below threshold + clearUserConsumption() + addUserConsumption(user, ResourceConsumption(Utils.byteStringAsBytes("70g"), 0, 0, 0)) + quotaManager.updateResourceConsumption() + assertFalse(quotaManager.isClusterOverloaded) + + clearUserConsumption() + } + + test("test cluster overload - custom factor changes the effective threshold") { + val user = UserIdentifier("tenant_01", "Jerry") + conf.set("celeborn.quota.cluster.diskBytesWritten", "100gb") + configService.refreshCache() + + // With factor=0.9, threshold = 90GB; 85GB should not trigger overload + conf.set("celeborn.quota.overload.factor", "0.9") + configService.refreshCache() + addUserConsumption(user, ResourceConsumption(Utils.byteStringAsBytes("85g"), 0, 0, 0)) + quotaManager.updateResourceConsumption() + assertFalse(quotaManager.isClusterOverloaded) + clearUserConsumption() + + // Same 85GB would trigger overload with a lower factor=0.8 (threshold=80GB) + conf.set("celeborn.quota.overload.factor", "0.8") + configService.refreshCache() + addUserConsumption(user, ResourceConsumption(Utils.byteStringAsBytes("85g"), 0, 0, 0)) + quotaManager.updateResourceConsumption() + assertTrue(quotaManager.isClusterOverloaded) + clearUserConsumption() + } + + test("test cluster overload - triggered by hdfs bytes exceeding threshold") { + val user = UserIdentifier("tenant_01", "Jerry") + conf.set("celeborn.quota.cluster.diskBytesWritten", "100gb") + conf.set("celeborn.quota.cluster.hdfsBytesWritten", "50gb") + conf.set("celeborn.quota.overload.factor", "0.8") + configService.refreshCache() + + // Disk usage is fine (10GB < 80GB threshold) but hdfs usage (45GB) exceeds hdfs threshold (40GB) + addUserConsumption( + user, + ResourceConsumption(Utils.byteStringAsBytes("10g"), 0, Utils.byteStringAsBytes("45g"), 0)) + quotaManager.updateResourceConsumption() + + assertTrue(quotaManager.isClusterOverloaded) + clearUserConsumption() + } + + test("test cluster overload - multiple dimensions, all below threshold") { + val user = UserIdentifier("tenant_01", "Jerry") + conf.set("celeborn.quota.cluster.diskBytesWritten", "100gb") + conf.set("celeborn.quota.cluster.diskFileCount", "10000") + conf.set("celeborn.quota.cluster.hdfsBytesWritten", "50gb") + conf.set("celeborn.quota.cluster.hdfsFileCount", "5000") + conf.set("celeborn.quota.overload.factor", "0.8") + configService.refreshCache() + + // All dimensions below their respective 80% thresholds + addUserConsumption( + user, + ResourceConsumption( + Utils.byteStringAsBytes("75g"), // disk: 75GB < 80GB threshold + 7000, // files: 7000 < 8000 threshold + Utils.byteStringAsBytes("35g"), // hdfs: 35GB < 40GB threshold + 3000 + ) + ) // hdfs files: 3000 < 4000 threshold + quotaManager.updateResourceConsumption() + + assertFalse(quotaManager.isClusterOverloaded) + clearUserConsumption() + } + def checkUserQuota(userIdentifier: UserIdentifier): CheckQuotaResponse = { quotaManager.checkUserQuotaStatus(userIdentifier) } diff --git a/service/src/main/java/org/apache/celeborn/server/common/service/config/DynamicConfig.java b/service/src/main/java/org/apache/celeborn/server/common/service/config/DynamicConfig.java index d55145f331b..ac0673996e6 100644 --- a/service/src/main/java/org/apache/celeborn/server/common/service/config/DynamicConfig.java +++ b/service/src/main/java/org/apache/celeborn/server/common/service/config/DynamicConfig.java @@ -169,6 +169,14 @@ public WorkerTagsMeta getWorkerTagsMeta() { ConfigType.STRING)); } + public Double getClusterOverloadQuotaFactor() { + return getValue( + CelebornConf.QUOTA_CLUSTER_OVERLOAD_FACTOR().key(), + CelebornConf.QUOTA_CLUSTER_OVERLOAD_FACTOR(), + Double.TYPE, + ConfigType.DOUBLE); + } + public StorageQuota getClusterStorageQuota() { return new StorageQuota( getValue( @@ -280,7 +288,8 @@ public enum ConfigType { BYTES, STRING, TIME_MS, - BOOLEAN + BOOLEAN, + DOUBLE } public static T convert(Class clazz, String value) {