[SPARK-51347] Enable Ingress and Service Support for Spark Driver

### What changes were proposed in this pull request?

This PR adds support for launching Ingress and Services with Spark Applications.

### Why are the changes needed?

There's common ask from user to support exposing driver via ingress and service to serve HTTP / HTTPS from outside the cluster - for example, to access Spark UI.

As we support podTemplateSupport already, this feature can also be handy for those who would like to expose additional service ports in the same pod. Therefore, this PR aims to introduce this feature in a more general way instead of just exposing Spark UI.

### Does this PR introduce _any_ user-facing change?

No - unreleased version

### How was this patch tested?

CIs

### Was this patch authored or co-authored using generative AI tooling?

No

Closes apache#159 from jiangzho/um.

Authored-by: Zhou JIANG <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>

Author: jiangzho

docs/spark_custom_resources.md

@@ -99,6 +99,46 @@ Please be advised that Spark still overrides necessary pod configuration in both
 more details,
 refer [Spark doc](https://spark.apache.org/docs/latest/running-on-kubernetes.html#pod-template).
 
+## Enable Additional Ingress for Driver
+
+Operator may create [Ingress](https://kubernetes.io/docs/concepts/services-networking/ingress/) for
+Spark driver of running applications on demand. For example, to expose Spark UI - which is by 
+default enabled on driver port 4040, you may configure 
+
+```yaml
+spec:
+  driverServiceIngressList:
+    - serviceMetadata:
+        name: "spark-ui-service"
+      serviceSpec:
+        ports:
+          - protocol: TCP
+            port: 80
+            targetPort: 4040
+      ingressMetadata:
+        name: "spark-ui-ingress"
+        annotations:
+          nginx.ingress.kubernetes.io/rewrite-target: /
+      ingressSpec:
+        ingressClassName: nginx-example
+        rules:
+          - http:
+              paths:
+                - path: "/"
+                  pathType: Prefix
+                  backend:
+                    service:
+                      name: spark-ui-service
+                      port:
+                        number: 80
+```
+
+Spark Operator by default would populate the `.spec.selector` field of the created Service to match 
+the driver labels. If `.ingressSpec.rules` is not provided, Spark Operator would also populate one
+default rule backed by the associated Service. It's recommended to always provide the ingress spec 
+to make sure it's compatible with your 
+[IngressController](https://kubernetes.io/docs/concepts/services-networking/ingress-controllers/).
+
 ## Understanding Failure Types
 
 In addition to the general `Failed` state (that driver pod fails or driver container exits

spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/ApplicationSpec.java

@@ -54,4 +54,5 @@ public class ApplicationSpec extends BaseSpec {
 
   protected BaseApplicationTemplateSpec driverSpec;
   protected BaseApplicationTemplateSpec executorSpec;
+  protected List<DriverServiceIngressSpec> driverServiceIngressList;
 }

spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/DriverServiceIngressSpec.java

@@ -0,0 +1,41 @@
+/*
+ * 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.k8s.operator.spec;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import io.fabric8.generator.annotation.Required;
+import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.fabric8.kubernetes.api.model.ServiceSpec;
+import io.fabric8.kubernetes.api.model.networking.v1.IngressSpec;
+import lombok.Builder;
+import lombok.Data;
+
+@Data
+@Builder
+@JsonInclude(JsonInclude.Include.NON_NULL)
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class DriverServiceIngressSpec {
+  @Required protected ObjectMeta serviceMetadata;
+  @Required protected ServiceSpec serviceSpec;
+
+  @Required protected ObjectMeta ingressMetadata;
+  protected IngressSpec ingressSpec;
+}

spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppResourceSpec.java

@@ -20,7 +20,10 @@
 package org.apache.spark.k8s.operator;
 
 import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
+import java.util.stream.Collectors;
 
 import scala.Tuple2;
 import scala.collection.immutable.HashMap;
@@ -40,6 +43,8 @@
 import org.apache.spark.deploy.k8s.KubernetesDriverSpec;
 import org.apache.spark.deploy.k8s.SparkPod;
 import org.apache.spark.deploy.k8s.submit.KubernetesClientUtils;
+import org.apache.spark.k8s.operator.spec.DriverServiceIngressSpec;
+import org.apache.spark.k8s.operator.utils.DriverServiceIngressUtils;
 
 /**
  * Resembles resources that would be directly launched by operator. Based on resolved
@@ -59,7 +64,9 @@ public class SparkAppResourceSpec {
   private final SparkAppDriverConf kubernetesDriverConf;
 
   public SparkAppResourceSpec(
-      SparkAppDriverConf kubernetesDriverConf, KubernetesDriverSpec kubernetesDriverSpec) {
+      SparkAppDriverConf kubernetesDriverConf,
+      KubernetesDriverSpec kubernetesDriverSpec,
+      List<DriverServiceIngressSpec> driverServiceIngressList) {
     this.kubernetesDriverConf = kubernetesDriverConf;
     String namespace = kubernetesDriverConf.sparkConf().get(Config.KUBERNETES_NAMESPACE().key());
     Map<String, String> confFilesMap =
@@ -86,6 +93,7 @@ public SparkAppResourceSpec(
     this.driverResources.add(
         KubernetesClientUtils.buildConfigMap(
             kubernetesDriverConf.configMapNameDriver(), confFilesMap, new HashMap<>()));
+    this.driverResources.addAll(configureDriverServerIngress(sparkPod, driverServiceIngressList));
     this.driverPreResources.forEach(r -> setNamespaceIfMissing(r, namespace));
     this.driverResources.forEach(r -> setNamespaceIfMissing(r, namespace));
   }
@@ -126,4 +134,15 @@ private SparkPod addConfigMap(SparkPod pod, Map<String, String> confFilesMap) {
             .build();
     return new SparkPod(podWithConfigMapVolume, containerWithConfigMapVolume);
   }
+
+  private List<HasMetadata> configureDriverServerIngress(
+      SparkPod pod, List<DriverServiceIngressSpec> driverServiceIngressList) {
+    if (driverServiceIngressList == null || driverServiceIngressList.isEmpty()) {
+      return Collections.emptyList();
+    }
+    return driverServiceIngressList.stream()
+        .map(spec -> DriverServiceIngressUtils.buildIngressService(spec, pod.pod().getMetadata()))
+        .flatMap(Collection::stream)
+        .collect(Collectors.toList());
+  }
 }

spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/SparkAppSubmissionWorker.java

@@ -20,6 +20,7 @@
 package org.apache.spark.k8s.operator;
 
 import java.math.BigInteger;
+import java.util.List;
 import java.util.Map;
 
 import scala.Option;
@@ -37,6 +38,7 @@
 import org.apache.spark.deploy.k8s.submit.PythonMainAppResource;
 import org.apache.spark.deploy.k8s.submit.RMainAppResource;
 import org.apache.spark.k8s.operator.spec.ApplicationSpec;
+import org.apache.spark.k8s.operator.spec.DriverServiceIngressSpec;
 import org.apache.spark.k8s.operator.utils.ModelUtils;
 
 /**
@@ -82,7 +84,7 @@ public class SparkAppSubmissionWorker {
   public SparkAppResourceSpec getResourceSpec(
       SparkApplication app, KubernetesClient client, Map<String, String> confOverrides) {
     SparkAppDriverConf appDriverConf = buildDriverConf(app, confOverrides);
-    return buildResourceSpec(appDriverConf, client);
+    return buildResourceSpec(appDriverConf, app.getSpec().getDriverServiceIngressList(), client);
   }
 
   protected SparkAppDriverConf buildDriverConf(
@@ -127,11 +129,14 @@ protected SparkAppDriverConf buildDriverConf(
   }
 
   protected SparkAppResourceSpec buildResourceSpec(
-      SparkAppDriverConf kubernetesDriverConf, KubernetesClient client) {
+      SparkAppDriverConf kubernetesDriverConf,
+      List<DriverServiceIngressSpec> driverServiceIngressList,
+      KubernetesClient client) {
     KubernetesDriverBuilder builder = new KubernetesDriverBuilder();
     KubernetesDriverSpec kubernetesDriverSpec =
         builder.buildFromFeatures(kubernetesDriverConf, client);
-    return new SparkAppResourceSpec(kubernetesDriverConf, kubernetesDriverSpec);
+    return new SparkAppResourceSpec(
+        kubernetesDriverConf, kubernetesDriverSpec, driverServiceIngressList);
   }
 
   /**

spark-submission-worker/src/main/java/org/apache/spark/k8s/operator/utils/DriverServiceIngressUtils.java

@@ -0,0 +1,95 @@
+/*
+ * 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.k8s.operator.utils;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.fabric8.kubernetes.api.model.ObjectMeta;
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
+import io.fabric8.kubernetes.api.model.Service;
+import io.fabric8.kubernetes.api.model.ServiceBuilder;
+import io.fabric8.kubernetes.api.model.networking.v1.Ingress;
+import io.fabric8.kubernetes.api.model.networking.v1.IngressBuilder;
+import io.fabric8.kubernetes.api.model.networking.v1.IngressSpec;
+import io.fabric8.kubernetes.api.model.networking.v1.IngressSpecBuilder;
+
+import org.apache.spark.k8s.operator.spec.DriverServiceIngressSpec;
+
+public final class DriverServiceIngressUtils {
+  private DriverServiceIngressUtils() {}
+
+  /** Build the full spec for ingress and service. */
+  public static List<HasMetadata> buildIngressService(
+      DriverServiceIngressSpec spec, ObjectMeta driverPodMetaData) {
+    List<HasMetadata> resources = new ArrayList<>(2);
+    Service service = buildService(spec, driverPodMetaData);
+    resources.add(service);
+    resources.add(buildIngress(spec, service));
+    return resources;
+  }
+
+  private static Service buildService(DriverServiceIngressSpec spec, ObjectMeta driverPodMetaData) {
+    ObjectMeta serviceMeta = new ObjectMetaBuilder(spec.getServiceMetadata()).build();
+    serviceMeta.setNamespace(driverPodMetaData.getNamespace());
+    Map<String, String> selectors = spec.getServiceSpec().getSelector();
+    if (selectors == null || selectors.isEmpty()) {
+      selectors = driverPodMetaData.getLabels();
+    }
+    return new ServiceBuilder()
+        .withMetadata(serviceMeta)
+        .withNewSpecLike(spec.getServiceSpec())
+        .withSelector(selectors)
+        .endSpec()
+        .build();
+  }
+
+  private static Ingress buildIngress(DriverServiceIngressSpec spec, Service service) {
+    ObjectMeta metadata = new ObjectMetaBuilder(spec.getIngressMetadata()).build();
+    IngressSpec ingressSpec = new IngressSpecBuilder(spec.getIngressSpec()).build();
+    if ((ingressSpec.getRules() == null || ingressSpec.getRules().isEmpty())
+        && service.getSpec().getPorts() != null
+        && !service.getSpec().getPorts().isEmpty()) {
+      // if no rule is provided, populate default path with backend to the associated service
+      ingressSpec =
+          new IngressSpecBuilder()
+              .addNewRule()
+              .withNewHttp()
+              .addNewPath()
+              .withPath("/")
+              .withPathType("ImplementationSpecific")
+              .withNewBackend()
+              .withNewService()
+              .withName(service.getMetadata().getName())
+              .withNewPort()
+              .withNumber(service.getSpec().getPorts().get(0).getPort())
+              .endPort()
+              .endService()
+              .endBackend()
+              .endPath()
+              .endHttp()
+              .endRule()
+              .build();
+    }
+    return new IngressBuilder().withMetadata(metadata).withSpec(ingressSpec).build();
+  }
+}

spark-submission-worker/src/test/java/org/apache/spark/k8s/operator/SparkAppResourceSpecTest.java

@@ -69,7 +69,8 @@ void testDriverResourceIncludesConfigMap() {
     when(mockSpec.pod()).thenReturn(sparkPod);
     when(mockSpec.systemProperties()).thenReturn(new HashMap<>());
 
-    SparkAppResourceSpec appResourceSpec = new SparkAppResourceSpec(mockConf, mockSpec);
+    SparkAppResourceSpec appResourceSpec =
+        new SparkAppResourceSpec(mockConf, mockSpec, Collections.emptyList());
 
     Assertions.assertEquals(2, appResourceSpec.getDriverResources().size());
     Assertions.assertEquals(1, appResourceSpec.getDriverPreResources().size());