feat(api): gzip encoding

Author: dszakallas

lib/agent/api/bufferStream.js

@@ -0,0 +1,28 @@
+var stream = require('stream')
+var util = require('util')
+
+function BufferStream (source) {
+  stream.Readable.call(this)
+  this._source = source
+  this._offset = 0
+  this._length = source.length
+  this.on('end', this._destroy)
+}
+
+util.inherits(BufferStream, stream.Readable)
+
+BufferStream.prototype._destroy = function () {
+  this._source = null
+}
+
+BufferStream.prototype._read = function (size) {
+  if (this._offset < this._length) {
+    this.push(this._source.slice(this._offset, (this._offset + size)))
+    this._offset += size
+  }
+  if (this._offset >= this._length) {
+    this.push(null)
+  }
+}
+
+module.exports = BufferStream

lib/agent/api/index.js

@@ -7,9 +7,11 @@ var debug = require('debug')('risingstack/trace:agent:api')
 var assign = require('lodash.assign')
 var HttpsProxyAgent = require('https-proxy-agent')
 var stringify = require('json-stringify-safe')
+var BufferStream = require('./bufferStream')
 
 var bl = require('bl')
 var libPackage = require('../../../package')
+var zlib = require('zlib')
 
 function CollectorApi (options) {
   this.COLLECTOR_API_SERVICE = url.resolve(options.collectorApiUrl, options.collectorApiServiceEndpoint)
@@ -66,22 +68,34 @@ CollectorApi.prototype._send = function (destinationUrl, data, callback, options
   var opts = url.parse(destinationUrl)
   var payload = stringify(data)
 
+  options = options || { }
+
   callback = callback || function () {}
 
+  var headers = {
+    'Authorization': 'Bearer ' + this.apiKey,
+    'Content-Type': 'application/json',
+    'X-Reporter-Version': libPackage.version,
+    'X-Reporter-Language': this.collectorLanguage
+  }
+
+  assign(headers, options.headers)
+
+  if (options.compress) {
+    headers['Content-Encoding'] = 'gzip'
+    // headers['Content-Type'] = 'application/octet-stream'
+  } else {
+    headers['Content-Length'] = Buffer.byteLength(payload)
+  }
+
   var requestOptions = {
     hostname: opts.hostname,
     port: opts.port,
     path: opts.path,
     method: 'POST',
     // if the proxy is not set, it will fallback to the default agent
     agent: this.proxyAgent,
-    headers: assign({
-      'Authorization': 'Bearer ' + this.apiKey,
-      'Content-Type': 'application/json',
-      'X-Reporter-Version': libPackage.version,
-      'X-Reporter-Language': this.collectorLanguage,
-      'Content-Length': Buffer.byteLength(payload)
-    }, options && options.headers)
+    headers: headers
   }
 
   var req = https.request(requestOptions, function (res) {
@@ -102,8 +116,14 @@ CollectorApi.prototype._send = function (destinationUrl, data, callback, options
     debug('#_send', '[Error] Connection error', error)
     callback(error)
   })
-  req.write(payload)
-  req.end()
+
+  if (options.compress) {
+    var stream = new BufferStream(new Buffer(payload))
+    stream.pipe(zlib.createGzip()).pipe(req)
+  } else {
+    req.write(payload)
+    req.end()
+  }
 }
 
 CollectorApi.prototype.sendRpmMetrics = function (data) {
@@ -281,8 +301,6 @@ CollectorApi.prototype.getService = function (cb) {
       }
       if (res.statusCode > 399) {
         debug('#getService', '[Error] Service responded with', res.statusCode)
-        console.log(setTimeout)
-        console.log(retryInterval)
         return setTimeout(function () {
           debug('#getService', 'Retrying with %d ms', retryInterval)
           console.log('adsfsadf')

lib/agent/control/control.js

@@ -1,4 +1,4 @@
-var debug = require('debug')('risingstack/trace')
+var debug = require('debug')('risingstack/trace:agent:control')
 
 var inherits = require('util').inherits
 var Agent = require('../agent')
@@ -24,7 +24,7 @@ Control.prototype.getUpdates = function () {
     latestCommandId: _this.latestCommandId
   }, function (err, result) {
     if (err) {
-      return debug(err)
+      return debug('#getUpdates', '[Error]', err)
     }
 
     _this.latestCommandId = result.latestCommandId

lib/agent/tracer/collector.js

@@ -412,7 +412,7 @@ Collector.prototype.end = function (briefcase, options) {
 }
 
 Collector.prototype.getTransactionId = function (briefcase) {
-  return briefcase.communication && briefcase.communication.transactionId
+  return briefcase && briefcase.communication && briefcase.communication.transactionId
 }
 
 module.exports = Collector

lib/agent/tracer/index.js

@@ -41,14 +41,17 @@ Tracer.prototype.send = function (isSync) {
       s: { n: this.serviceName, k: this.serviceKey },
       e: events
     }
+    data
     if (isSync === true) {
       this._api._sendSync(this.sampleUrl, data, {
+        compress: true,
         headers: {
           'x-document-schema': 'CollectorTransactionRaw:1.0.0'
         }
       })
     } else {
       this._api._send(this.sampleUrl, data, null, {
+        compress: true,
         headers: {
           'x-document-schema': 'CollectorTransactionRaw:1.0.0'
         }

lib/instrumentations/core/http/request.js

@@ -1,10 +1,11 @@
-var debug = require('debug')('risingstack/trace')
+var debug = require('debug')('risingstack/trace:agent:instrumentations')
 var url = require('url')
 var microtime = require('../../../optionalDependencies/microtime')
 
 var util = require('./util')
+var format = require('util').format
 
-function wrapRequest (originalHttpRequest, agent) {
+function httpClient (originalHttpRequest, agent) {
   var whiteListHosts = agent.getConfig().whiteListHosts
 
   return function wrappedRequest (requestParams) {
@@ -41,22 +42,16 @@ function wrapRequest (originalHttpRequest, agent) {
 
     var csCtx = cs.briefcase.csCtx
 
-    debug('trace event (cs); reqId: %s, child-id: %s',
-      csCtx.transactionId,
-      csCtx.communicationId
-    )
-
     // decorate headers
     requestParams.headers = requestParams.headers || {}
 
     if (cs.duffelBag.transactionId) {
       requestParams.headers['request-id'] = cs.duffelBag.transactionId
     }
 
-    if (collector.LEVELS.gte(cs.duffelBag.severity, collector.mustCollectSeverity)) {
-      debug('trace event (cs); reqId: %s, child commId: %s must collect',
-        csCtx.transactionId,
-        csCtx.communicationId)
+    var mustCollect = collector.LEVELS.gte(cs.duffelBag.severity, collector.mustCollectSeverity)
+
+    if (mustCollect) {
       requestParams.headers['x-must-collect'] = '1'
     }
 
@@ -68,13 +63,13 @@ function wrapRequest (originalHttpRequest, agent) {
     requestParams.headers['x-client-send'] = String(cs.duffelBag.timestamp)
     requestParams.headers['x-span-id'] = cs.duffelBag.communicationId
 
+    debug('#httpClient', format('CS(%s) [must-collect: %s]', csCtx.communicationId, mustCollect))
+
     var returned = originalHttpRequest.apply(this, arguments)
 
     // returns with error
     returned.on('error', function (err) {
-      debug('trace event (cr) on error; reqId: %s, child commId: %s must collect',
-      csCtx.transactionId,
-      csCtx.communicationId)
+      debug('#httpClient', format('CS(%s) network error: %s', csCtx.communicationId, err.code))
       collector.networkError(cs.briefcase, err)
     })
 
@@ -95,6 +90,8 @@ function wrapRequest (originalHttpRequest, agent) {
         targetServiceKey: serviceKey
       }
 
+      debug('#httpClient', format('CR(%s) [severity: %s]', csCtx.communicationId, severity))
+
       collector.clientRecv({
         protocol: 'http',
         status: incomingMessage.statusCode > 399 ? 'bad' : 'ok',
@@ -118,10 +115,10 @@ function wrapRequest (originalHttpRequest, agent) {
       }
     })
 
-    agent.storage.bindNew(returned)
+    agent.storage.bind(returned)
 
     return returned
   }
 }
 
-module.exports = wrapRequest
+module.exports = httpClient

lib/instrumentations/core/http/request.spec.js

@@ -60,7 +60,7 @@ describe('The http.request wrapper module', function () {
         get: this.sandbox.stub().returns({
           communication: {}
         }),
-        bindNew: this.sandbox.spy()
+        bind: this.sandbox.spy()
       }
     }
   })

lib/instrumentations/core/http/server.js

@@ -1,5 +1,5 @@
 var microtime = require('../../../optionalDependencies/microtime')
-var debug = require('debug')('risingstack/trace')
+var debug = require('debug')('risingstack/trace:agent:instrumentations')
 var Shimmer = require('../../../utils/shimmer')
 
 var util = require('./util')
@@ -32,7 +32,7 @@ function isStatusCodeIgnored (ignoreStatusCodes, statusCode) {
   return false
 }
 
-function wrapListener (listener, agent) {
+function httpServer (listener, agent) {
   var config = agent.getConfig()
   var collector = agent.tracer.collector
   var incomingEdgeMetrics = agent.incomingEdgeMetrics
@@ -50,7 +50,6 @@ function wrapListener (listener, agent) {
     })
 
     if (isSkippedHeader || isPathIgnored(ignorePaths, requestUrl)) {
-      debug('trace event (sr); request skipped because of ignore options', headers)
       return listener.apply(this, arguments)
     }
 
@@ -119,6 +118,8 @@ function wrapListener (listener, agent) {
             if (collector.LEVELS.gte(ss.duffelBag.severity, collector.mustCollectSeverity)) {
               response.setHeader('x-must-collect', '1')
             }
+          } else {
+            debug('#httpServer', '[Warning] failed to create SS, cannot intrument headers')
           }
           agent.rpmMetrics.addResponseTime(ssTime - srTime)
           agent.rpmMetrics.addStatusCode(response.statusCode)
@@ -137,4 +138,4 @@ function wrapListener (listener, agent) {
   }
 }
 
-module.exports = wrapListener
+module.exports = httpServer

lib/instrumentations/express.js

@@ -1,4 +1,4 @@
-var debug = require('debug')('risingstack/trace')
+var debug = require('debug')('risingstack/trace:agent:instrumentations')
 var Shimmer = require('../utils/shimmer')
 
 module.exports = function (express, agent) {
@@ -10,7 +10,7 @@ module.exports = function (express, agent) {
 
   // for now support only express@4
   if (!isVersion4) {
-    debug('express version is not supported, not wrapping error handler')
+    debug('#express', 'express version is not supported, not wrapping error handler')
     return express
   }
 

lib/instrumentations/pg.js

@@ -1,17 +1,17 @@
 var semver = require('semver')
-var debug = require('debug')('risingstack/trace')
+var debug = require('debug')('risingstack/trace:agent:instrumentations')
 var consts = require('../consts')
 var Shimmer = require('../utils/shimmer')
 var utils = require('./utils')
 
 function wrapNative (native, agent, version) {
   // We support it only versions gte 4.0.0
   if (!version) {
-    debug('trace: warning: Cannot determine postgres version. ' +
+    debug('#postgres', 'cannot determine postgres version. ' +
       '(No package.json?) Native queries won\'t be instrumented.'
     )
   } else if (!semver.satisfies(version, '>= 4.0.0')) {
-    debug('trace: warning: You are using node-postgres version ' +
+    debug('#postgres', 'you are using node-postgres version ' +
       version + ', (<4.0.0) so native queries won\'t be instrumented.'
     )
   } else {