fix(cloud): wait for unprocessed logs before shutdown (#2212)

* fix(cloud): wait for unprocessed logs before shutdown * refactor(console): remove fixed wait from console capture shutdown event * refactor(cloud): make unprocessed logs a counter
artilleryio · Oct 17, 2023 · be1e088 · be1e088
1 parent 1eac5df
commit be1e088
Show file tree

Hide file tree

Showing 3 changed files with 34 additions and 5 deletions.
diff --git a/packages/artillery/lib/console-capture.js b/packages/artillery/lib/console-capture.js
@@ -26,8 +26,6 @@ function setupConsoleCapture() {
       debug('onBeforeExit', sendFromIndex, outputLines.length);
       clearInterval(interval);
 
-      await sleep(5 * 1000);
-
       if (!truncated && sendFromIndex < outputLines.length) {
         const ts = Date.now();
         global.artillery.globalEvents.emit(

diff --git a/packages/artillery/lib/platform/cloud/cloud.js b/packages/artillery/lib/platform/cloud/cloud.js
@@ -6,6 +6,8 @@
 
 const debug = require('debug')('cloud');
 const request = require('got');
+const awaitOnEE = require('../../util/await-on-ee');
+const sleep = require('../../util/sleep');
 const util = require('node:util');
 
 class ArtilleryCloudPlugin {
@@ -30,6 +32,7 @@ class ArtilleryCloudPlugin {
     this.defaultHeaders = {
       'x-auth-token': this.apiKey
     };
+    this.unprocessedLogsCounter = 0;
 
     let testEndInfo = {};
     global.artillery.globalEvents.on('test:init', async (testInfo) => {
@@ -86,6 +89,7 @@ class ArtilleryCloudPlugin {
 
     global.artillery.globalEvents.on('logLines', async (lines, ts) => {
       debug('logLines event', ts);
+      this.unprocessedLogsCounter += 1;
 
       let text = '';
 
@@ -99,7 +103,13 @@ class ArtilleryCloudPlugin {
         text += util.format(...Object.keys(args).map((k) => args[k])) + '\n';
       }
 
-      await this._event('testrun:textlog', { lines: text, ts });
+      try {
+        await this._event('testrun:textlog', { lines: text, ts });
+      } catch (err) {
+        debug(err);
+      } finally {
+        this.unprocessedLogsCounter -= 1;
+      }
 
       debug('last 100 characters:');
       debug(text.slice(text.length - 100, text.length));
@@ -127,6 +137,15 @@ class ArtilleryCloudPlugin {
     global.artillery.ext({
       ext: 'onShutdown',
       method: async (opts) => {
+        // Wait for the last logLines events to be processed, as they can sometimes finish processing after shutdown has finished
+        await awaitOnEE(
+          global.artillery.globalEvents,
+          'logLines',
+          200,
+          1 * 1000 //wait at most 1 second for a final log lines event emitter to be fired
+        );
+        await this.waitOnUnprocessedLogs(2 * 1000); //just waiting for ee is not enough, as the api call takes time
+
         await this._event('testrun:end', {
           ts: testEndInfo.endTime,
           exitCode: global.artillery.suggestedExitCode || opts.exitCode
@@ -142,6 +161,16 @@ class ArtilleryCloudPlugin {
     return this;
   }
 
+  async waitOnUnprocessedLogs(maxWaitTime) {
+    let waitedTime = 0;
+    while (this.unprocessedLogsCounter > 0 && waitedTime < maxWaitTime) {
+      debug('waiting on unprocessed logs');
+      await sleep(500);
+      waitedTime += 500;
+    }
+    return true;
+  }
+
   async _event(eventName, eventPayload) {
     debug('☁️', eventName, eventPayload);
 

diff --git a/packages/artillery/lib/util/await-on-ee.js b/packages/artillery/lib/util/await-on-ee.js
@@ -1,19 +1,21 @@
 const sleep = require('./sleep');
 
-async function awaitOnEE(ee, message, pollMs = 1000) {
+async function awaitOnEE(ee, message, pollMs = 1000, maxWaitMs = Infinity) {
   let messageFired = false;
   let args = null;
+  let waitedMs = 0;
 
   ee.once(message, () => {
     messageFired = true;
     args = arguments;
   });
 
-  while (true) {
+  while (true && waitedMs < maxWaitMs) {
     if (messageFired) {
       break;
     }
     await sleep(pollMs);
+    waitedMs += pollMs;
   }
 
   return args;