inceptum/dist/db/ConnectionPool.js

"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
const prom_client_1 = require("prom-client");
const prometheus_extended_gauge_1 = require("prometheus-extended-gauge");
const generic_pool_1 = require("generic-pool");
/**
 * Sensible defaults for the connection pool options
 */
exports.DEFAULT_CONNECTION_POOL_OPTIONS = {
    max: 10,
    min: 2,
    maxWaitingClients: 10,
    testOnBorrow: true,
    acquireTimeoutMillis: 1000,
    evictionRunIntervalMillis: 30000,
    numTestsPerRun: 2,
    idleTimeoutMillis: 30000,
};
class ConnectionPool {
}
exports.ConnectionPool = ConnectionPool;
const connectErrorsCounter = new prom_client_1.Counter({
    name: 'db_pool_connect_errors_counter',
    help: 'Number of times a connection attempt fails',
    labelNames: ['poolName', 'readonly'],
});
const acquireErrorsCounter = new prom_client_1.Counter({
    name: 'db_pool_acquire_errors_counter',
    help: 'Number of times an acquire attempt fails',
    labelNames: ['poolName', 'readonly'],
});
const validateFailedCounter = new prom_client_1.Counter({
    name: 'db_pool_validate_failed_counter',
    help: 'Number of times a validation fails',
    labelNames: ['poolName', 'readonly'],
});
const activeGauge = new prometheus_extended_gauge_1.ExtendedGauge({
    name: 'db_pool_active_connections_gauge',
    help: 'Number of active connections in the pool',
    labelNames: ['poolName', 'readonly'],
    average: true,
    max: true,
    min: false,
    bucketSizeMillis: 1000,
    numBuckets: 60,
});
const totalGauge = new prom_client_1.Gauge({
    name: 'db_pool_total_connections_gauge',
    help: 'Number of total connections in the pool',
    labelNames: ['poolName', 'readonly'],
});
const acquireTimeHistogram = new prom_client_1.Histogram({
    name: 'db_pool_acquire_time_histogram',
    help: 'Time required to acquire a connection from the pool',
    labelNames: ['poolName', 'readonly'],
    buckets: [0.003, 0.005, 0.01, 0.05, 0.1, 0.3],
});
const connectTimeHistogram = new prom_client_1.Histogram({
    name: 'db_pool_connect_time_histogram',
    help: 'Time required to establish a new connection to the DB',
    labelNames: ['poolName', 'readonly'],
    buckets: [0.003, 0.005, 0.01, 0.05, 0.1, 0.3],
});
const useTimeHistogram = new prom_client_1.Histogram({
    name: 'db_pool_use_time_histogram',
    help: 'Time a connection is kept out of the pool',
    labelNames: ['poolName', 'readonly'],
    buckets: [0.003, 0.005, 0.01, 0.05, 0.1, 0.3],
});
const maxConnectionLimitGauge = new prom_client_1.Gauge({
    name: 'db_pool_max_pool_size_gauge',
    help: 'Max number of total connections in the pool',
    labelNames: ['poolName', 'readonly'],
});
class InstrumentedFactory {
    constructor(factory, name, readonly) {
        this.factory = factory;
        const labels = [name, readonly ? 'true' : 'false'];
        this.connectTimeHistogram = connectTimeHistogram.labels(...labels);
        this.connectErrorsCounter = connectErrorsCounter.labels(...labels);
        this.totalGauge = totalGauge.labels(...labels);
        this.validateFailedCounter = validateFailedCounter.labels(...labels);
    }
    async create() {
        const timer = this.connectTimeHistogram.startTimer();
        try {
            const connection = await this.factory.create();
            this.totalGauge.inc();
            return connection;
        }
        catch (e) {
            this.connectErrorsCounter.inc();
            throw e;
        }
        finally {
            timer();
        }
    }
    async destroy(client) {
        try {
            return await this.factory.destroy(client);
        }
        finally {
            this.totalGauge.dec();
        }
    }
    async validate(client) {
        if (this.factory.validate) {
            const resp = await this.factory.validate(client);
            if (!resp) {
                this.validateFailedCounter.inc();
            }
            return resp;
        }
        return true;
    }
}
exports.InstrumentedFactory = InstrumentedFactory;
var PoolStatus;
(function (PoolStatus) {
    PoolStatus[PoolStatus["NOT_STARTED"] = 0] = "NOT_STARTED";
    PoolStatus[PoolStatus["STARTED"] = 1] = "STARTED";
    PoolStatus[PoolStatus["STOPPED"] = 2] = "STOPPED";
})(PoolStatus || (PoolStatus = {}));
class InstrumentedConnectionPool extends ConnectionPool {
    constructor(factory, options, name, readonly) {
        super();
        this.status = PoolStatus.NOT_STARTED;
        this.name = name;
        this.readonly = readonly;
        this.options = Object.assign({}, exports.DEFAULT_CONNECTION_POOL_OPTIONS, options);
        // force to int because env values from k8s are string.
        const maxString = this.options.max.toString();
        this.options.max = parseInt(maxString, 10);
        const labels = [name, readonly ? 'true' : 'false'];
        const instrumentedFactory = new InstrumentedFactory(factory, name, readonly);
        this.pool = generic_pool_1.createPool(instrumentedFactory, this.getGenericPoolOptions());
        this.acquireTimeHistogram = acquireTimeHistogram.labels(...labels);
        this.acquireErrorsCounter = acquireErrorsCounter.labels(...labels);
        this.useTimeHistogram = useTimeHistogram.labels(...labels);
        this.activeGauge = activeGauge.labels(...labels);
        this.maxConnectionLimitGauge = maxConnectionLimitGauge.labels(...labels);
    }
    getName() {
        return this.name;
    }
    isReadonly() {
        return this.readonly;
    }
    async getConnection() {
        if (this.status !== PoolStatus.STARTED) {
            throw new Error(`Can't acquire connections from connection pool ${this.name}. The pool is not started`);
        }
        const timer = this.acquireTimeHistogram.startTimer();
        try {
            const connection = await this.pool.acquire();
            connection['__useTimer'] = this.useTimeHistogram.startTimer();
            this.activeGauge.inc();
            this.maxConnectionLimitGauge.set(this.options.max);
            return connection;
        }
        catch (e) {
            this.acquireErrorsCounter.inc();
            throw e;
        }
        finally {
            timer();
        }
    }
    getGenericPoolOptions() {
        return {
            acquireTimeoutMillis: this.options.acquireTimeoutMillis,
            autostart: false,
            evictionRunIntervalMillis: this.options.evictionRunIntervalMillis,
            fifo: false,
            idleTimeoutMillis: this.options.idleTimeoutMillis,
            max: this.options.max,
            maxWaitingClients: this.options.maxWaitingClients,
            min: this.options.min,
            numTestsPerRun: this.options.numTestsPerRun,
            softIdleTimeoutMillis: this.options.softIdleTimeoutMillis,
            testOnBorrow: this.options.testOnBorrow,
        };
    }
    release(connection) {
        this.activeGauge.dec();
        if (connection['__useTimer']) {
            connection['__useTimer']();
        }
        this.pool.release(connection);
    }
    async start() {
        if (this.status !== PoolStatus.NOT_STARTED) {
            throw new Error(`Can't start a connection pool that isn't in NOT_STARTED state. Pool: ${this.name}, Current Status: ${this.status}`);
        }
        this.status = PoolStatus.STARTED;
        this.pool['start']();
        // The pool needs to get into a good state. Waiting a bit has proven a good solution.
        return new Promise((resolve) => setTimeout(resolve, 30));
    }
    async stop() {
        if (this.status !== PoolStatus.STARTED) {
            throw new Error(`Can't stop a connection pool that isn't in STARTED state. Pool: ${this.name}, Current Status: ${this.status}`);
        }
        this.status = PoolStatus.STOPPED;
        await this.pool.drain();
        await this.pool.clear();
    }
    getOptions() {
        return Object.assign({}, this.options);
    }
}
exports.InstrumentedConnectionPool = InstrumentedConnectionPool;
//# sourceMappingURL=ConnectionPool.js.map