fix(collector): solves TRN-34, TRN-39

Author: dszakallas

lib/agent/api/index.js

@@ -303,7 +303,6 @@ CollectorApi.prototype.getService = function (cb) {
         debug('#getService', '[Error] Service responded with', res.statusCode)
         return setTimeout(function () {
           debug('#getService', 'Retrying with %d ms', retryInterval)
-          console.log('adsfsadf')
           _this.getService(cb)
         }, retryInterval)
       }

lib/agent/tracer/collector.js

@@ -47,11 +47,6 @@ function Collector (options) {
 
 Collector.prototype.LEVELS = levels
 
-Collector.prototype.CACHE_MODES = {
-  NORMAL: 0,
-  RETAIN_UNTIL_SS: 1
-}
-
 Collector.prototype.userSentEvent = function (briefcase, name, payload) {
   briefcase = briefcase || {}
   var communicationId = briefcase.communication && briefcase.communication.id
@@ -144,7 +139,11 @@ Collector.prototype.networkError = function (briefcase, error) {
 
   var self = this
 
-  this._withCache(communicationId, function (cache) {
+  var root = !communicationId
+
+  var cacheId = root ? briefcase.csCtx.communicationId : communicationId
+
+  this._withCache(cacheId, function (cache) {
     cache.buffer.push({
       t: EVENT_TYPE.ERROR,
       r: briefcase.csCtx.transactionId,
@@ -162,18 +161,36 @@ Collector.prototype.networkError = function (briefcase, error) {
     })
     cache.severity = severity = levels.greater(cache.severity, severity)
   })
+
+  if (root) {
+    this._flushCache(cacheId)
+  }
+
   return { briefcase: assign({ severity: severity }, briefcase) }
 }
 
-Collector.prototype.clientSend = function (payload, briefcase) {
+Collector.prototype.end = function (briefcase, options) {
+  var communicationId = briefcase.communication.id
+
+  if (options && options.skip) {
+    this._deleteCache(communicationId)
+  } else {
+    this._flushCache(communicationId)
+  }
+}
+
+Collector.prototype.clientSend = function (payload, briefcase, options) {
   briefcase = briefcase || {}
   var timestamp = microtime.now()
 
   var parentCommunicationId = briefcase.communication && briefcase.communication.id
 
   var transactionId = briefcase.communication && briefcase.communication.transactionId
 
+  var root = !parentCommunicationId
+
   if (!transactionId) {
+    root = true
     transactionId = uuid.v4()
   }
 
@@ -192,7 +209,7 @@ Collector.prototype.clientSend = function (payload, briefcase) {
     transactionId: transactionId
   }
 
-  var cacheId = communication.parentId
+  var cacheId = root ? communication.id : communication.parentId
 
   this._withCache(cacheId, function (cache) {
     cache.buffer.push({
@@ -210,6 +227,10 @@ Collector.prototype.clientSend = function (payload, briefcase) {
     cache.severity = severity = levels.greater(cache.severity, severity)
   })
 
+  if (root) {
+    this._lockCache(cacheId)
+  }
+
   var duffelBag = {
     transactionId: transactionId,
     timestamp: timestamp,
@@ -258,7 +279,9 @@ Collector.prototype.serverRecv = function (payload, duffelBag, options) {
       d: payload.data
     })
     cache.severity = severity
-  }, { mode: options && options.cacheMode })
+  })
+
+  this._lockCache(communication.id)
 
   return { briefcase: briefcase }
 }
@@ -284,11 +307,11 @@ Collector.prototype.serverSend = function (payload, briefcase, options) {
     targetServiceKey: this.serviceKey
   }
 
-  var self = this
-
-  this._withCache(communicationId, function (cache) {
-    cache.severity = severity = levels.greater(severity, cache.severity)
-    if (!options || !options.skip) {
+  if (options && options.skip) {
+    this._deleteCache(communicationId)
+  } else {
+    this._withCache(communicationId, function (cache) {
+      cache.severity = severity = levels.greater(severity, cache.severity)
       cache.buffer.push({
         t: EVENT_TYPE.SERVER_SEND,
         r: transactionId,
@@ -298,22 +321,20 @@ Collector.prototype.serverSend = function (payload, briefcase, options) {
         s: payload.status,
         d: payload.data
       })
-      if (severity <= self.mustCollectSeverity) {
-        Array.prototype.push.apply(self._mustCollectBuffer, cache.buffer.elements())
-      }
-    }
-  })
+    })
+    this._flushCache(communicationId)
+  }
 
-  this._deleteCache(communicationId)
   return { briefcase: assign({ severity: severity }, briefcase), duffelBag: duffelBag }
 }
 
 Collector.prototype.clientRecv = function (payload, duffelBag, briefcase) {
   briefcase = briefcase || {}
   var timestamp = microtime.now()
   if (!briefcase.csCtx) {
-    debug(this.name, 'Error: cannot collect CR event without a CS context. Ignoring')
-    return { briefcase: briefcase }
+    var err = new Error('cannot collect CR event without a CS context. Ignoring')
+    debug('#clientRecv', '[Warning]', err.toString())
+    return { briefcase: briefcase, error: err }
   }
 
   var communication = {
@@ -324,7 +345,11 @@ Collector.prototype.clientRecv = function (payload, duffelBag, briefcase) {
 
   var severity = levels.greater(briefcase.severity, duffelBag.severity)
 
-  this._withCache(communication.parentId, function (cache) {
+  var root = !communication.parentId
+
+  var cacheId = root ? communication.id : communication.parentId
+
+  this._withCache(cacheId, function (cache) {
     cache.buffer.push({
       t: EVENT_TYPE.CLIENT_RECV,
       r: communication.transactionId,
@@ -340,16 +365,17 @@ Collector.prototype.clientRecv = function (payload, duffelBag, briefcase) {
     cache.severity = severity = levels.greater(severity, cache.severity)
   })
 
+  if (root) {
+    this._flushCache(cacheId)
+  }
+
   return { briefcase: assign({ severity: severity }, briefcase) }
 }
 
 Collector.prototype._withCache = function (cacheId, cb, options) {
   cacheId = cacheId || 'root'
   options = options || {}
   if (this._eventBuffers[cacheId] == null) {
-    if (options.noCreate) {
-      return
-    }
     this._eventBuffers[cacheId] = {
       severity: this.defaultSeverity,
       buffer: new ExpiringBuffer(this._eventTtl)
@@ -358,9 +384,6 @@ Collector.prototype._withCache = function (cacheId, cb, options) {
     delete this._eventBuffers[cacheId].zombieTtl
     this._eventBuffers[cacheId].buffer = new ExpiringBuffer(this._eventTtl)
   }
-  if (options.mode === this.CACHE_MODES.RETAIN_UNTIL_SS) {
-    this._eventBuffers[cacheId].locked = true
-  }
   cb(this._eventBuffers[cacheId])
 }
 
@@ -369,6 +392,32 @@ Collector.prototype._deleteCache = function (cacheId) {
   delete this._eventBuffers[cacheId]
 }
 
+Collector.prototype._lockCache = function (cacheId) {
+  cacheId = cacheId || 'root'
+  if (this._eventBuffers[cacheId]) {
+    this._eventBuffers[cacheId].locked = true
+  }
+}
+
+Collector.prototype._unlockCache = function (cacheId) {
+  cacheId = cacheId || 'root'
+  if (this._eventBuffers[cacheId]) {
+    delete this._eventBuffers[cacheId].locked
+  }
+}
+
+Collector.prototype._flushCache = function (cacheId) {
+  cacheId = cacheId || 'root'
+  if (this._eventBuffers[cacheId]) {
+    if (this._eventBuffers[cacheId].severity <= this.mustCollectSeverity) {
+      Array.prototype.push.apply(
+        this._mustCollectBuffer,
+        this._eventBuffers[cacheId].buffer.elements())
+    }
+    delete this._eventBuffers[cacheId]
+  }
+}
+
 Collector.prototype.collect = function () {
   var result = []
   var self = this
@@ -397,20 +446,6 @@ Collector.prototype.collect = function () {
   return result
 }
 
-Collector.prototype.end = function (briefcase, options) {
-  var communicationId = briefcase.communication.id
-
-  var self = this
-  this._withCache(communicationId, function (cache) {
-    if (!options || !options.skip) {
-      if (cache.severity <= self.mustCollectSeverity) {
-        Array.prototype.push.apply(self._mustCollectBuffer, cache.buffer.elements())
-      }
-    }
-  }, { noCreate: true })
-  this._deleteCache(communicationId)
-}
-
 Collector.prototype.getTransactionId = function (briefcase) {
   return briefcase && briefcase.communication && briefcase.communication.transactionId
 }