[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 <dongjoon@apache.org>
Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

Author: dongjoon-hyun

build-tools/docs-utils/src/main/java/org/apache/spark/k8s/operator/utils/DocTable.java

@@ -63,7 +63,8 @@ public void flush(PrintWriter writer) {
   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

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. | 
 

spark-operator/src/main/java/org/apache/spark/k8s/operator/SparkOperator.java

@@ -106,7 +106,8 @@ public 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());
       }
     }

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.

spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/Utils.java

@@ -35,6 +35,7 @@
 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 @@
 /** 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 static String commonResourceLabelsStr() {
     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.

spark-operator/src/test/java/org/apache/spark/k8s/operator/SparkOperatorTest.java

@@ -29,12 +29,24 @@
 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.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 @@ void testUpdateWatchedNamespacesWithDynamicConfigEnabled() {
       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());
+      }
+    }
+  }
 }

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());
+  }
+}