[KYUUBI #7756][KUBERNETES] Support owner-scoped application watches

kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala

@@ -1462,6 +1462,22 @@ object KyuubiConf {
       .toSequence()
       .createWithDefault(Nil)
 
+  val KUBERNETES_APPLICATION_OWNER_SCOPED_WATCH_ENABLED: ConfigEntry[Boolean] =
+    buildConf("kyuubi.kubernetes.application.ownerScopedWatch.enabled")
+      .doc("Whether each Kyuubi server only watches Kubernetes applications in its watch scope. " +
+        "Kyuubi adds the `kyuubi.apache.org/watch-scope` label to all Spark applications it " +
+        "submits. CONNECTION and SERVER_LOCAL engines use the Kyuubi server address as their " +
+        "scope, while cross-server shared engines and batch applications use `global`. When " +
+        "enabled, each server watches applications in its own address scope and the global " +
+        "scope. " +
+        "Server addresses must be unique across Kyuubi servers. Applications without the watch " +
+        "scope label are not watched when this option is enabled.")
+      .version("1.13.0")
+      .audience(SERVER)
+      .immutable
+      .booleanConf
+      .createWithDefault(false)
+
   val KUBERNETES_CLIENT_DISPATCHER_VIRTUAL_THREADS_ENABLED: ConfigEntry[Boolean] =
     buildConf("kyuubi.kubernetes.client.dispatcher.virtualThreads.enabled")
       .doc("Whether the Kyuubi server Kubernetes HTTP client dispatcher uses virtual threads. " +

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

@@ -91,6 +91,9 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging {
 
   private var metadataManager: Option[MetadataManager] = _
 
+  private[engine] def ownerScopedWatchEnabled: Boolean =
+    kyuubiConf.get(KyuubiConf.KUBERNETES_APPLICATION_OWNER_SCOPED_WATCH_ENABLED)
+
   // Visible for testing
   private[engine] def checkKubernetesInfo(kubernetesInfo: KubernetesInfo): Unit = {
     val context = kubernetesInfo.context
@@ -113,15 +116,32 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging {
     KubernetesUtils.buildKubernetesClient(kubernetesConf) match {
       case Some(client) =>
         info(s"[$kubernetesInfo] Initialized Kubernetes Client connect to: ${client.getMasterUrl}")
-        val enginePodInformer = client.pods()
-          .withLabel(LABEL_KYUUBI_UNIQUE_KEY)
-          .inform(new SparkEnginePodEventHandler(kubernetesInfo))
+        val enginePods = client.pods().withLabel(LABEL_KYUUBI_UNIQUE_KEY)
+        val enginePodInformer = if (ownerScopedWatchEnabled) {
+          enginePods
+            .withLabelIn(
+              LABEL_KYUUBI_WATCH_SCOPE_KEY,
+              KubernetesUtils.serverAddressLabelValue,
+              GLOBAL_WATCH_SCOPE)
+            .inform(new SparkEnginePodEventHandler(kubernetesInfo))
+        } else {
+          enginePods.inform(new SparkEnginePodEventHandler(kubernetesInfo))
+        }
         info(s"[$kubernetesInfo] Start Kubernetes Client POD Informer.")
         enginePodInformers.put(kubernetesInfo, enginePodInformer)
         if (sparkAppUrlSource == KubernetesApplicationUrlSource.SVC) {
           info(s"[$kubernetesInfo] Start Kubernetes Client SVC Informer.")
-          val engineSvcInformer = client.services()
-            .inform(new SparkEngineSvcEventHandler(kubernetesInfo))
+          val engineServices = client.services()
+          val engineSvcInformer = if (ownerScopedWatchEnabled) {
+            engineServices
+              .withLabelIn(
+                LABEL_KYUUBI_WATCH_SCOPE_KEY,
+                KubernetesUtils.serverAddressLabelValue,
+                GLOBAL_WATCH_SCOPE)
+              .inform(new SparkEngineSvcEventHandler(kubernetesInfo))
+          } else {
+            engineServices.inform(new SparkEngineSvcEventHandler(kubernetesInfo))
+          }
           engineSvcInformers.put(kubernetesInfo, engineSvcInformer)
         }
         client
@@ -134,6 +154,10 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging {
     kyuubiConf = conf
     this.metadataManager = metadataManager
     info("Start initializing Kubernetes application operation.")
+    if (ownerScopedWatchEnabled) {
+      info(s"Kubernetes application owner-scoped watch is enabled for scopes " +
+        s"${KubernetesUtils.serverAddressLabelValue} and $GLOBAL_WATCH_SCOPE.")
+    }
     submitTimeout = conf.get(KyuubiConf.ENGINE_KUBERNETES_SUBMIT_TIMEOUT)
     // Defer cleaning terminated application information
     val retainPeriod = conf.get(KyuubiConf.KUBERNETES_TERMINATED_APPLICATION_RETAIN_PERIOD)
@@ -576,6 +600,8 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging {
 
 object KubernetesApplicationOperation extends Logging {
   val LABEL_KYUUBI_UNIQUE_KEY = "kyuubi-unique-tag"
+  val LABEL_KYUUBI_WATCH_SCOPE_KEY = "kyuubi.apache.org/watch-scope"
+  val GLOBAL_WATCH_SCOPE = "global"
   private val SPARK_APP_ID_LABEL = "spark-app-selector"
   private val SPARK_APP_NAME_LABEL = "spark-app-name"
   val KUBERNETES_SERVICE_HOST = "KUBERNETES_SERVICE_HOST"

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

@@ -28,11 +28,13 @@ import org.apache.commons.lang3.StringUtils
 
 import org.apache.kyuubi.{KyuubiException, Utils}
 import org.apache.kyuubi.config.KyuubiConf
-import org.apache.kyuubi.engine.KubernetesApplicationOperation.LABEL_KYUUBI_UNIQUE_KEY
+import org.apache.kyuubi.engine.KubernetesApplicationOperation.{GLOBAL_WATCH_SCOPE, LABEL_KYUUBI_UNIQUE_KEY, LABEL_KYUUBI_WATCH_SCOPE_KEY}
+import org.apache.kyuubi.engine.ShareLevel.{CONNECTION, SERVER_LOCAL}
 import org.apache.kyuubi.engine.flink.FlinkProcessBuilder
 import org.apache.kyuubi.engine.spark.SparkProcessBuilder
 import org.apache.kyuubi.server.metadata.MetadataManager
 import org.apache.kyuubi.service.AbstractService
+import org.apache.kyuubi.util.KubernetesUtils
 import org.apache.kyuubi.util.reflect.ReflectUtils._
 
 class KyuubiApplicationManager(metadataManager: Option[MetadataManager])
@@ -124,8 +126,17 @@ object KyuubiApplicationManager {
     conf.set(KyuubiConf.ENGINE_DEPLOY_YARN_MODE_TAGS.key, newTag)
   }
 
-  private def setupSparkK8sTag(tag: String, conf: KyuubiConf): Unit = {
+  private def setupSparkK8sTag(
+      tag: String,
+      conf: KyuubiConf,
+      engineShareLevel: Option[String]): Unit = {
     conf.set("spark.kubernetes.driver.label." + LABEL_KYUUBI_UNIQUE_KEY, tag)
+    val watchScope = engineShareLevel.map(ShareLevel.withName) match {
+      case Some(CONNECTION | SERVER_LOCAL) => KubernetesUtils.serverAddressLabelValue
+      case _ => GLOBAL_WATCH_SCOPE
+    }
+    conf.set("spark.kubernetes.driver.label." + LABEL_KYUUBI_WATCH_SCOPE_KEY, watchScope)
+    conf.set("spark.kubernetes.driver.service.label." + LABEL_KYUUBI_WATCH_SCOPE_KEY, watchScope)
   }
 
   private def setupFlinkYarnTag(tag: String, conf: KyuubiConf): Unit = {
@@ -206,13 +217,23 @@ object KyuubiApplicationManager {
       applicationType: String,
       resourceManager: Option[String],
       conf: KyuubiConf): Unit = {
+    tagApplication(applicationTag, applicationType, resourceManager, conf, None)
+  }
+
+  def tagApplication(
+      applicationTag: String,
+      applicationType: String,
+      resourceManager: Option[String],
+      conf: KyuubiConf,
+      engineShareLevel: Option[String]): Unit = {
     (applicationType.toUpperCase, resourceManager.map(_.toUpperCase())) match {
       case ("SPARK", Some("YARN")) => setupSparkYarnTag(applicationTag, conf)
-      case ("SPARK", Some(rm)) if rm.startsWith("K8S") => setupSparkK8sTag(applicationTag, conf)
+      case ("SPARK", Some(rm)) if rm.startsWith("K8S") =>
+        setupSparkK8sTag(applicationTag, conf, engineShareLevel)
       case ("SPARK", _) =>
         // if the master is not identified ahead, add all tags
         setupSparkYarnTag(applicationTag, conf)
-        setupSparkK8sTag(applicationTag, conf)
+        setupSparkK8sTag(applicationTag, conf, engineShareLevel)
       case ("FLINK", Some("YARN")) =>
         // running flink on other platforms is not yet supported
         setupFlinkYarnTag(applicationTag, conf)

kyuubi-server/src/main/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilder.scala

@@ -134,7 +134,12 @@ class SparkProcessBuilder(
     // complete `spark.master` if absent on kubernetes
     completeMasterUrl(conf)
 
-    KyuubiApplicationManager.tagApplication(engineRefId, shortName, clusterManager(), conf)
+    KyuubiApplicationManager.tagApplication(
+      engineRefId,
+      shortName,
+      clusterManager(),
+      conf,
+      Some(conf.get(KyuubiConf.ENGINE_SHARE_LEVEL)))
     val buffer = new mutable.ListBuffer[String]()
     buffer += executable
     buffer += CLASS

kyuubi-server/src/main/scala/org/apache/kyuubi/util/KubernetesUtils.scala

@@ -18,6 +18,7 @@
 package org.apache.kyuubi.util
 
 import java.io.File
+import java.net.{Inet4Address, InetAddress}
 import java.util.Locale
 
 import com.fasterxml.jackson.databind.ObjectMapper
@@ -34,6 +35,14 @@ import org.apache.kyuubi.config.KyuubiConf._
 
 object KubernetesUtils extends Logging {
   final val DRIVER_POD_NAME_MAX_LENGTH = 253
+
+  lazy val serverAddressLabelValue: String =
+    toServerAddressLabelValue(JavaUtils.findLocalInetAddress)
+
+  private[kyuubi] def toServerAddressLabelValue(address: InetAddress): String = address match {
+    case ipv4: Inet4Address => ipv4.getHostAddress
+    case ipv6 => ipv6.getAddress.map(byte => f"${byte & 0xFF}%02x").mkString
+  }
   final private val POD_UID_MAX_LENGTH = 36
   final private val POD_LOGS_DIRECTORY_SEPARATOR_LENGTH = 2
   final private val EXECUTOR_POD_NAME_RESERVED_LENGTH =

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

@@ -17,12 +17,15 @@
 
 package org.apache.kyuubi.engine
 
+import java.net.InetAddress
+
 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.util.KubernetesUtils
 
 class KubernetesApplicationOperationSuite extends KyuubiFunSuite {
 
@@ -50,6 +53,33 @@ class KubernetesApplicationOperationSuite extends KyuubiFunSuite {
     }
   }
 
+  test("server address should be a valid Kubernetes label value") {
+    assert(KubernetesUtils.toServerAddressLabelValue(
+      InetAddress.getByName("10.0.0.1")) === "10.0.0.1")
+    assert(KubernetesUtils.toServerAddressLabelValue(
+      InetAddress.getByName("2001:db8::1")) === "20010db8000000000000000000000001")
+  }
+
+  test("owner-scoped watch configuration") {
+    val operation = new KubernetesApplicationOperation()
+    operation.initialize(KyuubiConf(), None)
+    try {
+      assert(!operation.ownerScopedWatchEnabled)
+    } finally {
+      operation.stop()
+    }
+
+    val scopedOperation = new KubernetesApplicationOperation()
+    scopedOperation.initialize(
+      KyuubiConf().set(KyuubiConf.KUBERNETES_APPLICATION_OWNER_SCOPED_WATCH_ENABLED, true),
+      None)
+    try {
+      assert(scopedOperation.ownerScopedWatchEnabled)
+    } finally {
+      scopedOperation.stop()
+    }
+  }
+
   test("test check kubernetes info") {
     val kyuubiConf = KyuubiConf()
     kyuubiConf.set(KyuubiConf.KUBERNETES_CONTEXT_ALLOW_LIST.key, "1,2")

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

@@ -19,8 +19,10 @@ package org.apache.kyuubi.engine
 
 import org.apache.kyuubi.{KyuubiException, KyuubiFunSuite}
 import org.apache.kyuubi.config.KyuubiConf
-import org.apache.kyuubi.engine.KubernetesApplicationOperation.LABEL_KYUUBI_UNIQUE_KEY
+import org.apache.kyuubi.engine.KubernetesApplicationOperation.{GLOBAL_WATCH_SCOPE, LABEL_KYUUBI_UNIQUE_KEY, LABEL_KYUUBI_WATCH_SCOPE_KEY}
+import org.apache.kyuubi.engine.ShareLevel.{CONNECTION, GROUP, SERVER, SERVER_LOCAL, USER}
 import org.apache.kyuubi.engine.spark.SparkProcessBuilder
+import org.apache.kyuubi.util.KubernetesUtils
 
 class KyuubiApplicationManagerSuite extends KyuubiFunSuite {
   test("application access path") {
@@ -177,6 +179,10 @@ class KyuubiApplicationManagerSuite extends KyuubiFunSuite {
   test("Test kyuubi application Manager tag spark on kubernetes application") {
     val conf: KyuubiConf = KyuubiConf()
     val tag = "kyuubi-test-tag"
+    val watchScopePodLabel = "spark.kubernetes.driver.label." + LABEL_KYUUBI_WATCH_SCOPE_KEY
+    val watchScopeServiceLabel =
+      "spark.kubernetes.driver.service.label." + LABEL_KYUUBI_WATCH_SCOPE_KEY
+
     KyuubiApplicationManager.tagApplication(
       tag,
       "SPARK",
@@ -187,5 +193,32 @@ class KyuubiApplicationManagerSuite extends KyuubiFunSuite {
     val yarnTag = conf.getOption("spark.yarn.tags")
     assert(kubernetesTag.nonEmpty && tag.equals(kubernetesTag.get))
     assert(yarnTag.isEmpty)
+    assert(conf.getOption(watchScopePodLabel).contains(GLOBAL_WATCH_SCOPE))
+    assert(conf.getOption(watchScopeServiceLabel).contains(GLOBAL_WATCH_SCOPE))
+
+    Seq(CONNECTION, SERVER_LOCAL).foreach { shareLevel =>
+      conf.set(watchScopePodLabel, "overridden")
+      conf.set(watchScopeServiceLabel, "overridden")
+      KyuubiApplicationManager.tagApplication(
+        tag,
+        "SPARK",
+        Some("k8s://https://kyuubi-test:8443"),
+        conf,
+        Some(shareLevel.toString))
+      assert(conf.getOption(watchScopePodLabel).contains(KubernetesUtils.serverAddressLabelValue))
+      assert(conf.getOption(watchScopeServiceLabel)
+        .contains(KubernetesUtils.serverAddressLabelValue))
+    }
+
+    Seq(USER, GROUP, SERVER).foreach { shareLevel =>
+      KyuubiApplicationManager.tagApplication(
+        tag,
+        "SPARK",
+        Some("k8s://https://kyuubi-test:8443"),
+        conf,
+        Some(shareLevel.toString))
+      assert(conf.getOption(watchScopePodLabel).contains(GLOBAL_WATCH_SCOPE))
+      assert(conf.getOption(watchScopeServiceLabel).contains(GLOBAL_WATCH_SCOPE))
+    }
   }
 }

kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkBatchProcessBuilderSuite.scala

@@ -23,6 +23,7 @@ import java.util.UUID
 
 import org.apache.kyuubi.KyuubiFunSuite
 import org.apache.kyuubi.config.KyuubiConf
+import org.apache.kyuubi.engine.KubernetesApplicationOperation.{GLOBAL_WATCH_SCOPE, LABEL_KYUUBI_WATCH_SCOPE_KEY}
 
 class SparkBatchProcessBuilderSuite extends KyuubiFunSuite {
   test("spark batch conf should be converted with `spark.` prefix") {
@@ -39,6 +40,27 @@ class SparkBatchProcessBuilderSuite extends KyuubiFunSuite {
     assert(builder.commands.toSeq.contains("spark.kyuubi.key=value"))
   }
 
+  test("spark batch adds global Kubernetes watch scope labels") {
+    val conf = KyuubiConf(false)
+      .set("spark.master", "k8s://test:12345")
+    val builder = new SparkBatchProcessBuilder(
+      "kyuubi",
+      conf,
+      UUID.randomUUID().toString,
+      "test",
+      Some("test"),
+      "test",
+      Map(KyuubiConf.ENGINE_SHARE_LEVEL.key -> "CONNECTION"),
+      Seq.empty,
+      None)
+    val commands = builder.toString.split(' ')
+
+    assert(commands.contains(
+      s"spark.kubernetes.driver.label.$LABEL_KYUUBI_WATCH_SCOPE_KEY=$GLOBAL_WATCH_SCOPE"))
+    assert(commands.contains(
+      s"spark.kubernetes.driver.service.label.$LABEL_KYUUBI_WATCH_SCOPE_KEY=$GLOBAL_WATCH_SCOPE"))
+  }
+
   test("spark.kubernetes.file.upload.path supports placeholder") {
     val conf1 = KyuubiConf(false)
     conf1.set("spark.master", "k8s://test:12345")

kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilderSuite.scala

@@ -30,12 +30,14 @@ import org.scalatestplus.mockito.MockitoSugar
 import org.apache.kyuubi._
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.config.KyuubiConf._
+import org.apache.kyuubi.engine.KubernetesApplicationOperation.{GLOBAL_WATCH_SCOPE, LABEL_KYUUBI_WATCH_SCOPE_KEY}
 import org.apache.kyuubi.engine.ProcBuilder.KYUUBI_ENGINE_LOG_PATH_KEY
 import org.apache.kyuubi.engine.spark.SparkProcessBuilder._
 import org.apache.kyuubi.ha.HighAvailabilityConf
 import org.apache.kyuubi.ha.client.AuthTypes
 import org.apache.kyuubi.service.ServiceUtils
 import org.apache.kyuubi.util.AssertionUtils._
+import org.apache.kyuubi.util.KubernetesUtils
 import org.apache.kyuubi.util.command.CommandLineUtils._
 
 class SparkProcessBuilderSuite extends KerberizedTestHelper with MockitoSugar {
@@ -285,6 +287,34 @@ class SparkProcessBuilderSuite extends KerberizedTestHelper with MockitoSugar {
     assert(pb.toString.contains(engineRefId))
   }
 
+  test("SparkProcessBuilder adds Kubernetes watch scope labels") {
+    val globalCommands = new SparkProcessBuilder(
+      "kyuubi",
+      true,
+      conf.set(MASTER_KEY, "k8s://internal"))
+      .toString.split(' ')
+
+    assert(globalCommands.contains(
+      s"spark.kubernetes.driver.label.$LABEL_KYUUBI_WATCH_SCOPE_KEY=$GLOBAL_WATCH_SCOPE"))
+    assert(globalCommands.contains(
+      s"spark.kubernetes.driver.service.label.$LABEL_KYUUBI_WATCH_SCOPE_KEY=$GLOBAL_WATCH_SCOPE"))
+
+    val serverAddressLabelValue = KubernetesUtils.serverAddressLabelValue
+    val localCommands = new SparkProcessBuilder(
+      "kyuubi",
+      true,
+      conf
+        .set(MASTER_KEY, "k8s://internal")
+        .set(ENGINE_SHARE_LEVEL, "CONNECTION"))
+      .toString.split(' ')
+
+    assert(localCommands.contains(
+      s"spark.kubernetes.driver.label.$LABEL_KYUUBI_WATCH_SCOPE_KEY=$serverAddressLabelValue"))
+    assert(localCommands.contains(
+      s"spark.kubernetes.driver.service.label.$LABEL_KYUUBI_WATCH_SCOPE_KEY=" +
+        serverAddressLabelValue))
+  }
+
   test("SparkProcessBuilder build spark engine with SPARK_USER_NAME") {
     val proxyName = "kyuubi"
     val conf1 = KyuubiConf(false).set("spark.master", "k8s://test:12345")