lib: add tracing channels for http2 streams

PR-URL: #266
Reviewed-By: Juan José Arboleda <[email protected]>
Reviewed-By: Rafael Gonzaga <[email protected]>
PR-URL: #268

Author: santigimeno

lib/internal/http2/core.js

@@ -197,6 +197,9 @@ const onServerStreamErrorChannel = dc.channel('http2.server.stream.error');
 const onServerStreamFinishChannel = dc.channel('http2.server.stream.finish');
 const onServerStreamCloseChannel = dc.channel('http2.server.stream.close');
 
+const clientTracingChannels = dc.tracingChannel('http2.client');
+const serverTracingChannels = dc.tracingChannel('http2.server');
+
 let debug = require('internal/util/debuglog').debuglog('http2', (fn) => {
   debug = fn;
 });
@@ -258,6 +261,7 @@ const kSentHeaders = Symbol('sent-headers');
 const kSentTrailers = Symbol('sent-trailers');
 const kServer = Symbol('server');
 const kState = Symbol('state');
+const kTracingContext = Symbol('tracing-context');
 const kType = Symbol('type');
 const kWriteGeneric = Symbol('write-generic');
 
@@ -341,6 +345,31 @@ function emit(self, ...args) {
   ReflectApply(self.emit, self, args);
 }
 
+function createServerStream(session, handle, id, options, headers, endOfStream) {
+  // eslint-disable-next-line no-use-before-define
+  const stream = new ServerHttp2Stream(session, handle, id, options, headers);
+  if (endOfStream) {
+    stream.push(null);
+    stream[kState].endAfterHeaders = true;
+  }
+
+  if (headers[HTTP2_HEADER_METHOD] === HTTP2_METHOD_HEAD) {
+    // For head requests, there must not be a body...
+    // end the writable side immediately.
+    stream.end();
+    stream[kState].flags |= STREAM_FLAGS_HEAD_REQUEST;
+  }
+
+  if (onServerStreamStartChannel.hasSubscribers) {
+    onServerStreamStartChannel.publish({
+      stream,
+      headers,
+    });
+  }
+
+  return stream;
+}
+
 // Called when a new block of headers has been received for a given
 // stream. The stream may or may not be new. If the stream is new,
 // create the associated Http2Stream instance and emit the 'stream'
@@ -373,23 +402,24 @@ function onSessionHeaders(handle, id, cat, flags, headers, sensitiveHeaders) {
     }
     // session[kType] can be only one of two possible values
     if (type === NGHTTP2_SESSION_SERVER) {
-      // eslint-disable-next-line no-use-before-define
-      stream = new ServerHttp2Stream(session, handle, id, {}, obj);
-      if (endOfStream) {
-        stream.push(null);
-      }
-      if (obj[HTTP2_HEADER_METHOD] === HTTP2_METHOD_HEAD) {
-        // For head requests, there must not be a body...
-        // end the writable side immediately.
-        stream.end();
-        stream[kState].flags |= STREAM_FLAGS_HEAD_REQUEST;
-      }
-
-      if (onServerStreamStartChannel.hasSubscribers) {
-        onServerStreamStartChannel.publish({
-          stream,
-          headers: obj,
+      if (serverTracingChannels.hasSubscribers) {
+        const context = { headers: obj, flags };
+        serverTracingChannels.start.runStores(context, () => {
+          try {
+            stream = createServerStream(session, handle, id, {}, obj);
+            stream[kTracingContext] = context;
+            process.nextTick(emit, session, 'stream', stream, obj, flags, headers);
+          } catch (e) {
+            context.error = e;
+            serverTracingChannels.error.publish(context);
+            throw e;
+          } finally {
+            serverTracingChannels.end.publish(context);
+          }
         });
+      } else {
+        stream = createServerStream(session, handle, id, {}, obj, endOfStream);
+        process.nextTick(emit, session, 'stream', stream, obj, flags, headers);
       }
     } else {
       // eslint-disable-next-line no-use-before-define
@@ -405,10 +435,10 @@ function onSessionHeaders(handle, id, cat, flags, headers, sensitiveHeaders) {
         stream.push(null);
       }
       stream.end();
+      if (endOfStream)
+        stream[kState].endAfterHeaders = true;
+      process.nextTick(emit, session, 'stream', stream, obj, flags, headers);
     }
-    if (endOfStream)
-      stream[kState].endAfterHeaders = true;
-    process.nextTick(emit, session, 'stream', stream, obj, flags, headers);
   } else {
     let event;
     const status = obj[HTTP2_HEADER_STATUS];
@@ -444,9 +474,10 @@ function onSessionHeaders(handle, id, cat, flags, headers, sensitiveHeaders) {
         });
       }
     }
-  }
-  if (endOfStream) {
-    stream.push(null);
+
+    if (endOfStream) {
+      stream.push(null);
+    }
   }
 }
 
@@ -1786,7 +1817,31 @@ class ClientHttp2Session extends Http2Session {
   // associated Http2Stream instance.
   request(headers, options) {
     debugSessionObj(this, 'initiating request');
+    headers = ObjectAssign({ __proto__: null }, headers);
+    options = { ...options };
+    if (!clientTracingChannels.hasSubscribers) {
+      this._requestPrepare(headers, options);
+      return this._requestFinalize(headers, options);
+    }
+
+    this._requestPrepare(headers, options);
+    const context = { headers, options };
+    return clientTracingChannels.start.runStores(context, () => {
+      try {
+        const stream = this._requestFinalize(context.headers, context.options);
+        stream[kTracingContext] = context;
+        return stream;
+      } catch (e) {
+        context.error = e;
+        clientTracingChannels.error.publish(context);
+        throw e;
+      } finally {
+        clientTracingChannels.end.publish(context);
+      }
+    });
+  }
 
+  _requestPrepare(headers, options) {
     if (this.destroyed)
       throw new ERR_HTTP2_INVALID_SESSION();
 
@@ -1809,9 +1864,6 @@ class ClientHttp2Session extends Http2Session {
     assertIsObject(headers, 'headers');
     assertIsObject(options, 'options');
 
-    headers = ObjectAssign({ __proto__: null }, headers);
-    options = { ...options };
-
     if (headers[HTTP2_HEADER_METHOD] === undefined)
       headers[HTTP2_HEADER_METHOD] = HTTP2_METHOD_GET;
 
@@ -1843,7 +1895,9 @@ class ClientHttp2Session extends Http2Session {
     } else {
       validateBoolean(options.endStream, 'options.endStream');
     }
+  }
 
+  _requestFinalize(headers, options) {
     const headersList = mapToHeaders(headers);
 
     // eslint-disable-next-line no-use-before-define
@@ -1859,6 +1913,13 @@ class ClientHttp2Session extends Http2Session {
     if (options.waitForTrailers)
       stream[kState].flags |= STREAM_FLAGS_HAS_TRAILERS;
 
+    if (onClientStreamCreatedChannel.hasSubscribers) {
+      onClientStreamCreatedChannel.publish({
+        stream,
+        headers,
+      });
+    }
+
     const { signal } = options;
     if (signal) {
       validateAbortSignal(signal, 'options.signal');
@@ -1888,17 +1949,11 @@ class ClientHttp2Session extends Http2Session {
       onConnect();
     }
 
-    if (onClientStreamCreatedChannel.hasSubscribers) {
-      onClientStreamCreatedChannel.publish({
-        stream,
-        headers,
-      });
-    }
-
     return stream;
   }
 }
 
+
 function trackWriteState(stream, bytes) {
   const session = stream[kSession];
   stream[kState].writeQueueSize += bytes;
@@ -2001,18 +2056,31 @@ function closeStream(stream, code, rstStreamStatus = kSubmitRstStream) {
       stream.once('finish', finishFn);
   }
 
+  const context = stream[kTracingContext];
   if (type === NGHTTP2_SESSION_CLIENT) {
     if (onClientStreamCloseChannel.hasSubscribers) {
       onClientStreamCloseChannel.publish({
         stream,
         code,
       });
     }
-  } else if (onServerStreamCloseChannel.hasSubscribers) {
-    onServerStreamCloseChannel.publish({
-      stream,
-      code,
-    });
+
+    if (context) {
+      context.result = code;
+      clientTracingChannels.asyncEnd.publish(context);
+    }
+  } else {
+    if (onServerStreamCloseChannel.hasSubscribers) {
+      onServerStreamCloseChannel.publish({
+        stream,
+        code,
+      });
+    }
+
+    if (context) {
+      context.result = code;
+      serverTracingChannels.asyncEnd.publish(context);
+    }
   }
 }
 
@@ -2408,6 +2476,41 @@ class Http2Stream extends Duplex {
     }
     const hasHandle = handle !== undefined;
 
+    // RST code 8 not emitted as an error as its used by clients to signify
+    // abort and is already covered by aborted event, also allows more
+    // seamless compatibility with http1
+    if (err == null && code !== NGHTTP2_NO_ERROR && code !== NGHTTP2_CANCEL)
+      err = new ERR_HTTP2_STREAM_ERROR(nameForErrorCode[code] || code);
+
+    if (err) {
+      const context = this[kTracingContext];
+      if (session[kType] === NGHTTP2_SESSION_CLIENT) {
+        if (onClientStreamErrorChannel.hasSubscribers) {
+          onClientStreamErrorChannel.publish({
+            stream: this,
+            error: err,
+          });
+        }
+
+        if (context) {
+          context.error = err;
+          clientTracingChannels.error.publish(context);
+        }
+      } else {
+        if (onServerStreamErrorChannel.hasSubscribers) {
+          onServerStreamErrorChannel.publish({
+            stream: this,
+            error: err,
+          });
+        }
+
+        if (context) {
+          context.error = err;
+          serverTracingChannels.error.publish(context);
+        }
+      }
+    }
+
     if (!this.closed)
       closeStream(this, code, hasHandle ? kForceRstStream : kNoRstStream);
     this.push(null);
@@ -2423,12 +2526,6 @@ class Http2Stream extends Duplex {
     sessionState.writeQueueSize -= state.writeQueueSize;
     state.writeQueueSize = 0;
 
-    // RST code 8 not emitted as an error as its used by clients to signify
-    // abort and is already covered by aborted event, also allows more
-    // seamless compatibility with http1
-    if (err == null && code !== NGHTTP2_NO_ERROR && code !== NGHTTP2_CANCEL)
-      err = new ERR_HTTP2_STREAM_ERROR(nameForErrorCode[code] || code);
-
     this[kSession] = undefined;
     this[kHandle] = undefined;
 
@@ -2440,21 +2537,6 @@ class Http2Stream extends Duplex {
     setImmediate(() => {
       session[kMaybeDestroy]();
     });
-    if (err) {
-      if (session[kType] === NGHTTP2_SESSION_CLIENT) {
-        if (onClientStreamErrorChannel.hasSubscribers) {
-          onClientStreamErrorChannel.publish({
-            stream: this,
-            error: err,
-          });
-        }
-      } else if (onServerStreamErrorChannel.hasSubscribers) {
-        onServerStreamErrorChannel.publish({
-          stream: this,
-          error: err,
-        });
-      }
-    }
     callback(err);
   }
   // The Http2Stream can be destroyed if it has closed and if the readable