update writer to buffer by payload size instead of by trace count

benchmark/core.js

@@ -70,7 +70,7 @@ suite
   })
   .add('Writer#append', {
     onStart () {
-      writer = new Writer({}, 1000000)
+      writer = new Writer({}, {})
     },
     fn () {
       writer.append(spanStub)

src/config.js

@@ -32,7 +32,6 @@ class Config {
     this.url = url ? new URL(url) : new URL(`${protocol}://${hostname || 'localhost'}:${port}`)
     this.hostname = hostname || this.url.hostname
     this.flushInterval = flushInterval
-    this.bufferSize = 100000
     this.sampleRate = sampleRate
     this.logger = options.logger
     this.plugins = !!plugins

src/opentracing/tracer.js

@@ -30,7 +30,7 @@ class DatadogTracer extends Tracer {
     this._logInjection = config.logInjection
     this._analytics = config.analytics
     this._prioritySampler = new PrioritySampler(config.env)
-    this._writer = new Writer(this._prioritySampler, config.url, config.bufferSize)
+    this._writer = new Writer(this._prioritySampler, config.url)
     this._recorder = new Recorder(this._writer, config.flushInterval)
     this._recorder.init()
     this._sampler = new Sampler(config.sampleRate)

src/writer.js

@@ -6,12 +6,14 @@ const format = require('./format')
 const encode = require('./encode')
 const tracerVersion = require('../lib/version')
 
+const MAX_SIZE = 8 * 1024 * 1024 // 8MB
+
 class Writer {
-  constructor (prioritySampler, url, size) {
+  constructor (prioritySampler, url) {
     this._queue = []
     this._prioritySampler = prioritySampler
     this._url = url
-    this._size = size
+    this._size = 0
   }
 
   get length () {
@@ -36,11 +38,12 @@ class Writer {
 
       log.debug(() => `Adding encoded trace to buffer: ${buffer.toString('hex').match(/../g).join(' ')}`)
 
-      if (this.length < this._size) {
-        this._queue.push(buffer)
-      } else {
-        this._squeeze(buffer)
+      if (buffer.length + this._size > MAX_SIZE) {
+        this.flush()
       }
+
+      this._size += buffer.length
+      this._queue.push(buffer)
     }
   }
 
@@ -51,6 +54,7 @@ class Writer {
       this._request(data, this._queue.length)
 
       this._queue = []
+      this._size = 0
     }
   }
 
@@ -82,11 +86,6 @@ class Writer {
       })
       .catch(e => log.error(e))
   }
-
-  _squeeze (buffer) {
-    const index = Math.floor(Math.random() * this.length)
-    this._queue[index] = buffer
-  }
 }
 
 module.exports = Writer

test/config.spec.js

@@ -24,7 +24,6 @@ describe('Config', () => {
     expect(config).to.have.nested.property('url.hostname', 'localhost')
     expect(config).to.have.nested.property('url.port', '8126')
     expect(config).to.have.property('flushInterval', 2000)
-    expect(config).to.have.property('bufferSize', 100000)
     expect(config).to.have.property('sampleRate', 1)
     expect(config).to.have.deep.property('tags', {
       'runtime-id': ''

test/opentracing/tracer.spec.js

@@ -69,7 +69,6 @@ describe('Tracer', () => {
       service: 'service',
       url: 'http://test:7777',
       flushInterval: 2000,
-      bufferSize: 1000,
       sampleRate: 0.5,
       logger: 'logger',
       tags: {},
@@ -99,7 +98,7 @@ describe('Tracer', () => {
     tracer = new Tracer(config)
 
     expect(Writer).to.have.been.called
-    expect(Writer).to.have.been.calledWith(prioritySampler, config.url, config.bufferSize)
+    expect(Writer).to.have.been.calledWith(prioritySampler, config.url)
     expect(Recorder).to.have.been.calledWith(writer, config.flushInterval)
     expect(recorder.init).to.have.been.called
   })

test/writer.spec.js

@@ -69,7 +69,7 @@ describe('Writer', () => {
       './encode': encode,
       '../lib/version': 'tracerVersion'
     })
-    writer = new Writer(prioritySampler, url, 3)
+    writer = new Writer(prioritySampler, url)
   })
 
   describe('length', () => {
@@ -95,26 +95,17 @@ describe('Writer', () => {
       expect(writer._queue).to.be.empty
     })
 
-    it('should replace a random trace when full', () => {
-      writer._queue = new Array(1000)
+    it('should flush when full', () => {
+      writer.append(span)
+      writer._size = 8 * 1024 * 1024
       writer.append(span)
 
-      expect(writer.length).to.equal(1000)
+      expect(writer.length).to.equal(1)
       expect(writer._queue).to.deep.include('encoded')
     })
-  })
-
-  describe('drop', () => {
-    beforeEach(() => {
-      span.context = sinon.stub().returns({
-        _trace: trace,
-        _sampling: {
-          drop: true
-        }
-      })
-    })
 
-    it('should not append if being dropped', () => {
+    it('should not append if the span was dropped', () => {
+      span.context()._sampling.drop = true
       writer.append(span)
 
       expect(writer._queue).to.be.empty