add s3 and kinesis service extensions for aws-sdk instrumentation

Author: jj22ee

plugins/node/opentelemetry-instrumentation-aws-sdk/package.json

@@ -52,6 +52,7 @@
   },
   "devDependencies": {
     "@aws-sdk/client-dynamodb": "3.85.0",
+    "@aws-sdk/client-kinesis": "3.85.0",
     "@aws-sdk/client-lambda": "3.85.0",
     "@aws-sdk/client-s3": "3.85.0",
     "@aws-sdk/client-sns": "3.85.0",

plugins/node/opentelemetry-instrumentation-aws-sdk/src/services/ServicesExtensions.ts

@@ -24,6 +24,8 @@ import {
 import { DynamodbServiceExtension } from './dynamodb';
 import { SnsServiceExtension } from './sns';
 import { LambdaServiceExtension } from './lambda';
+import { S3ServiceExtension } from './s3';
+import { KinesisServiceExtension } from './kinesis';
 
 export class ServicesExtensions implements ServiceExtension {
   services: Map<string, ServiceExtension> = new Map();
@@ -33,6 +35,8 @@ export class ServicesExtensions implements ServiceExtension {
     this.services.set('SNS', new SnsServiceExtension());
     this.services.set('DynamoDB', new DynamodbServiceExtension());
     this.services.set('Lambda', new LambdaServiceExtension());
+    this.services.set('S3', new S3ServiceExtension());
+    this.services.set('Kinesis', new KinesisServiceExtension());
   }
 
   requestPreSpanHook(

plugins/node/opentelemetry-instrumentation-aws-sdk/src/services/kinesis.ts

@@ -0,0 +1,59 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ *
+ * 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
+ *
+ *      https://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.
+ */
+import { Attributes, Span, SpanKind, Tracer } from '@opentelemetry/api';
+import {
+  AwsSdkInstrumentationConfig,
+  NormalizedRequest,
+  NormalizedResponse,
+} from '../types';
+import { _AWS_KINESIS_STREAM_NAME } from '../utils';
+import { RequestMetadata, ServiceExtension } from './ServiceExtension';
+
+export class KinesisServiceExtension implements ServiceExtension {
+  requestPreSpanHook(
+    request: NormalizedRequest,
+    _config: AwsSdkInstrumentationConfig
+  ): RequestMetadata {
+    const streamName = request.commandInput.StreamName;
+
+    const spanKind: SpanKind = SpanKind.CLIENT;
+    let spanName: string | undefined;
+
+    const spanAttributes: Attributes = {};
+
+    if (streamName) {
+      spanAttributes[_AWS_KINESIS_STREAM_NAME] = streamName;
+    }
+
+    const isIncoming = false;
+
+    return {
+      isIncoming,
+      spanAttributes,
+      spanKind,
+      spanName,
+    };
+  }
+
+  requestPostSpanHook = (request: NormalizedRequest) => {};
+
+  responseHook = (
+    response: NormalizedResponse,
+    span: Span,
+    tracer: Tracer,
+    config: AwsSdkInstrumentationConfig
+  ) => {};
+}

plugins/node/opentelemetry-instrumentation-aws-sdk/src/services/s3.ts

@@ -0,0 +1,59 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ *
+ * 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
+ *
+ *      https://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.
+ */
+import { Attributes, Span, SpanKind, Tracer } from '@opentelemetry/api';
+import {
+  AwsSdkInstrumentationConfig,
+  NormalizedRequest,
+  NormalizedResponse,
+} from '../types';
+import { _AWS_S3_BUCKET } from '../utils';
+import { RequestMetadata, ServiceExtension } from './ServiceExtension';
+
+export class S3ServiceExtension implements ServiceExtension {
+  requestPreSpanHook(
+    request: NormalizedRequest,
+    _config: AwsSdkInstrumentationConfig
+  ): RequestMetadata {
+    const bucketName = request.commandInput.Bucket;
+
+    const spanKind: SpanKind = SpanKind.CLIENT;
+    let spanName: string | undefined;
+
+    const spanAttributes: Attributes = {};
+
+    if (bucketName) {
+      spanAttributes[_AWS_S3_BUCKET] = bucketName;
+    }
+
+    const isIncoming = false;
+
+    return {
+      isIncoming,
+      spanAttributes,
+      spanKind,
+      spanName,
+    };
+  }
+
+  requestPostSpanHook = (request: NormalizedRequest) => {};
+
+  responseHook = (
+    response: NormalizedResponse,
+    span: Span,
+    tracer: Tracer,
+    config: AwsSdkInstrumentationConfig
+  ) => {};
+}

plugins/node/opentelemetry-instrumentation-aws-sdk/src/utils.ts

@@ -22,6 +22,12 @@ import {
 } from '@opentelemetry/semantic-conventions';
 import { AttributeNames } from './enums';
 
+// TODO: Add these semantic attributes to:
+// - https://github.com/open-telemetry/opentelemetry-js/blob/main/packages/opentelemetry-semantic-conventions/src/trace/SemanticAttributes.ts
+// For S3, see specification: https://github.com/open-telemetry/semantic-conventions/blob/main/docs/object-stores/s3.md
+export const _AWS_S3_BUCKET = 'aws.s3.bucket';
+export const _AWS_KINESIS_STREAM_NAME = 'aws.kinesis.stream.name';
+
 const toPascalCase = (str: string): string =>
   typeof str === 'string' ? str.charAt(0).toUpperCase() + str.slice(1) : str;
 

plugins/node/opentelemetry-instrumentation-aws-sdk/test/kinesis.test.ts

@@ -0,0 +1,81 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ *
+ * 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
+ *
+ *      https://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.
+ */
+
+import {
+  getTestSpans,
+  registerInstrumentationTesting,
+} from '@opentelemetry/contrib-test-utils';
+import { AwsInstrumentation } from '../src';
+registerInstrumentationTesting(new AwsInstrumentation());
+
+import * as AWS from 'aws-sdk';
+import { AWSError } from 'aws-sdk';
+import * as nock from 'nock';
+
+import { SpanKind } from '@opentelemetry/api';
+import { ReadableSpan } from '@opentelemetry/sdk-trace-base';
+import { expect } from 'expect';
+import { _AWS_KINESIS_STREAM_NAME } from '../src/utils';
+
+const region = 'us-east-1';
+
+describe('Kinesis', () => {
+  let kinesis: AWS.Kinesis;
+  beforeEach(() => {
+    AWS.config.credentials = {
+      accessKeyId: 'test key id',
+      expired: false,
+      expireTime: new Date(),
+      secretAccessKey: 'test acc key',
+      sessionToken: 'test token',
+    };
+  });
+
+  describe('CreateStream', () => {
+    it('adds Stream Name', async () => {
+      kinesis = new AWS.Kinesis({ region: region });
+      const dummyStreamName = 'dummy-stream-name';
+
+      nock(`https://kinesis.${region}.amazonaws.com`)
+        .get('/')
+        .reply(200, 'null');
+
+      await kinesis
+        .createStream(
+          {
+            StreamName: dummyStreamName,
+            ShardCount: 3,
+          },
+          (err: AWSError) => {
+            expect(err).toBeFalsy();
+          }
+        )
+        .promise();
+
+      const testSpans = getTestSpans();
+      console.log(testSpans.length);
+      const creationSpans = testSpans.filter((s: ReadableSpan) => {
+        return s.name === 'Kinesis.CreateStream';
+      });
+      expect(creationSpans.length).toBe(1);
+      const publishSpan = creationSpans[0];
+      expect(publishSpan.attributes[_AWS_KINESIS_STREAM_NAME]).toBe(
+        dummyStreamName
+      );
+      expect(publishSpan.kind).toBe(SpanKind.CLIENT);
+    });
+  });
+});

plugins/node/opentelemetry-instrumentation-aws-sdk/test/s3.test.ts

@@ -0,0 +1,76 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ *
+ * 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
+ *
+ *      https://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.
+ */
+
+import {
+  getTestSpans,
+  registerInstrumentationTesting,
+} from '@opentelemetry/contrib-test-utils';
+import { AwsInstrumentation } from '../src';
+registerInstrumentationTesting(new AwsInstrumentation());
+
+import * as AWS from 'aws-sdk';
+import { AWSError } from 'aws-sdk';
+import * as nock from 'nock';
+
+import { SpanKind } from '@opentelemetry/api';
+import { ReadableSpan } from '@opentelemetry/sdk-trace-base';
+import { expect } from 'expect';
+import { _AWS_S3_BUCKET } from '../src/utils';
+
+const region = 'us-east-1';
+
+describe('S3', () => {
+  let s3: AWS.S3;
+  beforeEach(() => {
+    AWS.config.credentials = {
+      accessKeyId: 'test key id',
+      expired: false,
+      expireTime: new Date(),
+      secretAccessKey: 'test acc key',
+      sessionToken: 'test token',
+    };
+  });
+
+  describe('ListObjects', () => {
+    it('adds bucket Name', async () => {
+      s3 = new AWS.S3({ region: region });
+      const dummyBucketName = 'dummy-bucket-name';
+
+      nock(`https://s3.${region}.amazonaws.com`).get('/').reply(200, 'null');
+
+      await s3
+        .listObjects(
+          {
+            Bucket: dummyBucketName,
+          },
+          (err: AWSError) => {
+            expect(err).toBeFalsy();
+          }
+        )
+        .promise();
+
+      const testSpans = getTestSpans();
+      console.log(testSpans.length);
+      const creationSpans = testSpans.filter((s: ReadableSpan) => {
+        return s.name === 'S3.ListObjects';
+      });
+      expect(creationSpans.length).toBe(1);
+      const publishSpan = creationSpans[0];
+      expect(publishSpan.attributes[_AWS_S3_BUCKET]).toBe(dummyBucketName);
+      expect(publishSpan.kind).toBe(SpanKind.CLIENT);
+    });
+  });
+});