[SPARK-58910][K8S] Keep the executor pod template's service account

docs/running-on-kubernetes.md

@@ -920,7 +920,10 @@ See the [configuration page](configuration.html) for information on Spark config
   <td><code>(value of spark.kubernetes.authenticate.driver.serviceAccountName)</code></td>
   <td>
     Service account that is used when running the executor pod.
-    If this parameter is not setup, the fallback logic will use the driver's service account.
+    If this parameter is not setup, the fallback logic will use the value of
+    <code>spark.kubernetes.authenticate.driver.serviceAccountName</code>.
+    Both are ignored when the executor pod template already names a non-empty service account in
+    either <code>serviceAccount</code> or <code>serviceAccountName</code>.
   </td>
   <td>3.1.0</td>
 </tr>
@@ -1975,20 +1978,36 @@ See the below table for the full list of pod specifications that will be overwri
 </tr>
 <tr>
   <td>serviceAccount</td>
-  <td>Value of <code>spark.kubernetes.authenticate.driver.serviceAccountName</code></td>
+  <td>Value of <code>spark.kubernetes.authenticate.driver.serviceAccountName</code>; for executor
+  pods, of <code>spark.kubernetes.authenticate.executor.serviceAccountName</code>, falling back to
+  the driver's</td>
   <td>
-    Spark will override <code>serviceAccount</code> with the value of the spark configuration for only
-    driver pods, and only if the spark configuration is specified and no driver credentials are
-    submitted for Spark to mount as a secret. Executor pods will remain unaffected.
+    For driver pods Spark will override <code>serviceAccount</code> with the value of
+    <code>spark.kubernetes.authenticate.driver.serviceAccountName</code>, but only if that
+    configuration is set and no driver credentials are submitted for Spark to mount as a secret.
+    For executor pods Spark writes both fields only when the template leaves both of them empty,
+    and writes the same value into each; see
+    <code>spark.kubernetes.authenticate.executor.serviceAccountName</code> for which account that
+    is. When an executor pod template names an account in either field, Spark overwrites
+    neither, and warns if <code>spark.kubernetes.authenticate.executor.serviceAccountName</code>
+    named a different one.
   </td>
 </tr>
 <tr>
   <td>serviceAccountName</td>
-  <td>Value of <code>spark.kubernetes.authenticate.driver.serviceAccountName</code></td>
-  <td>
-    Spark will override <code>serviceAccountName</code> with the value of the spark configuration for only
-    driver pods, and only if the spark configuration is specified and no driver credentials are
-    submitted for Spark to mount as a secret. Executor pods will remain unaffected.
+  <td>Value of <code>spark.kubernetes.authenticate.driver.serviceAccountName</code>; for executor
+  pods, of <code>spark.kubernetes.authenticate.executor.serviceAccountName</code>, falling back to
+  the driver's</td>
+  <td>
+    For driver pods Spark will override <code>serviceAccountName</code> with the value of
+    <code>spark.kubernetes.authenticate.driver.serviceAccountName</code>, but only if that
+    configuration is set and no driver credentials are submitted for Spark to mount as a secret.
+    For executor pods Spark writes both fields only when the template leaves both of them empty,
+    and writes the same value into each; see
+    <code>spark.kubernetes.authenticate.executor.serviceAccountName</code> for which account that
+    is. When an executor pod template names an account in either field, Spark overwrites
+    neither, and warns if <code>spark.kubernetes.authenticate.executor.serviceAccountName</code>
+    named a different one.
   </td>
 </tr>
 <tr>

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala

@@ -383,8 +383,10 @@ private[spark] object Config extends Logging {
 
   val KUBERNETES_EXECUTOR_SERVICE_ACCOUNT_NAME =
     ConfigBuilder(s"$KUBERNETES_AUTH_EXECUTOR_CONF_PREFIX.serviceAccountName")
-      .doc("Service account that is used when running the executor pod." +
-        "If this parameter is not setup, the fallback logic will use the driver's service account.")
+      .doc("Service account that is used when running the executor pod. " +
+        "If this parameter is not setup, the fallback logic will use the value of " +
+        "spark.kubernetes.authenticate.driver.serviceAccountName. Both are ignored when the " +
+        "executor pod template already names a non-empty service account.")
       .version("3.1.0")
       .stringConf
       .createOptional

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/KubernetesUtils.scala

@@ -335,6 +335,18 @@ object KubernetesUtils extends Logging {
     }
   }
 
+  /**
+   * The service account the pod's spec names, if any. Both spec fields are read,
+   * `serviceAccountName` winning and an empty value counting as unset, matching Kubernetes'
+   * SetDefaults_PodSpec. A pod template is deserialized with no API-server defaulting, so it can
+   * name the account in either field and leave the other null.
+   */
+  private[spark] def podServiceAccount(pod: SparkPod): Option[String] = {
+    val spec = Option(pod.pod.getSpec)
+    spec.flatMap(s => Option(s.getServiceAccountName)).filter(_.nonEmpty)
+      .orElse(spec.flatMap(s => Option(s.getServiceAccount)).filter(_.nonEmpty))
+  }
+
   @Since("3.0.0")
   def buildPodWithServiceAccount(serviceAccount: Option[String], pod: SparkPod): Option[Pod] = {
     serviceAccount.map { account =>

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/DriverKubernetesCredentialsFeatureStep.scala

@@ -28,7 +28,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
+import org.apache.spark.deploy.k8s.KubernetesUtils.{buildPodWithServiceAccount, podServiceAccount}
 import org.apache.spark.internal.Logging
 import org.apache.spark.internal.LogKeys.{CONFIG, CONFIGS, PREFIX, SERVICE_ACCOUNT_NAME, VALUE}
 
@@ -81,15 +81,10 @@ private[spark] class DriverKubernetesCredentialsFeatureStep(kubernetesConf: Kube
     } else {
       // The credentials secret takes precedence over the driver service account: this branch never
       // applies the account, so warn that the pod keeps whatever its spec names, or the namespace
-      // default. Stay quiet when the spec already names the same account. Both spec fields are
-      // read, `serviceAccountName` winning, matching Kubernetes' SetDefaults_PodSpec: a pod
-      // template is deserialized with no API-server defaulting, so it can leave either one null.
-      val podSpec = Option(pod.pod.getSpec)
-      val podServiceAccount = podSpec.flatMap(s => Option(s.getServiceAccountName))
-        .filter(_.nonEmpty)
-        .orElse(podSpec.flatMap(s => Option(s.getServiceAccount)).filter(_.nonEmpty))
-      driverServiceAccount.filterNot(podServiceAccount.contains).foreach { account =>
-        val keptAccount = podServiceAccount
+      // default. Stay quiet when the spec already names the same account.
+      val specServiceAccount = podServiceAccount(pod)
+      driverServiceAccount.filterNot(specServiceAccount.contains).foreach { account =>
+        val keptAccount = specServiceAccount
           .map(name => log"the pod keeps ${MDC(SERVICE_ACCOUNT_NAME, name)}, named by its spec")
           .getOrElse(log"the pod falls back to the namespace's default account")
         logWarning(log"Not applying " +

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/ExecutorKubernetesCredentialsFeatureStep.scala

@@ -18,24 +18,63 @@ package org.apache.spark.deploy.k8s.features
 
 import org.apache.spark.deploy.k8s.{KubernetesConf, SparkPod}
 import org.apache.spark.deploy.k8s.Config.{KUBERNETES_DRIVER_SERVICE_ACCOUNT_NAME, KUBERNETES_EXECUTOR_SERVICE_ACCOUNT_NAME}
-import org.apache.spark.deploy.k8s.KubernetesUtils.buildPodWithServiceAccount
+import org.apache.spark.deploy.k8s.KubernetesUtils.{buildPodWithServiceAccount, podServiceAccount}
+import org.apache.spark.internal.Logging
+import org.apache.spark.internal.LogKeys.{CONFIG, SERVICE_ACCOUNT_NAME, VALUE}
 
 private[spark] class ExecutorKubernetesCredentialsFeatureStep(kubernetesConf: KubernetesConf)
-  extends KubernetesFeatureConfigStep {
+  extends KubernetesFeatureConfigStep with Logging {
 
   private lazy val driverServiceAccount = kubernetesConf.get(KUBERNETES_DRIVER_SERVICE_ACCOUNT_NAME)
   private lazy val executorServiceAccount =
     kubernetesConf.get(KUBERNETES_EXECUTOR_SERVICE_ACCOUNT_NAME)
 
   override def configurePod(pod: SparkPod): SparkPod = {
-      pod.copy(
-        // if not setup by the pod template, fallback to the executor's sa,
-        // if executor's sa is not setup, the last option is driver's sa.
-        pod = if (Option(pod.pod.getSpec.getServiceAccount).isEmpty) {
-          buildPodWithServiceAccount(executorServiceAccount
-            .orElse(driverServiceAccount), pod).getOrElse(pod.pod)
-        } else {
-          pod.pod
-        })
+    val templateServiceAccount = podServiceAccount(pod)
+    templateServiceAccount.foreach(reportAccountNotApplied)
+    pod.copy(
+      // if not setup by the pod template, fallback to the executor's sa,
+      // if executor's sa is not setup, the last option is driver's sa.
+      pod = if (templateServiceAccount.isEmpty) {
+        buildPodWithServiceAccount(executorServiceAccount.orElse(driverServiceAccount), pod)
+          .getOrElse(pod.pod)
+      } else {
+        pod.pod
+      })
+  }
+
+  /**
+   * The pod template's account takes precedence, so report a configured one that did not apply.
+   * An account set through the executor configuration is an explicit instruction that had no
+   * effect, so that is a warning. The driver's account only ever served as a fallback here, so a
+   * template superseding it is the documented outcome and says so at INFO. Either way, stay quiet
+   * when the template names the same account, and when the configured value is empty and so names
+   * no account at all.
+   *
+   * With the default pod allocator both repeat once per executor pod, since the feature steps are
+   * rebuilt for each one. Each warning corresponds to one pod launched with a different account
+   * than the configuration asked for. The INFO reports a correct configuration, and adds one line
+   * to the one `BasicExecutorFeatureStep` already logs per pod.
+   */
+  private def reportAccountNotApplied(templateAccount: String): Unit = {
+    executorServiceAccount match {
+      case Some(configured) =>
+        // An empty value counts as set on this path, which is long-standing; nothing to report.
+        if (configured.nonEmpty && configured != templateAccount) {
+          logWarning(log"Not applying " +
+            log"${MDC(CONFIG, KUBERNETES_EXECUTOR_SERVICE_ACCOUNT_NAME.key)}=" +
+            log"${MDC(VALUE, configured)} to the executor pod, because its pod template already " +
+            log"names ${MDC(SERVICE_ACCOUNT_NAME, templateAccount)}, which takes precedence. " +
+            log"Remove the account from the template to have Spark apply that configuration " +
+            log"instead.")
+        }
+      case None =>
+        driverServiceAccount.filter(a => a.nonEmpty && a != templateAccount).foreach { account =>
+          logInfo(log"The executor pod template names " +
+            log"${MDC(SERVICE_ACCOUNT_NAME, templateAccount)}, so the executor pods use it " +
+            log"rather than ${MDC(CONFIG, KUBERNETES_DRIVER_SERVICE_ACCOUNT_NAME.key)}=" +
+            log"${MDC(VALUE, account)}, which is only a fallback for executor pods.")
+        }
     }
+  }
 }