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

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

* fix:  update base on PR feedback

* fix:  lint issue

* Fix PR's feedback

* Fix lint issue

* Fix typo

---------

Co-authored-by: Marc Pichler <[email protected]>
Co-authored-by: Gerhard Stöbich <[email protected]>
Co-authored-by: Haddas Bronfman <[email protected]>

Author: 4 people

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');
+    });
   });
 });

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

@@ -0,0 +1,96 @@
+/*
+ * 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 * as assert from 'assert';
+
+import { PatchedRequest, _LAYERS_STORE_PROPERTY } from '../src/internal-types';
+import {
+  addNewStackLayer,
+  generateRoute,
+  replaceCurrentStackRoute,
+} from '../src/utils';
+
+describe('utils', () => {
+  describe('addNewStackLayer', () => {
+    it('should inject new array to symbol property if not exist', () => {
+      const fakeRequest = {} as PatchedRequest;
+
+      addNewStackLayer(fakeRequest);
+
+      assert.strictEqual(fakeRequest[_LAYERS_STORE_PROPERTY].length, 1);
+    });
+
+    it('should append new stack item if private symbol already exists', () => {
+      const stack = ['/first'];
+      const fakeRequest = {
+        [_LAYERS_STORE_PROPERTY]: stack,
+      } as PatchedRequest;
+
+      addNewStackLayer(fakeRequest);
+
+      assert.equal(fakeRequest[_LAYERS_STORE_PROPERTY], stack);
+      assert.strictEqual(fakeRequest[_LAYERS_STORE_PROPERTY].length, 2);
+    });
+
+    it('should return pop method to remove newly add stack', () => {
+      const fakeRequest = {} as PatchedRequest;
+
+      const pop = addNewStackLayer(fakeRequest);
+
+      assert.notStrictEqual(pop, undefined);
+
+      pop();
+
+      assert.strictEqual(fakeRequest[_LAYERS_STORE_PROPERTY].length, 0);
+    });
+
+    it('should prevent pop the same stack item multiple time', () => {
+      const fakeRequest = {} as PatchedRequest;
+
+      addNewStackLayer(fakeRequest); // add first stack item
+      const pop = addNewStackLayer(fakeRequest); // add second stack item
+
+      pop();
+      pop();
+
+      assert.strictEqual(fakeRequest[_LAYERS_STORE_PROPERTY].length, 1);
+    });
+  });
+
+  describe('replaceCurrentStackRoute', () => {
+    it('should replace the last stack item with new value', () => {
+      const fakeRequest = {
+        [_LAYERS_STORE_PROPERTY]: ['/first', '/second'],
+      } as PatchedRequest;
+
+      replaceCurrentStackRoute(fakeRequest, '/new_route');
+
+      assert.strictEqual(fakeRequest[_LAYERS_STORE_PROPERTY].length, 2);
+      assert.strictEqual(fakeRequest[_LAYERS_STORE_PROPERTY][1], '/new_route');
+    });
+  });
+
+  describe('generateRoute', () => {
+    it('should combine the stack and striped any slash between layer', () => {
+      const fakeRequest = {
+        [_LAYERS_STORE_PROPERTY]: ['/first/', '/second', '/third/'],
+      } as PatchedRequest;
+
+      const route = generateRoute(fakeRequest);
+
+      assert.strictEqual(route, '/first/second/third/');
+    });
+  });
+});