Adding unit test and one more integration test

Author: ifilonenko

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

@@ -512,7 +512,7 @@ package object config extends Logging {
   private[spark] val KUBERNETES_NODE_SELECTOR_PREFIX = "spark.kubernetes.node.selector."
 
   private[spark] val KUBERNETES_KERBEROS_SUPPORT =
-    ConfigBuilder("spark.kubernetes.kerberos")
+    ConfigBuilder("spark.kubernetes.kerberos.enabled")
       .doc("Specify whether your job is a job that will require a Delegation Token to access HDFS")
       .booleanConf
       .createWithDefault(false)

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/hadoopsteps/HadoopKerberosKeytabResolverStep.scala

@@ -41,7 +41,7 @@ import org.apache.spark.internal.Logging
   * assumes that the job user has either specified a principal and keytab or ran
   * $kinit before running spark-submit. With a TGT stored locally, by running
   * UGI.getCurrentUser you are able to obtain the current user, alternatively
-  * you can run UGI.logingUserFromKeytabAndReturnUGI and by running .doAs run
+  * you can run UGI.loginUserFromKeytabAndReturnUGI and by running .doAs run
   * as the logged into user instead of the current user. With the Job User principal
   * you then retrieve the delegation token from the NameNode and store values in
   * DelegationToken. Lastly, the class puts the data into a secret. All this is

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/hadoopsteps/HadoopStepsOrchestrator.scala

@@ -23,7 +23,6 @@ import org.apache.spark.deploy.kubernetes.{HadoopConfBootstrapImpl, OptionRequir
 import org.apache.spark.deploy.kubernetes.config._
 import org.apache.spark.internal.Logging
 
-
  /**
   * Returns the complete ordered list of steps required to configure the hadoop configurations.
   */

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/kubernetes/InitContainerResourceStagingServerSecretPluginSuite.scala

@@ -23,7 +23,7 @@ import scala.collection.JavaConverters._
 import org.apache.spark.deploy.kubernetes.constants._
 import org.apache.spark.SparkFunSuite
 
-class InitContainerResourceStagingServerSecretPluginSuite extends SparkFunSuite with BeforeAndAfter{
+private[spark] class InitContainerResourceStagingServerSecretPluginSuite extends SparkFunSuite with BeforeAndAfter{
   private val INIT_CONTAINER_SECRET_NAME = "init-secret"
   private val INIT_CONTAINER_SECRET_MOUNT = "/tmp/secret"
 

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/kubernetes/submit/ClientSuite.scala

@@ -17,22 +17,22 @@
 package org.apache.spark.deploy.kubernetes.submit
 
 import com.google.common.collect.Iterables
-import io.fabric8.kubernetes.api.model.{ContainerBuilder, DoneablePod, HasMetadata, Pod, PodBuilder, PodList, Secret, SecretBuilder}
+import io.fabric8.kubernetes.api.model._
+import io.fabric8.kubernetes.client.dsl.{MixedOperation, NamespaceListVisitFromServerGetDeleteRecreateWaitApplicable, PodResource}
 import io.fabric8.kubernetes.client.{KubernetesClient, Watch}
-import io.fabric8.kubernetes.client.dsl.{MixedOperation, NamespaceListVisitFromServerGetDeleteRecreateWaitApplicable, NamespaceVisitFromServerGetWatchDeleteRecreateWaitApplicable, PodResource, Resource}
-import org.mockito.{ArgumentCaptor, Mock, MockitoAnnotations}
+import org.apache.spark.deploy.kubernetes.constants._
+import org.apache.spark.deploy.kubernetes.submit.submitsteps.{DriverConfigurationStep, KubernetesDriverSpec}
+import org.apache.spark.{SparkConf, SparkFunSuite}
 import org.mockito.Mockito.{doReturn, verify, when}
 import org.mockito.invocation.InvocationOnMock
 import org.mockito.stubbing.Answer
+import org.mockito.{ArgumentCaptor, Mock, MockitoAnnotations}
 import org.scalatest.BeforeAndAfter
 import org.scalatest.mock.MockitoSugar._
-import scala.collection.JavaConverters._
 
-import org.apache.spark.{SparkConf, SparkFunSuite}
-import org.apache.spark.deploy.kubernetes.constants._
-import org.apache.spark.deploy.kubernetes.submit.submitsteps.{DriverConfigurationStep, KubernetesDriverSpec}
+import scala.collection.JavaConverters._
 
-class ClientSuite extends SparkFunSuite with BeforeAndAfter {
+private[spark] class ClientSuite extends SparkFunSuite with BeforeAndAfter {
 
   private val DRIVER_POD_UID = "pod-id"
   private val DRIVER_POD_API_VERSION = "v1"

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/PythonStepSuite.scala

@@ -22,7 +22,7 @@ import scala.collection.JavaConverters._
 
 import org.apache.spark.{SparkConf, SparkFunSuite}
 
-class PythonStepSuite extends SparkFunSuite with BeforeAndAfter {
+private[spark] class PythonStepSuite extends SparkFunSuite with BeforeAndAfter {
   private val FILE_DOWNLOAD_PATH = "/var/data/spark-files"
   private val PYSPARK_FILES = Seq(
     "hdfs://localhost:9000/app/files/file1.py",

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/hadoopsteps/HadoopConfMounterStepSuite.scala

@@ -0,0 +1,102 @@
+/*
+ * 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.kubernetes.submit.submitsteps.hadoopsteps
+
+import java.io.File
+import java.util.UUID
+
+import scala.collection.JavaConverters._
+
+import com.google.common.io.Files
+import io.fabric8.kubernetes.api.model._
+import org.apache.commons.io.FileUtils.readFileToString
+import org.mockito.{Mock, MockitoAnnotations}
+import org.mockito.Matchers.any
+import org.mockito.Mockito.when
+import org.mockito.invocation.InvocationOnMock
+import org.mockito.stubbing.Answer
+import org.scalatest.BeforeAndAfter
+
+import org.apache.spark.SparkFunSuite
+import org.apache.spark.deploy.kubernetes.{HadoopConfBootstrap, PodWithMainContainer}
+import org.apache.spark.deploy.kubernetes.constants.HADOOP_CONF_DIR_LOC
+import org.apache.spark.util.Utils
+
+
+private[spark] class HadoopConfMounterStepSuite extends SparkFunSuite with BeforeAndAfter{
+  private val CONFIG_MAP_NAME = "config-map"
+  private val HADOOP_CONF_DIR_VAL = "/etc/hadoop"
+  private val POD_LABEL = Map("bootstrap" -> "true")
+  private val DRIVER_CONTAINER_NAME = "driver-container"
+  private val TEMP_HADOOP_FILE = createTempFile("core-site.xml")
+  private val HADOOP_FILES = Seq(TEMP_HADOOP_FILE)
+
+  @Mock
+  private var hadoopConfBootstrap : HadoopConfBootstrap = _
+
+  before {
+    MockitoAnnotations.initMocks(this)
+    when(hadoopConfBootstrap.bootstrapMainContainerAndVolumes(
+      any[PodWithMainContainer])).thenAnswer(new Answer[PodWithMainContainer] {
+      override def answer(invocation: InvocationOnMock) : PodWithMainContainer = {
+        val pod = invocation.getArgumentAt(0, classOf[PodWithMainContainer])
+        pod.copy(
+          pod =
+            new PodBuilder(pod.pod)
+              .withNewMetadata()
+              .addToLabels("bootstrap", "true")
+              .endMetadata()
+              .withNewSpec().endSpec()
+              .build(),
+          mainContainer =
+            new ContainerBuilder()
+              .withName(DRIVER_CONTAINER_NAME).build()
+        )}})
+  }
+
+  test("Test of mounting hadoop_conf_dir files into HadoopConfigSpec") {
+    val hadoopConfStep = new HadoopConfMounterStep(
+      CONFIG_MAP_NAME,
+      HADOOP_FILES,
+      hadoopConfBootstrap,
+      HADOOP_CONF_DIR_VAL)
+    val expectedDriverSparkConf = Map(HADOOP_CONF_DIR_LOC -> HADOOP_CONF_DIR_VAL)
+    val expectedConfigMap = Map(
+      TEMP_HADOOP_FILE.toPath.getFileName.toString ->
+        readFileToString(TEMP_HADOOP_FILE)
+    )
+    val hadoopConfSpec = HadoopConfigSpec(
+      Map.empty[String, String],
+      new Pod(),
+      new Container(),
+      Map.empty[String, String],
+      None,
+      "",
+      "")
+    val returnContainerSpec = hadoopConfStep.configureContainers(hadoopConfSpec)
+    assert(expectedDriverSparkConf === returnContainerSpec.additionalDriverSparkConf)
+    assert(returnContainerSpec.driverContainer.getName == DRIVER_CONTAINER_NAME)
+    assert(returnContainerSpec.driverPod.getMetadata.getLabels.asScala === POD_LABEL)
+    assert(returnContainerSpec.configMapProperties === expectedConfigMap)
+  }
+  private def createTempFile(contents: String): File = {
+    val dir = Utils.createTempDir()
+    val file = new File(dir, s"${UUID.randomUUID().toString}")
+    Files.write(contents.getBytes, file)
+    file
+  }
+}

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/hadoopsteps/HadoopKerberosKeytabResolverStepSuite.scala

@@ -0,0 +1,77 @@
+/*
+ * 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.kubernetes.submit.submitsteps.hadoopsteps
+
+import java.io.File
+import java.util.UUID
+
+import scala.collection.JavaConverters._
+
+import com.google.common.io.Files
+import io.fabric8.kubernetes.api.model._
+
+import org.apache.spark.{SparkConf, SparkFunSuite}
+import org.apache.spark.deploy.kubernetes.constants._
+import org.apache.spark.util.Utils
+
+
+
+private[spark] class HadoopKerberosKeytabResolverStepSuite extends SparkFunSuite {
+  private val POD_LABEL = Map("bootstrap" -> "true")
+  private val DRIVER_CONTAINER_NAME = "driver-container"
+  private val TEMP_KEYTAB_FILE = createTempFile("keytab")
+  private val KERB_PRINCIPAL = "[email protected]"
+
+  // TODO: Require mocking of UGI methods
+  test("Testing keytab login") {
+    val keytabStep = new HadoopKerberosKeytabResolverStep(
+      new SparkConf(),
+      Some(KERB_PRINCIPAL),
+      Some(TEMP_KEYTAB_FILE))
+    val hadoopConfSpec = HadoopConfigSpec(
+      Map.empty[String, String],
+      new PodBuilder()
+        .withNewMetadata()
+        .addToLabels("bootstrap", "true")
+        .endMetadata()
+        .withNewSpec().endSpec()
+        .build(),
+      new ContainerBuilder().withName(DRIVER_CONTAINER_NAME).build(),
+      Map.empty[String, String],
+      None,
+      "",
+      "")
+    val returnContainerSpec = keytabStep.configureContainers(hadoopConfSpec)
+    assert(returnContainerSpec.additionalDriverSparkConf(HADOOP_KERBEROS_CONF_LABEL)
+        .contains(KERBEROS_SECRET_LABEL_PREFIX))
+    assert(returnContainerSpec.additionalDriverSparkConf(HADOOP_KERBEROS_CONF_SECRET) ===
+      HADOOP_KERBEROS_SECRET_NAME)
+    assert(returnContainerSpec.driverContainer.getName == DRIVER_CONTAINER_NAME)
+    assert(returnContainerSpec.driverPod.getMetadata.getLabels.asScala === POD_LABEL)
+    assert(returnContainerSpec.dtSecretLabel.contains(KERBEROS_SECRET_LABEL_PREFIX))
+    assert(returnContainerSpec.dtSecretName === HADOOP_KERBEROS_SECRET_NAME)
+    assert(returnContainerSpec.dtSecret.nonEmpty)
+    assert(returnContainerSpec.dtSecret.get.getMetadata.getName === HADOOP_KERBEROS_SECRET_NAME)
+  }
+
+  private def createTempFile(contents: String): File = {
+    val dir = Utils.createTempDir()
+    val file = new File(dir, s"${UUID.randomUUID().toString}")
+    Files.write(contents.getBytes, file)
+    file
+  }
+}

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/hadoopsteps/HadoopKerberosSecretResolverStepSuite.scala

@@ -0,0 +1,63 @@
+/*
+ * 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.kubernetes.submit.submitsteps.hadoopsteps
+
+import scala.collection.JavaConverters._
+
+import io.fabric8.kubernetes.api.model._
+
+import org.apache.spark.{SparkConf, SparkFunSuite}
+import org.apache.spark.deploy.kubernetes.constants._
+
+private[spark] class HadoopKerberosSecretResolverStepSuite extends SparkFunSuite {
+  private val CONFIG_MAP_NAME = "config-map"
+  private val HADOOP_CONF_DIR_VAL = "/etc/hadoop"
+  private val POD_LABEL = Map("bootstrap" -> "true")
+  private val DRIVER_CONTAINER_NAME = "driver-container"
+  private val TOKEN_SECRET_NAME = "secretName"
+  private val TOKEN_SECRET_LABEL = "secretLabel"
+
+  test("Testing kerberos with Secret") {
+    val keytabStep = new HadoopKerberosSecretResolverStep(
+      new SparkConf(),
+      TOKEN_SECRET_NAME,
+      TOKEN_SECRET_LABEL)
+    val expectedDriverSparkConf = Map(
+      HADOOP_KERBEROS_CONF_LABEL -> TOKEN_SECRET_LABEL,
+      HADOOP_KERBEROS_CONF_SECRET -> TOKEN_SECRET_NAME)
+    val hadoopConfSpec = HadoopConfigSpec(
+      Map.empty[String, String],
+      new PodBuilder()
+        .withNewMetadata()
+        .addToLabels("bootstrap", "true")
+        .endMetadata()
+        .withNewSpec().endSpec()
+        .build(),
+      new ContainerBuilder().withName(DRIVER_CONTAINER_NAME).build(),
+      Map.empty[String, String],
+      None,
+      "",
+      "")
+    val returnContainerSpec = keytabStep.configureContainers(hadoopConfSpec)
+    assert(expectedDriverSparkConf === returnContainerSpec.additionalDriverSparkConf)
+    assert(returnContainerSpec.driverContainer.getName == DRIVER_CONTAINER_NAME)
+    assert(returnContainerSpec.driverPod.getMetadata.getLabels.asScala === POD_LABEL)
+    assert(returnContainerSpec.dtSecret === None)
+    assert(returnContainerSpec.dtSecretLabel === TOKEN_SECRET_LABEL)
+    assert(returnContainerSpec.dtSecretName === TOKEN_SECRET_NAME)
+  }
+}

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/hadoopsteps/HadoopStepsOrchestratorSuite.scala

@@ -0,0 +1,72 @@
+/*
+ * 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.kubernetes.submit.submitsteps.hadoopsteps
+
+import org.apache.spark.{SparkConf, SparkFunSuite}
+import org.apache.spark.deploy.kubernetes.config._
+
+private[spark] class HadoopStepsOrchestratorSuite extends SparkFunSuite {
+  private val NAMESPACE = "testNamespace"
+  private val HADOOP_CONFIG_MAP = "hadoop-config-map"
+  private val HADOOP_CONF_DIR_VAL = "/etc/hadoop/conf"
+
+  test("Testing without Kerberos") {
+    val sparkTestConf = new SparkConf(true)
+      .set(KUBERNETES_KERBEROS_SUPPORT, false)
+    val hadoopOrchestrator = new HadoopStepsOrchestrator(
+      NAMESPACE,
+      HADOOP_CONFIG_MAP,
+      sparkTestConf,
+      HADOOP_CONF_DIR_VAL)
+    val steps = hadoopOrchestrator.getHadoopSteps()
+    assert(steps.length === 1)
+    assert(steps.head.isInstanceOf[HadoopConfMounterStep])
+    assert(true)
+  }
+
+//  test("Testing with Keytab Kerberos Login") {
+//    val sparkTestConf2 = new SparkConf(true)
+//      .set(KUBERNETES_KERBEROS_SUPPORT, true)
+//      .set(KUBERNETES_KERBEROS_KEYTAB, "keytab.file")
+//      .set(KUBERNETES_KERBEROS_PRINCIPAL, "user@kerberos")
+//    val hadoopOrchestrator = new HadoopStepsOrchestrator(
+//      NAMESPACE,
+//      HADOOP_CONFIG_MAP,
+//      sparkTestConf2,
+//      HADOOP_CONF_DIR_VAL)
+//    val steps = hadoopOrchestrator.getHadoopSteps()
+//    assert(steps.length === 2)
+//    assert(steps.head.isInstanceOf[HadoopConfMounterStep])
+//    assert(steps(1).isInstanceOf[HadoopKerberosKeytabResolverStep])
+//  }
+
+//  test("Testing with Keytab Kerberos Login") {
+//    val sparkTestConf3 = new SparkConf(true)
+//      .set(KUBERNETES_KERBEROS_SUPPORT, true)
+//      .set(KUBERNETES_KERBEROS_DT_SECRET_NAME, "dtSecret")
+//      .set(KUBERNETES_KERBEROS_DT_SECRET_LABEL, "dtLabel")
+//    val hadoopOrchestrator = new HadoopStepsOrchestrator(
+//      NAMESPACE,
+//      HADOOP_CONFIG_MAP,
+//      sparkTestConf3,
+//      HADOOP_CONF_DIR_VAL)
+//    val steps = hadoopOrchestrator.getHadoopSteps()
+//    assert(steps.length === 2)
+//    assert(steps.head.isInstanceOf[HadoopConfMounterStep])
+//    assert(steps(1).isInstanceOf[HadoopKerberosSecretResolverStep])
+//  }
+}