From 73f9afffcb5deb13a47483108fe1480c3b1c9748 Mon Sep 17 00:00:00 2001 From: ruanwenjun Date: Sun, 20 Sep 2026 17:19:45 +0800 Subject: [PATCH 1/4] [KYUUBI #7756][KUBERNETES] Support owner-scoped application watches --- docs/configuration/settings.md | 1 + .../org/apache/kyuubi/config/KyuubiConf.scala | 13 ++++++++ .../KubernetesApplicationOperation.scala | 29 +++++++++++++--- .../engine/KyuubiApplicationManager.scala | 8 ++++- .../spark/SparkBatchProcessBuilder.scala | 3 ++ .../apache/kyuubi/util/KubernetesUtils.scala | 9 +++++ .../KubernetesApplicationOperationSuite.scala | 33 +++++++++++++++++++ .../KyuubiApplicationManagerSuite.scala | 20 ++++++++++- .../spark/SparkBatchProcessBuilderSuite.scala | 25 ++++++++++++++ .../spark/SparkProcessBuilderSuite.scala | 18 ++++++++++ 10 files changed, 152 insertions(+), 7 deletions(-) diff --git a/docs/configuration/settings.md b/docs/configuration/settings.md index f18e6ba7da0..be47ff15416 100644 --- a/docs/configuration/settings.md +++ b/docs/configuration/settings.md @@ -368,6 +368,7 @@ You can configure the Kyuubi properties in `$KYUUBI_HOME/conf/kyuubi-defaults.co | Key | Default | Meaning | Type | Since | |----------------------------------------------------------------------|----------------------------------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|----------|--------| | kyuubi.kubernetes.application.cleanup.virtualThreads.enabled | false | Whether asynchronous Kubernetes application cleanup tasks in the Kyuubi server use virtual threads. Requires at least Java 21; Java 25 or later is recommended. Defaults to kyuubi.server.virtualThreads.enabled. | boolean | 1.13.0 | +| kyuubi.kubernetes.application.ownerScopedWatch.enabled | false | Whether each Kyuubi server only watches Kubernetes applications submitted by itself. When enabled, Kyuubi adds the server hostname as the `kyuubi.apache.org/server-name` label. The hostname must be a valid Kubernetes label value, unique across servers, and stable across restarts. This applies to all Spark applications submitted by Kyuubi, including SQL engines and batch applications. | boolean | 1.13.0 | | kyuubi.kubernetes.application.state.container | spark-kubernetes-driver | The container name to retrieve the application state from. | string | 1.8.1 | | kyuubi.kubernetes.application.state.source | POD | The source to retrieve the application state from. The valid values are pod and container. When the pod is in a terminated state, the container state will be ignored, and the application state will be determined based on the pod state. If the source is container and there is container inside the pod with the name of kyuubi.kubernetes.application.state.container, the application state will be from the matched container state. Otherwise, the application state will be from the pod state. | string | 1.8.1 | | kyuubi.kubernetes.authenticate.caCertFile | <undefined> | Path to the CA cert file for connecting to the Kubernetes API server over TLS from the kyuubi. Specify this as a path as opposed to a URI (i.e. do not provide a scheme) | string | 1.7.0 | diff --git a/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala b/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala index ba68133b8f0..e0e22bd269a 100644 --- a/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala +++ b/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala @@ -1462,6 +1462,19 @@ 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 submitted by itself. " + + "When enabled, Kyuubi adds the server hostname as the " + + "`kyuubi.apache.org/server-name` label. The hostname must be a valid Kubernetes label " + + "value, unique across servers, and stable across restarts. This applies to all Spark " + + "applications submitted by Kyuubi, including SQL engines and batch applications.") + .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. " + diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/KubernetesApplicationOperation.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/KubernetesApplicationOperation.scala index 85723bff0fd..bc90d474682 100644 --- a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/KubernetesApplicationOperation.scala +++ b/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,26 @@ 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 + .withLabel(LABEL_KYUUBI_SERVER_NAME_KEY, KubernetesUtils.serverName) + .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 + .withLabel(LABEL_KYUUBI_SERVER_NAME_KEY, KubernetesUtils.serverName) + .inform(new SparkEngineSvcEventHandler(kubernetesInfo)) + } else { + engineServices.inform(new SparkEngineSvcEventHandler(kubernetesInfo)) + } engineSvcInformers.put(kubernetesInfo, engineSvcInformer) } client @@ -134,6 +148,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 server " + + s"${KubernetesUtils.serverName}.") + } 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 +594,7 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging { object KubernetesApplicationOperation extends Logging { val LABEL_KYUUBI_UNIQUE_KEY = "kyuubi-unique-tag" + val LABEL_KYUUBI_SERVER_NAME_KEY = "kyuubi.apache.org/server-name" 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" diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/KyuubiApplicationManager.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/KyuubiApplicationManager.scala index ec4c83b47e2..9ea4e61057c 100644 --- a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/KyuubiApplicationManager.scala +++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/KyuubiApplicationManager.scala @@ -28,11 +28,12 @@ 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.{LABEL_KYUUBI_SERVER_NAME_KEY, LABEL_KYUUBI_UNIQUE_KEY} 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]) @@ -126,6 +127,11 @@ object KyuubiApplicationManager { private def setupSparkK8sTag(tag: String, conf: KyuubiConf): Unit = { conf.set("spark.kubernetes.driver.label." + LABEL_KYUUBI_UNIQUE_KEY, tag) + if (conf.get(KyuubiConf.KUBERNETES_APPLICATION_OWNER_SCOPED_WATCH_ENABLED)) { + val serverName = KubernetesUtils.serverName + conf.set("spark.kubernetes.driver.label." + LABEL_KYUUBI_SERVER_NAME_KEY, serverName) + conf.set("spark.kubernetes.driver.service.label." + LABEL_KYUUBI_SERVER_NAME_KEY, serverName) + } } private def setupFlinkYarnTag(tag: String, conf: KyuubiConf): Unit = { diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/spark/SparkBatchProcessBuilder.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/spark/SparkBatchProcessBuilder.scala index 4ca38ef3b7f..a12c305e3fc 100644 --- a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/spark/SparkBatchProcessBuilder.scala +++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/spark/SparkBatchProcessBuilder.scala @@ -50,6 +50,9 @@ class SparkBatchProcessBuilder( // complete `spark.master` if absent on kubernetes completeMasterUrl(batchKyuubiConf) batchConf.foreach(entry => { batchKyuubiConf.set(entry._1, entry._2) }) + batchKyuubiConf.set( + KyuubiConf.KUBERNETES_APPLICATION_OWNER_SCOPED_WATCH_ENABLED, + conf.get(KyuubiConf.KUBERNETES_APPLICATION_OWNER_SCOPED_WATCH_ENABLED)) // tag batch application KyuubiApplicationManager.tagApplication(batchId, "spark", clusterManager(), batchKyuubiConf) diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/util/KubernetesUtils.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/util/KubernetesUtils.scala index de9a38d14b9..0930c01c231 100644 --- a/kyuubi-server/src/main/scala/org/apache/kyuubi/util/KubernetesUtils.scala +++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/util/KubernetesUtils.scala @@ -34,6 +34,15 @@ import org.apache.kyuubi.config.KyuubiConf._ object KubernetesUtils extends Logging { final val DRIVER_POD_NAME_MAX_LENGTH = 253 + + lazy val serverName: String = validateServerName(JavaUtils.findLocalInetAddress.getHostName) + + private[kyuubi] def validateServerName(hostname: String): String = { + require( + hostname.matches("[A-Za-z0-9]([A-Za-z0-9_.-]{0,61}[A-Za-z0-9])?"), + s"Kyuubi server hostname '$hostname' is not a valid Kubernetes label value") + hostname + } 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 = diff --git a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/KubernetesApplicationOperationSuite.scala b/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/KubernetesApplicationOperationSuite.scala index 4cc79b583db..ae1135fd3c6 100644 --- a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/KubernetesApplicationOperationSuite.scala +++ b/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/KubernetesApplicationOperationSuite.scala @@ -23,6 +23,7 @@ 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 +51,38 @@ class KubernetesApplicationOperationSuite extends KyuubiFunSuite { } } + test("server name should be a valid Kubernetes label value") { + assert(KubernetesUtils.validateServerName("kyuubi-0") === "kyuubi-0") + assert(KubernetesUtils.validateServerName("kyuubi.server_0") === "kyuubi.server_0") + + intercept[IllegalArgumentException] { + KubernetesUtils.validateServerName("kyuubi/server") + } + intercept[IllegalArgumentException] { + KubernetesUtils.validateServerName("k" * 64) + } + } + + 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") diff --git a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/KyuubiApplicationManagerSuite.scala b/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/KyuubiApplicationManagerSuite.scala index bdb06cb9687..ce4f78cb9da 100644 --- a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/KyuubiApplicationManagerSuite.scala +++ b/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/KyuubiApplicationManagerSuite.scala @@ -19,8 +19,9 @@ 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.{LABEL_KYUUBI_SERVER_NAME_KEY, LABEL_KYUUBI_UNIQUE_KEY} import org.apache.kyuubi.engine.spark.SparkProcessBuilder +import org.apache.kyuubi.util.KubernetesUtils class KyuubiApplicationManagerSuite extends KyuubiFunSuite { test("application access path") { @@ -187,5 +188,22 @@ class KyuubiApplicationManagerSuite extends KyuubiFunSuite { val yarnTag = conf.getOption("spark.yarn.tags") assert(kubernetesTag.nonEmpty && tag.equals(kubernetesTag.get)) assert(yarnTag.isEmpty) + + val serverNamePodLabel = "spark.kubernetes.driver.label." + LABEL_KYUUBI_SERVER_NAME_KEY + val serverNameServiceLabel = + "spark.kubernetes.driver.service.label." + LABEL_KYUUBI_SERVER_NAME_KEY + assert(conf.getOption(serverNamePodLabel).isEmpty) + assert(conf.getOption(serverNameServiceLabel).isEmpty) + + conf.set(KyuubiConf.KUBERNETES_APPLICATION_OWNER_SCOPED_WATCH_ENABLED, true) + conf.set(serverNamePodLabel, "overridden") + conf.set(serverNameServiceLabel, "overridden") + KyuubiApplicationManager.tagApplication( + tag, + "SPARK", + Some("k8s://https://kyuubi-test:8443"), + conf) + assert(conf.getOption(serverNamePodLabel).contains(KubernetesUtils.serverName)) + assert(conf.getOption(serverNameServiceLabel).contains(KubernetesUtils.serverName)) } } diff --git a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkBatchProcessBuilderSuite.scala b/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkBatchProcessBuilderSuite.scala index f858a0f7784..627d0841d41 100644 --- a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkBatchProcessBuilderSuite.scala +++ b/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkBatchProcessBuilderSuite.scala @@ -23,6 +23,8 @@ import java.util.UUID import org.apache.kyuubi.KyuubiFunSuite import org.apache.kyuubi.config.KyuubiConf +import org.apache.kyuubi.engine.KubernetesApplicationOperation.LABEL_KYUUBI_SERVER_NAME_KEY +import org.apache.kyuubi.util.KubernetesUtils class SparkBatchProcessBuilderSuite extends KyuubiFunSuite { test("spark batch conf should be converted with `spark.` prefix") { @@ -39,6 +41,29 @@ class SparkBatchProcessBuilderSuite extends KyuubiFunSuite { assert(builder.commands.toSeq.contains("spark.kyuubi.key=value")) } + test("spark batch adds Kubernetes server name labels") { + val conf = KyuubiConf(false) + .set("spark.master", "k8s://test:12345") + .set(KyuubiConf.KUBERNETES_APPLICATION_OWNER_SCOPED_WATCH_ENABLED, true) + val builder = new SparkBatchProcessBuilder( + "kyuubi", + conf, + UUID.randomUUID().toString, + "test", + Some("test"), + "test", + Map.empty, + Seq.empty, + None) + val commands = builder.toString.split(' ') + val serverName = KubernetesUtils.serverName + + assert(commands.contains( + s"spark.kubernetes.driver.label.$LABEL_KYUUBI_SERVER_NAME_KEY=$serverName")) + assert(commands.contains( + s"spark.kubernetes.driver.service.label.$LABEL_KYUUBI_SERVER_NAME_KEY=$serverName")) + } + test("spark.kubernetes.file.upload.path supports placeholder") { val conf1 = KyuubiConf(false) conf1.set("spark.master", "k8s://test:12345") diff --git a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilderSuite.scala b/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilderSuite.scala index be24c53c20c..649a6875924 100644 --- a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilderSuite.scala +++ b/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.LABEL_KYUUBI_SERVER_NAME_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,22 @@ class SparkProcessBuilderSuite extends KerberizedTestHelper with MockitoSugar { assert(pb.toString.contains(engineRefId)) } + test("SparkProcessBuilder adds Kubernetes server name labels") { + val serverName = KubernetesUtils.serverName + val commands = new SparkProcessBuilder( + "kyuubi", + true, + conf + .set(MASTER_KEY, "k8s://internal") + .set(KUBERNETES_APPLICATION_OWNER_SCOPED_WATCH_ENABLED, true)) + .toString.split(' ') + + assert(commands.contains( + s"spark.kubernetes.driver.label.$LABEL_KYUUBI_SERVER_NAME_KEY=$serverName")) + assert(commands.contains( + s"spark.kubernetes.driver.service.label.$LABEL_KYUUBI_SERVER_NAME_KEY=$serverName")) + } + test("SparkProcessBuilder build spark engine with SPARK_USER_NAME") { val proxyName = "kyuubi" val conf1 = KyuubiConf(false).set("spark.master", "k8s://test:12345") From 66348b0815e4117e8b265aeda388c20a0b6ec6d7 Mon Sep 17 00:00:00 2001 From: ruanwenjun Date: Mon, 21 Sep 2026 12:07:38 +0800 Subject: [PATCH 2/4] [KYUUBI #7756][KUBERNETES] Refine owner-scoped application watches --- docs/configuration/settings.md | 52 +++++++++---------- .../org/apache/kyuubi/config/KyuubiConf.scala | 12 +++-- .../KubernetesApplicationOperation.scala | 17 ++++-- .../engine/KyuubiApplicationManager.scala | 31 ++++++++--- .../spark/SparkBatchProcessBuilder.scala | 3 -- .../engine/spark/SparkProcessBuilder.scala | 7 ++- .../apache/kyuubi/util/KubernetesUtils.scala | 11 ++-- .../KubernetesApplicationOperationSuite.scala | 17 +++--- .../KyuubiApplicationManagerSuite.scala | 48 +++++++++++------ .../spark/SparkBatchProcessBuilderSuite.scala | 13 ++--- .../spark/SparkProcessBuilderSuite.scala | 29 +++++++---- 11 files changed, 142 insertions(+), 98 deletions(-) diff --git a/docs/configuration/settings.md b/docs/configuration/settings.md index be47ff15416..23cd22cd25b 100644 --- a/docs/configuration/settings.md +++ b/docs/configuration/settings.md @@ -365,32 +365,32 @@ You can configure the Kyuubi properties in `$KYUUBI_HOME/conf/kyuubi-defaults.co ### Kubernetes -| Key | Default | Meaning | Type | Since | -|----------------------------------------------------------------------|----------------------------------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|----------|--------| -| kyuubi.kubernetes.application.cleanup.virtualThreads.enabled | false | Whether asynchronous Kubernetes application cleanup tasks in the Kyuubi server use virtual threads. Requires at least Java 21; Java 25 or later is recommended. Defaults to kyuubi.server.virtualThreads.enabled. | boolean | 1.13.0 | -| kyuubi.kubernetes.application.ownerScopedWatch.enabled | false | Whether each Kyuubi server only watches Kubernetes applications submitted by itself. When enabled, Kyuubi adds the server hostname as the `kyuubi.apache.org/server-name` label. The hostname must be a valid Kubernetes label value, unique across servers, and stable across restarts. This applies to all Spark applications submitted by Kyuubi, including SQL engines and batch applications. | boolean | 1.13.0 | -| kyuubi.kubernetes.application.state.container | spark-kubernetes-driver | The container name to retrieve the application state from. | string | 1.8.1 | -| kyuubi.kubernetes.application.state.source | POD | The source to retrieve the application state from. The valid values are pod and container. When the pod is in a terminated state, the container state will be ignored, and the application state will be determined based on the pod state. If the source is container and there is container inside the pod with the name of kyuubi.kubernetes.application.state.container, the application state will be from the matched container state. Otherwise, the application state will be from the pod state. | string | 1.8.1 | -| kyuubi.kubernetes.authenticate.caCertFile | <undefined> | Path to the CA cert file for connecting to the Kubernetes API server over TLS from the kyuubi. Specify this as a path as opposed to a URI (i.e. do not provide a scheme) | string | 1.7.0 | -| kyuubi.kubernetes.authenticate.clientCertFile | <undefined> | Path to the client cert file for connecting to the Kubernetes API server over TLS from the kyuubi. Specify this as a path as opposed to a URI (i.e. do not provide a scheme) | string | 1.7.0 | -| kyuubi.kubernetes.authenticate.clientKeyFile | <undefined> | Path to the client key file for connecting to the Kubernetes API server over TLS from the kyuubi. Specify this as a path as opposed to a URI (i.e. do not provide a scheme) | string | 1.7.0 | -| kyuubi.kubernetes.authenticate.oauthToken | <undefined> | The OAuth token to use when authenticating against the Kubernetes API server. Note that unlike, the other authentication options, this must be the exact string value of the token to use for the authentication. | string | 1.7.0 | -| kyuubi.kubernetes.authenticate.oauthTokenFile | <undefined> | Path to the file containing the OAuth token to use when authenticating against the Kubernetes API server. Specify this as a path as opposed to a URI (i.e. do not provide a scheme) | string | 1.7.0 | -| kyuubi.kubernetes.client.dispatcher.virtualThreads.enabled | false | Whether the Kyuubi server Kubernetes HTTP client dispatcher uses virtual threads. Requires at least Java 21; Java 25 or later is recommended. Defaults to kyuubi.server.virtualThreads.enabled. | boolean | 1.13.0 | -| kyuubi.kubernetes.client.initialize.list || The kubernetes client initialize list to register kubernetes resource informers during Kyuubi server startup. This ensures the Kyuubi server is promptly informed for any Kubernetes resource changes after startup. It is highly recommended to set it for multiple Kyuubi instances mode. The format is `context1:namespace1,context2:namespace2`. When the list is empty and Kyuubi runs in Kubernetes, the client for the configured namespace is initialized automatically with the in-cluster configuration. | seq | 1.11.0 | -| kyuubi.kubernetes.context | <undefined> | The desired context from your kubernetes config file used to configure the K8s client for interacting with the cluster. | string | 1.6.0 | -| kyuubi.kubernetes.context.allow.list || The allowed kubernetes context list, if it is empty, there is no kubernetes context limitation. | set | 1.8.0 | -| kyuubi.kubernetes.master.address | <undefined> | The internal Kubernetes master (API server) address to be used for kyuubi. | string | 1.7.0 | -| kyuubi.kubernetes.namespace | default | The default namespace used by the Kyuubi server's Kubernetes client to discover and manage engine pods, when the engine submission does not specify one (e.g. `spark.kubernetes.namespace`). It does not control the namespace where the Kyuubi server itself runs. | string | 1.7.0 | -| kyuubi.kubernetes.namespace.allow.list || The allowed kubernetes namespace list, if it is empty, there is no kubernetes namespace limitation. | set | 1.8.0 | -| kyuubi.kubernetes.spark.appUrlPattern | http://{{SPARK_DRIVER_SVC}}.{{KUBERNETES_NAMESPACE}}.svc:{{SPARK_UI_PORT}} | The pattern to generate the spark on kubernetes application UI URL. The pattern should contain placeholders for the application variables. Available placeholders are `{{SPARK_APP_ID}}`, `{{SPARK_DRIVER_SVC}}`, `{{SPARK_DRIVER_POD_IP}}`, `{{KUBERNETES_NAMESPACE}}`, `{{KUBERNETES_CONTEXT}}` and `{{SPARK_UI_PORT}}`. | string | 1.10.0 | -| kyuubi.kubernetes.spark.autoCreateFileUploadPath.enabled | false | If enabled, Kyuubi server will try to create the `spark.kubernetes.file.upload.path` with permission 777 before submitting the Spark application. | boolean | 1.11.0 | -| kyuubi.kubernetes.spark.cleanupTerminatedDriverPod.checkInterval | PT1M | Kyuubi server use guava cache as the cleanup trigger with time-based eviction, but the eviction would not happened until any get/put operation happened. This option schedule a daemon thread evict cache periodically. | duration | 1.8.1 | -| kyuubi.kubernetes.spark.cleanupTerminatedDriverPod.kind | NONE | Kyuubi server will delete the spark driver pod after the application terminates for kyuubi.kubernetes.terminatedApplicationRetainPeriod. Available options are NONE, ALL, COMPLETED and default value is None which means none of the pod will be deleted | string | 1.8.1 | -| kyuubi.kubernetes.spark.forciblyRewriteDriverPodName.enabled | false | Whether to forcibly rewrite Spark driver pod name with 'kyuubi--driver'. If disabled, Kyuubi will try to preserve the application name while satisfying K8s' pod name policy, but some vendors may have stricter pod name policies, thus the generated name may become illegal. | boolean | 1.8.1 | -| kyuubi.kubernetes.spark.forciblyRewriteExecutorPodNamePrefix.enabled | false | Whether to forcibly rewrite Spark executor pod name prefix with 'kyuubi-'. If disabled, Kyuubi will try to preserve the application name while satisfying K8s' pod name policy, but some vendors may have stricter Pod name policies, thus the generated name may become illegal. | boolean | 1.8.1 | -| kyuubi.kubernetes.terminatedApplicationRetainPeriod | PT5M | The period for which the Kyuubi server retains application information after the application terminates. | duration | 1.7.1 | -| kyuubi.kubernetes.trust.certificates | false | If set to true then client can submit to kubernetes cluster only with token | boolean | 1.7.0 | +| Key | Default | Meaning | Type | Since | +|----------------------------------------------------------------------|----------------------------------------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|----------|--------| +| kyuubi.kubernetes.application.cleanup.virtualThreads.enabled | false | Whether asynchronous Kubernetes application cleanup tasks in the Kyuubi server use virtual threads. Requires at least Java 21; Java 25 or later is recommended. Defaults to kyuubi.server.virtualThreads.enabled. | boolean | 1.13.0 | +| kyuubi.kubernetes.application.ownerScopedWatch.enabled | false | 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. | boolean | 1.13.0 | +| kyuubi.kubernetes.application.state.container | spark-kubernetes-driver | The container name to retrieve the application state from. | string | 1.8.1 | +| kyuubi.kubernetes.application.state.source | POD | The source to retrieve the application state from. The valid values are pod and container. When the pod is in a terminated state, the container state will be ignored, and the application state will be determined based on the pod state. If the source is container and there is container inside the pod with the name of kyuubi.kubernetes.application.state.container, the application state will be from the matched container state. Otherwise, the application state will be from the pod state. | string | 1.8.1 | +| kyuubi.kubernetes.authenticate.caCertFile | <undefined> | Path to the CA cert file for connecting to the Kubernetes API server over TLS from the kyuubi. Specify this as a path as opposed to a URI (i.e. do not provide a scheme) | string | 1.7.0 | +| kyuubi.kubernetes.authenticate.clientCertFile | <undefined> | Path to the client cert file for connecting to the Kubernetes API server over TLS from the kyuubi. Specify this as a path as opposed to a URI (i.e. do not provide a scheme) | string | 1.7.0 | +| kyuubi.kubernetes.authenticate.clientKeyFile | <undefined> | Path to the client key file for connecting to the Kubernetes API server over TLS from the kyuubi. Specify this as a path as opposed to a URI (i.e. do not provide a scheme) | string | 1.7.0 | +| kyuubi.kubernetes.authenticate.oauthToken | <undefined> | The OAuth token to use when authenticating against the Kubernetes API server. Note that unlike, the other authentication options, this must be the exact string value of the token to use for the authentication. | string | 1.7.0 | +| kyuubi.kubernetes.authenticate.oauthTokenFile | <undefined> | Path to the file containing the OAuth token to use when authenticating against the Kubernetes API server. Specify this as a path as opposed to a URI (i.e. do not provide a scheme) | string | 1.7.0 | +| kyuubi.kubernetes.client.dispatcher.virtualThreads.enabled | false | Whether the Kyuubi server Kubernetes HTTP client dispatcher uses virtual threads. Requires at least Java 21; Java 25 or later is recommended. Defaults to kyuubi.server.virtualThreads.enabled. | boolean | 1.13.0 | +| kyuubi.kubernetes.client.initialize.list || The kubernetes client initialize list to register kubernetes resource informers during Kyuubi server startup. This ensures the Kyuubi server is promptly informed for any Kubernetes resource changes after startup. It is highly recommended to set it for multiple Kyuubi instances mode. The format is `context1:namespace1,context2:namespace2`. When the list is empty and Kyuubi runs in Kubernetes, the client for the configured namespace is initialized automatically with the in-cluster configuration. | seq | 1.11.0 | +| kyuubi.kubernetes.context | <undefined> | The desired context from your kubernetes config file used to configure the K8s client for interacting with the cluster. | string | 1.6.0 | +| kyuubi.kubernetes.context.allow.list || The allowed kubernetes context list, if it is empty, there is no kubernetes context limitation. | set | 1.8.0 | +| kyuubi.kubernetes.master.address | <undefined> | The internal Kubernetes master (API server) address to be used for kyuubi. | string | 1.7.0 | +| kyuubi.kubernetes.namespace | default | The default namespace used by the Kyuubi server's Kubernetes client to discover and manage engine pods, when the engine submission does not specify one (e.g. `spark.kubernetes.namespace`). It does not control the namespace where the Kyuubi server itself runs. | string | 1.7.0 | +| kyuubi.kubernetes.namespace.allow.list || The allowed kubernetes namespace list, if it is empty, there is no kubernetes namespace limitation. | set | 1.8.0 | +| kyuubi.kubernetes.spark.appUrlPattern | http://{{SPARK_DRIVER_SVC}}.{{KUBERNETES_NAMESPACE}}.svc:{{SPARK_UI_PORT}} | The pattern to generate the spark on kubernetes application UI URL. The pattern should contain placeholders for the application variables. Available placeholders are `{{SPARK_APP_ID}}`, `{{SPARK_DRIVER_SVC}}`, `{{SPARK_DRIVER_POD_IP}}`, `{{KUBERNETES_NAMESPACE}}`, `{{KUBERNETES_CONTEXT}}` and `{{SPARK_UI_PORT}}`. | string | 1.10.0 | +| kyuubi.kubernetes.spark.autoCreateFileUploadPath.enabled | false | If enabled, Kyuubi server will try to create the `spark.kubernetes.file.upload.path` with permission 777 before submitting the Spark application. | boolean | 1.11.0 | +| kyuubi.kubernetes.spark.cleanupTerminatedDriverPod.checkInterval | PT1M | Kyuubi server use guava cache as the cleanup trigger with time-based eviction, but the eviction would not happened until any get/put operation happened. This option schedule a daemon thread evict cache periodically. | duration | 1.8.1 | +| kyuubi.kubernetes.spark.cleanupTerminatedDriverPod.kind | NONE | Kyuubi server will delete the spark driver pod after the application terminates for kyuubi.kubernetes.terminatedApplicationRetainPeriod. Available options are NONE, ALL, COMPLETED and default value is None which means none of the pod will be deleted | string | 1.8.1 | +| kyuubi.kubernetes.spark.forciblyRewriteDriverPodName.enabled | false | Whether to forcibly rewrite Spark driver pod name with 'kyuubi--driver'. If disabled, Kyuubi will try to preserve the application name while satisfying K8s' pod name policy, but some vendors may have stricter pod name policies, thus the generated name may become illegal. | boolean | 1.8.1 | +| kyuubi.kubernetes.spark.forciblyRewriteExecutorPodNamePrefix.enabled | false | Whether to forcibly rewrite Spark executor pod name prefix with 'kyuubi-'. If disabled, Kyuubi will try to preserve the application name while satisfying K8s' pod name policy, but some vendors may have stricter Pod name policies, thus the generated name may become illegal. | boolean | 1.8.1 | +| kyuubi.kubernetes.terminatedApplicationRetainPeriod | PT5M | The period for which the Kyuubi server retains application information after the application terminates. | duration | 1.7.1 | +| kyuubi.kubernetes.trust.certificates | false | If set to true then client can submit to kubernetes cluster only with token | boolean | 1.7.0 | ### Lineage diff --git a/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala b/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala index e0e22bd269a..abc4cce0d44 100644 --- a/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala +++ b/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala @@ -1464,11 +1464,13 @@ object KyuubiConf { val KUBERNETES_APPLICATION_OWNER_SCOPED_WATCH_ENABLED: ConfigEntry[Boolean] = buildConf("kyuubi.kubernetes.application.ownerScopedWatch.enabled") - .doc("Whether each Kyuubi server only watches Kubernetes applications submitted by itself. " + - "When enabled, Kyuubi adds the server hostname as the " + - "`kyuubi.apache.org/server-name` label. The hostname must be a valid Kubernetes label " + - "value, unique across servers, and stable across restarts. This applies to all Spark " + - "applications submitted by Kyuubi, including SQL engines and batch applications.") + .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 diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/KubernetesApplicationOperation.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/KubernetesApplicationOperation.scala index bc90d474682..45510463eb5 100644 --- a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/KubernetesApplicationOperation.scala +++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/KubernetesApplicationOperation.scala @@ -119,7 +119,10 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging { val enginePods = client.pods().withLabel(LABEL_KYUUBI_UNIQUE_KEY) val enginePodInformer = if (ownerScopedWatchEnabled) { enginePods - .withLabel(LABEL_KYUUBI_SERVER_NAME_KEY, KubernetesUtils.serverName) + .withLabelIn( + LABEL_KYUUBI_WATCH_SCOPE_KEY, + KubernetesUtils.serverAddress, + GLOBAL_WATCH_SCOPE) .inform(new SparkEnginePodEventHandler(kubernetesInfo)) } else { enginePods.inform(new SparkEnginePodEventHandler(kubernetesInfo)) @@ -131,7 +134,10 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging { val engineServices = client.services() val engineSvcInformer = if (ownerScopedWatchEnabled) { engineServices - .withLabel(LABEL_KYUUBI_SERVER_NAME_KEY, KubernetesUtils.serverName) + .withLabelIn( + LABEL_KYUUBI_WATCH_SCOPE_KEY, + KubernetesUtils.serverAddress, + GLOBAL_WATCH_SCOPE) .inform(new SparkEngineSvcEventHandler(kubernetesInfo)) } else { engineServices.inform(new SparkEngineSvcEventHandler(kubernetesInfo)) @@ -149,8 +155,8 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging { this.metadataManager = metadataManager info("Start initializing Kubernetes application operation.") if (ownerScopedWatchEnabled) { - info(s"Kubernetes application owner-scoped watch is enabled for server " + - s"${KubernetesUtils.serverName}.") + info(s"Kubernetes application owner-scoped watch is enabled for scopes " + + s"${KubernetesUtils.serverAddress} and $GLOBAL_WATCH_SCOPE.") } submitTimeout = conf.get(KyuubiConf.ENGINE_KUBERNETES_SUBMIT_TIMEOUT) // Defer cleaning terminated application information @@ -594,7 +600,8 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging { object KubernetesApplicationOperation extends Logging { val LABEL_KYUUBI_UNIQUE_KEY = "kyuubi-unique-tag" - val LABEL_KYUUBI_SERVER_NAME_KEY = "kyuubi.apache.org/server-name" + 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" diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/KyuubiApplicationManager.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/KyuubiApplicationManager.scala index 9ea4e61057c..0084c116eb5 100644 --- a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/KyuubiApplicationManager.scala +++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/KyuubiApplicationManager.scala @@ -28,7 +28,8 @@ 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_SERVER_NAME_KEY, 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 @@ -125,13 +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) - if (conf.get(KyuubiConf.KUBERNETES_APPLICATION_OWNER_SCOPED_WATCH_ENABLED)) { - val serverName = KubernetesUtils.serverName - conf.set("spark.kubernetes.driver.label." + LABEL_KYUUBI_SERVER_NAME_KEY, serverName) - conf.set("spark.kubernetes.driver.service.label." + LABEL_KYUUBI_SERVER_NAME_KEY, serverName) + val watchScope = engineShareLevel.map(ShareLevel.withName) match { + case Some(CONNECTION | SERVER_LOCAL) => KubernetesUtils.serverAddress + 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 = { @@ -212,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) diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/spark/SparkBatchProcessBuilder.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/spark/SparkBatchProcessBuilder.scala index a12c305e3fc..4ca38ef3b7f 100644 --- a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/spark/SparkBatchProcessBuilder.scala +++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/spark/SparkBatchProcessBuilder.scala @@ -50,9 +50,6 @@ class SparkBatchProcessBuilder( // complete `spark.master` if absent on kubernetes completeMasterUrl(batchKyuubiConf) batchConf.foreach(entry => { batchKyuubiConf.set(entry._1, entry._2) }) - batchKyuubiConf.set( - KyuubiConf.KUBERNETES_APPLICATION_OWNER_SCOPED_WATCH_ENABLED, - conf.get(KyuubiConf.KUBERNETES_APPLICATION_OWNER_SCOPED_WATCH_ENABLED)) // tag batch application KyuubiApplicationManager.tagApplication(batchId, "spark", clusterManager(), batchKyuubiConf) diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilder.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilder.scala index 813c42aad32..66c9961761e 100644 --- a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilder.scala +++ b/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 diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/util/KubernetesUtils.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/util/KubernetesUtils.scala index 0930c01c231..8dc1a575c1e 100644 --- a/kyuubi-server/src/main/scala/org/apache/kyuubi/util/KubernetesUtils.scala +++ b/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 @@ -35,13 +36,11 @@ import org.apache.kyuubi.config.KyuubiConf._ object KubernetesUtils extends Logging { final val DRIVER_POD_NAME_MAX_LENGTH = 253 - lazy val serverName: String = validateServerName(JavaUtils.findLocalInetAddress.getHostName) + lazy val serverAddress: String = toServerAddressLabelValue(JavaUtils.findLocalInetAddress) - private[kyuubi] def validateServerName(hostname: String): String = { - require( - hostname.matches("[A-Za-z0-9]([A-Za-z0-9_.-]{0,61}[A-Za-z0-9])?"), - s"Kyuubi server hostname '$hostname' is not a valid Kubernetes label value") - hostname + 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 diff --git a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/KubernetesApplicationOperationSuite.scala b/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/KubernetesApplicationOperationSuite.scala index ae1135fd3c6..59fdfd2f7c4 100644 --- a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/KubernetesApplicationOperationSuite.scala +++ b/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/KubernetesApplicationOperationSuite.scala @@ -17,6 +17,8 @@ package org.apache.kyuubi.engine +import java.net.InetAddress + import io.fabric8.kubernetes.api.model.{ContainerState, ContainerStateWaiting, PodBuilder} import org.apache.kyuubi.{KyuubiException, KyuubiFunSuite} @@ -51,16 +53,11 @@ class KubernetesApplicationOperationSuite extends KyuubiFunSuite { } } - test("server name should be a valid Kubernetes label value") { - assert(KubernetesUtils.validateServerName("kyuubi-0") === "kyuubi-0") - assert(KubernetesUtils.validateServerName("kyuubi.server_0") === "kyuubi.server_0") - - intercept[IllegalArgumentException] { - KubernetesUtils.validateServerName("kyuubi/server") - } - intercept[IllegalArgumentException] { - KubernetesUtils.validateServerName("k" * 64) - } + 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") { diff --git a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/KyuubiApplicationManagerSuite.scala b/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/KyuubiApplicationManagerSuite.scala index ce4f78cb9da..1fb00924688 100644 --- a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/KyuubiApplicationManagerSuite.scala +++ b/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/KyuubiApplicationManagerSuite.scala @@ -19,7 +19,8 @@ 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_SERVER_NAME_KEY, 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 @@ -178,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", @@ -188,22 +193,31 @@ 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.serverAddress)) + assert(conf.getOption(watchScopeServiceLabel).contains(KubernetesUtils.serverAddress)) + } - val serverNamePodLabel = "spark.kubernetes.driver.label." + LABEL_KYUUBI_SERVER_NAME_KEY - val serverNameServiceLabel = - "spark.kubernetes.driver.service.label." + LABEL_KYUUBI_SERVER_NAME_KEY - assert(conf.getOption(serverNamePodLabel).isEmpty) - assert(conf.getOption(serverNameServiceLabel).isEmpty) - - conf.set(KyuubiConf.KUBERNETES_APPLICATION_OWNER_SCOPED_WATCH_ENABLED, true) - conf.set(serverNamePodLabel, "overridden") - conf.set(serverNameServiceLabel, "overridden") - KyuubiApplicationManager.tagApplication( - tag, - "SPARK", - Some("k8s://https://kyuubi-test:8443"), - conf) - assert(conf.getOption(serverNamePodLabel).contains(KubernetesUtils.serverName)) - assert(conf.getOption(serverNameServiceLabel).contains(KubernetesUtils.serverName)) + 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)) + } } } diff --git a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkBatchProcessBuilderSuite.scala b/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkBatchProcessBuilderSuite.scala index 627d0841d41..dee6c57e973 100644 --- a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkBatchProcessBuilderSuite.scala +++ b/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkBatchProcessBuilderSuite.scala @@ -23,8 +23,7 @@ import java.util.UUID import org.apache.kyuubi.KyuubiFunSuite import org.apache.kyuubi.config.KyuubiConf -import org.apache.kyuubi.engine.KubernetesApplicationOperation.LABEL_KYUUBI_SERVER_NAME_KEY -import org.apache.kyuubi.util.KubernetesUtils +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") { @@ -41,10 +40,9 @@ class SparkBatchProcessBuilderSuite extends KyuubiFunSuite { assert(builder.commands.toSeq.contains("spark.kyuubi.key=value")) } - test("spark batch adds Kubernetes server name labels") { + test("spark batch adds global Kubernetes watch scope labels") { val conf = KyuubiConf(false) .set("spark.master", "k8s://test:12345") - .set(KyuubiConf.KUBERNETES_APPLICATION_OWNER_SCOPED_WATCH_ENABLED, true) val builder = new SparkBatchProcessBuilder( "kyuubi", conf, @@ -52,16 +50,15 @@ class SparkBatchProcessBuilderSuite extends KyuubiFunSuite { "test", Some("test"), "test", - Map.empty, + Map(KyuubiConf.ENGINE_SHARE_LEVEL.key -> "CONNECTION"), Seq.empty, None) val commands = builder.toString.split(' ') - val serverName = KubernetesUtils.serverName assert(commands.contains( - s"spark.kubernetes.driver.label.$LABEL_KYUUBI_SERVER_NAME_KEY=$serverName")) + s"spark.kubernetes.driver.label.$LABEL_KYUUBI_WATCH_SCOPE_KEY=$GLOBAL_WATCH_SCOPE")) assert(commands.contains( - s"spark.kubernetes.driver.service.label.$LABEL_KYUUBI_SERVER_NAME_KEY=$serverName")) + s"spark.kubernetes.driver.service.label.$LABEL_KYUUBI_WATCH_SCOPE_KEY=$GLOBAL_WATCH_SCOPE")) } test("spark.kubernetes.file.upload.path supports placeholder") { diff --git a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilderSuite.scala b/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilderSuite.scala index 649a6875924..400b4468cff 100644 --- a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilderSuite.scala +++ b/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilderSuite.scala @@ -30,7 +30,7 @@ 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.LABEL_KYUUBI_SERVER_NAME_KEY +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 @@ -287,20 +287,31 @@ class SparkProcessBuilderSuite extends KerberizedTestHelper with MockitoSugar { assert(pb.toString.contains(engineRefId)) } - test("SparkProcessBuilder adds Kubernetes server name labels") { - val serverName = KubernetesUtils.serverName - val commands = new SparkProcessBuilder( + 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 serverAddress = KubernetesUtils.serverAddress + val localCommands = new SparkProcessBuilder( "kyuubi", true, conf .set(MASTER_KEY, "k8s://internal") - .set(KUBERNETES_APPLICATION_OWNER_SCOPED_WATCH_ENABLED, true)) + .set(ENGINE_SHARE_LEVEL, "CONNECTION")) .toString.split(' ') - assert(commands.contains( - s"spark.kubernetes.driver.label.$LABEL_KYUUBI_SERVER_NAME_KEY=$serverName")) - assert(commands.contains( - s"spark.kubernetes.driver.service.label.$LABEL_KYUUBI_SERVER_NAME_KEY=$serverName")) + assert(localCommands.contains( + s"spark.kubernetes.driver.label.$LABEL_KYUUBI_WATCH_SCOPE_KEY=$serverAddress")) + assert(localCommands.contains( + s"spark.kubernetes.driver.service.label.$LABEL_KYUUBI_WATCH_SCOPE_KEY=$serverAddress")) } test("SparkProcessBuilder build spark engine with SPARK_USER_NAME") { From 1aea9283c4811196c85035ba7106ad5415b73a13 Mon Sep 17 00:00:00 2001 From: ruanwenjun Date: Mon, 21 Sep 2026 13:51:38 +0800 Subject: [PATCH 3/4] [KYUUBI #7756][KUBERNETES] Fix watch scope style --- .../src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala b/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala index abc4cce0d44..4dbf0ac75e3 100644 --- a/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala +++ b/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala @@ -1468,7 +1468,8 @@ object KyuubiConf { "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. " + + "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") From 61fc1be92be9ec2ce7bb1845b158531179b3c543 Mon Sep 17 00:00:00 2001 From: ruanwenjun Date: Mon, 21 Sep 2026 16:30:57 +0800 Subject: [PATCH 4/4] [KYUUBI #7756][KUBERNETES] Clarify server address label value name --- .../kyuubi/engine/KubernetesApplicationOperation.scala | 6 +++--- .../apache/kyuubi/engine/KyuubiApplicationManager.scala | 2 +- .../scala/org/apache/kyuubi/util/KubernetesUtils.scala | 3 ++- .../kyuubi/engine/KyuubiApplicationManagerSuite.scala | 5 +++-- .../kyuubi/engine/spark/SparkProcessBuilderSuite.scala | 7 ++++--- 5 files changed, 13 insertions(+), 10 deletions(-) diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/KubernetesApplicationOperation.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/KubernetesApplicationOperation.scala index 45510463eb5..2b8a24ddebd 100644 --- a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/KubernetesApplicationOperation.scala +++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/KubernetesApplicationOperation.scala @@ -121,7 +121,7 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging { enginePods .withLabelIn( LABEL_KYUUBI_WATCH_SCOPE_KEY, - KubernetesUtils.serverAddress, + KubernetesUtils.serverAddressLabelValue, GLOBAL_WATCH_SCOPE) .inform(new SparkEnginePodEventHandler(kubernetesInfo)) } else { @@ -136,7 +136,7 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging { engineServices .withLabelIn( LABEL_KYUUBI_WATCH_SCOPE_KEY, - KubernetesUtils.serverAddress, + KubernetesUtils.serverAddressLabelValue, GLOBAL_WATCH_SCOPE) .inform(new SparkEngineSvcEventHandler(kubernetesInfo)) } else { @@ -156,7 +156,7 @@ class KubernetesApplicationOperation extends ApplicationOperation with Logging { info("Start initializing Kubernetes application operation.") if (ownerScopedWatchEnabled) { info(s"Kubernetes application owner-scoped watch is enabled for scopes " + - s"${KubernetesUtils.serverAddress} and $GLOBAL_WATCH_SCOPE.") + s"${KubernetesUtils.serverAddressLabelValue} and $GLOBAL_WATCH_SCOPE.") } submitTimeout = conf.get(KyuubiConf.ENGINE_KUBERNETES_SUBMIT_TIMEOUT) // Defer cleaning terminated application information diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/KyuubiApplicationManager.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/KyuubiApplicationManager.scala index 0084c116eb5..ffa43de1d57 100644 --- a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/KyuubiApplicationManager.scala +++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/KyuubiApplicationManager.scala @@ -132,7 +132,7 @@ object KyuubiApplicationManager { 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.serverAddress + case Some(CONNECTION | SERVER_LOCAL) => KubernetesUtils.serverAddressLabelValue case _ => GLOBAL_WATCH_SCOPE } conf.set("spark.kubernetes.driver.label." + LABEL_KYUUBI_WATCH_SCOPE_KEY, watchScope) diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/util/KubernetesUtils.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/util/KubernetesUtils.scala index 8dc1a575c1e..f630b58e508 100644 --- a/kyuubi-server/src/main/scala/org/apache/kyuubi/util/KubernetesUtils.scala +++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/util/KubernetesUtils.scala @@ -36,7 +36,8 @@ import org.apache.kyuubi.config.KyuubiConf._ object KubernetesUtils extends Logging { final val DRIVER_POD_NAME_MAX_LENGTH = 253 - lazy val serverAddress: String = toServerAddressLabelValue(JavaUtils.findLocalInetAddress) + lazy val serverAddressLabelValue: String = + toServerAddressLabelValue(JavaUtils.findLocalInetAddress) private[kyuubi] def toServerAddressLabelValue(address: InetAddress): String = address match { case ipv4: Inet4Address => ipv4.getHostAddress diff --git a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/KyuubiApplicationManagerSuite.scala b/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/KyuubiApplicationManagerSuite.scala index 1fb00924688..909b61d8523 100644 --- a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/KyuubiApplicationManagerSuite.scala +++ b/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/KyuubiApplicationManagerSuite.scala @@ -205,8 +205,9 @@ class KyuubiApplicationManagerSuite extends KyuubiFunSuite { Some("k8s://https://kyuubi-test:8443"), conf, Some(shareLevel.toString)) - assert(conf.getOption(watchScopePodLabel).contains(KubernetesUtils.serverAddress)) - assert(conf.getOption(watchScopeServiceLabel).contains(KubernetesUtils.serverAddress)) + assert(conf.getOption(watchScopePodLabel).contains(KubernetesUtils.serverAddressLabelValue)) + assert(conf.getOption(watchScopeServiceLabel) + .contains(KubernetesUtils.serverAddressLabelValue)) } Seq(USER, GROUP, SERVER).foreach { shareLevel => diff --git a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilderSuite.scala b/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilderSuite.scala index 400b4468cff..92f9c2d0aa2 100644 --- a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilderSuite.scala +++ b/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilderSuite.scala @@ -299,7 +299,7 @@ class SparkProcessBuilderSuite extends KerberizedTestHelper with MockitoSugar { assert(globalCommands.contains( s"spark.kubernetes.driver.service.label.$LABEL_KYUUBI_WATCH_SCOPE_KEY=$GLOBAL_WATCH_SCOPE")) - val serverAddress = KubernetesUtils.serverAddress + val serverAddressLabelValue = KubernetesUtils.serverAddressLabelValue val localCommands = new SparkProcessBuilder( "kyuubi", true, @@ -309,9 +309,10 @@ class SparkProcessBuilderSuite extends KerberizedTestHelper with MockitoSugar { .toString.split(' ') assert(localCommands.contains( - s"spark.kubernetes.driver.label.$LABEL_KYUUBI_WATCH_SCOPE_KEY=$serverAddress")) + 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=$serverAddress")) + s"spark.kubernetes.driver.service.label.$LABEL_KYUUBI_WATCH_SCOPE_KEY=" + + serverAddressLabelValue)) } test("SparkProcessBuilder build spark engine with SPARK_USER_NAME") {