More accurate log processing. (#643)

* More accurate log processing.

The Docker logs can come in non normalized form. The line breaks can be not at end of line, the frame's boundary can split multibyte unicode symbols in the middle and so on.

This patch address this problems and normalize logs so, that they can be processed strictly line by line.

To achive this, the extra base class BaseConsumer have been added. That class normalize the incoming in method accept() logs and forward it to method process() for it's child classes. Other *Consumer classes have been reworked to be child to the BaseConsumer class and only do own work.

Adititionally, BaseConsumer have new withRemoveAnsiCodes(boolean) method for ability to disable ANSI color codes removing (true by default).

* Move normalization logic directly to FrameConsumerResultCallback

It's help to not ruin the API of *Consumer classes.

* Added some tests for log normalization in class FrameConsumerResultCallback.

* Fix Codacity warning

* More test for the FrameConsumerResultCallback class

* Stabilize tests

* Log consumers, that not derive BaseConsumer class, is not receive color codes now. Added record to the changelog.

* One more test for FrameConsumerResultCallback class.

* Fixes due to the code review recommendations

* Fixes due to the code review recommendations (Part 2)

* One more use case for FrameConsumerResultCallback class. And unit test for it.
If StreamType is STDERR or STDOUT the log always have newline at line end. Therefore preprocessor should trim it to be consistent with RAW type processing.

* Roll back previouse change doe to failing tests. Doing newline trimming directly in Slf4jLogConsumer class.

* Fixes due to the code review recommendations (Part 3)

Author: vektory79

CHANGELOG.md

@@ -6,6 +6,7 @@ All notable changes to this project will be documented in this file.
 ### Fixed
 - Fixed missing `commons-codec` dependency ([\#642](https://github.com/testcontainers/testcontainers-java/issues/642))
 - Fixed `HostPortWaitStrategy` throws `NumberFormatException` when port is exposed but not mapped ([\#640](https://github.com/testcontainers/testcontainers-java/issues/640))
+- Fixed log processing: multibyte unicode, linebreaks and ASCII color codes. Color codes can be turned on with `withRemoveAnsiCodes(false)` ([PR \#643](https://github.com/testcontainers/testcontainers-java/pull/643))
 
 ### Changed
 - Support multiple HTTP status codes for HttpWaitStrategy ([\#630](https://github.com/testcontainers/testcontainers-java/issues/630))

core/src/main/java/org/testcontainers/containers/output/BaseConsumer.java

@@ -0,0 +1,17 @@
+package org.testcontainers.containers.output;
+
+import lombok.Getter;
+import lombok.Setter;
+
+import java.util.function.Consumer;
+
+public abstract class BaseConsumer<SELF extends BaseConsumer<SELF>> implements Consumer<OutputFrame> {
+    @Getter
+    @Setter
+    private boolean removeColorCodes = true;
+
+    public SELF withRemoveAnsiCodes(boolean removeAnsiCodes) {
+        this.removeColorCodes = removeAnsiCodes;
+        return (SELF) this;
+    }
+}

core/src/main/java/org/testcontainers/containers/output/FrameConsumerResultCallback.java

@@ -2,28 +2,45 @@
 
 
 import com.github.dockerjava.api.model.Frame;
+import com.github.dockerjava.api.model.StreamType;
 import com.github.dockerjava.core.async.ResultCallbackTemplate;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.function.Consumer;
+import java.util.regex.Pattern;
 
 /**
  * This class can be used as a generic callback for docker-java commands that produce Frames.
  */
 public class FrameConsumerResultCallback extends ResultCallbackTemplate<FrameConsumerResultCallback, Frame> {
 
-    private final static Logger LOGGER = LoggerFactory.getLogger(FrameConsumerResultCallback.class);
+    private static final Logger LOGGER = LoggerFactory.getLogger(FrameConsumerResultCallback.class);
+
+    private static final byte[] EMPTY_LINE = new byte[0];
+
+    private static final Pattern ANSI_COLOR_PATTERN = Pattern.compile("\u001B\\[[0-9;]+m");
+
+    private static final String LINE_BREAK_REGEX = "((\\r?\\n)|(\\r))";
+
+    static final String LINE_BREAK_AT_END_REGEX = LINE_BREAK_REGEX + "$";
 
     private Map<OutputFrame.OutputType, Consumer<OutputFrame>> consumers;
 
     private CountDownLatch completionLatch = new CountDownLatch(1);
 
+    private StringBuilder logString = new StringBuilder();
+
+    private OutputFrame brokenFrame;
+
     public FrameConsumerResultCallback() {
         consumers = new HashMap<>();
     }
@@ -45,9 +62,13 @@ public void onNext(Frame frame) {
             if (outputFrame != null) {
                 Consumer<OutputFrame> consumer = consumers.get(outputFrame.getType());
                 if (consumer == null) {
-                    LOGGER.error("got frame with type " + frame.getStreamType() + ", for which no handler is configured");
-                } else {
-                    consumer.accept(outputFrame);
+                    LOGGER.error("got frame with type {}, for which no handler is configured", frame.getStreamType());
+                } else if (outputFrame.getBytes() != null && outputFrame.getBytes().length > 0) {
+                    if (frame.getStreamType() == StreamType.RAW) {
+                        processRawFrame(outputFrame, consumer);
+                    } else {
+                        processOtherFrame(outputFrame, consumer);
+                    }
                 }
             }
         }
@@ -63,8 +84,17 @@ public void onError(Throwable throwable) {
 
     @Override
     public void close() throws IOException {
+        OutputFrame lastLine = null;
+
+        if (logString.length() > 0) {
+            lastLine = new OutputFrame(OutputFrame.OutputType.STDOUT, logString.toString().getBytes());
+        }
+
         // send an END frame to every consumer... but only once per consumer.
         for (Consumer<OutputFrame> consumer : new HashSet<>(consumers.values())) {
+            if (lastLine != null) {
+                consumer.accept(lastLine);
+            }
             consumer.accept(OutputFrame.END);
         }
         super.close();
@@ -78,4 +108,72 @@ public void close() throws IOException {
     public CountDownLatch getCompletionLatch() {
         return completionLatch;
     }
+
+    private synchronized void processRawFrame(OutputFrame outputFrame, Consumer<OutputFrame> consumer) {
+        String utf8String = outputFrame.getUtf8String();
+        byte[] bytes = outputFrame.getBytes();
+
+        // Merging the strings by bytes to solve the problem breaking non-latin unicode symbols.
+        if (brokenFrame != null) {
+            bytes = merge(brokenFrame.getBytes(), bytes);
+            utf8String = new String(bytes);
+            brokenFrame = null;
+        }
+        // Logger chunks can break the string in middle of multibyte unicode character.
+        // Backup the bytes to reconstruct proper char sequence with bytes from next frame.
+        int lastCharacterType = Character.getType(utf8String.charAt(utf8String.length() - 1));
+        if (lastCharacterType == Character.OTHER_SYMBOL) {
+            brokenFrame = new OutputFrame(outputFrame.getType(), bytes);
+            return;
+        }
+
+        utf8String = processAnsiColorCodes(utf8String, consumer);
+        normalizeLogLines(utf8String, consumer);
+    }
+
+    private synchronized void processOtherFrame(OutputFrame outputFrame, Consumer<OutputFrame> consumer) {
+        String utf8String = outputFrame.getUtf8String();
+
+        utf8String = processAnsiColorCodes(utf8String, consumer);
+        consumer.accept(new OutputFrame(outputFrame.getType(), utf8String.getBytes()));
+    }
+
+    private void normalizeLogLines(String utf8String, Consumer<OutputFrame> consumer) {
+        // Reformat strings to normalize new lines.
+        List<String> lines = new ArrayList<>(Arrays.asList(utf8String.split(LINE_BREAK_REGEX)));
+        if (lines.isEmpty()) {
+            consumer.accept(new OutputFrame(OutputFrame.OutputType.STDOUT, EMPTY_LINE));
+            return;
+        }
+        if (utf8String.startsWith("\n") || utf8String.startsWith("\r")) {
+            lines.add(0, "");
+        }
+        if (utf8String.endsWith("\n") || utf8String.endsWith("\r")) {
+            lines.add("");
+        }
+        for (int i = 0; i < lines.size() - 1; i++) {
+            String line = lines.get(i);
+            if (i == 0 && logString.length() > 0) {
+                line = logString.toString() + line;
+                logString.setLength(0);
+            }
+            consumer.accept(new OutputFrame(OutputFrame.OutputType.STDOUT, line.getBytes()));
+        }
+        logString.append(lines.get(lines.size() - 1));
+    }
+
+    private String processAnsiColorCodes(String utf8String, Consumer<OutputFrame> consumer) {
+        if (!(consumer instanceof BaseConsumer) || ((BaseConsumer) consumer).isRemoveColorCodes()) {
+            return ANSI_COLOR_PATTERN.matcher(utf8String).replaceAll("");
+        }
+        return utf8String;
+    }
+
+
+    private byte[] merge(byte[] str1, byte[] str2) {
+        byte[] mergedString = new byte[str1.length + str2.length];
+        System.arraycopy(str1, 0, mergedString, 0, str1.length);
+        System.arraycopy(str2, 0, mergedString, str1.length, str2.length);
+        return mergedString;
+    }
 }

core/src/main/java/org/testcontainers/containers/output/Slf4jLogConsumer.java

@@ -2,18 +2,13 @@
 
 import org.slf4j.Logger;
 
-import java.util.function.Consumer;
-import java.util.regex.Pattern;
-
 /**
  * A consumer for container output that logs output to an SLF4J logger.
  */
-public class Slf4jLogConsumer implements Consumer<OutputFrame> {
+public class Slf4jLogConsumer extends BaseConsumer<Slf4jLogConsumer> {
     private final Logger logger;
     private String prefix = "";
 
-    private static final Pattern ANSI_CODE_PATTERN = Pattern.compile("\\[\\d[ABCD]");
-
     public Slf4jLogConsumer(Logger logger) {
         this.logger = logger;
     }
@@ -25,28 +20,19 @@ public Slf4jLogConsumer withPrefix(String prefix) {
 
     @Override
     public void accept(OutputFrame outputFrame) {
-        if (outputFrame != null) {
-            String utf8String = outputFrame.getUtf8String();
-
-            if (utf8String != null) {
-                OutputFrame.OutputType outputType = outputFrame.getType();
-                String message = utf8String.trim();
-
-                if (ANSI_CODE_PATTERN.matcher(message).matches()) {
-                    return;
-                }
-
-                switch (outputType) {
-                    case END:
-                        break;
-                    case STDOUT:
-                    case STDERR:
-                        logger.info("{}{}: {}", prefix, outputType, message);
-                        break;
-                    default:
-                        throw new IllegalArgumentException("Unexpected outputType " + outputType);
-                }
-            }
+        OutputFrame.OutputType outputType = outputFrame.getType();
+
+        String utf8String = outputFrame.getUtf8String();
+        utf8String = utf8String.replaceAll(FrameConsumerResultCallback.LINE_BREAK_AT_END_REGEX, "");
+        switch (outputType) {
+            case END:
+                break;
+            case STDOUT:
+            case STDERR:
+                logger.info("{}{}: {}", prefix, outputType, utf8String);
+                break;
+            default:
+                throw new IllegalArgumentException("Unexpected outputType " + outputType);
         }
     }
 }

core/src/main/java/org/testcontainers/containers/output/ToStringConsumer.java

@@ -5,21 +5,26 @@
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.nio.charset.Charset;
-import java.util.function.Consumer;
 
 /**
  * Created by rnorth on 26/03/2016.
  */
-public class ToStringConsumer implements Consumer<OutputFrame> {
+public class ToStringConsumer extends BaseConsumer<ToStringConsumer> {
+    private static final byte[] NEW_LINE = "\n".getBytes();
 
+    private boolean firstLine = true;
     private ByteArrayOutputStream stringBuffer = new ByteArrayOutputStream();
 
     @Override
     public void accept(OutputFrame outputFrame) {
         try {
             if (outputFrame.getBytes() != null) {
+                if (!firstLine) {
+                    stringBuffer.write(NEW_LINE);
+                }
                 stringBuffer.write(outputFrame.getBytes());
                 stringBuffer.flush();
+                firstLine = false;
             }
         } catch (IOException e) {
             throw new RuntimeException(e);

core/src/main/java/org/testcontainers/containers/output/WaitingConsumer.java

@@ -6,14 +6,13 @@
 import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
-import java.util.function.Consumer;
 import java.util.function.Predicate;
 
 /**
  * A consumer for container output that buffers lines in a {@link java.util.concurrent.BlockingDeque} and enables tests
  * to wait for a matching condition.
  */
-public class WaitingConsumer implements Consumer<OutputFrame> {
+public class WaitingConsumer extends BaseConsumer<WaitingConsumer> {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(WaitingConsumer.class);