journald: make maximum size of stream log lines configurable and bump it to 48K (#6838)

This adds a new setting LineMax= to journald.conf, and sets it by
default to 48K. When we convert stream-based stdout/stderr logging into
record-based log entries, read up to the specified amount of bytes
before forcing a line-break.

This also makes three related changes:

- When a NUL byte is read we'll not recognize this as alternative line
  break, instead of silently dropping everything after it. (see #4863)

- The reason for a line-break is now encoded in the log record, if it
  wasn't a plain newline. Specifically, we distuingish "nul",
  "line-max" and "eof", for line breaks due to NUL byte, due to the
  maximum line length as configured with LineMax= or due to end of
  stream. This data is stored in the new implicit _LINE_BREAK= field.
  It's not synthesized for plain \n line breaks.

- A randomized 128bit ID is assigned to each log stream.

With these three changes in place it's (mostly) possible to reconstruct
the original byte streams from log data, as (most) of the context of
the conversion from the byte stream to log records is saved now. (So,
the only bits we still drop are empty lines. Which might be something to
look into in a future change, and which is outside of the scope of this
work)

Fixes: https://bugs.freedesktop.org/show_bug.cgi?id=86465
See: #4863
Replaces: #4875

5 files changed, 159 insertions, 21 deletions

diff --git a/src/journal/journald-gperf.gperf b/src/journal/journald-gperf.gperf
index 522ad30cf3..6b05bda0ae 100644
--- a/src/journal/journald-gperf.gperf
+++ b/src/journal/journald-gperf.gperf
@@ -45,3 +45,4 @@ Journal.MaxLevelKMsg,       config_parse_log_level,  0, offsetof(Server, max_lev
 Journal.MaxLevelConsole,    config_parse_log_level,  0, offsetof(Server, max_level_console)
 Journal.MaxLevelWall,       config_parse_log_level,  0, offsetof(Server, max_level_wall)
 Journal.SplitMode,          config_parse_split_mode, 0, offsetof(Server, split_mode)
+Journal.LineMax,            config_parse_line_max,   0, offsetof(Server, line_max)
diff --git a/src/journal/journald-server.c b/src/journal/journald-server.c
index feef10c2db..27c2571cfc 100644
--- a/src/journal/journald-server.c
+++ b/src/journal/journald-server.c
@@ -88,6 +88,10 @@
 /* The period to insert between posting changes for coalescing */
 #define POST_CHANGE_TIMER_INTERVAL_USEC (250*USEC_PER_MSEC)
 
+/* Pick a good default that is likely to fit into AF_UNIX and AF_INET SOCK_DGRAM datagrams, and even leaves some room
+ * for a bit of additional metadata. */
+#define DEFAULT_LINE_MAX (48*1024)
+
 static int determine_path_usage(Server *s, const char *path, uint64_t *ret_used, uint64_t *ret_free) {
         _cleanup_closedir_ DIR *d = NULL;
         struct dirent *de;
@@ -1689,6 +1693,8 @@ int server_init(Server *s) {
         s->max_level_console = LOG_INFO;
         s->max_level_wall = LOG_EMERG;
 
+        s->line_max = DEFAULT_LINE_MAX;
+
         journal_reset_metrics(&s->system_storage.metrics);
         journal_reset_metrics(&s->runtime_storage.metrics);
 
@@ -1963,3 +1969,55 @@ static const char* const split_mode_table[_SPLIT_MAX] = {
 
 DEFINE_STRING_TABLE_LOOKUP(split_mode, SplitMode);
 DEFINE_CONFIG_PARSE_ENUM(config_parse_split_mode, split_mode, SplitMode, "Failed to parse split mode setting");
+
+int config_parse_line_max(
+                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) {
+
+        size_t *sz = data;
+        int r;
+
+        assert(filename);
+        assert(lvalue);
+        assert(rvalue);
+        assert(data);
+
+        if (isempty(rvalue))
+                /* Empty assignment means default */
+                *sz = DEFAULT_LINE_MAX;
+        else {
+                uint64_t v;
+
+                r = parse_size(rvalue, 1024, &v);
+                if (r < 0) {
+                        log_syntax(unit, LOG_ERR, filename, line, r, "Failed to parse LineMax= value, ignoring: %s", rvalue);
+                        return 0;
+                }
+
+                if (v < 79) {
+                        /* Why specify 79 here as minimum line length? Simply, because the most common traditional
+                         * terminal size is 80ch, and it might make sense to break one character before the natural
+                         * line break would occur on that. */
+                        log_syntax(unit, LOG_WARNING, filename, line, 0, "LineMax= too small, clamping to 79: %s", rvalue);
+                        *sz = 79;
+                } else if (v > (uint64_t) (SSIZE_MAX-1)) {
+                        /* So, why specify SSIZE_MAX-1 here? Because that's one below the largest size value read()
+                         * can return, and we need one extra byte for the trailing NUL byte. Of course IRL such large
+                         * memory allocations will fail anyway, hence this limit is mostly theoretical anyway, as we'll
+                         * fail much earlier anyway. */
+                        log_syntax(unit, LOG_WARNING, filename, line, 0, "LineMax= too large, clamping to %" PRIu64 ": %s", (uint64_t) (SSIZE_MAX-1), rvalue);
+                        *sz = SSIZE_MAX-1;
+                } else
+                        *sz = (size_t) v;
+        }
+
+        return 0;
+}
diff --git a/src/journal/journald-server.h b/src/journal/journald-server.h
index 6a4549b1ba..f4bdd588cb 100644
--- a/src/journal/journald-server.h
+++ b/src/journal/journald-server.h
@@ -169,6 +169,8 @@ struct Server {
 
         usec_t last_realtime_clock;
 
+        size_t line_max;
+
         /* Caching of client metadata */
         Hashmap *client_contexts;
         Prioq *client_contexts_lru;
@@ -192,6 +194,7 @@ void server_driver_message(Server *s, const char *message_id, const char *format
 const struct ConfigPerfItem* journald_gperf_lookup(const char *key, GPERF_LEN_TYPE length);
 
 int config_parse_storage(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_line_max(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);
 
 const char *storage_to_string(Storage s) _const_;
 Storage storage_from_string(const char *s) _pure_;
diff --git a/src/journal/journald-stream.c b/src/journal/journald-stream.c
index f074f0476f..a44c540f67 100644
--- a/src/journal/journald-stream.c
+++ b/src/journal/journald-stream.c
@@ -63,6 +63,16 @@ typedef enum StdoutStreamState {
         STDOUT_STREAM_RUNNING
 } StdoutStreamState;
 
+/* The different types of log record terminators: a real \n was read, a NUL character was read, the maximum line length
+ * was reached, or the end of the stream was reached */
+
+typedef enum LineBreak {
+        LINE_BREAK_NEWLINE,
+        LINE_BREAK_NUL,
+        LINE_BREAK_LINE_MAX,
+        LINE_BREAK_EOF,
+} LineBreak;
+
 struct StdoutStream {
         Server *server;
         StdoutStreamState state;
@@ -82,8 +92,9 @@ struct StdoutStream {
         bool fdstore:1;
         bool in_notify_queue:1;
 
-        char buffer[LINE_MAX+1];
+        char *buffer;
         size_t length;
+        size_t allocated;
 
         sd_event_source *event_source;
 
@@ -93,6 +104,8 @@ struct StdoutStream {
 
         LIST_FIELDS(StdoutStream, stdout_stream);
         LIST_FIELDS(StdoutStream, stdout_stream_notify_queue);
+
+        char id_field[sizeof("_STREAM_ID=")-1 + SD_ID128_STRING_MAX];
 };
 
 void stdout_stream_free(StdoutStream *s) {
@@ -122,6 +135,7 @@ void stdout_stream_free(StdoutStream *s) {
         free(s->identifier);
         free(s->unit_id);
         free(s->state_file);
+        free(s->buffer);
 
         free(s);
 }
@@ -172,12 +186,14 @@ static int stdout_stream_save(StdoutStream *s) {
                 "LEVEL_PREFIX=%i\n"
                 "FORWARD_TO_SYSLOG=%i\n"
                 "FORWARD_TO_KMSG=%i\n"
-                "FORWARD_TO_CONSOLE=%i\n",
+                "FORWARD_TO_CONSOLE=%i\n"
+                "STREAM_ID=%s\n",
                 s->priority,
                 s->level_prefix,
                 s->forward_to_syslog,
                 s->forward_to_kmsg,
-                s->forward_to_console);
+                s->forward_to_console,
+                s->id_field + strlen("_STREAM_ID="));
 
         if (!isempty(s->identifier)) {
                 _cleanup_free_ char *escaped;
@@ -234,8 +250,8 @@ fail:
         return log_error_errno(r, "Failed to save stream data %s: %m", s->state_file);
 }
 
-static int stdout_stream_log(StdoutStream *s, const char *p) {
-        struct iovec iovec[N_IOVEC_META_FIELDS + 5];
+static int stdout_stream_log(StdoutStream *s, const char *p, LineBreak line_break) {
+        struct iovec iovec[N_IOVEC_META_FIELDS + 7];
         int priority;
         char syslog_priority[] = "PRIORITY=\0";
         char syslog_facility[sizeof("SYSLOG_FACILITY=")-1 + DECIMAL_STR_MAX(int) + 1];
@@ -268,6 +284,8 @@ static int stdout_stream_log(StdoutStream *s, const char *p) {
 
         IOVEC_SET_STRING(iovec[n++], "_TRANSPORT=stdout");
 
+        IOVEC_SET_STRING(iovec[n++], s->id_field);
+
         syslog_priority[strlen("PRIORITY=")] = '0' + LOG_PRI(priority);
         IOVEC_SET_STRING(iovec[n++], syslog_priority);
 
@@ -282,6 +300,18 @@ static int stdout_stream_log(StdoutStream *s, const char *p) {
                         IOVEC_SET_STRING(iovec[n++], syslog_identifier);
         }
 
+        if (line_break != LINE_BREAK_NEWLINE) {
+                const char *c;
+
+                /* If this log message was generated due to an uncommon line break then mention this in the log
+                 * entry */
+
+                c =     line_break == LINE_BREAK_NUL ?      "_LINE_BREAK=nul" :
+                        line_break == LINE_BREAK_LINE_MAX ? "_LINE_BREAK=line-max" :
+                                                            "_LINE_BREAK=eof";
+                IOVEC_SET_STRING(iovec[n++], c);
+        }
+
         message = strappend("MESSAGE=", p);
         if (message)
                 IOVEC_SET_STRING(iovec[n++], message);
@@ -298,7 +328,7 @@ static int stdout_stream_log(StdoutStream *s, const char *p) {
         return 0;
 }
 
-static int stdout_stream_line(StdoutStream *s, char *p) {
+static int stdout_stream_line(StdoutStream *s, char *p, LineBreak line_break) {
         int r;
         char *orig;
 
@@ -308,6 +338,12 @@ static int stdout_stream_line(StdoutStream *s, char *p) {
         orig = p;
         p = strstrip(p);
 
+        /* line breaks by NUL, line max length or EOF are not permissible during the negotiation part of the protocol */
+        if (line_break != LINE_BREAK_NEWLINE && s->state != STDOUT_STREAM_RUNNING) {
+                log_warning("Control protocol line not properly terminated.");
+                return -EINVAL;
+        }
+
         switch (s->state) {
 
         case STDOUT_STREAM_IDENTIFIER:
@@ -390,7 +426,7 @@ static int stdout_stream_line(StdoutStream *s, char *p) {
                 return 0;
 
         case STDOUT_STREAM_RUNNING:
-                return stdout_stream_log(s, orig);
+                return stdout_stream_log(s, orig, line_break);
         }
 
         assert_not_reached("Unknown stream state");
@@ -409,21 +445,31 @@ static int stdout_stream_scan(StdoutStream *s, bool force_flush) {
         /* XXX: This function does nothing if (s->length == 0) */
 
         for (;;) {
-                char *end;
+                LineBreak line_break;
                 size_t skip;
-
-                end = memchr(p, '\n', remaining);
-                if (end)
-                        skip = end - p + 1;
-                else if (remaining >= sizeof(s->buffer) - 1) {
-                        end = p + sizeof(s->buffer) - 1;
+                char *end1, *end2;
+
+                end1 = memchr(p, '\n', remaining);
+                end2 = memchr(p, 0, end1 ? (size_t) (end1 - p) : remaining);
+
+                if (end2) {
+                        /* We found a NUL terminator */
+                        skip = end2 - p + 1;
+                        line_break = LINE_BREAK_NUL;
+                } else if (end1) {
+                        /* We found a \n terminator */
+                        *end1 = 0;
+                        skip = end1 - p + 1;
+                        line_break = LINE_BREAK_NEWLINE;
+                } else if (remaining >= s->server->line_max) {
+                        /* Force a line break after the maximum line length */
+                        *(p + s->server->line_max) = 0;
                         skip = remaining;
+                        line_break = LINE_BREAK_LINE_MAX;
                 } else
                         break;
 
-                *end = 0;
-
-                r = stdout_stream_line(s, p);
+                r = stdout_stream_line(s, p, line_break);
                 if (r < 0)
                         return r;
 
@@ -433,7 +479,7 @@ static int stdout_stream_scan(StdoutStream *s, bool force_flush) {
 
         if (force_flush && remaining > 0) {
                 p[remaining] = 0;
-                r = stdout_stream_line(s, p);
+                r = stdout_stream_line(s, p, LINE_BREAK_EOF);
                 if (r < 0)
                         return r;
 
@@ -451,6 +497,7 @@ static int stdout_stream_scan(StdoutStream *s, bool force_flush) {
 
 static int stdout_stream_process(sd_event_source *es, int fd, uint32_t revents, void *userdata) {
         StdoutStream *s = userdata;
+        size_t limit;
         ssize_t l;
         int r;
 
@@ -461,9 +508,20 @@ static int stdout_stream_process(sd_event_source *es, int fd, uint32_t revents,
                 goto terminate;
         }
 
-        l = read(s->fd, s->buffer+s->length, sizeof(s->buffer)-1-s->length);
-        if (l < 0) {
+        /* If the buffer is full already (discounting the extra NUL we need), add room for another 1K */
+        if (s->length + 1 >= s->allocated) {
+                if (!GREEDY_REALLOC(s->buffer, s->allocated, s->length + 1 + 1024)) {
+                        log_oom();
+                        goto terminate;
+                }
+        }
 
+        /* Try to make use of the allocated buffer in full, but never read more than the configured line size. Also,
+         * always leave room for a terminating NUL we might need to add. */
+        limit = MIN(s->allocated - 1, s->server->line_max);
+
+        l = read(s->fd, s->buffer + s->length, limit - s->length);
+        if (l < 0) {
                 if (errno == EAGAIN)
                         return 0;
 
@@ -490,11 +548,16 @@ terminate:
 
 static int stdout_stream_install(Server *s, int fd, StdoutStream **ret) {
         _cleanup_(stdout_stream_freep) StdoutStream *stream = NULL;
+        sd_id128_t id;
         int r;
 
         assert(s);
         assert(fd >= 0);
 
+        r = sd_id128_randomize(&id);
+        if (r < 0)
+                return log_error_errno(r, "Failed to generate stream ID: %m");
+
         stream = new0(StdoutStream, 1);
         if (!stream)
                 return log_oom();
@@ -502,6 +565,8 @@ static int stdout_stream_install(Server *s, int fd, StdoutStream **ret) {
         stream->fd = -1;
         stream->priority = LOG_INFO;
 
+        xsprintf(stream->id_field, "_STREAM_ID=" SD_ID128_FORMAT_STR, SD_ID128_FORMAT_VAL(id));
+
         r = getpeercred(fd, &stream->ucred);
         if (r < 0)
                 return log_error_errno(r, "Failed to determine peer credentials: %m");
@@ -575,7 +640,8 @@ static int stdout_stream_load(StdoutStream *stream, const char *fname) {
                 *level_prefix = NULL,
                 *forward_to_syslog = NULL,
                 *forward_to_kmsg = NULL,
-                *forward_to_console = NULL;
+                *forward_to_console = NULL,
+                *stream_id = NULL;
         int r;
 
         assert(stream);
@@ -595,6 +661,7 @@ static int stdout_stream_load(StdoutStream *stream, const char *fname) {
                            "FORWARD_TO_CONSOLE", &forward_to_console,
                            "IDENTIFIER", &stream->identifier,
                            "UNIT", &stream->unit_id,
+                           "STREAM_ID", &stream_id,
                            NULL);
         if (r < 0)
                 return log_error_errno(r, "Failed to read: %s", stream->state_file);
@@ -631,6 +698,14 @@ static int stdout_stream_load(StdoutStream *stream, const char *fname) {
                         stream->forward_to_console = r;
         }
 
+        if (stream_id) {
+                sd_id128_t id;
+
+                r = sd_id128_from_string(stream_id, &id);
+                if (r >= 0)
+                        xsprintf(stream->id_field, "_STREAM_ID=" SD_ID128_FORMAT_STR, SD_ID128_FORMAT_VAL(id));
+        }
+
         return 0;
 }
 
diff --git a/src/journal/journald.conf b/src/journal/journald.conf
index 2541b949be..03fce40e90 100644
--- a/src/journal/journald.conf
+++ b/src/journal/journald.conf
@@ -39,3 +39,4 @@
 #MaxLevelKMsg=notice
 #MaxLevelConsole=info
 #MaxLevelWall=emerg
+#LineMax=48K