Add stats support

lifion · Apr 26, 2019 · 328cd80 · 328cd80
1 parent 7505973
commit 328cd80
Show file tree

Hide file tree

Showing 7 changed files with 444 additions and 115 deletions.
diff --git a/lib/dynamodb-client.js b/lib/dynamodb-client.js
@@ -2,6 +2,7 @@
 
 const retry = require('async-retry');
 const { DynamoDB } = require('aws-sdk');
+const { reportException, reportSuccess: reportSuccesStats } = require('./stats');
 
 const privateData = new WeakMap();
 
@@ -21,6 +22,11 @@ function shouldBailRetry(err) {
   );
 }
 
+const reportSuccess = result => {
+  reportSuccesStats('dynamoDb');
+  return result;
+};
+
 class DynamoDbClient {
   constructor({ awsOptions, logger, tableName }) {
     const client = new DynamoDB(awsOptions);
@@ -51,10 +57,12 @@ class DynamoDbClient {
     return internal(this)
       .client.createTable(...args)
       .promise()
+      .then(reportSuccess)
       .catch(err => {
         const { code, message } = err;
         const error = new Error(message);
         error.code = code;
+        reportException(err, 'dynamoDb');
         throw error;
       });
   }
@@ -66,10 +74,14 @@ class DynamoDbClient {
         client
           .describeTable(...args)
           .promise()
+          .then(reportSuccess)
           .catch(err => {
             const { code } = err;
-            if (code === 'ResourceNotFoundException' || shouldBailRetry(err)) bail();
-            else throw err;
+            if (code === 'ResourceNotFoundException' || shouldBailRetry(err)) bail(err);
+            else {
+              reportException(err, 'dynamoDb');
+              throw err;
+            }
           }),
       retryOpts
     ).catch(err => {
@@ -87,9 +99,13 @@ class DynamoDbClient {
         docClient
           .get(...args)
           .promise()
+          .then(reportSuccess)
           .catch(err => {
             if (shouldBailRetry(err)) bail(err);
-            else throw err;
+            else {
+              reportException(err, 'dynamoDb');
+              throw err;
+            }
           }),
       retryOpts
     ).catch(err => {
@@ -107,9 +123,13 @@ class DynamoDbClient {
         client
           .listTagsOfResource(...args)
           .promise()
+          .then(reportSuccess)
           .catch(err => {
             if (shouldBailRetry(err)) bail(err);
-            else throw err;
+            else {
+              reportException(err, 'dynamoDb');
+              throw err;
+            }
           }),
       retryOpts
     ).catch(err => {
@@ -127,10 +147,14 @@ class DynamoDbClient {
         docClient
           .put(...args)
           .promise()
+          .then(reportSuccess)
           .catch(err => {
             const { code } = err;
             if (code === 'ConditionalCheckFailedException' || shouldBailRetry(err)) bail(err);
-            else throw err;
+            else {
+              reportException(err, 'dynamoDb');
+              throw err;
+            }
           }),
       retryOpts
     ).catch(err => {
@@ -145,10 +169,12 @@ class DynamoDbClient {
     return internal(this)
       .client.tagResource(...args)
       .promise()
+      .then(reportSuccess)
       .catch(err => {
         const { code, message } = err;
         const error = new Error(message);
         error.code = code;
+        reportException(err, 'dynamoDb');
         throw error;
       });
   }
@@ -160,10 +186,14 @@ class DynamoDbClient {
         docClient
           .update(...args)
           .promise()
+          .then(reportSuccess)
           .catch(err => {
             const { code } = err;
             if (code === 'ConditionalCheckFailedException' || shouldBailRetry(err)) bail(err);
-            else throw err;
+            else {
+              reportException(err, 'dynamoDb');
+              throw err;
+            }
           }),
       retryOpts
     ).catch(err => {
@@ -181,9 +211,13 @@ class DynamoDbClient {
         client
           .waitFor(...args)
           .promise()
+          .then(reportSuccess)
           .catch(err => {
             if (shouldBailRetry(err)) bail(err);
-            else throw err;
+            else {
+              reportException(err, 'dynamoDb');
+              throw err;
+            }
           }),
       retryOpts
     ).catch(err => {
@@ -198,10 +232,12 @@ class DynamoDbClient {
     return internal(this)
       .docClient.delete(...args)
       .promise()
+      .then(reportSuccess)
       .catch(err => {
         const { code, message } = err;
         const error = new Error(message);
         error.code = code;
+        reportException(err, 'dynamoDb');
         throw error;
       });
   }

diff --git a/lib/fan-out-consumer.js b/lib/fan-out-consumer.js
@@ -10,6 +10,7 @@ const { promisify } = require('util');
 const { RecordsDecoder } = require('./records');
 const { getStreamShards } = require('./stream');
 const { safeJsonParse, wait } = require('./utils');
+const { reportException, reportSuccess } = require('./stats');
 
 const AWS_API_TARGET = 'Kinesis_20131202.SubscribeToShard';
 const AWS_EVENT_STREAM = 'application/vnd.amazon.eventstream';
@@ -155,9 +156,11 @@ class FanOutConsumer {
       if (headers['content-type'] !== AWS_EVENT_STREAM || statusCode !== 200) {
         logger.error(`Subscription unsuccessful: ${statusCode}`);
         requestFlags.isEventStream = false;
+        reportException({ statusCode }, 'kinesis', streamName);
       } else {
         logger.debug('Subscription to shard is successful.');
         requestFlags.isEventStream = true;
+        reportSuccess('kinesis', streamName);
       }
     };
 

diff --git a/lib/index.js b/lib/index.js
@@ -18,6 +18,7 @@ const StateStore = require('./state-store');
 const { getRecordsEncoder } = require('./records');
 const { name: moduleName } = require('../package.json');
 const { noop } = require('./utils');
+const { getStats, reportRecordConsumed, startStatsEmitter } = require('./stats');
 
 const MAX_ENHANCED_CONSUMERS = 5;
 
@@ -38,6 +39,8 @@ function internal(instance) {
 
 async function ensureStreamInitialized(instance) {
   const privateProps = internal(instance);
+  startStatsEmitter(instance, privateProps);
+
   const { encryption, streamArn, streamCreatedOn, tags } = privateProps;
   if (streamArn && streamCreatedOn) return;
 
@@ -170,6 +173,7 @@ class Kinesis extends PassThrough {
       noRecordsPollDelay = 1000,
       pollDelay = 250,
       shardCount,
+      statsInterval,
       streamName,
       tags,
       useAutoCheckpoints = true,
@@ -197,7 +201,7 @@ class Kinesis extends PassThrough {
 
     Object.assign(internal(this), {
       awsOptions,
-      client: new KinesisClient({ awsOptions, logger: normLogger }),
+      client: new KinesisClient({ awsOptions, logger: normLogger, streamName }),
       compression,
       consumerGroup,
       consumerId: generate(),
@@ -209,6 +213,7 @@ class Kinesis extends PassThrough {
       pollDelay: normPollDelay >= 0 ? normPollDelay : 250,
       recordsEncoder: getRecordsEncoder(compression, 'Buffer'),
       shardCount: normShardCount >= 1 ? normShardCount : 1,
+      statsInterval,
       streamName,
       tags,
       useAutoCheckpoints: Boolean(useAutoCheckpoints),
@@ -229,7 +234,7 @@ class Kinesis extends PassThrough {
    */
   async startConsumer() {
     const privateProps = internal(this);
-    const { logger, useEnhancedFanOut } = privateProps;
+    const { logger, streamName, useEnhancedFanOut } = privateProps;
 
     await ensureStreamInitialized(this);
 
@@ -247,7 +252,10 @@ class Kinesis extends PassThrough {
 
     privateProps.pushToStream = (err, ...args) => {
       if (err) this.emit('error', err);
-      else this.push(...args);
+      else {
+        this.push(...args);
+        reportRecordConsumed(streamName);
+      }
     };
 
     const consumersManager = new ConsumersManager(privateProps);
@@ -277,7 +285,7 @@ class Kinesis extends PassThrough {
     const privateProps = internal(this);
     const { client, recordsEncoder } = privateProps;
     await ensureStreamInitialized(this);
-    return client.putRecord({
+    await client.putRecord({
       ...(await recordsEncoder(record)),
       StreamName: streamName || privateProps.streamName
     });
@@ -306,6 +314,15 @@ class Kinesis extends PassThrough {
       StreamName: streamName || privateProps.streamName
     });
   }
+
+  getStats() {
+    const { streamName } = internal(this);
+    return getStats(streamName);
+  }
+
+  static getStats() {
+    return getStats();
+  }
 }
 
 /**