feat(graceful): stop agents asynchronously

Author: Shadowbeetle

.npmignore

@@ -8,6 +8,7 @@ scripts
 test
 *.spec.js
 *.spec.e2e.js
+*.mock.js
 example
 
 # .gitignore contents

lib/agent/agent.js

@@ -15,10 +15,14 @@ Agent.prototype.start = function () {
   }
 }
 
-Agent.prototype.stop = function () {
+Agent.prototype.stop = function (callback) {
   if (this.timer) {
     this.timer.end()
   }
+
+  if (callback) {
+    process.nextTick(callback)
+  }
 }
 
 Agent.prototype.initialize = function () { }

lib/agent/api/index.js

@@ -103,7 +103,7 @@ CollectorApi.prototype._send = function (destinationUrl, data, callback, options
     res.pipe(bl(function (err, result) {
       if (err) {
         debug('#_send', '[Error]', err)
-        return
+        return callback(err)
       }
       callback(null, result)
     }))
@@ -126,37 +126,39 @@ CollectorApi.prototype._send = function (destinationUrl, data, callback, options
   }
 }
 
-CollectorApi.prototype.sendRpmMetrics = function (data) {
+CollectorApi.prototype.sendRpmMetrics = function (data, callback) {
+  callback = callback || function () {}
   if (!isNumber(this.serviceKey)) {
     debug('#sendRpmMetrics', '[Error] Service key not present, cannot send rpm metrics')
-    return
+    return callback(new Error('serviceKey is missing'))
   }
 
   var url = util.format(this.COLLECTOR_API_RPM_METRICS, this.serviceKey)
-  this._send(url, this._withInstanceInfo(data))
+  this._send(url, this._withInstanceInfo(data), callback)
 }
 
 CollectorApi.prototype.ping = function () {
   if (!isNumber(this.serviceKey)) {
     debug('#ping', '[Error] Service key not present, cannot do healthcheck')
-    return
   }
 
   var url = util.format(this.COLLECTOR_API_HEALTHCHECK, this.serviceKey)
   this._send(url, this._withInstanceInfo({}))
 }
 
-CollectorApi.prototype.sendApmMetrics = function (data) {
+CollectorApi.prototype.sendApmMetrics = function (data, callback) {
+  callback = callback || function () {}
   if (!isNumber(this.serviceKey)) {
     debug('#sendApmMetrics', '[Error] Service key not present, cannot send metrics')
-    return
+    return callback(new Error('serviceKey is missing'))
   }
 
   var url = util.format(this.COLLECTOR_API_METRICS, this.serviceKey)
-  this._send(url, this._withInstanceInfo(data))
+  this._send(url, this._withInstanceInfo(data), callback)
 }
 
 CollectorApi.prototype.sendMemorySnapshot = function (data, callback) {
+  callback = callback || function () {}
   if (!isNumber(this.serviceKey)) {
     debug('#sendMemorySnapshot', '[Error] Service key not present, cannot send heapdump')
     return callback(new Error('serviceKey is missing'))
@@ -167,6 +169,7 @@ CollectorApi.prototype.sendMemorySnapshot = function (data, callback) {
 }
 
 CollectorApi.prototype.sendCpuProfile = function (data, callback) {
+  callback = callback || function () {}
   if (!isNumber(this.serviceKey)) {
     debug('#sendCpuProfile', '[Error] Service key not present, cannot send cpu profile')
     return callback(new Error('serviceKey is missing'))
@@ -176,33 +179,36 @@ CollectorApi.prototype.sendCpuProfile = function (data, callback) {
   this._send(url, this._withInstanceInfo(data), callback)
 }
 
-CollectorApi.prototype.sendExternalEdgeMetrics = function (data) {
+CollectorApi.prototype.sendExternalEdgeMetrics = function (data, callback) {
+  callback = callback || function () {}
   if (!isNumber(this.serviceKey)) {
     debug('#sendExternalEdgeMetrics', '[Error] Service key not present, cannot send metrics')
-    return
+    return callback(new Error('serviceKey is missing'))
   }
 
   var url = util.format(this.COLLECTOR_API_EXTERNAL_EDGE_METRICS, this.serviceKey)
-  this._send(url, this._withInstanceInfo(data))
+  this._send(url, this._withInstanceInfo(data), callback)
 }
 
-CollectorApi.prototype.sendIncomingEdgeMetrics = function (data) {
+CollectorApi.prototype.sendIncomingEdgeMetrics = function (data, callback) {
+  callback = callback || function () {}
   if (!isNumber(this.serviceKey)) {
     debug('#sendIncomingEdgeMetrics', '[Error] Service key not present, cannot send metrics')
-    return
+    return callback(new Error('serviceKey is missing'))
   }
 
   var url = util.format(this.COLLECTOR_API_INCOMING_EDGE_METRICS, this.serviceKey)
   this._send(url, this._withInstanceInfo({
     timestamp: (new Date()).toISOString(),
     data: data
-  }))
+  }), callback)
 }
 
 CollectorApi.prototype.getUpdates = function (data, callback) {
+  callback = callback || function () {}
   if (!isNumber(this.serviceKey)) {
     debug('#getUpdates', '[Error] Service key not present, cannot get updates')
-    return
+    return callback(new Error('serviceKey is missing'))
   }
 
   var url = util.format(this.COLLECTOR_API_CONTROL, this.serviceKey)
@@ -222,17 +228,18 @@ CollectorApi.prototype.getUpdates = function (data, callback) {
   })
 }
 
-CollectorApi.prototype.sendCustomMetrics = function (data) {
+CollectorApi.prototype.sendCustomMetrics = function (data, callback) {
+  callback = callback || function () {}
   if (!isNumber(this.serviceKey)) {
     debug('#sendCustomMetrics', '[Error] Service key not present, cannot send metrics')
-    return
+    return callback(new Error('serviceKey is missing'))
   }
 
   var url = util.format(this.COLLECTOR_API_CUSTOM_METRICS, this.serviceKey)
   this._send(url, this._withInstanceInfo({
     timestamp: (new Date()).toISOString(),
     data: data
-  }))
+  }), callback)
 }
 
 CollectorApi.prototype.sendDependencies = function (dependencies) {

lib/agent/api/index.spec.js

@@ -1,7 +1,8 @@
 var util = require('util')
 var url = require('url')
 var https = require('https')
-
+var FakeReadable = require('../util').FakeStream.FakeReadable
+var FakeWritable = require('../util').FakeStream.FakeWritable
 var CollectorApi = require('./')
 
 var expect = require('chai').expect
@@ -10,6 +11,7 @@ var libPackage = require('../../../package')
 
 describe('The Trace CollectorApi module', function () {
   var defaultConfig
+
   beforeEach(function () {
     defaultConfig = {
       serviceName: 'testName',
@@ -116,6 +118,49 @@ describe('The Trace CollectorApi module', function () {
     expect(sendStub).to.have.been.calledWith(sendUrl, data)
   })
 
+  it('calls callback on successful send', function (done) {
+    var serviceKey = 12
+
+    var path = util.format(defaultConfig.collectorApiRpmMetricsEndpoint, serviceKey)
+    var sendUrl = url.resolve(defaultConfig.collectorApiUrl, path)
+
+    var collectorApi = CollectorApi.create(defaultConfig)
+    collectorApi.serviceKey = serviceKey
+
+    this.sandbox.stub(https, 'request', function (_, cb) {
+      process.nextTick(function () {
+        cb(new FakeReadable())
+      })
+      return new FakeWritable()
+    })
+
+    collectorApi._send(sendUrl, {}, function (err) {
+      done(err)
+    })
+  })
+
+  it('calls callback on send failure', function (done) {
+    var serviceKey = 12
+
+    var collectorApi = CollectorApi.create(defaultConfig)
+    collectorApi.serviceKey = serviceKey
+
+    this.sandbox.stub(https, 'request', function (_, cb) {
+      process.nextTick(function () {
+        cb(new FakeReadable(new Error('uh-oh')))
+      })
+      return new FakeWritable()
+    })
+
+    collectorApi._send('test-url', {}, function (err) {
+      if (err) {
+        done()
+      } else {
+        done('Should have been called with error')
+      }
+    })
+  })
+
   it('sends incomingEdgeMetrics with the required properties', function (done) {
     var serviceKey = 12
     var collectorApi = CollectorApi.create(defaultConfig)

lib/agent/index.js

@@ -111,11 +111,24 @@ Agent.prototype.start = function () {
   })
 }
 
-Agent.prototype.stop = function () {
+Agent.prototype.stop = function (callback) {
   debug('#stop', 'Stopping agents...')
-  this.agents.forEach(function (agent) {
-    agent.stop()
-    debug('#stop', agent.name + ' stopped')
+  var agents = this.agents
+  var counter = 1
+  var error
+
+  agents.forEach(function (agent) {
+    agent.stop(function (err) {
+      if (!error && err) {
+        error = err
+      }
+
+      if (counter >= agents.length) {
+        callback(error)
+      } else {
+        counter++
+      }
+    })
   })
 }
 

lib/agent/metrics/apm/index.js

@@ -45,7 +45,8 @@ function ApmMetrics (options) {
 
 inherits(ApmMetrics, Agent)
 
-ApmMetrics.prototype.sendMetrics = function () {
+ApmMetrics.prototype.sendMetrics = function (callback) {
+  callback = callback || function () {}
   var eventloop = this.getEventLoop()
   var gc = this.getGC()
 
@@ -63,7 +64,7 @@ ApmMetrics.prototype.sendMetrics = function () {
     databag.gc = gc
   }
 
-  this.collectorApi.sendApmMetrics(databag)
+  this.collectorApi.sendApmMetrics(databag, callback)
 
   this.reset()
 }
@@ -141,6 +142,11 @@ ApmMetrics.prototype.reset = function () {
   }
 }
 
+ApmMetrics.prototype.stop = function (callback) {
+  Agent.prototype.stop.call(this)
+  this.sendMetrics(callback)
+}
+
 function create (options) {
   return new ApmMetrics(options)
 }

lib/agent/metrics/custom/index.js

@@ -42,16 +42,22 @@ CustomMetrics.prototype.record = function (name, value) {
   }
 }
 
-CustomMetrics.prototype.sendMetrics = function () {
+CustomMetrics.prototype.sendMetrics = function (callback) {
+  callback = callback || function () {}
   this.collectorApi.sendCustomMetrics({
     incrementMetrics: this.incrementMetrics,
     recordMetrics: this.recordMetrics
-  })
+  }, callback)
 
   this.incrementMetrics = {}
   this.recordMetrics = {}
 }
 
+CustomMetrics.prototype.stop = function (callback) {
+  Agent.prototype.stop.call(this)
+  this.sendMetrics(callback)
+}
+
 function create (options) {
   return new CustomMetrics(options)
 }

lib/agent/metrics/externalEdge/index.js

@@ -70,8 +70,9 @@ ExternalEdgeMetrics.prototype.calculateTimes = function (items) {
   }
 }
 
-ExternalEdgeMetrics.prototype.sendMetrics = function () {
+ExternalEdgeMetrics.prototype.sendMetrics = function (callback) {
   var _this = this
+  callback = callback || function () {}
 
   var metrics = flatMap(Object.keys(this.metrics), function (protocol) {
     return Object.keys(_this.metrics[protocol]).map(function (hostName) {
@@ -96,20 +97,25 @@ ExternalEdgeMetrics.prototype.sendMetrics = function () {
 
   // if no metrics, don't send anything
   if (!metrics || !metrics.length) {
-    return
+    return callback()
   }
 
   this.collectorApi.sendExternalEdgeMetrics({
     timestamp: (new Date()).toISOString(),
     data: metrics
-  })
+  }, callback)
 }
 
 ExternalEdgeMetrics.prototype.EDGE_STATUS = {
   OK: 0,
   NOT_OK: 1
 }
 
+ExternalEdgeMetrics.prototype.stop = function (callback) {
+  Agent.prototype.stop.call(this)
+  this.sendMetrics(callback)
+}
+
 function create (options) {
   return new ExternalEdgeMetrics(options)
 }

lib/agent/metrics/incomingEdge/index.js

@@ -59,8 +59,9 @@ IncomingEdgeMetrics.prototype._calculateTimes = function (items) {
   }
 }
 
-IncomingEdgeMetrics.prototype.sendMetrics = function () {
+IncomingEdgeMetrics.prototype.sendMetrics = function (callback) {
   var _this = this
+  callback = callback || function () {}
 
   var metrics = flatMap(Object.keys(this.metrics), function (protocol) {
     return Object.keys(_this.metrics[protocol]).map(function (serviceKey) {
@@ -89,17 +90,22 @@ IncomingEdgeMetrics.prototype.sendMetrics = function () {
 
   // if no metrics, don't send anything
   if (!metrics || !metrics.length) {
-    return
+    return callback()
   }
 
-  this.collectorApi.sendIncomingEdgeMetrics(metrics)
+  this.collectorApi.sendIncomingEdgeMetrics(metrics, callback)
 }
 
 IncomingEdgeMetrics.prototype.EDGE_STATUS = {
   OK: 0,
   NOT_OK: 1
 }
 
+IncomingEdgeMetrics.prototype.stop = function (callback) {
+  Agent.prototype.stop.call(this)
+  this.sendMetrics(callback)
+}
+
 function create (options) {
   return new IncomingEdgeMetrics(options)
 }