Add lambda embedded metrics plugin

__tests__/LambdaEmbeddedMetricsPlugin.spec.ts

@@ -0,0 +1,179 @@
+import {v4 as uuid} from 'uuid';
+import Monitor from '../src/Monitor';
+import {LambdaEmbeddedMetricsPlugin} from '../src/plugins/LambdaEmbeddedMetricsPlugin';
+import {pascalCase} from 'pascal-case';
+import {StorageResolution} from 'aws-embedded-metrics';
+
+const pascalifyObject = (obj: Record<string, string>): Record<string, string> =>
+    Object.keys(obj).reduce((prev, curr) => {
+        prev[pascalCase(curr)] = obj[curr];
+        return prev;
+    }, {});
+
+const generateExpectedCall = (
+    serviceName: string,
+    metricName: string,
+    status: 'Start' | 'Success' | 'Failure' | 'SuccessExecutionTime' | 'FailureExecutionTime',
+    unit: 'Count' | 'Milliseconds',
+    value: number,
+    context: {},
+    tags: {},
+    storageResolution?: StorageResolution
+) => ({
+    ...context,
+    ...tags,
+    _aws: {
+        CloudWatchMetrics: [
+            {
+                Dimensions: [Object.keys(tags)],
+                Metrics: [
+                    {
+                        Name: `${metricName}${status}`,
+                        Unit: unit,
+                        ...(storageResolution && {StorageResolution: storageResolution}),
+                    },
+                ],
+                Namespace: serviceName,
+            },
+        ],
+    },
+    [`${metricName}${status}`]: value,
+});
+
+describe('LambdaEmbeddedMetricsPlugin', () => {
+    let plugin: LambdaEmbeddedMetricsPlugin;
+    let monitor: Monitor;
+    let serviceName: string;
+
+    function initMonitor(_opts?: LambdaEmbeddedMetricsPlugin) {
+        plugin = new LambdaEmbeddedMetricsPlugin({serviceName});
+        monitor = new Monitor({plugins: [plugin]});
+    }
+
+    beforeEach(() => {
+        jest.useFakeTimers();
+
+        serviceName = uuid();
+        initMonitor();
+    });
+
+    afterEach(() => {
+        jest.clearAllMocks();
+        jest.runOnlyPendingTimers();
+        jest.useRealTimers();
+    });
+
+    it('onSuccess', async () => {
+        const metricName = uuid();
+        const logSpy = jest.spyOn(console, 'log');
+        const tags = {[`dimension_${uuid()}`]: uuid(), [`dimension_${uuid()}`]: uuid()};
+        const expectedTags = {Service: serviceName, ...pascalifyObject(tags)};
+
+        const context = {[`property_${uuid()}`]: uuid(), [`property_${uuid()}`]: uuid()};
+        const expectedContext = pascalifyObject(context);
+
+        await monitor.monitored(metricName, async () => 123, {tags, context});
+
+        await monitor.flush(1000);
+        const {calls} = logSpy.mock;
+
+        const startCall = JSON.parse(calls[0][0]);
+        delete startCall._aws.Timestamp;
+        const successCall = JSON.parse(calls[1][0]);
+        delete successCall._aws.Timestamp;
+        const successExecutionTimeCall = JSON.parse(calls[2][0]);
+        delete successExecutionTimeCall._aws.Timestamp;
+
+        expect(startCall).toEqual(
+            expect.objectContaining(
+                generateExpectedCall(serviceName, metricName, 'Start', 'Count', 1, expectedContext, expectedTags)
+            )
+        );
+
+        expect(successCall).toEqual(
+            expect.objectContaining(
+                generateExpectedCall(serviceName, metricName, 'Success', 'Count', 1, expectedContext, expectedTags)
+            )
+        );
+
+        expect(successExecutionTimeCall).toEqual(
+            expect.objectContaining(
+                generateExpectedCall(
+                    serviceName,
+                    metricName,
+                    'SuccessExecutionTime',
+                    'Milliseconds',
+                    0,
+                    expectedContext,
+                    expectedTags,
+                    StorageResolution.High
+                )
+            )
+        );
+    });
+
+    it('onError', async () => {
+        const metricName = uuid();
+        const logSpy = jest.spyOn(console, 'log').mockImplementation(() => {});
+
+        const tags = {[`dimension_${uuid()}`]: uuid(), [`dimension_${uuid()}`]: uuid()};
+        const expectedTags = {Service: serviceName, ...pascalifyObject(tags)};
+
+        const context = {[`property_${uuid()}`]: uuid(), [`property_${uuid()}`]: uuid()};
+        const expectedContext = pascalifyObject(context);
+
+        try {
+            await monitor.monitored(
+                metricName,
+                async () => {
+                    throw new Error(uuid());
+                },
+                {tags, context}
+            );
+        } catch (err) {
+            console.log(err);
+
+            await monitor.flush(2000);
+
+            const {calls} = logSpy.mock;
+
+            const startCall = JSON.parse(calls[0][0]);
+            delete startCall._aws.Timestamp;
+            const failureCall = JSON.parse(calls[2][0]);
+            delete failureCall._aws.Timestamp;
+            const failureExecutionTimeCall = JSON.parse(calls[3][0]);
+            delete failureExecutionTimeCall._aws.Timestamp;
+
+            expect(startCall).toEqual(
+                expect.objectContaining(
+                    generateExpectedCall(serviceName, metricName, 'Start', 'Count', 1, expectedContext, expectedTags)
+                )
+            );
+
+            expect(failureCall).toEqual(
+                expect.objectContaining(
+                    generateExpectedCall(serviceName, metricName, 'Failure', 'Count', 1, expectedContext, expectedTags)
+                )
+            );
+
+            expect(failureExecutionTimeCall).toEqual(
+                expect.objectContaining(
+                    generateExpectedCall(
+                        serviceName,
+                        metricName,
+                        'FailureExecutionTime',
+                        'Milliseconds',
+                        0,
+                        expectedContext,
+                        expectedTags,
+                        StorageResolution.High
+                    )
+                )
+            );
+
+            return;
+        }
+
+        throw new Error('Function execution should fail');
+    });
+});

package.json

@@ -9,7 +9,7 @@
     "build": "rm -rf dist && tsc -p tsconfig.json",
     "example": "ts-node example/run.ts",
     "prepublishOnly": "yarn build",
-    "test": "jest --config jest.config.js",
+    "test": "AWS_EMF_ENVIRONMENT=Local jest --config jest.config.js",
     "prepare": "husky install"
   },
   "repository": {
@@ -23,7 +23,9 @@
   },
   "homepage": "https://github.com/Soluto/monitored#readme",
   "dependencies": {
+    "aws-embedded-metrics": "^4.1.0",
     "hot-shots": "^8.5.0",
+    "pascal-case": "^3.1.2",
     "prom-client": "^13.2.0"
   },
   "devDependencies": {

src/plugins/LambdaEmbeddedMetricsPlugin.ts

@@ -0,0 +1,95 @@
+import {pascalCase} from 'pascal-case';
+import {createMetricsLogger, StorageResolution, Unit} from 'aws-embedded-metrics';
+import {MonitoredPlugin, OnFailureOptions, OnStartOptions, OnSuccessOptions} from './types';
+import {timeoutPromise} from '../utils';
+
+export interface LambdaEmbeddedMetricsPluginOptions {
+    namespace?: string;
+    serviceName: string;
+}
+
+const pascalifyObject = (obj: Record<string, string>): Record<string, string> =>
+    Object.keys(obj).reduce((prev, curr) => {
+        prev[pascalCase(curr)] = obj[curr];
+        return prev;
+    }, {});
+
+// Docs: https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch_Embedded_Metric_Format_Specification.html
+export class LambdaEmbeddedMetricsPlugin implements MonitoredPlugin {
+    private promises: Promise<void>[] = [];
+
+    constructor(readonly opts: LambdaEmbeddedMetricsPluginOptions) {}
+
+    private async sendMetric<T extends string>(
+        name: T,
+        unit: Unit,
+        value: number,
+        tags?: Record<string, string>,
+        context?: Record<string, string> | undefined,
+        storageResolution?: StorageResolution
+    ): Promise<void> {
+        const metrics = createMetricsLogger();
+        metrics.setNamespace(this.opts.namespace ?? this.opts.serviceName);
+
+        const dimensions = {Service: this.opts.serviceName, ...pascalifyObject(tags ?? {})};
+        const properties = pascalifyObject(context ?? {});
+
+        metrics.setDimensions(dimensions);
+        Object.entries(properties).map(([x, y]) => metrics.setProperty(x, y));
+
+        metrics.putMetric(name, value, unit, storageResolution);
+
+        metrics.flushPreserveDimensions = false;
+        await metrics.flush();
+    }
+
+    async increment<T extends string>(
+        name: T,
+        value?: number | undefined,
+        tags?: Record<string, string> | undefined,
+        context?: Record<string, string> | undefined
+    ): Promise<void> {
+        await this.sendMetric(name, Unit.Count, value ?? 1, tags, context);
+    }
+
+    gauge(_name: string, _value?: number | undefined, _tags?: Record<string, string> | undefined): Promise<void> {
+        throw new Error('Method not implemented.');
+    }
+
+    async timing(
+        name: string,
+        value: number,
+        tags?: Record<string, string> | undefined,
+        context?: Record<string, string> | undefined
+    ): Promise<void> {
+        await this.sendMetric(name, Unit.Milliseconds, value, tags, context, StorageResolution.High);
+    }
+
+    async flush(timeout: number): Promise<boolean> {
+        try {
+            await timeoutPromise(
+                timeout,
+                Promise.all(this.promises),
+                'Timeout reached, stopped wait for pending log writes'
+            );
+            return true;
+        } catch (err) {
+            console.log('Failed to flush metrics', err);
+            return false;
+        }
+    }
+
+    onStart({scope, options}: OnStartOptions): void {
+        this.promises.push(this.increment(`${scope}Start`, 1, options?.tags, options?.context));
+    }
+
+    onSuccess({scope, options, executionTime}: OnSuccessOptions): void {
+        this.promises.push(this.increment(`${scope}Success`, 1, options?.tags, options?.context));
+        this.promises.push(this.timing(`${scope}SuccessExecutionTime`, executionTime, options?.tags, options?.context));
+    }
+
+    onFailure({scope, options, executionTime}: OnFailureOptions): void {
+        this.promises.push(this.increment(`${scope}Failure`, 1, options?.tags, options?.context));
+        this.promises.push(this.timing(`${scope}FailureExecutionTime`, executionTime, options?.tags, options?.context));
+    }
+}

yarn.lock

@@ -452,6 +452,11 @@
   dependencies:
     "@cspotcode/source-map-consumer" "0.8.0"
 
+"@datastructures-js/heap@^4.0.2":
+  version "4.3.1"
+  resolved "https://registry.yarnpkg.com/@datastructures-js/heap/-/heap-4.3.1.tgz#d3752b5664202b99b41815f6117e501d8f66c632"
+  integrity sha512-au4fYa4fprREES58FnMOTFjg8lCYpSenF5tBu8C/iweMaj02rAOZUqlLUCqR3HIWzNfgTeCmAiyRHdjJVHrsIQ==
+
 "@endemolshinegroup/cosmiconfig-typescript-loader@^3.0.2":
   version "3.0.2"
   resolved "https://registry.yarnpkg.com/@endemolshinegroup/cosmiconfig-typescript-loader/-/cosmiconfig-typescript-loader-3.0.2.tgz#eea4635828dde372838b0909693ebd9aafeec22d"
@@ -966,6 +971,13 @@ asynckit@^0.4.0:
   resolved "https://registry.yarnpkg.com/asynckit/-/asynckit-0.4.0.tgz#c79ed97f7f34cb8f2ba1bc9790bcc366474b4b79"
   integrity sha1-x57Zf380y48robyXkLzDZkdLS3k=
 
+aws-embedded-metrics@^4.1.0:
+  version "4.1.0"
+  resolved "https://registry.yarnpkg.com/aws-embedded-metrics/-/aws-embedded-metrics-4.1.0.tgz#2237e6681d0a9e32fb743974dc0e97e3844599a1"
+  integrity sha512-Yiscee2EfyiczIy9GFOSR0qzqD6kcD0HjQuBJRyO842SkKoFlxzOo/99OVEmg2odUS5XI8oxiS7HO0WTynkteg==
+  dependencies:
+    "@datastructures-js/heap" "^4.0.2"
+
 babel-jest@^27.5.1:
   version "27.5.1"
   resolved "https://registry.yarnpkg.com/babel-jest/-/babel-jest-27.5.1.tgz#a1bf8d61928edfefd21da27eb86a695bfd691444"
@@ -2542,6 +2554,13 @@ log-update@^4.0.0:
     slice-ansi "^4.0.0"
     wrap-ansi "^6.2.0"
 
+lower-case@^2.0.2:
+  version "2.0.2"
+  resolved "https://registry.yarnpkg.com/lower-case/-/lower-case-2.0.2.tgz#6fa237c63dbdc4a82ca0fd882e4722dc5e634e28"
+  integrity sha512-7fm3l3NAF9WfN6W3JOmf5drwpVqX78JtoGJ3A6W0a6ZnldM41w2fV5D490psKFTpMds8TJse/eHLFFsNHHjHgg==
+  dependencies:
+    tslib "^2.0.3"
+
 lru-cache@^6.0.0:
   version "6.0.0"
   resolved "https://registry.yarnpkg.com/lru-cache/-/lru-cache-6.0.0.tgz#6d6fe6570ebd96aaf90fcad1dafa3b2566db3a94"
@@ -2666,6 +2685,14 @@ natural-compare@^1.4.0:
   resolved "https://registry.yarnpkg.com/natural-compare/-/natural-compare-1.4.0.tgz#4abebfeed7541f2c27acfb29bdbbd15c8d5ba4f7"
   integrity sha1-Sr6/7tdUHywnrPspvbvRXI1bpPc=
 
+no-case@^3.0.4:
+  version "3.0.4"
+  resolved "https://registry.yarnpkg.com/no-case/-/no-case-3.0.4.tgz#d361fd5c9800f558551a8369fc0dcd4662b6124d"
+  integrity sha512-fgAN3jGAh+RoxUGZHTSOLJIqUc2wmoBwGR4tbpNAKmmovFoWq0OdRkb0VkldReO2a2iBT/OEulG9XSUc10r3zg==
+  dependencies:
+    lower-case "^2.0.2"
+    tslib "^2.0.3"
+
 node-int64@^0.4.0:
   version "0.4.0"
   resolved "https://registry.yarnpkg.com/node-int64/-/node-int64-0.4.0.tgz#87a9065cdb355d3182d8f94ce11188b825c68a3b"
@@ -2801,6 +2828,14 @@ [email protected]:
   resolved "https://registry.yarnpkg.com/parse5/-/parse5-6.0.1.tgz#e1a1c085c569b3dc08321184f19a39cc27f7c30b"
   integrity sha512-Ofn/CTFzRGTTxwpNEs9PP93gXShHcTq255nzRYSKe8AkVpZY7e1fpmTfOyoIvjP5HG7Z2ZM7VS9PPhQGW2pOpw==
 
+pascal-case@^3.1.2:
+  version "3.1.2"
+  resolved "https://registry.yarnpkg.com/pascal-case/-/pascal-case-3.1.2.tgz#b48e0ef2b98e205e7c1dae747d0b1508237660eb"
+  integrity sha512-uWlGT3YSnK9x3BQJaOdcZwrnV6hPpd8jFH1/ucpiLRPh/2zCVJKS19E4GvYHvaCcACn3foXZ0cLB9Wrx1KGe5g==
+  dependencies:
+    no-case "^3.0.4"
+    tslib "^2.0.3"
+
 path-exists@^4.0.0:
   version "4.0.0"
   resolved "https://registry.yarnpkg.com/path-exists/-/path-exists-4.0.0.tgz#513bdbe2d3b95d7762e8c1137efa195c6c61b5b3"
@@ -3434,6 +3469,11 @@ tslib@^2:
   resolved "https://registry.yarnpkg.com/tslib/-/tslib-2.3.1.tgz#e8a335add5ceae51aa261d32a490158ef042ef01"
   integrity sha512-77EbyPPpMz+FRFRuAFlWMtmgUWGe9UOG2Z25NqCwiIjRhOf5iKGuzSe5P2w1laq+FkRy4p+PCuVkJSGkzTEKVw==
 
+tslib@^2.0.3:
+  version "2.5.0"
+  resolved "https://registry.yarnpkg.com/tslib/-/tslib-2.5.0.tgz#42bfed86f5787aeb41d031866c8f402429e0fddf"
+  integrity sha512-336iVw3rtn2BUK7ORdIAHTyxHGRIHVReokCR3XjbckJMK7ms8FysBfhLR8IXnAgy7T0PTPNBWKiH514FOW/WSg==
+
 tslib@~2.1.0:
   version "2.1.0"
   resolved "https://registry.yarnpkg.com/tslib/-/tslib-2.1.0.tgz#da60860f1c2ecaa5703ab7d39bc05b6bf988b97a"