From 106d9e8c755f435de1942f77960566dcac3bebbe Mon Sep 17 00:00:00 2001
From: Santiago Gimeno <santiago.gimeno@gmail.com>
Date: Tue, 21 Jan 2025 16:41:07 +0100
Subject: [PATCH] lib: add metrics support for http2

---
 lib/internal/nsolid_diag.js                   |  59 ++++
 .../test-nsolid-http2-client-destroy.mjs      | 311 ++++++++++++++++++
 .../test-nsolid-http2-server-rst-stream.js    |  71 ++++
 ...est-nsolid-http2-server-session-destroy.js |  29 ++
 test/parallel/test-nsolid-http2.js            |  47 +++
 5 files changed, 517 insertions(+)
 create mode 100644 test/parallel/test-nsolid-http2-client-destroy.mjs
 create mode 100644 test/parallel/test-nsolid-http2-server-rst-stream.js
 create mode 100644 test/parallel/test-nsolid-http2-server-session-destroy.js
 create mode 100644 test/parallel/test-nsolid-http2.js

diff --git a/lib/internal/nsolid_diag.js b/lib/internal/nsolid_diag.js
index 77cd316acf..a3e340c62d 100644
--- a/lib/internal/nsolid_diag.js
+++ b/lib/internal/nsolid_diag.js
@@ -23,6 +23,8 @@ const dc = require('diagnostics_channel');
 const {
   kHttpClientAbortCount,
   kHttpClientCount,
+  kHttpServerAbortCount,
+  kHttpServerCount,
   kSpanHttpClient,
   kSpanHttpMethod,
   kSpanHttpReqUrl,
@@ -31,6 +33,9 @@ const {
 
 const undiciFetch = dc.tracingChannel('undici:fetch');
 
+// To lazy load the http2 constants
+let http2Constants;
+
 let tracingEnabled = false;
 
 const fetchSubscribeListener = (message, name) => {};
@@ -119,3 +124,57 @@ dc.subscribe('undici:request:error', ({ request, error }) => {
     }
   }
 });
+
+dc.subscribe('http2.client.stream.created', ({ stream }) => {
+  stream[nsolid_tracer_s] = now();
+});
+
+dc.subscribe('http2.client.stream.error', ({ stream }) => {
+  http2Constants ||= require('internal/http2/core').constants;
+  if (stream.rstCode !== http2Constants.NGHTTP2_CANCEL)
+    nsolid_counts[kHttpClientAbortCount]++;
+});
+
+dc.subscribe('http2.client.stream.finish', ({ stream }) => {
+  nsolid_counts[kHttpClientCount]++;
+  nsolidApi.pushClientBucket(now() - stream[nsolid_tracer_s]);
+  // We set this property to null to also signal that the stream has
+  // successfully finished and avoid double counting.
+  stream[nsolid_tracer_s] = null;
+});
+
+dc.subscribe('http2.client.stream.close', ({ stream, code }) => {
+  http2Constants ||= require('internal/http2/core').constants;
+  if ((code === http2Constants.NGHTTP2_CANCEL ||
+       code === http2Constants.NGHTTP2_NO_ERROR) &&
+      stream[nsolid_tracer_s] !== null) {
+    nsolid_counts[kHttpClientAbortCount]++;
+  }
+});
+
+dc.subscribe('http2.server.stream.start', ({ stream }) => {
+  stream[nsolid_tracer_s] = now();
+});
+
+dc.subscribe('http2.server.stream.error', ({ stream }) => {
+  http2Constants ||= require('internal/http2/core').constants;
+  if (stream.rstCode !== http2Constants.NGHTTP2_CANCEL)
+    nsolid_counts[kHttpServerAbortCount]++;
+});
+
+dc.subscribe('http2.server.stream.finish', ({ stream }) => {
+  nsolid_counts[kHttpServerCount]++;
+  nsolidApi.pushServerBucket(now() - stream[nsolid_tracer_s]);
+  // We set this property to null to also signal that the stream has
+  // successfully finished and avoid double counting.
+  stream[nsolid_tracer_s] = null;
+});
+
+dc.subscribe('http2.server.stream.close', ({ stream, code }) => {
+  http2Constants ||= require('internal/http2/core').constants;
+  if ((code === http2Constants.NGHTTP2_CANCEL ||
+       code === http2Constants.NGHTTP2_NO_ERROR) &&
+      stream[nsolid_tracer_s] !== null) {
+    nsolid_counts[kHttpServerAbortCount]++;
+  }
+});
diff --git a/test/parallel/test-nsolid-http2-client-destroy.mjs b/test/parallel/test-nsolid-http2-client-destroy.mjs
new file mode 100644
index 0000000000..2cb03d2d08
--- /dev/null
+++ b/test/parallel/test-nsolid-http2-client-destroy.mjs
@@ -0,0 +1,311 @@
+// Flags: --expose-internals
+import * as common from '../common/index.mjs';
+if (!common.hasCrypto)
+  common.skip('missing crypto');
+import assert from 'assert';
+import * as h2 from 'http2';
+import util from 'internal/http2/util';
+import { getEventListeners } from 'events';
+import nsolid from 'nsolid';
+
+const kSocket = util.kSocket;
+
+let httpClientAbortCount = 0;
+let httpServerAbortCount = 0;
+
+const tests = [];
+
+tests.push({
+  name: 'Test destroy before client operations',
+  test: (done) => {
+    return new Promise((resolve) => {
+      const server = h2.createServer();
+      server.listen(0, common.mustCall(() => {
+        const client = h2.connect(`http://localhost:${server.address().port}`);
+        const socket = client[kSocket];
+        socket.on('close', common.mustCall(() => {
+          assert(socket.destroyed);
+        }));
+
+        const req = client.request();
+        req.on('error', common.expectsError({
+          code: 'ERR_HTTP2_STREAM_CANCEL',
+          name: 'Error',
+          message: 'The pending stream has been canceled'
+        }));
+
+        client.destroy();
+
+        req.on('response', common.mustNotCall());
+
+        const sessionError = {
+          name: 'Error',
+          code: 'ERR_HTTP2_INVALID_SESSION',
+          message: 'The session has been destroyed'
+        };
+
+        assert.throws(() => client.setNextStreamID(), sessionError);
+        assert.throws(() => client.setLocalWindowSize(), sessionError);
+        assert.throws(() => client.ping(), sessionError);
+        assert.throws(() => client.settings({}), sessionError);
+        assert.throws(() => client.goaway(), sessionError);
+        assert.throws(() => client.request(), sessionError);
+        client.close();  // Should be a non-op at this point
+
+        // Wait for setImmediate call from destroy() to complete
+        // so that state.destroyed is set to true
+        setImmediate(() => {
+          assert.throws(() => client.setNextStreamID(), sessionError);
+          assert.throws(() => client.setLocalWindowSize(), sessionError);
+          assert.throws(() => client.ping(), sessionError);
+          assert.throws(() => client.settings({}), sessionError);
+          assert.throws(() => client.goaway(), sessionError);
+          assert.throws(() => client.request(), sessionError);
+          client.close();  // Should be a non-op at this point
+        });
+
+        req.resume();
+        req.on('end', common.mustNotCall());
+        req.on('close', common.mustCall(() => {
+          server.close();
+          httpClientAbortCount++;
+          resolve();
+        }));
+      }));
+    });
+  }
+});
+
+tests.push({
+  name: 'Test destroy before goaway',
+  test: () => {
+    return new Promise((resolve) => {
+      const server = h2.createServer();
+      server.on('stream', common.mustCall((stream) => {
+        stream.session.destroy();
+      }));
+
+      server.listen(0, common.mustCall(() => {
+        const client = h2.connect(`http://localhost:${server.address().port}`);
+
+        client.on('close', () => {
+          server.close();
+          // Calling destroy in here should not matter
+          client.destroy();
+          httpClientAbortCount++;
+          httpServerAbortCount++;
+          resolve();
+        });
+
+        client.request();
+      }));
+    });
+  }
+});
+
+tests.push({
+  name: 'Test destroy before connect',
+  test: () => {
+    return new Promise((resolve) => {
+      const server = h2.createServer();
+      server.on('stream', common.mustNotCall());
+
+      server.listen(0, common.mustCall(() => {
+        const client = h2.connect(`http://localhost:${server.address().port}`);
+
+        server.on('connection', common.mustCall(() => {
+          server.close();
+          client.close();
+          httpClientAbortCount++;
+          resolve();
+        }));
+
+        const req = client.request();
+        req.destroy();
+      }));
+    });
+  }
+});
+
+tests.push({
+  name: 'Destroy with AbortSignal',
+  test: () => {
+    return new Promise((resolve) => {
+      const server = h2.createServer();
+      const controller = new AbortController();
+
+      server.on('stream', common.mustNotCall());
+      server.listen(0, common.mustCall(() => {
+        const client = h2.connect(`http://localhost:${server.address().port}`);
+        client.on('close', common.mustCall());
+
+        const { signal } = controller;
+        assert.strictEqual(getEventListeners(signal, 'abort').length, 0);
+
+        client.on('error', common.mustCall(() => {
+          // After underlying stream dies, signal listener detached
+          assert.strictEqual(getEventListeners(signal, 'abort').length, 0);
+        }));
+
+        const req = client.request({}, { signal });
+
+        req.on('error', common.mustCall((err) => {
+          assert.strictEqual(err.code, 'ABORT_ERR');
+          assert.strictEqual(err.name, 'AbortError');
+        }));
+        req.on('close', common.mustCall(() => {
+          server.close();
+          httpClientAbortCount++;
+          resolve();
+        }));
+
+        assert.strictEqual(req.aborted, false);
+        assert.strictEqual(req.destroyed, false);
+        // Signal listener attached
+        assert.strictEqual(getEventListeners(signal, 'abort').length, 1);
+
+        controller.abort();
+
+        assert.strictEqual(req.aborted, false);
+        assert.strictEqual(req.destroyed, true);
+      }));
+    });
+  }
+});
+
+tests.push({
+  name: 'Pass an already destroyed signal to abort immediately',
+  test: async () => {
+    return new Promise((resolve) => {
+      const server = h2.createServer();
+      const controller = new AbortController();
+
+      server.on('stream', common.mustNotCall());
+      server.listen(0, common.mustCall(() => {
+        const client = h2.connect(`http://localhost:${server.address().port}`);
+        client.on('close', common.mustCall());
+
+        const { signal } = controller;
+        controller.abort();
+
+        assert.strictEqual(getEventListeners(signal, 'abort').length, 0);
+
+        client.on('error', common.mustCall(() => {
+          // After underlying stream dies, signal listener detached
+          assert.strictEqual(getEventListeners(signal, 'abort').length, 0);
+        }));
+
+        const req = client.request({}, { signal });
+        // Signal already aborted, so no event listener attached.
+        assert.strictEqual(getEventListeners(signal, 'abort').length, 0);
+
+        assert.strictEqual(req.aborted, false);
+        // Destroyed on same tick as request made
+        assert.strictEqual(req.destroyed, true);
+
+        req.on('error', common.mustCall((err) => {
+          assert.strictEqual(err.code, 'ABORT_ERR');
+          assert.strictEqual(err.name, 'AbortError');
+        }));
+        req.on('close', common.mustCall(() => {
+          server.close();
+          httpClientAbortCount++;
+          resolve();
+        }));
+      }));
+    });
+  }
+});
+
+tests.push({
+  name: 'Destroy ClientHttpSession with AbortSignal',
+  test: async () => {
+    async function testH2ConnectAbort(secure) {
+      return new Promise((resolve) => {
+        const server = secure ? h2.createSecureServer() : h2.createServer();
+        const controller = new AbortController();
+        server.on('stream', common.mustNotCall());
+        server.listen(0, common.mustCall(() => {
+          const { signal } = controller;
+          const protocol = secure ? 'https' : 'http';
+          const client = h2.connect(`${protocol}://localhost:${server.address().port}`, {
+            signal,
+          });
+          client.on('close', common.mustCall());
+          assert.strictEqual(getEventListeners(signal, 'abort').length, 1);
+          client.on('error', common.mustCall(common.mustCall((err) => {
+            assert.strictEqual(err.code, 'ABORT_ERR');
+            assert.strictEqual(err.name, 'AbortError');
+          })));
+          const req = client.request({}, {});
+          assert.strictEqual(getEventListeners(signal, 'abort').length, 1);
+          req.on('error', common.mustCall((err) => {
+            assert.strictEqual(err.code, 'ERR_HTTP2_STREAM_CANCEL');
+            assert.strictEqual(err.name, 'Error');
+            assert.strictEqual(req.aborted, false);
+            assert.strictEqual(req.destroyed, true);
+          }));
+          req.on('close', common.mustCall(() => {
+            server.close();
+            resolve();
+          }));
+          assert.strictEqual(req.aborted, false);
+          assert.strictEqual(req.destroyed, false);
+          // Signal listener attached
+          assert.strictEqual(getEventListeners(signal, 'abort').length, 1);
+          controller.abort();
+        }));
+      });
+    }
+    await testH2ConnectAbort(false);
+    httpClientAbortCount++;
+    await testH2ConnectAbort(true);
+    httpClientAbortCount++;
+  }
+});
+
+tests.push({
+  name: 'Destroy ClientHttp2Stream with AbortSignal',
+  test: async () => {
+    return new Promise((resolve) => {
+      const server = h2.createServer();
+      const controller = new AbortController();
+
+      server.on('stream', common.mustCall((stream) => {
+        stream.on('error', common.mustNotCall());
+        stream.on('close', common.mustCall(() => {
+          assert.strictEqual(stream.rstCode, h2.constants.NGHTTP2_CANCEL);
+          server.close();
+          httpClientAbortCount++;
+          httpServerAbortCount++;
+          resolve();
+        }));
+        controller.abort();
+      }));
+      server.listen(0, common.mustCall(() => {
+        const client = h2.connect(`http://localhost:${server.address().port}`);
+        client.on('close', common.mustCall());
+
+        const { signal } = controller;
+        const req = client.request({}, { signal });
+        assert.strictEqual(getEventListeners(signal, 'abort').length, 1);
+        req.on('error', common.mustCall((err) => {
+          assert.strictEqual(err.code, 'ABORT_ERR');
+          assert.strictEqual(err.name, 'AbortError');
+          client.close();
+        }));
+        req.on('close', common.mustCall());
+      }));
+    });
+  },
+});
+
+for (const { name, test } of tests) {
+  console.log(`${name}`);
+  await test();
+}
+
+assert.strictEqual(nsolid.traceStats.httpClientCount, 0);
+assert.strictEqual(nsolid.traceStats.httpClientAbortCount, httpClientAbortCount);
+assert.strictEqual(nsolid.traceStats.httpServerCount, 0);
+assert.strictEqual(nsolid.traceStats.httpServerAbortCount, httpServerAbortCount);
diff --git a/test/parallel/test-nsolid-http2-server-rst-stream.js b/test/parallel/test-nsolid-http2-server-rst-stream.js
new file mode 100644
index 0000000000..da9690af03
--- /dev/null
+++ b/test/parallel/test-nsolid-http2-server-rst-stream.js
@@ -0,0 +1,71 @@
+'use strict';
+
+const common = require('../common');
+if (!common.hasCrypto)
+  common.skip('missing crypto');
+const assert = require('assert');
+const http2 = require('http2');
+const Countdown = require('../common/countdown');
+const nsolid = require('nsolid');
+
+const {
+  NGHTTP2_CANCEL,
+  NGHTTP2_NO_ERROR,
+  NGHTTP2_PROTOCOL_ERROR,
+  NGHTTP2_REFUSED_STREAM,
+  NGHTTP2_INTERNAL_ERROR
+} = http2.constants;
+
+const tests = [
+  [NGHTTP2_NO_ERROR, false],
+  [NGHTTP2_NO_ERROR, false],
+  [NGHTTP2_PROTOCOL_ERROR, true, 'NGHTTP2_PROTOCOL_ERROR'],
+  [NGHTTP2_CANCEL, false],
+  [NGHTTP2_REFUSED_STREAM, true, 'NGHTTP2_REFUSED_STREAM'],
+  [NGHTTP2_INTERNAL_ERROR, true, 'NGHTTP2_INTERNAL_ERROR'],
+];
+
+const server = http2.createServer();
+server.on('stream', (stream, headers) => {
+  const test = tests.find((t) => t[0] === Number(headers.rstcode));
+  if (test[1]) {
+    stream.on('error', common.expectsError({
+      name: 'Error',
+      code: 'ERR_HTTP2_STREAM_ERROR',
+      message: `Stream closed with error code ${test[2]}`
+    }));
+  }
+  stream.close(headers.rstcode | 0);
+  stream.on('close', () => {
+    console.log('stream closed');
+  });
+});
+
+server.listen(0, common.mustCall(() => {
+  const client = http2.connect(`http://localhost:${server.address().port}`);
+
+  const countdown = new Countdown(tests.length, () => {
+    client.close();
+    server.close();
+    assert.strictEqual(nsolid.traceStats.httpClientCount, 0);
+    assert.strictEqual(nsolid.traceStats.httpClientAbortCount, 6);
+    assert.strictEqual(nsolid.traceStats.httpServerCount, 0);
+    assert.strictEqual(nsolid.traceStats.httpServerAbortCount, 6);
+  });
+
+  tests.forEach((test) => {
+    const req = client.request({
+      ':method': 'POST',
+      'rstcode': test[0]
+    });
+    req.on('close', common.mustCall(() => {
+      assert.strictEqual(req.rstCode, test[0]);
+      countdown.dec();
+    }));
+    req.on('aborted', common.mustCall());
+    if (test[1])
+      req.on('error', common.mustCall());
+    else
+      req.on('error', common.mustNotCall());
+  });
+}));
diff --git a/test/parallel/test-nsolid-http2-server-session-destroy.js b/test/parallel/test-nsolid-http2-server-session-destroy.js
new file mode 100644
index 0000000000..147f8ea573
--- /dev/null
+++ b/test/parallel/test-nsolid-http2-server-session-destroy.js
@@ -0,0 +1,29 @@
+'use strict';
+
+const common = require('../common');
+if (!common.hasCrypto)
+  common.skip('missing crypto');
+const assert = require('assert');
+const h2 = require('http2');
+const nsolid = require('nsolid');
+
+const server = h2.createServer();
+server.listen(0, common.localhostIPv4, common.mustCall(() => {
+  const afterConnect = common.mustCall((session) => {
+    session.request({ ':method': 'POST' }).end(common.mustCall(() => {
+      session.destroy();
+      server.close();
+    }));
+  });
+
+  const port = server.address().port;
+  const host = common.localhostIPv4;
+  h2.connect(`http://${host}:${port}`, afterConnect);
+}));
+
+process.on('beforeExit', common.mustCall(() => {
+  assert.strictEqual(nsolid.traceStats.httpClientCount, 0);
+  assert.strictEqual(nsolid.traceStats.httpClientAbortCount, 1);
+  assert.strictEqual(nsolid.traceStats.httpServerCount, 0);
+  assert.strictEqual(nsolid.traceStats.httpServerAbortCount, 1);
+}));
diff --git a/test/parallel/test-nsolid-http2.js b/test/parallel/test-nsolid-http2.js
new file mode 100644
index 0000000000..3a7c75962c
--- /dev/null
+++ b/test/parallel/test-nsolid-http2.js
@@ -0,0 +1,47 @@
+'use strict';
+
+const common = require('../common');
+if (!common.hasCrypto)
+  common.skip('missing crypto');
+const assert = require('assert');
+const http2 = require('http2');
+const nsolid = require('nsolid');
+
+const server = http2.createServer();
+server.on('stream', common.mustCall((stream, headers, flags) => {
+  stream.respond({ 'content-type': 'text/html' });
+  stream.end('test');
+}));
+
+
+server.listen(0, common.mustCall(() => {
+  const port = server.address().port;
+  const client = http2.connect(`http://localhost:${port}`);
+
+  const req = client.request();
+
+  req.on('response', common.mustCall((headers) => {
+    assert.strictEqual(headers[':status'], 200);
+    assert.strictEqual(headers['content-type'], 'text/html');
+  }));
+
+  let data = '';
+
+  req.setEncoding('utf8');
+  req.on('data', common.mustCallAtLeast((d) => data += d));
+  req.on('end', common.mustCall(() => {
+    server.close();
+    client.close();
+    assert.strictEqual(data, 'test');
+    assert.strictEqual(nsolid.traceStats.httpClientCount, 1);
+    assert.strictEqual(nsolid.traceStats.httpClientAbortCount, 0);
+    assert.strictEqual(nsolid.traceStats.httpServerCount, 1);
+    assert.strictEqual(nsolid.traceStats.httpServerAbortCount, 0);
+    // Wait for more than 3secs for the percentiles to be updated
+    setTimeout(() => {
+      const metrics = nsolid.metrics();
+      assert.ok(metrics.httpClientMedian > 0);
+      assert.ok(metrics.httpClient99Ptile > 0);
+    }, 3500);
+  }));
+}));