Allow providing an OAuth token for authenticating against k8s

docs/running-on-kubernetes.md

@@ -222,6 +222,14 @@ from the other deployment modes. See the [configuration page](configuration.html
     machine's disk.
   </td>
 </tr>
+<tr>
+  <td><code>spark.kubernetes.submit.oauthToken</code></td>
+  <td>(none)</td>
+  <td>
+    OAuth token to use when authenticating against the against the Kubernetes API server. Note that unlike the other
+    authentication options, this should be the exact string value of the token to use for the authentication.
+  </td>
+</tr>
 <tr>
   <td><code>spark.kubernetes.submit.serviceAccountName</code></td>
   <td><code>default</code></td>

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

@@ -21,6 +21,7 @@ import java.security.SecureRandom
 import java.util.ServiceLoader
 import java.util.concurrent.{CountDownLatch, TimeUnit}
 
+import com.google.common.base.Charsets
 import com.google.common.io.Files
 import com.google.common.util.concurrent.SettableFuture
 import io.fabric8.kubernetes.api.model._
@@ -131,6 +132,11 @@ private[spark] class Client(
     sparkConf.get(KUBERNETES_CLIENT_CERT_FILE).foreach {
       f => k8ConfBuilder = k8ConfBuilder.withClientCertFile(f)
     }
+    sparkConf.get(KUBERNETES_OAUTH_TOKEN).foreach { token =>
+      k8ConfBuilder = k8ConfBuilder.withOauthToken(token)
+      // Remove the oauth token from Spark conf so that its doesn't appear in the Spark UI.
+      sparkConf.set(KUBERNETES_OAUTH_TOKEN, "<present_but_redacted>")
+    }
 
     val k8ClientConfig = k8ConfBuilder.build
     Utils.tryWithResource(new DefaultKubernetesClient(k8ClientConfig)) { kubernetesClient =>

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

@@ -83,6 +83,18 @@ package object config {
       .stringConf
       .createOptional
 
+  private[spark] val KUBERNETES_OAUTH_TOKEN =
+    ConfigBuilder("spark.kubernetes.submit.oauthToken")
+      .doc("""
+          | OAuth token to use when authenticating against the
+          | against the Kubernetes API server. Note that unlike
+          | the other authentication options, this should be the
+          | exact string value of the token to use for the
+          | authentication.
+        """.stripMargin)
+      .stringConf
+      .createOptional
+
   private[spark] val KUBERNETES_SERVICE_ACCOUNT_NAME =
     ConfigBuilder("spark.kubernetes.submit.serviceAccountName")
       .doc("""