[SPARK-32067][k8s] Give pod template configmap a unique name

The pod template configmap always had the same name
for each job submitted.
This means if you schedule 2 spark jobs in the same
namespace there will be conflicts.

Author: stijndehaes

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

@@ -86,6 +86,7 @@ private[spark] object Constants {
   val DEFAULT_EXECUTOR_CONTAINER_NAME = "spark-kubernetes-executor"
   val MEMORY_OVERHEAD_MIN_MIB = 384L
   val NON_JVM_MEMORY_OVERHEAD_FACTOR = 0.4d
+  val KUBERNETES_MAX_NAME_LENGTH = 63
 
   // Hadoop Configuration
   val HADOOP_CONF_VOLUME = "hadoop-properties"

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

@@ -83,7 +83,7 @@ private[spark] class BasicExecutorFeatureStep(
     // hostname must be no longer than 63 characters, so take the last 63 characters of the pod
     // name as the hostname.  This preserves uniqueness since the end of name contains
     // executorId
-    val hostname = name.substring(Math.max(0, name.length - 63))
+    val hostname = name.substring(Math.max(0, name.length - KUBERNETES_MAX_NAME_LENGTH))
       // Remove non-word characters from the start of the hostname
       .replaceAll("^[^\\w]+", "")
       // Replace dangerous characters in the remaining string with a safe alternative.

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

@@ -39,13 +39,13 @@ private[spark] class DriverServiceFeatureStep(
       "managed via a Kubernetes service.")
 
   private val preferredServiceName = s"${kubernetesConf.resourceNamePrefix}$DRIVER_SVC_POSTFIX"
-  private val resolvedServiceName = if (preferredServiceName.length <= MAX_SERVICE_NAME_LENGTH) {
+  private val resolvedServiceName = if (preferredServiceName.length <= KUBERNETES_MAX_NAME_LENGTH) {
     preferredServiceName
   } else {
     val randomServiceId = KubernetesUtils.uniqueID(clock = clock)
     val shorterServiceName = s"spark-$randomServiceId$DRIVER_SVC_POSTFIX"
     logWarning(s"Driver's hostname would preferably be $preferredServiceName, but this is " +
-      s"too long (must be <= $MAX_SERVICE_NAME_LENGTH characters). Falling back to use " +
+      s"too long (must be <= $KUBERNETES_MAX_NAME_LENGTH characters). Falling back to use " +
       s"$shorterServiceName as the driver service's name.")
     shorterServiceName
   }
@@ -99,5 +99,4 @@ private[spark] object DriverServiceFeatureStep {
   val DRIVER_BIND_ADDRESS_KEY = config.DRIVER_BIND_ADDRESS.key
   val DRIVER_HOST_KEY = config.DRIVER_HOST_ADDRESS.key
   val DRIVER_SVC_POSTFIX = "-driver-svc"
-  val MAX_SERVICE_NAME_LENGTH = 63
 }

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

@@ -22,23 +22,39 @@ import java.nio.charset.StandardCharsets
 import com.google.common.io.Files
 import io.fabric8.kubernetes.api.model.{ConfigMapBuilder, ContainerBuilder, HasMetadata, PodBuilder}
 
-import org.apache.spark.deploy.k8s.{KubernetesConf, SparkPod}
+import org.apache.spark.deploy.k8s.{KubernetesConf, KubernetesUtils, SparkPod}
 import org.apache.spark.deploy.k8s.Config._
 import org.apache.spark.deploy.k8s.Constants._
+import org.apache.spark.internal.Logging
+import org.apache.spark.util.{Clock, SystemClock}
 
-private[spark] class PodTemplateConfigMapStep(conf: KubernetesConf)
-  extends KubernetesFeatureConfigStep {
+private[spark] class PodTemplateConfigMapStep
+(conf: KubernetesConf, clock: Clock = new SystemClock())
+  extends KubernetesFeatureConfigStep with Logging {
 
   private val hasTemplate = conf.contains(KUBERNETES_EXECUTOR_PODTEMPLATE_FILE)
 
+  private val preferredConfigmapName = s"${conf.resourceNamePrefix}-$POD_TEMPLATE_CONFIGMAP"
+  private val resolvedConfigmapName = if (preferredConfigmapName.length <= KUBERNETES_MAX_NAME_LENGTH)
+  {
+    preferredConfigmapName
+  } else {
+    val randomServiceId = KubernetesUtils.uniqueID(clock = clock)
+    val shorterTemplateConfigmapName = s"spark-$randomServiceId-$POD_TEMPLATE_CONFIGMAP"
+    logWarning(s"The pod template configmap name preferably be $preferredConfigmapName," +
+      s"but this is too long (must be <= $KUBERNETES_MAX_NAME_LENGTH characters). Falling back to" +
+      s"use $shorterTemplateConfigmapName as the pod template configmap name.")
+    shorterTemplateConfigmapName
+  }
+
   def configurePod(pod: SparkPod): SparkPod = {
     if (hasTemplate) {
       val podWithVolume = new PodBuilder(pod.pod)
           .editSpec()
             .addNewVolume()
               .withName(POD_TEMPLATE_VOLUME)
               .withNewConfigMap()
-                .withName(POD_TEMPLATE_CONFIGMAP)
+                .withName(resolvedConfigmapName)
                 .addNewItem()
                   .withKey(POD_TEMPLATE_KEY)
                   .withPath(EXECUTOR_POD_SPEC_TEMPLATE_FILE_NAME)
@@ -76,7 +92,7 @@ private[spark] class PodTemplateConfigMapStep(conf: KubernetesConf)
       val podTemplateString = Files.toString(new File(podTemplateFile), StandardCharsets.UTF_8)
       Seq(new ConfigMapBuilder()
           .withNewMetadata()
-            .withName(POD_TEMPLATE_CONFIGMAP)
+            .withName(resolvedConfigmapName)
           .endMetadata()
           .addToData(POD_TEMPLATE_KEY, podTemplateString)
         .build())
@@ -85,3 +101,4 @@ private[spark] class PodTemplateConfigMapStep(conf: KubernetesConf)
     }
   }
 }
+

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

@@ -33,8 +33,7 @@ import org.apache.spark.util.ManualClock
 class DriverServiceFeatureStepSuite extends SparkFunSuite {
 
   private val LONG_RESOURCE_NAME_PREFIX =
-    "a" * (DriverServiceFeatureStep.MAX_SERVICE_NAME_LENGTH -
-      DriverServiceFeatureStep.DRIVER_SVC_POSTFIX.length + 1)
+    "a" * (KUBERNETES_MAX_NAME_LENGTH - DriverServiceFeatureStep.DRIVER_SVC_POSTFIX.length + 1)
   private val DRIVER_LABELS = Map(
     "label1key" -> "label1value",
     "label2key" -> "label2value")

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

@@ -21,12 +21,16 @@ import java.nio.file.Files
 
 import io.fabric8.kubernetes.api.model.ConfigMap
 
+import org.apache.spark.deploy.k8s.Constants.{KUBERNETES_MAX_NAME_LENGTH, POD_TEMPLATE_CONFIGMAP}
 import org.apache.spark.{SparkConf, SparkFunSuite}
 import org.apache.spark.deploy.k8s._
 import org.apache.spark.util.Utils
 
 class PodTemplateConfigMapStepSuite extends SparkFunSuite {
 
+  private val LONG_RESOURCE_NAME_PREFIX =
+    "a" * (KUBERNETES_MAX_NAME_LENGTH - POD_TEMPLATE_CONFIGMAP.length + 1)
+
   test("Do nothing when executor template is not specified") {
     val conf = KubernetesTestConf.createDriverConf()
     val step = new PodTemplateConfigMapStep(conf)
@@ -56,8 +60,9 @@ class PodTemplateConfigMapStepSuite extends SparkFunSuite {
 
     assert(configuredPod.pod.getSpec.getVolumes.size() === 1)
     val volume = configuredPod.pod.getSpec.getVolumes.get(0)
+    val generatedResourceName = s"${kubernetesConf.resourceNamePrefix}-$POD_TEMPLATE_CONFIGMAP"
     assert(volume.getName === Constants.POD_TEMPLATE_VOLUME)
-    assert(volume.getConfigMap.getName === Constants.POD_TEMPLATE_CONFIGMAP)
+    assert(volume.getConfigMap.getName === generatedResourceName)
     assert(volume.getConfigMap.getItems.size() === 1)
     assert(volume.getConfigMap.getItems.get(0).getKey === Constants.POD_TEMPLATE_KEY)
     assert(volume.getConfigMap.getItems.get(0).getPath ===
@@ -70,7 +75,7 @@ class PodTemplateConfigMapStepSuite extends SparkFunSuite {
 
     val resources = step.getAdditionalKubernetesResources()
     assert(resources.size === 1)
-    assert(resources.head.getMetadata.getName === Constants.POD_TEMPLATE_CONFIGMAP)
+    assert(resources.head.getMetadata.getName === generatedResourceName)
     assert(resources.head.isInstanceOf[ConfigMap])
     val configMap = resources.head.asInstanceOf[ConfigMap]
     assert(configMap.getData.size() === 1)
@@ -84,4 +89,31 @@ class PodTemplateConfigMapStepSuite extends SparkFunSuite {
       (Constants.EXECUTOR_POD_SPEC_TEMPLATE_MOUNTPATH + "/" +
         Constants.EXECUTOR_POD_SPEC_TEMPLATE_FILE_NAME))
   }
+
+  test("SPARK-32067: Long prefixes should switch to using a generated unique name.") {
+    val templateFile = Files.createTempFile("pod-template", "yml").toFile
+    templateFile.deleteOnExit()
+
+    val sparkConf = new SparkConf(false)
+      .set(Config.KUBERNETES_EXECUTOR_PODTEMPLATE_FILE, templateFile.getAbsolutePath)
+    val kubernetesConf = KubernetesTestConf.createDriverConf(
+      sparkConf = sparkConf,
+      resourceNamePrefix = Some(LONG_RESOURCE_NAME_PREFIX))
+
+    Utils.tryWithResource(new PrintWriter(templateFile)) { writer =>
+      writer.write("pod-template-contents")
+    }
+
+    val step = new PodTemplateConfigMapStep(kubernetesConf)
+    val configuredPod = step.configurePod(SparkPod.initialPod())
+
+    assert(configuredPod.pod.getSpec.getVolumes.size() === 1)
+    val volume = configuredPod.pod.getSpec.getVolumes.get(0)
+    assert(!volume.getConfigMap.getName.startsWith(kubernetesConf.resourceNamePrefix))
+    val resources = step.getAdditionalKubernetesResources()
+    assert(resources.size === 1)
+    assert(!resources.head.getMetadata.getName.startsWith(kubernetesConf.resourceNamePrefix))
+    assert(volume.getConfigMap.getName === resources.head.getMetadata.getName)
+    assert(resources.head.getMetadata.getName.length <= KUBERNETES_MAX_NAME_LENGTH)
+  }
 }