Add support for monitoring watches

Author: skadefro

OpenFlow/src/Config.ts

@@ -111,6 +111,7 @@ export class Config {
         Config.nodered_limits_cpu = Config.getEnv("nodered_limits_cpu", ""); // 1000m = 1vCPU
         Config.prometheus_measure_nodeid = Config.parseBoolean(Config.getEnv("prometheus_measure_nodeid", "false"));
         Config.prometheus_measure_queued_messages = Config.parseBoolean(Config.getEnv("prometheus_measure_queued_messages", "false"));
+        Config.prometheus_measure__mongodb_watch = Config.parseBoolean(Config.getEnv("prometheus_measure__mongodb_watch", "false"));
         Config.validate_user_form = Config.getEnv("validate_user_form", "");
     }
     public static db: DatabaseConnection = null;
@@ -213,6 +214,7 @@ export class Config {
     public static nodered_limits_cpu: string = Config.getEnv("nodered_limits_cpu", ""); // 1000m = 1vCPU
     public static prometheus_measure_nodeid: boolean = Config.parseBoolean(Config.getEnv("prometheus_measure_nodeid", "false"));
     public static prometheus_measure_queued_messages: boolean = Config.parseBoolean(Config.getEnv("prometheus_measure_queued_messages", "false"));
+    public static prometheus_measure__mongodb_watch: boolean = Config.parseBoolean(Config.getEnv("prometheus_measure__mongodb_watch", "false"));
     public static validate_user_form: string = Config.getEnv("validate_user_form", "");
 
     public static baseurl(): string {

OpenFlow/src/WebSocketServer.ts

@@ -53,6 +53,11 @@ export class WebSocketServer {
         help: 'Total number messages waiting on reply from client',
         labelNames: ["clientid"]
     })
+    public static mongodb_watch_count = new client.Gauge({
+        name: 'mongodb_watch_count',
+        help: 'Total number af steams  watching for changes',
+        labelNames: ["agent", "clientid"]
+    })
     public static update_message_queue_count(cli: WebSocketServerClient) {
         if (!Config.prometheus_measure_queued_messages) return;
         const result: any = {};
@@ -76,6 +81,16 @@ export class WebSocketServer {
             WebSocketServer.message_queue_count.labels(cli.id, key).set(result[key]);
         });
     }
+    public static update_mongodb_watch_count(cli: WebSocketServerClient) {
+        if (!Config.prometheus_measure__mongodb_watch) return;
+        const result: any = {};
+        let total: number = 0;
+        WebSocketServer.mongodb_watch_count.reset();
+        for (let i = WebSocketServer._clients.length - 1; i >= 0; i--) {
+            const cli: WebSocketServerClient = WebSocketServer._clients[i];
+            WebSocketServer.mongodb_watch_count.labels(cli.clientagent, cli.id).set(cli.streamcount());
+        }
+    }
     static configure(logger: winston.Logger, server: http.Server, register: client.Registry): void {
         this._clients = [];
         this._logger = logger;
@@ -94,6 +109,7 @@ export class WebSocketServer {
         if (!NoderedUtil.IsNullUndefinded(register)) register.registerMetric(WebSocketServer.websocket_rate_limit);
         if (!NoderedUtil.IsNullUndefinded(register)) register.registerMetric(WebSocketServer.websocket_messages);
         if (!NoderedUtil.IsNullUndefinded(register)) register.registerMetric(WebSocketServer.message_queue_count);
+        if (!NoderedUtil.IsNullUndefinded(register)) register.registerMetric(WebSocketServer.mongodb_watch_count);
         if (!NoderedUtil.IsNullUndefinded(register)) register.registerMetric(DatabaseConnection.mongodb_query);
         if (!NoderedUtil.IsNullUndefinded(register)) register.registerMetric(DatabaseConnection.mongodb_query_count);
         if (!NoderedUtil.IsNullUndefinded(register)) register.registerMetric(DatabaseConnection.mongodb_aggregate);

OpenFlow/src/WebSocketServerClient.ts

@@ -180,6 +180,7 @@ export class WebSocketServerClient {
                 this._logger.error("WebSocketclient::Close " + error);
             }
         }
+        WebSocketServer.update_mongodb_watch_count(this);
     }
     public async CloseConsumer(queuename: string): Promise<void> {
         var old = this._queues.length;
@@ -444,6 +445,7 @@ export class WebSocketServerClient {
                 }
             }
         }
+        WebSocketServer.update_mongodb_watch_count(this);
     }
     async UnWatch(id: string, jwt: string): Promise<void> {
         this.CloseStream(id);
@@ -474,6 +476,7 @@ export class WebSocketServerClient {
                     this._logger.error("WebSocketclient::Watch::changeListener " + error + " " + this.id + "/" + this.clientagent);
                 }
             }, options);
+            WebSocketServer.update_mongodb_watch_count(this);
             return stream.id;
         } catch (error) {
             this._logger.error("WebSocketclient::Watch " + error + " " + this.id + "/" + this.clientagent);

OpenFlowNodeRED/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@openiap/nodered",
-  "version": "1.1.169",
+  "version": "1.1.171",
   "description": "Simple wrapper around NodeRed, RabbitMQ and MongoDB to support a more scaleable NodeRed implementation.\r Also the \"backend\" for [OpenRPA](https://github.com/skadefro/OpenRPA)",
   "main": "index.js",
   "scripts": {

VERSION

@@ -1 +1 @@
-1.1.170
+1.1.171

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@openiap/openflow",
-  "version": "1.1.169",
+  "version": "1.1.171",
   "description": "Simple wrapper around NodeRed, RabbitMQ and MongoDB to support a more scaleable NodeRed implementation.\r Also the \"backend\" for [OpenRPA](https://github.com/skadefro/OpenRPA)",
   "main": "index.js",
   "scripts": {