added additional logging

src/routes/queue-route.ts

@@ -33,6 +33,7 @@ router.post("/log", async (req: Request, res: Response) => {
         data: log,
       }))
     );
+    console.log("Received %d Logs", logs.length);
     res.json({
       success: true,
     });

src/utils/logstash-client.ts

@@ -1,11 +1,16 @@
 import Logstash from "./logstash";
 
 let errorLogger = (error: Error) => {
-  console.log(error);
+  console.error(error);
+};
+
+let connectedLogger = () => {
+  console.log("Connection Established With Logstash");
 };
 
 export default new Logstash({
   error_logger: errorLogger,
+  connected_logger: connectedLogger,
   host: process.env.LOGSTASH_HOST,
   port: process.env.LOGSTASH_PORT
     ? parseInt(process.env.LOGSTASH_PORT)

src/utils/logstash/index.ts

@@ -12,6 +12,7 @@ export default class LogstashTransport {
       ? new SecureConnection(options)
       : new PlainConnection(options);
     this.manager = new Manager(options, this.connection);
+    this.manager.on("connected", options.connected_logger.bind(this));
     this.manager.on("error", options.error_logger.bind(this));
     this.manager.start();
   }
@@ -23,4 +24,4 @@ export default class LogstashTransport {
   close() {
     this.manager.close();
   }
-};
+}

src/utils/logstash/types.d.ts

@@ -14,6 +14,7 @@ interface LogstashTransportSSLOptions {
 
 interface LogstashTransportOptions extends LogstashTransportSSLOptions {
   error_logger: (...args: any[]) => void;
+  connected_logger: () => void;
   host?: string;
   port?: number;
   node_name?: string;

src/utils/queue-batch-processor.ts

@@ -7,11 +7,12 @@ export default async (queue: Queue, logstash: Logstash) => {
   for (let job of jobs) {
     jobData.push(job.data);
   }
-  if(jobData.length>0) {
+  if (jobData.length > 0) {
     logstash.log(jobData, async () => {
-        for (let job of jobs) {
-          await job.remove();
-        }
-      });
+      for (let job of jobs) {
+        await job.remove();
+      }
+      console.log("Pushed %d Logs To Logstash", jobData.length);
+    });
   }
 };

src/utils/queue-client.ts

@@ -1,5 +1,5 @@
 import { Queue } from "bullmq";
-export default new Queue("logs", {
+const queue = new Queue("logs", {
   connection: {
     host: process.env.REDIS_HOST,
     username: process.env.REDIS_USERNAME,
@@ -11,3 +11,12 @@ export default new Queue("logs", {
   },
   prefix: process.env.QUEUE_PREFIX,
 });
+
+async function checkQueueHealth(queue: Queue) {
+  const jobCounts = await queue.getJobCounts();
+  console.log("Connection Established With Redis Queue", jobCounts);
+}
+
+checkQueueHealth(queue);
+
+export default queue;