Update prod

.env.sample

@@ -16,6 +16,9 @@ MONGO_ACCOUNTS_DATABASE_URI=mongodb://localhost:27017/hawk
 # MongoDB URL for connecting to events database
 MONGO_EVENTS_DATABASE_URI=mongodb://localhost:27017/hawk_events
 
+# MongoDB appName for per-worker load attribution (e.g. hawk-worker-limiter)
+MONGO_APP_NAME=
+
 # JWT secret key for projects tokens
 JWT_SECRET=qwerty
 
@@ -25,11 +28,14 @@ EVENT_SECRET=hell
 # @codex_bot webhook for reports
 CODEX_BOT_WEBHOOK=
 
-# address of prometheus pushgateway
-PROMETHEUS_PUSHGATEWAY_URL=
+# Port for VictoriaMetrics metrics endpoint.
+PROMETHEUS_METRICS_PORT=
+
+# Host for metrics endpoint binding.
+PROMETHEUS_METRICS_HOST=0.0.0.0
 
-# pushgateway push interval in ms
-PROMETHEUS_PUSHGATEWAY_INTERVAL=10000
+# Path for metrics endpoint.
+PROMETHEUS_METRICS_PATH=/metrics
 
 # Grouper memory log controls
 # Number of handled tasks between memory checkpoint logs

.env.test

@@ -22,8 +22,8 @@ EVENT_SECRET=hell
 # @codex_bot webhook for reports
 CODEX_BOT_WEBHOOK=
 
-# address of prometheus pushgateway
-PROMETHEUS_PUSHGATEWAY=
+# Port for VictoriaMetrics metrics endpoint.
+PROMETHEUS_METRICS_PORT=
 
 # Feature flags
 

lib/db/controller.ts

@@ -20,6 +20,11 @@ export class DatabaseController {
    */
   private readonly connectionUri: string;
 
+  /**
+   * Sent to MongoDB on handshake; overrides any `appName` query param in the URI
+   */
+  private readonly appName?: string;
+
   /**
    * GridFSBucket object
    * Used to store files in GridFS
@@ -30,12 +35,14 @@ export class DatabaseController {
    * Creates controller instance
    *
    * @param connectionUri - mongo URI for connection
+   * @param appName - MongoDB appName, defaults to `process.env.MONGO_APP_NAME`
    */
-  constructor(connectionUri: string) {
+  constructor(connectionUri: string, appName?: string) {
     if (!connectionUri) {
       throw new DatabaseConnectionError('Connection URI is not specified. Check .env');
     }
     this.connectionUri = connectionUri;
+    this.appName = appName ?? process.env.MONGO_APP_NAME;
   }
 
   /**
@@ -53,6 +60,7 @@ export class DatabaseController {
       this.connection = await connect(this.connectionUri, {
         useNewUrlParser: true,
         useUnifiedTopology: true,
+        ...(this.appName ? { appName: this.appName } : {}),
       });
       this.db = await this.connection.db();
 

lib/metrics.ts

@@ -1,97 +1,180 @@
 import * as client from 'prom-client';
-import os from 'os';
-import { nanoid } from 'nanoid';
+import * as http from 'http';
 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';
+const DEFAULT_METRICS_HOST = '0.0.0.0';
+const DEFAULT_METRICS_PATH = '/metrics';
+const MIN_PORT = 1;
+const MAX_PORT = 65535;
+const HTTP_OK = 200;
+const HTTP_NOT_FOUND = 404;
+const HTTP_INTERNAL_SERVER_ERROR = 500;
 
-let pushInterval: NodeJS.Timeout | null = null;
+let metricsServer: http.Server | null = null;
 let currentWorkerName = '';
 
 client.collectDefaultMetrics({ register });
 
 export { register, client };
 
 /**
- * Parse push interval from environment.
+ * Parse metrics endpoint port 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`);
+function getMetricsPort(): number | null {
+  const rawPort = process.env.PROMETHEUS_METRICS_PORT;
+
+  if (!rawPort) {
+    return null;
+  }
+
+  const port = Number(rawPort);
+
+  if (!Number.isInteger(port) || port < MIN_PORT || port > MAX_PORT) {
+    logger.warn(`[metrics] invalid PROMETHEUS_METRICS_PORT="${rawPort}"; expected an integer between ${MIN_PORT} and ${MAX_PORT}`);
+
+    return null;
+  }
+
+  return port;
+}
+
+/**
+ * Read metrics endpoint path from environment.
+ */
+function getMetricsPath(): string {
+  const rawPath = process.env.PROMETHEUS_METRICS_PATH;
+
+  if (!rawPath) {
+    return DEFAULT_METRICS_PATH;
+  }
+
+  const path = rawPath.trim();
+
+  if (!path) {
+    logger.warn(`[metrics] invalid PROMETHEUS_METRICS_PATH="${rawPath}", fallback to ${DEFAULT_METRICS_PATH}`);
+
+    return DEFAULT_METRICS_PATH;
+  }
+
+  if (!path.startsWith('/')) {
+    const normalizedPath = `/${path}`;
+
+    logger.warn(`[metrics] normalized PROMETHEUS_METRICS_PATH from "${rawPath}" to "${normalizedPath}"`);
+
+    return normalizedPath;
   }
 
-  return interval;
+  return path;
 }
 
 /**
- * Stop periodic push to pushgateway.
+ * Stop HTTP metrics endpoint.
  */
-export function stopMetricsPushing(): void {
-  if (!pushInterval) {
+export function stopMetricsServer(): void {
+  if (!metricsServer) {
     return;
   }
 
-  clearInterval(pushInterval);
-  pushInterval = null;
-  logger.info(`[metrics] stopped pushing metrics for worker=${currentWorkerName}`);
-  currentWorkerName = '';
+  const serverToStop = metricsServer;
+  const stoppedWorkerName = currentWorkerName;
+
+  if (!serverToStop.listening) {
+    logger.info(`[metrics] endpoint already stopped for worker=${stoppedWorkerName}`);
+
+    if (metricsServer === serverToStop) {
+      metricsServer = null;
+      currentWorkerName = '';
+    }
+
+    return;
+  }
+
+  serverToStop.close((error) => {
+    if (error) {
+      logger.error(`[metrics] failed to stop endpoint for worker=${stoppedWorkerName}: ${error.message}`);
+
+      return;
+    }
+
+    if (metricsServer === serverToStop) {
+      metricsServer = null;
+      currentWorkerName = '';
+    }
+
+    logger.info(`[metrics] stopped endpoint for worker=${stoppedWorkerName}`);
+  });
 }
 
 /**
- * Start periodic push to pushgateway.
+ * Start HTTP metrics endpoint for scraper-based monitoring.
  *
- * @param workerName - name of the worker for grouping.
+ * @param workerName - name of the worker for default metric labels.
  */
-export function startMetricsPushing(workerName: string): () => void {
-  const url = process.env.PROMETHEUS_PUSHGATEWAY_URL;
+export function startMetricsServer(workerName: string): () => void {
+  const port = getMetricsPort();
 
-  if (!url) {
-    return stopMetricsPushing;
+  if (!port) {
+    return stopMetricsServer;
   }
 
-  if (pushInterval) {
-    logger.warn(`[metrics] pushing is already started for worker=${currentWorkerName}, skip duplicate start for worker=${workerName}`);
+  if (metricsServer) {
+    logger.warn(`[metrics] endpoint is already started for worker=${currentWorkerName}, skip duplicate start for worker=${workerName}`);
 
-    return stopMetricsPushing;
+    return stopMetricsServer;
   }
 
-  const interval = getPushIntervalMs();
-  const hostname = os.hostname();
-  const id = nanoid(ID_SIZE);
-  const gateway = new client.Pushgateway(url, undefined, register);
+  const host = process.env.PROMETHEUS_METRICS_HOST || DEFAULT_METRICS_HOST;
+  const path = getMetricsPath();
+
+  register.setDefaultLabels({ worker: workerName });
 
+  const server = http.createServer(async (request, response) => {
+    const requestPath = request.url?.split('?')[0];
+
+    if (requestPath === '/-/healthy') {
+      response.writeHead(HTTP_OK, { 'Content-Type': 'text/plain' });
+      response.end('ok');
+
+      return;
+    }
+
+    if (request.method !== 'GET' || requestPath !== path) {
+      response.writeHead(HTTP_NOT_FOUND, { 'Content-Type': 'text/plain' });
+      response.end('not found');
+
+      return;
+    }
+
+    try {
+      response.writeHead(HTTP_OK, { 'Content-Type': register.contentType });
+      response.end(await register.metrics());
+    } catch (error) {
+      const message = error instanceof Error ? error.message : String(error);
+
+      logger.error(`[metrics] failed to render metrics: ${message}`);
+      response.writeHead(HTTP_INTERNAL_SERVER_ERROR, { 'Content-Type': 'text/plain' });
+      response.end('metrics error');
+    }
+  });
+
+  server.on('error', (error) => {
+    logger.error(`[metrics] endpoint error for worker=${workerName}: ${error.message}`);
+
+    if (metricsServer === server) {
+      metricsServer = null;
+      currentWorkerName = '';
+    }
+  });
+
+  metricsServer = server;
   currentWorkerName = workerName;
 
-  logger.info(`Start pushing metrics to ${url} every ${interval}ms (host: ${hostname}, id: ${id}, worker: ${workerName})`);
-
-  pushInterval = setInterval(() => {
-    gateway.pushAdd({
-      jobName: METRICS_JOB_NAME,
-      groupings: {
-        worker: workerName,
-        host: hostname,
-        id,
-      },
-    }, (err) => {
-      if (err) {
-        logger.error(`Metrics push error: ${err.message || err}`);
-      }
-    });
-  }, interval);
-
-  return stopMetricsPushing;
+  server.listen(port, host, () => {
+    logger.info(`[metrics] endpoint started for worker=${workerName} at http://${host}:${port}${path}`);
+  });
+
+  return stopMetricsServer;
 }