Merge pull request #269 from sematext/sc-10353-ecs-improvements

Author: Adnan Rahić

config/examples/aws-ecs-input-es-output.yml

@@ -5,7 +5,7 @@ input:
     module: input-aws-ecs
     port: 6666
     useIndexFromUrlPath: true
-    workers: 4
+    # workers: 4
 
 outputFilter: 
   aws-ecs-format: 
@@ -16,3 +16,4 @@ output:
   sematext-cloud:
     module: elasticsearch
     url: https://logsene-receiver.sematext.com
+    # debug: true

config/examples/file-input-json-parse-es-output.yml

@@ -0,0 +1,17 @@
+options:
+  includeOriginalLine: false
+
+input:
+  files:
+    - '/var/log'
+
+# this parses the message when it is a structured JSON
+parser:
+  json: 
+    enabled: true
+
+output:
+  sematext:
+    module: elasticsearch
+    url: https://logsene-receiver.sematext.com
+    index: <add your logs token>

lib/plugins/input/aws-ecs.js

@@ -4,6 +4,7 @@ const throng = require('throng')
 const extractTokenRegEx = /([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})/
 const tokenFormatRegEx = /[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}/
 const TokenBlacklist = require('../../util/token-blacklist.js')
+const zlib = require('zlib')
 
 class AwsEcs {
   constructor (config, eventEmitter) {
@@ -97,7 +98,6 @@ class AwsEcs {
     if (body.length === 0) {
       return
     }
-
     const self = this
     const docs = JSON.parse(body)
     if (docs && docs.length > 0) {
@@ -127,6 +127,8 @@ class AwsEcs {
       const path = req.url.split('/')
       let token = null
       let bodyIn = ''
+      const buffer = []
+
       if (self.config.useIndexFromUrlPath === true && path.length > 1) {
         if (path[1] && path[1].length > 31 && tokenFormatRegEx.test(path[1])) {
           const match = path[1].match(extractTokenRegEx)
@@ -147,25 +149,56 @@ class AwsEcs {
         res.end(`invalid logs token in url ${req.url}`)
         return
       }
-      req.on('data', function (data) {
-        bodyIn += String(data)
-      })
-      req.on('end', function endHandler () {
-        try {
-          self.parseRes(req.headers, bodyIn, token)
-        } catch (err) {
-          if (self.config.debug) {
-            consoleLogger.error('Error in ECS HttpHandler: ' + err)
+
+      const isGzip = req.headers['content-encoding'] === 'gzip'
+      // if the JSON content is gzipped this will unzip it and parse it into a stringified JSON string
+      if (isGzip) {
+        const gunzip = zlib.createGunzip()
+        req.pipe(gunzip)
+
+        gunzip.on('data', function onGzipHandler (data) {
+          buffer.push(data.toString())
+        })
+        gunzip.on('end', function endGzipHandler () {
+          try {
+            const body = buffer.join('')
+            self.parseRes(req.headers, body, token)
+          } catch (err) {
+            if (self.config.debug) {
+              consoleLogger.error('Error in ECS HttpHandler: ' + err)
+            }
+
+            res.writeHead(500, { 'Content-Type': 'text/plain' })
+            res.end(`Invalid gzip input: ${err}\n`)
+            return
           }
+          // send response to client
+          res.writeHead(200, { 'Content-Type': 'text/plain' })
+          res.end('OK\n')
+        })
 
-          res.writeHead(500, { 'Content-Type': 'text/plain' })
-          res.end(`Invalid json input: ${err}\n`)
-          return
-        }
-        // send response to client
-        res.writeHead(200, { 'Content-Type': 'text/plain' })
-        res.end('OK\n')
-      })
+        return
+      } else {
+        req.on('data', function onHandler (data) {
+          bodyIn += String(data)
+        })
+        req.on('end', function endHandler () {
+          try {
+            self.parseRes(req.headers, bodyIn, token)
+          } catch (err) {
+            if (self.config.debug) {
+              consoleLogger.error('Error in ECS HttpHandler: ' + err)
+            }
+
+            res.writeHead(500, { 'Content-Type': 'text/plain' })
+            res.end(`Invalid json input: ${err}\n`)
+            return
+          }
+          // send response to client
+          res.writeHead(200, { 'Content-Type': 'text/plain' })
+          res.end('OK\n')
+        })
+      }
     } catch (err) {
       res.statusCode = 500
       res.end()

lib/plugins/output-filter/aws-ecs-format.js

@@ -65,13 +65,13 @@ function formatAwsEcs (context, config, eventEmitter, log, callback) {
         }
 
         if (data && data._type) {
-          const type = `${data._type}`
           delete data['@timestamp']
           delete data.logSource
           delete data._type
-          log.type = type
-          log[type] = {}
-          Object.assign(log[type], data)
+
+          log.type = 'ecs'
+          log[log.source] = {}
+          Object.assign(log[log.source], data)
         }
         return callback(null, log)
       })