From d9868363bcfdc52ed43d75eb08db518faf23dfc1 Mon Sep 17 00:00:00 2001 From: sileshidev-lab Date: Wed, 5 Aug 2026 17:22:43 +0300 Subject: [PATCH] Ensure JobClusterActor's logs always contain the clusterName Log lines emitted from JobClusterActor didn't consistently include the cluster name, making it hard to trace behavior of a specific cluster in the logs. Manually passing `name` into all ~256 log call sites in this class would be tedious and error-prone (easy to miss new call sites going forward). Instead, override aroundReceive (which every message this actor processes passes through) to set the cluster name in SLF4J's MDC before delegating to the actual message handler, and clear it afterward. This makes the cluster name automatically available to every log line emitted while handling a message, for logging backends that render MDC values (structured/JSON encoders, or a pattern layout with %X{clusterName}). Added a test that exercises aroundReceive directly via TestActorRef and asserts MDC contains the cluster name during message handling and is cleared afterward. Fixes #385 --- .../master/jobcluster/JobClusterActor.java | 23 +++++++++++++ .../master/jobcluster/JobClusterAkkaTest.java | 33 +++++++++++++++++++ 2 files changed, 56 insertions(+) diff --git a/mantis-control-plane/mantis-control-plane-server/src/main/java/io/mantisrx/master/jobcluster/JobClusterActor.java b/mantis-control-plane/mantis-control-plane-server/src/main/java/io/mantisrx/master/jobcluster/JobClusterActor.java index 7d50d6113..b79d06e2f 100644 --- a/mantis-control-plane/mantis-control-plane-server/src/main/java/io/mantisrx/master/jobcluster/JobClusterActor.java +++ b/mantis-control-plane/mantis-control-plane-server/src/main/java/io/mantisrx/master/jobcluster/JobClusterActor.java @@ -167,10 +167,13 @@ import org.checkerframework.checker.nullness.qual.NonNull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.slf4j.MDC; import rx.Observable; import rx.functions.Action1; import rx.schedulers.Schedulers; import rx.subjects.BehaviorSubject; +import scala.PartialFunction; +import scala.runtime.BoxedUnit; /** @@ -186,6 +189,8 @@ public class JobClusterActor extends AbstractActorWithTimers implements IJobClus private static final Integer DEFAULT_LIMIT = 100; private static final Integer DEFAULT_ACTIVE_JOB_LIMIT = 5000; + private static final String MDC_KEY_CLUSTER_NAME = "clusterName"; + private final Logger logger = LoggerFactory.getLogger(JobClusterActor.class); private static final String CHECK_EXPIRED_TIMER_KEY = "EXPIRE_OLD_JOBS"; @@ -673,6 +678,24 @@ MetricGroupId getMetricGroupId(String name) { return new MetricGroupId("JobClusterActor", new BasicTag("jobCluster", name)); } + /** + * Wraps every message this actor processes so that the cluster name is available via + * SLF4J's MDC for the duration of that message's handling. This lets every log line + * emitted while handling a message automatically include the cluster name (for logging + * backends that render MDC values, e.g. structured/JSON encoders, or a pattern layout + * with %X{clusterName}), instead of requiring each of this class's log call sites to pass + * `name` explicitly. + */ + @Override + public void aroundReceive(PartialFunction receive, Object msg) { + MDC.put(MDC_KEY_CLUSTER_NAME, name); + try { + super.aroundReceive(receive, msg); + } finally { + MDC.remove(MDC_KEY_CLUSTER_NAME); + } + } + @Override public void preStart() throws Exception { logger.info("JobClusterActor {} started", name); diff --git a/mantis-control-plane/mantis-control-plane-server/src/test/java/io/mantisrx/master/jobcluster/JobClusterAkkaTest.java b/mantis-control-plane/mantis-control-plane-server/src/test/java/io/mantisrx/master/jobcluster/JobClusterAkkaTest.java index 385247586..430f76b7a 100644 --- a/mantis-control-plane/mantis-control-plane-server/src/test/java/io/mantisrx/master/jobcluster/JobClusterAkkaTest.java +++ b/mantis-control-plane/mantis-control-plane-server/src/test/java/io/mantisrx/master/jobcluster/JobClusterAkkaTest.java @@ -64,8 +64,11 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import akka.actor.AbstractActor.Receive; import akka.actor.ActorRef; import akka.actor.ActorSystem; +import akka.japi.pf.ReceiveBuilder; +import akka.testkit.TestActorRef; import akka.testkit.javadsl.TestKit; import com.netflix.mantis.master.scheduler.TestHelpers; import com.typesafe.config.Config; @@ -141,6 +144,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import org.junit.AfterClass; import org.junit.Before; import org.junit.BeforeClass; @@ -150,6 +154,7 @@ import org.junit.rules.TemporaryFolder; import org.mockito.Mockito; import org.mockito.stubbing.Answer; +import org.slf4j.MDC; import rx.schedulers.Schedulers; import rx.subjects.BehaviorSubject; @@ -331,6 +336,34 @@ private JobDefinition createJob(String name2) throws InvalidJobException { return createJob(name2, 0, MantisJobDurationType.Perpetual, null); } + // LOGGING TESTS //////////////////////////////////////////////////////////////////////////////// + @Test + public void testAroundReceivePutsClusterNameInMdcDuringMessageHandlingAndClearsItAfter() { + String clusterName = "mdcTestCluster"; + MantisSchedulerFactory schedulerMockFactory = mock(MantisSchedulerFactory.class); + + TestActorRef actorRef = TestActorRef.create( + system, + props(clusterName, jobStore, schedulerMockFactory, eventPublisher, costsCalculator, 0)); + JobClusterActor actor = actorRef.underlyingActor(); + + // Call aroundReceive directly (bypassing the mailbox/dispatcher) with a receive + // handler that captures the MDC value synchronously while "processing" the message, + // since MDC is thread-local and reading it from the test thread after the fact + // (post message-send) wouldn't observe the value set-and-cleared during handling. + AtomicReference observedDuringProcessing = new AtomicReference<>(); + Receive captureMdcReceive = ReceiveBuilder.create() + .matchAny(msg -> observedDuringProcessing.set(MDC.get("clusterName"))) + .build(); + + actor.aroundReceive(captureMdcReceive.onMessage(), "any-test-message"); + + assertEquals("MDC should contain the cluster name while the actor is handling a message", + clusterName, observedDuringProcessing.get()); + assertNull("MDC should be cleared once message handling completes", + MDC.get("clusterName")); + } + // CLUSTER CRUD TESTS /////////////////////////////////////////////////////////////////////////// @Test public void testJobClusterCreate() throws Exception {