journal-upload: a tool to push messages to systemd-journal-remote

2 files changed, 414 insertions, 0 deletions

diff --git a/src/journal-remote/journal-upload.c b/src/journal-remote/journal-upload.c
new file mode 100644
index 0000000000..e82f440eca
--- /dev/null
+++ b/src/journal-remote/journal-upload.c
@@ -0,0 +1,389 @@
+/*-*- Mode: C; c-basic-offset: 8; indent-tabs-mode: nil -*-*/
+
+/***
+  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 <stdio.h>
+#include <curl/curl.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+#include <getopt.h>
+
+#include "sd-daemon.h"
+
+#include "log.h"
+#include "util.h"
+#include "build.h"
+#include "journal-upload.h"
+
+static const char* arg_url;
+
+static void close_fd_input(Uploader *u);
+
+#define easy_setopt(curl, opt, value, level, cmd)                       \
+        {                                                               \
+                code = curl_easy_setopt(curl, opt, value);              \
+                if (code) {                                             \
+                        log_full(level,                                 \
+                                 "curl_easy_setopt " #opt " failed: %s", \
+                                  curl_easy_strerror(code));            \
+                        cmd;                                            \
+                }                                                       \
+        }
+
+int start_upload(Uploader *u,
+                 size_t (*input_callback)(void *ptr,
+                                          size_t size,
+                                          size_t nmemb,
+                                          void *userdata),
+                 void *data) {
+        CURLcode code;
+
+        assert(u);
+        assert(input_callback);
+
+        if (!u->header) {
+                struct curl_slist *h;
+
+                h = curl_slist_append(NULL, "Content-Type: application/vnd.fdo.journal");
+                if (!h)
+                        return log_oom();
+
+                h = curl_slist_append(h, "Transfer-Encoding: chunked");
+                if (!h) {
+                        curl_slist_free_all(h);
+                        return log_oom();
+                }
+
+                h = curl_slist_append(h, "Accept: text/plain");
+                if (!h) {
+                        curl_slist_free_all(h);
+                        return log_oom();
+                }
+
+                u->header = h;
+        }
+
+        if (!u->easy) {
+                CURL *curl;
+
+                curl = curl_easy_init();
+                if (!curl) {
+                        log_error("Call to curl_easy_init failed.");
+                        return -ENOSR;
+                }
+
+                /* tell it to POST to the URL */
+                easy_setopt(curl, CURLOPT_POST, 1L,
+                            LOG_ERR, return -EXFULL);
+
+                /* set where to read from */
+                easy_setopt(curl, CURLOPT_READFUNCTION, input_callback,
+                            LOG_ERR, return -EXFULL);
+
+                easy_setopt(curl, CURLOPT_READDATA, data,
+                            LOG_ERR, return -EXFULL);
+
+                /* use our special own mime type and chunked transfer */
+                easy_setopt(curl, CURLOPT_HTTPHEADER, u->header,
+                            LOG_ERR, return -EXFULL);
+
+                /* enable verbose for easier tracing */
+                easy_setopt(curl, CURLOPT_VERBOSE, 1L, LOG_WARNING, );
+
+                easy_setopt(curl, CURLOPT_USERAGENT,
+                            "systemd-journal-upload " PACKAGE_STRING,
+                            LOG_WARNING, );
+
+                u->easy = curl;
+        }
+
+        /* upload to this place */
+        code = curl_easy_setopt(u->easy, CURLOPT_URL, u->url);
+        if (code) {
+                log_error("curl_easy_setopt CURLOPT_URL failed: %s",
+                          curl_easy_strerror(code));
+                return -EXFULL;
+        }
+
+        u->uploading = true;
+
+        return 0;
+}
+
+static size_t fd_input_callback(void *buf, size_t size, size_t nmemb, void *userp) {
+        Uploader *u = userp;
+
+        ssize_t r;
+
+        assert(u);
+        assert(nmemb <= SSIZE_MAX / size);
+
+        if (u->input < 0)
+                return 0;
+
+        r = read(u->input, buf, size * nmemb);
+        log_debug("%s: allowed %zu, read %zu", __func__, size*nmemb, r);
+
+        if (r > 0)
+                return r;
+
+        u->uploading = false;
+        if (r == 0) {
+                log_debug("Reached EOF");
+                close_fd_input(u);
+                return 0;
+        } else {
+                log_error("Aborting transfer after read error on input: %m.");
+                return CURL_READFUNC_ABORT;
+        }
+}
+
+static void close_fd_input(Uploader *u) {
+        assert(u);
+
+        if (u->input >= 0)
+                close_nointr(u->input);
+        u->input = -1;
+}
+
+static int dispatch_fd_input(sd_event_source *event,
+                             int fd,
+                             uint32_t revents,
+                             void *userp) {
+        Uploader *u = userp;
+
+        assert(u);
+        assert(revents & EPOLLIN);
+        assert(fd >= 0);
+
+        if (u->uploading) {
+                log_warning("dispatch_fd_input called when uploading, ignoring.");
+                return 0;
+        }
+
+        return start_upload(u, fd_input_callback, u);
+}
+
+static int open_file_for_upload(Uploader *u, const char *filename) {
+        int fd, r;
+
+        if (streq(filename, "-"))
+                fd = STDIN_FILENO;
+        else {
+                fd = open(filename, O_RDONLY|O_CLOEXEC|O_NOCTTY);
+                if (fd < 0) {
+                        log_error("Failed to open %s: %m", filename);
+                        return -errno;
+                }
+        }
+
+        u->input = fd;
+
+        r = sd_event_add_io(u->events, &u->input_event,
+                            fd, EPOLLIN, dispatch_fd_input, u);
+        if (r < 0) {
+                if (r != -EPERM) {
+                        log_error("Failed to register input event: %s", strerror(-r));
+                        return r;
+                }
+
+                /* Normal files should just be consumed without polling. */
+                r = start_upload(u, fd_input_callback, u);
+        }
+        return r;
+}
+
+static int setup_uploader(Uploader *u, const char *url) {
+        int r;
+
+        assert(u);
+        assert(url);
+
+        memzero(u, sizeof(Uploader));
+        u->input = -1;
+
+        u->url = url;
+
+        r = sd_event_default(&u->events);
+        if (r < 0) {
+                log_error("sd_event_default failed: %s", strerror(-r));
+                return r;
+        }
+
+        return 0;
+}
+
+static void destroy_uploader(Uploader *u) {
+        assert(u);
+
+        curl_easy_cleanup(u->easy);
+        curl_slist_free_all(u->header);
+
+        u->input_event = sd_event_source_unref(u->input_event);
+
+        close_fd_input(u);
+
+        sd_event_unref(u->events);
+}
+
+static void help(void) {
+        printf("%s -u URL {FILE|-}...\n\n"
+               "Upload journal events to a remote server.\n\n"
+               "Options:\n"
+               "  --url=URL                Upload to this address\n"
+               "  -h --help                Show this help and exit\n"
+               "  --version                Print version string and exit\n"
+               , program_invocation_short_name);
+}
+
+static int parse_argv(int argc, char *argv[]) {
+        enum {
+                ARG_VERSION = 0x100,
+        };
+
+        static const struct option options[] = {
+                { "help",         no_argument,       NULL, 'h'                },
+                { "version",      no_argument,       NULL, ARG_VERSION        },
+                { "url",          required_argument, NULL, 'u'                },
+                {}
+        };
+
+        int c;
+
+        assert(argc >= 0);
+        assert(argv);
+
+        opterr = 0;
+
+        while ((c = getopt_long(argc, argv, "hu:", options, NULL)) >= 0)
+                switch(c) {
+                case 'h':
+                        help();
+                        return 0 /* done */;
+
+                case ARG_VERSION:
+                        puts(PACKAGE_STRING);
+                        puts(SYSTEMD_FEATURES);
+                        return 0 /* done */;
+
+                case 'u':
+                        if (arg_url) {
+                                log_error("cannot use more than one --url");
+                                return -EINVAL;
+                        }
+
+                        arg_url = optarg;
+                        break;
+
+                case '?':
+                        log_error("Unknown option %s.", argv[optind-1]);
+                        return -EINVAL;
+
+                case ':':
+                        log_error("Missing argument to %s.", argv[optind-1]);
+                        return -EINVAL;
+
+                default:
+                        assert_not_reached("Unhandled option code.");
+                }
+
+        if (!arg_url) {
+                log_error("Required --url/-u option missing.");
+                return -EINVAL;
+        }
+
+        if (optind >= argc) {
+                log_error("Input argument missing.");
+                return -EINVAL;
+        }
+
+        return 1;
+}
+
+
+int main(int argc, char **argv) {
+        Uploader u;
+        int r;
+
+        log_show_color(true);
+        log_parse_environment();
+
+        r = parse_argv(argc, argv);
+        if (r <= 0)
+                goto finish;
+
+        r = setup_uploader(&u, arg_url);
+        if (r < 0)
+                goto cleanup;
+
+        log_debug("%s running as pid "PID_FMT,
+                  program_invocation_short_name, getpid());
+        sd_notify(false,
+                  "READY=1\n"
+                  "STATUS=Processing input...");
+
+        while (true) {
+                if (u.input < 0) {
+                        if (optind >= argc)
+                                break;
+
+                        log_debug("Using %s as input.", argv[optind]);
+
+                        r = open_file_for_upload(&u, argv[optind++]);
+                        if (r < 0)
+                                goto cleanup;
+
+                }
+
+                r = sd_event_get_state(u.events);
+                if (r < 0)
+                        break;
+                if (r == SD_EVENT_FINISHED)
+                        break;
+
+                if (u.uploading) {
+                        CURLcode code;
+
+                        assert(u.easy);
+
+                        code = curl_easy_perform(u.easy);
+                        if (code) {
+                                log_error("Upload to %s failed: %s",
+                                          u.url, curl_easy_strerror(code));
+                                r = -EIO;
+                                break;
+                        } else
+                                log_debug("Upload finished successfully.");
+                }
+
+                r = sd_event_run(u.events, u.input >= 0 ? -1 : 0);
+                if (r < 0) {
+                        log_error("Failed to run event loop: %s", strerror(-r));
+                        break;
+                }
+        }
+
+cleanup:
+        sd_notify(false, "STATUS=Shutting down...");
+        destroy_uploader(&u);
+
+finish:
+        return r == 0 ? EXIT_SUCCESS : EXIT_FAILURE;
+}
diff --git a/src/journal-remote/journal-upload.h b/src/journal-remote/journal-upload.h
new file mode 100644
index 0000000000..68d85be6bc
--- /dev/null
+++ b/src/journal-remote/journal-upload.h
@@ -0,0 +1,25 @@
+#pragma once
+
+#include <inttypes.h>
+
+#include "sd-event.h"
+
+typedef struct Uploader {
+        sd_event *events;
+
+        const char *url;
+        CURL *easy;
+        bool uploading;
+        struct curl_slist *header;
+
+        int input;
+
+        sd_event_source *input_event;
+} Uploader;
+
+int start_upload(Uploader *u,
+                 size_t (*input_callback)(void *ptr,
+                                          size_t size,
+                                          size_t nmemb,
+                                          void *userdata),
+                 void *data);