ConsulKvMonitor.js

'use strict';

const EventEmitter = require('events');
const _ = require('lodash');

const AlreadyInitializedError = require('./Error').AlreadyInitializedError;
const WatchError = require('./Error').WatchError;
const WatchTimeoutError = require('./Error').WatchTimeoutError;
const kvFactory = require('./Factory');
const ConsulKvData = require('./ConsulKvData');

const DEFAULT_TIMEOUT_MSEC = 5000;
const HEALTH_FALLBACK_INTERVAL_MSEC = 1000;
const DEFAULT_RETRY_START_SERVICE_TIMEOUT_MSEC = 1000;
const X_CONSUL_HEADERS = ['x-consul-index', 'x-consul-knownleader', 'x-consul-lastcontact'];

/**
 * @emits ConsulKvMonitor#changed
 * @emits ConsulKvMonitor#error
 * @emits ConsulKvMonitor#healthy
 * @emits ConsulKvMonitor#unhealthy
 */
class ConsulKvMonitor extends EventEmitter {
    /**
     * @param {Object} options
     * @param {String} options.keysPrefix - namespace in consul to monitor
     * @param {Boolean} options.json - namespace in consul to monitor
     * @param {Number} [options.timeoutMsec=5000] - connection timeout to consul
     * @param {Consul} consul
     * @throws {TypeError} On invalid options format
     * @public
     */
    constructor(options, consul) {
        super();

        if (!_.isPlainObject(options)) {
            throw new TypeError('options must be an object');
        }

        if (!_.has(options, 'keysPrefix') || !_.isString(options.keysPrefix) || _.isEmpty(options.keysPrefix)) {
            throw new TypeError('options.keysPrefix must be set and be a non-empty string');
        }

        if (!_.has(options, 'timeoutMsec')) {
            this._timeoutMsec = DEFAULT_TIMEOUT_MSEC;
        } else {
            if (!_.isSafeInteger(options.timeoutMsec) || options.timeoutMsec <= 0) {
                throw new TypeError('options.timeoutMsec must be a positive integer if set');
            }

            this._timeoutMsec = options.timeoutMsec;
        }

        // duck typing check
        if (!_.isObject(consul) || !_.isFunction(consul.watch) ||
            !_.isObject(consul.kv) || !_.isFunction(consul.kv.get)
        ) {
            throw new TypeError('consul argument does not look like Consul object');
        }

        this._keysPrefix = options.keysPrefix;
        this._json = options.json;
        this._initialized = false;

        this._consul = consul;

        this._consulKvData = new ConsulKvData();
        this._consulHeaders = {};

        this._onWatcherChange = this._onWatcherChange.bind(this);
        this._onWatcherError = this._onWatcherError.bind(this);
        this._onWatcherEnd = this._onWatcherEnd.bind(this);
        this._retryStartService = this._retryStartService.bind(this);

        this._watchKvChange = null;
        this._setWatchUnealthy();
        this._setUninitialized();

        this._fallbackToWatchHealthyInterval = null;
        this._retryTimer = null;
    }

    isWatchHealthy() {
        return this._isWatchHealthy;
    }

    _setWatchHealthy() {
        this._isWatchHealthy = true;
    }

    _setWatchUnealthy() {
        this._isWatchHealthy = false;
    }

    isInitialized() {
        return this._initialized;
    }

    _setInitialized() {
        this._initialized = true;
    }

    _setUninitialized() {
        this._initialized = false;
    }

    _isWatcherRegistered() {
        return this._watchKvChange !== null;
    }

    /**
     * @returns {ConsulKvData}
     */
    getData() {
        return this._consulKvData;
    }

    /**
     * @returns {Object}
     */
    getConsulHeaders() {
        return this._consulHeaders;
    }

    /**
     * Starts service and resolves promise with initial list of KV data.
     *
     * Listens for changes after successful resolve.
     *
     * Promise will be rejected with:
     *   `AlreadyInitializedError` if service is already started.
     *   `WatchTimeoutError` if either initial data nor error received for 5000 msec
     *   `WatchError` on error from `consul` underlying method
     *
     * Rejection of promise means that watcher was stopped and no retries will be done.
     *
     * @returns {Promise.<Array>|Promise.<AlreadyInitializedError>|Promise.<WatchError>|Promise.<WatchTimeoutError>}
     * @public
     */
    start() {
        if (this._isWatcherRegistered()) {
            return Promise.reject(new AlreadyInitializedError('Service is already started'));
        }

        return this._registerWatcherAndWaitForInitialKvData()
            .then(kvData => {
                this._watchKvChange.on('change', this._onWatcherChange);
                this._watchKvChange.on('error', this._onWatcherError);
                this._watchKvChange.on('end', this._onWatcherEnd);

                this._setInitialized();
                this._setWatchHealthy();

                this._consulKvData = kvData;
                return kvData;
            });
    }

    /**
     * Stops service even if it is not started yet. Monitor becomes `uninitialized` and `unhealthy`.
     *
     * Does not listened for changes after execution.
     *
     * @returns {ConsulKvMonitor}
     * @public
     */
    stop() {
        if (this._retryTimer !== null) {
            clearTimeout(this._retryTimer);
            this._retryTimer = null;
        }

        if (!this._isWatcherRegistered()) {
            return this;
        }

        // we need to remove listener to prevent emitting of `end` event after stop of watcher
        this._watchKvChange.removeListener('end', this._onWatcherEnd);
        this._watchKvChange.end();
        this._watchKvChange = null;
        this._unsetFallbackToWatchHealthy();
        this._setUninitialized();
        this._setWatchUnealthy();
        return this;
    }

    /**
     * Registers `consul.watch` and assigns watcher to `this._watchKvChange` and waits for the
     * first successful response from consul with list of KV data.
     * On successful response resolves promise with array of kv records (it may be empty). Method doesn't
     * add listener for `change` event.
     *
     * Promise will be rejected with:
     *   `AlreadyInitializedError` if another `consul.watch` execution is found.
     *   `WatchTimeoutError` if either initial data nor error received for 5000 msec
     *   `WatchError` on error from `consul` underlying method
     *
     * Rejection of promise means that watch was stopped and `this._watchKvChange` was cleared.
     *
     * @returns {Promise.<Array>|Promise.<AlreadyInitializedError>|Promise.<WatchError>|Promise.<WatchTimeoutError>}
     * @private
     */
    _registerWatcherAndWaitForInitialKvData() {
        return new Promise((resolve, reject) => {
            if (this._watchKvChange !== null) {
                reject(new AlreadyInitializedError('Another `consul.watch` execution is found'));
            }

            this._watchKvChange = this._consul.watch({
                method: this._consul.kv.get,
                options: {
                    key: this._keysPrefix,
                    recurse: true,
                    wait: '60s',
                },
            });

            const firstChange = (data, response) => {
                this._watchKvChange.removeListener('error', firstError);
                clearTimeout(timerId);

                const {consulKvData, errors} = kvFactory.buildConsulKvData(data, this._json);

                for (const headerName of X_CONSUL_HEADERS) {
                    this._consulHeaders[headerName] = response.headers[headerName];
                }

                if (!_.isEmpty(errors)) {
                    this._emitFactoryErrors(errors);
                }

                resolve(consulKvData);
            };

            const firstError = (err) => {
                this._watchKvChange.removeListener('change', firstChange);
                this._watchKvChange.end();
                this._watchKvChange = null;
                clearTimeout(timerId);
                reject(new WatchError(err.message, {err}));
            };

            const timerId = setTimeout(() => {
                this._watchKvChange.removeListener('error', firstError);
                this._watchKvChange.removeListener('change', firstChange);
                this._watchKvChange.end();
                this._watchKvChange = null;
                reject(new WatchTimeoutError('Initial consul watch request was timed out'));
            }, this._timeoutMsec);

            this._watchKvChange.once('change', firstChange);
            this._watchKvChange.once('error', firstError);
        });
    }

    /**
     * This method receives list of key-value records sent by `consul.watch` in `consul` format. Performs
     * parsing of response.
     *
     * If service was unhealthy, it becomes healthy.
     *
     * @param {Array} data - list of key-value records after some changes
     * @param {IncomingMessage} response - response from Consul
     * @emits ConsulKvMonitor#changed actual array of key-value records
     * @private
     */
    _onWatcherChange(data, response) {
        let isHealthyStateChanged = false;
        if (!this.isWatchHealthy()) {
            this._setWatchHealthy();
            isHealthyStateChanged = true;
        }

        const {consulKvData, errors} = kvFactory.buildConsulKvData(data, this._json);

        this._consulKvData = consulKvData;
        for (const headerName of X_CONSUL_HEADERS) {
            this._consulHeaders[headerName] = response.headers[headerName];
        }
        if (isHealthyStateChanged) {
            this.emit('healthy');
        }
        this.emit('changed', consulKvData);

        if (!_.isEmpty(errors)) {
            this._emitFactoryErrors(errors);
        }
    }

    /**
     * This method receives an Error from Consul.
     *
     * If service was healthy, it becomes unhealthy.
     *
     * @param {Error} err
     * @emits ConsulKvMonitor#error `WatchError` error
     * @private
     */
    _onWatcherError(err) {
        this._unsetFallbackToWatchHealthy();

        if (this.isWatchHealthy()) {
            this._setWatchUnealthy();
            this.emit('unhealthy');
        }

        this._setFallbackToWatchHealthy();

        this.emit('error', new WatchError(err.message, {err}));
    }

    /**
     * This method is called when connection with Consul agent was refused.
     *
     * Monitor becomes `uninitialized` and `unhealthy`.
     *
     * @emits ConsulKvMonitor#emergencyStop
     * @private
     */
    async _onWatcherEnd() {
        this._unsetFallbackToWatchHealthy();
        this._setUninitialized();
        if (this.isWatchHealthy()) {
            this._setWatchUnealthy();
            this.emit('unhealthy');
        }
        this._watchKvChange = null;
        await this._retryStartService();
    }

    /**
     * This method receives a list of errors and emits them.
     *
     * @param {Error[]} errors
     * @emits ConsulKvMonitor#error
     * @private
     */
    _emitFactoryErrors(errors) {
        setImmediate(() => {
            errors.forEach(error => this.emit('error', error));
        });
    }

    /**
     * @private
     */
    _setFallbackToWatchHealthy() {
        if (this._fallbackToWatchHealthyInterval) {
            this._unsetFallbackToWatchHealthy();
        }

        const initialUpdateTime = this._watchKvChange.updateTime();

        this._fallbackToWatchHealthyInterval = setInterval(() => {
            if (this.isWatchHealthy()) {

                // watcher is currently ends or becomes `healthy`, unset fallback interval',
                this._unsetFallbackToWatchHealthy();

                return;
            }

            const lastUpdateTime = this._watchKvChange.updateTime();

            if (initialUpdateTime !== lastUpdateTime) {
                this._unsetFallbackToWatchHealthy();

                this._setWatchHealthy();
            }

        }, HEALTH_FALLBACK_INTERVAL_MSEC);
    }

    /**
     * @private
     */
    _unsetFallbackToWatchHealthy() {
        clearInterval(this._fallbackToWatchHealthyInterval);

        this._fallbackToWatchHealthyInterval = null;
    }

    async _retryStartService() {
        try {
            const consulKvData = await this.start();
            this._consulKvData = consulKvData;

            this.emit('healthy');
            this.emit('changed', consulKvData);
        } catch (err) {
            setImmediate(() => this.emit('error', err));

            this._retryTimer = setTimeout(this._retryStartService, DEFAULT_RETRY_START_SERVICE_TIMEOUT_MSEC);
        }
    }

}

module.exports = ConsulKvMonitor;