Merge pull request #1183 from CartoDB/feature/ch91877/remove-log-aggregation-in-metro
Metro: stop aggregating logs per request idremotes/origin/alasarr/gitignore
Daniel G. Aubert
4 years ago
committed by
GitHub
commit
d5c5d07507
@ -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)"
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
@ -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);
|
||||
};
|
||||
}
|
||||
|
||||
@ -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);
|
||||
}
|
||||
@ -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();
|
||||
}
|
||||
};
|
||||
Loading…
Reference in new issue