Skip to content

Commit 88c9921

Browse files
megastefmegastef
committed
use Sematext common schema, lowercase field names
1 parent 29bb7a2 commit 88c9921

1 file changed

Lines changed: 97 additions & 45 deletions

File tree

‎lib/plugins/input/journaldUpload.js‎

Lines changed: 97 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,60 @@ const extractTokenRegEx = /([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9
55
const tokenFormatRegEx = /[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}/
66
const maxIndexingErrors = 1 // failures before blacklisting tokens
77
var LRU = require('lru-cache')
8+
// journald fields for lowercase access
9+
const __CURSOR = '__CURSOR'
10+
const __REALTIME_TIMESTAMP = '__realtime_timestamp'
11+
const __SOURCE_REALTIME_TIMESTAMP = '_source_realtime_timestamp'
12+
const SYSLOG_IDENTIFIER = 'syslog_identifier'
13+
const _HOSTNAME = '_hostname'
14+
const _SYSTEMD_UNIT = '_systemd_unit'
15+
const PRIORITY = 'priority'
16+
const SYSLOG_FACILITY = 'syslog_facility'
17+
// field mapping for Sematext Common Schema
18+
const processFields = {
19+
'_pid': 'pid',
20+
'_uid': 'uid',
21+
'_gid': 'gid',
22+
'_cmdline': 'cmd',
23+
'_systemd_cgroup': 'cgroup'
24+
}
25+
// mapping for syslog priority and facility values
26+
const SEVERITY = [
27+
'emerg',
28+
'alert',
29+
'crit',
30+
'err',
31+
'warning',
32+
'notice',
33+
'info',
34+
'debug'
35+
]
36+
const FACILITY = [
37+
'kern',
38+
'user',
39+
'mail',
40+
'daemon',
41+
'auth',
42+
'syslog',
43+
'lpr',
44+
'news',
45+
'uucp',
46+
'cron',
47+
'authpriv',
48+
'ftp',
49+
'ntp',
50+
'logaudit',
51+
'logalert',
52+
'clock',
53+
'local0',
54+
'local1',
55+
'local2',
56+
'local3',
57+
'local4',
58+
'local5',
59+
'local6',
60+
'local7'
61+
]
862

963
function JournaldUpload (config, eventEmitter) {
1064
this.config = config
@@ -14,10 +68,12 @@ function JournaldUpload (config, eventEmitter) {
1468
max: 5000,
1569
maxAge: 10 * 60 * 1000
1670
})
17-
71+
if (config.useSematextCommonSchema === undefined) {
72+
config.useSematextCommonSchema = true
73+
}
1874
if (config.removeFields) {
1975
for (var i = 0; i < config.removeFields.length; i++) {
20-
this.removeFields[config.removeFields[i]] = true
76+
this.removeFields[config.removeFields[i].toLowerCase()] = true
2177
}
2278
}
2379
// set default filter
@@ -45,7 +101,7 @@ function JournaldUpload (config, eventEmitter) {
45101
self.invalidTokens.set(match[1], self.invalidTokens.get(match[1]))
46102
}
47103
if (self.invalidTokens.get(match[1]) >= maxIndexingErrors) {
48-
consoleLogger.log(`Invalid token added to blacklist ${match[1]}`)
104+
consoleLogger.log(`Invalid token added to blacklist ${match[1]}`)
49105
}
50106
}
51107
})
@@ -66,31 +122,44 @@ JournaldUpload.prototype.stop = function (cb) {
66122
cb()
67123
}
68124

125+
// Transforming fields to Sematext Common Schema https://sematext.com/docs/tags/common-schema/
69126
// see https://sematext.com/docs/agents/sematext-agent/processes/metadata/
70127
JournaldUpload.prototype.applySematextCommonSchema = function (log) {
71-
var fields = {
72-
'_PID': 'pid',
73-
'_UID': 'uid',
74-
'_GID': 'gid',
75-
'_CMDLINE': 'cmd'
128+
// use Sematext common schema os.host = hostname
129+
let hostname = log[_HOSTNAME]
130+
if (hostname) {
131+
log.os = {host: hostname}
132+
delete log[_HOSTNAME]
133+
}
134+
let timestamp = log[__REALTIME_TIMESTAMP] || log[__SOURCE_REALTIME_TIMESTAMP]
135+
if (timestamp) {
136+
var d = new Date(Number(fieldValue) / 1000)
137+
if (d instanceof Date && !isNaN(d)) {
138+
log['@timestamp'] = d
139+
}
76140
}
77-
if (log['_PID']){
141+
let prio = log[PRIORITY]
142+
let facility = log[log[SYSLOG_FACILITY]]
143+
// handling syslog priorit and facility values
144+
if (prio || facility) {
145+
log.facility = FACILITY[facility] || String(facility)
146+
log.severity = SEVERITY[prio] || String(prio)
147+
}
148+
if (log['_pid']) {
78149
log.process = {}
79150
} else {
80-
return
151+
return
81152
}
82-
for (field in fields) {
153+
for (field in processFields) {
83154
if (log[field]) {
84-
if (!log.process){
85-
log.process = {}
86-
}
87-
log.process[fields[field]] = log[field]
155+
log.process[processFields[field]] = log[field]
88156
delete log[field]
89157
}
90158
}
91159
}
160+
92161
JournaldUpload.prototype.emitEvent = function (log, token) {
93-
let systemdUnit = log['_SYSTEMD_UNIT']
162+
let systemdUnit = log[_SYSTEMD_UNIT]
94163
let config = this.config
95164
if (systemdUnit) {
96165
if (config.systemdUnitFilter !== undefined && config.systemdUnitFilter.include && !config.systemdUnitFilter.include.test(log['_SYSTEMD_UNIT'])) {
@@ -101,16 +170,18 @@ JournaldUpload.prototype.emitEvent = function (log, token) {
101170
}
102171
}
103172
this.addTags(log)
104-
let context = {
105-
sourceName: log['_SYSTEMD_UNIT'] || log['SYSLOG_IDENTIFIER'] || 'journald',
173+
let context = {
174+
sourceName: log[_SYSTEMD_UNIT] || log[SYSLOG_IDENTIFIER] || 'journald',
106175
name: 'journald'
107176
}
108-
177+
109178
if (token) {
110179
// set index, for elasticsearch output plugin
111180
context.index = token
112181
}
113-
this.applySematextCommonSchema(log)
182+
if (this.config.useSematextCommonSchema) {
183+
this.applySematextCommonSchema(log)
184+
}
114185
this.eventEmitter.emit('data.object', log, context)
115186
}
116187

@@ -148,33 +219,15 @@ JournaldUpload.prototype.parseBody = function (body, token) {
148219
let lines = body.split('\n')
149220
let log = null
150221
for (var i = 0; i < lines.length; i++) {
151-
if (lines[i].indexOf('__CURSOR') > -1) {
222+
if (lines[i].indexOf(__CURSOR) > -1) {
152223
log = {}
153224
}
154225
var index = lines[i].indexOf('=')
155226
if (index > -1 && lines[i].length > 0) {
156227
var fieldName = lines[i].substr(0, index)
157228
var fieldValue = lines[i].substr(index + 1, lines[i].length)
158-
if (fieldName === '__REALTIME_TIMESTAMP' || fieldName === '_SOURCE_REALTIME_TIMESTAMP') {
159-
var d = new Date(Number(fieldValue) / 1000)
160-
if (d instanceof Date && !isNaN(d)) {
161-
log['@timestamp'] = d
162-
}
163-
}
164-
// use Sematext common schema os.host = hostname
165-
if (fieldName === '_HOSTNAME') {
166-
log.os = {host: fieldValue}
167-
}
168-
// use Sematext common schema process.pid = _PID
169-
if (fieldName === '_PID') {
170-
log.process = {pid: fieldValue}
171-
}
172-
if (fieldName === 'MESSAGE') {
173-
log['message'] = fieldValue
174-
} else {
175-
if (!self.removeFields[fieldName]) {
176-
log[fieldName] = fieldValue
177-
}
229+
if (!self.removeFields[fieldName]) {
230+
log[fieldName.toLowerCase()] = fieldValue
178231
}
179232
} else {
180233
if (lines[i] === '' && log !== null) {
@@ -208,15 +261,14 @@ JournaldUpload.prototype.journaldHttpHandler = function (req, res) {
208261
}
209262
}
210263

211-
if ((self.config.useIndexFromUrlPath === true && !token) || self.invalidTokens.get(token) >= maxIndexingErrors) {
264+
if ((self.config.useIndexFromUrlPath === true && !token) || self.invalidTokens.get(token) >= maxIndexingErrors) {
212265
res.statusCode = 401
213-
res.end(`invalid logs token in url ${req.url}\n`)
266+
res.end(`invalid logs token in url ${req.url}`)
214267
return
215268
}
216-
269+
217270
req.on('data', function (data) {
218271
bodyIn += String(data)
219-
// onsole.log(data)
220272
})
221273

222274
req.on('end', function endHandler () {

0 commit comments

Comments
 (0)