Filter first and last log files on timestamp, and don't quit reading too early

4 files changed, 103 insertions, 45 deletions

diff --git a/container-core/src/main/java/com/yahoo/container/handler/LogHandler.java b/container-core/src/main/java/com/yahoo/container/handler/LogHandler.java
index 645c231531d..0b42b3a481b 100644
--- a/container-core/src/main/java/com/yahoo/container/handler/LogHandler.java
+++ b/container-core/src/main/java/com/yahoo/container/handler/LogHandler.java
@@ -29,10 +29,10 @@ public class LogHandler extends ThreadedHttpRequestHandler {
     @Override
     public HttpResponse handle(HttpRequest request) {
 
-        Instant earliestLogThreshold = Optional.ofNullable(request.getProperty("from"))
-                .map(Long::valueOf).map(Instant::ofEpochMilli).orElse(Instant.MIN);
-        Instant latestLogThreshold = Optional.ofNullable(request.getProperty("to"))
-                .map(Long::valueOf).map(Instant::ofEpochMilli).orElseGet(Instant::now);
+        Instant from = Optional.ofNullable(request.getProperty("from"))
+                               .map(Long::valueOf).map(Instant::ofEpochMilli).orElse(Instant.MIN);
+        Instant to = Optional.ofNullable(request.getProperty("to"))
+                             .map(Long::valueOf).map(Instant::ofEpochMilli).orElse(Instant.MAX);
 
         return new HttpResponse(200) {
             {
@@ -40,7 +40,7 @@ public class LogHandler extends ThreadedHttpRequestHandler {
             }
             @Override
             public void render(OutputStream outputStream) {
-                logReader.writeLogs(outputStream, earliestLogThreshold, latestLogThreshold);
+                logReader.writeLogs(outputStream, from, to);
             }
         };
     }
diff --git a/container-core/src/main/java/com/yahoo/container/handler/LogReader.java b/container-core/src/main/java/com/yahoo/container/handler/LogReader.java
index 6cb92244522..832b4ec6667 100644
--- a/container-core/src/main/java/com/yahoo/container/handler/LogReader.java
+++ b/container-core/src/main/java/com/yahoo/container/handler/LogReader.java
@@ -1,11 +1,17 @@
 // Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 package com.yahoo.container.handler;
 
-import com.yahoo.collections.Pair;
 import com.yahoo.vespa.defaults.Defaults;
 
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
 import java.io.OutputStream;
+import java.io.OutputStreamWriter;
 import java.io.UncheckedIOException;
 import java.nio.file.FileVisitResult;
 import java.nio.file.Files;
@@ -14,13 +20,23 @@ import java.nio.file.Paths;
 import java.nio.file.SimpleFileVisitor;
 import java.nio.file.attribute.BasicFileAttributes;
 import java.time.Instant;
-import java.util.Comparator;
-import java.util.LinkedList;
+import java.time.temporal.ChronoUnit;
+import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.regex.Pattern;
-import java.util.stream.Collectors;
+import java.util.zip.GZIPInputStream;
 import java.util.zip.GZIPOutputStream;
 
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.util.Comparator.comparing;
+
+/**
+ * @author olaaun
+ * @author freva
+ * @author jonmv
+ */
 class LogReader {
 
     private final Path logDirectory;
@@ -35,22 +51,61 @@ class LogReader {
         this.logFilePattern = logFilePattern;
     }
 
-    void writeLogs(OutputStream outputStream, Instant earliestLogThreshold, Instant latestLogThreshold) {
+    void writeLogs(OutputStream outputStream, Instant from, Instant to) {
         try {
-            for (Path file : getMatchingFiles(earliestLogThreshold, latestLogThreshold)) {
-                if (!file.toString().endsWith(".gz") && !(outputStream instanceof GZIPOutputStream)) {
-                    outputStream = new GZIPOutputStream(outputStream);
+            List<Path> logs = getMatchingFiles(from, to);
+            for (int i = 0; i < logs.size(); i++) {
+                Path log = logs.get(i);
+                boolean zipped = log.toString().endsWith(".gz");
+                try (InputStream in = Files.newInputStream(log)) {
+                    InputStream inProxy;
+
+                    // If the log needs filtering, possibly unzip (and rezip) it, and filter its lines on timestamp.
+                    if (i == 0 || i == logs.size() - 1) {
+                        ByteArrayOutputStream buffer = new ByteArrayOutputStream();
+                        try (BufferedReader reader = new BufferedReader(new InputStreamReader(zipped ? new GZIPInputStream(in) : in, UTF_8));
+                             BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(zipped ? new GZIPOutputStream(buffer) : buffer, UTF_8))) {
+                            for (String line; (line = reader.readLine()) != null; ) {
+                                String[] parts = line.split("\t");
+                                if (parts.length != 7)
+                                    continue;
+
+                                Instant at = Instant.EPOCH.plus((long) (Double.parseDouble(parts[0]) * 1_000_000), ChronoUnit.MICROS);
+                                if (at.isAfter(from) && at.isBefore(to)) {
+                                    writer.write(line);
+                                    writer.newLine();
+                                }
+                            }
+                        }
+                        inProxy = new ByteArrayInputStream(buffer.toByteArray());
+                    }
+                    else
+                        inProxy = in;
+
+                    // At the point when logs switch to un-zipped, replace the output stream with a zipping proxy.
+                    if ( ! zipped && ! (outputStream instanceof GZIPOutputStream))
+                        outputStream = new GZIPOutputStream(outputStream);
+
+                    inProxy.transferTo(outputStream);
                 }
-                Files.copy(file, outputStream);
             }
-            outputStream.close();
-        } catch (IOException e) {
+        }
+        catch (IOException e) {
             throw new UncheckedIOException(e);
         }
+        finally {
+            try {
+                outputStream.close();
+            }
+            catch (IOException e) {
+                throw new UncheckedIOException(e);
+            }
+        }
     }
 
-    private List<Path> getMatchingFiles(Instant earliestLogThreshold, Instant latestLogThreshold) {
-        final List<Pair<Path, Instant>> paths = new LinkedList<>();
+    /** Returns log files which may have relevant entries, sorted by modification time — the first and last must be filtered. */
+    private List<Path> getMatchingFiles(Instant from, Instant to) {
+        Map<Path, Instant> paths = new HashMap<>();
         try {
             Files.walkFileTree(logDirectory, new SimpleFileVisitor<>() {
 
@@ -61,12 +116,8 @@ class LogReader {
 
                 @Override
                 public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) {
-                    Instant lastModified = attrs.lastModifiedTime().toInstant();
-                    if (lastModified.isAfter(earliestLogThreshold) &&
-                            lastModified.isBefore(latestLogThreshold) &&
-                            logFilePattern.matcher(file.getFileName().toString()).matches()) {
-                        paths.add(new Pair<>(file, lastModified));
-                    }
+                    if (logFilePattern.matcher(file.getFileName().toString()).matches())
+                        paths.put(file, attrs.lastModifiedTime().toInstant());
 
                     return FileVisitResult.CONTINUE;
                 }
@@ -76,13 +127,20 @@ class LogReader {
                     return FileVisitResult.CONTINUE;
                 }
             });
-        } catch (IOException e) {
+        }
+        catch (IOException e) {
             throw new UncheckedIOException(e);
         }
 
-        return paths.stream()
-                .sorted(Comparator.comparing(Pair::getSecond))
-                .map(Pair::getFirst)
-                .collect(Collectors.toList());
+        List<Path> sorted = new ArrayList<>();
+        for (var entries = paths.entrySet().stream().sorted(comparing(Map.Entry::getValue)).iterator(); entries.hasNext(); ) {
+            var entry = entries.next();
+            if (entry.getValue().isAfter(from))
+                sorted.add(entry.getKey());
+            if (entry.getValue().isAfter(to))
+                break;
+        }
+        return sorted;
     }
+
 }
diff --git a/container-core/src/test/java/com/yahoo/container/handler/LogHandlerTest.java b/container-core/src/test/java/com/yahoo/container/handler/LogHandlerTest.java
index f0885451431..01dcb885a97 100644
--- a/container-core/src/test/java/com/yahoo/container/handler/LogHandlerTest.java
+++ b/container-core/src/test/java/com/yahoo/container/handler/LogHandlerTest.java
@@ -47,9 +47,9 @@ public class LogHandlerTest {
         }
 
         @Override
-        protected void writeLogs(OutputStream outputStream, Instant earliestLogThreshold, Instant latestLogThreshold)  {
+        protected void writeLogs(OutputStream outputStream, Instant from, Instant to)  {
             try {
-                if (latestLogThreshold.isAfter(Instant.ofEpochMilli(1000))) {
+                if (to.isAfter(Instant.ofEpochMilli(1000))) {
                     outputStream.write("newer log".getBytes());
                 } else {
                     outputStream.write("older log".getBytes());
diff --git a/container-core/src/test/java/com/yahoo/container/handler/LogReaderTest.java b/container-core/src/test/java/com/yahoo/container/handler/LogReaderTest.java
index 27300e5f3f5..c68facf4f01 100644
--- a/container-core/src/test/java/com/yahoo/container/handler/LogReaderTest.java
+++ b/container-core/src/test/java/com/yahoo/container/handler/LogReaderTest.java
@@ -18,6 +18,7 @@ import java.util.regex.Pattern;
 import java.util.zip.GZIPInputStream;
 import java.util.zip.GZIPOutputStream;
 
+import static java.nio.charset.StandardCharsets.UTF_8;
 import static org.junit.Assert.assertEquals;
 
 public class LogReaderTest {
@@ -25,15 +26,18 @@ public class LogReaderTest {
     private final FileSystem fileSystem = TestFileSystem.create();
     private final Path logDirectory = fileSystem.getPath("/opt/vespa/logs");
 
+    private static final String log1 = "0.1\t17491290-v6-1.ostk.bm2.prod.ne1.yahoo.com\t5480\tcontainer\tstdout\tinfo\tERROR: Bundle canary-application [71] Unable to get module class path. (java.lang.NullPointerException)\n";
+    private static final String log2 = "0.2\t17491290-v6-1.ostk.bm2.prod.ne1.yahoo.com\t5480\tcontainer\tstderr\twarning\tjava.lang.NullPointerException\\n\\tat org.apache.felix.framework.BundleRevisionImpl.calculateContentPath(BundleRevisionImpl.java:438)\\n\\tat org.apache.felix.framework.BundleRevisionImpl.initializeContentPath(BundleRevisionImpl.java:371)\n";
+
     @Before
     public void setup() throws IOException {
         Files.createDirectories(logDirectory.resolve("subfolder"));
 
         Files.setLastModifiedTime(
-                Files.write(logDirectory.resolve("log1.log.gz"), compress("This is one log file\n")),
+                Files.write(logDirectory.resolve("log1.log.gz"), compress(log1)),
                 FileTime.from(Instant.ofEpochMilli(123)));
         Files.setLastModifiedTime(
-                Files.write(logDirectory.resolve("subfolder/log2.log"), "This is another log file\n".getBytes()),
+                Files.write(logDirectory.resolve("subfolder/log2.log"), log2.getBytes(UTF_8)),
                 FileTime.from(Instant.ofEpochMilli(234)));
 
     }
@@ -42,31 +46,27 @@ public class LogReaderTest {
     public void testThatLogsOutsideRangeAreExcluded() throws Exception {
         ByteArrayOutputStream baos = new ByteArrayOutputStream();
         LogReader logReader = new LogReader(logDirectory, Pattern.compile(".*"));
-        logReader.writeLogs(baos, Instant.ofEpochMilli(235), Instant.ofEpochMilli(300));
-        String expected = "";
-        String actual = decompress(baos.toByteArray());
-        assertEquals(expected, actual);
+        logReader.writeLogs(baos, Instant.ofEpochMilli(150), Instant.ofEpochMilli(160));
+
+        assertEquals("", decompress(baos.toByteArray()));
     }
 
     @Test
     public void testThatLogsNotMatchingRegexAreExcluded() throws Exception {
         ByteArrayOutputStream baos = new ByteArrayOutputStream();
         LogReader logReader = new LogReader(logDirectory, Pattern.compile(".*2\\.log"));
-        logReader.writeLogs(baos, Instant.ofEpochMilli(21), Instant.now());
-        String expected = "This is another log file\n";
-        String actual = decompress(baos.toByteArray());
-        assertEquals(expected, actual);
+        logReader.writeLogs(baos, Instant.ofEpochMilli(0), Instant.ofEpochMilli(300));
+
+        assertEquals(log2, decompress(baos.toByteArray()));
     }
 
     @Test
     public void testZippedStreaming() throws IOException {
         ByteArrayOutputStream zippedBaos = new ByteArrayOutputStream();
         LogReader logReader = new LogReader(logDirectory, Pattern.compile(".*"));
-        logReader.writeLogs(zippedBaos, Instant.ofEpochMilli(21), Instant.now());
+        logReader.writeLogs(zippedBaos, Instant.ofEpochMilli(0), Instant.ofEpochMilli(300));
 
-        String expected = "This is one log file\nThis is another log file\n";
-        String actual = decompress(zippedBaos.toByteArray());
-        assertEquals(expected, actual);
+        assertEquals(log1 + log2, decompress(zippedBaos.toByteArray()));
     }
 
     private byte[] compress(String input) throws IOException {