Merge pull request #258 from sematext/sc-9544-journald-fix

bin/logagent-setup.sh

@@ -295,26 +295,19 @@ input:
     # 100MB pipe buffer
     maxBuffer: 100000000
 
-# lowercasing field names
-parser:
-  json:
-    enabled: true
-    transform: !!js/function >
-      function (sourceName, parsed) {
-        var keys = Object.keys(parsed)
-        keys.forEach(function(key) {
-          parsed[key.toLowerCase()] = parsed[key]
-          delete parsed[key]
-        })
-      }
-
 # here we parse journald logs and remove extra fields
 outputFilter:
   journald-format:
     module: journald-format
     # Run Logagent parser for the message field
     parseMessageField: true
 
+  lowercase-fields:
+    module: lowercase-fields
+    # JS regular expression to match log source name
+    matchSource: !!js/regexp .*
+    allFields: true
+
   removeFields:
     module: remove-fields
     # JS regular expression to match log source name

bin/logagent.js

@@ -81,6 +81,7 @@ var moduleAlias = {
   'ip-truncate-fields': '../lib/plugins/output-filter/ip-truncate-fields.js',
   'remove-fields': '../lib/plugins/output-filter/remove-fields.js',
   'rename-fields': '../lib/plugins/output-filter/rename-fields.js',
+  'lowercase-fields': '../lib/plugins/output-filter/lowercase-fields.js',
   'drop-events': '../lib/plugins/output-filter/dropEventsFilter.js',
   'docker-enrichment': '../lib/plugins/output-filter/docker-log-enrichment.js',
   'kubernetes-enrichment':

config/examples/command-input-journald-output-es.yml

@@ -1,30 +1,62 @@
+options:
+  # print stats every 60 seconds 
+  printStats: 60
+  # Enable/disable GeoIP lookups
+  geoipEnabled: false
+  # Directory to store Logagent status and temporary files
+  # this is equals to LOGS_TMP_DIR env variable 
+  diskBufferDir: /tmp/sematext-logagent
+  debug: true
+  suppress: false
+
 input:
-  journald-json: 
+  journald-json:
     module: command
+    # note journalctl -u unitName can filter logs for systemd-units
     command: journalctl -o json --since="$QUERY_TIME"
     sourceName: journald
 
-    # date format for replacing $QUERY_TIME and $NOW
-    # the following works for journalctl, the default is ISO 8601
-    dateFormat: YYYY-MM-DD HH:mm:ss
-
+    dateFormat: YYYY-MM-DD HH:mm:ss # date format for $QUERY_TIME and $NOW
+    restart: 1 # seconds to wait between runs
     # where to persist last $QUERY_TIME
-    # defaults to os.tmpdir()+'logagentLastQueryTimeFile'
-    # lastQueryTimeFile: /tmp/logagentLastQueryTimeFile
+    lastQueryTimeFile: /var/run/logagentLastQueryTimeFile
 
-    # value for $QUERY_TIME if nothing was persisted. Default below
-    # initialQueryTime: "2001-01-01 00:00:00"
+    # pull logs from one week ago initially
+    initialQueryTime: "$ONE_WEEK_AGO"
+    # 100MB pipe buffer
+    maxBuffer: 100000000
 
-    # memory for the pipe between the command and Logagent
-    # it should fit the maximum size of the command's stdout/stderr
-    # size in bytes. Default below
-    # maxBuffer: 50000000
+# here we parse journald logs and remove extra fields
+outputFilter:
+  journald-format:
+    module: journald-format
+    # Run Logagent parser for the message field
+    parseMessageField: true
 
-    # after finishing the command, wait for N seconds then run it again
-    restart: 1
+  lowercase-fields:
+    module: lowercase-fields # this fliter only lowercases root fields, not nested fields
+    # JS regular expression to match log source name
+    matchSource: !!js/regexp .*
+    allFields: true # this will lowercase all root fields
+    # fields:
+    #   - fieldName: SELINUX_CONTEXT # this will lowercase only certain root fields, this is an array and you can specify multiple fields
+
+  removeFields:
+    module: remove-fields
+    # JS regular expression to match log source name
+    matchSource: !!js/regexp .*
+    # Note: journald format converts to lower case
+    fields:
+      - __cursor
+      - __monotonic_timestamp
+      - _transport
 
 output:
-  es-local:
+  stdout: yaml
+
+  # index logs in Elasticsearch or Sematext Logs
+  elasticsearch: 
     module: elasticsearch
-    url: http://localhost:9200
-    index: journald_logs
+    url: url
+    # default index (Logs token) to use:
+    index: token

config/examples/output-filter-lowercase-fields.yml

@@ -0,0 +1,16 @@
+outputFilter:
+  lowercase-fields:
+    module: lowercase-fields
+    # JS regular expression to match log source name
+    matchSource: !!js/regexp .*
+    allFields: true # this will lowercase all fields
+    fields:
+      - fieldName: SELINUX_CONTEXT # this will lowercase only certain fields, this is an array and you can specify multiple fields
+
+# Exmple, input: 
+# {"USER": "{ name: root }", "SELINUX_CONTEXT": "unconfined", "message": "Client connect: root", "originalLine": "INFO Client connect: root"}
+# Example, output
+# {"user": "{ name: root }", "selinux_context": "unconfined", "message": "Client connect: root", "originalLine": "INFO Client connect: root"}
+
+# Note:
+# This filter will only lowercase root fields, not any nested field names!

lib/plugins/output-filter/journald-format.js

@@ -134,7 +134,7 @@ function applySematextCommonSchema (
             Object.assign(log[type], data)
           }
           callback(null, log)
-        }.bind(this)
+        }
       )
     } else {
       callback(null, log)

lib/plugins/output-filter/lowercase-fields.js

@@ -0,0 +1,43 @@
+const get = require('get-value')
+const unset = require('unset-value')
+const set = require('set-value')
+
+function renameFields (context, config, eventEmitter, data, callback) {
+  if (data === undefined) {
+    return callback(new Error('data is null'), null)
+  }
+  if (config.allFields !== true) {
+    config.allFields = false
+  }
+
+  try {
+    if (!config.matchSource.test(context.sourceName || data.logSource)) {
+      return callback(null, data)
+    }
+
+    if (config.allFields === true) {
+      const fields = Object.keys(data)
+      fields.forEach(field => {
+        const lowerCaseField = field.toLowerCase()
+        const fieldValue = get(data, field)
+        unset(data, field)
+        set(data, lowerCaseField, fieldValue)
+      })
+      return callback(null, data)
+    }
+
+    const fields = config.fields
+    fields.forEach(field => {
+      const { fieldName } = field
+      const lowerCaseField = fieldName.toLowerCase()
+      const fieldValue = get(data, fieldName)
+      unset(data, fieldName)
+      set(data, lowerCaseField, fieldValue)
+    })
+
+    callback(null, data)
+  } catch (ex) {
+    callback(ex, null)
+  }
+}
+module.exports = renameFields

patterns.yml

@@ -640,7 +640,7 @@ patterns:
     - type: flink[job|task]manager
       regex: !!js/regexp /^(\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2},\d{3})\s(INFO|info|WARN|ERROR|FATAL)\s+([\S]+)\s+-\s([\S|\s|\n]+)/
       fields: [ts, severity, processid, message]
-      dateFormat: YYYY-MM-DD HH:mm:ss:l
+      dateFormat: yyyy-MM-dd HH:mm:ss:l
  - # Apache Flink jobmanager and taskmanager
   sourceName: !!js/regexp /flink/
   blockStart: !!js/regexp /^\w{3}\s[\d|,|\s|\:]+/