From b2c5f5ecc9ddad262a7cbb56e5d20b84a31973b8 Mon Sep 17 00:00:00 2001
From: Zaid Abdulla <zaid@genieacs.com>
Date: Sat, 16 Sep 2017 22:43:17 -0700
Subject: [PATCH] Handle overload spikes gracefully

---
 lib/config.coffee |  1 +
 lib/cwmp.coffee   | 52 ++++++++++++++++++++++++++++++++++++++++-------
 2 files changed, 46 insertions(+), 7 deletions(-)

diff --git a/lib/config.coffee b/lib/config.coffee
index 1f57a73..377da1e 100644
--- a/lib/config.coffee
+++ b/lib/config.coffee
@@ -64,6 +64,7 @@ options = {
   COOKIES_PATH : {type : 'string'},
   LOG_FORMAT : {type : 'string', default : 'simple'},
   ACCESS_LOG_FORMAT : {type : 'string', default : ''},
+  MAX_CONCURRENT_REQUESTS : {type : 'int', default: 20},
 
   # XML configuration
   XML_RECOVER : {type : 'bool', default : false},
diff --git a/lib/cwmp.coffee b/lib/cwmp.coffee
index efd1e74..8b7d4a3 100644
--- a/lib/cwmp.coffee
+++ b/lib/cwmp.coffee
@@ -34,12 +34,20 @@ logger = require './logger'
 
 MAX_CYCLES = 4
 
+MAX_CONCURRENT_REQUESTS = config.get('MAX_CONCURRENT_REQUESTS')
+stats = {
+  concurrentRequests: 0
+  totalRequests: 0
+  droppedRequests: 0
+  initiatedSessions: 0
+}
 
 throwError = (err, httpResponse) ->
   if httpResponse
     currentSessions.delete(httpResponse.connection)
     httpResponse.writeHead(500, {'Connection' : 'close'})
     httpResponse.end("#{err.name}: #{err.message}")
+    stats.concurrentRequests -= 1
 
   throw err
 
@@ -69,6 +77,7 @@ writeResponse = (sessionContext, res) ->
       sessionContext.httpResponse.end(data)
       delete sessionContext.httpRequest
       delete sessionContext.httpResponse
+      stats.concurrentRequests -= 1
     )
   else
     res.headers['Content-Length'] = res.data.length
@@ -76,6 +85,7 @@ writeResponse = (sessionContext, res) ->
     sessionContext.httpResponse.end(res.data)
     delete sessionContext.httpRequest
     delete sessionContext.httpResponse
+    stats.concurrentRequests -= 1
 
 
 recordFault = (sessionContext, fault, provisions, channels) ->
@@ -575,12 +585,7 @@ sendAcsRequest = (sessionContext, id, acsRequest) ->
   writeResponse(sessionContext, res)
 
 
-getSession = (httpRequest, callback) ->
-  # Separation by comma is important as some devices don't comform to standard
-  COOKIE_REGEX = /\s*([a-zA-Z0-9\-_]+?)\s*=\s*"?([a-zA-Z0-9\-_]*?)"?\s*(,|;|$)/g
-  while match = COOKIE_REGEX.exec(httpRequest.headers.cookie)
-    sessionId = match[2] if match[1] == 'session'
-
+getSession = (httpRequest, sessionId, callback) ->
   return callback() if not sessionId?
 
   sessionContext = currentSessions.get(httpRequest.connection)?[sessionId]
@@ -618,6 +623,21 @@ onConnection = (socket) ->
       )
   )
 
+setInterval(() ->
+  if stats.droppedRequests
+    logger.warn({
+      message: 'Worker overloaded'
+      droppedRequests: stats.droppedRequests
+      totalRequests: stats.totalRequests
+      initiatedSessions: stats.initiatedSessions
+      pid: process.pid
+    })
+
+  stats.totalRequests = 0
+  stats.droppedRequests = 0
+  stats.initiatedSessions = 0
+, 10000).unref()
+
 
 getDueTasksAndFaultsAndOperations = (deviceId, timestamp, callback) ->
   cache.get("#{deviceId}_tasks_faults_operations", (err, res) ->
@@ -682,11 +702,25 @@ cacheDueTasksAndFaultsAndOperations = (deviceId, tasks, faults, operations, cach
 
 
 listener = (httpRequest, httpResponse) ->
+  stats.totalRequests += 1
   if httpRequest.method != 'POST'
     httpResponse.writeHead 405, {'Allow': 'POST'}
     httpResponse.end('405 Method Not Allowed')
     return
 
+  sessionId = null
+  # Separation by comma is important as some devices don't comform to standard
+  COOKIE_REGEX = /\s*([a-zA-Z0-9\-_]+?)\s*=\s*"?([a-zA-Z0-9\-_]*?)"?\s*(,|;|$)/g
+  while match = COOKIE_REGEX.exec(httpRequest.headers.cookie)
+    sessionId = match[2] if match[1] == 'session'
+
+  # If overloaded, ask CPE to retry in 60 seconds
+  if not sessionId and stats.concurrentRequests > MAX_CONCURRENT_REQUESTS
+    httpResponse.writeHead(503, {'Retry-after': 60})
+    httpResponse.end('503 Service Unavailable')
+    stats.droppedRequests += 1
+    return
+
   if httpRequest.headers['content-encoding']?
     switch httpRequest.headers['content-encoding']
       when 'gzip'
@@ -700,6 +734,7 @@ listener = (httpRequest, httpResponse) ->
   else
     stream = httpRequest
 
+  stats.concurrentRequests += 1
   chunks = []
   bytes = 0
 
@@ -721,15 +756,18 @@ listener = (httpRequest, httpResponse) ->
   stream.on('end', () ->
     rpc = null
     parseWarnings = []
-    getSession(httpRequest, f = (err, sessionContext) ->
+    getSession(httpRequest, sessionId, f = (err, sessionContext) ->
       return throwError(err, httpResponse) if err
+
       rpc ?= soap.request(httpRequest, sessionContext?.cwmpVersion, parseWarnings)
       if not sessionContext?
         if rpc.cpeRequest?.name isnt 'Inform'
           httpResponse.writeHead(400)
           httpResponse.end('Session is expired')
+          stats.concurrentRequests -= 1
           return
 
+        stats.initiatedSessions += 1
         deviceId = common.generateDeviceId(rpc.cpeRequest.deviceId)
         return session.init(deviceId, rpc.cwmpVersion, rpc.sessionTimeout ? config.get('SESSION_TIMEOUT', deviceId), (err, sessionContext) ->
           return throwError(err, httpResponse) if err