Rework log consolidation to simplify. Fix bug in not using paging to …

…fetch stream list

src/tw/com/commandline/Main.java

@@ -1,5 +1,6 @@
 package tw.com.commandline;
 
+import ch.qos.logback.classic.util.ContextInitializer;
 import com.amazonaws.regions.Regions;
 import com.amazonaws.services.cloudformation.model.Parameter;
 import org.apache.commons.cli.*;
@@ -13,7 +14,7 @@
 import java.util.Collection;
 
 public class Main {
-	private static final Logger logger = LoggerFactory.getLogger(Main.class);
+	private static Logger logger = null;//
 
 	private Options commandLineOptions;
 	private String[] args;
@@ -23,7 +24,11 @@ public class Main {
 	private Actions commandActions;
 
 	public Main(String[] args) {
-		this.args = args;
+
+		System.setProperty(ContextInitializer.CONFIG_FILE_PROPERTY, "logback.xml");
+        logger = LoggerFactory.getLogger(Main.class);
+
+        this.args = args;
 		executableName = "cfnassist";
 		commandLineOptions = new Options();
 		flags = new CommandFlags(executableName, commandLineOptions);

src/tw/com/entity/OutputLogEventDecorator.java

@@ -14,15 +14,10 @@ public OutputLogEventDecorator(OutputLogEvent outputLogEvent, String groupName,
         this.streamName = streamName;
     }
 
-    public long getTimestamp() {
-        return outputLogEvent.getTimestamp();
-    }
-
-    public String getMessage() {
-        return outputLogEvent.getMessage();
-    }
-
     public String getGroupName() {
         return groupName;
     }
+    public String toString() {
+        return String.format("%s %s", groupName, outputLogEvent.getMessage());
+    }
 }

src/tw/com/providers/LogClient.java

@@ -42,10 +42,25 @@ public Map<String, Map<String, String>> getGroupsWithTags() {
     }
 
     public List<LogStream> getStreamsFor(String groupName) {
-        logger.info("Get log streams for group " + groupName);
+        List<LogStream> accum = new LinkedList<>();
+        getStreamsFor(accum, groupName, "");
+        return accum;
+    }
+
+    public void getStreamsFor(List<LogStream> accum, String groupName, String token) {
         DescribeLogStreamsRequest request = new DescribeLogStreamsRequest().withLogGroupName(groupName);
-        DescribeLogStreamsResult result = theClient.describeLogStreams(request);
-        return result.getLogStreams();
+        if (!token.isEmpty()) {
+            request.setNextToken(token);
+        }
+        DescribeLogStreamsResult describeResult = theClient.describeLogStreams(request);
+        String nextToken = describeResult.getNextToken();
+        List<LogStream> logStreams = describeResult.getLogStreams();
+
+        logger.info(format("Got %s log streams for group %s and token: %s ", logStreams.size(), groupName, nextToken));
+        accum.addAll(logStreams);
+        if (!(nextToken==null) || token.equals(nextToken)) {
+            getStreamsFor(accum, groupName, nextToken);
+        }
     }
 
     public void deleteLogStream(String groupdName, String streamName) {

src/tw/com/repository/LogRepository.java

@@ -60,12 +60,13 @@ public void removeOldStreamsFor(String groupName, Duration duration) {
 
         List<LogStream> streams = logClient.getStreamsFor(groupName);
 
-        streams.stream().filter(logStream -> (logStream.getLastEventTimestamp()<when))
+        streams.stream().
+                filter(logStream -> (logStream.getLastEventTimestamp()<when))
                 .forEach(oldStream -> {
                     DateTime last = new DateTime(oldStream.getLastEventTimestamp());
-                    logger.info(format("Deleting stream %s from group %s, last event was %s", oldStream.getLogStreamName(),
-                            groupName, last));
-                    logClient.deleteLogStream(groupName, oldStream.getLogStreamName());
+                    String streamName = oldStream.getLogStreamName();
+                    logger.info(format("Deleting stream %s from group %s, last event was %s", streamName, groupName, last));
+                    logClient.deleteLogStream(groupName, streamName);
         });
     }
 
@@ -95,7 +96,10 @@ public Stream<String> fetchLogs(ProjectAndEnv projectAndEnv, Duration duration)
         List<Stream<OutputLogEventDecorator>> groupSteams = new LinkedList<>();
         groupNames.forEach(group -> {
             List<LogStream> awsLogStreams = logClient.getStreamsFor(group);
-            List<String> streamNames = awsLogStreams.stream().map(s -> s.getLogStreamName()).collect(Collectors.toList());
+            List<String> streamNames = awsLogStreams.stream().
+                    filter(stream -> stream.getLastEventTimestamp()>=when).
+                    map(stream -> stream.getLogStreamName()).collect(Collectors.toList());
+            logger.info(format("Got %s streams with events in scope", streamNames.size()));
             List<Stream<OutputLogEventDecorator>> fetchLogs = logClient.fetchLogs(group, streamNames, when);
             groupSteams.addAll(fetchLogs);
         });
@@ -117,10 +121,6 @@ public Stream<String> fetchLogs(ProjectAndEnv projectAndEnv, Duration duration)
     }
 
     private Stream<String> mapStream(Stream<OutputLogEventDecorator> outputLogEventStream) {
-        return outputLogEventStream.
-                map(entry -> {
-                    DateTime timestamp = new DateTime((entry.getTimestamp()));
-                    return String.format("%s %s %s", entry.getGroupName(), timestamp, entry.getMessage());
-                });
+        return outputLogEventStream.map(entry ->  entry.toString());
     }
 }

test/tw/com/unit/TestLogRepository.java

@@ -125,7 +125,8 @@ public void shouldFetchLogs() {
 
         Optional<String> entry = result.findFirst();
         assertTrue(entry.isPresent());
-        assertEquals(String.format("groupB %s TEST", timestamp.minusDays(days)), entry.get());
+        //assertEquals(String.format("groupB %s TEST", timestamp.minusDays(days)), entry.get());
+        assertEquals("groupB TEST", entry.get());
     }
 
     private LogStream createStream(long offset, String streamName) {