[SPARK-58203][K8S] Support configuring spark-managed UI service to work with spark.ui.port=0 #57350
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
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
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
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
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 |
|---|---|---|
| @@ -0,0 +1,117 @@ | ||
| /* | ||
| * 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.k8s.features | ||
|
|
||
| import scala.jdk.CollectionConverters._ | ||
|
|
||
| import io.fabric8.kubernetes.api.model.{HasMetadata, ServiceBuilder} | ||
|
|
||
| import org.apache.spark.deploy.k8s.{KubernetesDriverConf, SparkPod} | ||
| import org.apache.spark.deploy.k8s.Config.{ | ||
| KUBERNETES_DRIVER_SERVICE_IP_FAMILIES, | ||
| KUBERNETES_DRIVER_SERVICE_IP_FAMILY_POLICY, | ||
| KUBERNETES_DRIVER_UI_SERVICE_ENABLED, | ||
| KUBERNETES_DRIVER_UI_SERVICE_NAME, | ||
| KUBERNETES_DRIVER_UI_SERVICE_TYPE | ||
| } | ||
| import org.apache.spark.deploy.k8s.Constants._ | ||
| import org.apache.spark.internal.{config, Logging} | ||
|
|
||
| /** | ||
| * Optionally provisions a dedicated Kubernetes Service exposing only the Spark driver's Web UI | ||
| * port. | ||
| * | ||
| * At creation time, the Service's `targetPort` is set to the configured `spark.ui.port` | ||
| * (a placeholder when the user has requested a random port). Once the driver's Jetty server | ||
| * has bound, `K8sDriverUIServicePatcher` updates the Service's `targetPort` to reflect the | ||
| * actual bound port. | ||
| */ | ||
| private[spark] class DriverUIServiceFeatureStep(kubernetesConf: KubernetesDriverConf) | ||
| extends KubernetesFeatureConfigStep with Logging { | ||
| import DriverUIServiceFeatureStep._ | ||
|
|
||
| private val enabled = kubernetesConf.get(KUBERNETES_DRIVER_UI_SERVICE_ENABLED) | ||
| private lazy val serviceType = kubernetesConf.get(KUBERNETES_DRIVER_UI_SERVICE_TYPE) | ||
| private lazy val configuredUIPort = kubernetesConf.get(config.UI.UI_PORT) | ||
|
|
||
| /** | ||
| * Port value used when building the Service. When the user has requested a random UI port | ||
| * (`spark.ui.port=0`), the actual port is only known after the driver's Jetty server binds, | ||
| * so we substitute the default UI port (typically 4040) purely as a placeholder to satisfy | ||
| * Kubernetes' Service port validation (must be > 0). After the driver JVM starts, | ||
| * [[org.apache.spark.scheduler.cluster.k8s.K8sDriverUIServicePatcher]] updates the Service's | ||
| * `targetPort` to the real bound port. | ||
| */ | ||
| private lazy val servicePort: Int = if (configuredUIPort == 0) { | ||
| config.UI.UI_PORT.defaultValue.get | ||
| } else { | ||
| configuredUIPort | ||
| } | ||
|
|
||
| private lazy val serviceName: String = kubernetesConf.get(KUBERNETES_DRIVER_UI_SERVICE_NAME) | ||
| .getOrElse(kubernetesConf.driverUIServiceName) | ||
|
|
||
| // The UI Service reuses the driver Service IP family settings to keep the same IP family. | ||
| private lazy val ipFamilyPolicy = kubernetesConf.get(KUBERNETES_DRIVER_SERVICE_IP_FAMILY_POLICY) | ||
| private lazy val ipFamilies = | ||
| kubernetesConf.get(KUBERNETES_DRIVER_SERVICE_IP_FAMILIES).split(",").toList.asJava | ||
|
|
||
| override def configurePod(pod: SparkPod): SparkPod = pod | ||
|
|
||
| override def getAdditionalPodSystemProperties(): Map[String, String] = { | ||
| if (enabled) { | ||
| Map(KUBERNETES_DRIVER_UI_SERVICE_NAME_INTERNAL -> serviceName) | ||
| } else { | ||
| Map.empty | ||
| } | ||
| } | ||
|
|
||
| override def getAdditionalKubernetesResources(): Seq[HasMetadata] = { | ||
| if (!enabled) return Seq.empty | ||
|
|
||
| val uiService = new ServiceBuilder() | ||
| .withNewMetadata() | ||
| .withName(serviceName) | ||
| .addToAnnotations(kubernetesConf.serviceAnnotations.asJava) | ||
| .addToLabels(SPARK_APP_ID_LABEL, kubernetesConf.appId) | ||
| .addToLabels(kubernetesConf.serviceLabels.asJava) | ||
| .endMetadata() | ||
| .withNewSpec() | ||
| .withType(serviceType) | ||
|
Member
There was a problem hiding this comment. [P2] Preserve the configured IP family on the UI Service This Service does not set
Contributor
Author
There was a problem hiding this comment. @sunchao I have updated the code. UI service will reuse Driver Service's ip family to keep the same IP family. |
||
| .withIpFamilyPolicy(ipFamilyPolicy) | ||
| .withIpFamilies(ipFamilies) | ||
| .withSelector(kubernetesConf.labels.asJava) | ||
| .addNewPort() | ||
| .withName(UI_PORT_NAME) | ||
| .withPort(servicePort) | ||
| .withNewTargetPort(servicePort) | ||
| .endPort() | ||
| .endSpec() | ||
| .build() | ||
| Seq(uiService) | ||
| } | ||
| } | ||
|
|
||
| private[spark] object DriverUIServiceFeatureStep { | ||
| /** | ||
| * Internal spark conf key used to pass the UI service name from this feature step to the | ||
| * driver runtime (SparkContext) so `K8sDriverUIServicePatcher` can look up the Service to | ||
| * patch. | ||
| */ | ||
| val KUBERNETES_DRIVER_UI_SERVICE_NAME_INTERNAL = | ||
| "spark.kubernetes.driver.ui.service.name.internal" | ||
| } | ||
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
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 |
|---|---|---|
| @@ -0,0 +1,94 @@ | ||
| /* | ||
| * 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.scheduler.cluster.k8s | ||
|
|
||
| import scala.jdk.CollectionConverters._ | ||
|
|
||
| import io.fabric8.kubernetes.api.model.{IntOrString, ServiceBuilder} | ||
| import io.fabric8.kubernetes.client.KubernetesClient | ||
|
|
||
| import org.apache.spark.deploy.k8s.Constants.UI_PORT_NAME | ||
| import org.apache.spark.internal.Logging | ||
|
|
||
| /** | ||
| * Driver-side utility that updates the `targetPort` of the dedicated Spark UI Kubernetes Service | ||
| * (created by [[org.apache.spark.deploy.k8s.features.DriverUIServiceFeatureStep]]) to the actual | ||
| * port the driver's Jetty server bound to. | ||
| */ | ||
| private[k8s] object K8sDriverUIServicePatcher extends Logging { | ||
|
|
||
| /** | ||
| * Patch the targetPort of the given Service's `spark-ui` port entry to `actualPort`, if the | ||
| * current value differs. | ||
| * | ||
| * @param client Kubernetes client to use (typically the one already held by the backend). | ||
| * @param namespace Namespace where the Service lives. | ||
| * @param serviceName Name of the UI Service (from | ||
| * `spark.kubernetes.driver.ui.service.name.internal`). | ||
| * @param actualPort The actual port the driver's Jetty server bound to | ||
| * (typically `SparkUI.boundPort`). | ||
| */ | ||
| def patchTargetPort( | ||
| client: KubernetesClient, | ||
| namespace: String, | ||
| serviceName: String, | ||
| actualPort: Int): Unit = { | ||
| try { | ||
| val service = client.services() | ||
| .inNamespace(namespace) | ||
| .withName(serviceName) | ||
| .get() | ||
|
|
||
| if (service == null) { | ||
| logWarning(s"UI service '$serviceName' not found in namespace '$namespace'; " + | ||
| "skipping targetPort patch.") | ||
| return | ||
| } | ||
|
|
||
| val currentTargetPort = service.getSpec.getPorts.asScala | ||
| .find(_.getName == UI_PORT_NAME) | ||
| .map(_.getTargetPort.getIntVal.toInt) | ||
|
|
||
| currentTargetPort match { | ||
| case Some(existing) if existing == actualPort => | ||
| logInfo(s"UI service '$serviceName' targetPort already matches actual UI port " + | ||
| s"($actualPort); no patch needed.") | ||
|
|
||
| case _ => | ||
| val updated = new ServiceBuilder(service) | ||
| .editSpec() | ||
| .editMatchingPort(portBuilder => portBuilder.build().getName == UI_PORT_NAME) | ||
| .withTargetPort(new IntOrString(actualPort)) | ||
| .endPort() | ||
| .endSpec() | ||
| .build() | ||
|
|
||
| client.services() | ||
| .inNamespace(namespace) | ||
| .withName(serviceName) | ||
| .patch(updated) | ||
|
|
||
| logInfo(s"Patched UI service '$serviceName' targetPort " + | ||
| s"${currentTargetPort.map(_.toString).getOrElse("<unset>")} -> $actualPort") | ||
| } | ||
| } catch { | ||
| case e: Exception => | ||
| logError(s"Failed to patch UI service '$serviceName' targetPort to $actualPort", e) | ||
| throw e | ||
| } | ||
| } | ||
| } |
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
Oops, something went wrong.
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.
[P1] Handle port zero in the existing driver resources too
This placeholder only fixes the new dedicated Service. With
spark.ui.port=0,BasicDriverFeatureStepstill emitscontainerPort=0, andDriverServiceFeatureStepstill emitsport=0/targetPort=0for the mandatory headless Service. Kubernetes requires these numeric ports to be in1..65535, so the driver Pod/resources are rejected before this runtime patch can run. Please omit or substitute both existing UI port declarations and cover the completeKubernetesDriverBuilderoutput in a test.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.
@sunchao In fact, Because I think the internal service do not need to expose ports. So I use
spark.kubernetes.executor.useDriverPodIP=true and addorg.apache.spark.deploy.k8s.features.DriverServiceFeatureSteptospark.kubernetes.driver.pod.excludedFeatureSteps.If the driver service's port is configured to 0, then this is indeed necessary. However, it's unclear whether it's appropriate to do this in this PR, since this is a PR for adding a UI service. I'll add it to this PR if you think it's suitable.