BSNSS-1015: Add Kafka DataDog middleware package

packages/kafka-datadog/.npmignore

@@ -0,0 +1,4 @@
+.*
+test/
+tsconfig.json
+tslint.json

packages/kafka-datadog/LICENSE

@@ -0,0 +1,13 @@
+Copyright 2022 OVO Energy
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.

packages/kafka-datadog/README.md

@@ -0,0 +1,60 @@
+# Kafka DataDog
+
+[![License](https://img.shields.io/badge/License-Apache_2.0-blue.svg)](https://opensource.org/licenses/Apache-2.0)
+
+Middleware that instruments a Kafka consumer with datadog APM (Application Performance Monitoring) and metrics.
+
+## Installation
+
+```bash
+yarn add @ovotech/kafka-datadog
+```
+
+## Usage
+
+```typescript
+import { datadog, DependenciesContext } from '@ovotech/kafka-datadog';
+import { Middleware } from '@ovotech/castle';
+
+interface Dependencies {
+  someDependency: string;
+}
+
+const dependencies: Dependencies = {
+  someDependency: 'mock-dependency',
+};
+
+const createDependencyMiddleware = function<T>(dependencies: T): Middleware<DependenciesContext<T>> {
+  return function(next) {
+    return async function(ctx): Promise<void> {
+      await next({ ...ctx, dependencies });
+    };
+  };
+};
+
+const dependencyMiddleware = createDependencyMiddleware(dependencies);
+
+const middleware = (handler: Resolver<any>) => dependencyMiddleware(datadog(handler));
+```
+
+### Coding style (linting, etc) tests
+
+Code style is enforced by using a linter ([tslint](https://palantir.github.io/tslint/)) and [Prettier](https://prettier.io/).
+
+```bash
+yarn lint
+```
+
+## Deployment
+
+Deployment is preferment by lerna automatically on merge / push to master, but you'll need to bump the package version numbers yourself. Only updated packages with newer versions will be pushed to the npm registry.
+
+## Contributing
+
+Have a bug? File an issue with a simple example that reproduces this so we can take a look & confirm.
+
+Want to make a change? Submit a PR, explain why it's useful, and make sure you've updated the docs (this file) and the tests (see [test folder](test)).
+
+## License
+
+This project is licensed under Apache 2 - see the [LICENSE](LICENSE) file for details

packages/kafka-datadog/package.json

@@ -0,0 +1,42 @@
+{
+  "name": "@ovotech/kafka-datadog",
+  "version": "1.0.1",
+  "description": "Middleware that instruments a Kafka consumer with datadog APM (Application Performance Monitoring) and metrics",
+  "main": "dist/index.js",
+  "source": "src/index.ts",
+  "types": "dist/index.d.ts",
+  "author": "Theodore J H Jones <https://github.com/MrKiplin>",
+  "license": "Apache-2.0",
+  "keywords": [
+    "kafka",
+    "datadog",
+    "middleware"
+  ],
+  "scripts": {
+    "lint-prettier": "prettier --list-different {src,test}/**/*.ts",
+    "lint-tslint": "tslint --config tslint.json '{src,test}/**/*.ts'",
+    "lint": "yarn lint-prettier && yarn lint-tslint",
+    "build": "tsc -p ./tsconfig.build.json --outDir dist --declaration",
+    "test": "jest --runInBand"
+  },
+  "devDependencies": {
+    "@types/jest": "^24.0.13",
+    "@types/long": "^4.0.1",
+    "@types/node": "^11.11.4",
+    "jest": "^24.8.0",
+    "prettier": "^1.17.1",
+    "tslint": "^5.17.0",
+    "tslint-config-prettier": "^1.18.0",
+    "ts-retry-promise": "^0.6.0",
+    "typescript": "^3.7.0"
+  },
+  "jest": {
+    "preset": "../../jest-preset.json"
+  },
+  "dependencies": {
+    "@ovotech/castle": "^0.8.1"
+  },
+  "peerDependencies": {
+    "@ovotech/castle": "^0.8.1"
+  }
+}

packages/kafka-datadog/src/datadog.ts

@@ -0,0 +1,48 @@
+import { CastleEachBatchPayload, CastleEachMessagePayload, Middleware } from '@ovotech/castle';
+import { getConsumeMetadata, MessageWithMetadata } from './get-consume-metadata';
+
+export interface DependenciesContext<T> {
+  dependencies: T;
+}
+
+/**
+ * Middleware that instruments a Kafka consumer with datadog APM and metrics
+ */
+export const datadog: Middleware<
+  object,
+  DependenciesContext<any> & CastleEachMessagePayload<MessageWithMetadata> & CastleEachBatchPayload<MessageWithMetadata>
+> = next => ctx => {
+  const meta = getConsumeMetadata(ctx);
+  return ctx.dependencies.tracer.trace('kafka.consume', { resource: meta.topic }, async () => {
+    const start = Date.now();
+
+    const logger = ctx.dependencies.logger.withStaticMeta(meta);
+
+    try {
+      return await next({
+        ...ctx,
+        dependencies: {
+          ...ctx.dependencies,
+          logger,
+        },
+      });
+    } finally {
+      const now = Date.now();
+      ctx.dependencies.metricsTracker.timing('kafka_consumer.consume_time', now - start, {
+        topic: meta.topic,
+      });
+      const messages = ctx.batch ? ctx.batch.messages : [ctx.message];
+      messages.forEach(message => {
+        if (message.timestamp) {
+          ctx.dependencies.metricsTracker.timing(
+            'kafka_consumer.consumption_latency',
+            now - new Date(parseInt(message.timestamp)).getTime(),
+            {
+              topic: meta.topic,
+            },
+          );
+        }
+      });
+    }
+  });
+};

packages/kafka-datadog/src/get-consume-metadata.ts

@@ -0,0 +1,45 @@
+import { CastleEachMessagePayload, CastleEachBatchPayload } from '@ovotech/castle';
+
+export interface MessageWithMetadata {
+  metadata: { eventId: string; createdAt: Date; traceToken: string };
+}
+
+interface ConsumeMetadata {
+  topic: string;
+  topic_partition: string;
+  topic_key?: string;
+  trace_token?: string;
+  high_watermark?: string;
+  batch_length?: string;
+  batch_first_offset?: string;
+  batch_last_offset?: string;
+}
+
+export function getConsumeMetadata(
+  ctx: CastleEachMessagePayload<MessageWithMetadata> & CastleEachBatchPayload<MessageWithMetadata>,
+): ConsumeMetadata {
+  const info = (ctx.batch ?? ctx) as CastleEachBatchPayload['batch'] & CastleEachMessagePayload;
+  let meta: ConsumeMetadata = {
+    topic: info.topic,
+    topic_partition: `${info.partition}`,
+  };
+  if (ctx.message) {
+    meta = {
+      ...meta,
+      topic_key: ctx.message.key?.toString(),
+    };
+    const messageMetadata = ctx.message?.value?.metadata;
+    if (messageMetadata && messageMetadata.traceToken) {
+      meta.trace_token = messageMetadata.traceToken;
+    }
+  } else {
+    meta = {
+      ...meta,
+      high_watermark: info.highWatermark,
+      batch_length: info.messages.length.toString(),
+      batch_first_offset: info.firstOffset() ?? 'null',
+      batch_last_offset: info.lastOffset(),
+    };
+  }
+  return meta;
+}

packages/kafka-datadog/src/index.ts

@@ -0,0 +1 @@
+export { datadog, DependenciesContext } from './datadog';

packages/kafka-datadog/test/integration.spec.ts

@@ -0,0 +1,54 @@
+import { Castle, produce } from '@ovotech/castle';
+import { Schema } from 'avsc';
+import { HelloWorldV1 } from '../test/topics/__generated__/hello_world_v1.json';
+import * as HelloWorldSchema from '../test/topics/schemas/hello_world_v1.json';
+import { retry } from 'ts-retry-promise';
+
+interface Tags {
+  [key: string]: string;
+}
+
+interface MetricsTracker<L extends string> {
+  timing(stat: L, value: number | Date, tags?: Tags | string[]): void;
+}
+
+interface ExtendedGlobal extends NodeJS.Global {
+  castle: Castle;
+  metrics: MetricsTracker<string>;
+}
+
+declare let global: ExtendedGlobal;
+
+describe('datadog', () => {
+  beforeEach(() => {
+    jest.clearAllMocks();
+  });
+
+  const produceFunction = produce<HelloWorldV1>({
+    topic: 'hello_world_v1',
+    schema: HelloWorldSchema as Schema,
+  });
+
+  it('Issues a metric for consumption latency when a message is consumed', async () => {
+    await produceFunction(global.castle.producer, [
+      {
+        key: null,
+        value: {
+          message: 'Hello World',
+          metadata: {
+            eventId: 'test-event-id',
+            createdAt: new Date(),
+            traceToken: 'test-trace-token',
+          },
+        } as HelloWorldV1,
+      },
+    ]);
+
+    await retry(async () => {
+      expect(global.metrics.timing).toBeCalledWith('kafka_consumer.consumption_latency', expect.any(Number), {
+        topic: 'hello_world_v1',
+      });
+      return Promise.resolve();
+    });
+  });
+});

packages/kafka-datadog/test/topics/schemas/hello_world_v1.json

@@ -0,0 +1,38 @@
+{
+  "type": "record",
+  "name": "HelloWorldV1",
+  "namespace": "com.ovoenergy.kafka.boost.notification",
+  "fields": [
+    {
+      "name": "message",
+      "type": "string",
+      "doc": "Hello World"
+    },
+    {
+      "name": "metadata",
+      "type": {
+        "type": "record",
+        "name": "EventMetadata",
+        "namespace": "com.ovoenergy.kafka.common.event",
+        "fields": [
+          {
+            "name": "eventId",
+            "type": "string"
+          },
+          {
+            "name": "traceToken",
+            "type": "string"
+          },
+          {
+            "name": "createdAt",
+            "type": {
+              "type": "long",
+              "logicalType": "timestamp-millis"
+            }
+          }
+        ]
+      }
+    }
+  ],
+  "doc:": "A basic avro schema for testing purposes"
+}

packages/kafka-datadog/tsconfig.build.json

@@ -0,0 +1,5 @@
+{
+  "extends": "./tsconfig.json",
+  "include": ["src"],
+  "exclude": ["**/*.spec.ts"]
+}

packages/kafka-datadog/tsconfig.json

@@ -0,0 +1,22 @@
+{
+  "compilerOptions": {
+    "experimentalDecorators": true,
+    "emitDecoratorMetadata": true,
+    "forceConsistentCasingInFileNames": true,
+    "noImplicitReturns": true,
+    "noImplicitThis": true,
+    "noImplicitAny": true,
+    "resolveJsonModule": true,
+    "strictNullChecks": true,
+    "sourceMap": true,
+    "skipLibCheck": true,
+    "strict": true,
+    "module": "commonjs",
+    "target": "es2018",
+    "lib": ["es2015", "es2018", "esnext.asynciterable", "ES2019"],
+    "outDir": "dist",
+    "downlevelIteration": true
+  },
+  "include": ["src", "test"],
+  "exclude": ["node_modules", "dist"]
+}

packages/kafka-datadog/tslint.json

@@ -0,0 +1,3 @@
+{
+  "extends": "../../tslint.base.json"
+}