add job ids as correlation ids for log messages

Author: lassewesth

progress-tracking/src/main/java/org/neo4j/gds/core/utils/progress/BatchingProgressLogger.java

@@ -36,6 +36,7 @@ public class BatchingProgressLogger implements ProgressLogger {
     public static final long MAXIMUM_LOG_INTERVAL = (long) Math.pow(2, 13);
 
     private final Log log;
+    private final JobId jobId;
     private final Concurrency concurrency;
     private long taskVolume;
     private long batchSize;
@@ -58,12 +59,13 @@ static long calculateBatchSize(long taskVolume, Concurrency concurrency) {
         return Math.max(1, BitUtil.nextHighestPowerOfTwo(batchSize));
     }
 
-    public BatchingProgressLogger(Log log, Task task, Concurrency concurrency) {
-        this(log, task, calculateBatchSize(task, concurrency), concurrency);
+    public BatchingProgressLogger(Log log, JobId jobId, Task task, Concurrency concurrency) {
+        this(log, jobId, task, calculateBatchSize(task, concurrency), concurrency);
     }
 
-    public BatchingProgressLogger(Log log, Task task, long batchSize, Concurrency concurrency) {
+    public BatchingProgressLogger(Log log, JobId jobId, Task task, long batchSize, Concurrency concurrency) {
         this.log = log;
+        this.jobId = jobId;
         this.taskVolume = task.getProgress().volume();
         this.batchSize = batchSize;
         this.taskName = task.description();
@@ -136,44 +138,38 @@ private synchronized void doLogPercentage(Supplier<String> msgFactory, long prog
     }
 
     private void logProgress(int nextPercentage) {
-        logInfo(formatWithLocale("[%s] %s %d%%", Thread.currentThread().getName(), taskName, nextPercentage));
+        logMessage(formatWithLocale("%d%%", nextPercentage));
     }
 
     private void logProgressWithMessage(int nextPercentage, String msg) {
-        logInfo(
-            formatWithLocale("[%s] %s %d%% %s", Thread.currentThread().getName(), taskName, nextPercentage, msg)
-        );
+        logMessage(formatWithLocale("%d%% %s", nextPercentage, msg));
     }
 
     @Override
     public void logMessage(String msg) {
-        log.info("[%s] %s %s", Thread.currentThread().getName(), taskName, msg);
+        log.info("[%s] [%s] %s %s", jobId.asString(), Thread.currentThread().getName(), taskName, msg);
     }
 
     @Override
     public void logMessage(Supplier<String> msg) {
         logMessage(Objects.requireNonNull(msg.get()));
     }
 
-    private void logInfo(String message) {
-        log.info(message);
-    }
-
     @Override
     public void logDebug(Supplier<String> msg) {
         if (log.isDebugEnabled()) {
-            log.debug("[%s] %s %s", Thread.currentThread().getName(), taskName, msg.get());
+            log.debug("[%s] [%s] %s %s", jobId.asString(), Thread.currentThread().getName(), taskName, msg.get());
         }
     }
 
     @Override
     public void logWarning(String message) {
-        log.warn("[%s] %s %s", Thread.currentThread().getName(), taskName, message);
+        log.warn("[%s] [%s] %s %s", jobId.asString(), Thread.currentThread().getName(), taskName, message);
     }
 
     @Override
     public void logError(String message) {
-        log.error("[%s] %s %s", Thread.currentThread().getName(), taskName, message);
+        log.error("[%s] [%s] %s %s", jobId.asString(), Thread.currentThread().getName(), taskName, message);
     }
 
     @Override

progress-tracking/src/main/java/org/neo4j/gds/core/utils/progress/tasks/TaskProgressLogger.java

@@ -22,6 +22,7 @@
 import org.jetbrains.annotations.Nullable;
 import org.neo4j.gds.core.concurrency.Concurrency;
 import org.neo4j.gds.core.utils.progress.BatchingProgressLogger;
+import org.neo4j.gds.core.utils.progress.JobId;
 import org.neo4j.gds.core.utils.progress.ProgressLogger;
 import org.neo4j.gds.logging.Log;
 
@@ -32,14 +33,14 @@ public class TaskProgressLogger extends BatchingProgressLogger {
     private final Task baseTask;
     private final TaskVisitor loggingLeafTaskVisitor;
 
-    TaskProgressLogger(Log log, Task baseTask, Concurrency concurrency) {
-        super(log, baseTask, concurrency);
+    TaskProgressLogger(Log log, JobId jobId, Task baseTask, Concurrency concurrency) {
+        super(log, jobId, baseTask, concurrency);
         this.baseTask = baseTask;
         this.loggingLeafTaskVisitor = new LoggingLeafTaskVisitor(this);
-
     }
-    TaskProgressLogger(Log log, Task baseTask, Concurrency concurrency, TaskVisitor leafTaskVisitor) {
-        super(log, baseTask, concurrency);
+
+    TaskProgressLogger(Log log, JobId jobId, Task baseTask, Concurrency concurrency, TaskVisitor leafTaskVisitor) {
+        super(log, jobId, baseTask, concurrency);
         this.baseTask = baseTask;
         this.loggingLeafTaskVisitor = leafTaskVisitor;
     }

progress-tracking/src/main/java/org/neo4j/gds/core/utils/progress/tasks/TaskProgressTracker.java

@@ -69,10 +69,10 @@ public TaskProgressTracker(
         TaskRegistryFactory taskRegistryFactory,
         UserLogRegistryFactory userLogRegistryFactory
     ) {
-        this(baseTask, jobId, taskRegistryFactory, new TaskProgressLogger(log, baseTask, concurrency), userLogRegistryFactory);
+        this(baseTask, jobId, taskRegistryFactory, new TaskProgressLogger(log, jobId, baseTask, concurrency), userLogRegistryFactory);
     }
 
-    protected TaskProgressTracker(
+    public TaskProgressTracker(
         Task baseTask,
         JobId jobId,
         TaskRegistryFactory taskRegistryFactory,

progress-tracking/src/main/java/org/neo4j/gds/core/utils/progress/tasks/TaskTreeProgressTracker.java

@@ -41,6 +41,7 @@ public TaskTreeProgressTracker(
             taskRegistryFactory,
             new TaskProgressLogger(
                 log,
+                jobId,
                 baseTask,
                 concurrency,
                 new PassThroughTaskVisitor()

progress-tracking/src/test/java/org/neo4j/gds/core/utils/BatchingProgressLoggerTest.java renamed to progress-tracking/src/test/java/org/neo4j/gds/core/utils/progress/BatchingProgressLoggerTest.java

@@ -17,7 +17,7 @@
  * You should have received a copy of the GNU General Public License
  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  */
-package org.neo4j.gds.core.utils;
+package org.neo4j.gds.core.utils.progress;
 
 import org.assertj.core.api.junit.jupiter.SoftAssertionsExtension;
 import org.eclipse.collections.impl.utility.ListIterate;
@@ -27,7 +27,7 @@
 import org.neo4j.gds.compat.TestLog;
 import org.neo4j.gds.core.concurrency.Concurrency;
 import org.neo4j.gds.core.concurrency.RunWithConcurrency;
-import org.neo4j.gds.core.utils.progress.BatchingProgressLogger;
+import org.neo4j.gds.core.utils.progress.tasks.LeafTask;
 import org.neo4j.gds.core.utils.progress.tasks.Tasks;
 import org.neo4j.gds.logging.GdsTestLog;
 import org.neo4j.gds.logging.Log;
@@ -46,6 +46,9 @@
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.assertj.core.api.Assertions.fail;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 import static org.neo4j.gds.assertj.Extractors.removingThreadId;
 import static org.neo4j.gds.utils.StringFormatting.formatWithLocale;
 
@@ -60,6 +63,7 @@ void mustLogProgressOnlyAfterBatchSizeInvocations() {
         var concurrency = new Concurrency(1);
         var logger = new BatchingProgressLogger(
             log,
+            new JobId("some job id"),
             Tasks.leaf("foo", taskVolume),
             batchSize,
             concurrency
@@ -71,13 +75,13 @@ void mustLogProgressOnlyAfterBatchSizeInvocations() {
         }
 
         var threadName = Thread.currentThread().getName();
-        var messageTemplate = "[%s] foo %d%% %d";
+        var messageTemplate = "[%s] [%s] foo %d%% %d";
         var expectedMessages = List.of(
-            formatWithLocale(messageTemplate, threadName, 1 * batchSize * 100 / taskVolume, 1 * batchSize - 1),
-            formatWithLocale(messageTemplate, threadName, 2 * batchSize * 100 / taskVolume, 2 * batchSize - 1),
-            formatWithLocale(messageTemplate, threadName, 3 * batchSize * 100 / taskVolume, 3 * batchSize - 1),
-            formatWithLocale(messageTemplate, threadName, 4 * batchSize * 100 / taskVolume, 4 * batchSize - 1),
-            formatWithLocale(messageTemplate, threadName, 5 * batchSize * 100 / taskVolume, 5 * batchSize - 1)
+            formatWithLocale(messageTemplate, "some job id", threadName, 1 * batchSize * 100 / taskVolume, 1 * batchSize - 1),
+            formatWithLocale(messageTemplate, "some job id", threadName, 2 * batchSize * 100 / taskVolume, 2 * batchSize - 1),
+            formatWithLocale(messageTemplate, "some job id", threadName, 3 * batchSize * 100 / taskVolume, 3 * batchSize - 1),
+            formatWithLocale(messageTemplate, "some job id", threadName, 4 * batchSize * 100 / taskVolume, 4 * batchSize - 1),
+            formatWithLocale(messageTemplate, "some job id", threadName, 5 * batchSize * 100 / taskVolume, 5 * batchSize - 1)
         );
 
         var messages = log.getMessages("info");
@@ -93,6 +97,7 @@ void mustLogProgressOnlyAfterHittingOrExceedingBatchSize() {
         var concurrency = new Concurrency(1);
         var logger = new BatchingProgressLogger(
             log,
+            new JobId("a job id"),
             Tasks.leaf("foo", taskVolume),
             batchSize,
             concurrency
@@ -103,7 +108,7 @@ void mustLogProgressOnlyAfterHittingOrExceedingBatchSize() {
         }
 
         var threadName = Thread.currentThread().getName();
-        var messageTemplate = "[%s] foo %d%%";
+        var messageTemplate = "[%s] [%s] foo %d%%";
 
         var progressSteps = IntStream
             .iterate(0, i -> i < taskVolume, i -> i + progressStep)
@@ -113,7 +118,7 @@ void mustLogProgressOnlyAfterHittingOrExceedingBatchSize() {
 
         var expectedMessages = loggedProgressSteps.stream()
             .skip(1)
-            .map(i -> formatWithLocale(messageTemplate, threadName, i * 100 / taskVolume))
+            .map(i -> formatWithLocale(messageTemplate, "a job id", threadName, i * 100 / taskVolume))
             .collect(Collectors.toList());
 
         var messages = log.getMessages("info");
@@ -150,7 +155,7 @@ void shouldLogAfterResetWhereACallCountHigherThanBatchSizeIsLeftBehind() {
         var taskVolume = 1337;
 
         var log = new GdsTestLog();
-        var logger = new BatchingProgressLogger(log, Tasks.leaf("Test", taskVolume), concurrency); // batchSize is 13
+        var logger = new BatchingProgressLogger(log, new JobId(), Tasks.leaf("Test", taskVolume), concurrency); // batchSize is 13
         logger.reset(taskVolume);
         logger.logProgress(20); // callCount is 20, call count after logging == 20 - 13 = 7
         assertThat(log.getMessages(TestLog.INFO))
@@ -167,7 +172,7 @@ void shouldLogAfterResetWhereACallCountHigherThanBatchSizeIsLeftBehind() {
     void log100Percent() {
         var log = new GdsTestLog();
         var concurrency = new Concurrency(1);
-        var testProgressLogger = new BatchingProgressLogger(log, Tasks.leaf("Test"), concurrency);
+        var testProgressLogger = new BatchingProgressLogger(log, new JobId(), Tasks.leaf("Test"), concurrency);
         testProgressLogger.reset(1337);
         testProgressLogger.logFinishPercentage();
         assertThat(log.getMessages(TestLog.INFO))
@@ -179,7 +184,7 @@ void log100Percent() {
     void shouldLog100OnlyOnce() {
         var log = new GdsTestLog();
         var concurrency = new Concurrency(1);
-        var testProgressLogger = new BatchingProgressLogger(log, Tasks.leaf("Test"), concurrency);
+        var testProgressLogger = new BatchingProgressLogger(log, new JobId(), Tasks.leaf("Test"), concurrency);
         testProgressLogger.reset(1);
         testProgressLogger.logProgress(1);
         testProgressLogger.logFinishPercentage();
@@ -192,7 +197,7 @@ void shouldLog100OnlyOnce() {
     void shouldNotExceed100Percent() {
         var log = new GdsTestLog();
         var concurrency = new Concurrency(1);
-        var testProgressLogger = new BatchingProgressLogger(log, Tasks.leaf("Test"), concurrency);
+        var testProgressLogger = new BatchingProgressLogger(log, new JobId(), Tasks.leaf("Test"), concurrency);
         testProgressLogger.reset(1);
         testProgressLogger.logProgress(1); // reaches 100 %
         testProgressLogger.logProgress(1); // exceeds 100 %
@@ -205,6 +210,7 @@ void shouldNotExceed100Percent() {
     void closesThreadLocal() {
         var logger = new BatchingProgressLogger(
             Log.noOpLog(),
+            new JobId(),
             Tasks.leaf("foo", 42),
             new Concurrency(1)
         );
@@ -226,7 +232,7 @@ void closesThreadLocal() {
 
     private static List<Integer> performLogging(long taskVolume, Concurrency concurrency) {
         var log = new GdsTestLog();
-        var logger = new BatchingProgressLogger(log, Tasks.leaf("Test", taskVolume), concurrency);
+        var logger = new BatchingProgressLogger(log, new JobId("the_job_id"), Tasks.leaf("Test", taskVolume), concurrency);
         logger.reset(taskVolume);
 
         var batchSize = (int) BitUtil.ceilDiv(taskVolume, concurrency.value());
@@ -248,9 +254,69 @@ private static List<Integer> performLogging(long taskVolume, Concurrency concurr
         return log
             .getMessages(TestLog.INFO)
             .stream()
-            .map(progress -> progress.split(" ")[2].replace("%", ""))
+            .map(progress -> progress.split(" ")[3].replace("%", ""))
             .map(Integer::parseInt)
             .collect(Collectors.toList());
     }
 
+    @Test
+    void shouldPrependCorrelationIdToInfoLogMessages() {
+        var log = mock(Log.class);
+        var batchingProgressLogger = new BatchingProgressLogger(
+            log,
+            JobId.parse("my job id"),
+            new LeafTask("Monsieur Alfonse", 42),
+            new Concurrency(87)
+        );
+
+        batchingProgressLogger.logMessage("Swiftly, and with style");
+
+        verify(log).info("[%s] [%s] %s %s", "my job id", "Test worker", "Monsieur Alfonse", "Swiftly, and with style");
+    }
+
+    @Test
+    void shouldPrependCorrelationIdToDebugLogMessages() {
+        var log = mock(Log.class);
+        var batchingProgressLogger = new BatchingProgressLogger(
+            log,
+            JobId.parse("my job id"),
+            new LeafTask("Monsieur Alfonse", 42),
+            new Concurrency(87)
+        );
+
+        when(log.isDebugEnabled()).thenReturn(true);
+        batchingProgressLogger.logDebug("Swiftly, and with style");
+
+        verify(log).debug("[%s] [%s] %s %s", "my job id", "Test worker", "Monsieur Alfonse", "Swiftly, and with style");
+    }
+
+    @Test
+    void shouldPrependCorrelationIdToWarningLogMessages() {
+        var log = mock(Log.class);
+        var batchingProgressLogger = new BatchingProgressLogger(
+            log,
+            JobId.parse("my job id"),
+            new LeafTask("Monsieur Alfonse", 42),
+            new Concurrency(87)
+        );
+
+        batchingProgressLogger.logWarning("Swiftly, and with style");
+
+        verify(log).warn("[%s] [%s] %s %s", "my job id", "Test worker", "Monsieur Alfonse", "Swiftly, and with style");
+    }
+
+    @Test
+    void shouldPrependCorrelationIdToErrorLogMessages() {
+        var log = mock(Log.class);
+        var batchingProgressLogger = new BatchingProgressLogger(
+            log,
+            JobId.parse("my job id"),
+            new LeafTask("Monsieur Alfonse", 42),
+            new Concurrency(87)
+        );
+
+        batchingProgressLogger.logError("Swiftly, and with style");
+
+        verify(log).error("[%s] [%s] %s %s", "my job id", "Test worker", "Monsieur Alfonse", "Swiftly, and with style");
+    }
 }

progress-tracking/src/test/java/org/neo4j/gds/core/utils/progress/tasks/TaskProgressLoggerTest.java

@@ -21,6 +21,7 @@
 
 import org.junit.jupiter.api.Test;
 import org.neo4j.gds.core.concurrency.Concurrency;
+import org.neo4j.gds.core.utils.progress.JobId;
 import org.neo4j.gds.logging.Log;
 
 import java.util.List;
@@ -36,7 +37,7 @@ void shouldNotEliminateParentTaskIfCommonPreffix(){
         var taskAB = Tasks.task("A B", List.of(taskA));
         var task = Tasks.task("T", List.of(taskA));
 
-        var logger =new TaskProgressLogger(Log.noOpLog(),task,new Concurrency(1));
+        var logger =new TaskProgressLogger(Log.noOpLog(), new JobId(), task,new Concurrency(1));
 
         assertThatNoException().isThrownBy(
             ()-> {