further testing work

Author: ifilonenko

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

@@ -27,10 +27,8 @@ import org.apache.spark.internal.Logging
   * mounted the DT secret for executors as well.
   */
 private[spark] trait KerberosTokenBootstrapConf {
-  /**
-    * Bootstraps a main container with the Secret mounted as volumes and an ENV variable
-    * pointing to the mounted file containing the DT for Secure HDFS interaction
-    */
+  // Bootstraps a main container with the Secret mounted as volumes and an ENV variable
+  // pointing to the mounted file containing the DT for Secure HDFS interaction
   def bootstrapMainContainerAndVolumes(
   originalPodWithMainContainer: PodWithMainContainer)
   : PodWithMainContainer

resource-managers/kubernetes/integration-tests-spark-jobs/src/main/scala/org/apache/spark/deploy/kubernetes/integrationtest/jobs/HDFSTest.scala

@@ -40,6 +40,7 @@ private[spark] object HDFSTest{
       val end = System.currentTimeMillis()
       println("Iteration " + iter + " took " + (end-start) + " ms")
     }
+    println(s"Returned length(s) of: ${file.map(s => s.length).collect().mkString(",")}")
     // scalastyle:on println
     spark.stop()
   }

resource-managers/kubernetes/integration-tests/src/test/resources/hdfs-site.xml

@@ -104,8 +104,6 @@
     <name>dfs.namenode.datanode.registration.ip-hostname-check</name>
     <value>false</value>
   </property>
-
-
   <property>
     <name>dfs.datanode.data.dir.perm</name>
     <value>700</value> 

resource-managers/kubernetes/integration-tests/src/test/resources/krb5.conf

@@ -12,7 +12,7 @@ renew_lifetime = 7d
 forwardable = true
 rdns = false
 default_realm = CLUSTER.LOCAL
-# default_ccache_name = KEYRING:persistent:%{uid}
+# default_ccache_name = MEMORY
 
 [realms]
 CLUSTER.LOCAL = {

resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/KubernetesSuite.scala

@@ -87,16 +87,16 @@ private[spark] class SparkDockerImageBuilder
     if (exitCode != 0) {
       logInfo(s"exitCode: $exitCode")
     }
-//    buildImage("spark-base", BASE_DOCKER_FILE)
-//    buildImage("spark-driver", DRIVER_DOCKER_FILE)
-//    buildImage("spark-driver-py", DRIVERPY_DOCKER_FILE)
-//    buildImage("spark-executor", EXECUTOR_DOCKER_FILE)
-//    buildImage("spark-executor-py", EXECUTORPY_DOCKER_FILE)
-//    buildImage("spark-shuffle", SHUFFLE_SERVICE_DOCKER_FILE)
-//    buildImage("spark-resource-staging-server", STAGING_SERVER_DOCKER_FILE)
-//    buildImage("spark-init", INIT_CONTAINER_DOCKER_FILE)
-//    buildImage("spark-integration-test-asset-server", STATIC_ASSET_SERVER_DOCKER_FILE)
-//    buildImage("kerberos-test", KERBEROS_DOCKER_FILE)
+    buildImage("spark-base", BASE_DOCKER_FILE)
+    buildImage("spark-driver", DRIVER_DOCKER_FILE)
+    buildImage("spark-driver-py", DRIVERPY_DOCKER_FILE)
+    buildImage("spark-executor", EXECUTOR_DOCKER_FILE)
+    buildImage("spark-executor-py", EXECUTORPY_DOCKER_FILE)
+    buildImage("spark-shuffle", SHUFFLE_SERVICE_DOCKER_FILE)
+    buildImage("spark-resource-staging-server", STAGING_SERVER_DOCKER_FILE)
+    buildImage("spark-init", INIT_CONTAINER_DOCKER_FILE)
+    buildImage("spark-integration-test-asset-server", STATIC_ASSET_SERVER_DOCKER_FILE)
+    buildImage("kerberos-test", KERBEROS_DOCKER_FILE)
   }
 
   private def buildImage(name: String, dockerFile: String): Unit = {

resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/docker/SparkDockerImageBuilder.scala

@@ -28,9 +28,9 @@ import io.fabric8.kubernetes.client.Watcher.Action
 import org.apache.spark.internal.Logging
 
  /**
-  * This class if used to ensure that the Hadoop cluster that is launched is executed
+  * This class is used to ensure that the Hadoop cluster that is launched is executed
   * in this order: KDC --> NN --> DN --> Data-Populator and that each one of these nodes
-  * is running before launching the kerberos test.
+  * is running before launching the Kerberos test.
   */
 private[spark] class KerberosPodWatcherCache(
   kerberosUtils: KerberosUtils,
@@ -220,14 +220,12 @@ private[spark] class KerberosPodWatcherCache(
         while (!kdcIsUp) kdcRunning.await()
         while (!nnIsUp) nnRunning.await()
         while (!dnIsUp) dnRunning.await()
-        while (!hasInLogs(dnName, "Computing capacity for map BlockMap")) {
+        while (!hasInLogs(dnName, "Got finalize command for block pool")) {
           logInfo("Waiting on DN to be formatted")
           Thread.sleep(500)
         }
-        Thread.sleep(2000)
         dpIsUp = true
         logInfo(s"data-populator has signaled")
-
         try {
           dpRunning.signalAll()
         } finally {
@@ -245,6 +243,7 @@ private[spark] class KerberosPodWatcherCache(
       case _ if name.startsWith("data-populator") => "data-populator"
     }
   }
+
   private def hasInLogs(name: String, expectation: String): Boolean = {
     kubernetesClient
       .pods()

resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/kerberos/KerberosPodWatcherCache.scala

@@ -39,16 +39,21 @@ private[spark] class KerberosUtils(
   def loadFromYaml(resource: String): FileInputStream =
     new FileInputStream(new File(yamlLocation(resource)))
   private val regex = "REPLACE_ME".r
+  private val regexDP = "# default_ccache_name = MEMORY".r
+  private val defaultCacheDP = "default_ccache_name = KRBCONF"
   private def locationResolver(loc: String) = s"src/test/resources/$loc"
   private val kerberosFiles = Seq("krb5.conf", "core-site.xml", "hdfs-site.xml")
   private val kerberosConfTupList =
     kerberosFiles.map { file =>
-      (file, regex.replaceAllIn(readFileToString(new File(locationResolver(file))), namespace))}
+      (file, regex.replaceAllIn(readFileToString(new File(locationResolver(file))), namespace))} ++
+      Seq(("krb5-dp.conf", regexDP.replaceAllIn(regex.replaceAllIn(readFileToString(
+        new File(locationResolver("krb5.conf"))), namespace), defaultCacheDP)))
   private val KRB_VOLUME = "krb5-conf"
   private val KRB_FILE_DIR = "/tmp"
   private val KRB_CONFIG_MAP_NAME = "krb-config-map"
   private val PV_LABELS = Map("job" -> "kerberostest")
-  private val keyPaths: Seq[KeyToPath] = kerberosFiles.map(file =>
+  private val keyPaths: Seq[KeyToPath] = (kerberosFiles ++ Seq("krb5-dp.conf"))
+    .map(file =>
     new KeyToPathBuilder()
       .withKey(file)
       .withPath(file)
@@ -102,6 +107,10 @@ private[spark] class KerberosUtils(
                 .withName("TMP_KRB_LOC")
                 .withValue(s"$KRB_FILE_DIR/${kerberosFiles.head}")
                 .endEnv()
+              .addNewEnv()
+                .withName("TMP_KRB_DP_LOC")
+                .withValue(s"$KRB_FILE_DIR/krb5-dp.conf")
+                .endEnv()
               .addNewEnv()
                 .withName("TMP_CORE_LOC")
                 .withValue(s"$KRB_FILE_DIR/${kerberosFiles(1)}")