SparkJob resource support

pom.xml

@@ -336,7 +336,7 @@
       <dependency>
         <groupId>com.squareup.okhttp3</groupId>
         <artifactId>okhttp</artifactId>
-        <version>3.4.1</version>
+        <version>3.6.0</version>
       </dependency>
       <dependency>
         <groupId>org.bouncycastle</groupId>

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/KubernetesClientBuilder.scala → resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/ClientBuilder.scala

@@ -20,11 +20,12 @@ import java.io.File
 
 import com.google.common.base.Charsets
 import com.google.common.io.Files
-import io.fabric8.kubernetes.client.{Config, ConfigBuilder, DefaultKubernetesClient}
+import io.fabric8.kubernetes.client.{BaseClient, Config, ConfigBuilder, DefaultKubernetesClient}
+import okhttp3._
 
 import org.apache.spark.deploy.kubernetes.constants._
 
-private[spark] object KubernetesClientBuilder {
+private[spark] object ClientBuilder {
   private val API_SERVER_TOKEN = new File(Config.KUBERNETES_SERVICE_ACCOUNT_TOKEN_PATH)
   private val CA_CERT_FILE = new File(Config.KUBERNETES_SERVICE_ACCOUNT_CA_CRT_PATH)
 
@@ -34,7 +35,7 @@ private[spark] object KubernetesClientBuilder {
    * are picked up from canonical locations, as they are injected
    * into the pod's disk space.
    */
-  def buildFromWithinPod(
+  def buildK8sClientFromWithinPod(
       kubernetesNamespace: String): DefaultKubernetesClient = {
     var clientConfigBuilder = new ConfigBuilder()
       .withApiVersion("v1")
@@ -51,4 +52,14 @@ private[spark] object KubernetesClientBuilder {
     }
     new DefaultKubernetesClient(clientConfigBuilder.build)
   }
+
+  def buildOkhttpClientFromWithinPod(client: BaseClient): OkHttpClient = {
+    val field = classOf[BaseClient].getDeclaredField("httpClient")
+    try {
+      field.setAccessible(true)
+      field.get(client).asInstanceOf[OkHttpClient]
+    } finally {
+      field.setAccessible(false)
+    }
+  }
 }

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

@@ -0,0 +1,78 @@
+/*
+ * 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
+
+import java.nio.file.{Files, Paths}
+
+import io.fabric8.kubernetes.client.{Config, KubernetesClient}
+import scala.util.{Failure, Success, Try}
+
+import org.apache.spark.deploy.kubernetes.tpr.TPRCrudCalls
+
+private[spark] class SparkJobResourceClientFromWithinK8s(
+    client: KubernetesClient) extends TPRCrudCalls {
+
+  private val protocol: String = "https://"
+
+  // we can also get the host from the environment variable
+  private val kubeHost: String = {
+    val host = Try(sys.env("KUBERNETES_SERVICE_HOST")) match {
+      case Success(h) => Option(h)
+      case Failure(_) => None
+    }
+    host.map(h => h).getOrElse {
+      // Log a warning just in case, but this should almost certainly never happen
+      logWarning("Error while retrieving k8s host address")
+      "127.0.0.1"
+    }
+  }
+
+  // the port from the environment variable
+  private val kubeHostPort: String = {
+    val port = Try(sys.env("KUBERNETES_PORT_443_TCP_PORT")) match {
+      case Success(p) => Option(p)
+      case Failure(_) => None
+    }
+    port.map(p => p).getOrElse {
+      // Log a warning just in case, but this should almost certainly never happen
+      logWarning("Error while retrieving k8s host port")
+      "8001"
+    }
+  }
+
+  // Since this will be running inside a pod
+  // we can access the pods token and use it with the Authorization header when
+  // making rest calls to the k8s Api
+  override protected val kubeToken: Option[String] = {
+    val path = Paths.get(Config.KUBERNETES_SERVICE_ACCOUNT_TOKEN_PATH)
+    Try(new String(Files.readAllBytes(path))) match {
+      case Success(some) => Option(some)
+      case Failure(e: Throwable) => logError(s"${e.getMessage}")
+        None
+    }
+  }
+
+  override protected val k8sClient: KubernetesClient = client
+  override protected val kubeMaster: String = s"$protocol$kubeHost:$kubeHostPort"
+}
+
+private[spark] class SparkJobResourceClientFromOutsideK8s(
+    client: KubernetesClient) extends TPRCrudCalls {
+
+  override protected val k8sClient: KubernetesClient = client
+}

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

@@ -68,4 +68,9 @@ package object constants {
   private[spark] val DRIVER_CONTAINER_NAME = "spark-kubernetes-driver"
   private[spark] val KUBERNETES_SUBMIT_SSL_NAMESPACE = "kubernetes.submit"
   private[spark] val KUBERNETES_MASTER_INTERNAL_URL = "https://kubernetes.default.svc"
+
+  // TPR
+  private[spark] val TPR_API_VERSION = "apache.io/v1"
+  private[spark] val TPR_API_ENDPOINT = s"apis/%s/namespaces/%s/sparkjobs"
+  private[spark] val TPR_KIND = "SparkJob"
 }

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

@@ -0,0 +1,33 @@
+/*
+ * 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.tpr
+
+private[spark] object JobState extends Enumeration {
+  type JobState = Value
+
+  /*
+   * QUEUED - Spark Job has been queued to run
+   * SUBMITTED - Driver Pod deployed but tasks are not yet scheduled on worker pod(s)
+   * RUNNING - Task(s) have been allocated to worker pod(s) to run and Spark Job is now running
+   * FINISHED - Spark Job ran and exited cleanly, i.e, worker pod(s) and driver pod were
+   *            gracefully deleted
+   * FAILED - Spark Job Failed due to error
+   * KILLED - A user manually killed this Spark Job
+   */
+  val QUEUED, SUBMITTED, RUNNING, FINISHED, FAILED, KILLED = Value
+}

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

@@ -0,0 +1,46 @@
+/*
+ * 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.tpr
+
+import org.json4s.{CustomSerializer, JString}
+import org.json4s.JsonAST.JNull
+
+import org.apache.spark.deploy.kubernetes.tpr.JobState.JobState
+
+ /**
+  * JobState Serializer and Deserializer
+  * */
+object JobStateSerDe extends CustomSerializer[JobState](_ =>
+  ({
+    case JString("SUBMITTED") => JobState.SUBMITTED
+    case JString("QUEUED") => JobState.QUEUED
+    case JString("RUNNING") => JobState.RUNNING
+    case JString("FINISHED") => JobState.FINISHED
+    case JString("KILLED") => JobState.KILLED
+    case JString("FAILED") => JobState.FAILED
+    case JNull =>
+      throw new UnsupportedOperationException("No JobState Specified")
+  }, {
+    case JobState.FAILED => JString("FAILED")
+    case JobState.SUBMITTED => JString("SUBMITTED")
+    case JobState.KILLED => JString("KILLED")
+    case JobState.FINISHED => JString("FINISHED")
+    case JobState.QUEUED => JString("QUEUED")
+    case JobState.RUNNING => JString("RUNNING")
+  })
+)