This is an automated email from the ASF dual-hosted git repository. dongjoon pushed a commit to branch branch-2.4 in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-2.4 by this push: new 5a50e30 [SPARK-27872][K8S][2.4] Fix executor service account inconsistency 5a50e30 is described below commit 5a50e30881d11971efe91b95ca6a20bb33868d2a Author: nssalian <neelesh.sal...@stitchfix.com> AuthorDate: Sun Oct 11 20:37:41 2020 -0700 [SPARK-27872][K8S][2.4] Fix executor service account inconsistency ### What changes were proposed in this pull request? Similar patch to #24748 but applied to the branch-2.4. Backporting the fix to releases 2.4.x. Closes #29877 from nssalian/SPARK-27872. Authored-by: nssalian <neelesh.sal...@stitchfix.com> Signed-off-by: Dongjoon Hyun <dh...@apple.com> --- .../scala/org/apache/spark/deploy/k8s/Config.scala | 7 ++-- .../apache/spark/deploy/k8s/KubernetesUtils.scala | 13 +++++++ .../DriverKubernetesCredentialsFeatureStep.scala | 11 +----- .../ExecutorKubernetesCredentialsFeatureStep.scala | 43 ++++++++++++++++++++++ .../cluster/k8s/KubernetesExecutorBuilder.scala | 9 ++++- .../k8s/KubernetesExecutorBuilderSuite.scala | 9 ++++- .../k8s/integrationtest/BasicTestsSuite.scala | 7 ++++ .../k8s/integrationtest/KubernetesSuite.scala | 4 ++ 8 files changed, 87 insertions(+), 16 deletions(-) diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala index c7338a7..02471af 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala @@ -61,10 +61,9 @@ private[spark] object Config extends Logging { .stringConf .createOptional - val KUBERNETES_AUTH_DRIVER_CONF_PREFIX = - "spark.kubernetes.authenticate.driver" - val KUBERNETES_AUTH_DRIVER_MOUNTED_CONF_PREFIX = - "spark.kubernetes.authenticate.driver.mounted" + val KUBERNETES_AUTH_DRIVER_CONF_PREFIX = "spark.kubernetes.authenticate.driver" + val KUBERNETES_AUTH_EXECUTOR_CONF_PREFIX = "spark.kubernetes.authenticate.executor" + val KUBERNETES_AUTH_DRIVER_MOUNTED_CONF_PREFIX = "spark.kubernetes.authenticate.driver.mounted" val KUBERNETES_AUTH_CLIENT_MODE_PREFIX = "spark.kubernetes.authenticate" val OAUTH_TOKEN_CONF_SUFFIX = "oauthToken" val OAUTH_TOKEN_FILE_CONF_SUFFIX = "oauthTokenFile" diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/KubernetesUtils.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/KubernetesUtils.scala index 588cd9d..9e32d6e 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/KubernetesUtils.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/KubernetesUtils.scala @@ -16,6 +16,8 @@ */ package org.apache.spark.deploy.k8s +import io.fabric8.kubernetes.api.model.{Container, ContainerBuilder, ContainerStateRunning, ContainerStateTerminated, ContainerStateWaiting, ContainerStatus, Pod, PodBuilder} + import org.apache.spark.SparkConf import org.apache.spark.util.Utils @@ -60,4 +62,15 @@ private[spark] object KubernetesUtils { } def parseMasterUrl(url: String): String = url.substring("k8s://".length) + + def buildPodWithServiceAccount(serviceAccount: Option[String], pod: SparkPod): Option[Pod] = { + serviceAccount.map { account => + new PodBuilder(pod.pod) + .editOrNewSpec() + .withServiceAccount(account) + .withServiceAccountName(account) + .endSpec() + .build() + } + } } diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/DriverKubernetesCredentialsFeatureStep.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/DriverKubernetesCredentialsFeatureStep.scala index ff5ad66..6079381 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/DriverKubernetesCredentialsFeatureStep.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/DriverKubernetesCredentialsFeatureStep.scala @@ -27,6 +27,7 @@ import io.fabric8.kubernetes.api.model.{ContainerBuilder, HasMetadata, PodBuilde import org.apache.spark.deploy.k8s.{KubernetesConf, SparkPod} import org.apache.spark.deploy.k8s.Config._ import org.apache.spark.deploy.k8s.Constants._ +import org.apache.spark.deploy.k8s.KubernetesUtils.buildPodWithServiceAccount private[spark] class DriverKubernetesCredentialsFeatureStep(kubernetesConf: KubernetesConf[_]) extends KubernetesFeatureConfigStep { @@ -70,15 +71,7 @@ private[spark] class DriverKubernetesCredentialsFeatureStep(kubernetesConf: Kube override def configurePod(pod: SparkPod): SparkPod = { if (!shouldMountSecret) { - pod.copy( - pod = driverServiceAccount.map { account => - new PodBuilder(pod.pod) - .editOrNewSpec() - .withServiceAccount(account) - .withServiceAccountName(account) - .endSpec() - .build() - }.getOrElse(pod.pod)) + pod.copy(pod = buildPodWithServiceAccount(driverServiceAccount, pod).getOrElse(pod.pod)) } else { val driverPodWithMountedKubernetesCredentials = new PodBuilder(pod.pod) diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/ExecutorKubernetesCredentialsFeatureStep.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/ExecutorKubernetesCredentialsFeatureStep.scala new file mode 100644 index 0000000..8b9569d --- /dev/null +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/ExecutorKubernetesCredentialsFeatureStep.scala @@ -0,0 +1,43 @@ +/* + * 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.HasMetadata + +import org.apache.spark.deploy.k8s.{KubernetesConf, SparkPod} +import org.apache.spark.deploy.k8s.Config.KUBERNETES_SERVICE_ACCOUNT_NAME +import org.apache.spark.deploy.k8s.KubernetesUtils.buildPodWithServiceAccount + +private[spark] class ExecutorKubernetesCredentialsFeatureStep(kubernetesConf: KubernetesConf[_]) + extends KubernetesFeatureConfigStep { + private lazy val driverServiceAccount = kubernetesConf.get(KUBERNETES_SERVICE_ACCOUNT_NAME) + + override def configurePod(pod: SparkPod): SparkPod = { + pod.copy( + // if not setup by the pod template fallback to the driver's sa, + // last option is the default sa. + pod = if (Option(pod.pod.getSpec.getServiceAccount).isEmpty) { + buildPodWithServiceAccount(driverServiceAccount, pod).getOrElse(pod.pod) + } else { + pod.pod + }) + } + + override def getAdditionalPodSystemProperties(): Map[String, String] = Map.empty + + override def getAdditionalKubernetesResources(): Seq[HasMetadata] = Seq.empty +} 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 364b6fb..34c340a 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 @@ -18,12 +18,15 @@ package org.apache.spark.scheduler.cluster.k8s import org.apache.spark.deploy.k8s.{KubernetesConf, KubernetesExecutorSpecificConf, KubernetesRoleSpecificConf, SparkPod} import org.apache.spark.deploy.k8s.features._ -import org.apache.spark.deploy.k8s.features.{BasicExecutorFeatureStep, EnvSecretsFeatureStep, LocalDirsFeatureStep, MountSecretsFeatureStep} +import org.apache.spark.deploy.k8s.features.{BasicExecutorFeatureStep, EnvSecretsFeatureStep, ExecutorKubernetesCredentialsFeatureStep, LocalDirsFeatureStep, MountSecretsFeatureStep} private[spark] class KubernetesExecutorBuilder( provideBasicStep: (KubernetesConf [KubernetesExecutorSpecificConf]) => BasicExecutorFeatureStep = new BasicExecutorFeatureStep(_), + provideCredentialsStep: (KubernetesConf [KubernetesExecutorSpecificConf]) + => ExecutorKubernetesCredentialsFeatureStep = + new ExecutorKubernetesCredentialsFeatureStep(_), provideSecretsStep: (KubernetesConf[_ <: KubernetesRoleSpecificConf]) => MountSecretsFeatureStep = new MountSecretsFeatureStep(_), @@ -50,8 +53,10 @@ private[spark] class KubernetesExecutorBuilder( val volumesFeature = if (kubernetesConf.roleVolumes.nonEmpty) { Seq(provideVolumesStep(kubernetesConf)) } else Nil + val credentialsFeature = Seq(provideCredentialsStep(kubernetesConf)) - val allFeatures = baseFeatures ++ secretFeature ++ secretEnvFeature ++ volumesFeature + val allFeatures = + baseFeatures ++ secretFeature ++ secretEnvFeature ++ volumesFeature ++ credentialsFeature var executorPod = SparkPod.initialPod() for (feature <- allFeatures) { diff --git a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBuilderSuite.scala b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBuilderSuite.scala index 44fe4a2..97184d4 100644 --- a/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBuilderSuite.scala +++ b/resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBuilderSuite.scala @@ -24,6 +24,7 @@ import org.apache.spark.deploy.k8s.features._ class KubernetesExecutorBuilderSuite extends SparkFunSuite { private val BASIC_STEP_TYPE = "basic" + private val CREDENTIALS_STEP_TYPE = "creds" private val SECRETS_STEP_TYPE = "mount-secrets" private val ENV_SECRETS_STEP_TYPE = "env-secrets" private val LOCAL_DIRS_STEP_TYPE = "local-dirs" @@ -31,6 +32,8 @@ class KubernetesExecutorBuilderSuite extends SparkFunSuite { private val basicFeatureStep = KubernetesFeaturesTestUtils.getMockConfigStepForStepType( BASIC_STEP_TYPE, classOf[BasicExecutorFeatureStep]) + private val credentialsStep = KubernetesFeaturesTestUtils.getMockConfigStepForStepType( + CREDENTIALS_STEP_TYPE, classOf[ExecutorKubernetesCredentialsFeatureStep]) private val mountSecretsStep = KubernetesFeaturesTestUtils.getMockConfigStepForStepType( SECRETS_STEP_TYPE, classOf[MountSecretsFeatureStep]) private val envSecretsStep = KubernetesFeaturesTestUtils.getMockConfigStepForStepType( @@ -42,6 +45,7 @@ class KubernetesExecutorBuilderSuite extends SparkFunSuite { private val builderUnderTest = new KubernetesExecutorBuilder( _ => basicFeatureStep, + _ => credentialsStep, _ => mountSecretsStep, _ => envSecretsStep, _ => localDirsStep, @@ -62,7 +66,8 @@ class KubernetesExecutorBuilderSuite extends SparkFunSuite { Nil, Seq.empty[String]) validateStepTypesApplied( - builderUnderTest.buildFromFeatures(conf), BASIC_STEP_TYPE, LOCAL_DIRS_STEP_TYPE) + builderUnderTest.buildFromFeatures(conf), BASIC_STEP_TYPE, + CREDENTIALS_STEP_TYPE, LOCAL_DIRS_STEP_TYPE) } test("Apply secrets step if secrets are present.") { @@ -82,6 +87,7 @@ class KubernetesExecutorBuilderSuite extends SparkFunSuite { validateStepTypesApplied( builderUnderTest.buildFromFeatures(conf), BASIC_STEP_TYPE, + CREDENTIALS_STEP_TYPE, LOCAL_DIRS_STEP_TYPE, SECRETS_STEP_TYPE, ENV_SECRETS_STEP_TYPE) @@ -109,6 +115,7 @@ class KubernetesExecutorBuilderSuite extends SparkFunSuite { validateStepTypesApplied( builderUnderTest.buildFromFeatures(conf), BASIC_STEP_TYPE, + CREDENTIALS_STEP_TYPE, LOCAL_DIRS_STEP_TYPE, MOUNT_VOLUMES_STEP_TYPE) } diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/BasicTestsSuite.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/BasicTestsSuite.scala index 1e9f830..76221e4 100644 --- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/BasicTestsSuite.scala +++ b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/BasicTestsSuite.scala @@ -84,6 +84,13 @@ private[spark] trait BasicTestsSuite { k8sSuite: KubernetesSuite => }) } + test("All pods have the same service account by default", k8sTestTag) { + runSparkPiAndVerifyCompletion( + executorPodChecker = (executorPod: Pod) => { + doExecutorServiceAccountCheck(executorPod, kubernetesTestComponents.serviceAccountName) + }) + } + test("Run extraJVMOptions check on driver", k8sTestTag) { sparkAppConf .set("spark.driver.extraJavaOptions", "-Dspark.test.foo=spark.test.bar") diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/KubernetesSuite.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/KubernetesSuite.scala index d893433..1036589 100644 --- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/KubernetesSuite.scala +++ b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/k8s/integrationtest/KubernetesSuite.scala @@ -264,6 +264,10 @@ private[spark] class KubernetesSuite extends SparkFunSuite === baseMemory) } + protected def doExecutorServiceAccountCheck(executorPod: Pod, account: String): Unit = { + doBasicExecutorPodCheck(executorPod) + assert(executorPod.getSpec.getServiceAccount == kubernetesTestComponents.serviceAccountName) + } protected def doBasicDriverPyPodCheck(driverPod: Pod): Unit = { assert(driverPod.getMetadata.getName === driverPodName) --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org