Merge pull request #45 from CS3219-AY2425S1/feat/d4-queue-status-logging

add logging for queue status for D4

Author: SeeuSim

backend/matching/src/config.ts

@@ -12,3 +12,5 @@ export const DB_URL = `redis://${DB_USERNAME}:${DB_PASSWORD}@${DB_HOSTNAME}:${DB
 // export const DB_URL = `redis://${DB_HOSTNAME}:${DB_PORT}`;
 
 export const NODE_ENV = process.env.NODE_ENV;
+
+export const IS_MILESTONE_D4 = true;

backend/matching/src/controllers/match-request.ts

@@ -1,9 +1,10 @@
 import type { Request, Response } from 'express';
 import { StatusCodes } from 'http-status-codes';
 
-import { client } from '@/lib/db';
+import { client, logQueueStatus } from '@/lib/db';
 import type { IRedisClient, IRequestMatchPayload } from '@/types';
 import { createNotifSocket, queueingService } from '@/services';
+import { logger } from '@/lib/utils';
 
 let redisClient: IRedisClient;
 export const matchRequestController = async (req: Request, res: Response) => {
@@ -39,4 +40,6 @@ export const matchRequestController = async (req: Request, res: Response) => {
     socketPort: socketRoom,
     timestamp,
   });
+
+  logQueueStatus(logger, redisClient, `Queue Status Before Matching: <PLACEHOLDER>`);
 };

backend/matching/src/lib/db/index.ts

@@ -1 +1,2 @@
 export * from './client';
+export * from './log-queue-status';

backend/matching/src/lib/db/log-queue-status.ts

@@ -0,0 +1,23 @@
+import { IS_MILESTONE_D4 } from '@/config';
+
+import { client } from './client';
+import { STREAM_NAME } from './constants';
+
+export const getQueueStatusLog = async (redisClient: typeof client) => {
+  const queueStatus = await redisClient.xRange(STREAM_NAME, '-', '+');
+  const messages = queueStatus.map((v) => v.message);
+  return JSON.stringify(messages);
+};
+
+export const logQueueStatus = async (
+  // eslint-disable-next-line
+  logger: { info: (...m: any[]) => void },
+  redisClient: typeof client,
+  message: string
+) => {
+  if (!IS_MILESTONE_D4) {
+    return;
+  }
+  const queueStatusLog = await getQueueStatusLog(redisClient);
+  logger.info(message.replace('<PLACEHOLDER>', queueStatusLog));
+};

backend/matching/src/workers/cleaner.ts

@@ -1,4 +1,4 @@
-import { client } from '@/lib/db';
+import { client, logQueueStatus } from '@/lib/db';
 import { STREAM_CLEANER, STREAM_GROUP, STREAM_NAME } from '@/lib/db/constants';
 import { decodePoolTicket, getPoolKey } from '@/lib/utils';
 import { MATCH_SVC_EVENT } from '@/ws';
@@ -64,6 +64,8 @@ async function clean() {
       sendNotif([socketRoom], MATCH_SVC_EVENT.FAILED);
       sendNotif([socketRoom], MATCH_SVC_EVENT.DISCONNECT);
     }
+
+    await logQueueStatus(logger, redisClient, `Queue Status after Expiring Request: <PLACEHOLDER>`);
   }
 }
 

backend/matching/src/workers/matcher.ts

@@ -1,4 +1,4 @@
-import { client } from '@/lib/db';
+import { client, logQueueStatus } from '@/lib/db';
 import { POOL_INDEX, STREAM_GROUP, STREAM_NAME, STREAM_WORKER } from '@/lib/db/constants';
 import { decodePoolTicket, getPoolKey, getStreamId } from '@/lib/utils';
 import { getMatchItems } from '@/services';
@@ -69,6 +69,8 @@ async function processMatch(
       const { ...matchItems } = getMatchItems();
       sendNotif([requestorSocketPort, matchedSocketPort], MATCH_SVC_EVENT.SUCCESS, matchItems);
       sendNotif([requestorSocketPort, matchedSocketPort], MATCH_SVC_EVENT.DISCONNECT);
+
+      await logQueueStatus(logger, redisClient, `Queue Status After Matching: <PLACEHOLDER>`);
       return true;
     }
   }