This is an automated email from the ASF dual-hosted git repository.
nicholasjiang pushed a commit to branch branch-0.4
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
The following commit(s) were added to refs/heads/branch-0.4 by this push:
new f7c8c9dd7 [CELEBORN-1174][0.4] Introduce application dimension
resource consumption metrics
f7c8c9dd7 is described below
commit f7c8c9dd751208c579e76332ed33e03f06d6265b
Author: SteNicholas <[email protected]>
AuthorDate: Fri Feb 2 20:21:43 2024 +0800
[CELEBORN-1174][0.4] Introduce application dimension resource consumption
metrics
### What changes were proposed in this pull request?
Cherry pick #2161.
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 #2279 from SteNicholas/CELEBORN-1174-0.4.
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 | 115 +++++++++++++++------
.../clustermeta/ha/MasterStateMachineSuiteJ.java | 55 ++++++++--
.../celeborn/service/deploy/worker/Worker.scala | 97 +++++++++++++----
.../deploy/worker/storage/StorageManager.scala | 41 +++++---
14 files changed, 405 insertions(+), 112 deletions(-)
diff --git a/common/src/main/proto/TransportMessages.proto
b/common/src/main/proto/TransportMessages.proto
index d172d3e0a..c1f011278 100644
--- a/common/src/main/proto/TransportMessages.proto
+++ b/common/src/main/proto/TransportMessages.proto
@@ -512,6 +512,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 d681152f7..1a39ce6db 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
@@ -156,15 +156,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 fc725f3b8..0e4606399 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.util.Random
@@ -217,7 +218,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
@@ -287,9 +298,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,
worker1.toString.replaceAll("HeartbeatElapsedSeconds:.*\n", ""))
assertEquals(exp2,
worker2.toString.replaceAll("HeartbeatElapsedSeconds:.*\n", ""))
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 64b6f7e86..220ce2566 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, DiskInfo, FileInfo,
WorkerInfo}
@@ -62,7 +64,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)
@@ -167,10 +174,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 89dffbef0..49a37c345 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 03e1d21e3..c6abe41f2 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$;
@@ -26,6 +27,7 @@ import org.apache.celeborn.common.meta.DiskInfo;
import org.apache.celeborn.common.meta.WorkerInfo;
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 {
@@ -89,35 +91,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 78dc477bd..841ee2750 100644
--- a/master/src/main/proto/Resource.proto
+++ b/master/src/main/proto/Resource.proto
@@ -170,6 +170,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 4a0df0a1b..ad35ae005 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
@@ -42,8 +42,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.rpc.netty.NettyRpcEnv
-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}
@@ -783,6 +782,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)
@@ -792,6 +793,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 {
@@ -853,55 +878,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 cca324445..5f87d2a52 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
@@ -43,7 +43,7 @@ import org.apache.celeborn.common.protocol.{PartitionType,
PbRegisterWorkerRespo
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}
@@ -526,31 +526,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 {
@@ -561,6 +586,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)
@@ -570,6 +601,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 79b03c709..624925c1b 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
@@ -815,32 +815,43 @@ final private[worker] class StorageManager(conf:
CelebornConf, workerSource: Abs
fileInfos
.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[FileInfo],
+ 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,
