[thirdparty/systemd.git] / src / journal-remote / journal-upload-journal.c

/***
  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 <curl/curl.h>
#include <stdbool.h>

#include "alloc-util.h"
#include "journal-upload.h"
#include "log.h"
#include "utf8.h"
#include "util.h"
#include "sd-daemon.h"

/**
 * Write up to size bytes to buf. Return negative on error, and number of
 * bytes written otherwise. The last case is a kind of an error too.
 */
static ssize_t write_entry(char *buf, size_t size, Uploader *u) {
        int r;
        size_t pos = 0;

        assert(size <= SSIZE_MAX);

        for (;;) {

                switch(u->entry_state) {
                case ENTRY_CURSOR: {
                        u->current_cursor = mfree(u->current_cursor);

                        r = sd_journal_get_cursor(u->journal, &u->current_cursor);
                        if (r < 0)
                                return log_error_errno(r, "Failed to get cursor: %m");

                        r = snprintf(buf + pos, size - pos,
                                     "__CURSOR=%s\n", u->current_cursor);
                        if (pos + r > size)
                                /* not enough space */
                                return pos;

                        u->entry_state++;

                        if (pos + r == size) {
                                /* exactly one character short, but we don't need it */
                                buf[size - 1] = '\n';
                                return size;
                        }

                        pos += r;
                }       /* fall through */

                case ENTRY_REALTIME: {
                        usec_t realtime;

                        r = sd_journal_get_realtime_usec(u->journal, &realtime);
                        if (r < 0)
                                return log_error_errno(r, "Failed to get realtime timestamp: %m");

                        r = snprintf(buf + pos, size - pos,
                                     "__REALTIME_TIMESTAMP="USEC_FMT"\n", realtime);
                        if (r + pos > size)
                                /* not enough space */
                                return pos;

                        u->entry_state++;

                        if (r + pos == size) {
                                /* exactly one character short, but we don't need it */
                                buf[size - 1] = '\n';
                                return size;
                        }

                        pos += r;
                }       /* fall through */

                case ENTRY_MONOTONIC: {
                        usec_t monotonic;
                        sd_id128_t boot_id;

                        r = sd_journal_get_monotonic_usec(u->journal, &monotonic, &boot_id);
                        if (r < 0)
                                return log_error_errno(r, "Failed to get monotonic timestamp: %m");

                        r = snprintf(buf + pos, size - pos,
                                     "__MONOTONIC_TIMESTAMP="USEC_FMT"\n", monotonic);
                        if (r + pos > size)
                                /* not enough space */
                                return pos;

                        u->entry_state++;

                        if (r + pos == size) {
                                /* exactly one character short, but we don't need it */
                                buf[size - 1] = '\n';
                                return size;
                        }

                        pos += r;
                }       /* fall through */

                case ENTRY_BOOT_ID: {
                        sd_id128_t boot_id;
                        char sid[33];

                        r = sd_journal_get_monotonic_usec(u->journal, NULL, &boot_id);
                        if (r < 0)
                                return log_error_errno(r, "Failed to get monotonic timestamp: %m");

                        r = snprintf(buf + pos, size - pos,
                                     "_BOOT_ID=%s\n", sd_id128_to_string(boot_id, sid));
                        if (r + pos > size)
                                /* not enough space */
                                return pos;

                        u->entry_state++;

                        if (r + pos == size) {
                                /* exactly one character short, but we don't need it */
                                buf[size - 1] = '\n';
                                return size;
                        }

                        pos += r;
                }       /* fall through */

                case ENTRY_NEW_FIELD: {
                        u->field_pos = 0;

                        r = sd_journal_enumerate_data(u->journal,
                                                      &u->field_data,
                                                      &u->field_length);
                        if (r < 0)
                                return log_error_errno(r, "Failed to move to next field in entry: %m");
                        else if (r == 0) {
                                u->entry_state = ENTRY_OUTRO;
                                continue;
                        }

                        if (!utf8_is_printable_newline(u->field_data,
                                                       u->field_length, false)) {
                                u->entry_state = ENTRY_BINARY_FIELD_START;
                                continue;
                        }

                        u->entry_state++;
                }       /* fall through */

                case ENTRY_TEXT_FIELD:
                case ENTRY_BINARY_FIELD: {
                        bool done;
                        size_t tocopy;

                        done = size - pos > u->field_length - u->field_pos;
                        if (done)
                                tocopy = u->field_length - u->field_pos;
                        else
                                tocopy = size - pos;

                        memcpy(buf + pos,
                               (char*) u->field_data + u->field_pos,
                               tocopy);

                        if (done) {
                                buf[pos + tocopy] = '\n';
                                pos += tocopy + 1;
                                u->entry_state = ENTRY_NEW_FIELD;
                                continue;
                        } else {
                                u->field_pos += tocopy;
                                return size;
                        }
                }

                case ENTRY_BINARY_FIELD_START: {
                        const char *c;
                        size_t len;

                        c = memchr(u->field_data, '=', u->field_length);
                        if (!c || c == u->field_data) {
                                log_error("Invalid field.");
                                return -EINVAL;
                        }

                        len = c - (const char*)u->field_data;

                        /* need space for label + '\n' */
                        if (size - pos < len + 1)
                                return pos;

                        memcpy(buf + pos, u->field_data, len);
                        buf[pos + len] = '\n';
                        pos += len + 1;

                        u->field_pos = len + 1;
                        u->entry_state++;
                }       /* fall through */

                case ENTRY_BINARY_FIELD_SIZE: {
                        uint64_t le64;

                        /* need space for uint64_t */
                        if (size - pos < 8)
                                return pos;

                        le64 = htole64(u->field_length - u->field_pos);
                        memcpy(buf + pos, &le64, 8);
                        pos += 8;

                        u->entry_state++;
                        continue;
                }

                case ENTRY_OUTRO:
                        /* need space for '\n' */
                        if (size - pos < 1)
                                return pos;

                        buf[pos++] = '\n';
                        u->entry_state++;
                        u->entries_sent++;

                        return pos;

                default:
                        assert_not_reached("WTF?");
                }
        }
        assert_not_reached("WTF?");
}

static inline void check_update_watchdog(Uploader *u) {
        usec_t watchdog_usec;
        static usec_t before;
        usec_t after;
        usec_t elapsed_time;

        if (sd_watchdog_enabled(false, &watchdog_usec) < 0)
                return;
        if (u->reset_reference_timestamp) {
                before = now(CLOCK_MONOTONIC);
                u->reset_reference_timestamp = false;
        } else {
                after = now(CLOCK_MONOTONIC);
                elapsed_time = usec_sub(after, before);
                if (elapsed_time > watchdog_usec / 2) {
                        log_debug("Update watchdog timer");
                        sd_notify(false, "WATCHDOG=1");
                        u->reset_reference_timestamp = true;
                }
        }
}

static size_t journal_input_callback(void *buf, size_t size, size_t nmemb, void *userp) {
        Uploader *u = userp;
        int r;
        sd_journal *j;
        size_t filled = 0;
        ssize_t w;

        assert(u);
        assert(nmemb <= SSIZE_MAX / size);

        check_update_watchdog(u);

        j = u->journal;

        while (j && filled < size * nmemb) {
                if (u->entry_state == ENTRY_DONE) {
                        r = sd_journal_next(j);
                        if (r < 0) {
                                log_error_errno(r, "Failed to move to next entry in journal: %m");
                                return CURL_READFUNC_ABORT;
                        } else if (r == 0) {
                                if (u->input_event)
                                        log_debug("No more entries, waiting for journal.");
                                else {
                                        log_info("No more entries, closing journal.");
                                        close_journal_input(u);
                                }

                                u->uploading = false;

                                break;
                        }

                        u->entry_state = ENTRY_CURSOR;
                }

                w = write_entry((char*)buf + filled, size * nmemb - filled, u);
                if (w < 0)
                        return CURL_READFUNC_ABORT;
                filled += w;

                if (filled == 0) {
                        log_error("Buffer space is too small to write entry.");
                        return CURL_READFUNC_ABORT;
                } else if (u->entry_state != ENTRY_DONE)
                        /* This means that all available space was used up */
                        break;

                log_debug("Entry %zu (%s) has been uploaded.",
                          u->entries_sent, u->current_cursor);
        }

        return filled;
}

void close_journal_input(Uploader *u) {
        assert(u);

        if (u->journal) {
                log_debug("Closing journal input.");

                sd_journal_close(u->journal);
                u->journal = NULL;
        }
        u->timeout = 0;
}

static int process_journal_input(Uploader *u, int skip) {
        int r;

        if (u->uploading)
                return 0;

        r = sd_journal_next_skip(u->journal, skip);
        if (r < 0)
                return log_error_errno(r, "Failed to skip to next entry: %m");
        else if (r < skip)
                return 0;

        /* have data */
        u->entry_state = ENTRY_CURSOR;
        return start_upload(u, journal_input_callback, u);
}

int check_journal_input(Uploader *u) {
        if (u->input_event) {
                int r;

                r = sd_journal_process(u->journal);
                if (r < 0) {
                        log_error_errno(r, "Failed to process journal: %m");
                        close_journal_input(u);
                        return r;
                }

                if (r == SD_JOURNAL_NOP)
                        return 0;
        }

        return process_journal_input(u, 1);
}

static int dispatch_journal_input(sd_event_source *event,
                                  int fd,
                                  uint32_t revents,
                                  void *userp) {
        Uploader *u = userp;

        assert(u);

        if (u->uploading)
                return 0;

        log_debug("Detected journal input, checking for new data.");
        return check_journal_input(u);
}

int open_journal_for_upload(Uploader *u,
                            sd_journal *j,
                            const char *cursor,
                            bool after_cursor,
                            bool follow) {
        int fd, r, events;

        u->journal = j;

        sd_journal_set_data_threshold(j, 0);

        if (follow) {
                fd = sd_journal_get_fd(j);
                if (fd < 0)
                        return log_error_errno(fd, "sd_journal_get_fd failed: %m");

                events = sd_journal_get_events(j);

                r = sd_journal_reliable_fd(j);
                assert(r >= 0);
                if (r > 0)
                        u->timeout = -1;
                else
                        u->timeout = JOURNAL_UPLOAD_POLL_TIMEOUT;

                r = sd_event_add_io(u->events, &u->input_event,
                                    fd, events, dispatch_journal_input, u);
                if (r < 0)
                        return log_error_errno(r, "Failed to register input event: %m");

                log_debug("Listening for journal events on fd:%d, timeout %d",
                          fd, u->timeout == (uint64_t) -1 ? -1 : (int) u->timeout);
        } else
                log_debug("Not listening for journal events.");

        if (cursor) {
                r = sd_journal_seek_cursor(j, cursor);
                if (r < 0)
                        return log_error_errno(r, "Failed to seek to cursor %s: %m",
                                               cursor);
        }

        return process_journal_input(u, 1 + !!after_cursor);
}
Commit	Line	Data
b5efdb8a LP	1	/***
	2	This file is part of systemd.
	3
	4	Copyright 2014 Zbigniew Jędrzejewski-Szmek
	5
	6	systemd is free software; you can redistribute it and/or modify it
	7	under the terms of the GNU Lesser General Public License as published by
	8	the Free Software Foundation; either version 2.1 of the License, or
	9	(at your option) any later version.
	10
	11	systemd is distributed in the hope that it will be useful, but
	12	WITHOUT ANY WARRANTY; without even the implied warranty of
	13	MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
	14	Lesser General Public License for more details.
	15
	16	You should have received a copy of the GNU Lesser General Public License
	17	along with systemd; If not, see <http://www.gnu.org/licenses/>.
	18	***/
	19
eacbb4d3	20	#include <curl/curl.h>
cf0fbc49	21	#include <stdbool.h>
eacbb4d3	22
b5efdb8a LP	23	#include "alloc-util.h"
b5efdb8a LP	24	#include "journal-upload.h"
eacbb4d3 ZJS	25	#include "log.h"
eacbb4d3 ZJS	26	#include "utf8.h"
b5efdb8a	27	#include "util.h"
d79ca7a6	28	#include "sd-daemon.h"
eacbb4d3 ZJS	29
	30	/**
	31	* Write up to size bytes to buf. Return negative on error, and number of
	32	* bytes written otherwise. The last case is a kind of an error too.
	33	*/
	34	static ssize_t write_entry(char buf, size_t size, Uploader u) {
	35	int r;
	36	size_t pos = 0;
	37
	38	assert(size <= SSIZE_MAX);
	39
57255510	40	for (;;) {
eacbb4d3 ZJS	41
	42	switch(u->entry_state) {
	43	case ENTRY_CURSOR: {
a1e58e8e	44	u->current_cursor = mfree(u->current_cursor);
eacbb4d3	45
722b6795	46	r = sd_journal_get_cursor(u->journal, &u->current_cursor);
eb56eb9b MS	47	if (r < 0)
eb56eb9b MS	48	return log_error_errno(r, "Failed to get cursor: %m");
eacbb4d3 ZJS	49
eacbb4d3 ZJS	50	r = snprintf(buf + pos, size - pos,
722b6795	51	"__CURSOR=%s\n", u->current_cursor);
eacbb4d3 ZJS	52	if (pos + r > size)
	53	/* not enough space */
	54	return pos;
	55
313cefa1	56	u->entry_state++;
eacbb4d3 ZJS	57
	58	if (pos + r == size) {
	59	/* exactly one character short, but we don't need it */
	60	buf[size - 1] = '\n';
	61	return size;
	62	}
	63
	64	pos += r;
	65	} /* fall through */
	66
	67	case ENTRY_REALTIME: {
	68	usec_t realtime;
	69
	70	r = sd_journal_get_realtime_usec(u->journal, &realtime);
eb56eb9b MS	71	if (r < 0)
eb56eb9b MS	72	return log_error_errno(r, "Failed to get realtime timestamp: %m");
eacbb4d3 ZJS	73
	74	r = snprintf(buf + pos, size - pos,
	75	"__REALTIME_TIMESTAMP="USEC_FMT"\n", realtime);
	76	if (r + pos > size)
	77	/* not enough space */
	78	return pos;
	79
313cefa1	80	u->entry_state++;
eacbb4d3 ZJS	81
	82	if (r + pos == size) {
	83	/* exactly one character short, but we don't need it */
	84	buf[size - 1] = '\n';
	85	return size;
	86	}
	87
	88	pos += r;
	89	} /* fall through */
	90
	91	case ENTRY_MONOTONIC: {
	92	usec_t monotonic;
	93	sd_id128_t boot_id;
	94
	95	r = sd_journal_get_monotonic_usec(u->journal, &monotonic, &boot_id);
eb56eb9b MS	96	if (r < 0)
eb56eb9b MS	97	return log_error_errno(r, "Failed to get monotonic timestamp: %m");
eacbb4d3 ZJS	98
	99	r = snprintf(buf + pos, size - pos,
	100	"__MONOTONIC_TIMESTAMP="USEC_FMT"\n", monotonic);
	101	if (r + pos > size)
	102	/* not enough space */
	103	return pos;
	104
313cefa1	105	u->entry_state++;
eacbb4d3 ZJS	106
	107	if (r + pos == size) {
	108	/* exactly one character short, but we don't need it */
	109	buf[size - 1] = '\n';
	110	return size;
	111	}
	112
	113	pos += r;
	114	} /* fall through */
	115
	116	case ENTRY_BOOT_ID: {
	117	sd_id128_t boot_id;
	118	char sid[33];
	119
	120	r = sd_journal_get_monotonic_usec(u->journal, NULL, &boot_id);
eb56eb9b MS	121	if (r < 0)
eb56eb9b MS	122	return log_error_errno(r, "Failed to get monotonic timestamp: %m");
eacbb4d3 ZJS	123
	124	r = snprintf(buf + pos, size - pos,
	125	"_BOOT_ID=%s\n", sd_id128_to_string(boot_id, sid));
5ffa8c81	126	if (r + pos > size)
eacbb4d3 ZJS	127	/* not enough space */
	128	return pos;
	129
313cefa1	130	u->entry_state++;
eacbb4d3 ZJS	131
	132	if (r + pos == size) {
	133	/* exactly one character short, but we don't need it */
	134	buf[size - 1] = '\n';
	135	return size;
	136	}
	137
	138	pos += r;
	139	} /* fall through */
	140
	141	case ENTRY_NEW_FIELD: {
	142	u->field_pos = 0;
	143
	144	r = sd_journal_enumerate_data(u->journal,
	145	&u->field_data,
	146	&u->field_length);
eb56eb9b MS	147	if (r < 0)
	148	return log_error_errno(r, "Failed to move to next field in entry: %m");
	149	else if (r == 0) {
eacbb4d3 ZJS	150	u->entry_state = ENTRY_OUTRO;
	151	continue;
	152	}
	153
	154	if (!utf8_is_printable_newline(u->field_data,
	155	u->field_length, false)) {
	156	u->entry_state = ENTRY_BINARY_FIELD_START;
	157	continue;
	158	}
	159
313cefa1	160	u->entry_state++;
eacbb4d3 ZJS	161	} /* fall through */
	162
	163	case ENTRY_TEXT_FIELD:
	164	case ENTRY_BINARY_FIELD: {
	165	bool done;
	166	size_t tocopy;
	167
	168	done = size - pos > u->field_length - u->field_pos;
	169	if (done)
	170	tocopy = u->field_length - u->field_pos;
	171	else
	172	tocopy = size - pos;
	173
	174	memcpy(buf + pos,
	175	(char*) u->field_data + u->field_pos,
	176	tocopy);
	177
	178	if (done) {
	179	buf[pos + tocopy] = '\n';
	180	pos += tocopy + 1;
	181	u->entry_state = ENTRY_NEW_FIELD;
	182	continue;
	183	} else {
	184	u->field_pos += tocopy;
	185	return size;
	186	}
	187	}
	188
	189	case ENTRY_BINARY_FIELD_START: {
	190	const char *c;
	191	size_t len;
	192
	193	c = memchr(u->field_data, '=', u->field_length);
	194	if (!c \|\| c == u->field_data) {
	195	log_error("Invalid field.");
	196	return -EINVAL;
	197	}
	198
	199	len = c - (const char*)u->field_data;
	200
	201	/* need space for label + '\n' */
	202	if (size - pos < len + 1)
	203	return pos;
	204
	205	memcpy(buf + pos, u->field_data, len);
	206	buf[pos + len] = '\n';
	207	pos += len + 1;
	208
	209	u->field_pos = len + 1;
313cefa1	210	u->entry_state++;
eacbb4d3 ZJS	211	} /* fall through */
	212
	213	case ENTRY_BINARY_FIELD_SIZE: {
	214	uint64_t le64;
	215
	216	/* need space for uint64_t */
	217	if (size - pos < 8)
	218	return pos;
	219
	220	le64 = htole64(u->field_length - u->field_pos);
	221	memcpy(buf + pos, &le64, 8);
	222	pos += 8;
	223
313cefa1	224	u->entry_state++;
eacbb4d3 ZJS	225	continue;
	226	}
	227
	228	case ENTRY_OUTRO:
	229	/* need space for '\n' */
	230	if (size - pos < 1)
	231	return pos;
	232
	233	buf[pos++] = '\n';
313cefa1 VC	234	u->entry_state++;
313cefa1 VC	235	u->entries_sent++;
eacbb4d3 ZJS	236
	237	return pos;
	238
	239	default:
	240	assert_not_reached("WTF?");
	241	}
	242	}
	243	assert_not_reached("WTF?");
	244	}
	245
d79ca7a6 KC	246	static inline void check_update_watchdog(Uploader *u) {
	247	usec_t watchdog_usec;
	248	static usec_t before;
	249	usec_t after;
	250	usec_t elapsed_time;
	251
	252	if (sd_watchdog_enabled(false, &watchdog_usec) < 0)
	253	return;
	254	if (u->reset_reference_timestamp) {
	255	before = now(CLOCK_MONOTONIC);
	256	u->reset_reference_timestamp = false;
	257	} else {
	258	after = now(CLOCK_MONOTONIC);
	259	elapsed_time = usec_sub(after, before);
	260	if (elapsed_time > watchdog_usec / 2) {
	261	log_debug("Update watchdog timer");
	262	sd_notify(false, "WATCHDOG=1");
	263	u->reset_reference_timestamp = true;
	264	}
	265	}
	266	}
	267
eacbb4d3 ZJS	268	static size_t journal_input_callback(void buf, size_t size, size_t nmemb, void userp) {
	269	Uploader *u = userp;
	270	int r;
	271	sd_journal *j;
	272	size_t filled = 0;
	273	ssize_t w;
	274
	275	assert(u);
	276	assert(nmemb <= SSIZE_MAX / size);
	277
d79ca7a6 KC	278	check_update_watchdog(u);
d79ca7a6 KC	279
eacbb4d3 ZJS	280	j = u->journal;
	281
	282	while (j && filled < size * nmemb) {
	283	if (u->entry_state == ENTRY_DONE) {
	284	r = sd_journal_next(j);
	285	if (r < 0) {
c33b3297	286	log_error_errno(r, "Failed to move to next entry in journal: %m");
eacbb4d3 ZJS	287	return CURL_READFUNC_ABORT;
	288	} else if (r == 0) {
	289	if (u->input_event)
	290	log_debug("No more entries, waiting for journal.");
	291	else {
	292	log_info("No more entries, closing journal.");
	293	close_journal_input(u);
	294	}
	295
	296	u->uploading = false;
	297
	298	break;
	299	}
	300
	301	u->entry_state = ENTRY_CURSOR;
	302	}
	303
	304	w = write_entry((char)buf + filled, size nmemb - filled, u);
	305	if (w < 0)
	306	return CURL_READFUNC_ABORT;
	307	filled += w;
	308
	309	if (filled == 0) {
	310	log_error("Buffer space is too small to write entry.");
	311	return CURL_READFUNC_ABORT;
	312	} else if (u->entry_state != ENTRY_DONE)
	313	/* This means that all available space was used up */
	314	break;
	315
	316	log_debug("Entry %zu (%s) has been uploaded.",
722b6795	317	u->entries_sent, u->current_cursor);
eacbb4d3 ZJS	318	}
	319
	320	return filled;
	321	}
	322
	323	void close_journal_input(Uploader *u) {
	324	assert(u);
	325
	326	if (u->journal) {
	327	log_debug("Closing journal input.");
	328
	329	sd_journal_close(u->journal);
	330	u->journal = NULL;
	331	}
	332	u->timeout = 0;
	333	}
	334
	335	static int process_journal_input(Uploader *u, int skip) {
	336	int r;
	337
8a3db16d KC	338	if (u->uploading)
	339	return 0;
	340
eacbb4d3	341	r = sd_journal_next_skip(u->journal, skip);
eb56eb9b MS	342	if (r < 0)
	343	return log_error_errno(r, "Failed to skip to next entry: %m");
	344	else if (r < skip)
eacbb4d3 ZJS	345	return 0;
	346
	347	/* have data */
	348	u->entry_state = ENTRY_CURSOR;
	349	return start_upload(u, journal_input_callback, u);
	350	}
	351
	352	int check_journal_input(Uploader *u) {
	353	if (u->input_event) {
	354	int r;
	355
	356	r = sd_journal_process(u->journal);
	357	if (r < 0) {
da927ba9	358	log_error_errno(r, "Failed to process journal: %m");
eacbb4d3 ZJS	359	close_journal_input(u);
	360	return r;
	361	}
	362
	363	if (r == SD_JOURNAL_NOP)
	364	return 0;
	365	}
	366
	367	return process_journal_input(u, 1);
	368	}
	369
	370	static int dispatch_journal_input(sd_event_source *event,
	371	int fd,
	372	uint32_t revents,
	373	void *userp) {
	374	Uploader *u = userp;
	375
	376	assert(u);
	377
8a3db16d	378	if (u->uploading)
eacbb4d3	379	return 0;
eacbb4d3 ZJS	380
	381	log_debug("Detected journal input, checking for new data.");
	382	return check_journal_input(u);
	383	}
	384
	385	int open_journal_for_upload(Uploader *u,
	386	sd_journal *j,
	387	const char *cursor,
	388	bool after_cursor,
	389	bool follow) {
	390	int fd, r, events;
	391
	392	u->journal = j;
	393
	394	sd_journal_set_data_threshold(j, 0);
	395
	396	if (follow) {
	397	fd = sd_journal_get_fd(j);
eb56eb9b MS	398	if (fd < 0)
eb56eb9b MS	399	return log_error_errno(fd, "sd_journal_get_fd failed: %m");
eacbb4d3 ZJS	400
	401	events = sd_journal_get_events(j);
	402
	403	r = sd_journal_reliable_fd(j);
	404	assert(r >= 0);
	405	if (r > 0)
	406	u->timeout = -1;
	407	else
	408	u->timeout = JOURNAL_UPLOAD_POLL_TIMEOUT;
	409
	410	r = sd_event_add_io(u->events, &u->input_event,
	411	fd, events, dispatch_journal_input, u);
eb56eb9b MS	412	if (r < 0)
eb56eb9b MS	413	return log_error_errno(r, "Failed to register input event: %m");
eacbb4d3 ZJS	414
	415	log_debug("Listening for journal events on fd:%d, timeout %d",
	416	fd, u->timeout == (uint64_t) -1 ? -1 : (int) u->timeout);
	417	} else
	418	log_debug("Not listening for journal events.");
	419
	420	if (cursor) {
	421	r = sd_journal_seek_cursor(j, cursor);
ece174c5	422	if (r < 0)
eb56eb9b MS	423	return log_error_errno(r, "Failed to seek to cursor %s: %m",
eb56eb9b MS	424	cursor);
eacbb4d3 ZJS	425	}
	426
	427	return process_journal_input(u, 1 + !!after_cursor);
	428	}