trentm/mockapmserver.js Secret

## mockapmserver.js
// See: https://gist.github.com/trentm/4b8c54a8bdb1c1eba2ade871253860f6
//
// A very simple mock apm-server.
//
// It will listen on the default 8200 port, respond 200 to any request with a
// `{}` body, and will log (in Bunyan format) the HTTP request and response.
// Importantly that log will include the *request body*, which is useful to
// see what the APM Agent is sending to the server.
//
// Usage:
//      npm install -g bunyan  # doesn't have to be installed globally
//      node mockapmserver.js | bunyan
//

const http = require('http');
const url = require('url');
const util = require('util');
const zlib = require('zlib');

const bunyan = require('bunyan');

const OPT_PRETTY_PRINT_REQUEST_BODY = true
// Render text tree for transactions+spans in a single intake request.
const OPT_RENDER_WATERFALL = true
const OPT_HANG_ON_RESPONSE = false
const OPT_QUEUE_IS_FULL = false

const MAX_REQ_BODY_LENTGH = 2 * 10240
const MAX_RES_BODY_LENTGH = 1024
const log = bunyan.createLogger({
    name: 'mockapmserver',
    serializers: {
        err: bunyan.stdSerializers.err,
        req: function (req) {
            if (!req || !req.connection)
                return req;
            let repr = {
                method: req.method,
                // Accept `req.originalUrl` for expressjs usage.
                // https://expressjs.com/en/api.html#req.originalUrl
                url: req.originalUrl || req.url,
                headers: req.headers,
                remoteAddress: req.connection.remoteAddress,
                remotePort: req.connection.remotePort,
                bodyLength: req.body && req.body.length
            };
            if (req.body && req.body.length > MAX_REQ_BODY_LENTGH) {
                repr.body = req.body.slice(0, MAX_REQ_BODY_LENTGH) +
                    util.format(' ... (clipped at %d, full length is %d chars)',
                        MAX_REQ_BODY_LENTGH, req.body.length);
            } else if (OPT_PRETTY_PRINT_REQUEST_BODY) {
                if (typeof req.body === 'string' && req.body.startsWith('{')) {
                  try {
                    repr.body = req.body
                      .split(/\n/g)
                      .map(line => line.trim() && JSON.stringify(JSON.parse(line), null, 4))
                      .join('\n')
                  } catch {
                    repr.body = req.body
                  }
                }
            } else {
                repr.body = req.body
            }
            if (req.body && OPT_RENDER_WATERFALL) {
                repr.waterfall = '\n' + renderWaterfall(req.body)
            }
            return repr;
        },
        res: function (res) {
            if (!res || !res.statusCode) {
                return res;
            }
            let repr = {
                statusCode: res.statusCode,
                header: res._header
            };
            if (typeof(res._data) === 'string') {
                repr.bodyLength = res._data.length
                if (res._data.length > MAX_RES_BODY_LENTGH) {
                    repr.body = res._data.slice(0, MAX_RES_BODY_LENTGH) +
                        util.format(' ... (clipped at %d, full length is %d chars)',
                            MAX_RES_BODY_LENTGH, res._data.length);
                } else {
                    repr.body = res._data
                }
            }
            return repr
        }
    },
    level: 'debug',
    stream: process.stdout
});


// // LineStream
// var stream = require('stream');
// function LineStream(opts) {
//     if (!(this instanceof LineStream)) {
//         throw new Error('use `new LineStream(opts)`');
//     }
//     stream.Transform.call(this, opts);
//     this._buf = '';
// };
// util.inherits(LineStream, stream.Transform);
// LineStream.prototype._transform = function(chunk, encoding, done) {
//     // XXX need toString? Can we do this with raw Buffer for perf?
//     var data = this._buf + chunk.toString('utf8');
//     var lines = data.split(/\r?\n|\r(?!\n)/);
//     this._buf = lines.pop();
//     for (var i = 0; i < lines.length; i++) {
//         this.push(lines[i] + '\n');
//     }
//     done();
// };
// LineStream.prototype._flush = function(done) {
//     if (this._buf) {
//         this.push(this._buf);
//     }
//     done();
// };

function renderWaterfall(body) {
    const OPT_EXTRA_FIELDS = false

    function addChild(span, childSpan) {
        if (!span.children) {
            span.children = [childSpan]
        } else {
            span.children.push(childSpan)
        }
    }
    function renderError(error, prefix) {
        let r = `${prefix}error ${error.id.slice(0, 6)} "${error.exception.message}"`;
        if (OPT_EXTRA_FIELDS) {
            r += ` (type=${error.exception.type}, parent_id=${error.parent_id})`;
        }
        (error.children || []).forEach(s => {
            r += '\n' + renderSpanOrError(s, '  ' + prefix)
        })
        return r
    }
    function renderSpanOrError(spanOrError, prefix) {
        if (spanOrError.name === undefined) {
            return renderError(spanOrError, prefix)
        } else {
            return renderSpan(spanOrError, prefix)
        }
    }
    function renderSpan(span, prefix) {
        let r = `${prefix}span ${span.id.slice(0, 6)} "${span.name}"`;
        if (OPT_EXTRA_FIELDS) {
            r += ` (parent_id=${span.parent_id})`;
        }
        (span.children || []).forEach(s => {
            r += '\n' + renderSpanOrError(s, '  ' + prefix)
        })
        return r
    }
    function renderTx(tx) {
        let r = `transaction ${tx.id.slice(0, 6)} "${tx.name}"`;
        if (OPT_EXTRA_FIELDS) {
            r += ` (trace_id=${tx.trace_id})`;
        }
        (tx.children || []).forEach(s => {
            r += '\n' + renderSpanOrError(s, '`- ')
        })
        tx.orphans.forEach(s => {
            r += '\n' + renderSpanOrError(s, 'ORPHAN: ')
        })
        return r
    }

    const events = body
        .split(/\n/g)
        .filter(line => line.trim(line))
        .map(JSON.parse)
    const allSpansAndErrors = events
        .filter(e => e.span || e.error)
        .map(e => e.span || e.error)
        .sort((a, b) => a.timestamp - b.timestamp)
    const txs = events.filter(e => e.transaction)
        .map(e => e.transaction)
        .sort((a, b) => a.timestamp - b.timestamp)
        .map(tx => {
            tx.orphans = []
            const spans = allSpansAndErrors.filter(s => s.trace_id === tx.trace_id)
            const spanOrErrorFromId = { [tx.id]: tx }
            spans.forEach(s => { spanOrErrorFromId[s.id] = s })
            spans.forEach(s => {
                const parent = spanOrErrorFromId[s.parent_id]
                if (parent) {
                    addChild(parent, s)
                } else {
                    tx.orphans.push(s)
                }
            })
            return tx
        })
    const rendering = txs
        .map(tx => renderTx(tx))
        .join('\n')
    return rendering
}

const server = http.createServer(function (req, res) {
    // return // respond nevermore

    var parsedUrl = url.parse(req.url);

    var instream = req;
    if (req.headers['content-encoding'] === 'gzip') {
        instream = req.pipe(zlib.createGunzip())
    } else {
        instream.setEncoding('utf8');
    }

    // // Look at individual *lines* stream in on a chunked request.
    // lstream = new LineStream()
    // lstream.on('data', function (line) {
    //   console.warn('XXX line', line)
    // })

    if (OPT_QUEUE_IS_FULL) {
        // APM server "queue is full" error response example from:
        // https://www.elastic.co/guide/en/apm/server/current/events-api.html#events-api-errors
        setTimeout(function () {
            res.writeHead(503, {
                'Content-Type': 'application/json'
            })
            resBody = JSON.stringify({
                "errors": [
                    {
                        "message": "<json-schema-err>",
                        "document": "<ndjson-obj>"
                    },{
                        "message": "<json-schema-err>",
                        "document": "<ndjson-obj>"
                    },{
                        "message": "<json-decoding-err>",
                        "document": "<ndjson-obj>"
                    },{
                        "message": "queue is full"
                    },
                ],
                "accepted": 2320
            })
            res.end(resBody)
            res._data = resBody; // for audit logging
            log.info({req, res}, 'request')
        }, 2000)
    }

    let body = '';
    instream.on('data', (chunk) => {
        //console.warn('XXX got a chunk: %d bytes', chunk.length, Buffer.byteLength(chunk))
        body += chunk;
        // lstream.write(chunk)
    });
    instream.on('end', function () {
        req.body = body;
        var resBody;

        // TODO: add support for '/' path for server version detection
        // as done by https://github.com/elastic/apm-agent-python/pull/1194

        if (parsedUrl.pathname === '/config/v1/agents') {
            // Central config mocking.
            if (process.env.MOCK_CENTRAL_CONFIG_FAILS) {
                resBody = '{"ok":false,"message":"The requested resource is currently unavailable."}\n'
                res.writeHead(503)
            } else if (process.env.MOCK_CENTRAL_CONFIG_INVALID_JSON) {
                res.writeHead(200)
                resBody = '{"log_level":'
            } else {
                res.writeHead(200)
                // resBody = '{"log_level": "debug"}'
                resBody = '{}'
            }
        } else if (req.method === 'POST' && parsedUrl.pathname === '/intake/v2/events') {
            resBody = '{}'
            res.writeHead(202)
        } else if (req.method === 'POST' && parsedUrl.pathname === '/intake/v2/profile') {
            resBody = '{"accepted": 1}'
            res.writeHead(202)
        }
        if (OPT_HANG_ON_RESPONSE) { // Actually respond. Set to false to hang the request.
            log.warn('intentionally hanging response')
        } else {
            res.end(resBody);
        }

        res._data = resBody; // for audit logging
        log.info({req, res}, 'request')
    });
})

let PORT = 8200; // default APM server port
//PORT = 8201;
//PORT = 80
server.listen(PORT, function () {
    log.info('listening', server.address())
})
	// See: https://gist.github.com/trentm/4b8c54a8bdb1c1eba2ade871253860f6
	//
	// A very simple mock apm-server.
	//
	// It will listen on the default 8200 port, respond 200 to any request with a
	// `{}` body, and will log (in Bunyan format) the HTTP request and response.
	// Importantly that log will include the request body, which is useful to
	// see what the APM Agent is sending to the server.
	//
	// Usage:
	// npm install -g bunyan # doesn't have to be installed globally
	// node mockapmserver.js \| bunyan
	//

	const http = require('http');
	const url = require('url');
	const util = require('util');
	const zlib = require('zlib');

	const bunyan = require('bunyan');

	const OPT_PRETTY_PRINT_REQUEST_BODY = true
	// Render text tree for transactions+spans in a single intake request.
	const OPT_RENDER_WATERFALL = true
	const OPT_HANG_ON_RESPONSE = false
	const OPT_QUEUE_IS_FULL = false

	const MAX_REQ_BODY_LENTGH = 2 * 10240
	const MAX_RES_BODY_LENTGH = 1024
	const log = bunyan.createLogger({
	name: 'mockapmserver',
	serializers: {
	err: bunyan.stdSerializers.err,
	req: function (req) {
	if (!req \|\| !req.connection)
	return req;
	let repr = {
	method: req.method,
	// Accept `req.originalUrl` for expressjs usage.
	// https://expressjs.com/en/api.html#req.originalUrl
	url: req.originalUrl \|\| req.url,
	headers: req.headers,
	remoteAddress: req.connection.remoteAddress,
	remotePort: req.connection.remotePort,
	bodyLength: req.body && req.body.length
	};
	if (req.body && req.body.length > MAX_REQ_BODY_LENTGH) {
	repr.body = req.body.slice(0, MAX_REQ_BODY_LENTGH) +
	util.format(' ... (clipped at %d, full length is %d chars)',
	MAX_REQ_BODY_LENTGH, req.body.length);
	} else if (OPT_PRETTY_PRINT_REQUEST_BODY) {
	if (typeof req.body === 'string' && req.body.startsWith('{')) {
	try {
	repr.body = req.body
	.split(/\n/g)
	.map(line => line.trim() && JSON.stringify(JSON.parse(line), null, 4))
	.join('\n')
	} catch {
	repr.body = req.body
	}
	}
	} else {
	repr.body = req.body
	}
	if (req.body && OPT_RENDER_WATERFALL) {
	repr.waterfall = '\n' + renderWaterfall(req.body)
	}
	return repr;
	},
	res: function (res) {
	if (!res \|\| !res.statusCode) {
	return res;
	}
	let repr = {
	statusCode: res.statusCode,
	header: res._header
	};
	if (typeof(res._data) === 'string') {
	repr.bodyLength = res._data.length
	if (res._data.length > MAX_RES_BODY_LENTGH) {
	repr.body = res._data.slice(0, MAX_RES_BODY_LENTGH) +
	util.format(' ... (clipped at %d, full length is %d chars)',
	MAX_RES_BODY_LENTGH, res._data.length);
	} else {
	repr.body = res._data
	}
	}
	return repr
	}
	},
	level: 'debug',
	stream: process.stdout
	});


	// // LineStream
	// var stream = require('stream');
	// function LineStream(opts) {
	// if (!(this instanceof LineStream)) {
	// throw new Error('use `new LineStream(opts)`');
	// }
	// stream.Transform.call(this, opts);
	// this._buf = '';
	// };
	// util.inherits(LineStream, stream.Transform);
	// LineStream.prototype._transform = function(chunk, encoding, done) {
	// // XXX need toString? Can we do this with raw Buffer for perf?
	// var data = this._buf + chunk.toString('utf8');
	// var lines = data.split(/\r?\n\|\r(?!\n)/);
	// this._buf = lines.pop();
	// for (var i = 0; i < lines.length; i++) {
	// this.push(lines[i] + '\n');
	// }
	// done();
	// };
	// LineStream.prototype._flush = function(done) {
	// if (this._buf) {
	// this.push(this._buf);
	// }
	// done();
	// };

	function renderWaterfall(body) {
	const OPT_EXTRA_FIELDS = false

	function addChild(span, childSpan) {
	if (!span.children) {
	span.children = [childSpan]
	} else {
	span.children.push(childSpan)
	}
	}
	function renderError(error, prefix) {
	let r = `${prefix}error ${error.id.slice(0, 6)} "${error.exception.message}"`;
	if (OPT_EXTRA_FIELDS) {
	r += ` (type=${error.exception.type}, parent_id=${error.parent_id})`;
	}
	(error.children \|\| []).forEach(s => {
	r += '\n' + renderSpanOrError(s, ' ' + prefix)
	})
	return r
	}
	function renderSpanOrError(spanOrError, prefix) {
	if (spanOrError.name === undefined) {
	return renderError(spanOrError, prefix)
	} else {
	return renderSpan(spanOrError, prefix)
	}
	}
	function renderSpan(span, prefix) {
	let r = `${prefix}span ${span.id.slice(0, 6)} "${span.name}"`;
	if (OPT_EXTRA_FIELDS) {
	r += ` (parent_id=${span.parent_id})`;
	}
	(span.children \|\| []).forEach(s => {
	r += '\n' + renderSpanOrError(s, ' ' + prefix)
	})
	return r
	}
	function renderTx(tx) {
	let r = `transaction ${tx.id.slice(0, 6)} "${tx.name}"`;
	if (OPT_EXTRA_FIELDS) {
	r += ` (trace_id=${tx.trace_id})`;
	}
	(tx.children \|\| []).forEach(s => {
	r += '\n' + renderSpanOrError(s, '`- ')
	})
	tx.orphans.forEach(s => {
	r += '\n' + renderSpanOrError(s, 'ORPHAN: ')
	})
	return r
	}

	const events = body
	.split(/\n/g)
	.filter(line => line.trim(line))
	.map(JSON.parse)
	const allSpansAndErrors = events
	.filter(e => e.span \|\| e.error)
	.map(e => e.span \|\| e.error)
	.sort((a, b) => a.timestamp - b.timestamp)
	const txs = events.filter(e => e.transaction)
	.map(e => e.transaction)
	.sort((a, b) => a.timestamp - b.timestamp)
	.map(tx => {
	tx.orphans = []
	const spans = allSpansAndErrors.filter(s => s.trace_id === tx.trace_id)
	const spanOrErrorFromId = { [tx.id]: tx }
	spans.forEach(s => { spanOrErrorFromId[s.id] = s })
	spans.forEach(s => {
	const parent = spanOrErrorFromId[s.parent_id]
	if (parent) {
	addChild(parent, s)
	} else {
	tx.orphans.push(s)
	}
	})
	return tx
	})
	const rendering = txs
	.map(tx => renderTx(tx))
	.join('\n')
	return rendering
	}

	const server = http.createServer(function (req, res) {
	// return // respond nevermore

	var parsedUrl = url.parse(req.url);

	var instream = req;
	if (req.headers['content-encoding'] === 'gzip') {
	instream = req.pipe(zlib.createGunzip())
	} else {
	instream.setEncoding('utf8');
	}

	// // Look at individual lines stream in on a chunked request.
	// lstream = new LineStream()
	// lstream.on('data', function (line) {
	// console.warn('XXX line', line)
	// })

	if (OPT_QUEUE_IS_FULL) {
	// APM server "queue is full" error response example from:
	// https://www.elastic.co/guide/en/apm/server/current/events-api.html#events-api-errors
	setTimeout(function () {
	res.writeHead(503, {
	'Content-Type': 'application/json'
	})
	resBody = JSON.stringify({
	"errors": [
	{
	"message": "<json-schema-err>",
	"document": "<ndjson-obj>"
	},{
	"message": "<json-schema-err>",
	"document": "<ndjson-obj>"
	},{
	"message": "<json-decoding-err>",
	"document": "<ndjson-obj>"
	},{
	"message": "queue is full"
	},
	],
	"accepted": 2320
	})
	res.end(resBody)
	res._data = resBody; // for audit logging
	log.info({req, res}, 'request')
	}, 2000)
	}

	let body = '';
	instream.on('data', (chunk) => {
	//console.warn('XXX got a chunk: %d bytes', chunk.length, Buffer.byteLength(chunk))
	body += chunk;
	// lstream.write(chunk)
	});
	instream.on('end', function () {
	req.body = body;
	var resBody;

	// TODO: add support for '/' path for server version detection
	// as done by https://github.com/elastic/apm-agent-python/pull/1194

	if (parsedUrl.pathname === '/config/v1/agents') {
	// Central config mocking.
	if (process.env.MOCK_CENTRAL_CONFIG_FAILS) {
	resBody = '{"ok":false,"message":"The requested resource is currently unavailable."}\n'
	res.writeHead(503)
	} else if (process.env.MOCK_CENTRAL_CONFIG_INVALID_JSON) {
	res.writeHead(200)
	resBody = '{"log_level":'
	} else {
	res.writeHead(200)
	// resBody = '{"log_level": "debug"}'
	resBody = '{}'
	}
	} else if (req.method === 'POST' && parsedUrl.pathname === '/intake/v2/events') {
	resBody = '{}'
	res.writeHead(202)
	} else if (req.method === 'POST' && parsedUrl.pathname === '/intake/v2/profile') {
	resBody = '{"accepted": 1}'
	res.writeHead(202)
	}
	if (OPT_HANG_ON_RESPONSE) { // Actually respond. Set to false to hang the request.
	log.warn('intentionally hanging response')
	} else {
	res.end(resBody);
	}

	res._data = resBody; // for audit logging
	log.info({req, res}, 'request')
	});
	})

	let PORT = 8200; // default APM server port
	//PORT = 8201;
	//PORT = 80
	server.listen(PORT, function () {
	log.info('listening', server.address())
	})