stuff

1 files changed, 422 insertions, 0 deletions

diff --git a/src/grp-journal/grp-remote/systemd-journal-upload/journal-upload-journal.c b/src/grp-journal/grp-remote/systemd-journal-upload/journal-upload-journal.c
new file mode 100644
index 0000000000..eddfc384fa
--- /dev/null
+++ b/src/grp-journal/grp-remote/systemd-journal-upload/journal-upload-journal.c
@@ -0,0 +1,422 @@
+/***
+  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 "basic/alloc-util.h"
+#include "journal-upload.h"
+#include "basic/log.h"
+#include "basic/utf8.h"
+#include "basic/util.h"
+#include <systemd/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 after;
+        usec_t elapsed_time;
+
+        if (u->watchdog_usec <= 0)
+                return;
+
+        after = now(CLOCK_MONOTONIC);
+        elapsed_time = usec_sub(after, u->watchdog_timestamp);
+        if (elapsed_time > u->watchdog_usec / 2) {
+                log_debug("Update watchdog timer");
+                sd_notify(false, "WATCHDOG=1");
+                u->watchdog_timestamp = after;
+        }
+}
+
+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);
+}