Fix logging configuration

airbyte-commons/src/main/resources/log4j2.xml

@@ -10,9 +10,13 @@
             <PatternLayout pattern="${default-pattern}"/>
         </Console>
         <Routing name="LogSplit">
-            <Routes pattern="${ctx:context}-${ctx:job_root}">
+            <Routes pattern="$${ctx:job_root}">
+                <!-- Don't split logs if job_root isn't defined -->
+                <Route key="$${ctx:job_root}">
+                    <Null name="/dev/null"/>
+                </Route>
                 <Route>
-                    <File name="${ctx:context}-${ctx:job_id}" fileName="${ctx:job_root}/${ctx:job_log_filename}">
+                    <File name="${ctx:job_root}" fileName="${ctx:job_root}/${ctx:job_log_filename}">
                         <PatternLayout pattern="${default-worker-file-pattern}"/>
                     </File>
                 </Route>

airbyte-commons/src/test/java/io/airbyte/commons/logging/Log4j2ConfigTest.java

@@ -24,13 +24,13 @@
 
 package io.airbyte.commons.logging;
 
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import io.airbyte.commons.io.IOs;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
-import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
@@ -42,8 +42,6 @@
 
 public class Log4j2ConfigTest {
 
-  private static final Logger LOGGER = LoggerFactory.getLogger(Log4j2ConfigTest.class);
-
   private Path root;
 
   @BeforeEach
@@ -53,51 +51,73 @@ void setUp() throws IOException {
   }
 
   @Test
-  void testWorkerDispatch() {
+  void testWorkerDispatch() throws InterruptedException {
+    final Logger logger = LoggerFactory.getLogger("testWorkerDispatch");
+
     final String filename = "logs.log";
 
-    MDC.put("context", "worker");
-    MDC.put("job_root", root.toString());
-    MDC.put("job_log_filename", filename);
-    MDC.put("job_id", "1");
+    ExecutorService executor = Executors.newFixedThreadPool(1);
+    executor.submit(() -> {
+      MDC.put("context", "worker");
+      MDC.put("job_root", root.toString());
+      MDC.put("job_log_filename", filename);
+      MDC.put("job_id", "1");
+      logger.error("random message testWorkerDispatch");
+      MDC.clear();
+    });
 
-    LOGGER.error("random message");
+    executor.shutdown();
+    executor.awaitTermination(10, TimeUnit.SECONDS);
 
-    assertTrue(IOs.readFile(root, filename).contains("random message"));
+    assertTrue(IOs.readFile(root, filename).contains("random message testWorkerDispatch"));
   }
 
   @Test
   void testLogSeparateFiles() throws InterruptedException {
+    final Logger logger = LoggerFactory.getLogger("testLogSeparateFiles");
+
     final String filename = "logs.log";
     final Path root1 = root.resolve("1");
     final Path root2 = root.resolve("2");
 
-    CountDownLatch latch = new CountDownLatch(2);
     ExecutorService executor = Executors.newFixedThreadPool(2);
     executor.submit(() -> {
-      MDC.put("context", "worker");
       MDC.put("job_root", root1.toString());
       MDC.put("job_log_filename", filename);
       MDC.put("job_id", "1");
-      LOGGER.error("random message 1");
-      latch.countDown();
+      logger.error("random message 1");
     });
 
     executor.submit(() -> {
-      MDC.put("context", "worker");
       MDC.put("job_root", root2.toString());
       MDC.put("job_log_filename", filename);
       MDC.put("job_id", "2");
-      LOGGER.error("random message 2");
-      latch.countDown();
+      logger.error("random message 2");
     });
 
     executor.shutdown();
     executor.awaitTermination(10, TimeUnit.SECONDS);
-    latch.await();
 
     assertTrue(IOs.readFile(root1, filename).contains("random message 1"));
     assertTrue(IOs.readFile(root2, filename).contains("random message 2"));
   }
 
+  @Test
+  void testLogNoJobRoot() throws InterruptedException {
+    final Logger logger = LoggerFactory.getLogger("testWorkerDispatch");
+
+    final String filename = "logs.log";
+
+    ExecutorService executor = Executors.newFixedThreadPool(1);
+    executor.submit(() -> {
+      logger.error("random message testLogNoJobRoot");
+      MDC.clear();
+    });
+
+    executor.shutdown();
+    executor.awaitTermination(10, TimeUnit.SECONDS);
+
+    assertFalse(Files.exists(root.resolve(filename)));
+  }
+
 }

airbyte-scheduler/src/main/java/io/airbyte/scheduler/JobSubmitter.java

@@ -74,7 +74,6 @@ private void submitJob(Job job) {
           final Path logFilePath = workerRun.getJobRoot().resolve(WorkerConstants.LOG_FILENAME);
           persistence.updateLogPath(job.getId(), logFilePath);
           persistence.incrementAttempts(job.getId());
-          MDC.put("context", "worker");
           MDC.put("job_id", String.valueOf(job.getId()));
           MDC.put("job_root", logFilePath.getParent().toString());
           MDC.put("job_log_filename", logFilePath.getFileName().toString());

airbyte-scheduler/src/test/java/io/airbyte/scheduler/JobSubmitterTest.java

@@ -168,7 +168,6 @@ void testMDC() throws Exception {
 
     assertEquals(
         ImmutableMap.of(
-            "context", "worker",
             "job_id", "1",
             "job_root", workerRun.getJobRoot().toString(),
             "job_log_filename", WorkerConstants.LOG_FILENAME),