http: report request start and end with diagnostics_channel

PR-URL: #34895
Reviewed-By: Bryan English <bryan@bryanenglish.com>
Reviewed-By: Gerhard Stöbich <deb2001-github@yahoo.de>
Reviewed-By: Vladimir de Turckheim <vlad2t@hotmail.com>
Reviewed-By: Rich Trott <rtrott@gmail.com>
Reviewed-By: Gabriel Schulhof <gabriel.schulhof@intel.com>
Reviewed-By: Michael Dawson <midawson@redhat.com>

Author: Qard

benchmark/diagnostics_channel/http.js

@@ -0,0 +1,96 @@
+'use strict';
+constcommon=require('../common.js');
+constdc=require('diagnostics_channel');
+const{ AsyncLocalStorage }=require('async_hooks');
+consthttp=require('http');
+
+constbench=common.createBenchmark(main,{
+apm: ['none','diagnostics_channel','patch'],
+type: 'buffer',
+len: 1024,
+chunks: 4,
+connections: [50,500],
+chunkedEnc: 1,
+duration: 5
+});
+
+functionmain({ apm, connections, duration, type, len, chunks, chunkedEnc }){
+constdone={ none, patch, diagnostics_channel }[apm]();
+
+constserver=require('../fixtures/simple-http-server.js')
+.listen(common.PORT)
+.on('listening',()=>{
+constpath=`/${type}/${len}/${chunks}/normal/${chunkedEnc}`;
+bench.http({
+ path,
+ connections,
+ duration
+},()=>{
+server.close();
+if(done)done();
+});
+});
+}
+
+functionnone(){}
+
+functionpatch(){
+constals=newAsyncLocalStorage();
+consttimes=[];
+
+const{ emit }=http.Server.prototype;
+functionwrappedEmit(...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()
+});
+});
+}
+returnemit.apply(this,args);
+}
+http.Server.prototype.emit=wrappedEmit;
+
+return()=>{
+http.Server.prototype.emit=emit;
+};
+}
+
+functiondiagnostics_channel(){
+constals=newAsyncLocalStorage();
+consttimes=[];
+
+conststart=dc.channel('http.server.request.start');
+constfinish=dc.channel('http.server.response.finish');
+
+functiononStart(req){
+als.enterWith({
+url: req.url,
+start: process.hrtime.bigint()
+});
+}
+
+functiononFinish(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);
+};
+}

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;
 
+constdc=require('diagnostics_channel');
+constonRequestStartChannel=dc.channel('http.server.request.start');
+constonResponseFinishChannel=dc.channel('http.server.response.finish');
+
 constkServerResponse=Symbol('ServerResponse');
 constkServerResponseStatistics=Symbol('ServerResponseStatistics');
 
@@ -775,6 +779,15 @@ function clearRequestTimeout(req) {
 }
 
 functionresOnFinish(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);

test/parallel/test-diagnostics-channel-http-server-start.js

@@ -0,0 +1,65 @@
+'use strict';
+
+constcommon=require('../common');
+const{ AsyncLocalStorage }=require('async_hooks');
+constdc=require('diagnostics_channel');
+constassert=require('assert');
+consthttp=require('http');
+
+constincomingStartChannel=dc.channel('http.server.request.start');
+constoutgoingFinishChannel=dc.channel('http.server.response.finish');
+
+constals=newAsyncLocalStorage();
+letcontext;
+
+// Bind requests to an AsyncLocalStorage context
+incomingStartChannel.subscribe(common.mustCall((message)=>{
+als.enterWith(message);
+context=message;
+}));
+
+// When the request ends, verify the context has been maintained
+// and that the messages contain the expected data
+outgoingFinishChannel.subscribe(common.mustCall((message)=>{
+constdata={
+ request,
+ response,
+ server,
+socket: request.socket
+};
+
+// Context is maintained
+compare(als.getStore(),context);
+
+compare(context,data);
+compare(message,data);
+}));
+
+letrequest;
+letresponse;
+
+constserver=http.createServer(common.mustCall((req,res)=>{
+request=req;
+response=res;
+
+setTimeout(()=>{
+res.end('done');
+},1);
+}));
+
+server.listen(()=>{
+const{ port }=server.address();
+http.get(`http://localhost:${port}`,(res)=>{
+res.resume();
+res.on('end',()=>{
+server.close();
+});
+});
+});
+
+functioncompare(a,b){
+assert.strictEqual(a.request,b.request);
+assert.strictEqual(a.response,b.response);
+assert.strictEqual(a.socket,b.socket);
+assert.strictEqual(a.server,b.server);
+}