diff --git a/app.js b/app.js index 8e231c54..3f6e9e35 100755 --- a/app.js +++ b/app.js @@ -198,7 +198,7 @@ function addHandlers (listener, killTimeout) { process.on('unhandledRejection', (err) => exitProcess(err, listener, 'unhandledRejection', killTimeout)); process.on('ENOMEM', (err) => exitProcess(err, listener, 'ENOMEM', killTimeout)); process.on('SIGINT', () => exitProcess(null, listener, 'SIGINT', killTimeout)); - process.on('SIGTERM', () => exitProcess(null, listener, 'SIGINT', killTimeout)); + process.on('SIGTERM', () => exitProcess(null, listener, 'SIGTERM', killTimeout)); } addHandlers(listener, 45000); diff --git a/lib/api/api-router.js b/lib/api/api-router.js index 57881da7..473dbcb1 100644 --- a/lib/api/api-router.js +++ b/lib/api/api-router.js @@ -199,12 +199,12 @@ module.exports = class ApiRouter { const apiRouter = router({ mergeParams: true }); const { paths, middlewares = [] } = route; + apiRouter.use(user(this.metadataBackend)); apiRouter.use(initLogger({ logger: this.serverOptions.logger })); apiRouter.use(profiler({ enabled: this.serverOptions.useProfiler, statsClient: global.statsClient })); - apiRouter.use(user(this.metadataBackend)); middlewares.forEach(middleware => apiRouter.use(middleware())); @@ -236,10 +236,10 @@ function createTemplateMaps ({ redisPool, surrogateKeysCache, logger }) { const startTime = Date.now(); surrogateKeysCache.invalidate(new NamedMapsCacheEntry(user, templateName), (err) => { if (err) { - return logger.error(err, `Named map (${templateName}) invalidation failed, user: ${user}`); + return logger.error({ exception: err, 'cdb-user': user, template_id: templateName }, 'Named map invalidation failed'); } - logger.info({ user, type: 'named_map_invalidation', elapsed: Date.now() - startTime }, `Named map (${templateName}) invalidation success, user: ${user}`); + logger.info({ 'cdb-user': user, template_id: templateName, duration: (Date.now() - startTime) / 1000 }, 'Named map invalidation success'); }); } diff --git a/lib/api/map/preview-template-controller.js b/lib/api/map/preview-template-controller.js index 2301e7d4..b49e1e7a 100644 --- a/lib/api/map/preview-template-controller.js +++ b/lib/api/map/preview-template-controller.js @@ -343,8 +343,7 @@ function incrementMapViews ({ metadataBackend }) { mapConfigProvider.getMapConfig((err, mapConfig) => { if (err) { - err.message = `Failed to increment mapview count for user '${user}'. ${err.message}`; - logger.warn({ error: err }); + logger.warn({ exception: err }, 'Failed to increment mapview count'); return next(); } @@ -354,8 +353,7 @@ function incrementMapViews ({ metadataBackend }) { metadataBackend.incMapviewCount(user, statTag, (err) => { if (err) { - err.message = `Failed to increment mapview count for user '${user}'. ${err.message}`; - logger.warn({ error: err }); + logger.warn({ exception: err }, 'Failed to increment mapview count'); } next(); diff --git a/lib/api/middlewares/cache-channel-header.js b/lib/api/middlewares/cache-channel-header.js index 622aa053..bc4a8207 100644 --- a/lib/api/middlewares/cache-channel-header.js +++ b/lib/api/middlewares/cache-channel-header.js @@ -10,8 +10,7 @@ module.exports = function setCacheChannelHeader () { mapConfigProvider.getAffectedTables((err, affectedTables) => { if (err) { - err.message = `Error generating Cache Channel Header. ${err.message}`; - logger.warn({ error: err }); + logger.warn({ exception: err }, 'Error generating Cache Channel Header'); return next(); } diff --git a/lib/api/middlewares/cache-control-header.js b/lib/api/middlewares/cache-control-header.js index 5ae0d481..875f78f1 100644 --- a/lib/api/middlewares/cache-control-header.js +++ b/lib/api/middlewares/cache-control-header.js @@ -44,8 +44,7 @@ module.exports = function setCacheControlHeader ({ mapConfigProvider.getAffectedTables((err, affectedTables) => { if (err) { - err.message = `Error generating Cache Control Header. ${err.message}`; - logger.warn({ error: err }); + logger.warn({ exception: err }, 'Error generating Cache Control Header'); return next(); } diff --git a/lib/api/middlewares/error-middleware.js b/lib/api/middlewares/error-middleware.js index 9994a754..3890761f 100644 --- a/lib/api/middlewares/error-middleware.js +++ b/lib/api/middlewares/error-middleware.js @@ -7,7 +7,7 @@ module.exports = function errorMiddleware (/* options */) { const { logger } = res.locals; const errors = populateLimitErrors(Array.isArray(err) ? err : [err]); - logger.error({ error: errors }); + errors.forEach((err) => logger.error({ exception: err }, 'Error while handling the request')); setCommonHeaders(req, res, () => { const errorResponseBody = { diff --git a/lib/api/middlewares/increment-map-view-count.js b/lib/api/middlewares/increment-map-view-count.js index fb8bdacd..e070412b 100644 --- a/lib/api/middlewares/increment-map-view-count.js +++ b/lib/api/middlewares/increment-map-view-count.js @@ -7,8 +7,7 @@ module.exports = function incrementMapViewCount (metadataBackend) { metadataBackend.incMapviewCount(user, statTag, (err) => { if (err) { - err.message = `Failed to increment mapview count for user '${user}'. ${err.message}`; - logger.warn({ error: err }); + logger.warn({ exception: err }, 'Failed to increment mapview count'); } next(); diff --git a/lib/api/middlewares/last-modified-header.js b/lib/api/middlewares/last-modified-header.js index f2d41292..8a224c89 100644 --- a/lib/api/middlewares/last-modified-header.js +++ b/lib/api/middlewares/last-modified-header.js @@ -21,8 +21,7 @@ module.exports = function setLastModifiedHeader () { mapConfigProvider.getAffectedTables((err, affectedTables) => { if (err) { - err.message = `Error generating Last Modified Header. ${err.message}`; - logger.warn({ error: err }); + logger.warn({ exception: err }, 'Error generating Last Modified Header'); return next(); } diff --git a/lib/api/middlewares/logger.js b/lib/api/middlewares/logger.js index c081c8ea..b815595f 100644 --- a/lib/api/middlewares/logger.js +++ b/lib/api/middlewares/logger.js @@ -4,13 +4,9 @@ const uuid = require('uuid'); module.exports = function initLogger ({ logger }) { return function initLoggerMiddleware (req, res, next) { - const id = req.get('X-Request-Id') || uuid.v4(); - res.locals.logger = logger.child({ id }); - - res.locals.logger.info({ request: req }); - res.on('finish', () => res.locals.logger.info({ response: res })); - res.on('close', () => res.locals.logger.info({ end: true })); - + res.locals.logger = logger.child({ request_id: req.get('X-Request-Id') || uuid.v4(), 'cdb-user': res.locals.user }); + res.locals.logger.info({ client_request: req }, 'Incoming request'); + res.on('finish', () => res.locals.logger.info({ server_response: res, status: res.statusCode }, 'Response sent')); next(); }; }; diff --git a/lib/api/middlewares/metrics.js b/lib/api/middlewares/metrics.js index 6b7a9ff8..aea87d61 100644 --- a/lib/api/middlewares/metrics.js +++ b/lib/api/middlewares/metrics.js @@ -23,7 +23,7 @@ module.exports = function metrics ({ enabled, tags, metricsBackend }) { const { event, attributes } = getEventData(req, res, tags); metricsBackend.send(event, attributes) - .catch((err) => logger.error(err, `Failed to publish event "${event}"`)); + .catch((err) => logger.error({ exception: err, event }, 'Failed to publish event')); }); return next(); diff --git a/lib/api/middlewares/profiler.js b/lib/api/middlewares/profiler.js index f4af21d7..163046b6 100644 --- a/lib/api/middlewares/profiler.js +++ b/lib/api/middlewares/profiler.js @@ -21,7 +21,8 @@ module.exports = function profiler (options) { res.on('finish', () => { req.profiler.done('response'); req.profiler.end(); - logger.info({ stats: req.profiler.toJSON() }); + const stats = req.profiler.toJSON(); + logger.info({ stats, duration: stats.response / 1000 }, 'Request profiling stats'); try { // May throw due to dns, see: http://github.com/CartoDB/Windshaft/issues/166 diff --git a/lib/api/middlewares/surrogate-key-header.js b/lib/api/middlewares/surrogate-key-header.js index c8fd550d..3b33885f 100644 --- a/lib/api/middlewares/surrogate-key-header.js +++ b/lib/api/middlewares/surrogate-key-header.js @@ -17,8 +17,7 @@ module.exports = function setSurrogateKeyHeader ({ surrogateKeysCache }) { mapConfigProvider.getAffectedTables((err, affectedTables) => { if (err) { - err.message = `Erros generating Surrogate Key Header. ${err.message}`; - logger.warn({ error: err }); + logger.warn({ exception: err }, 'Error generating Surrogate Key Header'); return next(); } diff --git a/lib/api/middlewares/tag.js b/lib/api/middlewares/tag.js index de5404b1..7c6f182d 100644 --- a/lib/api/middlewares/tag.js +++ b/lib/api/middlewares/tag.js @@ -8,7 +8,7 @@ module.exports = function tag ({ tags }) { return function tagMiddleware (req, res, next) { const { logger } = res.locals; res.locals.tags = tags; - res.on('finish', () => logger.info({ tags: res.locals.tags })); + res.on('finish', () => logger.info({ tags: res.locals.tags }, 'Request tagged')); next(); }; diff --git a/lib/api/middlewares/user.js b/lib/api/middlewares/user.js index 155d3347..bf508085 100644 --- a/lib/api/middlewares/user.js +++ b/lib/api/middlewares/user.js @@ -6,10 +6,8 @@ module.exports = function user (metadataBackend) { const cdbRequest = new CdbRequest(); return function userMiddleware (req, res, next) { - const { logger } = res.locals; try { res.locals.user = getUserNameFromRequest(req, cdbRequest); - logger.info({ user: res.locals.user }); } catch (err) { return next(err); } diff --git a/lib/utils/common-headers.js b/lib/utils/common-headers.js index 5355d452..359e30dc 100644 --- a/lib/utils/common-headers.js +++ b/lib/utils/common-headers.js @@ -3,7 +3,7 @@ module.exports = function setCommonHeaders (req, res, callback) { const { logger } = res.locals; - res.set('X-Request-Id', logger.bindings().id); + res.set('X-Request-Id', logger.bindings().request_id); // TODO: x-layergroupid header?? @@ -39,8 +39,7 @@ module.exports = function setCommonHeaders (req, res, callback) { getStatTag({ res }, (err, statTag) => { if (err) { - err.message = `Error generating Stat Tag header: ${err.message}`; - logger.warn({ error: err }); + logger.warn({ exception: err }, 'Error generating Stat Tag header'); } if (statTag) { @@ -100,8 +99,7 @@ function getTemplateHash ({ res }) { try { templateHash = res.locals.mapConfigProvider.getTemplateHash().substring(0, 8); } catch (err) { - err.message = `Error generating Stat Tag header: ${err.message}`; - logger.warn({ error: err }); + logger.warn({ exception: err }, 'Error generating Stat Tag header'); } return templateHash; diff --git a/lib/utils/logger.js b/lib/utils/logger.js index 3acc1166..0f949858 100644 --- a/lib/utils/logger.js +++ b/lib/utils/logger.js @@ -15,10 +15,21 @@ module.exports = class Logger { const options = { base: null, // Do not bind hostname, pid and friends by default level: LOG_LEVEL || logLevelFromNodeEnv, + formatters: { + level (label) { + if (label === 'warn') { + return { levelname: 'warning' }; + } + + return { levelname: label }; + } + }, + messageKey: 'message', + timestamp: () => `,"timestamp":"${new Date(Date.now()).toISOString()}"`, serializers: { - request: requestSerializer, - response: responseSerializer, - error: (error) => Array.isArray(error) ? error.map((err) => errorSerializer(err)) : [errorSerializer(error)] + client_request: requestSerializer, + server_response: responseSerializer, + exception: errorSerializer } }; const dest = pino.destination({ sync: false }); // stdout diff --git a/metro/config.json b/metro/config.json new file mode 100644 index 00000000..26054e11 --- /dev/null +++ b/metro/config.json @@ -0,0 +1,94 @@ +{ + "metrics": { + "port": 9145, + "definitions": [ + { + "type": "counter", + "options": { + "name": "maps_api_requests_total", + "help": "MAPS API requests total" + }, + "valuePath": "server_response.statusCode", + "shouldMeasure": "({ value }) => Number.isFinite(value)", + "measure": "({ metric }) => metric.inc()" + }, + { + "type": "counter", + "options": { + "name": "maps_api_requests_ok_total", + "help": "MAPS API requests ok total" + }, + "valuePath": "server_response.statusCode", + "shouldMeasure": "({ value }) => value >= 200 && value < 400", + "measure": "({ metric }) => metric.inc()" + }, + { + "type": "counter", + "options": { + "name": "maps_api_requests_errors_total", + "help": "MAPS API requests errors total" + }, + "valuePath": "server_response.statusCode", + "shouldMeasure": "({ value }) => value >= 400", + "measure": "({ metric }) => metric.inc()" + }, + { + "type": "histogram", + "options": { + "name": "maps_api_response_time_total", + "help": "MAPS API response time total" + }, + "valuePath": "stats.response", + "shouldMeasure": "({ value }) => Number.isFinite(value)", + "measure": "({ metric, value }) => metric.observe(value)" + }, + { + "type": "counter", + "options": { + "name": "maps_api_requests", + "help": "MAPS API requests per user", + "labelNames": ["user", "http_code"] + }, + "labelPaths": ["cdb-user", "server_response.statusCode"], + "shouldMeasure": "({ labels }) => labels.every((label) => label !== undefined)", + "measure": "({ metric, labels }) => metric.labels(...labels).inc()" + }, + { + "type": "counter", + "options": { + "name": "maps_api_requests_ok", + "help": "MAPS API requests per user with success HTTP code", + "labelNames": ["user", "http_code"] + }, + "labelPaths": ["cdb-user", "server_response.statusCode"], + "valuePath": "server_response.statusCode", + "shouldMeasure": "({ labels, value }) => labels.every((label) => label !== undefined) && value >= 200 && value < 400", + "measure": "({ metric, labels }) => metric.labels(...labels).inc()" + }, + { + "type": "counter", + "options": { + "name": "maps_api_requests_errors", + "help": "MAPS API requests per user with error HTTP code", + "labelNames": ["user", "http_code"] + }, + "labelPaths": ["cdb-user", "server_response.statusCode"], + "valuePath": "server_response.statusCode", + "shouldMeasure": "({ labels, value }) => labels.every((label) => label !== undefined) && value >= 400", + "measure": "({ metric, labels }) => metric.labels(...labels).inc()" + }, + { + "type": "histogram", + "options": { + "name": "maps_api_response_time", + "help": "MAPS API response time total", + "labelNames": ["user"] + }, + "labelPaths": ["cdb-user"], + "valuePath": "stats.response", + "shouldMeasure": "({ labels, value }) => labels.every((label) => label !== undefined) && Number.isFinite(value)", + "measure": "({ metric, labels, value }) => metric.labels(...labels).observe(value)" + } + ] + } +} diff --git a/metro/index.js b/metro/index.js index 123e147f..d9425d12 100644 --- a/metro/index.js +++ b/metro/index.js @@ -1,29 +1,40 @@ 'use strict'; -const split = require('split2'); -const logCollector = require('./log-collector'); -const metricsCollector = require('./metrics-collector'); +const metro = require('./metro'); +const path = require('path'); +const fs = require('fs'); -const streams = [process.stdin, split(), logCollector(), metricsCollector(), process.stdout]; +const { CONFIG_PATH = path.resolve(__dirname, './config.json') } = process.env; +const existsConfigFile = fs.existsSync(CONFIG_PATH); -pipeline('pipe', streams); +if (!existsConfigFile) { + exit(4)(new Error(`Wrong path for CONFIG_PATH env variable: ${CONFIG_PATH} no such file`)); +} -process.on('SIGINT', exitProcess(0)); -process.on('SIGTERM', exitProcess(0)); -process.on('uncaughtException', exitProcess(1)); -process.on('unhandledRejection', exitProcess(1)); +let config; -function pipeline (action, streams) { - for (let index = 0; index < streams.length - 1; index++) { - const source = streams[index]; - const destination = streams[index + 1]; - source[action](destination); +if (existsConfigFile) { + config = fs.readFileSync(CONFIG_PATH); + try { + config = JSON.parse(config); + } catch (e) { + exit(5)(new Error('Wrong config format: invalid JSON')); } } -function exitProcess (code = 0) { - return function exitProcess (signal) { - pipeline('unpipe', streams); +metro({ metrics: config && config.metrics }) + .then(exit(0)) + .catch(exit(1)); + +process.on('uncaughtException', exit(2)); +process.on('unhandledRejection', exit(3)); + +function exit (code = 1) { + return function (err) { + if (err) { + console.error(err); + } + process.exit(code); }; } diff --git a/metro/log-collector.js b/metro/log-collector.js deleted file mode 100644 index 88123db9..00000000 --- a/metro/log-collector.js +++ /dev/null @@ -1,105 +0,0 @@ -'use strict' - -const fs = require('fs'); -const split = require('split2'); -const assingDeep = require('assign-deep'); -const { Transform } = require('stream'); -const DEV_ENVS = ['test', 'development']; -const dumpPath = `${__dirname}/dump.json`; - -let logs; - -const LEVELS = { - 10: 'trace', - 20: 'debug', - 30: 'info', - 40: 'warn', - 50: 'error', - 60: 'fatal' -}; - -module.exports = function logCollector () { - const stream = 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 + '\n'); - } - return callback(); - } - - const { id } = entry; - - if (id === undefined) { - entry.level = LEVELS[entry.level]; - this.push(`${JSON.stringify(entry)}\n`); - return callback(); - } - - if (logs.has(id)) { - const accEntry = logs.get(id); - const { end } = entry; - - if (end === true) { - accEntry.level = LEVELS[accEntry.level]; - accEntry.time = entry.time; - this.push(`${JSON.stringify(accEntry)}\n`); - logs.delete(id); - return callback(); - } - - if (accEntry.level > entry.level) { - delete entry.level; - } - - if (hasProperty(accEntry, 'error') && hasProperty(entry, 'error')) { - logs.set(id, assingDeep({}, accEntry, entry, { error: accEntry.error.concat(entry.error) })); - } else { - logs.set(id, assingDeep({}, accEntry, entry)); - } - } else { - logs.set(id, entry); - } - - callback(); - } - }); - - stream.on('pipe', () => { - if (!fs.existsSync(dumpPath)) { - logs = new Map(); - return; - } - - try { - const dump = require(dumpPath); - logs = new Map(dump); - } catch (err) { - console.error(`Cannot read the dump for unfinished logs: ${err}`); - logs = new Map(); - } - }); - - stream.on('unpipe', () => { - try { - fs.writeFileSync(dumpPath, JSON.stringify([...logs])); - } catch (err) { - console.error(`Cannot create a dump for unfinished logs: ${err}`); - } - }); - - return stream; -}; - -function hasProperty (obj, prop) { - return Object.prototype.hasOwnProperty.call(obj, prop); -} diff --git a/metro/metrics-collector.js b/metro/metrics-collector.js index e20377e8..168c504d 100644 --- a/metro/metrics-collector.js +++ b/metro/metrics-collector.js @@ -1,121 +1,102 @@ -'use strict' +'use strict'; const http = require('http'); const { Counter, Histogram, register } = require('prom-client'); -const split = require('split2'); -const { Transform } = require('stream'); +const flatten = require('flat'); +const { Transform, PassThrough } = require('stream'); const DEV_ENVS = ['test', 'development']; -const requestCounter = new Counter({ - name: 'maps_api_requests_total', - help: 'MAPS API requests total' -}); +const factory = { + counter: Counter, + histogram: Histogram +}; -const requestOkCounter = new Counter({ - name: 'maps_api_requests_ok_total', - help: 'MAPS API requests ok total' -}); +module.exports = class MetricsCollector { + constructor ({ port = 0, definitions } = {}) { + this._port = port; + this._definitions = definitions; + this._server = null; + this._stream = createTransformStream(this._definitions); + } -const requestErrorCounter = new Counter({ - name: 'maps_api_requests_errors_total', - help: 'MAPS API requests errors total' -}); + get stream () { + return this._stream; + } -const responseTimeHistogram = new Histogram({ - name: 'maps_api_response_time_total', - help: 'MAPS API response time total' -}); + start () { + return new Promise((resolve, reject) => { + this._server = http.createServer((req, res) => { + res.writeHead(200, { 'Content-Type': register.contentType }); + res.end(register.metrics()); + }); -const userRequestCounter = new Counter({ - name: 'maps_api_requests', - help: 'MAPS API requests per user', - labelNames: ['user', 'http_code'] -}); + this._server.once('error', err => reject(err)); + this._server.once('listening', () => resolve()); + this._server.listen(this._port); + }); + } -const userRequestOkCounter = new Counter({ - name: 'maps_api_requests_ok', - help: 'MAPS API requests per user with success HTTP code', - labelNames: ['user', 'http_code'] -}); + stop () { + return new Promise((resolve) => { + register.clear(); + if (!this._server) { + return resolve(); + } -const userRequestErrorCounter = new Counter({ - name: 'maps_api_requests_errors', - help: 'MAPS API requests per user with error HTTP code', - labelNames: ['user', 'http_code'] -}); + this._server.once('close', () => { + this._server = null; + resolve(); + }); -const userResponseTimeHistogram = new Histogram({ - name: 'maps_api_response_time', - help: 'MAPS API response time total', - labelNames: ['user'] -}); + this._server.close(); + }); + }; +}; + +function createTransformStream (definitions) { + if (typeof definitions !== 'object') { + return new PassThrough(); + } + + const metrics = []; + + for (const { type, options, valuePath, labelPaths, shouldMeasure, measure } of definitions) { + metrics.push({ + instance: new factory[type](options), + valuePath, + labelPaths, + shouldMeasure: eval(shouldMeasure), // eslint-disable-line no-eval + measure: eval(measure) // eslint-disable-line no-eval + }); + } -module.exports = function metricsCollector () { 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); + this.push(chunk + '\n'); } return callback(); } - const { request, response, stats } = entry; + const flatEntry = flatten(entry); - if (request === undefined || response === undefined || stats === undefined) { - this.push(chunk); - return callback(); - } + for (const metric of metrics) { + const value = flatEntry[metric.valuePath]; + const labels = Array.isArray(metric.labelPaths) && metric.labelPaths.map(path => flatEntry[path]); - const { statusCode, headers } = response; - const { 'carto-user': user } = headers; - - requestCounter.inc(); - - if (statusCode !== undefined && user !== undefined) { - userRequestCounter.labels(user, `${statusCode}`).inc(); - } - - if (statusCode >= 200 && statusCode < 400) { - requestOkCounter.inc(); - if (user !== undefined) { - userRequestOkCounter.labels(user, `${statusCode}`).inc(); - } - } else if (statusCode >= 400) { - requestErrorCounter.inc(); - if (user !== undefined) { - userRequestErrorCounter.labels(user, `${statusCode}`).inc(); + if (metric.shouldMeasure({ labels, value })) { + metric.measure({ metric: metric.instance, labels, value }); } } - const { response: responseTime } = stats; + this.push(`${JSON.stringify(entry)}\n`); - if (Number.isFinite(responseTime)) { - responseTimeHistogram.observe(responseTime); - userResponseTimeHistogram.labels(user).observe(responseTime); - } - - this.push(chunk); - callback(); + return callback(); } - }) + }); } - -const port = process.env.PORT || 9145; - -http - .createServer((req, res) => { - res.writeHead(200, { 'Content-Type': register.contentType }); - res.end(register.metrics()); - }) - .listen(port) - .unref(); diff --git a/metro/metro.js b/metro/metro.js new file mode 100644 index 00000000..241c49ee --- /dev/null +++ b/metro/metro.js @@ -0,0 +1,19 @@ +'use strict'; + +const util = require('util'); +const stream = require('stream'); +const pipeline = util.promisify(stream.pipeline); +const split = require('split2'); +const MetricsCollector = require('./metrics-collector'); + +module.exports = async function metro ({ input = process.stdin, output = process.stdout, metrics = {} } = {}) { + const metricsCollector = new MetricsCollector(metrics); + const { stream: metricsStream } = metricsCollector; + + try { + await metricsCollector.start(); + await pipeline(input, split(), metricsStream, output); + } finally { + await metricsCollector.stop(); + } +}; diff --git a/package-lock.json b/package-lock.json index bfc05c7e..2aebb203 100644 --- a/package-lock.json +++ b/package-lock.json @@ -814,9 +814,9 @@ "integrity": "sha1-MvxLn82vhF/N9+c7uXysImHwqwo=" }, "camshaft": { - "version": "0.66.0", - "resolved": "https://registry.npmjs.org/camshaft/-/camshaft-0.66.0.tgz", - "integrity": "sha512-25WFuUEiPD9Tw0XWSno/m7qJ7QqaEy9t/WbcXt+j+zMTOxvQ4RqpvCAhmUEbLgTLRyTCTvY8nqPtu4PaVJuIyw==", + "version": "0.67.0", + "resolved": "https://registry.npmjs.org/camshaft/-/camshaft-0.67.0.tgz", + "integrity": "sha512-IiE4y01qZmIJUm88Zy8bi6oNo2Y1lG4cYL/WPiqVT/HOqk5c2FBDpcOZU1LSK8M1KJmNX8zIaoUWVhs40w3HZQ==", "requires": { "async": "^1.5.2", "cartodb-psql": "0.14.0", diff --git a/package.json b/package.json index dbe54e23..7397e982 100644 --- a/package.json +++ b/package.json @@ -39,7 +39,7 @@ "assign-deep": "^1.0.1", "basic-auth": "2.0.0", "body-parser": "1.18.3", - "camshaft": "^0.66.0", + "camshaft": "^0.67.0", "cartodb-psql": "0.14.0", "cartodb-query-tables": "^0.7.0", "cartodb-redis": "^3.0.0",