[SPARK-23257][K8S] Kerberos Support for Spark on K8S #21669
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -65,11 +65,13 @@ private[spark] object Constants { | |
| val ENV_CLASSPATH = "SPARK_CLASSPATH" | ||
| val ENV_DRIVER_BIND_ADDRESS = "SPARK_DRIVER_BIND_ADDRESS" | ||
| val ENV_SPARK_CONF_DIR = "SPARK_CONF_DIR" | ||
| val ENV_SPARK_USER = "SPARK_USER" | ||
|
Contributor
There was a problem hiding this comment. I guess this is for setting the correct user. But I think hadoop libs should pick the correct user like in SparkContext where |
||
| // Spark app configs for containers | ||
| val SPARK_CONF_VOLUME = "spark-conf-volume" | ||
| val SPARK_CONF_DIR_INTERNAL = "/opt/spark/conf" | ||
| val SPARK_CONF_FILE_NAME = "spark.properties" | ||
| val SPARK_CONF_PATH = s"$SPARK_CONF_DIR_INTERNAL/$SPARK_CONF_FILE_NAME" | ||
| val ENV_HADOOP_TOKEN_FILE_LOCATION = "HADOOP_TOKEN_FILE_LOCATION" | ||
|
|
||
| // BINDINGS | ||
| val ENV_PYSPARK_PRIMARY = "PYSPARK_PRIMARY" | ||
|
|
@@ -81,4 +83,35 @@ private[spark] object Constants { | |
| val KUBERNETES_MASTER_INTERNAL_URL = "https://kubernetes.default.svc" | ||
| val DRIVER_CONTAINER_NAME = "spark-kubernetes-driver" | ||
| val MEMORY_OVERHEAD_MIN_MIB = 384L | ||
|
|
||
| // Hadoop Configuration | ||
| val HADOOP_FILE_VOLUME = "hadoop-properties" | ||
| val HADOOP_CONF_DIR_PATH = "/etc/hadoop/conf" | ||
| val ENV_HADOOP_CONF_DIR = "HADOOP_CONF_DIR" | ||
| val HADOOP_CONF_DIR_LOC = "spark.kubernetes.hadoop.conf.dir" | ||
| val HADOOP_CONFIG_MAP_SPARK_CONF_NAME = | ||
|
||
| "spark.kubernetes.hadoop.executor.hadoopConfigMapName" | ||
|
|
||
| // Kerberos Configuration | ||
| val KERBEROS_DELEGEGATION_TOKEN_SECRET_NAME = | ||
| "spark.kubernetes.kerberos.delegation-token-secret-name" | ||
| val KERBEROS_KEYTAB_SECRET_NAME = | ||
|
||
| "spark.kubernetes.kerberos.key-tab-secret-name" | ||
| val KERBEROS_KEYTAB_SECRET_KEY = | ||
| "spark.kubernetes.kerberos.key-tab-secret-key" | ||
| val KERBEROS_SPARK_USER_NAME = | ||
| "spark.kubernetes.kerberos.spark-user-name" | ||
| val KERBEROS_SECRET_LABEL_PREFIX = | ||
| "hadoop-tokens" | ||
| val SPARK_HADOOP_PREFIX = "spark.hadoop." | ||
| val HADOOP_SECURITY_AUTHENTICATION = | ||
| SPARK_HADOOP_PREFIX + "hadoop.security.authentication" | ||
|
|
||
| // Kerberos Token-Refresh Server | ||
| val KERBEROS_REFRESH_LABEL_KEY = "refresh-hadoop-tokens" | ||
|
||
| val KERBEROS_REFRESH_LABEL_VALUE = "yes" | ||
|
|
||
| // Hadoop credentials secrets for the Spark app. | ||
| val SPARK_APP_HADOOP_CREDENTIALS_BASE_DIR = "/mnt/secrets/hadoop-credentials" | ||
| val SPARK_APP_HADOOP_SECRET_VOLUME_NAME = "hadoop-secret" | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,52 @@ | ||
| /* | ||
| * 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 | ||
|
Contributor
There was a problem hiding this comment. Why this is not under the |
||
|
|
||
| import io.fabric8.kubernetes.api.model.HasMetadata | ||
|
|
||
| import org.apache.spark.deploy.k8s.{KubernetesConf, KubernetesRoleSpecificConf, SparkPod} | ||
| import org.apache.spark.deploy.k8s.Constants._ | ||
| import org.apache.spark.deploy.k8s.features.hadoopsteps.HadoopBootstrapUtil | ||
| import org.apache.spark.internal.Logging | ||
|
|
||
| /** | ||
| * This step is responsible for bootstraping the container with ConfigMaps | ||
| * containing Hadoop config files mounted as volumes and an ENV variable | ||
| * pointed to the mounted file directory. This is run by both the driver | ||
| * and executor, as they both require Hadoop config files. | ||
| */ | ||
| private[spark] class HadoopConfExecutorFeatureStep( | ||
| kubernetesConf: KubernetesConf[_ <: KubernetesRoleSpecificConf]) | ||
| extends KubernetesFeatureConfigStep with Logging{ | ||
|
|
||
| override def configurePod(pod: SparkPod): SparkPod = { | ||
| val maybeHadoopConfDir = kubernetesConf.sparkConf.getOption(HADOOP_CONF_DIR_LOC) | ||
| val maybeHadoopConfigMap = kubernetesConf.sparkConf.getOption(HADOOP_CONFIG_MAP_SPARK_CONF_NAME) | ||
| require(maybeHadoopConfDir.isDefined && maybeHadoopConfigMap.isDefined, | ||
| "Ensure that HADOOP_CONF_DIR is defined") | ||
| logInfo("HADOOP_CONF_DIR defined. Mounting Hadoop specific files") | ||
| HadoopBootstrapUtil.bootstrapHadoopConfDir( | ||
| maybeHadoopConfDir.get, | ||
| maybeHadoopConfigMap.get, | ||
| kubernetesConf.getTokenManager, | ||
| pod) | ||
| } | ||
|
|
||
| override def getAdditionalPodSystemProperties(): Map[String, String] = Map.empty | ||
|
Contributor
There was a problem hiding this comment. No need to address it here but it feels like these methods should have default implementations, given that lots of classes just don't do anything with them. |
||
|
|
||
| override def getAdditionalKubernetesResources(): Seq[HasMetadata] = Seq.empty | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,123 @@ | ||
| /* | ||
| * 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 scala.collection.JavaConverters._ | ||
|
|
||
| import io.fabric8.kubernetes.api.model.{ConfigMapBuilder, ContainerBuilder, HasMetadata, PodBuilder} | ||
|
|
||
| import org.apache.spark.deploy.k8s.{KubernetesConf, SparkPod} | ||
| import org.apache.spark.deploy.k8s.Constants._ | ||
| import org.apache.spark.deploy.k8s.KubernetesDriverSpecificConf | ||
| import org.apache.spark.deploy.k8s.features.hadoopsteps.{HadoopBootstrapUtil, HadoopConfigSpec, HadoopConfigurationStep} | ||
| import org.apache.spark.internal.Logging | ||
|
|
||
| /** | ||
| * This is the main method that runs the hadoopConfigurationSteps defined | ||
|
||
| * by the HadoopStepsOrchestrator. These steps are run to modify the | ||
| * SparkPod and Kubernetes Resources using the additive method of the feature steps | ||
| */ | ||
| private[spark] class HadoopGlobalFeatureDriverStep( | ||
|
||
| kubernetesConf: KubernetesConf[KubernetesDriverSpecificConf]) | ||
|
||
| extends KubernetesFeatureConfigStep with Logging { | ||
| private val hadoopTestOrchestrator = | ||
| kubernetesConf.getHadoopStepsOrchestrator | ||
|
||
| require(kubernetesConf.hadoopConfDir.isDefined && | ||
| hadoopTestOrchestrator.isDefined, "Ensure that HADOOP_CONF_DIR is defined") | ||
|
||
| private val hadoopSteps = | ||
| hadoopTestOrchestrator | ||
| .map(hto => hto.getHadoopSteps(kubernetesConf.getTokenManager)) | ||
| .getOrElse(Seq.empty[HadoopConfigurationStep]) | ||
|
|
||
| var currentHadoopSpec = HadoopConfigSpec( | ||
| podVolumes = Seq.empty, | ||
| containerEnvs = Seq.empty, | ||
| containerVMs = Seq.empty, | ||
| configMapProperties = Map.empty[String, String], | ||
| dtSecret = None, | ||
| dtSecretName = KERBEROS_DELEGEGATION_TOKEN_SECRET_NAME, | ||
| dtSecretItemKey = None, | ||
| jobUserName = None) | ||
|
|
||
| for (nextStep <- hadoopSteps) { | ||
| currentHadoopSpec = nextStep.configureHadoopSpec(currentHadoopSpec) | ||
| } | ||
|
|
||
| override def configurePod(pod: SparkPod): SparkPod = { | ||
| val hadoopBasedPod = new PodBuilder(pod.pod) | ||
| .editSpec() | ||
| .addAllToVolumes(currentHadoopSpec.podVolumes.asJava) | ||
| .endSpec() | ||
| .build() | ||
|
|
||
| val hadoopBasedContainer = new ContainerBuilder(pod.container) | ||
| .addAllToEnv(currentHadoopSpec.containerEnvs.asJava) | ||
| .addAllToVolumeMounts(currentHadoopSpec.containerVMs.asJava) | ||
| .build() | ||
|
|
||
| val hadoopBasedSparkPod = HadoopBootstrapUtil.bootstrapHadoopConfDir( | ||
| kubernetesConf.hadoopConfDir.get, | ||
| kubernetesConf.getHadoopConfigMapName, | ||
| kubernetesConf.getTokenManager, | ||
| SparkPod(hadoopBasedPod, hadoopBasedContainer)) | ||
|
|
||
| val maybeKerberosModification = | ||
| for { | ||
| secretItemKey <- currentHadoopSpec.dtSecretItemKey | ||
| userName <- currentHadoopSpec.jobUserName | ||
| } yield { | ||
| HadoopBootstrapUtil.bootstrapKerberosPod( | ||
| currentHadoopSpec.dtSecretName, | ||
| secretItemKey, | ||
| userName, | ||
| hadoopBasedSparkPod) | ||
| } | ||
| maybeKerberosModification.getOrElse( | ||
| HadoopBootstrapUtil.bootstrapSparkUserPod( | ||
| kubernetesConf.getTokenManager.getCurrentUser.getShortUserName, | ||
| hadoopBasedSparkPod)) | ||
| } | ||
|
|
||
| override def getAdditionalPodSystemProperties(): Map[String, String] = { | ||
| val maybeKerberosConfValues = | ||
| for { | ||
| secretItemKey <- currentHadoopSpec.dtSecretItemKey | ||
| userName <- currentHadoopSpec.jobUserName | ||
| } yield { | ||
| Map(KERBEROS_KEYTAB_SECRET_NAME -> currentHadoopSpec.dtSecretName, | ||
| KERBEROS_KEYTAB_SECRET_KEY -> secretItemKey, | ||
| KERBEROS_SPARK_USER_NAME -> userName) | ||
| } | ||
|
||
| val resolvedConfValues = maybeKerberosConfValues.getOrElse( | ||
| Map(KERBEROS_SPARK_USER_NAME -> | ||
| kubernetesConf.getTokenManager.getCurrentUser.getShortUserName) | ||
| ) | ||
| Map(HADOOP_CONFIG_MAP_SPARK_CONF_NAME -> kubernetesConf.getHadoopConfigMapName, | ||
| HADOOP_CONF_DIR_LOC -> kubernetesConf.hadoopConfDir.get) ++ resolvedConfValues | ||
| } | ||
|
|
||
| override def getAdditionalKubernetesResources(): Seq[HasMetadata] = { | ||
| val configMap = | ||
| new ConfigMapBuilder() | ||
| .withNewMetadata() | ||
| .withName(kubernetesConf.getHadoopConfigMapName) | ||
| .endMetadata() | ||
| .addToData(currentHadoopSpec.configMapProperties.asJava) | ||
| .build() | ||
| Seq(configMap) ++ currentHadoopSpec.dtSecret.toSeq | ||
| } | ||
| } | ||
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This check has been restrictive for customers in the past. There are cases where spark submit should not have the file locally and keytab should be mounted as a secret within the cluster for example.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This check can be removed, but I included it since I believed that the keytab shouldn't be stored as a secret for security reasons and should instead be only accessible from the JVM.