No functional change.
src/basic/format-util.h \
src/basic/nss-util.h \
src/basic/khash.h \
- src/basic/khash.c
+ src/basic/khash.c \
+ src/basic/journal-importer.h \
+ src/basic/journal-importer.c
nodist_libbasic_la_SOURCES = \
src/basic/errno-from-name.h \
--- /dev/null
+/***
+ This file is part of systemd.
+
+ Copyright 2014 Zbigniew Jędrzejewski-Szmek
+
+ systemd is free software; you can redistribute it and/or modify it
+ under the terms of the GNU Lesser General Public License as published by
+ the Free Software Foundation; either version 2.1 of the License, or
+ (at your option) any later version.
+
+ systemd is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ Lesser General Public License for more details.
+
+ You should have received a copy of the GNU Lesser General Public License
+ along with systemd; If not, see <http://www.gnu.org/licenses/>.
+***/
+
+#include <unistd.h>
+
+#include "alloc-util.h"
+#include "journal-importer.h"
+#include "fd-util.h"
+#include "parse-util.h"
+#include "string-util.h"
+
+enum {
+ IMPORTER_STATE_LINE = 0, /* waiting to read, or reading line */
+ IMPORTER_STATE_DATA_START, /* reading binary data header */
+ IMPORTER_STATE_DATA, /* reading binary data */
+ IMPORTER_STATE_DATA_FINISH, /* expecting newline */
+ IMPORTER_STATE_EOF, /* done */
+};
+
+static int iovw_put(struct iovec_wrapper *iovw, void* data, size_t len) {
+ if (!GREEDY_REALLOC(iovw->iovec, iovw->size_bytes, iovw->count + 1))
+ return log_oom();
+
+ iovw->iovec[iovw->count++] = (struct iovec) {data, len};
+ return 0;
+}
+
+static void iovw_free_contents(struct iovec_wrapper *iovw) {
+ iovw->iovec = mfree(iovw->iovec);
+ iovw->size_bytes = iovw->count = 0;
+}
+
+static void iovw_rebase(struct iovec_wrapper *iovw, char *old, char *new) {
+ size_t i;
+
+ for (i = 0; i < iovw->count; i++)
+ iovw->iovec[i].iov_base = (char*) iovw->iovec[i].iov_base - old + new;
+}
+
+size_t iovw_size(struct iovec_wrapper *iovw) {
+ size_t n = 0, i;
+
+ for (i = 0; i < iovw->count; i++)
+ n += iovw->iovec[i].iov_len;
+
+ return n;
+}
+
+void journal_importer_cleanup(JournalImporter *imp) {
+ if (imp->fd >= 0 && !imp->passive_fd) {
+ log_debug("Closing %s (fd=%d)", imp->name ?: "importer", imp->fd);
+ safe_close(imp->fd);
+ }
+
+ free(imp->buf);
+ iovw_free_contents(&imp->iovw);
+}
+
+static char* realloc_buffer(JournalImporter *imp, size_t size) {
+ char *b, *old = imp->buf;
+
+ b = GREEDY_REALLOC(imp->buf, imp->size, size);
+ if (!b)
+ return NULL;
+
+ iovw_rebase(&imp->iovw, old, imp->buf);
+
+ return b;
+}
+
+static int get_line(JournalImporter *imp, char **line, size_t *size) {
+ ssize_t n;
+ char *c = NULL;
+
+ assert(imp);
+ assert(imp->state == IMPORTER_STATE_LINE);
+ assert(imp->offset <= imp->filled);
+ assert(imp->filled <= imp->size);
+ assert(imp->buf == NULL || imp->size > 0);
+ assert(imp->fd >= 0);
+
+ for (;;) {
+ if (imp->buf) {
+ size_t start = MAX(imp->scanned, imp->offset);
+
+ c = memchr(imp->buf + start, '\n',
+ imp->filled - start);
+ if (c != NULL)
+ break;
+ }
+
+ imp->scanned = imp->filled;
+ if (imp->scanned >= DATA_SIZE_MAX) {
+ log_error("Entry is bigger than %u bytes.", DATA_SIZE_MAX);
+ return -E2BIG;
+ }
+
+ if (imp->passive_fd)
+ /* we have to wait for some data to come to us */
+ return -EAGAIN;
+
+ /* We know that imp->filled is at most DATA_SIZE_MAX, so if
+ we reallocate it, we'll increase the size at least a bit. */
+ assert_cc(DATA_SIZE_MAX < ENTRY_SIZE_MAX);
+ if (imp->size - imp->filled < LINE_CHUNK &&
+ !realloc_buffer(imp, MIN(imp->filled + LINE_CHUNK, ENTRY_SIZE_MAX)))
+ return log_oom();
+
+ assert(imp->buf);
+ assert(imp->size - imp->filled >= LINE_CHUNK ||
+ imp->size == ENTRY_SIZE_MAX);
+
+ n = read(imp->fd,
+ imp->buf + imp->filled,
+ imp->size - imp->filled);
+ if (n < 0) {
+ if (errno != EAGAIN)
+ log_error_errno(errno, "read(%d, ..., %zu): %m",
+ imp->fd,
+ imp->size - imp->filled);
+ return -errno;
+ } else if (n == 0)
+ return 0;
+
+ imp->filled += n;
+ }
+
+ *line = imp->buf + imp->offset;
+ *size = c + 1 - imp->buf - imp->offset;
+ imp->offset += *size;
+
+ return 1;
+}
+
+static int fill_fixed_size(JournalImporter *imp, void **data, size_t size) {
+
+ assert(imp);
+ assert(imp->state == IMPORTER_STATE_DATA_START ||
+ imp->state == IMPORTER_STATE_DATA ||
+ imp->state == IMPORTER_STATE_DATA_FINISH);
+ assert(size <= DATA_SIZE_MAX);
+ assert(imp->offset <= imp->filled);
+ assert(imp->filled <= imp->size);
+ assert(imp->buf != NULL || imp->size == 0);
+ assert(imp->buf == NULL || imp->size > 0);
+ assert(imp->fd >= 0);
+ assert(data);
+
+ while (imp->filled - imp->offset < size) {
+ int n;
+
+ if (imp->passive_fd)
+ /* we have to wait for some data to come to us */
+ return -EAGAIN;
+
+ if (!realloc_buffer(imp, imp->offset + size))
+ return log_oom();
+
+ n = read(imp->fd, imp->buf + imp->filled,
+ imp->size - imp->filled);
+ if (n < 0) {
+ if (errno != EAGAIN)
+ log_error_errno(errno, "read(%d, ..., %zu): %m", imp->fd,
+ imp->size - imp->filled);
+ return -errno;
+ } else if (n == 0)
+ return 0;
+
+ imp->filled += n;
+ }
+
+ *data = imp->buf + imp->offset;
+ imp->offset += size;
+
+ return 1;
+}
+
+static int get_data_size(JournalImporter *imp) {
+ int r;
+ void *data;
+
+ assert(imp);
+ assert(imp->state == IMPORTER_STATE_DATA_START);
+ assert(imp->data_size == 0);
+
+ r = fill_fixed_size(imp, &data, sizeof(uint64_t));
+ if (r <= 0)
+ return r;
+
+ imp->data_size = le64toh( *(uint64_t *) data );
+ if (imp->data_size > DATA_SIZE_MAX) {
+ log_error("Stream declares field with size %zu > DATA_SIZE_MAX = %u",
+ imp->data_size, DATA_SIZE_MAX);
+ return -EINVAL;
+ }
+ if (imp->data_size == 0)
+ log_warning("Binary field with zero length");
+
+ return 1;
+}
+
+static int get_data_data(JournalImporter *imp, void **data) {
+ int r;
+
+ assert(imp);
+ assert(data);
+ assert(imp->state == IMPORTER_STATE_DATA);
+
+ r = fill_fixed_size(imp, data, imp->data_size);
+ if (r <= 0)
+ return r;
+
+ return 1;
+}
+
+static int get_data_newline(JournalImporter *imp) {
+ int r;
+ char *data;
+
+ assert(imp);
+ assert(imp->state == IMPORTER_STATE_DATA_FINISH);
+
+ r = fill_fixed_size(imp, (void**) &data, 1);
+ if (r <= 0)
+ return r;
+
+ assert(data);
+ if (*data != '\n') {
+ log_error("expected newline, got '%c'", *data);
+ return -EINVAL;
+ }
+
+ return 1;
+}
+
+static int process_dunder(JournalImporter *imp, char *line, size_t n) {
+ const char *timestamp;
+ int r;
+
+ assert(line);
+ assert(n > 0);
+ assert(line[n-1] == '\n');
+
+ /* XXX: is it worth to support timestamps in extended format?
+ * We don't produce them, but who knows... */
+
+ timestamp = startswith(line, "__CURSOR=");
+ if (timestamp)
+ /* ignore __CURSOR */
+ return 1;
+
+ timestamp = startswith(line, "__REALTIME_TIMESTAMP=");
+ if (timestamp) {
+ long long unsigned x;
+ line[n-1] = '\0';
+ r = safe_atollu(timestamp, &x);
+ if (r < 0)
+ log_warning("Failed to parse __REALTIME_TIMESTAMP: '%s'", timestamp);
+ else
+ imp->ts.realtime = x;
+ return r < 0 ? r : 1;
+ }
+
+ timestamp = startswith(line, "__MONOTONIC_TIMESTAMP=");
+ if (timestamp) {
+ long long unsigned x;
+ line[n-1] = '\0';
+ r = safe_atollu(timestamp, &x);
+ if (r < 0)
+ log_warning("Failed to parse __MONOTONIC_TIMESTAMP: '%s'", timestamp);
+ else
+ imp->ts.monotonic = x;
+ return r < 0 ? r : 1;
+ }
+
+ timestamp = startswith(line, "__");
+ if (timestamp) {
+ log_notice("Unknown dunder line %s", line);
+ return 1;
+ }
+
+ /* no dunder */
+ return 0;
+}
+
+int journal_importer_process_data(JournalImporter *imp) {
+ int r;
+
+ switch(imp->state) {
+ case IMPORTER_STATE_LINE: {
+ char *line, *sep;
+ size_t n = 0;
+
+ assert(imp->data_size == 0);
+
+ r = get_line(imp, &line, &n);
+ if (r < 0)
+ return r;
+ if (r == 0) {
+ imp->state = IMPORTER_STATE_EOF;
+ return r;
+ }
+ assert(n > 0);
+ assert(line[n-1] == '\n');
+
+ if (n == 1) {
+ log_trace("Received empty line, event is ready");
+ return 1;
+ }
+
+ r = process_dunder(imp, line, n);
+ if (r != 0)
+ return r < 0 ? r : 0;
+
+ /* MESSAGE=xxx\n
+ or
+ COREDUMP\n
+ LLLLLLLL0011223344...\n
+ */
+ sep = memchr(line, '=', n);
+ if (sep) {
+ /* chomp newline */
+ n--;
+
+ r = iovw_put(&imp->iovw, line, n);
+ if (r < 0)
+ return r;
+ } else {
+ /* replace \n with = */
+ line[n-1] = '=';
+
+ imp->field_len = n;
+ imp->state = IMPORTER_STATE_DATA_START;
+
+ /* we cannot put the field in iovec until we have all data */
+ }
+
+ log_trace("Received: %.*s (%s)", (int) n, line, sep ? "text" : "binary");
+
+ return 0; /* continue */
+ }
+
+ case IMPORTER_STATE_DATA_START:
+ assert(imp->data_size == 0);
+
+ r = get_data_size(imp);
+ // log_debug("get_data_size() -> %d", r);
+ if (r < 0)
+ return r;
+ if (r == 0) {
+ imp->state = IMPORTER_STATE_EOF;
+ return 0;
+ }
+
+ imp->state = imp->data_size > 0 ?
+ IMPORTER_STATE_DATA : IMPORTER_STATE_DATA_FINISH;
+
+ return 0; /* continue */
+
+ case IMPORTER_STATE_DATA: {
+ void *data;
+ char *field;
+
+ assert(imp->data_size > 0);
+
+ r = get_data_data(imp, &data);
+ // log_debug("get_data_data() -> %d", r);
+ if (r < 0)
+ return r;
+ if (r == 0) {
+ imp->state = IMPORTER_STATE_EOF;
+ return 0;
+ }
+
+ assert(data);
+
+ field = (char*) data - sizeof(uint64_t) - imp->field_len;
+ memmove(field + sizeof(uint64_t), field, imp->field_len);
+
+ r = iovw_put(&imp->iovw, field + sizeof(uint64_t), imp->field_len + imp->data_size);
+ if (r < 0)
+ return r;
+
+ imp->state = IMPORTER_STATE_DATA_FINISH;
+
+ return 0; /* continue */
+ }
+
+ case IMPORTER_STATE_DATA_FINISH:
+ r = get_data_newline(imp);
+ // log_debug("get_data_newline() -> %d", r);
+ if (r < 0)
+ return r;
+ if (r == 0) {
+ imp->state = IMPORTER_STATE_EOF;
+ return 0;
+ }
+
+ imp->data_size = 0;
+ imp->state = IMPORTER_STATE_LINE;
+
+ return 0; /* continue */
+ default:
+ assert_not_reached("wtf?");
+ }
+}
+
+int journal_importer_push_data(JournalImporter *imp, const char *data, size_t size) {
+ assert(imp);
+ assert(imp->state != IMPORTER_STATE_EOF);
+
+ if (!realloc_buffer(imp, imp->filled + size)) {
+ log_error("Failed to store received data of size %zu "
+ "(in addition to existing %zu bytes with %zu filled): %s",
+ size, imp->size, imp->filled, strerror(ENOMEM));
+ return -ENOMEM;
+ }
+
+ memcpy(imp->buf + imp->filled, data, size);
+ imp->filled += size;
+
+ return 0;
+}
+
+void journal_importer_drop_iovw(JournalImporter *imp) {
+ size_t remain, target;
+
+ /* This function drops processed data that along with the iovw that points at it */
+
+ iovw_free_contents(&imp->iovw);
+
+ /* possibly reset buffer position */
+ remain = imp->filled - imp->offset;
+
+ if (remain == 0) /* no brainer */
+ imp->offset = imp->scanned = imp->filled = 0;
+ else if (imp->offset > imp->size - imp->filled &&
+ imp->offset > remain) {
+ memcpy(imp->buf, imp->buf + imp->offset, remain);
+ imp->offset = imp->scanned = 0;
+ imp->filled = remain;
+ }
+
+ target = imp->size;
+ while (target > 16 * LINE_CHUNK && imp->filled < target / 2)
+ target /= 2;
+ if (target < imp->size) {
+ char *tmp;
+
+ tmp = realloc(imp->buf, target);
+ if (!tmp)
+ log_warning("Failed to reallocate buffer to (smaller) size %zu",
+ target);
+ else {
+ log_debug("Reallocated buffer from %zu to %zu bytes",
+ imp->size, target);
+ imp->buf = tmp;
+ imp->size = target;
+ }
+ }
+}
+
+bool journal_importer_eof(const JournalImporter *imp) {
+ return imp->state == IMPORTER_STATE_EOF;
+}
--- /dev/null
+/***
+ This file is part of systemd.
+
+ Copyright 2016 Zbigniew Jędrzejewski-Szmek
+
+ systemd is free software; you can redistribute it and/or modify it
+ under the terms of the GNU Lesser General Public License as published by
+ the Free Software Foundation; either version 2.1 of the License, or
+ (at your option) any later version.
+
+ systemd is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ Lesser General Public License for more details.
+
+ You should have received a copy of the GNU Lesser General Public License
+ along with systemd; If not, see <http://www.gnu.org/licenses/>.
+***/
+
+#pragma once
+
+#include <stddef.h>
+#include <stdbool.h>
+#include <sys/uio.h>
+
+#include "time-util.h"
+
+/* Make sure not to make this smaller than the maximum coredump size.
+ * See COREDUMP_MAX in coredump.c */
+#define ENTRY_SIZE_MAX (1024*1024*770u)
+#define DATA_SIZE_MAX (1024*1024*768u)
+#define LINE_CHUNK 8*1024u
+
+struct iovec_wrapper {
+ struct iovec *iovec;
+ size_t size_bytes;
+ size_t count;
+};
+
+size_t iovw_size(struct iovec_wrapper *iovw);
+
+typedef struct JournalImporter {
+ int fd;
+ bool passive_fd;
+ char *name;
+
+ char *buf;
+ size_t size; /* total size of the buffer */
+ size_t offset; /* offset to the beginning of live data in the buffer */
+ size_t scanned; /* number of bytes since the beginning of data without a newline */
+ size_t filled; /* total number of bytes in the buffer */
+
+ size_t field_len; /* used for binary fields: the field name length */
+ size_t data_size; /* and the size of the binary data chunk being processed */
+
+ struct iovec_wrapper iovw;
+
+ int state;
+ dual_timestamp ts;
+} JournalImporter;
+
+void journal_importer_cleanup(JournalImporter *);
+int journal_importer_process_data(JournalImporter *);
+int journal_importer_push_data(JournalImporter *, const char *data, size_t size);
+void journal_importer_drop_iovw(JournalImporter *);
+bool journal_importer_eof(const JournalImporter *);
+
+static inline size_t journal_importer_bytes_remaining(const JournalImporter *imp) {
+ return imp->filled;
+}
#include "fileio.h"
#include "fs-util.h"
#include "io-util.h"
-#include "journald-native.h"
+#include "journal-importer.h"
#include "log.h"
#include "macro.h"
#include "missing.h"
#include "parse-util.h"
#include "string-util.h"
-#define LINE_CHUNK 8*1024u
-
void source_free(RemoteSource *source) {
if (!source)
return;
- if (source->fd >= 0 && !source->passive_fd) {
- log_debug("Closing fd:%d (%s)", source->fd, source->name);
- safe_close(source->fd);
- }
-
- free(source->name);
- free(source->buf);
- iovw_free_contents(&source->iovw);
+ journal_importer_cleanup(&source->importer);
log_debug("Writer ref count %i", source->writer->n_ref);
writer_unref(source->writer);
if (!source)
return NULL;
- source->fd = fd;
- source->passive_fd = passive_fd;
- source->name = name;
+ source->importer.fd = fd;
+ source->importer.passive_fd = passive_fd;
+ source->importer.name = name;
+
source->writer = writer;
return source;
}
-static char* realloc_buffer(RemoteSource *source, size_t size) {
- char *b, *old = source->buf;
-
- b = GREEDY_REALLOC(source->buf, source->size, size);
- if (!b)
- return NULL;
-
- iovw_rebase(&source->iovw, old, source->buf);
-
- return b;
-}
-
-static int get_line(RemoteSource *source, char **line, size_t *size) {
- ssize_t n;
- char *c = NULL;
-
- assert(source);
- assert(source->state == STATE_LINE);
- assert(source->offset <= source->filled);
- assert(source->filled <= source->size);
- assert(source->buf == NULL || source->size > 0);
- assert(source->fd >= 0);
-
- for (;;) {
- if (source->buf) {
- size_t start = MAX(source->scanned, source->offset);
-
- c = memchr(source->buf + start, '\n',
- source->filled - start);
- if (c != NULL)
- break;
- }
-
- source->scanned = source->filled;
- if (source->scanned >= DATA_SIZE_MAX) {
- log_error("Entry is bigger than %u bytes.", DATA_SIZE_MAX);
- return -E2BIG;
- }
-
- if (source->passive_fd)
- /* we have to wait for some data to come to us */
- return -EAGAIN;
-
- /* We know that source->filled is at most DATA_SIZE_MAX, so if
- we reallocate it, we'll increase the size at least a bit. */
- assert_cc(DATA_SIZE_MAX < ENTRY_SIZE_MAX);
- if (source->size - source->filled < LINE_CHUNK &&
- !realloc_buffer(source, MIN(source->filled + LINE_CHUNK, ENTRY_SIZE_MAX)))
- return log_oom();
-
- assert(source->buf);
- assert(source->size - source->filled >= LINE_CHUNK ||
- source->size == ENTRY_SIZE_MAX);
-
- n = read(source->fd,
- source->buf + source->filled,
- source->size - source->filled);
- if (n < 0) {
- if (errno != EAGAIN)
- log_error_errno(errno, "read(%d, ..., %zu): %m",
- source->fd,
- source->size - source->filled);
- return -errno;
- } else if (n == 0)
- return 0;
-
- source->filled += n;
- }
-
- *line = source->buf + source->offset;
- *size = c + 1 - source->buf - source->offset;
- source->offset += *size;
-
- return 1;
-}
-
-int push_data(RemoteSource *source, const char *data, size_t size) {
- assert(source);
- assert(source->state != STATE_EOF);
-
- if (!realloc_buffer(source, source->filled + size)) {
- log_error("Failed to store received data of size %zu "
- "(in addition to existing %zu bytes with %zu filled): %s",
- size, source->size, source->filled, strerror(ENOMEM));
- return -ENOMEM;
- }
-
- memcpy(source->buf + source->filled, data, size);
- source->filled += size;
-
- return 0;
-}
-
-static int fill_fixed_size(RemoteSource *source, void **data, size_t size) {
-
- assert(source);
- assert(source->state == STATE_DATA_START ||
- source->state == STATE_DATA ||
- source->state == STATE_DATA_FINISH);
- assert(size <= DATA_SIZE_MAX);
- assert(source->offset <= source->filled);
- assert(source->filled <= source->size);
- assert(source->buf != NULL || source->size == 0);
- assert(source->buf == NULL || source->size > 0);
- assert(source->fd >= 0);
- assert(data);
-
- while (source->filled - source->offset < size) {
- int n;
-
- if (source->passive_fd)
- /* we have to wait for some data to come to us */
- return -EAGAIN;
-
- if (!realloc_buffer(source, source->offset + size))
- return log_oom();
-
- n = read(source->fd, source->buf + source->filled,
- source->size - source->filled);
- if (n < 0) {
- if (errno != EAGAIN)
- log_error_errno(errno, "read(%d, ..., %zu): %m", source->fd,
- source->size - source->filled);
- return -errno;
- } else if (n == 0)
- return 0;
-
- source->filled += n;
- }
-
- *data = source->buf + source->offset;
- source->offset += size;
-
- return 1;
-}
-
-static int get_data_size(RemoteSource *source) {
- int r;
- void *data;
-
- assert(source);
- assert(source->state == STATE_DATA_START);
- assert(source->data_size == 0);
-
- r = fill_fixed_size(source, &data, sizeof(uint64_t));
- if (r <= 0)
- return r;
-
- source->data_size = le64toh( *(uint64_t *) data );
- if (source->data_size > DATA_SIZE_MAX) {
- log_error("Stream declares field with size %zu > DATA_SIZE_MAX = %u",
- source->data_size, DATA_SIZE_MAX);
- return -EINVAL;
- }
- if (source->data_size == 0)
- log_warning("Binary field with zero length");
-
- return 1;
-}
-
-static int get_data_data(RemoteSource *source, void **data) {
- int r;
-
- assert(source);
- assert(data);
- assert(source->state == STATE_DATA);
-
- r = fill_fixed_size(source, data, source->data_size);
- if (r <= 0)
- return r;
-
- return 1;
-}
-
-static int get_data_newline(RemoteSource *source) {
- int r;
- char *data;
-
- assert(source);
- assert(source->state == STATE_DATA_FINISH);
-
- r = fill_fixed_size(source, (void**) &data, 1);
- if (r <= 0)
- return r;
-
- assert(data);
- if (*data != '\n') {
- log_error("expected newline, got '%c'", *data);
- return -EINVAL;
- }
-
- return 1;
-}
-
-static int process_dunder(RemoteSource *source, char *line, size_t n) {
- const char *timestamp;
- int r;
-
- assert(line);
- assert(n > 0);
- assert(line[n-1] == '\n');
-
- /* XXX: is it worth to support timestamps in extended format?
- * We don't produce them, but who knows... */
-
- timestamp = startswith(line, "__CURSOR=");
- if (timestamp)
- /* ignore __CURSOR */
- return 1;
-
- timestamp = startswith(line, "__REALTIME_TIMESTAMP=");
- if (timestamp) {
- long long unsigned x;
- line[n-1] = '\0';
- r = safe_atollu(timestamp, &x);
- if (r < 0)
- log_warning("Failed to parse __REALTIME_TIMESTAMP: '%s'", timestamp);
- else
- source->ts.realtime = x;
- return r < 0 ? r : 1;
- }
-
- timestamp = startswith(line, "__MONOTONIC_TIMESTAMP=");
- if (timestamp) {
- long long unsigned x;
- line[n-1] = '\0';
- r = safe_atollu(timestamp, &x);
- if (r < 0)
- log_warning("Failed to parse __MONOTONIC_TIMESTAMP: '%s'", timestamp);
- else
- source->ts.monotonic = x;
- return r < 0 ? r : 1;
- }
-
- timestamp = startswith(line, "__");
- if (timestamp) {
- log_notice("Unknown dunder line %s", line);
- return 1;
- }
-
- /* no dunder */
- return 0;
-}
-
-static int process_data(RemoteSource *source) {
- int r;
-
- switch(source->state) {
- case STATE_LINE: {
- char *line, *sep;
- size_t n = 0;
-
- assert(source->data_size == 0);
-
- r = get_line(source, &line, &n);
- if (r < 0)
- return r;
- if (r == 0) {
- source->state = STATE_EOF;
- return r;
- }
- assert(n > 0);
- assert(line[n-1] == '\n');
-
- if (n == 1) {
- log_trace("Received empty line, event is ready");
- return 1;
- }
-
- r = process_dunder(source, line, n);
- if (r != 0)
- return r < 0 ? r : 0;
-
- /* MESSAGE=xxx\n
- or
- COREDUMP\n
- LLLLLLLL0011223344...\n
- */
- sep = memchr(line, '=', n);
- if (sep) {
- /* chomp newline */
- n--;
-
- r = iovw_put(&source->iovw, line, n);
- if (r < 0)
- return r;
- } else {
- /* replace \n with = */
- line[n-1] = '=';
-
- source->field_len = n;
- source->state = STATE_DATA_START;
-
- /* we cannot put the field in iovec until we have all data */
- }
-
- log_trace("Received: %.*s (%s)", (int) n, line, sep ? "text" : "binary");
-
- return 0; /* continue */
- }
-
- case STATE_DATA_START:
- assert(source->data_size == 0);
-
- r = get_data_size(source);
- // log_debug("get_data_size() -> %d", r);
- if (r < 0)
- return r;
- if (r == 0) {
- source->state = STATE_EOF;
- return 0;
- }
-
- source->state = source->data_size > 0 ?
- STATE_DATA : STATE_DATA_FINISH;
-
- return 0; /* continue */
-
- case STATE_DATA: {
- void *data;
- char *field;
-
- assert(source->data_size > 0);
-
- r = get_data_data(source, &data);
- // log_debug("get_data_data() -> %d", r);
- if (r < 0)
- return r;
- if (r == 0) {
- source->state = STATE_EOF;
- return 0;
- }
-
- assert(data);
-
- field = (char*) data - sizeof(uint64_t) - source->field_len;
- memmove(field + sizeof(uint64_t), field, source->field_len);
-
- r = iovw_put(&source->iovw, field + sizeof(uint64_t), source->field_len + source->data_size);
- if (r < 0)
- return r;
-
- source->state = STATE_DATA_FINISH;
-
- return 0; /* continue */
- }
-
- case STATE_DATA_FINISH:
- r = get_data_newline(source);
- // log_debug("get_data_newline() -> %d", r);
- if (r < 0)
- return r;
- if (r == 0) {
- source->state = STATE_EOF;
- return 0;
- }
-
- source->data_size = 0;
- source->state = STATE_LINE;
-
- return 0; /* continue */
- default:
- assert_not_reached("wtf?");
- }
-}
-
int process_source(RemoteSource *source, bool compress, bool seal) {
- size_t remain, target;
int r;
assert(source);
assert(source->writer);
- r = process_data(source);
+ r = journal_importer_process_data(&source->importer);
if (r <= 0)
return r;
/* We have a full event */
log_trace("Received full event from source@%p fd:%d (%s)",
- source, source->fd, source->name);
+ source, source->importer.fd, source->importer.name);
- if (!source->iovw.count) {
+ if (source->importer.iovw.count == 0) {
log_warning("Entry with no payload, skipping");
goto freeing;
}
- assert(source->iovw.iovec);
- assert(source->iovw.count);
+ assert(source->importer.iovw.iovec);
- r = writer_write(source->writer, &source->iovw, &source->ts, compress, seal);
+ r = writer_write(source->writer, &source->importer.iovw, &source->importer.ts, compress, seal);
if (r < 0)
log_error_errno(r, "Failed to write entry of %zu bytes: %m",
- iovw_size(&source->iovw));
+ iovw_size(&source->importer.iovw));
else
r = 1;
freeing:
- iovw_free_contents(&source->iovw);
-
- /* possibly reset buffer position */
- remain = source->filled - source->offset;
-
- if (remain == 0) /* no brainer */
- source->offset = source->scanned = source->filled = 0;
- else if (source->offset > source->size - source->filled &&
- source->offset > remain) {
- memcpy(source->buf, source->buf + source->offset, remain);
- source->offset = source->scanned = 0;
- source->filled = remain;
- }
-
- target = source->size;
- while (target > 16 * LINE_CHUNK && source->filled < target / 2)
- target /= 2;
- if (target < source->size) {
- char *tmp;
-
- tmp = realloc(source->buf, target);
- if (!tmp)
- log_warning("Failed to reallocate buffer to (smaller) size %zu",
- target);
- else {
- log_debug("Reallocated buffer from %zu to %zu bytes",
- source->size, target);
- source->buf = tmp;
- source->size = target;
- }
- }
-
+ journal_importer_drop_iovw(&source->importer);
return r;
}
#include "sd-event.h"
+#include "journal-importer.h"
#include "journal-remote-write.h"
-typedef enum {
- STATE_LINE = 0, /* waiting to read, or reading line */
- STATE_DATA_START, /* reading binary data header */
- STATE_DATA, /* reading binary data */
- STATE_DATA_FINISH, /* expecting newline */
- STATE_EOF, /* done */
-} source_state;
-
typedef struct RemoteSource {
- char *name;
- int fd;
- bool passive_fd;
-
- char *buf;
- size_t size; /* total size of the buffer */
- size_t offset; /* offset to the beginning of live data in the buffer */
- size_t scanned; /* number of bytes since the beginning of data without a newline */
- size_t filled; /* total number of bytes in the buffer */
-
- size_t field_len; /* used for binary fields: the field name length */
- size_t data_size; /* and the size of the binary data chunk being processed */
-
- struct iovec_wrapper iovw;
-
- source_state state;
- dual_timestamp ts;
+ JournalImporter importer;
Writer *writer;
} RemoteSource;
RemoteSource* source_new(int fd, bool passive_fd, char *name, Writer *writer);
-
-static inline size_t source_non_empty(RemoteSource *source) {
- assert(source);
-
- return source->filled;
-}
-
void source_free(RemoteSource *source);
-int push_data(RemoteSource *source, const char *data, size_t size);
int process_source(RemoteSource *source, bool compress, bool seal);
#include "alloc-util.h"
#include "journal-remote.h"
-int iovw_put(struct iovec_wrapper *iovw, void* data, size_t len) {
- if (!GREEDY_REALLOC(iovw->iovec, iovw->size_bytes, iovw->count + 1))
- return log_oom();
-
- iovw->iovec[iovw->count++] = (struct iovec) {data, len};
- return 0;
-}
-
-void iovw_free_contents(struct iovec_wrapper *iovw) {
- iovw->iovec = mfree(iovw->iovec);
- iovw->size_bytes = iovw->count = 0;
-}
-
-size_t iovw_size(struct iovec_wrapper *iovw) {
- size_t n = 0, i;
-
- for (i = 0; i < iovw->count; i++)
- n += iovw->iovec[i].iov_len;
-
- return n;
-}
-
-void iovw_rebase(struct iovec_wrapper *iovw, char *old, char *new) {
- size_t i;
-
- for (i = 0; i < iovw->count; i++)
- iovw->iovec[i].iov_base = (char*) iovw->iovec[i].iov_base - old + new;
-}
-
-/**********************************************************************
- **********************************************************************
- **********************************************************************/
-
static int do_rotate(JournalFile **f, bool compress, bool seal) {
int r = journal_file_rotate(f, compress, seal, NULL);
if (r < 0) {
***/
#include "journal-file.h"
+#include "journal-importer.h"
typedef struct RemoteServer RemoteServer;
-struct iovec_wrapper {
- struct iovec *iovec;
- size_t size_bytes;
- size_t count;
-};
-
-int iovw_put(struct iovec_wrapper *iovw, void* data, size_t len);
-void iovw_free_contents(struct iovec_wrapper *iovw);
-size_t iovw_size(struct iovec_wrapper *iovw);
-void iovw_rebase(struct iovec_wrapper *iovw, char *old, char *new);
-
typedef struct Writer {
JournalFile *journal;
JournalMetrics metrics;
if (*upload_data_size) {
log_trace("Received %zu bytes", *upload_data_size);
- r = push_data(source, upload_data, *upload_data_size);
+ r = journal_importer_push_data(&source->importer,
+ upload_data, *upload_data_size);
if (r < 0)
return mhd_respond_oom(connection);
/* The upload is finished */
- remaining = source_non_empty(source);
+ remaining = journal_importer_bytes_remaining(&source->importer);
if (remaining > 0) {
log_warning("Premature EOF byte. %zu bytes lost.", remaining);
return mhd_respondf(connection,
assert(fd >= 0 && fd < (ssize_t) s->sources_size);
source = s->sources[fd];
- assert(source->fd == fd);
+ assert(source->importer.fd == fd);
r = process_source(source, arg_compress, arg_seal);
- if (source->state == STATE_EOF) {
+ if (journal_importer_eof(&source->importer)) {
size_t remaining;
- log_debug("EOF reached with source fd:%d (%s)",
- source->fd, source->name);
+ log_debug("EOF reached with source %s (fd=%d)",
+ source->importer.name, source->importer.fd);
- remaining = source_non_empty(source);
+ remaining = journal_importer_bytes_remaining(&source->importer);
if (remaining > 0)
log_notice("Premature EOF. %zu bytes lost.", remaining);
- remove_source(s, source->fd);
+ remove_source(s, source->importer.fd);
log_debug("%zu active sources remaining", s->active);
return 0;
} else if (r == -E2BIG) {
/* Make sure event stays around even if source is destroyed */
sd_event_source_ref(event);
- r = handle_raw_source(event, source->fd, EPOLLIN, server);
+ r = handle_raw_source(event, source->importer.fd, EPOLLIN, server);
if (r != 1)
/* No more data for now */
sd_event_source_set_enabled(event, SD_EVENT_OFF);
void *userdata) {
RemoteSource *source = userdata;
- return handle_raw_source(event, source->fd, EPOLLIN, server);
+ return handle_raw_source(event, source->importer.fd, EPOLLIN, server);
}
static int accept_connection(const char* type, int fd,
#include "fd-util.h"
#include "fs-util.h"
#include "io-util.h"
+#include "journal-importer.h"
#include "journald-console.h"
#include "journald-kmsg.h"
#include "journald-native.h"
#include "journald-server.h"
-/* Make sure not to make this smaller than the maximum coredump
- * size. See COREDUMP_MAX in coredump.c */
-#define ENTRY_SIZE_MAX (1024*1024*770u)
-#define DATA_SIZE_MAX (1024*1024*768u)
-
bool valid_user_field(const char *p, size_t l, bool allow_protected);
void server_process_native_message(Server *s, const void *buffer, size_t buffer_size, const struct ucred *ucred, const struct timeval *tv, const char *label, size_t label_len);
projects / thirdparty / systemd.git / commitdiff
