diff options
| -rw-r--r-- | Makefile.am | 4 | ||||
| -rw-r--r-- | src/basic/journal-importer.c | 481 | ||||
| -rw-r--r-- | src/basic/journal-importer.h | 70 | ||||
| -rw-r--r-- | src/coredump/coredump.c | 2 | ||||
| -rw-r--r-- | src/journal-remote/journal-remote-parse.c | 431 | ||||
| -rw-r--r-- | src/journal-remote/journal-remote-parse.h | 35 | ||||
| -rw-r--r-- | src/journal-remote/journal-remote-write.c | 33 | ||||
| -rw-r--r-- | src/journal-remote/journal-remote-write.h | 12 | ||||
| -rw-r--r-- | src/journal-remote/journal-remote.c | 21 | ||||
| -rw-r--r-- | src/journal/journald-native.c | 1 | ||||
| -rw-r--r-- | src/journal/journald-native.h | 5 |
11 files changed, 582 insertions, 513 deletions
diff --git a/Makefile.am b/Makefile.am index c45755f36e..64343b6b26 100644 --- a/Makefile.am +++ b/Makefile.am @@ -960,7 +960,9 @@ libbasic_la_SOURCES = \ 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 \ diff --git a/src/basic/journal-importer.c b/src/basic/journal-importer.c new file mode 100644 index 0000000000..4c13e46a49 --- /dev/null +++ b/src/basic/journal-importer.c @@ -0,0 +1,481 @@ +/*** + 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; +} diff --git a/src/basic/journal-importer.h b/src/basic/journal-importer.h new file mode 100644 index 0000000000..b3e308dd6d --- /dev/null +++ b/src/basic/journal-importer.h @@ -0,0 +1,70 @@ +/*** + 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; +} diff --git a/src/coredump/coredump.c b/src/coredump/coredump.c index 4d66b59fd6..2d9819e5b0 100644 --- a/src/coredump/coredump.c +++ b/src/coredump/coredump.c @@ -47,7 +47,7 @@ #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" diff --git a/src/journal-remote/journal-remote-parse.c b/src/journal-remote/journal-remote-parse.c index 9ba9ee3fc0..79afe6604c 100644 --- a/src/journal-remote/journal-remote-parse.c +++ b/src/journal-remote/journal-remote-parse.c @@ -24,20 +24,11 @@ #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); @@ -65,442 +56,44 @@ RemoteSource* source_new(int fd, bool passive_fd, char *name, Writer *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; } diff --git a/src/journal-remote/journal-remote-parse.h b/src/journal-remote/journal-remote-parse.h index 1740a21f92..e3632528cf 100644 --- a/src/journal-remote/journal-remote-parse.h +++ b/src/journal-remote/journal-remote-parse.h @@ -21,34 +21,11 @@ #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; @@ -57,13 +34,5 @@ typedef struct RemoteSource { } 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); diff --git a/src/journal-remote/journal-remote-write.c b/src/journal-remote/journal-remote-write.c index 8729372aa3..734cad333f 100644 --- a/src/journal-remote/journal-remote-write.c +++ b/src/journal-remote/journal-remote-write.c @@ -20,39 +20,6 @@ #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) { diff --git a/src/journal-remote/journal-remote-write.h b/src/journal-remote/journal-remote-write.h index 53ba45fc04..e04af54e55 100644 --- a/src/journal-remote/journal-remote-write.h +++ b/src/journal-remote/journal-remote-write.h @@ -20,20 +20,10 @@ ***/ #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; diff --git a/src/journal-remote/journal-remote.c b/src/journal-remote/journal-remote.c index d0d8d936e3..202a5a3f97 100644 --- a/src/journal-remote/journal-remote.c +++ b/src/journal-remote/journal-remote.c @@ -512,7 +512,8 @@ static int process_http_upload( 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); @@ -542,7 +543,7 @@ static int process_http_upload( /* 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, @@ -1036,19 +1037,19 @@ static int handle_raw_source(sd_event_source *event, 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) { @@ -1072,7 +1073,7 @@ static int dispatch_raw_source_until_block(sd_event_source *event, /* 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); @@ -1105,7 +1106,7 @@ static int dispatch_blocking_source_event(sd_event_source *event, 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, diff --git a/src/journal/journald-native.c b/src/journal/journald-native.c index 0a1ce205c2..3c03b83754 100644 --- a/src/journal/journald-native.c +++ b/src/journal/journald-native.c @@ -27,6 +27,7 @@ #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" diff --git a/src/journal/journald-native.h b/src/journal/journald-native.h index c13b80aa4f..1ab415ac85 100644 --- a/src/journal/journald-native.h +++ b/src/journal/journald-native.h @@ -21,11 +21,6 @@ #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); |