feat(grouper): add Prometheus metrics

.env.sample

@@ -31,6 +31,12 @@ PROMETHEUS_PUSHGATEWAY_URL=
 # pushgateway push interval in ms
 PROMETHEUS_PUSHGATEWAY_INTERVAL=10000
 
+# Grouper memory log controls
+GROUPER_MEMORY_LOG_EVERY_TASKS=50
+GROUPER_MEMORY_GROWTH_WINDOW_TASKS=200
+GROUPER_MEMORY_GROWTH_WARN_MB=64
+GROUPER_MEMORY_HANDLE_GROWTH_WARN_MB=16
+
 # project token for error catching
 HAWK_CATCHER_TOKEN=
 
@@ -40,4 +46,4 @@ HAWK_CATCHER_TOKEN=
 IS_NOTIFIER_WORKER_ENABLED=false
 
 ## Url for telegram notifications about workspace blocks and unblocks 
-TELEGRAM_LIMITER_CHAT_URL=
+TELEGRAM_LIMITER_CHAT_URL=

lib/metrics.ts

@@ -0,0 +1,39 @@
+import * as client from 'prom-client';
+import os from 'os';
+import { nanoid } from 'nanoid';
+
+const register = new client.Registry();
+
+client.collectDefaultMetrics({ register });
+
+export { register, client };
+
+/**
+ * Start periodic push to pushgateway
+ *
+ * @param workerName - name of the worker for grouping
+ */
+export function startMetricsPushing(workerName: string): void {
+  const url = process.env.PROMETHEUS_PUSHGATEWAY_URL;
+  const interval = parseInt(process.env.PROMETHEUS_PUSHGATEWAY_INTERVAL || '10000');
+
+  if (!url) {
+    return;
+  }
+
+  const hostname = os.hostname();
+  const ID_SIZE = 5;
+  const id = nanoid(ID_SIZE);
+
+  const gateway = new client.Pushgateway(url, [], register);
+
+  console.log(`Start pushing metrics to ${url} every ${interval}ms (host: ${hostname}, id: ${id})`);
+
+  setInterval(() => {
+    gateway.pushAdd({ jobName: 'workers', groupings: { worker: workerName, host: hostname, id } }, (err) => {
+      if (err) {
+        console.error('Metrics push error:', err);
+      }
+    });
+  }, interval);
+}

runner.ts

@@ -9,6 +9,7 @@ import * as utils from './lib/utils';
 import { Worker } from './lib/worker';
 import HawkCatcher from '@hawk.so/nodejs';
 import * as dotenv from 'dotenv';
+import { startMetricsPushing } from './lib/metrics';
 
 dotenv.config();
 
@@ -57,19 +58,17 @@ class WorkerRunner {
       .then((workerConstructors) => {
         this.constructWorkers(workerConstructors);
       })
-      // .then(() => {
-      //   try {
-      //     this.startMetrics();
-      //   } catch (e) {
-      //     HawkCatcher.send(e);
-      //     console.error(`Metrics not started: ${e}`);
-      //   }
-      //
-      //   return Promise.resolve();
-      // })
       .then(() => {
         return this.startWorkers();
       })
+      .then(() => {
+        try {
+          this.startMetrics();
+        } catch (e) {
+          HawkCatcher.send(e);
+          console.error(`Metrics not started: ${e}`);
+        }
+      })
       .then(() => {
         this.observeProcess();
       })
@@ -82,67 +81,15 @@ class WorkerRunner {
   /**
    * Run metrics exporter
    */
-  // private startMetrics(): void {
-  //   if (!process.env.PROMETHEUS_PUSHGATEWAY_URL) {
-  //     return;
-  //   }
-  //
-  //   const PUSH_INTERVAL = parseInt(process.env.PROMETHEUS_PUSHGATEWAY_INTERVAL);
-  //
-  //   if (isNaN(PUSH_INTERVAL)) {
-  //     throw new Error('PROMETHEUS_PUSHGATEWAY_INTERVAL is invalid or not set');
-  //   }
-  //
-  //   const collectDefaultMetrics = promClient.collectDefaultMetrics;
-  //   const Registry = promClient.Registry;
-  //
-  //   const register = new Registry();
-  //   const startGcStats = gcStats(register);
-  //
-  //   const hostname = os.hostname();
-  //
-  //   const ID_SIZE = 5;
-  //   const id = nanoid(ID_SIZE);
-  //
-  //   // eslint-disable-next-line node/no-deprecated-api
-  //   const instance = url.parse(process.env.PROMETHEUS_PUSHGATEWAY_URL).host;
-  //
-  //   // Initialize metrics for workers
-  //   this.workers.forEach((worker) => {
-  //     // worker.initMetrics();
-  //     worker.getMetrics().forEach((metric: promClient.Counter<string>) => register.registerMetric(metric));
-  //   });
-  //
-  //   collectDefaultMetrics({ register });
-  //   startGcStats();
-  //
-  //   this.gateway = new promClient.Pushgateway(process.env.PROMETHEUS_PUSHGATEWAY_URL, null, register);
-  //
-  //   console.log(`Start pushing metrics to ${process.env.PROMETHEUS_PUSHGATEWAY_URL}`);
-  //
-  //   // Pushing metrics to the pushgateway every PUSH_INTERVAL
-  //   this.pushIntervalNumber = setInterval(() => {
-  //     this.workers.forEach((worker) => {
-  //       if (!this.gateway || !instance) {
-  //         return;
-  //       }
-  //       // Use pushAdd not to overwrite previous metrics
-  //       this.gateway.pushAdd({
-  //         jobName: 'workers',
-  //         groupings: {
-  //           worker: worker.type.replace('/', '_'),
-  //           host: hostname,
-  //           id,
-  //         },
-  //       }, (err?: Error) => {
-  //         if (err) {
-  //           HawkCatcher.send(err);
-  //           console.log(`Error of pushing metrics to gateway: ${err}`);
-  //         }
-  //       });
-  //     });
-  //   }, PUSH_INTERVAL);
-  // }
+  private startMetrics(): void {
+    if (!process.env.PROMETHEUS_PUSHGATEWAY_URL) {
+      return;
+    }
+
+    this.workers.forEach((worker) => {
+      startMetricsPushing(worker.type.replace('/', '_'));
+    });
+  }
 
   /**
    * Dynamically loads workers through the yarn workspaces