This repository was archived by the owner on Jan 9, 2020. It is now read-only.
Use readiness probe instead of client-side ping. #75
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -31,16 +31,14 @@ import io.fabric8.kubernetes.client.Watcher.Action | |
| import org.apache.commons.codec.binary.Base64 | ||
| import scala.collection.JavaConverters._ | ||
| import scala.collection.mutable | ||
|
|
||
| import org.apache.spark.{SecurityManager, SparkConf, SparkException, SSLOptions} | ||
| import org.apache.spark.deploy.kubernetes.config._ | ||
| 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.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) | ||
|
|
||
| def run(): Unit = { | ||
| val (driverSubmitSslOptions, isKeyStoreLocalFile) = parseDriverSubmitSslOptions() | ||
| val parsedCustomLabels = parseCustomLabels(customLabels) | ||
|
|
@@ -127,6 +121,11 @@ private[spark] class Client( | |
| .pods() | ||
| .withLabels(driverKubernetesSelectors) | ||
| .watch(podWatcher)) { _ => | ||
| val probePingHttpGet = new HTTPGetActionBuilder() | ||
| .withScheme(if (driverSubmitSslOptions.enabled) "HTTPS" else "HTTP") | ||
|
There was a problem hiding this comment. these need to be all-caps? looks kinda weird to me
Author
There was a problem hiding this comment. It does need to be capitalized, the Kubernetes API will throw an error otherwise. |
||
| .withPath("/v1/submissions/ping") | ||
| .withNewPort(SUBMISSION_SERVER_PORT_NAME) | ||
| .build() | ||
| kubernetesClient.pods().createNew() | ||
| .withNewMetadata() | ||
| .withName(kubernetesAppId) | ||
|
|
@@ -162,6 +161,8 @@ private[spark] class Client( | |
| .endEnv() | ||
| .addToEnv(sslEnvs: _*) | ||
| .withPorts(containerPorts.asJava) | ||
| .withNewReadinessProbe().withHttpGet(probePingHttpGet).endReadinessProbe() | ||
| .withNewLivenessProbe().withHttpGet(probePingHttpGet).endLivenessProbe() | ||
|
There was a problem hiding this comment. do we need https://kubernetes.io/docs/api-reference/v1/definitions/#_v1_probe I'm also not sure that having both is necessary -- we only rely on the readiness probe in this PR?
Member
There was a problem hiding this comment. The liveness could be useful to restart our server if it fails for some reason, but we would need an initial delay seconds, since otherwise, we'll have it fail early on, before the server starts listening. |
||
| .endContainer() | ||
| .endSpec() | ||
| .done() | ||
|
|
@@ -308,114 +309,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) { | ||
| 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() | ||
|
There was a problem hiding this comment. comment that this ping is a final check of the service liveness before submitting the job (in addition to the k8s checks). Might also be worth try/catching it and logging on failure that even though the k8s service is active we are unable to connect to the driver's rest service from the submitter |
||
| 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 => | ||
| } | ||
| } | ||
| } | ||
|
|
||
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Currently our PodWatcher watches the driver pod to determine its liveness for uploading files to it. What we more directly depend on those is the service -- when the readiness probe goes to success and the pod gets added to the service, that seems like the more specific trigger.
Proposal: what if we change the
DriverPodWatcherto watch for the service gaining a backing pod, instead of the driver pod being ready? That more directly monitors for the condition we need to submit the job.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We would need an extra layer of nesting to do that, since we want to delay the service creation for as long as possible. So we want to only make the service after we make the pod, so we'll need a
Watchfor the pod being created to trigger creating the service, and another watch for theServicehaving the backing pod. It's doable, just not sure how to write it in a way that makes the nesting easy to reason about.Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It's worth discussing what the ramifications could be for creating the service up front and perhaps having the node ports open for the duration of the launch - in the worst case, if the launch times out. If we created the service outside of the watch it would be easier to design this where we don't have two layers of nesting and futures to listen for, but there is the aforementioned tradeoff to consider.
In the meantime - @ash211 @foxish how about we merge this PR as is and follow up on that point moving forward?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I'm good with making this PR's change be to watch on the readiness probe of the driver pod, and potentially in a subsequent PR we change that to be a watch on the driver service directly if that makes more sense.
Filed #76 as a followup for discussion
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I also agree to let this go in for now, perhaps, we could table this fine-grained approach at the next meeting.