cloudtrail_partitions

cloudtrail_partitions.js

'use strict';

console.log('Loading function');

const aws = require('aws-sdk');
aws.config.update({region: 'us-east-1'});

const s3 = new aws.S3({ apiVersion: '2006-03-01' });
const ddb = new aws.DynamoDB({apiVersion: '2012-10-08'});
const athena = new aws.Athena();

exports.handler = (event, context, callback) => {
    const bucket = event.Records[0].s3.bucket.name;
    const key = decodeURIComponent(event.Records[0].s3.object.key.replace(/\+/g, ' '));
    if (key.includes("$folder$")) {
      return;
    }
    const params = {
        Bucket: bucket,
        Key: key,
    };
    var parts = key.split("/");
    var region = parts[3];
    var year = parts[4];
    var month = parts[5];
    var day = parts[6];
    var date = `${region}--${year}--${month}--${day}`;
    
    var getParams = {
      TableName: 'ProcessCloudTrailLogs',
      Key: {
        'date' : {S: date},
      }
    };

    ddb.getItem(getParams, function(err, data) {
      if (err) {
        console.log("Error retrieving item:", err) 
      } else {
        if (data.Item) {
          console.log("Already added partition:", date);
          return;
        } else {
          console.log("Couldn't find date: " + date);
          var query = `ALTER TABLE cloudtrail_logs ADD PARTITION (region='${region}',year='${year}',month='${month}',day='${day}') location 's3://azs-cloudtrail/AWSLogs/ACCOUNTNUMBER/CloudTrail/${region}/${year}/${month}/${day}'`;
          console.log("Inserting partition to Athena: ", query);
          var athenaParams = {
            QueryString: query,
            ResultConfiguration: {
              OutputLocation: 's3://aws-athena-query-results-ACCOUNTNUMBER-us-east-1/',
            },
            QueryExecutionContext: {
              Database: 'default'
            }
          };
          var putParams = {
            TableName: 'ProcessCloudTrailLogs',
            Item: {
              'date' : {S: date},
            }
          };
          athena.startQueryExecution(athenaParams, function(err, data) {
            if (err) {
              console.log("Failure:", err, err.stack);
            } else {
              console.log("Success:", data.QueryExecutionId);
            }
          });
          console.log("Recording date in DynamoDB: " + date);
          ddb.putItem(putParams, function(err, data) {
            if (err) {
              console.log("Error:", err);
            } else {
              console.log("Put item:", date);
            }
          });
        }
      }
    });
};