Implementing demons on Node.js

gkozlenko June 21, 2016 at 21:51

Implementing demons on Node.js

Not so long ago, I had the honor to make the implementation of demons (workers) on Node.js for use in all projects developed by our company. We often develop projects where it is necessary to process and distribute video files on several servers in a distributed manner, so we need to have a ready-made tool for this.

Formulation of the problem:

Development should be on Node.js. We have been using this platform for the development of all our projects for a long time, so here it is a reasonable choice.
All nodes in the cluster must be equivalent. There should be no special manager or master. Otherwise, stopping the wizard may cause the entire cluster to stop.
Tasks should be in a MySQL table. It is much more flexible and informative than using any MQ. You can always access all tasks, evaluate the queue, reassign them to another node, etc.
Each worker should be able to handle several tasks at once.

Immediately provide a link to the GitHub of what happened: ( https://github.com/pipll/node-daemons ).

Running Workers

Each worker is a separate Node.js process. To create worker processes, the cluster built-in module is used . He also controls the fall of workers and re-launches them.

Worker Launch Code

'use strict';
const config = require('./config/config');
const _ = require('lodash');
const path = require('path');
const cluster = require('cluster');
const logger = require('log4js').getLogger('app');
const models = require('./models');
// Таймер для проверки количества воркеров и остановки мастера
let shutdownInterval = null;
if (cluster.isMaster) {
    // Запускаем воркеры
    _.each(config.workers, (conf, name) => {
        if (conf.enabled) {
            startWorker(name);
        }
    });
} else {
    // Инициализируем воркер
    let name = process.env.WORKER_NAME;
    let WorkerClass = require(path.join(__dirname, 'workers', name + '.js'));
    let worker = null;
    if (WorkerClass) {
        worker = new WorkerClass(name, config.workers[name]);
        // Запускаем воркер
        worker.start();
        // Подписываемся на событие, когда воркер остановлен
        worker.on('stop', () => {
            process.exit();
        });
    }
    // Подписываемся на события от мастера
    process.on('message', message => {
        if ('shutdown' === message) {
            if (worker) {
                worker.stop();
            } else {
                process.exit();
            }
        }
    });
}
// Shutdown
process.on('SIGTERM', shutdownCluster);
process.on('SIGINT', shutdownCluster);
// Метод запуска воркера
function startWorker(name) {
    let worker = cluster.fork({WORKER_NAME: name}).on('online', () => {
        logger.info('Start %s worker #%d.', name, worker.id);
    }).on('exit', status => {
        // Когда воркер был остановлен
        if ((worker.exitedAfterDisconnect || worker.suicide) === true || status === 0) {
            // Если воркер был остановлен контролируемо, то ни чего не делаем
            logger.info('Worker %s #%d was killed.', name, worker.id);
        } else {
            // Если воркер был остановлен неконтролируемо, то запускаем его еще раз
            logger.warn('Worker %s #%d was died. Replace it with a new one.', name, worker.id);
            startWorker(name);
        }
    });
}
// Метод остановки кластера
function shutdownCluster() {
    if (cluster.isMaster) {
        clearInterval(shutdownInterval);
        if (_.size(cluster.workers) > 0) {
            // Посылаем сигнал останова каждому воркеру
            logger.info('Shutdown workers:', _.size(cluster.workers));
            _.each(cluster.workers, worker => {
                try {
                    worker.send('shutdown');
                } catch (err) {
                    logger.warn('Cannot send shutdown message to worker:', err);
                }
            });
            // Ожидаем останов всех воркеров
            shutdownInterval = setInterval(() => {
                if (_.size(cluster.workers) === 0) {
                    process.exit();
                }
            }, config.shutdownInterval);
        } else {
            process.exit();
        }
    }
}

I would like to draw attention to several points:

To start the worker, the fork of the process is used, and the WORKER_NAMEname of the launched worker is passed in the environment variable .
When the worker shuts down uncontrollably, we restart it.
To control workers in a controlled way, we send them a shutdown signal . The worker responds to this event and, after completing the task, does process.exit().
The master monitors the number of workers with help setIntervaland when all the workers are stopped doing process.exit().

Base Worker Component

This component is designed to perform periodic processes and does not work with the task queue.

Base Worker Code

'use strict';
const _ = require('lodash');
const Promise = require('bluebird');
const log4js = require('log4js');
const EventEmitter = require('events');
const WorkerStates = require('./worker_states');
class Worker extends EventEmitter {
    constructor(name, conf) {
        super();
        this.name = name;
        // Значения настроек по умолчанию
        this.conf = _.defaults({}, conf, {
            sleep: 1000 // Задержка между запусками
        });
        this.logger = log4js.getLogger('worker-' + name);
        // Флаг останова воркера
        this.stopped = true;
        // Таймер для задержки между запусками loop мотода
        this.timer = null;
        // Состояние воркера
        this.state = null;
    }
    // Метод запуска воркера
    start() {
        this.logger.info('Start');
        this.stopped = false;
        this.state = WorkerStates.STATE_IDLE;
        return this._startLoop();
    }
    // Метод останова воркера
    stop() {
        this.logger.info('Stop');
        this.stopped = true;
        if (this.state === WorkerStates.STATE_IDLE) {
            // Останавливаем таймер задержки
            if (this.timer) {
                clearTimeout(this.timer);
                this.timer = null;
            }
            this.state = WorkerStates.STATE_STOP;
            // Вызываем событие останова воркера
            this.emit('stop');
        }
    }
    // Метод для выполнения бизнес задач
    loop() {
        return Promise.resolve();
    }
    // Метод запуска и обработки loop метода
    _startLoop() {
        this.state = WorkerStates.STATE_WORK;
        return this.loop().catch(err => {
            this.logger.warn('Loop error:', err);
        }).finally(() => {
            this.state = WorkerStates.STATE_IDLE;
            if (!this.stopped) {
                // Делаем повторный запуск по окончании работы loop метода
                this.timer = setTimeout(() => {
                    this._startLoop();
                }, this.conf.sleep);
            } else {
                this.state = WorkerStates.STATE_STOP;
                // Вызываем событие останова воркера
                this.emit('stop');
            }
        });
    }
}
module.exports = Worker;

The simplest worker code might look like this:

'use strict';
const Promise = require('bluebird');
const Worker = require('../components/worker');
class Sample extends Worker {
    loop() {
        this.logger.info("Loop method");
        return Promise.resolve().delay(30000);
    }
}
module.exports = Sample;

Some features:

loopThe method is intended for inheritance in descendants and implementation of business tasks. The return value of this method should be Promise.
At the end of the loopmethod, it starts again after the time specified in the worker's settings.
Worker has three states:
- STATE_IDLE - during a pause between loopmethod starts .
- STATE_WORK - while the loopmethod is running .
- STATE_STOP - after stopping the worker.

Task Processing Worker Component

This is the main component designed for parallel processing of tasks from the MySQL table.

Task Processing Worker Code

'use strict';
const config = require('../config/config');
const _ = require('lodash');
const Promise = require('bluebird');
const Worker = require('./worker');
const WorkerStates = require('./worker_states');
const models = require('../models');
class TaskWorker extends Worker {
    constructor(name, conf) {
        super(name, conf);
        // Значения настроек по умолчанию
        this.conf = _.defaults({}, this.conf, {
            maxAttempts: 3, // Максимальное количество попыток запуска задачи
            delayRatio: 300000, // Базовое значение отсрочки запуска задачи
            count: 1, // Максимальное количество одновременно обрабатываемых задач
            queue: '', // Название очереди для получения задач
            update: 3000 // Интервал обновления статуса задачи
        });
        // Счетчик одновременно обрабатываемых задач
        this.count = 0;
    }
    loop() {
        if (this.count < this.conf.count && !this.stopped) {
            // Получение очередной задачи
            return this._getTask().then(task => {
                if (task) {
                    // Увеличение счетчика одновременно обрабатываемых задач
                    this.count++;
                    // Запуск периодического обновления статуса задачи
                    let interval = setInterval(() => {
                        return models.sequelize.transaction(t => {
                            return task.touch({transaction: t});
                        });
                    }, this.conf.update);
                    // Запуск метода обработки задачи
                    this.handleTask(task.get({plain: true})).then(() => {
                        // Завершение задачи
                        return models.sequelize.transaction(t => {
                            return task.complete({transaction: t}).then(() => {
                                this.logger.info('Task completed:', task.id);
                            });
                        });
                    }).catch(err => {
                        // Задача не была выполнена - перезапуск задачи
                        this.logger.warn('Handle error:', err);
                        return this.delay(task).then(delay => {
                            return models.sequelize.transaction(t => {
                                return task.fail(delay, {transaction: t}).then(() => {
                                    this.logger.warn('Task failed:', task.id);
                                });
                            });
                        });
                    }).finally(() => {
                        clearInterval(interval);
                        this.count--;
                    }).done();
                    return null;
                }
            });
        } else {
            return Promise.resolve();
        }
    }
    // Метод обработки задачи
    handleTask() {
        return Promise.resolve();
    }
    // Метод вычисления отсрочки задачи после неуспешного запуска
    delay(task) {
        return Promise.resolve().then(() => {
            return task.attempts * this.conf.delayRatio;
        });
    }
    // Метод получения задачи для обработки
    _getTask() {
        return models.sequelize.transaction({autocommit: false}, t => {
            return models.Task.scope({
                method: ['forWork', this.conf.queue, config.node_id]
            }).find({transaction: t, lock: t.LOCK.UPDATE}).then(task => {
                if (task) {
                    return task.work(config.node_id, {transaction: t});
                }
            });
        });
    }
    _startLoop() {
        this.state = WorkerStates.STATE_WORK;
        return this.loop().catch(err => {
            this.logger.warn('Loop error:', err);
        }).finally(() => {
            if (this.count === 0) {
                this.state = WorkerStates.STATE_IDLE;
            }
            if (this.stopped && this.count === 0) {
                this.state = WorkerStates.STATE_STOP;
                this.emit('stop');
            } else {
                this.timer = setTimeout(() => {
                    this._startLoop();
                }, this.conf.sleep);
            }
        });
    }
}
module.exports = TaskWorker;

The simplest worker code might look like this:

'use strict';
const Promise = require('bluebird');
const TaskWorker = require('../components/task_worker');
class Sample extends TaskWorker {
    handleTask(task) {
        this.logger.info('Sample Task:', task);
        return Promise.resolve().delay(30000);
    }
}
module.exports = Sample;

Features of the component:

To obtain a task from the database, the design SELECT ... FOR UPDATEand subsequent UPDATEentries in the database are used in the same transaction with the auto commit disabled. This allows you to get exclusive access to the task even with simultaneous requests from several servers.
During the processing of the task, a periodic process of updating the status of the task in the database is launched. This is necessary to distinguish a long-running task from an unexpectedly completed task without updating the status.
The status of the processed task is determined by the status of the Promisereturned method handleTask. If successful, the task is marked as completed. Otherwise, the task is marked as failed and starts with a delay set in the method delay.

Task Model

The sequelize module is used to work with database models . All tasks are in the tasks table . The table has the following structure:

Field	A type	Description
`id`	integer, autoincrement	Task id
`node_id`	integer, nullable	ID of the node for which the task is intended
`queue`	string	Task queue
`status`	enum	Task status
`attempts`	integer	Number of tasks to run
`priority`	integer	Task priority
`body`	string	Task body in JSON format
`start_at`	datetime, nullable	Date and time of the start of processing the task
`finish_at`	datetime, nullable	Date and time of completion of task processing (TTL analog)
`worker_node_id`	integer, nullable	ID of the node that started processing the task
`worker_started_at`	datetime, nullable	Date and time of the start of processing the task
`checked_at`	datetime, nullable	Date and time of updating the status of the task
`created_at`	datetime, nullable	Date and time of task creation
`updated_at`	datetime, nullable	Date and time of task change

A task can be assigned to either a specific node or any node in the cluster. This is adjusted by the field node_idduring task creation. The task can be started with a delay (field start_at) and with a limited processing time (field finish_at).

Different workers work with different task queues defined in the field queue. Processing priority is set in the field priority(the more, the higher the priority). The number of task restarts is stored in the field attempts. In the body of the task (field body), in JSON format, the parameters necessary for the worker are transferred.

The field checked_atacts as a sign of a working task. Its value changes all the time during the task. If the field value checked_athas not changed for a long time, and the task is in the working status , then the task is considered failed and its status changes to failure .

Task Model Code

'use strict';
const moment = require('moment');
module.exports = function(sequelize, Sequelize) {
    return sequelize.define('Task', {
        id: {
            type: Sequelize.INTEGER,
            primaryKey: true,
            autoIncrement: true
        },
        node_id: {
            type: Sequelize.INTEGER
        },
        queue: {
            type: Sequelize.STRING,
            allowNull: false
        },
        status: {
            type: Sequelize.ENUM,
            values: ['pending', 'working', 'done', 'failure'],
            defaultValue: 'pending',
            allowNull: false
        },
        attempts: {
            type: Sequelize.INTEGER,
            defaultValue: 0,
            allowNull: false
        },
        priority: {
            type: Sequelize.INTEGER,
            defaultValue: 10,
            allowNull: false
        },
        body: {
            type: Sequelize.TEXT,
            set: function(body) {
                return this.setDataValue('body', JSON.stringify(body));
            },
            get: function() {
                try {
                    return JSON.parse(this.getDataValue('body'));
                } catch (e) {
                    return null;
                }
            }
        },
        start_at: {
            type: Sequelize.DATE
        },
        finish_at: {
            type: Sequelize.DATE
        },
        worker_node_id: {
            type: Sequelize.INTEGER
        },
        worker_started_at: {
            type: Sequelize.DATE
        },
        checked_at: {
            type: Sequelize.DATE
        }
    }, {
        tableName: 'tasks',
        freezeTableName: true,
        underscored: true,
        scopes: {
            forWork: function(queue, node_id) {
                return {
                    where: {
                        node_id: {
                            $or: [
                                null,
                                node_id
                            ]
                        },
                        queue: queue,
                        status: 'pending',
                        start_at: {
                            $or: [
                                null,
                                {
                                    $lt: moment().toDate()
                                }
                            ]
                        },
                        finish_at: {
                            $or: [
                                null,
                                {
                                    $gte: moment().toDate()
                                }
                            ]
                        }
                    },
                    order: [
                        ['priority', 'DESC'],
                        ['attempts', 'ASC'],
                        [sequelize.fn('IFNULL', sequelize.col('start_at'), sequelize.col('created_at')), 'ASC']
                    ]
                };
            }
        },
        instanceMethods: {
            fail: function(delay, options) {
                this.start_at = delay ? moment().add(delay, 'ms').toDate() : null;
                this.attempts = sequelize.literal('attempts + 1');
                this.status = 'failure';
                return this.save(options);
            },
            complete: function(options) {
                this.status = 'done';
                return this.save(options);
            },
            work: function(node_id, options) {
                this.status = 'working';
                this.worker_node_id = node_id;
                this.worker_started_at = moment().toDate();
                return this.save(options);
            },
            check: function(options) {
                this.checked_at = moment().toDate();
                return this.save(options);
            }
        }
    });
};

Task life cycle

All tasks go through the following life cycle:

A new task is created with the status pending .
When the time comes for processing the task, the first free worker gets it, transfers it to working status and fills in the fields worker_node_idand worker_started_at.
During the processing of the task, the worker, with some periodicity (by default, every 10 seconds), updates the field checked_atfor notification of correct operation.
At the end of work on the task, there may be several options for the development of events:
4.1. If the task completed successfully, then the task is transferred to the done status .
4.2. If the task was completed, the task is transferred to the failure status , the number increases attemptsand the launch is delayed for a specified amount of time (calculated in the method delaybased on the number of attempts and setting delayRatio).

The project also has a built-in module Managerthat runs on each node and does the following task processing:

Puts suspended tasks into failure status .
Translates failed tasks into pending status for new processing.
Deletes failed tasks that have already expired.
Deletes successfully completed tasks with a delay of 1 hour (can be configured).
Deletes failed completed tasks with the spent number of launch attempts with a 3-day delay (can be configured).

In addition, this module works with enabling / disabling nodes.

Manager Module Code

'use strict';
const _ = require('lodash');
const moment = require('moment');
const Promise = require('bluebird');
const Worker = require('../components/worker');
const models = require('../models');
const config = require('../config/config');
class Manager extends Worker {
    constructor(name, conf) {
        super(name, conf);
        this.conf = _.defaults({}, this.conf, {
            maxUpdate: 30000,       // 30 seconds
            maxCompleted: 3600000,  // 1 hour
            maxFailed: 259200000    // 3 days
        });
    }
    loop() {
        return models.sequelize.transaction(t => {
            return Promise.resolve()
                .then(() => {
                    return this._checkCurrentNode(t);
                })
                .then(() => {
                    return this._activateNodes(t);
                })
                .then(() => {
                    return this._pauseNodes(t);
                })
                .then(() => {
                    return this._restoreFrozenTasks(t);
                })
                .then(() => {
                    return this._restoreFailedTasks(t);
                })
                .then(() => {
                    return this._deleteDeadTasks(t);
                })
                .then(() => {
                    return this._deleteCompletedTasks(t);
                })
                .then(() => {
                    return this._deleteFailedTasks(t);
                });
        });
    }
    _checkCurrentNode(t) {
        return models.Node.findById(config.node_id, {transaction: t}).then(node => {
            if (node) {
                return node.check();
            }
        });
    }
    _activateNodes(t) {
        return models.Node.update({
            is_active: true
        }, {
            where: {
                is_active: false,
                checked_at: {
                    $gte: moment().subtract(2 * this.conf.sleep).toDate()
                }
            },
            transaction: t
        }).spread(count => {
            if (count > 0) {
                this.logger.info('Activate nodes:', count);
            }
        });
    }
    _pauseNodes(t) {
        return models.Node.update({
            is_active: false
        }, {
            where: {
                is_active: true,
                checked_at: {
                    $lt: moment().subtract(2 * this.conf.sleep).toDate()
                }
            },
            transaction: t
        }).spread(count => {
            if (count > 0) {
                this.logger.info('Pause nodes:', count);
            }
        });
    }
    _restoreFrozenTasks(t) {
        return models.Task.update({
            status: 'failure',
            attempts: models.sequelize.literal('attempts + 1')
        }, {
            where: {
                status: 'working',
                checked_at: {
                    $lt: moment().subtract(this.conf.maxUpdate).toDate()
                }
            },
            transaction: t
        }).spread(count => {
            if (count > 0) {
                this.logger.info('Restore frozen tasks:', count);
            }
        });
    }
    _restoreFailedTasks(t) {
        let where = [{status: 'failure'}];
        let conditions = this._failedTasksConditions();
        if (conditions.length) {
            where.push({$or: conditions});
        }
        return models.Task.update({
            status: 'pending',
            worker_node_id: null,
            worker_started_at: null
        }, {
            where: where,
            transaction: t
        }).spread(count => {
            if (count > 0) {
                this.logger.info('Restore failure tasks:', count);
            }
        });
    }
    _deleteDeadTasks(t) {
        return models.Task.destroy({
            where: {
                status: 'pending',
                finish_at: {
                    $lt: moment().toDate()
                }
            },
            transaction: t
        }).then(count => {
            if (count > 0) {
                this.logger.info('Delete dead tasks:', count);
            }
        });
    }
    _deleteCompletedTasks(t) {
        return models.Task.destroy({
            where: {
                status: 'done',
                checked_at: {
                    $lt: moment().subtract(this.conf.maxCompleted).toDate()
                }
            },
            transaction: t
        }).then(count => {
            if (count > 0) {
                this.logger.info('Delete completed tasks:', count);
            }
        });
    }
    _deleteFailedTasks(t) {
        let where = [
            {status: 'failure'},
            {checked_at: {
                $lt: moment().subtract(this.conf.maxFailed).toDate()
            }}
        ];
        let conditions = this._failedTasksConditions();
        if (conditions.length) {
            where.push({$or: conditions});
        }
        return models.Task.destroy({
            where: where,
            transaction: t
        }).then(count => {
            if (count > 0) {
                this.logger.info('Delete failed tasks:', count);
            }
        });
    }
    _failedTasksConditions() {
        let conditions = [];
        _.each(config.workers, (worker) => {
            if (worker.queue) {
                let item = {queue: worker.queue};
                if (worker.maxAttempts !== undefined) {
                    item.attempts = {
                        $lt: worker.maxAttempts
                    };
                }
                conditions.push(item);
            }
        });
        return conditions;
    }
}
module.exports = Manager;

Conclusions and plans for the future

In general, it turned out to be a good and fairly reliable tool for working with background tasks, which we can share with the community. We developed the basic ideas and principles used in this project over several years of work on various projects.

The project development plans:

Hotreload workers without restarting the master process. So that you can update the code of individual workers without interfering with the work of others.
Adding information about the progress of the task (field progress) and adding a method to the module TaskWorkerto update this information.
Creation of various interfaces for working with tasks and nodes:
- CLI
- Web
- API
Creating basic workers:
- Processing video files.
- File replication between servers for reliable storage.
Tests for workers.

I will be glad to constructive criticism and answer questions of interest in the comments.

Tags:

Implementing demons on Node.js