add-athena-support

.eslintrc.js

@@ -38,6 +38,8 @@ module.exports = {
     "class-methods-use-this": 0,
     "implicit-arrow-linebreak": 0,
     "import/prefer-default-export": 0,
+    "no-nested-ternary": 0,
+    "prefer-destructuring": 0,
   },
   "globals": {
     "fetch": false,

package.json

@@ -44,7 +44,10 @@
         "KMS_REGIONS": "us-west-2",
         "NODE_ENV": "test",
         "AWS_ACCESS_KEY_ID": "notarealaccesskeyid",
-        "AWS_SECRET_ACCESS_KEY": "notarealsecretaccesskey"
+        "AWS_SECRET_ACCESS_KEY": "notarealsecretaccesskey",
+        "SCHEDULER_ROLE_ARN": "scheduler-arn",
+        "BUS_ARN": "bus-arn",
+        "MASTER_KEY_ARN": "kms-arn"
       }
     }
   },
@@ -57,6 +60,7 @@
     "url": "https://github.com/jgilbert01/aws-lambda-stream.git"
   },
   "devDependencies": {
+    "@aws-sdk/client-athena": "^3.450.0",
     "@aws-sdk/client-cloudwatch": "^3.450.0",
     "@aws-sdk/client-dynamodb": "^3.450.0",
     "@aws-sdk/client-eventbridge": "^3.450.0",
@@ -65,8 +69,8 @@
     "@aws-sdk/client-kms": "^3.450.0",
     "@aws-sdk/client-lambda": "^3.450.0",
     "@aws-sdk/client-s3": "^3.450.0",
-    "@aws-sdk/client-secrets-manager": "^3.450.0",
     "@aws-sdk/client-scheduler": "^3.450.0",
+    "@aws-sdk/client-secrets-manager": "^3.450.0",
     "@aws-sdk/client-sns": "^3.450.0",
     "@aws-sdk/client-sqs": "^3.450.0",
     "@aws-sdk/client-sts": "^3.450.0",
@@ -92,6 +96,7 @@
     "better-npm-run": "^0.1.1",
     "bluebird": "^3.7.1",
     "chai": "^4.2.0",
+    "date-fns": "^4.1.0",
     "debug": "^4.1.1",
     "eslint": "^6.7.0",
     "eslint-config-airbnb-base": "^14.0.0",
@@ -129,4 +134,4 @@
   "dependencies": {
     "object-sizeof": "^2.6.0"
   }
-}
+}

src/connectors/athena.js

@@ -0,0 +1,58 @@
+/* eslint import/no-extraneous-dependencies: ["error", {"devDependencies": true}] */
+import { AthenaClient, StartQueryExecutionCommand } from '@aws-sdk/client-athena';
+import { NodeHttpHandler } from '@smithy/node-http-handler';
+import Promise from 'bluebird';
+import { omit, pick } from 'lodash';
+import { defaultDebugLogger } from '../utils/log';
+
+class Connector {
+  constructor({
+    debug,
+    pipelineId,
+    timeout = Number(process.env.ATHENA_TIMEOUT) || Number(process.env.TIMEOUT) || 1000,
+    additionalClientOpts = {},
+    ...opt
+  }) {
+    this.debug = (msg) => debug('%j', msg);
+    this.client = Connector.getClient(pipelineId, debug, timeout, additionalClientOpts);
+    this.opt = opt;
+  }
+
+  static clients = {};
+
+  static getClient(pipelineId, debug, timeout, additionalClientOpts) {
+    const addlRequestHandlerOpts = pick(additionalClientOpts, ['requestHandler']);
+    const addlClientOpts = omit(additionalClientOpts, ['requestHandler']);
+
+    if (!this.clients[pipelineId]) {
+      this.clients[pipelineId] = new AthenaClient({
+        requestHandler: new NodeHttpHandler({
+          requestTimeout: timeout,
+          connectionTimeout: timeout,
+          ...addlRequestHandlerOpts,
+        }),
+        logger: defaultDebugLogger(debug),
+        maxAttempts: 5,
+        ...addlClientOpts,
+      });
+    }
+    return this.clients[pipelineId];
+  }
+
+  startQueryExecution(inputParams, ctx) {
+    const params = {
+      ...inputParams,
+    };
+
+    return this._sendCommand(new StartQueryExecutionCommand(params), ctx);
+  }
+
+  _sendCommand(command, ctx) {
+    this.opt.metrics?.capture(this.client, command, 'athena', this.opt, ctx);
+    return Promise.resolve(this.client.send(command))
+      .tap(this.debug)
+      .tapCatch(this.debug);
+  }
+}
+
+export default Connector;