/*-*- Mode: C; c-basic-offset: 8; indent-tabs-mode: nil -*-*/ /*** 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 "journal-remote-parse.h" #include "journald-native.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); log_debug("Writer ref count %i", source->writer->n_ref); writer_unref(source->writer); sd_event_source_unref(source->event); sd_event_source_unref(source->buffer_event); free(source); } /** * Initialize zero-filled source with given values. On success, takes * ownerhship of fd and writer, otherwise does not touch them. */ RemoteSource* source_new(int fd, bool passive_fd, char *name, Writer *writer) { RemoteSource *source; log_debug("Creating source for %sfd:%d (%s)", passive_fd ? "passive " : "", fd, name); assert(fd >= 0); source = new0(RemoteSource, 1); if (!source) return NULL; source->fd = fd; source->passive_fd = passive_fd; source->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); while (true) { 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); 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); if (!source->iovw.count) { log_warning("Entry with no payload, skipping"); goto freeing; } assert(source->iovw.iovec); assert(source->iovw.count); r = writer_write(source->writer, &source->iovw, &source->ts, compress, seal); if (r < 0) log_error_errno(r, "Failed to write entry of %zu bytes: %m", iovw_size(&source->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 && remain < 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; } } return r; }