index.js

'use strict';

console.log('Loading event');

var logGroupName = 'vpc-flow-logs',
    region = 'eu-west-1';

var AWS = require('aws-sdk'),
    Promise = require('bluebird'),
    async = require('async'),
    util = require('util'),
    _ = require('lodash'),
    ec2 = new AWS.EC2({region: region}),
    cloudwatchlogs = new AWS.CloudWatchLogs({region: region});

exports.handler = function(event, context) {
    console.log('Received event:', JSON.stringify(event, null, 2));

    var params = { Filters: [ {Name: 'attachment.instance-owner-id', Values: [ 'amazon-elb', ]} ] };

    async.auto({
        registerMetricFilters: function(registerMetricFiltersCallback) {

            ec2.describeNetworkInterfaces(params, function(err, data) {
                if (err) {
                    console.log(err, err.stack);
                } else {
                    async.eachLimit(data.NetworkInterfaces, 1,  function(eni, eniCallback) {
                        var elbName = eni.Description.replace(/ELB /, '');

                        async.eachLimit([
                            {
                                metricName: 'vpc-flow-log-packets-elb-' + elbName,
                                metricNamespace: 'LogMetrics',
                                metricValue: '$packets'
                            },
                            {
                                metricName: 'vpc-flow-log-bytes-elb-' + elbName,
                                metricNamespace: 'LogMetrics',
                                metricValue: '$bytes'
                            },
                        ], 1,  function(metrics, metricsCallback) {
                            var params = {
                                filterName: 'vpc-flow-logs-elb-' + elbName + '-' + eni.NetworkInterfaceId + '-' + eni.PrivateIpAddress.replace(/\./g, '-'),
                                filterPattern: '[version, account_id, interface_id="' + eni.NetworkInterfaceId + '", srcaddr, dstaddr, srcport, dstport, protocol, packets, bytes, start, end, action, log_status, ...]',
                                logGroupName: 'vpc-flow-logs',
                                metricTransformations: [metrics]
                            };
                            console.log(params);
                            cloudwatchlogs.putMetricFilter(params, function(err, data) {
                                if (err) console.log(err, err.stack); // an error occurred
                                else     console.log(data);           // successful response
                                return metricsCallback(err, data);
                            });
                        }, function(data) {
                           return eniCallback(data);
                        });
                    }, function(err, data) {
                        return registerMetricFiltersCallback(err, data);
                    });
                }
            });
        },
        cleanup: ['registerMetricFilters', function(cleanupCallback) {
            console.log('cleanup');
            cloudwatchlogs.describeMetricFilters({
                logGroupName: logGroupName,
            }, function(errMetricFilters, metricFilters) {
                var mapping = {};

                if (errMetricFilters) {
                    console.log(errMetricFilters, errMetricFilters.stack);
                    return cleanupCallback(errMetricFilters);
               }

                metricFilters.metricFilters.forEach(function(filter) {
                    var m = filter.filterName.match(/vpc-flow-logs-elb-(.*)-(eni-[a-zA-Z0-9]+)-([0-9]{1,3}-[0-9]{1,3}-[0-9]{1,3}-[0-9]{1,3})$/);
                    if (m) {
                        mapping[m[2]] = { elb: m[1], eni: m[2], filterName: filter.filterName } ;
                    } else {
                        console.log('Invalid filter skipping - ' + filter.filterName);
                    }
                });
                var networkInterfaceList = Object.keys(mapping);

                async.eachLimit(networkInterfaceList, 5, function(networkInterfaceId, callbackEni) {
                    ec2.describeNetworkInterfaces({NetworkInterfaceIds: [networkInterfaceId] }, function(errEni, data) {
                        if (errEni) {
                            var filterName = mapping[networkInterfaceId].filterName;
                            networkInterfaceList.splice(networkInterfaceList.indexOf(networkInterfaceId), 1);

                            if (errEni.code == 'InvalidNetworkInterfaceID.NotFound') {

                                if (mapping[networkInterface]) {
                                   console.log("Removing network interface - " + networkInterfaceId, { filterName: filterName, logGroupName: logGroupName });
                                   delete mapping[networkInterfaceId];
                                   return cloudwatchlogs.deleteMetricFilter({ filterName: filterName, logGroupName: logGroupName }, callbackEni);
                                }
                            }

                        }
                        return callbackEni(errEni);
                    });
                }, cleanupCallback);
            });

        }]
    }, function () {
        console.log('complete');
    });
};

package.json

{
  "name": "amazon-elb-lambda-vpc-flow-statistics",
  "version": "1.0.0",
  "description": "Amazon ECS cluster service group management",
  "main": "index.js",
  "scripts": {
    "start": "node invoke.js",
    "test": "echo \"Error: no test specified\" && exit 1"
  },
  "repository": {
    "type": "git",
    "url": "https://github.com/photobox/services-operations"
  },
  "author": "andrew.johnstone",
  "license": "ISC",
  "bugs": {
    "url": "https://github.com/photobox/services-operations/issues"
  },
  "homepage": "https://github.com/photobox/services-operations",
  "dependencies": {
    "async": "^1.4.2",
    "aws-sdk": "^2.2.9",
    "lodash": "^3.10.1",
    "moment": "^2.10.6"
  }
}