[SERVER] Add Kubernetes engine Pod discovery latency metric

docs/monitor/metrics.md

@@ -63,6 +63,7 @@ These metrics include:
 | `kyuubi.operation.exec_time`                     | `${operationType}`                     | histogram | 1.7.0  | <div style='width: 150pt;word-wrap: break-word;white-space: normal'>  execution time histogram for the operation `${operationType}`, now only `ExecuteStatement` is enabled.  </div>                                                                                                       |
 | `kyuubi.engine.total`                            |                                        | counter   | 1.2.0  | <div style='width: 150pt;word-wrap: break-word;white-space: normal'>  cumulative created engines</div>                                                                                                                                                                                     |
 | `kyuubi.engine.startup.time`                     |                                        | histogram | 1.12.0 | <div style='width: 150pt;word-wrap: break-word;white-space: normal'> startup time histogram for successfully created engines, from submitting a new engine to the engine being registered and discovered by Kyuubi server </div>                                                           |
+| `kyuubi.engine.kubernetes.pod.discovery.latency` |                                        | histogram | 1.13.0 | Time in milliseconds from Spark engine Pod creation to the informer ADD callback. Recorded for each ADD event, including existing Pods discovered during initial listing. UPDATE and DELETE events do not contribute samples.                                                              |
 | `kyuubi.engine.timeout`                          |                                        | counter   | 1.2.0  | <div style='width: 150pt;word-wrap: break-word;white-space: normal'>  cumulative timeout engines</div>                                                                                                                                                                                     |
 | `kyuubi.engine.failed`                           | `${user}`                              | counter   | 1.2.0  | <div style='width: 150pt;word-wrap: break-word;white-space: normal'>  cumulative explicitly failed engine count for a `${user}`</div>                                                                                                                                                      |
 | `kyuubi.engine.failed`                           | `${errorType}`                         | counter   | 1.2.0  | <div style='width: 150pt;word-wrap: break-word;white-space: normal'> cumulative explicitly failed engine count for a particular `${errorType}`, e.g. `ClassNotFoundException`</div>                                                                                                        |

kyuubi-metrics/src/main/scala/org/apache/kyuubi/metrics/MetricsConstants.scala

@@ -58,6 +58,8 @@ object MetricsConstants {
   final private val ENGINE = KYUUBI + "engine."
   final val ENGINE_FAIL: String = ENGINE + "failed"
   final val ENGINE_STARTUP_TIME: String = ENGINE + "startup.time"
+  final val ENGINE_KUBERNETES_POD_DISCOVERY_LATENCY: String =
+    ENGINE + "kubernetes.pod.discovery.latency"
   final val ENGINE_TIMEOUT: String = ENGINE + "timeout"
   final val ENGINE_TOTAL: String = ENGINE + "total"
 

kyuubi-server/src/main/scala/org/apache/kyuubi/engine/KubernetesApplicationOperation.scala

@@ -17,6 +17,8 @@
 
 package org.apache.kyuubi.engine
 
+import java.time.Instant
+import java.time.format.DateTimeParseException
 import java.util.Locale
 import java.util.concurrent.{ConcurrentHashMap, ExecutorService, ScheduledExecutorService, TimeUnit}
 
@@ -37,6 +39,8 @@ import org.apache.kyuubi.config.KyuubiConf.KubernetesCleanupDriverPodStrategy.{A
 import org.apache.kyuubi.engine.ApplicationState.{isTerminated, ApplicationState, FAILED, FINISHED, KILLED, NOT_FOUND, PENDING, RUNNING, UNKNOWN}
 import org.apache.kyuubi.engine.KubernetesApplicationUrlSource._
 import org.apache.kyuubi.engine.KubernetesResourceEventTypes.KubernetesResourceEventType
+import org.apache.kyuubi.metrics.MetricsConstants.ENGINE_KUBERNETES_POD_DISCOVERY_LATENCY
+import org.apache.kyuubi.metrics.MetricsSystem
 import org.apache.kyuubi.operation.OperationState
 import org.apache.kyuubi.server.metadata.MetadataManager
 import org.apache.kyuubi.server.metadata.api.KubernetesEngineInfo
@@ -359,7 +363,26 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging {
     extends ResourceEventHandler[Pod] {
 
     override def onAdd(pod: Pod): Unit = {
+      val observedAt = System.currentTimeMillis()
       if (isSparkEnginePod(pod)) {
+        Option(pod.getMetadata.getCreationTimestamp) match {
+          case Some(creationTimestamp) =>
+            try {
+              val latency = observedAt - Instant.parse(creationTimestamp).toEpochMilli
+              if (latency >= 0) {
+                MetricsSystem.tracing(_.updateHistogram(
+                  ENGINE_KUBERNETES_POD_DISCOVERY_LATENCY,
+                  latency))
+              }
+            } catch {
+              case e: DateTimeParseException =>
+                warn(s"Invalid creation timestamp for engine pod ${pod.getMetadata.getName}", e)
+            }
+          case None =>
+            warn(s"[$kubernetesInfo] Missing creation timestamp for engine pod " +
+              s"${pod.getMetadata.getNamespace}/${pod.getMetadata.getName}, " +
+              "skipping pod discovery latency metric")
+        }
         val eventType = KubernetesResourceEventTypes.ADD
         updateApplicationState(kubernetesInfo, pod, eventType)
         val appState = toApplicationState(pod, appStateSource, appStateContainer, eventType)

kyuubi-server/src/test/scala/org/apache/kyuubi/engine/KubernetesApplicationOperationSuite.scala

@@ -17,15 +17,113 @@
 
 package org.apache.kyuubi.engine
 
+import java.time.Instant
+
 import io.fabric8.kubernetes.api.model.{ContainerState, ContainerStateWaiting, PodBuilder}
 
 import org.apache.kyuubi.{KyuubiException, KyuubiFunSuite}
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.engine.ApplicationState.{FAILED, FINISHED, PENDING}
 import org.apache.kyuubi.engine.KubernetesApplicationOperation.LABEL_KYUUBI_UNIQUE_KEY
+import org.apache.kyuubi.metrics.MetricsConstants.ENGINE_KUBERNETES_POD_DISCOVERY_LATENCY
+import org.apache.kyuubi.metrics.MetricsSystem
 
 class KubernetesApplicationOperationSuite extends KyuubiFunSuite {
 
+  test("record pod discovery latency only for ADD events") {
+    val operation = new KubernetesApplicationOperation()
+    val metricsSystem = new MetricsSystem
+    val conf = KyuubiConf()
+    operation.initialize(conf, None)
+    metricsSystem.initialize(conf)
+    metricsSystem.start()
+    val createdAt = Instant.now().minusSeconds(180)
+    val pod = new PodBuilder()
+      .withNewMetadata()
+      .withName("delayed-driver")
+      .withCreationTimestamp(createdAt.toString)
+      .addToLabels(LABEL_KYUUBI_UNIQUE_KEY, "delayed-app")
+      .addToLabels("spark-app-selector", "spark-application")
+      .endMetadata()
+      .withNewStatus()
+      .withPhase("Running")
+      .endStatus()
+      .build()
+    val handler = new operation.SparkEnginePodEventHandler(KubernetesInfo())
+    val histogram = MetricsSystem.getMetricsRegistry.get
+      .histogram(ENGINE_KUBERNETES_POD_DISCOVERY_LATENCY)
+    try {
+      handler.onUpdate(pod, pod)
+      assert(histogram.getCount == 0)
+      val before = System.currentTimeMillis()
+      handler.onAdd(pod)
+      val after = System.currentTimeMillis()
+      assert(histogram.getCount == 1)
+      val latency = histogram.getSnapshot.getMax
+      assert(latency >= before - createdAt.toEpochMilli)
+      assert(latency <= after - createdAt.toEpochMilli)
+      handler.onAdd(pod)
+      assert(histogram.getCount == 2)
+      handler.onUpdate(pod, pod)
+      handler.onDelete(pod, false)
+      assert(histogram.getCount == 2)
+    } finally {
+      operation.stop()
+      metricsSystem.stop()
+    }
+  }
+
+  test("skip invalid pod discovery samples without preventing pod event handling") {
+    val operation = new KubernetesApplicationOperation()
+    val metricsSystem = new MetricsSystem
+    val conf = KyuubiConf()
+    operation.initialize(conf, None)
+    metricsSystem.initialize(conf)
+    metricsSystem.start()
+    val histogram = MetricsSystem.getMetricsRegistry.get
+      .histogram(ENGINE_KUBERNETES_POD_DISCOVERY_LATENCY)
+    val handler = new operation.SparkEnginePodEventHandler(KubernetesInfo())
+    try {
+      Seq(null, "invalid", Instant.now().plusSeconds(3600).toString).zipWithIndex.foreach {
+        case (creationTimestamp, index) =>
+          val pod = new PodBuilder()
+            .withNewMetadata()
+            .withName(s"driver-$index")
+            .withCreationTimestamp(creationTimestamp)
+            .addToLabels(LABEL_KYUUBI_UNIQUE_KEY, s"app-$index")
+            .addToLabels("spark-app-selector", "spark-application")
+            .endMetadata()
+            .withNewStatus()
+            .withPhase("Succeeded")
+            .endStatus()
+            .build()
+          handler.onAdd(pod)
+          assert(histogram.getCount == 0)
+          assert(operation.cleanupTerminatedAppInfoTrigger.getIfPresent(s"app-$index") == FINISHED)
+      }
+      val deletedPod = new PodBuilder()
+        .withNewMetadata()
+        .withName("deleted-driver")
+        .withCreationTimestamp(Instant.now().minusSeconds(180).toString)
+        .addToLabels(LABEL_KYUUBI_UNIQUE_KEY, "deleted-app")
+        .addToLabels("spark-app-selector", "spark-application")
+        .endMetadata()
+        .withNewStatus()
+        .withPhase("Running")
+        .endStatus()
+        .build()
+      handler.onDelete(deletedPod, false)
+      assert(histogram.getCount == 0)
+      deletedPod.getMetadata.getLabels.put(LABEL_KYUUBI_UNIQUE_KEY, "non-engine-app")
+      deletedPod.getMetadata.getLabels.remove("spark-app-selector")
+      handler.onAdd(deletedPod)
+      assert(histogram.getCount == 0)
+    } finally {
+      operation.stop()
+      metricsSystem.stop()
+    }
+  }
+
   test("mark terminated application received from pod add event") {
     val operation = new KubernetesApplicationOperation()
     operation.initialize(KyuubiConf(), None)