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