fix(collector): add back reservoir sampler

Author: dszakallas

lib/agent/tracer/collector.js

@@ -6,6 +6,7 @@ var ExpiringBuffer = require('./expiringBuffer')
 var uuid = require('node-uuid')
 var assign = require('lodash.assign')
 var levels = require('./severity')
+var ReservoirSampler = require('./reservoirSampler')
 
 var EVENT_TYPE = {
   CLIENT_SEND: 'cs',
@@ -37,12 +38,13 @@ function Collector (options) {
   this.serviceKey = options.serviceKey
   this.mustCollectSeverity = levels.ERROR
   this.defaultSeverity = levels.INFO
+  this.samplerLimit = options.samplerLimit || 100
 
   // init required variables
   this._noStack = options.noStack
   this._eventTtl = options.eventTtl || 1
   this._eventBuffers = { }
-  this._mustCollectBuffer = []
+  this._sampler = new ReservoirSampler(this.samplerLimit)
 }
 
 Collector.prototype.LEVELS = levels
@@ -410,16 +412,13 @@ 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())
+      this._sampler.add(this._eventBuffers[cacheId].buffer.elements())
     }
     delete this._eventBuffers[cacheId]
   }
 }
 
 Collector.prototype.collect = function () {
-  var result = []
   var self = this
   Object.keys(this._eventBuffers).forEach(function (key) {
     if (self._eventBuffers[key].locked === true) {
@@ -430,7 +429,7 @@ Collector.prototype.collect = function () {
         delete self._eventBuffers[key]
       }
     } else if (levels.gte(self._eventBuffers[key].severity, self.mustCollectSeverity)) {
-      Array.prototype.push.apply(result, self._eventBuffers[key].buffer.elements())
+      self._sampler.add(self._eventBuffers[key].buffer.elements())
       delete self._eventBuffers[key].buffer
       self._eventBuffers[key].zombieTtl = self._eventTtl
     } else {
@@ -440,10 +439,10 @@ Collector.prototype.collect = function () {
       }
     }
   })
-
-  Array.prototype.push.apply(result, this._mustCollectBuffer)
-  this._mustCollectBuffer = []
-  return result
+  return this._sampler.flush().reduce(function (acc, x) {
+    Array.prototype.push.apply(acc, x)
+    return acc
+  }, [])
 }
 
 Collector.prototype.getTransactionId = function (briefcase) {

lib/agent/tracer/collector.spec.js

@@ -646,4 +646,38 @@ describe('Collector', function () {
       expect(tracer.collect()[0].d.r.stack).to.not.exist
     })
   })
+
+  describe('Sampling', function () {
+    it('happens when there is a big amount of data', function () {
+      this.sandbox.stub(microtime, 'now', function () {
+        return 2
+      })
+      var tracer = new Collector(assign({ samplerLimit: 10 }, options))
+      function report () {
+        var sr = {
+          payload: { },
+          duffelBag: {
+            communicationId: 'commId',
+            transactionId: 'trId',
+            severity: tracer.mustCollectSeverity + 1
+          }
+        }
+        var ss = {
+          payload: { },
+          briefcase: {
+            communication: {
+              id: sr.duffelBag.communicationId,
+              transactionId: sr.duffelBag.transactionId
+            },
+            severity: tracer.mustCollectSeverity
+          }
+        }
+
+        tracer.serverRecv(sr.payload, sr.duffelBag)
+        tracer.serverSend(ss.payload, ss.briefcase)
+      }
+      for (var i = 0; i < 100; ++i) { report() }
+      expect(tracer.collect().length).to.be.eql(20)
+    })
+  })
 })

lib/agent/tracer/reservoirSampler.js

@@ -0,0 +1,37 @@
+
+function Reservoir (limit) {
+  this.MAX_ITEMS = limit || 10
+
+  this._itemsSeen = 0
+  this._data = []
+}
+
+Reservoir.prototype.add = function (item) {
+  var inserted = true
+  if (this._itemsSeen < this.MAX_ITEMS) {
+    this._data.push(item)
+    inserted = true
+  } else {
+    inserted = this._replace(item)
+  }
+  this._itemsSeen++
+  return inserted
+}
+
+Reservoir.prototype.flush = function () {
+  var data = this._data
+  this._data = []
+  this._itemsSeen = 0
+  return data
+}
+
+Reservoir.prototype._replace = function (item) {
+  var toReplaceIndex = Math.floor(Math.random() * (this._itemsSeen + 2))
+  if (toReplaceIndex < this.MAX_ITEMS) {
+    this._data[toReplaceIndex] = item
+    return true
+  }
+  return false
+}
+
+module.exports = Reservoir

lib/agent/tracer/reservoirSampler.spec.js

@@ -0,0 +1,40 @@
+var expect = require('chai').expect
+var ReservoirSampler = require('./reservoirSampler')
+
+describe('The Reservoir Sampler', function () {
+  var itemToAdd = {
+    name: 'John Coin'
+  }
+
+  describe('when seen count is below the size limit', function () {
+    it('adds the item', function () {
+      var sampler = new ReservoirSampler(3)
+      var isAdded = sampler.add(itemToAdd)
+
+      expect(isAdded).to.eql(true)
+      expect(sampler.flush()).to.eql([itemToAdd])
+    })
+  })
+
+  describe('when seen count is above the size limit', function () {
+    it('adds the item if selected', function () {
+      this.sandbox.stub(Math, 'random', function () {
+        return 0
+      })
+      var sampler = new ReservoirSampler(1)
+      sampler._itemsSeen = 1
+      var isAdded = sampler.add(itemToAdd)
+      expect(isAdded).to.eql(true)
+    })
+
+    it('discards the item if not selected', function () {
+      this.sandbox.stub(Math, 'random', function () {
+        return 1
+      })
+      var sampler = new ReservoirSampler(1)
+      sampler._itemsSeen = 1
+      var isAdded = sampler.add(itemToAdd)
+      expect(isAdded).to.eql(false)
+    })
+  })
+})