Add Patching Mechanism with AWS SDK telemetry improvements

aws-distro-opentelemetry-node-autoinstrumentation/src/aws-attribute-keys.ts

@@ -1,6 +1,12 @@
 // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
 
+import {
+  SEMATTRS_AWS_DYNAMODB_TABLE_NAMES,
+  SEMATTRS_MESSAGING_DESTINATION,
+  SEMATTRS_MESSAGING_URL,
+} from '@opentelemetry/semantic-conventions';
+
 // Utility class holding attribute keys with special meaning to AWS components
 export const AWS_ATTRIBUTE_KEYS: { [key: string]: string } = {
   AWS_SPAN_KIND: 'aws.span.kind',
@@ -20,13 +26,11 @@ export const AWS_ATTRIBUTE_KEYS: { [key: string]: string } = {
   AWS_IS_LOCAL_ROOT: 'aws.is.local.root',
 
   // Divergence from Java/Python
-  // TODO: Audit this: These will most definitely be different in JavaScript.
-  //   For example:
-  //     - `messaging.url` for AWS_QUEUE_URL
-  //     - `aws.dynamodb.table_names` for AWS_TABLE_NAME
-  AWS_BUCKET_NAME: 'aws.bucket.name',
-  AWS_QUEUE_URL: 'aws.queue.url',
-  AWS_QUEUE_NAME: 'aws.queue.name',
-  AWS_STREAM_NAME: 'aws.stream.name',
-  AWS_TABLE_NAME: 'aws.table.name',
+  // For consistency between ADOT SDK languages, the attribute Key name is named similarly to Java/Python,
+  // while the value is different to accommodate the actual attribute set from OTel JS instrumentations
+  AWS_BUCKET_NAME: 'aws.s3.bucket',
+  AWS_QUEUE_URL: SEMATTRS_MESSAGING_URL,
+  AWS_QUEUE_NAME: SEMATTRS_MESSAGING_DESTINATION,
+  AWS_STREAM_NAME: 'aws.kinesis.stream.name',
+  AWS_TABLE_NAMES: SEMATTRS_AWS_DYNAMODB_TABLE_NAMES,
 };

aws-distro-opentelemetry-node-autoinstrumentation/src/aws-metric-attribute-generator.ts

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 import { Attributes, AttributeValue, diag, SpanKind } from '@opentelemetry/api';
-import { Resource, defaultServiceName } from '@opentelemetry/resources';
+import { defaultServiceName, Resource } from '@opentelemetry/resources';
 import { ReadableSpan } from '@opentelemetry/sdk-trace-base';
 import {
   SEMATTRS_DB_CONNECTION_STRING,
@@ -340,10 +340,13 @@ export class AwsMetricAttributeGenerator implements MetricAttributeGenerator {
     let remoteResourceIdentifier: AttributeValue | undefined;
 
     if (AwsSpanProcessingUtil.isAwsSDKSpan(span)) {
-      if (AwsSpanProcessingUtil.isKeyPresent(span, AWS_ATTRIBUTE_KEYS.AWS_TABLE_NAME)) {
+      if (
+        AwsSpanProcessingUtil.isKeyPresent(span, AWS_ATTRIBUTE_KEYS.AWS_TABLE_NAMES) &&
+        (span.attributes[AWS_ATTRIBUTE_KEYS.AWS_TABLE_NAMES] as string[]).length === 1
+      ) {
         remoteResourceType = NORMALIZED_DYNAMO_DB_SERVICE_NAME + '::Table';
         remoteResourceIdentifier = AwsMetricAttributeGenerator.escapeDelimiters(
-          span.attributes[AWS_ATTRIBUTE_KEYS.AWS_TABLE_NAME]
+          (span.attributes[AWS_ATTRIBUTE_KEYS.AWS_TABLE_NAMES] as string[])[0]
         );
       } else if (AwsSpanProcessingUtil.isKeyPresent(span, AWS_ATTRIBUTE_KEYS.AWS_STREAM_NAME)) {
         remoteResourceType = NORMALIZED_KINESIS_SERVICE_NAME + '::Stream';

aws-distro-opentelemetry-node-autoinstrumentation/src/aws-opentelemetry-configurator.ts

@@ -2,10 +2,8 @@
 // SPDX-License-Identifier: Apache-2.0
 
 import { diag } from '@opentelemetry/api';
-import {
-  getNodeAutoInstrumentations,
-  getResourceDetectors as getResourceDetectorsFromEnv,
-} from '@opentelemetry/auto-instrumentations-node';
+import { getResourceDetectors as getResourceDetectorsFromEnv } from '@opentelemetry/auto-instrumentations-node';
+import { Instrumentation } from '@opentelemetry/instrumentation';
 import { awsEc2Detector, awsEcsDetector, awsEksDetector } from '@opentelemetry/resource-detector-aws';
 import {
   Detector,
@@ -37,8 +35,9 @@ const APPLICATION_SIGNALS_ENABLED_CONFIG: string = 'OTEL_AWS_APPLICATION_SIGNALS
  */
 export class AwsOpentelemetryConfigurator {
   private resource: Resource;
+  private instrumentations: Instrumentation[];
 
-  constructor() {
+  constructor(instrumentations: Instrumentation[]) {
     /*
      * Set and Detect Resources via Resource Detectors
      *
@@ -88,6 +87,8 @@ export class AwsOpentelemetryConfigurator {
     autoResource = autoResource.merge(detectResourcesSync(internalConfig));
 
     this.resource = autoResource;
+
+    this.instrumentations = instrumentations;
   }
 
   private customizeVersions(autoResource: Resource): Resource {
@@ -106,14 +107,14 @@ export class AwsOpentelemetryConfigurator {
     if (this.isApplicationSignalsEnabled()) {
       // TODO: This is a placeholder config. This will be replaced with an ADOT config in a future commit.
       config = {
-        instrumentations: getNodeAutoInstrumentations(),
+        instrumentations: this.instrumentations,
         resource: this.resource,
         autoDetectResources: false,
       };
     } else {
       // Default experience config
       config = {
-        instrumentations: getNodeAutoInstrumentations(),
+        instrumentations: this.instrumentations,
         resource: this.resource,
         autoDetectResources: false,
       };

aws-distro-opentelemetry-node-autoinstrumentation/src/patches/aws/services/ServiceExtension.ts

@@ -0,0 +1,38 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+import { DiagLogger, Span, SpanAttributes, SpanKind, Tracer } from '@opentelemetry/api';
+import {
+  AwsSdkInstrumentationConfig,
+  NormalizedRequest,
+  NormalizedResponse,
+} from '@opentelemetry/instrumentation-aws-sdk';
+
+// The OpenTelemetry Authors code
+// We need to copy these interfaces that are not exported by Opentelemetry
+export interface RequestMetadata {
+  // isIncoming - if true, then the operation callback / promise should be bind with the operation's span
+  isIncoming: boolean;
+  spanAttributes?: SpanAttributes;
+  spanKind?: SpanKind;
+  spanName?: string;
+}
+
+export interface ServiceExtension {
+  // called before request is sent, and before span is started
+  requestPreSpanHook: (
+    request: NormalizedRequest,
+    config: AwsSdkInstrumentationConfig,
+    diag: DiagLogger
+  ) => RequestMetadata;
+
+  // called before request is sent, and after span is started
+  requestPostSpanHook?: (request: NormalizedRequest) => void;
+
+  responseHook?: (
+    response: NormalizedResponse,
+    span: Span,
+    tracer: Tracer,
+    config: AwsSdkInstrumentationConfig
+  ) => void;
+}

aws-distro-opentelemetry-node-autoinstrumentation/src/patches/aws/services/index.ts

@@ -0,0 +1,5 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+export { KinesisServiceExtension } from './kinesis';
+export { S3ServiceExtension } from './s3';

aws-distro-opentelemetry-node-autoinstrumentation/src/patches/aws/services/kinesis.ts

@@ -0,0 +1,44 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+import { Attributes, Span, SpanKind, Tracer } from '@opentelemetry/api';
+import {
+  AwsSdkInstrumentationConfig,
+  NormalizedRequest,
+  NormalizedResponse,
+} from '@opentelemetry/instrumentation-aws-sdk';
+import { RequestMetadata, ServiceExtension } from './ServiceExtension';
+
+// The OpenTelemetry Authors code
+// This file's contents are being contributed to upstream
+// - https://github.com/open-telemetry/opentelemetry-js-contrib/pull/2361
+
+const _AWS_KINESIS_STREAM_NAME = 'aws.kinesis.stream.name';
+
+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) => {};
+}

aws-distro-opentelemetry-node-autoinstrumentation/src/patches/aws/services/s3.ts

@@ -0,0 +1,44 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+import { Attributes, Span, SpanKind, Tracer } from '@opentelemetry/api';
+import {
+  AwsSdkInstrumentationConfig,
+  NormalizedRequest,
+  NormalizedResponse,
+} from '@opentelemetry/instrumentation-aws-sdk';
+import { RequestMetadata, ServiceExtension } from './ServiceExtension';
+
+// The OpenTelemetry Authors code
+// This file's contents are being contributed to upstream
+// - https://github.com/open-telemetry/opentelemetry-js-contrib/pull/2361
+
+const _AWS_S3_BUCKET = 'aws.s3.bucket';
+
+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) => {};
+}

aws-distro-opentelemetry-node-autoinstrumentation/src/patches/instrumentation-patch.ts

@@ -0,0 +1,30 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+import { Instrumentation } from '@opentelemetry/instrumentation';
+import { AwsInstrumentation } from '@opentelemetry/instrumentation-aws-sdk';
+import { KinesisServiceExtension, S3ServiceExtension } from './aws/services';
+
+export function applyInstrumentationPatches(instrumentations: Instrumentation[]): void {
+  /*
+  Apply patches to upstream instrumentation libraries.
+
+  This method is invoked to apply changes to upstream instrumentation libraries, typically when changes to upstream
+  are required on a timeline that cannot wait for upstream release. Generally speaking, patches should be short-term
+  local solutions that are comparable to long-term upstream solutions.
+
+  Where possible, automated testing should be run to catch upstream changes resulting in broken patches
+  */
+  instrumentations.forEach(instrumentation => {
+    if (instrumentation.instrumentationName === '@opentelemetry/instrumentation-aws-sdk') {
+      // Access private property servicesExtensions of AwsInstrumentation
+      // eslint-disable-next-line @typescript-eslint/ban-ts-comment
+      // @ts-ignore
+      const services = (instrumentation as AwsInstrumentation).servicesExtensions?.services;
+      if (services) {
+        services.set('S3', new S3ServiceExtension());
+        services.set('Kinesis', new KinesisServiceExtension());
+      }
+    }
+  });
+}

aws-distro-opentelemetry-node-autoinstrumentation/src/register.ts

@@ -2,8 +2,11 @@
 // SPDX-License-Identifier: Apache-2.0
 
 import { DiagConsoleLogger, diag } from '@opentelemetry/api';
+import { getNodeAutoInstrumentations } from '@opentelemetry/auto-instrumentations-node';
+import { Instrumentation } from '@opentelemetry/instrumentation';
 import * as opentelemetry from '@opentelemetry/sdk-node';
 import { AwsOpentelemetryConfigurator } from './aws-opentelemetry-configurator';
+import { applyInstrumentationPatches } from './patches/instrumentation-patch';
 
 diag.setLogger(new DiagConsoleLogger(), opentelemetry.core.getEnv().OTEL_LOG_LEVEL);
 
@@ -31,7 +34,14 @@ if (!process.env.OTEL_PROPAGATORS) {
   process.env.OTEL_PROPAGATORS = 'xray,tracecontext,b3,b3multi';
 }
 
-const configurator: AwsOpentelemetryConfigurator = new AwsOpentelemetryConfigurator();
+const instrumentations: Instrumentation[] = getNodeAutoInstrumentations();
+
+// Apply instrumentation patches by default
+if (process.env.AWS_APPLY_PATCHES === undefined || process.env.AWS_APPLY_PATCHES?.toLowerCase() === 'true') {
+  applyInstrumentationPatches(instrumentations);
+}
+
+const configurator: AwsOpentelemetryConfigurator = new AwsOpentelemetryConfigurator(instrumentations);
 const configuration: Partial<opentelemetry.NodeSDKConfiguration> = configurator.configure();
 
 const sdk: opentelemetry.NodeSDK = new opentelemetry.NodeSDK(configuration);

aws-distro-opentelemetry-node-autoinstrumentation/test/aws-metric-attribute-generator.test.ts

@@ -760,20 +760,30 @@ describe('AwsMetricAttributeGeneratorTest', () => {
     validateRemoteResourceAttributes('AWS::Kinesis::Stream', 'aws_stream_name');
     mockAttribute(AWS_ATTRIBUTE_KEYS.AWS_STREAM_NAME, undefined);
 
-    // Validate behaviour of AWS_TABLE_NAME attribute, then remove it.
-    mockAttribute(AWS_ATTRIBUTE_KEYS.AWS_TABLE_NAME, 'aws_table_name');
+    // Validate behaviour of AWS_TABLE_NAMES attribute with one table name, then remove it.
+    mockAttribute(AWS_ATTRIBUTE_KEYS.AWS_TABLE_NAMES, ['aws_table_name']);
     validateRemoteResourceAttributes('AWS::DynamoDB::Table', 'aws_table_name');
-    mockAttribute(AWS_ATTRIBUTE_KEYS.AWS_TABLE_NAME, undefined);
+    mockAttribute(AWS_ATTRIBUTE_KEYS.AWS_TABLE_NAMES, undefined);
 
-    // Validate behaviour of AWS_TABLE_NAME attribute with special chars(|), then remove it.
-    mockAttribute(AWS_ATTRIBUTE_KEYS.AWS_TABLE_NAME, 'aws_table|name');
+    // Validate behaviour of AWS_TABLE_NAMES attribute with no table name, then remove it.
+    mockAttribute(AWS_ATTRIBUTE_KEYS.AWS_TABLE_NAMES, []);
+    validateRemoteResourceAttributes(undefined, undefined);
+    mockAttribute(AWS_ATTRIBUTE_KEYS.AWS_TABLE_NAMES, undefined);
+
+    // Validate behaviour of AWS_TABLE_NAMES attribute with two table names, then remove it.
+    mockAttribute(AWS_ATTRIBUTE_KEYS.AWS_TABLE_NAMES, ['aws_table_name1', 'aws_table_name2']);
+    validateRemoteResourceAttributes(undefined, undefined);
+    mockAttribute(AWS_ATTRIBUTE_KEYS.AWS_TABLE_NAMES, undefined);
+
+    // Validate behaviour of AWS_TABLE_NAMES attribute with special chars(|), then remove it.
+    mockAttribute(AWS_ATTRIBUTE_KEYS.AWS_TABLE_NAMES, ['aws_table|name']);
     validateRemoteResourceAttributes('AWS::DynamoDB::Table', 'aws_table^|name');
-    mockAttribute(AWS_ATTRIBUTE_KEYS.AWS_TABLE_NAME, undefined);
+    mockAttribute(AWS_ATTRIBUTE_KEYS.AWS_TABLE_NAMES, undefined);
 
-    // Validate behaviour of AWS_TABLE_NAME attribute with special chars(^), then remove it.
-    mockAttribute(AWS_ATTRIBUTE_KEYS.AWS_TABLE_NAME, 'aws_table^name');
+    // Validate behaviour of AWS_TABLE_NAMES attribute with special chars(^), then remove it.
+    mockAttribute(AWS_ATTRIBUTE_KEYS.AWS_TABLE_NAMES, ['aws_table^name']);
     validateRemoteResourceAttributes('AWS::DynamoDB::Table', 'aws_table^^name');
-    mockAttribute(AWS_ATTRIBUTE_KEYS.AWS_TABLE_NAME, undefined);
+    mockAttribute(AWS_ATTRIBUTE_KEYS.AWS_TABLE_NAMES, undefined);
   });
 
   it('testDBClientSpanWithRemoteResourceAttributes', () => {
@@ -1010,7 +1020,7 @@ describe('AwsMetricAttributeGeneratorTest', () => {
     mockAttribute(SEMATTRS_PEER_SERVICE, undefined);
   }
 
-  function validateRemoteResourceAttributes(type: string, identifier: string): void {
+  function validateRemoteResourceAttributes(type: string | undefined, identifier: string | undefined): void {
     // Client, Producer and Consumer spans should generate the expected remote resource attributes
     (spanDataMock as any).kind = SpanKind.CLIENT;
     let actualAttributes: Attributes = GENERATOR.generateMetricAttributeMapFromSpan(spanDataMock, resource)[

aws-distro-opentelemetry-node-autoinstrumentation/test/patches/instrumentation-patch.test.ts

@@ -0,0 +1,36 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+import { getNodeAutoInstrumentations } from '@opentelemetry/auto-instrumentations-node';
+import { Instrumentation } from '@opentelemetry/instrumentation';
+import { AwsInstrumentation } from '@opentelemetry/instrumentation-aws-sdk';
+import { expect } from 'expect';
+import { applyInstrumentationPatches } from './../../src/patches/instrumentation-patch';
+
+describe('InstrumentationPatchTest', () => {
+  it('PatchAwsSdkInstrumentation', () => {
+    const instrumentations: Instrumentation[] = getNodeAutoInstrumentations();
+    applyInstrumentationPatches(instrumentations);
+
+    const filteredInstrumentations: Instrumentation[] = instrumentations.filter(
+      instrumentation => instrumentation.instrumentationName === '@opentelemetry/instrumentation-aws-sdk'
+    );
+    expect(filteredInstrumentations.length).toEqual(1);
+
+    const awsSdkInstrumentation = filteredInstrumentations[0];
+    // eslint-disable-next-line @typescript-eslint/ban-ts-comment
+    // @ts-ignore
+    const services: Map<string, ServiceExtension> = (awsSdkInstrumentation as AwsInstrumentation).servicesExtensions
+      ?.services;
+    // Not from patching
+    expect(services.has('SQS')).toBeTruthy();
+    expect(services.has('SNS')).toBeTruthy();
+    expect(services.has('DynamoDB')).toBeTruthy();
+    expect(services.has('Lambda')).toBeTruthy();
+    // From patching
+    expect(services.has('S3')).toBeTruthy();
+    expect(services.has('Kinesis')).toBeTruthy();
+    // Sanity check
+    expect(services.has('InvalidService')).toBeFalsy();
+  });
+});