multi logs parser

iterative · Jun 5, 2022 · cb1e92d · cb1e92d
1 parent d381dce
commit cb1e92d
Show file tree

Hide file tree

Showing 6 changed files with 137 additions and 94 deletions.
diff --git a/bin/cml/runner.js b/bin/cml/runner.js
@@ -241,18 +241,23 @@ const runLocal = async (opts) => {
   }
 
   const dataHandler = async (data) => {
-    const log = await cml.parseRunnerLog({ data });
-    log && winston.info('runner status', log);
-
-    if (log && log.status === 'job_started') {
-      RUNNER_JOBS_RUNNING.push({ id: log.job, date: log.date });
-    } else if (log && log.status === 'job_ended') {
-      const { job: jobId } = log;
-      RUNNER_JOBS_RUNNING = RUNNER_JOBS_RUNNING.filter(
-        (job) => job.id !== jobId
-      );
+    const logs = await cml.parseRunnerLog({ data });
+    for (const idx in logs) {
+      const log = logs[idx];
+      log && winston.info('runner status', log);
+
+      if (log.status === 'job_started') {
+        RUNNER_JOBS_RUNNING.push({ id: log.job, date: log.date });
+      }
+
+      if (log.status === 'job_ended') {
+        const { job: jobId } = log;
+        RUNNER_JOBS_RUNNING = RUNNER_JOBS_RUNNING.filter(
+          (job) => job.id !== jobId
+        );
 
-      if (single) await shutdown({ ...opts, reason: 'single job' });
+        if (single) await shutdown({ ...opts, reason: 'single job' });
+      }
     }
   };
 
@@ -318,6 +323,11 @@ const runLocal = async (opts) => {
 };
 
 const run = async (opts) => {
+  process.on('unhandledRejection', (reason) =>
+    shutdown({ ...opts, error: new Error(reason) })
+  );
+  process.on('uncaughtException', (error) => shutdown({ ...opts, error }));
+
   ['SIGTERM', 'SIGINT', 'SIGQUIT'].forEach((signal) => {
     process.on(signal, () => shutdown({ ...opts, reason: signal }));
   });

diff --git a/src/cml.js b/src/cml.js
@@ -259,23 +259,36 @@ class CML {
 
   async parseRunnerLog(opts = {}) {
     let { data } = opts;
-    if (!data) return;
+    if (!data) return [];
 
     data = data.toString('utf8');
 
-    const date = new Date();
-    const log = {
-      level: 'info',
-      date: date.toISOString(),
-      repo: this.repo
-    };
-
-    await getDriver(this).runnerParseLog({ data, log });
-
-    if (!log.status) return;
+    const logs = [];
+    const entities = await getDriver(this).runnerParseLogEntities();
+    for (const [entity, rule] of Object.entries(entities)) {
+      const regex = new RegExp(rule);
+      if (regex.test(data)) {
+        const date = new Date();
+        const log = {
+          id: 'dummy',
+          status: entity,
+          date: date.toISOString(),
+          repo: this.repo
+        };
+
+        if (entity === 'job_ended') log.success = false;
+
+        log.level = log.success ? 'info' : 'error';
+
+        if (entity === 'job_ended_succeded') {
+          logs[logs.length - 1].success = true;
+        } else {
+          logs.push(log);
+        }
+      }
+    }
 
-    log.level = log.success ? 'info' : 'error';
-    return log;
+    return logs;
   }
 
   async startRunner(opts = {}) {

diff --git a/src/cml.test.js b/src/cml.test.js
@@ -79,25 +79,24 @@ describe('Github tests', () => {
   test('Runner logs', async () => {
     const cml = new CML();
     cml.driver = 'github';
-    let log = await cml.parseRunnerLog({ data: 'Listening for Jobs' });
-    expect(log.status).toBe('ready');
-
-    log = await cml.parseRunnerLog({ data: 'Running job' });
-    expect(log.status).toBe('job_started');
-
-    log = await cml.parseRunnerLog({
-      data: 'completed with result: Succeeded'
+    let logs = await cml.parseRunnerLog({
+      data: `
+2022-06-05 16:25:56Z: Listening for Jobs
+2022-06-05 16:26:35Z: Running job: train
+2022-06-05 16:28:03Z: Job train completed with result: Failed
+`
     });
-    expect(log.status).toBe('job_ended');
-    expect(log.success).toBe(true);
-
-    log = await cml.parseRunnerLog({ data: 'completed with result: Failed' });
-    expect(log.status).toBe('job_ended');
-    expect(log.success).toBe(false);
-
-    log = await cml.parseRunnerLog({ data: 'completed with result: Canceled' });
-    expect(log.status).toBe('job_ended');
-    expect(log.success).toBe(false);
+    expect(logs.length).toBe(3);
+    expect(logs[0].status).toBe('ready');
+    expect(logs[1].status).toBe('job_started');
+    expect(logs[2].status).toBe('job_ended');
+    expect(logs[2].success).toBe(false);
+
+    logs = await cml.parseRunnerLog({
+      data: '2022-06-05 16:28:03Z: Job train completed with result: Succeeded'
+    });
+    expect(logs[0].status).toBe('job_ended');
+    expect(logs[0].success).toBe(true);
   });
 });
 
@@ -224,26 +223,63 @@ describe('Gitlab tests', () => {
   test('Runner logs', async () => {
     const cml = new CML();
     cml.driver = 'gitlab';
-    let log = await cml.parseRunnerLog({
-      data: '{"level":"info","msg":"Starting runner for https://gitlab.com with token 2SGFrnGt ...","time":"2021-07-02T16:45:05Z"}'
-    });
-    expect(log.status).toBe('ready');
-
-    log = await cml.parseRunnerLog({
-      data: '{"job":1396213069,"level":"info","msg":"Checking for jobs... received","repo_url":"https://gitlab.com/iterative.ai/fashion_mnist.git","runner":"2SGFrnGt","time":"2021-07-02T16:45:47Z"}'
+    let logs = await cml.parseRunnerLog({
+      data: `
+{"level":"info","msg":"Starting runner for https://gitlab.com with token 2SGFrnGt ...","time":"2021-07-02T16:45:05Z"}
+{"job":1396213069,"level":"info","msg":"Checking for jobs... received","repo_url":"https://gitlab.com/iterative.ai/fashion_mnist.git","runner":"2SGFrnGt","time":"2021-07-02T16:45:47Z"}
+{"duration_s":120.0120526,"job":1396213069,"level":"warning","msg":"Job failed: execution took longer than 2m0s seconds","project":27856642,"runner":"2SGFrnGt","time":"2021-07-02T16:47:47Z"}
+`
     });
-    expect(log.status).toBe('job_started');
+    expect(logs.length).toBe(3);
+    expect(logs[0].status).toBe('ready');
+    expect(logs[1].status).toBe('job_started');
+    expect(logs[2].status).toBe('job_ended');
+    expect(logs[2].success).toBe(false);
 
-    log = await cml.parseRunnerLog({
+    logs = await cml.parseRunnerLog({
       data: '{"duration_s":7.706165838,"job":2177867438,"level":"info","msg":"Job succeeded","project":27939020,"runner":"fe36krFK","time":"2022-03-08T18:12:57+01:00"}'
     });
-    expect(log.status).toBe('job_ended');
-    expect(log.success).toBe(true);
+    expect(logs[0].status).toBe('job_ended');
+    expect(logs[0].success).toBe(true);
+  });
+});
+
+describe('Bitbucket tests', () => {
+  const OLD_ENV = process.env;
+
+  const { TEST_BITBUCKET_TOKEN: TOKEN = 'DUMMY' } = process.env;
+
+  beforeEach(() => {
+    jest.resetModules();
 
-    log = await cml.parseRunnerLog({
-      data: '{"duration_s":120.0120526,"job":1396213069,"level":"warning","msg":"Job failed: execution took longer than 2m0s seconds","project":27856642,"runner":"2SGFrnGt","time":"2021-07-02T16:47:47Z"}'
+    process.env = {};
+    process.env.REPO_TOKEN = TOKEN;
+  });
+
+  afterAll(() => {
+    process.env = OLD_ENV;
+  });
+
+  test('Runner logs', async () => {
+    const cml = new CML();
+    cml.driver = 'bitbucket';
+    let logs = await cml.parseRunnerLog({
+      data: `
+[2022-06-05 17:23:41,945] Updating runner status to "ONLINE" and checking for new steps assigned to the runner after 0 seconds and then every 30 seconds.
+[2022-06-05 17:24:12,246] Getting step StepId{accountUuid={XXXXX-XXX-XXX-XXXXXXXX}, repositoryUuid={XXXXX-XXX-XXX-XXXXXXXX}, pipelineUuid={XXXXX-XXX-XXX-XXXXXXXX}, stepUuid={XXXXX-XXX-XXX-XXXXXXXX}}.
+[2022-06-05 17:24:53,466] Completing step with result Result{status=FAILED, error=None}.
+`
+    });
+    expect(logs.length).toBe(3);
+    expect(logs[0].status).toBe('ready');
+    expect(logs[1].status).toBe('job_started');
+    expect(logs[2].status).toBe('job_ended');
+    expect(logs[2].success).toBe(false);
+
+    logs = await cml.parseRunnerLog({
+      data: '[2022-06-05 17:24:53,466] Completing step with result Result{status=PASSED, error=None}.'
     });
-    expect(log.status).toBe('job_ended');
-    expect(log.success).toBe(false);
+    expect(logs[0].status).toBe('job_ended');
+    expect(logs[0].success).toBe(true);
   });
 });
diff --git a/src/drivers/bitbucket_cloud.js b/src/drivers/bitbucket_cloud.js
@@ -266,18 +266,13 @@ class BitbucketCloud {
     return href;
   }
 
-  runnerParseLog({ data, log }) {
-    const id = 'bb';
-    if (data.includes('Getting step StepId{accountUuid={')) {
-      log.job = id;
-      log.status = 'job_started';
-    } else if (data.includes('Completing step with result Result{status=')) {
-      log.job = id;
-      log.status = 'job_ended';
-      log.success = data.includes('status=PASSED');
-    } else if (data.includes('Updating runner status to "ONLINE"')) {
-      log.status = 'ready';
-    }
+  runnerParseLogEntities() {
+    return {
+      ready: /Updating runner status to "ONLINE"/,
+      job_started: /Getting step StepId/,
+      job_ended: /Completing step with result/,
+      job_ended_succeded: /Completing step with result Result{status=PASSED/
+    };
   }
 
   async prAutoMerge({ pullRequestId, mergeMode, mergeMessage }) {

diff --git a/src/drivers/github.js b/src/drivers/github.js
@@ -329,18 +329,13 @@ class Github {
     return this.parseRunner(runner);
   }
 
-  runnerParseLog({ data, log }) {
-    const id = 'gh';
-    if (data.includes('Running job')) {
-      log.job = id;
-      log.status = 'job_started';
-    } else if (data.includes('completed with result')) {
-      log.job = id;
-      log.status = 'job_ended';
-      log.success = data.includes('Succeeded');
-    } else if (data.includes('Listening for Jobs')) {
-      log.status = 'ready';
-    }
+  runnerParseLogEntities() {
+    return {
+      ready: /Listening for Jobs/,
+      job_started: /Running job/,
+      job_ended: /completed with result/,
+      job_ended_succeded: /completed with result: Succeeded/
+    };
   }
 
   parseRunner(runner) {

diff --git a/src/drivers/gitlab.js b/src/drivers/gitlab.js
@@ -246,19 +246,13 @@ class Gitlab {
     };
   }
 
-  runnerParseLog({ data, log }) {
-    const { msg, job, duration_s: duration } = JSON.parse(data);
-
-    if (msg.endsWith('received')) {
-      log.job = job;
-      log.status = 'job_started';
-    } else if (duration) {
-      log.job = job;
-      log.status = 'job_ended';
-      log.success = msg.includes('Job succeeded');
-    } else if (msg.includes('Starting runner for')) {
-      log.status = 'ready';
-    }
+  runnerParseLogEntities() {
+    return {
+      ready: /Starting runner for/,
+      job_started: /"job":.+received/,
+      job_ended: /"duration_s":/,
+      job_ended_succeded: /"duration_s":.+Job succeeded/
+    };
   }
 
   async prCreate(opts = {}) {