feat(edges): add generic topology support

e2e/reportApmMetrics.spec.js

@@ -23,6 +23,14 @@ test('should report apm metrics', function (t) {
     function (uri, requestBody) {
       t.pass('collector sent rpm metrics')
     })
+  serviceMocks.mockEdgeMetricsRequest(
+    TRACE_COLLECTOR_API_URL,
+    TRACE_API_KEY_TEST,
+    TRACE_SERVICE_KEY_TEST,
+    1,
+    function (uri, requestBody) {
+      t.pass('collector sent edge metrics')
+    })
   serviceMocks.mockApmMetricsRequest(
     TRACE_COLLECTOR_API_URL,
     TRACE_API_KEY_TEST,

e2e/reportHttpRequest.spec.js

@@ -27,6 +27,11 @@ test('should report http requests', function (t) {
     TRACE_API_KEY_TEST,
     42,
     Number.MAX_SAFE_INTEGER) // set the times parameter high so the http mock catches all
+  serviceMocks.mockEdgeMetricsRequest(
+    TRACE_COLLECTOR_API_URL,
+    TRACE_API_KEY_TEST,
+    42,
+    Number.MAX_SAFE_INTEGER)
   serviceMocks.mockHttpTransactionRequest(
     TRACE_COLLECTOR_API_URL,
     TRACE_API_KEY_TEST,

e2e/utils/serviceMocks.js

@@ -42,6 +42,17 @@ function mockRpmMetricsRequest (url, apiKey, serviceKey, maxTimes, callback) {
     .reply(callback || 200)
 }
 
+function mockEdgeMetricsRequest (url, apiKey, serviceKey, maxTimes, callback) {
+  return nock(url, {
+    reqheaders: {
+      'Authorization': 'Bearer ' + apiKey
+    }
+  })
+    .post('/service/42/edge-metrics')
+    .times(maxTimes)
+    .reply(callback || 200)
+}
+
 function mockHttpTransactionRequest (url, apiKey, callback) {
   return nock(url, {
     reqheaders: {
@@ -56,5 +67,6 @@ module.exports = {
   mockServiceKeyRequest: mockServiceKeyRequest,
   mockApmMetricsRequest: mockApmMetricsRequest,
   mockRpmMetricsRequest: mockRpmMetricsRequest,
+  mockEdgeMetricsRequest: mockEdgeMetricsRequest,
   mockHttpTransactionRequest: mockHttpTransactionRequest
 }

lib/agent/api/index.js

@@ -15,6 +15,7 @@ function CollectorApi (options) {
   this.COLLECTOR_API_SERVICE = url.resolve(options.collectorApiUrl, options.collectorApiServiceEndpoint)
   this.COLLECTOR_API_METRICS = url.resolve(options.collectorApiUrl, options.collectorApiApmMetricsEndpoint)
   this.COLLECTOR_API_RPM_METRICS = url.resolve(options.collectorApiUrl, options.collectorApiRpmMetricsEndpoint)
+  this.COLLECTOR_API_EDGE_METRICS = url.resolve(options.collectorApiUrl, options.collectorApiEdgeMetricsEndpoint)
 
   this.apiKey = options.apiKey
   this.processId = options.processId
@@ -38,6 +39,13 @@ CollectorApi.prototype.sendSync = function (data) {
   })
 }
 
+CollectorApi.prototype._sendWithPid = function (destinationUrl, data) {
+  this._send(destinationUrl, assign({
+    hostname: this.hostname,
+    pid: this.processId
+  }, data))
+}
+
 CollectorApi.prototype._send = function (destinationUrl, data) {
   var opts = url.parse(destinationUrl)
   var payload = JSON.stringify(data)
@@ -77,7 +85,7 @@ CollectorApi.prototype.sendRpmMetrics = function (data) {
   }
 
   var url = util.format(this.COLLECTOR_API_RPM_METRICS, this.serviceKey)
-  this._send(url, assign({ hostname: this.hostname, pid: this.processId }, data))
+  this._sendWithPid(url, data)
 }
 
 CollectorApi.prototype.sendApmMetrics = function (data) {
@@ -87,12 +95,22 @@ CollectorApi.prototype.sendApmMetrics = function (data) {
   }
 
   var url = util.format(this.COLLECTOR_API_METRICS, this.serviceKey)
-  this._send(url, assign({ hostname: this.hostname, pid: this.processId }, data))
+  this._sendWithPid(url, data)
+}
+
+CollectorApi.prototype.sendEdgeMetrics = function (data) {
+  if (isNaN(this.serviceKey)) {
+    debug('Service id not present, cannot send metrics')
+    return
+  }
+
+  var url = util.format(this.COLLECTOR_API_EDGE_METRICS, this.serviceKey)
+  this._sendWithPid(url, data)
 }
 
 CollectorApi.prototype.sendSamples = function (data) {
   var url = this.COLLECTOR_API_SAMPLE
-  this._send(url, assign({ hostname: this.hostname, pid: this.processId }, data))
+  this._sendWithPid(url, data)
 }
 
 CollectorApi.prototype._getRetryInterval = function () {

lib/agent/api/index.spec.js

@@ -17,7 +17,8 @@ describe('The Trace CollectorApi module', function () {
     collectorApiApmMetricsEndpoint: '/service/%s/apm-metrics',
     collectorApiRpmMetricsEndpoint: '/service/%s/rpm-metrics',
     hostname: 'test.org',
-    processId: '7777'
+    processId: '7777',
+    collectorApiEdgeMetricsEndpoint: '/service/%s/edge-metrics'
   }
 
   it('can be instantiated w/ serviceName and apiKey', function () {

lib/agent/index.js

@@ -40,6 +40,11 @@ function Agent (options) {
     config: this.config
   })
 
+  this.edgeMetrics = Metrics.edge.create({
+    collectorApi: this.collectorApi,
+    config: this.config
+  })
+
   this.collectorApi.getService(function (err, serviceKey) {
     if (err) {
       return debug(err.message)
@@ -83,7 +88,7 @@ Agent.prototype.serverReceive = function (data) {
   span.parent = parentId
   span.events.push({
     id: spanId,
-    time: data.time || microtime.now(),
+    time: data.time || this.getMicrotime(),
     type: Agent.SERVER_RECV
   })
 }
@@ -100,7 +105,7 @@ Agent.prototype.serverSend = function (data) {
   span.statusCode = data.statusCode
   span.events.push({
     id: spanId,
-    time: data.time || microtime.now(),
+    time: data.time || this.getMicrotime(),
     type: Agent.SERVER_SEND
   })
 
@@ -123,7 +128,7 @@ Agent.prototype.clientSend = function (data) {
     return
   }
 
-  data.time = data.time || microtime.now()
+  data.time = data.time || this.getMicrotime()
 
   if (data.err) {
     span.isForceSampled = true
@@ -156,11 +161,23 @@ Agent.prototype.clientSend = function (data) {
 Agent.prototype.clientReceive = function (data) {
   var span = this.findSpan(data.id)
 
+  this.edgeMetrics.report({
+    targetHost: data.host,
+    targetServiceKey: data.targetServiceKey,
+    protocol: data.protocol,
+    networkDelay: {
+      incoming: data.networkDelayIncoming,
+      outgoing: data.networkDelayOutgoing
+    },
+    status: data.status,
+    responseTime: data.responseTime
+  })
+
   if (!span) {
     return
   }
 
-  data.time = data.time || microtime.now()
+  data.time = data.time || this.getMicrotime()
 
   span.events.push({
     host: data.host,
@@ -183,7 +200,7 @@ Agent.prototype.onCrash = function (data) {
   span.isForceSampled = true
 
   span.events.push({
-    time: microtime.now(),
+    time: this.getMicrotime(),
     id: spanId,
     type: 'err',
     data: {
@@ -214,7 +231,7 @@ Agent.prototype.report = function (name, userData) {
 
   var dataToSend = {
     id: spanId,
-    time: microtime.now(),
+    time: this.getMicrotime(),
     type: 'us',
     data: {
       name: name,
@@ -249,7 +266,7 @@ Agent.prototype.reportError = function (errorName, error) {
 
   var dataToSend = {
     id: spanId,
-    time: microtime.now(),
+    time: this.getMicrotime(),
     type: 'us',
     data: {
       name: errorName,
@@ -354,6 +371,10 @@ Agent.prototype.generateId = function () {
   return uuid.v4()
 }
 
+Agent.prototype.getMicrotime = function () {
+  return microtime.now()
+}
+
 Agent.prototype._send = function (options) {
   debug('sending logs to the trace service')
   if (this.spans.length > 0) {

lib/agent/metrics/edge/index.js

@@ -0,0 +1,119 @@
+var consts = require('../../../consts')
+
+function EdgeMetrics (options) {
+  var _this = this
+  this.collectorApi = options.collectorApi
+  this.config = options.config
+  this.collectInterval = this.config.collectInterval
+
+  // metrics
+  this.metrics = {}
+
+  this.interval = setInterval(function () {
+    _this.sendMetrics()
+  }, this.collectInterval)
+}
+
+EdgeMetrics.prototype.initHost = function (data) {
+  if (!this.metrics[data.protocol][data.targetHost]) {
+    this.metrics[data.protocol][data.targetHost] = {
+      targetServiceKey: data.targetServiceKey,
+      responseTime: [],
+      networkDelayIncoming: [],
+      networkDelayOutgoing: [],
+      status: {
+        ok: 0,
+        notOk: 0
+      }
+    }
+  }
+  return this.metrics[data.protocol][data.targetHost]
+}
+
+EdgeMetrics.prototype.initProtocol = function (data) {
+  if (!this.metrics[data.protocol]) {
+    this.metrics[data.protocol] = {}
+  }
+  return this.metrics[data.protocol]
+}
+
+EdgeMetrics.prototype.report = function (data) {
+  this.initProtocol(data)
+  var edge = this.initHost(data)
+
+  if (data.networkDelay.incoming) {
+    edge.networkDelayIncoming.push(data.networkDelay.incoming)
+  }
+
+  if (data.networkDelay.outgoing) {
+    edge.networkDelayOutgoing.push(data.networkDelay.outgoing)
+  }
+
+  edge.responseTime.push(data.responseTime)
+
+  if (data.status === consts.EDGE_STATUS.OK) {
+    edge.status.ok += 1
+  } else if (data.status === consts.EDGE_STATUS.NOT_OK) {
+    edge.status.notOk += 1
+  }
+}
+
+EdgeMetrics.prototype.calculateTimes = function (items) {
+  var sorted = items.sort(function (a, b) {
+    return a - b
+  })
+
+  var medianElementIndex = Math.round(sorted.length / 2) - 1
+  var ninetyFiveElementIndex = Math.round(sorted.length * 0.95) - 1
+
+  return {
+    median: sorted[medianElementIndex],
+    ninetyFive: sorted[ninetyFiveElementIndex]
+  }
+}
+
+EdgeMetrics.prototype.sendMetrics = function () {
+  var _this = this
+
+  var metrics = Object.keys(this.metrics).map(function (protocol) {
+    var targetHosts = Object.keys(_this.metrics[protocol]).map(function (hostName) {
+      var host = _this.metrics[protocol][hostName]
+      return {
+        name: hostName,
+        metrics: {
+          targetServiceKey: host.targetServiceKey,
+          responseTime: _this.calculateTimes(host.responseTime),
+          networkDelayIncoming: _this.calculateTimes(host.networkDelayIncoming),
+          networkDelayOutgoing: _this.calculateTimes(host.networkDelayOutgoing),
+          status: {
+            ok: host.status.ok,
+            notOk: host.status.notOk
+          }
+        }
+      }
+    })
+
+    return {
+      protocol: protocol,
+      targetHosts: targetHosts
+    }
+  })
+
+  this.metrics = {}
+
+  // if no metrics, don't send anything
+  if (!metrics || !metrics.length) {
+    return
+  }
+
+  this.collectorApi.sendEdgeMetrics({
+    timestamp: (new Date()).toISOString(),
+    hostMetrics: metrics
+  })
+}
+
+function create (options) {
+  return new EdgeMetrics(options)
+}
+
+module.exports.create = create