Add Patching Mechanism with AWS SDK telemetry improvements

aws-distro-opentelemetry-node-autoinstrumentation/package.json

@@ -30,13 +30,17 @@
     "prewatch": "npm run precompile",
     "prepublishOnly": "npm run compile",
     "tdd": "yarn test -- --watch-extensions ts --watch",
-    "test": "nyc ts-mocha --timeout 10000 -p tsconfig.json 'test/**/*.ts'",
+    "test": "nyc ts-mocha --timeout 10000 -p tsconfig.json --require '@opentelemetry/contrib-test-utils' 'test/**/*.ts'",
     "watch": "tsc -w"
   },
   "bugs": {
     "url": "https://github.com/aws-observability/aws-otel-js-instrumentation/issues"
   },
   "devDependencies": {
+    "@aws-sdk/client-kinesis": "3.85.0",
+    "@aws-sdk/client-s3": "3.85.0",
+    "@aws-sdk/client-sqs": "3.85.0",
+    "@opentelemetry/contrib-test-utils": "^0.40.0",
     "@types/mocha": "7.0.2",
     "@types/node": "18.6.5",
     "@types/sinon": "10.0.18",

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

@@ -1,11 +1,7 @@
 // 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';
+import { SEMATTRS_AWS_DYNAMODB_TABLE_NAMES } from '@opentelemetry/semantic-conventions';
 
 // Utility class holding attribute keys with special meaning to AWS components
 export const AWS_ATTRIBUTE_KEYS: { [key: string]: string } = {
@@ -25,12 +21,10 @@ export const AWS_ATTRIBUTE_KEYS: { [key: string]: string } = {
   // Used for JavaScript workaround - attribute for pre-calculated value of isLocalRoot
   AWS_IS_LOCAL_ROOT: 'aws.is.local.root',
 
-  // Divergence from Java/Python
-  // 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,
+  // Naming divergence from Java/Python
+  AWS_S3_BUCKET: 'aws.s3.bucket',
+  AWS_SQS_QUEUE_URL: 'aws.sqs.queue.url',
+  AWS_SQS_QUEUE_NAME: 'aws.sqs.queue.name',
+  AWS_KINESIS_STREAM_NAME: 'aws.kinesis.stream.name',
+  AWS_DYNAMODB_TABLE_NAMES: SEMATTRS_AWS_DYNAMODB_TABLE_NAMES,
 };

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

@@ -340,33 +340,33 @@ export class AwsMetricAttributeGenerator implements MetricAttributeGenerator {
     let remoteResourceIdentifier: AttributeValue | undefined;
 
     if (AwsSpanProcessingUtil.isAwsSDKSpan(span)) {
+      const awsTableNames: AttributeValue | undefined = span.attributes[AWS_ATTRIBUTE_KEYS.AWS_DYNAMODB_TABLE_NAMES];
       if (
-        AwsSpanProcessingUtil.isKeyPresent(span, AWS_ATTRIBUTE_KEYS.AWS_TABLE_NAMES) &&
-        (span.attributes[AWS_ATTRIBUTE_KEYS.AWS_TABLE_NAMES] as string[]).length === 1
+        AwsSpanProcessingUtil.isKeyPresent(span, AWS_ATTRIBUTE_KEYS.AWS_DYNAMODB_TABLE_NAMES) &&
+        Array.isArray(awsTableNames) &&
+        awsTableNames.length === 1
       ) {
         remoteResourceType = NORMALIZED_DYNAMO_DB_SERVICE_NAME + '::Table';
-        remoteResourceIdentifier = AwsMetricAttributeGenerator.escapeDelimiters(
-          (span.attributes[AWS_ATTRIBUTE_KEYS.AWS_TABLE_NAMES] as string[])[0]
-        );
-      } else if (AwsSpanProcessingUtil.isKeyPresent(span, AWS_ATTRIBUTE_KEYS.AWS_STREAM_NAME)) {
+        remoteResourceIdentifier = AwsMetricAttributeGenerator.escapeDelimiters(awsTableNames[0]);
+      } else if (AwsSpanProcessingUtil.isKeyPresent(span, AWS_ATTRIBUTE_KEYS.AWS_KINESIS_STREAM_NAME)) {
         remoteResourceType = NORMALIZED_KINESIS_SERVICE_NAME + '::Stream';
         remoteResourceIdentifier = AwsMetricAttributeGenerator.escapeDelimiters(
-          span.attributes[AWS_ATTRIBUTE_KEYS.AWS_STREAM_NAME]
+          span.attributes[AWS_ATTRIBUTE_KEYS.AWS_KINESIS_STREAM_NAME]
         );
-      } else if (AwsSpanProcessingUtil.isKeyPresent(span, AWS_ATTRIBUTE_KEYS.AWS_BUCKET_NAME)) {
+      } else if (AwsSpanProcessingUtil.isKeyPresent(span, AWS_ATTRIBUTE_KEYS.AWS_S3_BUCKET)) {
         remoteResourceType = NORMALIZED_S3_SERVICE_NAME + '::Bucket';
         remoteResourceIdentifier = AwsMetricAttributeGenerator.escapeDelimiters(
-          span.attributes[AWS_ATTRIBUTE_KEYS.AWS_BUCKET_NAME]
+          span.attributes[AWS_ATTRIBUTE_KEYS.AWS_S3_BUCKET]
         );
-      } else if (AwsSpanProcessingUtil.isKeyPresent(span, AWS_ATTRIBUTE_KEYS.AWS_QUEUE_NAME)) {
+      } else if (AwsSpanProcessingUtil.isKeyPresent(span, AWS_ATTRIBUTE_KEYS.AWS_SQS_QUEUE_NAME)) {
         remoteResourceType = NORMALIZED_SQS_SERVICE_NAME + '::Queue';
         remoteResourceIdentifier = AwsMetricAttributeGenerator.escapeDelimiters(
-          span.attributes[AWS_ATTRIBUTE_KEYS.AWS_QUEUE_NAME]
+          span.attributes[AWS_ATTRIBUTE_KEYS.AWS_SQS_QUEUE_NAME]
         );
-      } else if (AwsSpanProcessingUtil.isKeyPresent(span, AWS_ATTRIBUTE_KEYS.AWS_QUEUE_URL)) {
+      } else if (AwsSpanProcessingUtil.isKeyPresent(span, AWS_ATTRIBUTE_KEYS.AWS_SQS_QUEUE_URL)) {
         remoteResourceType = NORMALIZED_SQS_SERVICE_NAME + '::Queue';
         remoteResourceIdentifier = SqsUrlParser.getQueueName(
-          AwsMetricAttributeGenerator.escapeDelimiters(span.attributes[AWS_ATTRIBUTE_KEYS.AWS_QUEUE_URL])
+          AwsMetricAttributeGenerator.escapeDelimiters(span.attributes[AWS_ATTRIBUTE_KEYS.AWS_SQS_QUEUE_URL])
         );
       }
     } else if (AwsSpanProcessingUtil.isDBSpan(span)) {
@@ -482,8 +482,8 @@ export class AwsMetricAttributeGenerator implements MetricAttributeGenerator {
     return AwsMetricAttributeGenerator.escapeDelimiters(address) + (port !== '' ? '|' + port : '');
   }
 
-  private static escapeDelimiters(input: string | AttributeValue | undefined): string | undefined {
-    if (input === undefined) {
+  private static escapeDelimiters(input: string | AttributeValue | undefined | null): string | undefined {
+    if (typeof input !== 'string') {
       return undefined;
     }
 

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

@@ -2,8 +2,20 @@
 // SPDX-License-Identifier: Apache-2.0
 // Modifications Copyright The OpenTelemetry Authors. Licensed under the Apache License 2.0 License.
 
-import { diag } from '@opentelemetry/api';
+import { TextMapPropagator, diag } from '@opentelemetry/api';
+import { getPropagator } from '@opentelemetry/auto-configuration-propagators';
 import { getResourceDetectors as getResourceDetectorsFromEnv } from '@opentelemetry/auto-instrumentations-node';
+import { ENVIRONMENT, TracesSamplerValues, getEnv, getEnvWithoutDefaults } from '@opentelemetry/core';
+import { OTLPMetricExporter as OTLPGrpcOTLPMetricExporter } from '@opentelemetry/exporter-metrics-otlp-grpc';
+import {
+  AggregationTemporalityPreference,
+  OTLPMetricExporter as OTLPHttpOTLPMetricExporter,
+} from '@opentelemetry/exporter-metrics-otlp-http';
+import { OTLPTraceExporter as OTLPGrpcTraceExporter } from '@opentelemetry/exporter-trace-otlp-grpc';
+import { OTLPTraceExporter as OTLPHttpTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';
+import { OTLPTraceExporter as OTLPProtoTraceExporter } from '@opentelemetry/exporter-trace-otlp-proto';
+import { ZipkinExporter } from '@opentelemetry/exporter-zipkin';
+import { AWSXRayIdGenerator } from '@opentelemetry/id-generator-aws-xray';
 import { Instrumentation } from '@opentelemetry/instrumentation';
 import { awsEc2Detector, awsEcsDetector, awsEksDetector } from '@opentelemetry/resource-detector-aws';
 import {
@@ -65,8 +77,23 @@ const DEFAULT_METRIC_EXPORT_INTERVAL_MILLIS: number = 60000;
 export class AwsOpentelemetryConfigurator {
   private resource: Resource;
   private instrumentations: Instrumentation[];
-
-  constructor(instrumentations: Instrumentation[]) {
+  private idGenerator: IdGenerator;
+  private sampler: Sampler;
+  private spanProcessors: SpanProcessor[];
+  private propagator: TextMapPropagator;
+
+  /**
+   * The constructor will setup the AwsOpentelemetryConfigurator object to be able to provide a
+   * configuration for ADOT JavaScript Auto-Instrumentation.
+   *
+   * The `instrumentations` are the desired Node auto-instrumentations to be used when using ADOT JavaScript.
+   * The auto-Instrumentions are usually populated from OTel's `getNodeAutoInstrumentations()` method from the
+   * `@opentelemetry/auto-instrumentations-node` NPM package, and may have instrumentation patching applied.
+   *
+   * @constructor
+   * @param {Instrumentation[]} instrumentations - Auto-Instrumentations to be added to the ADOT Config
+   */
+  public constructor(instrumentations: Instrumentation[]) {
     /*
      * Set and Detect Resources via Resource Detectors
      *
@@ -117,6 +144,19 @@ export class AwsOpentelemetryConfigurator {
     this.resource = autoResource;
 
     this.instrumentations = instrumentations;
+    this.propagator = getPropagator();
+
+    // TODO: Consider removing AWSXRayIdGenerator as it is not needed
+    // Similarly to Java, always use AWS X-Ray Id Generator
+    // https://github.com/aws-observability/aws-otel-java-instrumentation/blob/a011b8cc29ee32b7f668c04ccfdf64cd30de467c/awsagentprovider/src/main/java/software/amazon/opentelemetry/javaagent/providers/AwsTracerCustomizerProvider.java#L36
+    this.idGenerator = new AWSXRayIdGenerator();
+
+    this.sampler = AwsOpentelemetryConfigurator.customizeSampler(customBuildSamplerFromEnv(this.resource));
+
+    // default SpanProcessors with Span Exporters wrapped inside AwsMetricAttributesSpanExporter
+    const awsSpanProcessorProvider: AwsSpanProcessorProvider = new AwsSpanProcessorProvider(this.resource);
+    this.spanProcessors = awsSpanProcessorProvider.getSpanProcessors();
+    AwsOpentelemetryConfigurator.customizeSpanProcessors(this.spanProcessors, this.resource);
   }
 
   private customizeVersions(autoResource: Resource): Resource {

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

@@ -1,31 +1,32 @@
 // 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';
-
+import { Attributes, SpanKind } from '@opentelemetry/api';
+import { AwsSdkInstrumentationConfig, NormalizedRequest } from '@opentelemetry/instrumentation-aws-sdk';
+import { AWS_ATTRIBUTE_KEYS } from '../../../aws-attribute-keys';
+import { RequestMetadata, ServiceExtension } from '../../../third-party/otel/aws/services/ServiceExtension';
+
+/*
+This file's contents are being contributed to upstream
+- https://github.com/open-telemetry/opentelemetry-js-contrib/pull/2361
+
+This class is a service extension to be used for the AWS JavaScript SDK instrumentation patch for Kinesis.
+The instrumentation patch adds this extension to the upstream's Map of known extension for Kinesis.
+Extensions allow for custom logic for adding service-specific information to spans, such as attributes.
+Specifically, we are adding logic to add the `aws.kinesis.stream.name` attribute, to be used to generate
+RemoteTarget and achieve parity with the Java/Python instrumentation.
+*/
 export class KinesisServiceExtension implements ServiceExtension {
   requestPreSpanHook(request: NormalizedRequest, _config: AwsSdkInstrumentationConfig): RequestMetadata {
-    const streamName = request.commandInput.StreamName;
+    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;
+      spanAttributes[AWS_ATTRIBUTE_KEYS.AWS_KINESIS_STREAM_NAME] = streamName;
     }
 
     const isIncoming = false;
@@ -37,8 +38,4 @@ export class KinesisServiceExtension implements ServiceExtension {
       spanName,
     };
   }
-
-  requestPostSpanHook = (request: NormalizedRequest) => {};
-
-  responseHook = (response: NormalizedResponse, span: Span, tracer: Tracer, config: AwsSdkInstrumentationConfig) => {};
 }

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

@@ -1,31 +1,32 @@
 // 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';
-
+import { Attributes, SpanKind } from '@opentelemetry/api';
+import { AwsSdkInstrumentationConfig, NormalizedRequest } from '@opentelemetry/instrumentation-aws-sdk';
+import { AWS_ATTRIBUTE_KEYS } from '../../../aws-attribute-keys';
+import { RequestMetadata, ServiceExtension } from '../../../third-party/otel/aws/services/ServiceExtension';
+
+/*
+This file's contents are being contributed to upstream
+- https://github.com/open-telemetry/opentelemetry-js-contrib/pull/2361
+
+This class is a service extension to be used for the AWS JavaScript SDK instrumentation patch for S3.
+The instrumentation patch adds this extension to the upstream's Map of known extension for S3.
+Extensions allow for custom logic for adding service-specific information to spans, such as attributes.
+Specifically, we are adding logic to add the `aws.s3.bucket` attribute, to be used to generate
+RemoteTarget and achieve parity with the Java/Python instrumentation.
+*/
 export class S3ServiceExtension implements ServiceExtension {
   requestPreSpanHook(request: NormalizedRequest, _config: AwsSdkInstrumentationConfig): RequestMetadata {
-    const bucketName = request.commandInput.Bucket;
+    const bucketName = request.commandInput?.Bucket;
 
     const spanKind: SpanKind = SpanKind.CLIENT;
     let spanName: string | undefined;
 
     const spanAttributes: Attributes = {};
 
     if (bucketName) {
-      spanAttributes[_AWS_S3_BUCKET] = bucketName;
+      spanAttributes[AWS_ATTRIBUTE_KEYS.AWS_S3_BUCKET] = bucketName;
     }
 
     const isIncoming = false;
@@ -37,8 +38,4 @@ export class S3ServiceExtension implements ServiceExtension {
       spanName,
     };
   }
-
-  requestPostSpanHook = (request: NormalizedRequest) => {};
-
-  responseHook = (response: NormalizedResponse, span: Span, tracer: Tracer, config: AwsSdkInstrumentationConfig) => {};
 }

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

@@ -1,9 +1,15 @@
 // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
 
+import { AttributeValue } from '@opentelemetry/api';
 import { Instrumentation } from '@opentelemetry/instrumentation';
-import { AwsInstrumentation } from '@opentelemetry/instrumentation-aws-sdk';
-import { KinesisServiceExtension, S3ServiceExtension } from './aws/services';
+import { AwsSdkInstrumentationConfig, NormalizedRequest } from '@opentelemetry/instrumentation-aws-sdk';
+import { SEMATTRS_MESSAGING_URL } from '@opentelemetry/semantic-conventions';
+import { AWS_ATTRIBUTE_KEYS } from '../aws-attribute-keys';
+import { SqsUrlParser } from '../sqs-url-parser';
+import { RequestMetadata } from '../third-party/otel/aws/services/ServiceExtension';
+import { KinesisServiceExtension } from './aws/services/kinesis';
+import { S3ServiceExtension } from './aws/services/s3';
 
 export function applyInstrumentationPatches(instrumentations: Instrumentation[]): void {
   /*
@@ -20,10 +26,34 @@ export function applyInstrumentationPatches(instrumentations: Instrumentation[])
       // Access private property servicesExtensions of AwsInstrumentation
       // eslint-disable-next-line @typescript-eslint/ban-ts-comment
       // @ts-ignore
-      const services = (instrumentation as AwsInstrumentation).servicesExtensions?.services;
+      const services: Map<string, ServiceExtension> | undefined = (instrumentation as any).servicesExtensions?.services;
       if (services) {
         services.set('S3', new S3ServiceExtension());
         services.set('Kinesis', new KinesisServiceExtension());
+        const sqsServiceExtension: any = services.get('SQS');
+        // It is not expected that `sqsServiceExtension` is undefined
+        if (sqsServiceExtension) {
+          const requestPreSpanHook = sqsServiceExtension.requestPreSpanHook;
+          // Save original `requestPreSpanHook` under a similar name, to be invoked by the patched hook
+          sqsServiceExtension._requestPreSpanHook = requestPreSpanHook;
+          // The patched hook will populate the 'aws.sqs.queue.url' and 'aws.sqs.queue.name' attributes according to spec
+          // from the 'messaging.url' attribute
+          const patchedRequestPreSpanHook = (
+            request: NormalizedRequest,
+            _config: AwsSdkInstrumentationConfig
+          ): RequestMetadata => {
+            const requestMetadata: RequestMetadata = sqsServiceExtension._requestPreSpanHook(request, _config);
+            // It is not expected that `requestMetadata.spanAttributes` can possibly be undefined, but still be careful anyways
+            if (requestMetadata.spanAttributes) {
+              const queueUrl: AttributeValue | undefined = requestMetadata.spanAttributes[SEMATTRS_MESSAGING_URL];
+              requestMetadata.spanAttributes[AWS_ATTRIBUTE_KEYS.AWS_SQS_QUEUE_URL] = queueUrl;
+              requestMetadata.spanAttributes[AWS_ATTRIBUTE_KEYS.AWS_SQS_QUEUE_NAME] =
+                SqsUrlParser.getQueueName(queueUrl);
+            }
+            return requestMetadata;
+          };
+          sqsServiceExtension.requestPreSpanHook = patchedRequestPreSpanHook;
+        }
       }
     }
   });

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

@@ -45,10 +45,8 @@ setAwsDefaultEnvironmentVariables();
 
 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);
-}
+// Apply instrumentation patches
+applyInstrumentationPatches(instrumentations);
 
 const configurator: AwsOpentelemetryConfigurator = new AwsOpentelemetryConfigurator(instrumentations);
 const configuration: Partial<opentelemetry.NodeSDKConfiguration> = configurator.configure();

aws-distro-opentelemetry-node-autoinstrumentation/src/sqs-url-parser.ts

@@ -1,6 +1,8 @@
 // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
 
+import { AttributeValue } from '@opentelemetry/api';
+
 const HTTP_SCHEMA: string = 'http://';
 const HTTPS_SCHEMA: string = 'https://';
 
@@ -16,8 +18,8 @@ export class SqsUrlParser {
    * /'s (excluding schema), the second part should be a 12-digit account id, and the third part
    * should be a valid queue name, per SQS naming conventions.
    */
-  public static getQueueName(url: string | undefined): string | undefined {
-    if (url === undefined) {
+  public static getQueueName(url: AttributeValue | undefined): string | undefined {
+    if (typeof url !== 'string') {
       return undefined;
     }
     url = url.replace(HTTP_SCHEMA, '').replace(HTTPS_SCHEMA, '');