CartoDB-SQL-API/app.js

#!/usr/bin/env node

'use strict';

const fs = require('fs');
const path = require('path');
const fqdn = require('@carto/fqdn-sync');

const argv = require('yargs')
    .usage('Usage: node $0 <environment> [options]')
    .help('h')
    .example(
        'node $0 production -c /etc/sql-api/config.js',
        'start server in production environment with /etc/sql-api/config.js as config file'
    )
    .alias('h', 'help')
    .alias('c', 'config')
    .nargs('c', 1)
    .describe('c', 'Load configuration from path')
    .argv;

const environmentArg = argv._[0] || process.env.NODE_ENV || 'development';
const configurationFile = path.resolve(argv.config || './config/environments/' + environmentArg + '.js');

if (!fs.existsSync(configurationFile)) {
    console.error('Configuration file "%s" does not exist', configurationFile);
    process.exit(1);
}

global.settings = require(configurationFile);

const ENVIRONMENT = argv._[0] || process.env.NODE_ENV || global.settings.environment;
process.env.NODE_ENV = ENVIRONMENT;

const availableEnvironments = ['development', 'production', 'test', 'staging'];

// sanity check arguments
if (availableEnvironments.indexOf(ENVIRONMENT) === -1) {
  console.error("node app.js [environment]");
  console.error("Available environments: " + availableEnvironments.join(', '));
  process.exit(1);
}

global.settings.api_hostname = fqdn.hostname();

global.log4js = require('log4js');
const log4jsConfig = {
    appenders: [],
    replaceConsole: true
};

if (global.settings.log_filename) {
    const logFilename = path.resolve(global.settings.log_filename);
    const logDirectory = path.dirname(logFilename);
    if (!fs.existsSync(logDirectory)) {
        console.error("Log filename directory does not exist: " + logDirectory);
        process.exit(1);
    }
    console.log("Logs will be written to " + logFilename);
    log4jsConfig.appenders.push(
        { type: "file", absolute: true, filename: logFilename }
    );
} else {
    log4jsConfig.appenders.push(
        { type: "console", layout: { type:'basic' } }
    );
}

global.log4js.configure(log4jsConfig);
global.logger = global.log4js.getLogger();

if (!global.settings.routes) {
    console.error('Missing environment paramenter "routes". Please review your configuration file.');
    console.error("Available environments: " + availableEnvironments.join(', '));
    process.exit(1);
}

const version = require("./package").version;

const StatsClient = require('./app/stats/client');
if (global.settings.statsd) {
    // Perform keyword substitution in statsd
    if (global.settings.statsd.prefix) {
        global.settings.statsd.prefix = global.settings.statsd.prefix.replace(/:host/, fqdn.reverse());
    }
}
const statsClient = StatsClient.getInstance(global.settings.statsd);

const serverFactory = require('./app/server');

const server = serverFactory(statsClient);
const listener = server.listen(global.settings.node_port, global.settings.node_host);
listener.on('listening', function() {
    console.info("Using Node.js %s", process.version);
    console.info('Using configuration file "%s"', configurationFile);
    console.log(
        "CartoDB SQL API %s listening on %s:%s PID=%d (%s)",
        version, global.settings.node_host, global.settings.node_port, process.pid, ENVIRONMENT
    );
});

process.on('uncaughtException', function(err) {
    global.logger.error('Uncaught exception: ' + err.stack);
});

process.on('SIGHUP', function() {
    global.log4js.clearAndShutdownAppenders(function() {
        global.log4js.configure(log4jsConfig);
        global.logger = global.log4js.getLogger();
        console.log('Log files reloaded');
    });

    if (server.batch && server.batch.logger) {
        server.batch.logger.reopenFileStreams();
    }

    if (server.dataIngestionLogger) {
        server.dataIngestionLogger.reopenFileStreams();
    }
});

addHandlers({ killTimeout: 45000 });

function addHandlers({ killTimeout }) {
    // FIXME: minimize the number of 'uncaughtException' before uncomment the following line
    // process.on('uncaughtException', exitProcess(listener, logger, killTimeout));
    process.on('unhandledRejection', exitProcess({ killTimeout }));
    process.on('SIGINT', exitProcess({ killTimeout }));
    process.on('SIGTERM', exitProcess({ killTimeout }));
}

function exitProcess ({ killTimeout }) {
    return function exitProcessFn (signal) {
        scheduleForcedExit({ killTimeout });

        let code = 0;

        if (!['SIGINT', 'SIGTERM'].includes(signal)) {
            const err = signal instanceof Error ? signal : new Error(signal);
            signal = undefined;
            code = 1;

            global.logger.fatal(err);
        } else {
            global.logger.info(`Process has received signal: ${signal}`);
        }

        listener.close(function () {
            server.batch.stop(function () {
                server.batch.drain(function (err) {
                    if (err) {
                        global.logger.error(err);
                        return process.exit(1);
                    }

                    global.logger.info(`Process is going to exit with code: ${code}`);
                    global.log4js.shutdown(function () {
                        server.batch.logger.end(function () {
                            process.exit(code);
                        });
                    });
                });
            });
        });
    };
}

function scheduleForcedExit ({ killTimeout }) {
    // Schedule exit if there is still ongoing work to deal with
    const killTimer = setTimeout(() => {
        global.logger.info('Process didn\'t close on time. Force exit');
        process.exit(1);
    }, killTimeout);

    // Don't keep the process open just for this
    killTimer.unref();
}

function isGteMinVersion(version, minVersion) {
    const versionMatch = /[a-z]?([0-9]*)/.exec(version);
    if (versionMatch) {
        const majorVersion = parseInt(versionMatch[1], 10);
        if (Number.isFinite(majorVersion)) {
            return majorVersion >= minVersion;
        }
    }
    return false;
}

setInterval(function memoryUsageMetrics () {
    let memoryUsage = process.memoryUsage();

    Object.keys(memoryUsage).forEach(property => {
        statsClient.gauge(`sqlapi.memory.${property}`, memoryUsage[property]);
    });
}, 5000);

function getCPUUsage (oldUsage) {
    let usage;

    if (oldUsage && oldUsage._start) {
        usage = Object.assign({}, process.cpuUsage(oldUsage._start.cpuUsage));
        usage.time = Date.now() - oldUsage._start.time;
    } else {
        usage = Object.assign({}, process.cpuUsage());
        usage.time = process.uptime() * 1000; // s to ms
    }

    usage.percent = (usage.system + usage.user) / (usage.time * 10);

    Object.defineProperty(usage, '_start', {
        value: {
            cpuUsage: process.cpuUsage(),
            time: Date.now()
        }
    });

    return usage;
}

let previousCPUUsage = getCPUUsage();
setInterval(function cpuUsageMetrics () {
    const CPUUsage = getCPUUsage(previousCPUUsage);

    Object.keys(CPUUsage).forEach(property => {
        statsClient.gauge(`sqlapi.cpu.${property}`, CPUUsage[property]);
    });

    previousCPUUsage = CPUUsage;
}, 5000);

if (global.gc && isGteMinVersion(process.version, 6)) {
    const gcInterval = Number.isFinite(global.settings.gc_interval) ?
        global.settings.gc_interval :
        10000;

    if (gcInterval > 0) {
        setInterval(function gcForcedCycle() {
            global.gc();
        }, gcInterval);
    }
}

const gcStats = require('gc-stats')();

gcStats.on('stats', function ({ pauseMS, gctype }) {
    statsClient.timing('sqlapi.gc', pauseMS);
    statsClient.timing(`sqlapi.gctype.${getGCTypeValue(gctype)}`, pauseMS);
});

function getGCTypeValue (type) {
    // 1: Scavenge (minor GC)
    // 2: Mark/Sweep/Compact (major GC)
    // 4: Incremental marking
    // 8: Weak/Phantom callback processing
    // 15: All
    let value;

    switch (type) {
        case 1:
            value = 'Scavenge';
            break;
        case 2:
            value = 'MarkSweepCompact';
            break;
        case 4:
            value = 'IncrementalMarking';
            break;
        case 8:
            value = 'ProcessWeakCallbacks';
            break;
        case 15:
            value = 'All';
            break;
        default:
            value = 'Unkown';
            break;
    }

    return value;
}