feat: emit latency metric in RestHook Lambda

package.json

@@ -34,6 +34,7 @@
   },
   "dependencies": {
     "@types/aws-lambda": "^8.10.92",
+    "aws-embedded-metrics": "^2.0.4",
     "aws-sdk": "^2.1000.0",
     "axios": "^0.21.4",
     "fhir-works-on-aws-authz-rbac": "5.0.0",
@@ -42,6 +43,7 @@
     "fhir-works-on-aws-routing": "6.3.0",
     "fhir-works-on-aws-search-es": "3.9.2",
     "lodash": "^4.17.21",
+    "moment": "^2.29.1",
     "serverless-http": "^2.7.0",
     "tslib": "^2.3.1",
     "yargs": "^16.2.0"

src/subscriptions/restHookLambda/restHook.ts

@@ -2,6 +2,8 @@ import axios from 'axios';
 import { makeLogger } from 'fhir-works-on-aws-interface';
 import { SQSEvent } from 'aws-lambda';
 import { SubscriptionNotification } from 'fhir-works-on-aws-search-es';
+import { metricScope, Unit } from 'aws-embedded-metrics';
+import moment from 'moment';
 import ensureAsyncInit from '../../index';
 import { AllowListInfo, getAllowListHeaders } from './allowListUtil';
 
@@ -19,15 +21,29 @@ const mergeRequestHeaders = (allowListHeader: string[], channelHeader: string[])
     const mergeHeader = (header: string) => {
         const colonIndex = header.indexOf(':') === -1 ? header.length : header.indexOf(':');
         const headerKey = header.substring(0, colonIndex);
-        const headerValue = header.substring(colonIndex + 1);
-        mergedHeader[headerKey] = headerValue;
+        mergedHeader[headerKey] = header.substring(colonIndex + 1);
     };
 
     allowListHeader.forEach((header) => mergeHeader(header));
     channelHeader.forEach((header) => mergeHeader(header));
     return mergedHeader;
 };
 
+/**
+ * Push latency metric to CloudWatch
+ * @param messages
+ */
+const pushLatencyMetric = metricScope((metrics) => async (messages: SubscriptionNotification[]): Promise<void> => {
+    const currentTime = moment(new Date());
+    messages.forEach((message: SubscriptionNotification) => {
+        metrics.putMetric(
+            'SubscriptionEndToEndLatency',
+            currentTime.diff(moment(new Date(message.matchedResource.lastUpdated)), 'milliseconds'),
+            Unit.Milliseconds,
+        );
+    });
+});
+
 export default class RestHookHandler {
     readonly enableMultitenancy: boolean;
 
@@ -41,10 +57,13 @@ export default class RestHookHandler {
     ): Promise<any> {
         await ensureAsyncInit(allowListPromise);
         const allowList = await allowListPromise;
-        const notificationPromises = event.Records.map((record: any) => {
+        const messages = event.Records.map((record: any): SubscriptionNotification => {
             const body = JSON.parse(record.body);
-            logger.debug(body);
-            const message: SubscriptionNotification = JSON.parse(body.Message);
+            return JSON.parse(body.Message);
+        });
+        // Latency is reported before HTTP call since the external endpoint latency is out of our control.
+        await pushLatencyMetric(messages);
+        const notificationPromises = messages.map((message: SubscriptionNotification) => {
             const { endpoint, channelHeader, channelPayload, matchedResource, tenantId } = message;
             const allowListHeaders = getAllowListHeaders(allowList, endpoint, {
                 enableMultitenancy: this.enableMultitenancy,

yarn.lock

@@ -2847,6 +2847,11 @@ aws-elasticsearch-connector@^8.2.0:
     aws4 "^1.10.0"
     lodash.get "^4.4.2"
 
+aws-embedded-metrics@^2.0.4:
+  version "2.0.4"
+  resolved "https://registry.yarnpkg.com/aws-embedded-metrics/-/aws-embedded-metrics-2.0.4.tgz#6c7dab11a839bb301351058d63664dae600467da"
+  integrity sha512-b4+3BC8e3DE3HdZkmzlozvS8wU45w9M+4eAd2R4Z62N+ihiCuwndEh15Yj/scCKRq/FTO3rfKnzlvUMsuJtzLg==
+
 aws-sdk@^2.1000.0, aws-sdk@^2.834.0:
   version "2.1015.0"
   resolved "https://registry.yarnpkg.com/aws-sdk/-/aws-sdk-2.1015.0.tgz#fc69c0620a34c1d77173123769fbc725d867b3cc"