app/include/cluster.js

'use strict'
/**
 * Copyright (c) 2017~2019, OBCon Inc.
 * All rights reserved.
 */

/**
 * @file
 * @copyright 2017~2019, OBCon Inc.
 * @author gye hyun james kim [pnuskgh@gmail.com]
 */

let os = require('os');
let cluster = require('cluster');
let moment = require('moment');
let { v1: uuidv1 } = require('uuid');

// ✅ 누락돼서 터졌던 caches 정의
const caches = require('./cache');

/**
 * 안전하게 worker.send 하기 (죽은 worker/닫힌 IPC 채널 방지)
 */
function safeSend(worker, msg) {
  try {
    if (!worker) return false;
    // node 버전에 따라 worker.isConnected()가 안전
    if (typeof worker.isConnected === 'function' && !worker.isConnected()) return false;
    if (worker.process && worker.process.connected === false) return false;
    worker.send(msg);
    return true;
  } catch (e) {
    // ERR_IPC_CHANNEL_CLOSED / EPIPE는 “죽은 채널에 보냈다”라서 무시해도 됨
    return false;
  }
}

class Cluster {
  constructor() {
    // cluster.schedulingPolicy = cluster.SCHED_NONE;      //--- 워커 스케쥴을 OS에 맡긴다.
    cluster.schedulingPolicy = cluster.SCHED_RR;        //---워커 스케쥴을 Round Robin 방식으로 한다.

    this.init_event();
  }

  get cpus() {
    return os.cpus().length;
  }

  init_event() {
    if (cluster.isMaster) {
      this.init_event_master();
    } else {
      this.init_event_worker();
    }
  }

  init_event_master() {
    //--- master:setup > master:fork > master:online
    //--- worker:disconnect > master:disconnect > master:exit
    cluster.on("setup", function(mySettings){
      // console.log("Master :: Master %s is setup", JSON.stringify(mySettings));
    });

    cluster.on("fork", function(worker) {
      // console.log("Master :: Worker %s is fork", worker.id);
    });

    cluster.on("online", function(worker) {
      console.log("Master :: Worker %s is online", worker.id);
    });

    cluster.on("disconnect", function(worker) {
      console.log("Master :: Worker %s is disconnect", worker.id);
    });

    cluster.on("exit", function(worker, code, signal) {
      console.log("Master :: Worker %s is exit", worker.id);
      console.log("Master :: Worker %s is exit code : %s", worker.id, code);
      console.log("Master :: Worker %s is exit signal : %s", worker.id, signal);
    });

    cluster.on("listening", function(worker, address) {
      console.log("Master :: Worker %s is listening", worker.id);
    });

    //--- Master에게 온 메시지를 모든 worker에게 전달한다.
    // ✅ 기존 코드: 죽은 worker에게도 send → EPIPE/IPC_CLOSED 폭탄
    cluster.on('message', function(_workerSender, req) {
      for (const idx in cluster.workers) {
        safeSend(cluster.workers[idx], req);
      }
    });
  }

  init_event_worker() {
    //--- worker:disconnect > master:disconnect > master:exit
    cluster.worker.on("disconnect", function() {
      console.log("Worker :: Worker %s disconnect", cluster.worker.id);
    });

    cluster.worker.on("error", function(error) {
      console.log("Worker :: Worker %s error", cluster.worker.id);
    });

    cluster.worker.on("exit", function(code, signal) {
      console.log("Worker :: Worker %s exit", cluster.worker.id);
    });

    cluster.worker.on("listening", function(address) {
      console.log("Worker :: Worker %s listening", cluster.worker.id);
    });

    //--- Cache에 저장된 handler에게 메시지 처리를 위임 한다.
    cluster.worker.on("message", function(req) {
      if (typeof(req && req.headers) !== 'undefined') {
        if ((req.headers.module == 'cluster') && (req.headers.action == 'workerInit')) {
          global.appl.worker = req.body;
        }
      } else {
        const handler = caches.getCache(req && req.type, req && req.path);
        if (handler == null) {
          console.log("%s - %s handler가 없습니다.", req && req.type, req && req.path);
        } else {
          console.log("%s - %s 요청 처리", req.type, req.path);
          handler(req);
        }
      }
    });

    cluster.worker.on("online", function() {
      console.log("Worker :: Worker %s online", cluster.worker.id);
    });
  }

  /**
   * config.cluster 설정 기반으로 worker fork
   * (PM2가 아닌 직접 실행 모드에서 사용)
   */
  forks() {
    if (!cluster.isMaster) return;

    const countHttp = Number((global.config && global.config.cluster && global.config.cluster.worker_count_http) || 0);
    const countTcp = Number((global.config && global.config.cluster && global.config.cluster.worker_count_tcp) || 0);
    const countUdp = Number((global.config && global.config.cluster && global.config.cluster.worker_count_udp) || 0);
    const countModbus = Number((global.config && global.config.cluster && global.config.cluster.worker_count_modbus) || 0);
    const countProxy = Number((global.config && global.config.cluster && global.config.cluster.worker_count_proxy) || 0);

    let total = 0;
    if (global.appl && global.appl.type === 'obcon_proxy') {
      total = Math.max(1, countProxy || 1);
    } else {
      total = countHttp + countTcp + countUdp + countModbus;
      if (total <= 0) total = 1;
    }

    for (let seq = 1; seq <= total; seq++) {
      const worker = cluster.fork();
      const workerType = this._getWorkerType(seq, {
        countHttp,
        countTcp,
        countUdp,
        countModbus,
        countProxy
      });

      const workerInfo = {
        id: worker.id,
        type: workerType,
        seq
      };

      worker.on('online', () => {
        safeSend(worker, {
          headers: { module: 'cluster', action: 'workerInit' },
          body: workerInfo,
          ts: moment().format('YYYY-MM-DD HH:mm:ss.SSS'),
          uuid: uuidv1()
        });
      });
    }
  }

  _getWorkerType(seq, counts) {
    if (global.appl && global.appl.type === 'obcon_proxy') return 'proxy';

    if (seq <= counts.countHttp) return 'http';
    if (seq <= counts.countHttp + counts.countTcp) return 'tcp';
    if (seq <= counts.countHttp + counts.countTcp + counts.countUdp) return 'udp';
    if (seq <= counts.countHttp + counts.countTcp + counts.countUdp + counts.countModbus) return 'modbus';
    return 'http';
  }

  /**
   * Worker에 들어온 요청(type/path)을 처리할 handler 등록
   */
  setHandler(type, path, handler) {
    caches.setCache(type, path, handler);
  }

  /**
   * workerInit 브로드캐스트
   */
  sendWorkerInit(workerInfo) {
    if (cluster.isMaster) {
      const msg = {
        headers: { module: 'cluster', action: 'workerInit' },
        body: workerInfo,
        ts: moment().format('YYYY-MM-DD HH:mm:ss.SSS'),
        uuid: uuidv1()
      };

      for (const idx in cluster.workers) {
        safeSend(cluster.workers[idx], msg);
      }
    }
  }

  /**
   * 임의 메시지 브로드캐스트 (마스터만)
   */
  broadcast(req) {
    if (!cluster.isMaster) return;

    for (const idx in cluster.workers) {
      safeSend(cluster.workers[idx], req);
    }
  }
}

module.exports = new Cluster();