feat(profiling): adding heapdump (#37)

gergelyke · web-flow · commit 607f128d2a72 · 2016-06-28T15:19:28.000+02:00
* feat(profiling): adding heapdump

* feat(control): adding remote control
diff --git a/e2e/reportApmMetrics.spec.js b/e2e/reportApmMetrics.spec.js
@@ -33,6 +33,11 @@ test('should report apm metrics', function (t) {
     TRACE_API_KEY_TEST,
     42,
     Number.MAX_SAFE_INTEGER)
+  serviceMocks.mockControlRequest(
+    TRACE_COLLECTOR_API_URL,
+    TRACE_API_KEY_TEST,
+    42,
+    Number.MAX_SAFE_INTEGER)
   serviceMocks.mockApmMetricsRequest(
     TRACE_COLLECTOR_API_URL,
     TRACE_API_KEY_TEST,
diff --git a/e2e/reportHttpRequest.spec.js b/e2e/reportHttpRequest.spec.js
@@ -38,6 +38,11 @@ test('should report http requests', function (t) {
     TRACE_API_KEY_TEST,
     42,
     Number.MAX_SAFE_INTEGER)
+  serviceMocks.mockControlRequest(
+    TRACE_COLLECTOR_API_URL,
+    TRACE_API_KEY_TEST,
+    42,
+    Number.MAX_SAFE_INTEGER)
   serviceMocks.mockHttpTransactionRequest(
     TRACE_COLLECTOR_API_URL,
     TRACE_API_KEY_TEST,
diff --git a/e2e/utils/serviceMocks.js b/e2e/utils/serviceMocks.js
@@ -64,6 +64,19 @@ function mockIncomingEdgeMetricsRequest (url, apiKey, serviceKey, maxTimes, call
     .reply(callback || 200)
 }
 
+function mockControlRequest (url, apiKey, serviceKey, maxTimes, callback) {
+  return nock(url, {
+    reqheaders: {
+      'Authorization': 'Bearer ' + apiKey
+    }
+  })
+    .post('/service/42/control')
+    .times(maxTimes)
+    .reply(callback || 200, {
+      commands: []
+    })
+}
+
 function mockHttpTransactionRequest (url, apiKey, callback) {
   return nock(url, {
     reqheaders: {
@@ -80,5 +93,6 @@ module.exports = {
   mockRpmMetricsRequest: mockRpmMetricsRequest,
   mockExternalEdgeMetricsRequest: mockExternalEdgeMetricsRequest,
   mockIncomingEdgeMetricsRequest: mockIncomingEdgeMetricsRequest,
-  mockHttpTransactionRequest: mockHttpTransactionRequest
+  mockHttpTransactionRequest: mockHttpTransactionRequest,
+  mockControlRequest: mockControlRequest
 }
diff --git a/lib/agent/api/index.js b/lib/agent/api/index.js
@@ -17,6 +17,8 @@ function CollectorApi (options) {
   this.COLLECTOR_API_INCOMING_EDGE_METRICS = url.resolve(options.collectorApiUrl, options.collectorApiIncomingEdgeMetricsEndpoint)
   this.COLLECTOR_API_EXTERNAL_EDGE_METRICS = url.resolve(options.collectorApiUrl, options.collectorApiExternalEdgeMetricsEndpoint)
   this.COLLECTOR_API_HEALTHCHECK = url.resolve(options.collectorApiUrl, options.collectorApiHealthcheckEndpoint)
+  this.COLLECTOR_API_PROFILER_MEMORY_HEAPDUMP = url.resolve(options.collectorApiUrl, options.collectorApiProfilerMemoryHeapdumpEndpoint)
+  this.COLLECTOR_API_CONTROL = url.resolve(options.collectorApiUrl, options.collectorApiControlEndpoint)
 
   this.collectorLanguage = options.collectorLanguage
   this.apiKey = options.apiKey
@@ -48,10 +50,12 @@ CollectorApi.prototype._withInstanceInfo = function (data) {
   }, data)
 }
 
-CollectorApi.prototype._send = function (destinationUrl, data) {
+CollectorApi.prototype._send = function (destinationUrl, data, callback) {
   var opts = url.parse(destinationUrl)
   var payload = JSON.stringify(data)
 
+  callback = callback || function () {}
+
   var req = https.request({
     hostname: opts.hostname,
     port: opts.port,
@@ -66,12 +70,13 @@ CollectorApi.prototype._send = function (destinationUrl, data) {
     }
   }, function (res) {
     res.setEncoding('utf8')
-    res.pipe(bl(function (err) {
+    res.pipe(bl(function (err, result) {
       if (err) {
         debug('There was an error when connecting to the Trace API', err)
         return
       }
 
+      callback(null, result)
       debug('HTTP Traces sent successfully')
     }))
   })
@@ -81,6 +86,7 @@ CollectorApi.prototype._send = function (destinationUrl, data) {
   req.on('error', function (error) {
     console.error('error: [trace]', 'There was an error connecting to the Trace servers. Make sure your servers can reach', opts.hostname)
     debug('error connecting to the Trace servers', error)
+    callback(error)
   })
   req.write(payload)
   req.end()
@@ -116,6 +122,16 @@ CollectorApi.prototype.sendApmMetrics = function (data) {
   this._send(url, this._withInstanceInfo(data))
 }
 
+CollectorApi.prototype.sendMemorySnapshot = function (data) {
+  if (!isNumber(this.serviceKey)) {
+    debug('Service id not present, cannot send heapdump')
+    return
+  }
+
+  var url = util.format(this.COLLECTOR_API_PROFILER_MEMORY_HEAPDUMP, this.serviceKey)
+  this._send(url, this._withInstanceInfo(data))
+}
+
 CollectorApi.prototype.sendExternalEdgeMetrics = function (data) {
   if (!isNumber(this.serviceKey)) {
     debug('Service id not present, cannot send metrics')
@@ -139,6 +155,29 @@ CollectorApi.prototype.sendIncomingEdgeMetrics = function (data) {
   }))
 }
 
+CollectorApi.prototype.getUpdates = function (data, callback) {
+  if (!isNumber(this.serviceKey)) {
+    debug('Service id not present, cannot get updates')
+    return
+  }
+
+  var url = util.format(this.COLLECTOR_API_CONTROL, this.serviceKey)
+
+  this._send(url, this._withInstanceInfo({
+    latestCommandId: data.latestCommandId
+  }), function (err, response) {
+    if (err) {
+      return callback(err)
+    }
+
+    try {
+      callback(null, JSON.parse(response.toString('utf8')))
+    } catch (ex) {
+      return callback(ex)
+    }
+  })
+}
+
 CollectorApi.prototype.sendSamples = function (samples, sync) {
   var url = this.COLLECTOR_API_SAMPLE
   var metadata = {
diff --git a/lib/agent/api/index.spec.js b/lib/agent/api/index.spec.js
@@ -22,6 +22,8 @@ describe('The Trace CollectorApi module', function () {
     collectorApiIncomingEdgeMetricsEndpoint: '/service/%s/edge-incoming',
     collectorApiExternalEdgeMetricsEndpoint: '/service/%s/edge-external',
     collectorApiHealthcheckEndpoint: '/service/%s/healthcheck',
+    collectorApiProfilerMemoryHeapdumpEndpoint: '/service/%s/memory-heapdump',
+    collectorApiControlEndpoint: '/service/%s/control',
     system: {
       hostname: 'test.org',
       processVersion: '4.3.1',
diff --git a/lib/agent/control/control.js b/lib/agent/control/control.js
@@ -0,0 +1,40 @@
+var debug = require('debug')('risingstack/trace')
+
+function Control (options) {
+  var _this = this
+  this.collectorApi = options.collectorApi
+  this.controlBus = options.controlBus
+  this.config = options.config
+  this.updateInterval = this.config.updateInterval
+  this.latestCommandId = undefined
+
+  this.getUpdates()
+
+  this.interval = setInterval(function () {
+    _this.getUpdates()
+  }, this.updateInterval)
+}
+
+Control.prototype.getUpdates = function () {
+  var _this = this
+  this.collectorApi.getUpdates({
+    latestCommandId: _this.latestCommandId
+  }, function (err, result) {
+    if (err) {
+      return debug(err)
+    }
+
+    _this.latestCommandId = result.latestCommandId
+
+    result.commands = result.commands || []
+    result.commands.forEach(function (command) {
+      _this.controlBus.emit(command.command, command.data)
+    })
+  })
+}
+
+function create (options) {
+  return new Control(options)
+}
+
+module.exports.create = create
diff --git a/lib/agent/control/control.spec.js b/lib/agent/control/control.spec.js
@@ -0,0 +1,35 @@
+var expect = require('chai').expect
+
+var Control = require('./')
+
+describe('The Control module', function () {
+  it('interprets commands', function () {
+    var collectorApi = {
+      getUpdates: function (options, cb) {
+        cb(null, {
+          commands: [{
+            id: 123,
+            command: 'memory-heapdump'
+          }]
+        })
+      }
+    }
+
+    var controlBus = {
+      on: function () {
+
+      },
+      emit: this.sandbox.spy()
+    }
+
+    var profiler = Control.create({
+      collectorApi: collectorApi,
+      controlBus: controlBus,
+      config: {
+        updateInterval: 1
+      }
+    })
+    profiler.getUpdates()
+    expect(controlBus.emit).to.be.calledWith('memory-heapdump', undefined)
+  })
+})
diff --git a/lib/agent/control/index.js b/lib/agent/control/index.js
@@ -0,0 +1 @@
+module.exports = require('./control')
diff --git a/lib/agent/index.js b/lib/agent/index.js
@@ -1,7 +1,9 @@
+var EventEmitter = require('events').EventEmitter
+
 var debug = require('debug')('risingstack/trace')
-var microtime = require('../optionalDependencies/microtime')
 var uuid = require('node-uuid')
 var cls = require('continuation-local-storage')
+var microtime = require('../optionalDependencies/microtime')
 
 var assign = require('lodash.assign')
 
@@ -10,6 +12,10 @@ var Metrics = require('./metrics')
 var Healthcheck = require('./healthcheck')
 var consts = require('../consts')
 var ReservoirSampler = require('./reservoir_sampler')
+var Profiler = require('./profiler')
+var Control = require('./control')
+
+var controlBus = new EventEmitter()
 
 var REQUEST_ID = 'request-id'
 var PARENT_COMM_ID = 'parent-comm-id'
@@ -55,6 +61,18 @@ function Agent (options) {
     config: this.config
   })
 
+  this.memoryProfiler = Profiler.memory.create({
+    collectorApi: this.collectorApi,
+    config: this.config,
+    controlBus: controlBus
+  })
+
+  this.control = Control.create({
+    collectorApi: this.collectorApi,
+    config: this.config,
+    controlBus: controlBus
+  })
+
   this.collectorApi.getService(function (err, serviceKey) {
     if (err) {
       return debug(err.message)
diff --git a/lib/agent/index.spec.js b/lib/agent/index.spec.js
@@ -34,7 +34,8 @@ describe('The Trace agent', function () {
     responseTime = 100
     collectorApi = {
       sendSamples: this.sandbox.spy(),
-      getService: this.sandbox.spy()
+      getService: this.sandbox.spy(),
+      getUpdates: this.sandbox.spy()
     }
     rpmMetrics = {
       addResponseTime: this.sandbox.spy(),
diff --git a/lib/agent/profiler/index.js b/lib/agent/profiler/index.js
@@ -0,0 +1 @@
+module.exports.memory = require('./memory')
diff --git a/lib/agent/profiler/memory/index.js b/lib/agent/profiler/memory/index.js
@@ -0,0 +1,46 @@
+var os = require('os')
+var fs = require('fs')
+
+var debug = require('debug')('risingstack/trace')
+
+var heapdump = require('../../../optionalDependencies/heapdump')
+
+function MemoryProfiler (options) {
+  var _this = this
+  this.collectorApi = options.collectorApi
+  this.controlBus = options.controlBus
+
+  this.controlBus.on('memory-heapdump', function (data) {
+    _this.sendSnapshot()
+  })
+}
+
+MemoryProfiler.prototype.sendSnapshot = function () {
+  var _this = this
+  var tempdir = os.tmpdir()
+  var now = Date.now()
+  var heapSnapshotName = now + '.heapsnapshot'
+  var snapshotPath = tempdir + '/' + heapSnapshotName
+  heapdump.writeSnapshot(snapshotPath, function (err) {
+    if (err) {
+      return debug('Error writing the heapsnapshot', err)
+    }
+
+    fs.readFile(snapshotPath, 'utf-8', function (err, data) {
+      if (err) {
+        return debug('Error reading the heapsnapshot', err)
+      }
+
+      _this.collectorApi.sendMemorySnapshot({
+        heapSnapshot: data,
+        time: now
+      })
+    })
+  })
+}
+
+function create (options) {
+  return new MemoryProfiler(options)
+}
+
+module.exports.create = create
diff --git a/lib/agent/profiler/memory/index.spec.js b/lib/agent/profiler/memory/index.spec.js
@@ -0,0 +1,44 @@
+var expect = require('chai').expect
+var os = require('os')
+var fs = require('fs')
+var EventEmitter = require('events').EventEmitter
+
+var heapdump = require('../../../optionalDependencies/heapdump')
+
+var MemoryProfiler = require('./')
+
+describe('The Memory Profiler module', function () {
+  it('sends heapdump', function () {
+    var msgBus = new EventEmitter()
+    var collectorApi = {
+      sendMemorySnapshot: this.sandbox.spy()
+    }
+
+    var snapShotContent = 'very-snapshot'
+    var now = 1234
+
+    var profiler = MemoryProfiler.create({
+      collectorApi: collectorApi,
+      controlBus: msgBus
+    })
+
+    this.sandbox.stub(heapdump, 'writeSnapshot', function (path, cb) {
+      cb()
+    })
+
+    this.sandbox.stub(Date, 'now', function () {
+      return now
+    })
+
+    this.sandbox.stub(fs, 'readFile', function (path, encoding, cb) {
+      cb(null, snapShotContent)
+    })
+
+    profiler.sendSnapshot()
+
+    expect(collectorApi.sendMemorySnapshot).to.be.calledWith({
+      heapSnapshot: snapShotContent,
+      time: now
+    })
+  })
+})
diff --git a/lib/config.js b/lib/config.js
@@ -1,6 +1,7 @@
 var config = {}
 
 config.collectInterval = 2 * 60 * 1000
+config.updateInterval = 15 * 1000
 config.healthcheckInterval = 10 * 1000
 
 config.collectorLanguage = 'nodejs'
@@ -13,6 +14,8 @@ config.collectorApiEdgeMetricsEndpoint = '/service/%s/edge-metrics'
 config.collectorApiIncomingEdgeMetricsEndpoint = '/service/%s/edge-incoming'
 config.collectorApiExternalEdgeMetricsEndpoint = '/service/%s/edge-external'
 config.collectorApiHealthcheckEndpoint = '/service/%s/healthcheck'
+config.collectorApiProfilerMemoryHeapdumpEndpoint = '/service/%s/memory-heapdump'
+config.collectorApiControlEndpoint = '/service/%s/control'
 
 config.configPath = 'trace.config'
 
diff --git a/lib/index.js b/lib/index.js
diff --git a/lib/optionalDependencies/heapdump.js b/lib/optionalDependencies/heapdump.js
diff --git a/lib/utils/configReader.js b/lib/utils/configReader.js
diff --git a/package.json b/package.json
