runtime-v1, runtime-v2: Capture docker output in separate threads.

it/runtime-v2/src/test/java/com/walmartlabs/concord/it/runtime/v2/DockerIT.java

@@ -0,0 +1,29 @@
+package com.walmartlabs.concord.it.runtime.v2;
+
+/*-
+ * *****
+ * Concord
+ * -----
+ * Copyright (C) 2017 - 2021 Walmart Inc.
+ * -----
+ * Licensed 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.
+ * =====
+ */
+
+/**
+ * Docker ITs implementation to come after DIND support added to
+ * testcontainers-concord
+ */
+public class DockerIT {
+
+}

it/runtime-v2/src/test/java/com/walmartlabs/concord/it/runtime/v2/ITConstants.java

@@ -20,17 +20,22 @@
  * =====
  */
 
+import com.google.common.base.Strings;
+
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.Properties;
 
 public final class ITConstants {
 
     public static final String PROJECT_VERSION;
+    public static final String DOCKER_ANSIBLE_IMAGE;
     public static final long DEFAULT_TEST_TIMEOUT = 120000;
 
     static {
         PROJECT_VERSION = getProperties("version.properties").getProperty("project.version");
+
+        DOCKER_ANSIBLE_IMAGE = env("IT_DOCKER_ANSIBLE_IMAGE", "walmartlabs/concord-ansible");
     }
 
     private static Properties getProperties(String path) {
@@ -43,6 +48,14 @@ private static Properties getProperties(String path) {
         }
     }
 
+    private static String env(String k, String def) {
+        String v = System.getenv(k);
+        if (Strings.isNullOrEmpty(v)) {
+            return def;
+        }
+        return v;
+    }
+
     private ITConstants() {
     }
 }

it/server/src/test/java/com/walmartlabs/concord/it/server/DockerIT.java

@@ -147,4 +147,40 @@ public void testPullRetry() throws Exception {
         byte[] ab = getLog(pir.getInstanceId());
         assertLogAtLeast(".*Error pulling the image.*", 2, ab);
     }
+
+    @Test
+    public void testDockerLogCaptureLimit() throws Exception {
+        byte[] payload = archive(DockerIT.class.getResource("dockerCaptureLimit").toURI());
+
+        Map<String, Object> input = new HashMap<>();
+        input.put("archive", payload);
+        input.put("arguments.image", ITConstants.DOCKER_ANSIBLE_IMAGE);
+        StartProcessResponse spr = start(input);
+
+        ProcessApi processApi = new ProcessApi(getApiClient());
+        ProcessEntry pir = waitForCompletion(processApi, spr.getInstanceId());
+        assertNotNull(pir.getLogFileName());
+
+        byte[] ab = getLog(pir.getLogFileName());
+        assertLog(".*stdout loop 10000.*", ab);
+        assertLog(".*stderr loop 10000.*", ab);
+    }
+
+    @Test
+    public void testDockerLogCaptureLimitV2() throws Exception {
+        byte[] payload = archive(DockerIT.class.getResource("dockerCaptureLimitV2").toURI());
+
+        Map<String, Object> input = new HashMap<>();
+        input.put("archive", payload);
+        input.put("arguments.image", ITConstants.DOCKER_ANSIBLE_IMAGE);
+        StartProcessResponse spr = start(input);
+
+        ProcessApi processApi = new ProcessApi(getApiClient());
+        ProcessEntry pir = waitForCompletion(processApi, spr.getInstanceId());
+        assertNotNull(pir.getLogFileName());
+
+        byte[] ab = getLog(pir.getLogFileName());
+        assertLog(".*stdout loop 10000.*", ab);
+        assertLog(".*stderr loop 10000.*", ab);
+    }
 }

it/server/src/test/resources/com/walmartlabs/concord/it/server/dockerCaptureLimit/concord.yml

@@ -0,0 +1,14 @@
+flows:
+  default:
+    - task: docker
+      in:
+        image: ${image}
+        debug: true
+        forcePull: false
+        stderr: myErrout
+        cmd: |
+          for i in {1..10000}
+          do
+            echo "stdout loop $i"
+            >&2 echo "stderr loop $i"
+          done

it/server/src/test/resources/com/walmartlabs/concord/it/server/dockerCaptureLimitV2/concord.yml

@@ -0,0 +1,17 @@
+flows:
+  default:
+    - task: docker
+      in:
+        image: ${image}
+        debug: true
+        forcePull: false
+        stderr: myErrout
+        cmd: |
+          for i in {1..10000}
+          do
+            echo "stdout loop $i"
+            >&2 echo "stderr loop $i"
+          done
+
+configuration:
+  runtime: concord-v2

runtime/v1/impl/src/main/java/com/walmartlabs/concord/runner/engine/DockerServiceImpl.java

@@ -79,12 +79,14 @@ public int start(Context ctx, DockerContainerSpec spec, LogCallback outCallback,
                 Process p = dp.start();
 
                 LogCapture c = new LogCapture(outCallback);
-                streamToLog(p.getInputStream(), c);
-                if (errCallback != null) {
-                    streamToLog(p.getErrorStream(), errCallback);
-                }
+                Thread outCaptureThread = startDockerOutThread(p.getInputStream(), c);
+                Thread errCaptureThread = startDockerOutThread(p.getErrorStream(), errCallback);
 
                 result = p.waitFor();
+
+                interruptThread(errCaptureThread);
+                interruptThread(outCaptureThread);
+
                 if (result == SUCCESS_EXIT_CODE || retryCount == 0 || tryCount >= retryCount) {
                     return result;
                 }
@@ -119,6 +121,30 @@ private DockerProcessBuilder.DockerProcess build(Context ctx, DockerContainerSpe
         return b.build();
     }
 
+    private static Thread startDockerOutThread(InputStream i, LogCallback c) {
+        if (c == null) {
+            return null;
+        }
+
+        Thread t = new Thread(() -> {
+            try {
+                streamToLog(i, c);
+            } catch (IOException e) {
+                log.error("Error reading docker log stream", e);
+            }
+        });
+
+        t.start();
+
+        return t;
+    }
+
+    private static void interruptThread(Thread t) {
+        if (t != null) {
+            t.interrupt();
+        }
+    }
+
     private static Map<String, String> createEffectiveEnv(Map<String, String> env, boolean exposeDockerDaemon) {
         Map<String, String> m = new HashMap<>();
 

runtime/v2/runner/src/main/java/com/walmartlabs/concord/runtime/v2/runner/DefaultDockerService.java

@@ -80,12 +80,14 @@ public int start(DockerContainerSpec spec, LogCallback outCallback, LogCallback
                 Process p = dp.start();
 
                 LogCapture c = new LogCapture(outCallback);
-                streamToLog(p.getInputStream(), c);
-                if (errCallback != null) {
-                    streamToLog(p.getErrorStream(), errCallback);
-                }
+                Thread outCaptureThread = startDockerOutThread(p.getInputStream(), c);
+                Thread errCaptureThread = startDockerOutThread(p.getErrorStream(), errCallback);
 
                 result = p.waitFor();
+
+                interruptThread(errCaptureThread);
+                interruptThread(outCaptureThread);
+
                 if (result == SUCCESS_EXIT_CODE || retryCount == 0 || tryCount >= retryCount) {
                     return result;
                 }
@@ -115,6 +117,30 @@ private DockerProcess build(DockerContainerSpec spec) throws IOException {
         return b.build();
     }
 
+    private static Thread startDockerOutThread(InputStream i, LogCallback c) {
+        if (c == null) {
+            return null;
+        }
+
+        Thread t = new Thread(() -> {
+            try {
+                streamToLog(i, c);
+            } catch (IOException e) {
+                log.error("Error reading docker log stream", e);
+            }
+        });
+
+        t.start();
+
+        return t;
+    }
+
+    private static void interruptThread(Thread t) {
+        if (t != null) {
+            t.interrupt();
+        }
+    }
+
     private static boolean needRetry(List<String> lines) {
         for (String l : lines) {
             for (Pattern p : REGISTRY_ERROR_PATTERNS) {