fix(collector): add expiry to locks to prevent memory leak in incomplete instrumentations

Author: dszakallas

lib/agent/tracer/collector.js

@@ -44,6 +44,7 @@ function Collector (options) {
   // init required variables
   this._noStack = options.noStack
   this._eventTtl = options.eventTtl || 1
+  this._lockExpriryMs = 240000
   this._eventBuffers = { }
   this._sampler = reservoirSampler.create(this.samplerLimit)
 }
@@ -374,6 +375,7 @@ Collector.prototype._addLockToCache = function (cacheId) {
   cacheId = cacheId || 'root'
   if (this._eventBuffers[cacheId]) {
     ++this._eventBuffers[cacheId].locks
+    this._eventBuffers[cacheId].lastLocked = Date.now()
   }
 }
 
@@ -418,26 +420,28 @@ Collector.prototype.collect = function () {
   var lockedCount = 0
   Object.keys(this._eventBuffers).forEach(function (key) {
     if (self._eventBuffers[key].locks > 0) {
-      ++lockedCount
-      return
-    } else {
-      if (!self._eventBuffers[key].skipped &&
-          some(self._eventBuffers[key].severityBuf.elements(), function (severity) {
-            return levels.gte(severity, self.mustCollectSeverity)
-          })) {
-        var events = self._eventBuffers[key].eventBuf.flush()
-        if (events.length) {
-          self._sampler.add(events)
-        }
+      var deltaSinceLastLocked = Date.now() - self._eventBuffers[key].lastLocked
+      if (deltaSinceLastLocked < self._lockExpriryMs) {
+        ++lockedCount
+        return
       }
-      self._eventBuffers[key].eventBuf.expire()
-      self._eventBuffers[key].severityBuf.expire()
-
-      if (self._eventBuffers[key].skipped ||
-            self._eventBuffers[key].severityBuf.isEmpty()) {
-        delete self._eventBuffers[key]
+    }
+    if (!self._eventBuffers[key].skipped &&
+        some(self._eventBuffers[key].severityBuf.elements(), function (severity) {
+          return levels.gte(severity, self.mustCollectSeverity)
+        })) {
+      var events = self._eventBuffers[key].eventBuf.flush()
+      if (events.length) {
+        self._sampler.add(events)
       }
     }
+    self._eventBuffers[key].eventBuf.expire()
+    self._eventBuffers[key].severityBuf.expire()
+
+    if (self._eventBuffers[key].skipped ||
+          self._eventBuffers[key].severityBuf.isEmpty()) {
+      delete self._eventBuffers[key]
+    }
   })
   debug('#collect', lockedCount + ' buffers locked')
   return this._sampler.flush().reduce(function (acc, x) {

lib/agent/tracer/collector.spec.js

@@ -12,6 +12,13 @@ var options = {
 }
 
 describe('Collector', function () {
+  var clock
+  beforeEach(function () {
+    clock = this.sandbox.useFakeTimers()
+  })
+  afterEach(function () {
+    clock.restore()
+  })
   describe('SR', function () {
     var payload = {
       protocol: 'http',
@@ -399,6 +406,37 @@ describe('Collector', function () {
       tracer.clientSend(payload, briefcase)
       expect(tracer.collect().length).to.eql(0)
     })
+
+    it('is forcefully collected after some time (lock expired)', function () {
+      var tracer = new Collector(options)
+      var payload = {
+        protocol: 'http',
+        action: 'action',
+        resource: 'resource',
+        host: 'host',
+        data: { },
+        severity: tracer.mustCollectSeverity
+      }
+      var briefcase = {
+        communication: {
+          id: 'communicationId',
+          transactionId: 'transactionId'
+        }
+      }
+      var id = 'uuid'
+
+      this.sandbox.stub(uuid, 'v4', function () {
+        return id
+      })
+
+      this.sandbox.stub(microtime, 'now', function () {
+        return 2
+      })
+      tracer.clientSend(payload, briefcase)
+      clock.tick(tracer._lockExpriryMs)
+      expect(tracer.collect().length).to.eql(1)
+      expect(tracer.collect().length).to.eql(0)
+    })
     it('is not collected (in itself) when not in a transaction', function () {
       var tracer = new Collector(options)
       var payload = { }