Add executor feature step to create executor kubernetes service

Author: EnricoMi

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala

@@ -442,6 +442,19 @@ private[spark] object Config extends Logging {
       .toSequence
       .createWithDefault(Nil)
 
+  val KUBERNETES_EXECUTOR_ENABLE_SERVICE =
+    ConfigBuilder("spark.kubernetes.executor.enableService")
+      .doc("If a Kubernetes service is created for the executor. " +
+        "The executor pod creates a Kubernetes service that allows to connect to executor ports " +
+        "via the Kubernetes service instead of the pod host IP. Connecting to such ports " +
+        "instantly fails with 'connection refused' error once the executor got decommissioned. " +
+        "Connection to the port via the pod host IP instead fails with a 'connection timeout' " +
+        "which is set via NETWORK_TIMEOUT and defaults to 2 minutes. " +
+        "The kubernetes service provides access to the executors shuffle service.")
+      .version("4.1.0")
+      .booleanConf
+      .createWithDefault(false)
+
   val KUBERNETES_EXECUTOR_DECOMMISSION_LABEL =
     ConfigBuilder("spark.kubernetes.executor.decommissionLabel")
       .doc("Label to apply to a pod which is being decommissioned." +

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/features/ExecutorServiceFeatureStep.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.deploy.k8s.features
+
+import scala.jdk.CollectionConverters._
+
+import io.fabric8.kubernetes.api.model.{ContainerBuilder, HasMetadata, ServiceBuilder}
+
+import org.apache.spark.SparkException
+import org.apache.spark.deploy.k8s.{KubernetesExecutorConf, SparkPod}
+import org.apache.spark.internal.config.SHUFFLE_SERVICE_PORT
+
+class ExecutorServiceFeatureStep(conf: KubernetesExecutorConf) extends KubernetesFeatureConfigStep {
+  private val spark_app_selector_label = "spark-app-selector"
+  private val spark_exec_id_label = "spark-exec-id"
+  private val service_selector_labels = Set(spark_app_selector_label, spark_exec_id_label)
+
+  private lazy val sparkAppSelector = getLabel(spark_app_selector_label)
+  private lazy val sparkExecId = getLabel(spark_exec_id_label)
+  // name length is 8 + 38 + 6 + 10 = 62
+  // which fits in KUBERNETES_DNS_LABEL_NAME_MAX_LENGTH = 63
+  private lazy val serviceName = s"svc-$sparkAppSelector-exec-$sparkExecId"
+
+  private def getLabel(label: String): String = {
+    val value = conf.labels.get(label)
+    value.getOrElse(
+      throw new SparkException(s"This feature step requires label $label")
+    )
+  }
+
+  override def configurePod(pod: SparkPod): SparkPod = {
+    SparkPod(
+      pod.pod,
+      // tell the executor entry point its Kubernetes service name
+      new ContainerBuilder(pod.container)
+        .addNewEnv()
+        .withName("EXECUTOR_SERVICE_NAME")
+        .withValue(serviceName)
+        .endEnv()
+        .build())
+  }
+
+  override def getAdditionalKubernetesResources(): Seq[HasMetadata] = {
+    val selector = conf.labels
+      .filter { case (key, _) => service_selector_labels.contains(key) }
+
+    // kubernetes executor service provides access to the executor's shuffle service
+    val portName = "spark-shuffle-service"
+    val port = conf.sparkConf.get(SHUFFLE_SERVICE_PORT)
+
+    val service = new ServiceBuilder()
+      .withNewMetadata()
+      .withName(serviceName)
+      .endMetadata()
+      .withNewSpec()
+      .withSelector(selector.asJava)
+      .addNewPort()
+      .withName(portName)
+      .withPort(port)
+      .withNewTargetPort(port)
+      .endPort()
+      .endSpec()
+      .build()
+
+    Seq(service)
+  }
+}

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocator.scala

@@ -463,6 +463,7 @@ class ExecutorPodsAllocator(
         kubernetesClient.pods().inNamespace(namespace).resource(podWithAttachedContainer).create()
       try {
         addOwnerReference(createdExecutorPod, resources)
+        kubernetesClient.resourceList(resources: _*).forceConflicts().serverSideApply()
         resources
           .filter(_.getKind == "PersistentVolumeClaim")
           .foreach { resource =>

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBuilder.scala

@@ -65,14 +65,20 @@ private[spark] class KubernetesExecutorBuilder {
         }
       }
 
+    val optionalFeatures = Seq(
+      Some(conf.get(Config.KUBERNETES_EXECUTOR_ENABLE_SERVICE))
+        .filter(enabled => enabled)
+        .map(_ => new ExecutorServiceFeatureStep(conf))
+    ).flatten
+
     val allFeatures = Seq(
       new BasicExecutorFeatureStep(conf, secMgr, resourceProfile),
       new ExecutorKubernetesCredentialsFeatureStep(conf),
       new MountSecretsFeatureStep(conf),
       new EnvSecretsFeatureStep(conf),
       new MountVolumesFeatureStep(conf),
       new HadoopConfExecutorFeatureStep(conf),
-      new LocalDirsFeatureStep(conf)) ++ userFeatures
+      new LocalDirsFeatureStep(conf)) ++ optionalFeatures ++ userFeatures
 
     val features = allFeatures.filterNot(f =>
       conf.get(Config.KUBERNETES_EXECUTOR_POD_EXCLUDED_FEATURE_STEPS).contains(f.getClass.getName))

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/ExecutorPodsAllocatorSuite.scala

@@ -25,14 +25,15 @@ import scala.jdk.CollectionConverters._
 
 import io.fabric8.kubernetes.api.model._
 import io.fabric8.kubernetes.client.{KubernetesClient, KubernetesClientException}
-import io.fabric8.kubernetes.client.dsl.PodResource
+import io.fabric8.kubernetes.client.dsl.{NamespaceListVisitFromServerGetDeleteRecreateWaitApplicable, PodResource, ServerSideApplicable}
 import org.mockito.{Mock, MockitoAnnotations}
 import org.mockito.ArgumentMatchers.{any, anyString, eq => meq}
 import org.mockito.Mockito.{never, times, verify, when}
 import org.mockito.invocation.InvocationOnMock
 import org.mockito.stubbing.Answer
 import org.scalatest.BeforeAndAfter
 import org.scalatest.PrivateMethodTester._
+import org.scalatestplus.mockito.MockitoSugar.mock
 
 import org.apache.spark.{SecurityManager, SparkConf, SparkException, SparkFunSuite}
 import org.apache.spark.deploy.k8s.{KubernetesExecutorConf, KubernetesExecutorSpec}
@@ -144,6 +145,11 @@ class ExecutorPodsAllocatorSuite extends SparkFunSuite with BeforeAndAfter {
       conf, secMgr, executorBuilder, kubernetesClient, snapshotsStore, waitForExecutorPodsClock)
     when(schedulerBackend.getExecutorIds()).thenReturn(Seq.empty)
     podsAllocatorUnderTest.start(TEST_SPARK_APP_ID, schedulerBackend)
+    val apl = mock[NamespaceListVisitFromServerGetDeleteRecreateWaitApplicable[HasMetadata]]
+    val ssa = mock[ServerSideApplicable[java.util.List[HasMetadata]]]
+    when(apl.forceConflicts()).thenReturn(ssa)
+    when(kubernetesClient.resourceList()).thenReturn(apl)
+    when(kubernetesClient.resourceList(any[HasMetadata]())).thenReturn(apl)
     when(kubernetesClient.persistentVolumeClaims()).thenReturn(persistentVolumeClaims)
     when(persistentVolumeClaims.inNamespace("default")).thenReturn(pvcWithNamespace)
     when(pvcWithNamespace.withLabel(any(), any())).thenReturn(labeledPersistentVolumeClaims)

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBuilderSuite.scala

@@ -16,12 +16,16 @@
  */
 package org.apache.spark.scheduler.cluster.k8s
 
+import scala.jdk.CollectionConverters.IterableHasAsScala
+
+import io.fabric8.kubernetes.api.model.Service
 import io.fabric8.kubernetes.client.KubernetesClient
+import org.mockito.Mockito.mock
 
 import org.apache.spark.{SecurityManager, SparkConf}
 import org.apache.spark.deploy.k8s._
 import org.apache.spark.deploy.k8s.features.KubernetesExecutorCustomFeatureConfigStep
-import org.apache.spark.internal.config.ConfigEntry
+import org.apache.spark.internal.config.{ConfigEntry, SHUFFLE_SERVICE_PORT}
 import org.apache.spark.resource.ResourceProfile
 
 class KubernetesExecutorBuilderSuite extends PodBuilderSuite {
@@ -65,6 +69,47 @@ class KubernetesExecutorBuilderSuite extends PodBuilderSuite {
     val defaultProfile = ResourceProfile.getOrCreateDefaultProfile(sparkConf)
     new KubernetesExecutorBuilder().buildFromFeatures(conf, secMgr, client, defaultProfile).pod
   }
+
+  test("SPARK-XXXXX: check executor kubernetes spec with service disabled by default") {
+    val sparkConf = baseConf
+    val conf = KubernetesTestConf.createExecutorConf(sparkConf = sparkConf)
+    val secMgr = new SecurityManager(sparkConf)
+    val client = mock(classOf[KubernetesClient])
+    val profile = ResourceProfile.getOrCreateDefaultProfile(sparkConf)
+    val spec = new KubernetesExecutorBuilder().buildFromFeatures(conf, secMgr, client, profile)
+
+    val containerEnvs = spec.pod.container.getEnv.asScala
+    assert(!containerEnvs.exists(_.getName === "EXECUTOR_SERVICE_NAME"))
+
+    assert(spec.executorKubernetesResources.size === 0)
+  }
+
+  test("SPARK-XXXXX: check executor kubernetes spec with service enabled") {
+    Seq(None, Some(1234)).foreach { somePort =>
+      val sparkConf = baseConf.clone.set(Config.KUBERNETES_EXECUTOR_ENABLE_SERVICE, true)
+      somePort.foreach(sparkConf.set(SHUFFLE_SERVICE_PORT, _))
+      val conf = KubernetesTestConf.createExecutorConf(sparkConf = sparkConf)
+      val secMgr = new SecurityManager(sparkConf)
+      val client = mock(classOf[KubernetesClient])
+      val profile = ResourceProfile.getOrCreateDefaultProfile(sparkConf)
+      val spec = new KubernetesExecutorBuilder().buildFromFeatures(conf, secMgr, client, profile)
+
+      val containerEnvs = spec.pod.container.getEnv.asScala
+      assert(containerEnvs.exists(_.getName === "EXECUTOR_SERVICE_NAME"))
+      val containerEnv = containerEnvs.filter(_.getName === "EXECUTOR_SERVICE_NAME").head
+      assert(containerEnv.getValue === "svc-appId-exec-1")
+
+      assert(spec.executorKubernetesResources.size === 1)
+      val resource = spec.executorKubernetesResources.head
+      assert(resource.getKind === "Service")
+      val service = resource.asInstanceOf[Service]
+      assert(service.getMetadata.getName === "svc-appId-exec-1")
+      assert(service.getSpec.getPorts.size() === 1)
+      val port = service.getSpec.getPorts.get(0)
+      assert(port.getName === "spark-shuffle-service")
+      assert(port.getPort === somePort.getOrElse(SHUFFLE_SERVICE_PORT.defaultValue.get))
+    }
+  }
 }
 
 /**

resource-managers/kubernetes/docker/src/main/dockerfiles/spark/entrypoint.sh

@@ -102,7 +102,8 @@ case "$1" in
       --executor-id $SPARK_EXECUTOR_ID
       --cores $SPARK_EXECUTOR_CORES
       --app-id $SPARK_APPLICATION_ID
-      --hostname $SPARK_EXECUTOR_POD_IP
+      ${EXECUTOR_SERVICE_NAME:+--bind-address $SPARK_EXECUTOR_POD_IP}
+      --hostname ${EXECUTOR_SERVICE_NAME:-$SPARK_EXECUTOR_POD_IP}
       --resourceProfileId $SPARK_RESOURCE_PROFILE_ID
       --podName $SPARK_EXECUTOR_POD_NAME
     )