Runner detect ACPI shutdown

bin/cml/runner.js

@@ -1,19 +1,20 @@
 const { join } = require('path');
 const { homedir } = require('os');
 const fs = require('fs').promises;
-const { SpotNotifier } = require('ec2-spot-notification');
+const net = require('net');
 const kebabcaseKeys = require('kebabcase-keys');
 const timestring = require('timestring');
 const winston = require('winston');
+
 const CML = require('../../src/cml').default;
 const { randid, sleep } = require('../../src/utils');
 const tf = require('../../src/terraform');
 
 let cml;
 let RUNNER;
-let RUNNER_JOBS_RUNNING = [];
 let RUNNER_SHUTTING_DOWN = false;
 let RUNNER_TIMER = 0;
+const RUNNER_JOBS_RUNNING = [];
 const GH_5_MIN_TIMEOUT = (72 * 60 - 5) * 60 * 1000;
 
 const shutdown = async (opts) => {
@@ -46,14 +47,15 @@ const shutdown = async (opts) => {
 
   const retryWorkflows = async () => {
     try {
-      if (!noRetry) {
-        if (RUNNER_JOBS_RUNNING.length > 0) {
-          await Promise.all(
-            RUNNER_JOBS_RUNNING.map(
-              async (job) => await cml.pipelineRestart({ jobId: job.id })
-            )
-          );
-        }
+      if (!noRetry && RUNNER_JOBS_RUNNING.length > 0) {
+        winston.info(`Still pending jobs, retrying workflow...`);
+
+        await Promise.all(
+          RUNNER_JOBS_RUNNING.map(
+            async (job) =>
+              await cml.pipelineRerun({ id: job.pipeline, jobId: job.id })
+          )
+        );
       }
     } catch (err) {
       winston.error(err);
@@ -240,21 +242,36 @@ const runLocal = async (opts) => {
     await tf.saveTfState({ tfstate, path });
   }
 
+  if (process.platform === 'linux') {
+    const acpiSock = net.connect('/var/run/acpid.socket');
+    acpiSock.on('connect', () => {
+      winston.info('Connected to acpid service.');
+    });
+    acpiSock.on('error', (err) => {
+      winston.warn(
+        `Error connecting to ACPI socket: ${err.message}. The acpid.service helps with instance termination detection.`
+      );
+    });
+    acpiSock.on('data', (buf) => {
+      const data = buf.toString().toLowerCase();
+      if (data.includes('power') && data.includes('button')) {
+        shutdown({ ...opts, reason: 'ACPI shutdown' });
+      }
+    });
+  }
+
   const dataHandler = async (data) => {
-    const logs = await cml.parseRunnerLog({ data });
+    const logs = await cml.parseRunnerLog({ data, name });
     for (const log of logs) {
       winston.info('runner status', log);
 
       if (log.status === 'job_started') {
-        RUNNER_JOBS_RUNNING.push({ id: log.job, date: log.date });
+        const { job: id, pipeline, date } = log;
+        RUNNER_JOBS_RUNNING.push({ id, pipeline, date });
       }
 
       if (log.status === 'job_ended') {
-        const { job: jobId } = log;
-        RUNNER_JOBS_RUNNING = RUNNER_JOBS_RUNNING.filter(
-          (job) => job.id !== jobId
-        );
-
+        RUNNER_JOBS_RUNNING.pop();
         if (single) await shutdown({ ...opts, reason: 'single job' });
       }
     }
@@ -295,16 +312,6 @@ const runLocal = async (opts) => {
   }
 
   if (!noRetry) {
-    try {
-      winston.info(`EC2 id ${await SpotNotifier.instanceId()}`);
-      SpotNotifier.on('termination', () =>
-        shutdown({ ...opts, reason: 'spot_termination' })
-      );
-      SpotNotifier.start();
-    } catch (err) {
-      winston.warn('SpotNotifier can not be started.');
-    }
-
     if (cml.driver === 'github') {
       const watcherSeventyTwo = setInterval(() => {
         RUNNER_JOBS_RUNNING.forEach((job) => {