Merge branch 'improvement/UTAPI-73/wait_for_redis_before_starting_server' into user/tmacro/utapi_metrics

.github/workflows/tests.yaml

@@ -103,7 +103,8 @@ jobs:
         test:
         - name: run unit tests
           command: yarn test
-          env: {}
+          env:
+            UTAPI_METRICS_ENABLED: 'true'
         - name: run client tests
           command: bash ./.github/scripts/run_ft_tests.bash false ft_test:client
           env: {}

libV2/cache/backend/redis.js

@@ -19,7 +19,10 @@ class RedisCache {
         moduleLogger.debug('Connecting to redis...');
         this._redis = new RedisClient(this._options);
         this._redis.connect();
-        return true;
+        return new Promise((resolve, reject) => {
+            this._redis.once('ready', () => resolve(true));
+            this._redis.once('error', error => reject(error));
+        });
     }
 
     async disconnect() {

libV2/models/RequestContext.js

@@ -3,6 +3,7 @@ const Joi = require('@hapi/joi');
 const { buildModel } = require('./Base');
 const { apiOperations } = require('../server/spec');
 const ResponseContainer = require('./ResponseContainer');
+const { httpRequestDurationSeconds } = require('../server/metrics');
 
 const apiTags = Object.keys(apiOperations);
 const apiOperationIds = Object.values(apiOperations)
@@ -21,6 +22,7 @@ const contextSchema = {
     logger: Joi.any(),
     request: Joi.any(),
     results: Joi.any(),
+    requestTimer: Joi.any(),
 };
 
 const RequestContextModel = buildModel('RequestContext', contextSchema);
@@ -34,6 +36,10 @@ class RequestContext extends RequestContextModel {
         const tag = request.swagger.operation['x-router-controller'];
         const { operationId } = request.swagger.operation;
 
+        const requestTimer = tag !== 'internal'
+            ? httpRequestDurationSeconds.startTimer({ action: operationId })
+            : null;
+
         request.logger.logger.addDefaultFields({
             tag,
             operationId,
@@ -50,6 +56,7 @@ class RequestContext extends RequestContextModel {
             encrypted,
             results: new ResponseContainer(),
             logger: request.logger,
+            requestTimer,
         });
     }
 

libV2/server/index.js

@@ -28,6 +28,7 @@ class UtapiServer extends Process {
         app.use(middleware.loggerMiddleware);
         await initializeOasTools(spec, app);
         app.use(middleware.errorMiddleware);
+        app.use(middleware.httpMetricsMiddleware);
         app.use(middleware.responseLoggerMiddleware);
         return app;
     }

libV2/server/metrics.js

@@ -0,0 +1,20 @@
+const promClient = require('prom-client');
+
+const httpRequestsTotal = new promClient.Counter({
+    name: 'utapi_http_requests_total',
+    help: 'Total number of HTTP requests',
+    labelNames: ['action', 'code'],
+});
+
+const httpRequestDurationSeconds = new promClient.Histogram({
+    name: 'utapi_http_request_duration_seconds',
+    help: 'Duration of HTTP requests in seconds',
+    labelNames: ['action', 'code'],
+    // buckets for response time from 0.1ms to 60s
+    buckets: [0.0001, 0.005, 0.015, 0.05, 0.1, 0.2, 0.3, 0.4, 0.5, 1.0, 5.0, 15.0, 30.0, 60.0],
+});
+
+module.exports = {
+    httpRequestDurationSeconds,
+    httpRequestsTotal,
+};

libV2/server/middleware.js

@@ -6,6 +6,7 @@ const config = require('../config');
 const { logger, buildRequestLogger } = require('../utils');
 const errors = require('../errors');
 const { translateAndAuthorize } = require('../vault');
+const metricHandlers = require('./metrics');
 
 const oasOptions = {
     controllers: path.join(__dirname, './API/'),
@@ -55,6 +56,23 @@ function responseLoggerMiddleware(req, res, next) {
     }
 }
 
+function httpMetricsMiddleware(request, response, next) {
+    // If the request.ctx is undefined then this is an internal oasTools request (/_/docs)
+    // No metrics should be pushed
+    if (config.metrics.enabled && request.ctx && request.ctx.tag !== 'internal') {
+        metricHandlers.httpRequestsTotal
+            .labels({
+                action: request.ctx.operationId,
+                code: response.statusCode,
+            }).inc(1);
+        request.ctx.requestTimer({ code: response.statusCode });
+    }
+
+    if (next) {
+        next();
+    }
+}
+
 // next is purposely not called as all error responses are handled here
 // eslint-disable-next-line no-unused-vars
 function errorMiddleware(err, req, res, next) {
@@ -82,7 +100,7 @@ function errorMiddleware(err, req, res, next) {
         code,
         message,
     });
-    responseLoggerMiddleware(req, res);
+    responseLoggerMiddleware(req, res, () => httpMetricsMiddleware(req, res));
 }
 
 // eslint-disable-next-line no-unused-vars
@@ -158,5 +176,6 @@ module.exports = {
         responseLoggerMiddleware,
         authV4Middleware,
         clientIpLimitMiddleware,
+        httpMetricsMiddleware,
     },
 };

libV2/tasks/BaseTask.js

@@ -48,6 +48,11 @@ class BaseTask extends Process {
         }
 
         if (this._enableMetrics) {
+            promClient.collectDefaultMetrics({
+                timeout: 10000,
+                gcDurationBuckets: [0.001, 0.01, 0.1, 1, 2, 5],
+            });
+
             this._metricsHandlers = {
                 ...this._registerDefaultMetricHandlers(),
                 ...this._registerMetricHandlers(),

libV2/tasks/CreateCheckpoint.js

@@ -29,8 +29,26 @@ class CreateCheckpoint extends BaseTask {
             labelNames: ['origin', 'containerName'],
         });
 
+        const getLastCheckpoint = this._getLastCheckpoint.bind(this);
+        const lastCheckpoint = new promClient.Gauge({
+            name: 'utapi_create_checkpoint_last_checkpoint_seconds',
+            help: 'Timestamp of the last successfully created checkpoint',
+            labelNames: ['origin', 'containerName'],
+            async collect() {
+                try {
+                    const timestamp = await getLastCheckpoint();
+                    if (timestamp !== null) {
+                        this.set(timestamp);
+                    }
+                } catch (error) {
+                    logger.error('error during metric collection', { error });
+                }
+            },
+        });
+
         return {
             created,
+            lastCheckpoint,
         };
     }
 
@@ -55,6 +73,25 @@ class CreateCheckpoint extends BaseTask {
         }
     }
 
+    async _getLastCheckpoint() {
+        const resp = await this.withWarp10(async warp10 => warp10.fetch({
+            className: 'utapi.checkpoint.master',
+            labels: {
+                node: warp10.nodeId,
+            },
+            start: 'now',
+            stop: -1,
+        }));
+
+        if (resp.result && (resp.result.length === 0 || resp.result[0] === '' || resp.result[0] === '[]')) {
+            return null;
+        }
+
+        const result = JSON.parse(resp.result[0])[0];
+        const timestamp = result.v[0][0];
+        return timestamp / 1000000;// Convert timestamp from microseconds to seconds
+    }
+
     async _execute(timestamp) {
         logger.debug('creating checkpoints', { checkpointTimestamp: timestamp });
         const status = await this.withWarp10(async warp10 => {

libV2/tasks/CreateSnapshot.js

@@ -29,8 +29,26 @@ class CreateSnapshot extends BaseTask {
             labelNames: ['origin', 'containerName'],
         });
 
+        const getLastSnapshot = this._getLastSnapshot.bind(this);
+        const lastSnapshot = new promClient.Gauge({
+            name: 'utapi_create_snapshot_last_snapshot_seconds',
+            help: 'Timestamp of the last successfully created snapshot',
+            labelNames: ['origin', 'containerName'],
+            async collect() {
+                try {
+                    const timestamp = await getLastSnapshot();
+                    if (timestamp !== null) {
+                        this.set(timestamp);
+                    }
+                } catch (error) {
+                    logger.error('error during metric collection', { error });
+                }
+            },
+        });
+
         return {
             created,
+            lastSnapshot,
         };
     }
 
@@ -55,6 +73,25 @@ class CreateSnapshot extends BaseTask {
         }
     }
 
+    async _getLastSnapshot() {
+        const resp = await this.withWarp10(async warp10 => warp10.fetch({
+            className: 'utapi.snapshot.master',
+            labels: {
+                node: warp10.nodeId,
+            },
+            start: 'now',
+            stop: -1,
+        }));
+
+        if (resp.result && (resp.result.length === 0 || resp.result[0] === '' || resp.result[0] === '[]')) {
+            return null;
+        }
+
+        const result = JSON.parse(resp.result[0])[0];
+        const timestamp = result.v[0][0];
+        return timestamp / 1000000;// Convert timestamp from microseconds to seconds
+    }
+
     async _execute(timestamp) {
         logger.debug('creating snapshots', { snapshotTimestamp: timestamp });
 

libV2/tasks/DiskUsage.js

@@ -1,4 +1,6 @@
 const async = require('async');
+const Path = require('path');
+const fs = require('fs');
 const promClient = require('prom-client');
 const BaseTask = require('./BaseTask');
 const config = require('../config');
@@ -55,9 +57,15 @@ class MonitorDiskUsage extends BaseTask {
             labelNames: ['origin', 'containerName'],
         });
 
-        const diskUsage = new promClient.Gauge({
+        const leveldbBytes = new promClient.Gauge({
-            name: 'utapi_monitor_disk_usage_bytes',
+            name: 'utapi_monitor_disk_usage_leveldb_bytes',
-            help: 'Total bytes used by warp 10',
+            help: 'Total bytes used by warp 10 leveldb',
+            labelNames: ['origin', 'containerName'],
+        });
+
+        const datalogBytes = new promClient.Gauge({
+            name: 'utapi_monitor_disk_usage_datalog_bytes',
+            help: 'Total bytes used by warp 10 datalog',
             labelNames: ['origin', 'containerName'],
         });
 
@@ -75,7 +83,8 @@ class MonitorDiskUsage extends BaseTask {
 
         return {
             isLocked,
-            diskUsage,
+            leveldbBytes,
+            datalogBytes,
             hardLimitRatio,
             hardLimitSetting,
         };
@@ -85,7 +94,8 @@ class MonitorDiskUsage extends BaseTask {
      * Metrics for MonitorDiskUsage
      * @typedef {Object} MonitorDiskUsageMetrics
      * @property {boolean} isLocked - Indicates if writes have been disabled for the monitored warp10
-     * @property {number} diskUsage - Total bytes used by warp 10
+     * @property {number} leveldbBytes - Total bytes used by warp 10 leveldb
+     * @property {number} datalogBytes - Total bytes used by warp 10 datalog
      * @property {number} hardLimitRatio - Percent of the hard limit used by warp 10
      * @property {number} hardLimitSetting - The hard limit setting in bytes
      */
@@ -104,8 +114,12 @@ class MonitorDiskUsage extends BaseTask {
             this._metricsHandlers.isLocked.set(metrics.isLocked ? 1 : 0);
         }
 
-        if (metrics.diskUsage !== undefined) {
+        if (metrics.leveldbBytes !== undefined) {
-            this._metricsHandlers.diskUsage.set(metrics.diskUsage);
+            this._metricsHandlers.leveldbBytes.set(metrics.leveldbBytes);
+        }
+
+        if (metrics.datalogBytes !== undefined) {
+            this._metricsHandlers.datalogBytes.set(metrics.datalogBytes);
         }
 
         if (metrics.hardLimitRatio !== undefined) {
@@ -129,9 +143,13 @@ class MonitorDiskUsage extends BaseTask {
         return this._program.lock !== undefined;
     }
 
-    _getUsage() {
+    // eslint-disable-next-line class-methods-use-this
-        moduleLogger.debug(`calculating disk usage for ${this._path}`);
+    async _getUsage(path) {
-        return getFolderSize(this._path);
+        moduleLogger.debug(`calculating disk usage for ${path}`);
+        if (!fs.existsSync(path)) {
+            throw Error(`failed to calculate usage for non-existent path ${path}`);
+        }
+        return getFolderSize(path);
     }
 
     async _expireMetrics(timestamp) {
@@ -249,16 +267,19 @@ class MonitorDiskUsage extends BaseTask {
             return;
         }
 
-        let size = null;
+        let leveldbBytes = null;
+        let datalogBytes = null;
         try {
-            size = await this._getUsage();
+            leveldbBytes = await this._getUsage(Path.join(this._path, 'leveldb'));
+            datalogBytes = await this._getUsage(Path.join(this._path, 'datalog'));
         } catch (error) {
             moduleLogger.error(`error calculating disk usage for ${this._path}`, { error });
             return;
         }
 
-        this._pushMetrics({ diskUsage: size });
+        this._pushMetrics({ leveldbBytes, datalogBytes });
 
+        const size = leveldbBytes + datalogBytes;
         if (this._hardLimit !== null) {
             moduleLogger.info(`warp 10 leveldb using ${formatDiskSize(size)} of disk space`, { usage: size });
 

tests/functional/hardLimit/testHardLimit.js

@@ -1,4 +1,5 @@
 const assert = require('assert');
+const fs = require('fs');
 const sinon = require('sinon');
 const uuid = require('uuid');
 const promClient = require('prom-client');
@@ -17,6 +18,7 @@ describe('Test MonitorDiskUsage hard limit', function () {
 
     beforeEach(async () => {
         path = `/tmp/diskusage-${uuid.v4()}`;
+        fs.mkdirSync(`${path}/datalog`, { recursive: true });
         promClient.register.clear();
         task = new MonitorDiskUsage({ warp10: warp10Clients, enableMetrics: true });
         await task.setup();
@@ -27,7 +29,7 @@ describe('Test MonitorDiskUsage hard limit', function () {
     afterEach(async () => task.join());
 
     it('should trigger a database lock if above the limit', async () => {
-        fillDir(path, { count: 1, size: 100 });
+        fillDir(`${path}/leveldb`, { count: 1, size: 100 });
         task._hardLimit = 1;
         const checkSpy = sinon.spy(task, '_checkHardLimit');
         const lockSpy = sinon.spy(task, '_disableWarp10Updates');
@@ -43,7 +45,7 @@ describe('Test MonitorDiskUsage hard limit', function () {
     });
 
     it('should trigger a database unlock if below the limit', async () => {
-        fillDir(path, { count: 1, size: 100 });
+        fillDir(`${path}/leveldb`, { count: 1, size: 100 });
         task._hardLimit = 10240;
         const checkSpy = sinon.spy(task, '_checkHardLimit');
         const lockSpy = sinon.spy(task, '_disableWarp10Updates');
@@ -56,7 +58,7 @@ describe('Test MonitorDiskUsage hard limit', function () {
     });
 
     it('should not throw when failing to calculate disk usage', async () => {
-        fillDir(path, { count: 1, size: 100 });
+        fillDir(`${path}/leveldb`, { count: 1, size: 100 });
         task._hardLimit = 1;
         sinon.stub(task, '_getUsage').throws();
         const _task = task.execute();

tests/functional/v2/task/testDiskUsage.js

@@ -1,4 +1,5 @@
 const assert = require('assert');
+const fs = require('fs');
 const promClient = require('prom-client');
 const uuid = require('uuid');
 
@@ -9,9 +10,10 @@ const { fillDir } = require('../../../utils/v2Data');
 const { assertMetricValue } = require('../../../utils/prom');
 
 class MonitorDiskUsageShim extends MonitorDiskUsage {
-    async _getUsage() {
+    async _getUsage(path) {
-        this.usage = await super._getUsage();
+        const usage = await super._getUsage(path);
-        return this.usage;
+        this.usage = (this.usage || 0) + usage;
+        return usage;
     }
 }
 
@@ -45,6 +47,7 @@ describe('Test MonitorDiskUsage', () => {
 
     beforeEach(async () => {
         path = `/tmp/diskusage-${uuid.v4()}`;
+        fs.mkdirSync(path);
         task = new MonitorDiskUsageShim({ warp10: [], enableMetrics: true });
         task._path = path;
         task._enabled = true;
@@ -56,12 +59,23 @@ describe('Test MonitorDiskUsage', () => {
         promClient.register.clear();
     });
 
-    testCases.map(testCase =>
+    testCases.map(testCase => {
         it(`should calculate disk usage for ${testCase.count} files of ${testCase.size} bytes each`,
             async () => {
-                fillDir(path, testCase);
+                fillDir(`${path}/leveldb`, testCase);
+                fillDir(`${path}/datalog`, testCase);
                 await task._execute();
-                assert.strictEqual(task.usage, testCase.expected + emptyDirSize + (emptyFileSize * testCase.count));
+                const expectedSingleSize = emptyDirSize + testCase.expected + (emptyFileSize * testCase.count);
-                await assertMetricValue('utapi_monitor_disk_usage_bytes', task.usage);
+                const expectedTotalSize = expectedSingleSize * 2;
-            }));
+                assert.strictEqual(task.usage, expectedTotalSize);
+                // Should equal the usage minus the empty datalog
+                await assertMetricValue('utapi_monitor_disk_usage_leveldb_bytes', expectedSingleSize);
+                await assertMetricValue('utapi_monitor_disk_usage_datalog_bytes', expectedSingleSize);
+            });
+    });
+
+    it('should fail if a subpath does not exists', () => {
+        assert.doesNotThrow(() => task._execute());
+        assert.strictEqual(task.usage, undefined);
+    });
 });

tests/unit/v2/models/testRequestContext.js

@@ -10,6 +10,7 @@ const templateExpected = opts => ({
     operationId: 'healthcheck',
     tag: 'internal',
     encrypted: false,
+    requestTimer: null,
     ...(opts || {}),
 });
 

tests/unit/v2/server/testMiddleware.js

@@ -1,8 +1,11 @@
 const assert = require('assert');
 const sinon = require('sinon');
+const promClient = require('prom-client');
 
 const { middleware } = require('../../../../libV2/server/middleware');
 const { templateRequest, ExpressResponseStub } = require('../../../utils/v2Data');
+const RequestContext = require('../../../../libV2/models/RequestContext');
+const { getMetricValues, assertMetricValue } = require('../../../utils/prom');
 
 describe('Test middleware', () => {
     it('should build a request logger', next => {
@@ -86,4 +89,55 @@ describe('Test middleware', () => {
             }));
         });
     });
+
+    describe('test httpMetricsMiddleware', () => {
+        let resp;
+
+        beforeEach(() => {
+            resp = new ExpressResponseStub();
+            resp.status(200);
+        });
+
+        afterEach(() => {
+            promClient.register.clear();
+        });
+
+        it('should increment the counter if not an internal route', async () => {
+            const req = templateRequest({
+                swagger: {
+                    operation: {
+                        'x-router-controller': 'metrics',
+                        'operationId': 'listMetrics',
+                    },
+                },
+            });
+
+            req.ctx = new RequestContext(req);
+            middleware.httpMetricsMiddleware(req, resp);
+            await assertMetricValue('utapi_http_requests_total', 1);
+            const duration = await getMetricValues('utapi_http_request_duration_seconds');
+            // 14 defined buckets + 1 for Infinity
+            assert.strictEqual(
+                duration.filter(metric => metric.metricName === 'utapi_http_request_duration_seconds_bucket').length,
+                15,
+            );
+            const count = duration.filter(metric => metric.metricName === 'utapi_http_request_duration_seconds_count');
+            assert.deepStrictEqual(count, [{
+                labels: {
+                    action: 'listMetrics',
+                    code: 200,
+                },
+                metricName: 'utapi_http_request_duration_seconds_count',
+                value: 1,
+            }]);
+            assert.strictEqual(count[0].value, 1);
+        });
+
+        it('should not increment the counter if an internal route', async () => {
+            const req = templateRequest();
+            req.ctx = new RequestContext(req);
+            middleware.httpMetricsMiddleware(req, resp);
+            assert.rejects(() => getMetricValues('utapi_http_requests_total'));
+        });
+    });
 });

warpscript/utapi/createCheckpoint.mc2

@@ -115,7 +115,16 @@
       %> FOREACH
      %>
     <%
-       DROP 0 STOP
+      // If no new events were found
+      // - drop the empty list
+      // - write a new master checkpoint
+      // - return 0
+      DROP
+      NEWGTS $master_checkpoint_class RENAME
+      $endTimestamp NaN NaN NaN 0 ADDVALUE
+      { 'node' $nodeId } RELABEL
+      $write_token UPDATE
+      0 STOP
     %> IFTE
 
     0 'checkpoints' STORE