Fyllut implementing Node.js Cluster for improved performance

.nais/fyllut/config.yaml

@@ -50,7 +50,8 @@ spec:
       memory: 256Mi
   prometheus:
     enabled: true  # default: false. Pod will now be scraped for metrics by Prometheus.
-    path: /fyllut/internal/metrics # Path where prometheus metrics are served.
+    path: /node-cluster-metrics # Path where prometheus metrics are served.
+    port: 4001
   accessPolicy:
     inbound:
       rules:

packages/fyllut-backend/src/logger.js

@@ -1,13 +1,15 @@
 import correlator from 'express-correlation-id';
+import nodeProcess from 'node:process';
 import { createLogger, format, transports } from 'winston';
 
-const correlationIdFormat = format((info) => {
+const commonFormat = format((info) => {
   info.correlation_id = correlator.getId();
+  info.node_worker_pid = nodeProcess.pid;
   return info;
 });
 
 export const logger = createLogger({
   level: process.env.FYLLUT_BACKEND_LOGLEVEL || (process.env.NODE_ENV === 'test' ? 'warning' : 'info'),
-  format: format.combine(correlationIdFormat(), format.json()),
+  format: format.combine(commonFormat(), format.json()),
   transports: [new transports.Console()],
 });

packages/fyllut-backend/src/metrics-server.ts

@@ -0,0 +1,18 @@
+import express from 'express';
+import { AggregatorRegistry } from 'prom-client';
+
+const metricsServer = express();
+const aggregatorRegistry = new AggregatorRegistry();
+
+metricsServer.get('/node-cluster-metrics', async (req, res) => {
+  try {
+    const metrics = await aggregatorRegistry.clusterMetrics();
+    res.set('Content-Type', aggregatorRegistry.contentType);
+    res.send(metrics);
+  } catch (ex: any) {
+    res.statusCode = 500;
+    res.send(ex.message);
+  }
+});
+
+export default metricsServer;

packages/fyllut-backend/src/middleware/httpRequestLogger.js

@@ -1,6 +1,7 @@
 import ecsFormat from '@elastic/ecs-morgan-format';
 import correlator from 'express-correlation-id';
 import morgan from 'morgan';
+import nodeProcess from 'node:process';
 import { config } from '../config/config';
 import { isEnabled } from '../logging';
 import { clean } from '../utils/logCleaning.js';
@@ -16,6 +17,7 @@ const httpRequestLogger = morgan(
         ...logEntry,
         level: res.statusCode < 500 ? 'Info' : 'Error',
         correlation_id: correlator.getId(),
+        node_worker_pid: nodeProcess.pid,
       }),
     );
   },

packages/fyllut-backend/src/server.js

@@ -1,17 +1,45 @@
+import nodeCluster from 'node:cluster';
+import nodeOs from 'node:os';
+import nodeProcess from 'node:process';
 import { createApp } from './app';
 import { logger } from './logger.js';
+import metricsServer from './metrics-server';
 import './utils/errorToJson.js';
 
+const setupNodeCluster = !!process.env.NAIS_CLUSTER_NAME || process.env.ENABLE_CLUSTER === 'true';
+const numCPUs = nodeOs.availableParallelism();
 const port = parseInt(process.env.PORT || '8080');
+const metricsPort = parseInt(process.env.METRICS_PORT || '4001');
 
-const app = createApp();
+let viteApp;
 
-logger.info(`serving on ${port}`);
-if (import.meta.env.PROD) {
-  app.listen(port);
+if (setupNodeCluster && nodeCluster.isPrimary) {
+  logger.info(`Primary ${nodeProcess.pid} is running, will spawn ${numCPUs} workers`);
+
+  // Start server for aggregated metrics
+  metricsServer.listen(metricsPort);
+  logger.info(`Metrics server for node cluster listening to ${metricsPort}`);
+
+  // Fork workers
+  for (let i = 0; i < numCPUs; i++) {
+    nodeCluster.fork();
+  }
+
+  nodeCluster.on('exit', (_worker, code, signal) => {
+    logger.info(`Worker ${nodeProcess.pid} died`, { signal, code });
+  });
+} else {
+  const app = createApp();
+  if (import.meta.env.PROD) {
+    app.listen(port);
+    logger.info(`Worker ${process.pid} started, listening to ${port}`);
+  } else {
+    logger.info(`serving on ${port}`);
+  }
+  viteApp = app;
 }
 
 //Play nice with nais, force node to delay quiting to ensure no traffic is incoming
 process.on('SIGTERM', () => setTimeout(() => logger.debug('Har sovet i 30 sekunder'), 30000));
 
-export const viteNodeApp = app;
+export const viteNodeApp = viteApp;

packages/fyllut-backend/src/services/AppMetrics.ts

@@ -1,46 +1,40 @@
-import { Counter, Histogram, Registry, collectDefaultMetrics } from 'prom-client';
+import { Counter, Histogram, collectDefaultMetrics, register as defaultRegister } from 'prom-client';
 import { logger } from '../logger';
 
 class AppMetrics {
-  private readonly _register: Registry;
   private readonly _exstreamPdfRequestsCounter: Counter;
   private readonly _exstreamPdfFailuresCounter: Counter;
   private readonly _outgoingRequestDuration: Histogram;
 
   constructor() {
     logger.info('Initializing metrics client');
 
-    this._register = new Registry();
-
     this._exstreamPdfRequestsCounter = new Counter({
       name: 'fyllut_exstream_pdf_requests_total',
       help: 'Number of exstream pdf requests',
       labelNames: [],
-      registers: [this._register],
     });
 
     this._exstreamPdfFailuresCounter = new Counter({
       name: 'fyllut_exstream_pdf_failures_total',
       help: 'Number of exstream pdf requests which failed',
       labelNames: [],
-      registers: [this._register],
     });
 
     this._outgoingRequestDuration = new Histogram({
       name: 'fyllut_outgoing_request_duration_seconds',
       help: 'Request duration for outgoing requests made by FyllUt',
       labelNames: ['service', 'method', 'error'],
       buckets: [0.7, 0.8, 0.9, 1.0, 1.2, 1.5, 2.0, 5.0, 15.0],
-      registers: [this._register],
     });
     this._outgoingRequestDuration.zero({ service: 'exstream', method: 'createPdf', error: 'false' });
     this._outgoingRequestDuration.zero({ service: 'exstream', method: 'createPdf', error: 'true' });
 
-    collectDefaultMetrics({ register: this._register });
+    collectDefaultMetrics();
   }
 
   public get register() {
-    return this._register;
+    return defaultRegister;
   }
 
   public get exstreamPdfRequestsCounter() {