Merge pull request #200 from skadefro/master

Close 1.4.4

OpenFlow/src/Config.ts

@@ -59,6 +59,7 @@ export class Config {
         Config.auto_hourly_housekeeping = Config.parseBoolean(Config.getEnv("auto_hourly_housekeeping", "false"));
         Config.housekeeping_update_usage_hourly = Config.parseBoolean(Config.getEnv("housekeeping_update_usage_hourly", "false"));
         Config.housekeeping_update_usersize_hourly = Config.parseBoolean(Config.getEnv("housekeeping_update_usersize_hourly", "true"));
+        Config.housekeeping_skip_collections = Config.getEnv("housekeeping_skip_collections", "");
         Config.workitem_queue_monitoring_enabled = Config.parseBoolean(Config.getEnv("workitem_queue_monitoring_enabled", "true"));
         Config.workitem_queue_monitoring_interval = parseInt(Config.getEnv("workitem_queue_monitoring_interval", (30 * 1000).toString())); // 30 sec
 
@@ -237,6 +238,7 @@ export class Config {
     public static auto_hourly_housekeeping: boolean = Config.parseBoolean(Config.getEnv("auto_hourly_housekeeping", "true"));
     public static housekeeping_update_usage_hourly: boolean = Config.parseBoolean(Config.getEnv("housekeeping_update_usage_hourly", "false"));
     public static housekeeping_update_usersize_hourly: boolean = Config.parseBoolean(Config.getEnv("housekeeping_update_usersize_hourly", "true"));
+    public static housekeeping_skip_collections: string = Config.getEnv("housekeeping_skip_collections", "");
     public static workitem_queue_monitoring_enabled: boolean = Config.parseBoolean(Config.getEnv("workitem_queue_monitoring_enabled", "true"));
     public static workitem_queue_monitoring_interval: number = parseInt(Config.getEnv("workitem_queue_monitoring_interval", (30 * 1000).toString())); // 30 sec
 

OpenFlow/src/DatabaseConnection.ts

@@ -260,11 +260,15 @@ export class DatabaseConnection extends events.EventEmitter {
                     }
                     if (!NoderedUtil.IsNullEmpty(wiq.robotqueue) && !NoderedUtil.IsNullEmpty(wiq.workflowid)) {
                         Logger.instanse.verbose("[workitems] Send invoke message to robot queue " + wiq.workflowid);
-                        await amqpwrapper.Instance().send(null, wiq.robotqueue, payload, 5000, null, null, 2);
+                        let expiration = (Config.amqp_requeue_time / 2, 10) | 0;
+                        if (expiration < 500) expiration = 500;
+                        await amqpwrapper.Instance().send(null, wiq.robotqueue, payload, expiration, null, null, 2);
                     }
                     if (!NoderedUtil.IsNullEmpty(wiq.amqpqueue)) {
                         Logger.instanse.verbose("[workitems] Send invoke message to amqp queue " + wiq.amqpqueue);
-                        await amqpwrapper.Instance().send(null, wiq.amqpqueue, payload, 5000, null, null, 2);
+                        let expiration = (Config.amqp_requeue_time / 2, 10) | 0;
+                        if (expiration < 500) expiration = 500;
+                        await amqpwrapper.Instance().send(null, wiq.amqpqueue, payload, expiration, null, null, 2);
                     }
                 }
             }

OpenFlow/src/Messages/Message.ts

@@ -138,7 +138,7 @@ export class Message {
                     await this.GetNoderedInstance(span);
                     break;
                 case "housekeeping":
-                    await this.Housekeeping(false, false, false, span);
+                    await this.Housekeeping(span);
                     break;
                 case "updateworkitemqueue":
                     await this.UpdateWorkitemQueue(span);
@@ -656,10 +656,7 @@ export class Message {
                     if (Config.enable_openflow_amqp) {
                         cli.Send(await QueueClient.SendForProcessing(this, this.priority));
                     } else {
-                        // await this.Housekeeping(false, false, false, span);
-                        Message.lastHouseKeeping = null;
-                        var msg = JSON.parse(this.data);
-                        await this.Housekeeping(msg.skipnodered, msg.skipcalculatesize, msg.skipupdateusersize, span);
+                        await this.Housekeeping(span);
                         cli.Send(this);
                     }
                     break;
@@ -3559,7 +3556,33 @@ export class Message {
         if (diffminutes < 60) return false;
         return true;
     }
-    public async Housekeeping(skipNodered: boolean, skipCalculateSize: boolean, skipUpdateUserSize: boolean, parent: Span): Promise<void> {
+    private async Housekeeping(parent: Span): Promise<void> {
+        this.Reply();
+        const span: Span = Logger.otel.startSubSpan("message.GetNoderedInstance", parent);
+        let msg: any;
+        try {
+            msg = JSON.parse(this.data);
+            Message.lastHouseKeeping = null;
+            if (NoderedUtil.IsNullEmpty(msg.skipnodered)) msg.skipnodered = false;
+            if (NoderedUtil.IsNullEmpty(msg.skipcalculatesize)) msg.skipcalculatesize = false;
+            if (NoderedUtil.IsNullEmpty(msg.skipupdateusersize)) msg.skipupdateusersize = false;
+            await this._Housekeeping(msg.skipnodered, msg.skipcalculatesize, msg.skipupdateusersize, span);
+        } catch (error) {
+            span?.recordException(error);
+            this.data = "";
+            await handleError(null, error);
+            if (msg !== null && msg !== undefined) msg.error = error.message ? error.message : error;
+        }
+        try {
+            this.data = JSON.stringify(msg);
+        } catch (error) {
+            span?.recordException(error);
+            this.data = "";
+            await handleError(null, error);
+        }
+        Logger.otel.endSpan(span);
+    }
+    public async _Housekeeping(skipNodered: boolean, skipCalculateSize: boolean, skipUpdateUserSize: boolean, parent: Span): Promise<void> {
         if (Message.lastHouseKeeping == null) {
             Message.lastHouseKeeping = new Date();
             Message.lastHouseKeeping.setDate(Message.lastHouseKeeping.getDate() - 1);
@@ -3764,8 +3787,15 @@ export class Message {
                 collections = collections.filter(x => x.name.indexOf("system.") === -1);
                 let totalusage = 0;
                 let index = 0;
+                let skip_collections = [];
+                if (!NoderedUtil.IsNullEmpty(Config.housekeeping_skip_collections)) skip_collections = Config.housekeeping_skip_collections.split(",")
                 for (let col of collections) {
                     if (col.name == "fs.chunks") continue;
+                    if (skip_collections.indexOf(col.name) > -1) {
+                        Logger.instanse.debug("[housekeeping][" + col.name + "] skipped due to housekeeping_skip_collections setting");
+                        continue;
+                    }
+
                     index++;
                     let aggregates: any = [
                         {

OpenFlow/src/index.ts

@@ -59,7 +59,7 @@ function doHouseKeeping() {
     var msg2 = new Message(); msg2.jwt = Crypt.rootToken();
     var h = dt.getHours();
     var skipUpdateUsage: boolean = !(dt.getHours() == 1 || dt.getHours() == 13);
-    msg2.Housekeeping(false, skipUpdateUsage, skipUpdateUsage, null).catch((error) => Logger.instanse.error(error));
+    msg2._Housekeeping(false, skipUpdateUsage, skipUpdateUsage, null).catch((error) => Logger.instanse.error(error));
 
     // var dt = new Date(new Date().toISOString());
     // var msg = new Message(); msg.jwt = Crypt.rootToken();

OpenFlowNodeRED/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@openiap/nodered",
-  "version": "1.4.3",
+  "version": "1.4.4",
   "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": {

OpenFlowNodeRED/src/nodered/nodes/api.html

@@ -1796,8 +1796,67 @@
 </script>
 
 
+<script type="text/x-red" data-template-name="api drop collection">
+    <div class="form-row">
+        <label><i class="fa fa-list"></i> Collection name</label>
+        <input type="hidden" id="node-input-collectionametype">
+        <input style="width:70%" type="text" id="node-input-collectioname">
+    </div>
+    <div class="form-row">
+        <label><i class="fa fa-tag"></i> Name</label>
+        <input type="text" id="node-input-name" placeholder="Name">
+    </div>
+</script>
+<script type="text/x-red" data-help-name="api drop collection">
+    <p>drop collection in the database <br>
+    </p>
+</script>
+<script type="text/javascript">
+    // https://fontawesome.com/v4.7/icon/list
+    RED.nodes.registerType('api drop collection', {
+        category: 'api',
+        color: "#E9967A",
+        paletteLabel: 'drop collection',
+        icon: "font-awesome/fa-list",
+        defaults: {
+            name: { value: "" },
+            collectioname: { value: "collectionname", validate: validate("collectionametype"), required: true },
+            collectionametype: { value: "msg", required: true },
+        },
+        inputs: 1,
+        outputs: 1,
+        label: function () {
+            return this.name || "drop collection";
+        },
+        labelStyle: function () {
+            return this.name ? "node_label_italic" : "";
+        },
+        oneditprepare: function () {
+            if (this.collectionametype === null) $("#node-input-collectionametype").val('msg');
+            if (this.collectionametype === 'val') $("#node-input-collectionametype").val('str');
+            $("#node-input-collectioname").typedInput({
+                default: 'msg',
+                typeField: $("#node-input-collectionametype"),
+                types: ['msg', 'str']
+            });
+        }
+    });
+</script>
+
 
 <script type="text/x-red" data-template-name="api housekeeping">
+    <div class="form-row">
+        <label><i class="fa fa-tag"></i> Skip Nodered</label>
+        <input type="checkbox" id="node-input-skipnodered" style="width: auto;">
+    </div>
+    <div class="form-row">
+        <label><i class="fa fa-tag"></i> Skip Calculate Size</label>
+        <input type="checkbox" id="node-input-skipcalculatesize" style="width: auto;">
+    </div>
+    <div class="form-row">
+        <label><i class="fa fa-tag"></i> Skip Update User Size</label>
+        <input type="checkbox" id="node-input-skipupdateusersize" style="width: auto;">
+    </div>
     <div class="form-row">
         <label><i class="fa fa-tag"></i> Name</label>
         <input type="text" id="node-input-name" placeholder="Name">
@@ -1816,6 +1875,9 @@
         icon: "font-awesome/fa-list",
         defaults: {
             name: { value: "" },
+            skipnodered: { value: false },
+            skipcalculatesize: { value: false },
+            skipupdateusersize: { value: false },
         },
         inputs: 1,
         outputs: 1,

OpenFlowNodeRED/src/nodered/nodes/api.ts

@@ -30,6 +30,8 @@ export = function (RED: Red) {
     RED.nodes.registerType("api upload file", api.upload_file);
 
     RED.nodes.registerType("api list collections", api.list_collections);
+    RED.nodes.registerType("api drop collection", api.drop_collection);
+
     RED.nodes.registerType("api housekeeping", api.housekeeping);
 
     RED.httpAdmin.get("/api_roles", RED.auth.needsPermission('serial.read'), api.get_api_roles);

OpenFlowNodeRED/src/nodered/nodes/api_nodes.ts

@@ -1430,10 +1430,43 @@ export class list_collections {
 
 
 
+export interface Idrop_collection {
+    name: string;
+    collectioname: string;
+}
+export class drop_collection {
+    public node: Red = null;
+    public name: string;
+    constructor(public config: Idrop_collection) {
+        RED.nodes.createNode(this, config);
+        this.node = this;
+        this.name = config.name;
+        this.node.on("input", this.oninput);
+        this.node.on("close", this.onclose);
+    }
+    async oninput(msg: any) {
+        try {
+            this.node.status({});
+            let priority: number = 1;
+            if (!NoderedUtil.IsNullEmpty(msg.priority)) { priority = msg.priority; }
+            const collectionname: any = await Util.EvaluateNodeProperty<string>(this, msg, "collectioname");
+            await NoderedUtil.DropCollection({ collectionname, priority });
+            this.node.send(msg);
+            this.node.status({});
+        } catch (error) {
+            NoderedUtil.HandleError(this, error, msg);
+        }
+    }
+    onclose() {
+    }
+}
+
 
 export interface Ihousekeeping {
     name: string;
-    results: string;
+    skipnodered: boolean
+    skipcalculatesize: boolean;
+    skipupdateusersize: boolean;
 }
 export class housekeeping {
     public node: Red = null;
@@ -1450,8 +1483,10 @@ export class housekeeping {
             let priority: number = 1;
             if (!NoderedUtil.IsNullEmpty(msg.priority)) { priority = msg.priority; }
 
+            const { skipnodered, skipcalculatesize, skipupdateusersize } = this.config;
+
             this.node.status({ fill: "blue", shape: "dot", text: "Running house keeping" });
-            await NoderedUtil.HouseKeeping({ jwt: msg.jwt, priority });
+            await NoderedUtil.HouseKeeping({ skipnodered, skipcalculatesize, skipupdateusersize, jwt: msg.jwt, priority });
             this.node.send(msg);
             this.node.status({ fill: "green", shape: "dot", text: "Complete" });
         } catch (error) {

VERSION

@@ -1 +1 @@
-1.4.3
+1.4.4

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@openiap/openflow",
-  "version": "1.4.3",
+  "version": "1.4.4",
   "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": {