fix(metrics): validate push interval, add push cleanup, and prevent retry double-counting in grouper

Author: Kuchizu

lib/metrics.ts

@@ -1,39 +1,97 @@
 import * as client from 'prom-client';
 import os from 'os';
 import { nanoid } from 'nanoid';
+import createLogger from './logger';
 
 const register = new client.Registry();
+const logger = createLogger();
+
+const DEFAULT_PUSH_INTERVAL_MS = 10_000;
+const ID_SIZE = 5;
+const METRICS_JOB_NAME = 'workers';
+
+let pushInterval: NodeJS.Timeout | null = null;
+let currentWorkerName = '';
 
 client.collectDefaultMetrics({ register });
 
 export { register, client };
 
 /**
- * Start periodic push to pushgateway
+ * Parse push interval from environment.
+ */
+function getPushIntervalMs(): number {
+  const rawInterval = process.env.PROMETHEUS_PUSHGATEWAY_INTERVAL;
+  const parsedInterval = rawInterval === undefined
+    ? DEFAULT_PUSH_INTERVAL_MS
+    : Number(rawInterval);
+
+  const interval = Number.isFinite(parsedInterval) && parsedInterval > 0
+    ? parsedInterval
+    : DEFAULT_PUSH_INTERVAL_MS;
+
+  if (rawInterval !== undefined && interval !== parsedInterval) {
+    logger.warn(`[metrics] invalid PROMETHEUS_PUSHGATEWAY_INTERVAL="${rawInterval}", fallback to ${DEFAULT_PUSH_INTERVAL_MS}ms`);
+  }
+
+  return interval;
+}
+
+/**
+ * Stop periodic push to pushgateway.
+ */
+export function stopMetricsPushing(): void {
+  if (!pushInterval) {
+    return;
+  }
+
+  clearInterval(pushInterval);
+  pushInterval = null;
+  logger.info(`[metrics] stopped pushing metrics for worker=${currentWorkerName}`);
+  currentWorkerName = '';
+}
+
+/**
+ * Start periodic push to pushgateway.
  *
- * @param workerName - name of the worker for grouping
+ * @param workerName - name of the worker for grouping.
  */
-export function startMetricsPushing(workerName: string): void {
+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;
+    return stopMetricsPushing;
   }
 
+  if (pushInterval) {
+    logger.warn(`[metrics] pushing is already started for worker=${currentWorkerName}, skip duplicate start for worker=${workerName}`);
+
+    return stopMetricsPushing;
+  }
+
+  const interval = getPushIntervalMs();
   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})`);
+  currentWorkerName = workerName;
+
+  logger.info(`Start pushing metrics to ${url} every ${interval}ms (host: ${hostname}, id: ${id}, worker: ${workerName})`);
 
-  setInterval(() => {
-    gateway.pushAdd({ jobName: 'workers', groupings: { worker: workerName, host: hostname, id } }, (err) => {
+  pushInterval = setInterval(() => {
+    gateway.pushAdd({
+      jobName: METRICS_JOB_NAME,
+      groupings: {
+        worker: workerName,
+        host: hostname,
+        id,
+      },
+    }, (err) => {
       if (err) {
-        console.error('Metrics push error:', err);
+        logger.error(`Metrics push error: ${err.message || err}`);
       }
     });
   }, interval);
+
+  return stopMetricsPushing;
 }

runner.ts

@@ -41,9 +41,9 @@ class WorkerRunner {
   // private gateway?: promClient.Pushgateway;
 
   /**
-   * number returned by setInterval() of metrics push function
+   * Metrics push cleanup callback.
    */
-  private pushIntervalNumber?: ReturnType<typeof setInterval>;
+  private stopMetricsPushing?: () => void;
 
   /**
    * Create runner instance
@@ -86,9 +86,21 @@ class WorkerRunner {
       return;
     }
 
-    this.workers.forEach((worker) => {
-      startMetricsPushing(worker.type.replace('/', '_'));
-    });
+    if (this.workers.length === 0) {
+      return;
+    }
+
+    const workerTypes = Array.from(new Set(this.workers.map((worker) => {
+      return worker.type.replace('/', '_');
+    })));
+
+    const workerTypeForMetrics = workerTypes.length === 1 ? workerTypes[0] : 'multi_worker_process';
+
+    if (workerTypes.length > 1) {
+      console.warn(`[metrics] ${workerTypes.length} workers are running in one process; pushing metrics as "${workerTypeForMetrics}" to avoid duplicated attribution`);
+    }
+
+    this.stopMetricsPushing = startMetricsPushing(workerTypeForMetrics);
   }
 
   /**
@@ -224,7 +236,8 @@ class WorkerRunner {
   private async stopWorker(worker: Worker): Promise<void> {
     try {
       // stop pushing metrics
-      clearInterval(this.pushIntervalNumber);
+      this.stopMetricsPushing?.();
+      this.stopMetricsPushing = undefined;
       await worker.finish();
 
       console.log(