diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala index 4ffd617c4e4ce..9c79ef77b8c28 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Constants.scala @@ -121,6 +121,8 @@ object Constants { "spark.kubernetes.executor.hadoopConfigMapName" // Kerberos Configuration + val KRB_CONFIG_MAP_NAME = + "spark.kubernetes.executor.krbConfigMapName" val KERBEROS_DT_SECRET_NAME = "spark.kubernetes.kerberos.dt-secret-name" val KERBEROS_DT_SECRET_KEY = diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/KerberosConfDriverFeatureStep.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/KerberosConfDriverFeatureStep.scala index 2da8dcd6261d9..2d257632a3475 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/KerberosConfDriverFeatureStep.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/KerberosConfDriverFeatureStep.scala @@ -215,14 +215,17 @@ private[spark] class KerberosConfDriverFeatureStep(kubernetesConf: KubernetesDri } override def getAdditionalPodSystemProperties(): Map[String, String] = { + val props = scala.collection.mutable.Map.empty[String, String] // If a submission-local keytab is provided, update the Spark config so that it knows the // path of the keytab in the driver container. if (needKeytabUpload) { val ktName = new File(keytab.get).getName() - Map(KEYTAB.key -> s"$KERBEROS_KEYTAB_MOUNT_POINT/$ktName") - } else { - Map.empty + props += (KEYTAB.key -> s"$KERBEROS_KEYTAB_MOUNT_POINT/$ktName") + } + if (hasKerberosConf) { + props += (KRB_CONFIG_MAP_NAME -> krb5CMap.getOrElse(newConfigMapName)) } + props.toMap } override def getAdditionalKubernetesResources(): Seq[HasMetadata] = { diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/KerberosConfExecutorFeatureStep.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/KerberosConfExecutorFeatureStep.scala new file mode 100644 index 0000000000000..28f0f17f219a7 --- /dev/null +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/KerberosConfExecutorFeatureStep.scala @@ -0,0 +1,59 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.spark.deploy.k8s.features + +import io.fabric8.kubernetes.api.model.{ContainerBuilder, PodBuilder, VolumeBuilder} + +import org.apache.spark.deploy.k8s.{KubernetesConf, SparkPod} +import org.apache.spark.deploy.k8s.Constants._ + +/** + * Mounts the krb5.conf ConfigMap on the executor pod. + */ +private[spark] class KerberosConfExecutorFeatureStep(conf: KubernetesConf) + extends KubernetesFeatureConfigStep { + + private val krbConfigMapName = conf.getOption(KRB_CONFIG_MAP_NAME) + + override def configurePod(original: SparkPod): SparkPod = { + original.transform { case pod if krbConfigMapName.isDefined => + val configMapVolume = new VolumeBuilder() + .withName(KRB_FILE_VOLUME) + .withNewConfigMap() + .withName(krbConfigMapName.get) + .endConfigMap() + .build() + + val podWithVolume = new PodBuilder(pod.pod) + .editSpec() + .addNewVolumeLike(configMapVolume) + .endVolume() + .endSpec() + .build() + + val containerWithMount = new ContainerBuilder(pod.container) + .addNewVolumeMount() + .withName(KRB_FILE_VOLUME) + .withMountPath(KRB_FILE_DIR_PATH + "/krb5.conf") + .withSubPath("krb5.conf") + .endVolumeMount() + .build() + + SparkPod(podWithVolume, containerWithMount) + } + } +} diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackend.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackend.scala index 27a0b320cbf9a..d0aa61ccc0c09 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackend.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackend.scala @@ -16,6 +16,8 @@ */ package org.apache.spark.scheduler.cluster.k8s +import java.io.File +import java.nio.file.Files import java.util.concurrent.{ScheduledExecutorService, TimeUnit} import java.util.concurrent.atomic.AtomicInteger @@ -98,6 +100,35 @@ private[spark] class KubernetesClusterSchedulerBackend( kubernetesClient.configMaps().inNamespace(namespace).resource(configMap).create() } + /** + * Publishes the krb5 ConfigMap name into [[KRB_CONFIG_MAP_NAME]] so executor pods can mount it. + * Cluster mode already has it set by the driver step; client mode creates or reuses it here. + */ + private[k8s] def setUpExecutorKrb5ConfigMap(driverPod: Option[Pod]): Unit = { + // Already set by the driver feature step (cluster mode), nothing to do. + if (conf.getOption(KRB_CONFIG_MAP_NAME).isEmpty) { + conf.get(KUBERNETES_KERBEROS_KRB5_CONFIG_MAP) match { + case Some(existingMap) => + // User supplied an existing ConfigMap, just publish its name to executors. + conf.set(KRB_CONFIG_MAP_NAME, existingMap) + case None => + conf.get(KUBERNETES_KERBEROS_KRB5_FILE).foreach { localPath => + val file = new File(localPath) + val configMapName = KubernetesClientUtils + .configMapName(s"spark-krb5-${KubernetesUtils.uniqueID()}") + val labels = + Map(SPARK_APP_ID_LABEL -> applicationId(), + SPARK_ROLE_LABEL -> SPARK_POD_EXECUTOR_ROLE) + val configMap = KubernetesClientUtils.buildConfigMap( + configMapName, Map(file.getName -> Files.readString(file.toPath)), labels) + KubernetesUtils.addOwnerReference(driverPod.orNull, Seq(configMap)) + kubernetesClient.configMaps().inNamespace(namespace).resource(configMap).create() + conf.set(KRB_CONFIG_MAP_NAME, configMapName) + } + } + } + } + /** * Get an application ID associated with the job. * This returns the string value of spark.app.id if set, otherwise @@ -115,6 +146,7 @@ private[spark] class KubernetesClusterSchedulerBackend( // allocation is asynchronous (background thread pool), so requesting executors first // can race with ConfigMap creation, causing transient "configmap ... not found" mounts. if (!conf.get(KUBERNETES_EXECUTOR_DISABLE_CONFIGMAP)) { + setUpExecutorKrb5ConfigMap(podAllocator.driverPod) setUpExecutorConfigMap(podAllocator.driverPod) } val defaultProfile = scheduler.sc.resourceProfileManager.defaultResourceProfile diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBuilder.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBuilder.scala index 2253c07df116e..b8024e2ba6a63 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBuilder.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBuilder.scala @@ -72,6 +72,7 @@ private[spark] class KubernetesExecutorBuilder { new EnvSecretsFeatureStep(conf), new MountVolumesFeatureStep(conf), new HadoopConfExecutorFeatureStep(conf), + new KerberosConfExecutorFeatureStep(conf), new LocalDirsFeatureStep(conf)) ++ userFeatures val features = allFeatures.filterNot(f => diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/KerberosConfDriverFeatureStepSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/KerberosConfDriverFeatureStepSuite.scala index 0da39b30e3883..30eae1d0c75b6 100644 --- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/KerberosConfDriverFeatureStepSuite.scala +++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/KerberosConfDriverFeatureStepSuite.scala @@ -48,7 +48,7 @@ class KerberosConfDriverFeatureStepSuite extends SparkFunSuite { new SparkConf(false).set(KUBERNETES_KERBEROS_KRB5_CONFIG_MAP, configMap)) checkPodForKrbConf(step.configurePod(SparkPod.initialPod()), configMap) - assert(step.getAdditionalPodSystemProperties().isEmpty) + assert(step.getAdditionalPodSystemProperties() === Map(KRB_CONFIG_MAP_NAME -> configMap)) assert(filter[ConfigMap](step.getAdditionalKubernetesResources()).isEmpty) } @@ -64,7 +64,8 @@ class KerberosConfDriverFeatureStepSuite extends SparkFunSuite { assert(confMap.getData().keySet().asScala === Set(krbConf.getName())) checkPodForKrbConf(step.configurePod(SparkPod.initialPod()), confMap.getMetadata().getName()) - assert(step.getAdditionalPodSystemProperties().isEmpty) + assert(step.getAdditionalPodSystemProperties() === + Map(KRB_CONFIG_MAP_NAME -> confMap.getMetadata().getName())) } test("create keytab secret if client keytab file used") { diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/KerberosConfExecutorFeatureStepSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/KerberosConfExecutorFeatureStepSuite.scala new file mode 100644 index 0000000000000..b780225890e49 --- /dev/null +++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/KerberosConfExecutorFeatureStepSuite.scala @@ -0,0 +1,89 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.deploy.k8s.features + +import java.io.File +import java.nio.file.Files + +import org.apache.spark.{SparkConf, SparkFunSuite} +import org.apache.spark.deploy.k8s._ +import org.apache.spark.deploy.k8s.Config._ +import org.apache.spark.deploy.k8s.Constants._ +import org.apache.spark.util.Utils + +class KerberosConfExecutorFeatureStepSuite extends SparkFunSuite { + import SecretVolumeUtils._ + + test("SPARK-50758: mount krb5 ConfigMap when KRB_CONFIG_MAP_NAME is set") { + val executorSparkConf = new SparkConf(false).set(KRB_CONFIG_MAP_NAME, "testCM") + val executorConf = KubernetesTestConf.createExecutorConf(sparkConf = executorSparkConf) + val initial = SparkPod.initialPod() + val executorPod = new KerberosConfExecutorFeatureStep(executorConf).configurePod(initial) + checkPod(executorPod, hasKrb5 = true) + } + + test("SPARK-50758: no-op when KRB_CONFIG_MAP_NAME is not set") { + val executorConf = KubernetesTestConf.createExecutorConf(sparkConf = new SparkConf(false)) + val initial = SparkPod.initialPod() + val executorPod = new KerberosConfExecutorFeatureStep(executorConf).configurePod(initial) + checkPod(executorPod, hasKrb5 = false) + } + + test("SPARK-50758: mount krb5 ConfigMap when driver step publishes its name") { + val tmpDir = Utils.createTempDir() + val krbConf = File.createTempFile("krb5", ".conf", tmpDir) + Files.writeString(krbConf.toPath, "some data") + + Seq( + // (sparkConf, expectMount) + (new SparkConf(false).set(KUBERNETES_KERBEROS_KRB5_CONFIG_MAP, "userCM"), true), + (new SparkConf(false).set(KUBERNETES_KERBEROS_KRB5_FILE, krbConf.getAbsolutePath), true), + (new SparkConf(false), false) + ).foreach { case (driverSparkConf, expectMount) => + + val driverConf = KubernetesTestConf.createDriverConf(sparkConf = driverSparkConf) + val driverStep = new KerberosConfDriverFeatureStep(driverConf) + + val executorSparkConf = new SparkConf(false) + val additionalProps = driverStep.getAdditionalPodSystemProperties() + if (expectMount) { + assert(additionalProps.contains(KRB_CONFIG_MAP_NAME), + s"Driver step must publish $KRB_CONFIG_MAP_NAME when krb5 conf is provided") + additionalProps.foreach { case (k, v) => executorSparkConf.set(k, v) } + } else { + assert(!additionalProps.contains(KRB_CONFIG_MAP_NAME)) + } + + val executorConf = KubernetesTestConf.createExecutorConf(sparkConf = executorSparkConf) + val initial = SparkPod.initialPod() + val executorPod = new KerberosConfExecutorFeatureStep(executorConf).configurePod(initial) + checkPod(executorPod, expectMount) + } + } + + private def checkPod(pod: SparkPod, hasKrb5: Boolean): Unit = { + val mountPath = KRB_FILE_DIR_PATH + "/krb5.conf" + if (hasKrb5) { + assert(podHasVolume(pod.pod, KRB_FILE_VOLUME)) + assert(containerHasVolume(pod.container, KRB_FILE_VOLUME, mountPath)) + } else { + assert(!podHasVolume(pod.pod, KRB_FILE_VOLUME)) + assert(!containerHasVolume(pod.container, KRB_FILE_VOLUME, mountPath)) + } + } +} diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala index 64734cc6c612a..168a024f2eabc 100644 --- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala +++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackendSuite.scala @@ -16,6 +16,8 @@ */ package org.apache.spark.scheduler.cluster.k8s +import java.io.File +import java.nio.file.Files import java.util.Arrays import java.util.concurrent.TimeUnit @@ -43,6 +45,7 @@ import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages.{Decommis import org.apache.spark.scheduler.cluster.CoarseGrainedSchedulerBackend import org.apache.spark.scheduler.cluster.k8s.ExecutorLifecycleTestUtils.TEST_SPARK_APP_ID import org.apache.spark.storage.{BlockManager, BlockManagerMaster} +import org.apache.spark.util.Utils class KubernetesClusterSchedulerBackendSuite extends SparkFunSuite with BeforeAndAfter { @@ -510,4 +513,55 @@ class KubernetesClusterSchedulerBackendSuite extends SparkFunSuite with BeforeAn } assert(schedulerBackendUnderTest.supportsExecutorHold) } + + test("SPARK-50758: setUpExecutorKrb5ConfigMap publishes or creates the krb5 ConfigMap") { + // 1. No krb5 config at all: no-op. + schedulerBackendUnderTest.setUpExecutorKrb5ConfigMap(None) + assert(sparkConf.getOption(KRB_CONFIG_MAP_NAME).isEmpty) + verify(configMapsWithNamespace, never()).resource(any[ConfigMap]()) + + // 2. Cluster mode: the driver step already set KRB_CONFIG_MAP_NAME. + sparkConf.set(KRB_CONFIG_MAP_NAME, "alreadySetByDriverStep") + sparkConf.set(KUBERNETES_KERBEROS_KRB5_FILE, "/does/not/matter") + try { + schedulerBackendUnderTest.setUpExecutorKrb5ConfigMap(None) + assert(sparkConf.get(KRB_CONFIG_MAP_NAME) === "alreadySetByDriverStep") + verify(configMapsWithNamespace, never()).resource(any[ConfigMap]()) + } finally { + sparkConf.remove(KRB_CONFIG_MAP_NAME) + sparkConf.remove(KUBERNETES_KERBEROS_KRB5_FILE.key) + } + + // 3. User-provided ConfigMap name is just published, nothing is created. + sparkConf.set(KUBERNETES_KERBEROS_KRB5_CONFIG_MAP, "userKrbCM") + try { + schedulerBackendUnderTest.setUpExecutorKrb5ConfigMap(None) + assert(sparkConf.get(KRB_CONFIG_MAP_NAME) === "userKrbCM") + verify(configMapsWithNamespace, never()).resource(any[ConfigMap]()) + } finally { + sparkConf.remove(KUBERNETES_KERBEROS_KRB5_CONFIG_MAP.key) + sparkConf.remove(KRB_CONFIG_MAP_NAME) + } + + // 4. Client mode: a local krb5.conf file triggers ConfigMap creation. + val tmpDir = Utils.createTempDir() + val krb5 = File.createTempFile("krb5", ".conf", tmpDir) + Files.writeString(krb5.toPath, "some krb5 data") + sparkConf.set(KUBERNETES_KERBEROS_KRB5_FILE, krb5.getAbsolutePath) + try { + schedulerBackendUnderTest.setUpExecutorKrb5ConfigMap(None) + val captor = ArgumentCaptor.forClass(classOf[ConfigMap]) + verify(configMapsWithNamespace).resource(captor.capture()) + verify(configMapResource).create() + val created = captor.getValue + val labels = created.getMetadata.getLabels.asScala + assert(labels(SPARK_APP_ID_LABEL) === TEST_SPARK_APP_ID) + assert(labels(SPARK_ROLE_LABEL) === SPARK_POD_EXECUTOR_ROLE) + assert(created.getData.keySet().asScala === Set(krb5.getName)) + assert(sparkConf.get(KRB_CONFIG_MAP_NAME) === created.getMetadata.getName) + } finally { + sparkConf.remove(KUBERNETES_KERBEROS_KRB5_FILE.key) + sparkConf.remove(KRB_CONFIG_MAP_NAME) + } + } }