From c25c86c6134ded3c81e8682eca25e4ad21314216 Mon Sep 17 00:00:00 2001 From: mcheah Date: Wed, 1 Feb 2017 18:20:57 -0800 Subject: [PATCH 01/19] Use readiness probe instead of client-side ping. Keep one ping() just as a sanity check, but otherwise set up the readiness probe to report the container as ready only when the ping endpoint can be reached. Also add a liveliness probe for convenience and symmetry. --- .../spark/deploy/kubernetes/Client.scala | 186 ++++++++---------- .../spark/deploy/kubernetes/Retry.scala | 42 ---- 2 files changed, 87 insertions(+), 141 deletions(-) delete mode 100644 resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Retry.scala diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala index fed9334dbbab4..f5845ca27bb06 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala @@ -162,6 +162,20 @@ private[spark] class Client( .endEnv() .addToEnv(sslEnvs: _*) .withPorts(containerPorts.asJava) + .withNewReadinessProbe() + .withNewHttpGet() + .withScheme(if (driverSubmitSslOptions.enabled) "HTTPS" else "HTTP") + .withPath("/v1/submissions/ping") + .withNewPort(SUBMISSION_SERVER_PORT_NAME) + .endHttpGet() + .endReadinessProbe() + .withNewLivenessProbe() + .withNewHttpGet() + .withScheme(if (driverSubmitSslOptions.enabled) "HTTPS" else "HTTP") + .withPath("/v1/submissions/ping") + .withNewPort(SUBMISSION_SERVER_PORT_NAME) + .endHttpGet() + .endLivenessProbe() .endContainer() .endSpec() .done() @@ -308,114 +322,88 @@ private[spark] class Client( driverKubernetesSelectors: java.util.Map[String, String]) extends Watcher[Pod] { override def eventReceived(action: Action, pod: Pod): Unit = { if ((action == Action.ADDED || action == Action.MODIFIED) - && pod.getStatus.getPhase == "Running" - && !submitCompletedFuture.isDone) { - if (!submitPending.getAndSet(true)) { - pod.getStatus - .getContainerStatuses - .asScala - .find(status => - status.getName == DRIVER_CONTAINER_NAME && status.getReady) match { - case Some(_) => - val ownerRefs = Seq(new OwnerReferenceBuilder() - .withName(pod.getMetadata.getName) - .withUid(pod.getMetadata.getUid) - .withApiVersion(pod.getApiVersion) - .withKind(pod.getKind) - .withController(true) - .build()) - - applicationSecrets.foreach(secret => { - secret.getMetadata.setOwnerReferences(ownerRefs.asJava) - kubernetesClient.secrets().createOrReplace(secret) - }) - - val driverSubmissionServicePort = new ServicePortBuilder() - .withName(SUBMISSION_SERVER_PORT_NAME) - .withPort(SUBMISSION_SERVER_PORT) - .withNewTargetPort(SUBMISSION_SERVER_PORT) - .build() - val service = kubernetesClient.services().createNew() - .withNewMetadata() - .withName(kubernetesAppId) - .withLabels(driverKubernetesSelectors) - .withOwnerReferences(ownerRefs.asJava) - .endMetadata() - .withNewSpec() - .withType("NodePort") - .withSelector(driverKubernetesSelectors) - .withPorts(driverSubmissionServicePort) - .endSpec() - .done() + && pod.getStatus.getPhase == "Running" + && !submitCompletedFuture.isDone) { + pod.getStatus + .getContainerStatuses + .asScala + .find(status => + status.getName == DRIVER_CONTAINER_NAME && status.getReady) match { + case Some(_) => + if (!submitPending.getAndSet(true)) { try { - sparkConf.getOption("spark.app.id").foreach { id => - logWarning(s"Warning: Provided app id in spark.app.id as $id will be" + - s" overridden as $kubernetesAppId") - } - sparkConf.set(KUBERNETES_DRIVER_POD_NAME, kubernetesAppId) - sparkConf.set(KUBERNETES_DRIVER_SERVICE_NAME, service.getMetadata.getName) - sparkConf.set("spark.app.id", kubernetesAppId) - sparkConf.setIfMissing("spark.app.name", appName) - sparkConf.setIfMissing("spark.driver.port", DEFAULT_DRIVER_PORT.toString) - sparkConf.setIfMissing("spark.blockmanager.port", - DEFAULT_BLOCKMANAGER_PORT.toString) - val driverSubmitter = buildDriverSubmissionClient(kubernetesClient, service, + val ownerRefs = Seq(new OwnerReferenceBuilder() + .withName(pod.getMetadata.getName) + .withUid(pod.getMetadata.getUid) + .withApiVersion(pod.getApiVersion) + .withKind(pod.getKind) + .withController(true) + .build()) + applicationSecrets.foreach(secret => { + secret.getMetadata.setOwnerReferences(ownerRefs.asJava) + kubernetesClient.secrets().createOrReplace(secret) + }) + val driverSubmissionServicePort = new ServicePortBuilder() + .withName(SUBMISSION_SERVER_PORT_NAME) + .withPort(SUBMISSION_SERVER_PORT) + .withNewTargetPort(SUBMISSION_SERVER_PORT) + .build() + val service = kubernetesClient.services().createNew() + .withNewMetadata() + .withName(kubernetesAppId) + .withLabels(driverKubernetesSelectors) + .withOwnerReferences(ownerRefs.asJava) + .endMetadata() + .withNewSpec() + .withType("NodePort") + .withSelector(driverKubernetesSelectors) + .withPorts(driverSubmissionServicePort) + .endSpec() + .done() + + try { + sparkConf.getOption("spark.app.id").foreach { id => + logWarning(s"Warning: Provided app id in spark.app.id as $id will be" + + s" overridden as $kubernetesAppId") + } + sparkConf.set(KUBERNETES_DRIVER_POD_NAME, kubernetesAppId) + sparkConf.set(KUBERNETES_DRIVER_SERVICE_NAME, service.getMetadata.getName) + sparkConf.set("spark.app.id", kubernetesAppId) + sparkConf.setIfMissing("spark.app.name", appName) + sparkConf.setIfMissing("spark.driver.port", DEFAULT_DRIVER_PORT.toString) + sparkConf.setIfMissing("spark.blockmanager.port", + DEFAULT_BLOCKMANAGER_PORT.toString) + val driverSubmitter = buildDriverSubmissionClient(kubernetesClient, service, driverSubmitSslOptions) - val ping = Retry.retry(5, 5.seconds) { driverSubmitter.ping() - } - ping onFailure { - case t: Throwable => - submitCompletedFuture.setException(t) - kubernetesClient.services().delete(service) - } - val submitComplete = ping.flatMap { _ => - Future { - sparkConf.set("spark.driver.host", pod.getStatus.getPodIP) - val submitRequest = buildSubmissionRequest() - driverSubmitter.submitApplication(submitRequest) - } - } - submitComplete onFailure { - case t: Throwable => - submitCompletedFuture.setException(t) - kubernetesClient.services().delete(service) - } - val adjustServicePort = submitComplete.flatMap { _ => - Future { - // After submitting, adjust the service to only expose the Spark UI - val uiServicePort = new ServicePortBuilder() - .withName(UI_PORT_NAME) - .withPort(uiPort) - .withNewTargetPort(uiPort) - .build() - kubernetesClient.services().withName(kubernetesAppId).edit() - .editSpec() - .withType("ClusterIP") - .withPorts(uiServicePort) - .endSpec() - .done - } - } - adjustServicePort onSuccess { - case _ => - submitCompletedFuture.set(true) - } - adjustServicePort onFailure { - case throwable: Throwable => - submitCompletedFuture.setException(throwable) - kubernetesClient.services().delete(service) + val submitRequest = buildSubmissionRequest() + driverSubmitter.submitApplication(submitRequest) + // After submitting, adjust the service to only expose the Spark UI + val uiServicePort = new ServicePortBuilder() + .withName(UI_PORT_NAME) + .withPort(uiPort) + .withNewTargetPort(uiPort) + .build() + kubernetesClient.services().withName(kubernetesAppId).edit().editSpec() + .withType("ClusterIP") + .withPorts(uiServicePort) + .endSpec() + .done() + submitCompletedFuture.set(true) + } catch { + case e: Throwable => + Utils.tryLogNonFatalError({ + kubernetesClient.services().delete(service) + }) + throw e } } catch { case e: Throwable => submitCompletedFuture.setException(e) - Utils.tryLogNonFatalError({ - kubernetesClient.services().delete(service) - }) throw e } - case None => - } + } + case None => } } } diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Retry.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Retry.scala deleted file mode 100644 index e5ce0bcd606b2..0000000000000 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Retry.scala +++ /dev/null @@ -1,42 +0,0 @@ -/* - * 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.kubernetes - -import scala.concurrent.{ExecutionContext, Future} -import scala.concurrent.duration.Duration - -private[spark] object Retry { - - private def retryableFuture[T] - (times: Int, interval: Duration) - (f: => Future[T]) - (implicit executionContext: ExecutionContext): Future[T] = { - f recoverWith { - case _ if times > 0 => { - Thread.sleep(interval.toMillis) - retryableFuture(times - 1, interval)(f) - } - } - } - - def retry[T] - (times: Int, interval: Duration) - (f: => T) - (implicit executionContext: ExecutionContext): Future[T] = { - retryableFuture(times, interval)(Future[T] { f }) - } -} From e231105b9114ae47017a4c58077e3f83b6949b09 Mon Sep 17 00:00:00 2001 From: mcheah Date: Wed, 1 Feb 2017 18:33:52 -0800 Subject: [PATCH 02/19] Extract common HTTP get action --- .../spark/deploy/kubernetes/Client.scala | 21 +++++++------------ 1 file changed, 7 insertions(+), 14 deletions(-) diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala index f5845ca27bb06..6d2f7a6195715 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala @@ -127,6 +127,11 @@ private[spark] class Client( .pods() .withLabels(driverKubernetesSelectors) .watch(podWatcher)) { _ => + val probePingHttpGet = new HTTPGetActionBuilder() + .withScheme(if (driverSubmitSslOptions.enabled) "HTTPS" else "HTTP") + .withPath("/v1/submissions/ping") + .withNewPort(SUBMISSION_SERVER_PORT_NAME) + .build() kubernetesClient.pods().createNew() .withNewMetadata() .withName(kubernetesAppId) @@ -162,20 +167,8 @@ private[spark] class Client( .endEnv() .addToEnv(sslEnvs: _*) .withPorts(containerPorts.asJava) - .withNewReadinessProbe() - .withNewHttpGet() - .withScheme(if (driverSubmitSslOptions.enabled) "HTTPS" else "HTTP") - .withPath("/v1/submissions/ping") - .withNewPort(SUBMISSION_SERVER_PORT_NAME) - .endHttpGet() - .endReadinessProbe() - .withNewLivenessProbe() - .withNewHttpGet() - .withScheme(if (driverSubmitSslOptions.enabled) "HTTPS" else "HTTP") - .withPath("/v1/submissions/ping") - .withNewPort(SUBMISSION_SERVER_PORT_NAME) - .endHttpGet() - .endLivenessProbe() + .withNewReadinessProbe().withHttpGet(probePingHttpGet).endReadinessProbe() + .withNewLivenessProbe().withHttpGet(probePingHttpGet).endLivenessProbe() .endContainer() .endSpec() .done() From 0ad223f7bcd24f8fa14c3165b6ffdea23ce017e2 Mon Sep 17 00:00:00 2001 From: mcheah Date: Wed, 1 Feb 2017 18:36:14 -0800 Subject: [PATCH 03/19] Remove some code --- .../scala/org/apache/spark/deploy/kubernetes/Client.scala | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala index 6d2f7a6195715..05dabdb8c41db 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala @@ -31,8 +31,6 @@ import io.fabric8.kubernetes.client.Watcher.Action import org.apache.commons.codec.binary.Base64 import scala.collection.JavaConverters._ import scala.collection.mutable -import scala.concurrent.{ExecutionContext, Future} -import scala.concurrent.duration.DurationInt import org.apache.spark.{SecurityManager, SparkConf, SparkException, SSLOptions} import org.apache.spark.deploy.kubernetes.config._ @@ -40,7 +38,7 @@ import org.apache.spark.deploy.kubernetes.constants._ import org.apache.spark.deploy.rest.{AppResource, ContainerAppResource, KubernetesCreateSubmissionRequest, RemoteAppResource, TarGzippedData, UploadedAppResource} import org.apache.spark.deploy.rest.kubernetes._ import org.apache.spark.internal.Logging -import org.apache.spark.util.{ThreadUtils, Utils} +import org.apache.spark.util.Utils private[spark] class Client( sparkConf: SparkConf, @@ -74,10 +72,6 @@ private[spark] class Client( private val serviceAccount = sparkConf.get(KUBERNETES_SERVICE_ACCOUNT_NAME) private val customLabels = sparkConf.get(KUBERNETES_DRIVER_LABELS) - private implicit val retryableExecutionContext = ExecutionContext - .fromExecutorService( - ThreadUtils.newDaemonSingleThreadExecutor("kubernetes-client-retryable-futures")) - def run(): Unit = { val (driverSubmitSslOptions, isKeyStoreLocalFile) = parseDriverSubmitSslOptions() val parsedCustomLabels = parseCustomLabels(customLabels) From 6699072562257797eb8b16b76e7fc99b09402476 Mon Sep 17 00:00:00 2001 From: mcheah Date: Thu, 2 Feb 2017 12:24:55 -0800 Subject: [PATCH 04/19] Add delay to liveliness check --- .../scala/org/apache/spark/deploy/kubernetes/Client.scala | 6 +++++- .../org/apache/spark/deploy/kubernetes/constants.scala | 1 + 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala index 05dabdb8c41db..0ef0680c24fde 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala @@ -162,7 +162,10 @@ private[spark] class Client( .addToEnv(sslEnvs: _*) .withPorts(containerPorts.asJava) .withNewReadinessProbe().withHttpGet(probePingHttpGet).endReadinessProbe() - .withNewLivenessProbe().withHttpGet(probePingHttpGet).endLivenessProbe() + .withNewLivenessProbe() + .withHttpGet(probePingHttpGet) + .withInitialDelaySeconds(DRIVER_INITIAL_LIVELINESS_CHECK_DELAY_SECONDS) + .endLivenessProbe() .endContainer() .endSpec() .done() @@ -335,6 +338,7 @@ private[spark] class Client( .withPort(SUBMISSION_SERVER_PORT) .withNewTargetPort(SUBMISSION_SERVER_PORT) .build() + val service = kubernetesClient.services().createNew() .withNewMetadata() .withName(kubernetesAppId) diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/constants.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/constants.scala index 027cc3c022b4e..d16d7527cd3bf 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/constants.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/constants.scala @@ -67,4 +67,5 @@ package object constants { // Miscellaneous private[spark] val DRIVER_CONTAINER_NAME = "spark-kubernetes-driver" private[spark] val KUBERNETES_SUBMIT_SSL_NAMESPACE = "kubernetes.submit" + private[spark] val DRIVER_INITIAL_LIVELINESS_CHECK_DELAY_SECONDS = 10 } From ef8179a56f973208a270240e10694203e5dec52e Mon Sep 17 00:00:00 2001 From: mcheah Date: Thu, 2 Feb 2017 17:19:53 -0800 Subject: [PATCH 05/19] Fix merge conflicts. --- .../spark/deploy/kubernetes/Client.scala | 96 ------------------- .../spark/deploy/kubernetes/Retry.scala | 54 ----------- 2 files changed, 150 deletions(-) delete mode 100644 resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Retry.scala diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala index 5bddde78d7e1d..4e09fd21a6d0b 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala @@ -368,15 +368,7 @@ private[spark] class Client( DEFAULT_BLOCKMANAGER_PORT.toString) val driverSubmitter = buildDriverSubmissionClient(kubernetesClient, service, driverSubmitSslOptions) -<<<<<<< HEAD -||||||| merged common ancestors - val ping = Retry.retry(5, 5.seconds) { -======= - val ping = Retry.retry(5, 5.seconds, - Some("Failed to contact the driver server")) { ->>>>>>> apache-spark-on-k8s/k8s-support-alternate-incremental driverSubmitter.ping() -<<<<<<< HEAD val submitRequest = buildSubmissionRequest() driverSubmitter.submitApplication(submitRequest) // After submitting, adjust the service to only expose the Spark UI @@ -397,94 +389,6 @@ private[spark] class Client( kubernetesClient.services().delete(service) }) throw e -||||||| merged common ancestors - } - ping onFailure { - case t: Throwable => - submitCompletedFuture.setException(t) - kubernetesClient.services().delete(service) - } - val submitComplete = ping.flatMap { _ => - Future { - sparkConf.set("spark.driver.host", pod.getStatus.getPodIP) - val submitRequest = buildSubmissionRequest() - driverSubmitter.submitApplication(submitRequest) - } - } - submitComplete onFailure { - case t: Throwable => - submitCompletedFuture.setException(t) - kubernetesClient.services().delete(service) - } - val adjustServicePort = submitComplete.flatMap { _ => - Future { - // After submitting, adjust the service to only expose the Spark UI - val uiServicePort = new ServicePortBuilder() - .withName(UI_PORT_NAME) - .withPort(uiPort) - .withNewTargetPort(uiPort) - .build() - kubernetesClient.services().withName(kubernetesAppId).edit() - .editSpec() - .withType("ClusterIP") - .withPorts(uiServicePort) - .endSpec() - .done - } - } - adjustServicePort onSuccess { - case _ => - submitCompletedFuture.set(true) - } - adjustServicePort onFailure { - case throwable: Throwable => - submitCompletedFuture.setException(throwable) - kubernetesClient.services().delete(service) -======= - } - ping onFailure { - case t: Throwable => - logError("Ping failed to the driver server", t) - submitCompletedFuture.setException(t) - kubernetesClient.services().delete(service) - } - val submitComplete = ping.flatMap { _ => - Future { - sparkConf.set("spark.driver.host", pod.getStatus.getPodIP) - val submitRequest = buildSubmissionRequest() - driverSubmitter.submitApplication(submitRequest) - } - } - submitComplete onFailure { - case t: Throwable => - submitCompletedFuture.setException(t) - kubernetesClient.services().delete(service) - } - val adjustServicePort = submitComplete.flatMap { _ => - Future { - // After submitting, adjust the service to only expose the Spark UI - val uiServicePort = new ServicePortBuilder() - .withName(UI_PORT_NAME) - .withPort(uiPort) - .withNewTargetPort(uiPort) - .build() - kubernetesClient.services().withName(kubernetesAppId).edit() - .editSpec() - .withType("ClusterIP") - .withPorts(uiServicePort) - .endSpec() - .done - } - } - adjustServicePort onSuccess { - case _ => - submitCompletedFuture.set(true) - } - adjustServicePort onFailure { - case throwable: Throwable => - submitCompletedFuture.setException(throwable) - kubernetesClient.services().delete(service) ->>>>>>> apache-spark-on-k8s/k8s-support-alternate-incremental } } catch { case e: Throwable => diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Retry.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Retry.scala deleted file mode 100644 index 378583b29c547..0000000000000 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Retry.scala +++ /dev/null @@ -1,54 +0,0 @@ -/* - * 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.kubernetes - -import scala.concurrent.{ExecutionContext, Future} -import scala.concurrent.duration.Duration - -import org.apache.spark.SparkException -import org.apache.spark.internal.Logging - -private[spark] object Retry extends Logging { - - private def retryableFuture[T] - (attempt: Int, maxAttempts: Int, interval: Duration, retryMessage: Option[String]) - (f: => Future[T]) - (implicit executionContext: ExecutionContext): Future[T] = { - f recoverWith { - case error: Throwable => - if (attempt <= maxAttempts) { - retryMessage.foreach { message => - logWarning(s"$message - attempt $attempt of $maxAttempts", error) - } - Thread.sleep(interval.toMillis) - retryableFuture(attempt + 1, maxAttempts, interval, retryMessage)(f) - } else { - Future.failed(retryMessage.map(message => - new SparkException(s"$message - reached $maxAttempts attempts," + - s" and aborting task.", error) - ).getOrElse(error)) - } - } - } - - def retry[T] - (times: Int, interval: Duration, retryMessage: Option[String] = None) - (f: => T) - (implicit executionContext: ExecutionContext): Future[T] = { - retryableFuture(1, times, interval, retryMessage)(Future[T] { f }) - } -} From deb6386bd82f2d6a8f6efd24d424d2b3e2e14c1c Mon Sep 17 00:00:00 2001 From: mcheah Date: Thu, 2 Feb 2017 18:07:35 -0800 Subject: [PATCH 06/19] Fix more merge conflicts --- .../spark/deploy/kubernetes/Client.scala | 182 +----------------- 1 file changed, 8 insertions(+), 174 deletions(-) diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala index 0c8ec32e8bb45..4132ee32d0e27 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala @@ -124,24 +124,12 @@ private[spark] class Client( Utils.tryWithResource(kubernetesClient .pods() .withLabels(driverKubernetesSelectors) -<<<<<<< HEAD - .watch(podWatcher)) { _ => + .watch(loggingWatch)) { _ => val probePingHttpGet = new HTTPGetActionBuilder() .withScheme(if (driverSubmitSslOptions.enabled) "HTTPS" else "HTTP") .withPath("/v1/submissions/ping") .withNewPort(SUBMISSION_SERVER_PORT_NAME) .build() - kubernetesClient.pods().createNew() - .withNewMetadata() - .withName(kubernetesAppId) -||||||| merged common ancestors - .watch(podWatcher)) { _ => - kubernetesClient.pods().createNew() - .withNewMetadata() - .withName(kubernetesAppId) -======= - .watch(loggingWatch)) { _ => - // launch driver pod with inner watch to upload jars when it's ready val submitCompletedFuture = SettableFuture.create[Boolean] val submitPending = new AtomicBoolean(false) @@ -154,7 +142,6 @@ private[spark] class Client( driverKubernetesSelectors) Utils.tryWithResource(kubernetesClient .pods() ->>>>>>> apache-spark-on-k8s/k8s-support-alternate-incremental .withLabels(driverKubernetesSelectors) .watch(podWatcher)) { _ => kubernetesClient.pods().createNew() @@ -166,74 +153,6 @@ private[spark] class Client( .withRestartPolicy("OnFailure") .addNewVolume() .withName(SUBMISSION_APP_SECRET_VOLUME_NAME) -<<<<<<< HEAD - .withMountPath(secretDirectory) - .withReadOnly(true) - .endVolumeMount() - .addToVolumeMounts(sslVolumeMounts: _*) - .addNewEnv() - .withName(ENV_SUBMISSION_SECRET_LOCATION) - .withValue(s"$secretDirectory/$SUBMISSION_APP_SECRET_NAME") - .endEnv() - .addNewEnv() - .withName(ENV_SUBMISSION_SERVER_PORT) - .withValue(SUBMISSION_SERVER_PORT.toString) - .endEnv() - .addToEnv(sslEnvs: _*) - .withPorts(containerPorts.asJava) - .withNewReadinessProbe().withHttpGet(probePingHttpGet).endReadinessProbe() - .withNewLivenessProbe() - .withHttpGet(probePingHttpGet) - .withInitialDelaySeconds(DRIVER_INITIAL_LIVELINESS_CHECK_DELAY_SECONDS) - .endLivenessProbe() - .endContainer() - .endSpec() - .done() - var submitSucceeded = false - try { - submitCompletedFuture.get(driverSubmitTimeoutSecs, TimeUnit.SECONDS) - submitSucceeded = true - } catch { - case e: TimeoutException => - val finalErrorMessage: String = buildSubmitFailedErrorMessage(kubernetesClient, e) - logError(finalErrorMessage, e) - throw new SparkException(finalErrorMessage, e) - } finally { - if (!submitSucceeded) { - Utils.tryLogNonFatalError { - kubernetesClient.pods.withName(kubernetesAppId).delete() -||||||| merged common ancestors - .withMountPath(secretDirectory) - .withReadOnly(true) - .endVolumeMount() - .addToVolumeMounts(sslVolumeMounts: _*) - .addNewEnv() - .withName(ENV_SUBMISSION_SECRET_LOCATION) - .withValue(s"$secretDirectory/$SUBMISSION_APP_SECRET_NAME") - .endEnv() - .addNewEnv() - .withName(ENV_SUBMISSION_SERVER_PORT) - .withValue(SUBMISSION_SERVER_PORT.toString) - .endEnv() - .addToEnv(sslEnvs: _*) - .withPorts(containerPorts.asJava) - .endContainer() - .endSpec() - .done() - var submitSucceeded = false - try { - submitCompletedFuture.get(driverSubmitTimeoutSecs, TimeUnit.SECONDS) - submitSucceeded = true - } catch { - case e: TimeoutException => - val finalErrorMessage: String = buildSubmitFailedErrorMessage(kubernetesClient, e) - logError(finalErrorMessage, e) - throw new SparkException(finalErrorMessage, e) - } finally { - if (!submitSucceeded) { - Utils.tryLogNonFatalError { - kubernetesClient.pods.withName(kubernetesAppId).delete() -======= .withNewSecret() .withSecretName(submitServerSecret.getMetadata.getName) .endSecret() @@ -260,6 +179,11 @@ private[spark] class Client( .endEnv() .addToEnv(sslEnvs: _*) .withPorts(containerPorts.asJava) + .withNewReadinessProbe().withHttpGet(probePingHttpGet).endReadinessProbe() + .withNewLivenessProbe() + .withHttpGet(probePingHttpGet) + .withInitialDelaySeconds(DRIVER_INITIAL_LIVELINESS_CHECK_DELAY_SECONDS) + .endLivenessProbe() .endContainer() .endSpec() .done() @@ -473,7 +397,8 @@ private[spark] class Client( val driverSubmitter = buildDriverSubmissionClient(kubernetesClient, service, driverSubmitSslOptions) driverSubmitter.ping() -<<<<<<< HEAD + logInfo(s"Submitting local resources to driver pod for application " + + s"$kubernetesAppId ...") val submitRequest = buildSubmissionRequest() driverSubmitter.submitApplication(submitRequest) // After submitting, adjust the service to only expose the Spark UI @@ -494,97 +419,6 @@ private[spark] class Client( kubernetesClient.services().delete(service) }) throw e -||||||| merged common ancestors - } - ping onFailure { - case t: Throwable => - logError("Ping failed to the driver server", t) - submitCompletedFuture.setException(t) - kubernetesClient.services().delete(service) - } - val submitComplete = ping.flatMap { _ => - Future { - sparkConf.set("spark.driver.host", pod.getStatus.getPodIP) - val submitRequest = buildSubmissionRequest() - driverSubmitter.submitApplication(submitRequest) - } - } - submitComplete onFailure { - case t: Throwable => - submitCompletedFuture.setException(t) - kubernetesClient.services().delete(service) - } - val adjustServicePort = submitComplete.flatMap { _ => - Future { - // After submitting, adjust the service to only expose the Spark UI - val uiServicePort = new ServicePortBuilder() - .withName(UI_PORT_NAME) - .withPort(uiPort) - .withNewTargetPort(uiPort) - .build() - kubernetesClient.services().withName(kubernetesAppId).edit() - .editSpec() - .withType("ClusterIP") - .withPorts(uiServicePort) - .endSpec() - .done - } - } - adjustServicePort onSuccess { - case _ => - submitCompletedFuture.set(true) - } - adjustServicePort onFailure { - case throwable: Throwable => - submitCompletedFuture.setException(throwable) - kubernetesClient.services().delete(service) -======= - } - ping onFailure { - case t: Throwable => - logError("Ping failed to the driver server", t) - submitCompletedFuture.setException(t) - kubernetesClient.services().delete(service) - } - val submitComplete = ping.flatMap { _ => - Future { - sparkConf.set("spark.driver.host", pod.getStatus.getPodIP) - val submitRequest = buildSubmissionRequest() - logInfo(s"Submitting local resources to driver pod for application " + - s"$kubernetesAppId ...") - driverSubmitter.submitApplication(submitRequest) - } - } - submitComplete onFailure { - case t: Throwable => - submitCompletedFuture.setException(t) - kubernetesClient.services().delete(service) - } - val adjustServicePort = submitComplete.flatMap { _ => - Future { - // After submitting, adjust the service to only expose the Spark UI - val uiServicePort = new ServicePortBuilder() - .withName(UI_PORT_NAME) - .withPort(uiPort) - .withNewTargetPort(uiPort) - .build() - kubernetesClient.services().withName(kubernetesAppId).edit() - .editSpec() - .withType("ClusterIP") - .withPorts(uiServicePort) - .endSpec() - .done - } - } - adjustServicePort onSuccess { - case _ => - submitCompletedFuture.set(true) - } - adjustServicePort onFailure { - case throwable: Throwable => - submitCompletedFuture.setException(throwable) - kubernetesClient.services().delete(service) ->>>>>>> apache-spark-on-k8s/k8s-support-alternate-incremental } } catch { case e: Throwable => From afb2a084d19750420d8471888f40b397ef7b59e1 Mon Sep 17 00:00:00 2001 From: mcheah Date: Thu, 2 Feb 2017 18:15:13 -0800 Subject: [PATCH 07/19] Fix more merge conflicts --- .../main/scala/org/apache/spark/deploy/kubernetes/Client.scala | 1 - 1 file changed, 1 deletion(-) diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala index 4132ee32d0e27..8362b43b7de30 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala @@ -202,7 +202,6 @@ private[spark] class Client( Utils.tryLogNonFatalError { kubernetesClient.pods.withName(kubernetesAppId).delete() } ->>>>>>> apache-spark-on-k8s/k8s-support-alternate-incremental } } } From dcc9c539cc6a53be7cb1b7a603826dcc2c0fc632 Mon Sep 17 00:00:00 2001 From: mcheah Date: Fri, 3 Feb 2017 17:43:40 -0800 Subject: [PATCH 08/19] Revamp readiness check logic --- .../spark/deploy/kubernetes/Client.scala | 587 +++++++++++------- 1 file changed, 368 insertions(+), 219 deletions(-) diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala index 8362b43b7de30..2a971b9041515 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala @@ -18,8 +18,8 @@ package org.apache.spark.deploy.kubernetes import java.io.{File, FileInputStream} import java.security.{KeyStore, SecureRandom} -import java.util.concurrent.{CountDownLatch, TimeoutException, TimeUnit} -import java.util.concurrent.atomic.AtomicBoolean +import java.util +import java.util.concurrent.{CountDownLatch, TimeUnit} import javax.net.ssl.{SSLContext, TrustManagerFactory, X509TrustManager} import com.google.common.base.Charsets @@ -104,126 +104,379 @@ private[spark] class Client( .withData(Map((SUBMISSION_APP_SECRET_NAME, secretBase64String)).asJava) .withType("Opaque") .done() - val (sslEnvs, sslVolumes, sslVolumeMounts, sslSecrets) = configureSsl(kubernetesClient, - driverSubmitSslOptions, - isKeyStoreLocalFile) try { - val driverKubernetesSelectors = (Map( - SPARK_DRIVER_LABEL -> kubernetesAppId, - SPARK_APP_ID_LABEL -> kubernetesAppId, - SPARK_APP_NAME_LABEL -> appName) - ++ parsedCustomLabels).asJava - val containerPorts = buildContainerPorts() - - // start outer watch for status logging of driver pod - val driverPodCompletedLatch = new CountDownLatch(1) - // only enable interval logging if in waitForAppCompletion mode - val loggingInterval = if (waitForAppCompletion) sparkConf.get(REPORT_INTERVAL) else 0 - val loggingWatch = new LoggingPodStatusWatcher(driverPodCompletedLatch, kubernetesAppId, - loggingInterval) - Utils.tryWithResource(kubernetesClient - .pods() - .withLabels(driverKubernetesSelectors) - .watch(loggingWatch)) { _ => - val probePingHttpGet = new HTTPGetActionBuilder() - .withScheme(if (driverSubmitSslOptions.enabled) "HTTPS" else "HTTP") - .withPath("/v1/submissions/ping") - .withNewPort(SUBMISSION_SERVER_PORT_NAME) - .build() - // launch driver pod with inner watch to upload jars when it's ready - val submitCompletedFuture = SettableFuture.create[Boolean] - val submitPending = new AtomicBoolean(false) - val podWatcher = new DriverPodWatcher( - submitCompletedFuture, - submitPending, - kubernetesClient, - driverSubmitSslOptions, - Array(submitServerSecret) ++ sslSecrets, - driverKubernetesSelectors) + val (sslEnvs, sslVolumes, sslVolumeMounts, sslSecrets) = configureSsl(kubernetesClient, + driverSubmitSslOptions, + isKeyStoreLocalFile) + try { + // start outer watch for status logging of driver pod + val driverPodCompletedLatch = new CountDownLatch(1) + // only enable interval logging if in waitForAppCompletion mode + val loggingInterval = if (waitForAppCompletion) sparkConf.get(REPORT_INTERVAL) else 0 + val loggingWatch = new LoggingPodStatusWatcher(driverPodCompletedLatch, kubernetesAppId, + loggingInterval) Utils.tryWithResource(kubernetesClient .pods() - .withLabels(driverKubernetesSelectors) - .watch(podWatcher)) { _ => - kubernetesClient.pods().createNew() - .withNewMetadata() - .withName(kubernetesAppId) - .withLabels(driverKubernetesSelectors) - .endMetadata() - .withNewSpec() - .withRestartPolicy("OnFailure") - .addNewVolume() - .withName(SUBMISSION_APP_SECRET_VOLUME_NAME) - .withNewSecret() - .withSecretName(submitServerSecret.getMetadata.getName) - .endSecret() - .endVolume - .addToVolumes(sslVolumes: _*) - .withServiceAccount(serviceAccount) - .addNewContainer() - .withName(DRIVER_CONTAINER_NAME) - .withImage(driverDockerImage) - .withImagePullPolicy("IfNotPresent") - .addNewVolumeMount() - .withName(SUBMISSION_APP_SECRET_VOLUME_NAME) - .withMountPath(secretDirectory) - .withReadOnly(true) - .endVolumeMount() - .addToVolumeMounts(sslVolumeMounts: _*) - .addNewEnv() - .withName(ENV_SUBMISSION_SECRET_LOCATION) - .withValue(s"$secretDirectory/$SUBMISSION_APP_SECRET_NAME") - .endEnv() - .addNewEnv() - .withName(ENV_SUBMISSION_SERVER_PORT) - .withValue(SUBMISSION_SERVER_PORT.toString) - .endEnv() - .addToEnv(sslEnvs: _*) - .withPorts(containerPorts.asJava) - .withNewReadinessProbe().withHttpGet(probePingHttpGet).endReadinessProbe() - .withNewLivenessProbe() - .withHttpGet(probePingHttpGet) - .withInitialDelaySeconds(DRIVER_INITIAL_LIVELINESS_CHECK_DELAY_SECONDS) - .endLivenessProbe() - .endContainer() - .endSpec() - .done() - var submitSucceeded = false + .withName(kubernetesAppId) + .watch(loggingWatch)) { _ => + val (driverPod, driverService) = launchDriverKubernetesComponents( + kubernetesClient, + parsedCustomLabels, + submitServerSecret, + driverSubmitSslOptions, + sslSecrets, + sslVolumes, + sslVolumeMounts, + sslEnvs, + isKeyStoreLocalFile) try { - submitCompletedFuture.get(driverSubmitTimeoutSecs, TimeUnit.SECONDS) - submitSucceeded = true - logInfo(s"Finished launching local resources to application $kubernetesAppId") + submitApplicationToDriverServer(kubernetesClient, driverSubmitSslOptions, + driverService) + // wait if configured to do so + if (waitForAppCompletion) { + logInfo(s"Waiting for application $kubernetesAppId to finish...") + driverPodCompletedLatch.await() + logInfo(s"Application $kubernetesAppId finished.") + } else { + logInfo(s"Application $kubernetesAppId successfully launched.") + } } catch { - case e: TimeoutException => - val finalErrorMessage: String = buildSubmitFailedErrorMessage(kubernetesClient, e) - logError(finalErrorMessage, e) - throw new SparkException(finalErrorMessage, e) - } finally { - if (!submitSucceeded) { + case e: Throwable => Utils.tryLogNonFatalError { - kubernetesClient.pods.withName(kubernetesAppId).delete() + kubernetesClient.pods().delete(driverPod) } - } + Utils.tryLogNonFatalError { + kubernetesClient.services().delete(driverService) + } + throw e } } - - // wait if configured to do so - if (waitForAppCompletion) { - logInfo(s"Waiting for application $kubernetesAppId to finish...") - driverPodCompletedLatch.await() - logInfo(s"Application $kubernetesAppId finished.") - } else { - logInfo(s"Application $kubernetesAppId successfully launched.") + } finally { + Utils.tryLogNonFatalError { + // Secrets may have been mutated so delete by name to avoid problems with not having + // the latest version. + sslSecrets.foreach { secret => + kubernetesClient.secrets().withName(secret.getMetadata.getName).delete() + } + kubernetesClient.secrets().delete(sslSecrets: _*) } } } finally { Utils.tryLogNonFatalError { - kubernetesClient.secrets().delete(submitServerSecret) + kubernetesClient.secrets().withName(submitServerSecret.getMetadata.getName).delete() } + } + } + } + + private def submitApplicationToDriverServer( + kubernetesClient: KubernetesClient, + driverSubmitSslOptions: SSLOptions, + driverService: Service) = { + sparkConf.getOption("spark.app.id").foreach { id => + logWarning(s"Warning: Provided app id in spark.app.id as $id will be" + + s" overridden as $kubernetesAppId") + } + sparkConf.set(KUBERNETES_DRIVER_POD_NAME, kubernetesAppId) + sparkConf.set(KUBERNETES_DRIVER_SERVICE_NAME, driverService.getMetadata.getName) + sparkConf.set("spark.app.id", kubernetesAppId) + sparkConf.setIfMissing("spark.app.name", appName) + sparkConf.setIfMissing("spark.driver.port", DEFAULT_DRIVER_PORT.toString) + sparkConf.setIfMissing("spark.blockmanager.port", + DEFAULT_BLOCKMANAGER_PORT.toString) + val driverSubmitter = buildDriverSubmissionClient(kubernetesClient, driverService, + driverSubmitSslOptions) + // Sanity check to see if the driver submitter is even reachable. + driverSubmitter.ping() + logInfo(s"Submitting local resources to driver pod for application " + + s"$kubernetesAppId ...") + val submitRequest = buildSubmissionRequest() + driverSubmitter.submitApplication(submitRequest) + logInfo("Successfully submitted local resources and driver configuration to" + + " driver pod.") + // After submitting, adjust the service to only expose the Spark UI + val uiServicePort = new ServicePortBuilder() + .withName(UI_PORT_NAME) + .withPort(uiPort) + .withNewTargetPort(uiPort) + .build() + kubernetesClient.services().withName(kubernetesAppId).edit().editSpec() + .withType("ClusterIP") + .withPorts(uiServicePort) + .endSpec() + .done() + logInfo("Finished submitting application to Kubernetes.") + } + + private def launchDriverKubernetesComponents( + kubernetesClient: KubernetesClient, + parsedCustomLabels: Map[String, String], + submitServerSecret: Secret, + driverSubmitSslOptions: SSLOptions, + sslSecrets: Array[Secret], + sslVolumes: Array[Volume], + sslVolumeMounts: Array[VolumeMount], + sslEnvs: Array[EnvVar], + isKeyStoreLocalFile: Boolean): (Pod, Service) = { + val endpointsReadyFuture = SettableFuture.create[Endpoints] + val endpointsReadyWatcher = new DriverEndpointsReadyWatcher(endpointsReadyFuture) + val serviceReadyFuture = SettableFuture.create[Service] + val serviceReadyWatcher = new DriverServiceReadyWatcher(serviceReadyFuture) + val podReadyFuture = SettableFuture.create[Pod] + val podWatcher = new DriverPodReadyWatcher(podReadyFuture) + val (readyPod, readyService) = Utils.tryWithResource(kubernetesClient + .pods() + .withName(kubernetesAppId) + .watch(podWatcher)) { _ => + Utils.tryWithResource(kubernetesClient + .services() + .withName(kubernetesAppId) + .watch(serviceReadyWatcher)) { _ => + Utils.tryWithResource(kubernetesClient + .endpoints() + .withName(kubernetesAppId) + .watch(endpointsReadyWatcher)) { _ => + val driverKubernetesSelectors = (Map( + SPARK_DRIVER_LABEL -> kubernetesAppId, + SPARK_APP_ID_LABEL -> kubernetesAppId, + SPARK_APP_NAME_LABEL -> appName) + ++ parsedCustomLabels).asJava + val driverService = createDriverService( + kubernetesClient, + driverKubernetesSelectors, + submitServerSecret) + val driverPod = try { + createDriverPod( + kubernetesClient, + driverKubernetesSelectors, + submitServerSecret, + driverSubmitSslOptions, + sslVolumes, + sslVolumeMounts, + sslEnvs) + } catch { + case e: Throwable => + Utils.tryLogNonFatalError { + kubernetesClient.services().delete(driverService) + } + throw e + } + try { + waitForReadyKubernetesComponents(kubernetesClient, endpointsReadyFuture, + serviceReadyFuture, podReadyFuture) + } catch { + case e: Throwable => + Utils.tryLogNonFatalError { + kubernetesClient.services().delete(driverService) + } + Utils.tryLogNonFatalError { + kubernetesClient.pods().delete(driverPod) + } + throw new SparkException("Timed out while waiting for a Kubernetes component to be" + + " ready.", e) + } + (driverPod, driverService) + } + } + } + val resolvedDriverService = try { + val driverPodOwnerRef = new OwnerReferenceBuilder() + .withName(readyPod.getMetadata.getName) + .withUid(readyPod.getMetadata.getUid) + .withApiVersion(readyPod.getApiVersion) + .withKind(readyPod.getKind) + .withController(true) + .build() + sslSecrets.foreach(secret => { + kubernetesClient.secrets().withName(secret.getMetadata.getName).edit() + .editMetadata() + .addToOwnerReferences(driverPodOwnerRef) + .endMetadata() + .done() + }) + kubernetesClient.secrets().withName(submitServerSecret.getMetadata.getName).edit() + .editMetadata() + .addToOwnerReferences(driverPodOwnerRef) + .endMetadata() + .done() + kubernetesClient.services().withName(readyService.getMetadata.getName).edit() + .editMetadata() + .addToOwnerReferences(driverPodOwnerRef) + .endMetadata() + .done() + } catch { + case e: Throwable => Utils.tryLogNonFatalError { - kubernetesClient.secrets().delete(sslSecrets: _*) + kubernetesClient.services().delete(readyService) } + Utils.tryLogNonFatalError { + kubernetesClient.pods().delete(readyPod) + } + throw e + } + (readyPod, resolvedDriverService) + } + + private def waitForReadyKubernetesComponents( + kubernetesClient: KubernetesClient, + endpointsReadyFuture: SettableFuture[Endpoints], + serviceReadyFuture: SettableFuture[Service], + podReadyFuture: SettableFuture[Pod]) = { + try { + podReadyFuture.get(driverSubmitTimeoutSecs, TimeUnit.SECONDS) + logInfo("Driver pod successfully created in Kubernetes cluster.") + } catch { + case e: Throwable => + val finalErrorMessage: String = buildSubmitFailedErrorMessage(kubernetesClient, e) + logError(finalErrorMessage, e) + throw new SparkException(finalErrorMessage, e) + } + try { + serviceReadyFuture.get(driverSubmitTimeoutSecs, TimeUnit.SECONDS) + logInfo("Driver service created successfully in Kubernetes.") + } catch { + case e: Throwable => + throw new SparkException(s"The driver service was not ready" + + s" in $driverSubmitTimeoutSecs seconds.", e) + } + try { + endpointsReadyFuture.get(driverSubmitTimeoutSecs, TimeUnit.SECONDS) + logInfo("Driver endpoints ready to receive application submission") + } catch { + case e: Throwable => + throw new SparkException(s"The driver service endpoint was not ready" + + s" in $driverSubmitTimeoutSecs seconds.", e) + } + } + + private def createDriverService( + kubernetesClient: KubernetesClient, + driverKubernetesSelectors: java.util.Map[String, String], + submitServerSecret: Secret): Service = { + val driverSubmissionServicePort = new ServicePortBuilder() + .withName(SUBMISSION_SERVER_PORT_NAME) + .withPort(SUBMISSION_SERVER_PORT) + .withNewTargetPort(SUBMISSION_SERVER_PORT) + .build() + kubernetesClient.services().createNew() + .withNewMetadata() + .withName(kubernetesAppId) + .withLabels(driverKubernetesSelectors) + .endMetadata() + .withNewSpec() + .withType("NodePort") + .withSelector(driverKubernetesSelectors) + .withPorts(driverSubmissionServicePort) + .endSpec() + .done() + } + + private def createDriverPod( + kubernetesClient: KubernetesClient, + driverKubernetesSelectors: util.Map[String, String], + submitServerSecret: Secret, + driverSubmitSslOptions: SSLOptions, + sslVolumes: Array[Volume], + sslVolumeMounts: Array[VolumeMount], + sslEnvs: Array[EnvVar]) = { + val containerPorts = buildContainerPorts() + val probePingHttpGet = new HTTPGetActionBuilder() + .withScheme(if (driverSubmitSslOptions.enabled) "HTTPS" else "HTTP") + .withPath("/v1/submissions/ping") + .withNewPort(SUBMISSION_SERVER_PORT_NAME) + .build() + kubernetesClient.pods().createNew() + .withNewMetadata() + .withName(kubernetesAppId) + .withLabels(driverKubernetesSelectors) + .endMetadata() + .withNewSpec() + .withRestartPolicy("OnFailure") + .addNewVolume() + .withName(SUBMISSION_APP_SECRET_VOLUME_NAME) + .withNewSecret() + .withSecretName(submitServerSecret.getMetadata.getName) + .endSecret() + .endVolume() + .addToVolumes(sslVolumes: _*) + .withServiceAccount(serviceAccount) + .addNewContainer() + .withName(DRIVER_CONTAINER_NAME) + .withImage(driverDockerImage) + .withImagePullPolicy("IfNotPresent") + .addNewVolumeMount() + .withName(SUBMISSION_APP_SECRET_VOLUME_NAME) + .withMountPath(secretDirectory) + .withReadOnly(true) + .endVolumeMount() + .addToVolumeMounts(sslVolumeMounts: _*) + .addNewEnv() + .withName(ENV_SUBMISSION_SECRET_LOCATION) + .withValue(s"$secretDirectory/$SUBMISSION_APP_SECRET_NAME") + .endEnv() + .addNewEnv() + .withName(ENV_SUBMISSION_SERVER_PORT) + .withValue(SUBMISSION_SERVER_PORT.toString) + .endEnv() + .addToEnv(sslEnvs: _*) + .withPorts(containerPorts.asJava) + .withNewReadinessProbe() + .withHttpGet(probePingHttpGet) + .endReadinessProbe() + .withNewLivenessProbe() + .withHttpGet(probePingHttpGet) + .withInitialDelaySeconds(DRIVER_INITIAL_LIVELINESS_CHECK_DELAY_SECONDS) + .endLivenessProbe() + .endContainer() + .endSpec() + .done() + } + + private class DriverPodReadyWatcher(resolvedDriverPod: SettableFuture[Pod]) extends Watcher[Pod] { + override def eventReceived(action: Action, pod: Pod): Unit = { + if ((action == Action.ADDED || action == Action.MODIFIED) + && pod.getStatus.getPhase == "Running" + && !resolvedDriverPod.isDone) { + pod.getStatus + .getContainerStatuses + .asScala + .find(status => + status.getName == DRIVER_CONTAINER_NAME && status.getReady) + .foreach { _ => resolvedDriverPod.set(pod) } } } + + override def onClose(cause: KubernetesClientException): Unit = { + logDebug("Driver pod readiness watch closed.", cause) + } + } + + private class DriverEndpointsReadyWatcher(resolvedDriverEndpoints: SettableFuture[Endpoints]) + extends Watcher[Endpoints] { + override def eventReceived(action: Action, endpoints: Endpoints): Unit = { + if ((action == Action.ADDED) || (action == Action.MODIFIED) + && !resolvedDriverEndpoints.isDone) { + resolvedDriverEndpoints.set(endpoints) + } + } + + override def onClose(cause: KubernetesClientException): Unit = { + logDebug("Driver endpoints readiness watch closed.", cause) + } + } + + private class DriverServiceReadyWatcher(resolvedDriverService: SettableFuture[Service]) + extends Watcher[Service] { + override def eventReceived(action: Action, service: Service): Unit = { + if ((action == Action.ADDED) || (action == Action.MODIFIED) + && !resolvedDriverService.isDone) { + resolvedDriverService.set(service) + } + } + + override def onClose(cause: KubernetesClientException): Unit = { + logDebug("Driver service readiness watch closed.", cause) + } } private def parseDriverSubmitSslOptions(): (SSLOptions, Boolean) = { @@ -307,18 +560,10 @@ private[spark] class Client( .withName(ENV_SUBMISSION_USE_SSL) .withValue("true") .build() - val sslSecrets = kubernetesClient.secrets().createNew() - .withNewMetadata() - .withName(sslSecretsName) - .endMetadata() - .withData(sslSecretsMap.asJava) - .withType("Opaque") - .done() - secrets += sslSecrets val sslVolume = new VolumeBuilder() .withName(SUBMISSION_SSL_SECRETS_VOLUME_NAME) .withNewSecret() - .withSecretName(sslSecrets.getMetadata.getName) + .withSecretName(sslSecretsName) .endSecret() .build() val sslVolumeMount = new VolumeMountBuilder() @@ -326,120 +571,24 @@ private[spark] class Client( .withReadOnly(true) .withMountPath(sslSecretsDirectory) .build() + val sslSecrets = kubernetesClient.secrets().createNew() + .withNewMetadata() + .withName(sslSecretsName) + .endMetadata() + .withData(sslSecretsMap.asJava) + .withType("Opaque") + .done() + secrets += sslSecrets (sslEnvs.toArray, Array(sslVolume), Array(sslVolumeMount), secrets.toArray) } else { (Array[EnvVar](), Array[Volume](), Array[VolumeMount](), Array[Secret]()) } } - private class DriverPodWatcher( - submitCompletedFuture: SettableFuture[Boolean], - submitPending: AtomicBoolean, - kubernetesClient: KubernetesClient, - driverSubmitSslOptions: SSLOptions, - applicationSecrets: Array[Secret], - driverKubernetesSelectors: java.util.Map[String, String]) extends Watcher[Pod] { - override def eventReceived(action: Action, pod: Pod): Unit = { - if ((action == Action.ADDED || action == Action.MODIFIED) - && pod.getStatus.getPhase == "Running" - && !submitCompletedFuture.isDone) { - pod.getStatus - .getContainerStatuses - .asScala - .find(status => - status.getName == DRIVER_CONTAINER_NAME && status.getReady) match { - case Some(_) => - if (!submitPending.getAndSet(true)) { - try { - val ownerRefs = Seq(new OwnerReferenceBuilder() - .withName(pod.getMetadata.getName) - .withUid(pod.getMetadata.getUid) - .withApiVersion(pod.getApiVersion) - .withKind(pod.getKind) - .withController(true) - .build()) - applicationSecrets.foreach(secret => { - secret.getMetadata.setOwnerReferences(ownerRefs.asJava) - kubernetesClient.secrets().createOrReplace(secret) - }) - val driverSubmissionServicePort = new ServicePortBuilder() - .withName(SUBMISSION_SERVER_PORT_NAME) - .withPort(SUBMISSION_SERVER_PORT) - .withNewTargetPort(SUBMISSION_SERVER_PORT) - .build() - - val service = kubernetesClient.services().createNew() - .withNewMetadata() - .withName(kubernetesAppId) - .withLabels(driverKubernetesSelectors) - .withOwnerReferences(ownerRefs.asJava) - .endMetadata() - .withNewSpec() - .withType("NodePort") - .withSelector(driverKubernetesSelectors) - .withPorts(driverSubmissionServicePort) - .endSpec() - .done() - - try { - sparkConf.getOption("spark.app.id").foreach { id => - logWarning(s"Warning: Provided app id in spark.app.id as $id will be" + - s" overridden as $kubernetesAppId") - } - sparkConf.set(KUBERNETES_DRIVER_POD_NAME, kubernetesAppId) - sparkConf.set(KUBERNETES_DRIVER_SERVICE_NAME, service.getMetadata.getName) - sparkConf.set("spark.app.id", kubernetesAppId) - sparkConf.setIfMissing("spark.app.name", appName) - sparkConf.setIfMissing("spark.driver.port", DEFAULT_DRIVER_PORT.toString) - sparkConf.setIfMissing("spark.blockmanager.port", - DEFAULT_BLOCKMANAGER_PORT.toString) - val driverSubmitter = buildDriverSubmissionClient(kubernetesClient, service, - driverSubmitSslOptions) - driverSubmitter.ping() - logInfo(s"Submitting local resources to driver pod for application " + - s"$kubernetesAppId ...") - val submitRequest = buildSubmissionRequest() - driverSubmitter.submitApplication(submitRequest) - // After submitting, adjust the service to only expose the Spark UI - val uiServicePort = new ServicePortBuilder() - .withName(UI_PORT_NAME) - .withPort(uiPort) - .withNewTargetPort(uiPort) - .build() - kubernetesClient.services().withName(kubernetesAppId).edit().editSpec() - .withType("ClusterIP") - .withPorts(uiServicePort) - .endSpec() - .done() - submitCompletedFuture.set(true) - } catch { - case e: Throwable => - Utils.tryLogNonFatalError({ - kubernetesClient.services().delete(service) - }) - throw e - } - } catch { - case e: Throwable => - submitCompletedFuture.setException(e) - throw e - } - } - case None => - } - } - } - - override def onClose(e: KubernetesClientException): Unit = { - if (!submitCompletedFuture.isDone) { - submitCompletedFuture.setException(e) - } - } - } private def buildSubmitFailedErrorMessage( - kubernetesClient: DefaultKubernetesClient, - e: TimeoutException): String = { + kubernetesClient: KubernetesClient, + e: Throwable): String = { val driverPod = try { kubernetesClient.pods().withName(kubernetesAppId).get() } catch { From 606accb905967cb6ae6e64672cd4af7cd81fa98d Mon Sep 17 00:00:00 2001 From: mcheah Date: Fri, 3 Feb 2017 17:56:40 -0800 Subject: [PATCH 09/19] Add addresses ready condition to endpoints watch --- .../main/scala/org/apache/spark/deploy/kubernetes/Client.scala | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala index 2a971b9041515..db40d7d79f4e8 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala @@ -455,6 +455,8 @@ private[spark] class Client( extends Watcher[Endpoints] { override def eventReceived(action: Action, endpoints: Endpoints): Unit = { if ((action == Action.ADDED) || (action == Action.MODIFIED) + && endpoints.getSubsets.asScala.nonEmpty + && endpoints.getSubsets.asScala.exists(_.getAddresses.asScala.nonEmpty) && !resolvedDriverEndpoints.isDone) { resolvedDriverEndpoints.set(endpoints) } @@ -585,7 +587,6 @@ private[spark] class Client( } } - private def buildSubmitFailedErrorMessage( kubernetesClient: KubernetesClient, e: Throwable): String = { From ec990b2fb81bff7cd68ab7238489894c36cc452d Mon Sep 17 00:00:00 2001 From: mcheah Date: Mon, 6 Feb 2017 14:27:15 -0800 Subject: [PATCH 10/19] Rearrange the logic some more. --- .../spark/deploy/kubernetes/Client.scala | 112 ++++++++++-------- 1 file changed, 65 insertions(+), 47 deletions(-) diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala index db40d7d79f4e8..bb21fe5f5b91a 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala @@ -129,9 +129,26 @@ private[spark] class Client( sslVolumeMounts, sslEnvs, isKeyStoreLocalFile) + val ownerReferenceConfiguredDriverService = try { + configureOwnerReferences( + kubernetesClient, + submitServerSecret, + sslSecrets, + driverPod, + driverService) + } catch { + case e: Throwable => + Utils.tryLogNonFatalError { + kubernetesClient.services().delete(driverService) + } + Utils.tryLogNonFatalError { + kubernetesClient.pods().delete(driverPod) + } + throw new SparkException("Failed to set owner references to the driver pod.", e) + } try { submitApplicationToDriverServer(kubernetesClient, driverSubmitSslOptions, - driverService) + ownerReferenceConfiguredDriverService) // wait if configured to do so if (waitForAppCompletion) { logInfo(s"Waiting for application $kubernetesAppId to finish...") @@ -143,12 +160,12 @@ private[spark] class Client( } catch { case e: Throwable => Utils.tryLogNonFatalError { - kubernetesClient.pods().delete(driverPod) + kubernetesClient.services().delete(ownerReferenceConfiguredDriverService) } Utils.tryLogNonFatalError { - kubernetesClient.services().delete(driverService) + kubernetesClient.pods().delete(driverPod) } - throw e + throw new SparkException("Failed to submit the application to the driver pod.", e) } } } finally { @@ -158,7 +175,6 @@ private[spark] class Client( sslSecrets.foreach { secret => kubernetesClient.secrets().withName(secret.getMetadata.getName).delete() } - kubernetesClient.secrets().delete(sslSecrets: _*) } } } finally { @@ -221,10 +237,15 @@ private[spark] class Client( val endpointsReadyFuture = SettableFuture.create[Endpoints] val endpointsReadyWatcher = new DriverEndpointsReadyWatcher(endpointsReadyFuture) val serviceReadyFuture = SettableFuture.create[Service] + val driverKubernetesSelectors = (Map( + SPARK_DRIVER_LABEL -> kubernetesAppId, + SPARK_APP_ID_LABEL -> kubernetesAppId, + SPARK_APP_NAME_LABEL -> appName) + ++ parsedCustomLabels).asJava val serviceReadyWatcher = new DriverServiceReadyWatcher(serviceReadyFuture) val podReadyFuture = SettableFuture.create[Pod] val podWatcher = new DriverPodReadyWatcher(podReadyFuture) - val (readyPod, readyService) = Utils.tryWithResource(kubernetesClient + Utils.tryWithResource(kubernetesClient .pods() .withName(kubernetesAppId) .watch(podWatcher)) { _ => @@ -236,11 +257,6 @@ private[spark] class Client( .endpoints() .withName(kubernetesAppId) .watch(endpointsReadyWatcher)) { _ => - val driverKubernetesSelectors = (Map( - SPARK_DRIVER_LABEL -> kubernetesAppId, - SPARK_APP_ID_LABEL -> kubernetesAppId, - SPARK_APP_NAME_LABEL -> appName) - ++ parsedCustomLabels).asJava val driverService = createDriverService( kubernetesClient, driverKubernetesSelectors, @@ -259,11 +275,12 @@ private[spark] class Client( Utils.tryLogNonFatalError { kubernetesClient.services().delete(driverService) } - throw e + throw new SparkException("Failed to create the driver pod.", e) } try { waitForReadyKubernetesComponents(kubernetesClient, endpointsReadyFuture, serviceReadyFuture, podReadyFuture) + (driverPod, driverService) } catch { case e: Throwable => Utils.tryLogNonFatalError { @@ -275,46 +292,47 @@ private[spark] class Client( throw new SparkException("Timed out while waiting for a Kubernetes component to be" + " ready.", e) } - (driverPod, driverService) } } } - val resolvedDriverService = try { - val driverPodOwnerRef = new OwnerReferenceBuilder() - .withName(readyPod.getMetadata.getName) - .withUid(readyPod.getMetadata.getUid) - .withApiVersion(readyPod.getApiVersion) - .withKind(readyPod.getKind) - .withController(true) - .build() - sslSecrets.foreach(secret => { - kubernetesClient.secrets().withName(secret.getMetadata.getName).edit() - .editMetadata() - .addToOwnerReferences(driverPodOwnerRef) - .endMetadata() - .done() - }) - kubernetesClient.secrets().withName(submitServerSecret.getMetadata.getName).edit() - .editMetadata() - .addToOwnerReferences(driverPodOwnerRef) - .endMetadata() - .done() - kubernetesClient.services().withName(readyService.getMetadata.getName).edit() + } + + /** + * Sets the owner reference for all the kubernetes components to link to the driver pod. + * + * @return The driver service after it has been adjusted to reflect the new owner + * reference. + */ + private def configureOwnerReferences( + kubernetesClient: KubernetesClient, + submitServerSecret: Secret, + sslSecrets: Array[Secret], + driverPod: Pod, + driverService: Service): Service = { + val driverPodOwnerRef = new OwnerReferenceBuilder() + .withName(driverPod.getMetadata.getName) + .withUid(driverPod.getMetadata.getUid) + .withApiVersion(driverPod.getApiVersion) + .withKind(driverPod.getKind) + .withController(true) + .build() + sslSecrets.foreach(secret => { + kubernetesClient.secrets().withName(secret.getMetadata.getName).edit() .editMetadata() - .addToOwnerReferences(driverPodOwnerRef) - .endMetadata() + .addToOwnerReferences(driverPodOwnerRef) + .endMetadata() .done() - } catch { - case e: Throwable => - Utils.tryLogNonFatalError { - kubernetesClient.services().delete(readyService) - } - Utils.tryLogNonFatalError { - kubernetesClient.pods().delete(readyPod) - } - throw e - } - (readyPod, resolvedDriverService) + }) + kubernetesClient.secrets().withName(submitServerSecret.getMetadata.getName).edit() + .editMetadata() + .addToOwnerReferences(driverPodOwnerRef) + .endMetadata() + .done() + kubernetesClient.services().withName(driverService.getMetadata.getName).edit() + .editMetadata() + .addToOwnerReferences(driverPodOwnerRef) + .endMetadata() + .done() } private def waitForReadyKubernetesComponents( From 977ca4217ba8bc766f9a90f12e2ee718e46b4d4d Mon Sep 17 00:00:00 2001 From: mcheah Date: Wed, 8 Feb 2017 11:26:50 -0800 Subject: [PATCH 11/19] Remove liveness probe, retry against servers --- .../spark/deploy/kubernetes/Client.scala | 9 ++---- .../rest/kubernetes/HttpClientUtil.scala | 1 + .../kubernetes/MultiServerFeignTarget.scala | 30 ++++++++++++++++--- .../integrationtest/KubernetesSuite.scala | 3 +- 4 files changed, 31 insertions(+), 12 deletions(-) diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala index bb21fe5f5b91a..82987860bb1e0 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala @@ -438,13 +438,7 @@ private[spark] class Client( .endEnv() .addToEnv(sslEnvs: _*) .withPorts(containerPorts.asJava) - .withNewReadinessProbe() - .withHttpGet(probePingHttpGet) - .endReadinessProbe() - .withNewLivenessProbe() - .withHttpGet(probePingHttpGet) - .withInitialDelaySeconds(DRIVER_INITIAL_LIVELINESS_CHECK_DELAY_SECONDS) - .endLivenessProbe() + .withNewReadinessProbe().withHttpGet(probePingHttpGet).endReadinessProbe() .endContainer() .endSpec() .done() @@ -758,6 +752,7 @@ private[spark] class Client( } HttpClientUtil.createClient[KubernetesSparkRestApi]( uris = nodeUrls, + maxRetriesPerServer = 10, sslSocketFactory = sslContext.getSocketFactory, trustContext = trustManager) } diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/HttpClientUtil.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/HttpClientUtil.scala index 1cabfbad656eb..733ce48b26b79 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/HttpClientUtil.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/HttpClientUtil.scala @@ -33,6 +33,7 @@ private[spark] object HttpClientUtil { def createClient[T: ClassTag]( uris: Array[String], + maxRetriesPerServer: Int = 1, sslSocketFactory: SSLSocketFactory = SSLContext.getDefault.getSocketFactory, trustContext: X509TrustManager = null, readTimeoutMillis: Int = 20000, diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/MultiServerFeignTarget.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/MultiServerFeignTarget.scala index fea7f057cfa1b..fe5b7a0d94337 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/MultiServerFeignTarget.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/MultiServerFeignTarget.scala @@ -20,13 +20,19 @@ import feign.{Request, RequestTemplate, RetryableException, Retryer, Target} import scala.reflect.ClassTag import scala.util.Random +import org.apache.spark.internal.Logging + private[kubernetes] class MultiServerFeignTarget[T : ClassTag]( - private val servers: Seq[String]) extends Target[T] with Retryer { + private val servers: Seq[String], + private val maxRetriesPerServer: Int = 1) extends Target[T] with Retryer with Logging { require(servers.nonEmpty, "Must provide at least one server URI.") private val threadLocalShuffledServers = new ThreadLocal[Seq[String]] { override def initialValue(): Seq[String] = Random.shuffle(servers) } + private val threadLocalCurrentAttempt = new ThreadLocal[Int] { + override def initialValue(): Int = 0 + } override def `type`(): Class[T] = { implicitly[ClassTag[T]].runtimeClass.asInstanceOf[Class[T]] @@ -54,14 +60,30 @@ private[kubernetes] class MultiServerFeignTarget[T : ClassTag]( requestTemplate.request() } + override def url(): String = threadLocalShuffledServers.get.head + override def continueOrPropagate(e: RetryableException): Unit = { - threadLocalShuffledServers.set(threadLocalShuffledServers.get.drop(1)) - if (threadLocalShuffledServers.get.isEmpty) { - throw e + threadLocalCurrentAttempt.set(threadLocalCurrentAttempt.get + 1) + val currentAttempt = threadLocalCurrentAttempt.get + if (threadLocalCurrentAttempt.get < maxRetriesPerServer) { + logWarning(s"Attempt $currentAttempt of $maxRetriesPerServer failed for" + + s" server ${url()}. Retrying request...", e) + } else { + val previousUrl = url() + threadLocalShuffledServers.set(threadLocalShuffledServers.get.drop(1)) + if (threadLocalShuffledServers.get.isEmpty) { + logError(s"Failed request to all servers $maxRetriesPerServer times.", e) + throw e + } else { + logWarning(s"Failed request to $previousUrl $maxRetriesPerServer times." + + s" Trying to access ${url()} instead.", e) + threadLocalCurrentAttempt.set(0) + } } } def reset(): Unit = { threadLocalShuffledServers.set(Random.shuffle(servers)) + threadLocalCurrentAttempt.set(0) } } diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/KubernetesSuite.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/KubernetesSuite.scala index c5458eccf830d..394f02333467b 100644 --- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/KubernetesSuite.scala +++ b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/KubernetesSuite.scala @@ -122,11 +122,12 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter { }) } + /** override def afterAll(): Unit = { if (!System.getProperty("spark.docker.test.persistMinikube", "false").toBoolean) { Minikube.deleteMinikube() } - } + }*/ private def getSparkMetricsService(sparkBaseAppName: String): SparkRestApiV1 = { val serviceName = minikubeKubernetesClient.services() From 7be274a9b764207510059da24d3c141f7cc15a6d Mon Sep 17 00:00:00 2001 From: mcheah Date: Wed, 8 Feb 2017 11:29:26 -0800 Subject: [PATCH 12/19] Fix compiler error --- .../spark/deploy/rest/kubernetes/MultiServerFeignTarget.scala | 2 -- 1 file changed, 2 deletions(-) diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/MultiServerFeignTarget.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/MultiServerFeignTarget.scala index fe5b7a0d94337..627a71b0e7cd1 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/MultiServerFeignTarget.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/MultiServerFeignTarget.scala @@ -38,8 +38,6 @@ private[kubernetes] class MultiServerFeignTarget[T : ClassTag]( implicitly[ClassTag[T]].runtimeClass.asInstanceOf[Class[T]] } - override def url(): String = threadLocalShuffledServers.get.head - /** * Cloning the target is done on every request, for use on the current * thread - thus it's important that clone returns a "fresh" target. From d243dbfa12fd15cd269650f18d0cd24bb44feaf9 Mon Sep 17 00:00:00 2001 From: mcheah Date: Wed, 8 Feb 2017 11:30:15 -0800 Subject: [PATCH 13/19] Fix another compiler error --- .../deploy/kubernetes/integrationtest/minikube/Minikube.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/minikube/Minikube.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/minikube/Minikube.scala index b42f97952394e..2ef55e1752787 100644 --- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/minikube/Minikube.scala +++ b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/minikube/Minikube.scala @@ -123,7 +123,7 @@ private[spark] object Minikube extends Logging { .build() val sslContext = SSLUtils.sslContext(kubernetesConf) val trustManager = SSLUtils.trustManagers(kubernetesConf)(0).asInstanceOf[X509TrustManager] - HttpClientUtil.createClient[T](Array(url), sslContext.getSocketFactory, trustManager) + HttpClientUtil.createClient[T](Array(url), 5, sslContext.getSocketFactory, trustManager) } def executeMinikubeSsh(command: String): Unit = { From bfb71c76283b35d9fb2abfcf3c8d890cbe588c4f Mon Sep 17 00:00:00 2001 From: mcheah Date: Wed, 8 Feb 2017 11:34:42 -0800 Subject: [PATCH 14/19] Delay between retries. Remove unintended test modification --- .../spark/deploy/rest/kubernetes/MultiServerFeignTarget.scala | 4 +++- .../deploy/kubernetes/integrationtest/KubernetesSuite.scala | 3 +-- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/MultiServerFeignTarget.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/MultiServerFeignTarget.scala index 627a71b0e7cd1..18bc1fc832b03 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/MultiServerFeignTarget.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/MultiServerFeignTarget.scala @@ -24,7 +24,8 @@ import org.apache.spark.internal.Logging private[kubernetes] class MultiServerFeignTarget[T : ClassTag]( private val servers: Seq[String], - private val maxRetriesPerServer: Int = 1) extends Target[T] with Retryer with Logging { + private val maxRetriesPerServer: Int = 1, + private val delayBetweenRetriesMillis = 1000) extends Target[T] with Retryer with Logging { require(servers.nonEmpty, "Must provide at least one server URI.") private val threadLocalShuffledServers = new ThreadLocal[Seq[String]] { @@ -66,6 +67,7 @@ private[kubernetes] class MultiServerFeignTarget[T : ClassTag]( if (threadLocalCurrentAttempt.get < maxRetriesPerServer) { logWarning(s"Attempt $currentAttempt of $maxRetriesPerServer failed for" + s" server ${url()}. Retrying request...", e) + Thread.sleep(delayBetweenRetriesMillis) } else { val previousUrl = url() threadLocalShuffledServers.set(threadLocalShuffledServers.get.drop(1)) diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/KubernetesSuite.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/KubernetesSuite.scala index 394f02333467b..c5458eccf830d 100644 --- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/KubernetesSuite.scala +++ b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/KubernetesSuite.scala @@ -122,12 +122,11 @@ private[spark] class KubernetesSuite extends SparkFunSuite with BeforeAndAfter { }) } - /** override def afterAll(): Unit = { if (!System.getProperty("spark.docker.test.persistMinikube", "false").toBoolean) { Minikube.deleteMinikube() } - }*/ + } private def getSparkMetricsService(sparkBaseAppName: String): SparkRestApiV1 = { val serviceName = minikubeKubernetesClient.services() From a95a2d6ee06c04180be4e328b7c93773c0510ffe Mon Sep 17 00:00:00 2001 From: mcheah Date: Wed, 8 Feb 2017 11:35:24 -0800 Subject: [PATCH 15/19] FIx another compiler error --- .../spark/deploy/rest/kubernetes/MultiServerFeignTarget.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/MultiServerFeignTarget.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/MultiServerFeignTarget.scala index 18bc1fc832b03..51313e00ce2da 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/MultiServerFeignTarget.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/MultiServerFeignTarget.scala @@ -25,7 +25,7 @@ import org.apache.spark.internal.Logging private[kubernetes] class MultiServerFeignTarget[T : ClassTag]( private val servers: Seq[String], private val maxRetriesPerServer: Int = 1, - private val delayBetweenRetriesMillis = 1000) extends Target[T] with Retryer with Logging { + private val delayBetweenRetriesMillis: Int = 1000) extends Target[T] with Retryer with Logging { require(servers.nonEmpty, "Must provide at least one server URI.") private val threadLocalShuffledServers = new ThreadLocal[Seq[String]] { From 6b0275600bff1c28a7f2a3b30902770228f59192 Mon Sep 17 00:00:00 2001 From: mcheah Date: Wed, 8 Feb 2017 11:38:51 -0800 Subject: [PATCH 16/19] Extract method --- .../spark/deploy/kubernetes/Client.scala | 27 ++++++++++--------- 1 file changed, 15 insertions(+), 12 deletions(-) diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala index 82987860bb1e0..06e38a6aa0958 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala @@ -138,12 +138,7 @@ private[spark] class Client( driverService) } catch { case e: Throwable => - Utils.tryLogNonFatalError { - kubernetesClient.services().delete(driverService) - } - Utils.tryLogNonFatalError { - kubernetesClient.pods().delete(driverPod) - } + cleanupPodAndService(kubernetesClient, driverPod, driverService) throw new SparkException("Failed to set owner references to the driver pod.", e) } try { @@ -159,12 +154,8 @@ private[spark] class Client( } } catch { case e: Throwable => - Utils.tryLogNonFatalError { - kubernetesClient.services().delete(ownerReferenceConfiguredDriverService) - } - Utils.tryLogNonFatalError { - kubernetesClient.pods().delete(driverPod) - } + cleanupPodAndService(kubernetesClient, driverPod, + ownerReferenceConfiguredDriverService) throw new SparkException("Failed to submit the application to the driver pod.", e) } } @@ -185,6 +176,18 @@ private[spark] class Client( } } + private def cleanupPodAndService( + kubernetesClient: KubernetesClient, + driverPod: Pod, + driverService: Service): Unit = { + Utils.tryLogNonFatalError { + kubernetesClient.services().delete(driverService) + } + Utils.tryLogNonFatalError { + kubernetesClient.pods().delete(driverPod) + } + } + private def submitApplicationToDriverServer( kubernetesClient: KubernetesClient, driverSubmitSslOptions: SSLOptions, From 9f11ad1d57935f5bdf28f5386782c5fde0f02c6b Mon Sep 17 00:00:00 2001 From: mcheah Date: Wed, 8 Feb 2017 13:23:12 -0800 Subject: [PATCH 17/19] Address comments --- .../scala/org/apache/spark/deploy/kubernetes/constants.scala | 1 - .../apache/spark/deploy/rest/kubernetes/HttpClientUtil.scala | 4 ++-- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/constants.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/constants.scala index d16d7527cd3bf..027cc3c022b4e 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/constants.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/constants.scala @@ -67,5 +67,4 @@ package object constants { // Miscellaneous private[spark] val DRIVER_CONTAINER_NAME = "spark-kubernetes-driver" private[spark] val KUBERNETES_SUBMIT_SSL_NAMESPACE = "kubernetes.submit" - private[spark] val DRIVER_INITIAL_LIVELINESS_CHECK_DELAY_SECONDS = 10 } diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/HttpClientUtil.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/HttpClientUtil.scala index 733ce48b26b79..c3227f88c1280 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/HttpClientUtil.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/HttpClientUtil.scala @@ -46,12 +46,12 @@ private[spark] object HttpClientUtil { .registerModule(new DefaultScalaModule) .setDateFormat(JacksonMessageWriter.makeISODateFormat) objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false) - val target = new MultiServerFeignTarget[T](uris) + val target = new MultiServerFeignTarget[T](uris, maxRetriesPerServer) val baseHttpClient = new feign.okhttp.OkHttpClient(httpClientBuilder.build()) val resetTargetHttpClient = new Client { override def execute(request: Request, options: Options): Response = { val response = baseHttpClient.execute(request, options) - if (response.status() >= 200 && response.status() < 300) { + if (response.status() / 100 == 2) { target.reset() } response From ce6bb035008d0f62930a9d271d4982b8177447d0 Mon Sep 17 00:00:00 2001 From: mcheah Date: Wed, 8 Feb 2017 14:46:13 -0800 Subject: [PATCH 18/19] Deduplicate node addresses, use lower initial connect timeout --- .../scala/org/apache/spark/deploy/kubernetes/Client.scala | 5 +++-- .../apache/spark/deploy/rest/kubernetes/HttpClientUtil.scala | 4 ++-- .../kubernetes/integrationtest/minikube/Minikube.scala | 2 +- 3 files changed, 6 insertions(+), 5 deletions(-) diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala index 06e38a6aa0958..6de3cb89f27af 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala @@ -745,7 +745,7 @@ private[spark] class Client( node.getSpec.getUnschedulable) .flatMap(_.getStatus.getAddresses.asScala.map(address => { s"$urlScheme://${address.getAddress}:$servicePort" - })).toArray + })).toSet require(nodeUrls.nonEmpty, "No nodes found to contact the driver!") val (trustManager, sslContext): (X509TrustManager, SSLContext) = if (driverSubmitSslOptions.enabled) { @@ -757,7 +757,8 @@ private[spark] class Client( uris = nodeUrls, maxRetriesPerServer = 10, sslSocketFactory = sslContext.getSocketFactory, - trustContext = trustManager) + trustContext = trustManager, + connectTimeoutMillis = 5000) } private def buildSslConnectionConfiguration(driverSubmitSslOptions: SSLOptions) = { diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/HttpClientUtil.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/HttpClientUtil.scala index c3227f88c1280..576f7058f20ee 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/HttpClientUtil.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/rest/kubernetes/HttpClientUtil.scala @@ -32,7 +32,7 @@ import org.apache.spark.status.api.v1.JacksonMessageWriter private[spark] object HttpClientUtil { def createClient[T: ClassTag]( - uris: Array[String], + uris: Set[String], maxRetriesPerServer: Int = 1, sslSocketFactory: SSLSocketFactory = SSLContext.getDefault.getSocketFactory, trustContext: X509TrustManager = null, @@ -46,7 +46,7 @@ private[spark] object HttpClientUtil { .registerModule(new DefaultScalaModule) .setDateFormat(JacksonMessageWriter.makeISODateFormat) objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false) - val target = new MultiServerFeignTarget[T](uris, maxRetriesPerServer) + val target = new MultiServerFeignTarget[T](uris.toSeq, maxRetriesPerServer) val baseHttpClient = new feign.okhttp.OkHttpClient(httpClientBuilder.build()) val resetTargetHttpClient = new Client { override def execute(request: Request, options: Options): Response = { diff --git a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/minikube/Minikube.scala b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/minikube/Minikube.scala index 2ef55e1752787..736b92cc2d628 100644 --- a/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/minikube/Minikube.scala +++ b/resource-managers/kubernetes/integration-tests/src/test/scala/org/apache/spark/deploy/kubernetes/integrationtest/minikube/Minikube.scala @@ -123,7 +123,7 @@ private[spark] object Minikube extends Logging { .build() val sslContext = SSLUtils.sslContext(kubernetesConf) val trustManager = SSLUtils.trustManagers(kubernetesConf)(0).asInstanceOf[X509TrustManager] - HttpClientUtil.createClient[T](Array(url), 5, sslContext.getSocketFactory, trustManager) + HttpClientUtil.createClient[T](Set(url), 5, sslContext.getSocketFactory, trustManager) } def executeMinikubeSsh(command: String): Unit = { From e929b50ceb111a1591fdbf277024bdbfab6b0a5c Mon Sep 17 00:00:00 2001 From: Andrew Ash Date: Wed, 8 Feb 2017 17:05:15 -0800 Subject: [PATCH 19/19] Drop maxRetriesPerServer from 10 to 3 --- .../main/scala/org/apache/spark/deploy/kubernetes/Client.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala index 6de3cb89f27af..0c01a56f7b10b 100644 --- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala +++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/kubernetes/Client.scala @@ -755,7 +755,7 @@ private[spark] class Client( } HttpClientUtil.createClient[KubernetesSparkRestApi]( uris = nodeUrls, - maxRetriesPerServer = 10, + maxRetriesPerServer = 3, sslSocketFactory = sslContext.getSocketFactory, trustContext = trustManager, connectTimeoutMillis = 5000)