Fix worker logs not writing to independent files (#365)

airbytehq · Sep 21, 2020 · 281cd11 · 281cd11
1 parent 0acd1c4
commit 281cd11
Show file tree

Hide file tree

Showing 2 changed files with 42 additions and 4 deletions.
diff --git a/airbyte-commons/src/main/resources/log4j2.xml b/airbyte-commons/src/main/resources/log4j2.xml
@@ -9,9 +9,9 @@
         <Console name="Default" target="SYSTEM_OUT">
             <PatternLayout pattern="${default-pattern}"/>
         </Console>
-        <Routing name="Context">
-            <Routes pattern="${ctx:context}">
-                <Route key="worker">
+        <Routing name="LogSplit">
+            <Routes pattern="${ctx:context}-${ctx:job_root}">
+                <Route>
                     <File name="${ctx:context}-${ctx:job_id}" fileName="${ctx:job_root}/${ctx:job_log_filename}">
                         <PatternLayout pattern="${default-worker-file-pattern}"/>
                     </File>
@@ -24,7 +24,7 @@
     <Loggers>
         <Root level="DEBUG">
             <AppenderRef ref="Default"/>
-            <AppenderRef ref="Context"/>
+            <AppenderRef ref="LogSplit"/>
         </Root>
 
         <Logger name="org.eclipse.jetty" level="INFO" />

diff --git a/airbyte-commons/src/test/java/io/airbyte/commons/logging/Log4j2ConfigTest.java b/airbyte-commons/src/test/java/io/airbyte/commons/logging/Log4j2ConfigTest.java
@@ -30,6 +30,10 @@
 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;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.slf4j.Logger;
@@ -62,4 +66,38 @@ void testWorkerDispatch() {
     assertTrue(IOs.readFile(root, filename).contains("random message"));
   }
 
+  @Test
+  void testLogSeparateFiles() throws InterruptedException {
+    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();
+    });
+
+    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();
+    });
+
+    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"));
+  }
+
 }