Stage 3 Integration tests pass

ifilonenko · ifilonenko · commit c6b11f86839a · 2017-08-03T15:17:07.000-07:00
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesClusterSchedulerBackend.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesClusterSchedulerBackend.scala
@@ -132,8 +132,6 @@ private[spark] class KubernetesClusterSchedulerBackend(
   private implicit val requestExecutorContext = ExecutionContext.fromExecutorService(
     ThreadUtils.newDaemonCachedThreadPool("kubernetes-executor-requests"))
 
-  private val maybeMountedHadoopSecret = conf.get(MOUNTED_HADOOP_SECRET_CONF)
-
   private val driverPod = try {
     kubernetesClient.pods().inNamespace(kubernetesNamespace).
       withName(kubernetesDriverPodName).get()
diff --git a/resource-managers/kubernetes/integration-tests-spark-jobs/src/main/scala/org/apache/spark/deploy/kubernetes/integrationtest/jobs/HDFSTest.scala b/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"File contents: ${file.map(s => s.toString).collect().mkString(",")}")
     println(s"Returned length(s) of: ${file.map(s => s.length).collect().mkString(",")}")
     // scalastyle:on println
     spark.stop()
diff --git a/resource-managers/kubernetes/integration-tests/kerberos-yml/test-env.sh b/resource-managers/kubernetes/integration-tests/kerberos-yml/test-env.sh
@@ -19,6 +19,7 @@ until /usr/bin/kinit -kt /var/keytabs/hdfs.keytab hdfs/nn.${NAMESPACE}.svc.clust
       --conf spark.kubernetes.executor.docker.image=spark-executor:latest \
       --conf spark.kubernetes.initcontainer.docker.image=spark-init:latest \
       --conf spark.hadoop.fs.defaultFS=hdfs://nn.${NAMESPACE}.svc.cluster.local:9000 \
+      --conf spark.hadoop.dfs.data.transfer.protection=authentication \
       --conf spark.kubernetes.kerberos=true \
       --conf spark.kubernetes.kerberos.keytab=/var/keytabs/hdfs.keytab \
       --conf spark.kubernetes.kerberos.principal=hdfs/nn.${NAMESPACE}.svc.cluster.local@CLUSTER.LOCAL \
diff --git a/resource-managers/kubernetes/integration-tests/src/test/resources/core-site.xml b/resource-managers/kubernetes/integration-tests/src/test/resources/core-site.xml
@@ -31,4 +31,8 @@
     <name>fs.defaultFS</name>
     <value>hdfs://nn.REPLACE_ME.svc.cluster.local:9000</value>
   </property>
+  <property>
+    <name>hadoop.rpc.protection</name>
+    <value>authentication</value>
+  </property>
 </configuration>
diff --git a/resource-managers/kubernetes/integration-tests/src/test/resources/hdfs-site.xml b/resource-managers/kubernetes/integration-tests/src/test/resources/hdfs-site.xml
@@ -67,9 +67,8 @@
   <!-- DataNode security config -->
   <property>
     <name>dfs.data.transfer.protection</name>
-    <value>authentication</value>
+    <value>integrity</value>
   </property>
-   
   <property>
     <name>dfs.datanode.address</name>
     <value>0.0.0.0:10019</value>
@@ -128,6 +127,18 @@
     <name>dfs.datanode.kerberos.principal</name>
     <value>hdfs/dn1.REPLACE_ME.svc.cluster.local@CLUSTER.LOCAL</value>
   </property>
+  <property>
+    <name>dfs.encrypt.data.transfer</name>
+    <value>true</value>
+  </property>
+  <property>
+    <name>dfs.encrypt.data.transfer.cipher.suites</name>
+    <value>AES/CTR/NoPadding</value>
+  </property>
+  <property>
+    <name>dfs.encrypt.data.transfer.cipher.key.bitlength</name>
+    <value>256</value>
+  </property>
 
   <!-- Web Authentication config -->
   <property>
diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/KubernetesSuite.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/KubernetesSuite.scala
@@ -78,7 +78,7 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
   }
 
   after {
-    kubernetesTestComponents.deleteKubernetesResources()
+    kubernetesTestComponents.deleteKubernetesPVs()
     kubernetesTestComponents.deleteNamespace()
   }
 
@@ -105,7 +105,9 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter {
       Map("spark-app-locator" -> APP_LOCATOR_LABEL))
     driverWatcherCache.start()
     driverWatcherCache.stop()
-    val expectedLogOnCompletion = Seq("Returned length(s) of:")
+    val expectedLogOnCompletion = Seq(
+        "Returned length(s) of: 1",
+        "File contents: [This is an awesome word count file]")
     val driverPod = kubernetesClient
       .pods()
       .withLabel("spark-app-locator", APP_LOCATOR_LABEL)
diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/KubernetesTestComponents.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/KubernetesTestComponents.scala
@@ -39,7 +39,21 @@ private[spark] class KubernetesTestComponents(defaultClient: DefaultKubernetesCl
       .endMetadata()
       .done()
   }
-
+  def deleteKubernetesPVs(): Unit = {
+      // Temporary hack until client library for fabric8 is updated to get around
+      // the NPE that comes about when I do .list().getItems().asScala
+      try {
+        val pvList = kubernetesClient.persistentVolumes().list().getItems().asScala
+        if (pvList.nonEmpty) {
+          kubernetesClient.persistentVolumes().delete()
+          Eventually.eventually(KubernetesSuite.TIMEOUT, KubernetesSuite.INTERVAL) {
+            require(!pvList.exists(_.getMetadata.getNamespace == namespace))
+          }
+        }
+      } catch {
+        case ex: java.lang.NullPointerException =>
+      }
+  }
   def deleteNamespace(): Unit = {
     defaultClient.namespaces.withName(namespace).delete()
     Eventually.eventually(KubernetesSuite.TIMEOUT, KubernetesSuite.INTERVAL) {
@@ -52,27 +66,6 @@ private[spark] class KubernetesTestComponents(defaultClient: DefaultKubernetesCl
     }
   }
 
-  def deleteKubernetesResources(): Unit = {
-    kubernetesClient.persistentVolumes().delete()
-    Eventually.eventually(KubernetesSuite.TIMEOUT, KubernetesSuite.INTERVAL) {
-      val persistentList = kubernetesClient
-        .persistentVolumes()
-        .list()
-        .getItems()
-        .asScala
-      require(!persistentList.exists(_.getMetadata.getNamespace == namespace))
-    }
-    kubernetesClient.configMaps().delete()
-    Eventually.eventually(KubernetesSuite.TIMEOUT, KubernetesSuite.INTERVAL) {
-      val configMapsList = kubernetesClient
-        .configMaps()
-        .list()
-        .getItems()
-        .asScala
-      require(!configMapsList.exists(_.getMetadata.getNamespace == namespace))
-    }
-  }
-
   def newSparkConf(): SparkConf = {
     new SparkConf(true)
       .setMaster(s"k8s://${kubernetesClient.getMasterUrl}")
diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/docker/SparkDockerImageBuilder.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/docker/SparkDockerImageBuilder.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 = {

Original file line number	Diff line number	Diff line change
`@@ -40,6 +40,7 @@ private[spark] object HDFSTest{`
`40`	`40`	`val end = System.currentTimeMillis()`
`41`	`41`	`println("Iteration " + iter + " took " + (end-start) + " ms")`
`42`	`42`	`}`
	`43`	`+ println(s"File contents: ${file.map(s => s.toString).collect().mkString(",")}")`
`43`	`44`	`println(s"Returned length(s) of: ${file.map(s => s.length).collect().mkString(",")}")`
`44`	`45`	`// scalastyle:on println`
`45`	`46`	`spark.stop()`