AWS ELB log parsing

alb-parser.js

'use strict';

const _ = require('lodash');
const StreamParser = require('./stream-parser');

class ApplicationLoadBalancerParser extends StreamParser {
	constructor() {
		super(parse);
	}
}

// Log format
// type timestamp elb client:port target:port request_processing_time target_processing_time response_processing_time elb_status_code target_status_code received_bytes sent_bytes "request" "user_agent" ssl_cipher ssl_protocol target_group_arn trace_id

const regx = /^(\S+) (\S+) (\S+) (\S+) (\S+) (\S+) (\S+) (\S+) (\S+) (\S+) (\S+) (\S+) \"([^\"]+)\" \"([^\"]+)\" (\S+) (\S+) (\S+) (\S+)$/;
const fields = [
	'type',
	'timestamp',
	'elb',
	'client',
	'target',
	'requestProcessingTime',
	'targetProcessingTime',
	'responseProcessingTime',
	'elbStatusCode',
	'targetStatusCode',
	'receivedBytes',
	'sentBytes',
	'request',
	'userAgent',
	'sslCipher',
	'sslProtocol',
	'targetGroupArn',
	'traceId'
];

function parse(lines) {
	return lines
		.map((line) => parseLine(line))
		.filter((line) => Boolean(line));
}

function parseLine(line) {
	const matches = regx.exec(line);
	if (!matches) {
		return;
	}

	const obj = _.zipObject(fields, matches.slice(1, matches.length));
	obj.elbStatusCode = parseInt(obj.elbStatusCode, 10);
	obj.receivedBytes = parseInt(obj.receivedBytes, 10);
	obj.requestProcessingTime = parseFloat(obj.requestProcessingTime, 10);
	obj.responseProcessingTime = parseFloat(obj.responseProcessingTime, 10);
	obj.sentBytes = parseInt(obj.sentBytes, 10);
	obj.targetProcessingTime = parseFloat(obj.targetProcessingTime, 10);
	obj.targetStatusCode = parseInt(obj.targetStatusCode, 10);
	return obj;
}

module.exports.ApplicationLoadBalancerParser = ApplicationLoadBalancerParser;
module.exports.parseLine = parseLine;
module.exports.parse = parse;

elb-parser.js

'use strict';

const _ = require('lodash');
const StreamParser = require('./stream-parser');

class ElasticLoadBalancerParser extends StreamParser {
	constructor() {
		super(parse);
	}
}

// Log format
// timestamp elb client:port backend:port request_processing_time backend_processing_time response_processing_time elb_status_code backend_status_code received_bytes sent_bytes "request" "user_agent" ssl_cipher ssl_protocol

const regx = /^(\S+) (\S+) (\S+) (\S+) (\S+) (\S+) (\S+) (\S+) (\S+) (\S+) (\S+) \"([^\"]+)\" \"([^\"]+)\" (\S+) (\S+)$/;
const fields = [
	'timestamp',
	'elb',
	'client',
	'backed',
	'requestProcessingTime',
	'backendProcessingTime',
	'responseProcessingTime',
	'elbStatusCode',
	'backendStatusCode',
	'receivedBytes',
	'sentBytes',
	'request',
	'userAgent',
	'sslCipher',
	'sslProtocol'
];

function parse(lines) {
	return lines
		.map((line) => parseLine(line))
		.filter((line) => Boolean(line));
}

function parseLine(line) {
	const matches = regx.exec(line);
	if (!matches) {
		return;
	}

	const obj = _.zipObject(fields, matches.slice(1, matches.length));
	obj.elbStatusCode = parseInt(obj.elbStatusCode, 10);
	obj.receivedBytes = parseInt(obj.receivedBytes, 10);
	obj.requestProcessingTime = parseFloat(obj.requestProcessingTime, 10);
	obj.responseProcessingTime = parseFloat(obj.responseProcessingTime, 10);
	obj.sentBytes = parseInt(obj.sentBytes, 10);
	obj.backendProcessingTime = parseFloat(obj.backendProcessingTime, 10);
	obj.backendStatusCode = parseInt(obj.backendStatusCode, 10);
	return obj;
}

module.exports.ElasticLoadBalancerParser = ElasticLoadBalancerParser;
module.exports.parseLine = parseLine;
module.exports.parse = parse;

example.js

	const p = new parser.ApplicationLoadBalancerParser();
	const result = [];

  p.on('readable', () => {
    let record;
    while ((record = p.read())) {
      result.push(record);
    }
  });

  fs.createReadStream(__dirname + '/../fixtures/alb-log.gz')
    .pipe(zlib.createGunzip())
    .pipe(p)
    .on('end', () => {
      console.log(result);    
    });

stream-parser.js

'use strict';

const TransformStream = require('stream').Transform;

class StreamParser extends TransformStream {
	constructor(parse) {
		super({objectMode: true});
		this.parse = parse;
	}

	_transform(chunk, encoding, done) {
		encoding = encoding || 'utf8';

		if (Buffer.isBuffer(chunk)) {
			chunk = encoding === 'buffer' ? chunk.toString() : chunk.toString(encoding);
		}

		if (this._lastLineBuffer) {
			chunk = this._lastLineBuffer + chunk;
		}

		const lines = chunk.split('\n');
		this._lastLineBuffer = lines.splice(lines.length - 1, 1)[0];

		this.parse(lines).forEach(this.push.bind(this));
		done();
	}

	_flush(done) {
		if (this._lastLineBuffer) {
			this.parse(this._lastLineBuffer).forEach(this.push.bind(this));
			this._lastLineBuffer = null;
		}

		done();
	}
}

module.exports = StreamParser;