Scott.solmonson/batch item failure metric

src/index.spec.ts

@@ -9,7 +9,11 @@ import {
   sendDistributionMetricWithDate,
   _metricsQueue,
 } from "./index";
-import { incrementErrorsMetric, incrementInvocationsMetric } from "./metrics/enhanced-metrics";
+import {
+  incrementErrorsMetric,
+  incrementInvocationsMetric,
+  incrementBatchItemFailureMetric,
+} from "./metrics/enhanced-metrics";
 import { LogLevel, setLogLevel } from "./utils";
 import { HANDLER_STREAMING, STREAM_RESPONSE } from "./constants";
 import { PassThrough } from "stream";
@@ -22,6 +26,9 @@ jest.mock("./metrics/enhanced-metrics");
 
 const mockedIncrementErrors = incrementErrorsMetric as jest.Mock<typeof incrementErrorsMetric>;
 const mockedIncrementInvocations = incrementInvocationsMetric as jest.Mock<typeof incrementInvocationsMetric>;
+const mockedIncrementBatchItemFailures = incrementBatchItemFailureMetric as jest.Mock<
+  typeof incrementBatchItemFailureMetric
+>;
 
 const mockARN = "arn:aws:lambda:us-east-1:123497598159:function:my-test-lambda";
 const mockContext = {
@@ -88,6 +95,7 @@ describe("datadog", () => {
 
     mockedIncrementErrors.mockClear();
     mockedIncrementInvocations.mockClear();
+    mockedIncrementBatchItemFailures.mockClear();
   });
   afterEach(() => {
     process.env = oldEnv;
@@ -379,6 +387,65 @@ describe("datadog", () => {
     expect(mockedIncrementErrors).toBeCalledWith(expect.anything(), mockContext);
   });
 
+  it("increments batch item failures enhanced metric", async () => {
+    const lambdaResponse: any = {
+      batchItemFailures: [{ itemIdentifier: "abc123" }, { itemIdentifier: "def456" }],
+    };
+
+    const wrapped = datadog(async () => {
+      return lambdaResponse;
+    });
+
+    const lambdaResult = await wrapped({}, mockContext, () => {});
+
+    expect(lambdaResult).toEqual(lambdaResponse);
+
+    expect(mockedIncrementBatchItemFailures).toBeCalledTimes(1);
+    expect(mockedIncrementInvocations).toBeCalledTimes(1);
+
+    expect(mockedIncrementBatchItemFailures).toBeCalledWith(expect.anything(), 2, mockContext);
+    expect(mockedIncrementInvocations).toBeCalledWith(expect.anything(), mockContext);
+  });
+
+  it("sets batch item failures enhanced metric to zero if list is empty", async () => {
+    const lambdaResponse: any = {
+      batchItemFailures: [],
+    };
+
+    const wrapped = datadog(async () => {
+      return lambdaResponse;
+    });
+
+    const lambdaResult = await wrapped({}, mockContext, () => {});
+
+    expect(lambdaResult).toEqual(lambdaResponse);
+
+    expect(mockedIncrementBatchItemFailures).toBeCalledTimes(1);
+    expect(mockedIncrementInvocations).toBeCalledTimes(1);
+
+    expect(mockedIncrementBatchItemFailures).toBeCalledWith(expect.anything(), 0, mockContext);
+    expect(mockedIncrementInvocations).toBeCalledWith(expect.anything(), mockContext);
+  });
+
+  it("doesn't increment batch item failures if it's not a failure response", async () => {
+    const lambdaResponse: any = {
+      foo: "bar",
+    };
+
+    const wrapped = datadog(async () => {
+      return lambdaResponse;
+    });
+
+    const lambdaResult = await wrapped({}, mockContext, () => {});
+
+    expect(lambdaResult).toEqual(lambdaResponse);
+
+    expect(mockedIncrementBatchItemFailures).toBeCalledTimes(0);
+    expect(mockedIncrementInvocations).toBeCalledTimes(1);
+
+    expect(mockedIncrementInvocations).toBeCalledWith(expect.anything(), mockContext);
+  });
+
   it("doesn't increment errors or invocations with config false setting", async () => {
     const handlerError: Handler = (event, context, callback) => {
       throw Error("Some error");

src/index.ts

@@ -3,14 +3,25 @@ import { HANDLER_STREAMING, STREAM_RESPONSE } from "./constants";
 import {
   incrementErrorsMetric,
   incrementInvocationsMetric,
+  incrementBatchItemFailureMetric,
   KMSService,
   MetricsConfig,
   MetricsListener,
   MetricsQueue,
 } from "./metrics";
 import { TraceConfig, TraceListener } from "./trace";
 import { subscribeToDC } from "./runtime";
-import { logDebug, Logger, LogLevel, promisifiedHandler, setSandboxInit, setLogger, setLogLevel } from "./utils";
+import {
+  isBatchItemFailure,
+  batchItemFailureCount,
+  logDebug,
+  Logger,
+  LogLevel,
+  promisifiedHandler,
+  setSandboxInit,
+  setLogger,
+  setLogLevel,
+} from "./utils";
 import { getEnhancedMetricTags } from "./metrics/enhanced-metrics";
 import { DatadogTraceHeaders } from "./trace/context/extractor";
 
@@ -190,6 +201,9 @@ export function datadog<TEvent, TResult>(
           if (responseIs5xxError) {
             incrementErrorsMetric(metricsListener, context);
           }
+          if (isBatchItemFailure(localResult)) {
+            incrementBatchItemFailureMetric(metricsListener, batchItemFailureCount(localResult), context);
+          }
         }
         return localResult;
       };

src/metrics/enhanced-metrics.ts

@@ -81,3 +81,12 @@ export function incrementInvocationsMetric(listener: MetricsListener, context: C
 export function incrementErrorsMetric(listener: MetricsListener, context: Context): void {
   incrementEnhancedMetric(listener, "errors", context);
 }
+
+export function incrementBatchItemFailureMetric(listener: MetricsListener, count: number, context: Context): void {
+  listener.sendDistributionMetric(
+    "aws.lambda.enhanced.batch_item_failures",
+    count,
+    true,
+    ...getEnhancedMetricTags(context),
+  );
+}

src/metrics/index.ts

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

src/utils/batch-item-failures.ts

@@ -0,0 +1,11 @@
+export function isBatchItemFailure(lambdaResponse: any): boolean {
+  return (
+    typeof lambdaResponse === "object" &&
+    "batchItemFailures" in lambdaResponse &&
+    Array.isArray(lambdaResponse.batchItemFailures)
+  );
+}
+
+export function batchItemFailureCount(lambdaResponse: any): number {
+  return lambdaResponse?.batchItemFailures?.length || 0; // Guard clause in case someone calls this without checking isBatchItemFailure
+}

src/utils/index.ts

@@ -3,3 +3,4 @@ export { wrap, promisifiedHandler } from "./handler";
 export { Timer } from "./timer";
 export { logWarning, logError, logDebug, Logger, setLogLevel, setLogger, LogLevel } from "./log";
 export { tagObject } from "./tag-object";
+export { batchItemFailureCount, isBatchItemFailure } from "./batch-item-failures";