zhengchenyu commented on code in PR #57350: URL: https://github.com/apache/spark/pull/57350#discussion_r3710397572
########## 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: The driver has the same issue. Furthermore, by default, `spark.kubernetes.driver.service.deleteOnTermination`=true will delete the UI service when stops. I think this could be modified in a separate PR: if `spark.kubernetes.driver.service.deleteOnTermination`=false is configured, then the selector will be dropped. Actually, I don't think it's necessary to change it since the default value is already true. ########## 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: The driver service also has the same issue, which can lead to resource leaks. However, compared to that, using LoadBalancer/NodePort might result in wasted resources. Perhaps we could move the service creation to after the driver starts, but that's outside the scope of this PR discussion. ########## resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/DriverUIServiceFeatureStep.scala: ########## @@ -0,0 +1,167 @@ +/* + * 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. + * + * When the user requests a random UI port (`spark.ui.port=0`), the actual port is unknown at + * submission time, so the Service's `targetPort` uses a placeholder. To avoid routing traffic + * to the wrong endpoint during that window (in `hostNetwork` mode a pod endpoint is the node IP, + * so the placeholder `targetPort` could reach an unrelated driver co-located on the same node), + * the Service is created *without* a selector, leaving it endpointless. Once the driver's Jetty + * server has bound, `K8sDriverUIServicePatcher` patches the selector and the actual `targetPort` + * together. When the port is fixed up front, the Service is created with its selector and final + * `targetPort` immediately. + */ +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) Review Comment: I will change it to be the same as the driver, and no longer use separate configuration. -- 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]
