upsert

Author: turboFei

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

@@ -38,6 +38,7 @@ import org.apache.kyuubi.engine.ApplicationState.{isTerminated, ApplicationState
 import org.apache.kyuubi.engine.KubernetesResourceEventTypes.KubernetesResourceEventType
 import org.apache.kyuubi.operation.OperationState
 import org.apache.kyuubi.server.KyuubiServer
+import org.apache.kyuubi.server.metadata.api.KubernetesMetadata
 import org.apache.kyuubi.session.KyuubiSessionManager
 import org.apache.kyuubi.util.{KubernetesUtils, ThreadUtils}
 
@@ -336,7 +337,7 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging {
         updateApplicationState(kubernetesInfo, newPod, eventType)
         val appState = toApplicationState(newPod, appStateSource, appStateContainer, eventType)
         if (isTerminated(appState)) {
-          markApplicationTerminated(newPod, eventType)
+          markApplicationTerminated(kubernetesInfo, newPod, eventType)
         }
         KubernetesApplicationAuditLogger.audit(
           eventType,
@@ -354,7 +355,7 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging {
       if (isSparkEnginePod(pod)) {
         val eventType = KubernetesResourceEventTypes.DELETE
         updateApplicationState(kubernetesInfo, pod, eventType)
-        markApplicationTerminated(pod, eventType)
+        markApplicationTerminated(kubernetesInfo, pod, eventType)
         KubernetesApplicationAuditLogger.audit(
           eventType,
           kubernetesInfo,
@@ -450,13 +451,24 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging {
   }
 
   private def markApplicationTerminated(
+      kubernetesInfo: KubernetesInfo,
       pod: Pod,
       eventType: KubernetesResourceEventType): Unit = synchronized {
     val key = pod.getMetadata.getLabels.get(LABEL_KYUUBI_UNIQUE_KEY)
+    val (appState, appError) =
+      toApplicationStateAndError(pod, appStateSource, appStateContainer, eventType)
+    // upsert the kubernetes metadata when the application is terminated
+    metadataManager.foreach(_.upsertKubernetesMetadata(
+      KubernetesMetadata(
+        identifier = key,
+        context = kubernetesInfo.context,
+        namespace = kubernetesInfo.namespace,
+        podName = pod.getMetadata.getName,
+        appId = pod.getMetadata.getLabels.get(SPARK_APP_ID_LABEL),
+        appState = appState.toString,
+        appError = appError)))
     if (cleanupTerminatedAppInfoTrigger.getIfPresent(key) == null) {
-      cleanupTerminatedAppInfoTrigger.put(
-        key,
-        toApplicationState(pod, appStateSource, appStateContainer, eventType))
+      cleanupTerminatedAppInfoTrigger.put(key, appState)
     }
   }
 

kyuubi-server/src/main/scala/org/apache/kyuubi/server/metadata/api/KubernetesMetadata.scala

@@ -23,7 +23,7 @@ package org.apache.kyuubi.server.metadata.api
  * @param identifier the kubernetes unique identifier.
  * @param context the kubernetes context.
  * @param namespace the kubernetes namespace.
- * @param pod the kubernetes pod name.
+ * @param podName the kubernetes pod name.
  * @param appId the application id.
  * @param appState the application state.
  * @param appError the application error diagnose.
@@ -34,7 +34,7 @@ case class KubernetesMetadata(
     identifier: String,
     context: Option[String],
     namespace: Option[String],
-    pod: String,
+    podName: String,
     appId: String,
     appState: String,
     appError: Option[String],

kyuubi-server/src/main/scala/org/apache/kyuubi/server/metadata/jdbc/JDBCMetadataStore.scala

@@ -429,7 +429,8 @@ class JDBCMetadataStore(conf: KyuubiConf) extends MetadataStore with Logging {
          |pod_name,
          |app_id,
          |app_state,
-         |app_error
+         |app_error,
+         |create_time
          |)
          |VALUES(?, ?, ?, ?, ?, ?, ?)
          |WHERE NOT EXISTS (
@@ -443,10 +444,11 @@ class JDBCMetadataStore(conf: KyuubiConf) extends MetadataStore with Logging {
         metadata.identifier,
         metadata.context.orNull,
         metadata.namespace.orNull,
-        metadata.pod,
+        metadata.podName,
         metadata.appId,
         metadata.appState,
         metadata.appError.orNull,
+        System.currentTimeMillis(),
         metadata.identifier)
     }
   }
@@ -465,7 +467,7 @@ class JDBCMetadataStore(conf: KyuubiConf) extends MetadataStore with Logging {
       setClauses += "namespace = ?"
       params += namespace
     }
-    Option(metadata.pod).foreach { pod =>
+    Option(metadata.podName).foreach { pod =>
       setClauses += "pod = ?"
       params += pod
     }
@@ -605,7 +607,7 @@ class JDBCMetadataStore(conf: KyuubiConf) extends MetadataStore with Logging {
           identifier = identifier,
           context = context,
           namespace = namespace,
-          pod = pod,
+          podName = pod,
           appId = appId,
           appState = appState,
           appError = appError,