Improve housekeeping

Author: skadefro

OpenFlow/src/Messages/Message.ts

@@ -4331,7 +4331,28 @@ export class Message {
         m.data = JSON.stringify(l);
         cli.Send(m);
     }
+    public static lastHouseKeeping: Date = null;
+    public static ReadyForHousekeeping(): boolean {
+        const date = new Date();
+        const a: number = (date as any) - (Message.lastHouseKeeping as any);
+        const diffminutes = a / (1000 * 60);
+        // const diffhours = a / (1000 * 60 * 60);
+        if (diffminutes < 60) return false;
+        return true;
+    }
     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);
+        }
+        if (!Message.ReadyForHousekeeping()) {
+            const date = new Date();
+            const a: number = (date as any) - (Message.lastHouseKeeping as any);
+            const diffminutes = a / (1000 * 60);
+            Logger.instanse.debug("[housekeeping] Skipping housekeeping, to early for next run (ran " + diffminutes + " minutes ago)");
+            return;
+        }
+        Message.lastHouseKeeping = new Date();
         const span: Span = Logger.otel.startSubSpan("message.QueueMessage", parent);
         try {
             if (!skipNodered) await this.GetNoderedInstance(span)

OpenFlow/src/amqpwrapper.ts

@@ -7,6 +7,7 @@ import { Span } from "@opentelemetry/api";
 import { Logger } from "./Logger";
 import events = require("events");
 import { Auth } from "./Auth";
+import { Message } from "./Messages/Message";
 type QueueOnMessage = (msg: string, options: QueueMessageOptions, ack: any, done: any) => void;
 interface IHashTable<T> {
     [key: string]: T;
@@ -559,6 +560,10 @@ export class amqpwrapper extends events.EventEmitter {
             done();
         }, undefined);
     }
+    IsMyconsumerTag(consumerTag: string) {
+        var q = this.queues.filter(q => q.consumerTag == consumerTag);
+        return q.length != 0;
+    }
     async AddOFExchange() {
         if (!Config.enable_openflow_amqp) return;
         await this.AddExchangeConsumer("openflow", "fanout", "", null, null, async (msg: any, options: QueueMessageOptions, ack: any, done: any) => {
@@ -574,6 +579,22 @@ export class amqpwrapper extends events.EventEmitter {
                     case "clearcache":
                         Auth.clearCache();
                         break;
+                    case "housekeeping":
+                        if (this.IsMyconsumerTag(options.consumerTag)) {
+                            ack();
+                            done();
+                            try {
+                                var dt = new Date(new Date().toISOString());
+                                var msg2 = new Message(); msg2.jwt = Crypt.rootToken();
+                                var skipUpdateUsage: boolean = !(dt.getHours() == 1 || dt.getHours() == 13);
+                                await msg2.Housekeeping(false, skipUpdateUsage, skipUpdateUsage, null);
+                            } catch (error) {
+                            }
+                            return;
+                        }
+                        console.log("Reset house keeping, someone else is doing it now");
+                        Message.lastHouseKeeping = new Date();
+                        break;
                     case "shutdown":
                         try {
                             await Config.db.shutdown();

OpenFlow/src/index.ts

@@ -189,24 +189,38 @@ async function initDatabase(): Promise<boolean> {
         Logger.otel.endSpan(span);
 
         if (Config.auto_hourly_housekeeping) {
+            // Every 15 minutes, give and take a few minutes, send out a message to do house keeping, if ready
+            const randomNum = (Math.floor(Math.random() * 100) + 1);
+            console.log("Housekeeping every 15 minutes plus " + randomNum + " seconds");
             housekeeping = setInterval(async () => {
-                try {
-                    var dt = new Date(new Date().toISOString());
-                    var msg = new Message(); msg.jwt = Crypt.rootToken();
-                    var skipUpdateUsage: boolean = !(dt.getHours() == 1 || dt.getHours() == 13);
-                    await msg.Housekeeping(false, skipUpdateUsage, skipUpdateUsage, null);
-                } catch (error) {
+                if (Config.enable_openflow_amqp) {
+                    if (Config.enable_openflow_amqp) {
+                        if (!Message.ReadyForHousekeeping()) {
+                            return;
+                        }
+                        amqpwrapper.Instance().send("openflow", "", { "command": "housekeeping" }, 20000, null, "", 1);
+                    }
+                } else {
+                    try {
+                        var dt = new Date(new Date().toISOString());
+                        var msg = new Message(); msg.jwt = Crypt.rootToken();
+                        var skipUpdateUsage: boolean = !(dt.getHours() == 1 || dt.getHours() == 13);
+                        await msg.Housekeeping(false, skipUpdateUsage, skipUpdateUsage, null);
+                    } catch (error) {
+                    }
                 }
-            }, 3600000);
+            }, (15 * 60 * 1000) + (randomNum * 1000));
+            // If I'm first and noone else has run it, lets trigger it now
+            const randomNum2 = (Math.floor(Math.random() * 10) + 1);
+            console.log("Trigger first Housekeeping in " + randomNum2 + " seconds");
             setTimeout(async () => {
-                var dt = new Date(new Date().toISOString());
-                var msg = new Message(); msg.jwt = Crypt.rootToken();
-                var skipUpdateUsage: boolean = !(dt.getHours() == 1 || dt.getHours() == 13);
-                var skipUpdateUserSize: boolean = !(dt.getHours() == 1 || dt.getHours() == 13);
-                if (Config.housekeeping_update_usage_hourly) skipUpdateUsage = false;
-                if (Config.housekeeping_update_usersize_hourly) skipUpdateUserSize = false;
-                await msg.Housekeeping(false, skipUpdateUsage, skipUpdateUserSize, null);
-            }, 5000);
+                if (Config.enable_openflow_amqp) {
+                    if (!Message.ReadyForHousekeeping()) {
+                        return;
+                    }
+                    amqpwrapper.Instance().send("openflow", "", { "command": "housekeeping" }, 20000, null, "", 1);
+                }
+            }, randomNum2 * 1000);
         }
         return true;
     } catch (error) {