fix: 保证finalhandler时执行正常

Author: wescai

src/agent.js

@@ -1,5 +1,15 @@
 const { EventEmitter } = require('events')
 
-class Agent extends EventEmitter {}
+const Transaction = require('./transaction')
+
+class Agent extends EventEmitter {
+  constructor() {
+    super()
+    this._transaction = new Transaction()
+  }
+  get transaction() {
+    return this._transaction
+  }
+}
 
 module.exports = Agent

src/constants.js

@@ -8,7 +8,8 @@
 const MODULE_TYPE = {
   /** Web server framework module, such as Express or Koa. */
   WEB_FRAMEWORK: 'web-framework',
-  PROXY: 'proxy'
+  PROXY: 'proxy',
+  TRANSACTION: 'transaction'
 }
 
 exports.MODULE_TYPE = MODULE_TYPE

src/instrumentation/core/http.js

@@ -0,0 +1,54 @@
+const utils = require('../../utils')
+
+function wrapEmitWithTransaction(agent, emit) {
+  const { transaction } = agent
+
+  return function wrappedHandler(evnt, request, response) {
+    transaction.init()
+
+    function instrumentedFinish() {
+      response.removeListener('finish', instrumentedFinish)
+      request.removeListener('aborted', instrumentedFinish)
+
+      // 状态码
+      if (response.statusCode != null) {
+        const statusCode = String(response.statusCode)
+        const path = transaction.path || request.path
+        if (/^\d+$/.test(statusCode)) {
+          const context = request.headers['x-apigateway-context']
+          const latency = Date.now() - transaction.start
+          const data = {
+            latency,
+            path,
+            method: request.method,
+            statusCode
+          }
+          agent.emit('responseFinish', context, data)
+        }
+      }
+      transaction.end()
+    }
+    // response结束时上报状态码和耗时
+    response.once('finish', instrumentedFinish)
+    request.once('aborted', instrumentedFinish)
+
+    return emit.apply(this, arguments)
+  }
+}
+
+module.exports = function initialize(agent, http) {
+  if (!http) {
+    return false
+  }
+
+  utils.wrapMethod(http.Server && http.Server.prototype, 'emit', function wrapEmit(emit) {
+    var txStarter = wrapEmitWithTransaction(agent, emit)
+    return function wrappedEmit(evnt) {
+      // 针对request事件做特殊逻辑
+      if (evnt === 'request') {
+        return txStarter.apply(this, arguments)
+      }
+      return emit.apply(this, arguments)
+    }
+  })
+}

src/instrumentation/express.js

@@ -38,27 +38,9 @@ module.exports = function initialize(agent, express) {
         utils.wrapMethod(layer, 'handle', function(func) {
           const { route } = layer
           const { path } = route
-          return function(request, response) {
-            function finish() {
-              response.removeListener('finish', finish)
-              request.removeListener('aborted', finish)
-              // 状态码
-              if (response.statusCode != null) {
-                const responseCode = String(response.statusCode)
-                if (/^\d+$/.test(responseCode)) {
-                  const context = request.headers['x-apigateway-context']
-                  agent.emit('responseFinish', context, request.method, path, responseCode)
-                }
-              }
-            }
-
-            // response结束时上报状态码和耗时
-            response.once('finish', finish)
-            request.once('aborted', finish)
-
-            const handle = func.apply(this, arguments)
-            return handle
-          }
+          const { transaction } = agent
+          transaction.path = path
+          return func
         })
       }
       return sourceRoute

src/instrumentation/tencent-serverless-http.js

@@ -1,16 +1,16 @@
 const utils = require('../utils')
-const { REUQEST_START_KEY } = require('../constants')
 const report = require('../report')
 
 module.exports = function initialize(agent, httpProxy) {
   utils.wrapMethod(httpProxy, 'proxy', function wrapRoute(fn) {
-    return function(server, event, context) {
-      context[REUQEST_START_KEY] = Date.now()
+    return function() {
+      const { transaction } = agent
+      transaction.init()
       const proxy = fn.apply(this, arguments)
       return new Promise(function(resolve) {
-        agent.once('responseFinish', function(ctx, method, path, responseCode) {
+        agent.once('responseFinish', function(ctx, data) {
           if (ctx) {
-            report.reportHttp(ctx, method, path, responseCode).then(
+            report.reportHttp(ctx, data).then(
               function() {
                 resolve(proxy)
               },

src/instrumentations.js

@@ -3,6 +3,7 @@ const { MODULE_TYPE } = require('./constants')
 module.exports = function instrumentations() {
   return {
     express: { type: MODULE_TYPE.WEB_FRAMEWORK },
+    finalhandler: { type: MODULE_TYPE.WEB_FRAMEWORK },
     'tencent-serverless-http': { type: MODULE_TYPE.PROXY }
   }
 }

src/report.js

@@ -1,6 +1,5 @@
 const { Capi } = require('@tencent-sdk/capi')
 const logger = require('./logger')
-const { REUQEST_START_KEY } = require('./constants')
 
 // 字符串转16进制
 function str2hex(str) {
@@ -14,7 +13,7 @@ function str2hex(str) {
   return arr.join('')
 }
 
-exports.reportHttp = async function(context, method, path, statusCode) {
+exports.reportHttp = async function(context, { latency, method, path, statusCode }) {
   try {
     context = JSON.parse(decodeURIComponent(context))
     // 自定级监控上报的指标名只支持【a-zA-Z0-9_-】，所以把path进行转义上报
@@ -55,7 +54,6 @@ exports.reportHttp = async function(context, method, path, statusCode) {
       host: 'monitor.tencentcloudapi.com'
     }
 
-    const latency = Date.now() - context[REUQEST_START_KEY]
     const keyPrefix = `${method}_${path}`
     const Metrics = [
       { MetricName: 'request', Value: 1 },

src/shimmer.js

@@ -3,12 +3,25 @@ const fs = require('fs')
 const utils = require('./utils')
 const INSTRUMENTATIONS = require('./instrumentations')()
 
-function _firstPartyInstrumentation(agent, fileName, nodule) {
+const { MODULE_TYPE } = require('./constants')
+
+const CORE_INSTRUMENTATION = {
+  http: {
+    type: MODULE_TYPE.TRANSACTION,
+    file: 'http.js'
+  },
+  https: {
+    type: MODULE_TYPE.TRANSACTION,
+    file: 'http.js'
+  }
+}
+
+function _firstPartyInstrumentation(agent, fileName, nodule, moduleName) {
   if (!fs.existsSync(fileName)) {
     return
   }
   try {
-    return require(fileName)(agent, nodule)
+    return require(fileName)(agent, nodule, moduleName)
   } catch (error) {
     agent.emit('responseFinish')
   }
@@ -73,6 +86,19 @@ const shimmer = {
   },
 
   bootstrapInstrumentation: function bootstrapInstrumentation(agent) {
+    // Instrument each of the core modules.
+    Object.keys(CORE_INSTRUMENTATION).forEach(function forEachCore(mojule) {
+      const core = CORE_INSTRUMENTATION[mojule]
+      const filePath = path.join(__dirname, 'instrumentation', 'core', core.file)
+      let uninstrumented = null
+
+      try {
+        uninstrumented = require(mojule)
+      } catch (err) {}
+
+      _firstPartyInstrumentation(agent, filePath, uninstrumented)
+    })
+
     // 注册所有注入模块
     Object.keys(INSTRUMENTATIONS).forEach(function forEachInstrumentation(moduleName) {
       const instrInfo = INSTRUMENTATIONS[moduleName]

src/transaction/index.js

@@ -0,0 +1,19 @@
+class Transaction {
+  get start() {
+    return this._start
+  }
+  set Path(path) {
+    this._path = path
+  }
+  get path() {
+    return this._path
+  }
+  init() {
+    this._start = Date.now()
+  }
+  end() {
+    this._start = null
+    this._path = null
+  }
+}
+module.exports = Transaction