[thirdparty/systemd.git] / src / shared / journal-importer.c

/* SPDX-License-Identifier: LGPL-2.1+ */

#include <errno.h>
#include <unistd.h>

#include "alloc-util.h"
#include "escape.h"
#include "fd-util.h"
#include "io-util.h"
#include "journal-file.h"
#include "journal-importer.h"
#include "journal-util.h"
#include "parse-util.h"
#include "string-util.h"
#include "unaligned.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 (iovw->count >= ENTRY_FIELD_COUNT_MAX)
                return -E2BIG;

        if (!GREEDY_REALLOC(iovw->iovec, iovw->size_bytes, iovw->count + 1))
                return log_oom();

        iovw->iovec[iovw->count++] = IOVEC_MAKE(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->name);
        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 || 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)
                        return log_error_errno(SYNTHETIC_ERRNO(ENOBUFS),
                                               "Entry is bigger than %u bytes.",
                                               DATA_SIZE_MAX);

                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(IN_SET(imp->state, IMPORTER_STATE_DATA_START, IMPORTER_STATE_DATA, IMPORTER_STATE_DATA_FINISH));
        assert(size <= DATA_SIZE_MAX);
        assert(imp->offset <= imp->filled);
        assert(imp->filled <= imp->size);
        assert(imp->buf || imp->size == 0);
        assert(!imp->buf || 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 = unaligned_read_le64(data);
        if (imp->data_size > DATA_SIZE_MAX)
                return log_error_errno(SYNTHETIC_ERRNO(EINVAL),
                                       "Stream declares field with size %zu > DATA_SIZE_MAX = %u",
                                       imp->data_size, DATA_SIZE_MAX);
        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') {
                char buf[4];
                int l;

                l = cescape_char(*data, buf);
                return log_error_errno(SYNTHETIC_ERRNO(EINVAL),
                                       "Expected newline, got '%.*s'", l, buf);
        }

        return 1;
}

static int process_special_field(JournalImporter *imp, char *line) {
        const char *value;
        char buf[CELLESCAPE_DEFAULT_LENGTH];
        int r;

        assert(line);

        value = startswith(line, "__CURSOR=");
        if (value)
                /* ignore __CURSOR */
                return 1;

        value = startswith(line, "__REALTIME_TIMESTAMP=");
        if (value) {
                uint64_t x;

                r = safe_atou64(value, &x);
                if (r < 0)
                        return log_warning_errno(r, "Failed to parse __REALTIME_TIMESTAMP '%s': %m",
                                                 cellescape(buf, sizeof buf, value));
                else if (!VALID_REALTIME(x)) {
                        log_warning("__REALTIME_TIMESTAMP out of range, ignoring: %"PRIu64, x);
                        return -ERANGE;
                }

                imp->ts.realtime = x;
                return 1;
        }

        value = startswith(line, "__MONOTONIC_TIMESTAMP=");
        if (value) {
                uint64_t x;

                r = safe_atou64(value, &x);
                if (r < 0)
                        return log_warning_errno(r, "Failed to parse __MONOTONIC_TIMESTAMP '%s': %m",
                                                 cellescape(buf, sizeof buf, value));
                else if (!VALID_MONOTONIC(x)) {
                        log_warning("__MONOTONIC_TIMESTAMP out of range, ignoring: %"PRIu64, x);
                        return -ERANGE;
                }

                imp->ts.monotonic = x;
                return 1;
        }

        /* Just a single underline, but it needs special treatment too. */
        value = startswith(line, "_BOOT_ID=");
        if (value) {
                r = sd_id128_from_string(value, &imp->boot_id);
                if (r < 0)
                        return log_warning_errno(r, "Failed to parse _BOOT_ID '%s': %m",
                                                 cellescape(buf, sizeof buf, value));

                /* store the field in the usual fashion too */
                return 0;
        }

        value = startswith(line, "__");
        if (value) {
                log_notice("Unknown dunder line __%s, ignoring.", cellescape(buf, sizeof buf, value));
                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 0;
                }
                assert(n > 0);
                assert(line[n-1] == '\n');

                if (n == 1) {
                        log_trace("Received empty line, event is ready");
                        return 1;
                }

                /* MESSAGE=xxx\n
                   or
                   COREDUMP\n
                   LLLLLLLL0011223344...\n
                */
                sep = memchr(line, '=', n);
                if (sep) {
                        /* chomp newline */
                        n--;

                        if (!journal_field_valid(line, sep - line, true)) {
                                char buf[64], *t;

                                t = strndupa(line, sep - line);
                                log_debug("Ignoring invalid field: \"%s\"",
                                          cellescape(buf, sizeof buf, t));

                                return 0;
                        }

                        line[n] = '\0';
                        r = process_special_field(imp, line);
                        if (r != 0)
                                return r < 0 ? r : 0;

                        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))
                return log_error_errno(SYNTHETIC_ERRNO(ENOMEM),
                                       "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));

        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;
}
Commit	Line	Data
53e1b683	1	/* SPDX-License-Identifier: LGPL-2.1+ */
b18453ed	2
dccca82b	3	#include <errno.h>
b18453ed ZJS	4	#include <unistd.h>
	5
	6	#include "alloc-util.h"
b778252b	7	#include "escape.h"
b18453ed	8	#include "fd-util.h"
e6a7ec4b	9	#include "io-util.h"
41b0b127	10	#include "journal-file.h"
e6a7ec4b	11	#include "journal-importer.h"
1e448731	12	#include "journal-util.h"
b18453ed ZJS	13	#include "parse-util.h"
b18453ed ZJS	14	#include "string-util.h"
f652c62d	15	#include "unaligned.h"
b18453ed ZJS	16
	17	enum {
	18	IMPORTER_STATE_LINE = 0, /* waiting to read, or reading line */
	19	IMPORTER_STATE_DATA_START, /* reading binary data header */
	20	IMPORTER_STATE_DATA, /* reading binary data */
	21	IMPORTER_STATE_DATA_FINISH, /* expecting newline */
	22	IMPORTER_STATE_EOF, /* done */
	23	};
	24
	25	static int iovw_put(struct iovec_wrapper iovw, void data, size_t len) {
ef4d6abe ZJS	26	if (iovw->count >= ENTRY_FIELD_COUNT_MAX)
	27	return -E2BIG;
	28
b18453ed ZJS	29	if (!GREEDY_REALLOC(iovw->iovec, iovw->size_bytes, iovw->count + 1))
	30	return log_oom();
	31
e6a7ec4b	32	iovw->iovec[iovw->count++] = IOVEC_MAKE(data, len);
b18453ed ZJS	33	return 0;
	34	}
	35
	36	static void iovw_free_contents(struct iovec_wrapper *iovw) {
	37	iovw->iovec = mfree(iovw->iovec);
	38	iovw->size_bytes = iovw->count = 0;
	39	}
	40
	41	static void iovw_rebase(struct iovec_wrapper iovw, char old, char *new) {
	42	size_t i;
	43
	44	for (i = 0; i < iovw->count; i++)
	45	iovw->iovec[i].iov_base = (char*) iovw->iovec[i].iov_base - old + new;
	46	}
	47
	48	size_t iovw_size(struct iovec_wrapper *iovw) {
	49	size_t n = 0, i;
	50
	51	for (i = 0; i < iovw->count; i++)
	52	n += iovw->iovec[i].iov_len;
	53
	54	return n;
	55	}
	56
	57	void journal_importer_cleanup(JournalImporter *imp) {
	58	if (imp->fd >= 0 && !imp->passive_fd) {
	59	log_debug("Closing %s (fd=%d)", imp->name ?: "importer", imp->fd);
	60	safe_close(imp->fd);
	61	}
	62
2ddb70d2	63	free(imp->name);
b18453ed ZJS	64	free(imp->buf);
	65	iovw_free_contents(&imp->iovw);
	66	}
	67
	68	static char* realloc_buffer(JournalImporter *imp, size_t size) {
	69	char b, old = imp->buf;
	70
	71	b = GREEDY_REALLOC(imp->buf, imp->size, size);
	72	if (!b)
	73	return NULL;
	74
	75	iovw_rebase(&imp->iovw, old, imp->buf);
	76
	77	return b;
	78	}
	79
	80	static int get_line(JournalImporter imp, char line, size_t size) {
	81	ssize_t n;
	82	char *c = NULL;
	83
	84	assert(imp);
	85	assert(imp->state == IMPORTER_STATE_LINE);
	86	assert(imp->offset <= imp->filled);
	87	assert(imp->filled <= imp->size);
234519ae	88	assert(!imp->buf \|\| imp->size > 0);
b18453ed ZJS	89	assert(imp->fd >= 0);
	90
	91	for (;;) {
	92	if (imp->buf) {
	93	size_t start = MAX(imp->scanned, imp->offset);
	94
	95	c = memchr(imp->buf + start, '\n',
	96	imp->filled - start);
	97	if (c != NULL)
	98	break;
	99	}
	100
	101	imp->scanned = imp->filled;
baaa35ad	102	if (imp->scanned >= DATA_SIZE_MAX)
ef4d6abe	103	return log_error_errno(SYNTHETIC_ERRNO(ENOBUFS),
baaa35ad ZJS	104	"Entry is bigger than %u bytes.",
baaa35ad ZJS	105	DATA_SIZE_MAX);
b18453ed ZJS	106
	107	if (imp->passive_fd)
	108	/* we have to wait for some data to come to us */
	109	return -EAGAIN;
	110
	111	/* We know that imp->filled is at most DATA_SIZE_MAX, so if
	112	we reallocate it, we'll increase the size at least a bit. */
	113	assert_cc(DATA_SIZE_MAX < ENTRY_SIZE_MAX);
	114	if (imp->size - imp->filled < LINE_CHUNK &&
	115	!realloc_buffer(imp, MIN(imp->filled + LINE_CHUNK, ENTRY_SIZE_MAX)))
	116	return log_oom();
	117
	118	assert(imp->buf);
	119	assert(imp->size - imp->filled >= LINE_CHUNK \|\|
	120	imp->size == ENTRY_SIZE_MAX);
	121
	122	n = read(imp->fd,
	123	imp->buf + imp->filled,
	124	imp->size - imp->filled);
	125	if (n < 0) {
	126	if (errno != EAGAIN)
	127	log_error_errno(errno, "read(%d, ..., %zu): %m",
	128	imp->fd,
	129	imp->size - imp->filled);
	130	return -errno;
	131	} else if (n == 0)
	132	return 0;
	133
	134	imp->filled += n;
	135	}
	136
	137	*line = imp->buf + imp->offset;
	138	*size = c + 1 - imp->buf - imp->offset;
	139	imp->offset += *size;
	140
	141	return 1;
	142	}
	143
	144	static int fill_fixed_size(JournalImporter imp, void *data, size_t size) {
	145
	146	assert(imp);
3742095b	147	assert(IN_SET(imp->state, IMPORTER_STATE_DATA_START, IMPORTER_STATE_DATA, IMPORTER_STATE_DATA_FINISH));
b18453ed ZJS	148	assert(size <= DATA_SIZE_MAX);
	149	assert(imp->offset <= imp->filled);
	150	assert(imp->filled <= imp->size);
234519ae LP	151	assert(imp->buf \|\| imp->size == 0);
234519ae LP	152	assert(!imp->buf \|\| imp->size > 0);
b18453ed ZJS	153	assert(imp->fd >= 0);
	154	assert(data);
	155
	156	while (imp->filled - imp->offset < size) {
	157	int n;
	158
	159	if (imp->passive_fd)
	160	/* we have to wait for some data to come to us */
	161	return -EAGAIN;
	162
	163	if (!realloc_buffer(imp, imp->offset + size))
	164	return log_oom();
	165
	166	n = read(imp->fd, imp->buf + imp->filled,
	167	imp->size - imp->filled);
	168	if (n < 0) {
	169	if (errno != EAGAIN)
	170	log_error_errno(errno, "read(%d, ..., %zu): %m", imp->fd,
	171	imp->size - imp->filled);
	172	return -errno;
	173	} else if (n == 0)
	174	return 0;
	175
	176	imp->filled += n;
	177	}
	178
	179	*data = imp->buf + imp->offset;
	180	imp->offset += size;
	181
	182	return 1;
	183	}
	184
	185	static int get_data_size(JournalImporter *imp) {
	186	int r;
	187	void *data;
	188
	189	assert(imp);
	190	assert(imp->state == IMPORTER_STATE_DATA_START);
	191	assert(imp->data_size == 0);
	192
	193	r = fill_fixed_size(imp, &data, sizeof(uint64_t));
	194	if (r <= 0)
	195	return r;
	196
f652c62d	197	imp->data_size = unaligned_read_le64(data);
baaa35ad ZJS	198	if (imp->data_size > DATA_SIZE_MAX)
	199	return log_error_errno(SYNTHETIC_ERRNO(EINVAL),
	200	"Stream declares field with size %zu > DATA_SIZE_MAX = %u",
	201	imp->data_size, DATA_SIZE_MAX);
b18453ed ZJS	202	if (imp->data_size == 0)
	203	log_warning("Binary field with zero length");
	204
	205	return 1;
	206	}
	207
	208	static int get_data_data(JournalImporter imp, void *data) {
	209	int r;
	210
	211	assert(imp);
	212	assert(data);
	213	assert(imp->state == IMPORTER_STATE_DATA);
	214
	215	r = fill_fixed_size(imp, data, imp->data_size);
	216	if (r <= 0)
	217	return r;
	218
	219	return 1;
	220	}
	221
	222	static int get_data_newline(JournalImporter *imp) {
	223	int r;
	224	char *data;
	225
	226	assert(imp);
	227	assert(imp->state == IMPORTER_STATE_DATA_FINISH);
	228
	229	r = fill_fixed_size(imp, (void**) &data, 1);
	230	if (r <= 0)
	231	return r;
	232
	233	assert(data);
	234	if (*data != '\n') {
b778252b ZJS	235	char buf[4];
	236	int l;
	237
	238	l = cescape_char(*data, buf);
baaa35ad ZJS	239	return log_error_errno(SYNTHETIC_ERRNO(EINVAL),
baaa35ad ZJS	240	"Expected newline, got '%.*s'", l, buf);
b18453ed ZJS	241	}
	242
	243	return 1;
	244	}
	245
c0b6ada7 ZJS	246	static int process_special_field(JournalImporter imp, char line) {
c0b6ada7 ZJS	247	const char *value;
cca24fc3	248	char buf[CELLESCAPE_DEFAULT_LENGTH];
b18453ed ZJS	249	int r;
	250
	251	assert(line);
b18453ed	252
c0b6ada7 ZJS	253	value = startswith(line, "__CURSOR=");
c0b6ada7 ZJS	254	if (value)
b18453ed ZJS	255	/* ignore __CURSOR */
	256	return 1;
	257
c0b6ada7 ZJS	258	value = startswith(line, "__REALTIME_TIMESTAMP=");
c0b6ada7 ZJS	259	if (value) {
41b0b127	260	uint64_t x;
cca24fc3	261
c0b6ada7	262	r = safe_atou64(value, &x);
b18453ed	263	if (r < 0)
cca24fc3	264	return log_warning_errno(r, "Failed to parse __REALTIME_TIMESTAMP '%s': %m",
c0b6ada7	265	cellescape(buf, sizeof buf, value));
41b0b127 ZJS	266	else if (!VALID_REALTIME(x)) {
	267	log_warning("__REALTIME_TIMESTAMP out of range, ignoring: %"PRIu64, x);
	268	return -ERANGE;
	269	}
	270
	271	imp->ts.realtime = x;
	272	return 1;
b18453ed ZJS	273	}
b18453ed ZJS	274
c0b6ada7 ZJS	275	value = startswith(line, "__MONOTONIC_TIMESTAMP=");
c0b6ada7 ZJS	276	if (value) {
41b0b127	277	uint64_t x;
cca24fc3	278
c0b6ada7	279	r = safe_atou64(value, &x);
b18453ed	280	if (r < 0)
cca24fc3	281	return log_warning_errno(r, "Failed to parse __MONOTONIC_TIMESTAMP '%s': %m",
c0b6ada7	282	cellescape(buf, sizeof buf, value));
41b0b127 ZJS	283	else if (!VALID_MONOTONIC(x)) {
	284	log_warning("__MONOTONIC_TIMESTAMP out of range, ignoring: %"PRIu64, x);
	285	return -ERANGE;
	286	}
	287
	288	imp->ts.monotonic = x;
	289	return 1;
b18453ed ZJS	290	}
b18453ed ZJS	291
c0b6ada7 ZJS	292	/* Just a single underline, but it needs special treatment too. */
	293	value = startswith(line, "_BOOT_ID=");
	294	if (value) {
	295	r = sd_id128_from_string(value, &imp->boot_id);
	296	if (r < 0)
	297	return log_warning_errno(r, "Failed to parse _BOOT_ID '%s': %m",
	298	cellescape(buf, sizeof buf, value));
	299
	300	/* store the field in the usual fashion too */
	301	return 0;
	302	}
	303
	304	value = startswith(line, "__");
	305	if (value) {
	306	log_notice("Unknown dunder line __%s, ignoring.", cellescape(buf, sizeof buf, value));
b18453ed ZJS	307	return 1;
	308	}
	309
	310	/* no dunder */
	311	return 0;
	312	}
	313
	314	int journal_importer_process_data(JournalImporter *imp) {
	315	int r;
	316
	317	switch(imp->state) {
	318	case IMPORTER_STATE_LINE: {
	319	char line, sep;
	320	size_t n = 0;
	321
	322	assert(imp->data_size == 0);
	323
	324	r = get_line(imp, &line, &n);
	325	if (r < 0)
	326	return r;
	327	if (r == 0) {
	328	imp->state = IMPORTER_STATE_EOF;
d74dc4f2	329	return 0;
b18453ed ZJS	330	}
	331	assert(n > 0);
	332	assert(line[n-1] == '\n');
	333
	334	if (n == 1) {
	335	log_trace("Received empty line, event is ready");
	336	return 1;
	337	}
	338
b18453ed ZJS	339	/* MESSAGE=xxx\n
	340	or
	341	COREDUMP\n
	342	LLLLLLLL0011223344...\n
	343	*/
	344	sep = memchr(line, '=', n);
	345	if (sep) {
	346	/* chomp newline */
	347	n--;
	348
1e448731 ZJS	349	if (!journal_field_valid(line, sep - line, true)) {
	350	char buf[64], *t;
	351
	352	t = strndupa(line, sep - line);
	353	log_debug("Ignoring invalid field: \"%s\"",
c0b6ada7	354	cellescape(buf, sizeof buf, t));
1e448731 ZJS	355
	356	return 0;
	357	}
	358
bcac9822	359	line[n] = '\0';
c0b6ada7	360	r = process_special_field(imp, line);
bcac9822 ZJS	361	if (r != 0)
	362	return r < 0 ? r : 0;
	363
b18453ed ZJS	364	r = iovw_put(&imp->iovw, line, n);
	365	if (r < 0)
	366	return r;
	367	} else {
	368	/* replace \n with = */
	369	line[n-1] = '=';
	370
	371	imp->field_len = n;
	372	imp->state = IMPORTER_STATE_DATA_START;
	373
	374	/* we cannot put the field in iovec until we have all data */
	375	}
	376
	377	log_trace("Received: %.*s (%s)", (int) n, line, sep ? "text" : "binary");
	378
	379	return 0; /* continue */
	380	}
	381
	382	case IMPORTER_STATE_DATA_START:
	383	assert(imp->data_size == 0);
	384
	385	r = get_data_size(imp);
	386	// log_debug("get_data_size() -> %d", r);
	387	if (r < 0)
	388	return r;
	389	if (r == 0) {
	390	imp->state = IMPORTER_STATE_EOF;
	391	return 0;
	392	}
	393
	394	imp->state = imp->data_size > 0 ?
	395	IMPORTER_STATE_DATA : IMPORTER_STATE_DATA_FINISH;
	396
	397	return 0; /* continue */
	398
	399	case IMPORTER_STATE_DATA: {
	400	void *data;
	401	char *field;
	402
	403	assert(imp->data_size > 0);
	404
	405	r = get_data_data(imp, &data);
	406	// log_debug("get_data_data() -> %d", r);
	407	if (r < 0)
	408	return r;
	409	if (r == 0) {
	410	imp->state = IMPORTER_STATE_EOF;
	411	return 0;
	412	}
	413
	414	assert(data);
	415
	416	field = (char*) data - sizeof(uint64_t) - imp->field_len;
	417	memmove(field + sizeof(uint64_t), field, imp->field_len);
	418
	419	r = iovw_put(&imp->iovw, field + sizeof(uint64_t), imp->field_len + imp->data_size);
	420	if (r < 0)
	421	return r;
	422
	423	imp->state = IMPORTER_STATE_DATA_FINISH;
	424
	425	return 0; /* continue */
	426	}
	427
428	case IMPORTER_STATE_DATA_FINISH:
429	r = get_data_newline(imp);
430	// log_debug("get_data_newline() -> %d", r);
431	if (r < 0)
432	return r;
433	if (r == 0) {
434	imp->state = IMPORTER_STATE_EOF;
435	return 0;
436	}
437
438	imp->data_size = 0;
439	imp->state = IMPORTER_STATE_LINE;
440
441	return 0; /* continue */
442	default:
443	assert_not_reached("wtf?");
444	}
445	}
446
447	int journal_importer_push_data(JournalImporter imp, const char data, size_t size) {
448	assert(imp);
449	assert(imp->state != IMPORTER_STATE_EOF);
450
baaa35ad ZJS	451	if (!realloc_buffer(imp, imp->filled + size))
	452	return log_error_errno(SYNTHETIC_ERRNO(ENOMEM),
	453	"Failed to store received data of size %zu "
	454	"(in addition to existing %zu bytes with %zu filled): %s",
	455	size, imp->size, imp->filled,
	456	strerror(ENOMEM));
b18453ed ZJS	457
	458	memcpy(imp->buf + imp->filled, data, size);
	459	imp->filled += size;
	460
	461	return 0;
	462	}
	463
	464	void journal_importer_drop_iovw(JournalImporter *imp) {
	465	size_t remain, target;
	466
	467	/* This function drops processed data that along with the iovw that points at it */
	468
	469	iovw_free_contents(&imp->iovw);
	470
	471	/* possibly reset buffer position */
	472	remain = imp->filled - imp->offset;
	473
	474	if (remain == 0) /* no brainer */
	475	imp->offset = imp->scanned = imp->filled = 0;
	476	else if (imp->offset > imp->size - imp->filled &&
	477	imp->offset > remain) {
	478	memcpy(imp->buf, imp->buf + imp->offset, remain);
	479	imp->offset = imp->scanned = 0;
	480	imp->filled = remain;
	481	}
	482
	483	target = imp->size;
	484	while (target > 16 * LINE_CHUNK && imp->filled < target / 2)
	485	target /= 2;
	486	if (target < imp->size) {
	487	char *tmp;
	488
	489	tmp = realloc(imp->buf, target);
	490	if (!tmp)
	491	log_warning("Failed to reallocate buffer to (smaller) size %zu",
	492	target);
	493	else {
	494	log_debug("Reallocated buffer from %zu to %zu bytes",
	495	imp->size, target);
	496	imp->buf = tmp;
	497	imp->size = target;
	498	}
	499	}
	500	}
	501
	502	bool journal_importer_eof(const JournalImporter *imp) {
	503	return imp->state == IMPORTER_STATE_EOF;
	504	}