Support using PEM files to configure SSL for driver submission (apache-spark-on-k8s#173)

* Support configuring SSL using PEM files.

* Address some missed comments

* Fix import ordering

* Slight rewording of comments

* Fix scalastyle

(cherry picked from commit 078697f)

Author: mccheah

docs/running-on-kubernetes.md

@@ -127,16 +127,24 @@ Spark supports using TLS to encrypt the traffic in this bootstrapping process. I
 whenever possible. 
 
 See the [security page](security.html) and [configuration](configuration.html) sections for more information on
-configuring TLS; use the prefix `spark.ssl.kubernetes.submission` in configuring the TLS-related fields in the context
+configuring TLS; use the prefix `spark.ssl.kubernetes.driversubmitserver` in configuring the TLS-related fields in the context
 of submitting to Kubernetes. For example, to set the trustStore used when the local machine communicates with the driver
-pod in starting the application, set `spark.ssl.kubernetes.submission.trustStore`.
+pod in starting the application, set `spark.ssl.kubernetes.driversubmitserver.trustStore`.
 
 One note about the keyStore is that it can be specified as either a file on the client machine or a file in the
-container image's disk. Thus `spark.ssl.kubernetes.submission.keyStore` can be a URI with a scheme of either `file:`
+container image's disk. Thus `spark.ssl.kubernetes.driversubmitserver.keyStore` can be a URI with a scheme of either `file:`
 or `local:`. A scheme of `file:` corresponds to the keyStore being located on the client machine; it is mounted onto
 the driver container as a [secret volume](https://kubernetes.io/docs/user-guide/secrets/). When the URI has the scheme
 `local:`, the file is assumed to already be on the container's disk at the appropriate path.
 
+Finally, the submission server and client can be configured to use PEM files instead of Java keyStores. When using
+this mode, set `spark.ssl.kubernetes.driversubmitserver.keyPem` and
+`spark.ssl.kubernetes.driversubmitserver.serverCertPem` to configure the key and certificate files on the driver
+submission server. These files can be uploaded from the submitter's machine if they have no scheme or a scheme of
+`file:`, or they can be located on the container's disk if they have the scheme `local:`. The client's certificate
+file should be provided via setting `spark.ssl.kubernetes.driversubmitserver.clientCertPem`, and this file must be
+located on the submitting machine's local disk.
+
 ### Submission of Local Files through Ingress/External controller
 
 Kubernetes pods run with their own IP address space. If Spark is run in cluster mode, the driver pod may not be

resource-managers/kubernetes/core/pom.xml

@@ -87,6 +87,10 @@
       <groupId>com.google.guava</groupId>
       <artifactId>guava</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.bouncycastle</groupId>
+      <artifactId>bcpkix-jdk15on</artifactId>
+    </dependency>
     <!-- End of shaded deps. -->
 
   </dependencies>

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

@@ -161,7 +161,7 @@ private[spark] class Client(
           driverServiceManager.handleSubmissionError(
             new SparkException("Submission shutting down early...")))
         try {
-          val sslConfigurationProvider = new SslConfigurationProvider(
+          val sslConfigurationProvider = new DriverSubmitSslConfigurationProvider(
             sparkConf, kubernetesAppId, kubernetesClient, kubernetesResourceCleaner)
           val submitServerSecret = kubernetesClient.secrets().createNew()
             .withNewMetadata()
@@ -182,7 +182,7 @@ private[spark] class Client(
           configureOwnerReferences(
             kubernetesClient,
             submitServerSecret,
-            sslConfiguration.sslSecrets,
+            sslConfiguration.sslSecret,
             driverPod,
             driverService)
           submitApplicationToDriverServer(
@@ -209,7 +209,6 @@ private[spark] class Client(
           Utils.tryLogNonFatalError {
             driverServiceManager.stop()
           }
-
           // Remove the shutdown hooks that would be redundant
           Utils.tryLogNonFatalError {
             ShutdownHookManager.removeShutdownHook(resourceCleanShutdownHook)
@@ -236,7 +235,7 @@ private[spark] class Client(
   private def submitApplicationToDriverServer(
       kubernetesClient: KubernetesClient,
       driverServiceManager: DriverServiceManager,
-      sslConfiguration: SslConfiguration,
+      sslConfiguration: DriverSubmitSslConfiguration,
       driverService: Service,
       submitterLocalFiles: Iterable[String],
       submitterLocalJars: Iterable[String],
@@ -298,7 +297,7 @@ private[spark] class Client(
       customLabels: Map[String, String],
       customAnnotations: Map[String, String],
       submitServerSecret: Secret,
-      sslConfiguration: SslConfiguration): (Pod, Service) = {
+      sslConfiguration: DriverSubmitSslConfiguration): (Pod, Service) = {
     val driverKubernetesSelectors = (Map(
       SPARK_DRIVER_LABEL -> kubernetesAppId,
       SPARK_APP_ID_LABEL -> kubernetesAppId,
@@ -349,7 +348,7 @@ private[spark] class Client(
   private def configureOwnerReferences(
       kubernetesClient: KubernetesClient,
       submitServerSecret: Secret,
-      sslSecrets: Array[Secret],
+      sslSecret: Option[Secret],
       driverPod: Pod,
       driverService: Service): Service = {
     val driverPodOwnerRef = new OwnerReferenceBuilder()
@@ -359,7 +358,7 @@ private[spark] class Client(
       .withKind(driverPod.getKind)
       .withController(true)
       .build()
-    sslSecrets.foreach(secret => {
+    sslSecret.foreach(secret => {
       val updatedSecret = kubernetesClient.secrets().withName(secret.getMetadata.getName).edit()
         .editMetadata()
         .addToOwnerReferences(driverPodOwnerRef)
@@ -425,10 +424,10 @@ private[spark] class Client(
       driverKubernetesSelectors: Map[String, String],
       customAnnotations: Map[String, String],
       submitServerSecret: Secret,
-      sslConfiguration: SslConfiguration): Pod = {
+      sslConfiguration: DriverSubmitSslConfiguration): Pod = {
     val containerPorts = buildContainerPorts()
     val probePingHttpGet = new HTTPGetActionBuilder()
-      .withScheme(if (sslConfiguration.sslOptions.enabled) "HTTPS" else "HTTP")
+      .withScheme(if (sslConfiguration.enabled) "HTTPS" else "HTTP")
       .withPath("/v1/submissions/ping")
       .withNewPort(SUBMISSION_SERVER_PORT_NAME)
       .build()
@@ -452,7 +451,7 @@ private[spark] class Client(
             .withSecretName(submitServerSecret.getMetadata.getName)
             .endSecret()
           .endVolume()
-        .addToVolumes(sslConfiguration.sslPodVolumes: _*)
+        .addToVolumes(sslConfiguration.sslPodVolume.toSeq: _*)
         .withServiceAccount(serviceAccount.getOrElse("default"))
         .addNewContainer()
           .withName(DRIVER_CONTAINER_NAME)
@@ -463,7 +462,7 @@ private[spark] class Client(
             .withMountPath(secretDirectory)
             .withReadOnly(true)
             .endVolumeMount()
-          .addToVolumeMounts(sslConfiguration.sslPodVolumeMounts: _*)
+          .addToVolumeMounts(sslConfiguration.sslPodVolumeMount.toSeq: _*)
           .addNewEnv()
             .withName(ENV_SUBMISSION_SECRET_LOCATION)
             .withValue(s"$secretDirectory/$SUBMISSION_APP_SECRET_NAME")
@@ -661,7 +660,7 @@ private[spark] class Client(
       kubernetesClient: KubernetesClient,
       driverServiceManager: DriverServiceManager,
       service: Service,
-      sslConfiguration: SslConfiguration): KubernetesSparkRestApi = {
+      sslConfiguration: DriverSubmitSslConfiguration): KubernetesSparkRestApi = {
     val serviceUris = driverServiceManager.getDriverServiceSubmissionServerUris(service)
     require(serviceUris.nonEmpty, "No uris found to contact the driver!")
     HttpClientUtil.createClient[KubernetesSparkRestApi](