lambda2cw - analyze S3 log files and put custom metric data to CloudWatch

lambda2cw.coffee

console.log 'lambda2cw Started'

aws = require 'aws-sdk'
s3 = new aws.S3
  apiVersion: '2006-03-01'
cw = new aws.CloudWatch
  apiVersion: '2010-08-01'

putMetrics = (metrics,namespace,callback) ->
  metricdata=metrics.splice(0,20) # Max 20 metric data per request
  cw.putMetricData
    MetricData: metricdata
    Namespace: "S3requests"
    (err,data) ->
      if err
        context.done -2,err
      else
        console.log(metricdata.length+" metrics pushed. "+metrics.length+" left")
        if metrics.length > 0
          setTimeout ()->
            putMetrics(metrics,namespace,callback)
          ,1000
        else
          callback()

exports.handler = (event, context) ->
#  console.log 'Received event:'
#  console.log JSON.stringify event, null, '  '

  bucket = event.Records[0].s3.bucket.name
  if not event.Records[0].s3.object.key.match(/^logs\//)
    return context.done null,"not logs"

  key=event.Records[0].s3.object.key

  console.log "processing s3://#{bucket}/#{key}"

  s3.getObject
    Bucket: bucket
    Key: key
    (err,data) ->
      if err
        context.done -1,err
      else
        logs=data.Body.toString().split(/\n/)
        metrics=[]
        for log in logs
          fields=log.split(/\ /)
          if fields.length > 1
            date=fields[2].replace(/^\[/,'').replace(/:/,' ')
            timestamp=Date.parse(date+" UTC")
            if fields[7].match "GET"
              action="GET"
            else if fields[7].match /(PUT|COPY|POST|LIST)/
              action="OTHERS"
            else if fields[7].match "DELETE"
              action="DELETE"
            else
              action="UNKOWN"
            metrics.push
              MetricName: action
              Dimensions: [
                Name: "bucket"
                Value: fields[1]
              ]
              Timestamp: timestamp/1000
              Unit: "Count"
              Value: 1

        putMetrics metrics,"S3requests",() ->
          context.done null,"done"

return

lambda2cw.js

(function() {
  var aws, cw, putMetrics, s3;

  console.log('lambda2cw Started');

  aws = require('aws-sdk');

  s3 = new aws.S3({
    apiVersion: '2006-03-01'
  });

  cw = new aws.CloudWatch({
    apiVersion: '2010-08-01'
  });

  putMetrics = function(metrics, namespace, callback) {
    var metricdata;
    metricdata = metrics.splice(0, 20);
    return cw.putMetricData({
      MetricData: metricdata,
      Namespace: "S3requests"
    }, function(err, data) {
      if (err) {
        return context.done(-2, err);
      } else {
        console.log(metricdata.length + " metrics pushed. " + metrics.length + " left");
        if (metrics.length > 0) {
          return setTimeout(function() {
            return putMetrics(metrics, namespace, callback);
          }, 1000);
        } else {
          return callback();
        }
      }
    });
  };

  exports.handler = function(event, context) {
    var bucket, key;
    bucket = event.Records[0].s3.bucket.name;
    if (!event.Records[0].s3.object.key.match(/^logs\//)) {
      return context.done(null, "not logs");
    }
    key = event.Records[0].s3.object.key;
    console.log("processing s3://" + bucket + "/" + key);
    return s3.getObject({
      Bucket: bucket,
      Key: key
    }, function(err, data) {
      var action, date, fields, log, logs, metrics, timestamp, _i, _len;
      if (err) {
        return context.done(-1, err);
      } else {
        logs = data.Body.toString().split(/\n/);
        metrics = [];
        for (_i = 0, _len = logs.length; _i < _len; _i++) {
          log = logs[_i];
          fields = log.split(/\ /);
          if (fields.length > 1) {
            date = fields[2].replace(/^\[/, '').replace(/:/, ' ');
            timestamp = Date.parse(date + " UTC");
            if (fields[7].match("GET")) {
              action = "GET";
            } else if (fields[7].match(/(PUT|COPY|POST|LIST)/)) {
              action = "OTHERS";
            } else if (fields[7].match("DELETE")) {
              action = "DELETE";
            } else {
              action = "UNKOWN";
            }
            metrics.push({
              MetricName: action,
              Dimensions: [
                {
                  Name: "bucket",
                  Value: fields[1]
                }
              ],
              Timestamp: timestamp / 1000,
              Unit: "Count",
              Value: 1
            });
          }
        }
        return putMetrics(metrics, "S3requests", function() {
          return context.done(null, "done");
        });
      }
    });
  };

  return;

}).call(this);