fix(connect): fix wrong rpcMetada.route value not handle nested route

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

@@ -23,7 +23,13 @@ import {
   ConnectNames,
   ConnectTypes,
 } from './enums/AttributeNames';
-import { Use, UseArgs, UseArgs2 } from './internal-types';
+import {
+  _LAYERS_STORE_PROPERTY,
+  PatchedRequest,
+  Use,
+  UseArgs,
+  UseArgs2,
+} from './internal-types';
 import { VERSION } from './version';
 import {
   InstrumentationBase,
@@ -35,6 +41,22 @@ import { SemanticAttributes } from '@opentelemetry/semantic-conventions';
 
 export const ANONYMOUS_NAME = 'anonymous';
 
+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 replaceCurrentStackRoute = (request: PatchedRequest, value?: string) => {
+  if (value) {
+    request[_LAYERS_STORE_PROPERTY].splice(-1, 1, value);
+  }
+};
+
 /** Connect instrumentation for OpenTelemetry */
 export class ConnectInstrumentation extends InstrumentationBase<Server> {
   constructor(config: InstrumentationConfig = {}) {
@@ -65,6 +87,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,13 +145,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 = req[_LAYERS_STORE_PROPERTY].reduce(
+          (acc, sub) => acc.replace(/\/+$/, '') + sub
+        );
       }
       let spanName = '';
       if (routeName) {
@@ -180,4 +212,33 @@ 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];
+      addNewStackLayer(req);
+
+      function completeStack() {
+        req[_LAYERS_STORE_PROPERTY].pop();
+      }
+      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,14 @@
  * 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('__ot_middlewares');
 
 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/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');
+    });
   });
 });