fix: Surface metrics to main thread

near · Sep 14, 2023 · 6983e21 · 6983e21
1 parent 2fd35a1
commit 6983e21
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 4 deletions.
diff --git a/runner/src/index.ts b/runner/src/index.ts
@@ -13,6 +13,12 @@ const STREAM_HANDLER_THROTTLE_MS = 500;
 
 type StreamHandlers = Record<string, Worker>;
 
+interface Metric {
+  type: 'UNPROCESSED_STREAM_MESSAGES' | 'EXECUTION_DURATION'
+  labels: Record<string, string>
+  value: number
+};
+
 void (async function main () {
   try {
     const streamHandlers: StreamHandlers = {};
@@ -27,7 +33,11 @@ void (async function main () {
 
         const worker = new Worker('./dist/worker.js');
         worker.postMessage({ streamKey });
-        // const handler = processStream(streamKey);
+
+        worker.on('message', (message: Metric) => {
+          metrics[message.type].labels(message.labels).set(message.value);
+        });
+
         streamHandlers[streamKey] = worker;
       });
 

diff --git a/runner/src/worker.ts b/runner/src/worker.ts
@@ -2,7 +2,6 @@ import { parentPort } from 'worker_threads';
 
 import Indexer from './indexer';
 import RedisClient from './redis-client';
-import * as metrics from './metrics';
 
 const indexer = new Indexer('mainnet');
 const redisClient = new RedisClient();
@@ -46,8 +45,16 @@ parentPort?.on('message', async ({ streamKey }) => {
 
       const unprocessedMessages = await redisClient.getUnprocessedStreamMessages(streamKey);
 
-      metrics.UNPROCESSED_STREAM_MESSAGES.labels({ indexer: indexerName, type: streamType }).set(unprocessedMessages?.length ?? 0);
-      metrics.EXECUTION_DURATION.labels({ indexer: indexerName, type: streamType }).set(performance.now() - startTime);
+      parentPort?.postMessage({
+        type: 'UNPROCESSED_STREAM_MESSAGES',
+        labels: { indexer: indexerName, type: streamType },
+        value: unprocessedMessages?.length ?? 0,
+      });
+      parentPort?.postMessage({
+        type: 'EXECUTION_DURATION',
+        labels: { indexer: indexerName, type: streamType },
+        value: performance.now() - startTime,
+      });
 
       console.log(`Success: ${indexerName}`);
     } catch (err) {