feat: monitor failed http requests, mark as frozen on lambda freeze

default-vars.yml

@@ -14,6 +14,7 @@ versionFunctions: false
 serverless-offline:
   host: 0.0.0.0
   port: 21012
+  skipCacheInvalidation: false
 
 serverless-iot-local:
   host: 0.0.0.0

src/blockchain-adapter/ethereum.ts

@@ -31,7 +31,9 @@ export = function getNetworkAdapters ({ networkName, privateKey }) {
       networkName,
       privateKey,
       pollingInterval: 10000,
-      etherscan: true,
+      etherscan: {
+        maxRequestsPerSecond: 5,
+      },
       autostart: false,
       maxPriceInWei: MAX_PRICE_IN_WEI,
     }

src/env.ts

@@ -76,6 +76,7 @@ export default class Env {
   public _X_AMZN_TRACE_ID:string
   public AWS_ACCOUNT_ID: string
   public SESSION_TTL?: number
+  public ABORT_REQUESTS_ON_FREEZE?: boolean
 
   constructor(props:any) {
     props = clone(props)

src/globals.ts

@@ -1,104 +1,83 @@
-import http from 'http'
-import { parse as parseURL } from 'url'
+import { install as installSourceMapSupport } from 'source-map-support'
+
+installSourceMapSupport()
+
 // @ts-ignore
-// import Promise from 'bluebird'
+import Promise from 'bluebird'
 
-// global.Promise = Promise
+global.Promise = Promise
 
-import pick from 'lodash/pick'
-import once from 'lodash/once'
+import http from 'http'
+import { parse as parseURL } from 'url'
 import AWS from 'aws-sdk'
-import AWSXRay from 'aws-xray-sdk-core'
-import mockery from 'mockery'
-import { install as installSourceMapSupport } from 'source-map-support'
-import { createLogger } from './logger'
-
-const xrayIsOn = process.env.TRADLE_BUILD !== '1' && process.env._X_AMZN_TRACE_ID
 
-const logFailedHttpRequests = () => {
-  const logger = createLogger('global:http')
-  const mkHttpReq = http.request.bind(http)
-  http.request = (...args) => {
-    const req = mkHttpReq(...args)
-    const start = Date.now()
-    let [opts] = args
-    if (typeof opts === 'string') {
-      opts = parseURL(opts)
-    }
+AWS.config.setPromisesDependency(Promise)
 
-    req.on('error', error => {
-      const details:any = pick(opts, [
-        'port',
-        'path',
-        'host',
-        'protocol',
-        'hostname',
-        'hash',
-        'search',
-        'query',
-        'pathname',
-        'href'
-      ])
-
-      details.duration = Date.now() - start
-      logger.error('request failed', details)
-    })
-
-    return req
-  }
-}
+import AWSXRay from 'aws-xray-sdk-core'
 
-once(() => {
-  process.env.XRAY_IS_ON = xrayIsOn ? '1' : ''
+const xrayIsOn = process.env.TRADLE_BUILD !== '1' && process.env._X_AMZN_TRACE_ID
+process.env.XRAY_IS_ON = xrayIsOn ? '1' : ''
 
-  installSourceMapSupport()
-  logFailedHttpRequests()
+import mockery from 'mockery'
+import once from 'lodash/once'
+import { createLogger } from './logger'
+import { requestInterceptor } from './request-interceptor'
 
-  // AWS.config.setPromisesDependency(Promise)
+const warn = (...args) => {
+  // no need to pollute with this anymore
 
-  if (xrayIsOn) {
-    // tslint-disable-rule: no-console
-    console.warn('capturing all http requests with AWSXRay')
-    AWSXRay.captureHTTPsGlobal(http)
-  } else if (process.env.AWS_LAMBDA_FUNCTION_NAME) {
-    console.warn('AWSXray is off')
-  }
+  // if (!process.env.IS_OFFLINE) {
+  //   console.warn(...args)
+  // }
+}
 
+if (xrayIsOn) {
+  // tslint-disable-rule: no-console
+  warn('capturing all http requests with AWSXRay')
+  AWSXRay.captureHTTPsGlobal(http)
+} else if (process.env.AWS_LAMBDA_FUNCTION_NAME) {
+  warn('AWSXray is off')
+}
 
-  // process.on('unhandledRejection', function (reason, promise) {
-  //   console.error('possibly unhandled rejection', reason)
-  // })
+if (!process.env.IS_OFFLINE) {
+  const logger = createLogger('global:http')
 
-  const warn = (...args) => {
-    // no need to pollute with this anymore
+  requestInterceptor.disable()
+  requestInterceptor.enable()
+  requestInterceptor.on('error', reqInfo => {
+    if (reqInfo.freezeId) {
+      logger.debug('frozen request thawed and failed', reqInfo)
+    } else {
+      logger.error('request failed', reqInfo)
+    }
+  })
+}
 
-    // if (!process.env.IS_OFFLINE) {
-    //   console.warn(...args)
-    // }
-  }
+// process.on('unhandledRejection', function (reason, promise) {
+//   console.error('possibly unhandled rejection', reason)
+// })
 
-  mockery.enable({
-    warnOnReplace: false,
-    warnOnUnregistered: false
-  })
+mockery.enable({
+  warnOnReplace: false,
+  warnOnUnregistered: false
+})
 
-  // mockery.registerMock('@tradle/serverless', './')
+// mockery.registerMock('@tradle/serverless', './')
 
-  warn('disabling "scrypt" as it is an unneeded dep (here) of ethereumjs-wallet')
-  mockery.registerMock('scrypt', {})
+warn('disabling "scrypt" as it is an unneeded dep (here) of ethereumjs-wallet')
+mockery.registerMock('scrypt', {})
 
-  // https://github.com/Qix-/node-error-ex
-  warn(`replacing "error-ex" as it bluebird doesn't recognize its errors as Error objects`)
-  mockery.registerMock('error-ex', name => {
-    return class CustomError extends Error {
-      public name: string
-      constructor(message) {
-        super(message)
-        this.name = name
-      }
+// https://github.com/Qix-/node-error-ex
+warn(`replacing "error-ex" as it bluebird doesn't recognize its errors as Error objects`)
+mockery.registerMock('error-ex', name => {
+  return class CustomError extends Error {
+    public name: string
+    constructor(message) {
+      super(message)
+      this.name = name
     }
-  })
-})()
+  }
+})
 
 // if (process.env.IS_OFFLINE || process.env.IS_LOCAL || process.env.NODE_ENV === 'test') {
 //   warn('disabling "aws-xray-sdk" as this is a local environment')

src/lambda.ts

@@ -54,6 +54,7 @@ import {
 } from './types'
 
 import { warmup } from './middleware/warmup'
+import { requestInterceptor, RequestInfo } from './request-interceptor'
 
 const NOT_FOUND = new Error('nothing here')
 
@@ -63,9 +64,10 @@ const { commit } = require('./version')
 
 type Contextualized<T> = (ctx: T, next: Function) => any|void
 
-interface ServiceCallsSummary {
+interface PendingCallsSummary {
   start?: number
   duration?: number
+  httpRequests?: RequestInfo[]
   services: {
     [name: string]: any[]
   }
@@ -369,7 +371,18 @@ Previous exit stack: ${this.lastExitStack}`)
 
     const pendingServiceCalls = this._dumpPendingServiceCalls()
     if (!_.isEmpty(pendingServiceCalls.services)) {
-      this.logger.debug('service calls pending', pendingServiceCalls)
+      this.logger.debug('pending service calls', pendingServiceCalls)
+    }
+
+    requestInterceptor.freeze(this.requestId)
+
+    const pendingHttpRequests = this._dumpPendingHTTPRequests()
+    if (pendingHttpRequests.length) {
+      this.logger.debug('pending http requests', pendingHttpRequests)
+      if (this.env.ABORT_REQUESTS_ON_FREEZE) {
+        this.logger.warn(`aborting ${pendingHttpRequests.length} pending http requests`)
+        requestInterceptor.abortPending()
+      }
     }
 
     if (err) {
@@ -725,7 +738,7 @@ Previous exit stack: ${this.lastExitStack}`)
     service.$startRecording()
   }
 
-  private _dumpPendingServiceCalls = ():ServiceCallsSummary => {
+  private _dumpPendingServiceCalls = ():PendingCallsSummary => {
     try {
       // should never fail cause of this
       return this.__dumpPendingServiceCalls()
@@ -738,7 +751,19 @@ Previous exit stack: ${this.lastExitStack}`)
     }
   }
 
-  private _dumpServiceCalls = ():ServiceCallsSummary => {
+  private _dumpPendingHTTPRequests = ():RequestInfo[] => {
+    try {
+      return requestInterceptor.getPending()
+    } catch (err) {
+      this.logger.error('failed to dump pending http requests', {
+        error: err.stack
+      })
+
+      return []
+    }
+  }
+
+  private _dumpServiceCalls = ():PendingCallsSummary => {
     try {
       // should never fail cause of this
       return this.__dumpServiceCalls()
@@ -751,11 +776,12 @@ Previous exit stack: ${this.lastExitStack}`)
     }
   }
 
-  private __dumpPendingServiceCalls = (): ServiceCallsSummary => {
-    const summary: ServiceCallsSummary = {
+  private __dumpPendingServiceCalls = (): PendingCallsSummary => {
+    const summary: PendingCallsSummary = {
       start: Infinity,
       duration: 0,
       services: {},
+      httpRequests: requestInterceptor.getPending(),
     }
 
     forEachInstantiatedRecordableService(this.aws, (service, name) => {
@@ -779,8 +805,8 @@ Previous exit stack: ${this.lastExitStack}`)
     return summary
   }
 
-  private __dumpServiceCalls = ():ServiceCallsSummary => {
-    const summary:ServiceCallsSummary = {
+  private __dumpServiceCalls = ():PendingCallsSummary => {
+    const summary:PendingCallsSummary = {
       start: Infinity,
       duration: 0,
       services: {},

src/middleware/onmessage.ts

@@ -30,5 +30,6 @@ export const onMessage = (lambda: Lambda, { onSuccess, onError }) => {
     logger.debug(`preprocessed ${count} messages`)
     await next()
     await Promise.mapSeries(successes, success => onSuccess({ ...success, clientId }))
+    logger.debug(`postprocessed ${count} messages`)
   }
 }