node-cluster.ts

import * as cluster from 'cluster';
import * as os from 'os';
import { Logger } from '@nestjs/common';

export class Cluster {
  static register(workers: number, logger: Logger, callback: CallableFunction): void {
    const cpuCount = os.cpus().length;

    if (cluster.isMaster) {
      logger.log(`master process started on ${process.pid}`);

      if (workers > cpuCount) {
        workers = cpuCount;
      }

      for (let i = 0; i < workers; i++) {
        cluster.fork();
      }

      process.on('SIGINT', function () {
        Cluster.exitGracefully(logger);
      });

      process.on('SIGTERM', function () {
        Cluster.exitGracefully(logger);
      });

      cluster.on('online', function (worker) {
        logger.log(`worker ${worker.process.pid} is online`);
      });

      cluster.on('exit', (worker, code, signal) => {
        logger.warn(`Worker ${worker.process.pid} died. code: ${code}, signal: ${signal} Restarting...`);
        cluster.fork();
      });
    } else {
      callback();
    }
  }

  static exitGracefully(logger) {
    logger.log('cluster is shutting down...');
    const { workers } = cluster;

    Object.entries(workers).forEach(([id, worker]) => {
      logger.log(`worker ${id} is shutting down...`);
      worker.kill('SIGTERM');
    });

    process.exit(0);
  }
}


async function initMain() {
  if (cluster.isMaster) {
    logger.log('Initializing the master process...');
    await initializeConfiguration();
    logger.log('Master Process connected to the config service');
    const shouldUseClusterMode = config.get('REPORTING_ETL_CLUSTER_MODE') === 'true';
    const numberOfNodes = Number(config.get('REPORTING_ETL_CLUSTER_NODES_COUNT') || 8);

    if (shouldUseClusterMode) {
      logger.log(`service is configured to run in cluster mode`);
      logger.log(`service is configured to run with ${numberOfNodes} nodes`);
      Cluster.register(numberOfNodes, logger, bootstrap);
    } else {
      logger.log(`service is configured to run in a single-node mode`);
      await bootstrap();
    }
  } else {
    await bootstrap();
  }
}