feat: support sending metrics before handler initialization

integration_tests/send-metrics.js

@@ -1,5 +1,12 @@
 const { datadog, sendDistributionMetric } = require("datadog-lambda-js");
 
+sendDistributionMetric(
+  "serverless.integration_test.outside_handler",
+  1,
+  "tagkey:tagvalue",
+  `eventsource:outside_handler`,
+);
+
 async function handle(event, context) {
   const responsePayload = { message: "hello, dog!" };
 

integration_tests/snapshots/logs/async-metrics_node16.log

@@ -15,6 +15,17 @@ START
   ],
   "v": 1
 }
+{
+  "e": XXXX,
+  "m": "serverless.integration_test.outside_handler",
+  "t": [
+    "tagkey:tagvalue",
+    "eventsource:outside_handler",
+    "dd_lambda_layer:datadog-nodev16.XX.X"
+  ],
+  "v": 1
+}
+XXXX-XX-XX XX:XX:XX.XXX	INFO	[dd.trace_id=XXXX dd.span_id=XXXX] Processed APIGateway request
 {
   "e": XXXX,
   "m": "serverless.integration_test.execution",
@@ -25,7 +36,6 @@ START
   ],
   "v": 1
 }
-XXXX-XX-XX XX:XX:XX.XXX	INFO	[dd.trace_id=XXXX dd.span_id=XXXX] Processed APIGateway request
 END Duration: XXXX ms (init: XXXX ms) Memory Used: XXXX MB
 START
 {
@@ -43,6 +53,7 @@ START
   ],
   "v": 1
 }
+XXXX-XX-XX XX:XX:XX.XXX	INFO	[dd.trace_id=XXXX dd.span_id=XXXX] Processed SNS request
 {
   "e": XXXX,
   "m": "serverless.integration_test.records_processed",
@@ -63,9 +74,9 @@ START
   ],
   "v": 1
 }
-XXXX-XX-XX XX:XX:XX.XXX	INFO	[dd.trace_id=XXXX dd.span_id=XXXX] Processed SNS request
 END Duration: XXXX ms Memory Used: XXXX MB
 START
+XXXX-XX-XX XX:XX:XX.XXX	INFO	[dd.trace_id=XXXX dd.span_id=XXXX] Processed SQS request
 {
   "e": XXXX,
   "m": "aws.lambda.enhanced.invocations",
@@ -111,5 +122,4 @@ START
   ],
   "v": 1
 }
-XXXX-XX-XX XX:XX:XX.XXX	INFO	[dd.trace_id=XXXX dd.span_id=XXXX] Processed SQS request
 END Duration: XXXX ms Memory Used: XXXX MB

integration_tests/snapshots/logs/async-metrics_node18.log

@@ -15,6 +15,17 @@ START
   ],
   "v": 1
 }
+{
+  "e": XXXX,
+  "m": "serverless.integration_test.outside_handler",
+  "t": [
+    "tagkey:tagvalue",
+    "eventsource:outside_handler",
+    "dd_lambda_layer:datadog-nodev18.XX.X"
+  ],
+  "v": 1
+}
+XXXX-XX-XX XX:XX:XX.XXX	INFO	[dd.trace_id=XXXX dd.span_id=XXXX] Processed APIGateway request
 {
   "e": XXXX,
   "m": "serverless.integration_test.execution",
@@ -25,7 +36,6 @@ START
   ],
   "v": 1
 }
-XXXX-XX-XX XX:XX:XX.XXX	INFO	[dd.trace_id=XXXX dd.span_id=XXXX] Processed APIGateway request
 END Duration: XXXX ms (init: XXXX ms) Memory Used: XXXX MB
 START
 {
@@ -43,6 +53,7 @@ START
   ],
   "v": 1
 }
+XXXX-XX-XX XX:XX:XX.XXX	INFO	[dd.trace_id=XXXX dd.span_id=XXXX] Processed SNS request
 {
   "e": XXXX,
   "m": "serverless.integration_test.records_processed",
@@ -63,7 +74,6 @@ START
   ],
   "v": 1
 }
-XXXX-XX-XX XX:XX:XX.XXX	INFO	[dd.trace_id=XXXX dd.span_id=XXXX] Processed SNS request
 END Duration: XXXX ms Memory Used: XXXX MB
 START
 {
@@ -81,6 +91,7 @@ START
   ],
   "v": 1
 }
+XXXX-XX-XX XX:XX:XX.XXX	INFO	[dd.trace_id=XXXX dd.span_id=XXXX] Processed SQS request
 {
   "e": XXXX,
   "m": "serverless.integration_test.records_processed",
@@ -111,5 +122,4 @@ START
   ],
   "v": 1
 }
-XXXX-XX-XX XX:XX:XX.XXX	INFO	[dd.trace_id=XXXX dd.span_id=XXXX] Processed SQS request
 END Duration: XXXX ms Memory Used: XXXX MB

integration_tests/snapshots/logs/async-metrics_node20.log

@@ -15,7 +15,16 @@ START
   ],
   "v": 1
 }
-XXXX-XX-XX XX:XX:XX.XXX	INFO	[dd.trace_id=XXXX dd.span_id=XXXX] Processed APIGateway request
+{
+  "e": XXXX,
+  "m": "serverless.integration_test.outside_handler",
+  "t": [
+    "tagkey:tagvalue",
+    "eventsource:outside_handler",
+    "dd_lambda_layer:datadog-nodev20.XX.X"
+  ],
+  "v": 1
+}
 {
   "e": XXXX,
   "m": "serverless.integration_test.execution",
@@ -26,8 +35,10 @@ XXXX-XX-XX XX:XX:XX.XXX	INFO	[dd.trace_id=XXXX dd.span_id=XXXX] Processed APIGat
   ],
   "v": 1
 }
+XXXX-XX-XX XX:XX:XX.XXX	INFO	[dd.trace_id=XXXX dd.span_id=XXXX] Processed APIGateway request
 END Duration: XXXX ms (init: XXXX ms) Memory Used: XXXX MB
 START
+XXXX-XX-XX XX:XX:XX.XXX	INFO	[dd.trace_id=XXXX dd.span_id=XXXX] Processed SNS request
 {
   "e": XXXX,
   "m": "aws.lambda.enhanced.invocations",
@@ -43,7 +54,6 @@ START
   ],
   "v": 1
 }
-XXXX-XX-XX XX:XX:XX.XXX	INFO	[dd.trace_id=XXXX dd.span_id=XXXX] Processed SNS request
 {
   "e": XXXX,
   "m": "serverless.integration_test.records_processed",
@@ -66,7 +76,6 @@ XXXX-XX-XX XX:XX:XX.XXX	INFO	[dd.trace_id=XXXX dd.span_id=XXXX] Processed SNS re
 }
 END Duration: XXXX ms Memory Used: XXXX MB
 START
-XXXX-XX-XX XX:XX:XX.XXX	INFO	[dd.trace_id=XXXX dd.span_id=XXXX] Processed SQS request
 {
   "e": XXXX,
   "m": "aws.lambda.enhanced.invocations",
@@ -82,6 +91,7 @@ XXXX-XX-XX XX:XX:XX.XXX	INFO	[dd.trace_id=XXXX dd.span_id=XXXX] Processed SQS re
   ],
   "v": 1
 }
+XXXX-XX-XX XX:XX:XX.XXX	INFO	[dd.trace_id=XXXX dd.span_id=XXXX] Processed SQS request
 {
   "e": XXXX,
   "m": "serverless.integration_test.records_processed",

src/index.spec.ts

@@ -2,7 +2,13 @@ import http from "http";
 import nock from "nock";
 
 import { Context, Handler } from "aws-lambda";
-import { datadog, getTraceHeaders, sendDistributionMetric, sendDistributionMetricWithDate } from "./index";
+import {
+  datadog,
+  getTraceHeaders,
+  sendDistributionMetric,
+  sendDistributionMetricWithDate,
+  _metricsQueue,
+} from "./index";
 import { incrementErrorsMetric, incrementInvocationsMetric } from "./metrics/enhanced-metrics";
 import { LogLevel, setLogLevel } from "./utils";
 import { HANDLER_STREAMING, STREAM_RESPONSE } from "./constants";
@@ -407,6 +413,7 @@ describe("datadog", () => {
     const logger = {
       debug: jest.fn(),
       error: jest.fn(),
+      warn: jest.fn(),
     };
 
     const wrapped = datadog(handler, { forceWrap: true, logger: logger, debugLogging: true });
@@ -440,3 +447,25 @@ describe("datadog", () => {
     expect(wrapped[HANDLER_STREAMING]).toBe(undefined);
   });
 });
+
+describe("sendDistributionMetric", () => {
+  beforeEach(() => {
+    _metricsQueue.reset();
+    setLogLevel(LogLevel.NONE);
+  });
+  it("enqueues a metric for later processing when metrics listener is not initialized", () => {
+    sendDistributionMetric("metric", 1, "first-tag", "second-tag");
+    expect(_metricsQueue.length).toBe(1);
+  });
+});
+
+describe("sendDistributionMetricWithDate", () => {
+  beforeEach(() => {
+    _metricsQueue.reset();
+    setLogLevel(LogLevel.NONE);
+  });
+  it("enqueues a metric for later processing when metrics listener is not initialized", () => {
+    sendDistributionMetricWithDate("metric", 1, new Date(), "first-tag", "second-tag");
+    expect(_metricsQueue.length).toBe(1);
+  });
+});

src/index.ts

@@ -6,19 +6,11 @@ import {
   KMSService,
   MetricsConfig,
   MetricsListener,
+  MetricsQueue,
 } from "./metrics";
 import { TraceConfig, TraceListener } from "./trace";
 import { subscribeToDC } from "./runtime";
-import {
-  logDebug,
-  logError,
-  Logger,
-  LogLevel,
-  promisifiedHandler,
-  setSandboxInit,
-  setLogger,
-  setLogLevel,
-} from "./utils";
+import { logDebug, Logger, LogLevel, promisifiedHandler, setSandboxInit, setLogger, setLogLevel } from "./utils";
 import { getEnhancedMetricTags } from "./metrics/enhanced-metrics";
 import { DatadogTraceHeaders } from "./trace/context/extractor";
 
@@ -90,6 +82,8 @@ export const defaultConfig: Config = {
   localTesting: false,
 } as const;
 
+export const _metricsQueue: MetricsQueue = new MetricsQueue();
+
 let currentMetricsListener: MetricsListener | undefined;
 let currentTraceListener: TraceListener | undefined;
 
@@ -150,6 +144,8 @@ export function datadog<TEvent, TResult>(
       if (finalConfig.enhancedMetrics) {
         incrementInvocationsMetric(metricsListener, context);
       }
+
+      sendQueueMetrics(metricsListener);
     } catch (err) {
       if (err instanceof Error) {
         logDebug("Failed to start listeners", err);
@@ -259,9 +255,10 @@ export function sendDistributionMetricWithDate(name: string, value: number, metr
 
   if (currentMetricsListener !== undefined) {
     currentMetricsListener.sendDistributionMetricWithDate(name, value, metricTime, false, ...tags);
-  } else {
-    logError("handler not initialized");
+    return;
   }
+
+  _metricsQueue.push({ name, value, metricTime, tags });
 }
 
 /**
@@ -275,8 +272,22 @@ export function sendDistributionMetric(name: string, value: number, ...tags: str
 
   if (currentMetricsListener !== undefined) {
     currentMetricsListener.sendDistributionMetric(name, value, false, ...tags);
-  } else {
-    logError("handler not initialized");
+    return;
+  }
+
+  _metricsQueue.push({ name, value, tags });
+}
+
+function sendQueueMetrics(listener: MetricsListener) {
+  while (_metricsQueue.length > 0) {
+    const metric = _metricsQueue.shift()!; // This will always exist.
+    const { name, value, metricTime, tags } = metric;
+    if (metricTime !== undefined) {
+      listener.sendDistributionMetricWithDate(name, value, metricTime, false, ...tags);
+      return;
+    }
+
+    listener.sendDistributionMetric(name, value, false, ...tags);
   }
 }
 

src/metrics/index.ts

@@ -1,3 +1,4 @@
 export { MetricsConfig, MetricsListener } from "./listener";
+export { MetricsQueue } from "./queue";
 export { KMSService } from "./kms-service";
 export { incrementErrorsMetric, incrementInvocationsMetric } from "./enhanced-metrics";

src/metrics/queue.spec.ts

@@ -0,0 +1,68 @@
+import { LogLevel, logDebug, setLogLevel, setLogger } from "../utils";
+import { METRICS_QUEUE_LIMIT, MetricsQueue } from "./queue";
+
+describe("MetricsQueue", () => {
+  const logger = {
+    debug: jest.fn(),
+    error: jest.fn(),
+    warn: jest.fn(),
+  };
+  describe("push", () => {
+    beforeEach(() => {
+      setLogLevel(LogLevel.NONE);
+      setLogger(logger);
+    });
+    it("resets metrics queue when its full", () => {
+      setLogLevel(LogLevel.WARNING);
+      const queue = new MetricsQueue();
+      for (let i = 0; i < METRICS_QUEUE_LIMIT + 1; i++) {
+        queue.push({ name: "metric", tags: [], value: i });
+      }
+
+      // The queue should have been reset and only contain the last metric
+      expect(queue.length).toBe(1);
+      expect(logger.warn).toHaveBeenLastCalledWith(
+        '{"status":"warning","message":"datadog:Metrics queue is full, dropping all metrics."}',
+      );
+    });
+
+    it("enqueue metric", () => {
+      setLogLevel(LogLevel.DEBUG);
+      const queue = new MetricsQueue();
+      queue.push({ name: "metric", tags: [], value: 1 });
+      expect(queue.length).toBe(1);
+      expect(logger.debug).toHaveBeenLastCalledWith(
+        '{"status":"debug","message":"datadog:Metrics Listener was not initialized. Enqueuing metric for later processing."}',
+      );
+    });
+  });
+
+  describe("shift", () => {
+    it("returns undefined when queue is empty", () => {
+      const queue = new MetricsQueue();
+      expect(queue.shift()).toBeUndefined();
+    });
+
+    it("returns the first element in the queue", () => {
+      const queue = new MetricsQueue();
+      queue.push({ name: "metric", tags: [], value: 1 });
+      queue.push({ name: "metric", tags: [], value: 2 });
+      expect(queue.shift()).toEqual({ name: "metric", tags: [], value: 1 });
+    });
+  });
+
+  it("resets the queue", () => {
+    const queue = new MetricsQueue();
+    queue.push({ name: "metric", tags: [], value: 1 });
+    queue.push({ name: "metric", tags: [], value: 2 });
+    queue.reset();
+    expect(queue.length).toBe(0);
+  });
+
+  it("returns the length of the queue", () => {
+    const queue = new MetricsQueue();
+    queue.push({ name: "metric", tags: [], value: 1 });
+    queue.push({ name: "metric", tags: [], value: 2 });
+    expect(queue.length).toBe(2);
+  });
+});