This is an automated email from the ASF dual-hosted git repository.
nicholasjiang 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 05fa11b3a [CELEBORN-1174] Introduce application dimension resource
consumption metrics
05fa11b3a is described below
commit 05fa11b3a0557d6035aa62163f9d8ff584293879
Author: SteNicholas <[email protected]>
AuthorDate: Thu Feb 1 15:24:29 2024 +0800
[CELEBORN-1174] Introduce application dimension resource consumption metrics
### What changes were proposed in this pull request?
Introduce application dimension resource consumption metrics for
`ResourceConsumptionSource`.
### Why are the changes needed?
`ResourceConsumption` namespace metrics are generated for each user and
they are identified using a metric tag at present. It's recommended to
introduce application dimension resource consumption metrics that expose
application dimension resource consumption of Master and Worker. By monitoring
resource consumption in the application dimension, you can obtain the actual
situation of application resource consumption.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
- `WorkerInfoSuite#WorkerInfo toString output`
- `PbSerDeUtilsTest#fromAndToPbResourceConsumption`
- `MasterStateMachineSuitej#testObjSerde`
Closes #2161 from SteNicholas/CELEBORN-1174.
Authored-by: SteNicholas <[email protected]>
Signed-off-by: SteNicholas <[email protected]>
---
common/src/main/proto/TransportMessages.proto | 1 +
.../common/metrics/source/AbstractSource.scala | 26 ++++-
.../metrics/source/ResourceConsumptionSource.scala | 2 +
.../common/quota/ResourceConsumption.scala | 46 ++++++++-
.../apache/celeborn/common/util/PbSerDeUtils.scala | 25 ++++-
.../celeborn/common/meta/WorkerInfoSuite.scala | 17 ++-
.../celeborn/common/util/PbSerDeUtilsTest.scala | 18 +++-
docs/monitoring.md | 4 +-
.../deploy/master/clustermeta/MetaUtil.java | 69 +++++++++----
master/src/main/proto/Resource.proto | 1 +
.../celeborn/service/deploy/master/Master.scala | 114 +++++++++++++++------
.../clustermeta/ha/MasterStateMachineSuiteJ.java | 55 ++++++++--
.../celeborn/service/deploy/worker/Worker.scala | 97 ++++++++++++++----
.../deploy/worker/storage/StorageManager.scala | 41 +++++---
14 files changed, 405 insertions(+), 111 deletions(-)
diff --git a/common/src/main/proto/TransportMessages.proto
b/common/src/main/proto/TransportMessages.proto
index 534445fb3..8d7f2d83b 100644
--- a/common/src/main/proto/TransportMessages.proto
+++ b/common/src/main/proto/TransportMessages.proto
@@ -580,6 +580,7 @@ message PbResourceConsumption {
int64 diskFileCount = 2;
int64 hdfsBytesWritten = 3;
int64 hdfsFileCount = 4;
+ map<string, PbResourceConsumption> subResourceConsumptions = 5;
}
message PbAppDiskUsage {
diff --git
a/common/src/main/scala/org/apache/celeborn/common/metrics/source/AbstractSource.scala
b/common/src/main/scala/org/apache/celeborn/common/metrics/source/AbstractSource.scala
index 5ff032c57..8d76414c1 100644
---
a/common/src/main/scala/org/apache/celeborn/common/metrics/source/AbstractSource.scala
+++
b/common/src/main/scala/org/apache/celeborn/common/metrics/source/AbstractSource.scala
@@ -172,12 +172,30 @@ abstract class AbstractSource(conf: CelebornConf, role:
String)
val namedGauge = iter.next()
if (namedGauge.name.equals(name) &&
namedGauge.labelString.equals(labelString)) {
iter.remove()
- metricRegistry.remove(metricNameWithCustomizedLabels(name, labels))
+ removeGaugeMetric(name, namedGauge)
return
}
}
}
+ def removeGauge(name: String, labelKey: String, labelVal: String): Unit = {
+ val labels = Map(labelKey -> labelVal) ++ staticLabels
+
+ val iter = namedGauges.iterator()
+ while (iter.hasNext) {
+ val namedGauge = iter.next()
+ if (namedGauge.name.equals(name) &&
labels.toSet.subsetOf(namedGauge.labels.toSet)) {
+ iter.remove()
+ removeGaugeMetric(name, namedGauge)
+ return
+ }
+ }
+ }
+
+ def removeGaugeMetric(name: String, namedGauge: NamedGauge[_]): Unit = {
+ metricRegistry.remove(metricNameWithCustomizedLabelString(name,
namedGauge.labelString))
+ }
+
override def sample[T](metricsName: String, key: String)(f: => T): T = {
sample(metricsName, key, Map.empty[String, String])(f)
}
@@ -414,6 +432,12 @@ abstract class AbstractSource(conf: CelebornConf, role:
String)
metricsName + MetricLabels.labelString(labels ++ staticLabels)
}
}
+
+ protected def metricNameWithCustomizedLabelString(
+ metricsName: String,
+ labelString: String): String = {
+ metricsName + labelString
+ }
}
class TimerSupplier(val slidingWindowSize: Int)
diff --git
a/common/src/main/scala/org/apache/celeborn/common/metrics/source/ResourceConsumptionSource.scala
b/common/src/main/scala/org/apache/celeborn/common/metrics/source/ResourceConsumptionSource.scala
index df33310bb..88a4b9858 100644
---
a/common/src/main/scala/org/apache/celeborn/common/metrics/source/ResourceConsumptionSource.scala
+++
b/common/src/main/scala/org/apache/celeborn/common/metrics/source/ResourceConsumptionSource.scala
@@ -33,4 +33,6 @@ object ResourceConsumptionSource {
val HDFS_FILE_COUNT = "hdfsFileCount"
val HDFS_BYTES_WRITTEN = "hdfsBytesWritten"
+
+ val APPLICATION_LABEL = "applicationId"
}
diff --git
a/common/src/main/scala/org/apache/celeborn/common/quota/ResourceConsumption.scala
b/common/src/main/scala/org/apache/celeborn/common/quota/ResourceConsumption.scala
index 61a1b5318..10d1114b9 100644
---
a/common/src/main/scala/org/apache/celeborn/common/quota/ResourceConsumption.scala
+++
b/common/src/main/scala/org/apache/celeborn/common/quota/ResourceConsumption.scala
@@ -17,13 +17,18 @@
package org.apache.celeborn.common.quota
-import org.apache.celeborn.common.util.Utils
+import java.util
+
+import scala.collection.JavaConverters._
+
+import org.apache.celeborn.common.util.{CollectionUtils, Utils}
case class ResourceConsumption(
diskBytesWritten: Long,
diskFileCount: Long,
hdfsBytesWritten: Long,
- hdfsFileCount: Long) {
+ hdfsFileCount: Long,
+ var subResourceConsumptions: util.Map[String, ResourceConsumption] = null)
{
def add(other: ResourceConsumption): ResourceConsumption = {
ResourceConsumption(
@@ -33,10 +38,43 @@ case class ResourceConsumption(
hdfsFileCount + other.hdfsFileCount)
}
+ def addSubResourceConsumptions(otherSubResourceConsumptions: Map[
+ String,
+ ResourceConsumption]): Map[String, ResourceConsumption] = {
+ if (CollectionUtils.isNotEmpty(subResourceConsumptions)) {
+ subResourceConsumptions.asScala.foldRight(otherSubResourceConsumptions)(
+ (subResourceConsumption, resourceConsumptions) => {
+ if (resourceConsumptions.contains(subResourceConsumption._1)) {
+ resourceConsumptions + (subResourceConsumption._1 ->
subResourceConsumption._2.add(
+ resourceConsumptions(
+ subResourceConsumption._1)))
+ } else {
+ resourceConsumptions + (subResourceConsumption._1 ->
subResourceConsumption._2)
+ }
+ })
+ } else {
+ otherSubResourceConsumptions
+ }
+ }
+
+ def addWithSubResourceConsumptions(other: (ResourceConsumption, Map[String,
ResourceConsumption]))
+ : (ResourceConsumption, Map[String, ResourceConsumption]) = {
+ (add(other._1), addSubResourceConsumptions(other._2))
+ }
+
override def toString: String = {
+ val subResourceConsumptionString =
+ if (CollectionUtils.isEmpty(subResourceConsumptions)) {
+ "empty"
+ } else {
+ subResourceConsumptions.asScala.map { case (identifier,
resourceConsumption) =>
+ s"$identifier -> $resourceConsumption"
+ }.mkString("(", ",", ")")
+ }
s"ResourceConsumption(diskBytesWritten:
${Utils.bytesToString(diskBytesWritten)}," +
- s" diskFileCount: ${diskFileCount}," +
+ s" diskFileCount: $diskFileCount," +
s" hdfsBytesWritten: ${Utils.bytesToString(hdfsBytesWritten)}," +
- s" hdfsFileCount: ${hdfsFileCount})"
+ s" hdfsFileCount: $hdfsFileCount," +
+ s" subResourceConsumptions: $subResourceConsumptionString)"
}
}
diff --git
a/common/src/main/scala/org/apache/celeborn/common/util/PbSerDeUtils.scala
b/common/src/main/scala/org/apache/celeborn/common/util/PbSerDeUtils.scala
index d7e10fecf..6c4892c12 100644
--- a/common/src/main/scala/org/apache/celeborn/common/util/PbSerDeUtils.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/util/PbSerDeUtils.scala
@@ -169,15 +169,36 @@ object PbSerDeUtils {
pbResourceConsumption.getDiskBytesWritten,
pbResourceConsumption.getDiskFileCount,
pbResourceConsumption.getHdfsBytesWritten,
- pbResourceConsumption.getHdfsFileCount)
+ pbResourceConsumption.getHdfsFileCount,
+
fromPbSubResourceConsumptions(pbResourceConsumption.getSubResourceConsumptionsMap))
- def toPbResourceConsumption(resourceConsumption: ResourceConsumption):
PbResourceConsumption =
+ def toPbResourceConsumption(resourceConsumption: ResourceConsumption):
PbResourceConsumption = {
PbResourceConsumption.newBuilder
.setDiskBytesWritten(resourceConsumption.diskBytesWritten)
.setDiskFileCount(resourceConsumption.diskFileCount)
.setHdfsBytesWritten(resourceConsumption.hdfsBytesWritten)
.setHdfsFileCount(resourceConsumption.hdfsFileCount)
+ .putAllSubResourceConsumptions(toPbSubResourceConsumptions(
+ resourceConsumption.subResourceConsumptions))
.build
+ }
+
+ def fromPbSubResourceConsumptions(pbSubResourceConsumptions: util.Map[
+ String,
+ PbResourceConsumption]): util.Map[String, ResourceConsumption] =
+ if (CollectionUtils.isEmpty(pbSubResourceConsumptions))
+ null
+ else pbSubResourceConsumptions.asScala.map { case (key,
pbResourceConsumption) =>
+ (key, fromPbResourceConsumption(pbResourceConsumption))
+ }.asJava
+
+ def toPbSubResourceConsumptions(subResourceConsumptions: util.Map[String,
ResourceConsumption])
+ : util.Map[String, PbResourceConsumption] =
+ if (CollectionUtils.isEmpty(subResourceConsumptions))
+ new util.HashMap[String, PbResourceConsumption]
+ else subResourceConsumptions.asScala.map { case (key, resourceConsumption)
=>
+ (key, toPbResourceConsumption(resourceConsumption))
+ }.asJava
def fromPbUserResourceConsumption(pbUserResourceConsumption: util.Map[
String,
diff --git
a/common/src/test/scala/org/apache/celeborn/common/meta/WorkerInfoSuite.scala
b/common/src/test/scala/org/apache/celeborn/common/meta/WorkerInfoSuite.scala
index 1691c91ba..9eede93ed 100644
---
a/common/src/test/scala/org/apache/celeborn/common/meta/WorkerInfoSuite.scala
+++
b/common/src/test/scala/org/apache/celeborn/common/meta/WorkerInfoSuite.scala
@@ -22,6 +22,7 @@ import java.util.{Map => jMap}
import java.util.concurrent.{Future, ThreadLocalRandom}
import java.util.concurrent.atomic.AtomicInteger
+import scala.collection.JavaConverters._
import scala.collection.mutable.ArrayBuffer
import scala.concurrent.duration._
import scala.util.Random
@@ -218,7 +219,17 @@ class WorkerInfoSuite extends CelebornFunSuite {
JavaUtils.newConcurrentHashMap[UserIdentifier, ResourceConsumption]()
userResourceConsumption.put(
UserIdentifier("tenant1", "name1"),
- ResourceConsumption(20971520, 1, 52428800, 1))
+ ResourceConsumption(
+ 20971520,
+ 1,
+ 52428800,
+ 1,
+ Map(
+ "application_1697697127390_2171854" -> ResourceConsumption(
+ 20971520,
+ 1,
+ 52428800,
+ 1)).asJava))
val conf = new CelebornConf()
val endpointAddress = new RpcEndpointAddress(new RpcAddress("localhost",
12345), "mockRpc")
var rpcEnv: RpcEnv = null
@@ -288,9 +299,9 @@ class WorkerInfoSuite extends CelebornFunSuite {
| DiskInfo1: DiskInfo(maxSlots: 0, committed shuffles 0, running
applications 0, shuffleAllocations: Map(), mountPoint: disk1, usableSpace:
2048.0 MiB, avgFlushTime: 1 ns, avgFetchTime: 1 ns, activeSlots: 10,
storageType: SSD) status: HEALTHY dirs $placeholder
| DiskInfo2: DiskInfo(maxSlots: 0, committed shuffles 0, running
applications 0, shuffleAllocations: Map(), mountPoint: disk2, usableSpace:
2048.0 MiB, avgFlushTime: 2 ns, avgFetchTime: 2 ns, activeSlots: 20,
storageType: SSD) status: HEALTHY dirs $placeholder
|UserResourceConsumption: $placeholder
- | UserIdentifier: `tenant1`.`name1`, ResourceConsumption:
ResourceConsumption(diskBytesWritten: 20.0 MiB, diskFileCount: 1,
hdfsBytesWritten: 50.0 MiB, hdfsFileCount: 1)
+ | UserIdentifier: `tenant1`.`name1`, ResourceConsumption:
ResourceConsumption(diskBytesWritten: 20.0 MiB, diskFileCount: 1,
hdfsBytesWritten: 50.0 MiB, hdfsFileCount: 1, subResourceConsumptions:
(application_1697697127390_2171854 -> ResourceConsumption(diskBytesWritten:
20.0 MiB, diskFileCount: 1, hdfsBytesWritten: 50.0 MiB, hdfsFileCount: 1,
subResourceConsumptions: empty)))
|WorkerRef: null
- |""".stripMargin;
+ |""".stripMargin
assertEquals(
exp1,
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 de475b8ee..15b15ce7f 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
@@ -20,6 +20,8 @@ package org.apache.celeborn.common.util
import java.io.File
import java.util
+import scala.collection.JavaConverters._
+
import org.apache.celeborn.CelebornFunSuite
import org.apache.celeborn.common.identity.UserIdentifier
import org.apache.celeborn.common.meta.{DeviceInfo, DiskFileInfo, DiskInfo,
FileInfo, ReduceFileMeta, WorkerEventInfo, WorkerInfo, WorkerStatus}
@@ -73,7 +75,12 @@ class PbSerDeUtilsTest extends CelebornFunSuite {
val cache = JavaUtils.newConcurrentHashMap[String, UserIdentifier]()
val resourceConsumption1 = ResourceConsumption(1000, 2000, 3000, 4000)
- val resourceConsumption2 = ResourceConsumption(2000, 4000, 6000, 8000)
+ val resourceConsumption2 = ResourceConsumption(
+ 2000,
+ 4000,
+ 6000,
+ 8000,
+ Map("appld2" -> ResourceConsumption(2000, 4000, 6000, 8000)).asJava)
val userResourceConsumption = new util.HashMap[UserIdentifier,
ResourceConsumption]()
userResourceConsumption.put(userIdentifier1, resourceConsumption1)
userResourceConsumption.put(userIdentifier2, resourceConsumption2)
@@ -187,10 +194,15 @@ class PbSerDeUtilsTest extends CelebornFunSuite {
}
test("fromAndToPbResourceConsumption") {
- val pbResourceConsumption =
PbSerDeUtils.toPbResourceConsumption(resourceConsumption1)
+ testFromAndToPbResourceConsumption(resourceConsumption1)
+ testFromAndToPbResourceConsumption(resourceConsumption2)
+ }
+
+ def testFromAndToPbResourceConsumption(resourceConsumption:
ResourceConsumption): Unit = {
+ val pbResourceConsumption =
PbSerDeUtils.toPbResourceConsumption(resourceConsumption)
val restoredResourceConsumption =
PbSerDeUtils.fromPbResourceConsumption(pbResourceConsumption)
- assert(restoredResourceConsumption.equals(resourceConsumption1))
+ assert(restoredResourceConsumption.equals(resourceConsumption))
}
test("fromAndToPbUserResourceConsumption") {
diff --git a/docs/monitoring.md b/docs/monitoring.md
index 0f206a558..96e84ac6d 100644
--- a/docs/monitoring.md
+++ b/docs/monitoring.md
@@ -121,7 +121,8 @@ These metrics are exposed by Celeborn master.
- namespace=ResourceConsumption
- **notes:**
- - This metrics data is generated for each user and they are identified
using a metric tag.
+ - This metrics data is generated for each user and they are identified
using a metric tag.
+ - This metrics also include subResourceConsumptions generated for each
application of user and they are identified using `applicationId` tag.
- diskFileCount
- diskBytesWritten
- hdfsFileCount
@@ -296,6 +297,7 @@ These metrics are exposed by Celeborn worker.
- namespace=ResourceConsumption
- **notes:**
- This metrics data is generated for each user and they are identified
using a metric tag.
+ - This metrics also include subResourceConsumptions generated for each
application of user and they are identified using `applicationId` tag.
- diskFileCount
- diskBytesWritten
- hdfsFileCount
diff --git
a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/MetaUtil.java
b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/MetaUtil.java
index 6ec1f5068..c44fb0107 100644
---
a/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/MetaUtil.java
+++
b/master/src/main/java/org/apache/celeborn/service/deploy/master/clustermeta/MetaUtil.java
@@ -19,6 +19,7 @@ package org.apache.celeborn.service.deploy.master.clustermeta;
import java.util.HashMap;
import java.util.Map;
+import java.util.stream.Collectors;
import org.apache.celeborn.common.identity.UserIdentifier;
import org.apache.celeborn.common.identity.UserIdentifier$;
@@ -27,6 +28,7 @@ import org.apache.celeborn.common.meta.WorkerInfo;
import org.apache.celeborn.common.meta.WorkerStatus;
import org.apache.celeborn.common.protocol.StorageInfo;
import org.apache.celeborn.common.quota.ResourceConsumption;
+import org.apache.celeborn.common.util.CollectionUtils;
import org.apache.celeborn.common.util.Utils;
public class MetaUtil {
@@ -90,35 +92,64 @@ public class MetaUtil {
return map;
}
+ public static ResourceConsumption fromPbResourceConsumption(
+ ResourceProtos.ResourceConsumption pbResourceConsumption) {
+ return new ResourceConsumption(
+ pbResourceConsumption.getDiskBytesWritten(),
+ pbResourceConsumption.getDiskFileCount(),
+ pbResourceConsumption.getHdfsBytesWritten(),
+ pbResourceConsumption.getHdfsFileCount(),
+
fromPbSubResourceConsumptions(pbResourceConsumption.getSubResourceConsumptionMap()));
+ }
+
+ public static ResourceProtos.ResourceConsumption toPbResourceConsumption(
+ ResourceConsumption resourceConsumption) {
+ return ResourceProtos.ResourceConsumption.newBuilder()
+ .setDiskBytesWritten(resourceConsumption.diskBytesWritten())
+ .setDiskFileCount(resourceConsumption.diskFileCount())
+ .setHdfsBytesWritten(resourceConsumption.hdfsBytesWritten())
+ .setHdfsFileCount(resourceConsumption.hdfsFileCount())
+ .putAllSubResourceConsumption(
+
toPbSubResourceConsumptions(resourceConsumption.subResourceConsumptions()))
+ .build();
+ }
+
+ public static Map<String, ResourceConsumption> fromPbSubResourceConsumptions(
+ Map<String, ResourceProtos.ResourceConsumption>
pbSubResourceConsumptions) {
+ return CollectionUtils.isEmpty(pbSubResourceConsumptions)
+ ? null
+ : pbSubResourceConsumptions.entrySet().stream()
+ .collect(
+ Collectors.toMap(
+ Map.Entry::getKey,
+ resourceConsumption ->
+
fromPbResourceConsumption(resourceConsumption.getValue())));
+ }
+
+ public static Map<String, ResourceProtos.ResourceConsumption>
toPbSubResourceConsumptions(
+ Map<String, ResourceConsumption> subResourceConsumptions) {
+ return CollectionUtils.isEmpty(subResourceConsumptions)
+ ? new HashMap<>()
+ : subResourceConsumptions.entrySet().stream()
+ .collect(
+ Collectors.toMap(
+ Map.Entry::getKey,
+ resourceConsumption ->
+
toPbResourceConsumption(resourceConsumption.getValue())));
+ }
+
public static Map<UserIdentifier, ResourceConsumption>
fromPbUserResourceConsumption(
Map<String, ResourceProtos.ResourceConsumption>
pbUserResourceConsumption) {
Map<UserIdentifier, ResourceConsumption> map = new HashMap<>();
pbUserResourceConsumption.forEach(
- (k, v) -> {
- ResourceConsumption resourceConsumption =
- new ResourceConsumption(
- v.getDiskBytesWritten(),
- v.getDiskFileCount(),
- v.getHdfsBytesWritten(),
- v.getHdfsFileCount());
- map.put(UserIdentifier$.MODULE$.apply(k), resourceConsumption);
- });
+ (k, v) -> map.put(UserIdentifier$.MODULE$.apply(k),
fromPbResourceConsumption(v)));
return map;
}
public static Map<String, ResourceProtos.ResourceConsumption>
toPbUserResourceConsumption(
Map<UserIdentifier, ResourceConsumption> userResourceConsumption) {
Map<String, ResourceProtos.ResourceConsumption> map = new HashMap<>();
- userResourceConsumption.forEach(
- (k, v) ->
- map.put(
- k.toString(),
- ResourceProtos.ResourceConsumption.newBuilder()
- .setDiskBytesWritten(v.diskBytesWritten())
- .setDiskFileCount(v.diskFileCount())
- .setHdfsBytesWritten(v.hdfsBytesWritten())
- .setHdfsFileCount(v.hdfsFileCount())
- .build()));
+ userResourceConsumption.forEach((k, v) -> map.put(k.toString(),
toPbResourceConsumption(v)));
return map;
}
diff --git a/master/src/main/proto/Resource.proto
b/master/src/main/proto/Resource.proto
index 9ee195283..01635c187 100644
--- a/master/src/main/proto/Resource.proto
+++ b/master/src/main/proto/Resource.proto
@@ -203,6 +203,7 @@ message ResourceConsumption {
required int64 diskFileCount = 2;
required int64 hdfsBytesWritten = 3;
required int64 hdfsFileCount = 4;
+ map<string, ResourceConsumption> subResourceConsumption = 5;
}
enum Status {
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 4bb789822..79971f22b 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
@@ -43,7 +43,7 @@ import
org.apache.celeborn.common.protocol.message.{ControlMessages, StatusCode}
import org.apache.celeborn.common.protocol.message.ControlMessages._
import org.apache.celeborn.common.quota.{QuotaManager, ResourceConsumption}
import org.apache.celeborn.common.rpc._
-import org.apache.celeborn.common.util.{CelebornHadoopUtils, JavaUtils,
PbSerDeUtils, ThreadUtils, Utils}
+import org.apache.celeborn.common.util.{CelebornHadoopUtils, CollectionUtils,
JavaUtils, PbSerDeUtils, ThreadUtils, Utils}
import org.apache.celeborn.server.common.{HttpService, Service}
import
org.apache.celeborn.service.deploy.master.clustermeta.SingleMasterMetaManager
import org.apache.celeborn.service.deploy.master.clustermeta.ha.{HAHelper,
HAMasterMetaManager, MetaHandler}
@@ -844,6 +844,8 @@ private[celeborn] class Master(
nonEagerHandler.submit(new Runnable {
override def run(): Unit = {
statusSystem.handleAppLost(appId, requestId)
+ // Resource consumption source should remove lose application gauges.
+ removeAppResourceConsumption(appId)
logInfo(s"Removed application $appId")
if (hasHDFSStorage) {
checkAndCleanExpiredAppDirsOnHDFS(appId)
@@ -853,6 +855,30 @@ private[celeborn] class Master(
})
}
+ private def removeAppResourceConsumption(appId: String): Unit = {
+ removeResourceConsumptionGauge(
+ ResourceConsumptionSource.DISK_FILE_COUNT,
+ appId)
+ removeResourceConsumptionGauge(
+ ResourceConsumptionSource.DISK_BYTES_WRITTEN,
+ appId)
+ removeResourceConsumptionGauge(
+ ResourceConsumptionSource.HDFS_FILE_COUNT,
+ appId)
+ removeResourceConsumptionGauge(
+ ResourceConsumptionSource.HDFS_BYTES_WRITTEN,
+ appId)
+ }
+
+ private def removeResourceConsumptionGauge(
+ resourceConsumptionName: String,
+ appId: String): Unit = {
+ resourceConsumptionSource.removeGauge(
+ resourceConsumptionName,
+ ResourceConsumptionSource.APPLICATION_LABEL,
+ appId)
+ }
+
private def checkAndCleanExpiredAppDirsOnHDFS(expiredDir: String = ""): Unit
= {
if (hadoopFs == null) {
try {
@@ -914,55 +940,77 @@ private[celeborn] class Master(
statusSystem.handleRemoveWorkersUnavailableInfo(unavailableWorkers,
requestId)
}
- private def computeUserResourceConsumption(userIdentifier: UserIdentifier)
- : ResourceConsumption = {
- val current = System.currentTimeMillis()
- if (userResourceConsumptions.containsKey(userIdentifier)) {
- val resourceConsumptionAndUpdateTime =
userResourceConsumptions.get(userIdentifier)
- if (current - resourceConsumptionAndUpdateTime._2 >
masterResourceConsumptionInterval) {
- val newResourceConsumption = statusSystem.workers.asScala.flatMap {
workerInfo =>
- workerInfo.userResourceConsumption.asScala.get(userIdentifier)
- }.foldRight(ResourceConsumption(0, 0, 0, 0))(_ add _)
- userResourceConsumptions.put(userIdentifier, (newResourceConsumption,
current))
- newResourceConsumption
- } else {
- resourceConsumptionAndUpdateTime._1
- }
- } else {
- val newResourceConsumption = statusSystem.workers.asScala.flatMap {
workerInfo =>
- workerInfo.userResourceConsumption.asScala.get(userIdentifier)
- }.foldRight(ResourceConsumption(0, 0, 0, 0))(_ add _)
- userResourceConsumptions.put(userIdentifier, (newResourceConsumption,
current))
- newResourceConsumption
+ private def handleResourceConsumption(userIdentifier: UserIdentifier):
ResourceConsumption = {
+ val userResourceConsumption =
computeUserResourceConsumption(userIdentifier)
+ gaugeResourceConsumption(userIdentifier)
+ val subResourceConsumptions =
userResourceConsumption.subResourceConsumptions
+ if (CollectionUtils.isNotEmpty(subResourceConsumptions)) {
+ subResourceConsumptions.asScala.keys.foreach {
gaugeResourceConsumption(userIdentifier, _) }
}
+ userResourceConsumption
}
- private def handleCheckQuota(
+ private def gaugeResourceConsumption(
userIdentifier: UserIdentifier,
- context: RpcCallContext): Unit = {
-
+ applicationId: String = null): Unit = {
+ val resourceConsumptionLabel =
+ if (applicationId == null) userIdentifier.toMap
+ else userIdentifier.toMap + (ResourceConsumptionSource.APPLICATION_LABEL
-> applicationId)
resourceConsumptionSource.addGauge(
ResourceConsumptionSource.DISK_FILE_COUNT,
- userIdentifier.toMap) { () =>
- computeUserResourceConsumption(userIdentifier).diskFileCount
+ resourceConsumptionLabel) { () =>
+ computeResourceConsumption(userIdentifier).diskFileCount
}
resourceConsumptionSource.addGauge(
ResourceConsumptionSource.DISK_BYTES_WRITTEN,
- userIdentifier.toMap) { () =>
- computeUserResourceConsumption(userIdentifier).diskBytesWritten
+ resourceConsumptionLabel) { () =>
+ computeResourceConsumption(userIdentifier).diskBytesWritten
}
resourceConsumptionSource.addGauge(
ResourceConsumptionSource.HDFS_FILE_COUNT,
- userIdentifier.toMap) { () =>
- computeUserResourceConsumption(userIdentifier).hdfsFileCount
+ resourceConsumptionLabel) { () =>
+ computeResourceConsumption(userIdentifier).hdfsFileCount
}
resourceConsumptionSource.addGauge(
ResourceConsumptionSource.HDFS_BYTES_WRITTEN,
- userIdentifier.toMap) { () =>
- computeUserResourceConsumption(userIdentifier).hdfsBytesWritten
+ resourceConsumptionLabel) { () =>
+ computeResourceConsumption(userIdentifier).hdfsBytesWritten
}
+ }
- val userResourceConsumption =
computeUserResourceConsumption(userIdentifier)
+ private def computeResourceConsumption(
+ userIdentifier: UserIdentifier,
+ applicationId: String = null): ResourceConsumption = {
+ val newResourceConsumption = computeUserResourceConsumption(userIdentifier)
+ if (applicationId == null) {
+ val current = System.currentTimeMillis()
+ if (userResourceConsumptions.containsKey(userIdentifier)) {
+ val resourceConsumptionAndUpdateTime =
userResourceConsumptions.get(userIdentifier)
+ if (current - resourceConsumptionAndUpdateTime._2 <=
masterResourceConsumptionInterval) {
+ return resourceConsumptionAndUpdateTime._1
+ }
+ }
+ userResourceConsumptions.put(userIdentifier, (newResourceConsumption,
current))
+ newResourceConsumption
+ } else {
+ newResourceConsumption.subResourceConsumptions.get(applicationId)
+ }
+ }
+
+ private def computeUserResourceConsumption(
+ userIdentifier: UserIdentifier): ResourceConsumption = {
+ val (resourceConsumption, subResourceConsumptions) =
statusSystem.workers.asScala.flatMap {
+ workerInfo =>
workerInfo.userResourceConsumption.asScala.get(userIdentifier)
+ }.foldRight((ResourceConsumption(0, 0, 0, 0), Map.empty[String,
ResourceConsumption]))(
+ _ addWithSubResourceConsumptions _)
+ resourceConsumption.subResourceConsumptions =
subResourceConsumptions.asJava
+ resourceConsumption
+ }
+
+ private def handleCheckQuota(
+ userIdentifier: UserIdentifier,
+ context: RpcCallContext): Unit = {
+ val userResourceConsumption = handleResourceConsumption(userIdentifier)
val quota = quotaManager.getQuota(userIdentifier)
val (isAvailable, reason) =
quota.checkQuotaSpaceAvailable(userIdentifier, userResourceConsumption)
diff --git
a/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MasterStateMachineSuiteJ.java
b/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MasterStateMachineSuiteJ.java
index b202961da..79fb047c5 100644
---
a/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MasterStateMachineSuiteJ.java
+++
b/master/src/test/java/org/apache/celeborn/service/deploy/master/clustermeta/ha/MasterStateMachineSuiteJ.java
@@ -19,6 +19,7 @@ package
org.apache.celeborn.service.deploy.master.clustermeta.ha;
import java.io.File;
import java.io.IOException;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
@@ -116,11 +117,23 @@ public class MasterStateMachineSuiteJ extends
RatisBaseSuiteJ {
Map<UserIdentifier, ResourceConsumption> userResourceConsumption1 =
JavaUtils.newConcurrentHashMap();
userResourceConsumption1.put(
- new UserIdentifier("tenant1", "name1"), new ResourceConsumption(1000,
1, 1000, 1));
+ new UserIdentifier("tenant1", "name1"), new ResourceConsumption(1000,
1, 1000, 1, null));
userResourceConsumption1.put(
- new UserIdentifier("tenant1", "name2"), new ResourceConsumption(2000,
2, 2000, 2));
+ new UserIdentifier("tenant1", "name2"),
+ new ResourceConsumption(
+ 2000,
+ 2,
+ 2000,
+ 2,
+ Collections.singletonMap("appId2", new ResourceConsumption(2000,
2, 2000, 2, null))));
userResourceConsumption1.put(
- new UserIdentifier("tenant1", "name3"), new ResourceConsumption(3000,
3, 3000, 3));
+ new UserIdentifier("tenant1", "name3"),
+ new ResourceConsumption(
+ 3000,
+ 3,
+ 3000,
+ 3,
+ Collections.singletonMap("appId3", new ResourceConsumption(2000,
2, 2000, 2, null))));
Map<String, DiskInfo> disks2 = new HashMap<>();
disks2.put("disk1", new DiskInfo("disk1", 64 * 1024 * 1024 * 1024L, 100,
100, 0));
@@ -129,11 +142,23 @@ public class MasterStateMachineSuiteJ extends
RatisBaseSuiteJ {
Map<UserIdentifier, ResourceConsumption> userResourceConsumption2 =
JavaUtils.newConcurrentHashMap();
userResourceConsumption2.put(
- new UserIdentifier("tenant2", "name1"), new ResourceConsumption(1000,
1, 1000, 1));
+ new UserIdentifier("tenant2", "name1"), new ResourceConsumption(1000,
1, 1000, 1, null));
userResourceConsumption2.put(
- new UserIdentifier("tenant2", "name2"), new ResourceConsumption(2000,
2, 2000, 2));
+ new UserIdentifier("tenant2", "name2"),
+ new ResourceConsumption(
+ 2000,
+ 2,
+ 2000,
+ 2,
+ Collections.singletonMap("appId2", new ResourceConsumption(2000,
2, 2000, 2, null))));
userResourceConsumption2.put(
- new UserIdentifier("tenant2", "name3"), new ResourceConsumption(3000,
3, 3000, 3));
+ new UserIdentifier("tenant2", "name3"),
+ new ResourceConsumption(
+ 3000,
+ 3,
+ 3000,
+ 3,
+ Collections.singletonMap("appId3", new ResourceConsumption(2000,
2, 2000, 2, null))));
Map<String, DiskInfo> disks3 = new HashMap<>();
disks3.put("disk1", new DiskInfo("disk1", 64 * 1024 * 1024 * 1024L, 100,
100, 0));
@@ -142,11 +167,23 @@ public class MasterStateMachineSuiteJ extends
RatisBaseSuiteJ {
Map<UserIdentifier, ResourceConsumption> userResourceConsumption3 =
JavaUtils.newConcurrentHashMap();
userResourceConsumption3.put(
- new UserIdentifier("tenant3", "name1"), new ResourceConsumption(1000,
1, 1000, 1));
+ new UserIdentifier("tenant3", "name1"), new ResourceConsumption(1000,
1, 1000, 1, null));
userResourceConsumption3.put(
- new UserIdentifier("tenant3", "name2"), new ResourceConsumption(2000,
2, 2000, 2));
+ new UserIdentifier("tenant3", "name2"),
+ new ResourceConsumption(
+ 2000,
+ 2,
+ 2000,
+ 2,
+ Collections.singletonMap("appId2", new ResourceConsumption(2000,
2, 2000, 2, null))));
userResourceConsumption3.put(
- new UserIdentifier("tenant3", "name3"), new ResourceConsumption(3000,
3, 3000, 3));
+ new UserIdentifier("tenant3", "name3"),
+ new ResourceConsumption(
+ 3000,
+ 3,
+ 3000,
+ 3,
+ Collections.singletonMap("appId3", new ResourceConsumption(2000,
2, 2000, 2, null))));
WorkerInfo info1 = new WorkerInfo("host1", 1, 2, 3, 10, disks1,
userResourceConsumption1);
WorkerInfo info2 = new WorkerInfo("host2", 4, 5, 6, 11, disks2,
userResourceConsumption2);
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 ec170e64a..383820543 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
@@ -44,7 +44,7 @@ import
org.apache.celeborn.common.protocol.PbWorkerStatus.State
import org.apache.celeborn.common.protocol.message.ControlMessages._
import org.apache.celeborn.common.quota.ResourceConsumption
import org.apache.celeborn.common.rpc._
-import org.apache.celeborn.common.util.{CelebornExitKind, JavaUtils,
ShutdownHookManager, ThreadUtils, Utils}
+import org.apache.celeborn.common.util.{CelebornExitKind, CollectionUtils,
JavaUtils, ShutdownHookManager, ThreadUtils, Utils}
// Can Remove this if celeborn don't support scala211 in future
import org.apache.celeborn.common.util.FunctionConverter._
import org.apache.celeborn.server.common.{HttpService, Service}
@@ -532,31 +532,56 @@ private[celeborn] class Worker(
private def handleResourceConsumption(): util.Map[UserIdentifier,
ResourceConsumption] = {
val resourceConsumptionSnapshot =
storageManager.userResourceConsumptionSnapshot()
- resourceConsumptionSnapshot.foreach { case (userIdentifier, _) =>
- resourceConsumptionSource.addGauge(
- ResourceConsumptionSource.DISK_FILE_COUNT,
- userIdentifier.toMap) { () =>
- workerInfo.userResourceConsumption.get(userIdentifier).diskFileCount
- }
- 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
+ resourceConsumptionSnapshot.foreach { case (userIdentifier,
userResourceConsumption) =>
+ gaugeResourceConsumption(userIdentifier)
+ val subResourceConsumptions =
userResourceConsumption.subResourceConsumptions
+ if (CollectionUtils.isNotEmpty(subResourceConsumptions)) {
+ subResourceConsumptions.asScala.keys.foreach {
gaugeResourceConsumption(userIdentifier, _) }
}
}
workerInfo.updateThenGetUserResourceConsumption(resourceConsumptionSnapshot.asJava)
}
+ private def gaugeResourceConsumption(
+ userIdentifier: UserIdentifier,
+ applicationId: String = null): Unit = {
+ var resourceConsumptionLabel = userIdentifier.toMap
+ if (applicationId != null)
+ resourceConsumptionLabel += (ResourceConsumptionSource.APPLICATION_LABEL
-> applicationId)
+ resourceConsumptionSource.addGauge(
+ ResourceConsumptionSource.DISK_FILE_COUNT,
+ resourceConsumptionLabel) { () =>
+ computeResourceConsumption(userIdentifier, applicationId).diskFileCount
+ }
+ resourceConsumptionSource.addGauge(
+ ResourceConsumptionSource.DISK_BYTES_WRITTEN,
+ resourceConsumptionLabel) { () =>
+ computeResourceConsumption(userIdentifier,
applicationId).diskBytesWritten
+ }
+ resourceConsumptionSource.addGauge(
+ ResourceConsumptionSource.HDFS_FILE_COUNT,
+ resourceConsumptionLabel) { () =>
+ computeResourceConsumption(userIdentifier, applicationId).hdfsFileCount
+ }
+ resourceConsumptionSource.addGauge(
+ ResourceConsumptionSource.HDFS_BYTES_WRITTEN,
+ resourceConsumptionLabel) { () =>
+ computeResourceConsumption(userIdentifier,
applicationId).hdfsBytesWritten
+ }
+ }
+
+ private def computeResourceConsumption(
+ userIdentifier: UserIdentifier,
+ applicationId: String = null): ResourceConsumption = {
+ var resourceConsumption =
workerInfo.userResourceConsumption.get(userIdentifier)
+ if (applicationId != null) {
+ resourceConsumption =
resourceConsumption.subResourceConsumptions.getOrDefault(
+ applicationId,
+ ResourceConsumption(0, 0, 0, 0))
+ }
+ resourceConsumption
+ }
+
@VisibleForTesting
def cleanup(expiredShuffleKeys: JHashSet[String], threadPool:
ThreadPoolExecutor): Unit =
synchronized {
@@ -567,6 +592,12 @@ private[celeborn] class Worker(
shuffleMapperAttempts.remove(shuffleKey)
shuffleCommitInfos.remove(shuffleKey)
workerInfo.releaseSlots(shuffleKey)
+ val applicationId = Utils.splitShuffleKey(shuffleKey)._1
+ if (!workerInfo.getApplicationIdSet.contains(applicationId)) {
+ // When the running applications does not contain the application
corresponding to expired shuffle key,
+ // resource consumption source should remove lose application gauges.
+ removeAppResourceConsumption(applicationId)
+ }
logInfo(s"Cleaned up expired shuffle $shuffleKey")
}
partitionsSorter.cleanup(expiredShuffleKeys)
@@ -576,6 +607,30 @@ private[celeborn] class Worker(
})
}
+ private def removeAppResourceConsumption(applicationId: String): Unit = {
+ removeResourceConsumptionGauge(
+ ResourceConsumptionSource.DISK_FILE_COUNT,
+ applicationId)
+ removeResourceConsumptionGauge(
+ ResourceConsumptionSource.DISK_BYTES_WRITTEN,
+ applicationId)
+ removeResourceConsumptionGauge(
+ ResourceConsumptionSource.HDFS_FILE_COUNT,
+ applicationId)
+ removeResourceConsumptionGauge(
+ ResourceConsumptionSource.HDFS_BYTES_WRITTEN,
+ applicationId)
+ }
+
+ private def removeResourceConsumptionGauge(
+ resourceConsumptionName: String,
+ applicationId: String): Unit = {
+ resourceConsumptionSource.removeGauge(
+ resourceConsumptionName,
+ ResourceConsumptionSource.APPLICATION_LABEL,
+ applicationId)
+ }
+
override def getWorkerInfo: String = {
val sb = new StringBuilder
sb.append("====================== WorkerInfo of Worker
===========================\n")
diff --git
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala
index eafc3e5a7..ea84272bf 100644
---
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala
+++
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala
@@ -738,32 +738,43 @@ final private[worker] class StorageManager(conf:
CelebornConf, workerSource: Abs
diskFileInfos
.asScala
.toList
- .flatMap { case (_, fileInfoMaps) =>
+ .flatMap { case (shuffleKey, fileInfoMaps) =>
// userIdentifier -> fileInfo
fileInfoMaps.values().asScala.map { fileInfo =>
- (fileInfo.getUserIdentifier, fileInfo)
+ (fileInfo.getUserIdentifier,
(Utils.splitShuffleKey(shuffleKey)._1, fileInfo))
}
}
- // userIdentifier -> List((userIdentifier, fileInfo))
+ // userIdentifier -> List((userIdentifier, (applicationId, fileInfo))))
.groupBy(_._1)
.map { case (userIdentifier, userWithFileInfoList) =>
// collect resource consumed by each user on this worker
- val resourceConsumption = {
- val userFileInfos = userWithFileInfoList.map(_._2)
- val diskFileInfos = userFileInfos.filter(!_.isHdfs)
- val hdfsFileInfos = userFileInfos.filter(_.isHdfs)
-
- val diskBytesWritten = diskFileInfos.map(_.getFileLength).sum
- val diskFileCount = diskFileInfos.size
- val hdfsBytesWritten = hdfsFileInfos.map(_.getFileLength).sum
- val hdfsFileCount = hdfsFileInfos.size
- ResourceConsumption(diskBytesWritten, diskFileCount,
hdfsBytesWritten, hdfsFileCount)
- }
- (userIdentifier, resourceConsumption)
+ val userFileInfos = userWithFileInfoList.map(_._2)
+ (
+ userIdentifier,
+ resourceConsumption(
+ userFileInfos.map(_._2),
+ userFileInfos.groupBy(_._1).map {
+ case (applicationId, appWithFileInfoList) =>
+ (applicationId,
resourceConsumption(appWithFileInfoList.map(_._2)))
+ }.asJava))
}
}
}
+ def resourceConsumption(
+ fileInfos: List[DiskFileInfo],
+ subResourceConsumptions: util.Map[String, ResourceConsumption] = null)
+ : ResourceConsumption = {
+ val diskFileInfos = fileInfos.filter(!_.isHdfs)
+ val hdfsFileInfos = fileInfos.filter(_.isHdfs)
+ ResourceConsumption(
+ diskFileInfos.map(_.getFileLength).sum,
+ diskFileInfos.size,
+ hdfsFileInfos.map(_.getFileLength).sum,
+ hdfsFileInfos.size,
+ subResourceConsumptions)
+ }
+
def notifyFileInfoCommitted(
shuffleKey: String,
fileName: String,
