Revisit submission process after alpha release

[ash211]

The submission client has grown quite complicated for really just posting files/jars to a file server and running those files. Maybe we can use what we've learned from this alpha release to rearchitect the submission process for local resources.

Desired gains would be:

• simpler submission process
• make pods less stateful so k8s can restart the driver on failure
• eliminate OOM concerns from the submission server (OOM in Rest server when launching a job with large resources #146)

[mccheah]

@foxish I would think relying on volumes would be the right way to go, but my understanding of volumes is that they have to be backed by some kind of network storage and the burden is on the user to create these storage units themselves. I was wondering if Kubernetes has a built-in default storage for assets that can be uploaded and accessed via API server calls. For example, ConfigMap objects allow for storing arbitrary properties that can then be mounted as a volume into any given pod. ConfigMap appears to be strictly for lightweight content however, such as for application configuration - hence the name of the construct. Jars are too heavyweight for ConfigMaps. YARN's equivalent of this requires a hadoop filesystem API compatible storage area, such as an HDFS cluster.

We could make Spark deploy its own temporary file server. The submission process would turn into deploying the file server, shipping the content to it, then starting the driver pod which pulls the requisite files from the file server. The advantage of this over our current approach is twofold. Firstly, it separates the concerns of jar storage and application startup. Secondly, if we use a stock file server image (or even allow the user to provide their own) we don't have to worry about the data storage at all apart from how to give it the data.

If we could use the API server directly, however, this makes it a lot easier to handle security. Right now a lot of the complexity in the submission code is due to setting up SSL to talk to the driver. We would need a similar setup for SSL to talk to a temporary file server, and authentication in front of the file server so that only the submitting application can access its contents. If we were to rely on a Kubernetes resource in a Kubernetes namespace, we could rely on the Kubernetes primitives to handle the security for us, and we can re-use the SSL certificate that the user had given us to set up the Kubernetes client in general.

[mccheah]

Here's some ideas we could think about:

Current Implementation

Submission is done in two phases:

1. Create the driver pod and expose it on a front door that is reachable by the submission client
2. Send the driver pod dependencies from the user's local disk as well as the Spark application configuration, and start the job

Current Implementation Problems

The current implementation is very complex. Here's a list of the things we've run into while developing it:

1. Securing the submission step is very difficult — Supporting SSL end to end requires bootstrapping keystores / PEM files on every submission
2. Difficult to have a generic way to expose the driver pod on a front door (that is, outside the cluster) - need the extra abstraction layer of DriverServiceManager to capture annotation-based + other ways of exposing the driver pod
3. Unable to handle driver pod restart, since it holds state from the submitter that can't be recreated without the submitter re-submitting.

Goal of an ideal solution

The goal of an ideal solution would be to remove the need to expose the driver pod on a front door at all. This would greatly reduce the complexity of the submission process, removing both of the above concerns.

Proposed Alternative: File Staging Server

An alternative solution is to deploy a separate file staging server that can be shared by all Spark jobs, and use that to bootstrap Spark applications that run on the cluster. The approach works as follows:

1. Use kubectl or a custom script packaged with Spark to deploy a file staging server..
   1. The staging server would support two methods:
      1. registerDependencies(appId, dependencyStream, applicationKubernetesCredentials): Token
         1. Registers an application to the submission server. Note that this does not start the application itself. The file server stores the information on its local disk and returns a unique secure token that the driver can use for retrieving these dependencies. The appId is checked to make sure an application with a given ID doesn't register dependencies twice.
      2. getDependencies(token): (KubernetesCredentials, DependencyStream)
         1. Retrieve the dependencies for a given application. The token indicates which application's dependencies should be fetched.
   2. Note that we get and put the Kubernetes Credentials as well in order to bypass needing to use secrets to get the driver pod's kubernetes credentials (see Allow the driver pod's credentials to be shipped from the submission client (squashed) #192). This also helps for the file server being able to watch the API server for the driver pod's status and to handle cleaning up when the application terminates (see "Open Questions" below)
   3. One neat idea is to use kubectl entirely instead of using any custom scripts for this
      1. Support for customizing the REST server can be provided using a ConfigMap, where the keys of the config map are documented
      2. We can provide a standard YML file describing the standard deployment of one of these REST servers out of the box
      3. We can provide alternative YML files that can be used as guidelines for setting it up to use SSL
2. The spark submission client pushes local dependencies to the staging server and starts the driver pod
   1. The submission client is given the staging server URL in its configuration, say spark.kubernetes.driver.stagingServerUrl
      1. This URL can be arbitrary, as long as it routes to a staging server
      2. Users have complete flexibility as to how they've exposed their staging server that they created above.
      3. It also becomes one-time setup instead of every-job setup provide routing into the cluster
      4. The staging server's port should be well-documented so that it's easy for users to reason about exposing it
         1. We can use a named port on the pod
         2. Provide a service YML file that lines up nicely with the file server's YML file that exposes the file server over NodePort
   2. The submission client generates a token and pushes local dependencies + driver pod kubernetes credentials to the staging server using the registerDependencies endpoint.
      1. We still need to consider client-side SSL configurations here, although at this point it's just a matter of providing a certificate file / truststore JKS
   3. The submission client creates the driver pod which pulls its dependencies
      1. Dependency pulling from the file server can be done using an init-container
         1. The init-container is configured with the file server URL and the application token. The former is passed in an environment variable and the latter is given as a Kubernetes secret.
         2. The init-container downloads the dependencies to an external dependencies folder
         3. If the init-container needs to talk to the file server over SSL, it will need to be handed the certificate using a Kubernetes secret
      2. The main driver container runs the main application class with the classpath constructed as the external dependencies folder, plus jars that were pre-mounted on the docker image (i.e. local:// dependencies)
         1. Passing Spark properties is tricky
            1. We would prefer to run just straightforward Java in the driver container, making it easy to customize and not be tied to needing to run a Spark-specific class
            2. Need to pass environment variables denoting what the system properties and the classpath are - the driver submission client will need to build this given the dependencies the init-container downloaded, plus container-local dependencies - and set -D flags for the Spark system properties required by the application's SparkConf.

Advantages Of Suggested V2

• Driver docker image can be reduced to just packaging Spark + an intelligently formatted Java command, eliminating the rest submission server
• No need for code that exposes an arbitrary pod, which we have seen is a problem that is difficult to overcome in a fully generic way
• SSL into the cluster is set up once for all applications, on the staging server
  • Each application doesn't have to worry about provisioning its own key and certificate that is specific to the application
  • SSL is configured once on the file staging server and the certificate can just be distributed for use across all Spark applications
• Driver pod can be restarted and the init-container fetches the lost dependencies automatically, allowing recovery from failed driver pods
• Staging server can be hosted inside or outside the k8s cluster
• Queued submission allows jobs to queue on the file staging server before running in the cluster

Disadvantages Of Suggested V2

• Requires a separate step to specifically be run by the cluster administrator
• Difficult to ensure high availability for the file server
  • Ideally we could run N file servers and contacting any of them for pushing and pulling would achieve the same result. Unfortunately the N file servers would need to all agree on the applications that they are recording.
• Large files go through two hops (submission -> staging server -> driver) instead of one hop currently (submission -> driver)

Open Questions

• When do we clean up the data on the file server?
  • A naive implementation cleans up as soon as the application has finished the init-container. However, this makes it impossible for retries to fetch the dependencies again.
  • Idea: File server watches the Kubernetes API for the lifecycle of driver pods
    • When the file server first receives dependencies, start a timer and watch the API server for pods with the expected application ID
    • If the driver pod doesn't start in some time T, evict the files
    • Otherwise, keep the files around as long as the driver pod remains alive
    • When the driver pod exits, continue monitoring the API server for pods with the given label matching the app ID
    • If no retries enter the cluster after some time T, evict the files
    • This can be done with the suggested API since the file server will receive the credentials that can be used to watch the driver pod.
• How do we configure the SSL certificates the staging server uses?

[kimoonkim]

I just found out that k8s version 1.5 supports kubectl cp command.

~$ kubectl cp -h
Copy files and directories to and from containers.

Examples:
  # !!!Important Note!!!
  # Requires that the 'tar' binary is present in your container
  # image.  If 'tar' is not present, 'kubectl cp' will fail.

  # Copy /tmp/foo_dir local directory to /tmp/bar_dir in a remote pod in the default namespace
  kubectl cp /tmp/foo_dir <some-pod>:/tmp/bar_dir

  # Copy /tmp/foo local file to /tmp/bar in a remote pod in a specific container
  kubectl cp /tmp/foo <some-pod>:/tmp/bar -c <specific-container>

  # Copy /tmp/foo local file to /tmp/bar in a remote pod in namespace <some-namespace>
  kubectl cp /tmp/foo <some-namespace>/<some-pod>:/tmp/bar

  # Copy /tmp/foo from a remote pod to /tmp/bar locally
  kubectl cp <some-namespace>/<some-pod>:/tmp/foo /tmp/bar

Options:
  -c, --container='': Container name. If omitted, the first container in the pod will be chosen

Usage:
  kubectl cp <file-spec-src> <file-spec-dest> [options]

Use "kubectl options" for a list of global command-line options (applies to all commands).
~$ kubectl version
Client Version: version.Info{Major:"1", Minor:"5", GitVersion:"v1.5.4", GitCommit:"7243c69eb523aa4377bce883e7c0dd76b84709a1", GitTreeState:"clean", BuildDate:"2017-03-07T23:53:09Z", GoVersion:"go1.7.4", Compiler:"gc", Platform:"darwin/amd64"}
Server Version: version.Info{Major:"1", Minor:"5", GitVersion:"v1.5.4", GitCommit:"7243c69eb523aa4377bce883e7c0dd76b84709a1", GitTreeState:"clean", BuildDate:"2017-03-07T23:34:32Z", GoVersion:"go1.7.4", Compiler:"gc", Platform:"linux/amd64"}

Here is the upstream issue discussing it. Any chance we can piggyback on the underlying mechanism?

[mccheah]

I think the problem with this is that the files need to be present at the time the driver starts running. kubectl cp would only bootstrap the data after the driver has started. We could, for example, have an init-container that stalls until certain files have been sent up to the driver via something like kubectl cp, but I would prefer a more explicit mechanism than that.

[ash211]

From @erikerlandson : ideally when running a job with no local dependencies (e.g. files/jars are baked into the image or hosted in an accessible location e.g. hdfs:// or https://), we wouldn't require a staging server to be present to launch the job.

I'd agree with this -- seems like the file server should be hosting only "heavyweight" data, so jobs that have only "lightweight" startup configuration can avoid the hit to a file server entirely. This allows sites to choose not to run the file server entirely.

[foxish]

Went over the proposal in detail (sorry about the delay)
This is a great improvement.

A few comments:

• Following our talk on making the staging server a more generic component, being able to execute kubectl cp -h and somehow setting ACLs on the uploaded file might be useful. I mean copying files not into the driver pod, which has the flaw you pointed out in Revisit submission process after alpha release #167 (comment), but into the staging server. I'll be opening up an upstream issue to facilitate something like this in some "recommended" fashion and will link this.
• There needs to be a timeout/expiry for the files and periodic cleanup in the staging server. (maybe this is in the PRs already)
• In the long term, we need some way to monitor the staging server and see how well it's doing (in terms of disk space and seeing which "applications" have submitted files).

[mccheah]

Following our talk on making the staging server a more generic component, being able to execute kubectl cp -h and somehow setting ACLs on the uploaded file might be useful. I mean copying files not into the driver pod, which has the flaw you pointed out in #167 (comment), but into the staging server. I'll be opening up an upstream issue to facilitate something like this in some "recommended" fashion and will link this.

What are your thoughts on the tradeoffs between using kubectl cp versus using a rest API for uploads? Using kubectl cp introduces asymmetry in that downloads are over REST while uploads are over this kubectl cp mechanism. And if using kubectl cp how does the staging server know the paths to provide data from? If there's inherent advantages it may be worthwhile, however.

I like the existing proposal of having the interaction be REST-ful because the API boundaries are clearer. For example for ACLs, the file staging server simply registers every upload with a unique token and the unique token has to be provided in order to retrieve the items that were uploaded. However once we want the staging server to have write-once-read-many semantics then we might need a more flexible mechanism here.

[mccheah]

There needs to be a timeout/expiry for the files and periodic cleanup in the staging server. (maybe this is in the PRs already)

Yep - plan to work on this later, but this is why the staging server is provided Kubernetes credentials and labels to watch pods for their healthiness and eventually clean up items when pods with the given label finish without restarting for a period of time.

An alternative to the above is that the staging server should not need to be given Kubernetes credentials and should just operate as a "system administrator" component. Therefore the staging server would have inherent privileges to watch all pods on all namespaces in the cluster.

[foxish]

What are your thoughts on the tradeoffs between using kubectl cp versus using a rest API for uploads?

I did not mean for it to be a replacement, but as a method of staging files in addition to the REST API. It does need some thought about the file paths for example; maybe through setting up some convention by which the file path is derived.

kubectl cp ~/pod.yaml nginx1:/tmp/ 

The mechanism by which kubectl cp operates, in the above case is:

1. Creates a tar archive of the file/directory we're sending.
2. The exec command un-tars the archive which is in POST data on the server end.

POST https://api-server/api/v1/namespaces/default/pods/nginx1/exec?command=tar&command=xf&command=-&command=-C&command=%2Ftmp&container=nginx&container=nginx&stderr=true&stdin=true&stdout=true

The inherent advantage is probably just that it is more generic and can be used by other applications. I am starting to think that we could use the underlying mechanism used by kubectl cp to get files up to the staging server from the client because it would let us avoid some of the work around pushing those files up in Scala code and the exposing of a service endpoint outside. The downside of course is that it's traveling via the APIServer and we don't know how bad that may be in practice.

[mccheah]

I did not mean for it to be a replacement, but as a method of staging files in addition to the REST API. It does need some thought about the file paths for example; maybe through setting up some convention by which the file path is derived.

Makes sense. Does it therefore makes sense to have the REST API be the first prototype for accessing this server, and consider using kubectl and some of the unresolved questions around it when the staging server is used in other contexts?

[mccheah]

Out of curiosity - how does the tarball data get placed into the exec command that the API server sends to the pod? Does it convert the tar's bytes from the POST into a raw string that gets passed to the tar command over standard in? i.e. does the resolved command just become:

tar -xzvf - < <tar bytes>

We would want to stream the bytes from the client to the server, but if the above is true then the API server would need to hold the tar in memory at some point.

[foxish]

The APIServer is not holding the file in memory and is streaming the contents to the pod. So, the entire request is passed through and the tar extraction happens on the pod end.

[ash211]

This is coming in as part of the V2 submission process, with the cutover PR happening in #286 after a series of PRs from Matt earlier.

Closing this issue as its purpose was to discuss the design, which we've now settled on.