Merge branch 'improvement/CLDSRV-717-prom-cluster' into tmp/octopus/w/9.1/improvement/CLDSRV-717-prom-cluster

Author: bert-e

lib/Config.js

@@ -1032,6 +1032,10 @@ class Config extends EventEmitter {
                    'bad config: clusters must be a positive integer');
             this.clusters = config.clusters;
         }
+        if (process.env.S3BACKEND === 'mem') {
+            this.clusters = 1;
+        }
+        this.isCluster = this.clusters > 1;
 
         if (config.usEastBehavior !== undefined) {
             throw new Error('bad config: usEastBehavior key is deprecated. ' +

lib/server.js

@@ -67,7 +67,7 @@ class S3Server {
     constructor(config, worker) {
         this.config = config;
         this.worker = worker;
-        this.cluster = true;
+        this.cluster = config.isCluster;
         this.servers = [];
         http.globalAgent = new HttpAgent({
             keepAlive: true,
@@ -303,13 +303,23 @@ class S3Server {
             workerId: this.worker ? this.worker.id : undefined,
             workerPid: this.worker ? this.worker.process.pid : undefined,
         });
-        // Will close all servers, cause disconnect event on master and kill
+        // Will close all servers, cause disconnect event on primary and kill
         // worker process with 'SIGTERM'.
         if (this.worker) {
             this.worker.kill();
         }
     }
 
+    startServer(listenOn, port, routeRequest) {
+        if (listenOn.length > 0) {
+            listenOn.forEach(item => {
+                this._startServer(routeRequest.bind(this), item.port, item.ip);
+            });
+        } else if (port) {
+            this._startServer(routeRequest.bind(this), port);
+        }
+    }
+
     initiateStartup(log) {
         series([
             next => metadata.setup(next),
@@ -329,30 +339,14 @@ class S3Server {
             }
 
             // Start API server(s)
-            if (this.config.listenOn.length > 0) {
-                this.config.listenOn.forEach(item => {
-                    this._startServer(this.routeRequest.bind(this), item.port, item.ip);
-                });
-            } else if (this.config.port) {
-                this._startServer(this.routeRequest.bind(this), this.config.port);
-            }
+            this.startServer(this.config.listenOn, this.config.port, this.routeRequest);
 
             // Start internal API server(s)
-            if (this.config.internalListenOn.length > 0) {
-                this.config.internalListenOn.forEach(item => {
-                    this._startServer(this.internalRouteRequest.bind(this), item.port, item.ip);
-                });
-            } else if (this.config.internalPort) {
-                this._startServer(this.internalRouteRequest.bind(this), this.config.internalPort);
-            }
+            this.startServer(this.config.internalListenOn, this.config.internalPort, this.internalRouteRequest);
 
-            // Start metrics server(s)
-            if (this.config.metricsListenOn.length > 0) {
-                this.config.metricsListenOn.forEach(item => {
-                    this._startServer(this.routeAdminRequest.bind(this), item.port, item.ip);
-                });
-            } else {
-                this._startServer(this.routeAdminRequest.bind(this), this.config.metricsPort);
+            // Start metrics server(s) only if not cluster mode worker
+            if (!this.cluster && !this.worker) {
+                this.startServer(this.config.metricsListenOn, this.config.metricsPort, this.routeAdminRequest);
             }
 
             // Start quota service health checks
@@ -378,18 +372,13 @@ class S3Server {
 }
 
 function main() {
-    // TODO: change config to use workers prop. name for clarity
-    let workers = _config.clusters || 1;
-    if (process.env.S3BACKEND === 'mem') {
-        workers = 1;
-    }
-    this.cluster = workers > 1;
-    if (!this.cluster) {
+    const workers = _config.clusters;
+    if (!_config.isCluster) {
         process.env.REPORT_TOKEN = _config.reportToken;
         const server = new S3Server(_config);
         server.initiateStartup(logger.newRequestLogger());
     }
-    if (this.cluster && cluster.isMaster) {
+    if (_config.isCluster && cluster.isPrimary) {
         for (let n = 0; n < workers; n++) {
             const worker = cluster.fork();
             logger.info('new worker forked', {
@@ -430,12 +419,20 @@ function main() {
                 workerPid: worker.process.pid,
             });
         });
+
+        const metricServer = new S3Server(_config);
+        metricServer.startServer(_config.metricsListenOn,
+            _config.metricsPort, metricServer.routeAdminRequest);
     }
-    if (this.cluster && cluster.isWorker) {
+    if (_config.isCluster && cluster.isWorker) {
         const server = new S3Server(_config, cluster.worker);
         server.initiateStartup(logger.newRequestLogger());
     }
-    monitoringClient.collectDefaultMetrics({ timeout: 5000 });
+
+    // Avoid default metrics on cluster primary as it only aggregate workers
+    if (!_config.isCluster || cluster.isWorker) {
+        monitoringClient.collectDefaultMetrics({ timeout: 5000 });
+    }
 }
 
 module.exports = main;

lib/utilities/monitoringHandler.js

@@ -215,16 +215,27 @@ function writeResponse(res, error, results, cb) {
     });
 }
 
+const registry = config.isCluster ? new client.AggregatorRegistry() : client.register;
+const getMetrics = config.isCluster ?
+    registry.clusterMetrics.bind(registry) : registry.metrics.bind(registry);
 
 async function routeHandler(req, res, cb) {
     if (req.method !== 'GET') {
         return cb(errors.BadRequest, []);
     }
-    const promMetrics = await client.register.metrics();
+    let promMetrics;
+    try {
+        // Catch timeout on IPC between worker and primary
+        // prom-client has a 5s hardcoded timeout
+        promMetrics = await getMetrics();
+    } catch (err) {
+        return cb(err, { message: err.toString() });
+    }
+
     const contentLen = Buffer.byteLength(promMetrics, 'utf8');
     res.writeHead(200, {
         'Content-Length': contentLen,
-        'Content-Type': client.register.contentType,
+        'Content-Type': registry.contentType,
     });
     res.end(promMetrics);
     return undefined;

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@zenko/cloudserver",
-  "version": "9.0.21",
+  "version": "9.0.22",
   "description": "Zenko CloudServer, an open-source Node.js implementation of a server handling the Amazon S3 protocol",
   "main": "index.js",
   "engines": {