Merge branch 'main' into fix-1618-express-rpcMetadata

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

@@ -194,20 +194,16 @@ export class DynamodbServiceExtension implements ServiceExtension {
   responseHook(
     response: NormalizedResponse,
     span: Span,
-    tracer: Tracer,
-    config: AwsSdkInstrumentationConfig
+    _tracer: Tracer,
+    _config: AwsSdkInstrumentationConfig
   ) {
-    const operation = response.request.commandName;
-
-    if (operation === 'BatchGetItem') {
-      if (Array.isArray(response.data?.ConsumedCapacity)) {
-        span.setAttribute(
-          SemanticAttributes.AWS_DYNAMODB_CONSUMED_CAPACITY,
-          response.data.ConsumedCapacity.map(
-            (x: { [DictionaryKey: string]: any }) => JSON.stringify(x)
-          )
-        );
-      }
+    if (response.data?.ConsumedCapacity) {
+      span.setAttribute(
+        SemanticAttributes.AWS_DYNAMODB_CONSUMED_CAPACITY,
+        toArray(response.data.ConsumedCapacity).map(
+          (x: { [DictionaryKey: string]: any }) => JSON.stringify(x)
+        )
+      );
     }
 
     if (response.data?.ItemCollectionMetrics) {
@@ -241,3 +237,7 @@ export class DynamodbServiceExtension implements ServiceExtension {
     }
   }
 }
+
+function toArray<T>(values: T | T[]): T[] {
+  return Array.isArray(values) ? values : [values];
+}

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

@@ -633,4 +633,78 @@ describe('DynamoDB', () => {
       );
     });
   });
+
+  describe('ConsumedCapacity', () => {
+    it('should populate ConsumedCapacity attributes when they exist', done => {
+      mockV2AwsSend(responseMockSuccess, {
+        ConsumedCapacity: {
+          TableName: 'test-table',
+          CapacityUnits: 0.5,
+          Table: { CapacityUnits: 0.5 },
+        },
+      } as AWS.DynamoDB.Types.PutItemOutput);
+
+      const dynamodb = new AWS.DynamoDB.DocumentClient();
+      dynamodb.put(
+        {
+          TableName: 'test-table',
+          Item: { key1: 'val1' },
+          ReturnConsumedCapacity: 'INDEXES',
+        },
+        (err: AWSError, data: AWS.DynamoDB.DocumentClient.PutItemOutput) => {
+          const spans = getTestSpans();
+          expect(spans.length).toStrictEqual(1);
+          const attrs = spans[0].attributes;
+          expect(attrs[SemanticAttributes.DB_SYSTEM]).toStrictEqual(
+            DbSystemValues.DYNAMODB
+          );
+          expect(attrs[SemanticAttributes.DB_OPERATION]).toStrictEqual(
+            'PutItem'
+          );
+          expect(
+            attrs[SemanticAttributes.AWS_DYNAMODB_CONSUMED_CAPACITY]
+          ).toStrictEqual([
+            JSON.stringify({
+              TableName: 'test-table',
+              CapacityUnits: 0.5,
+              Table: { CapacityUnits: 0.5 },
+            }),
+          ]);
+          expect(err).toBeFalsy();
+          done();
+        }
+      );
+    });
+
+    it('should not populate ConsumedCapacity attributes when it is not returned', done => {
+      mockV2AwsSend(responseMockSuccess, {
+        ConsumedCapacity: undefined,
+      } as AWS.DynamoDB.Types.PutItemOutput);
+
+      const dynamodb = new AWS.DynamoDB.DocumentClient();
+      dynamodb.put(
+        {
+          TableName: 'test-table',
+          Item: { key1: 'val1' },
+          ReturnConsumedCapacity: 'NONE',
+        },
+        (err: AWSError, data: AWS.DynamoDB.DocumentClient.PutItemOutput) => {
+          const spans = getTestSpans();
+          expect(spans.length).toStrictEqual(1);
+          const attrs = spans[0].attributes;
+          expect(attrs[SemanticAttributes.DB_SYSTEM]).toStrictEqual(
+            DbSystemValues.DYNAMODB
+          );
+          expect(attrs[SemanticAttributes.DB_OPERATION]).toStrictEqual(
+            'PutItem'
+          );
+          expect(attrs).not.toHaveProperty(
+            SemanticAttributes.AWS_DYNAMODB_CONSUMED_CAPACITY
+          );
+          expect(err).toBeFalsy();
+          done();
+        }
+      );
+    });
+  });
 });

plugins/node/opentelemetry-instrumentation-connect/src/instrumentation.ts

@@ -23,7 +23,7 @@ import {
   ConnectNames,
   ConnectTypes,
 } from './enums/AttributeNames';
-import { Use, UseArgs, UseArgs2 } from './internal-types';
+import { PatchedRequest, Use, UseArgs, UseArgs2 } from './internal-types';
 import { VERSION } from './version';
 import {
   InstrumentationBase,
@@ -32,6 +32,11 @@ import {
   isWrapped,
 } from '@opentelemetry/instrumentation';
 import { SemanticAttributes } from '@opentelemetry/semantic-conventions';
+import {
+  replaceCurrentStackRoute,
+  addNewStackLayer,
+  generateRoute,
+} from './utils';
 
 export const ANONYMOUS_NAME = 'anonymous';
 
@@ -65,6 +70,9 @@ export class ConnectInstrumentation extends InstrumentationBase<Server> {
     if (!isWrapped(patchedApp.use)) {
       this._wrap(patchedApp, 'use', this._patchUse.bind(this));
     }
+    if (!isWrapped(patchedApp.handle)) {
+      this._wrap(patchedApp, 'handle', this._patchHandle.bind(this));
+    }
   }
 
   private _patchConstructor(original: () => Server): () => Server {
@@ -120,14 +128,20 @@ export class ConnectInstrumentation extends InstrumentationBase<Server> {
       if (!instrumentation.isEnabled()) {
         return (middleWare as any).apply(this, arguments);
       }
-      const [resArgIdx, nextArgIdx] = isErrorMiddleware ? [2, 3] : [1, 2];
+      const [reqArgIdx, resArgIdx, nextArgIdx] = isErrorMiddleware
+        ? [1, 2, 3]
+        : [0, 1, 2];
+      const req = arguments[reqArgIdx] as PatchedRequest;
       const res = arguments[resArgIdx] as ServerResponse;
       const next = arguments[nextArgIdx] as NextFunction;
 
+      replaceCurrentStackRoute(req, routeName);
+
       const rpcMetadata = getRPCMetadata(context.active());
       if (routeName && rpcMetadata?.type === RPCType.HTTP) {
-        rpcMetadata.route = routeName;
+        rpcMetadata.route = generateRoute(req);
       }
+
       let spanName = '';
       if (routeName) {
         spanName = `request handler - ${routeName}`;
@@ -180,4 +194,30 @@ export class ConnectInstrumentation extends InstrumentationBase<Server> {
       return original.apply(this, args as UseArgs2);
     };
   }
+
+  public _patchHandle(original: Server['handle']): Server['handle'] {
+    const instrumentation = this;
+    return function (this: Server): ReturnType<Server['handle']> {
+      const [reqIdx, outIdx] = [0, 2];
+      const req = arguments[reqIdx] as PatchedRequest;
+      const out = arguments[outIdx];
+      const completeStack = addNewStackLayer(req);
+
+      if (typeof out === 'function') {
+        arguments[outIdx] = instrumentation._patchOut(
+          out as NextFunction,
+          completeStack
+        );
+      }
+
+      return (original as any).apply(this, arguments);
+    };
+  }
+
+  public _patchOut(out: NextFunction, completeStack: () => void): NextFunction {
+    return function nextFunction(this: NextFunction, ...args: any[]): void {
+      completeStack();
+      return Reflect.apply(out, this, args);
+    };
+  }
 }

plugins/node/opentelemetry-instrumentation-connect/src/internal-types.ts

@@ -14,9 +14,16 @@
  * limitations under the License.
  */
 
-import type { HandleFunction, Server } from 'connect';
+import type { HandleFunction, IncomingMessage, Server } from 'connect';
+
+export const _LAYERS_STORE_PROPERTY: unique symbol = Symbol(
+  'opentelemetry.instrumentation-connect.request-route-stack'
+);
 
 export type UseArgs1 = [HandleFunction];
 export type UseArgs2 = [string, HandleFunction];
 export type UseArgs = UseArgs1 | UseArgs2;
 export type Use = (...args: UseArgs) => Server;
+export type PatchedRequest = {
+  [_LAYERS_STORE_PROPERTY]: string[];
+} & IncomingMessage;

plugins/node/opentelemetry-instrumentation-connect/src/utils.ts

@@ -0,0 +1,55 @@
+/*
+ * 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 { diag } from '@opentelemetry/api';
+import { _LAYERS_STORE_PROPERTY, PatchedRequest } from './internal-types';
+
+export const addNewStackLayer = (request: PatchedRequest) => {
+  if (Array.isArray(request[_LAYERS_STORE_PROPERTY]) === false) {
+    Object.defineProperty(request, _LAYERS_STORE_PROPERTY, {
+      enumerable: false,
+      value: [],
+    });
+  }
+  request[_LAYERS_STORE_PROPERTY].push('/');
+
+  const stackLength = request[_LAYERS_STORE_PROPERTY].length;
+
+  return () => {
+    if (stackLength === request[_LAYERS_STORE_PROPERTY].length) {
+      request[_LAYERS_STORE_PROPERTY].pop();
+    } else {
+      diag.warn('Connect: Trying to pop the stack multiple time');
+    }
+  };
+};
+
+export const replaceCurrentStackRoute = (
+  request: PatchedRequest,
+  newRoute?: string
+) => {
+  if (newRoute) {
+    request[_LAYERS_STORE_PROPERTY].splice(-1, 1, newRoute);
+  }
+};
+
+// generage route from existing stack on request object.
+// splash between stack layer will be dedup
+// ["/first/", "/second", "/third/"] => /first/second/thrid/
+export const generateRoute = (request: PatchedRequest) => {
+  return request[_LAYERS_STORE_PROPERTY].reduce(
+    (acc, sub) => acc.replace(/\/+$/, '') + sub
+  );
+};

plugins/node/opentelemetry-instrumentation-connect/test/instrumentation.test.ts

@@ -243,5 +243,88 @@ describe('connect', () => {
         changedRootSpan.spanContext().spanId
       );
     });
+
+    it('should append nested route in RpcMetadata', async () => {
+      const rootSpan = tracer.startSpan('root span');
+      const rpcMetadata: RPCMetadata = { type: RPCType.HTTP, span: rootSpan };
+      app.use((req, res, next) => {
+        return context.with(
+          setRPCMetadata(
+            trace.setSpan(context.active(), rootSpan),
+            rpcMetadata
+          ),
+          next
+        );
+      });
+
+      const nestedApp = connect();
+
+      app.use('/foo/', nestedApp);
+      nestedApp.use('/bar/', (req, res, next) => {
+        next();
+      });
+
+      await httpRequest.get(`http://localhost:${PORT}/foo/bar`);
+      rootSpan.end();
+
+      assert.strictEqual(rpcMetadata.route, '/foo/bar/');
+    });
+
+    it('should use latest match route when multiple route is match', async () => {
+      const rootSpan = tracer.startSpan('root span');
+      const rpcMetadata: RPCMetadata = { type: RPCType.HTTP, span: rootSpan };
+      app.use((req, res, next) => {
+        return context.with(
+          setRPCMetadata(
+            trace.setSpan(context.active(), rootSpan),
+            rpcMetadata
+          ),
+          next
+        );
+      });
+
+      app.use('/foo', (req, res, next) => {
+        next();
+      });
+
+      app.use('/foo/bar', (req, res, next) => {
+        next();
+      });
+
+      await httpRequest.get(`http://localhost:${PORT}/foo/bar`);
+      rootSpan.end();
+
+      assert.strictEqual(rpcMetadata.route, '/foo/bar');
+    });
+
+    it('should use latest match route when multiple route is match (with nested app)', async () => {
+      const rootSpan = tracer.startSpan('root span');
+      const rpcMetadata: RPCMetadata = { type: RPCType.HTTP, span: rootSpan };
+      app.use((req, res, next) => {
+        return context.with(
+          setRPCMetadata(
+            trace.setSpan(context.active(), rootSpan),
+            rpcMetadata
+          ),
+          next
+        );
+      });
+
+      const nestedApp = connect();
+
+      app.use('/foo/', nestedApp);
+      nestedApp.use('/bar/', (req, res, next) => {
+        next();
+      });
+
+      app.use('/foo/bar/test', (req, res, next) => {
+        next();
+      });
+
+      await httpRequest.get(`http://localhost:${PORT}/foo/bar/test`);
+      rootSpan.end();
+
+      assert.strictEqual(rpcMetadata.route, '/foo/bar/test');
+    });
   });
 });