sunchao commented on code in PR #57350: URL: https://github.com/apache/spark/pull/57350#discussion_r3706591625
########## resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/K8sDriverUIServicePatcher.scala: ########## @@ -0,0 +1,81 @@ +/* + * 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 io.fabric8.kubernetes.client.dsl.base.{PatchContext, PatchType} + +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 { + + /** + * Bring the endpointless placeholder Service online: set its `spark-ui` `targetPort` to the + * actual bound port and install the `selector` in the same patch, so it only starts routing + * once it points at the correct port. + * + * @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 servicePort The Service's stable `port` for the `spark-ui` entry (strategic merge key). + * @param actualPort The actual port the driver's Jetty server bound to + * (typically `SparkUI.boundPort`). + * @param selector Selector matching the driver pod, withheld at creation time. + */ + def patchTargetPortAndSelector( + client: KubernetesClient, + namespace: String, + serviceName: String, + servicePort: Int, + actualPort: Int, + selector: Map[String, String]): Unit = { + try { + val patch = new ServiceBuilder() + .withNewSpec() + .withSelector(selector.asJava) Review Comment: [P1] Withdraw the UI Service selector before stopping Jetty This installs the real selector but never withdraws it. With `hostNetwork=true`, `spark.kubernetes.driver.ui.service.type=NodePort` (or `LoadBalancer`), and `spark.kubernetes.driver.service.deleteOnTermination=false`, `SparkContext.stop()` closes the UI at `SparkContext.scala:2350` before scheduler/backend shutdown at `SparkContext.scala:2360`, while `KubernetesClusterSchedulerBackend.stop()` leaves this Service and its selector intact. The driver JVM can continue running after `sc.stop()`, so its Pod remains a Ready endpoint at the shared node IP. If another application binds the released port on that node, the first application's externally reachable Service now routes to the other application's UI. This is distinct from the already-fixed startup placeholder window. Please clear the selector or delete the dedicated UI Service before releasing the UI port, regardless of the headless driver Service's retention setting. ########## resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/DriverUIServiceFeatureStep.scala: ########## @@ -0,0 +1,176 @@ +/* + * 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. + * + * The actual UI port is never known at submission time: even with a fixed `spark.ui.port`, Jetty + * may bind a different port after a collision (`Utils.startServiceOnPort` retries the next ports), + * and when TLS is enabled the reachable port is the secure connector's port, not the configured + * one. Routing to a stale port is especially dangerous in `hostNetwork` mode, where a pod endpoint + * is the node IP, so a wrong `targetPort` could reach an unrelated driver co-located on the same + * node. To avoid that window, the Service is always created *without* a selector, leaving it + * endpointless, with the configured (or default) UI port as a placeholder `port`/`targetPort` + * purely to satisfy Kubernetes' Service port validation (must be > 0). Once the driver's Jetty + * server has bound, [[org.apache.spark.scheduler.cluster.k8s.K8sDriverUIServicePatcher]] patches + * the selector and the actual `targetPort` (`SparkUI.boundPort`) together, so the Service only + * starts routing once it points at the correct port. When the Spark UI is disabled the Service is + * not created at all; and if the driver dies before its UI binds, the created Service simply stays + * endpointless and routes nowhere. This reconciliation requires `patch services` RBAC on the + * driver's ServiceAccount whenever the feature is enabled. + */ +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) + + private val active: Boolean = if (enabled && !kubernetesConf.get(config.UI.UI_ENABLED)) { + logWarning(s"Ignoring ${KUBERNETES_DRIVER_UI_SERVICE_ENABLED.key}=true because " + + s"${config.UI.UI_ENABLED.key}=false; no driver UI Service will be created.") + false + } else { + enabled + } + + /** + * Placeholder port used when building the Service. The real bound port is only known after the + * driver's Jetty server binds (and may differ from `configuredUIPort` after a collision or when + * TLS is enabled), so we substitute the configured UI port, or the default (typically 4040) when + * a random port was requested (`spark.ui.port=0`), purely 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) { + configuredUIPort + } else { + config.UI.UI_PORT.defaultValue.get + } + + 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] = { + // These properties drive the runtime patch that installs the withheld selector and the actual + // bound `targetPort` once the driver's UI has bound. They are always needed while the feature + // is active, since the Service is created endpointless regardless of the configured port. + if (active) { + Map( + KUBERNETES_DRIVER_UI_SERVICE_NAME_INTERNAL -> serviceName, + KUBERNETES_DRIVER_UI_SERVICE_PORT_INTERNAL -> servicePort.toString, + KUBERNETES_DRIVER_UI_SERVICE_SELECTOR_INTERNAL -> encodeSelector(kubernetesConf.labels)) + } else { + Map.empty + } + } + + override def getAdditionalKubernetesResources(): Seq[HasMetadata] = { + if (!active) 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) Review Comment: [P2] Clean up external UI Services when the driver exits before SparkContext A `LoadBalancer` or `NodePort` Service is created by `KubernetesClientApplication` before the driver application constructs `SparkContext`. If application startup exits or throws before a context exists, `KubernetesClusterSchedulerBackend.stop()` never runs, so no Service cleanup occurs even when `spark.kubernetes.driver.service.deleteOnTermination=true`. Owner references do not help until the driver Pod is deleted; Spark's Kubernetes documentation says completed driver Pods remain until eventual garbage collection or manual cleanup. Every early failure can therefore retain a cloud load balancer/IP or an allocated NodePort for the Pod retention period. The existing headless driver Service did not allocate either external resource. Please defer external Service provisioning until the driver initializes, or clean it up when its Pod reaches a terminal phase. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
