Staging server for receiving application dependencies.

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]
+}