lib: create diagnostics_channel module

benchmark/diagnostics_channel/http.js

@@ -0,0 +1,96 @@
+'use strict';
+const common = require('../common.js');
+const dc = require('diagnostics_channel');
+const { AsyncLocalStorage } = require('async_hooks');
+const http = require('http');
+
+const bench = common.createBenchmark(main, {
+  apm: ['none', 'diagnostics_channel', 'patch'],
+  type: 'buffer',
+  len: 1024,
+  chunks: 4,
+  connections: [50, 500],
+  chunkedEnc: 1,
+  duration: 5
+});
+
+function main({ apm, connections, duration, type, len, chunks, chunkedEnc }) {
+  const done = { none, patch, diagnostics_channel }[apm]();
+
+  const server = require('../fixtures/simple-http-server.js')
+    .listen(common.PORT)
+    .on('listening', () => {
+      const path = `/${type}/${len}/${chunks}/normal/${chunkedEnc}`;
+      bench.http({
+        path,
+        connections,
+        duration
+      }, () => {
+        server.close();
+        if (done) done();
+      });
+    });
+}
+
+function none() {}
+
+function patch() {
+  const als = new AsyncLocalStorage();
+  const times = [];
+
+  const { emit } = http.Server.prototype;
+  function wrappedEmit(...args) {
+    const [name, req, res] = args;
+    if (name === 'request') {
+      als.enterWith({
+        url: req.url,
+        start: process.hrtime.bigint()
+      });
+
+      res.on('finish', () => {
+        times.push({
+          ...als.getStore(),
+          statusCode: res.statusCode,
+          end: process.hrtime.bigint()
+        });
+      });
+    }
+    return emit.apply(this, args);
+  }
+  http.Server.prototype.emit = wrappedEmit;
+
+  return () => {
+    http.Server.prototype.emit = emit;
+  };
+}
+
+function diagnostics_channel() {
+  const als = new AsyncLocalStorage();
+  const times = [];
+
+  const start = dc.channel('http.server.request.start');
+  const finish = dc.channel('http.server.response.finish');
+
+  function onStart(req) {
+    als.enterWith({
+      url: req.url,
+      start: process.hrtime.bigint()
+    });
+  }
+
+  function onFinish(res) {
+    times.push({
+      ...als.getStore(),
+      statusCode: res.statusCode,
+      end: process.hrtime.bigint()
+    });
+  }
+
+  start.subscribe(onStart);
+  finish.subscribe(onFinish);
+
+  return () => {
+    start.unsubscribe(onStart);
+    finish.unsubscribe(onFinish);
+  };
+}

benchmark/diagnostics_channel/publish.js

@@ -0,0 +1,29 @@
+'use strict';
+const common = require('../common.js');
+const dc = require('diagnostics_channel');
+
+const bench = common.createBenchmark(main, {
+  n: [1e8],
+  subscribers: [0, 1, 10],
+});
+
+function noop() {}
+
+function main({ n, subscribers }) {
+  const channel = dc.channel('test');
+  for (let i = 0; i < subscribers; i++) {
+    channel.subscribe(noop);
+  }
+
+  const data = {
+    foo: 'bar'
+  };
+
+  bench.start();
+  for (let i = 0; i < n; i++) {
+    if (channel.hasSubscribers) {
+      channel.publish(data);
+    }
+  }
+  bench.end(n);
+}

benchmark/diagnostics_channel/subscribe.js

@@ -0,0 +1,19 @@
+'use strict';
+const common = require('../common.js');
+const dc = require('diagnostics_channel');
+
+const bench = common.createBenchmark(main, {
+  n: [1e8],
+});
+
+function noop() {}
+
+function main({ n }) {
+  const channel = dc.channel('channel.0');
+
+  bench.start();
+  for (let i = 0; i < n; i++) {
+    channel.subscribe(noop);
+  }
+  bench.end(n);
+}

doc/api/diagnostics_channel.md

@@ -0,0 +1,180 @@
+# Diagnostics Channel
+
+<!--introduced_in=REPLACEME-->
+
+> Stability: 1 - Experimental
+
+<!-- source_link=lib/diagnostics_channel.js -->
+
+The `diagnostics_channel` module provides an API to create named channels
+to report arbitrary message data for diagnostics purposes.
+
+It can be accessed using:
+
+```js
+const diagnostics_channel = require('diagnostics_channel');
+```
+
+It is intended that a module writer wanting to report diagnostics messages
+will create one or many top-level channels to report messages through.
+Channels may also be acquired at runtime but it is not encouraged
+due to the additional overhead of doing so. Channels may be exported for
+convenience, but as long as the name is known it can be acquired anywhere.
+
+If you intend for your module to produce diagnostics data for others to
+consume it is recommended that you include documentation of what named
+channels are used along with the shape of the message data. Channel names
+should generally include the module name to avoid collisions with data from
+other modules.
+
+## Public API
+
+### Overview
+
+Following is a simple overview of the public API.
+
+```js
+const diagnostics_channel = require('diagnostics_channel');
+
+// Get a reusable channel object
+const channel = diagnostics_channel.channel('my-channel');
+
+// Subscribe to the channel
+channel.subscribe((message, name) => {
+  // Received data
+});
+
+// Check if the channel has an active subscriber
+if (channel.hasSubscribers) {
+  // Publish data to the channel
+  channel.publish({
+    some: 'data'
+  });
+}
+```
+
+#### `diagnostics_channel.hasSubscribers(name)`
+
+* `name` {string|symbol} The channel name
+* Returns: {boolean} If there are active subscribers
+
+Check if there are active subscribers to the named channel. This is helpful if
+the message you want to send might be expensive to prepare.
+
+This API is optional but helpful when trying to publish messages from very
+performance-senstive code.
+
+```js
+const diagnostics_channel = require('diagnostics_channel');
+
+if (diagnostics_channel.hasSubscribers('my-channel')) {
+  // There are subscribers, prepare and publish message
+}
+```
+
+#### `diagnostics_channel.channel(name)`
+
+* `name` {string|symbol} The channel name
+* Returns: {Channel} The named channel object
+
+This is the primary entry-point for anyone wanting to interact with a named
+channel. It produces a channel object which is optimized to reduce overhead at
+publish time as much as possible.
+
+```js
+const diagnostics_channel = require('diagnostics_channel');
+
+const channel = diagnostics_channel.channel('my-channel');
+```
+
+### Class: `Channel`
+
+The class `Channel` represents an individual named channel within the data
+pipeline. It is use to track subscribers and to publish messages when there
+are subscribers present. It exists as a separate object to avoid channel
+lookups at publish time, enabling very fast publish speeds and allowing
+for heavy use while incurring very minimal cost. Channels are created with
+[`diagnostics_channel.channel(name)`][], constructing a channel directly
+with `new Channel(name)` is not supported.
+
+#### `channel.hasSubscribers`
+
+* Returns: {boolean} If there are active subscribers
+
+Check if there are active subscribers to this channel. This is helpful if
+the message you want to send might be expensive to prepare.
+
+This API is optional but helpful when trying to publish messages from very
+performance-senstive code.
+
+```js
+const diagnostics_channel = require('diagnostics_channel');
+
+const channel = diagnostics_channel.channel('my-channel');
+
+if (channel.hasSubscribers) {
+  // There are subscribers, prepare and publish message
+}
+```
+
+#### `channel.publish(message)`
+
+* `message` {any} The message to send to the channel subscribers
+
+Publish a message to any subscribers to the channel. This will trigger
+message handlers synchronously so they will execute within the same context.
+
+```js
+const diagnostics_channel = require('diagnostics_channel');
+
+const channel = diagnostics_channel.channel('my-channel');
+
+channel.publish({
+  some: 'message'
+});
+```
+
+#### `channel.subscribe(onMessage)`
+
+* `onMessage` {Function} The handler to receive channel messages
+  * `message` {any} The message data
+  * `name` {string|symbol} The name of the channel
+
+Register a message handler to subscribe to this channel. This message handler
+will be run synchronously whenever a message is published to the channel. Any
+errors thrown in the message handler will trigger an [`'uncaughtException'`][].
+
+```js
+const diagnostics_channel = require('diagnostics_channel');
+
+const channel = diagnostics_channel.channel('my-channel');
+
+channel.subscribe((message, name) => {
+  // Received data
+});
+```
+
+#### `channel.unsubscribe(onMessage)`
+
+* `onMessage` {Function} The previous subscribed handler to remove
+
+Remove a message handler previously registered to this channel with
+[`channel.subscribe(onMessage)`][].
+
+```js
+const diagnostics_channel = require('diagnostics_channel');
+
+const channel = diagnostics_channel.channel('my-channel');
+
+function onMessage(message, name) {
+  // Received data
+}
+
+channel.subscribe(onMessage);
+
+channel.unsubscribe(onMessage);
+```
+
+[`diagnostics_channel.channel(name)`]: #diagnostics_channel_diagnostics_channel_channel_name
+[`channel.subscribe(onMessage)`]: #diagnostics_channel_channel_subscribe_onmessage
+[`'uncaughtException'`]: process.md#process_event_uncaughtexception

doc/api/index.md

@@ -23,6 +23,7 @@
 * [Crypto](crypto.md)
 * [Debugger](debugger.md)
 * [Deprecated APIs](deprecations.md)
+* [Diagnostics Channel](diagnostics_channel.md)
 * [DNS](dns.md)
 * [Domain](domain.md)
 * [Errors](errors.md)

lib/_http_server.js

@@ -84,6 +84,10 @@ const { observerCounts, constants } = internalBinding('performance');
 const { setTimeout, clearTimeout } = require('timers');
 const { NODE_PERFORMANCE_ENTRY_TYPE_HTTP } = constants;
 
+const dc = require('diagnostics_channel');
+const onRequestStartChannel = dc.channel('http.server.request.start');
+const onResponseFinishChannel = dc.channel('http.server.response.finish');
+
 const kServerResponse = Symbol('ServerResponse');
 const kServerResponseStatistics = Symbol('ServerResponseStatistics');
 
@@ -775,6 +779,15 @@ function clearRequestTimeout(req) {
 }
 
 function resOnFinish(req, res, socket, state, server) {
+  if (onResponseFinishChannel.hasSubscribers) {
+    onResponseFinishChannel.publish({
+      request: req,
+      response: res,
+      socket,
+      server
+    });
+  }
+
   // Usually the first incoming element should be our request.  it may
   // be that in the case abortIncoming() was called that the incoming
   // array will be empty.
@@ -862,6 +875,15 @@ function parserOnIncoming(server, socket, state, req, keepAlive) {
   res.shouldKeepAlive = keepAlive;
   DTRACE_HTTP_SERVER_REQUEST(req, socket);
 
+  if (onRequestStartChannel.hasSubscribers) {
+    onRequestStartChannel.publish({
+      request: req,
+      response: res,
+      socket,
+      server
+    });
+  }
+
   if (socket._httpMessage) {
     // There are already pending outgoing res, append.
     state.outgoing.push(res);