feat: logs processing speed increased

Author: makeros

bench/bench.sh

@@ -0,0 +1,2 @@
+#!/bin/bash
+time node producer.js 1000000 | node --trace-uncaught ../index.js log -c ../__mocks__/custom-schema.json -v

bench/producer.js

@@ -0,0 +1,5 @@
+'strict'
+
+for (let i = 0; i < process.argv.slice(2)[0]; i++) {
+  process.stdout.write('{"level":30,"time":1531171074631,"msg":"hello world","nested":{"mock":666},"test2":"red","pid":657,"hostname":"box","name":"app","v":1}' + '\n')
+}

lib/gelf-transformer.js

@@ -1,10 +1,15 @@
 const fastJsonParse = require('fast-json-parse')
-const pipeline = require('./pipeline')
-const pump = require('pump')
+const processLogs = require('./process-logs')
+const streamLogs = require('./stream-logs')
 const split = require('split2')
+const { pipeline } = require('readable-stream')
 
 module.exports = function (opts) {
-  const pipe = pipeline(opts)
-  pump(process.stdin, split(fastJsonParse), pipe)
+  pipeline(
+    process.stdin,
+    split(fastJsonParse),
+    processLogs(opts),
+    ...streamLogs(opts)
+  )
   process.on('SIGINT', function () { process.exit(0) })
 }

lib/pipeline.js

@@ -0,0 +1,30 @@
+const transformer = require('./transformer')
+const { stringify } = require('./utils')
+const { Transform } = require('readable-stream')
+
+module.exports = function processLogs (opts) {
+  const transform = transformer(opts)
+  const stringified = stringify(opts)
+
+  return new Transform({
+    writableObjectMode: true,
+    decodeStrings: false,
+    transform: function (data, enc, cb) {
+      if (data.value !== null) {
+        this.push(stringified(transform(data.value)))
+      }
+      cb()
+    },
+    construct (callback) {
+      this.data = ''
+      callback()
+    },
+    flush (callback) {
+      try {
+        this.push('')
+      } catch (err) {
+        callback(err)
+      }
+    }
+  })
+}

lib/process-logs.js

@@ -0,0 +1,39 @@
+const Transport = require('./transport')
+const { Transform } = require('readable-stream')
+
+module.exports = function processLogs (opts) {
+  const streams = []
+  if (opts.verbose) {
+    streams.push(getVerbose())
+  }
+  if (opts.transportToGraylog) {
+    streams.push(getTransport(opts))
+  }
+  return streams
+}
+
+function getVerbose () {
+  return new Transform({
+    writableObjectMode: true,
+    decodeStrings: false,
+    transform: function (data, enc, cb) {
+      setImmediate(function () { process.stdout.write(data + '\n') })
+      cb()
+    }
+  })
+}
+
+function getTransport (opts) {
+  console.log('transport is prepared')
+  const transport = new Transport(opts)
+  return new Transform({
+    writableObjectMode: true,
+    decodeStrings: false,
+    transform: function (data, enc, cb) {
+      setImmediate(function () {
+        transport.emit('log', data)
+      })
+      cb()
+    }
+  })
+}

lib/stream-logs.js

@@ -1,10 +1,11 @@
 const get = require('lodash.get')
 
-module.exports = function (data, customSchema) {
-  return Object.entries(customSchema.properties)
-    .reduce((acc, entry) => {
-      const sourcePath = entry[1].source ? entry[1].source : entry[0]
-      acc[entry[0]] = get(data, sourcePath, undefined)
-      return acc
-    }, {})
+module.exports = function customGelf (properties) {
+  return function (data) {
+    const _data = {}
+    for (const property in properties) {
+      _data[property] = get(data, properties[property].source, data[property])
+    }
+    return _data
+  }
 }

lib/transformer/custom-gelf.js

@@ -3,18 +3,17 @@ const buildCustomGelf = require('./custom-gelf')
 
 module.exports = function (opts) {
   const sanitizeData = setMessageField(opts)
-
-  return function (data) {
-    return {
-      ...buildStandardGelf(sanitizeData(data)),
-      ...opts.customSchema ? buildCustomGelf(data, opts.customSchema) : {}
-    }
+  const customProperties = opts.customSchema ? opts.customSchema.properties : {}
+  const build = buildCustomGelf(customProperties)
+  return function (value) {
+    return Object.assign(buildStandardGelf(sanitizeData(value)), build(value))
   }
 }
 
 function setMessageField (opts) {
+  const _field = opts.messageField
   return function (data) {
-    data.msg = data[opts.messageField] ? data[opts.messageField] : 'No msg property found or msg is empty'
+    data.msg = data[_field] !== undefined ? data[_field] : 'No msg property found or msg is empty'
     return data
   }
 }

lib/transformer/index.js

@@ -7,15 +7,13 @@ const EventEmitter = require('events').EventEmitter
 const pick = require('lodash.pick')
 const utils = require('./utils')
 
-var Transport = function (opts) {
+const Transport = function (opts) {
   const self = this
 
   self.config = pick(opts, ['host', 'port', 'maxChunkSize'])
   self.stringify = utils.stringify(opts)
 
-  self.on('log', function (gelf) {
-    const msg = self.stringify(gelf)
-
+  self.on('log', function (msg) {
     self.compress(msg, function (buffer) {
       self.processMessage(buffer)
     })
@@ -39,7 +37,7 @@ Transport.prototype.prepareDatagrams = function (chunks, cb) {
   const length = chunks.length
 
   const createDatagramArray = function (msgId) {
-    for (var i = 0; i < chunks.length; i++) {
+    for (let i = 0; i < chunks.length; i++) {
       datagrams[i] = Buffer.from(gelfBytes.concat(msgId, i, length, chunks[i]))
     }
 
@@ -98,7 +96,7 @@ Transport.prototype.processMessage = function (msg) {
 Transport.prototype.sendMultipleMessages = function (datagrams) {
   const self = this
 
-  for (var i = 0; i < datagrams.length; i++) {
+  for (let i = 0; i < datagrams.length; i++) {
     self.sendMessage(datagrams[i])
   }
 }

lib/transport.js

@@ -11,20 +11,15 @@ const mapPinoToSyslogLevel = {
   50: syslogLevel.error,
   60: syslogLevel.critical
 }
-function pinoLevelToSyslogLevel (level) {
+
+exports.pinoLevelToSyslogLevel = function pinoLevelToSyslogLevel (level) {
   return mapPinoToSyslogLevel[level] || syslogLevel.critical
 }
 
-const stringify = function (opts) {
-  const schema = standardSchema
+exports.stringify = function stringify (opts) {
   if (opts.customSchema) {
-    schema.properties = Object.assign(schema.properties, opts.customSchema.properties)
+    Object.assign(standardSchema.properties, opts.customSchema.properties)
   }
 
-  return fastJsonStringify(schema)
-}
-
-module.exports = {
-  pinoLevelToSyslogLevel,
-  stringify
+  return fastJsonStringify(standardSchema)
 }