Event loop monitoring to take action when it's under pressure

README.md

NodeJS Event Loop Monitoring

This utility library aims to monitor the event loop and report any lags above a certain threshold to a consumer. It is using heavy to monitor the event loop. It is also logging lags from time to time.

Use case

In my case, I'm using this utility because I've a lot of jobs spread accross consumers, those consumers need to manage what they can('t) do. When the event loop becomes laggy it means the single thread (event loop) is too heavily used and that space for I/O is lacking. Hence I'm registering a listener that breaks the parrallelism of the process.

Why am I not using a fork? Because I want to avoid too many I/O, hence I'm storing plenty of jobs in memory at once and then treating them in parallel because each of them as I/O involved. Using IPC in my case might lead to jobs being lost which I'm not ready to trade againt Rabbit.

Usage

const monitor = require('event-loop-monitoring.js');
monitor.underPressure(function () {
  console.log('Event loop is under pressure.');
});
monitor.start();

Params

LOG_INTERVAL = 5000 // Do not log more than once within 5 seconds
INTERVAL_MEASURE = 500, // Check event-loop delays each 500ms
MAX_LAG = 70, // Tolerated lag of the Event Loop is 70ms

If you have any suggestion, heads up :)

event-loop-monitoring.js

const Heavy = require('heavy'),
    LOG_INTERVAL = 5000,
    INTERVAL_MEASURE = 500,
    MAX_LAG = 70,
    service = {},
    underPressureCallbacks = [];

let BENCHMARK_STARTED = false,
    heavyInstance;

service.isOk = function cpuCurrentLoad() {
    if (!heavyInstance || !heavyInstance.load) {
        return true;
    }
    return heavyInstance.load.eventLoopDelay < MAX_LAG;
};

service.underPressure = function (cb) {
    if (typeof cb !== 'function') {
        throw new Error('underPressure callback must be a function.');
    }
    underPressureCallbacks.push(cb);
};

function benchmark() {
    heavyInstance = new Heavy({
        sampleInterval: INTERVAL_MEASURE,
    });

    heavyInstance.start();

    setInterval(logIfAboveThreshold, LOG_INTERVAL);
    setInterval(() => {
        if (underPressureCallbacks.length > 0 && !service.isOk()) {
            underPressureCallbacks.forEach(Function.prototype.call, Function.prototype.call);
        }
    }, INTERVAL_MEASURE + 10);
}

service.start = function () {
    if (BENCHMARK_STARTED) {
        return;
    }
    benchmark();
    BENCHMARK_STARTED = true;
};

module.exports = service;

function logIfAboveThreshold() {
    if (!heavyInstance || !heavyInstance.load) {
        return;
    }
    if (heavyInstance.load.eventLoopDelay > MAX_LAG) {
        console.log('Event loop delay above threshold', {
            threshold: MAX_LAG,
            eventLoopDelayedByMS: heavyInstance.load.eventLoopDelay,
            heap_used: heavyInstance.load.heapUsed,
            rss: heavyInstance.load.rss,
        });
    }
}