Introducing @carto/metro, the carto logs and metrics transport.

lib/api/api-router.js

@@ -199,20 +199,21 @@ module.exports = class ApiRouter {
             const apiRouter = router({ mergeParams: true });
             const { paths, middlewares = [] } = route;
 
-            middlewares.forEach(middleware => apiRouter.use(middleware()));
-
             apiRouter.use(initLogger({ logger: this.serverOptions.logger }));
-            apiRouter.use(initializeStatusCode());
-            apiRouter.use(bodyParser.json());
-            apiRouter.use(servedByHostHeader());
-            apiRouter.use(clientHeader());
             apiRouter.use(profiler({
                 enabled: this.serverOptions.useProfiler,
                 statsClient: global.statsClient
             }));
+            apiRouter.use(user(this.metadataBackend));
+
+            middlewares.forEach(middleware => apiRouter.use(middleware()));
+
+            apiRouter.use(initializeStatusCode());
+            apiRouter.use(bodyParser.json());
+            apiRouter.use(servedByHostHeader());
+            apiRouter.use(clientHeader());
             apiRouter.use(lzmaMiddleware());
             apiRouter.use(cors());
-            apiRouter.use(user(this.metadataBackend));
 
             this.templateRouter.route(apiRouter, route.template);
             this.mapRouter.route(apiRouter, route.map);

metro/index.js

@@ -0,0 +1,11 @@
+'use strict';
+
+const split = require('split2');
+const logCollector = require('./log-collector');
+const metricsCollector = require('./metrics-collector');
+
+process.stdin
+    .pipe(split())
+    .pipe(logCollector())
+    .pipe(metricsCollector())
+    .pipe(process.stdout);

metro/log-collector.js

@@ -2,7 +2,7 @@
 
 const split = require('split2');
 const assingDeep = require('assign-deep');
-const { Transform } = require('readable-stream');
+const { Transform } = require('stream');
 const DEV_ENVS = ['test', 'development'];
 const logs = new Map();
 
@@ -15,9 +15,9 @@ const LEVELS = {
     60: 'fatal'
 }
 
-function logTransport () {
+module.exports = function logCollector () {
     return new Transform({
-        transform: function transform (chunk, enc, callback) {
+        transform (chunk, enc, callback) {
             let entry;
 
             try {
@@ -76,8 +76,3 @@ function logTransport () {
 function hasProperty(obj, prop) {
     return Object.prototype.hasOwnProperty.call(obj, prop)
 }
-
-process.stdin
-    .pipe(split())
-    .pipe(logTransport())
-    .pipe(process.stdout);

metro/metrics-collector.js

@@ -0,0 +1,119 @@
+'use strict'
+
+const express = require('express');
+const { Counter, Histogram, register } = require('prom-client');
+const split = require('split2');
+const { Transform } = require('stream');
+const DEV_ENVS = ['test', 'development'];
+
+const requestCounter = new Counter({
+    name: 'maps_api_requests_total',
+    help: 'MAPS API requests total'
+});
+
+const requestOkCounter = new Counter({
+    name: 'maps_api_requests_ok_total',
+    help: 'MAPS API requests ok total'
+});
+
+const requestErrorCounter = new Counter({
+    name: 'maps_api_requests_errors_total',
+    help: 'MAPS API requests errors total'
+});
+
+const responseTimeHistogram = new Histogram({
+    name: 'maps_api_response_time_total',
+    help: 'MAPS API response time total'
+});
+
+const userRequestCounter = new Counter({
+    name: 'maps_api_requests',
+    help: 'MAPS API requests per user',
+    labelNames: ['user', 'http_code']
+});
+
+const userRequestOkCounter = new Counter({
+    name: 'maps_api_requests_ok',
+    help: 'MAPS API requests per user with success HTTP code',
+    labelNames: ['user', 'http_code']
+});
+
+const userRequestErrorCounter = new Counter({
+    name: 'maps_api_requests_errors',
+    help: 'MAPS API requests per user with error HTTP code',
+    labelNames: ['user', 'http_code']
+});
+
+const userResponseTimeHistogram = new Histogram({
+    name: 'maps_api_response_time',
+    help: 'MAPS API response time total',
+    labelNames: ['user']
+});
+
+module.exports = function updateMetrics () {
+    return new Transform({
+        transform (chunk, enc, callback) {
+            let entry;
+
+            try {
+                entry = JSON.parse(chunk);
+                const { level, time } = entry;
+
+                if (level === undefined && time === undefined) {
+                    throw new Error('Entry log is not valid');
+                }
+            } catch (e) {
+                if (DEV_ENVS.includes(process.env.NODE_ENV)) {
+                    this.push(chunk);
+                }
+                return callback();
+            }
+
+            const { request, response, stats } = entry;
+
+            if (request === undefined || response === undefined || stats === undefined) {
+                this.push(chunk + '\n');
+                return callback();
+            }
+
+            const { statusCode, headers } = response;
+            const { 'carto-user': user } = headers;
+
+            if (statusCode === undefined || headers === undefined || user === undefined) {
+                this.push(chunk + '\n');
+                return callback();
+            }
+
+            requestCounter.inc();
+            userRequestCounter.labels(user, `${statusCode}`).inc();
+
+            if (statusCode >= 200 && statusCode < 400) {
+                requestOkCounter.inc();
+                userRequestOkCounter.labels(user, `${statusCode}`).inc();
+            } else {
+                requestErrorCounter.inc();
+                userRequestErrorCounter.labels(user, `${statusCode}`).inc();
+            }
+
+            const { response: responseTime } = stats;
+
+            if (Number.isFinite(responseTime)) {
+                responseTimeHistogram.observe(responseTime);
+                userResponseTimeHistogram.labels(user).observe(responseTime);
+            }
+
+            this.push(chunk);
+            callback();
+        }
+    })
+}
+
+const port = process.env.PORT || 9144;
+const app = express();
+
+app.get('/metrics', (req, res) => {
+    res.set('Content-Type', register.contentType);
+    res.end(register.metrics());
+});
+
+app.listen(port);

package-lock.json

@@ -719,6 +719,11 @@
       "integrity": "sha512-Phlt0plgpIIBOGTT/ehfFnbNlfsDEiqmzE2KRXoX1bLIlir4X/MR+zSyBEkL05ffWgnRSf/DXv+WrUAVr93/ow==",
       "dev": true
     },
+    "bintrees": {
+      "version": "1.0.1",
+      "resolved": "https://registry.npmjs.org/bintrees/-/bintrees-1.0.1.tgz",
+      "integrity": "sha1-DmVcm5wkNeqraL9AJyJtK1WjRSQ="
+    },
     "body-parser": {
       "version": "1.18.3",
       "resolved": "https://registry.npmjs.org/body-parser/-/body-parser-1.18.3.tgz",
@@ -5409,6 +5414,14 @@
       "integrity": "sha512-7PiHtLll5LdnKIMw100I+8xJXR5gW2QwWYkT6iJva0bXitZKa/XMrSbdmg3r2Xnaidz9Qumd0VPaMrZlF9V9sA==",
       "dev": true
     },
+    "prom-client": {
+      "version": "12.0.0",
+      "resolved": "https://registry.npmjs.org/prom-client/-/prom-client-12.0.0.tgz",
+      "integrity": "sha512-JbzzHnw0VDwCvoqf8y1WDtq4wSBAbthMB1pcVI/0lzdqHGJI3KBJDXle70XK+c7Iv93Gihqo0a5LlOn+g8+DrQ==",
+      "requires": {
+        "tdigest": "^0.1.1"
+      }
+    },
     "propagate": {
       "version": "1.0.0",
       "resolved": "https://registry.npmjs.org/propagate/-/propagate-1.0.0.tgz",
@@ -6398,6 +6411,14 @@
         }
       }
     },
+    "tdigest": {
+      "version": "0.1.1",
+      "resolved": "https://registry.npmjs.org/tdigest/-/tdigest-0.1.1.tgz",
+      "integrity": "sha1-Ljyyw56kSeVdHmzZEReszKRYgCE=",
+      "requires": {
+        "bintrees": "1.0.1"
+      }
+    },
     "test-exclude": {
       "version": "5.2.3",
       "resolved": "https://registry.npmjs.org/test-exclude/-/test-exclude-5.2.3.tgz",

package.json

@@ -53,6 +53,7 @@
     "lzma": "2.3.2",
     "node-statsd": "0.1.1",
     "pino": "^6.3.1",
+    "prom-client": "^12.0.0",
     "queue-async": "1.1.0",
     "redis-mpool": "^0.8.0",
     "request": "2.87.0",