feat: add instrumentation for aws-sdk S3 client

.ci/tav.json

@@ -2,6 +2,7 @@
   "// todo": "We want versions=['20','19','18','16','14','12','10','8'], but versions*modules needs to be <256 for the GH Actions jobs limit",
   "versions": [ "20", "18", "16", "14", "12", "10", "8" ],
   "modules": [
+    "@aws-sdk/client-s3",
     "@elastic/elasticsearch",
     "@elastic/elasticsearch-canary",
     "@hapi/hapi",

.tav.yml

@@ -75,8 +75,8 @@ redis:
   name: redis
   versions: '>=4.0.0 <5.0.0'
   commands:
-   - node test/instrumentation/modules/redis.test.js
-   - node test/instrumentation/modules/redis4-legacy.test.js
+    - node test/instrumentation/modules/redis.test.js
+    - node test/instrumentation/modules/redis4-legacy.test.js
 
 # We want these version ranges:
 #   # v3.1.3 is broken in older versions of Node because of https://github.com/luin/ioredis/commit/d5867f7c7f03a770a8c0ca5680fdcbfcaf8488e7
@@ -537,6 +537,22 @@ aws-sdk:
     - node test/instrumentation/modules/aws-sdk/sqs.test.js
     - node test/instrumentation/modules/aws-sdk/dynamodb.test.js
 
+'@aws-sdk/client-s3':
+  # We want this version range:
+  #   versions: '>=3 <4'
+  # However, @awk-sdk/client-s3 releases *very* frequently (almost every day) and there
+  # is no need to test *all* those releases. Instead we statically list a subset
+  # of versions to test.
+  #
+  # Maintenance note: This should be updated periodically using:
+  #   ./dev-utils/aws-sdk-s3-client-tav-versions.sh
+  #
+  # Test v3.0.0, every N=41 of 210 releases, and current latest
+  versions: '3.0.0 || 3.36.0 || 3.86.0 || 3.171.0 || 3.245.0 || 3.315.0 || 3.321.1 || >3.321.1 <4'
+  commands:
+    - node test/instrumentation/modules/@aws-sdk/client-s3.test.js
+  node: '>=14'
+
 # - undici@4.7.0 added its diagnostics_channel support.
 # - In undici@4.7.1 the `request.origin` property was added, which we need
 #   in the 'undici:request:create' diagnostic message.

CHANGELOG.asciidoc

@@ -40,6 +40,8 @@ Notes:
 [float]
 ===== Features
 
+* Add support for @aws-sdk/client-s3 ({pull}3287[#3287])
+
 * Add <<capture-body>> support for Fastify instrumentation.
   Contributed by @xxzefgh. ({pull}2681[#2681])
 

dev-utils/aws-sdk-s3-client-tav-versions.sh

@@ -0,0 +1,23 @@
+#!/bin/sh
+#
+# Calculate and emit the "versions:" block of ".tav.yml" for aws-sdk.
+# This will include:
+# - the first supported release (2.858.0)
+# - the latest current release
+# - and ~5 releases in between
+
+npm info -j @aws-sdk/client-s3 | node -e '
+    var semver = require("semver");
+    var chunks = [];
+    process.stdin
+        .resume()
+        .on("data", (chunk) => { chunks.push(chunk) })
+        .on("end", () => {
+            var input = JSON.parse(chunks.join(""));
+            var vers = input.versions.filter(v => semver.satisfies(v, ">=3 <4"));
+            var modulus = Math.floor((vers.length - 2) / 5);
+            console.log("  # Test v3.0.0, every N=%d of %d releases, and current latest.", modulus, vers.length);
+            vers = vers.filter((v, idx, arr) => idx % modulus === 0 || idx === arr.length - 1);
+            console.log("  versions: '\''%s || >%s <4'\''", vers.join(" || "), vers[vers.length-1])
+        })
+'

docs/supported-technologies.asciidoc

@@ -139,6 +139,7 @@ The Node.js agent will automatically instrument the following modules to give yo
 |=======================================================================
 |Module |Version |Note
 |https://www.npmjs.com/package/aws-sdk[aws-sdk] |>1 <3 |Will instrument SQS send/receive/delete messages, all S3 methods, all DynamoDB methods, and the SNS publish method
+|https://www.npmjs.com/package/@aws-sdk/client-s3[@aws-sdk/client-s3] |>=3 <4 |Will instrument all S3 methods
 |https://www.npmjs.com/package/cassandra-driver[cassandra-driver] |>=3.0.0 <5 |Will instrument all queries
 |https://www.npmjs.com/package/elasticsearch[elasticsearch] |>=8.0.0 |Will instrument all queries
 |https://www.npmjs.com/package/@elastic/elasticsearch[@elastic/elasticsearch] |>=7.0.0 <9.0.0 |Will instrument all queries

lib/instrumentation/index.js

@@ -31,6 +31,7 @@ const nodeSupportsAsyncLocalStorage = semver.satisfies(process.versions.node, '>
 const nodeHasInstrumentableFetch = typeof (global.fetch) === 'function'
 
 var MODULES = [
+  '@aws-sdk/smithy-client', // Instrument the base client which all AWS-SDK v3 clients extends
   ['@elastic/elasticsearch', '@elastic/elasticsearch-canary'],
   '@node-redis/client/dist/lib/client',
   '@node-redis/client/dist/lib/client/commands-queue',

lib/instrumentation/modules/@aws-sdk/client-s3.js

@@ -0,0 +1,209 @@
+/*
+ * Copyright Elasticsearch B.V. and other contributors where applicable.
+ * Licensed under the BSD 2-Clause License; you may not use this file except in
+ * compliance with the BSD 2-Clause License.
+ */
+
+'use strict'
+
+const constants = require('../../../constants')
+const NAME = 'S3'
+const TYPE = 'storage'
+const SUBTYPE = 's3'
+const elasticAPMStash = Symbol('elasticAPMStash')
+
+/**
+ * Gets the region from the ARN
+ *
+ * @param {String} s3Arn
+ * @returns {String}
+ */
+function regionFromS3Arn (s3Arn) {
+  return s3Arn.split(':')[3]
+}
+
+/**
+ * Return an APM "resource" string for the bucket, Access Point ARN, or Outpost
+ * ARN. ARNs are normalized to a shorter resource name.
+ * Known ARN patterns:
+ * - arn:aws:s3:<region>:<account-id>:accesspoint/<accesspoint-name>
+ * - arn:aws:s3-outposts:<region>:<account>:outpost/<outpost-id>/bucket/<bucket-name>
+ * - arn:aws:s3-outposts:<region>:<account>:outpost/<outpost-id>/accesspoint/<accesspoint-name>
+ *
+ * In general that is:
+ *    arn:$partition:$service:$region:$accountId:$resource
+ *
+ * This parses using the same "split on colon" used by the JavaScript AWS SDK v3.
+ * https://github.com/aws/aws-sdk-js-v3/blob/v3.18.0/packages/util-arn-parser/src/index.ts#L14-L37
+ *
+ * @param {String} bucket The bucket string
+ * @returns {String | null}
+ */
+function resourceFromBucket (bucket) {
+  let resource = null
+  if (bucket) {
+    resource = bucket
+    if (resource.startsWith('arn:')) {
+      resource = bucket.split(':').slice(5).join(':')
+    }
+  }
+  return resource
+}
+
+/**
+ * Returns middlewares to instrument an S3Client instance
+ *
+ * @param {import('@aws-sdk/client-s3').S3Client} client
+ * @param {any} agent
+ * @returns {import('./smithy-client').AWSMiddlewareEntry[]}
+ */
+function s3MiddlewareFactory (client, agent) {
+  return [
+    {
+      middleware: (next, context) => async (args) => {
+        const input = args.input
+        const bucket = input && input.Bucket
+        const resource = resourceFromBucket(bucket)
+        const span = agent._instrumentation.currSpan()
+
+        if (!span) {
+          return await next(args)
+        }
+        // The given span comes with the operation name and we need to
+        // add the resource if applies
+        if (resource) {
+          span.name += ' ' + resource
+          span.setServiceTarget('s3', resource)
+        }
+
+        // As for now OTel spec defines attributes for operations that require a Bucket
+        // if that changes we should review this guard
+        // https://github.com/open-telemetry/opentelemetry-specification/blob/v1.20.0/semantic_conventions/trace/instrumentation/aws-sdk.yml#L435
+        if (bucket) {
+          const otelAttrs = span._getOTelAttributes()
+
+          otelAttrs['aws.s3.bucket'] = bucket
+
+          if (input.Key) {
+            otelAttrs['aws.s3.key'] = input.Key
+          }
+        }
+
+        let err
+        let result
+        let response
+        let statusCode
+        try {
+          result = await next(args)
+          response = result && result.response
+          statusCode = response && response.statusCode
+        } catch (ex) {
+          // Save the error for use in `finally` below, but re-throw it to
+          // not impact code flow.
+          err = ex
+
+          // This code path happens with a GetObject conditional request
+          // that returns a 304 Not Modified.
+          statusCode = err && err.$metadata && err.$metadata.httpStatusCode
+          throw ex
+        } finally {
+          if (statusCode) {
+            span._setOutcomeFromHttpStatusCode(statusCode)
+          } else {
+            span._setOutcomeFromErrorCapture(constants.OUTCOME_FAILURE)
+          }
+          if (err && (!statusCode || statusCode >= 400)) {
+            agent.captureError(err, { skipOutcome: true })
+          }
+
+          // Set the httpContext
+          if (statusCode) {
+            const httpContext = {
+              status_code: statusCode
+            }
+
+            if (response && response.headers && response.headers['content-length']) {
+              const encodedBodySize = Number(response.headers['content-length'])
+              if (!isNaN(encodedBodySize)) {
+                httpContext.response = { encoded_body_size: encodedBodySize }
+              }
+            }
+            span.setHttpContext(httpContext)
+          }
+
+          // Configuring `new S3Client({useArnRegion:true})` allows one to
+          // use an Access Point bucket ARN for a region *other* than the
+          // one for which the client is configured. Therefore, we attempt
+          // to get the bucket region from the ARN first.
+          const config = client.config
+          let useArnRegion
+          if (typeof config.useArnRegion === 'boolean') {
+            useArnRegion = config.useArnRegion
+          } else {
+            useArnRegion = await config.useArnRegion()
+          }
+
+          let region
+          if (useArnRegion && bucket && bucket.startsWith('arn:')) {
+            region = regionFromS3Arn(args.input.Bucket)
+          } else {
+            region = typeof config.region === 'boolean' ? region : await config.region()
+          }
+
+          // Destination context.
+          const destContext = {
+            address: context[elasticAPMStash].hostname,
+            port: context[elasticAPMStash].port,
+            service: {
+              name: SUBTYPE,
+              type: TYPE
+            }
+          }
+          if (resource) {
+            destContext.service.resource = resource
+          }
+
+          if (region) {
+            destContext.cloud = { region }
+          }
+          span._setDestinationContext(destContext)
+
+          span.end()
+        }
+
+        return result
+      },
+      options: { step: 'initialize', priority: 'high', name: 'elasticAPMSpan' }
+    },
+    {
+      middleware: (next, context) => async (args) => {
+        const req = args.request
+        let port = req.port
+
+        // Resolve port for HTTP(S) protocols
+        if (port === undefined) {
+          if (req.protocol === 'https:') {
+            port = 443
+          } else if (req.protocol === 'http:') {
+            port = 80
+          }
+        }
+
+        context[elasticAPMStash] = {
+          protocol: req.protocol,
+          hostname: req.hostname,
+          port: port
+        }
+        return next(args)
+      },
+      options: { step: 'finalizeRequest', name: 'elasticAPMHTTPInfo' }
+    }
+  ]
+}
+
+module.exports = {
+  S3_NAME: NAME,
+  S3_TYPE: TYPE,
+  S3_SUBTYPE: SUBTYPE,
+  s3MiddlewareFactory
+}