Staging server for receiving application dependencies. (#212)

* Staging server for receiving application dependencies.

* Add unit test for file writing

* Minor fixes

* Remove getting credentials from the API

We still want to post them because in the future we can use these
credentials to monitor the API server and handle cleaning up the data
accordingly.

* Generalize to resource staging server outside of Spark

* Update code documentation

* Val instead of var

* Fix naming, remove unused import

* Move suites from integration test package to core

* Use TrieMap instead of locks

* Address comments

* Fix imports

* Change paths, use POST instead of PUT

* Use a resource identifier as well as a resource secret

Author: mccheah

pom.xml

@@ -137,6 +137,7 @@
     <parquet.version>1.8.1</parquet.version>
     <hive.parquet.version>1.6.0</hive.parquet.version>
     <feign.version>8.18.0</feign.version>
+    <retrofit.version>2.2.0</retrofit.version>
     <bouncycastle.version>1.54</bouncycastle.version>
     <jetty.version>9.2.16.v20160414</jetty.version>
     <javaxservlet.version>3.1.0</javaxservlet.version>
@@ -327,6 +328,21 @@
         <artifactId>feign-jaxrs</artifactId>
         <version>${feign.version}</version>
       </dependency>
+      <dependency>
+        <groupId>com.squareup.retrofit2</groupId>
+        <artifactId>retrofit</artifactId>
+        <version>${retrofit.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>com.squareup.retrofit2</groupId>
+        <artifactId>converter-jackson</artifactId>
+        <version>${retrofit.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>com.squareup.retrofit2</groupId>
+        <artifactId>converter-scalars</artifactId>
+        <version>${retrofit.version}</version>
+      </dependency>
       <dependency>
         <groupId>org.bouncycastle</groupId>
         <artifactId>bcpkix-jdk15on</artifactId>
@@ -686,6 +702,11 @@
         <artifactId>jersey-client</artifactId>
         <version>${jersey.version}</version>
       </dependency>
+      <dependency>
+        <groupId>org.glassfish.jersey.media</groupId>
+        <artifactId>jersey-media-multipart</artifactId>
+        <version>${jersey.version}</version>
+      </dependency>
       <dependency>
         <groupId>javax.ws.rs</groupId>
         <artifactId>javax.ws.rs-api</artifactId>

resource-managers/kubernetes/core/pom.xml

@@ -60,10 +60,31 @@
       <groupId>com.netflix.feign</groupId>
       <artifactId>feign-okhttp</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.glassfish.jersey.containers</groupId>
+      <artifactId>jersey-container-servlet</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.glassfish.jersey.media</groupId>
+      <artifactId>jersey-media-multipart</artifactId>
+    </dependency>
     <dependency>
       <groupId>com.netflix.feign</groupId>
       <artifactId>feign-jackson</artifactId>
     </dependency>
+    <dependency>
+      <groupId>com.squareup.retrofit2</groupId>
+      <artifactId>retrofit</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>com.squareup.retrofit2</groupId>
+      <artifactId>converter-jackson</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>com.squareup.retrofit2</groupId>
+      <artifactId>converter-scalars</artifactId>
+    </dependency>
+
     <dependency>
       <groupId>com.netflix.feign</groupId>
       <artifactId>feign-jaxrs</artifactId>

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/v2/ResourceStagingServer.scala

@@ -0,0 +1,61 @@
+/*
+ * 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.rest.kubernetes.v2
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.jaxrs.json.JacksonJaxbJsonProvider
+import com.fasterxml.jackson.module.scala.DefaultScalaModule
+import org.eclipse.jetty.server.{Server, ServerConnector}
+import org.eclipse.jetty.servlet.{ServletContextHandler, ServletHolder}
+import org.eclipse.jetty.util.thread.QueuedThreadPool
+import org.glassfish.jersey.media.multipart.MultiPartFeature
+import org.glassfish.jersey.server.ResourceConfig
+import org.glassfish.jersey.servlet.ServletContainer
+
+private[spark] class ResourceStagingServer(
+    port: Int,
+    serviceInstance: ResourceStagingService) {
+
+  private var jettyServer: Option[Server] = None
+
+  def start(): Unit = synchronized {
+    val threadPool = new QueuedThreadPool
+    val contextHandler = new ServletContextHandler()
+    val jsonProvider = new JacksonJaxbJsonProvider()
+    jsonProvider.setMapper(new ObjectMapper().registerModule(new DefaultScalaModule))
+    val resourceConfig = new ResourceConfig().registerInstances(
+      serviceInstance,
+      jsonProvider,
+      new MultiPartFeature)
+    val servletHolder = new ServletHolder("main", new ServletContainer(resourceConfig))
+    contextHandler.setContextPath("/api/")
+    contextHandler.addServlet(servletHolder, "/*")
+    threadPool.setDaemon(true)
+    val server = new Server(threadPool)
+    val connector = new ServerConnector(server)
+    connector.setPort(port)
+    server.addConnector(connector)
+    server.setHandler(contextHandler)
+    server.start()
+    jettyServer = Some(server)
+  }
+
+  def stop(): Unit = synchronized {
+    jettyServer.foreach(_.stop())
+    jettyServer = None
+  }
+}

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/v2/ResourceStagingService.scala

@@ -0,0 +1,85 @@
+/*
+ * 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.rest.kubernetes.v2
+
+import java.io.InputStream
+import javax.ws.rs.{Consumes, GET, HeaderParam, Path, PathParam, POST, Produces}
+import javax.ws.rs.core.{MediaType, StreamingOutput}
+
+import org.glassfish.jersey.media.multipart.FormDataParam
+
+import org.apache.spark.deploy.rest.KubernetesCredentials
+
+/**
+ * Service that receives application data that can be retrieved later on. This is primarily used
+ * in the context of Spark, but the concept is generic enough to be used for arbitrary applications.
+ * The use case is to have a place for Kubernetes application submitters to bootstrap dynamic,
+ * heavyweight application data for pods. Application submitters may have data stored on their
+ * local disks that they want to provide to the pods they create through the API server. ConfigMaps
+ * are one way to provide this data, but the data in ConfigMaps are stored in etcd which cannot
+ * maintain data in the hundreds of megabytes in size.
+ * <p>
+ * The general use case is for an application submitter to ship the dependencies to the server via
+ * {@link uploadResources}; the application submitter will then receive a unique secure token.
+ * The application submitter then ought to convert the token into a secret, and use this secret in
+ * a pod that fetches the uploaded dependencies via {@link downloadResources}. An application can
+ * provide multiple resource bundles simply by hitting the upload endpoint multiple times and
+ * downloading each bundle with the appropriate secret.
+ */
+@Path("/v0")
+private[spark] trait ResourceStagingService {
+
+  /**
+   * Register a resource with the dependency service, so that pods with the given labels can
+   * retrieve them when they run.
+   *
+   * @param resources Application resources to upload, compacted together in tar + gzip format.
+   *                  The tarball should contain the files laid out in a flat hierarchy, without
+   *                  any directories. We take a stream here to avoid holding these entirely in
+   *                  memory.
+   * @param podLabels Labels of pods to monitor. When no more pods are running with the given label,
+   *                  after some period of time, these dependencies will be cleared.
+   * @param podNamespace Namespace of pods to monitor.
+   * @param kubernetesCredentials These credentials are primarily used to monitor the progress of
+   *                              the application. When the application shuts down normally, shuts
+   *                              down abnormally and does not restart, or fails to start entirely,
+   *                              the data uploaded through this endpoint is cleared.
+   * @return A unique token that should be provided when retrieving these dependencies later.
+   */
+  @POST
+  @Consumes(Array(MediaType.MULTIPART_FORM_DATA, MediaType.APPLICATION_JSON, MediaType.TEXT_PLAIN))
+  @Produces(Array(MediaType.APPLICATION_JSON))
+  @Path("/resources")
+  def uploadResources(
+      @FormDataParam("podLabels") podLabels: Map[String, String],
+      @FormDataParam("podNamespace") podNamespace: String,
+      @FormDataParam("resources") resources: InputStream,
+      @FormDataParam("kubernetesCredentials") kubernetesCredentials: KubernetesCredentials)
+      : StagedResourceIdentifier
+
+  /**
+   * Download an application's resources. The resources are provided as a stream, where the stream's
+   * underlying data matches the stream that was uploaded in uploadResources.
+   */
+  @GET
+  @Consumes(Array(MediaType.APPLICATION_JSON))
+  @Produces(Array(MediaType.APPLICATION_OCTET_STREAM))
+  @Path("/resources/{resourceId}")
+  def downloadResources(
+      @PathParam("resourceId") resourceId: String,
+      @HeaderParam("Authorization") resourceSecret: String): StreamingOutput
+}

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/v2/ResourceStagingServiceImpl.scala

@@ -0,0 +1,98 @@
+/*
+ * 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.rest.kubernetes.v2
+
+import java.io.{File, FileOutputStream, InputStream, OutputStream}
+import java.security.SecureRandom
+import java.util.UUID
+import javax.ws.rs.{NotAuthorizedException, NotFoundException}
+import javax.ws.rs.core.StreamingOutput
+
+import com.google.common.io.{BaseEncoding, ByteStreams, Files}
+import scala.collection.concurrent.TrieMap
+
+import org.apache.spark.SparkException
+import org.apache.spark.deploy.rest.KubernetesCredentials
+import org.apache.spark.internal.Logging
+import org.apache.spark.util.Utils
+
+private[spark] class ResourceStagingServiceImpl(dependenciesRootDir: File)
+    extends ResourceStagingService with Logging {
+
+  private val SECURE_RANDOM = new SecureRandom()
+  // TODO clean up these resources based on the driver's lifecycle
+  private val stagedResources = TrieMap.empty[String, StagedResources]
+
+  override def uploadResources(
+      podLabels: Map[String, String],
+      podNamespace: String,
+      resources: InputStream,
+      kubernetesCredentials: KubernetesCredentials): StagedResourceIdentifier = {
+    val resourceId = UUID.randomUUID().toString
+    val secretBytes = new Array[Byte](1024)
+    SECURE_RANDOM.nextBytes(secretBytes)
+    val resourceSecret = resourceId + "-" + BaseEncoding.base64().encode(secretBytes)
+
+    val namespaceDir = new File(dependenciesRootDir, podNamespace)
+    val resourcesDir = new File(namespaceDir, resourceId)
+    try {
+      if (!resourcesDir.exists()) {
+        if (!resourcesDir.mkdirs()) {
+          throw new SparkException("Failed to create dependencies directory for application" +
+            s" at ${resourcesDir.getAbsolutePath}")
+        }
+      }
+      // TODO encrypt the written data with the secret.
+      val resourcesTgz = new File(resourcesDir, "resources.data")
+      Utils.tryWithResource(new FileOutputStream(resourcesTgz)) { ByteStreams.copy(resources, _) }
+      stagedResources(resourceId) = StagedResources(
+        resourceSecret,
+        podLabels,
+        podNamespace,
+        resourcesTgz,
+        kubernetesCredentials)
+      StagedResourceIdentifier(resourceId, resourceSecret)
+    } catch {
+      case e: Throwable =>
+        if (!resourcesDir.delete()) {
+          logWarning(s"Failed to delete application directory $resourcesDir.")
+        }
+        throw e
+    }
+  }
+
+  override def downloadResources(resourceId: String, resourceSecret: String): StreamingOutput = {
+    val resource = stagedResources
+        .get(resourceId)
+        .getOrElse(throw new NotFoundException(s"No resource bundle found with id $resourceId"))
+    if (!resource.resourceSecret.equals(resourceSecret)) {
+      throw new NotAuthorizedException(s"Unauthorized to download resource with id $resourceId")
+    }
+    new StreamingOutput {
+      override def write(outputStream: OutputStream) = {
+        Files.copy(resource.resourcesFile, outputStream)
+      }
+    }
+  }
+}
+
+private case class StagedResources(
+  resourceSecret: String,
+  podLabels: Map[String, String],
+  podNamespace: String,
+  resourcesFile: File,
+  kubernetesCredentials: KubernetesCredentials)

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/v2/ResourceStagingServiceRetrofit.scala

@@ -0,0 +1,42 @@
+/*
+ * 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.rest.kubernetes.v2
+
+import okhttp3.{RequestBody, ResponseBody}
+import retrofit2.Call
+import retrofit2.http.{Multipart, Path, Streaming}
+
+/**
+ * Retrofit-compatible variant of {@link ResourceStagingService}. For documentation on
+ * how to use this service, see the aforementioned JAX-RS based interface.
+ */
+private[spark] trait ResourceStagingServiceRetrofit {
+
+  @Multipart
+  @retrofit2.http.POST("/api/v0/resources/")
+  def uploadResources(
+      @retrofit2.http.Part("podLabels") podLabels: RequestBody,
+      @retrofit2.http.Part("podNamespace") podNamespace: RequestBody,
+      @retrofit2.http.Part("resources") resources: RequestBody,
+      @retrofit2.http.Part("kubernetesCredentials")
+          kubernetesCredentials: RequestBody): Call[StagedResourceIdentifier]
+
+  @Streaming
+  @retrofit2.http.GET("/api/v0/resources/{resourceId}")
+  def downloadResources(@Path("resourceId") resourceId: String,
+      @retrofit2.http.Header("Authorization") resourceSecret: String): Call[ResponseBody]
+}