logging: buffer partial messages to journald

Ensures that log lines less than or equal to 8k in size are logged as a single journald message by buffering partial messages from subsequent read calls until a newline is observed (or buffer is full).

Fixes: #242
Signed-off-by: Cory Snyder <[email protected]>

src/ctr_logging.c

@@ -1,6 +1,7 @@
 #define _GNU_SOURCE
 #include "ctr_logging.h"
 #include "cli.h"
+#include "config.h"
 #include <string.h>
 
 // if the systemd development files were found, we can log to systemd
@@ -230,10 +231,19 @@ bool write_to_logs(stdpipe_t pipe, char *buf, ssize_t num_read)
 
 
 /* write to systemd journal. If the pipe is stdout, write with notice priority,
- * otherwise, write with error priority
+ * otherwise, write with error priority. Partial lines (that don't end in a newline) are buffered
+ * between invocations. A 0 buflen argument forces a buffered partial line to be flushed.
  */
 int write_journald(int pipe, char *buf, ssize_t buflen)
 {
+	static char stdout_partial_buf[STDIO_BUF_SIZE];
+	static size_t stdout_partial_buf_len = 0;
+	static char stderr_partial_buf[STDIO_BUF_SIZE];
+	static size_t stderr_partial_buf_len = 0;
+
+	char *partial_buf;
+	size_t *partial_buf_len;
+
 	/* When using writev_buffer_append_segment, we should never approach the number of
 	 * entries necessary to flush the buffer. Therefore, the fd passed in is for /dev/null
 	 */
@@ -246,30 +256,47 @@ int write_journald(int pipe, char *buf, ssize_t buflen)
 	 * to be changed if this convention is changed.
 	 */
 	const char *message_priority = "PRIORITY=6";
-	if (pipe == STDERR_PIPE)
+	if (pipe == STDERR_PIPE) {
 		message_priority = "PRIORITY=3";
+		partial_buf = stderr_partial_buf;
+		partial_buf_len = &stderr_partial_buf_len;
+	} else {
+		partial_buf = stdout_partial_buf;
+		partial_buf_len = &stdout_partial_buf_len;
+	}
 
 	ptrdiff_t line_len = 0;
 
-	while (buflen > 0) {
+	while (buflen > 0 || *partial_buf_len > 0) {
 		writev_buffer_t bufv = {0};
 
-		bool partial = get_line_len(&line_len, buf, buflen);
+		bool partial = buflen == 0 || get_line_len(&line_len, buf, buflen);
+
+		/* If this is a partial line, and we have capacity to buffer it, buffer it and return.
+		 * The capacity of the partial_buf is one less than its size so that we can always add
+		 * a null terminating char later */
+		if (buflen && partial && ((unsigned long)line_len < (STDIO_BUF_SIZE - *partial_buf_len))) {
+			memcpy(partial_buf + *partial_buf_len, buf, line_len);
+			*partial_buf_len += line_len;
+			return 0;
+		}
+
 		/* sd_journal_* doesn't have an option to specify the number of bytes to write in the message, and instead writes the
 		 * entire string. Copying every line doesn't make very much sense, so instead we do this tmp_line_end
 		 * hack to emulate separate strings.
 		 */
 		char tmp_line_end = buf[line_len];
 		buf[line_len] = '\0';
 
-		_cleanup_free_ char *message = g_strdup_printf("MESSAGE=%s", buf);
-		if (writev_buffer_append_segment(dev_null, &bufv, message, line_len + MESSAGE_EQ_LEN) < 0)
+		ssize_t msg_len = line_len + MESSAGE_EQ_LEN + *partial_buf_len;
+		partial_buf[*partial_buf_len] = '\0';
+		_cleanup_free_ char *message = g_strdup_printf("MESSAGE=%s%s", partial_buf, buf);
+		if (writev_buffer_append_segment(dev_null, &bufv, message, msg_len) < 0)
 			return -1;
 
 		/* Restore state of the buffer */
 		buf[line_len] = tmp_line_end;
 
-
 		if (writev_buffer_append_segment(dev_null, &bufv, container_id_full, cuuid_len + CID_FULL_EQ_LEN) < 0)
 			return -1;
 
@@ -301,6 +328,7 @@ int write_journald(int pipe, char *buf, ssize_t buflen)
 
 		buf += line_len;
 		buflen -= line_len;
+		*partial_buf_len = 0;
 	}
 	return 0;
 }

src/ctr_stdio.c

@@ -11,6 +11,7 @@
 
 static gboolean tty_hup_timeout_scheduled = false;
 static bool read_stdio(int fd, stdpipe_t pipe, gboolean *eof);
+static void drain_log_buffers(stdpipe_t pipe);
 static gboolean tty_hup_timeout_cb(G_GNUC_UNUSED gpointer user_data);
 
 
@@ -90,12 +91,23 @@ void drain_stdio()
 		while (read_stdio(mainfd_stdout, STDOUT_PIPE, NULL))
 			;
 	}
+	drain_log_buffers(STDOUT_PIPE);
 	if (mainfd_stderr != -1) {
 		g_unix_set_fd_nonblocking(mainfd_stderr, TRUE, NULL);
 		while (read_stdio(mainfd_stderr, STDERR_PIPE, NULL))
 			;
 	}
-	return;
+	drain_log_buffers(STDERR_PIPE);
+}
+
+/* the journald log writer is buffering partial lines so that whole log lines are emitted
+ * to the journal as a unit. this flushes those buffers */
+static void drain_log_buffers(stdpipe_t pipe)
+{
+	/* We pass a single byte buffer because write_to_logs expects that there is one
+	   byte of capacity beyond the buflen that we specify */
+	char buf;
+	write_to_logs(pipe, &buf, 0);
 }
 
 static bool read_stdio(int fd, stdpipe_t pipe, gboolean *eof)