feat: add diagnostic channels for request lifecycle

lib/application.js

@@ -15,6 +15,8 @@
 
 var finalhandler = require('finalhandler');
 var debug = require('debug')('express:application');
+var diagnostics = require('./diagnostics');
+var onFinished = require('on-finished');
 var View = require('./view');
 var http = require('node:http');
 var methods = require('./utils').methods;
@@ -174,6 +176,24 @@ app.handle = function handle(req, res, callback) {
     res.locals = Object.create(null);
   }
 
+  // publish diagnostic event: request started
+  if (diagnostics.requestStart.hasSubscribers) {
+    diagnostics.requestStart.publish({ req: req, res: res });
+  }
+
+  // publish diagnostic events on response finish
+  if (diagnostics.requestFinish.hasSubscribers || diagnostics.requestError.hasSubscribers) {
+    onFinished(res, function (err) {
+      if (err && diagnostics.requestError.hasSubscribers) {
+        diagnostics.requestError.publish({ req: req, res: res, error: err });
+      }
+
+      if (diagnostics.requestFinish.hasSubscribers) {
+        diagnostics.requestFinish.publish({ req: req, res: res });
+      }
+    });
+  }
+
   this.router.handle(req, res, done);
 };
 

lib/diagnostics.js

@@ -0,0 +1,46 @@
+/*!
+ * express
+ * Copyright(c) 2009-2013 TJ Holowaychuk
+ * Copyright(c) 2014-2015 Douglas Christopher Wilson
+ * MIT Licensed
+ */
+
+'use strict';
+
+/**
+ * Module dependencies.
+ * @private
+ */
+
+var dc = require('node:diagnostics_channel');
+
+/**
+ * Diagnostic channels for Express request lifecycle.
+ *
+ * These channels allow APM tools, monitoring systems, and
+ * instrumentation libraries to observe Express request
+ * processing without monkey-patching.
+ *
+ * @private
+ */
+
+module.exports = {
+  /**
+   * Published when Express begins handling an incoming request,
+   * before routing. Message: { req, res }
+   */
+  requestStart: dc.channel('express.request.start'),
+
+  /**
+   * Published when the response has been fully sent to the client.
+   * Message: { req, res }
+   */
+  requestFinish: dc.channel('express.request.finish'),
+
+  /**
+   * Published when a connection error occurs during request
+   * processing (e.g. ECONNRESET, ECONNABORTED).
+   * Message: { req, res, error }
+   */
+  requestError: dc.channel('express.request.error')
+};

test/diagnostics-channel.js

@@ -0,0 +1,221 @@
+'use strict'
+
+var assert = require('node:assert')
+var dc = require('node:diagnostics_channel')
+var express = require('..')
+var request = require('supertest')
+
+describe('diagnostics channels', function () {
+  describe('express.request.start', function () {
+    it('should publish before routing begins', function (done) {
+      var app = express()
+      var published = false
+
+      app.get('/', function (req, res) {
+        res.send('ok')
+      })
+
+      var channel = dc.channel('express.request.start')
+      function onMessage(message) {
+        published = true
+        assert.ok(message.req, 'message should contain req')
+        assert.ok(message.res, 'message should contain res')
+        assert.strictEqual(message.req.url, '/')
+      }
+      channel.subscribe(onMessage)
+
+      request(app)
+        .get('/')
+        .expect(200, function (err) {
+          channel.unsubscribe(onMessage)
+          if (err) return done(err)
+          assert.ok(published, 'express.request.start should have been published')
+          done()
+        })
+    })
+
+    it('should publish for every request', function (done) {
+      var app = express()
+      var count = 0
+
+      app.get('/', function (req, res) {
+        res.send('ok')
+      })
+
+      var channel = dc.channel('express.request.start')
+      function onMessage() {
+        count++
+      }
+      channel.subscribe(onMessage)
+
+      request(app)
+        .get('/')
+        .expect(200, function (err) {
+          if (err) { channel.unsubscribe(onMessage); return done(err) }
+          request(app)
+            .get('/')
+            .expect(200, function (err) {
+              channel.unsubscribe(onMessage)
+              if (err) return done(err)
+              assert.strictEqual(count, 2, 'should publish for each request')
+              done()
+            })
+        })
+    })
+
+    it('should not add overhead when no subscribers', function (done) {
+      var app = express()
+
+      app.get('/', function (req, res) {
+        res.send('ok')
+      })
+
+      // no subscribers — just verify the request works
+      request(app)
+        .get('/')
+        .expect(200, done)
+    })
+  })
+
+  describe('express.request.finish', function () {
+    it('should publish after response is sent', function (done) {
+      var app = express()
+      var published = false
+
+      app.get('/', function (req, res) {
+        res.send('ok')
+      })
+
+      var channel = dc.channel('express.request.finish')
+      function onMessage(message) {
+        published = true
+        assert.ok(message.req, 'message should contain req')
+        assert.ok(message.res, 'message should contain res')
+        assert.strictEqual(message.res.statusCode, 200)
+      }
+      channel.subscribe(onMessage)
+
+      request(app)
+        .get('/')
+        .expect(200, function (err) {
+          // give onFinished a tick to fire
+          setImmediate(function () {
+            channel.unsubscribe(onMessage)
+            if (err) return done(err)
+            assert.ok(published, 'express.request.finish should have been published')
+            done()
+          })
+        })
+    })
+
+    it('should publish for error responses', function (done) {
+      var app = express()
+      var finishStatus = null
+
+      app.get('/', function (req, res) {
+        res.status(500).send('Internal Server Error')
+      })
+
+      var channel = dc.channel('express.request.finish')
+      function onMessage(message) {
+        finishStatus = message.res.statusCode
+      }
+      channel.subscribe(onMessage)
+
+      request(app)
+        .get('/')
+        .expect(500, function (err) {
+          setImmediate(function () {
+            channel.unsubscribe(onMessage)
+            if (err) return done(err)
+            assert.strictEqual(finishStatus, 500, 'should capture error status code')
+            done()
+          })
+        })
+    })
+
+    it('should publish for 404 responses', function (done) {
+      var app = express()
+      var finishStatus = null
+
+      // no routes — will 404
+
+      var channel = dc.channel('express.request.finish')
+      function onMessage(message) {
+        finishStatus = message.res.statusCode
+      }
+      channel.subscribe(onMessage)
+
+      request(app)
+        .get('/nonexistent')
+        .expect(404, function (err) {
+          setImmediate(function () {
+            channel.unsubscribe(onMessage)
+            if (err) return done(err)
+            assert.strictEqual(finishStatus, 404, 'should capture 404 status')
+            done()
+          })
+        })
+    })
+  })
+
+  describe('express.request.error', function () {
+    it('should not publish on normal responses', function (done) {
+      var app = express()
+      var errorPublished = false
+
+      app.get('/', function (req, res) {
+        res.send('ok')
+      })
+
+      var channel = dc.channel('express.request.error')
+      function onMessage() {
+        errorPublished = true
+      }
+      channel.subscribe(onMessage)
+
+      request(app)
+        .get('/')
+        .expect(200, function (err) {
+          setImmediate(function () {
+            channel.unsubscribe(onMessage)
+            if (err) return done(err)
+            assert.ok(!errorPublished, 'express.request.error should not publish on success')
+            done()
+          })
+        })
+    })
+  })
+
+  describe('channel isolation', function () {
+    it('should publish start before finish', function (done) {
+      var app = express()
+      var order = []
+
+      app.get('/', function (req, res) {
+        res.send('ok')
+      })
+
+      var startChannel = dc.channel('express.request.start')
+      var finishChannel = dc.channel('express.request.finish')
+
+      function onStart() { order.push('start') }
+      function onFinish() { order.push('finish') }
+
+      startChannel.subscribe(onStart)
+      finishChannel.subscribe(onFinish)
+
+      request(app)
+        .get('/')
+        .expect(200, function (err) {
+          setImmediate(function () {
+            startChannel.unsubscribe(onStart)
+            finishChannel.unsubscribe(onFinish)
+            if (err) return done(err)
+            assert.deepStrictEqual(order, ['start', 'finish'], 'start should fire before finish')
+            done()
+          })
+        })
+    })
+  })
+})