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 f3551e1 [SPARK-57950] Support `spark.redaction.regex` and redact
sensitive values in operator logs
f3551e1 is described below
commit f3551e102cfff92233d8662b3ab4d180f550ba8b
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Sun Jul 5 21:30:17 2026 -0700
[SPARK-57950] Support `spark.redaction.regex` and redact sensitive values
in operator logs
### What changes were proposed in this pull request?
This PR aims to redact sensitive configuration values in operator logs.
- Add a new config option, `spark.redaction.regex`, reusing Apache Spark's
redaction key and default pattern (`(?i)secret|password|token|access[.]?key`).
- Add `Utils.redactSensitiveInfo` which replaces a value with
`*********(redacted)` when its key or value matches the pattern.
- Apply redaction to the configuration logging on startup (`spark.logConf`).
- Escape pipes in `DocTable` cells and regenerate
`docs/config_properties.md`.
### Why are the changes needed?
When `spark.logConf` is enabled, the operator prints all configuration
properties without filtering, exposing sensitive values like `*.password` or
`spark.hadoop.fs.s3a.access.key` in plain text.
### Does this PR introduce _any_ user-facing change?
Yes, sensitive configuration values are now printed as
`*********(redacted)`, and the pattern is customizable via
`spark.redaction.regex`.
### How was this patch tested?
Pass the CIs with the newly added test cases.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Fable 5
Closes #746 from dongjoon-hyun/SPARK-57950.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../apache/spark/k8s/operator/utils/DocTable.java | 3 +-
docs/config_properties.md | 1 +
.../apache/spark/k8s/operator/SparkOperator.java | 3 +-
.../k8s/operator/config/SparkOperatorConf.java | 16 +++
.../org/apache/spark/k8s/operator/utils/Utils.java | 34 +++++++
.../spark/k8s/operator/SparkOperatorTest.java | 107 ++++++++++++++++++++
.../apache/spark/k8s/operator/utils/UtilsTest.java | 108 +++++++++++++++++++++
7 files changed, 270 insertions(+), 2 deletions(-)
diff --git
a/build-tools/docs-utils/src/main/java/org/apache/spark/k8s/operator/utils/DocTable.java
b/build-tools/docs-utils/src/main/java/org/apache/spark/k8s/operator/utils/DocTable.java
index bb511d9..57faf3d 100644
---
a/build-tools/docs-utils/src/main/java/org/apache/spark/k8s/operator/utils/DocTable.java
+++
b/build-tools/docs-utils/src/main/java/org/apache/spark/k8s/operator/utils/DocTable.java
@@ -63,7 +63,8 @@ public class DocTable {
private String joinRow(List<String> elements) {
StringBuilder stringBuilder = new StringBuilder(ROW_SEPARATOR);
for (String element : elements) {
- stringBuilder.append(element).append(ROW_SEPARATOR);
+ // Escape pipes in cell content so that they do not break the Markdown
table layout
+ stringBuilder.append(element.replace("|", "\\|")).append(ROW_SEPARATOR);
}
if (elements.size() < columns) {
// Append empty cells to end if needed
diff --git a/docs/config_properties.md b/docs/config_properties.md
index f82f2b9..9c14468 100644
--- a/docs/config_properties.md
+++ b/docs/config_properties.md
@@ -51,4 +51,5 @@
| spark.kubernetes.operator.reconciler.trimStateTransitionHistoryEnabled |
Boolean | true | true | When enabled, operator would trim state transition
history when a new attempt starts, keeping previous attempt summary only. |
| spark.kubernetes.operator.watchedNamespaces | String | default | true |
Comma-separated list of namespaces that the operator would be watching for
Spark resources. If set to '*', operator would watch all namespaces. |
| spark.logConf | Boolean | false | true | When enabled, operator will print
configurations |
+ | spark.redaction.regex | String | (?i)secret\|password\|token\|access[.]?key
| true | Regex to decide which parts of configuration properties contain
sensitive information, whose values would be redacted in operator logs. This
reuses Spark's redaction configuration key and default pattern. |
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 3dcb9ad..911117a 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
@@ -106,7 +106,8 @@ public class SparkOperator {
this.registeredOperators = new ArrayList<>();
this.registeredOperators.add(registerSparkOperator());
if (SparkOperatorConf.LOG_CONF.getValue()) {
- for (var entry : SparkOperatorConfManager.INSTANCE.getAll().entrySet()) {
+ for (var entry :
+
Utils.redactSensitiveInfo(SparkOperatorConfManager.INSTANCE.getAll()).entrySet())
{
log.info("{} = {}", entry.getKey(), entry.getValue());
}
}
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 8b95d0f..03ab285 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
@@ -40,6 +40,22 @@ public final class SparkOperatorConf {
.defaultValue(false)
.build();
+ /**
+ * Regex to decide which parts of configuration properties contain sensitive
information, whose
+ * values are redacted in operator logs. Reuses Spark's {@code
spark.redaction.regex}
+ * configuration key and default pattern.
+ */
+ public static final ConfigOption<String> REDACTION_REGEX =
+ ConfigOption.<String>builder()
+ .key("spark.redaction.regex")
+ .description(
+ "Regex to decide which parts of configuration properties contain
sensitive "
+ + "information, whose values would be redacted in operator
logs. This reuses "
+ + "Spark's redaction configuration key and default pattern.")
+ .typeParameterClass(String.class)
+ .defaultValue("(?i)secret|password|token|access[.]?key")
+ .build();
+
/**
* Interval (in seconds) between periodic System.gc() invocations. Set to 0
or a negative value
* to disable.
diff --git
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/Utils.java
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/Utils.java
index 024a0e2..8921234 100644
---
a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/Utils.java
+++
b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/Utils.java
@@ -35,6 +35,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.regex.Pattern;
import java.util.stream.Collectors;
import io.fabric8.kubernetes.api.model.HasMetadata;
@@ -51,6 +52,9 @@ import
org.apache.spark.k8s.operator.listeners.SparkClusterStatusListener;
/** Utility class for common operations. */
public final class Utils {
+ /** Replacement text for redacted sensitive values, same as Spark's
redaction utility. */
+ public static final String REDACTION_REPLACEMENT_TEXT =
"*********(redacted)";
+
private Utils() {}
/**
@@ -232,6 +236,36 @@ public final class Utils {
return commonLabels;
}
+ /**
+ * Redacts sensitive values (e.g. secrets, passwords, tokens) in the given
properties for safe
+ * logging, following the semantics of Spark's {@code spark.redaction.regex}
based redaction. The
+ * redaction pattern is resolved from the {@code spark.redaction.regex}
entry of the given
+ * properties, falling back to the operator configuration ({@link
+ * SparkOperatorConf#REDACTION_REGEX}). An entry is redacted when its key or
value matches the
+ * pattern.
+ *
+ * @param props The properties to redact.
+ * @return A Map containing the given entries with sensitive values redacted.
+ */
+ public static Map<String, String> redactSensitiveInfo(Map<?, ?> props) {
+ Map<String, String> conf = new HashMap<>();
+ props.forEach((k, v) -> conf.put(String.valueOf(k), String.valueOf(v)));
+ String regex =
+ conf.getOrDefault(
+ SparkOperatorConf.REDACTION_REGEX.getKey(),
+ SparkOperatorConf.REDACTION_REGEX.getValue());
+ Pattern pattern = Pattern.compile(regex);
+ Map<String, String> redacted = new HashMap<>();
+ for (Map.Entry<String, String> entry : conf.entrySet()) {
+ if (pattern.matcher(entry.getKey()).find() ||
pattern.matcher(entry.getValue()).find()) {
+ redacted.put(entry.getKey(), REDACTION_REPLACEMENT_TEXT);
+ } else {
+ redacted.put(entry.getKey(), entry.getValue());
+ }
+ }
+ return redacted;
+ }
+
/**
* Creates a SecondaryToPrimaryMapper that maps secondary resources to their
primary resources
* based on a common label.
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 a110fdd..7c55c33 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
@@ -29,12 +29,24 @@ import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;
import static org.mockito.Mockito.when;
+import java.util.List;
+import java.util.Map;
import java.util.Set;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Consumer;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.javaoperatorsdk.operator.Operator;
import io.javaoperatorsdk.operator.RegisteredController;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.Filter;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.LoggerContext;
+import org.apache.logging.log4j.core.appender.AbstractAppender;
+import org.apache.logging.log4j.core.config.Configuration;
+import org.apache.logging.log4j.core.config.Property;
+import org.apache.logging.log4j.core.layout.PatternLayout;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.mockito.MockedConstruction;
@@ -43,6 +55,7 @@ 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.SparkOperatorConfManager;
import org.apache.spark.k8s.operator.metrics.MetricsService;
import org.apache.spark.k8s.operator.metrics.MetricsSystem;
import org.apache.spark.k8s.operator.metrics.MetricsSystemFactory;
@@ -252,4 +265,98 @@ class SparkOperatorTest {
setConfigKey(SparkOperatorConf.DYNAMIC_CONFIG_SOURCE,
dynamicConfigSource);
}
}
+
+ @Test
+ void testLogConfRedactsSensitiveValues() {
+ MetricsSystem mockMetricsSystem = mock(MetricsSystem.class);
+ KubernetesClient mockClient = mock(KubernetesClient.class);
+ TestLogAppender appender = installAppender();
+ try (MockedStatic<MetricsSystemFactory> mockMetricsSystemFactory =
+ mockStatic(MetricsSystemFactory.class);
+ MockedStatic<KubernetesClientFactory> mockKubernetesClientFactory =
+ mockStatic(KubernetesClientFactory.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<KubernetesMetricsInterceptor>
interceptorMockedConstruction =
+ mockConstruction(KubernetesMetricsInterceptor.class)) {
+ setConfigKey(SparkOperatorConf.LOG_CONF, true);
+ SparkOperatorConfManager.INSTANCE.refresh(
+ Map.of("spark.dummy.db.password", "super-sensitive-value"));
+ mockMetricsSystemFactory
+ .when(MetricsSystemFactory::createMetricsSystem)
+ .thenReturn(mockMetricsSystem);
+ mockKubernetesClientFactory
+ .when(() -> KubernetesClientFactory.buildKubernetesClient(any()))
+ .thenReturn(mockClient);
+
+ SparkOperator sparkOperator = new SparkOperator();
+ Assertions.assertNotNull(sparkOperator);
+ Assertions.assertEquals(1, operatorConstruction.constructed().size());
+ Assertions.assertEquals(1,
sparkAppReconcilerConstruction.constructed().size());
+ Assertions.assertEquals(1,
probeServiceConstruction.constructed().size());
+ Assertions.assertEquals(1,
metricsServiceConstruction.constructed().size());
+ Assertions.assertEquals(1,
interceptorMockedConstruction.constructed().size());
+
+ String redactedText =
org.apache.spark.util.Utils$.MODULE$.REDACTION_REPLACEMENT_TEXT();
+ Assertions.assertTrue(
+ appender.messages.stream()
+ .anyMatch(m -> m.equals("spark.dummy.db.password = " +
redactedText)),
+ "Expected redacted log for the sensitive key, got: " +
appender.messages);
+ Assertions.assertTrue(
+ appender.messages.stream().noneMatch(m ->
m.contains("super-sensitive-value")),
+ "Sensitive value must not appear in operator logs");
+ } finally {
+ setConfigKey(SparkOperatorConf.LOG_CONF, false);
+ SparkOperatorConfManager.INSTANCE.refresh(Map.of());
+ removeAppender(appender);
+ }
+ }
+
+ private static TestLogAppender installAppender() {
+ TestLogAppender appender = new
TestLogAppender("SparkOperatorTestLogAppender");
+ appender.start();
+ LoggerContext ctx = (LoggerContext) LogManager.getContext(false);
+ Configuration config = ctx.getConfiguration();
+ // The test log4j2 configuration has a global WARN threshold filter;
detach it so that the
+ // INFO level configuration logging can reach the appender.
+ appender.removedGlobalFilter = config.getFilter();
+ if (appender.removedGlobalFilter != null) {
+ config.removeFilter(appender.removedGlobalFilter);
+ }
+ config.getRootLogger().addAppender(appender, Level.ALL, null);
+ ctx.updateLoggers();
+ return appender;
+ }
+
+ private static void removeAppender(TestLogAppender appender) {
+ LoggerContext ctx = (LoggerContext) LogManager.getContext(false);
+ Configuration config = ctx.getConfiguration();
+ config.getRootLogger().removeAppender(appender.getName());
+ if (appender.removedGlobalFilter != null) {
+ config.addFilter(appender.removedGlobalFilter);
+ }
+ ctx.updateLoggers();
+ appender.stop();
+ }
+
+ private static final class TestLogAppender extends AbstractAppender {
+ private final List<String> messages = new CopyOnWriteArrayList<>();
+ private Filter removedGlobalFilter;
+
+ private TestLogAppender(String name) {
+ super(name, null, PatternLayout.createDefaultLayout(), false,
Property.EMPTY_ARRAY);
+ }
+
+ @Override
+ public void append(LogEvent event) {
+ if (SparkOperator.class.getName().equals(event.getLoggerName())) {
+ messages.add(event.getMessage().getFormattedMessage());
+ }
+ }
+ }
}
diff --git
a/spark-operator/src/test/java/org/apache/spark/k8s/operator/utils/UtilsTest.java
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/utils/UtilsTest.java
new file mode 100644
index 0000000..40e7e50
--- /dev/null
+++
b/spark-operator/src/test/java/org/apache/spark/k8s/operator/utils/UtilsTest.java
@@ -0,0 +1,108 @@
+/*
+ * 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.Map;
+import java.util.Properties;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import org.apache.spark.k8s.operator.config.SparkOperatorConf;
+import org.apache.spark.k8s.operator.config.SparkOperatorConfManager;
+
+class UtilsTest {
+ private static final String REDACTED_TEXT =
+ org.apache.spark.util.Utils$.MODULE$.REDACTION_REPLACEMENT_TEXT();
+ private static final String REDACTION_REGEX_KEY =
SparkOperatorConf.REDACTION_REGEX.getKey();
+
+ @Test
+ void redactionRegexConfigReusesSparkRedactionConfig() {
+ assertEquals(
+
org.apache.spark.internal.config.package$.MODULE$.SECRET_REDACTION_PATTERN().key(),
+ SparkOperatorConf.REDACTION_REGEX.getKey());
+ assertEquals(
+ org.apache.spark.internal.config.package$
+ .MODULE$
+ .SECRET_REDACTION_PATTERN()
+ .defaultValueString(),
+ SparkOperatorConf.REDACTION_REGEX.getDefaultValue());
+ }
+
+ @AfterEach
+ void resetConfigOverrides() {
+ SparkOperatorConfManager.INSTANCE.refresh(Map.of());
+ }
+
+ @Test
+ void redactSensitiveInfoMasksSensitiveKeysWithSparkDefaultPattern() {
+ Map<String, String> props =
+ Map.of(
+ "spark.hadoop.fs.s3a.access.key", "sensitive-access-key",
+ "spark.ssl.keyStorePassword", "sensitive-keystore-value",
+ "my.service.token", "sensitive-service-value",
+ "custom.api.secret", "sensitive-api-value",
+ "spark.kubernetes.operator.name", "spark-kubernetes-operator");
+ Map<String, String> redacted = Utils.redactSensitiveInfo(props);
+ assertEquals(REDACTED_TEXT,
redacted.get("spark.hadoop.fs.s3a.access.key"));
+ assertEquals(REDACTED_TEXT, redacted.get("spark.ssl.keyStorePassword"));
+ assertEquals(REDACTED_TEXT, redacted.get("my.service.token"));
+ assertEquals(REDACTED_TEXT, redacted.get("custom.api.secret"));
+ assertEquals("spark-kubernetes-operator",
redacted.get("spark.kubernetes.operator.name"));
+ assertEquals(props.keySet(), redacted.keySet());
+ }
+
+ @Test
+ void redactSensitiveInfoAcceptsProperties() {
+ Properties props = new Properties();
+ props.setProperty("my.db.password", "sensitive-db-value");
+ props.setProperty("spark.kubernetes.operator.namespace", "default");
+ Map<String, String> redacted = Utils.redactSensitiveInfo(props);
+ assertEquals(REDACTED_TEXT, redacted.get("my.db.password"));
+ assertEquals("default",
redacted.get("spark.kubernetes.operator.namespace"));
+ }
+
+ @Test
+ void redactSensitiveInfoHonorsRedactionRegexFromGivenProperties() {
+ Map<String, String> props =
+ Map.of(
+ REDACTION_REGEX_KEY, "(?i)confidential",
+ "custom.confidential.conf", "custom-value",
+ "spark.kubernetes.operator.namespace", "default");
+ Map<String, String> redacted = Utils.redactSensitiveInfo(props);
+ assertEquals(REDACTED_TEXT, redacted.get("custom.confidential.conf"));
+ assertEquals("default",
redacted.get("spark.kubernetes.operator.namespace"));
+ }
+
+ @Test
+ void redactSensitiveInfoHonorsRedactionRegexFromOperatorConfig() {
+ SparkOperatorConfManager.INSTANCE.refresh(Map.of(REDACTION_REGEX_KEY,
"(?i)confidential"));
+ Map<String, String> props =
+ Map.of(
+ "custom.confidential.conf", "custom-value",
+ "spark.kubernetes.operator.namespace", "default");
+ Map<String, String> redacted = Utils.redactSensitiveInfo(props);
+ assertEquals(REDACTED_TEXT, redacted.get("custom.confidential.conf"));
+ assertEquals("default",
redacted.get("spark.kubernetes.operator.namespace"));
+ assertEquals(props.keySet(), redacted.keySet());
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
