[SPARK-53405] Add metrics recording for latency of Spark app state transition

### What changes were proposed in this pull request?

This PR adds metrics tracking state transition latency for Spark Application in format of

```
sparkapp.latency.from.<fromState>.to.<toState>
```

### Why are the changes needed?

Latency measuring would be useful to analyze the performance from scheduling / orchestration perspective. For example, to analyze which state causes significant overhad and therefore optimize at cluster/app level.

### Does this PR introduce _any_ user-facing change?

More metrics becomes available.

### How was this patch tested?

CIs. New unit test added.

### Was this patch authored or co-authored using generative AI tooling?

No

Closes apache#299 from jiangzho/state_transition.

Authored-by: Zhou JIANG <[email protected]>
Signed-off-by: Peter Toth <[email protected]>

Author: jiangzho

docs/configuration.md

@@ -91,6 +91,20 @@ via [Codahale JVM Metrics](https://javadoc.io/doc/com.codahale.metrics/metrics-j
 | kubernetes.client.`ResourceName`.`Method`                 | Meter      | Tracking the rates of HTTP request for a combination of one Kubernetes resource and one http method                      |
 | kubernetes.client.`NamespaceName`.`ResourceName`.`Method` | Meter      | Tracking the rates of HTTP request for a combination of one namespace-scoped Kubernetes resource and one http method     |
 
+### Latency for State Transition
+
+Spark Operator also measures the latency between each state transition for apps, in the format of
+
+| Metrics Name                                         | Type  | Description                                                      |
+|------------------------------------------------------|-------|------------------------------------------------------------------|
+| sparkapp.latency.from.`<fromState>`.to.`<toState>`   | Timer | Tracking latency for app of transition from one state to another |
+
+The latency metrics can be used to provide insights about time spent in each state. For example, a
+long latency between `DriverRequested` and `DriverStarted` indicates overhead for driver pod to be
+scheduled. Latency between `DriverStarted` and `DriverReady` indicates overhead to pull image, to
+run init containers and to start SparkSession. These metrics can be used to analyze the overhead
+from multiple dimensions.
+
 ### Forward Metrics to Prometheus
 
 In this section, we will show you how to forward Spark Operator metrics

spark-operator/src/main/java/org/apache/spark/k8s/operator/SparkOperator.java

@@ -46,6 +46,7 @@
 import org.apache.spark.k8s.operator.metrics.MetricsService;
 import org.apache.spark.k8s.operator.metrics.MetricsSystem;
 import org.apache.spark.k8s.operator.metrics.MetricsSystemFactory;
+import org.apache.spark.k8s.operator.metrics.SparkAppStatusRecorderSource;
 import org.apache.spark.k8s.operator.metrics.healthcheck.SentinelManager;
 import org.apache.spark.k8s.operator.metrics.source.KubernetesMetricsInterceptor;
 import org.apache.spark.k8s.operator.metrics.source.OperatorJosdkMetrics;
@@ -83,7 +84,10 @@ public SparkOperator() {
         KubernetesClientFactory.buildKubernetesClient(getClientInterceptors(metricsSystem));
     this.appSubmissionWorker = new SparkAppSubmissionWorker();
     this.clusterSubmissionWorker = new SparkClusterSubmissionWorker();
-    this.sparkAppStatusRecorder = new SparkAppStatusRecorder(getAppStatusListener());
+    SparkAppStatusRecorderSource recorderSource = new SparkAppStatusRecorderSource();
+    this.metricsSystem.registerSource(recorderSource);
+    this.sparkAppStatusRecorder =
+        new SparkAppStatusRecorder(getAppStatusListener(), recorderSource);
     this.sparkClusterStatusRecorder = new SparkClusterStatusRecorder(getClusterStatusListener());
     this.registeredSparkControllers = new HashSet<>();
     this.watchedNamespaces = getWatchedNamespaces();

spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/BaseOperatorSource.java

@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.k8s.operator.metrics;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import com.codahale.metrics.Counter;
+import com.codahale.metrics.Gauge;
+import com.codahale.metrics.Histogram;
+import com.codahale.metrics.MetricRegistry;
+import com.codahale.metrics.Timer;
+import lombok.RequiredArgsConstructor;
+
+@RequiredArgsConstructor
+public class BaseOperatorSource {
+  protected final MetricRegistry metricRegistry;
+  protected final Map<String, Histogram> histograms = new ConcurrentHashMap<>();
+  protected final Map<String, Counter> counters = new ConcurrentHashMap<>();
+  protected final Map<String, Gauge<?>> gauges = new ConcurrentHashMap<>();
+  protected final Map<String, Timer> timers = new ConcurrentHashMap<>();
+
+  protected Histogram getHistogram(String metricNamePrefix, String... names) {
+    Histogram histogram;
+    String metricName = MetricRegistry.name(metricNamePrefix, names).toLowerCase();
+    if (histograms.containsKey(metricName)) {
+      histogram = histograms.get(metricName);
+    } else {
+      histogram = metricRegistry.histogram(metricName);
+      histograms.put(metricName, histogram);
+    }
+    return histogram;
+  }
+
+  protected Counter getCounter(String metricNamePrefix, String... names) {
+    Counter counter;
+    String metricName = MetricRegistry.name(metricNamePrefix, names).toLowerCase();
+    if (counters.containsKey(metricName)) {
+      counter = counters.get(metricName);
+    } else {
+      counter = metricRegistry.counter(metricName);
+      counters.put(metricName, counter);
+    }
+    return counter;
+  }
+
+  protected Gauge<?> getGauge(Gauge<?> defaultGauge, String metricNamePrefix, String... names) {
+    Gauge<?> gauge;
+    String metricName = MetricRegistry.name(metricNamePrefix, names).toLowerCase();
+    if (gauges.containsKey(metricName)) {
+      gauge = gauges.get(metricName);
+    } else {
+      gauge = defaultGauge;
+      gauges.put(metricName, defaultGauge);
+    }
+    return gauge;
+  }
+
+  protected Timer getTimer(String metricNamePrefix, String... names) {
+    Timer timer;
+    String metricName = MetricRegistry.name(metricNamePrefix, names).toLowerCase();
+    if (timers.containsKey(metricName)) {
+      timer = timers.get(metricName);
+    } else {
+      timer = metricRegistry.timer(metricName);
+      timers.put(metricName, timer);
+    }
+    return timer;
+  }
+
+  protected String getMetricNamePrefix(Class<?> klass) {
+    return klass.getSimpleName();
+  }
+}

spark-operator/src/main/java/org/apache/spark/k8s/operator/metrics/SparkAppStatusRecorderSource.java

@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.k8s.operator.metrics;
+
+import java.time.Duration;
+import java.time.Instant;
+
+import com.codahale.metrics.MetricRegistry;
+
+import org.apache.spark.k8s.operator.status.ApplicationState;
+import org.apache.spark.k8s.operator.status.ApplicationStatus;
+import org.apache.spark.metrics.source.Source;
+
+public class SparkAppStatusRecorderSource extends BaseOperatorSource implements Source {
+
+  public static final String RESOURCE_TYPE = "sparkapp";
+  public static final String LATENCY_METRIC_FORMAT = "latency.from.%s.to.%s";
+
+  public SparkAppStatusRecorderSource() {
+    super(new MetricRegistry());
+  }
+
+  @Override
+  public String sourceName() {
+    return "SparkAppStatusRecorder";
+  }
+
+  @Override
+  public MetricRegistry metricRegistry() {
+    return metricRegistry;
+  }
+
+  public void recordStatusUpdateLatency(
+      final ApplicationStatus status, final ApplicationState newState) {
+    ApplicationState currentState = status.getCurrentState();
+    if (currentState != null) {
+      Duration duration =
+          Duration.between(
+              Instant.parse(currentState.getLastTransitionTime()),
+              Instant.parse(newState.getLastTransitionTime()));
+      getTimer(
+              RESOURCE_TYPE,
+              String.format(
+                  LATENCY_METRIC_FORMAT,
+                  currentState.getCurrentStateSummary().name(),
+                  newState.getCurrentStateSummary().name()))
+          .update(duration);
+    }
+  }
+}