SparkJob resource support #126

iyanuobidele · 2017-02-17T16:21:32Z

This will eventually close #111.

Initial implementation of the CRUD + Watch calls for the SparkJob resource. This will be the base I'll be making patches against.

/cc
@foxish @mccheah @ash211 @erikerlandson @ssuchter

iyanuobidele · 2017-02-17T16:23:37Z

...ers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/tpr/TPRCrudCalls.scala

+import org.apache.spark.SparkException
+import org.apache.spark.util.ThreadUtils
+
+/**


will take this comment out

iyanuobidele · 2017-02-17T16:24:38Z

sbin/driver.yaml

+spec:
+  image: "driver-image"
+  state: "completed"
+  num-executors: 10


iyanuobidele · 2017-02-17T16:24:48Z

sbin/kubernetes-resource.yaml

+kind: ThirdPartyResource
+description: "A resource that manages a spark job"
+versions:
+  - name: v1


iyanuobidele · 2017-02-17T16:25:43Z

...ers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/tpr/TPRCrudCalls.scala

+  protected val kubeToken: Option[String] = None
+
+  implicit val formats: Formats = DefaultFormats + JobStateSerDe
+


one extra "/n" to take out

iyanuobidele · 2017-02-17T16:26:47Z

...rs/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/tpr/JobStateSerDe.scala

+    case JobState.QUEUED => JString("QUEUED")
+    case JobState.RUNNING => JString("RUNNING")
+  })
+)


iyanuobidele · 2017-02-17T16:27:01Z

...anagers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/tpr/JobState.scala

+   * KILLED - A user manually killed this Spark Job
+   */
+  val QUEUED, SUBMITTED, RUNNING, FINISHED, FAILED, KILLED = Value
+}


ash211 · 2017-02-17T16:31:27Z

...nagers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/ClientBuilder.scala

  }
+
+  def buildOkhttpClientFromWithinPod(client: BaseClient): OkHttpClient = {
+    val field = classOf[BaseClient].getDeclaredField("httpClient")


why do we need this reflection to grab the httpClient out of the BaseClient? Is this what we have to do because the fabric8 client doesn't support TPR yet?

Yes you're exactly right.

Can we move forward the Kubernetes client to support third party resources?

ash211 · 2017-02-17T16:36:45Z

...rs/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/tpr/JobStateSerDe.scala

+    case JString("RUNNING") => JobState.RUNNING
+    case JString("FINISHED") => JobState.FINISHED
+    case JString("KILLED") => JobState.KILLED
+    case JString("FAILED") => JobState.FAILED


there's probably a nicer way to do this rather than the manual mapping -- I think Scala case classes have an apply method that creates the class you expect?

ash211 · 2017-02-17T16:37:41Z

...ers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/tpr/TPRCrudCalls.scala

+  def getJobObject(name: String): SparkJobState = {
+    val request = completeRequest(new Request.Builder()
+      .get()
+      .url(s"$kubeMaster/${TPR_API_ENDPOINT.format(TPR_API_VERSION, namespace)}/$name"))


pull these URLs out to statics. Maybe use Feign with the JAX-RS annotations instead?

Sounds good. Thanks for the quick review. I'll look into your comments.

iyanuobidele · 2017-02-17T16:57:24Z

...ers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/tpr/TPRCrudCalls.scala

+    * This method relies on the assumption of one sparkjob per namespace
+    */
+  def watchJobObject(): Future[WatchObject] = {
+    val watchClient = httpClient.newBuilder().readTimeout(0, TimeUnit.MILLISECONDS).build()


this should probably be a clone not newBuilder

foxish · 2017-02-17T17:38:40Z

Thanks @iyanuobidele!
A quick comment on the schema we want.
We want to follow the conventions of the other Kubernetes API objects, which means having all the items we discussed in #111 to be nested under a status field.
The way I imagine it is:

apiVersion: "apache.org/v1" # apache.org, conforming with the project itself.
kind: "SparkJob"
metadata:
  name: "spark-job-1"
status:
  image: "driver-image"
  state: "completed"
  numExecutors: 10 # camelcase, instead of num-executors
  ... 
  ... 
  # other status items

I think it previously was nested under spec when we thought we'd want those fields to be mutable. The current thinking is to first use the TPR for reporting status via kubectl and the dashboard, and later extending it to other items. This will still give us the opportunity in future to add spec and expose some other fields there at a later stage.

mccheah

This review is at a first glance - much of this code assumes the Kubernetes client cannot support third party resources directly. We should look into contributing to the Kubernetes client project. If that doesn't make sense for our use case then we can move forward with this implementation.

mccheah · 2017-02-17T20:25:48Z

...nagers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/ClientBuilder.scala

 import org.apache.spark.deploy.kubernetes.constants._

-private[spark] object KubernetesClientBuilder {
+private[spark] object ClientBuilder {


Any particular reason for the name change?

mccheah · 2017-02-17T20:27:02Z

...s/core/src/main/scala/org/apache/spark/deploy/kubernetes/SparkJobResourceFromWithinK8s.scala

+  private val protocol: String = "https://"
+
+  // we can also get the host from the environment variable
+  private val kubeHost: String = {


Use what we already have in constants.scala, which provides a hostname that can be DNS resolved within a pod.

mccheah · 2017-02-17T20:27:42Z

...s/core/src/main/scala/org/apache/spark/deploy/kubernetes/SparkJobResourceFromWithinK8s.scala

+      case Failure(_) => None
+    }
+    host.map(h => h).getOrElse {
+      // Log a warning just in case, but this should almost certainly never happen


Throwing an exception here would be best, but this shouldn't be necessary once we use the proper hostname.

mccheah · 2017-02-17T20:28:08Z

...s/core/src/main/scala/org/apache/spark/deploy/kubernetes/SparkJobResourceFromWithinK8s.scala

+
+  // the port from the environment variable
+  private val kubeHostPort: String = {
+    val port = Try(sys.env("KUBERNETES_PORT_443_TCP_PORT")) match {


I believe using the hostname with DNS resolution removes the need to specify the port.

mccheah · 2017-02-17T20:30:52Z

...ers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/tpr/TPRCrudCalls.scala

+  protected implicit val ec: ThreadPoolExecutor = ThreadUtils
+    .newDaemonCachedThreadPool("tpr-watcher-pool")
+
+  private def executeBlocking(cb: => WatchObject): Future[WatchObject] = {


This once again seems like something the kubernetes client should provide for us. Although, it's unclear how the kubernetes client would be able to support completely arbitrary resource types.

Thanks Matt for taking a look at this. This is mostly a port of work we did a couple months ago. At the time, we considered making a contribution upstream, but we decided to go the crud way to save some cycles.

I'll spend sometime looking into what it'll take to get a contribution upstream and maybe we could also talk about this at the next sig.

iyanuobidele · 2017-02-18T00:46:11Z

Sure @foxish, I'll take note of that. I'll be making patches to this shortly.

ash211 · 2017-03-16T08:32:11Z

Please rebase onto branch-2.1-kubernetes and send this PR into that branch instead of k8s-support-alternate-incremental which is now deprecated.

foxish · 2017-04-25T00:05:53Z

@iyanuobidele, will you have time to pick this PR up again?

iyanuobidele · 2017-05-19T02:29:10Z

closing this and opening a new PR to cover this.

foxish and others added 4 commits February 16, 2017 16:37

[revert] adding new yamls for TPR

3776898

initial port from foxish-fork

f339dc2

second iteration from foxish-fork

1fd1633

initial sparkResource implementation

c198d9f

iyanuobidele commented Feb 17, 2017

View reviewed changes

ash211 reviewed Feb 17, 2017

View reviewed changes

iyanuobidele commented Feb 17, 2017

View reviewed changes

mccheah suggested changes Feb 17, 2017

View reviewed changes

mccheah mentioned this pull request Feb 22, 2017

K8s support watch tpr foxish/spark#12

Closed

foxish mentioned this pull request Feb 24, 2017

Update fabric8 kubernetes-client with support for TPR #153

Closed

iyanuobidele force-pushed the tpr-support branch 2 times, most recently from 05d8885 to c198d9f Compare May 1, 2017 21:08

iyanuobidele closed this May 19, 2017

iyanuobidele mentioned this pull request May 19, 2017

TPR Support #284

Closed

		protected val kubeToken: Option[String] = None

		implicit val formats: Formats = DefaultFormats + JobStateSerDe

SparkJob resource support #126

SparkJob resource support #126

Uh oh!

Conversation

iyanuobidele commented Feb 17, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

foxish commented Feb 17, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mccheah left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

iyanuobidele commented Feb 18, 2017

Uh oh!

ash211 commented Mar 16, 2017

Uh oh!

foxish commented Apr 25, 2017

Uh oh!

iyanuobidele commented May 19, 2017

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

foxish commented Feb 17, 2017 •

edited

Loading