Share secret name between drivers and executors when mounting files u…

…sing a secret-populated volume

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

@@ -20,13 +20,14 @@ package org.apache.spark.deploy.k8s.features
 import java.io.File
 import java.net.URI
 import java.nio.file.Paths
+import java.util.Locale
 
 import scala.collection.JavaConverters._
 
 import com.google.common.io.{BaseEncoding, Files}
 import io.fabric8.kubernetes.api.model.{ContainerBuilder, HasMetadata, PodBuilder, SecretBuilder}
 
-import org.apache.spark.deploy.k8s.{KubernetesConf, KubernetesDriverConf, SparkPod}
+import org.apache.spark.deploy.k8s.{KubernetesConf, KubernetesDriverConf, KubernetesUtils, SparkPod}
 import org.apache.spark.deploy.k8s.Config._
 import org.apache.spark.deploy.k8s.Constants._
 import org.apache.spark.deploy.k8s.submit.{JavaMainAppResource, PythonMainAppResource, RMainAppResource}
@@ -56,7 +57,10 @@ private[spark] abstract class MountLocalFilesFeatureStep(conf: KubernetesConf)
 
   private val enabled = conf.get(KUBERNETES_SECRET_FILE_MOUNT_ENABLED)
 
-  private val secretName = s"${conf.resourceNamePrefix}-mounted-files"
+  // Secret name needs to be the same for drivers and executors because both will have a volume
+  // populated by the secret, but Spark's k8s client will only store the secret configured on the
+  // driver. If the secret names don't match, executors will fail to mount the volume.
+  private val secretName = s"${secretNamePrefix()}-mounted-files"
 
   private val mountPath = conf.get(KUBERNETES_SECRET_FILE_MOUNT_PATH)
 
@@ -132,4 +136,15 @@ private[spark] abstract class MountLocalFilesFeatureStep(conf: KubernetesConf)
       case _ => false
     }
   }
+
+  // Like KubernetesConf#getResourceNamePrefix but unique per app, not per resource.
+  private def secretNamePrefix(): String = {
+    s"${conf.appName}"
+      .trim
+      .toLowerCase(Locale.ROOT)
+      .replaceAll("\\s+", "-")
+      .replaceAll("\\.", "-")
+      .replaceAll("[^a-z0-9\\-]", "")
+      .replaceAll("-+", "-")
+  }
 }

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/k8s/features/MountLocalFilesFeatureStepSuite.scala

@@ -52,6 +52,7 @@ class MountLocalFilesFeatureStepSuite extends SparkFunSuite with BeforeAndAfter
       "https://localhost:9000/file4.txt")
     localFiles = Seq(firstLocalFile, secondLocalFile)
     val sparkConf = new SparkConf(false)
+      .set("spark.app.name", "test.app")
       .set("spark.files", sparkFiles.mkString(","))
       .set(KUBERNETES_SECRET_FILE_MOUNT_ENABLED, true)
       .set(KUBERNETES_SECRET_FILE_MOUNT_PATH, "/var/data/spark-submitted-files")
@@ -69,7 +70,7 @@ class MountLocalFilesFeatureStepSuite extends SparkFunSuite with BeforeAndAfter
     assert(configuredPod.pod.getSpec.getVolumes.size === 1)
     val volume = configuredPod.pod.getSpec.getVolumes.get(0)
     assert(volume.getName === "submitted-files")
-    assert(volume.getSecret.getSecretName === s"${kubernetesConf.resourceNamePrefix}-mounted-files")
+    assert(volume.getSecret.getSecretName === "test-app-mounted-files")
     assert(configuredPod.container.getVolumeMounts.size === 1)
     val volumeMount = configuredPod.container.getVolumeMounts.get(0)
     assert(volumeMount.getName === "submitted-files")
@@ -98,7 +99,7 @@ class MountLocalFilesFeatureStepSuite extends SparkFunSuite with BeforeAndAfter
     assert(secrets.size === 1)
     assert(secrets.head.isInstanceOf[Secret])
     val secret = secrets.head.asInstanceOf[Secret]
-    assert(secret.getMetadata.getName === s"${kubernetesConf.resourceNamePrefix}-mounted-files")
+    assert(secret.getMetadata.getName === s"test-app-mounted-files")
     val secretData = secret.getData.asScala
     assert(secretData.size === 2)
     assert(decodeToUtf8(secretData("file1.txt")) === "a")