feat: 🎸 add health check

package.json

@@ -36,6 +36,9 @@
     "sleepjs": "3.0.1",
     "semver": "7.3.5"
   },
+  "optionalDependencies": { 
+      "fsevents": "2.1.2"
+  },
   "scripts": {
     "release:version": "lerna version --exact --conventional-commits",
     "release:publish": "lerna publish from-package",

packages/core/package.json

@@ -19,6 +19,7 @@
     "filename-regex": "2.0.1",
     "filtrex": "1.0.0",
     "from": "0.1.7",
+    "get-stream": "6.0.1",
     "global-modules": "2.0.0",
     "http-shutdown": "1.2.2",
     "load-json-file": "6.2.0",

packages/core/src/server/index.js

@@ -8,6 +8,7 @@ import debug from 'debug';
 import knownPipeline from './knownPipeline';
 import unknownPipeline from './unknownPipeline';
 import serverInformation from './serverInformation';
+import serverHealth from './serverHealth';
 import errorHandler from './errorHandler';
 import settings from '../settings';
 import { RX_FILENAME } from '../constants';
@@ -17,6 +18,7 @@ import {
     httpConnectionOpen,
     httpRequestDurationMicroseconds,
     aggregatorRegistry,
+    changeState,
 } from './metrics';
 
 function isPipeline() {
@@ -28,6 +30,10 @@ function methodMatch(values) {
     return (values.indexOf(this.method) !== -1);
 }
 
+function routeMatch(values) {
+    return (values.indexOf(this.pathName) !== -1);
+}
+
 const signals = ['SIGINT', 'SIGTERM'];
 
 function createServer(ezs, serverPort, serverPath, workerId) {
@@ -38,12 +44,14 @@ function createServer(ezs, serverPort, serverPath, workerId) {
         request.serverPath = serverPath;
         request.urlParsed = parse(request.url, true);
         request.pathName = request.urlParsed.pathname;
+        request.routeMatch = routeMatch;
         request.methodMatch = methodMatch;
         request.isPipeline = isPipeline;
         const stopTimer = httpRequestDurationMicroseconds.startTimer();
         eos(response, () => stopTimer());
         next();
     });
+    app.use(serverHealth(ezs));
     app.use(metrics(ezs));
     app.use(serverInformation(ezs));
     app.use(unknownPipeline(ezs));
@@ -73,9 +81,12 @@ function createServer(ezs, serverPort, serverPath, workerId) {
         });
     });
     signals.forEach((signal) => process.on(signal, () => {
-        debug('ezs')(`Signal received, stoping server with PID ${process.pid}`);
+        changeState('Stopping');
+        debug('ezs')(`Signal received, stopping server with PID ${process.pid}`);
         server.shutdown(() => process.exit(0));
     }));
+    process.on('beforeExit', () => changeState('Stopped'));
+    changeState('Started');
     debug('ezs')(`Server starting with PID ${process.pid} and listening on port ${serverPort}`);
     return server;
 }

packages/core/src/server/metrics.js

@@ -7,9 +7,48 @@ import {
     AggregatorRegistry,
     collectDefaultMetrics,
 } from 'prom-client';
+import getStream from 'get-stream';
 import settings from '../settings';
 
+const metricsRootServer = {
+    hostname: '0.0.0.0',
+    port: settings.port + 1,
+    path: '/',
+    method: 'GET',
+    headers: {
+        'Content-Type': 'text/plain',
+    }
+};
+
+
 
+export const ezsServerState = new Gauge({
+    name: 'ezs_server_state',
+    help: 'Number of different states for each server',
+    labelNames: ['pid'],
+});
+
+const states = ['Starting', 'Started', 'Stopping', 'Stopped'];
+export const changeState = (stateLabel) => {
+    const stateIndex = states.indexOf(stateLabel);
+    ezsServerState.labels(process.pid).set(stateIndex);
+};
+
+export const getSlashMetricsStream = () => new Promise((resolve, reject) => {
+    http.request(metricsRootServer, resolve).on('error', reject).end();
+});
+
+export const getState = async () => {
+    try {
+        const metricsStream = await getSlashMetricsStream();
+        const metricsString = await getStream(metricsStream);
+        return metricsString;
+    } catch(e) {
+        console.error(e);
+        return '';
+    }
+}
+changeState('Starting');
 
 export const aggregatorRegistry = new AggregatorRegistry();
 
@@ -79,12 +118,7 @@ export const ezsStreamDurationMicroseconds = new Histogram({
 
 let collected = false;
 export const metrics = () => (request, response, next) => {
-    const {
-        port,
-        metricsEnable,
-    } = settings;
-
-    if (!metricsEnable) {
+    if (!settings.metricsEnable) {
         return next();
     }
     if (!collected) {
@@ -107,17 +141,7 @@ export const metrics = () => (request, response, next) => {
         }).catch(next);
     }
 
-    const options = {
-        hostname: '0.0.0.0',
-        port: port + 1,
-        path: '/',
-        method: 'GET',
-        headers: {
-            'Content-Type': 'text/plain',
-        }
-    };
-
-    request.pipe(http.request(options, (source) => source.pipe(response)));
+    request.pipe(http.request(metricsRootServer, (source) => source.pipe(response)));
     return true;
 };
 

packages/core/src/server/serverHealth.js

@@ -0,0 +1,46 @@
+import debug from 'debug';
+import settings from '../settings';
+import {
+    getState,
+} from './metrics';
+
+
+const serverHealth =  () => (request, response, next) => {
+    if (!settings.metricsEnable) {
+        return next();
+    }
+    if (!request.methodMatch(['GET', 'OPTIONS', 'HEAD']) || !request.routeMatch(['/live', '/ready'])) {
+        return next();
+    }
+    request.catched = true;
+    debug('ezs')(`Create middleware 'serverHealth' for ${request.method} ${request.pathName}`);
+
+    return getState()
+        .then(
+            (metrics) => {
+                const states = metrics
+                    .split('\n')
+                    .filter(line => (line.search(/ezs_server_state/) === 0))
+                    .map(line => parseInt(line.split(' ').pop(), 10));
+                const statusCode = states.every(x => (x === 1)) ? 200 : 500;
+                const responseObject = {
+                    statusCode,
+                    states,
+                };
+                const responseBody = JSON.stringify(responseObject);
+                const responseHeaders = {
+                    'Access-Control-Allow-Origin': '*',
+                    'Access-Control-Allow-Methods': 'GET, OPTIONS',
+                    'Access-Control-Allow-Headers': '*',
+                    'Content-Type': 'application/json',
+                    'Content-Length': Buffer.byteLength(responseBody),
+                };
+                response.writeHead(statusCode, responseHeaders);
+                response.write(responseBody);
+                response.end();
+                return next();
+            }).catch(next);
+
+};
+
+export default serverHealth;

packages/core/src/server/serverInformation.js

@@ -177,7 +177,7 @@ const collectAll = async (ezs, request) => {
 };
 
 const serverInformation =  (ezs) => (request, response, next) => {
-    if (!request.methodMatch(['GET', 'OPTIONS', 'HEAD']) || request.pathName !== '/') {
+    if (!request.methodMatch(['GET', 'OPTIONS', 'HEAD']) || !request.routeMatch(['/'])) {
         return next();
     }
     request.catched = true;

packages/core/test/metrics.js

@@ -73,6 +73,18 @@ describe(' through server(s)', () => {
         });
     });
 
+    describe('health check #1', () => {
+        it('call /ready', (done) => {
+            fetch('http://127.0.0.1:33340/ready ')
+                .then((res) => res.json())
+                .then((json) => {
+                    assert(json.statusCode, 200);
+                    done();
+                });
+        });
+    });
+
+
 
 });