add comprehensive test for supported template features

Author: onursatici

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

@@ -105,7 +105,7 @@ private[spark] class BasicDriverFeatureStep(
           .withNewFieldRef("v1", "status.podIP")
           .build())
         .endEnv()
-      .withNewResources()
+      .editOrNewResources()
         .addToRequests("cpu", driverCpuQuantity)
         .addToLimits(maybeCpuLimitQuantity.toMap.asJava)
         .addToRequests("memory", driverMemoryQuantity)
@@ -121,7 +121,7 @@ private[spark] class BasicDriverFeatureStep(
         .endMetadata()
       .withNewSpec()
         .withRestartPolicy("Never")
-        .withNodeSelector(conf.nodeSelector().asJava)
+        .addToNodeSelector(conf.nodeSelector().asJava)
         .addToImagePullSecrets(conf.imagePullSecrets(): _*)
         .endSpec()
       .build()

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

@@ -132,7 +132,7 @@ private[spark] class BasicExecutorFeatureStep(
       .withName(Option(pod.container.getName).getOrElse(DEFAULT_EXECUTOR_CONTAINER_NAME))
       .withImage(executorContainerImage)
       .withImagePullPolicy(kubernetesConf.imagePullPolicy())
-      .withNewResources()
+      .editOrNewResources()
         .addToRequests("memory", executorMemoryQuantity)
         .addToLimits("memory", executorMemoryQuantity)
         .addToRequests("cpu", executorCpuQuantity)
@@ -170,7 +170,7 @@ private[spark] class BasicExecutorFeatureStep(
       .editOrNewSpec()
         .withHostname(hostname)
         .withRestartPolicy("Never")
-        .withNodeSelector(kubernetesConf.nodeSelector().asJava)
+        .addToNodeSelector(kubernetesConf.nodeSelector().asJava)
         .addToImagePullSecrets(kubernetesConf.imagePullSecrets(): _*)
         .endSpec()
       .build()

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBuilderSuite.scala

@@ -168,31 +168,96 @@ class KubernetesExecutorBuilderSuite extends SparkFunSuite {
       new PodBuilder()
         .withNewMetadata()
           .addToLabels("test-label-key", "test-label-value")
+          .addToAnnotations("test-annotation-key", "test-annotation-value")
+          .withNamespace("namespace")
+          .addNewOwnerReference()
+            .withController(true)
+            .withName("owner-reference")
+            .endOwnerReference()
           .endMetadata()
         .withNewSpec()
+          .withDnsPolicy("dns-policy")
+          .withHostAliases(new HostAliasBuilder().withHostnames("hostname").build())
+          .withImagePullSecrets(
+            new LocalObjectReferenceBuilder().withName("local-reference").build())
+          .withInitContainers(new ContainerBuilder().withName("init-container").build())
+          .withNodeName("node-name")
+          .withNodeSelector(Map("node-selector-key" -> "node-selector-value").asJava)
+          .withSchedulerName("scheduler")
+          .withNewSecurityContext()
+            .withRunAsUser(1000L)
+            .endSecurityContext()
+          .withServiceAccount("service-account")
+          .withSubdomain("subdomain")
+          .withTolerations(new TolerationBuilder()
+            .withKey("toleration-key")
+            .withOperator("Equal")
+            .withEffect("NoSchedule")
+            .build())
+          .addNewVolume()
+            .withNewHostPath()
+            .withPath("/test")
+            .endHostPath()
+            .withName("test-volume")
+            .endVolume()
           .addNewContainer()
+            .withArgs("arg")
+            .withCommand("command")
+            .addNewEnv()
+              .withName("env-key")
+              .withValue("env-value")
+              .endEnv()
+            .withImagePullPolicy("Always")
             .withName("executor-container")
+            .withNewResources()
+              .withLimits(Map("gpu" -> new QuantityBuilder().withAmount("1").build()).asJava)
+              .endResources()
+            .withNewSecurityContext()
+              .withRunAsNonRoot(true)
+              .endSecurityContext()
+            .withStdin(true)
+            .withTerminationMessagePath("termination-message-path")
+            .withTerminationMessagePolicy("termination-message-policy")
             .addToVolumeMounts(
               new VolumeMountBuilder()
                 .withName("test-volume")
                 .withMountPath("/test")
                 .build())
             .endContainer()
-          .addNewVolume()
-            .withNewHostPath()
-              .withPath("/test")
-              .endHostPath()
-            .withName("test-volume")
-            .endVolume()
           .endSpec()
         .build())
 
-    assert(pod.pod.getMetadata.getLabels.containsKey("test-label-key"))
-    assert(!pod.pod.getSpec.getContainers.asScala.exists(_.getName == "executor-container"))
-    assert(pod.pod.getMetadata.getLabels.get("test-label-key") === "test-label-value")
-    assert(pod.container.getName === "executor-container")
+    val metadata = pod.pod.getMetadata
+    assert(metadata.getLabels.containsKey("test-label-key"))
+    assert(metadata.getAnnotations.containsKey("test-annotation-key"))
+    assert(metadata.getNamespace === "namespace")
+    assert(metadata.getOwnerReferences.asScala.exists(_.getName == "owner-reference"))
+    val spec = pod.pod.getSpec
+    assert(!spec.getContainers.asScala.exists(_.getName == "executor-container"))
+    assert(spec.getDnsPolicy === "dns-policy")
+    assert(spec.getHostAliases.asScala.exists(_.getHostnames.asScala.exists(_ == "hostname")))
+    assert(spec.getImagePullSecrets.asScala.exists(_.getName == "local-reference"))
+    assert(spec.getInitContainers.asScala.exists(_.getName == "init-container"))
+    assert(spec.getNodeName == "node-name")
+    assert(spec.getNodeSelector.get("node-selector-key") === "node-selector-value")
+    assert(spec.getSchedulerName === "scheduler")
+    assert(spec.getSecurityContext.getRunAsUser === 1000L)
+    assert(spec.getServiceAccount === "service-account")
+    assert(spec.getSubdomain === "subdomain")
+    assert(spec.getTolerations.asScala.exists(_.getKey == "toleration-key"))
+    assert(spec.getVolumes.asScala.exists(_.getName == "test-volume"))
+    val container = pod.container
+    assert(container.getName === "executor-container")
+    assert(container.getArgs.contains("arg"))
+    assert(container.getCommand.equals(List("command").asJava))
+    assert(container.getEnv.asScala.exists(_.getName == "env-key"))
+    assert(container.getResources.getLimits.get("gpu") ===
+      new QuantityBuilder().withAmount("1").build())
+    assert(container.getSecurityContext.getRunAsNonRoot)
+    assert(container.getStdin)
+    assert(container.getTerminationMessagePath === "termination-message-path")
+    assert(container.getTerminationMessagePolicy === "termination-message-policy")
     assert(pod.container.getVolumeMounts.asScala.exists(_.getName == "test-volume"))
-    assert(pod.pod.getSpec.getVolumes.asScala.exists(_.getName == "test-volume"))
   }
 
   private def constructPodWithPodTemplate(pod: Pod) : SparkPod = {