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

/* SPDX-License-Identifier: LGPL-2.1+ */
/***
  Copyright © 2014 Zbigniew Jędrzejewski-Szmek
***/

#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) {
                        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(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) {
                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') {
                char buf[4];
                int l;

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

        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)) {
                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;
}
Commit	Line	Data
53e1b683	1	/* SPDX-License-Identifier: LGPL-2.1+ */
b18453ed	2	/***
96b2fb93	3	Copyright © 2014 Zbigniew Jędrzejewski-Szmek
b18453ed ZJS	4	***/
b18453ed ZJS	5
dccca82b	6	#include <errno.h>
b18453ed ZJS	7	#include <unistd.h>
	8
	9	#include "alloc-util.h"
b778252b	10	#include "escape.h"
b18453ed	11	#include "fd-util.h"
e6a7ec4b	12	#include "io-util.h"
41b0b127	13	#include "journal-file.h"
e6a7ec4b	14	#include "journal-importer.h"
1e448731	15	#include "journal-util.h"
b18453ed ZJS	16	#include "parse-util.h"
b18453ed ZJS	17	#include "string-util.h"
f652c62d	18	#include "unaligned.h"
b18453ed ZJS	19
	20	enum {
	21	IMPORTER_STATE_LINE = 0, /* waiting to read, or reading line */
	22	IMPORTER_STATE_DATA_START, /* reading binary data header */
	23	IMPORTER_STATE_DATA, /* reading binary data */
	24	IMPORTER_STATE_DATA_FINISH, /* expecting newline */
	25	IMPORTER_STATE_EOF, /* done */
	26	};
	27
	28	static int iovw_put(struct iovec_wrapper iovw, void data, size_t len) {
	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;
	102	if (imp->scanned >= DATA_SIZE_MAX) {
	103	log_error("Entry is bigger than %u bytes.", DATA_SIZE_MAX);
	104	return -E2BIG;
	105	}
	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);
b18453ed ZJS	198	if (imp->data_size > DATA_SIZE_MAX) {
	199	log_error("Stream declares field with size %zu > DATA_SIZE_MAX = %u",
	200	imp->data_size, DATA_SIZE_MAX);
	201	return -EINVAL;
	202	}
	203	if (imp->data_size == 0)
	204	log_warning("Binary field with zero length");
	205
	206	return 1;
	207	}
	208
	209	static int get_data_data(JournalImporter imp, void *data) {
	210	int r;
	211
	212	assert(imp);
	213	assert(data);
	214	assert(imp->state == IMPORTER_STATE_DATA);
	215
	216	r = fill_fixed_size(imp, data, imp->data_size);
	217	if (r <= 0)
	218	return r;
	219
	220	return 1;
	221	}
	222
	223	static int get_data_newline(JournalImporter *imp) {
	224	int r;
	225	char *data;
	226
	227	assert(imp);
	228	assert(imp->state == IMPORTER_STATE_DATA_FINISH);
	229
	230	r = fill_fixed_size(imp, (void**) &data, 1);
	231	if (r <= 0)
	232	return r;
	233
	234	assert(data);
	235	if (*data != '\n') {
b778252b ZJS	236	char buf[4];
	237	int l;
	238
	239	l = cescape_char(*data, buf);
	240	log_error("Expected newline, got '%.*s'", l, buf);
b18453ed ZJS	241	return -EINVAL;
	242	}
	243
	244	return 1;
	245	}
	246
c0b6ada7 ZJS	247	static int process_special_field(JournalImporter imp, char line) {
c0b6ada7 ZJS	248	const char *value;
cca24fc3	249	char buf[CELLESCAPE_DEFAULT_LENGTH];
b18453ed ZJS	250	int r;
	251
	252	assert(line);
b18453ed	253
c0b6ada7 ZJS	254	value = startswith(line, "__CURSOR=");
c0b6ada7 ZJS	255	if (value)
b18453ed ZJS	256	/* ignore __CURSOR */
	257	return 1;
	258
c0b6ada7 ZJS	259	value = startswith(line, "__REALTIME_TIMESTAMP=");
c0b6ada7 ZJS	260	if (value) {
41b0b127	261	uint64_t x;
cca24fc3	262
c0b6ada7	263	r = safe_atou64(value, &x);
b18453ed	264	if (r < 0)
cca24fc3	265	return log_warning_errno(r, "Failed to parse __REALTIME_TIMESTAMP '%s': %m",
c0b6ada7	266	cellescape(buf, sizeof buf, value));
41b0b127 ZJS	267	else if (!VALID_REALTIME(x)) {
	268	log_warning("__REALTIME_TIMESTAMP out of range, ignoring: %"PRIu64, x);
	269	return -ERANGE;
	270	}
	271
	272	imp->ts.realtime = x;
	273	return 1;
b18453ed ZJS	274	}
b18453ed ZJS	275
c0b6ada7 ZJS	276	value = startswith(line, "__MONOTONIC_TIMESTAMP=");
c0b6ada7 ZJS	277	if (value) {
41b0b127	278	uint64_t x;
cca24fc3	279
c0b6ada7	280	r = safe_atou64(value, &x);
b18453ed	281	if (r < 0)
cca24fc3	282	return log_warning_errno(r, "Failed to parse __MONOTONIC_TIMESTAMP '%s': %m",
c0b6ada7	283	cellescape(buf, sizeof buf, value));
41b0b127 ZJS	284	else if (!VALID_MONOTONIC(x)) {
	285	log_warning("__MONOTONIC_TIMESTAMP out of range, ignoring: %"PRIu64, x);
	286	return -ERANGE;
	287	}
	288
	289	imp->ts.monotonic = x;
	290	return 1;
b18453ed ZJS	291	}
b18453ed ZJS	292
c0b6ada7 ZJS	293	/* Just a single underline, but it needs special treatment too. */
	294	value = startswith(line, "_BOOT_ID=");
	295	if (value) {
	296	r = sd_id128_from_string(value, &imp->boot_id);
	297	if (r < 0)
	298	return log_warning_errno(r, "Failed to parse _BOOT_ID '%s': %m",
	299	cellescape(buf, sizeof buf, value));
	300
	301	/* store the field in the usual fashion too */
	302	return 0;
	303	}
	304
	305	value = startswith(line, "__");
	306	if (value) {
	307	log_notice("Unknown dunder line __%s, ignoring.", cellescape(buf, sizeof buf, value));
b18453ed ZJS	308	return 1;
	309	}
	310
	311	/* no dunder */
	312	return 0;
	313	}
	314
	315	int journal_importer_process_data(JournalImporter *imp) {
	316	int r;
	317
	318	switch(imp->state) {
	319	case IMPORTER_STATE_LINE: {
	320	char line, sep;
	321	size_t n = 0;
	322
	323	assert(imp->data_size == 0);
	324
	325	r = get_line(imp, &line, &n);
	326	if (r < 0)
	327	return r;
	328	if (r == 0) {
	329	imp->state = IMPORTER_STATE_EOF;
d74dc4f2	330	return 0;
b18453ed ZJS	331	}
	332	assert(n > 0);
	333	assert(line[n-1] == '\n');
	334
	335	if (n == 1) {
	336	log_trace("Received empty line, event is ready");
	337	return 1;
	338	}
	339
b18453ed ZJS	340	/* MESSAGE=xxx\n
	341	or
	342	COREDUMP\n
	343	LLLLLLLL0011223344...\n
	344	*/
	345	sep = memchr(line, '=', n);
	346	if (sep) {
	347	/* chomp newline */
	348	n--;
	349
1e448731 ZJS	350	if (!journal_field_valid(line, sep - line, true)) {
	351	char buf[64], *t;
	352
	353	t = strndupa(line, sep - line);
	354	log_debug("Ignoring invalid field: \"%s\"",
c0b6ada7	355	cellescape(buf, sizeof buf, t));
1e448731 ZJS	356
	357	return 0;
	358	}
	359
bcac9822	360	line[n] = '\0';
c0b6ada7	361	r = process_special_field(imp, line);
bcac9822 ZJS	362	if (r != 0)
	363	return r < 0 ? r : 0;
	364
b18453ed ZJS	365	r = iovw_put(&imp->iovw, line, n);
	366	if (r < 0)
	367	return r;
	368	} else {
	369	/* replace \n with = */
	370	line[n-1] = '=';
	371
	372	imp->field_len = n;
	373	imp->state = IMPORTER_STATE_DATA_START;
	374
	375	/* we cannot put the field in iovec until we have all data */
	376	}
	377
	378	log_trace("Received: %.*s (%s)", (int) n, line, sep ? "text" : "binary");
	379
	380	return 0; /* continue */
	381	}
	382
	383	case IMPORTER_STATE_DATA_START:
	384	assert(imp->data_size == 0);
	385
	386	r = get_data_size(imp);
	387	// log_debug("get_data_size() -> %d", r);
	388	if (r < 0)
	389	return r;
	390	if (r == 0) {
	391	imp->state = IMPORTER_STATE_EOF;
	392	return 0;
	393	}
	394
	395	imp->state = imp->data_size > 0 ?
	396	IMPORTER_STATE_DATA : IMPORTER_STATE_DATA_FINISH;
	397
	398	return 0; /* continue */
	399
	400	case IMPORTER_STATE_DATA: {
	401	void *data;
	402	char *field;
	403
	404	assert(imp->data_size > 0);
	405
	406	r = get_data_data(imp, &data);
	407	// log_debug("get_data_data() -> %d", r);
	408	if (r < 0)
	409	return r;
	410	if (r == 0) {
	411	imp->state = IMPORTER_STATE_EOF;
	412	return 0;
	413	}
	414
	415	assert(data);
	416
	417	field = (char*) data - sizeof(uint64_t) - imp->field_len;
	418	memmove(field + sizeof(uint64_t), field, imp->field_len);
	419
	420	r = iovw_put(&imp->iovw, field + sizeof(uint64_t), imp->field_len + imp->data_size);
	421	if (r < 0)
	422	return r;
	423
	424	imp->state = IMPORTER_STATE_DATA_FINISH;
	425
	426	return 0; /* continue */
	427	}
	428
429	case IMPORTER_STATE_DATA_FINISH:
430	r = get_data_newline(imp);
431	// log_debug("get_data_newline() -> %d", r);
432	if (r < 0)
433	return r;
434	if (r == 0) {
435	imp->state = IMPORTER_STATE_EOF;
436	return 0;
437	}
438
439	imp->data_size = 0;
440	imp->state = IMPORTER_STATE_LINE;
441
442	return 0; /* continue */
443	default:
444	assert_not_reached("wtf?");
445	}
446	}
447
448	int journal_importer_push_data(JournalImporter imp, const char data, size_t size) {
449	assert(imp);
450	assert(imp->state != IMPORTER_STATE_EOF);
451
452	if (!realloc_buffer(imp, imp->filled + size)) {
453	log_error("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, strerror(ENOMEM));
456	return -ENOMEM;
457	}
458
459	memcpy(imp->buf + imp->filled, data, size);
460	imp->filled += size;
461
462	return 0;
463	}
464
465	void journal_importer_drop_iovw(JournalImporter *imp) {
466	size_t remain, target;
467
468	/* This function drops processed data that along with the iovw that points at it */
469
470	iovw_free_contents(&imp->iovw);
471
472	/* possibly reset buffer position */
473	remain = imp->filled - imp->offset;
474
475	if (remain == 0) /* no brainer */
476	imp->offset = imp->scanned = imp->filled = 0;
477	else if (imp->offset > imp->size - imp->filled &&
478	imp->offset > remain) {
479	memcpy(imp->buf, imp->buf + imp->offset, remain);
480	imp->offset = imp->scanned = 0;
481	imp->filled = remain;
482	}
483
484	target = imp->size;
485	while (target > 16 * LINE_CHUNK && imp->filled < target / 2)
486	target /= 2;
487	if (target < imp->size) {
488	char *tmp;
489
490	tmp = realloc(imp->buf, target);
491	if (!tmp)
492	log_warning("Failed to reallocate buffer to (smaller) size %zu",
493	target);
494	else {
495	log_debug("Reallocated buffer from %zu to %zu bytes",
496	imp->size, target);
497	imp->buf = tmp;
498	imp->size = target;
499	}
500	}
501	}
502
503	bool journal_importer_eof(const JournalImporter *imp) {
504	return imp->state == IMPORTER_STATE_EOF;
505	}