logs-management: Add option to submit logs to system journal (#16485)
* Add systemd-journal logging of FLB_TAIL, FLB_SERIAL, FLB_WEB_LOG, FLB_DOCKER_EV and FLB_MQTT. * Refactor flb_collect_logs_cb() to use switch instead of if * Add configurable prefix to systemd journal fields generated by logs-management.plugin * Add configuration option to enable submission to system journal for each (applicable) log source type * Add serial bitrate as system journal field * Don't fail to build if systemd-devel is missing * Add log source and log type as system journal fields * Update README.md * Add prio_level config option for kernel logs * Add missing config option 'Min_Bytes' to FLB_SERIAL input
Dimitris P committed
Nov 30, 2023 at 10:53 UTC
94607d7d4bef6a7707d3f476c064baf97672ee29
15 files changed
+448
-268
Makefile.am
+1
@@ -1411,6 +1411,7 @@ if ENABLE_LOGSMANAGEMENT_TESTS
1411
endif
1412
logs_management_plugin_LDADD = \
1413
$(NETDATA_COMMON_LIBS) \
1414
+ $(OPTIONAL_SYSTEMD_LIBS) \
1415
$(NULL)
1416
endif
1417
logsmanagement/README.md
+3
-1
@@ -118,8 +118,9 @@ There are some fundamental configuration options that are common to all log coll
118
| `disk space limit` | Equivalent value in `logsmanagement.d.conf` (`500 MiB` by default). | Maximum disk space that all compressed logs in database can occupy (per log source). Once exceeded, oldest BLOB of logs will be truncated for new logs to be written over. Each log source database can contain a maximum of 10 BLOBs at any point, so each truncation equates to a deletion of about 10% of the oldest logs. The number of BLOBS will be configurable in a future release.
119
| `collected logs total chart enable` | Equivalent value in `logsmanagement.d.conf` (`no` by default). | Chart that shows the number of log records collected for this log source, since the last Netdata agent restart. Useful for debugging purposes.
120
| `collected logs rate chart enable` | Equivalent value in `logsmanagement.d.conf` (`yes` by default). | Chart that shows the rate that log records are collected at for this log source.
121
+| `submit logs to system journal = no` | Equivalent value in `logsmanagement.d.conf` (`no` by default). Available only for `flb_tail`, `flb_web_log`, `flb_serial`, `flb_docker_events` and `flb_mqtt`. | If enabled, it will submit the collected logs to the system journal.
122
122
-There are also one setting that cannot be set per log source, but can only be defined in `logsmanagement.d.conf`:
123
+There is also one setting that cannot be set per log source, but can only be defined in `logsmanagement.d.conf`:
124
125
| Configuration Option | Default | Description |
126
| :------------: | :------------: | ------------ |
@@ -152,6 +153,7 @@ This collector will collect logs from the kernel message log buffer. See also do
153
154
| Configuration Option | Description |
155
| :------------: | ------------ |
156
+| `prio level` | Drop kernel logs with priority higher than `prio level`. Default value is 8, so no logs will be dropped.
157
| `severity chart` | Enable chart showing Syslog Severity values of collected logs. Severity values are in the range of 0 to 7 inclusive.|
158
| `subsystem chart` | Enable chart showing which subsystems generated the logs.|
159
| `device chart` | Enable chart showing which devices generated the logs.|
logsmanagement/defaults.h
+4
@@ -30,6 +30,10 @@
30
#endif
31
#define ENABLE_COLLECTED_LOGS_RATE_DEFAULT CONFIG_BOOLEAN_YES /**< Default value to enable (or not) metrics of rate of collected log records **/
32
33
+#define SD_JOURNAL_FIELD_PREFIX "LOGS_MANAG_" /**< Default systemd journal field prefix for sources that log to the system journal */
34
+
35
+#define SD_JOURNAL_SEND_DEFAULT CONFIG_BOOLEAN_NO /**< Default value to enable (or not) submission of logs to the system journal (where applicable) **/
36
+
37
/* -------------------------------------------------------------------------- */
38
39
logsmanagement/file_info.h
+7
-3
@@ -45,6 +45,9 @@ typedef struct flb_tail_config {
45
int use_inotify;
46
} Flb_tail_config_t;
47
48
+typedef struct flb_kmsg_config {
49
+ char *prio_level;
50
+} Flb_kmsg_config_t;
51
52
typedef struct flb_serial_config {
53
char *bitrate;
@@ -78,11 +81,9 @@ typedef struct flb_output_config {
81
} Flb_output_config_t;
82
83
struct File_info {
81
- /* TODO: Struct needs refactoring, as a lot of members take up memory that
82
- * is not used, depending on the type of the log source. */
84
85
/* Struct members core to any log source type */
85
- const char *chartname; /**< Top level chart name for this log source on web dashboard **/
86
+ const char *chartname; /**< Top level chart name for this log source on web dashboard **/
87
char *filename; /**< Full path of log source **/
88
const char *file_basename; /**< Basename of log source **/
89
const char *stream_guid; /**< Streaming input GUID **/
@@ -93,6 +94,7 @@ struct File_info {
94
int update_every; /**< Interval (in sec) of how often to collect and update charts **/
95
int update_timeout; /**< Timeout to update charts after, since last update */
96
int use_log_timestamp; /**< Use log timestamps instead of collection timestamps, if available **/
97
+ int do_sd_journal_send; /**< Write to system journal - not applicable to all log source types **/
98
struct Chart_meta *chart_meta;
99
LOG_SRC_STATE state; /**< State of log source, used to sync status among threads **/
100
@@ -154,6 +156,8 @@ typedef struct {
156
int buff_flush_to_db_interval;
157
int enable_collected_logs_total;
158
int enable_collected_logs_rate;
159
+ char *sd_journal_field_prefix;
160
+ int do_sd_journal_send;
161
} g_logs_manag_config_t;
162
163
extern g_logs_manag_config_t g_logs_manag_config;
logsmanagement/flb_plugin.c
+359
-258
@@ -17,6 +17,13 @@
17
#include "../fluent-bit/lib/monkey/include/monkey/mk_core/mk_list.h"
18
#include <dlfcn.h>
19
20
+#ifdef HAVE_SYSTEMD
21
+#include <systemd/sd-journal.h>
22
+#define SD_JOURNAL_SEND_DEFAULT_FIELDS \
23
+ "%s_LOG_SOURCE=%s" , sd_journal_field_prefix, log_src_t_str[p_file_info->log_source], \
24
+ "%s_LOG_TYPE=%s" , sd_journal_field_prefix, log_src_type_t_str[p_file_info->log_type]
25
+#endif
26
+
27
#define LOG_REC_KEY "msg" /**< key to represent log message field in most log sources **/
28
#define LOG_REC_KEY_SYSTEMD "MESSAGE" /**< key to represent log message field in systemd log source **/
29
#define SYSLOG_TIMESTAMP_SIZE 16
@@ -120,9 +127,13 @@ static void *flb_lib_handle = NULL;
127
128
static struct flb_lib_out_cb *fwd_input_out_cb = NULL;
129
130
+static const char *sd_journal_field_prefix = SD_JOURNAL_FIELD_PREFIX;
131
+
132
extern netdata_mutex_t stdout_mut;
133
125
-int flb_init(flb_srvc_config_t flb_srvc_config, const char *const stock_config_dir){
134
+int flb_init(flb_srvc_config_t flb_srvc_config,
135
+ const char *const stock_config_dir,
136
+ const char *const new_sd_journal_field_prefix){
137
int rc = 0;
138
char *dl_error;
139
@@ -198,6 +209,9 @@ int flb_init(flb_srvc_config_t flb_srvc_config, const char *const stock_config_d
209
rc = -1;
210
goto do_return;
211
}
212
+
213
+ if(new_sd_journal_field_prefix && *new_sd_journal_field_prefix)
214
+ sd_journal_field_prefix = new_sd_journal_field_prefix;
215
216
do_return:
217
freez(flb_lib_path);
@@ -403,7 +417,8 @@ static int flb_collect_logs_cb(void *record, size_t size, void *data){
417
// continue;
418
} while(p < pend);
419
}
406
- if(unlikely(p_file_info == NULL)) goto skip_collect_and_drop_logs;
420
+ if(unlikely(p_file_info == NULL))
421
+ goto skip_collect_and_drop_logs;
422
423
424
uv_mutex_lock(&p_file_info->flb_tmp_buff_mut);
@@ -412,295 +427,296 @@ static int flb_collect_logs_cb(void *record, size_t size, void *data){
427
428
p = x->via.map.ptr;
429
pend = x->via.map.ptr + x->via.map.size;
415
- do{
416
-
417
- /* FLB_TAIL, FLB_WEB_LOG and FLB_SERIAL case */
418
- if( p_file_info->log_type == FLB_TAIL ||
419
- p_file_info->log_type == FLB_WEB_LOG ||
420
- p_file_info->log_type == FLB_SERIAL){
421
- if( !strncmp(p->key.via.str.ptr, LOG_REC_KEY, (size_t) p->key.via.str.size) ||
422
- /* The following line is in case we collect systemd logs
423
- * (tagged as "MESSAGE") or docker_events (tagged as
424
- * "message") via a "Forward" source to an FLB_TAIL
425
- * parent. */
426
- !strncasecmp(p->key.via.str.ptr, LOG_REC_KEY_SYSTEMD, (size_t) p->key.via.str.size)){
427
-
428
- message = (char *) p->val.via.str.ptr;
429
- message_size = p->val.via.str.size;
430
-
431
- if(p_file_info->log_type == FLB_WEB_LOG){
432
- parse_web_log_line( (Web_log_parser_config_t *) p_file_info->parser_config->gen_config,
433
- message, message_size, &line_parsed);
434
-
435
- if(likely(p_file_info->use_log_timestamp)){
436
- timestamp = line_parsed.timestamp * MSEC_PER_SEC; // convert to msec from sec
437
-
438
- { /* ------------------ FIXME ------------------------
439
- * Temporary kludge so that metrics don't break when
440
- * a new record has timestamp before the current one.
441
- */
442
- static msec_t previous_timestamp = 0;
443
- if((((long long) timestamp - (long long) previous_timestamp) < 0))
444
- timestamp = previous_timestamp;
445
-
446
- previous_timestamp = timestamp;
430
+ do{
431
+ switch(p_file_info->log_type){
432
+
433
+ case FLB_TAIL:
434
+ case FLB_WEB_LOG:
435
+ case FLB_SERIAL:
436
+ {
437
+ if( !strncmp(p->key.via.str.ptr, LOG_REC_KEY, (size_t) p->key.via.str.size) ||
438
+ /* The following line is in case we collect systemd logs
439
+ * (tagged as "MESSAGE") or docker_events (tagged as
440
+ * "message") via a "Forward" source to an FLB_TAIL parent. */
441
+ !strncasecmp(p->key.via.str.ptr, LOG_REC_KEY_SYSTEMD, (size_t) p->key.via.str.size)){
442
+
443
+ message = (char *) p->val.via.str.ptr;
444
+ message_size = p->val.via.str.size;
445
+
446
+ if(p_file_info->log_type == FLB_WEB_LOG){
447
+ parse_web_log_line( (Web_log_parser_config_t *) p_file_info->parser_config->gen_config,
448
+ message, message_size, &line_parsed);
449
+
450
+ if(likely(p_file_info->use_log_timestamp)){
451
+ timestamp = line_parsed.timestamp * MSEC_PER_SEC; // convert to msec from sec
452
+
453
+ { /* ------------------ FIXME ------------------------
454
+ * Temporary kludge so that metrics don't break when
455
+ * a new record has timestamp before the current one.
456
+ */
457
+ static msec_t previous_timestamp = 0;
458
+ if((((long long) timestamp - (long long) previous_timestamp) < 0))
459
+ timestamp = previous_timestamp;
460
+
461
+ previous_timestamp = timestamp;
462
+ }
463
}
464
}
449
- }
450
-
451
- new_tmp_text_size = message_size + 1; // +1 for '\n'
465
453
- m_assert(message_size, "message_size is 0");
454
- m_assert(message, "message is NULL");
455
- }
456
- ++p;
457
- continue;
458
- } /* FLB_TAIL, FLB_WEB_LOG and FLB_SERIAL case end */
459
-
460
- /* FLB_KMSG case */
461
- if(p_file_info->log_type == FLB_KMSG){
462
- if(unlikely(skip_kmsg_log_buffering)){
463
- static time_t netdata_start_time = 0;
464
- if (!netdata_start_time) netdata_start_time = now_boottime_sec();
465
- if(now_boottime_sec() - netdata_start_time < KERNEL_LOGS_COLLECT_INIT_WAIT)
466
- goto skip_collect_and_drop_logs;
467
- else skip_kmsg_log_buffering = 0;
468
- }
466
+ new_tmp_text_size = message_size + 1; // +1 for '\n'
467
470
- /* NOTE/WARNING:
471
- * kmsg timestamps are tricky. The timestamp will be
472
- * *wrong** if the system has gone into hibernation since
473
- * last boot and "p_file_info->use_log_timestamp" is set.
474
- * Even if "p_file_info->use_log_timestamp" is NOT set, we
475
- * need to use now_realtime_msec() as Fluent Bit timestamp
476
- * will also be wrong. */
477
- if( !strncmp(p->key.via.str.ptr, "sec", (size_t) p->key.via.str.size)){
478
- if(p_file_info->use_log_timestamp){
479
- timestamp += (now_realtime_sec() - now_boottime_sec() + p->val.via.i64) * MSEC_PER_SEC;
468
+ m_assert(message_size, "message_size is 0");
469
+ m_assert(message, "message is NULL");
470
}
481
- else if(!timestamp)
482
- timestamp = now_realtime_msec();
483
- }
484
- else if(!strncmp(p->key.via.str.ptr, "usec", (size_t) p->key.via.str.size) &&
485
- p_file_info->use_log_timestamp){
486
- timestamp += p->val.via.i64 / USEC_PER_MS;
471
+
472
+ break;
473
}
488
- else if(!strncmp(p->key.via.str.ptr, LOG_REC_KEY, (size_t) p->key.via.str.size)){
489
- message = (char *) p->val.via.str.ptr;
490
- message_size = p->val.via.str.size;
474
492
- m_assert(message, "message is NULL");
493
- m_assert(message_size, "message_size is 0");
475
+ case FLB_KMSG:
476
+ {
477
+ if(unlikely(skip_kmsg_log_buffering)){
478
+ static time_t netdata_start_time = 0;
479
+ if (!netdata_start_time) netdata_start_time = now_boottime_sec();
480
+ if(now_boottime_sec() - netdata_start_time < KERNEL_LOGS_COLLECT_INIT_WAIT)
481
+ goto skip_collect_and_drop_logs;
482
+ else skip_kmsg_log_buffering = 0;
483
+ }
484
495
- new_tmp_text_size += message_size + 1; // +1 for '\n'
496
- }
497
- else if(!strncmp(p->key.via.str.ptr, "priority", (size_t) p->key.via.str.size)){
498
- kmsg_sever = (int) p->val.via.u64;
499
- }
500
- ++p;
501
- continue;
502
- } /* FLB_KMSG case end */
485
+ /* NOTE/WARNING:
486
+ * kmsg timestamps are tricky. The timestamp will be
487
+ * *wrong** if the system has gone into hibernation since
488
+ * last boot and "p_file_info->use_log_timestamp" is set.
489
+ * Even if "p_file_info->use_log_timestamp" is NOT set, we
490
+ * need to use now_realtime_msec() as Fluent Bit timestamp
491
+ * will also be wrong. */
492
+ if( !strncmp(p->key.via.str.ptr, "sec", (size_t) p->key.via.str.size)){
493
+ if(p_file_info->use_log_timestamp){
494
+ timestamp += (now_realtime_sec() - now_boottime_sec() + p->val.via.i64) * MSEC_PER_SEC;
495
+ }
496
+ else if(!timestamp)
497
+ timestamp = now_realtime_msec();
498
+ }
499
+ else if(!strncmp(p->key.via.str.ptr, "usec", (size_t) p->key.via.str.size) &&
500
+ p_file_info->use_log_timestamp){
501
+ timestamp += p->val.via.i64 / USEC_PER_MS;
502
+ }
503
+ else if(!strncmp(p->key.via.str.ptr, LOG_REC_KEY, (size_t) p->key.via.str.size)){
504
+ message = (char *) p->val.via.str.ptr;
505
+ message_size = p->val.via.str.size;
506
504
- /* FLB_SYSTEMD or FLB_SYSLOG case */
505
- if( p_file_info->log_type == FLB_SYSTEMD ||
506
- p_file_info->log_type == FLB_SYSLOG){
507
-
508
- if( p_file_info->use_log_timestamp && !strncmp( p->key.via.str.ptr,
509
- "SOURCE_REALTIME_TIMESTAMP",
510
- (size_t) p->key.via.str.size)){
507
+ m_assert(message, "message is NULL");
508
+ m_assert(message_size, "message_size is 0");
509
512
- m_assert(p->val.via.str.size - 3 == TIMESTAMP_MS_STR_SIZE - 1,
513
- "p->val.via.str.size - 3 != TIMESTAMP_MS_STR_SIZE");
514
-
515
- strncpyz(timestamp_str, p->val.via.str.ptr, (size_t) p->val.via.str.size);
510
+ new_tmp_text_size += message_size + 1; // +1 for '\n'
511
+ }
512
+ else if(!strncmp(p->key.via.str.ptr, "priority", (size_t) p->key.via.str.size)){
513
+ kmsg_sever = (int) p->val.via.u64;
514
+ }
515
517
- char *endptr = NULL;
518
- timestamp = str2ll(timestamp_str, &endptr);
519
- timestamp = *endptr ? 0 : timestamp / USEC_PER_MS;
516
+ break;
517
}
521
- else if(!strncmp(p->key.via.str.ptr, "PRIVAL", (size_t) p->key.via.str.size)){
522
- m_assert(p->val.via.str.size <= 3, "p->val.via.str.size > 3");
523
- strncpyz(syslog_prival, p->val.via.str.ptr, (size_t) p->val.via.str.size);
524
- syslog_prival_size = (size_t) p->val.via.str.size;
525
-
526
- m_assert(syslog_prival, "syslog_prival is NULL");
527
- }
528
- else if(!strncmp(p->key.via.str.ptr, "PRIORITY", (size_t) p->key.via.str.size)){
529
- m_assert(p->val.via.str.size <= 1, "p->val.via.str.size > 1");
530
- strncpyz(syslog_severity, p->val.via.str.ptr, (size_t) p->val.via.str.size);
531
-
532
- m_assert(syslog_severity, "syslog_severity is NULL");
533
- }
534
- else if(!strncmp(p->key.via.str.ptr, "SYSLOG_FACILITY", (size_t) p->key.via.str.size)){
535
- m_assert(p->val.via.str.size <= 2, "p->val.via.str.size > 2");
536
- strncpyz(syslog_facility, p->val.via.str.ptr, (size_t) p->val.via.str.size);
537
-
538
- m_assert(syslog_facility, "syslog_facility is NULL");
539
- }
540
- else if(!strncmp(p->key.via.str.ptr, "SYSLOG_TIMESTAMP", (size_t) p->key.via.str.size)){
541
- syslog_timestamp = (char *) p->val.via.str.ptr;
542
- syslog_timestamp_size = p->val.via.str.size;
518
544
- m_assert(syslog_timestamp, "syslog_timestamp is NULL");
545
- m_assert(syslog_timestamp_size, "syslog_timestamp_size is 0");
546
-
547
- new_tmp_text_size += syslog_timestamp_size;
548
- }
549
- else if(!strncmp(p->key.via.str.ptr, "HOSTNAME", (size_t) p->key.via.str.size)){
550
- hostname = (char *) p->val.via.str.ptr;
551
- hostname_size = p->val.via.str.size;
519
+ case FLB_SYSTEMD:
520
+ case FLB_SYSLOG:
521
+ {
522
+ if( p_file_info->use_log_timestamp && !strncmp( p->key.via.str.ptr,
523
+ "SOURCE_REALTIME_TIMESTAMP",
524
+ (size_t) p->key.via.str.size)){
525
+
526
+ m_assert(p->val.via.str.size - 3 == TIMESTAMP_MS_STR_SIZE - 1,
527
+ "p->val.via.str.size - 3 != TIMESTAMP_MS_STR_SIZE");
528
+
529
+ strncpyz(timestamp_str, p->val.via.str.ptr, (size_t) p->val.via.str.size);
530
+
531
+ char *endptr = NULL;
532
+ timestamp = str2ll(timestamp_str, &endptr);
533
+ timestamp = *endptr ? 0 : timestamp / USEC_PER_MS;
534
+ }
535
+ else if(!strncmp(p->key.via.str.ptr, "PRIVAL", (size_t) p->key.via.str.size)){
536
+ m_assert(p->val.via.str.size <= 3, "p->val.via.str.size > 3");
537
+ strncpyz(syslog_prival, p->val.via.str.ptr, (size_t) p->val.via.str.size);
538
+ syslog_prival_size = (size_t) p->val.via.str.size;
539
+
540
+ m_assert(syslog_prival, "syslog_prival is NULL");
541
+ }
542
+ else if(!strncmp(p->key.via.str.ptr, "PRIORITY", (size_t) p->key.via.str.size)){
543
+ m_assert(p->val.via.str.size <= 1, "p->val.via.str.size > 1");
544
+ strncpyz(syslog_severity, p->val.via.str.ptr, (size_t) p->val.via.str.size);
545
+
546
+ m_assert(syslog_severity, "syslog_severity is NULL");
547
+ }
548
+ else if(!strncmp(p->key.via.str.ptr, "SYSLOG_FACILITY", (size_t) p->key.via.str.size)){
549
+ m_assert(p->val.via.str.size <= 2, "p->val.via.str.size > 2");
550
+ strncpyz(syslog_facility, p->val.via.str.ptr, (size_t) p->val.via.str.size);
551
+
552
+ m_assert(syslog_facility, "syslog_facility is NULL");
553
+ }
554
+ else if(!strncmp(p->key.via.str.ptr, "SYSLOG_TIMESTAMP", (size_t) p->key.via.str.size)){
555
+ syslog_timestamp = (char *) p->val.via.str.ptr;
556
+ syslog_timestamp_size = p->val.via.str.size;
557
+
558
+ m_assert(syslog_timestamp, "syslog_timestamp is NULL");
559
+ m_assert(syslog_timestamp_size, "syslog_timestamp_size is 0");
560
+
561
+ new_tmp_text_size += syslog_timestamp_size;
562
+ }
563
+ else if(!strncmp(p->key.via.str.ptr, "HOSTNAME", (size_t) p->key.via.str.size)){
564
+ hostname = (char *) p->val.via.str.ptr;
565
+ hostname_size = p->val.via.str.size;
566
553
- m_assert(hostname, "hostname is NULL");
554
- m_assert(hostname_size, "hostname_size is 0");
567
+ m_assert(hostname, "hostname is NULL");
568
+ m_assert(hostname_size, "hostname_size is 0");
569
556
- new_tmp_text_size += hostname_size + 1; // +1 for ' ' char
557
- }
558
- else if(!strncmp(p->key.via.str.ptr, "SYSLOG_IDENTIFIER", (size_t) p->key.via.str.size)){
559
- syslog_identifier = (char *) p->val.via.str.ptr;
560
- syslog_identifier_size = p->val.via.str.size;
570
+ new_tmp_text_size += hostname_size + 1; // +1 for ' ' char
571
+ }
572
+ else if(!strncmp(p->key.via.str.ptr, "SYSLOG_IDENTIFIER", (size_t) p->key.via.str.size)){
573
+ syslog_identifier = (char *) p->val.via.str.ptr;
574
+ syslog_identifier_size = p->val.via.str.size;
575
562
- new_tmp_text_size += syslog_identifier_size;
563
- }
564
- else if(!strncmp(p->key.via.str.ptr, "PID", (size_t) p->key.via.str.size)){
565
- pid = (char *) p->val.via.str.ptr;
566
- pid_size = p->val.via.str.size;
576
+ new_tmp_text_size += syslog_identifier_size;
577
+ }
578
+ else if(!strncmp(p->key.via.str.ptr, "PID", (size_t) p->key.via.str.size)){
579
+ pid = (char *) p->val.via.str.ptr;
580
+ pid_size = p->val.via.str.size;
581
568
- new_tmp_text_size += pid_size;
569
- }
570
- else if(!strncmp(p->key.via.str.ptr, LOG_REC_KEY_SYSTEMD, (size_t) p->key.via.str.size)){
571
-
572
- message = (char *) p->val.via.str.ptr;
573
- message_size = p->val.via.str.size;
582
+ new_tmp_text_size += pid_size;
583
+ }
584
+ else if(!strncmp(p->key.via.str.ptr, LOG_REC_KEY_SYSTEMD, (size_t) p->key.via.str.size)){
585
+
586
+ message = (char *) p->val.via.str.ptr;
587
+ message_size = p->val.via.str.size;
588
575
- m_assert(message, "message is NULL");
576
- m_assert(message_size, "message_size is 0");
589
+ m_assert(message, "message is NULL");
590
+ m_assert(message_size, "message_size is 0");
591
578
- new_tmp_text_size += message_size;
592
+ new_tmp_text_size += message_size;
593
+ }
594
+
595
+ break;
596
}
580
- ++p;
581
- continue;
582
- } /* FLB_SYSTEMD or FLB_SYSLOG case end */
597
584
- /* FLB_DOCKER_EV case */
585
- if(p_file_info->log_type == FLB_DOCKER_EV){
586
- if(!strncmp(p->key.via.str.ptr, "time", (size_t) p->key.via.str.size)){
587
- docker_ev_time = p->val.via.i64;
598
+ case FLB_DOCKER_EV:
599
+ {
600
+ if(!strncmp(p->key.via.str.ptr, "time", (size_t) p->key.via.str.size)){
601
+ docker_ev_time = p->val.via.i64;
602
589
- m_assert(docker_ev_time, "docker_ev_time is 0");
590
- }
591
- else if(!strncmp(p->key.via.str.ptr, "timeNano", (size_t) p->key.via.str.size)){
592
- docker_ev_timeNano = p->val.via.i64;
603
+ m_assert(docker_ev_time, "docker_ev_time is 0");
604
+ }
605
+ else if(!strncmp(p->key.via.str.ptr, "timeNano", (size_t) p->key.via.str.size)){
606
+ docker_ev_timeNano = p->val.via.i64;
607
594
- m_assert(docker_ev_timeNano, "docker_ev_timeNano is 0");
608
+ m_assert(docker_ev_timeNano, "docker_ev_timeNano is 0");
609
596
- if(likely(p_file_info->use_log_timestamp))
597
- timestamp = docker_ev_timeNano / NSEC_PER_MSEC;
598
- }
599
- else if(!strncmp(p->key.via.str.ptr, "Type", (size_t) p->key.via.str.size)){
600
- docker_ev_type = (char *) p->val.via.str.ptr;
601
- docker_ev_type_size = p->val.via.str.size;
610
+ if(likely(p_file_info->use_log_timestamp))
611
+ timestamp = docker_ev_timeNano / NSEC_PER_MSEC;
612
+ }
613
+ else if(!strncmp(p->key.via.str.ptr, "Type", (size_t) p->key.via.str.size)){
614
+ docker_ev_type = (char *) p->val.via.str.ptr;
615
+ docker_ev_type_size = p->val.via.str.size;
616
603
- m_assert(docker_ev_type, "docker_ev_type is NULL");
604
- m_assert(docker_ev_type_size, "docker_ev_type_size is 0");
617
+ m_assert(docker_ev_type, "docker_ev_type is NULL");
618
+ m_assert(docker_ev_type_size, "docker_ev_type_size is 0");
619
606
- // debug_log("docker_ev_type: %.*s", docker_ev_type_size, docker_ev_type);
607
- }
608
- else if(!strncmp(p->key.via.str.ptr, "Action", (size_t) p->key.via.str.size)){
609
- docker_ev_action = (char *) p->val.via.str.ptr;
610
- docker_ev_action_size = p->val.via.str.size;
620
+ // debug_log("docker_ev_type: %.*s", docker_ev_type_size, docker_ev_type);
621
+ }
622
+ else if(!strncmp(p->key.via.str.ptr, "Action", (size_t) p->key.via.str.size)){
623
+ docker_ev_action = (char *) p->val.via.str.ptr;
624
+ docker_ev_action_size = p->val.via.str.size;
625
612
- m_assert(docker_ev_action, "docker_ev_action is NULL");
613
- m_assert(docker_ev_action_size, "docker_ev_action_size is 0");
626
+ m_assert(docker_ev_action, "docker_ev_action is NULL");
627
+ m_assert(docker_ev_action_size, "docker_ev_action_size is 0");
628
615
- // debug_log("docker_ev_action: %.*s", docker_ev_action_size, docker_ev_action);
616
- }
617
- else if(!strncmp(p->key.via.str.ptr, "id", (size_t) p->key.via.str.size)){
618
- docker_ev_id = (char *) p->val.via.str.ptr;
619
- docker_ev_id_size = p->val.via.str.size;
629
+ // debug_log("docker_ev_action: %.*s", docker_ev_action_size, docker_ev_action);
630
+ }
631
+ else if(!strncmp(p->key.via.str.ptr, "id", (size_t) p->key.via.str.size)){
632
+ docker_ev_id = (char *) p->val.via.str.ptr;
633
+ docker_ev_id_size = p->val.via.str.size;
634
621
- m_assert(docker_ev_id, "docker_ev_id is NULL");
622
- m_assert(docker_ev_id_size, "docker_ev_id_size is 0");
635
+ m_assert(docker_ev_id, "docker_ev_id is NULL");
636
+ m_assert(docker_ev_id_size, "docker_ev_id_size is 0");
637
624
- // debug_log("docker_ev_id: %.*s", docker_ev_id_size, docker_ev_id);
625
- }
626
- else if(!strncmp(p->key.via.str.ptr, "Actor", (size_t) p->key.via.str.size)){
627
- // debug_log( "msg key:[%.*s]val:[%.*s]", (int) p->key.via.str.size,
628
- // p->key.via.str.ptr,
629
- // (int) p->val.via.str.size,
630
- // p->val.via.str.ptr);
631
- if(likely(p->val.type == MSGPACK_OBJECT_MAP && p->val.via.map.size != 0)){
632
- msgpack_object_kv* ac = p->val.via.map.ptr;
633
- msgpack_object_kv* const ac_pend= p->val.via.map.ptr + p->val.via.map.size;
634
- do{
635
- if(!strncmp(ac->key.via.str.ptr, "ID", (size_t) ac->key.via.str.size)){
636
- docker_ev_id = (char *) ac->val.via.str.ptr;
637
- docker_ev_id_size = ac->val.via.str.size;
638
-
639
- m_assert(docker_ev_id, "docker_ev_id is NULL");
640
- m_assert(docker_ev_id_size, "docker_ev_id_size is 0");
641
-
642
- // debug_log("docker_ev_id: %.*s", docker_ev_id_size, docker_ev_id);
643
- }
644
- else if(!strncmp(ac->key.via.str.ptr, "Attributes", (size_t) ac->key.via.str.size)){
645
- if(likely(ac->val.type == MSGPACK_OBJECT_MAP && ac->val.via.map.size != 0)){
646
- msgpack_object_kv* att = ac->val.via.map.ptr;
647
- msgpack_object_kv* const att_pend = ac->val.via.map.ptr + ac->val.via.map.size;
648
- do{
649
- if(unlikely(++docker_ev_attr.size > docker_ev_attr.max_size)){
650
- docker_ev_attr.max_size = docker_ev_attr.size;
651
- docker_ev_attr.key = reallocz(docker_ev_attr.key,
652
- docker_ev_attr.max_size * sizeof(char *));
653
- docker_ev_attr.val = reallocz(docker_ev_attr.val,
654
- docker_ev_attr.max_size * sizeof(char *));
655
- docker_ev_attr.key_size = reallocz(docker_ev_attr.key_size,
656
- docker_ev_attr.max_size * sizeof(size_t));
657
- docker_ev_attr.val_size = reallocz(docker_ev_attr.val_size,
658
- docker_ev_attr.max_size * sizeof(size_t));
659
- }
660
-
661
- docker_ev_attr.key[docker_ev_attr.size - 1] = (char *) att->key.via.str.ptr;
662
- docker_ev_attr.val[docker_ev_attr.size - 1] = (char *) att->val.via.str.ptr;
663
- docker_ev_attr.key_size[docker_ev_attr.size - 1] = (size_t) att->key.via.str.size;
664
- docker_ev_attr.val_size[docker_ev_attr.size - 1] = (size_t) att->val.via.str.size;
665
-
666
- att++;
667
- continue;
668
- } while(att < att_pend);
638
+ // debug_log("docker_ev_id: %.*s", docker_ev_id_size, docker_ev_id);
639
+ }
640
+ else if(!strncmp(p->key.via.str.ptr, "Actor", (size_t) p->key.via.str.size)){
641
+ // debug_log( "msg key:[%.*s]val:[%.*s]", (int) p->key.via.str.size,
642
+ // p->key.via.str.ptr,
643
+ // (int) p->val.via.str.size,
644
+ // p->val.via.str.ptr);
645
+ if(likely(p->val.type == MSGPACK_OBJECT_MAP && p->val.via.map.size != 0)){
646
+ msgpack_object_kv* ac = p->val.via.map.ptr;
647
+ msgpack_object_kv* const ac_pend= p->val.via.map.ptr + p->val.via.map.size;
648
+ do{
649
+ if(!strncmp(ac->key.via.str.ptr, "ID", (size_t) ac->key.via.str.size)){
650
+ docker_ev_id = (char *) ac->val.via.str.ptr;
651
+ docker_ev_id_size = ac->val.via.str.size;
652
+
653
+ m_assert(docker_ev_id, "docker_ev_id is NULL");
654
+ m_assert(docker_ev_id_size, "docker_ev_id_size is 0");
655
+
656
+ // debug_log("docker_ev_id: %.*s", docker_ev_id_size, docker_ev_id);
657
}
670
- }
671
- ac++;
672
- continue;
673
- } while(ac < ac_pend);
658
+ else if(!strncmp(ac->key.via.str.ptr, "Attributes", (size_t) ac->key.via.str.size)){
659
+ if(likely(ac->val.type == MSGPACK_OBJECT_MAP && ac->val.via.map.size != 0)){
660
+ msgpack_object_kv* att = ac->val.via.map.ptr;
661
+ msgpack_object_kv* const att_pend = ac->val.via.map.ptr + ac->val.via.map.size;
662
+ do{
663
+ if(unlikely(++docker_ev_attr.size > docker_ev_attr.max_size)){
664
+ docker_ev_attr.max_size = docker_ev_attr.size;
665
+ docker_ev_attr.key = reallocz(docker_ev_attr.key,
666
+ docker_ev_attr.max_size * sizeof(char *));
667
+ docker_ev_attr.val = reallocz(docker_ev_attr.val,
668
+ docker_ev_attr.max_size * sizeof(char *));
669
+ docker_ev_attr.key_size = reallocz(docker_ev_attr.key_size,
670
+ docker_ev_attr.max_size * sizeof(size_t));
671
+ docker_ev_attr.val_size = reallocz(docker_ev_attr.val_size,
672
+ docker_ev_attr.max_size * sizeof(size_t));
673
+ }
674
+
675
+ docker_ev_attr.key[docker_ev_attr.size - 1] = (char *) att->key.via.str.ptr;
676
+ docker_ev_attr.val[docker_ev_attr.size - 1] = (char *) att->val.via.str.ptr;
677
+ docker_ev_attr.key_size[docker_ev_attr.size - 1] = (size_t) att->key.via.str.size;
678
+ docker_ev_attr.val_size[docker_ev_attr.size - 1] = (size_t) att->val.via.str.size;
679
+
680
+ att++;
681
+ continue;
682
+ } while(att < att_pend);
683
+ }
684
+ }
685
+ ac++;
686
+ continue;
687
+ } while(ac < ac_pend);
688
+ }
689
}
690
+
691
+ break;
692
}
676
- ++p;
677
- continue;
678
- }
679
- /* FLB_DOCKER_EV case end */
693
681
- /* FLB_MQTT case */
682
- if(p_file_info->log_type == FLB_MQTT){
683
- if(!strncmp(p->key.via.str.ptr, "topic", (size_t) p->key.via.str.size)){
684
- mqtt_topic = (char *) p->val.via.str.ptr;
685
- mqtt_topic_size = (size_t) p->val.via.str.size;
694
+ case FLB_MQTT:
695
+ {
696
+ if(!strncmp(p->key.via.str.ptr, "topic", (size_t) p->key.via.str.size)){
697
+ mqtt_topic = (char *) p->val.via.str.ptr;
698
+ mqtt_topic_size = (size_t) p->val.via.str.size;
699
687
- while(0 == (message_size = dl_msgpack_object_print_buffer(mqtt_message, mqtt_message_size_max, *x)))
688
- mqtt_message = reallocz(mqtt_message, (mqtt_message_size_max += 10));
700
+ while(0 == (message_size = dl_msgpack_object_print_buffer(mqtt_message, mqtt_message_size_max, *x)))
701
+ mqtt_message = reallocz(mqtt_message, (mqtt_message_size_max += 10));
702
690
- new_tmp_text_size = message_size + 1; // +1 for '\n'
703
+ new_tmp_text_size = message_size + 1; // +1 for '\n'
704
692
- m_assert(message_size, "message_size is 0");
693
- m_assert(mqtt_message, "mqtt_message is NULL");
705
+ m_assert(message_size, "message_size is 0");
706
+ m_assert(mqtt_message, "mqtt_message is NULL");
707
695
- break; // watch out, MQTT requires a 'break' here, as we parse the entire 'x' msgpack_object
708
+ break; // watch out, MQTT requires a 'break' here, as we parse the entire 'x' msgpack_object
709
+ }
710
+ else m_assert(0, "missing mqtt topic");
711
+
712
+ break;
713
}
697
- else m_assert(0, "missing mqtt topic");
714
699
- ++p;
700
- continue;
715
+ default:
716
+ break;
717
}
702
-
703
- } while(p < pend);
718
+
719
+ } while(++p < pend);
720
}
721
}
722
@@ -739,7 +755,8 @@ static int flb_collect_logs_cb(void *record, size_t size, void *data){
755
p_file_info->parser_metrics->web_log);
756
757
// TODO: Fix: Metrics will still be collected if circ_buff_prepare_write() returns 0.
742
- if(unlikely(!circ_buff_prepare_write(buff, new_tmp_text_size))) goto skip_collect_and_drop_logs;
758
+ if(unlikely(!circ_buff_prepare_write(buff, new_tmp_text_size)))
759
+ goto skip_collect_and_drop_logs;
760
761
size_t tmp_item_off = buff->in->text_size;
762
@@ -749,6 +766,49 @@ static int flb_collect_logs_cb(void *record, size_t size, void *data){
766
buff->in->data[tmp_item_off++] = '\n';
767
m_assert(tmp_item_off == new_tmp_text_size, "tmp_item_off should be == new_tmp_text_size");
768
buff->in->text_size = new_tmp_text_size;
769
+
770
+#ifdef HAVE_SYSTEMD
771
+ if(p_file_info->do_sd_journal_send){
772
+ if(p_file_info->log_type == FLB_WEB_LOG){
773
+ sd_journal_send(
774
+ SD_JOURNAL_SEND_DEFAULT_FIELDS,
775
+ *line_parsed.vhost ? "%sWEB_LOG_VHOST=%s" : "_%s=%s", sd_journal_field_prefix, line_parsed.vhost,
776
+ line_parsed.port ? "%sWEB_LOG_PORT=%d" : "_%s=%d", sd_journal_field_prefix, line_parsed.port,
777
+ *line_parsed.req_scheme ? "%sWEB_LOG_REQ_SCHEME=%s" : "_%s=%s", sd_journal_field_prefix, line_parsed.req_scheme,
778
+ *line_parsed.req_client ? "%sWEB_LOG_REQ_CLIENT=%s" : "_%s=%s", sd_journal_field_prefix, line_parsed.req_client,
779
+ "%sWEB_LOG_REQ_METHOD=%s" , sd_journal_field_prefix, line_parsed.req_method,
780
+ *line_parsed.req_URL ? "%sWEB_LOG_REQ_URL=%s" : "_%s=%s", sd_journal_field_prefix, line_parsed.req_URL,
781
+ *line_parsed.req_proto ? "%sWEB_LOG_REQ_PROTO=%s" : "_%s=%s", sd_journal_field_prefix, line_parsed.req_proto,
782
+ line_parsed.req_size ? "%sWEB_LOG_REQ_SIZE=%d" : "_%s=%d", sd_journal_field_prefix, line_parsed.req_size,
783
+ line_parsed.req_proc_time ? "%sWEB_LOG_REC_PROC_TIME=%d" : "_%s=%d", sd_journal_field_prefix, line_parsed.req_proc_time,
784
+ line_parsed.resp_code ? "%sWEB_LOG_RESP_CODE=%d" : "_%s=%d", sd_journal_field_prefix ,line_parsed.resp_code,
785
+ line_parsed.ups_resp_time ? "%sWEB_LOG_UPS_RESP_TIME=%d" : "_%s=%d", sd_journal_field_prefix ,line_parsed.ups_resp_time,
786
+ *line_parsed.ssl_proto ? "%sWEB_LOG_SSL_PROTO=%s" : "_%s=%s", sd_journal_field_prefix ,line_parsed.ssl_proto,
787
+ *line_parsed.ssl_cipher ? "%sWEB_LOB_SSL_CIPHER=%s" : "_%s=%s", sd_journal_field_prefix ,line_parsed.ssl_cipher,
788
+ LOG_REC_KEY_SYSTEMD "=%.*s", (int) message_size, message,
789
+ NULL
790
+ );
791
+ }
792
+ else if(p_file_info->log_type == FLB_SERIAL){
793
+ Flb_serial_config_t *serial_config = (Flb_serial_config_t *) p_file_info->flb_config;
794
+ sd_journal_send(
795
+ SD_JOURNAL_SEND_DEFAULT_FIELDS,
796
+ serial_config->bitrate && *serial_config->bitrate ?
797
+ "%sSERIAL_BITRATE=%s" : "_%s=%s", sd_journal_field_prefix, serial_config->bitrate,
798
+ LOG_REC_KEY_SYSTEMD "=%.*s", (int) message_size, message,
799
+ NULL
800
+ );
801
+ }
802
+ else{
803
+ sd_journal_send(
804
+ SD_JOURNAL_SEND_DEFAULT_FIELDS,
805
+ LOG_REC_KEY_SYSTEMD "=%.*s", (int) message_size, message,
806
+ NULL
807
+ );
808
+ }
809
+ }
810
+#endif
811
+
812
} /* FLB_TAIL, FLB_WEB_LOG and FLB_SERIAL case end */
813
814
/* FLB_KMSG case */
@@ -903,7 +963,8 @@ static int flb_collect_logs_cb(void *record, size_t size, void *data){
963
964
/* Metrics extracted, now prepare circular buffer for write */
965
// TODO: Fix: Metrics will still be collected if circ_buff_prepare_write() returns 0.
906
- if(unlikely(!circ_buff_prepare_write(buff, new_tmp_text_size))) goto skip_collect_and_drop_logs;
966
+ if(unlikely(!circ_buff_prepare_write(buff, new_tmp_text_size)))
967
+ goto skip_collect_and_drop_logs;
968
969
size_t tmp_item_off = buff->in->text_size;
970
@@ -1029,9 +1090,11 @@ static int flb_collect_logs_cb(void *record, size_t size, void *data){
1090
1091
/* Metrics extracted, now prepare circular buffer for write */
1092
// TODO: Fix: Metrics will still be collected if circ_buff_prepare_write() returns 0.
1032
- if(unlikely(!circ_buff_prepare_write(buff, new_tmp_text_size))) goto skip_collect_and_drop_logs;
1093
+ if(unlikely(!circ_buff_prepare_write(buff, new_tmp_text_size)))
1094
+ goto skip_collect_and_drop_logs;
1095
1096
size_t tmp_item_off = buff->in->text_size;
1097
+ message_size = new_tmp_text_size - 1 - tmp_item_off;
1098
1099
if(likely(*docker_ev_datetime)){
1100
memcpy(&buff->in->data[tmp_item_off], docker_ev_datetime, docker_ev_datetime_size - 1);
@@ -1075,6 +1138,20 @@ static int flb_collect_logs_cb(void *record, size_t size, void *data){
1138
buff->in->data[tmp_item_off++] = '\n';
1139
m_assert(tmp_item_off == new_tmp_text_size, "tmp_item_off should be == new_tmp_text_size");
1140
buff->in->text_size = new_tmp_text_size;
1141
+
1142
+#ifdef HAVE_SYSTEMD
1143
+ if(p_file_info->do_sd_journal_send){
1144
+ sd_journal_send(
1145
+ SD_JOURNAL_SEND_DEFAULT_FIELDS,
1146
+ "%sDOCKER_EVENTS_TYPE=%.*s", sd_journal_field_prefix, (int) docker_ev_type_size, docker_ev_type,
1147
+ "%sDOCKER_EVENTS_ACTION=%.*s", sd_journal_field_prefix, (int) docker_ev_action_size, docker_ev_action,
1148
+ "%sDOCKER_EVENTS_ID=%.*s", sd_journal_field_prefix, (int) docker_ev_id_size, docker_ev_id,
1149
+ LOG_REC_KEY_SYSTEMD "=%.*s", (int) message_size, &buff->in->data[tmp_item_off - 1 - message_size],
1150
+ NULL
1151
+ );
1152
+ }
1153
+#endif
1154
+
1155
} /* FLB_DOCKER_EV case end */
1156
1157
/* FLB_MQTT case */
@@ -1087,7 +1164,8 @@ static int flb_collect_logs_cb(void *record, size_t size, void *data){
1164
dictionary_set_advanced(p_file_info->parser_metrics->mqtt->topic, key, mqtt_topic_size + 1, &item, sizeof(item), NULL);
1165
1166
// TODO: Fix: Metrics will still be collected if circ_buff_prepare_write() returns 0.
1090
- if(unlikely(!circ_buff_prepare_write(buff, new_tmp_text_size))) goto skip_collect_and_drop_logs;
1167
+ if(unlikely(!circ_buff_prepare_write(buff, new_tmp_text_size)))
1168
+ goto skip_collect_and_drop_logs;
1169
1170
size_t tmp_item_off = buff->in->text_size;
1171
@@ -1097,6 +1175,18 @@ static int flb_collect_logs_cb(void *record, size_t size, void *data){
1175
buff->in->data[tmp_item_off++] = '\n';
1176
m_assert(tmp_item_off == new_tmp_text_size, "tmp_item_off should be == new_tmp_text_size");
1177
buff->in->text_size = new_tmp_text_size;
1178
+
1179
+#ifdef HAVE_SYSTEMD
1180
+ if(p_file_info->do_sd_journal_send){
1181
+ sd_journal_send(
1182
+ SD_JOURNAL_SEND_DEFAULT_FIELDS,
1183
+ "%sMQTT_TOPIC=%s", key,
1184
+ LOG_REC_KEY_SYSTEMD "=%.*s", (int) message_size, mqtt_message,
1185
+ NULL
1186
+ );
1187
+ }
1188
+#endif
1189
+
1190
}
1191
else m_assert(0, "missing mqtt topic");
1192
}
@@ -1110,10 +1200,10 @@ skip_collect_and_drop_logs:
1200
memset(&result.data, 0, sizeof(msgpack_object));
1201
}
1202
1113
- if(p_file_info) uv_mutex_unlock(&p_file_info->flb_tmp_buff_mut);
1203
+ if(p_file_info)
1204
+ uv_mutex_unlock(&p_file_info->flb_tmp_buff_mut);
1205
1206
flb_lib_free(record);
1116
- // FLB_OUTPUT_RETURN(FLB_OK); // Watch out! This breaks output - won't flush all pending logs
1207
return 0;
1208
1209
}
@@ -1177,12 +1267,18 @@ int flb_add_input(struct File_info *const p_file_info){
1267
}
1268
case FLB_KMSG: {
1269
debug_log( "Setting up FLB_KMSG collector");
1270
+
1271
+ Flb_kmsg_config_t *kmsg_config = (Flb_kmsg_config_t *) p_file_info->flb_config;
1272
+ if(unlikely(!kmsg_config ||
1273
+ !kmsg_config->prio_level ||
1274
+ !*kmsg_config->prio_level)) return CONFIG_READ_ERROR;
1275
1276
/* Set up kmsg input */
1277
p_file_info->flb_input = flb_input(ctx, "kmsg", NULL);
1278
if(p_file_info->flb_input < 0 ) return FLB_INPUT_ERROR;
1279
if(flb_input_set(ctx, p_file_info->flb_input,
1280
"Tag", tag_s,
1281
+ "Prio_Level", kmsg_config->prio_level,
1282
NULL) != 0) return FLB_INPUT_SET_ERROR;
1283
1284
break;
@@ -1302,8 +1398,12 @@ int flb_add_input(struct File_info *const p_file_info){
1398
debug_log( "Setting up FLB_SERIAL collector");
1399
1400
Flb_serial_config_t *serial_config = (Flb_serial_config_t *) p_file_info->flb_config;
1305
- if(unlikely(!serial_config || !serial_config->bitrate || !*serial_config->bitrate ||
1306
- !serial_config->min_bytes || !p_file_info->filename)) return CONFIG_READ_ERROR;
1401
+ if(unlikely(!serial_config ||
1402
+ !serial_config->bitrate ||
1403
+ !*serial_config->bitrate ||
1404
+ !serial_config->min_bytes ||
1405
+ !*serial_config->min_bytes ||
1406
+ !p_file_info->filename)) return CONFIG_READ_ERROR;
1407
1408
/* Set up serial input */
1409
p_file_info->flb_input = flb_input(ctx, "serial", NULL);
@@ -1312,6 +1412,7 @@ int flb_add_input(struct File_info *const p_file_info){
1412
"Tag", tag_s,
1413
"File", p_file_info->filename,
1414
"Bitrate", serial_config->bitrate,
1415
+ "Min_Bytes", serial_config->min_bytes,
1416
"Separator", serial_config->separator,
1417
"Format", serial_config->format,
1418
NULL) != 0) return FLB_INPUT_SET_ERROR;
logsmanagement/flb_plugin.h
+3
-1
@@ -22,7 +22,9 @@ typedef struct {
22
*coro_stack_size;
23
} flb_srvc_config_t ;
24
25
-int flb_init(flb_srvc_config_t flb_srvc_config, const char *const stock_config_dir);
25
+int flb_init(flb_srvc_config_t flb_srvc_config,
26
+ const char *const stock_config_dir,
27
+ const char *const new_sd_journal_field_prefix);
28
int flb_run(void);
29
void flb_terminate(void);
30
void flb_complete_item_timer_timeout_cb(uv_timer_t *handle);
logsmanagement/logsmanag_config.c
+30
-2
@@ -21,7 +21,9 @@ g_logs_manag_config_t g_logs_manag_config = {
21
.disk_space_limit_in_mib = DISK_SPACE_LIMIT_DEFAULT,
22
.buff_flush_to_db_interval = SAVE_BLOB_TO_DB_DEFAULT,
23
.enable_collected_logs_total = ENABLE_COLLECTED_LOGS_TOTAL_DEFAULT,
24
- .enable_collected_logs_rate = ENABLE_COLLECTED_LOGS_RATE_DEFAULT
24
+ .enable_collected_logs_rate = ENABLE_COLLECTED_LOGS_RATE_DEFAULT,
25
+ .sd_journal_field_prefix = SD_JOURNAL_FIELD_PREFIX,
26
+ .do_sd_journal_send = SD_JOURNAL_SEND_DEFAULT
27
};
28
29
static logs_manag_db_mode_t db_mode_str_to_db_mode(const char *const db_mode_str){
@@ -311,7 +313,6 @@ int logs_manag_config_load( flb_srvc_config_t *p_flb_srvc_config,
313
section,
314
"circular buffer drop logs if full",
315
g_logs_manag_config.circ_buff_drop_logs);
314
-
316
317
g_logs_manag_config.compression_acceleration = appconfig_get_number(
318
&logsmanagement_d_conf,
@@ -330,6 +331,18 @@ int logs_manag_config_load( flb_srvc_config_t *p_flb_srvc_config,
331
section,
332
"collected logs rate chart enable",
333
g_logs_manag_config.enable_collected_logs_rate);
334
+
335
+ g_logs_manag_config.do_sd_journal_send = appconfig_get_boolean(
336
+ &logsmanagement_d_conf,
337
+ section,
338
+ "submit logs to system journal",
339
+ g_logs_manag_config.do_sd_journal_send);
340
+
341
+ g_logs_manag_config.sd_journal_field_prefix = appconfig_get(
342
+ &logsmanagement_d_conf,
343
+ section,
344
+ "systemd journal fields prefix",
345
+ g_logs_manag_config.sd_journal_field_prefix);
346
347
if(!rc){
348
collector_info("CONFIG: [%s] update every: %d", section, g_logs_manag_config.update_every);
@@ -340,6 +353,8 @@ int logs_manag_config_load( flb_srvc_config_t *p_flb_srvc_config,
353
collector_info("CONFIG: [%s] compression acceleration: %d", section, g_logs_manag_config.compression_acceleration);
354
collector_info("CONFIG: [%s] collected logs total chart enable: %d", section, g_logs_manag_config.enable_collected_logs_total);
355
collector_info("CONFIG: [%s] collected logs rate chart enable: %d", section, g_logs_manag_config.enable_collected_logs_rate);
356
+ collector_info("CONFIG: [%s] submit logs to system journal: %d", section, g_logs_manag_config.do_sd_journal_send);
357
+ collector_info("CONFIG: [%s] systemd journal fields prefix: %s", section, g_logs_manag_config.sd_journal_field_prefix);
358
}
359
360
@@ -800,6 +815,13 @@ static void config_section_init(uv_loop_t *main_loop,
815
collector_info("[%s]: BLOB max size = %lld", p_file_info->chartname, (long long)p_file_info->blob_max_size);
816
817
818
+ /* -------------------------------------------------------------------------
819
+ * Read configuration about sending logs to system journal.
820
+ * ------------------------------------------------------------------------- */
821
+ p_file_info->do_sd_journal_send = appconfig_get_boolean(&log_management_config, config_section->name,
822
+ "submit logs to system journal",
823
+ g_logs_manag_config.do_sd_journal_send);
824
+
825
/* -------------------------------------------------------------------------
826
* Read collected logs chart configuration.
827
* ------------------------------------------------------------------------- */
@@ -936,6 +958,12 @@ static void config_section_init(uv_loop_t *main_loop,
958
}
959
}
960
else if(p_file_info->log_type == FLB_KMSG){
961
+ Flb_kmsg_config_t *kmsg_config = callocz(1, sizeof(Flb_kmsg_config_t));
962
+
963
+ kmsg_config->prio_level = appconfig_get(&log_management_config, config_section->name, "prio level", "8");
964
+
965
+ p_file_info->flb_config = kmsg_config;
966
+
967
if(appconfig_get_boolean(&log_management_config, config_section->name, "severity chart", CONFIG_BOOLEAN_NO)) {
968
p_file_info->parser_config->chart_config |= CHART_SYSLOG_SEVER;
969
}
logsmanagement/logsmanagement.c
+1
-1
@@ -162,7 +162,7 @@ int main(int argc, char **argv) {
162
if(logs_manag_config_load(&flb_srvc_config, &p_forward_in_config, g_update_every))
163
exit(1);
164
165
- if(flb_init(flb_srvc_config, get_stock_config_dir())){
165
+ if(flb_init(flb_srvc_config, get_stock_config_dir(), g_logs_manag_config.sd_journal_field_prefix)){
166
collector_error("flb_init() failed - logs management will be disabled");
167
exit(1);
168
}
logsmanagement/stock_conf/logsmanagement.d.conf
+2
@@ -7,6 +7,8 @@
7
compression acceleration = 1
8
collected logs total chart enable = no
9
collected logs rate chart enable = yes
10
+ submit logs to system journal = no
11
+ systemd journal fields prefix = LOGS_MANAG_
12
13
[db]
14
db mode = none
logsmanagement/stock_conf/logsmanagement.d/default.conf
+21
@@ -33,6 +33,9 @@
33
# circular buffer flush to db = 6
34
# disk space limit MiB = 500
35
36
+ ## Drop kernel logs with priority higher than prio_level.
37
+ # prio level = 8
38
+
39
## Charts to enable
40
# collected logs total chart enable = no
41
# collected logs rate chart enable = yes
@@ -102,6 +105,9 @@
105
## Use default Docker socket UNIX path: /var/run/docker.sock
106
log path = auto
107
108
+ ## Submit structured log entries to the system journal
109
+ # submit logs to system journal = no
110
+
111
## Charts to enable
112
# collected logs total chart enable = no
113
# collected logs rate chart enable = yes
@@ -176,6 +182,9 @@
182
## Detect errors such as illegal port numbers or response codes.
183
verify parsed logs = yes
184
185
+ ## Submit structured log entries to the system journal
186
+ # submit logs to system journal = no
187
+
188
## Charts to enable
189
# collected logs total chart enable = no
190
# collected logs rate chart enable = yes
@@ -229,6 +238,9 @@
238
## Detect errors such as illegal port numbers or response codes.
239
verify parsed logs = yes
240
241
+ ## Submit structured log entries to the system journal
242
+ # submit logs to system journal = no
243
+
244
## Charts to enable
245
# collected logs total chart enable = no
246
# collected logs rate chart enable = yes
@@ -276,6 +288,9 @@
288
## Use inotify instead of file stat watcher. Set to 'no' to reduce CPU usage.
289
use inotify = yes
290
291
+ ## Submit structured log entries to the system journal
292
+ # submit logs to system journal = no
293
+
294
## Charts to enable
295
# collected logs total chart enable = no
296
# collected logs rate chart enable = yes
@@ -352,6 +367,9 @@
367
## Use inotify instead of file stat watcher. Set to 'no' to reduce CPU usage.
368
use inotify = yes
369
370
+ ## Submit structured log entries to the system journal
371
+ # submit logs to system journal = no
372
+
373
## Charts to enable
374
# collected logs total chart enable = no
375
# collected logs rate chart enable = yes
@@ -413,6 +431,9 @@
431
## Use inotify instead of file stat watcher. Set to 'no' to reduce CPU usage.
432
use inotify = yes
433
434
+ ## Submit structured log entries to the system journal
435
+ # submit logs to system journal = no
436
+
437
## Charts to enable
438
# collected logs total chart enable = no
439
# collected logs rate chart enable = yes
logsmanagement/stock_conf/logsmanagement.d/example_forward.conf
+6
@@ -51,6 +51,9 @@
51
# circular buffer flush to db = 6
52
# disk space limit MiB = 500
53
54
+ ## Submit structured log entries to the system journal
55
+ # submit logs to system journal = no
56
+
57
## Streaming input settings.
58
log source = forward
59
stream guid = 6ce266f5-2704-444d-a301-2423b9d30736
@@ -81,6 +84,9 @@
84
# circular buffer flush to db = 6
85
# disk space limit MiB = 500
86
87
+ ## Submit structured log entries to the system journal
88
+ # submit logs to system journal = no
89
+
90
## Streaming input settings.
91
log source = forward
92
stream guid = 6ce266f5-2704-444d-a301-2423b9d30737
logsmanagement/stock_conf/logsmanagement.d/example_mqtt.conf
+3
@@ -22,6 +22,9 @@
22
# listen = 0.0.0.0
23
# port = 1883
24
25
+ ## Submit structured log entries to the system journal
26
+ # submit logs to system journal = no
27
+
28
## Charts to enable
29
# collected logs total chart enable = no
30
# collected logs rate chart enable = yes
logsmanagement/stock_conf/logsmanagement.d/example_serial.conf
+3
@@ -24,6 +24,9 @@
24
# separator = X
25
# format = json
26
27
+ ## Submit structured log entries to the system journal
28
+ # submit logs to system journal = no
29
+
30
## Charts to enable
31
# collected logs total chart enable = no
32
# collected logs rate chart enable = yes
logsmanagement/stock_conf/logsmanagement.d/example_syslog.conf
+3
@@ -28,6 +28,9 @@
28
## Use inotify instead of file stat watcher. Set to 'no' to reduce CPU usage.
29
use inotify = yes
30
31
+ ## Submit structured log entries to the system journal
32
+ # submit logs to system journal = no
33
+
34
## Charts to enable
35
# collected logs total chart enable = no
36
# collected logs rate chart enable = yes
logsmanagement/unit_test/unit_test.c
+2
-2
@@ -663,7 +663,7 @@ static int test_flb_init(){
663
fprintf(stderr, "Testing flb_init() with wrong stock_config_dir...\n");
664
665
SUPRESS_STDERR();
666
- rc = flb_init(flb_srvc_config, "/tmp");
666
+ rc = flb_init(flb_srvc_config, "/tmp", "example_prefix_");
667
UNSUPRESS_STDERR();
668
if(!rc){
669
fprintf(stderr, "- Error, flb_init() should fail but it returns %d.\n", rc);
@@ -672,7 +672,7 @@ static int test_flb_init(){
672
673
fprintf(stderr, "Testing flb_init() with correct stock_config_dir...\n");
674
675
- rc = flb_init(flb_srvc_config, get_stock_config_dir());
675
+ rc = flb_init(flb_srvc_config, get_stock_config_dir(), "example_prefix_");
676
if(rc){
677
fprintf(stderr, "- Error, flb_init() should fail but it returns %d.\n", rc);
678
++errors;