[CELEBORN-1800] Introduce ApplicationTotalCount and ApplicationFallbackCount metric to record the total and fallback count of application

Author: SteNicholas

assets/grafana/celeborn-dashboard.json

@@ -1472,7 +1472,7 @@
             "type": "prometheus",
             "uid": "${DS_PROMETHEUS}"
           },
-          "description": "The total count of shuffle including celeborn shuffle and spark built-in shuffle.",
+          "description": "The total count of shuffle including celeborn shuffle and engine built-in shuffle.",
           "fieldConfig": {
             "defaults": {
               "color": {
@@ -1646,6 +1646,186 @@
           ],
           "title": "metrics_ShuffleFallbackCount_Value",
           "type": "timeseries"
+        },
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${DS_PROMETHEUS}"
+          },
+          "description": "The total count of application running with celeborn shuffle and engine built-in shuffle.",
+          "fieldConfig": {
+            "defaults": {
+              "color": {
+                "mode": "palette-classic"
+              },
+              "custom": {
+                "axisCenteredZero": false,
+                "axisColorMode": "text",
+                "axisLabel": "",
+                "axisPlacement": "auto",
+                "barAlignment": 0,
+                "drawStyle": "line",
+                "fillOpacity": 0,
+                "gradientMode": "none",
+                "hideFrom": {
+                  "legend": false,
+                  "tooltip": false,
+                  "viz": false
+                },
+                "lineInterpolation": "linear",
+                "lineWidth": 1,
+                "pointSize": 5,
+                "scaleDistribution": {
+                  "type": "linear"
+                },
+                "showPoints": "auto",
+                "spanNulls": false,
+                "stacking": {
+                  "group": "A",
+                  "mode": "none"
+                },
+                "thresholdsStyle": {
+                  "mode": "off"
+                }
+              },
+              "mappings": [],
+              "thresholds": {
+                "mode": "absolute",
+                "steps": [
+                  {
+                    "color": "green"
+                  },
+                  {
+                    "color": "red",
+                    "value": 80
+                  }
+                ]
+              }
+            },
+            "overrides": []
+          },
+          "gridPos": {
+            "h": 8,
+            "w": 12,
+            "x": 0,
+            "y": 90
+          },
+          "id": 234,
+          "options": {
+            "legend": {
+              "calcs": [],
+              "displayMode": "list",
+              "placement": "bottom",
+              "showLegend": true
+            },
+            "tooltip": {
+              "maxHeight": 600,
+              "mode": "single",
+              "sort": "none"
+            }
+          },
+          "targets": [
+            {
+              "datasource": {
+                "type": "prometheus",
+                "uid": "${DS_PROMETHEUS}"
+              },
+              "expr": "metrics_ApplicationTotalCount_Value{instance=~\"${instance}\"}",
+              "legendFormat": "${baseLegend}",
+              "range": true,
+              "refId": "A"
+            }
+          ],
+          "title": "metrics_ApplicationTotalCount_Value",
+          "type": "timeseries"
+        },
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${DS_PROMETHEUS}"
+          },
+          "description": "The count of application fallbacks.",
+          "fieldConfig": {
+            "defaults": {
+              "color": {
+                "mode": "palette-classic"
+              },
+              "custom": {
+                "axisCenteredZero": false,
+                "axisColorMode": "text",
+                "axisLabel": "",
+                "axisPlacement": "auto",
+                "barAlignment": 0,
+                "drawStyle": "line",
+                "fillOpacity": 0,
+                "gradientMode": "none",
+                "hideFrom": {
+                  "legend": false,
+                  "tooltip": false,
+                  "viz": false
+                },
+                "lineInterpolation": "linear",
+                "lineWidth": 1,
+                "pointSize": 5,
+                "scaleDistribution": {
+                  "type": "linear"
+                },
+                "showPoints": "auto",
+                "spanNulls": false,
+                "stacking": {
+                  "group": "A",
+                  "mode": "none"
+                },
+                "thresholdsStyle": {
+                  "mode": "off"
+                }
+              },
+              "mappings": [],
+              "thresholds": {
+                "mode": "absolute",
+                "steps": [
+                  {
+                    "color": "green"
+                  }
+                ]
+              }
+            },
+            "overrides": []
+          },
+          "gridPos": {
+            "h": 8,
+            "w": 12,
+            "x": 12,
+            "y": 90
+          },
+          "id": 235,
+          "options": {
+            "legend": {
+              "calcs": [],
+              "displayMode": "list",
+              "placement": "bottom",
+              "showLegend": true
+            },
+            "tooltip": {
+              "maxHeight": 600,
+              "mode": "single",
+              "sort": "none"
+            }
+          },
+          "targets": [
+            {
+              "datasource": {
+                "type": "prometheus",
+                "uid": "${DS_PROMETHEUS}"
+              },
+              "expr": "metrics_ApplicationFallbackCount_Value{instance=~\"${instance}\"}",
+              "legendFormat": "${baseLegend}",
+              "range": true,
+              "refId": "A"
+            }
+          ],
+          "title": "metrics_ApplicationFallbackCount_Value",
+          "type": "timeseries"
         }
       ],
       "title": "Master",

client-flink/common/src/main/java/org/apache/celeborn/plugin/flink/RemoteShuffleMaster.java

@@ -102,6 +102,7 @@ public void registerJob(JobShuffleContext context) {
           celebornAppId = FlinkUtils.toCelebornAppId(lifecycleManagerTimestamp, jobID);
           LOG.info("CelebornAppId: {}", celebornAppId);
           lifecycleManager = new LifecycleManager(celebornAppId, conf);
+          lifecycleManager.applicationCount().increment();
           this.shuffleResourceTracker = new ShuffleResourceTracker(executor, lifecycleManager);
         }
       }
@@ -113,7 +114,10 @@ public void registerJob(JobShuffleContext context) {
             ShuffleFallbackPolicyRunner.getActivatedFallbackPolicy(context, conf, lifecycleManager);
         if (shuffleFallbackPolicy.isPresent()) {
           LOG.warn("Fallback to vanilla Flink NettyShuffleMaster for job: {}.", jobID);
-          jobFallbackPolicies.put(jobID, shuffleFallbackPolicy.get().getClass().getName());
+          String jobFallbackPolicy = shuffleFallbackPolicy.get().getClass().getName();
+          jobFallbackPolicies.put(jobID, jobFallbackPolicy);
+          lifecycleManager.computeFallbackCounts(
+              lifecycleManager.applicationFallbackCounts(), jobFallbackPolicy);
           nettyShuffleMaster().registerJob(context);
           return;
         }
@@ -161,9 +165,8 @@ public CompletableFuture<ShuffleDescriptor> registerPartitionWithProducer(
           String jobFallbackPolicy = jobFallbackPolicies.get(jobID);
           if (jobFallbackPolicy != null) {
             try {
-              lifecycleManager
-                  .shuffleFallbackCounts()
-                  .compute(jobFallbackPolicy, (key, value) -> value == null ? 1L : value + 1L);
+              lifecycleManager.computeFallbackCounts(
+                  lifecycleManager.shuffleFallbackCounts(), jobFallbackPolicy);
               return nettyShuffleMaster()
                   .registerPartitionWithProducer(jobID, partitionDescriptor, producerDescriptor)
                   .get();

client-spark/spark-2/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java

@@ -22,6 +22,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 
 import scala.Int;
+import scala.Option;
 
 import org.apache.spark.*;
 import org.apache.spark.internal.config.package$;
@@ -58,6 +59,8 @@ public class SparkShuffleManager implements ShuffleManager {
   private volatile SortShuffleManager _sortShuffleManager;
   private final ConcurrentHashMap.KeySetView<Integer, Boolean> sortShuffleIds =
       ConcurrentHashMap.newKeySet();
+  private final ConcurrentHashMap.KeySetView<String, Boolean> fallbackApps =
+      ConcurrentHashMap.newKeySet();
   private final CelebornShuffleFallbackPolicyRunner fallbackPolicyRunner;
 
   private long sendBufferPoolCheckInterval;
@@ -97,6 +100,7 @@ private void initializeLifecycleManager(String appId) {
         if (lifecycleManager == null) {
           appUniqueId = celebornConf.appUniqueIdWithUUIDSuffix(appId);
           lifecycleManager = new LifecycleManager(appUniqueId, celebornConf);
+          lifecycleManager.applicationCount().increment();
           lifecycleManager.registerCancelShuffleCallback(SparkUtils::cancelShuffle);
           if (celebornConf.clientStageRerunEnabled()) {
             MapOutputTrackerMaster mapOutputTracker =
@@ -119,9 +123,19 @@ public <K, V, C> ShuffleHandle registerShuffle(
     initializeLifecycleManager(appId);
 
     lifecycleManager.shuffleCount().increment();
-    if (fallbackPolicyRunner.applyFallbackPolicies(dependency, lifecycleManager)) {
+    Option<ShuffleFallbackPolicy> fallbackPolicyOpt =
+        fallbackPolicyRunner.getActivatedFallbackPolicy(dependency, lifecycleManager);
+    if (fallbackPolicyOpt.isDefined()) {
       logger.warn("Fallback to SortShuffleManager!");
       sortShuffleIds.add(shuffleId);
+      String shuffleFallbackPolicy = fallbackPolicyOpt.get().getClass().getName();
+      lifecycleManager.computeFallbackCounts(
+          lifecycleManager.shuffleFallbackCounts(), shuffleFallbackPolicy);
+      if (!fallbackApps.contains(appId)) {
+        fallbackApps.add(appId);
+        lifecycleManager.computeFallbackCounts(
+            lifecycleManager.applicationFallbackCounts(), shuffleFallbackPolicy);
+      }
       return sortShuffleManager().registerShuffle(shuffleId, numMaps, dependency);
     } else {
       lifecycleManager.registerAppShuffleDeterminate(

client-spark/spark-2/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleFallbackPolicyRunner.scala

@@ -34,29 +34,15 @@ class CelebornShuffleFallbackPolicyRunner(conf: CelebornConf) extends Logging {
   private val shuffleFallbackPolicies =
     ShuffleFallbackPolicyFactory.getShuffleFallbackPolicies.asScala
 
-  def applyFallbackPolicies[K, V, C](
+  def getActivatedFallbackPolicy[K, V, C](
       dependency: ShuffleDependency[K, V, C],
-      lifecycleManager: LifecycleManager): Boolean = {
+      lifecycleManager: LifecycleManager): Option[ShuffleFallbackPolicy] = {
     val fallbackPolicy =
       shuffleFallbackPolicies.find(_.needFallback(dependency, conf, lifecycleManager))
-    if (fallbackPolicy.isDefined) {
-      if (FallbackPolicy.NEVER.equals(shuffleFallbackPolicy)) {
-        throw new CelebornIOException(
-          "Fallback to spark built-in shuffle implementation is prohibited.")
-      } else {
-        lifecycleManager.shuffleFallbackCounts.compute(
-          fallbackPolicy.get.getClass.getName,
-          new BiFunction[String, java.lang.Long, java.lang.Long] {
-            override def apply(k: String, v: java.lang.Long): java.lang.Long = {
-              if (v == null) {
-                1L
-              } else {
-                v + 1L
-              }
-            }
-          })
-      }
+    if (fallbackPolicy.isDefined && FallbackPolicy.NEVER.equals(shuffleFallbackPolicy)) {
+      throw new CelebornIOException(
+        "Fallback to spark built-in shuffle implementation is prohibited.")
     }
-    fallbackPolicy.isDefined
+    fallbackPolicy
   }
 }

client-spark/spark-3/src/main/java/org/apache/spark/shuffle/celeborn/SparkShuffleManager.java

@@ -21,6 +21,8 @@
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
 
+import scala.Option;
+
 import org.apache.spark.*;
 import org.apache.spark.internal.config.package$;
 import org.apache.spark.launcher.SparkLauncher;
@@ -82,6 +84,8 @@ public class SparkShuffleManager implements ShuffleManager {
   private volatile SortShuffleManager _sortShuffleManager;
   private final ConcurrentHashMap.KeySetView<Integer, Boolean> sortShuffleIds =
       ConcurrentHashMap.newKeySet();
+  private final ConcurrentHashMap.KeySetView<String, Boolean> fallbackApps =
+      ConcurrentHashMap.newKeySet();
   private final CelebornShuffleFallbackPolicyRunner fallbackPolicyRunner;
 
   private long sendBufferPoolCheckInterval;
@@ -139,6 +143,7 @@ private void initializeLifecycleManager(String appId) {
         if (lifecycleManager == null) {
           appUniqueId = celebornConf.appUniqueIdWithUUIDSuffix(appId);
           lifecycleManager = new LifecycleManager(appUniqueId, celebornConf);
+          lifecycleManager.applicationCount().increment();
           lifecycleManager.registerCancelShuffleCallback(SparkUtils::cancelShuffle);
           if (celebornConf.clientStageRerunEnabled()) {
             MapOutputTrackerMaster mapOutputTracker =
@@ -162,7 +167,9 @@ public <K, V, C> ShuffleHandle registerShuffle(
     initializeLifecycleManager(appId);
 
     lifecycleManager.shuffleCount().increment();
-    if (fallbackPolicyRunner.applyFallbackPolicies(dependency, lifecycleManager)) {
+    Option<ShuffleFallbackPolicy> fallbackPolicyOpt =
+        fallbackPolicyRunner.getActivatedFallbackPolicy(dependency, lifecycleManager);
+    if (fallbackPolicyOpt.isDefined()) {
       if (conf.getBoolean("spark.dynamicAllocation.enabled", false)
           && !conf.getBoolean("spark.shuffle.service.enabled", false)) {
         logger.error(
@@ -174,6 +181,14 @@ public <K, V, C> ShuffleHandle registerShuffle(
         logger.warn("Fallback to vanilla Spark SortShuffleManager for shuffle: {}", shuffleId);
       }
       sortShuffleIds.add(shuffleId);
+      String shuffleFallbackPolicy = fallbackPolicyOpt.get().getClass().getName();
+      lifecycleManager.computeFallbackCounts(
+          lifecycleManager.shuffleFallbackCounts(), shuffleFallbackPolicy);
+      if (!fallbackApps.contains(appId)) {
+        fallbackApps.add(appId);
+        lifecycleManager.computeFallbackCounts(
+            lifecycleManager.applicationFallbackCounts(), shuffleFallbackPolicy);
+      }
       return sortShuffleManager().registerShuffle(shuffleId, dependency);
     } else {
       lifecycleManager.registerAppShuffleDeterminate(

client-spark/spark-3/src/main/scala/org/apache/spark/shuffle/celeborn/CelebornShuffleFallbackPolicyRunner.scala

@@ -32,27 +32,15 @@ class CelebornShuffleFallbackPolicyRunner(conf: CelebornConf) extends Logging {
   private val shuffleFallbackPolicies =
     ShuffleFallbackPolicyFactory.getShuffleFallbackPolicies.asScala
 
-  def applyFallbackPolicies[K, V, C](
+  def getActivatedFallbackPolicy[K, V, C](
       dependency: ShuffleDependency[K, V, C],
-      lifecycleManager: LifecycleManager): Boolean = {
+      lifecycleManager: LifecycleManager): Option[ShuffleFallbackPolicy] = {
     val fallbackPolicy =
       shuffleFallbackPolicies.find(_.needFallback(dependency, conf, lifecycleManager))
-    if (fallbackPolicy.isDefined) {
-      if (FallbackPolicy.NEVER.equals(shuffleFallbackPolicy)) {
-        throw new CelebornIOException(
-          "Fallback to spark built-in shuffle implementation is prohibited.")
-      } else {
-        lifecycleManager.shuffleFallbackCounts.compute(
-          fallbackPolicy.get.getClass.getName,
-          (_, v) => {
-            if (v == null) {
-              1L
-            } else {
-              v + 1L
-            }
-          })
-      }
+    if (fallbackPolicy.isDefined && FallbackPolicy.NEVER.equals(shuffleFallbackPolicy)) {
+      throw new CelebornIOException(
+        "Fallback to spark built-in shuffle implementation is prohibited.")
     }
-    fallbackPolicy.isDefined
+    fallbackPolicy
   }
 }