feat(core): add specific context key to store rpc metadata (#2208)

Author: vmarchaud

packages/opentelemetry-core/src/index.ts

@@ -27,6 +27,7 @@ export * from './platform';
 export * from './propagation/composite';
 export * from './trace/HttpTraceContextPropagator';
 export * from './trace/IdGenerator';
+export * from './trace/rpc-metadata';
 export * from './trace/sampler/AlwaysOffSampler';
 export * from './trace/sampler/AlwaysOnSampler';
 export * from './trace/sampler/ParentBasedSampler';

packages/opentelemetry-core/src/trace/rpc-metadata.ts

@@ -0,0 +1,48 @@
+/*
+ * 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 { Context, createContextKey, Span } from '@opentelemetry/api';
+
+const RPC_METADATA_KEY = createContextKey(
+  'OpenTelemetry SDK Context Key RPC_METADATA'
+);
+
+export enum RPCType {
+  HTTP = 'http',
+}
+
+type HTTPMetadata = {
+  type: RPCType.HTTP;
+  route?: string;
+  span: Span;
+};
+
+/**
+ * Allows for future rpc metadata to be used with this mechanism
+ */
+export type RPCMetadata = HTTPMetadata;
+
+export function setRPCMetadata(context: Context, meta: RPCMetadata): Context {
+  return context.setValue(RPC_METADATA_KEY, meta);
+}
+
+export function deleteRPCMetadata(context: Context): Context {
+  return context.deleteValue(RPC_METADATA_KEY);
+}
+
+export function getRPCMetadata(context: Context): RPCMetadata | undefined {
+  return context.getValue(RPC_METADATA_KEY) as RPCMetadata | undefined;
+}

packages/opentelemetry-instrumentation-http/src/http.ts

@@ -50,6 +50,7 @@ import {
   isWrapped,
   safeExecuteInTheMiddle,
 } from '@opentelemetry/instrumentation';
+import { RPCMetadata, RPCType, setRPCMetadata } from '@opentelemetry/core';
 
 /**
  * Http instrumentation instrumentation for Opentelemetry
@@ -409,29 +410,72 @@ export class HttpInstrumentation extends InstrumentationBase<Http> {
         spanOptions,
         ctx
       );
+      const rpcMetadata: RPCMetadata = {
+        type: RPCType.HTTP,
+        span,
+      };
 
-      return context.with(trace.setSpan(ctx, span), () => {
-        context.bind(request);
-        context.bind(response);
+      return context.with(
+        setRPCMetadata(trace.setSpan(ctx, span), rpcMetadata),
+        () => {
+          context.bind(request);
+          context.bind(response);
 
-        if (instrumentation._getConfig().requestHook) {
-          instrumentation._callRequestHook(span, request);
-        }
-        if (instrumentation._getConfig().responseHook) {
-          instrumentation._callResponseHook(span, response);
-        }
+          if (instrumentation._getConfig().requestHook) {
+            instrumentation._callRequestHook(span, request);
+          }
+          if (instrumentation._getConfig().responseHook) {
+            instrumentation._callResponseHook(span, response);
+          }
 
-        // Wraps end (inspired by:
-        // https://github.com/GoogleCloudPlatform/cloud-trace-nodejs/blob/master/src/instrumentations/instrumentation-connect.ts#L75)
-        const originalEnd = response.end;
-        response.end = function (
-          this: http.ServerResponse,
-          ..._args: ResponseEndArgs
-        ) {
-          response.end = originalEnd;
-          // Cannot pass args of type ResponseEndArgs,
-          const returned = safeExecuteInTheMiddle(
-            () => response.end.apply(this, arguments as never),
+          // Wraps end (inspired by:
+          // https://github.com/GoogleCloudPlatform/cloud-trace-nodejs/blob/master/src/instrumentations/instrumentation-connect.ts#L75)
+          const originalEnd = response.end;
+          response.end = function (
+            this: http.ServerResponse,
+            ..._args: ResponseEndArgs
+          ) {
+            response.end = originalEnd;
+            // Cannot pass args of type ResponseEndArgs,
+            const returned = safeExecuteInTheMiddle(
+              () => response.end.apply(this, arguments as never),
+              error => {
+                if (error) {
+                  utils.setSpanWithError(span, error);
+                  instrumentation._closeHttpSpan(span);
+                  throw error;
+                }
+              }
+            );
+
+            const attributes = utils.getIncomingRequestAttributesOnResponse(
+              request,
+              response
+            );
+
+            span
+              .setAttributes(attributes)
+              .setStatus(utils.parseResponseStatus(response.statusCode));
+
+            if (instrumentation._getConfig().applyCustomAttributesOnSpan) {
+              safeExecuteInTheMiddle(
+                () =>
+                  instrumentation._getConfig().applyCustomAttributesOnSpan!(
+                    span,
+                    request,
+                    response
+                  ),
+                () => {},
+                true
+              );
+            }
+
+            instrumentation._closeHttpSpan(span);
+            return returned;
+          };
+
+          return safeExecuteInTheMiddle(
+            () => original.apply(this, [event, ...args]),
             error => {
               if (error) {
                 utils.setSpanWithError(span, error);
@@ -440,44 +484,8 @@ export class HttpInstrumentation extends InstrumentationBase<Http> {
               }
             }
           );
-
-          const attributes = utils.getIncomingRequestAttributesOnResponse(
-            request,
-            response
-          );
-
-          span
-            .setAttributes(attributes)
-            .setStatus(utils.parseResponseStatus(response.statusCode));
-
-          if (instrumentation._getConfig().applyCustomAttributesOnSpan) {
-            safeExecuteInTheMiddle(
-              () =>
-                instrumentation._getConfig().applyCustomAttributesOnSpan!(
-                  span,
-                  request,
-                  response
-                ),
-              () => {},
-              true
-            );
-          }
-
-          instrumentation._closeHttpSpan(span);
-          return returned;
-        };
-
-        return safeExecuteInTheMiddle(
-          () => original.apply(this, [event, ...args]),
-          error => {
-            if (error) {
-              utils.setSpanWithError(span, error);
-              instrumentation._closeHttpSpan(span);
-              throw error;
-            }
-          }
-        );
-      });
+        }
+      );
     };
   }
 

packages/opentelemetry-instrumentation-http/src/utils.ts

@@ -18,6 +18,7 @@ import {
   SpanStatusCode,
   Span,
   SpanStatus,
+  context,
 } from '@opentelemetry/api';
 import {
   NetTransportValues,
@@ -31,7 +32,7 @@ import {
   RequestOptions,
   ServerResponse,
 } from 'http';
-import { Socket } from 'net';
+import { getRPCMetadata, RPCType } from '@opentelemetry/core';
 import * as url from 'url';
 import { AttributeNames } from './enums/AttributeNames';
 import { Err, IgnoreMatcher, ParsedRequestOptions } from './types';
@@ -465,25 +466,15 @@ export const getIncomingRequestAttributes = (
  * @param {(ServerResponse & { socket: Socket; })} response the response object
  */
 export const getIncomingRequestAttributesOnResponse = (
-  request: IncomingMessage & { __ot_middlewares?: string[] },
-  response: ServerResponse & { socket: Socket }
+  request: IncomingMessage,
+  response: ServerResponse
 ): SpanAttributes => {
   // take socket from the request,
   // since it may be detached from the response object in keep-alive mode
   const { socket } = request;
   const { statusCode, statusMessage } = response;
   const { localAddress, localPort, remoteAddress, remotePort } = socket;
-  const { __ot_middlewares } = (request as unknown) as {
-    [key: string]: unknown;
-  };
-  const route = Array.isArray(__ot_middlewares)
-    ? __ot_middlewares
-        .filter(path => path !== '/')
-        .map(path => {
-          return path[0] === '/' ? path : '/' + path;
-        })
-        .join('')
-    : undefined;
+  const rpcMetadata = getRPCMetadata(context.active());
 
   const attributes: SpanAttributes = {
     [SemanticAttributes.NET_HOST_IP]: localAddress,
@@ -494,8 +485,8 @@ export const getIncomingRequestAttributesOnResponse = (
     [AttributeNames.HTTP_STATUS_TEXT]: (statusMessage || '').toUpperCase(),
   };
 
-  if (route !== undefined) {
-    attributes[SemanticAttributes.HTTP_ROUTE] = route;
+  if (rpcMetadata?.type === RPCType.HTTP && rpcMetadata.route !== undefined) {
+    attributes[SemanticAttributes.HTTP_ROUTE] = rpcMetadata.route;
   }
   return attributes;
 };

packages/opentelemetry-instrumentation-http/test/functionals/http-enable.test.ts

@@ -41,6 +41,7 @@ import { ContextManager } from '@opentelemetry/api';
 import { AsyncHooksContextManager } from '@opentelemetry/context-async-hooks';
 import type { ClientRequest, IncomingMessage, ServerResponse } from 'http';
 import { isWrapped } from '@opentelemetry/instrumentation';
+import { getRPCMetadata, RPCType } from '@opentelemetry/core';
 
 const instrumentation = new HttpInstrumentation();
 instrumentation.enable();
@@ -840,5 +841,32 @@ describe('HttpInstrumentation', () => {
         });
       });
     });
+    describe('rpc metadata', () => {
+      beforeEach(() => {
+        memoryExporter.reset();
+        instrumentation.setConfig({ requireParentforOutgoingSpans: true });
+        instrumentation.enable();
+      });
+
+      afterEach(() => {
+        server.close();
+        instrumentation.disable();
+      });
+
+      it('should set rpc metadata for incoming http request', async () => {
+        server = http.createServer((request, response) => {
+          const rpcMemadata = getRPCMetadata(context.active());
+          assert(typeof rpcMemadata !== 'undefined');
+          assert(rpcMemadata.type === RPCType.HTTP);
+          assert(rpcMemadata.span.setAttribute('key', 'value'));
+          response.end('Test Server Response');
+        });
+        await new Promise<void>(resolve => server.listen(serverPort, resolve));
+        await httpRequest.get(`${protocol}://${hostname}:${serverPort}`);
+        const spans = memoryExporter.getFinishedSpans();
+        assert.strictEqual(spans.length, 1);
+        assert.strictEqual(spans[0].attributes.key, 'value');
+      });
+    });
   });
 });

packages/opentelemetry-instrumentation-http/test/functionals/utils.test.ts

@@ -19,6 +19,7 @@ import {
   ROOT_CONTEXT,
   SpanKind,
   TraceFlags,
+  context,
 } from '@opentelemetry/api';
 import { BasicTracerProvider, Span } from '@opentelemetry/tracing';
 import { SemanticAttributes } from '@opentelemetry/semantic-conventions';
@@ -31,6 +32,8 @@ import * as url from 'url';
 import { IgnoreMatcher } from '../../src/types';
 import * as utils from '../../src/utils';
 import { AttributeNames } from '../../src/enums/AttributeNames';
+import { RPCType, setRPCMetadata } from '@opentelemetry/core';
+import { AsyncHooksContextManager } from '@opentelemetry/context-async-hooks';
 
 describe('Utility', () => {
   describe('parseResponseStatus()', () => {
@@ -285,16 +288,30 @@ describe('Utility', () => {
   });
 
   describe('getIncomingRequestAttributesOnResponse()', () => {
-    it('should correctly parse the middleware stack if present', () => {
+    it('should correctly parse the middleware stack if present', done => {
+      context.setGlobalContextManager(new AsyncHooksContextManager().enable());
       const request = {
-        __ot_middlewares: ['/test', '/toto', '/'],
         socket: {},
-      } as IncomingMessage & { __ot_middlewares?: string[] };
-
-      const attributes = utils.getIncomingRequestAttributesOnResponse(request, {
-        socket: {},
-      } as ServerResponse & { socket: Socket });
-      assert.deepEqual(attributes[SemanticAttributes.HTTP_ROUTE], '/test/toto');
+      } as IncomingMessage;
+      context.with(
+        setRPCMetadata(context.active(), {
+          type: RPCType.HTTP,
+          route: '/user/:id',
+          span: (null as unknown) as Span,
+        }),
+        () => {
+          const attributes = utils.getIncomingRequestAttributesOnResponse(
+            request,
+            {} as ServerResponse
+          );
+          assert.deepStrictEqual(
+            attributes[SemanticAttributes.HTTP_ROUTE],
+            '/user/:id'
+          );
+          context.disable();
+          return done();
+        }
+      );
     });
 
     it('should succesfully process without middleware stack', () => {