Merge branch 'branch-2.2-kubernetes' into add-testing

Author: foxish

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/initcontainer/InitContainerConfigurationStepsOrchestrator.scala

@@ -20,7 +20,7 @@ import org.apache.spark.SparkConf
 import org.apache.spark.deploy.kubernetes.{InitContainerResourceStagingServerSecretPluginImpl, OptionRequirements, SparkPodInitContainerBootstrapImpl}
 import org.apache.spark.deploy.kubernetes.config._
 import org.apache.spark.deploy.kubernetes.constants._
-import org.apache.spark.deploy.kubernetes.submit.SubmittedDependencyUploaderImpl
+import org.apache.spark.deploy.kubernetes.submit.{KubernetesFileUtils, SubmittedDependencyUploaderImpl}
 import org.apache.spark.deploy.rest.kubernetes.{ResourceStagingServerSslOptionsProviderImpl, RetrofitClientFactoryImpl}
 import org.apache.spark.util.Utils
 
@@ -62,6 +62,12 @@ private[spark] class InitContainerConfigurationStepsOrchestrator(
     submissionSparkConf.get(RESOURCE_STAGING_SERVER_INTERNAL_SSL_ENABLED)
       .orElse(submissionSparkConf.get(RESOURCE_STAGING_SERVER_SSL_ENABLED))
       .getOrElse(false)
+
+  OptionRequirements.requireSecondIfFirstIsDefined(
+    KubernetesFileUtils.getOnlySubmitterLocalFiles(sparkJars).headOption,
+    resourceStagingServerUri,
+    "Local JARs were provided, however no resource staging server URI was found.")
+
   OptionRequirements.requireNandDefined(
     maybeResourceStagingServerInternalClientCert,
     maybeResourceStagingServerInternalTrustStore,

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/kubernetes/submit/submitsteps/initcontainer/InitContainerConfigurationStepsOrchestratorSuite.scala

@@ -44,6 +44,80 @@ class InitContainerConfigurationStepsOrchestratorSuite extends SparkFunSuite {
   private val INIT_CONTAINER_CONFIG_MAP_KEY = "spark-init-config-map-key"
   private val STAGING_SERVER_URI = "http://localhost:8000"
 
+  test ("error thrown if local jars provided without resource staging server") {
+    val sparkConf = new SparkConf(true)
+      .set(KUBERNETES_DRIVER_LABELS, s"$DEPRECATED_CUSTOM_LABEL_KEY=$DEPRECATED_CUSTOM_LABEL_VALUE")
+      .set(s"$KUBERNETES_DRIVER_LABEL_PREFIX$CUSTOM_LABEL_KEY", CUSTOM_LABEL_VALUE)
+
+    assert(sparkConf.get(RESOURCE_STAGING_SERVER_URI).isEmpty)
+
+    val thrown = intercept[IllegalArgumentException] {
+      val orchestrator = new InitContainerConfigurationStepsOrchestrator(
+        NAMESPACE,
+        APP_RESOURCE_PREFIX,
+        SPARK_JARS,
+        SPARK_FILES,
+        JARS_DOWNLOAD_PATH,
+        FILES_DOWNLOAD_PATH,
+        DOCKER_IMAGE_PULL_POLICY,
+        DRIVER_LABELS,
+        INIT_CONTAINER_CONFIG_MAP_NAME,
+        INIT_CONTAINER_CONFIG_MAP_KEY,
+        sparkConf)
+    }
+
+    assert(thrown.getMessage contains "Local JARs were provided, however no resource staging" +
+      " server URI was found.")
+  }
+
+  test ("error not thrown with non-local jars and resource staging server provided") {
+    val sparkConf = new SparkConf(true)
+      .set(KUBERNETES_DRIVER_LABELS, s"$DEPRECATED_CUSTOM_LABEL_KEY=$DEPRECATED_CUSTOM_LABEL_VALUE")
+      .set(s"$KUBERNETES_DRIVER_LABEL_PREFIX$CUSTOM_LABEL_KEY", CUSTOM_LABEL_VALUE)
+      .set(RESOURCE_STAGING_SERVER_URI, STAGING_SERVER_URI)
+
+    val orchestrator = new InitContainerConfigurationStepsOrchestrator(
+      NAMESPACE,
+      APP_RESOURCE_PREFIX,
+      SPARK_JARS.take(1),
+      SPARK_FILES,
+      JARS_DOWNLOAD_PATH,
+      FILES_DOWNLOAD_PATH,
+      DOCKER_IMAGE_PULL_POLICY,
+      DRIVER_LABELS,
+      INIT_CONTAINER_CONFIG_MAP_NAME,
+      INIT_CONTAINER_CONFIG_MAP_KEY,
+      sparkConf)
+    val initSteps : Seq[InitContainerConfigurationStep] =
+        orchestrator.getAllConfigurationSteps()
+    assert(initSteps.length == 2)
+    assert(initSteps.head.isInstanceOf[BaseInitContainerConfigurationStep])
+    assert(initSteps(1).isInstanceOf[SubmittedResourcesInitContainerConfigurationStep])
+  }
+
+  test ("error not thrown with non-local jars and no resource staging server provided") {
+    val sparkConf = new SparkConf(true)
+      .set(KUBERNETES_DRIVER_LABELS, s"$DEPRECATED_CUSTOM_LABEL_KEY=$DEPRECATED_CUSTOM_LABEL_VALUE")
+      .set(s"$KUBERNETES_DRIVER_LABEL_PREFIX$CUSTOM_LABEL_KEY", CUSTOM_LABEL_VALUE)
+
+    val orchestrator = new InitContainerConfigurationStepsOrchestrator(
+      NAMESPACE,
+      APP_RESOURCE_PREFIX,
+      SPARK_JARS.take(1),
+      SPARK_FILES,
+      JARS_DOWNLOAD_PATH,
+      FILES_DOWNLOAD_PATH,
+      DOCKER_IMAGE_PULL_POLICY,
+      DRIVER_LABELS,
+      INIT_CONTAINER_CONFIG_MAP_NAME,
+      INIT_CONTAINER_CONFIG_MAP_KEY,
+      sparkConf)
+    val initSteps : Seq[InitContainerConfigurationStep] =
+        orchestrator.getAllConfigurationSteps()
+    assert(initSteps.length == 1)
+    assert(initSteps.head.isInstanceOf[BaseInitContainerConfigurationStep])
+  }
+
   test ("including step to contact resource staging server") {
     val sparkConf = new SparkConf(true)
       .set(KUBERNETES_DRIVER_LABELS, s"$DEPRECATED_CUSTOM_LABEL_KEY=$DEPRECATED_CUSTOM_LABEL_VALUE")
@@ -77,7 +151,7 @@ class InitContainerConfigurationStepsOrchestratorSuite extends SparkFunSuite {
     val orchestrator = new InitContainerConfigurationStepsOrchestrator(
       NAMESPACE,
       APP_RESOURCE_PREFIX,
-      SPARK_JARS,
+      SPARK_JARS.take(1),
       SPARK_FILES,
       JARS_DOWNLOAD_PATH,
       FILES_DOWNLOAD_PATH,

sbin/build-push-docker-images.sh

@@ -0,0 +1,72 @@
+#!/usr/bin/env bash
+
+# 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.
+#
+
+# This script builds and pushes docker images when run from a release of Spark
+# with Kubernetes support.
+
+declare -A path=( [spark-driver]=dockerfiles/driver/Dockerfile \
+                  [spark-executor]=dockerfiles/executor/Dockerfile \
+                  [spark-driver-py]=dockerfiles/driver-py/Dockerfile \
+                  [spark-executor-py]=dockerfiles/executor-py/Dockerfile \
+                  [spark-init]=dockerfiles/init-container/Dockerfile \
+                  [spark-shuffle]=dockerfiles/shuffle-service/Dockerfile \
+                  [spark-resource-staging-server]=dockerfiles/resource-staging-server/Dockerfile )
+
+function build {
+  docker build -t spark-base -f dockerfiles/spark-base/Dockerfile .
+  for image in "${!path[@]}"; do
+    docker build -t ${REPO}/$image:${TAG} -f ${path[$image]} .
+  done
+}
+
+
+function push {
+  for image in "${!path[@]}"; do
+    docker push ${REPO}/$image:${TAG}
+  done
+}
+
+function usage {
+  echo "Usage: ./sbin/build-push-docker-images.sh -r <repo> -t <tag> build"
+  echo "       ./sbin/build-push-docker-images.sh -r <repo> -t <tag> push"
+  echo "for example: ./sbin/build-push-docker-images.sh -r docker.io/kubespark -t v2.2.0 push"
+}
+
+if [[ "$@" = *--help ]] || [[ "$@" = *-h ]]; then
+  usage
+  exit 0
+fi
+
+while getopts r:t: option
+do
+ case "${option}"
+ in
+ r) REPO=${OPTARG};;
+ t) TAG=${OPTARG};;
+ esac
+done
+
+if [ -z "$REPO" ] || [ -z "$TAG" ]; then
+    usage
+else
+  case "${@: -1}" in
+    build) build;;
+    push) push;;
+    *) usage;;
+  esac
+fi