This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/spark-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new 5283a09 [SPARK-57466] Support `s.k.o.dynamicConfig.source` to use
mounted-`ConfigMap`
5283a09 is described below
commit 5283a09f751f00b3a8f04a6ce04c1ec31647cf38
Author: Attila Mészáros <[email protected]>
AuthorDate: Tue Jun 30 13:49:05 2026 -0700
[SPARK-57466] Support `s.k.o.dynamicConfig.source` to use
mounted-`ConfigMap`
### What changes were proposed in this pull request?
This PR adds a second, mount-based source for dynamic configuration
overrides and makes the dynamic-config mechanism pluggable, while retaining the
existing ConfigMap-informer behavior for backwards compatibility.
A new config option `spark.kubernetes.operator.dynamicConfig.source`
selects the source:
- **`configMap`** (default) — the existing behavior: a dedicated
`Operator` runs `SparkOperatorConfigMapReconciler`, watching a ConfigMap via a
Kubernetes informer. Requires RBAC to read ConfigMaps.
- **`file`** — a new `DynamicConfigMonitor` periodically reloads a
properties file mounted from a ConfigMap volume, requiring no extra RBAC.
Both paths funnel through the same
`SparkOperatorConfManager.refresh(...)` + watched-namespace update logic, so
runtime behavior is identical once overrides are applied.
Key changes:
- **New `DynamicConfigMonitor`** (`config/DynamicConfigMonitor.java`) — a
single-thread scheduled reader of a properties file with `start()` / `stop()` /
`isRunning()` lifecycle and change detection against the last loaded
snapshot.
- **`SparkOperator`** — `registerDynamicConfig()` dispatches on
`dynamicConfig.source`: the `file` source returns a `DynamicConfigMonitor`; the
`configMap` source registers the second informer `Operator` (restored
`registerSparkOperatorConfMonitor()` / `overrideConfigMonitorConfigs()`).
The operator still tracks a `List<Operator>` plus an optional monitor.
- **`SparkOperatorConf`** — added `dynamicConfig.source`,
`dynamicConfig.filePath`, `dynamicConfig.reloadIntervalSeconds`; retained
`dynamicConfig.selector` and `dynamicConfig.reconcilerParallelism`.
- **Probes** (`HealthProbe`, `ReadinessProbe`, `ProbeService`,
`ProbeUtil`) — health/readiness cover all registered operators (so the informer
operator is checked under `configMap`) and additionally the monitor's running
state under `file`.
- **Helm chart**:
- `dynamicConfig.source` added to `values.yaml` (default `configMap`).
- The ConfigMap `Role`/`RoleBinding` is created only when `enable &&
source == configMap`.
- The dynamic-config volume and mount are added only when `enable &&
source == file`.
- The dynamic ConfigMap renders a single
`spark-operator-dynamic.properties` key for the `file` source, or raw per-key
data for the `configMap` source.
### Why are the changes needed?
The informer-based source requires the operator to hold cluster/namespace
RBAC on ConfigMaps and runs a second JOSDK `Operator` purely to watch one
ConfigMap. The mount-based source removes that RBAC requirement and the extra
informer by letting the kubelet project ConfigMap changes onto a volume
that the operator polls. Rather than forcing a migration, this PR offers the
mount-based approach as an opt-in alternative while keeping the informer as
the default, so existing deployments are unaffected.
### Does this PR introduce _any_ user-facing change?
Yes, additive and backwards-compatible:
- New `spark.kubernetes.operator.dynamicConfig.source` option (default
`configMap`). With the default, behavior is unchanged from the released
operator.
- New `dynamicConfig.filePath` and `dynamicConfig.reloadIntervalSeconds`
options, used only by the `file` source.
- New `dynamicConfig.source` Helm value (default `configMap`). Opting
into `file` switches the chart to mount the ConfigMap as a volume (rendered
under a single `spark-operator-dynamic.properties` key) and drops the ConfigMap
RBAC `Role`/`RoleBinding`.
No change for users who do not set `dynamicConfig.source`.
### How was this patch tested?
- Unit tests for both sources:
- `configMap` — `SparkOperatorConfigMapReconcilerTest` and the
configMap-source case in `SparkOperatorTest` (asserts a second `Operator` is
registered and no `DynamicConfigMonitor`).
- `file` — `DynamicConfigMonitorTest` (initial load, no-op reload when
unchanged, refresh + namespace update on change, `isRunning()` lifecycle) and
the file-source case in `SparkOperatorTest`.
- Updated `HealthProbeTest`, `ReadinessProbeTest`, `ProbeServiceTest` for
the `List<Operator>` + monitor signatures.
- e2e `watched-namespaces` chainsaw test runs against the `file` source
(`tests/e2e/helm/dynamic-config-values{,-2}.yaml` set `source: file`);
ConfigMaps use the `spark-operator-dynamic.properties` layout and the
propagation
wait was raised to account for kubelet volume-sync plus the reload
interval.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
Closes #707 from csviri/dynamic-config-map-name-informer-mount.
Authored-by: Attila Mészáros <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.github/workflows/build_and_test.yml | 17 +++
.../templates/_helpers.tpl | 1 +
.../templates/operator-rbac.yaml | 2 +-
.../templates/spark-operator.yaml | 18 +++
.../helm/spark-kubernetes-operator/values.yaml | 6 +
docs/config_properties.md | 7 +-
gradle/libs.versions.toml | 2 +
spark-operator/build.gradle | 1 +
.../apache/spark/k8s/operator/SparkOperator.java | 54 ++++++-
.../k8s/operator/config/DynamicConfigMonitor.java | 162 +++++++++++++++++++++
.../k8s/operator/config/SparkOperatorConf.java | 105 ++++++++++++-
.../spark/k8s/operator/probe/HealthProbe.java | 6 +
.../spark/k8s/operator/probe/ProbeService.java | 16 +-
.../spark/k8s/operator/probe/ReadinessProbe.java | 16 +-
.../spark/k8s/operator/SparkOperatorTest.java | 75 +++++++++-
.../operator/config/DynamicConfigMonitorTest.java | 151 +++++++++++++++++++
.../spark/k8s/operator/probe/HealthProbeTest.java | 20 ++-
.../spark/k8s/operator/probe/ProbeServiceTest.java | 7 +-
.../k8s/operator/probe/ReadinessProbeTest.java | 52 +++++--
tests/e2e/helm/dynamic-config-values-file.yaml | 40 +++++
.../e2e/watched-namespaces-file/chainsaw-test.yaml | 77 ++++++++++
.../e2e/watched-namespaces-file/spark-example.yaml | 31 ++++
.../spark-operator-dynamic-config-1.yaml | 35 +++++
23 files changed, 857 insertions(+), 44 deletions(-)
diff --git a/.github/workflows/build_and_test.yml
b/.github/workflows/build_and_test.yml
index d75e552..790fe3e 100644
--- a/.github/workflows/build_and_test.yml
+++ b/.github/workflows/build_and_test.yml
@@ -122,6 +122,9 @@ jobs:
- mode: selector
test-group: driver-start-timeout
include:
+ - kubernetes-version: "1.36.0"
+ mode: dynamic-file
+ test-group: watched-namespaces-file
- kubernetes-version: "1.36.0"
mode: static
test-group: pi-with-comet
@@ -188,6 +191,20 @@ jobs:
if: matrix.mode == 'dynamic'
run: |
chainsaw test --test-dir ./tests/e2e/${{ matrix.test-group }}
--parallel 1
+ - name: Run Spark K8s Operator on K8S with Dynamic Configuration Enabled
(File source)
+ if: matrix.mode == 'dynamic-file'
+ run: |
+ eval $(minikube docker-env)
+ ./gradlew buildDockerImage
+ helm install spark --create-namespace -f \
+ build-tools/helm/spark-kubernetes-operator/values.yaml -f \
+ tests/e2e/helm/dynamic-config-values-file.yaml \
+ build-tools/helm/spark-kubernetes-operator/
+ minikube docker-env --unset
+ - name: Run E2E Test with Dynamic Configuration Enabled (File source)
+ if: matrix.mode == 'dynamic-file'
+ run: |
+ chainsaw test --test-dir ./tests/e2e/${{ matrix.test-group }}
--parallel 1
- name: Run Spark K8s Operator on K8S with Resource Selector Enabled
if: matrix.mode == 'selector'
run: |
diff --git a/build-tools/helm/spark-kubernetes-operator/templates/_helpers.tpl
b/build-tools/helm/spark-kubernetes-operator/templates/_helpers.tpl
index 37d58df..94452ca 100644
--- a/build-tools/helm/spark-kubernetes-operator/templates/_helpers.tpl
+++ b/build-tools/helm/spark-kubernetes-operator/templates/_helpers.tpl
@@ -114,6 +114,7 @@ Default property overrides
spark.kubernetes.operator.namespace={{ .Release.Namespace }}
spark.kubernetes.operator.name={{- include "spark-operator.name" . }}
spark.kubernetes.operator.dynamicConfig.enabled={{
.Values.operatorConfiguration.dynamicConfig.enable }}
+spark.kubernetes.operator.dynamicConfig.source={{
.Values.operatorConfiguration.dynamicConfig.source }}
spark.kubernetes.operator.metrics.port={{ include "spark-operator.metricsPort"
. }}
spark.kubernetes.operator.health.probePort={{ include
"spark-operator.probePort" . }}
{{- if .Values.workloadResources.namespaces.overrideWatchedNamespaces }}
diff --git
a/build-tools/helm/spark-kubernetes-operator/templates/operator-rbac.yaml
b/build-tools/helm/spark-kubernetes-operator/templates/operator-rbac.yaml
index 59d155b..9c32745 100644
--- a/build-tools/helm/spark-kubernetes-operator/templates/operator-rbac.yaml
+++ b/build-tools/helm/spark-kubernetes-operator/templates/operator-rbac.yaml
@@ -118,7 +118,7 @@ metadata:
{{- template "spark-operator.operatorRbacRules" $ }}
---
{{- end }}
-{{- if .Values.operatorConfiguration.dynamicConfig.enable }}
+{{- if and .Values.operatorConfiguration.dynamicConfig.enable (eq
.Values.operatorConfiguration.dynamicConfig.source "configMap") }}
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
diff --git
a/build-tools/helm/spark-kubernetes-operator/templates/spark-operator.yaml
b/build-tools/helm/spark-kubernetes-operator/templates/spark-operator.yaml
index e8aeb3b..1cf189d 100644
--- a/build-tools/helm/spark-kubernetes-operator/templates/spark-operator.yaml
+++ b/build-tools/helm/spark-kubernetes-operator/templates/spark-operator.yaml
@@ -148,6 +148,11 @@ spec:
mountPath: /opt/spark-operator/conf
- name: logs-volume
mountPath: /opt/spark-operator/logs
+ {{- if and .Values.operatorConfiguration.dynamicConfig.enable (eq
.Values.operatorConfiguration.dynamicConfig.source "file") }}
+ - name: spark-operator-dynamic-config-volume
+ mountPath: /opt/spark-operator/dynamic-conf
+ readOnly: true
+ {{- end }}
{{- with
.Values.operatorDeployment.operatorPod.operatorContainer.volumeMounts }}
{{- toYaml . | nindent 12 }}
{{- end }}
@@ -170,6 +175,11 @@ spec:
name: spark-kubernetes-operator-configuration
- name: logs-volume
emptyDir: { }
+ {{- if and .Values.operatorConfiguration.dynamicConfig.enable (eq
.Values.operatorConfiguration.dynamicConfig.source "file") }}
+ - name: spark-operator-dynamic-config-volume
+ configMap:
+ name: spark-kubernetes-operator-dynamic-configuration
+ {{- end }}
{{- with .Values.operatorDeployment.operatorPod.volumes }}
{{- toYaml . | nindent 8 }}
{{- end }}
@@ -215,8 +225,16 @@ metadata:
{{- include "spark-operator.dynamicConfigLabels" . | nindent 4 }}
annotations:
{{- toYaml .Values.operatorConfiguration.dynamicConfig.annotations |
nindent 4 }}
+{{- if eq .Values.operatorConfiguration.dynamicConfig.source "file" }}
+data:
+ spark-operator-dynamic.properties: |
+ {{- range $key, $value := .Values.operatorConfiguration.dynamicConfig.data
}}
+ {{ $key }}={{ $value }}
+ {{- end }}
+{{- else }}
{{- with .Values.operatorConfiguration.dynamicConfig.data }}
data:
{{- toYaml . | nindent 2 }}
{{- end }}
{{- end }}
+{{- end }}
diff --git a/build-tools/helm/spark-kubernetes-operator/values.yaml
b/build-tools/helm/spark-kubernetes-operator/values.yaml
index 17d7537..68fddac 100644
--- a/build-tools/helm/spark-kubernetes-operator/values.yaml
+++ b/build-tools/helm/spark-kubernetes-operator/values.yaml
@@ -206,6 +206,12 @@ operatorConfiguration:
dynamicConfig:
# Enable this for hot properties loading.
enable: false
+ # Source of the dynamic config overrides when enabled. Supported values:
+ # configMap - (default) watch a ConfigMap via a Kubernetes informer.
Requires the operator
+ # to have RBAC to read ConfigMaps (created by this chart).
+ # file - periodically reload a properties file mounted from a
ConfigMap. Requires no
+ # extra RBAC; the ConfigMap is mounted into the operator pod
as a volume.
+ source: configMap
# Enable this to create a config map for hot property loading
create: false
annotations:
diff --git a/docs/config_properties.md b/docs/config_properties.md
index 7929487..f82f2b9 100644
--- a/docs/config_properties.md
+++ b/docs/config_properties.md
@@ -10,9 +10,12 @@
| spark.kubernetes.operator.api.secondaryResourceCreateMaxAttempts | Long | 3
| false | Maximal number of retry attempts of requesting secondary resource for
Spark application. This would be performed on top of k8s client
spark.kubernetes.operator.retry.maxAttempts to overcome potential conflicting
reconcile on the same SparkApplication, as well as API server errors
(408/500/502/503/504) and network-level timeouts. Exponential backoff with
jitter is applied before retrying on 409 (Confl [...]
| spark.kubernetes.operator.api.secondaryResourceCreateMaxBackoffMillis |
Long | 40000 | false | Maximum backoff (in milliseconds) between retries when
creating secondary resources for Spark application. |
| spark.kubernetes.operator.api.statusPatchMaxAttempts | Long | 3 | false |
Maximal number of retry attempts of requests to k8s server for resource status
update. This would be performed on top of k8s client
spark.kubernetes.operator.retry.maxAttempts to overcome potential conflicting
update on the same SparkApplication. This should be positive number. |
- | spark.kubernetes.operator.dynamicConfig.enabled | Boolean | false | false |
When enabled, operator would use config map as source of truth for config
property override. The config map need to be created in
spark.kubernetes.operator.namespace, and labeled with operator name. |
+ | spark.kubernetes.operator.dynamicConfig.enabled | Boolean | false | false |
When enabled, operator would load config property overrides dynamically at
runtime. The source of the overrides is controlled by
spark.kubernetes.operator.dynamicConfig.source. |
+ | spark.kubernetes.operator.dynamicConfig.filePath | String |
/opt/spark-operator/dynamic-conf/spark-operator-dynamic.properties | false |
Path of the properties file holding dynamic configuration overrides. Used by
the 'file' source, typically populated by mounting a ConfigMap as a volume. |
| spark.kubernetes.operator.dynamicConfig.reconcilerParallelism | Integer | 1
| false | Parallelism for dynamic config reconciler. Unbounded pool would be
used if set to non-positive number. |
- | spark.kubernetes.operator.dynamicConfig.selector | String |
app.kubernetes.io/name=spark-kubernetes-operator,app.kubernetes.io/component=operator-dynamic-config-overrides
| false | The selector str applied to dynamic config map. |
+ | spark.kubernetes.operator.dynamicConfig.reloadIntervalSeconds | Long | 60 |
false | Interval (in seconds) at which the dynamic config file is re-read. Used
by the 'file' source. |
+ | spark.kubernetes.operator.dynamicConfig.selector | String |
app.kubernetes.io/name=spark-kubernetes-operator,app.kubernetes.io/component=operator-dynamic-config-overrides
| false | The selector str applied to dynamic config map. Used by the
'configMap' source. |
+ | spark.kubernetes.operator.dynamicConfig.source | String | configMap | false
| Source of dynamic config overrides when
spark.kubernetes.operator.dynamicConfig.enabled is true. Supported values:
'configMap' (default) watches a ConfigMap via a Kubernetes informer and
requires RBAC to read ConfigMaps; 'file' periodically reloads a properties file
mounted from a ConfigMap and requires no extra RBAC. |
| spark.kubernetes.operator.health.probePort | Integer | 19091 | false | The
port used for health/readiness check probe status. |
| spark.kubernetes.operator.health.sentinelExecutorPoolSize | Integer | 3 |
false | Size of executor service in Sentinel Managers to check the health of
sentinel resources. |
| spark.kubernetes.operator.health.sentinelResourceReconciliationDelaySeconds
| Integer | 60 | true | Allowed max time(seconds) between spec update and
reconciliation for sentinel resources. |
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
index 2a2c403..ae88c90 100644
--- a/gradle/libs.versions.toml
+++ b/gradle/libs.versions.toml
@@ -29,6 +29,7 @@ slf4j = "2.0.18"
junit = "6.1.0"
jacoco = "0.8.14"
mockito = "5.23.0"
+awaitility = "4.3.0"
# Build Analysis
checkstyle = "13.4.2"
@@ -64,6 +65,7 @@ metrics-jvm = { group = "io.dropwizard.metrics", name =
"metrics-jvm", version.r
spark-core = { group = "org.apache.spark", name = "spark-core_2.13",
version.ref = "spark"}
spark-kubernetes = { group = "org.apache.spark", name =
"spark-kubernetes_2.13", version.ref = "spark"}
mockito-core = { group = "org.mockito", name = "mockito-core", version.ref =
"mockito"}
+awaitility = { group = "org.awaitility", name = "awaitility", version.ref =
"awaitility"}
junit-bom = { group = "org.junit", name = "junit-bom", version.ref = "junit"}
junit-jupiter = { group = "org.junit.jupiter", name = "junit-jupiter",
version.ref = "junit"}
junit-platform-launcher = { group = "org.junit.platform", name =
"junit-platform-launcher"}
diff --git a/spark-operator/build.gradle b/spark-operator/build.gradle
index 3c0efce..4c0c33a 100644
--- a/spark-operator/build.gradle
+++ b/spark-operator/build.gradle
@@ -71,6 +71,7 @@ dependencies {
testCompileOnly(libs.spotbugs.annotations)
testImplementation(libs.mockito.core)
testImplementation(libs.kube.api.test.client.inject)
+ testImplementation(libs.awaitility)
}
jar.dependsOn shadowJar
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/SparkOperator.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/SparkOperator.java
index 2f8f98f..3dcb9ad 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/SparkOperator.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/SparkOperator.java
@@ -23,6 +23,7 @@ import static
org.apache.spark.k8s.operator.utils.Utils.getAppStatusListener;
import static
org.apache.spark.k8s.operator.utils.Utils.getClusterStatusListener;
import static org.apache.spark.k8s.operator.utils.Utils.getWatchedNamespaces;
+import java.nio.file.Paths;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
@@ -43,6 +44,7 @@ import
io.javaoperatorsdk.operator.api.config.ControllerConfigurationOverrider;
import lombok.extern.slf4j.Slf4j;
import org.apache.spark.k8s.operator.client.KubernetesClientFactory;
+import org.apache.spark.k8s.operator.config.DynamicConfigMonitor;
import org.apache.spark.k8s.operator.config.SparkOperatorConf;
import org.apache.spark.k8s.operator.config.SparkOperatorConfManager;
import org.apache.spark.k8s.operator.config.SparkOperatorConfigMapReconciler;
@@ -59,6 +61,7 @@ import
org.apache.spark.k8s.operator.reconciler.SparkClusterReconciler;
import org.apache.spark.k8s.operator.utils.SparkAppStatusRecorder;
import org.apache.spark.k8s.operator.utils.SparkClusterStatusRecorder;
import org.apache.spark.k8s.operator.utils.StringUtils;
+import org.apache.spark.k8s.operator.utils.Utils;
/**
* Entry point for Spark Operator. Bootstrap the operator app by starting
watch and reconciler for
@@ -82,6 +85,7 @@ public class SparkOperator {
private final MetricsService metricsService;
private final ExecutorService metricsResourcesSingleThreadPool;
private final ScheduledExecutorService periodicGcScheduler;
+ private final DynamicConfigMonitor dynamicConfigMonitor;
/** Constructs a new SparkOperator, initializing all its components. */
public SparkOperator() {
@@ -106,14 +110,13 @@ public class SparkOperator {
log.info("{} = {}", entry.getKey(), entry.getValue());
}
}
- if (SparkOperatorConf.DYNAMIC_CONFIG_ENABLED.getValue()) {
- this.registeredOperators.add(registerSparkOperatorConfMonitor());
- }
+ this.dynamicConfigMonitor = registerDynamicConfig();
this.metricsResourcesSingleThreadPool =
Executors.newSingleThreadExecutor();
this.probeService =
new ProbeService(
registeredOperators,
Arrays.asList(sparkApplicationSentinelManager,
sparkClusterSentinelManager),
+ dynamicConfigMonitor,
null);
this.metricsService = new MetricsService(metricsSystem,
metricsResourcesSingleThreadPool);
long periodicGcIntervalSeconds =
SparkOperatorConf.PERIODIC_GC_INTERVAL_SECONDS.getValue();
@@ -175,7 +178,44 @@ public class SparkOperator {
}
/**
- * Registers a monitor for dynamic configuration changes via ConfigMaps.
+ * Wires up dynamic configuration loading when enabled. Two sources are
supported, selected by
+ * {@code spark.kubernetes.operator.dynamicConfig.source}:
+ *
+ * <ul>
+ * <li>{@code configMap} (default) - registers a {@link
SparkOperatorConfigMapReconciler} on a
+ * dedicated {@link Operator} that watches a ConfigMap via a
Kubernetes informer. The
+ * operator is appended to {@link #registeredOperators}.
+ * <li>{@code file} - returns a {@link DynamicConfigMonitor} that
periodically reloads a
+ * properties file mounted from a ConfigMap, requiring no extra RBAC.
+ * </ul>
+ *
+ * @return a {@link DynamicConfigMonitor} for the {@code file} source, or
{@code null} when
+ * dynamic config is disabled or the {@code configMap} source is used.
+ */
+ protected DynamicConfigMonitor registerDynamicConfig() {
+ if
(Boolean.FALSE.equals(SparkOperatorConf.DYNAMIC_CONFIG_ENABLED.getValue())) {
+ return null;
+ }
+ String source = SparkOperatorConf.DYNAMIC_CONFIG_SOURCE.getValue();
+ if ("file".equalsIgnoreCase(source)) {
+ log.info("Starting dynamic config from mounted file source.");
+ return new DynamicConfigMonitor(
+ Paths.get(SparkOperatorConf.DYNAMIC_CONFIG_FILE_PATH.getValue()),
+
Duration.ofSeconds(SparkOperatorConf.getDynamicConfigReloadIntervalSeconds()),
+ Utils::getWatchedNamespaces,
+ this::updateWatchingNamespaces);
+ }
+ if (!"configMap".equalsIgnoreCase(source)) {
+ log.warn(
+ "Unknown dynamic config source '{}', falling back to 'configMap'
informer source.",
+ source);
+ }
+ registeredOperators.add(registerSparkOperatorConfMonitor());
+ return null;
+ }
+
+ /**
+ * Registers a monitor for dynamic configuration changes via a ConfigMap
informer.
*
* @return The Operator instance for the config monitor.
*/
@@ -264,7 +304,7 @@ public class SparkOperator {
}
/**
- * Overrides the configuration for the dynamic config monitor.
+ * Overrides the configuration for the dynamic config monitor (configMap
informer source).
*
* @param overrider The ConfigurationServiceOverrider to apply changes to.
*/
@@ -331,6 +371,10 @@ public class SparkOperator {
for (Operator operator : sparkOperator.registeredOperators) {
operator.start();
}
+
+ if (sparkOperator.dynamicConfigMonitor != null) {
+ sparkOperator.dynamicConfigMonitor.start();
+ }
sparkOperator.probeService.start();
// Single thread queue to ensure MetricsService starts after the
MetricsSystem
sparkOperator.metricsResourcesSingleThreadPool.submit(sparkOperator.metricsSystem::start);
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/DynamicConfigMonitor.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/DynamicConfigMonitor.java
new file mode 100644
index 0000000..b3f8118
--- /dev/null
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/DynamicConfigMonitor.java
@@ -0,0 +1,162 @@
+/*
+ * 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.IOException;
+import java.io.InputStream;
+import java.io.UncheckedIOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Periodically reloads dynamic configuration overrides from a properties file
on disk. The file is
+ * expected to be populated by mounting a ConfigMap as a volume into the
operator pod, so changes
+ * applied to the ConfigMap propagate to disk without requiring a Kubernetes
informer. When the
+ * file contents change, {@link SparkOperatorConfManager} is refreshed and the
caller-supplied
+ * namespace updater is invoked.
+ */
+@Slf4j
+public class DynamicConfigMonitor {
+
+ private final Path configFile;
+ private final Duration reloadInterval;
+ private final Supplier<Set<String>> watchedNamespacesSupplier;
+ private final Consumer<Set<String>> namespaceUpdater;
+ private final ScheduledExecutorService scheduler;
+ private final boolean ownsScheduler;
+
+ private final AtomicReference<Map<String, String>> lastLoaded = new
AtomicReference<>(Map.of());
+ private final AtomicReference<ScheduledFuture<?>> scheduledTask = new
AtomicReference<>();
+
+ public DynamicConfigMonitor(
+ Path configFile,
+ Duration reloadInterval,
+ Supplier<Set<String>> watchedNamespacesSupplier,
+ Consumer<Set<String>> namespaceUpdater) {
+ this(configFile, reloadInterval, watchedNamespacesSupplier,
namespaceUpdater, null);
+ }
+
+ DynamicConfigMonitor(
+ Path configFile,
+ Duration reloadInterval,
+ Supplier<Set<String>> watchedNamespacesSupplier,
+ Consumer<Set<String>> namespaceUpdater,
+ ScheduledExecutorService scheduler) {
+ this.configFile = configFile;
+ this.reloadInterval = reloadInterval;
+ this.watchedNamespacesSupplier = watchedNamespacesSupplier;
+ this.namespaceUpdater = namespaceUpdater;
+ if (scheduler == null) {
+ this.scheduler =
+ Executors.newSingleThreadScheduledExecutor(
+ r -> {
+ Thread t = new Thread(r, "spark-operator-dynamic-config");
+ t.setDaemon(true);
+ return t;
+ });
+ this.ownsScheduler = true;
+ } else {
+ this.scheduler = scheduler;
+ this.ownsScheduler = false;
+ }
+ }
+
+ /**
+ * Schedules periodic reloads at the configured interval. The first reload
runs through the
+ * scheduler with no initial delay (rather than synchronously here), so a
failing initial load
+ * never blocks operator startup.
+ */
+ public void start() {
+ log.info(
+ "Starting dynamic config monitor on {} with reload interval {}",
+ configFile,
+ reloadInterval);
+ long millis = reloadInterval.toMillis();
+ scheduledTask.set(
+ scheduler.scheduleAtFixedRate(this::reloadSafely, 0, millis,
TimeUnit.MILLISECONDS));
+ }
+
+ /** Stops the scheduler if it was created internally. */
+ public void stop() {
+ log.info("Stopping dynamic config monitor");
+ if (ownsScheduler) {
+ scheduler.shutdownNow();
+ }
+ }
+
+ /**
+ * Returns true once {@link #start()} has scheduled the periodic reload and
the underlying
+ * scheduler is still running. Because {@link #reloadSafely()} swallows
reload failures, the
+ * scheduled task only becomes done when it is cancelled (e.g. via {@link
#stop()}).
+ */
+ public boolean isRunning() {
+ ScheduledFuture<?> task = scheduledTask.get();
+ return task != null && !task.isDone() && !scheduler.isShutdown();
+ }
+
+ private void reloadSafely() {
+ try {
+ reload();
+ } catch (RuntimeException e) {
+ log.error("Failed to reload dynamic config from {}", configFile, e);
+ }
+ }
+
+ private void reload() {
+ Map<String, String> current = readProperties();
+ if (current.equals(lastLoaded.get())) {
+ return;
+ }
+ log.info(
+ "Detected dynamic config change in {}, applying {} overrides",
configFile, current.size());
+ SparkOperatorConfManager.INSTANCE.refresh(current);
+ lastLoaded.set(current);
+ namespaceUpdater.accept(watchedNamespacesSupplier.get());
+ }
+
+ private Map<String, String> readProperties() {
+ if (!Files.isRegularFile(configFile)) {
+ return Map.of();
+ }
+ Properties properties = new Properties();
+ try (InputStream in = Files.newInputStream(configFile)) {
+ properties.load(in);
+ } catch (IOException e) {
+ throw new UncheckedIOException("Failed to read dynamic config file " +
configFile, e);
+ }
+ Map<String, String> result = new HashMap<>(properties.size());
+ properties.forEach((k, v) -> result.put(String.valueOf(k),
String.valueOf(v)));
+ return Map.copyOf(result);
+ }
+}
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java
index 68bf5e3..8b95d0f 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/config/SparkOperatorConf.java
@@ -235,32 +235,71 @@ public final class SparkOperatorConf {
.build();
/**
- * When enabled, operator would use config map as source of truth for config
property override.
- * The config map need to be created in spark.kubernetes.operator.namespace,
and labeled with
- * operator name.
+ * When enabled, operator would load config property overrides dynamically
at runtime. The source
+ * of the overrides is controlled by
spark.kubernetes.operator.dynamicConfig.source.
*/
public static final ConfigOption<Boolean> DYNAMIC_CONFIG_ENABLED =
ConfigOption.<Boolean>builder()
.key("spark.kubernetes.operator.dynamicConfig.enabled")
.enableDynamicOverride(false)
.description(
- "When enabled, operator would use config map as source of truth
for config "
- + "property override. The config map need to be created in "
- + "spark.kubernetes.operator.namespace, and labeled with
operator name.")
+ "When enabled, operator would load config property overrides
dynamically at "
+ + "runtime. The source of the overrides is controlled by "
+ + "spark.kubernetes.operator.dynamicConfig.source.")
.typeParameterClass(Boolean.class)
.defaultValue(false)
.build();
- /** The selector str applied to dynamic config map. */
+ /** Source of dynamic config overrides: {@code configMap} (informer) or
{@code file} (mount). */
+ public static final ConfigOption<String> DYNAMIC_CONFIG_SOURCE =
+ ConfigOption.<String>builder()
+ .key("spark.kubernetes.operator.dynamicConfig.source")
+ .enableDynamicOverride(false)
+ .description(
+ "Source of dynamic config overrides when "
+ + "spark.kubernetes.operator.dynamicConfig.enabled is true.
Supported values: "
+ + "'configMap' (default) watches a ConfigMap via a
Kubernetes informer and "
+ + "requires RBAC to read ConfigMaps; 'file' periodically
reloads a properties "
+ + "file mounted from a ConfigMap and requires no extra
RBAC.")
+ .typeParameterClass(String.class)
+ .defaultValue("configMap")
+ .build();
+
+ /** The selector str applied to dynamic config map (used by the {@code
configMap} source). */
public static final ConfigOption<String> DYNAMIC_CONFIG_SELECTOR =
ConfigOption.<String>builder()
.key("spark.kubernetes.operator.dynamicConfig.selector")
.enableDynamicOverride(false)
- .description("The selector str applied to dynamic config map.")
+ .description(
+ "The selector str applied to dynamic config map. Used by the
'configMap' source.")
.typeParameterClass(String.class)
.defaultValue(Utils.labelsAsStr(Utils.defaultOperatorConfigLabels()))
.build();
+ /** Path of the properties file that holds dynamic config overrides. */
+ public static final ConfigOption<String> DYNAMIC_CONFIG_FILE_PATH =
+ ConfigOption.<String>builder()
+ .key("spark.kubernetes.operator.dynamicConfig.filePath")
+ .enableDynamicOverride(false)
+ .description(
+ "Path of the properties file holding dynamic configuration
overrides. Used by the "
+ + "'file' source, typically populated by mounting a
ConfigMap as a volume.")
+ .typeParameterClass(String.class)
+
.defaultValue("/opt/spark-operator/dynamic-conf/spark-operator-dynamic.properties")
+ .build();
+
+ /** Interval at which the dynamic config file is re-read (used by the {@code
file} source). */
+ public static final ConfigOption<Long>
DYNAMIC_CONFIG_RELOAD_INTERVAL_SECONDS =
+ ConfigOption.<Long>builder()
+ .key("spark.kubernetes.operator.dynamicConfig.reloadIntervalSeconds")
+ .enableDynamicOverride(false)
+ .description(
+ "Interval (in seconds) at which the dynamic config file is
re-read. Used by the "
+ + "'file' source.")
+ .typeParameterClass(Long.class)
+ .defaultValue(60L)
+ .build();
+
/**
* Parallelism for dynamic config reconciler. Unbounded pool would be used
if set to non-positive
* number.
@@ -718,6 +757,17 @@ public final class SparkOperatorConf {
ensureNonNegativeIntFor(RECONCILER_RATE_LIMITER_MAX_LOOP_FOR_PERIOD));
}
+ /**
+ * Returns the dynamic config file reload interval (in seconds) for the
{@code file} source,
+ * ensuring the configured value is positive. A non-positive value would be
rejected by the
+ * scheduler; the option's default is used instead in that case.
+ *
+ * @return The validated, positive reload interval in seconds.
+ */
+ public static long getDynamicConfigReloadIntervalSeconds() {
+ return ensurePositiveLongFor(DYNAMIC_CONFIG_RELOAD_INTERVAL_SECONDS);
+ }
+
/**
* Ensures that the integer value of a ConfigOption is non-negative.
*
@@ -738,6 +788,18 @@ public final class SparkOperatorConf {
return ensureValid(configOption.getValue(), configOption.getDescription(),
1, 1);
}
+ /**
+ * Ensures that the long value of a ConfigOption is positive, falling back
to the option's
+ * default value when the configured value is non-positive.
+ *
+ * @param configOption The ConfigOption to check.
+ * @return The positive long value, or the option's default if the
configured value is invalid.
+ */
+ private static long ensurePositiveLongFor(ConfigOption<Long> configOption) {
+ return ensureValid(
+ configOption.getValue(), configOption.getDescription(), 1L,
configOption.getDefaultValue());
+ }
+
/**
* Ensures that a given integer value is within a valid range.
*
@@ -763,4 +825,31 @@ public final class SparkOperatorConf {
}
return value;
}
+
+ /**
+ * Ensures that a given long value is within a valid range.
+ *
+ * @param value The value to validate.
+ * @param description A description of the value for logging purposes.
+ * @param minValue The minimum allowed value (inclusive).
+ * @param defaultValue The default value to use if the provided value is
invalid.
+ * @return The validated value, or the default value if invalid.
+ */
+ private static long ensureValid(
+ long value, String description, long minValue, long defaultValue) {
+ if (value < minValue) {
+ if (defaultValue < minValue) {
+ throw new IllegalArgumentException(
+ "Default value for " + description + " must be greater than " +
minValue);
+ }
+ log.warn(
+ "Requested {} should be greater than {}. Requested: {}, using {}
(default) instead",
+ description,
+ minValue,
+ value,
+ defaultValue);
+ return defaultValue;
+ }
+ return value;
+ }
}
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/HealthProbe.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/HealthProbe.java
index 64a80c0..dbe8782 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/HealthProbe.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/HealthProbe.java
@@ -40,6 +40,7 @@ import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
+import org.apache.spark.k8s.operator.config.DynamicConfigMonitor;
import org.apache.spark.k8s.operator.metrics.healthcheck.SentinelManager;
/** Health probe for the operator. */
@@ -49,6 +50,7 @@ import
org.apache.spark.k8s.operator.metrics.healthcheck.SentinelManager;
public class HealthProbe implements HttpHandler {
private final List<Operator> operators;
private final List<SentinelManager<?>> sentinelManagers;
+ private final DynamicConfigMonitor dynamicConfigMonitor;
/**
* Checks the overall health of the operator, including all registered
operators and sentinel
@@ -72,6 +74,10 @@ public class HealthProbe implements HttpHandler {
}
}
+ if (dynamicConfigMonitor != null && !dynamicConfigMonitor.isRunning()) {
+ log.error("Dynamic config monitor is not running.");
+ return false;
+ }
return true;
}
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/ProbeService.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/ProbeService.java
index c21f4a2..c7e0837 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/ProbeService.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/ProbeService.java
@@ -33,6 +33,7 @@ import io.javaoperatorsdk.operator.Operator;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
+import org.apache.spark.k8s.operator.config.DynamicConfigMonitor;
import org.apache.spark.k8s.operator.metrics.healthcheck.SentinelManager;
import org.apache.spark.k8s.operator.utils.HttpMethodFilter;
@@ -49,17 +50,26 @@ public class ProbeService {
*
* @param operators A list of Operator instances to monitor.
* @param sentinelManagers A list of SentinelManager instances to monitor.
+ * @param dynamicConfigMonitor optional dynamic config monitor whose running
state is included in
+ * the health and readiness checks. May be {@code null} when dynamic
config is disabled or the
+ * configMap informer source is used.
* @param executor The Executor to use for the HTTP server.
*/
public ProbeService(
- List<Operator> operators, List<SentinelManager<?>> sentinelManagers,
Executor executor) {
+ List<Operator> operators,
+ List<SentinelManager<?>> sentinelManagers,
+ DynamicConfigMonitor dynamicConfigMonitor,
+ Executor executor) {
try {
this.server = HttpServer.create(new
InetSocketAddress(OPERATOR_PROBE_PORT.getValue()), 0);
} catch (IOException e) {
throw new IllegalStateException("Failed to create Probe Service Server",
e);
}
- server.createContext(READYZ, new
ReadinessProbe(operators)).getFilters().add(FILTER);
- server.createContext(HEALTHZ, new HealthProbe(operators, sentinelManagers))
+ server.createContext(READYZ, new ReadinessProbe(operators,
dynamicConfigMonitor))
+ .getFilters().add(FILTER);
+ server
+ .createContext(
+ HEALTHZ, new HealthProbe(operators, sentinelManagers,
dynamicConfigMonitor))
.getFilters().add(FILTER);
server.createContext(
"/",
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/ReadinessProbe.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/ReadinessProbe.java
index a4778ca..1948f3f 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/ReadinessProbe.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/probe/ReadinessProbe.java
@@ -32,18 +32,25 @@ import com.sun.net.httpserver.HttpHandler;
import io.javaoperatorsdk.operator.Operator;
import lombok.extern.slf4j.Slf4j;
+import org.apache.spark.k8s.operator.config.DynamicConfigMonitor;
+
/** Readiness probe for the operator. */
@Slf4j
public class ReadinessProbe implements HttpHandler {
private final List<Operator> operators;
+ private final DynamicConfigMonitor dynamicConfigMonitor;
/**
* Constructs a new ReadinessProbe.
*
* @param operators A list of Operator instances to check for readiness.
+ * @param dynamicConfigMonitor optional dynamic config monitor whose running
state participates in
+ * the readiness check. May be {@code null} when dynamic config is
disabled or the configMap
+ * informer source is used.
*/
- public ReadinessProbe(List<Operator> operators) {
+ public ReadinessProbe(List<Operator> operators, DynamicConfigMonitor
dynamicConfigMonitor) {
this.operators = operators;
+ this.dynamicConfigMonitor = dynamicConfigMonitor;
}
/**
@@ -57,11 +64,18 @@ public class ReadinessProbe implements HttpHandler {
Optional<Boolean> operatorsAreReady = areOperatorsStarted(operators);
if (operatorsAreReady.isEmpty() || !operatorsAreReady.get()) {
sendMessage(httpExchange, HTTP_BAD_REQUEST, "spark operators are not
ready yet");
+ return;
+ }
+
+ if (dynamicConfigMonitor != null && !dynamicConfigMonitor.isRunning()) {
+ sendMessage(httpExchange, HTTP_BAD_REQUEST, "dynamic config monitor is
not running yet");
+ return;
}
if (!passRbacCheck()) {
sendMessage(
httpExchange, HTTP_FORBIDDEN, "required rbac test failed, operators
are not ready");
+ return;
}
sendMessage(httpExchange, HTTP_OK, "started");
diff --git
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/SparkOperatorTest.java
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/SparkOperatorTest.java
index b8b5799..a110fdd 100644
---
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/SparkOperatorTest.java
+++
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/SparkOperatorTest.java
@@ -41,8 +41,8 @@ import org.mockito.MockedConstruction;
import org.mockito.MockedStatic;
import org.apache.spark.k8s.operator.client.KubernetesClientFactory;
+import org.apache.spark.k8s.operator.config.DynamicConfigMonitor;
import org.apache.spark.k8s.operator.config.SparkOperatorConf;
-import org.apache.spark.k8s.operator.config.SparkOperatorConfigMapReconciler;
import org.apache.spark.k8s.operator.metrics.MetricsService;
import org.apache.spark.k8s.operator.metrics.MetricsSystem;
import org.apache.spark.k8s.operator.metrics.MetricsSystemFactory;
@@ -55,10 +55,11 @@ import org.apache.spark.k8s.operator.utils.Utils;
class SparkOperatorTest {
@Test
- void testOperatorConstructionWithDynamicConfigEnabled() {
+ void testOperatorConstructionWithDynamicConfigFileSource() {
MetricsSystem mockMetricsSystem = mock(MetricsSystem.class);
KubernetesClient mockClient = mock(KubernetesClient.class);
boolean dynamicConfigEnabled =
SparkOperatorConf.DYNAMIC_CONFIG_ENABLED.getValue();
+ String dynamicConfigSource =
SparkOperatorConf.DYNAMIC_CONFIG_SOURCE.getValue();
try (MockedStatic<MetricsSystemFactory> mockMetricsSystemFactory =
mockStatic(MetricsSystemFactory.class);
@@ -68,15 +69,16 @@ class SparkOperatorTest {
MockedConstruction<Operator> operatorConstruction =
mockConstruction(Operator.class);
MockedConstruction<SparkAppReconciler> sparkAppReconcilerConstruction =
mockConstruction(SparkAppReconciler.class);
- MockedConstruction<SparkOperatorConfigMapReconciler>
configReconcilerConstruction =
- mockConstruction(SparkOperatorConfigMapReconciler.class);
MockedConstruction<ProbeService> probeServiceConstruction =
mockConstruction(ProbeService.class);
MockedConstruction<MetricsService> metricsServiceConstruction =
mockConstruction(MetricsService.class);
+ MockedConstruction<DynamicConfigMonitor>
dynamicConfigMonitorConstruction =
+ mockConstruction(DynamicConfigMonitor.class);
MockedConstruction<KubernetesMetricsInterceptor>
interceptorMockedConstruction =
mockConstruction(KubernetesMetricsInterceptor.class)) {
setConfigKey(SparkOperatorConf.DYNAMIC_CONFIG_ENABLED, true);
+ setConfigKey(SparkOperatorConf.DYNAMIC_CONFIG_SOURCE, "file");
mockMetricsSystemFactory
.when(MetricsSystemFactory::createMetricsSystem)
.thenReturn(mockMetricsSystem);
@@ -87,11 +89,12 @@ class SparkOperatorTest {
SparkOperator sparkOperator = new SparkOperator();
Assertions.assertEquals(1,
sparkOperator.registeredSparkControllers.size());
- Assertions.assertEquals(2, operatorConstruction.constructed().size());
+ // Only the main operator is registered; the file source uses a
DynamicConfigMonitor.
+ Assertions.assertEquals(1, operatorConstruction.constructed().size());
Assertions.assertEquals(1,
sparkAppReconcilerConstruction.constructed().size());
- Assertions.assertEquals(1,
configReconcilerConstruction.constructed().size());
Assertions.assertEquals(1,
probeServiceConstruction.constructed().size());
Assertions.assertEquals(1,
metricsServiceConstruction.constructed().size());
+ Assertions.assertEquals(1,
dynamicConfigMonitorConstruction.constructed().size());
Assertions.assertEquals(1,
interceptorMockedConstruction.constructed().size());
verify(mockMetricsSystem).registerSource(interceptorMockedConstruction.constructed().get(0));
@@ -100,6 +103,56 @@ class SparkOperatorTest {
verify(sparkAppOperator).register(eq(sparkAppReconciler),
any(Consumer.class));
} finally {
setConfigKey(SparkOperatorConf.DYNAMIC_CONFIG_ENABLED,
dynamicConfigEnabled);
+ setConfigKey(SparkOperatorConf.DYNAMIC_CONFIG_SOURCE,
dynamicConfigSource);
+ }
+ }
+
+ @Test
+ void testOperatorConstructionWithDynamicConfigConfigMapSource() {
+ MetricsSystem mockMetricsSystem = mock(MetricsSystem.class);
+ KubernetesClient mockClient = mock(KubernetesClient.class);
+ boolean dynamicConfigEnabled =
SparkOperatorConf.DYNAMIC_CONFIG_ENABLED.getValue();
+ String dynamicConfigSource =
SparkOperatorConf.DYNAMIC_CONFIG_SOURCE.getValue();
+
+ try (MockedStatic<MetricsSystemFactory> mockMetricsSystemFactory =
+ mockStatic(MetricsSystemFactory.class);
+ MockedStatic<KubernetesClientFactory> mockKubernetesClientFactory =
+ mockStatic(KubernetesClientFactory.class);
+ MockedStatic<Utils> mockUtils = mockStatic(Utils.class);
+ MockedConstruction<Operator> operatorConstruction =
mockConstruction(Operator.class);
+ MockedConstruction<SparkAppReconciler> sparkAppReconcilerConstruction =
+ mockConstruction(SparkAppReconciler.class);
+ MockedConstruction<ProbeService> probeServiceConstruction =
+ mockConstruction(ProbeService.class);
+ MockedConstruction<MetricsService> metricsServiceConstruction =
+ mockConstruction(MetricsService.class);
+ MockedConstruction<DynamicConfigMonitor>
dynamicConfigMonitorConstruction =
+ mockConstruction(DynamicConfigMonitor.class);
+ MockedConstruction<KubernetesMetricsInterceptor>
interceptorMockedConstruction =
+ mockConstruction(KubernetesMetricsInterceptor.class)) {
+ setConfigKey(SparkOperatorConf.DYNAMIC_CONFIG_ENABLED, true);
+ setConfigKey(SparkOperatorConf.DYNAMIC_CONFIG_SOURCE, "configMap");
+ mockMetricsSystemFactory
+ .when(MetricsSystemFactory::createMetricsSystem)
+ .thenReturn(mockMetricsSystem);
+ mockKubernetesClientFactory
+ .when(() -> KubernetesClientFactory.buildKubernetesClient(any()))
+ .thenReturn(mockClient);
+
mockUtils.when(Utils::getWatchedNamespaces).thenReturn(Set.of("namespace-1"));
+
+ SparkOperator sparkOperator = new SparkOperator();
+ Assertions.assertEquals(1,
sparkOperator.registeredSparkControllers.size());
+ // The configMap informer source registers a second Operator and no
DynamicConfigMonitor.
+ Assertions.assertEquals(2, operatorConstruction.constructed().size());
+ Assertions.assertEquals(1,
sparkAppReconcilerConstruction.constructed().size());
+ Assertions.assertEquals(1,
probeServiceConstruction.constructed().size());
+ Assertions.assertEquals(1,
metricsServiceConstruction.constructed().size());
+ Assertions.assertEquals(0,
dynamicConfigMonitorConstruction.constructed().size());
+ Assertions.assertEquals(1,
interceptorMockedConstruction.constructed().size());
+
verify(mockMetricsSystem).registerSource(interceptorMockedConstruction.constructed().get(0));
+ } finally {
+ setConfigKey(SparkOperatorConf.DYNAMIC_CONFIG_ENABLED,
dynamicConfigEnabled);
+ setConfigKey(SparkOperatorConf.DYNAMIC_CONFIG_SOURCE,
dynamicConfigSource);
}
}
@@ -119,6 +172,8 @@ class SparkOperatorTest {
mockConstruction(ProbeService.class);
MockedConstruction<MetricsService> metricsServiceConstruction =
mockConstruction(MetricsService.class);
+ MockedConstruction<DynamicConfigMonitor>
dynamicConfigMonitorConstruction =
+ mockConstruction(DynamicConfigMonitor.class);
MockedConstruction<KubernetesMetricsInterceptor>
interceptorMockedConstruction =
mockConstruction(KubernetesMetricsInterceptor.class)) {
setConfigKey(SparkOperatorConf.DYNAMIC_CONFIG_ENABLED, false);
@@ -134,6 +189,7 @@ class SparkOperatorTest {
Assertions.assertEquals(1,
sparkAppReconcilerConstruction.constructed().size());
Assertions.assertEquals(1,
probeServiceConstruction.constructed().size());
Assertions.assertEquals(1,
metricsServiceConstruction.constructed().size());
+ Assertions.assertEquals(0,
dynamicConfigMonitorConstruction.constructed().size());
Assertions.assertEquals(1,
interceptorMockedConstruction.constructed().size());
verify(mockMetricsSystem).registerSource(interceptorMockedConstruction.constructed().get(0));
} finally {
@@ -149,6 +205,7 @@ class SparkOperatorTest {
var registeredController = mock(RegisteredController.class);
when(registeredController.allowsNamespaceChanges()).thenReturn(true);
boolean dynamicConfigEnabled =
SparkOperatorConf.DYNAMIC_CONFIG_ENABLED.getValue();
+ String dynamicConfigSource =
SparkOperatorConf.DYNAMIC_CONFIG_SOURCE.getValue();
try (MockedStatic<MetricsSystemFactory> mockMetricsSystemFactory =
mockStatic(MetricsSystemFactory.class);
@@ -166,15 +223,16 @@ class SparkOperatorTest {
});
MockedConstruction<SparkAppReconciler> sparkAppReconcilerConstruction =
mockConstruction(SparkAppReconciler.class);
- MockedConstruction<SparkOperatorConfigMapReconciler>
configReconcilerConstruction =
- mockConstruction(SparkOperatorConfigMapReconciler.class);
MockedConstruction<ProbeService> probeServiceConstruction =
mockConstruction(ProbeService.class);
MockedConstruction<MetricsService> metricsServiceConstruction =
mockConstruction(MetricsService.class);
+ MockedConstruction<DynamicConfigMonitor>
dynamicConfigMonitorConstruction =
+ mockConstruction(DynamicConfigMonitor.class);
MockedConstruction<KubernetesMetricsInterceptor>
interceptorMockedConstruction =
mockConstruction(KubernetesMetricsInterceptor.class)) {
setConfigKey(SparkOperatorConf.DYNAMIC_CONFIG_ENABLED, true);
+ setConfigKey(SparkOperatorConf.DYNAMIC_CONFIG_SOURCE, "file");
mockMetricsSystemFactory
.when(MetricsSystemFactory::createMetricsSystem)
.thenReturn(mockMetricsSystem);
@@ -191,6 +249,7 @@ class SparkOperatorTest {
verifyNoMoreInteractions(registeredController);
} finally {
setConfigKey(SparkOperatorConf.DYNAMIC_CONFIG_ENABLED,
dynamicConfigEnabled);
+ setConfigKey(SparkOperatorConf.DYNAMIC_CONFIG_SOURCE,
dynamicConfigSource);
}
}
}
diff --git
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/config/DynamicConfigMonitorTest.java
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/config/DynamicConfigMonitorTest.java
new file mode 100644
index 0000000..8801eb3
--- /dev/null
+++
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/config/DynamicConfigMonitorTest.java
@@ -0,0 +1,151 @@
+/*
+ * 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 static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+class DynamicConfigMonitorTest {
+
+ @TempDir Path tempDir;
+
+ private DynamicConfigMonitor monitor;
+
+ @AfterEach
+ void tearDown() {
+ if (monitor != null) {
+ monitor.stop();
+ }
+ SparkOperatorConfManager.INSTANCE.refresh(Map.of());
+ }
+
+ @Test
+ void initialLoadAppliesOverridesAndNotifiesNamespaceUpdater() throws
IOException {
+ String configKey = SparkOperatorConf.RECONCILER_INTERVAL_SECONDS.getKey();
+ Path configFile = tempDir.resolve("spark-operator-dynamic.properties");
+ Files.writeString(configFile, configKey + "=60\n");
+
+ Set<String> watchedNamespaces = Set.of("ns-a", "ns-b");
+ AtomicReference<Set<String>> updaterCalledWith = new AtomicReference<>();
+
+ monitor =
+ new DynamicConfigMonitor(
+ configFile,
+ Duration.ofMillis(50),
+ () -> watchedNamespaces,
+ ns -> updaterCalledWith.set(new HashSet<>(ns)));
+ monitor.start();
+
+ assertTrue(monitor.isRunning());
+ await()
+ .atMost(Duration.ofSeconds(5))
+ .untilAsserted(
+ () -> {
+ assertEquals("60",
SparkOperatorConfManager.INSTANCE.getValue(configKey));
+ assertEquals(watchedNamespaces, updaterCalledWith.get());
+ });
+ }
+
+ @Test
+ void detectsFileChangeOnNextPoll() throws IOException {
+ String configKey = SparkOperatorConf.RECONCILER_INTERVAL_SECONDS.getKey();
+ Path configFile = tempDir.resolve("spark-operator-dynamic.properties");
+ Files.writeString(configFile, configKey + "=60\n");
+
+ AtomicReference<Set<String>> updaterCalledWith = new AtomicReference<>();
+ monitor =
+ new DynamicConfigMonitor(
+ configFile,
+ Duration.ofMillis(50),
+ Set::of,
+ ns -> updaterCalledWith.set(new HashSet<>(ns)));
+ monitor.start();
+ await()
+ .atMost(Duration.ofSeconds(5))
+ .untilAsserted(
+ () -> assertEquals("60",
SparkOperatorConfManager.INSTANCE.getValue(configKey)));
+
+ Files.writeString(configFile, configKey + "=120\n");
+
+ await()
+ .atMost(Duration.ofSeconds(5))
+ .untilAsserted(
+ () -> assertEquals("120",
SparkOperatorConfManager.INSTANCE.getValue(configKey)));
+ }
+
+ @Test
+ void handlesMissingFileGracefully() {
+ Path configFile = tempDir.resolve("does-not-exist.properties");
+ AtomicReference<Set<String>> updaterCalledWith = new AtomicReference<>();
+
+ monitor =
+ new DynamicConfigMonitor(
+ configFile,
+ Duration.ofMillis(50),
+ Set::of,
+ ns -> updaterCalledWith.set(new HashSet<>(ns)));
+ monitor.start();
+
+ assertTrue(monitor.isRunning());
+
assertFalse(SparkOperatorConfManager.INSTANCE.getAll().containsKey("missing"));
+ }
+
+ @Test
+ void noOpWhenFileUnchanged() throws IOException {
+ String configKey = SparkOperatorConf.RECONCILER_INTERVAL_SECONDS.getKey();
+ Path configFile = tempDir.resolve("spark-operator-dynamic.properties");
+ Files.writeString(configFile, configKey + "=60\n");
+
+ AtomicReference<Integer> updaterCallCount = new AtomicReference<>(0);
+ monitor =
+ new DynamicConfigMonitor(
+ configFile,
+ Duration.ofMillis(50),
+ Set::of,
+ ns -> updaterCallCount.updateAndGet(c -> c + 1));
+ monitor.start();
+ await()
+ .atMost(Duration.ofSeconds(5))
+ .untilAsserted(
+ () ->
+ assertEquals(
+ 1, updaterCallCount.get(), "namespace updater invoked once
on initial load"));
+ int countAfterStart = updaterCallCount.get();
+
+ await().pollDelay(Duration.ofMillis(300)).until(() -> true);
+ assertEquals(
+ countAfterStart, updaterCallCount.get(), "namespace updater not
re-invoked without change");
+ }
+}
diff --git
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/probe/HealthProbeTest.java
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/probe/HealthProbeTest.java
index a48f9b1..4f73a7c 100644
---
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/probe/HealthProbeTest.java
+++
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/probe/HealthProbeTest.java
@@ -42,6 +42,7 @@ import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.apache.spark.k8s.operator.config.DynamicConfigMonitor;
import org.apache.spark.k8s.operator.metrics.healthcheck.SentinelManager;
@EnableKubernetesMockClient(crud = true)
@@ -107,7 +108,7 @@ class HealthProbeTest {
@Test
void testHealthProbeWithInformerHealthWithMultiOperators() {
- HealthProbe healthyProbe = new HealthProbe(operators, List.of());
+ HealthProbe healthyProbe = new HealthProbe(operators, List.of(), null);
isRunning.set(true);
assertFalse(
healthyProbe.isHealthy(),
@@ -127,7 +128,7 @@ class HealthProbeTest {
@Test
void testHealthProbeWithInformerHealthWithSingleOperator() {
- HealthProbe healthyProbe = new HealthProbe(List.of(operator), List.of());
+ HealthProbe healthyProbe = new HealthProbe(List.of(operator), List.of(),
null);
assertFalse(healthyProbe.isHealthy(), "Health Probe should fail when
operator is not running");
isRunning.set(true);
unhealthyEventSources.put(
@@ -141,7 +142,7 @@ class HealthProbeTest {
@Test
void testHealthProbeWithSentinelHealthWithMultiOperators() {
var sentinelManager = mock(SentinelManager.class);
- HealthProbe healthyProbe = new HealthProbe(operators,
List.of(sentinelManager));
+ HealthProbe healthyProbe = new HealthProbe(operators,
List.of(sentinelManager), null);
isRunning.set(true);
isRunning2.set(true);
when(sentinelManager.allSentinelsAreHealthy()).thenReturn(false);
@@ -152,6 +153,19 @@ class HealthProbeTest {
assertTrue(healthyProbe.isHealthy(), "Healthy Probe should pass");
}
+ @Test
+ void testHealthProbeWithDynamicConfigMonitor() {
+ var dynamicConfigMonitor = mock(DynamicConfigMonitor.class);
+ HealthProbe healthyProbe = new HealthProbe(List.of(operator), List.of(),
dynamicConfigMonitor);
+ isRunning.set(true);
+ when(dynamicConfigMonitor.isRunning()).thenReturn(false);
+ assertFalse(
+ healthyProbe.isHealthy(),
+ "Healthy Probe should fail when dynamic config monitor is not
running");
+ when(dynamicConfigMonitor.isRunning()).thenReturn(true);
+ assertTrue(healthyProbe.isHealthy(), "Healthy Probe should pass");
+ }
+
private static InformerWrappingEventSourceHealthIndicator
informerHealthIndicator(
Map<String, Status> informerStatuses) {
Map<String, InformerHealthIndicator> informers = new HashMap<>();
diff --git
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/probe/ProbeServiceTest.java
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/probe/ProbeServiceTest.java
index f960d7a..03e7539 100644
---
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/probe/ProbeServiceTest.java
+++
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/probe/ProbeServiceTest.java
@@ -55,7 +55,8 @@ class ProbeServiceTest {
when(runtimeInfo.unhealthyInformerWrappingEventSourceHealthIndicator())
.thenReturn(new HashMap<>());
when(sentinelManager.allSentinelsAreHealthy()).thenReturn(true);
- ProbeService probeService = new ProbeService(List.of(operator),
List.of(sentinelManager), null);
+ ProbeService probeService =
+ new ProbeService(List.of(operator), List.of(sentinelManager), null,
null);
probeService.start();
hitHealthyEndpoint();
probeService.stop();
@@ -80,7 +81,7 @@ class ProbeServiceTest {
.thenReturn(new HashMap<>());
when(sentinelManager.allSentinelsAreHealthy()).thenReturn(true);
ProbeService probeService =
- new ProbeService(List.of(operator, operator1),
List.of(sentinelManager), null);
+ new ProbeService(List.of(operator, operator1),
List.of(sentinelManager), null, null);
probeService.start();
hitHealthyEndpoint();
probeService.stop();
@@ -106,7 +107,7 @@ class ProbeServiceTest {
.thenReturn(new HashMap<>());
when(operator1.getKubernetesClient()).thenReturn(client);
ProbeService probeService =
- new ProbeService(List.of(operator, operator1),
List.of(sentinelManager), null);
+ new ProbeService(List.of(operator, operator1),
List.of(sentinelManager), null, null);
probeService.start();
hitStartedUpEndpoint();
probeService.stop();
diff --git
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/probe/ReadinessProbeTest.java
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/probe/ReadinessProbeTest.java
index 2c2e3f3..470e98d 100644
---
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/probe/ReadinessProbeTest.java
+++
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/probe/ReadinessProbeTest.java
@@ -19,51 +19,83 @@
package org.apache.spark.k8s.operator.probe;
+import static java.net.HttpURLConnection.HTTP_BAD_REQUEST;
import static java.net.HttpURLConnection.HTTP_OK;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.io.IOException;
import java.io.OutputStream;
-import java.util.Arrays;
+import java.util.List;
+import java.util.Optional;
import com.sun.net.httpserver.HttpExchange;
-import io.fabric8.kubernetes.client.KubernetesClient;
import io.javaoperatorsdk.operator.Operator;
import io.javaoperatorsdk.operator.RuntimeInfo;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
+import org.apache.spark.k8s.operator.config.DynamicConfigMonitor;
import org.apache.spark.k8s.operator.utils.ProbeUtil;
class ReadinessProbeTest {
- KubernetesClient client;
HttpExchange httpExchange;
@BeforeEach
void beforeEach() {
OutputStream outputStream = mock(OutputStream.class);
httpExchange = mock(HttpExchange.class);
- client = mock(KubernetesClient.class);
when(httpExchange.getResponseBody()).thenReturn(outputStream);
}
@Test
void testHandleSucceed() throws IOException {
Operator operator = mock(Operator.class);
- Operator sparkConfMonitor = mock(Operator.class);
RuntimeInfo runtimeInfo = mock(RuntimeInfo.class);
- RuntimeInfo sparkConfMonitorRuntimeInfo = mock(RuntimeInfo.class);
when(operator.getRuntimeInfo()).thenReturn(runtimeInfo);
when(runtimeInfo.isStarted()).thenReturn(true);
-
when(sparkConfMonitor.getRuntimeInfo()).thenReturn(sparkConfMonitorRuntimeInfo);
- when(sparkConfMonitorRuntimeInfo.isStarted()).thenReturn(true);
- when(sparkConfMonitor.getKubernetesClient()).thenReturn(client);
- ReadinessProbe readinessProbe = new
ReadinessProbe(Arrays.asList(operator));
+ ReadinessProbe readinessProbe = new ReadinessProbe(List.of(operator),
null);
try (var mockedStatic = Mockito.mockStatic(ProbeUtil.class)) {
+ mockedStatic
+ .when(() -> ProbeUtil.areOperatorsStarted(any()))
+ .thenReturn(Optional.of(true));
readinessProbe.handle(httpExchange);
mockedStatic.verify(() -> ProbeUtil.sendMessage(httpExchange, HTTP_OK,
"started"));
}
}
+
+ @Test
+ void testHandleSucceedWithRunningDynamicConfigMonitor() throws IOException {
+ Operator operator = mock(Operator.class);
+ DynamicConfigMonitor dynamicConfigMonitor =
mock(DynamicConfigMonitor.class);
+ when(dynamicConfigMonitor.isRunning()).thenReturn(true);
+ ReadinessProbe readinessProbe = new ReadinessProbe(List.of(operator),
dynamicConfigMonitor);
+ try (var mockedStatic = Mockito.mockStatic(ProbeUtil.class)) {
+ mockedStatic
+ .when(() -> ProbeUtil.areOperatorsStarted(any()))
+ .thenReturn(Optional.of(true));
+ readinessProbe.handle(httpExchange);
+ mockedStatic.verify(() -> ProbeUtil.sendMessage(httpExchange, HTTP_OK,
"started"));
+ }
+ }
+
+ @Test
+ void testHandleFailsWhenDynamicConfigMonitorNotRunning() throws IOException {
+ Operator operator = mock(Operator.class);
+ DynamicConfigMonitor dynamicConfigMonitor =
mock(DynamicConfigMonitor.class);
+ when(dynamicConfigMonitor.isRunning()).thenReturn(false);
+ ReadinessProbe readinessProbe = new ReadinessProbe(List.of(operator),
dynamicConfigMonitor);
+ try (var mockedStatic = Mockito.mockStatic(ProbeUtil.class)) {
+ mockedStatic
+ .when(() -> ProbeUtil.areOperatorsStarted(any()))
+ .thenReturn(Optional.of(true));
+ readinessProbe.handle(httpExchange);
+ mockedStatic.verify(
+ () ->
+ ProbeUtil.sendMessage(
+ httpExchange, HTTP_BAD_REQUEST, "dynamic config monitor is
not running yet"));
+ }
+ }
}
diff --git a/tests/e2e/helm/dynamic-config-values-file.yaml
b/tests/e2e/helm/dynamic-config-values-file.yaml
new file mode 100644
index 0000000..568477a
--- /dev/null
+++ b/tests/e2e/helm/dynamic-config-values-file.yaml
@@ -0,0 +1,40 @@
+# 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.
+
+workloadResources:
+ namespaces:
+ overrideWatchedNamespaces: false
+ data:
+ - "spark-1"
+ - "spark-2"
+ role:
+ create: true
+ roleBinding:
+ create: true
+
+operatorDeployment:
+ operatorPod:
+ operatorContainer:
+ resources:
+ requests:
+ cpu: "0.1"
+
+operatorConfiguration:
+ dynamicConfig:
+ enable: true
+ source: file
+ create: true
+ data:
+ spark.kubernetes.operator.watchedNamespaces: "default"
diff --git a/tests/e2e/watched-namespaces-file/chainsaw-test.yaml
b/tests/e2e/watched-namespaces-file/chainsaw-test.yaml
new file mode 100644
index 0000000..830899b
--- /dev/null
+++ b/tests/e2e/watched-namespaces-file/chainsaw-test.yaml
@@ -0,0 +1,77 @@
+#
+# 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.
+#
+
+# Validates dynamic configuration using the 'file' source: the operator
periodically reloads a
+# properties file mounted from a ConfigMap and applies watched-namespace
changes when the file
+# content changes. The operator under test is installed by CI with
+# operatorConfiguration.dynamicConfig.source=file (see
tests/e2e/helm/dynamic-config-values-file.yaml).
+apiVersion: chainsaw.kyverno.io/v1alpha1
+kind: Test
+metadata:
+ name: spark-operator-dynamic-configuration-file-validation
+spec:
+ steps:
+ - try:
+ - apply:
+ file: spark-operator-dynamic-config-1.yaml
+ - sleep:
+ duration: 150s
+ - script:
+ content:
+ kubectl logs -n default $(kubectl get pods -o=name -l
app.kubernetes.io/component=operator-deployment,app.kubernetes.io/name=spark-kubernetes-operator)
+ check:
+ (contains($stdout, 'Updating operator namespaces to [default,
spark-1]')): true
+ - apply:
+ bindings:
+ - name: SPARK_APP_NAMESPACE
+ value: spark-1
+ file: spark-example.yaml
+ - assert:
+ bindings:
+ - name: SPARK_APP_NAMESPACE
+ value: spark-1
+ timeout: 60s
+ file: "../assertions/spark-application/spark-state-transition.yaml"
+ - apply:
+ bindings:
+ - name: SPARK_APP_NAMESPACE
+ value: spark-2
+ file: spark-example.yaml
+ - sleep:
+ duration: 150s
+ - script:
+ content:
+ kubectl get sparkapplication spark-job-succeeded-test -n spark-2
-o json | jq ".status"
+ check:
+ (contains($stdout, 'null')): true
+ catch:
+ - podLogs:
+ namespace: default
+ selector:
app.kubernetes.io/component=operator-deployment,app.kubernetes.io/name=spark-kubernetes-operator
+ - describe:
+ apiVersion: spark.apache.org/v1
+ kind: SparkApplication
+ namespace: spark-1
+ - describe:
+ apiVersion: spark.apache.org/v1
+ kind: SparkApplication
+ namespace: spark-2
+ finally:
+ - script:
+ content: |
+ kubectl delete sparkapplication spark-job-succeeded-test -n
spark-1 --ignore-not-found=true
+ kubectl delete sparkapplication spark-job-succeeded-test -n
spark-2 --ignore-not-found=true
diff --git a/tests/e2e/watched-namespaces-file/spark-example.yaml
b/tests/e2e/watched-namespaces-file/spark-example.yaml
new file mode 100644
index 0000000..54c798f
--- /dev/null
+++ b/tests/e2e/watched-namespaces-file/spark-example.yaml
@@ -0,0 +1,31 @@
+#
+# 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.
+#
+
+apiVersion: spark.apache.org/v1
+kind: SparkApplication
+metadata:
+ name: spark-job-succeeded-test
+ namespace: ($SPARK_APP_NAMESPACE)
+spec:
+ mainClass: "org.apache.spark.examples.SparkPi"
+ jars: "local:///opt/spark/examples/jars/spark-examples.jar"
+ sparkConf:
+ spark.executor.instances: "1"
+ spark.kubernetes.container.image: "apache/spark:{{SPARK_VERSION}}-scala"
+ spark.kubernetes.authenticate.driver.serviceAccountName: "spark"
+ runtimeVersions:
+ sparkVersion: "4.1.2"
diff --git
a/tests/e2e/watched-namespaces-file/spark-operator-dynamic-config-1.yaml
b/tests/e2e/watched-namespaces-file/spark-operator-dynamic-config-1.yaml
new file mode 100644
index 0000000..643dc70
--- /dev/null
+++ b/tests/e2e/watched-namespaces-file/spark-operator-dynamic-config-1.yaml
@@ -0,0 +1,35 @@
+#
+# 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.
+#
+
+apiVersion: v1
+data:
+ spark-operator-dynamic.properties: |
+ spark.kubernetes.operator.watchedNamespaces=default, spark-1
+kind: ConfigMap
+metadata:
+ annotations:
+ helm.sh/resource-policy: keep
+ meta.helm.sh/release-name: spark-kubernetes-operator
+ meta.helm.sh/release-namespace: default
+ labels:
+ app.kubernetes.io/component: operator-dynamic-config-overrides
+ app.kubernetes.io/managed-by: Helm
+ app.kubernetes.io/name: spark-kubernetes-operator
+ app.kubernetes.io/version: 1.0.0-SNAPSHOT
+ helm.sh/chart: spark-kubernetes-operator-1.8.0-dev
+ name: spark-kubernetes-operator-dynamic-configuration
+ namespace: default
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
