Improve robustness of Throughput calculation, 100% flow coverage and refactoring improvements

#10
#11

Author: tmitchel2

.flowconfig

@@ -9,3 +9,4 @@
 [libs]
 
 [options]
+suppress_comment= \\(.\\|\n\\)*\\$FlowIgnore

package.json

@@ -15,6 +15,7 @@
     "mocha": "--require resources/mocha-bootload src/**/__tests__/**/*.js"
   },
   "scripts": {
+    "localtest": "babel-node ./scripts/localtest.js",
     "test": "npm run lint && npm run check",
     "lint": "eslint src",
     "check": "flow check",

scripts/localtest.js

@@ -0,0 +1,34 @@
+/* @flow */
+/* eslint-disable */
+// $FlowIgnore
+import babelPolyfill from 'babel-polyfill';
+/* eslint-enable */
+import config from '../src/Config';
+// $FlowIgnore
+import dotenv from 'dotenv';
+import DynamoDB from '../src/DynamoDB';
+import CloudWatch from '../src/CloudWatch';
+import CapacityCalculator from '../src/CapacityCalculator';
+import { log } from '../src/Global';
+
+// Load environment variables
+dotenv.config({path: 'config.env'});
+
+async function localtestAsync() {
+  let tableName = 'Contacts';
+
+  log('Getting table description', tableName);
+  let db = new DynamoDB(config.connection.dynamoDB);
+  let tableDescription = await db.describeTableAsync({TableName: tableName});
+
+  log('Getting table consumed capacity description', tableName);
+  let cw = new CloudWatch(config.connection.cloudWatch);
+  let cc = new CapacityCalculator(cw);
+  let consumedCapacity = await cc.describeTableConsumedCapacityAsync(tableDescription.Table);
+
+  log(JSON.stringify({
+    consumedCapacity
+  }, null, 2));
+}
+
+localtestAsync();

scripts/start.js

@@ -1,10 +1,12 @@
+/* eslint-disable no-console */
+
 try {
   var lambda = require('../dist/index.js');
 
   process.chdir('./dist');
 
   var context = {
-    succeed: function(data) {
+    succeed: data => {
       try {
         if (data) {
           console.log(JSON.stringify(data));
@@ -14,7 +16,7 @@ try {
         console.error(e);
       }
     },
-    fail: function(e) {
+    fail: e => {
       console.log(e.stack);
       console.error(e);
     }

src/CapacityCalculator.js

@@ -0,0 +1,142 @@
+/* @flow */
+import { json, stats, warning, invariant } from './Global';
+import CloudWatch from './CloudWatch';
+import type {
+  TableDescription, GetMetricStatisticsResponse, Dimension,
+  TableConsumedCapacityDescription } from './FlowTypes';
+
+export default class CapacityCalculator {
+  _cw: CloudWatch;
+
+  constructor(cloudWatch: CloudWatch) {
+    invariant(typeof cloudWatch !== 'undefined', 'Parameter \'cloudWatch\' is not set');
+    this._cw = cloudWatch;
+  }
+
+  async describeTableConsumedCapacityAsync(params: TableDescription)
+    : Promise<TableConsumedCapacityDescription> {
+    let sw = stats
+      .timer('DynamoDB.describeTableConsumedCapacityAsync')
+      .start();
+
+    try {
+      invariant(typeof params !== 'undefined', 'Parameter \'params\' is not set');
+
+      // Make all the requests concurrently
+      let tableRead = this.getConsumedCapacityAsync(true, params.TableName, null);
+      let tableWrite = this.getConsumedCapacityAsync(false, params.TableName, null);
+
+      let gsiReads = (params.GlobalSecondaryIndexes || [])
+        .map(gsi => this.getConsumedCapacityAsync(true, params.TableName, gsi.IndexName));
+
+      let gsiWrites = (params.GlobalSecondaryIndexes || [])
+        .map(gsi => this.getConsumedCapacityAsync(false, params.TableName, gsi.IndexName));
+
+      // Await on the results
+      let tableConsumedRead = await tableRead;
+      let tableConsumedWrite = await tableWrite;
+      let gsiConsumedReads = await Promise.all(gsiReads);
+      let gsiConsumedWrites = await Promise.all(gsiWrites);
+
+      // Format results
+      let gsis = gsiConsumedReads.map((read, i) => {
+        let write = gsiConsumedWrites[i];
+        return {
+          // $FlowIgnore: The indexName is not null in this case
+          IndexName: read.globalSecondaryIndexName,
+          ConsumedThroughput: {
+            ReadCapacityUnits: read.value,
+            WriteCapacityUnits: write.value
+          }
+        };
+      });
+
+      return {
+        TableName: params.TableName,
+        ConsumedThroughput: {
+          ReadCapacityUnits: tableConsumedRead.value,
+          WriteCapacityUnits: tableConsumedWrite.value
+        },
+        GlobalSecondaryIndexes: gsis
+      };
+    } catch (ex) {
+      warning(JSON.stringify({
+        class: 'CapacityCalculator',
+        function: 'describeTableConsumedCapacityAsync',
+        params,
+      }, null, json.padding));
+      throw ex;
+    } finally {
+      sw.end();
+    }
+  }
+
+  async getConsumedCapacityAsync(
+    isRead: boolean, tableName: string, globalSecondaryIndexName: ?string) {
+    try {
+      invariant(typeof isRead !== 'undefined', 'Parameter \'isRead\' is not set');
+      invariant(typeof tableName !== 'undefined', 'Parameter \'tableName\' is not set');
+      invariant(typeof globalSecondaryIndexName !== 'undefined',
+        'Parameter \'globalSecondaryIndexName\' is not set');
+
+      // These values determine how many minutes worth of metrics
+      let durationMinutes = 5;
+      let periodMinutes = 1;
+
+      let EndTime = new Date();
+      let StartTime = new Date();
+      StartTime.setTime(EndTime - (60000 * durationMinutes));
+      let MetricName = isRead ? 'ConsumedReadCapacityUnits' : 'ConsumedWriteCapacityUnits';
+      let Dimensions = this.getDimensions(tableName, globalSecondaryIndexName);
+      let params = {
+        Namespace: 'AWS/DynamoDB',
+        MetricName,
+        Dimensions,
+        StartTime,
+        EndTime,
+        Period: (periodMinutes * 60),
+        Statistics: [ 'Average' ],
+        Unit: 'Count'
+      };
+
+      let statistics = await this._cw.getMetricStatisticsAsync(params);
+      let value = this.getProjectedValue(statistics);
+      return {
+        tableName,
+        globalSecondaryIndexName,
+        value
+      };
+    } catch (ex) {
+      warning(JSON.stringify({
+        class: 'CapacityCalculator',
+        function: 'getConsumedCapacityAsync',
+        isRead, tableName, globalSecondaryIndexName,
+      }, null, json.padding));
+      throw ex;
+    }
+  }
+
+  getProjectedValue(data: GetMetricStatisticsResponse) {
+    if (data.Datapoints.length === 0) {
+      return 0;
+    }
+
+    // Default algorithm for projecting a good value for the current ConsumedThroughput is:
+    // 1. Query 5 average readings each spanning a minute
+    // 2. Select the Max value from those 5 readings
+    let averages = data.Datapoints.map(dp => dp.Average);
+    let value = Math.max(...averages);
+    return value;
+  }
+
+  getDimensions(tableName: string, globalSecondaryIndexName: ?string): Dimension[] {
+    if (globalSecondaryIndexName) {
+      return [
+        { Name: 'TableName', Value: tableName},
+        { Name: 'GlobalSecondaryIndex', Value: globalSecondaryIndexName}
+      ];
+    }
+
+    return [ { Name: 'TableName', Value: tableName} ];
+  }
+}

src/CloudWatch.js

@@ -1,18 +1,23 @@
+/* @flow */
 import AWS from 'aws-sdk-promise';
-import {
-  json,
-  stats,
-  warning,
-  invariant } from '../src/Global';
+import { json, stats, warning, invariant } from './Global';
+import type {
+  CloudWatchOptions,
+  GetMetricStatisticsRequest,
+  GetMetricStatisticsResponse,
+} from './FlowTypes';
 
 export default class CloudWatch {
-  constructor(cloudWatchOptions) {
+  _cw: AWS.CloudWatch;
+
+  constructor(cloudWatchOptions: CloudWatchOptions) {
     invariant(typeof cloudWatchOptions !== 'undefined',
       'Parameter \'cloudWatchOptions\' is not set');
     this._cw = new AWS.CloudWatch(cloudWatchOptions);
   }
 
-  async getMetricStatisticsAsync(params) {
+  async getMetricStatisticsAsync(params: GetMetricStatisticsRequest)
+    : Promise<GetMetricStatisticsResponse> {
     let sw = stats.timer('CloudWatch.getMetricStatisticsAsync').start();
     try {
       invariant(typeof params !== 'undefined',