functional worker

Author: lfroment0

src/config/config.ts

@@ -0,0 +1,7 @@
+import * as os from 'os';
+
+export const config = {
+    redisHost: process.env.REDIS_HOST || '127.0.0.1',
+    redisPort: Number(process.env.REDIS_PORT) || 6379,
+    defaultQueue: process.env.DEFAULT_QUEUE || 'zmon:queue:default',
+}

src/index.ts

@@ -1,14 +1,18 @@
 import { getQueueClient } from './redis/redisHandler';
 import { startServer } from './server/server';
 import ZmonWorker from './worker/zmon-worker';
+import { config } from './config/config';
 
 // start health endpoint server
-startServer()
+const redisClient = getQueueClient(config.redisHost, config.redisPort);
 
-// connect to redis
-const redisClient = getQueueClient();
-const queueName = 'zmon:queue:default';
+const cb = () => {
+    console.log('\nterminate Redis connection');
+    redisClient.end();
+}
+
+startServer(cb);
 
 // start worker
-const w = new ZmonWorker(redisClient, queueName);
+const w = new ZmonWorker(redisClient, config.defaultQueue);
 w.startWorker();

src/server/server.ts

@@ -7,27 +7,29 @@ const app = express()
 
 app.get('/', (req, res) => {
     res.send('ok')
-})
+});
 
 const server = createServer(app)
 
-function onSignal(): Promise<any> {
-    console.log('server is starting cleanup')
-    return Promise.all([
-       null
-    ]);
+function onSignal(cb: () => any):  () => Promise<any>{
+    console.log('server is starting cleanup');
+    return function (): Promise<any> {
+        return Promise.all([
+            cb()
+        ]);
+    }
 }
 
 async function onHealthCheck() {
     // checks if the system is healthy, like the db connection is live
     // resolves, if health, rejects if not
 }
 
-export function startServer() {
+export function startServer(cb: () => any) {
     createTerminus(server, {
         signal: 'SIGINT',
         healthChecks: { '/healthcheck': onHealthCheck },
-        onSignal,
+        onSignal: onSignal(cb),
     });
 
     server.listen(3000)

src/worker/exec.spec.ts

@@ -1,4 +1,4 @@
-import { execCheck } from './exec';
+import { execZmonScript } from './exec';
 
 function mockHttp() {
     return {response: true};
@@ -24,7 +24,7 @@ describe('Worker', () => {
         `;
         const ctx = {};
 
-        const res = execCheck(ctx, check, entity);
+        const res = execZmonScript(ctx, check, {entity});
         expect(res).toEqual('kube_pod');
     });
 
@@ -36,8 +36,9 @@ describe('Worker', () => {
         `;
         let res;
         let err;
+
         try {
-            res = execCheck({}, check, entity);
+            res = execZmonScript({}, check, {entity});
         } catch (e) {
             err = e;
         }
@@ -61,13 +62,13 @@ describe('Worker', () => {
         `;
         const ctx = {};
 
-        const res = execCheck(ctx, check, entity);
+        const res = execZmonScript(ctx, check, {entity});
         expect(res).toEqual({type: entity.type, team: entity.team});
     });
 
     it('should make an http call', () => {
         const ctx = {http: mockHttp}
-        const res = execCheck(ctx, check, entity);
+        const res = execZmonScript(ctx, check, {entity});
         expect(res).toEqual({response: true});
     });
 
@@ -82,7 +83,7 @@ describe('Worker', () => {
         let err;
 
         try {
-            res = execCheck(ctx, check, entity);
+            res = execZmonScript(ctx, check, {entity});
         } catch (e) {
             err = e;
         }
@@ -95,7 +96,7 @@ describe('Worker', () => {
         const ctx = {}
         let res;
         try {
-            res = execCheck(ctx, check, entity);
+            res = execZmonScript(ctx, check, {entity});
         } catch (e) {}
 
         expect(res).toBeUndefined();

src/worker/exec.ts

@@ -1,18 +1,26 @@
 import { Context, createContext, CreateContextOptions, runInNewContext, RunningScriptOptions } from 'vm';
+import { performance } from 'perf_hooks';
 
-export function execCheck(ctx: Context, checkDefinition: string, entity: any) {
-    const script = `(${checkDefinition})();`
+export function execZmonScript(ctx: Context, script: string, additionalCtx: {value: any} | {entity: Object}) {
+    const scriptToExec = `(${script})();`
     const opt: CreateContextOptions = {
         name: 'checkId: xxx',
         codeGeneration: {
             strings: false
         },
     };
 
-    const localCtx = {...ctx, entity};
+    const localCtx = {...ctx, ...additionalCtx};
     const newCtx = createContext(localCtx, opt);
 
-    return runInNewContext(script, newCtx);
+    const t0 = Date.now();
+    const result = runInNewContext(scriptToExec, newCtx);
+    const t1 = Date.now();
+
+    return {
+        td: t1 - t0,
+        result,
+    };
 }
 
 

src/worker/zmon-worker.ts

@@ -1,52 +1,102 @@
 import { RedisClient } from 'redis';
-import { execCheck } from './exec';
-import execCtx from '../exec-ctx/ctx';
 import { Context } from 'vm';
+import { promisify } from 'util';
+
+import execCtx from '../exec-ctx/ctx';
+import { execZmonScript } from './exec';
+
 
 export default class ZmonWorker {
-    private queueName: string
     private queueConn: RedisClient;
     private execCtx: Context;
+    private brpopAsync: Function;
+    private queueName: string;
+
     private tasks = {
         'check_and_notify': this.checkAndNotify,
         // 'trial_run': trialRun,
         // 'cleanup': cleanup,
-    }
+    };
 
     constructor(client: RedisClient, queueName: string) {
         this.queueConn = client;
-        this.queueName = queueName;
         this.execCtx = execCtx;
+        this.brpopAsync = promisify(client.brpop).bind(client);
+        this.queueName = queueName;
     }
 
-    startWorker() {
-        console.log("worker started")
-        this.consumeQueue();
-        // while (true) {
-            // this.consumeQueue();
-        // }
+    async startWorker() {
+        console.log('worker started');
+        let i = 0;
+        while (true) {
+            const res = await this.consumeQueue();
+            const [queueName, msg] = res;
+            const parsedMsg = JSON.parse(msg);
+            const {check_id, entity, command} = parsedMsg.body.args[0];
+            const alertList = parsedMsg.body.args[1];
+            const checkResult = this.executeZmonTask(command, check_id, entity);
+            this.storeCheckResult(check_id, entity.id, JSON.stringify(checkResult));
+            alertList.forEach((alert: any) => {
+                const startTime = Date.now();
+                const {td, result: alertRes} = execZmonScript(this.execCtx, alert.condition, {value: checkResult});
+                const alertObj = {
+                    exc: 1,
+                    downtimes: [],
+                    captures: {},
+                    start_time: startTime,
+                    worker: 'nodejs-worker',
+                    ts: Date.now(),
+                    value: alertRes,
+                    td,
+                };
+                if (!!alertRes) {
+                    this.storeAlertResult(alert.id, entity.id, JSON.stringify(alertObj));
+                    console.log(`alert ${alert.id} triggered on entity ${entity.id}`);
+                } else {
+                    this.deleteAlertResult(alert.id, entity.id);
+                }
+            });
+        }
     }
 
-    consumeQueue() {
-        this.queueConn.brpop(this.queueName, 0, this.executeZmonTask);
+    private async consumeQueue() {
+        const checkId = '';
+        const entityId = '';
+        return await this.brpopAsync(this.queueName, 0);
     }
 
-    executeZmonTask(err: Error | null, res: [string, string]) {
-        if (err !== null) {
-            return err;
-        }
+    private executeZmonTask(checkScript: string, checkId: string, entity: any): any {
+
+        const {td, result} = execZmonScript(execCtx, checkScript, {entity});
+        return result;
 
-        const [queueName, msg] = res;
-        console.log(msg);
     }
 
-    checkAndNotify(req: any) {
+    private checkAndNotify(req: any) {
         console.log(req);
         // const script = '';
         // const entity = {hello: 'world'};
         // const checkId = req['check_id'];
         // const entityId = req['entity']['id'];
         //
-        // execCheck(this.execCtx, script, entity)
+        // execZmonScript(this.execCtx, script, entity)
+    }
+
+    private storeCheckResult(checkId: number, entityId: string, res: string) {
+        const key = `zmon:checks:${checkId}:${entityId}`;
+        this.queueConn.lpush(key, res, console.log);
+        this.queueConn.ltrim(key, 0, 20 - 1);
+    }
+
+    private storeAlertResult(alertID: number, entityID: string, value: string) {
+        const storageName = `zmon:alerts:${alertID}:${entityID}`;
+
+        this.queueConn.set(storageName, value, console.log);
+    }
+
+    private deleteAlertResult(alertID: number, entityID: string) {
+        const storageName = `zmon:alerts:${alertID}:${entityID}`;
+
+        this.queueConn.del(storageName, console.log);
     }
 }