[SPARK-48786] Define Config Loading Framework for Spark Operator Controller

.github/.licenserc.yaml

@@ -17,5 +17,6 @@ header:
     - '**/*.gradle'
     - gradlew
     - 'build/**'
+    - '**/*.json'
 
   comment: on-failure

build.gradle

@@ -7,6 +7,7 @@ buildscript {
   dependencies {
     classpath "com.github.spotbugs.snom:spotbugs-gradle-plugin:${spotBugsGradlePluginVersion}"
     classpath "com.diffplug.spotless:spotless-plugin-gradle:${spotlessPluginVersion}"
+    classpath "com.github.johnrengelman:shadow:${shadowJarPluginVersion}"
   }
 }
 

dev/.rat-excludes

@@ -14,3 +14,4 @@ TAGS
 RELEASE
 build
 .*gradle
+.*json

gradle.properties

@@ -23,6 +23,9 @@ fabric8Version=6.12.1
 commonsLang3Version=3.14.0
 commonsIOVersion=2.16.1
 lombokVersion=1.18.32
+operatorSDKVersion=4.9.0
+okHttpVersion=4.12.0
+dropwizardMetricsVersion=4.2.25
 
 # Spark
 scalaVersion=2.13
@@ -42,3 +45,6 @@ pmdVersion=6.55.0
 spotBugsGradlePluginVersion=6.0.12
 spotBugsVersion=4.8.4
 spotlessPluginVersion=6.25.0
+
+# Packaging
+shadowJarPluginVersion=8.1.1

settings.gradle

@@ -1,3 +1,4 @@
 rootProject.name = 'apache-spark-kubernetes-operator'
 include 'spark-operator-api'
 include 'spark-submission-worker'
+include 'spark-operator'

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

@@ -26,6 +26,7 @@ public class Constants {
   public static final String LABEL_SPARK_OPERATOR_NAME = "spark.operator/name";
   public static final String LABEL_SENTINEL_RESOURCE = "spark.operator/sentinel";
   public static final String LABEL_RESOURCE_NAME = "app.kubernetes.io/name";
+  public static final String LABEL_COMPONENT_NAME = "app.kubernetes.io/component";
   public static final String LABEL_SPARK_ROLE_NAME = "spark-role";
   public static final String LABEL_SPARK_ROLE_DRIVER_VALUE = "driver";
   public static final String LABEL_SPARK_ROLE_EXECUTOR_VALUE = "executor";

spark-operator/build.gradle

@@ -0,0 +1,58 @@
+apply plugin: 'com.github.johnrengelman.shadow'
+
+dependencies {
+  implementation project(":spark-operator-api")
+  implementation project(":spark-submission-worker")
+
+  implementation("io.javaoperatorsdk:operator-framework:$operatorSDKVersion") {
+    exclude group: 'com.squareup.okio'
+    exclude group: 'io.fabric8'
+  }
+  implementation("io.fabric8:kubernetes-httpclient-okhttp:$fabric8Version") {
+    exclude group: 'com.squareup.okhttp3'
+  }
+  implementation("com.squareup.okhttp3:okhttp:$okHttpVersion")
+  implementation("com.squareup.okhttp3:logging-interceptor:$okHttpVersion")
+  implementation("com.github.spotbugs:spotbugs-annotations:$spotBugsVersion")
+  // logging
+  implementation("org.apache.logging.log4j:log4j-api:$log4jVersion")
+  implementation("org.apache.logging.log4j:log4j-core:$log4jVersion")
+  implementation("org.apache.logging.log4j:log4j-slf4j-impl:$log4jVersion")
+  implementation("org.apache.logging.log4j:log4j-1.2-api:$log4jVersion")
+  implementation("org.apache.logging.log4j:log4j-layout-template-json:$log4jVersion")
+
+  compileOnly("org.projectlombok:lombok:$lombokVersion")
+
+  annotationProcessor("org.projectlombok:lombok:$lombokVersion")
+  annotationProcessor("com.github.spotbugs:spotbugs-annotations:$spotBugsVersion")
+
+  testImplementation("io.javaoperatorsdk:operator-framework-junit-5:$operatorSDKVersion") {
+    exclude group: 'com.squareup.okio'
+    exclude group: 'io.fabric8'
+  }
+  testImplementation platform("org.junit:junit-bom:$junitVersion")
+  testImplementation("org.junit.jupiter:junit-jupiter:$junitVersion")
+  testImplementation("org.mockito:mockito-core:$mockitoVersion")
+}
+
+test {
+  useJUnitPlatform()
+}
+
+jar.dependsOn shadowJar
+
+jar {
+  zip64 = true
+  archiveVersion.set('')
+}
+
+
+shadowJar {
+  zip64 = true
+  mergeServiceFiles()
+  transform(com.github.jengelman.gradle.plugins.shadow.transformers.Log4j2PluginsCacheFileTransformer)
+}
+
+description = "Spark Kubernetes Operator"
+def artifact = "spark-kubernetes-operator"
+archivesBaseName = artifact

spark-operator/src/main/java/org/apache/spark/k8s/operator/config/ConfigOption.java

@@ -0,0 +1,105 @@
+/*
+ * 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.config;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import lombok.ToString;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+
+import org.apache.spark.k8s.operator.utils.ModelUtils;
+
+/** Config options for Spark Operator. Supports primitive and serialized JSON */
+@RequiredArgsConstructor
+@AllArgsConstructor
+@EqualsAndHashCode
+@ToString
+@Builder
+@Slf4j
+public class ConfigOption<T> {
+  @Getter @Builder.Default private boolean enableDynamicOverride = true;
+  @Getter private String key;
+  @Getter private String description;
+  private T defaultValue;
+  private Class<T> typeParameterClass;
+
+  public T getValue() {
+    T resolvedValue = resolveValue();
+    if (log.isDebugEnabled()) {
+      log.debug("Resolved value for property {}={}", key, resolvedValue);
+    }
+    return resolveValue();
+  }
+
+  private T resolveValue() {
+    try {
+      String value = SparkOperatorConfManager.INSTANCE.getValue(key);
+      if (!enableDynamicOverride) {
+        value = SparkOperatorConfManager.INSTANCE.getInitialValue(key);
+      }
+      if (StringUtils.isNotEmpty(value)) {
+        if (typeParameterClass.isPrimitive() || typeParameterClass == String.class) {
+          return (T) resolveValueToPrimitiveType(typeParameterClass, value);
+        } else {
+          return ModelUtils.objectMapper.readValue(value, typeParameterClass);
+        }
+      } else {
+        return defaultValue;
+      }
+    } catch (NumberFormatException | JsonProcessingException t) {
+      log.error(
+          "Failed to resolve value for config key {}, using default value {}",
+          key,
+          defaultValue,
+          t);
+      return defaultValue;
+    }
+  }
+
+  public static Object resolveValueToPrimitiveType(Class clazz, String value) {
+    if (Boolean.class == clazz || Boolean.TYPE == clazz) {
+      return Boolean.parseBoolean(value);
+    }
+    if (Byte.class == clazz || Byte.TYPE == clazz) {
+      return Byte.parseByte(value);
+    }
+    if (Short.class == clazz || Short.TYPE == clazz) {
+      return Short.parseShort(value);
+    }
+    if (Integer.class == clazz || Integer.TYPE == clazz) {
+      return Integer.parseInt(value);
+    }
+    if (Long.class == clazz || Long.TYPE == clazz) {
+      return Long.parseLong(value);
+    }
+    if (Float.class == clazz || Float.TYPE == clazz) {
+      return Float.parseFloat(value);
+    }
+    if (Double.class == clazz || Double.TYPE == clazz) {
+      return Double.parseDouble(value);
+    }
+    return value;
+  }
+}

spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConfManager.java

@@ -0,0 +1,101 @@
+/*
+ * 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.config;
+
+import java.io.FileInputStream;
+import java.io.InputStream;
+import java.util.Map;
+import java.util.Properties;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+
+/**
+ * Loads ConfigOption from properties file. In addition, loads hot properties override from config
+ * map if dynamic config is enabled.
+ */
+@Slf4j
+public class SparkOperatorConfManager {
+  public static final String BASE_CONFIG_DIR = "/opt/spark-operator/conf/";
+  public static final String INITIAL_CONFIG_FILE_PATH =
+      BASE_CONFIG_DIR + "spark-operator.properties";
+
+  public static final String METRICS_CONFIG_FILE_PATH = BASE_CONFIG_DIR + "metrics.properties";
+
+  public static final String INITIAL_CONFIG_FILE_PATH_PROPS_KEY =
+      "spark.kubernetes.operator.basePropertyFileName";
+
+  public static final String METRICS_CONFIG_FILE_PATH_PROPS_KEY =
+      "spark.kubernetes.operator.metrics.propertyFileName";
+
+  public static final SparkOperatorConfManager INSTANCE = new SparkOperatorConfManager();
+  protected final Properties initialConfig;
+  protected final Properties metricsConfig;
+  protected Properties configOverrides;
+
+  protected SparkOperatorConfManager() {
+    this.initialConfig = new Properties();
+    this.configOverrides = new Properties();
+    this.metricsConfig = new Properties();
+    initialize();
+  }
+
+  public String getValue(String key) {
+    String currentValue = configOverrides.getProperty(key);
+    return StringUtils.isEmpty(currentValue) ? getInitialValue(key) : currentValue;
+  }
+
+  public String getInitialValue(String key) {
+    return initialConfig.getProperty(key);
+  }
+
+  public void refresh(Map<String, String> updatedConfig) {
+    synchronized (this) {
+      this.configOverrides = new Properties();
+      configOverrides.putAll(updatedConfig);
+    }
+  }
+
+  public Properties getMetricsProperties() {
+    return metricsConfig;
+  }
+
+  private void initialize() {
+    initialConfig.putAll(System.getProperties());
+    Properties properties =
+        getProperties(
+            System.getProperty(INITIAL_CONFIG_FILE_PATH_PROPS_KEY, INITIAL_CONFIG_FILE_PATH));
+    initialConfig.putAll(properties);
+    Properties metricsProperties =
+        getProperties(
+            System.getProperty(METRICS_CONFIG_FILE_PATH_PROPS_KEY, METRICS_CONFIG_FILE_PATH));
+    metricsConfig.putAll(metricsProperties);
+  }
+
+  private Properties getProperties(String filePath) {
+    Properties properties = new Properties();
+    try (InputStream inputStream = new FileInputStream(filePath)) {
+      properties.load(inputStream);
+    } catch (Exception e) {
+      log.error("Failed to load properties from {}.", filePath, e);
+    }
+    return properties;
+  }
+}

spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConfigMapReconciler.java

@@ -0,0 +1,82 @@
+/*
+ * 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.config;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Function;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.javaoperatorsdk.operator.api.config.informer.InformerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.Context;
+import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusHandler;
+import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusUpdateControl;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
+import io.javaoperatorsdk.operator.api.reconciler.EventSourceInitializer;
+import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
+import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
+import io.javaoperatorsdk.operator.processing.event.source.EventSource;
+import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * This serves dynamic configuration for Spark Operator. When enabled, Operator assumes config file
+ * is located in given config map. It would keep watch the config map & apply changes when update is
+ * detected.
+ */
+@ControllerConfiguration
+@RequiredArgsConstructor
+@Slf4j
+public class SparkOperatorConfigMapReconciler
+    implements Reconciler<ConfigMap>,
+        ErrorStatusHandler<ConfigMap>,
+        EventSourceInitializer<ConfigMap> {
+  private final Function<Set<String>, Boolean> namespaceUpdater;
+  private final String operatorNamespace;
+  private final Function<Void, Set<String>> watchedNamespacesGetter;
+
+  @Override
+  public ErrorStatusUpdateControl<ConfigMap> updateErrorStatus(
+      ConfigMap resource, Context<ConfigMap> context, Exception e) {
+    log.error("Failed to reconcile dynamic config change.");
+    return ErrorStatusUpdateControl.noStatusUpdate();
+  }
+
+  @Override
+  public Map<String, EventSource> prepareEventSources(EventSourceContext<ConfigMap> context) {
+    EventSource configMapEventSource =
+        new InformerEventSource<>(
+            InformerConfiguration.from(ConfigMap.class, context)
+                .withNamespaces(operatorNamespace)
+                .build(),
+            context);
+    return EventSourceInitializer.nameEventSources(configMapEventSource);
+  }
+
+  @Override
+  public UpdateControl<ConfigMap> reconcile(ConfigMap resource, Context<ConfigMap> context)
+      throws Exception {
+    SparkOperatorConfManager.INSTANCE.refresh(resource.getData());
+    namespaceUpdater.apply(watchedNamespacesGetter.apply(null));
+    return UpdateControl.noUpdate();
+  }
+}