updating consumer

consumer/src/Metrics.ts

@@ -0,0 +1,48 @@
+
+import { Router } from 'express';
+import { RedisStreamService } from './shared/services/RedisStreamService';
+import { requireEnv } from './shared/utils/requireEnv';
+
+const stream_name = requireEnv('STREAM_NAME');
+
+export class MetricsManager {
+
+    private static processTime = new Map<string, number[]>();
+
+    static onProcess(id: string, time: number) {
+        const target = this.processTime.get(id);
+        if (!target) {
+            this.processTime.set(id, [time]);
+        } else {
+            target.push(time);
+            if (target.length > 1000) target.splice(0, target.length - 1000);
+        }
+    }
+
+    static get() {
+        return Array.from(this.processTime.entries());
+    }
+
+}
+
+export const metricsRouter = Router();
+
+metricsRouter.get('/queue', async (req, res) => {
+    try {
+        const size = await RedisStreamService.getQueueInfo(stream_name);
+        res.json({ size });
+    } catch (ex) {
+        console.error(ex);
+        res.status(500).json({ error: ex.message });
+    }
+})
+
+metricsRouter.get('/durations', async (req, res) => {
+    try {
+        const durations = MetricsManager.get();
+        res.json({ durations });
+    } catch (ex) {
+        console.error(ex);
+        res.status(500).json({ error: ex.message });
+    }
+})

consumer/src/index.ts

@@ -13,27 +13,19 @@ import express from 'express';
 
 import { ProjectLimitModel } from './shared/schema/project/ProjectsLimits';
 import { ProjectCountModel } from './shared/schema/project/ProjectsCounts';
+import { MetricsManager, metricsRouter } from './Metrics';
 
 
 const app = express();
 
-let durations: number[] = [];
-
-app.get('/status', async (req, res) => {
-    try {
-        return res.json({ status: 'ALIVE', durations })
-    } catch (ex) {
-        console.error(ex);
-        return res.setStatus(500).json({ error: ex.message });
-    }
-})
+app.use('/metrics', metricsRouter);
 
 app.listen(process.env.PORT);
 
 connectDatabase(requireEnv('MONGO_CONNECTION_STRING'));
 main();
 
-
+const CONSUMER_NAME = `CONSUMER_${process.env.NODE_APP_INSTANCE || 'DEFAULT'}`
 
 async function main() {
 
@@ -43,7 +35,7 @@ async function main() {
     const group_name = requireEnv('GROUP_NAME') as any; // Checks are inside "startReadingLoop"
 
     await RedisStreamService.startReadingLoop({
-        stream_name, group_name, consumer_name: `CONSUMER_${process.env.NODE_APP_INSTANCE || 'DEFAULT'}`
+        stream_name, group_name, consumer_name: CONSUMER_NAME
     }, processStreamEntry);
 
 }
@@ -73,18 +65,13 @@ async function processStreamEntry(data: Record<string, string>) {
             await process_visit(data, sessionHash);
         }
 
-        // console.log('Entry processed in', duration, 'ms');
-
     } catch (ex: any) {
         console.error('ERROR PROCESSING STREAM EVENT', ex.message);
     }
 
     const duration = Date.now() - start;
 
-    durations.push(duration);
-    if (durations.length > 1000) {
-        durations = durations.splice(500);
-    }
+    MetricsManager.onProcess(CONSUMER_NAME, duration);
 
 }
 

consumer/src/shared/services/RedisStreamService.ts

@@ -29,9 +29,20 @@ export class RedisStreamService {
         await this.client.connect();
     }
 
+
+    static async getQueueInfo(stream_name: string) {
+        try {
+            const size = await this.client.xLen(stream_name);
+            return size;
+        } catch (ex) {
+            console.error(ex);
+            return 0;
+        }
+    }
+
     static async readFromStream(stream_name: string, group_name: string, consumer_name: string, process_function: (content: Record<string, string>) => Promise<any>) {
 
-        const result: xReadGgroupResult = await this.client.xReadGroup(group_name, consumer_name, [{ key: stream_name, id: '>' }], { COUNT: 5, BLOCK: 10000 });
+        const result: xReadGgroupResult = await this.client.xReadGroup(group_name, consumer_name, [{ key: stream_name, id: '>' }], { COUNT: 5, BLOCK: 2000 });
 
         if (!result) {
             setTimeout(() => this.readFromStream(stream_name, group_name, consumer_name, process_function), 10);