Bugfix + features for lessening forwarding load

README.md

@@ -130,6 +130,9 @@ Reload after changes: `sudo systemctl reload systemd-netlogd`
 | `UseSysLogMsgId=` | Extract `SYSLOG_MSGID` from journal | `false` |
 | `ExcludeSyslogFacility=` | Space-separated facility list to exclude | None |
 | `ExcludeSyslogLevel=` | Space-separated level list to exclude | None |
+| `ExcludeUnits=` | Space-separated list of systemd units to exclude | None |
+| `ChunkMaxLineCount=` | Maximum lines to read & forward from the journal at once | `100` |
+| `ChunkDelaySec=` | Delay after ChunkMaxLineCount hit | `1s` |
 
 **Facilities:** `kern`, `user`, `mail`, `daemon`, `auth`, `syslog`, `lpr`, `news`, `uucp`, `cron`, `authpriv`, `ftp`, `ntp`, `security`, `console`, `solaris-cron`, `local0`-`local7`
 

conf/netlogd.conf.in

@@ -18,3 +18,7 @@
 #SendBuffer=
 #ExcludeSyslogFacility=
 #ExcludeSyslogLevel=
+#ExcludeUnits=
+#ChunkMaxLineCount=100
+#ChunkDelaySec=1
+

doc/index.rst

@@ -80,6 +80,9 @@ Option                        Type    Default       Description
 ``UseSysLogMsgId=``           bool    ``false``     Extract and use ``SYSLOG_MSGID`` field from journal entries.
 ``ExcludeSyslogFacility=``    list    –             Space-separated list of facilities to exclude (e.g., ``auth authpriv``).
 ``ExcludeSyslogLevel=``       list    –             Space-separated list of log levels to exclude (e.g., ``debug info``).
+``ExcludeUnits=``             list    –             Space-separated list of systemd units to exclude (e.g., ``logrotate.service avahi-daemon.service``).
+``ChunkMaxLineCount=``        int     100           Maximum number of lines to read from the journal in a single chunk (mainly during initial bulk send). Must be greater than zero. Between chunks, update the saved position on disk, notify the watchdog, and wait ``ChunkDelaySec`` before reading the next chunk.
+``ChunkDelaySec=``            sec     1             Delay between reading chunks from the journal. See also ``ChunkMaxLineCount``.
 ============================  ======  ============  ================================================================================================
 
 **Facilities**: ``kern``, ``user``, ``mail``, ``daemon``, ``auth``, ``syslog``, ``lpr``, ``news``, ``uucp``, ``cron``, ``authpriv``, ``ftp``, ``ntp``, ``security``, ``console``, ``solaris-cron``, ``local0``–``local7``.

src/netlog/netlog-conf.c

@@ -10,6 +10,7 @@
 #include "parse-util.h"
 #include "sd-resolve.h"
 #include "string-util.h"
+#include "strv.h"
 
 int config_parse_netlog_remote_address(const char *unit,
                                        const char *filename,
@@ -282,6 +283,87 @@ int config_parse_syslog_level(const char *unit,
         return 0;
 }
 
+int config_parse_string_list(const char *unit,
+                             const char *filename,
+                             unsigned line,
+                             const char *section,
+                             unsigned section_line,
+                             const char *lvalue,
+                             int ltype,
+                             const char *rvalue,
+                             void *data,
+                             void *userdata) {
+        char **strv = strv_new(NULL, NULL);
+        char ***strv_p = data;
+        int r;
+
+        assert(filename);
+        assert(lvalue);
+        assert(rvalue);
+        assert(strv_p);
+
+        if (!strv)
+                return log_oom();
+
+        for (const char *p = rvalue;;) {
+                char *word = NULL;
+
+                r = extract_first_word(&p, &word, NULL, EXTRACT_QUOTES|EXTRACT_RELAX);
+                if (r < 0) {
+                        strv_free(strv);
+                        log_syntax(unit, LOG_WARNING, filename, line, r, "Failed to parse %s= specifier '%s', ignoring: %m", lvalue, rvalue);
+                        return 0;
+                }
+                if (r == 0)
+                        break;
+
+                r = strv_consume(&strv, word);
+                if (r < 0) {
+                        strv_free(strv);
+                        return log_oom();
+                }
+        }
+
+        if (*strv_p)
+                strv_free(*strv_p);
+        *strv_p = strv;
+        return 0;
+}
+
+int config_parse_chunk_max_line_count(const char *unit,
+                                      const char *filename,
+                                      unsigned line,
+                                      const char *section,
+                                      unsigned section_line,
+                                      const char *lvalue,
+                                      int ltype,
+                                      const char *rvalue,
+                                      void *data,
+                                      void *userdata) {
+        unsigned *line_count = data;
+        unsigned u;
+        int r;
+
+        assert(filename);
+        assert(lvalue);
+        assert(rvalue);
+        assert(data);
+        assert(userdata);
+
+        r = safe_atou(rvalue, &u);
+        if (r < 0) {
+                log_syntax(unit, LOG_WARNING, filename, line, -r, "Failed to parse '%s=%s', ignoring.", lvalue, rvalue);
+                return 0;
+        }
+        if (u == 0) {
+                log_syntax(unit, LOG_WARNING, filename, line, 0, "%s= must be greater than zero, ignoring.", lvalue);
+                return 0;
+        }
+
+        *line_count = u;
+        return 0;
+}
+
 int manager_parse_config_file(Manager *m) {
         int r;
 

src/netlog/netlog-conf.h

@@ -84,4 +84,26 @@ int config_parse_syslog_level(const char *unit,
                               void *data,
                               void *userdata);
 
+int config_parse_string_list(const char *unit,
+                             const char *filename,
+                             unsigned line,
+                             const char *section,
+                             unsigned section_line,
+                             const char *lvalue,
+                             int ltype,
+                             const char *rvalue,
+                             void *data,
+                             void *userdata);
+
+int config_parse_chunk_max_line_count(const char *unit,
+                                      const char *filename,
+                                      unsigned line,
+                                      const char *section,
+                                      unsigned section_line,
+                                      const char *lvalue,
+                                      int ltype,
+                                      const char *rvalue,
+                                      void *data,
+                                      void *userdata);
+
 int manager_parse_config_file(Manager *m);

src/netlog/netlog-gperf.gperf

@@ -34,3 +34,6 @@ Network.NoDelay,                  config_parse_bool,                      0, off
 Network.SendBuffer,               config_parse_iec_size,                  0, offsetof(Manager, send_buffer)
 Network.ExcludeSyslogFacility,    config_parse_syslog_facility,           0, offsetof(Manager, excluded_syslog_facilities)
 Network.ExcludeSyslogLevel,       config_parse_syslog_level,              0, offsetof(Manager, excluded_syslog_levels)
+Network.ExcludeUnits,             config_parse_string_list,               0, offsetof(Manager, excluded_systemd_units)
+Network.ChunkMaxLineCount,        config_parse_chunk_max_line_count,      0, offsetof(Manager, chunk_max_line_count)
+Network.ChunkDelaySec,            config_parse_sec,                       0, offsetof(Manager, chunk_delay_interval)

src/netlog/netlog-journal.c

@@ -3,11 +3,13 @@
 #include "netlog-journal.h"
 
 #include <systemd/sd-journal.h>
+#include <systemd/sd-daemon.h>
 
 #include "alloc-util.h"
 #include "netlog-manager.h"
 #include "netlog-state.h"
 #include "parse-util.h"
+#include "strv.h"
 
 /* Default severity LOG_NOTICE */
 #define JOURNAL_DEFAULT_SEVERITY LOG_PRI(LOG_NOTICE)
@@ -91,6 +93,7 @@ static int parse_fieldv(
 static int parse_journal_fields(Manager *m,
                                 char **message,
                                 char **identifier,
+                                char **unit,
                                 char **hostname,
                                 char **pid,
                                 char **facility,
@@ -101,14 +104,15 @@ static int parse_journal_fields(Manager *m,
         size_t length;
         int r;
         size_t hostname_len = 0, identifier_len = 0, message_len = 0, priority_len = 0, facility_len = 0,
-                structured_data_len = 0, msgid_len = 0, pid_len = 0;
+                structured_data_len = 0, msgid_len = 0, pid_len = 0, unit_len = 0;
         const ParseFieldVec fields[] = {
                 PARSE_FIELD_VEC_ENTRY("_PID=",                        pid,               &pid_len              ),
                 PARSE_FIELD_VEC_ENTRY("MESSAGE=",                     message,           &message_len          ),
                 PARSE_FIELD_VEC_ENTRY("PRIORITY=",                    priority,          &priority_len         ),
                 PARSE_FIELD_VEC_ENTRY("_HOSTNAME=",                   hostname,          &hostname_len         ),
                 PARSE_FIELD_VEC_ENTRY("SYSLOG_FACILITY=",             facility,          &facility_len         ),
                 PARSE_FIELD_VEC_ENTRY("SYSLOG_IDENTIFIER=",           identifier,        &identifier_len       ),
+                PARSE_FIELD_VEC_ENTRY("_SYSTEMD_UNIT=",               unit,              &unit_len             ),
                 PARSE_FIELD_VEC_ENTRY("SYSLOG_STRUCTURED_DATA=",      structured_data,   &structured_data_len  ),
                 PARSE_FIELD_VEC_ENTRY("SYSLOG_MSGID",                 msgid,             &msgid_len            ),
         };
@@ -124,7 +128,7 @@ static int parse_journal_fields(Manager *m,
                 return 0;
         }
 
-        return r;
+        return r < 0 ? r : 1;
 }
 
 static int parse_syslog_severity(Manager *m, const char *priority, unsigned *sev) {
@@ -177,9 +181,26 @@ static int parse_syslog_facility(Manager *m, const char *facility, unsigned *fac
         return 0;
 }
 
+static int check_systemd_unit(Manager *m, const char* unit) {
+        int r;
+
+        if (!unit)
+                return 0;
+
+        if (m->excluded_systemd_units) {
+                r = strv_contains(m->excluded_systemd_units, unit);
+                if (r) {
+                        log_debug("Skipping message with excluded systemd unit %s.", unit);
+                        return 1; /* filtered */
+                }
+        }
+
+        return 0;
+}
+
 static int journal_read_input(Manager *m) {
         _cleanup_free_ char *facility = NULL, *identifier = NULL, *priority = NULL, *message = NULL, *pid = NULL,
-                *hostname = NULL, *structured_data = NULL, *msgid = NULL, *cursor = NULL;
+                *hostname = NULL, *structured_data = NULL, *msgid = NULL, *cursor = NULL, *unit = NULL;
         unsigned sev = JOURNAL_DEFAULT_SEVERITY;
         unsigned fac = JOURNAL_DEFAULT_FACILITY;
         struct timeval tv, *tvp = NULL;
@@ -195,7 +216,8 @@ static int journal_read_input(Manager *m) {
 
         log_debug("Reading from journal cursor=%s", cursor);
 
-        r = parse_journal_fields(m, &message, &identifier, &hostname, &pid, &facility, &priority, &structured_data, &msgid);
+        r = parse_journal_fields(m, &message, &identifier, &unit, &hostname, &pid, &facility, &priority,
+                                 &structured_data, &msgid);
         if (r < 0)
                 return log_error_errno(r, "Failed to get journal fields: %m");
         if (r == 0)
@@ -227,6 +249,10 @@ static int journal_read_input(Manager *m) {
         if (r > 0) /* filtered */
                 return 0;
 
+        r = check_systemd_unit(m, unit);
+        if (r > 0) /* filtered */
+                return 0;
+
         return manager_push_to_network(m,
                                        sev,
                                        fac,
@@ -241,6 +267,8 @@ static int journal_read_input(Manager *m) {
 static int journal_process_input(Manager *m) {
         _cleanup_free_ char *cursor = NULL;
         int r;
+        unsigned count = 0;
+        bool delay = false;
 
         assert(m);
         assert(m->journal);
@@ -262,6 +290,11 @@ static int journal_process_input(Manager *m) {
 
                         break;
                 }
+                if (++count >= m->chunk_max_line_count) {
+                        log_debug("Reached chunk max line count %u, delaying next read.", m->chunk_max_line_count);
+                        delay = true;
+                        break;
+                }
         }
 
         r = sd_journal_get_cursor(m->journal, &cursor);
@@ -274,7 +307,38 @@ static int journal_process_input(Manager *m) {
         m->last_cursor = cursor;
         cursor = NULL;
 
-        return state_update_cursor(m);
+        r = state_update_cursor(m);
+        if (r < 0)
+                return r; /* error */
+        else if (delay)
+                return 1; /* delay next read */
+        else
+                return r;
+}
+
+int journal_process_with_delay(Manager *m) {
+        int r, input_r;
+
+        input_r = journal_process_input(m);
+        if (input_r == 1) {
+                log_trace("Pausing journal for a delay.");
+                r = sd_event_source_set_time_relative(m->event_journal_delay, m->chunk_delay_interval);
+                if (r < 0)
+                        return log_error_errno(r, "Failed to set next chunk timer event: %m");
+                r = sd_event_source_set_enabled(m->event_journal_delay, SD_EVENT_ONESHOT);
+                if (r < 0)
+                        return log_error_errno(r, "Failed to enable next chunk timer event: %m");
+                r = sd_event_source_set_enabled(m->event_journal_input, SD_EVENT_OFF);
+                if (r < 0)
+                        return log_error_errno(r, "Failed to disable journal input event: %m");
+                sd_notify(false, "STATUS=Paused (inter-chunk delay).");
+        } else {
+                r = sd_event_source_set_enabled(m->event_journal_input, SD_EVENT_ON);
+                if (r < 0)
+                        return log_error_errno(r, "Failed to enable journal input event: %m");
+                sd_notify(false, "STATUS=Ready.");
+        }
+        return input_r;
 }
 
 int journal_event_handler(sd_event_source *event, int fd, uint32_t revents, void *userp) {
@@ -305,7 +369,17 @@ int journal_event_handler(sd_event_source *event, int fd, uint32_t revents, void
         if (r == SD_JOURNAL_NOP)
                 return 0;
 
-        return journal_process_input(m);
+        return journal_process_with_delay(m);
+}
+
+int journal_event_resume(sd_event_source *event, uint64_t usec, void *userp) {
+        Manager *m = userp;
+
+        assert(m);
+        assert(m->journal);
+        log_trace("Resuming journal input after delay.");
+
+        return journal_process_with_delay(m);
 }
 
 void journal_close_input(Manager *m) {
@@ -362,6 +436,10 @@ int journal_monitor_listen(Manager *m) {
         if (r < 0)
                 return log_error_errno(r, "Failed to register input event: %m");
 
+        r = sd_event_add_time(m->event, &m->event_journal_delay, CLOCK_MONOTONIC, UINT64_MAX, 0, journal_event_resume, m);
+        if (r < 0)
+                return log_error_errno(r, "Failed to register timer-based resume send event: %m");
+
         /* ignore failure */
         if (!m->last_cursor)
                 (void) state_load_cursor(m);

src/netlog/netlog-journal.h

@@ -8,4 +8,6 @@ typedef struct Manager Manager;
 
 int journal_monitor_listen(Manager *m);
 int journal_event_handler(sd_event_source *event, int fd, uint32_t revents, void *userp);
+int journal_event_resume(sd_event_source *event, uint64_t usec, void *userp);
+int journal_process_with_delay(Manager *m);
 void journal_close_input(Manager *m);