feat: Remove executor metrics on stop (#557)

When an Executor is stopped, the metrics exposed from that worker still
persist. As metrics are aggregated across all workers, the metrics for a
given indexer would be aggregated across all previous workers, creating
incorrect metrics.

runner/src/metrics.ts

@@ -67,10 +67,14 @@ export const METRICS = {
 };
 
 const aggregatorRegistry = new AggregatorRegistry();
-const workerMetrics: Record<number, string> = {};
+const workerMetrics = new Map<number, string>();
 
 export const registerWorkerMetrics = (workerId: number, metrics: string): void => {
-  workerMetrics[workerId] = metrics;
+  workerMetrics.set(workerId, metrics);
+};
+
+export const deregisterWorkerMetrics = (workerId: number): void => {
+  workerMetrics.delete(workerId);
 };
 
 export const startServer = async (): Promise<void> => {
@@ -81,7 +85,7 @@ export const startServer = async (): Promise<void> => {
   app.get('/metrics', async (_req, res) => {
     res.set('Content-Type', aggregatorRegistry.contentType);
 
-    const metrics = await AggregatorRegistry.aggregate(Object.values(workerMetrics)).metrics();
+    const metrics = await AggregatorRegistry.aggregate(Array.from(workerMetrics.values())).metrics();
     res.send(metrics);
   });
 

runner/src/stream-handler/stream-handler.ts

@@ -1,7 +1,7 @@
 import path from 'path';
 import { Worker, isMainThread } from 'worker_threads';
 
-import { registerWorkerMetrics } from '../metrics';
+import { registerWorkerMetrics, deregisterWorkerMetrics } from '../metrics';
 import Indexer from '../indexer';
 
 export enum Status {
@@ -60,6 +60,8 @@ export default class StreamHandler {
   }
 
   async stop (): Promise<void> {
+    deregisterWorkerMetrics(this.worker.threadId);
+
     await this.worker.terminate();
   }