http2: allow client request header injection via diagnostics_channel

doc/api/diagnostics_channel.md

@@ -1699,6 +1699,23 @@ Emitted when server sends a response.
 
 > Stability: 1 - Experimental
 
+##### Event: `'http2.client.request.headers'`
+
+* `headers` {HTTP/2 Headers Object | string\[]}
+* `session` {http2.ClientHttp2Session}
+
+Emitted when `ClientHttp2Session#request()` is called, before the request
+headers are validated and serialized. The `headers` object is the one passed
+to `request()` by the caller (or a newly created object when no headers were
+passed), and it can be mutated in place to add or modify request headers
+before they are sent. This allows subscribers such as tracing instrumentation
+to inject context headers (e.g. `traceparent`) into outgoing HTTP/2 requests.
+Headers injected this way are subject to the same validation as headers
+passed directly to `request()`. To take effect, headers must be mutated in
+place. Unlike `'http2.client.stream.created'`, which is published after
+the request headers have been serialized, mutations made by subscribers to
+this channel affect the headers that are sent.
+
 ##### Event: `'http2.client.stream.created'`
 
 * `stream` {ClientHttp2Stream}

lib/internal/http2/core.js

@@ -197,6 +197,7 @@ const { StreamPipe } = internalBinding('stream_pipe');
 const { _connectionListener: httpConnectionListener } = http;
 
 const dc = require('diagnostics_channel');
+const onClientRequestHeadersChannel = dc.channel('http2.client.request.headers');
 const onClientStreamCreatedChannel = dc.channel('http2.client.stream.created');
 const onClientStreamStartChannel = dc.channel('http2.client.stream.start');
 const onClientStreamErrorChannel = dc.channel('http2.client.stream.error');
@@ -1906,30 +1907,37 @@ class ClientHttp2Session extends Http2Session {
 
     this[kUpdateTimer]();
 
+    let headers = headersParam;
+    if (onClientRequestHeadersChannel.hasSubscribers) {
+      if (headers === undefined)
+        headers = { __proto__: null };
+      onClientRequestHeadersChannel.publish({ headers, session: this });
+    }
+
     let headersList;
     let headersObject;
     let rawHeaders;
     let scheme;
     let authority;
     let method;
 
-    if (ArrayIsArray(headersParam)) {
+    if (ArrayIsArray(headers)) {
       ({
         rawHeaders,
         headersList,
         scheme,
         authority,
         method,
-      } = prepareRequestHeadersArray(headersParam, this));
-    } else if (!!headersParam && typeof headersParam === 'object') {
+      } = prepareRequestHeadersArray(headers, this));
+    } else if (!!headers && typeof headers === 'object') {
       ({
         headersObject,
         headersList,
         scheme,
         authority,
         method,
-      } = prepareRequestHeadersObject(headersParam, this));
-    } else if (headersParam === undefined) {
+      } = prepareRequestHeadersObject(headers, this));
+    } else if (headers === undefined) {
       ({
         headersObject,
         headersList,

test/parallel/test-diagnostics-channel-http2-client-request-headers.js

@@ -0,0 +1,172 @@
+'use strict';
+
+const common = require('../common');
+if (!common.hasCrypto)
+  common.skip('missing crypto');
+
+const assert = require('assert');
+const { once } = require('events');
+const dc = require('diagnostics_channel');
+const http2 = require('http2');
+
+const kPublishCount = 6;
+
+let currentSession;
+let injection;
+
+dc.subscribe('http2.client.request.headers', common.mustCall(({ headers, session }) => {
+  assert.strictEqual(session, currentSession);
+  assert.strictEqual(session.constructor.name, 'ClientHttp2Session');
+  assert.ok(!!headers && typeof headers === 'object');
+  if (injection !== undefined)
+    injection(headers);
+}, kPublishCount));
+
+function doRequest(client, headers) {
+  return new Promise((resolve, reject) => {
+    const stream = client.request(headers);
+    stream.setEncoding('utf8');
+    let body = '';
+    stream.on('data', (chunk) => { body += chunk; });
+    stream.on('end', () => resolve(JSON.parse(body)));
+    stream.on('error', reject);
+  });
+}
+
+async function main() {
+  const server = http2.createServer();
+  server.on('stream', (stream, headers) => {
+    // Echo the received request headers back in the response body.
+    stream.respond({ 'content-type': 'application/json' });
+    stream.end(JSON.stringify(headers));
+  });
+  server.listen(0);
+  await once(server, 'listening');
+  const port = server.address().port;
+
+  // Scenario 1: request() called while the session is still connecting,
+  // with an object headers argument.
+  {
+    const client = http2.connect(`http://localhost:${port}`);
+    currentSession = client;
+    injection = (headers) => {
+      headers.traceparent = '00-0af7651916cd43dd8f8ba5c1d1f77f0f-00f067aa0ba902b7-01';
+    };
+    // Also make sure the snapshot published to 'http2.client.stream.created'
+    // reflects the injected header, since serialization happens after the
+    // 'http2.client.request.headers' publish.
+    const onCreated = common.mustCall(({ headers }) => {
+      assert.strictEqual(
+        headers.traceparent,
+        '00-0af7651916cd43dd8f8ba5c1d1f77f0f-00f067aa0ba902b7-01');
+    }, 1);
+    dc.subscribe('http2.client.stream.created', onCreated);
+    const received = await doRequest(client, { ':path': '/connecting' });
+    dc.unsubscribe('http2.client.stream.created', onCreated);
+    assert.strictEqual(
+      received.traceparent,
+      '00-0af7651916cd43dd8f8ba5c1d1f77f0f-00f067aa0ba902b7-01');
+    assert.strictEqual(received[':path'], '/connecting');
+    client.close();
+    await once(client, 'close');
+  }
+
+  // Scenario 2: request() called on an already-connected session.
+  {
+    const client = http2.connect(`http://localhost:${port}`);
+    await once(client, 'connect');
+    currentSession = client;
+    injection = (headers) => {
+      headers.traceparent = '00-0af7651916cd43dd8f8ba5c1d1f77f0f-b7ad6b7169203331-01';
+    };
+    const received = await doRequest(client, { ':path': '/connected' });
+    assert.strictEqual(
+      received.traceparent,
+      '00-0af7651916cd43dd8f8ba5c1d1f77f0f-b7ad6b7169203331-01');
+    assert.strictEqual(received[':path'], '/connected');
+    client.close();
+    await once(client, 'close');
+  }
+
+  // Scenario 3: no headers argument at all. The subscriber must still be
+  // able to inject headers into the fresh object that is published.
+  {
+    const client = http2.connect(`http://localhost:${port}`);
+    currentSession = client;
+    let sawFreshObject;
+    injection = (headers) => {
+      sawFreshObject = headers[':path'] === undefined;
+      headers.traceparent = '00-0af7651916cd43dd8f8ba5c1d1f77f0f-00f067aa0ba902b7-01';
+    };
+    const received = await doRequest(client);
+    assert.ok(sawFreshObject);
+    assert.strictEqual(
+      received.traceparent,
+      '00-0af7651916cd43dd8f8ba5c1d1f77f0f-00f067aa0ba902b7-01');
+    assert.strictEqual(received[':path'], '/');
+    client.close();
+    await once(client, 'close');
+  }
+
+  // Scenario 4: a subscriber that does not mutate the headers must have no
+  // effect on the request.
+  {
+    const client = http2.connect(`http://localhost:${port}`);
+    await once(client, 'connect');
+    currentSession = client;
+    injection = undefined;
+    const received = await doRequest(client, {
+      ':path': '/no-mutation',
+      'x-test': 'value',
+    });
+    assert.strictEqual(received.traceparent, undefined);
+    assert.deepStrictEqual(received, {
+      ':method': 'GET',
+      ':scheme': 'http',
+      ':authority': `localhost:${port}`,
+      ':path': '/no-mutation',
+      'x-test': 'value',
+    });
+    client.close();
+    await once(client, 'close');
+  }
+
+  // Scenario 5: an array of raw headers can be mutated in place.
+  {
+    const client = http2.connect(`http://localhost:${port}`);
+    currentSession = client;
+    let sawArray;
+    injection = (headers) => {
+      sawArray = Array.isArray(headers);
+      headers.push('traceparent', '00-0af7651916cd43dd8f8ba5c1d1f77f0f-00f067aa0ba902b7-01');
+    };
+    const received = await doRequest(client, [':path', '/array']);
+    assert.ok(sawArray);
+    assert.strictEqual(
+      received.traceparent,
+      '00-0af7651916cd43dd8f8ba5c1d1f77f0f-00f067aa0ba902b7-01');
+    assert.strictEqual(received[':path'], '/array');
+    client.close();
+    await once(client, 'close');
+  }
+
+  // Scenario 6: injected headers go through the same validation as headers
+  // passed directly to request(). Injecting an unknown pseudo-header throws
+  // the same error that passing it directly would.
+  {
+    const client = http2.connect(`http://localhost:${port}`);
+    currentSession = client;
+    injection = (headers) => {
+      headers[':foo'] = 'invalid';
+    };
+    assert.throws(() => client.request({ ':path': '/invalid' }),
+                  { code: 'ERR_HTTP2_INVALID_PSEUDOHEADER' });
+    injection = undefined;
+    client.destroy();
+  }
+
+  server.close();
+  await once(server, 'close');
+}
+
+main().then(common.mustCall());