Make sure some logs that are long are in fact returned in its entirety (#163)

Author: eiriksm

src/createCloudJob.ts

@@ -7,6 +7,9 @@ import * as sleep from 'await-sleep'
 import Publisher from './publisher'
 import { Runlog } from './RunLog'
 import promisify from './promisify'
+
+import * as crypto from 'crypto'
+
 const clusterName = 'violinist-cluster'
 const sleepWhilePolling = 5000
 // 3 hours (180 mins) hours with the interval above.
@@ -16,6 +19,49 @@ const createLogGroup = (taskDefinition) => {
   return util.format('/ecs/%s', taskDefinition)
 }
 
+async function getAllLogEvents ({ logs, logGroupName, logStreamName, startTime = 0, startFromHead = true, pageLimit = 1000 }) {
+  let nextToken
+  const events = []
+  const seen = new Set()
+  let guard = 0
+
+  while (true) {
+    const res = await logs.getLogEvents({
+      logGroupName,
+      logStreamName,
+      startTime,
+      startFromHead,
+      limit: pageLimit,
+      nextToken
+    }).promise()
+
+    for (const e of res.events) {
+      // We only want unique events, so we use a Set to track seen event IDs.
+      // This prevents duplicates in the final events array.
+      const hash = crypto.createHash('sha1')
+      hash.update(e.message + e.timestamp)
+      e.eventId = hash.digest('hex')
+      if (!seen.has(e.eventId)) {
+        seen.add(e.eventId)
+        events.push(e)
+      }
+    }
+
+    // If the forward token didn't change, there are no more pages.
+    if (res.nextForwardToken === nextToken) {
+      break
+    };
+    nextToken = res.nextForwardToken
+
+    // Safety to avoid accidental infinite loops
+    if (++guard > 50) {
+      throw new Error('Pagination guard tripped')
+    }
+  }
+
+  return events
+}
+
 const createEcsName = (data) => {
   // Should be named like this:
   // PHP version 7.1 => 71
@@ -131,13 +177,14 @@ const createCloudJob = (config, job: Job, gitRev) => {
       while (true) {
         try {
           retries++
-          const list = await watchClient.getLogEvents({
-            limit: 100,
+          events = await getAllLogEvents({
+            logs: watchClient,
             logGroupName: createLogGroup(taskDefinition),
             logStreamName: util.format('ecs/%s/%s', name, arnParts[2])
-          }).promise()
-          events = list.events
-          if (events.length) {
+          })
+          // We require the first one to start with "[{" since thats the opening of the JSON stream.
+          // But also, the last one should end with "}]"
+          if (events.length && events[0].message.startsWith('[{') && events[events.length - 1].message.endsWith('}]')) {
             break
           }
         } catch (logErr) {

test/src/fakeWatchClient.ts

@@ -8,15 +8,16 @@ export default class fakeWatchClient extends fakeAwsBase {
 
     getPromiseOutput() {
         return {
+            nextForwardToken: 'nextToken',
             events: [
                 {
-                    message: 'One log message',
+                    message: '[{"message: "One log message"',
 
                 },
                 {
-                    message: 'Log message 2'
+                    message: '},"message":"Log message 2"}]',
                 }
             ]
         }
     }
-}
+}