You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/03/17 05:12:28 UTC
(spark-kubernetes-operator) branch main updated: [SPARK-56005] Propagate MDC context in SparkClusterReconciler
This is an automated email from the ASF dual-hosted git repository.
dongjoon 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 d3eac5d [SPARK-56005] Propagate MDC context in SparkClusterReconciler
d3eac5d is described below
commit d3eac5dacb36ea3c6b25e53eadb9d34df1b51ac1
Author: Aaruna Godthi <aa...@gmail.com>
AuthorDate: Mon Mar 16 22:12:21 2026 -0700
[SPARK-56005] Propagate MDC context in SparkClusterReconciler
### What changes were proposed in this pull request?
Propagate MDC (Mapped Diagnostic Context) in `SparkClusterReconciler`, matching the existing pattern in `SparkAppReconciler`. Generalize `LoggingUtils.TrackedMDC` to accept `BaseResource` so both `SparkApplication` and `SparkCluster` reconcilers share the same MDC mechanism.
This PR adds `resource.name` and `resource.namespace` MDC keys (new for both resource types) and generalizes the existing `resource.app.attemptId` key to work with both `SparkApplication` and `SparkCluster`.
MDC keys set for each reconciliation:
| Key | Value | Before this PR |
|---|---|---|
| `resource.name` | Resource metadata name | Not set |
| `resource.namespace` | Resource metadata namespace | Not set |
| `resource.app.attemptId` | Current attempt ID from status | SparkApplication only |
### Why are the changes needed?
`SparkClusterReconciler` lacked MDC context that `SparkAppReconciler` already had, making it difficult to correlate cluster reconciliation log entries with specific `SparkCluster` resources in multi-tenant environments. Additionally,
`SparkAppReconciler` only logged `attemptId` but not the resource name or namespace, limiting its diagnostic usefulness.
### Does this PR introduce _any_ user-facing change?
Yes. Log output from both reconcilers now includes richer MDC context.
**SparkApplication** (before — only `attemptId`):
```
WARN {resource.app.attemptId=3} o.a.s.k.o.r.SparkAppReconciler [main] Start app reconciliation.
```
**SparkApplication** (after — adds `resource.name` and `resource.namespace`):
```
WARN {resource.app.attemptId=3, resource.name=spark-pi, resource.namespace=default} o.a.s.k.o.r.SparkAppReconciler [main] Start app reconciliation.
```
**SparkCluster** (before — no MDC context):
```
WARN {} o.a.s.k.o.r.SparkClusterReconciler [main] Start cluster reconciliation.
```
**SparkCluster** (after):
```
WARN {resource.app.attemptId=0, resource.name=my-cluster, resource.namespace=spark-ns} o.a.s.k.o.r.SparkClusterReconciler [main] Start cluster reconciliation.
```
### How was this patch tested?
Added `LoggingUtilsTest` with 4 test cases that use a custom Log4j2 `TestListAppender` to capture log events and verify MDC context appears in both the `LogEvent.getContextData()` and the formatted log output string.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code claude-opus-4-6
Closes #550 from aaruna/mdc-spark-cluster-reconciler.
Authored-by: Aaruna Godthi <aa...@gmail.com>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../reconciler/SparkClusterReconciler.java | 3 +
.../spark/k8s/operator/utils/LoggingUtils.java | 48 +++-
.../spark/k8s/operator/utils/LoggingUtilsTest.java | 244 +++++++++++++++++++++
3 files changed, 284 insertions(+), 11 deletions(-)
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkClusterReconciler.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkClusterReconciler.java
index a94d80a..6dc1b21 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkClusterReconciler.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/reconciler/SparkClusterReconciler.java
@@ -76,6 +76,7 @@ public class SparkClusterReconciler implements Reconciler<SparkCluster>, Cleaner
SparkCluster sparkCluster, Context<SparkCluster> context) throws Exception {
LoggingUtils.TrackedMDC trackedMDC = new LoggingUtils.TrackedMDC();
try {
+ trackedMDC.set(sparkCluster);
if (sentinelManager.handleSentinelResourceReconciliation(sparkCluster, context.getClient())) {
return UpdateControl.noUpdate();
}
@@ -109,6 +110,7 @@ public class SparkClusterReconciler implements Reconciler<SparkCluster>, Cleaner
SparkCluster sparkCluster, Context<SparkCluster> context, Exception e) {
LoggingUtils.TrackedMDC trackedMDC = new LoggingUtils.TrackedMDC();
try {
+ trackedMDC.set(sparkCluster);
context
.getRetryInfo()
.ifPresent(
@@ -178,6 +180,7 @@ public class SparkClusterReconciler implements Reconciler<SparkCluster>, Cleaner
public DeleteControl cleanup(SparkCluster sparkCluster, Context<SparkCluster> context) {
LoggingUtils.TrackedMDC trackedMDC = new LoggingUtils.TrackedMDC();
try {
+ trackedMDC.set(sparkCluster);
log.info("Cleaning up resources for SparkCluster.");
SparkClusterContext ctx = new SparkClusterContext(sparkCluster, context, submissionWorker);
List<ClusterReconcileStep> cleanupSteps = new ArrayList<>();
diff --git a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/LoggingUtils.java b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/LoggingUtils.java
index d438c6f..257fc27 100644
--- a/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/LoggingUtils.java
+++ b/spark-operator/src/main/java/org/apache/spark/k8s/operator/utils/LoggingUtils.java
@@ -25,34 +25,60 @@ import java.util.concurrent.locks.ReentrantLock;
import org.slf4j.MDC;
-import org.apache.spark.k8s.operator.SparkApplication;
-import org.apache.spark.k8s.operator.status.ApplicationAttemptSummary;
+import org.apache.spark.k8s.operator.BaseResource;
+import org.apache.spark.k8s.operator.spec.BaseSpec;
+import org.apache.spark.k8s.operator.status.BaseAttemptInfo;
+import org.apache.spark.k8s.operator.status.BaseAttemptSummary;
+import org.apache.spark.k8s.operator.status.BaseState;
+import org.apache.spark.k8s.operator.status.BaseStatus;
/** Utility class for logging. */
public class LoggingUtils {
/** Utility class for managing MDC (Mapped Diagnostic Context) for logging. */
public static final class TrackedMDC {
public static final String AppAttemptIdKey = "resource.app.attemptId";
+ public static final String ResourceNameKey = "resource.name";
+ public static final String ResourceNamespaceKey = "resource.namespace";
private final ReentrantLock lock = new ReentrantLock();
private final Set<String> keys = new HashSet<>();
/**
- * Sets the MDC (Mapped Diagnostic Context) with the application attempt ID if available.
+ * Sets the MDC (Mapped Diagnostic Context) with resource name, namespace, and
+ * application-specific attempt ID if available.
*
- * @param application The SparkApplication object.
+ * @param resource The BaseResource object (SparkApplication or SparkCluster).
*/
- public void set(final SparkApplication application) {
- if (application != null && application.getStatus() != null) {
- try {
- lock.lock();
- ApplicationAttemptSummary summary = application.getStatus().getCurrentAttemptSummary();
+ public void set(final BaseResource<
+ ?,
+ ? extends BaseAttemptSummary<? extends BaseAttemptInfo>,
+ ? extends BaseState<?>,
+ ? extends BaseSpec,
+ ? extends BaseStatus<?, ?, ?>> resource) {
+ if (resource == null) {
+ return;
+ }
+ try {
+ lock.lock();
+ if (resource.getMetadata() != null) {
+ if (resource.getMetadata().getName() != null) {
+ MDC.put(ResourceNameKey, resource.getMetadata().getName());
+ keys.add(ResourceNameKey);
+ }
+ if (resource.getMetadata().getNamespace() != null) {
+ MDC.put(ResourceNamespaceKey, resource.getMetadata().getNamespace());
+ keys.add(ResourceNamespaceKey);
+ }
+ }
+ if (resource.getStatus() != null) {
+ BaseAttemptSummary<? extends BaseAttemptInfo> summary =
+ resource.getStatus().getCurrentAttemptSummary();
if (summary != null && summary.getAttemptInfo() != null) {
MDC.put(AppAttemptIdKey, String.valueOf(summary.getAttemptInfo().getId()));
keys.add(AppAttemptIdKey);
}
- } finally {
- lock.unlock();
}
+ } finally {
+ lock.unlock();
}
}
diff --git a/spark-operator/src/test/java/org/apache/spark/k8s/operator/utils/LoggingUtilsTest.java b/spark-operator/src/test/java/org/apache/spark/k8s/operator/utils/LoggingUtilsTest.java
new file mode 100644
index 0000000..b2f8e62
--- /dev/null
+++ b/spark-operator/src/test/java/org/apache/spark/k8s/operator/utils/LoggingUtilsTest.java
@@ -0,0 +1,244 @@
+/*
+ * 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 static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import io.fabric8.kubernetes.api.model.ObjectMeta;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+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.Property;
+import org.apache.logging.log4j.core.layout.PatternLayout;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.MDC;
+
+import org.apache.spark.k8s.operator.SparkApplication;
+import org.apache.spark.k8s.operator.SparkCluster;
+import org.apache.spark.k8s.operator.status.ApplicationAttemptInfo;
+import org.apache.spark.k8s.operator.status.ApplicationAttemptSummary;
+import org.apache.spark.k8s.operator.status.ApplicationState;
+import org.apache.spark.k8s.operator.status.ApplicationStatus;
+import org.apache.spark.k8s.operator.status.ClusterStatus;
+
+class LoggingUtilsTest {
+
+ private static final Logger LOG = LogManager.getLogger(LoggingUtilsTest.class);
+
+ @AfterEach
+ void cleanup() {
+ MDC.clear();
+ }
+
+ @Test
+ void testSparkApplicationLogContainsMDC() throws InterruptedException {
+ TestListAppender appender = installAppender(1);
+ try {
+ SparkApplication app = createSparkApp("spark-pi", "default", 3);
+ LoggingUtils.TrackedMDC trackedMDC = new LoggingUtils.TrackedMDC();
+ trackedMDC.set(app);
+ LOG.warn("Start app reconciliation.");
+ trackedMDC.reset();
+
+ assertTrue(
+ appender.latch.await(5, TimeUnit.SECONDS), "Timed out waiting for log event");
+
+ LogEvent event = appender.events.get(0);
+ assertEquals(
+ "spark-pi",
+ event.getContextData().getValue(LoggingUtils.TrackedMDC.ResourceNameKey));
+ assertEquals(
+ "default",
+ event.getContextData().getValue(LoggingUtils.TrackedMDC.ResourceNamespaceKey));
+ assertEquals(
+ "3", event.getContextData().getValue(LoggingUtils.TrackedMDC.AppAttemptIdKey));
+
+ String formatted = appender.formattedMessages.get(0);
+ assertTrue(formatted.contains("resource.name=spark-pi"), formatted);
+ assertTrue(formatted.contains("resource.namespace=default"), formatted);
+ assertTrue(formatted.contains("resource.app.attemptId=3"), formatted);
+ } finally {
+ removeAppender(appender);
+ }
+ }
+
+ @Test
+ void testSparkClusterLogContainsMDC() throws InterruptedException {
+ TestListAppender appender = installAppender(1);
+ try {
+ SparkCluster cluster = createSparkCluster("my-cluster", "spark-ns");
+ LoggingUtils.TrackedMDC trackedMDC = new LoggingUtils.TrackedMDC();
+ trackedMDC.set(cluster);
+ LOG.warn("Start cluster reconciliation.");
+ trackedMDC.reset();
+
+ assertTrue(
+ appender.latch.await(5, TimeUnit.SECONDS), "Timed out waiting for log event");
+
+ LogEvent event = appender.events.get(0);
+ assertEquals(
+ "my-cluster",
+ event.getContextData().getValue(LoggingUtils.TrackedMDC.ResourceNameKey));
+ assertEquals(
+ "spark-ns",
+ event.getContextData().getValue(LoggingUtils.TrackedMDC.ResourceNamespaceKey));
+ assertEquals(
+ "0", event.getContextData().getValue(LoggingUtils.TrackedMDC.AppAttemptIdKey));
+
+ String formatted = appender.formattedMessages.get(0);
+ assertTrue(formatted.contains("resource.name=my-cluster"), formatted);
+ assertTrue(formatted.contains("resource.namespace=spark-ns"), formatted);
+ } finally {
+ removeAppender(appender);
+ }
+ }
+
+ @Test
+ void testResetClearsMDCFromLogOutput() throws InterruptedException {
+ TestListAppender appender = installAppender(1);
+ try {
+ SparkApplication app = createSparkApp("spark-pi", "default", 3);
+ LoggingUtils.TrackedMDC trackedMDC = new LoggingUtils.TrackedMDC();
+ trackedMDC.set(app);
+ trackedMDC.reset();
+ LOG.warn("After reset.");
+
+ assertTrue(
+ appender.latch.await(5, TimeUnit.SECONDS), "Timed out waiting for log event");
+
+ LogEvent event = appender.events.get(0);
+ assertNull(
+ event.getContextData().getValue(LoggingUtils.TrackedMDC.ResourceNameKey));
+ assertNull(
+ event.getContextData().getValue(LoggingUtils.TrackedMDC.ResourceNamespaceKey));
+ assertNull(
+ event.getContextData().getValue(LoggingUtils.TrackedMDC.AppAttemptIdKey));
+
+ String formatted = appender.formattedMessages.get(0);
+ assertTrue(
+ formatted.contains("{}"), "Expected empty MDC in log output, got: " + formatted);
+ } finally {
+ removeAppender(appender);
+ }
+ }
+
+ @Test
+ void testSetWithNullDoesNotSetMDC() throws InterruptedException {
+ TestListAppender appender = installAppender(1);
+ try {
+ LoggingUtils.TrackedMDC trackedMDC = new LoggingUtils.TrackedMDC();
+ trackedMDC.set(null);
+ LOG.warn("Null resource.");
+
+ assertTrue(
+ appender.latch.await(5, TimeUnit.SECONDS), "Timed out waiting for log event");
+
+ LogEvent event = appender.events.get(0);
+ assertNull(
+ event.getContextData().getValue(LoggingUtils.TrackedMDC.ResourceNameKey));
+ assertNull(
+ event.getContextData().getValue(LoggingUtils.TrackedMDC.ResourceNamespaceKey));
+ assertNull(
+ event.getContextData().getValue(LoggingUtils.TrackedMDC.AppAttemptIdKey));
+ } finally {
+ removeAppender(appender);
+ }
+ }
+
+ private static TestListAppender installAppender(int expectedEvents) {
+ PatternLayout layout =
+ PatternLayout.newBuilder().withPattern("%p %X %m%n").build();
+ TestListAppender appender =
+ new TestListAppender("TestListAppender", layout, expectedEvents);
+ appender.start();
+ LoggerContext ctx = (LoggerContext) LogManager.getContext(false);
+ ctx.getConfiguration().getRootLogger().addAppender(appender, Level.ALL, null);
+ ctx.updateLoggers();
+ return appender;
+ }
+
+ private static void removeAppender(TestListAppender appender) {
+ LoggerContext ctx = (LoggerContext) LogManager.getContext(false);
+ ctx.getConfiguration().getRootLogger().removeAppender(appender.getName());
+ ctx.updateLoggers();
+ appender.stop();
+ }
+
+ private static SparkApplication createSparkApp(
+ String name, String namespace, long attemptId) {
+ SparkApplication app = new SparkApplication();
+ ObjectMeta meta = new ObjectMeta();
+ meta.setName(name);
+ meta.setNamespace(namespace);
+ app.setMetadata(meta);
+ ApplicationAttemptInfo attemptInfo = new ApplicationAttemptInfo(attemptId, 0, 0, 0);
+ ApplicationAttemptSummary summary = new ApplicationAttemptSummary(attemptInfo);
+ ApplicationStatus status =
+ new ApplicationStatus(
+ new ApplicationState(),
+ Map.of(0L, new ApplicationState()),
+ null,
+ summary);
+ app.setStatus(status);
+ return app;
+ }
+
+ private static SparkCluster createSparkCluster(String name, String namespace) {
+ SparkCluster cluster = new SparkCluster();
+ ObjectMeta meta = new ObjectMeta();
+ meta.setName(name);
+ meta.setNamespace(namespace);
+ cluster.setMetadata(meta);
+ cluster.setStatus(new ClusterStatus());
+ return cluster;
+ }
+
+ static class TestListAppender extends AbstractAppender {
+ final List<LogEvent> events = new CopyOnWriteArrayList<>();
+ final List<String> formattedMessages = new CopyOnWriteArrayList<>();
+ final CountDownLatch latch;
+
+ TestListAppender(String name, PatternLayout layout, int expectedEvents) {
+ super(name, null, layout, true, Property.EMPTY_ARRAY);
+ this.latch = new CountDownLatch(expectedEvents);
+ }
+
+ @Override
+ public void append(LogEvent event) {
+ events.add(event.toImmutable());
+ formattedMessages.add(
+ new String(getLayout().toByteArray(event), StandardCharsets.UTF_8));
+ latch.countDown();
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org