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

15 files changed, 448 insertions, 268 deletions

diff --git a/Makefile.am b/Makefile.am
index 547b5933e0..aff19318ae 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -1411,6 +1411,7 @@ if ENABLE_LOGSMANAGEMENT_TESTS
 endif
     logs_management_plugin_LDADD = \
         $(NETDATA_COMMON_LIBS) \
+        $(OPTIONAL_SYSTEMD_LIBS) \
         $(NULL)
 endif
 
diff --git a/logsmanagement/README.md b/logsmanagement/README.md
index b489070ac2..5c3982dca7 100644
--- a/logsmanagement/README.md
+++ b/logsmanagement/README.md
@@ -118,8 +118,9 @@ There are some fundamental configuration options that are common to all log coll
 | `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.
 | `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.
 | `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.
+| `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.
 
-There are also one setting that cannot be set per log source, but can only be defined in `logsmanagement.d.conf`:
+There is also one setting that cannot be set per log source, but can only be defined in `logsmanagement.d.conf`:
 
 |  Configuration Option | Default 		| Description  |
 |      :------------:  	| :------------:  | ------------ |
@@ -152,6 +153,7 @@ This collector will collect logs from the kernel message log buffer. See also do
 
 |  Configuration Option | Description  |
 |      :------------:  	| ------------ |
+| `prio level` | Drop kernel logs with priority higher than `prio level`. Default value is 8, so no logs will be dropped.
 | `severity chart` | Enable chart showing Syslog Severity values of collected logs. Severity values are in the range of 0 to 7 inclusive.|
 | `subsystem chart` | Enable chart showing which subsystems generated the logs.|
 | `device chart` | Enable chart showing which devices generated the logs.|
diff --git a/logsmanagement/defaults.h b/logsmanagement/defaults.h
index e0f56b6add..f0b651cbab 100644
--- a/logsmanagement/defaults.h
+++ b/logsmanagement/defaults.h
@@ -30,6 +30,10 @@
 #endif
 #define ENABLE_COLLECTED_LOGS_RATE_DEFAULT CONFIG_BOOLEAN_YES   /**< Default value to enable (or not) metrics of rate of collected log records **/
 
+#define SD_JOURNAL_FIELD_PREFIX "LOGS_MANAG_"                   /**< Default systemd journal field prefix for sources that log to the system journal */
+
+#define SD_JOURNAL_SEND_DEFAULT CONFIG_BOOLEAN_NO               /**< Default value to enable (or not) submission of logs to the system journal (where applicable) **/
+
 /* -------------------------------------------------------------------------- */
 
 
diff --git a/logsmanagement/file_info.h b/logsmanagement/file_info.h
index cf3c240036..751b874448 100644
--- a/logsmanagement/file_info.h
+++ b/logsmanagement/file_info.h
@@ -45,6 +45,9 @@ typedef struct flb_tail_config {
     int use_inotify;
 } Flb_tail_config_t;
 
+typedef struct flb_kmsg_config {
+    char *prio_level;
+} Flb_kmsg_config_t;
 
 typedef struct flb_serial_config {
     char *bitrate;
@@ -78,11 +81,9 @@ typedef struct flb_output_config {
 } Flb_output_config_t;
 
 struct File_info {
-    /* TODO: Struct needs refactoring, as a lot of members take up memory that
-     * is not used, depending on the type of the log source. */
 
     /* Struct members core to any log source type */
-    const char *chartname;                         /**< Top level chart name for this log source on web dashboard **/ 
+    const char *chartname;                          /**< Top level chart name for this log source on web dashboard **/ 
     char *filename;                                 /**< Full path of log source **/
     const char *file_basename;                      /**< Basename of log source **/
     const char *stream_guid;                        /**< Streaming input GUID **/
@@ -93,6 +94,7 @@ struct File_info {
     int update_every;                               /**< Interval (in sec) of how often to collect and update charts **/
     int update_timeout;                             /**< Timeout to update charts after, since last update */
     int use_log_timestamp;                          /**< Use log timestamps instead of collection timestamps, if available **/
+    int do_sd_journal_send;                         /**< Write to system journal - not applicable to all log source types **/
     struct Chart_meta *chart_meta;
     LOG_SRC_STATE state;                            /**< State of log source, used to sync status among threads **/
 
@@ -154,6 +156,8 @@ typedef struct {
     int buff_flush_to_db_interval;
     int enable_collected_logs_total;
     int enable_collected_logs_rate;
+    char *sd_journal_field_prefix;
+    int do_sd_journal_send;
 } g_logs_manag_config_t;
 
 extern g_logs_manag_config_t g_logs_manag_config;
diff --git a/logsmanagement/flb_plugin.c b/logsmanagement/flb_plugin.c
index d33b01410e..493749ed4a 100644
--- a/logsmanagement/flb_plugin.c
+++ b/logsmanagement/flb_plugin.c
@@ -17,6 +17,13 @@
 #include "../fluent-bit/lib/monkey/include/monkey/mk_core/mk_list.h"
 #include <dlfcn.h>
 
+#ifdef HAVE_SYSTEMD
+#include <systemd/sd-journal.h>
+#define SD_JOURNAL_SEND_DEFAULT_FIELDS \
+            "%s_LOG_SOURCE=%s"  , sd_journal_field_prefix, log_src_t_str[p_file_info->log_source], \
+            "%s_LOG_TYPE=%s"    , sd_journal_field_prefix, log_src_type_t_str[p_file_info->log_type]
+#endif
+
 #define LOG_REC_KEY "msg" /**< key to represent log message field in most log sources **/
 #define LOG_REC_KEY_SYSTEMD "MESSAGE" /**< key to represent log message field in systemd log source **/
 #define SYSLOG_TIMESTAMP_SIZE 16
@@ -120,9 +127,13 @@ static void *flb_lib_handle = NULL;
 
 static struct flb_lib_out_cb *fwd_input_out_cb = NULL;
 
+static const char *sd_journal_field_prefix = SD_JOURNAL_FIELD_PREFIX;
+
 extern netdata_mutex_t stdout_mut;
 
-int flb_init(flb_srvc_config_t flb_srvc_config, const char *const stock_config_dir){
+int flb_init(flb_srvc_config_t flb_srvc_config, 
+            const char *const stock_config_dir, 
+            const char *const new_sd_journal_field_prefix){
     int rc = 0;
     char *dl_error;
 
@@ -198,6 +209,9 @@ int flb_init(flb_srvc_config_t flb_srvc_config, const char *const stock_config_d
             rc = -1;
             goto do_return;
         }
+    
+    if(new_sd_journal_field_prefix && *new_sd_journal_field_prefix)
+        sd_journal_field_prefix = new_sd_journal_field_prefix;
 
 do_return:
     freez(flb_lib_path);
@@ -403,7 +417,8 @@ static int flb_collect_logs_cb(void *record, size_t size, void *data){
                     // continue;
                 } while(p < pend);
             }
-            if(unlikely(p_file_info == NULL)) goto skip_collect_and_drop_logs;
+            if(unlikely(p_file_info == NULL)) 
+                goto skip_collect_and_drop_logs;
             
 
             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){
 
             p = x->via.map.ptr;
             pend = x->via.map.ptr + x->via.map.size;
-            do{              
-                
-                /* FLB_TAIL, FLB_WEB_LOG and FLB_SERIAL case */
-                if( p_file_info->log_type == FLB_TAIL || 
-                    p_file_info->log_type == FLB_WEB_LOG || 
-                    p_file_info->log_type == FLB_SERIAL){
-                    if( !strncmp(p->key.via.str.ptr, LOG_REC_KEY, (size_t) p->key.via.str.size) ||
-                        /* The following line is in case we collect systemd logs 
-                         * (tagged as "MESSAGE") or docker_events (tagged as
-                         * "message") via a "Forward" source to an FLB_TAIL 
-                         * parent. */
-                        !strncasecmp(p->key.via.str.ptr, LOG_REC_KEY_SYSTEMD, (size_t) p->key.via.str.size)){
-
-                        message = (char *) p->val.via.str.ptr;
-                        message_size = p->val.via.str.size;
-
-                        if(p_file_info->log_type == FLB_WEB_LOG){
-                            parse_web_log_line( (Web_log_parser_config_t *) p_file_info->parser_config->gen_config, 
-                                                message, message_size, &line_parsed);
-
-                            if(likely(p_file_info->use_log_timestamp)){
-                                timestamp = line_parsed.timestamp * MSEC_PER_SEC; // convert to msec from sec
-
-                                { /* ------------------ FIXME ------------------------  
-                                   * Temporary kludge so that metrics don't break when
-                                   * a new record has timestamp before the current one. 
-                                   */
-                                    static msec_t previous_timestamp = 0;
-                                    if((((long long) timestamp - (long long) previous_timestamp) < 0))
-                                        timestamp = previous_timestamp;
-                                    
-                                    previous_timestamp = timestamp;
+            do{
+                switch(p_file_info->log_type){
+
+                    case FLB_TAIL:
+                    case FLB_WEB_LOG:
+                    case FLB_SERIAL: 
+                    {
+                        if( !strncmp(p->key.via.str.ptr, LOG_REC_KEY, (size_t) p->key.via.str.size) ||
+                            /* The following line is in case we collect systemd logs 
+                            * (tagged as "MESSAGE") or docker_events (tagged as
+                            * "message") via a "Forward" source to an FLB_TAIL parent. */
+                            !strncasecmp(p->key.via.str.ptr, LOG_REC_KEY_SYSTEMD, (size_t) p->key.via.str.size)){
+
+                            message = (char *) p->val.via.str.ptr;
+                            message_size = p->val.via.str.size;
+
+                            if(p_file_info->log_type == FLB_WEB_LOG){
+                                parse_web_log_line( (Web_log_parser_config_t *) p_file_info->parser_config->gen_config, 
+                                                    message, message_size, &line_parsed);
+
+                                if(likely(p_file_info->use_log_timestamp)){
+                                    timestamp = line_parsed.timestamp * MSEC_PER_SEC; // convert to msec from sec
+
+                                    { /* ------------------ FIXME ------------------------  
+                                    * Temporary kludge so that metrics don't break when
+                                    * a new record has timestamp before the current one. 
+                                    */
+                                        static msec_t previous_timestamp = 0;
+                                        if((((long long) timestamp - (long long) previous_timestamp) < 0))
+                                            timestamp = previous_timestamp;
+                                        
+                                        previous_timestamp = timestamp;
+                                    }
                                 }
                             }
-                        }
-
-                        new_tmp_text_size = message_size + 1; // +1 for '\n'
 
-                        m_assert(message_size, "message_size is 0");
-                        m_assert(message, "message is NULL");
-                    }
-                    ++p;
-                    continue;
-                } /* FLB_TAIL, FLB_WEB_LOG and FLB_SERIAL case end */
-
-                /* FLB_KMSG case */
-                if(p_file_info->log_type == FLB_KMSG){
-                    if(unlikely(skip_kmsg_log_buffering)){
-                        static time_t netdata_start_time = 0;
-                        if (!netdata_start_time) netdata_start_time = now_boottime_sec();
-                        if(now_boottime_sec() - netdata_start_time < KERNEL_LOGS_COLLECT_INIT_WAIT) 
-                            goto skip_collect_and_drop_logs;
-                        else skip_kmsg_log_buffering = 0;
-                    }
+                            new_tmp_text_size = message_size + 1; // +1 for '\n'
 
-                    /* NOTE/WARNING: 
-                     * kmsg timestamps are tricky. The timestamp will be 
-                     * *wrong** if the system has gone into hibernation since 
-                     * last boot and "p_file_info->use_log_timestamp" is set. 
-                     * Even if "p_file_info->use_log_timestamp" is NOT set, we 
-                     * need to use now_realtime_msec() as Fluent Bit timestamp
-                     * will also be wrong. */
-                    if( !strncmp(p->key.via.str.ptr, "sec", (size_t) p->key.via.str.size)){
-                        if(p_file_info->use_log_timestamp){
-                            timestamp += (now_realtime_sec() - now_boottime_sec() + p->val.via.i64) * MSEC_PER_SEC;
+                            m_assert(message_size, "message_size is 0");
+                            m_assert(message, "message is NULL");
                         }
-                        else if(!timestamp)
-                            timestamp = now_realtime_msec();
-                    }
-                    else if(!strncmp(p->key.via.str.ptr, "usec", (size_t) p->key.via.str.size) && 
-                            p_file_info->use_log_timestamp){
-                        timestamp += p->val.via.i64 / USEC_PER_MS;
+
+                        break;
                     }
-                    else if(!strncmp(p->key.via.str.ptr, LOG_REC_KEY, (size_t) p->key.via.str.size)){
-                        message = (char *) p->val.via.str.ptr;
-                        message_size = p->val.via.str.size;
 
-                        m_assert(message, "message is NULL");
-                        m_assert(message_size, "message_size is 0");
+                    case FLB_KMSG:
+                    {
+                        if(unlikely(skip_kmsg_log_buffering)){
+                            static time_t netdata_start_time = 0;
+                            if (!netdata_start_time) netdata_start_time = now_boottime_sec();
+                            if(now_boottime_sec() - netdata_start_time < KERNEL_LOGS_COLLECT_INIT_WAIT) 
+                                goto skip_collect_and_drop_logs;
+                            else skip_kmsg_log_buffering = 0;
+                        }
 
-                        new_tmp_text_size += message_size + 1; // +1 for '\n'
-                    }
-                    else if(!strncmp(p->key.via.str.ptr, "priority", (size_t) p->key.via.str.size)){
-                        kmsg_sever = (int) p->val.via.u64;
-                    }
-                    ++p;
-                    continue;
-                } /* FLB_KMSG case end */
+                        /* NOTE/WARNING: 
+                        * kmsg timestamps are tricky. The timestamp will be 
+                        * *wrong** if the system has gone into hibernation since 
+                        * last boot and "p_file_info->use_log_timestamp" is set. 
+                        * Even if "p_file_info->use_log_timestamp" is NOT set, we 
+                        * need to use now_realtime_msec() as Fluent Bit timestamp
+                        * will also be wrong. */
+                        if( !strncmp(p->key.via.str.ptr, "sec", (size_t) p->key.via.str.size)){
+                            if(p_file_info->use_log_timestamp){
+                                timestamp += (now_realtime_sec() - now_boottime_sec() + p->val.via.i64) * MSEC_PER_SEC;
+                            }
+                            else if(!timestamp)
+                                timestamp = now_realtime_msec();
+                        }
+                        else if(!strncmp(p->key.via.str.ptr, "usec", (size_t) p->key.via.str.size) && 
+                                p_file_info->use_log_timestamp){
+                            timestamp += p->val.via.i64 / USEC_PER_MS;
+                        }
+                        else if(!strncmp(p->key.via.str.ptr, LOG_REC_KEY, (size_t) p->key.via.str.size)){
+                            message = (char *) p->val.via.str.ptr;
+                            message_size = p->val.via.str.size;
 
-                /* FLB_SYSTEMD or FLB_SYSLOG case */
-                if( p_file_info->log_type == FLB_SYSTEMD || 
-                    p_file_info->log_type == FLB_SYSLOG){
-                        
-                    if( p_file_info->use_log_timestamp && !strncmp( p->key.via.str.ptr, 
-                                                                    "SOURCE_REALTIME_TIMESTAMP", 
-                                                                    (size_t) p->key.via.str.size)){
+                            m_assert(message, "message is NULL");
+                            m_assert(message_size, "message_size is 0");
 
-                        m_assert(p->val.via.str.size - 3 == TIMESTAMP_MS_STR_SIZE - 1, 
-                                "p->val.via.str.size - 3 != TIMESTAMP_MS_STR_SIZE");
-                        
-                        strncpyz(timestamp_str, p->val.via.str.ptr, (size_t) p->val.via.str.size);
+                            new_tmp_text_size += message_size + 1; // +1 for '\n'
+                        }
+                        else if(!strncmp(p->key.via.str.ptr, "priority", (size_t) p->key.via.str.size)){
+                            kmsg_sever = (int) p->val.via.u64;
+                        }
 
-                        char *endptr = NULL;
-                        timestamp = str2ll(timestamp_str, &endptr);
-                        timestamp = *endptr ? 0 : timestamp / USEC_PER_MS;
+                        break;
                     }
-                    else if(!strncmp(p->key.via.str.ptr, "PRIVAL", (size_t) p->key.via.str.size)){
-                        m_assert(p->val.via.str.size <= 3, "p->val.via.str.size > 3");
-                        strncpyz(syslog_prival, p->val.via.str.ptr, (size_t) p->val.via.str.size);
-                        syslog_prival_size = (size_t) p->val.via.str.size;
-                        
-                        m_assert(syslog_prival, "syslog_prival is NULL");
-                    }
-                    else if(!strncmp(p->key.via.str.ptr, "PRIORITY", (size_t) p->key.via.str.size)){
-                        m_assert(p->val.via.str.size <= 1, "p->val.via.str.size > 1");
-                        strncpyz(syslog_severity, p->val.via.str.ptr, (size_t) p->val.via.str.size);
-                        
-                        m_assert(syslog_severity, "syslog_severity is NULL");
-                    }
-                    else if(!strncmp(p->key.via.str.ptr, "SYSLOG_FACILITY", (size_t) p->key.via.str.size)){
-                        m_assert(p->val.via.str.size <= 2, "p->val.via.str.size > 2");
-                        strncpyz(syslog_facility, p->val.via.str.ptr, (size_t) p->val.via.str.size);
-                        
-                        m_assert(syslog_facility, "syslog_facility is NULL");
-                    }
-                    else if(!strncmp(p->key.via.str.ptr, "SYSLOG_TIMESTAMP", (size_t) p->key.via.str.size)){
-                        syslog_timestamp = (char *) p->val.via.str.ptr;
-                        syslog_timestamp_size = p->val.via.str.size;
 
-                        m_assert(syslog_timestamp, "syslog_timestamp is NULL");
-                        m_assert(syslog_timestamp_size, "syslog_timestamp_size is 0");
-                        
-                        new_tmp_text_size += syslog_timestamp_size;
-                    }
-                    else if(!strncmp(p->key.via.str.ptr, "HOSTNAME", (size_t) p->key.via.str.size)){
-                        hostname = (char *) p->val.via.str.ptr;
-                        hostname_size = p->val.via.str.size;
+                    case FLB_SYSTEMD:
+                    case FLB_SYSLOG:
+                    {
+                        if( p_file_info->use_log_timestamp && !strncmp( p->key.via.str.ptr, 
+                                                                        "SOURCE_REALTIME_TIMESTAMP", 
+                                                                        (size_t) p->key.via.str.size)){
+
+                            m_assert(p->val.via.str.size - 3 == TIMESTAMP_MS_STR_SIZE - 1, 
+                                    "p->val.via.str.size - 3 != TIMESTAMP_MS_STR_SIZE");
+                            
+                            strncpyz(timestamp_str, p->val.via.str.ptr, (size_t) p->val.via.str.size);
+
+                            char *endptr = NULL;
+                            timestamp = str2ll(timestamp_str, &endptr);
+                            timestamp = *endptr ? 0 : timestamp / USEC_PER_MS;
+                        }
+                        else if(!strncmp(p->key.via.str.ptr, "PRIVAL", (size_t) p->key.via.str.size)){
+                            m_assert(p->val.via.str.size <= 3, "p->val.via.str.size > 3");
+                            strncpyz(syslog_prival, p->val.via.str.ptr, (size_t) p->val.via.str.size);
+                            syslog_prival_size = (size_t) p->val.via.str.size;
+                            
+                            m_assert(syslog_prival, "syslog_prival is NULL");
+                        }
+                        else if(!strncmp(p->key.via.str.ptr, "PRIORITY", (size_t) p->key.via.str.size)){
+                            m_assert(p->val.via.str.size <= 1, "p->val.via.str.size > 1");
+                            strncpyz(syslog_severity, p->val.via.str.ptr, (size_t) p->val.via.str.size);
+                            
+                            m_assert(syslog_severity, "syslog_severity is NULL");
+                        }
+                        else if(!strncmp(p->key.via.str.ptr, "SYSLOG_FACILITY", (size_t) p->key.via.str.size)){
+                            m_assert(p->val.via.str.size <= 2, "p->val.via.str.size > 2");
+                            strncpyz(syslog_facility, p->val.via.str.ptr, (size_t) p->val.via.str.size);
+                            
+                            m_assert(syslog_facility, "syslog_facility is NULL");
+                        }
+                        else if(!strncmp(p->key.via.str.ptr, "SYSLOG_TIMESTAMP", (size_t) p->key.via.str.size)){
+                            syslog_timestamp = (char *) p->val.via.str.ptr;
+                            syslog_timestamp_size = p->val.via.str.size;
+
+                            m_assert(syslog_timestamp, "syslog_timestamp is NULL");
+                            m_assert(syslog_timestamp_size, "syslog_timestamp_size is 0");
+                            
+                            new_tmp_text_size += syslog_timestamp_size;
+                        }
+                        else if(!strncmp(p->key.via.str.ptr, "HOSTNAME", (size_t) p->key.via.str.size)){
+                            hostname = (char *) p->val.via.str.ptr;
+                            hostname_size = p->val.via.str.size;
 
-                        m_assert(hostname, "hostname is NULL");
-                        m_assert(hostname_size, "hostname_size is 0");
+                            m_assert(hostname, "hostname is NULL");
+                            m_assert(hostname_size, "hostname_size is 0");
 
-                        new_tmp_text_size += hostname_size + 1; // +1 for ' ' char
-                    }
-                    else if(!strncmp(p->key.via.str.ptr, "SYSLOG_IDENTIFIER", (size_t) p->key.via.str.size)){
-                        syslog_identifier = (char *) p->val.via.str.ptr;
-                        syslog_identifier_size = p->val.via.str.size;
+                            new_tmp_text_size += hostname_size + 1; // +1 for ' ' char
+                        }
+                        else if(!strncmp(p->key.via.str.ptr, "SYSLOG_IDENTIFIER", (size_t) p->key.via.str.size)){
+                            syslog_identifier = (char *) p->val.via.str.ptr;
+                            syslog_identifier_size = p->val.via.str.size;
 
-                        new_tmp_text_size += syslog_identifier_size;
-                    }
-                    else if(!strncmp(p->key.via.str.ptr, "PID", (size_t) p->key.via.str.size)){
-                        pid = (char *) p->val.via.str.ptr;
-                        pid_size = p->val.via.str.size;
+                            new_tmp_text_size += syslog_identifier_size;
+                        }
+                        else if(!strncmp(p->key.via.str.ptr, "PID", (size_t) p->key.via.str.size)){
+                            pid = (char *) p->val.via.str.ptr;
+                            pid_size = p->val.via.str.size;
 
-                        new_tmp_text_size += pid_size;
-                    }
-                    else if(!strncmp(p->key.via.str.ptr, LOG_REC_KEY_SYSTEMD, (size_t) p->key.via.str.size)){
-                        
-                        message = (char *) p->val.via.str.ptr;
-                        message_size = p->val.via.str.size;
+                            new_tmp_text_size += pid_size;
+                        }
+                        else if(!strncmp(p->key.via.str.ptr, LOG_REC_KEY_SYSTEMD, (size_t) p->key.via.str.size)){
+                            
+                            message = (char *) p->val.via.str.ptr;
+                            message_size = p->val.via.str.size;
 
-                        m_assert(message, "message is NULL");
-                        m_assert(message_size, "message_size is 0");
+                            m_assert(message, "message is NULL");
+                            m_assert(message_size, "message_size is 0");
 
-                        new_tmp_text_size += message_size;
+                            new_tmp_text_size += message_size;
+                        }
+                        
+                        break;
                     }
-                    ++p;
-                    continue;
-                } /* FLB_SYSTEMD or FLB_SYSLOG case end */
 
-                /* FLB_DOCKER_EV case */
-                if(p_file_info->log_type == FLB_DOCKER_EV){ 
-                    if(!strncmp(p->key.via.str.ptr, "time", (size_t) p->key.via.str.size)){
-                        docker_ev_time = p->val.via.i64;
+                    case FLB_DOCKER_EV:
+                    { 
+                        if(!strncmp(p->key.via.str.ptr, "time", (size_t) p->key.via.str.size)){
+                            docker_ev_time = p->val.via.i64;
 
-                        m_assert(docker_ev_time, "docker_ev_time is 0");
-                    }
-                    else if(!strncmp(p->key.via.str.ptr, "timeNano", (size_t) p->key.via.str.size)){
-                        docker_ev_timeNano = p->val.via.i64;
+                            m_assert(docker_ev_time, "docker_ev_time is 0");
+                        }
+                        else if(!strncmp(p->key.via.str.ptr, "timeNano", (size_t) p->key.via.str.size)){
+                            docker_ev_timeNano = p->val.via.i64;
 
-                        m_assert(docker_ev_timeNano, "docker_ev_timeNano is 0");
+                            m_assert(docker_ev_timeNano, "docker_ev_timeNano is 0");
 
-                        if(likely(p_file_info->use_log_timestamp))
-                            timestamp = docker_ev_timeNano / NSEC_PER_MSEC;
-                    }
-                    else if(!strncmp(p->key.via.str.ptr, "Type", (size_t) p->key.via.str.size)){
-                        docker_ev_type = (char *) p->val.via.str.ptr;
-                        docker_ev_type_size = p->val.via.str.size;
+                            if(likely(p_file_info->use_log_timestamp))
+                                timestamp = docker_ev_timeNano / NSEC_PER_MSEC;
+                        }
+                        else if(!strncmp(p->key.via.str.ptr, "Type", (size_t) p->key.via.str.size)){
+                            docker_ev_type = (char *) p->val.via.str.ptr;
+                            docker_ev_type_size = p->val.via.str.size;
 
-                        m_assert(docker_ev_type, "docker_ev_type is NULL");
-                        m_assert(docker_ev_type_size, "docker_ev_type_size is 0");
+                            m_assert(docker_ev_type, "docker_ev_type is NULL");
+                            m_assert(docker_ev_type_size, "docker_ev_type_size is 0");
 
-                        // debug_log("docker_ev_type: %.*s", docker_ev_type_size, docker_ev_type);
-                    }
-                    else if(!strncmp(p->key.via.str.ptr, "Action", (size_t) p->key.via.str.size)){
-                        docker_ev_action = (char *) p->val.via.str.ptr;
-                        docker_ev_action_size = p->val.via.str.size;
+                            // debug_log("docker_ev_type: %.*s", docker_ev_type_size, docker_ev_type);
+                        }
+                        else if(!strncmp(p->key.via.str.ptr, "Action", (size_t) p->key.via.str.size)){
+                            docker_ev_action = (char *) p->val.via.str.ptr;
+                            docker_ev_action_size = p->val.via.str.size;
 
-                        m_assert(docker_ev_action, "docker_ev_action is NULL");
-                        m_assert(docker_ev_action_size, "docker_ev_action_size is 0");
+                            m_assert(docker_ev_action, "docker_ev_action is NULL");
+                            m_assert(docker_ev_action_size, "docker_ev_action_size is 0");
 
-                        // debug_log("docker_ev_action: %.*s", docker_ev_action_size, docker_ev_action);
-                    }
-                    else if(!strncmp(p->key.via.str.ptr, "id", (size_t) p->key.via.str.size)){
-                        docker_ev_id = (char *) p->val.via.str.ptr;
-                        docker_ev_id_size = p->val.via.str.size;
+                            // debug_log("docker_ev_action: %.*s", docker_ev_action_size, docker_ev_action);
+                        }
+                        else if(!strncmp(p->key.via.str.ptr, "id", (size_t) p->key.via.str.size)){
+                            docker_ev_id = (char *) p->val.via.str.ptr;
+                            docker_ev_id_size = p->val.via.str.size;
 
-                        m_assert(docker_ev_id, "docker_ev_id is NULL");
-                        m_assert(docker_ev_id_size, "docker_ev_id_size is 0");
+                            m_assert(docker_ev_id, "docker_ev_id is NULL");
+                            m_assert(docker_ev_id_size, "docker_ev_id_size is 0");
 
-                        // debug_log("docker_ev_id: %.*s", docker_ev_id_size, docker_ev_id);
-                    }
-                    else if(!strncmp(p->key.via.str.ptr, "Actor", (size_t) p->key.via.str.size)){
-                        // debug_log( "msg key:[%.*s]val:[%.*s]", (int)   p->key.via.str.size, 
-                        //                                                                     p->key.via.str.ptr, 
-                        //                                                             (int)   p->val.via.str.size, 
-                        //                                                                     p->val.via.str.ptr);
-                        if(likely(p->val.type == MSGPACK_OBJECT_MAP && p->val.via.map.size != 0)){
-                            msgpack_object_kv* ac = p->val.via.map.ptr;
-                            msgpack_object_kv* const ac_pend= p->val.via.map.ptr + p->val.via.map.size;
-                            do{
-                                if(!strncmp(ac->key.via.str.ptr, "ID", (size_t) ac->key.via.str.size)){
-                                    docker_ev_id = (char *) ac->val.via.str.ptr;
-                                    docker_ev_id_size = ac->val.via.str.size;
-
-                                    m_assert(docker_ev_id, "docker_ev_id is NULL");
-                                    m_assert(docker_ev_id_size, "docker_ev_id_size is 0");
-
-                                    // debug_log("docker_ev_id: %.*s", docker_ev_id_size, docker_ev_id);
-                                }
-                                else if(!strncmp(ac->key.via.str.ptr, "Attributes", (size_t) ac->key.via.str.size)){
-                                    if(likely(ac->val.type == MSGPACK_OBJECT_MAP && ac->val.via.map.size != 0)){
-                                        msgpack_object_kv* att = ac->val.via.map.ptr;
-                                        msgpack_object_kv* const att_pend = ac->val.via.map.ptr + ac->val.via.map.size;
-                                        do{
-                                            if(unlikely(++docker_ev_attr.size > docker_ev_attr.max_size)){
-                                                docker_ev_attr.max_size = docker_ev_attr.size;
-                                                docker_ev_attr.key = reallocz(docker_ev_attr.key, 
-                                                                            docker_ev_attr.max_size * sizeof(char *));
-                                                docker_ev_attr.val = reallocz(docker_ev_attr.val, 
-                                                                            docker_ev_attr.max_size * sizeof(char *));
-                                                docker_ev_attr.key_size = reallocz(docker_ev_attr.key_size, 
-                                                                            docker_ev_attr.max_size * sizeof(size_t));
-                                                docker_ev_attr.val_size = reallocz(docker_ev_attr.val_size, 
-                                                                            docker_ev_attr.max_size * sizeof(size_t));
-                                            }
-
-                                            docker_ev_attr.key[docker_ev_attr.size - 1] =  (char *) att->key.via.str.ptr;
-                                            docker_ev_attr.val[docker_ev_attr.size - 1] =  (char *) att->val.via.str.ptr;
-                                            docker_ev_attr.key_size[docker_ev_attr.size - 1] = (size_t) att->key.via.str.size;
-                                            docker_ev_attr.val_size[docker_ev_attr.size - 1] = (size_t) att->val.via.str.size;
-
-                                            att++;