[SPARK-51507] Support creating config map that can be mounted by Spark pods for apps

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

This PR introduces a new field `configMapSpecs`, which enables user to create configmap(s) which can be later mounted on driver and/or executors with values specified in SparkApplications. It also adds corresponding docs and unit tests.

### Why are the changes needed?

There's a common user demand for the feature to create lightweight values / files override using configmap. This new feature saves the trouble for user to create configmap(s) beforehand - the config map would share the lifecycle of the corresponding Spark app.

User may mount the ConfigMap as needed in driver and executor pod templates - one example is provided in the doc. In future we may further enhance this flow by mounting the created config map automatically to Spark pods. This can be achieved by supporting volume type of ConfigMap in Spark Kubernetes, or by adding mutating webhook & alter pod templates after creation. For now, this PR opens up the possibility of creating config map and mounting them with one single spec.

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

No - not yet released

### How was this patch tested?

Pass the CIs

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

No

Closes #166 from jiangzho/cm.

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

Author: jiangzho

docs/spark_custom_resources.md

@@ -138,6 +138,62 @@ default rule backed by the associated Service. It's recommended to always provid
 to make sure it's compatible with your 
 [IngressController](https://kubernetes.io/docs/concepts/services-networking/ingress-controllers/).
 
+## Create and Mount ConfigMap
+
+It is possible to ask operator to create configmap so they can be used by driver and/or executor 
+pods on the fly. `configMapSpecs` allows you to specify the desired metadata and data as string 
+literals for the configmap(s) to be created.
+
+```yaml
+spec:
+  configMapSpecs:
+    - name: "example-config-map"
+      data:
+        foo: "bar"
+```
+
+Like other app-specific resources, the created configmap has owner reference to Spark driver and
+therefore shares the same lifecycle and garbage collection mechanism with the associated app.  
+
+This feature can be used to create lightweight override config files for given Spark app. For 
+example, below snippet would create and mount a configmap with metrics property file, then use it
+in SparkConf:   
+
+```yaml
+spec:
+  sparkConf:
+    spark.metrics.conf: "/etc/metrics/metrics.properties"
+  driverSpec:
+    podTemplateSpec:
+      spec:
+        containers:
+          - volumeMounts:
+              - name: "config-override"
+                mountPath: "/etc/metrics"
+                readOnly: true
+        volumes:
+          - name: config-override
+            configMap:
+              name: metrics-configmap
+  executorSpec:
+    podTemplateSpec:
+      spec:
+        containers:
+          - volumeMounts:
+              - name: "config-override"
+                mountPath: "/etc/metrics"
+                readOnly: true
+        volumes:
+          - name: config-override
+            configMap:
+              name: metrics-configmap
+  configMapSpecs:
+    - name: "metrics-configmap"
+      data:
+        metrics.properties: "*.sink.jmx.class=org.apache.spark.metrics.sink.JmxSink\n"
+
+```
+
 ## 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

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

spark-operator-api/src/main/java/org/apache/spark/k8s/operator/spec/ConfigMapSpec.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 java.util.Map;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import io.fabric8.generator.annotation.Required;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+@Builder
+@JsonInclude(JsonInclude.Include.NON_NULL)
+public class ConfigMapSpec {
+  @Required protected String name;
+  protected Map<String, String> labels;
+  protected Map<String, String> annotations;
+  protected Map<String, String> data;
+}

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

@@ -43,7 +43,9 @@
 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.ConfigMapSpec;
 import org.apache.spark.k8s.operator.spec.DriverServiceIngressSpec;
+import org.apache.spark.k8s.operator.utils.ConfigMapSpecUtils;
 import org.apache.spark.k8s.operator.utils.DriverServiceIngressUtils;
 
 /**
@@ -66,7 +68,8 @@ public class SparkAppResourceSpec {
   public SparkAppResourceSpec(
       SparkAppDriverConf kubernetesDriverConf,
       KubernetesDriverSpec kubernetesDriverSpec,
-      List<DriverServiceIngressSpec> driverServiceIngressList) {
+      List<DriverServiceIngressSpec> driverServiceIngressList,
+      List<ConfigMapSpec> configMapSpecs) {
     this.kubernetesDriverConf = kubernetesDriverConf;
     String namespace = kubernetesDriverConf.sparkConf().get(Config.KUBERNETES_NAMESPACE().key());
     Map<String, String> confFilesMap =
@@ -93,6 +96,7 @@ public SparkAppResourceSpec(
     this.driverResources.add(
         KubernetesClientUtils.buildConfigMap(
             kubernetesDriverConf.configMapNameDriver(), confFilesMap, new HashMap<>()));
+    this.driverPreResources.addAll(ConfigMapSpecUtils.buildConfigMaps(configMapSpecs));
     this.driverResources.addAll(configureDriverServerIngress(sparkPod, driverServiceIngressList));
     this.driverPreResources.forEach(r -> setNamespaceIfMissing(r, namespace));
     this.driverResources.forEach(r -> setNamespaceIfMissing(r, namespace));

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

@@ -38,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.ConfigMapSpec;
 import org.apache.spark.k8s.operator.spec.DriverServiceIngressSpec;
 import org.apache.spark.k8s.operator.utils.ModelUtils;
 
@@ -84,7 +85,11 @@ public class SparkAppSubmissionWorker {
   public SparkAppResourceSpec getResourceSpec(
       SparkApplication app, KubernetesClient client, Map<String, String> confOverrides) {
     SparkAppDriverConf appDriverConf = buildDriverConf(app, confOverrides);
-    return buildResourceSpec(appDriverConf, app.getSpec().getDriverServiceIngressList(), client);
+    return buildResourceSpec(
+        appDriverConf,
+        app.getSpec().getDriverServiceIngressList(),
+        app.getSpec().getConfigMapSpecs(),
+        client);
   }
 
   protected SparkAppDriverConf buildDriverConf(
@@ -131,12 +136,13 @@ protected SparkAppDriverConf buildDriverConf(
   protected SparkAppResourceSpec buildResourceSpec(
       SparkAppDriverConf kubernetesDriverConf,
       List<DriverServiceIngressSpec> driverServiceIngressList,
+      List<ConfigMapSpec> configMapSpecs,
       KubernetesClient client) {
     KubernetesDriverBuilder builder = new KubernetesDriverBuilder();
     KubernetesDriverSpec kubernetesDriverSpec =
         builder.buildFromFeatures(kubernetesDriverConf, client);
     return new SparkAppResourceSpec(
-        kubernetesDriverConf, kubernetesDriverSpec, driverServiceIngressList);
+        kubernetesDriverConf, kubernetesDriverSpec, driverServiceIngressList, configMapSpecs);
   }
 
   /**

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

@@ -0,0 +1,51 @@
+/*
+ * 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.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.api.model.ConfigMapBuilder;
+
+import org.apache.spark.k8s.operator.spec.ConfigMapSpec;
+
+public final class ConfigMapSpecUtils {
+  private ConfigMapSpecUtils() {}
+
+  public static List<ConfigMap> buildConfigMaps(List<ConfigMapSpec> configMapMountList) {
+    if (configMapMountList == null || configMapMountList.isEmpty()) {
+      return Collections.emptyList();
+    }
+    return configMapMountList.stream()
+        .map(
+            c ->
+                new ConfigMapBuilder()
+                    .editOrNewMetadata()
+                    .withName(c.getName())
+                    .withLabels(c.getLabels())
+                    .withAnnotations(c.getAnnotations())
+                    .endMetadata()
+                    .addToData(c.getData())
+                    .build())
+        .collect(Collectors.toList());
+  }
+}

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

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

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

@@ -0,0 +1,70 @@
+/*
+ * 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 static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import org.junit.jupiter.api.Test;
+
+import org.apache.spark.k8s.operator.spec.ConfigMapSpec;
+
+class ConfigMapSpecUtilsTest {
+
+  @Test
+  void buildConfigMaps() {
+
+    assertEquals(Collections.emptyList(), ConfigMapSpecUtils.buildConfigMaps(null));
+    assertEquals(
+        Collections.emptyList(), ConfigMapSpecUtils.buildConfigMaps(Collections.emptyList()));
+
+    Map<String, String> labels = Map.of("foo-label-key", "foo-label-value");
+    Map<String, String> annotations = Map.of("foo-annotation-key", "foo-annotation-value");
+    Map<String, String> configMapData = new HashMap<>();
+    // literal value
+    configMapData.put("foo", "bar");
+    // escaped property file
+    configMapData.put(
+        "conf.properties", "spark.foo.keyOne=\"valueOne\"\nspark.foo.keyTwo=\"valueTwo\"");
+    // escaped yaml file
+    configMapData.put("conf.yaml", "conf:\n  foo: bar\n  nestedField:\n    keyName: value");
+    ConfigMapSpec mount =
+        ConfigMapSpec.builder()
+            .name("test-config-map")
+            .labels(labels)
+            .annotations(annotations)
+            .data(configMapData)
+            .build();
+    List<ConfigMap> configMaps =
+        ConfigMapSpecUtils.buildConfigMaps(Collections.singletonList(mount));
+    assertEquals(1, configMaps.size());
+    ConfigMap configMap = configMaps.get(0);
+    assertEquals("test-config-map", configMap.getMetadata().getName());
+    assertEquals("foo-label-value", configMap.getMetadata().getLabels().get("foo-label-key"));
+    assertEquals(
+        "foo-annotation-value", configMap.getMetadata().getAnnotations().get("foo-annotation-key"));
+    assertEquals(configMapData, configMap.getData());
+  }
+}