diff options
| -rw-r--r-- | Makefile-man.am | 2 | ||||
| -rw-r--r-- | Makefile.am | 14 | ||||
| -rw-r--r-- | man/systemd-journal-remote.xml | 289 | ||||
| -rw-r--r-- | src/journal/journal-gatewayd.c | 1 | ||||
| -rw-r--r-- | src/journal/journal-remote-parse.c | 415 | ||||
| -rw-r--r-- | src/journal/journal-remote-parse.h | 60 | ||||
| -rw-r--r-- | src/journal/journal-remote-write.c | 124 | ||||
| -rw-r--r-- | src/journal/journal-remote-write.h | 51 | ||||
| -rw-r--r-- | src/journal/journal-remote.c | 738 |
9 files changed, 1693 insertions, 1 deletions
diff --git a/Makefile-man.am b/Makefile-man.am index 346bc63248..eefd5eaa12 100644 --- a/Makefile-man.am +++ b/Makefile-man.am @@ -68,6 +68,7 @@ MANPAGES += \ man/systemd-halt.service.8 \ man/systemd-inhibit.1 \ man/systemd-initctl.service.8 \ + man/systemd-journal-remote.8 \ man/systemd-journald.service.8 \ man/systemd-machine-id-setup.1 \ man/systemd-notify.1 \ @@ -1462,6 +1463,7 @@ EXTRA_DIST += \ man/systemd-inhibit.xml \ man/systemd-initctl.service.xml \ man/systemd-journal-gatewayd.service.xml \ + man/systemd-journal-remote.xml \ man/systemd-journald.service.xml \ man/systemd-localed.service.xml \ man/systemd-logind.service.xml \ diff --git a/Makefile.am b/Makefile.am index fed8561ede..923e3c4720 100644 --- a/Makefile.am +++ b/Makefile.am @@ -3130,6 +3130,17 @@ systemd_cat_SOURCES = \ systemd_cat_LDADD = \ libsystemd-journal-core.la +systemd_journal_remote_SOURCES = \ + src/journal/journal-remote-parse.h \ + src/journal/journal-remote-parse.c \ + src/journal/journal-remote-write.h \ + src/journal/journal-remote-write.c \ + src/journal/journal-remote.c + +systemd_journal_remote_LDADD = \ + libsystemd-internal.la \ + libsystemd-journal-core.la + # using _CFLAGS = in the conditional below would suppress AM_CFLAGS journalctl_CFLAGS = \ $(AM_CFLAGS) @@ -3380,7 +3391,8 @@ noinst_LTLIBRARIES += \ libsystemd-journal-internal.la rootlibexec_PROGRAMS += \ - systemd-journald + systemd-journald \ + systemd-journal-remote rootbin_PROGRAMS += \ journalctl diff --git a/man/systemd-journal-remote.xml b/man/systemd-journal-remote.xml new file mode 100644 index 0000000000..036d0bcb4a --- /dev/null +++ b/man/systemd-journal-remote.xml @@ -0,0 +1,289 @@ +<?xml version='1.0'?> <!--*-nxml-*--> +<!DOCTYPE refentry PUBLIC "-//OASIS//DTD DocBook XML V4.2//EN" +"http://www.oasis-open.org/docbook/xml/4.2/docbookx.dtd"> + +<!-- +This file is part of systemd. + +Copyright 2012 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/>. +--> + +<refentry id="systemd-journal-remote"> + + <refentryinfo> + <title>systemd-journal-remote</title> + <productname>systemd</productname> + + <authorgroup> + <author> + <contrib>Developer</contrib> + <firstname>Zbigniew</firstname> + <surname>Jędrzejewski-Szmek</surname> + <email>zbyszek@in.waw.pl</email> + </author> + </authorgroup> + </refentryinfo> + + <refmeta> + <refentrytitle>systemd-journal-remote</refentrytitle> + <manvolnum>8</manvolnum> + </refmeta> + + <refnamediv> + <refname>systemd-journal-remote</refname> + <refpurpose>Stream journal messages over the network</refpurpose> + </refnamediv> + + <refsynopsisdiv> + <cmdsynopsis> + <command>systemd-journal-remote</command> + <arg choice="opt" rep="repeat">OPTIONS</arg> + <arg choice="opt" rep="norepeat">-o/--output=DIR|FILE</arg> + <arg choice="opt" rep="repeat">SOURCES</arg> + </cmdsynopsis> + </refsynopsisdiv> + + <refsect1> + <title>Description</title> + + <para> + <filename>systemd-journal-remote</filename> is a command to + receive serialized journal events and store them to the journal. + Input streams must be in the + <ulink url="http://www.freedesktop.org/wiki/Software/systemd/export"> + Journal Export Format + </ulink>, + i.e. like the output from + <command>journalctl --output=export</command>. + </para> + </refsect1> + + <refsect1> + <title>Sources</title> + + <para> + Sources can be either "active" + (<command>systemd-journal-remote</command> requests and pulls + the data), or "passive" + (<command>systemd-journal-remote</command> waits for a + connection and than receives events pushed by the other side). + </para> + + <para> + <command>systemd-journal-remote</command> can read more than one + event stream at a time. They will be interleaved in the output + file. In case of "active" connections, each "source" is one + stream, and in case of "passive" connections each connection can + result in a separate stream. Sockets can be configured in + "accept" mode (i.e. only one connection), or "listen" mode (i.e. + multiple connections, each resulting in a stream). + </para> + + <para> + When there are no more connections, and no more can be created + (there are no listening sockets), then + <command>systemd-journal-remote</command> will exit. + </para> + + <para>Active sources can be specified in the following + ways:</para> + + <variablelist> + <varlistentry> + <listitem><para>When <option>-</option> is given as a + positional argument, events will be read from standard input. + Other positional arguments will be treated as filenames + to open and read from.</para></listitem> + </varlistentry> + + <varlistentry> + <term><option>--url=<replaceable>ADDRESS</replaceable></option></term> + + <listitem><para>With the + <option>--url=<replaceable>ADDRESS</replaceable></option> option, + events will be retrieved using HTTP from + <replaceable>ADDRESS</replaceable>. This URL should refer to the + root of a remote + <citerefentry><refentrytitle>systemd-journal-gatewayd</refentrytitle><manvolnum>8</manvolnum></citerefentry> + instance (e.g. <ulink>http://some.host:19531/</ulink> or + <ulink>https://some.host:19531/</ulink>).</para></listitem> + </varlistentry> + </variablelist> + + <para>Passive sources can be specified in the following + ways:</para> + + <variablelist> + <varlistentry> + <term><option>--listen-raw=<replaceable>ADDRESS</replaceable></option></term> + + <listitem><para><option>ADDRESS</option> must be an address + suitable for <option>ListenStream=</option> (c.f. + <citerefentry><refentrytitle>systemd.socket</refentrytitle><manvolnum>5</manvolnum></citerefentry>). + <command>systemd-journal-remote</command> will listen on this + socket for connections. Each connection is expected to be a + stream of journal events.</para> + </listitem> + </varlistentry> + + <varlistentry> + <term><varname>$LISTEN_FDS</varname></term> + + <listitem><para><command>systemd-journal-remote</command> + supports the + <varname>$LISTEN_FDS</varname>/<varname>$LISTEN_PID</varname> + protocol. Open sockets inherited through socket + activation behave like those opened with + <option>--listen-raw=</option> described above. + </para> + </listitem> + </varlistentry> + + </variablelist> + </refsect1> + + <refsect1> + <title>Sinks</title> + + <para>The location of the output journal can be specified + with <option>-o</option> or <option>--output=</option>. + </para> + + <variablelist> + <varlistentry> + <term><option>--output=<replaceable>FILE</replaceable></option></term> + + <listitem><para>Will write to this journal. The filename must + end with <filename>.journal</filename>. The file will be + created if it does not exist. When necessary (journal file + full, or corrupted) the file will be renamed following normal + journald rules and a new journal file will be created in it's + stead. </para></listitem> + </varlistentry> + + <varlistentry> + <term><option>--output=<replaceable>DIR</replaceable></option></term> + + <listitem><para>Will create journal files underneath directory + <replaceable>DIR</replaceable>. The directory must exist. When + necessary (journal files over size, or corrupted) journal + files will be rotated following normal journald rules. Names + of files underneath <replaceable>DIR</replaceable> will be + generated using the rules described below.</para></listitem> + </varlistentry> + </variablelist> + + <para>If <option>--output=</option> is not used, output directory + <filename>/var/log/journal/<replaceable>machine-id</replaceable>/</filename> + will be used, where <replaceable>machine-id</replaceable> is the + identifier of the current system (see + <citerefentry><refentrytitle>machine-id</refentrytitle><manvolnum>5</manvolnum></citerefentry>). + In case the output file is not specified, journal files will be + created underneath the selected directory. Files will be called + <filename>remote-<replaceable>variable</replaceable>.journal</filename>, + where the <replaceable>variable</replaceable> part is generated + based on what passive and active sources are specified. It is + recommended to give a full output filename.</para> + + <para>In case of "active" sources, if the hostname is known it + will be used in the <replaceable>variable</replaceable> part. + Otherwise, local address and port number will be used, or + <literal>stdin</literal> for events passed over standard + input, and <literal>multiple</literal> if more than one source + is specified.</para> + </refsect1> + + <refsect1> + <title>Options</title> + + <para>The following options are understood:</para> + + <variablelist> + <varlistentry> + <term><option>--help</option></term> + <term><option>-h</option></term> + + <listitem><para>Print a short help + text and exit.</para></listitem> + </varlistentry> + + <varlistentry> + <term><option>--version</option></term> + + <listitem><para>Print a short version + string and exit.</para></listitem> + </varlistentry> + + <varlistentry> + <term><option>--compress</option></term> + <term><option>--no-compress</option></term> + + <listitem><para>Compress or not, respectively, the data in the + journal using XZ.</para></listitem> + </varlistentry> + + <varlistentry> + <term><option>--seal</option></term> + <term><option>--no-seal</option></term> + + <listitem><para>Periodically sign or not, respectively, the + data in the journal using Forward Secure Sealing. + </para></listitem> + </varlistentry> + + <varlistentry> + <term><option>--getter=<replaceable>PROG --option1 --option2</replaceable></option></term> + + <listitem><para>Program to invoke to retrieve data. Journal + event stream must be generated on standard output.</para> + + <para>Examples:</para> + + <programlisting>--getter='curl "-HAccept: application/vnd.fdo.journal" https://some.host:19531/'</programlisting> + + <programlisting>--getter='wget --header="Accept: application/vnd.fdo.journal" -O- https://some.host:19531/'</programlisting> + </listitem> + </varlistentry> + </variablelist> + </refsect1> + + <refsect1> + <title>Examples</title> + <para>Copy local journal events to a different journal directory: + <programlisting> +journalctl -o export | systemd-journal-remote -o /tmp/dir - + </programlisting> + </para> + + <para>Retrieve events from a remote + <citerefentry><refentrytitle>systemd-journal-gatewayd</refentrytitle><manvolnum>8</manvolnum></citerefentry> + instance and store them in + <filename>/var/log/journal/some.host/remote-some~host.journal</filename>: + <programlisting> +systemd-journal-remote --url http://some.host:19531/ + </programlisting> + </para> + </refsect1> + + <refsect1> + <title>See Also</title> + <para> + <citerefentry><refentrytitle>journalctl</refentrytitle><manvolnum>1</manvolnum></citerefentry>, + <citerefentry><refentrytitle>systemd-journald.service</refentrytitle><manvolnum>8</manvolnum></citerefentry>, + <citerefentry><refentrytitle>systemd-journal-gatewayd.service</refentrytitle><manvolnum>8</manvolnum></citerefentry>. + </para> + </refsect1> +</refentry> diff --git a/src/journal/journal-gatewayd.c b/src/journal/journal-gatewayd.c index ac16a7cf26..d47b27ef7f 100644 --- a/src/journal/journal-gatewayd.c +++ b/src/journal/journal-gatewayd.c @@ -78,6 +78,7 @@ static const char* const mime_types[_OUTPUT_MODE_MAX] = { static RequestMeta *request_meta(void **connection_cls) { RequestMeta *m; + assert(connection_cls); if (*connection_cls) return *connection_cls; diff --git a/src/journal/journal-remote-parse.c b/src/journal/journal-remote-parse.c new file mode 100644 index 0000000000..ee2260c5a9 --- /dev/null +++ b/src/journal/journal-remote-parse.c @@ -0,0 +1,415 @@ +/*-*- 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 "journal-remote-parse.h" +#include "journald-native.h" + +#define LINE_CHUNK 1024u + +void source_free(RemoteSource *source) { + if (!source) + return; + + if (source->fd >= 0) { + log_debug("Closing fd:%d (%s)", source->fd, source->name); + close(source->fd); + } + free(source->name); + free(source->buf); + iovw_free_contents(&source->iovw); + free(source); +} + +static int get_line(RemoteSource *source, char **line, size_t *size) { + ssize_t n, remain; + char *c; + char *newbuf = NULL; + size_t newsize = 0; + + assert(source); + assert(source->state == STATE_LINE); + assert(source->filled <= source->size); + assert(source->buf == NULL || source->size > 0); + + c = memchr(source->buf, '\n', source->filled); + if (c != NULL) + goto docopy; + + resize: + if (source->size - source->filled < LINE_CHUNK) { + // XXX: add check for maximum line length + + if (!GREEDY_REALLOC(source->buf, source->size, + source->filled + LINE_CHUNK)) + return log_oom(); + } + assert(source->size - source->filled >= LINE_CHUNK); + + n = read(source->fd, source->buf + source->filled, + source->size - source->filled); + if (n < 0) { + if (errno != EAGAIN && errno != EWOULDBLOCK) + log_error("read(%d, ..., %zd): %m", source->fd, + source->size - source->filled); + return -errno; + } else if (n == 0) + return 0; + + c = memchr(source->buf + source->filled, '\n', n); + source->filled += n; + + if (c == NULL) + goto resize; + + docopy: + *line = source->buf; + *size = c + 1 - source->buf; + + /* Check if something remains */ + remain = source->buf + source->filled - c - 1; + assert(remain >= 0); + if (remain) { + newsize = MAX(remain, LINE_CHUNK); + newbuf = malloc(newsize); + if (!newbuf) + return log_oom(); + memcpy(newbuf, c + 1, remain); + } + source->buf = newbuf; + source->size = newsize; + source->filled = remain; + + return 1; +} + +static int fill_fixed_size(RemoteSource *source, void **data, size_t size) { + int n; + char *newbuf = NULL; + size_t newsize = 0, remain; + + assert(source); + assert(source->state == STATE_DATA_START || + source->state == STATE_DATA || + source->state == STATE_DATA_FINISH); + assert(size <= DATA_SIZE_MAX); + assert(source->filled <= source->size); + assert(source->buf != NULL || source->size == 0); + assert(source->buf == NULL || source->size > 0); + assert(data); + + while(source->filled < size) { + if (!GREEDY_REALLOC(source->buf, source->size, size)) + return log_oom(); + + n = read(source->fd, source->buf + source->filled, + source->size - source->filled); + if (n < 0) { + if (errno != EAGAIN && errno != EWOULDBLOCK) + log_error("read(%d, ..., %zd): %m", source->fd, + source->size - source->filled); + return -errno; + } else if (n == 0) + return 0; + + source->filled += n; + } + + *data = source->buf; + + /* Check if something remains */ + assert(size <= source->filled); + remain = source->filled - size; + if (remain) { + newsize = MAX(remain, LINE_CHUNK); + newbuf = malloc(newsize); + if (!newbuf) + return log_oom(); + memcpy(newbuf, source->buf + size, remain); + } + source->buf = newbuf; + source->size = newsize; + source->filled = remain; + + return 1; +} + +static int get_data_size(RemoteSource *source) { + int r; + void _cleanup_free_ *data = NULL; + + assert(source); + assert(source->state == STATE_DATA_START); + assert(source->data_size == 0); + + r = fill_fixed_size(source, &data, sizeof(uint64_t)); + if (r <= 0) + return r; + + source->data_size = le64toh( *(uint64_t *) data ); + if (source->data_size > DATA_SIZE_MAX) { + log_error("Stream declares field with size %zu > %u == DATA_SIZE_MAX", + source->data_size, DATA_SIZE_MAX); + return -EINVAL; + } + if (source->data_size == 0) + log_warning("Binary field with zero length"); + + return 1; +} + +static int get_data_data(RemoteSource *source, void **data) { + int r; + + assert(source); + assert(data); + assert(source->state == STATE_DATA); + + r = fill_fixed_size(source, data, source->data_size); + if (r <= 0) + return r; + + return 1; +} + +static int get_data_newline(RemoteSource *source) { + int r; + char _cleanup_free_ *data = NULL; + + assert(source); + assert(source->state == STATE_DATA_FINISH); + + r = fill_fixed_size(source, (void**) &data, 1); + if (r <= 0) + return r; + + assert(data); + if (*data != '\n') { + log_error("expected newline, got '%c'", *data); + return -EINVAL; + } + + return 1; +} + +static int process_dunder(RemoteSource *source, char *line, size_t n) { + const char *timestamp; + int r; + + assert(line); + assert(n > 0); + assert(line[n-1] == '\n'); + + /* XXX: is it worth to support timestamps in extended format? + * We don't produce them, but who knows... */ + + timestamp = startswith(line, "__CURSOR="); + if (timestamp) + /* ignore __CURSOR */ + return 1; + + timestamp = startswith(line, "__REALTIME_TIMESTAMP="); + if (timestamp) { + long long unsigned x; + line[n-1] = '\0'; + r = safe_atollu(timestamp, &x); + if (r < 0) + log_warning("Failed to parse __REALTIME_TIMESTAMP: '%s'", timestamp); + else + source->ts.realtime = x; + return r < 0 ? r : 1; + } + + timestamp = startswith(line, "__MONOTONIC_TIMESTAMP="); + if (timestamp) { + long long unsigned x; + line[n-1] = '\0'; + r = safe_atollu(timestamp, &x); + if (r < 0) + log_warning("Failed to parse __MONOTONIC_TIMESTAMP: '%s'", timestamp); + else + source->ts.monotonic = x; + return r < 0 ? r : 1; + } + + timestamp = startswith(line, "__"); + if (timestamp) { + log_notice("Unknown dunder line %s", line); + return 1; + } + + /* no dunder */ + return 0; +} + +int process_data(RemoteSource *source) { + int r; + + switch(source->state) { + case STATE_LINE: { + char *line, *sep; + size_t n; + + assert(source->data_size == 0); + + r = get_line(source, &line, &n); + if (r < 0) + return r; + if (r == 0) { + source->state = STATE_EOF; + return r; + } + assert(n > 0); + assert(line[n-1] == '\n'); + + if (n == 1) { + log_debug("Received empty line, event is ready"); + free(line); + return 1; + } + + r = process_dunder(source, line, n); + if (r != 0) { + free(line); + return r < 0 ? r : 0; + } + + /* MESSAGE=xxx\n + or + COREDUMP\n + LLLLLLLL0011223344...\n + */ + sep = memchr(line, '=', n); + if (sep) + /* chomp newline */ + n--; + else + /* replace \n with = */ + line[n-1] = '='; + log_debug("Received: %.*s", (int) n, line); + + r = iovw_put(&source->iovw, line, n); + if (r < 0) { + log_error("Failed to put line in iovect"); + free(line); + return r; + } + + if (!sep) + source->state = STATE_DATA_START; + return 0; /* continue */ + } + + case STATE_DATA_START: + assert(source->data_size == 0); + + r = get_data_size(source); + log_debug("get_data_size() -> %d", r); + if (r < 0) + return r; + if (r == 0) { + source->state = STATE_EOF; + return 0; + } + + source->state = source->data_size > 0 ? + STATE_DATA : STATE_DATA_FINISH; + + return 0; /* continue */ + + case STATE_DATA: { + void *data; + + assert(source->data_size > 0); + + r = get_data_data(source, &data); + log_debug("get_data_data() -> %d", r); + if (r < 0) + return r; + if (r == 0) { + source->state = STATE_EOF; + return 0; + } + + assert(data); + + r = iovw_put(&source->iovw, data, source->data_size); + if (r < 0) { + log_error("failed to put binary buffer in iovect"); + return r; + } + + source->state = STATE_DATA_FINISH; + + return 0; /* continue */ + } + + case STATE_DATA_FINISH: + r = get_data_newline(source); + log_debug("get_data_newline() -> %d", r); + if (r < 0) + return r; + if (r == 0) { + source->state = STATE_EOF; + return 0; + } + + source->data_size = 0; + source->state = STATE_LINE; + + return 0; /* continue */ + default: + assert_not_reached("wtf?"); + } +} + +int process_source(RemoteSource *source, Writer *writer, bool compress, bool seal) { + int r; + + assert(source); + assert(writer); + + r = process_data(source); + if (r <= 0) + return r; + + /* We have a full event */ + log_info("Received a full event from source@%p fd:%d (%s)", + source, source->fd, source->name); + + if (!source->iovw.count) { + log_warning("Entry with no payload, skipping"); + goto freeing; + } + + assert(source->iovw.iovec); + assert(source->iovw.count); + + r = writer_write(writer, &source->iovw, &source->ts, compress, seal); + if (r < 0) + log_error("Failed to write entry of %zu bytes: %s", + iovw_size(&source->iovw), strerror(-r)); + else + r = 1; + + freeing: + iovw_free_contents(&source->iovw); + return r; +} diff --git a/src/journal/journal-remote-parse.h b/src/journal/journal-remote-parse.h new file mode 100644 index 0000000000..3bda97e2d4 --- /dev/null +++ b/src/journal/journal-remote-parse.h @@ -0,0 +1,60 @@ +/*-*- 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/>. +***/ + +#pragma once + +#include "sd-event.h" +#include "journal-remote-write.h" + +typedef enum { + STATE_LINE = 0, /* waiting to read, or reading line */ + STATE_DATA_START, /* reading binary data header */ + STATE_DATA, /* reading binary data */ + STATE_DATA_FINISH, /* expecting newline */ + STATE_EOF, /* done */ +} source_state; + +typedef struct RemoteSource { + char* name; + int fd; + + char *buf; + size_t size; + size_t filled; + size_t data_size; + + struct iovec_wrapper iovw; + + source_state state; + dual_timestamp ts; + + sd_event_source *event; +} RemoteSource; + +static inline int source_non_empty(RemoteSource *source) { + assert(source); + + return source->filled > 0; +} + +void source_free(RemoteSource *source); +int process_data(RemoteSource *source); +int process_source(RemoteSource *source, Writer *writer, bool compress, bool seal); diff --git a/src/journal/journal-remote-write.c b/src/journal/journal-remote-write.c new file mode 100644 index 0000000000..4d142bdc97 --- /dev/null +++ b/src/journal/journal-remote-write.c @@ -0,0 +1,124 @@ +/*-*- Mode: C; c-basic-offset: 8; indent-tabs-mode: nil -*-*/ + +/*** + This file is part of systemd. + + Copyright 2012 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 "journal-remote-write.h" + +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++] = (struct iovec) {data, len}; + return 0; +} + +void iovw_free_contents(struct iovec_wrapper *iovw) { + for (size_t j = 0; j < iovw->count; j++) + free(iovw->iovec[j].iov_base); + free(iovw->iovec); + iovw->iovec = NULL; + iovw->size_bytes = iovw->count = 0; +} + +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; +} + +/********************************************************************** + ********************************************************************** + **********************************************************************/ + +static int do_rotate(JournalFile **f, bool compress, bool seal) { + int r = journal_file_rotate(f, compress, seal); + if (r < 0) { + if (*f) + log_error("Failed to rotate %s: %s", (*f)->path, + strerror(-r)); + else + log_error("Failed to create rotated journal: %s", + strerror(-r)); + } + + return r; +} + +int writer_init(Writer *s) { + assert(s); + + s->journal = NULL; + + memset(&s->metrics, 0xFF, sizeof(s->metrics)); + + s->mmap = mmap_cache_new(); + if (!s->mmap) + return log_oom(); + + s->seqnum = 0; + + return 0; +} + +int writer_close(Writer *s) { + if (s->journal) + journal_file_close(s->journal); + if (s->mmap) + mmap_cache_unref(s->mmap); + return 0; +} + +int writer_write(Writer *s, + struct iovec_wrapper *iovw, + dual_timestamp *ts, + bool compress, + bool seal) { + int r; + + assert(s); + assert(iovw); + assert(iovw->count > 0); + + if (journal_file_rotate_suggested(s->journal, 0)) { + log_info("%s: Journal header limits reached or header out-of-date, rotating", + s->journal->path); + r = do_rotate(&s->journal, compress, seal); + if (r < 0) + return r; + } + + r = journal_file_append_entry(s->journal, ts, iovw->iovec, iovw->count, + &s->seqnum, NULL, NULL); + if (r >= 0) + return 1; + + log_info("%s: Write failed, rotating", s->journal->path); + r = do_rotate(&s->journal, compress, seal); + if (r < 0) + return r; + + log_debug("Retrying write."); + r = journal_file_append_entry(s->journal, ts, iovw->iovec, iovw->count, + &s->seqnum, NULL, NULL); + return r < 0 ? r : 1; +} diff --git a/src/journal/journal-remote-write.h b/src/journal/journal-remote-write.h new file mode 100644 index 0000000000..8798216415 --- /dev/null +++ b/src/journal/journal-remote-write.h @@ -0,0 +1,51 @@ +/*-*- 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/>. +***/ + +#pragma once + +#include <stdlib.h> + +#include "journal-file.h" + +struct iovec_wrapper { + struct iovec *iovec; + size_t size_bytes; + size_t count; +}; + +int iovw_put(struct iovec_wrapper *iovw, void* data, size_t len); +void iovw_free_contents(struct iovec_wrapper *iovw); +size_t iovw_size(struct iovec_wrapper *iovw); + +typedef struct Writer { + JournalFile *journal; + JournalMetrics metrics; + MMapCache *mmap; + uint64_t seqnum; +} Writer; + +int writer_init(Writer *s); +int writer_close(Writer *s); +int writer_write(Writer *s, + struct iovec_wrapper *iovw, + dual_timestamp *ts, + bool compress, + bool seal); diff --git a/src/journal/journal-remote.c b/src/journal/journal-remote.c new file mode 100644 index 0000000000..f8979daca6 --- /dev/null +++ b/src/journal/journal-remote.c @@ -0,0 +1,738 @@ +/*-*- Mode: C; c-basic-offset: 8; indent-tabs-mode: nil -*-*/ + +/*** + This file is part of systemd. + + Copyright 2012 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 <errno.h> +#include <fcntl.h> +#include <inttypes.h> +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <sys/prctl.h> +#include <sys/socket.h> +#include <sys/stat.h> +#include <sys/types.h> +#include <unistd.h> +#include <getopt.h> + +#include "sd-daemon.h" +#include "sd-event.h" +#include "journal-file.h" +#include "journald-native.h" +#include "socket-util.h" +#include "mkdir.h" +#include "build.h" +#include "macro.h" +#include "strv.h" + +#include "journal-remote-parse.h" +#include "journal-remote-write.h" + +#define REMOTE_JOURNAL_PATH "/var/log/journal/" SD_ID128_FORMAT_STR "/remote-%s.journal" + +static char* arg_output = NULL; +static char* arg_url = NULL; +static char* arg_getter = NULL; +static bool arg_stdin = false; +static char* arg_listen_raw = NULL; +static int arg_compress = true; +static int arg_seal = false; + +/********************************************************************** + ********************************************************************** + **********************************************************************/ + +static int spawn_child(const char* child, char** argv) { + int fd[2]; + pid_t parent_pid, child_pid; + int r; + + if (pipe(fd) < 0) { + log_error("Failed to create pager pipe: %m"); + return -errno; + } + + parent_pid = getpid(); + + child_pid = fork(); + if (child_pid < 0) { + r = -errno; + log_error("Failed to fork: %m"); + close_pipe(fd); + return r; + } + + /* In the child */ + if (child_pid == 0) { + r = dup2(fd[1], STDOUT_FILENO); + if (r < 0) { + log_error("Failed to dup pipe to stdout: %m"); + _exit(EXIT_FAILURE); + } + + r = close_pipe(fd); + if (r < 0) + log_warning("Failed to close pipe fds: %m"); + + /* Make sure the child goes away when the parent dies */ + if (prctl(PR_SET_PDEATHSIG, SIGTERM) < 0) + _exit(EXIT_FAILURE); + + /* Check whether our parent died before we were able + * to set the death signal */ + if (getppid() != parent_pid) + _exit(EXIT_SUCCESS); + + execvp(child, argv); + log_error("Failed to exec child %s: %m", child); + _exit(EXIT_FAILURE); + } + + r = close(fd[1]); + if (r < 0) + log_warning("Failed to close write end of pipe: %m"); + + return fd[0]; +} + +static int spawn_curl(char* url) { + char **argv = STRV_MAKE("curl", + "-HAccept: application/vnd.fdo.journal", + "--silent", + "--show-error", + url); + int r; + + r = spawn_child("curl", argv); + if (r < 0) + log_error("Failed to spawn curl: %m"); + return r; +} + +static int spawn_getter(char *getter, char *url) { + int r; + char _cleanup_strv_free_ **words = NULL, **words2 = NULL; + + assert(getter); + words = strv_split_quoted(getter); + if (!words) + return log_oom(); + + r = spawn_child(words[0], words); + if (r < 0) + log_error("Failed to spawn getter %s: %m", getter); + + return r; +} + +static int open_output(Writer *s, const char* url) { + char _cleanup_free_ *name, *output = NULL; + char *c; + int r; + + assert(url); + name = strdup(url); + if (!name) + return log_oom(); + + for(c = name; *c; c++) { + if (*c == '/' || *c == ':' || *c == ' ') + *c = '~'; + else if (*c == '?') { + *c = '\0'; + break; + } + } + + if (!arg_output) { + sd_id128_t machine; + r = sd_id128_get_machine(&machine); + if (r < 0) { + log_error("failed to determine machine ID128: %s", strerror(-r)); + return r; + } + + r = asprintf(&output, REMOTE_JOURNAL_PATH, + SD_ID128_FORMAT_VAL(machine), name); + if (r < 0) + return log_oom(); + } else { + r = is_dir(arg_output); + if (r > 0) { + r = asprintf(&output, + "%s/remote-%s.journal", arg_output, name); + if (r < 0) + return log_oom(); + } else { + output = strdup(arg_output); + if (!output) + return log_oom(); + } + } + + r = journal_file_open_reliably(output, + O_RDWR|O_CREAT, 0640, + arg_compress, arg_seal, + &s->metrics, + s->mmap, + NULL, &s->journal); + if (r < 0) + log_error("Failed to open output journal %s: %s", + arg_output, strerror(-r)); + else + log_info("Opened output file %s", s->journal->path); + return r; +} + +typedef struct RemoteServer { + RemoteSource **sources; + ssize_t sources_size; + ssize_t active; + + sd_event *events; + sd_event_source *sigterm_event, *sigint_event, *listen_event; + + Writer writer; +} RemoteServer; + +static int dispatch_raw_source_event(sd_event_source *event, + int fd, + uint32_t revents, + void *userdata); +static int dispatch_raw_connection_event(sd_event_source *event, + int fd, + uint32_t revents, + void *userdata); + +static int get_source_for_fd(RemoteServer *s, int fd, RemoteSource **source) { + assert(fd >= 0); + assert(source); + + if (!GREEDY_REALLOC0_T(s->sources, s->sources_size, fd + 1)) + return log_oom(); + + if (s->sources[fd] == NULL) { + s->sources[fd] = new0(RemoteSource, 1); + if (!s->sources[fd]) + return log_oom(); + s->sources[fd]->fd = -1; + s->active++; + } + + *source = s->sources[fd]; + return 0; +} + +static int remove_source(RemoteServer *s, int fd) { + RemoteSource *source; + + assert(s); + assert(fd >= 0); + assert(fd < s->sources_size); + + source = s->sources[fd]; + if (source) { + source_free(source); + s->sources[fd] = NULL; + s->active--; + } + + close(fd); + + return 0; +} + +static int add_source(RemoteServer *s, int fd, const char* name) { + RemoteSource *source = NULL; + char *realname; + int r; + + assert(s); + assert(fd >= 0); + + if (name) { + realname = strdup(name); + if (!realname) + return log_oom(); + } else { + r = asprintf(&realname, "fd:%d", fd); + if (r < 0) + return log_oom(); + } + + log_debug("Creating source for fd:%d (%s)", fd, name); + + r = get_source_for_fd(s, fd, &source); + if (r < 0) { + log_error("Failed to create source for fd:%d (%s)", fd, name); + return r; + } + assert(source); + assert(source->fd < 0); + source->fd = fd; + + r = sd_event_add_io(s->events, &source->event, + fd, EPOLLIN, dispatch_raw_source_event, s); + if (r < 0) { + log_error("Failed to register event source for fd:%d: %s", + fd, strerror(-r)); + goto error; + } + + return 1; /* work to do */ + + error: + remove_source(s, fd); + return r; +} + +static int setup_raw_socket(RemoteServer *s, const char *address) { + int fd, r; + + fd = make_socket_fd(LOG_INFO, address, SOCK_STREAM | SOCK_CLOEXEC); + if (fd < 0) + return fd; + + r = sd_event_add_io(s->events, &s->listen_event, fd, EPOLLIN, + dispatch_raw_connection_event, s); + if (r < 0) { + close(fd); + return r; + } + + s->active ++; + return 0; +} + +/********************************************************************** + ********************************************************************** + **********************************************************************/ + +static int dispatch_sigterm(sd_event_source *event, + const struct signalfd_siginfo *si, + void *userdata) { + RemoteServer *s = userdata; + + assert(s); + + log_received_signal(LOG_INFO, si); + + sd_event_exit(s->events, 0); + return 0; +} + +static int setup_signals(RemoteServer *s) { + sigset_t mask; + int r; + + assert(s); + + assert_se(sigemptyset(&mask) == 0); + sigset_add_many(&mask, SIGINT, SIGTERM, -1); + assert_se(sigprocmask(SIG_SETMASK, &mask, NULL) == 0); + + r = sd_event_add_signal(s->events, &s->sigterm_event, SIGTERM, dispatch_sigterm, s); + if (r < 0) + return r; + + r = sd_event_add_signal(s->events, &s->sigint_event, SIGINT, dispatch_sigterm, s); + if (r < 0) + return r; + + return 0; +} + +static int remoteserver_init(RemoteServer *s) { + int r, n, fd; + const char *output_name = NULL; + + assert(s); + + sd_event_default(&s->events); + + setup_signals(s); + + n = sd_listen_fds(true); + if (n < 0) { + log_error("Failed to read listening file descriptors from environment: %s", + strerror(-n)); + return n; + } else + log_info("Received %d descriptors", n); + + for (fd = SD_LISTEN_FDS_START; fd < SD_LISTEN_FDS_START + n; fd++) { + if (sd_is_socket(fd, AF_UNSPEC, 0, false)) { + assert_not_reached("not implemented"); + } else if (sd_is_socket(fd, AF_UNSPEC, 0, true)) { + log_info("Received a connection socket (fd:%d)", fd); + + r = add_source(s, fd, NULL); + output_name = "socket"; + } else { + log_error("Unknown socket passed on fd:%d", fd); + return -EINVAL; + } + } + + if (arg_url) { + char _cleanup_free_ *url = NULL; + char _cleanup_strv_free_ **urlv = strv_new(arg_url, "/entries", NULL); + if (!urlv) + return log_oom(); + url = strv_join(urlv, ""); + if (!url) + return log_oom(); + + if (arg_getter) { + log_info("Spawning getter %s...", url); + fd = spawn_getter(arg_getter, url); + } else { + log_info("Spawning curl %s...", url); + fd = spawn_curl(url); + } + if (fd < 0) + return fd; + + r = add_source(s, fd, arg_url); + if (r < 0) + return r; + + output_name = arg_url; + } + + if (arg_listen_raw) { + log_info("Listening on a socket..."); + r = setup_raw_socket(s, arg_listen_raw); + if (r < 0) + return r; + + output_name = arg_listen_raw; + } + + if (arg_stdin) { + log_info("Reading standard input..."); + r = add_source(s, STDIN_FILENO, "stdin"); + if (r < 0) + return r; + + output_name = "stdin"; + } + + if (s->active == 0) { + log_error("Zarro sources specified"); + return -EINVAL; + } + + if (!!n + !!arg_url + !!arg_listen_raw + !!arg_stdin > 1) + output_name = "multiple"; + + r = writer_init(&s->writer); + if (r < 0) + return r; + + r = open_output(&s->writer, output_name); + return r; +} + +static int server_destroy(RemoteServer *s) { + int r; + ssize_t i; + + r = writer_close(&s->writer); + + assert(s->sources_size == 0 || s->sources); + for(i = 0; i < s->sources_size; i++) + remove_source(s, i); + + free(s->sources); + + sd_event_source_unref(s->sigterm_event); + sd_event_source_unref(s->sigint_event); + sd_event_source_unref(s->listen_event); + sd_event_unref(s->events); + + /* fds that we're listening on remain open... */ + + return r; +} + +/********************************************************************** + ********************************************************************** + **********************************************************************/ + +static int dispatch_raw_source_event(sd_event_source *event, + int fd, + uint32_t revents, + void *userdata) { + + RemoteServer *s = userdata; + RemoteSource *source; + int r; + + assert(fd < s->sources_size); + source = s->sources[fd]; + assert(source->fd == fd); + + r = process_source(source, &s->writer, arg_compress, arg_seal); + if (source->state == STATE_EOF) { + log_info("EOF reached with source fd:%d (%s)", + source->fd, source->name); + if (source_non_empty(source)) + log_warning("EOF reached with incomplete data"); + remove_source(s, source->fd); + log_info("%zd active source remaining", s->active); + } else if (r == -E2BIG) { + log_error("Entry too big, skipped"); + r = 1; + } + + return r; +} + +static int dispatch_raw_connection_event(sd_event_source *event, + int fd, + uint32_t revents, + void *userdata) { + RemoteServer *s = userdata; + + SocketAddress addr = { + .size = sizeof(union sockaddr_union), + .type = SOCK_STREAM, + }; + int fd2, r; + + log_debug("Accepting new connection on fd:%d", fd); + fd2 = accept4(fd, &addr.sockaddr.sa, &addr.size, SOCK_NONBLOCK|SOCK_CLOEXEC); + if (fd2 < 0) { + log_error("accept() on fd:%d failed: %m", fd); + return -errno; + } + + switch(socket_address_family(&addr)) { + case AF_INET: + case AF_INET6: { + char* _cleanup_free_ a = NULL; + + r = socket_address_print(&addr, &a); + if (r < 0) { + log_error("socket_address_print(): %s", strerror(-r)); + close(fd2); + return r; + } + + log_info("Accepted %s connection from %s", + socket_address_family(&addr) == AF_INET ? "IP" : "IPv6", + a); + break; + }; + default: + log_error("Connection with unsupported family %d", + socket_address_family(&addr)); + close(fd2); + return -EINVAL; + } + + r = add_source(s, fd2, NULL); + if (r < 0) + log_error("failed to create source from fd:%d: %s", fd2, strerror(-r)); + + return r; +} + + +/********************************************************************** + ********************************************************************** + **********************************************************************/ + +static int help(void) { + printf("%s [OPTIONS...]\n\n" + "Write external journal events to a journal file.\n\n" + "Options:\n" + " --url=URL Read events from systemd-journal-gatewayd at URL\n" + " --getter=COMMAND Read events from the output of COMMAND\n" + " --listen-raw=ADDR Listen for connections at ADDR\n" + " --stdin Read events from standard input\n" + " -o --output=FILE|DIR Write output to FILE or DIR/external-*.journal\n" + " --[no-]compress Use XZ-compression in the output journal (default: yes)\n" + " --[no-]seal Use Event sealing in the output journal (default: no)\n" + " -h --help Show this help and exit\n" + " --version Print version string and exit\n" + "\n" + "Note: file descriptors from sd_listen_fds() will be consumed, too.\n" + , program_invocation_short_name); + + return 0; +} + +static int parse_argv(int argc, char *argv[]) { + enum { + ARG_VERSION = 0x100, + ARG_URL, + ARG_LISTEN_RAW, + ARG_STDIN, + ARG_GETTER, + ARG_COMPRESS, + ARG_NO_COMPRESS, + ARG_SEAL, + ARG_NO_SEAL, + }; + + static const struct option options[] = { + { "help", no_argument, NULL, 'h' }, + { "version", no_argument, NULL, ARG_VERSION }, + { "url", required_argument, NULL, ARG_URL }, + { "getter", required_argument, NULL, ARG_GETTER }, + { "listen-raw", required_argument, NULL, ARG_LISTEN_RAW }, + { "stdin", no_argument, NULL, ARG_STDIN }, + { "output", required_argument, NULL, 'o' }, + { "compress", no_argument, NULL, ARG_COMPRESS }, + { "no-compress", no_argument, NULL, ARG_NO_COMPRESS }, + { "seal", no_argument, NULL, ARG_SEAL }, + { "no-seal", no_argument, NULL, ARG_NO_SEAL }, + {} + }; + + int c; + + assert(argc >= 0); + assert(argv); + + while ((c = getopt_long(argc, argv, "ho:", options, NULL)) >= 0) + switch(c) { + case 'h': + help(); + return 0 /* done */; + + case ARG_VERSION: + puts(PACKAGE_STRING); + puts(SYSTEMD_FEATURES); + return 0 /* done */; + + case ARG_URL: + if (arg_url) { + log_error("cannot currently set more than one --url"); + return -EINVAL; + } + + arg_url = optarg; + break; + + case ARG_GETTER: + if (arg_getter) { + log_error("cannot currently use --getter more than once"); + return -EINVAL; + } + + arg_getter = optarg; + break; + + case ARG_LISTEN_RAW: + if (arg_listen_raw) { + log_error("cannot currently use --listen-raw more than once"); + return -EINVAL; + } + + arg_listen_raw = optarg; + break; + + case ARG_STDIN: + arg_stdin = true; + break; + + case 'o': + if (arg_output) { + log_error("cannot use --output/-o more than once"); + return -EINVAL; + } + + arg_output = optarg; + break; + + case ARG_COMPRESS: + arg_compress = true; + break; + case ARG_NO_COMPRESS: + arg_compress = false; + break; + case ARG_SEAL: + arg_seal = true; + break; + case ARG_NO_SEAL: + arg_seal = false; + break; + + case '?': + return -EINVAL; + + default: + log_error("Unknown option code %c", c); + return -EINVAL; + } + + if (optind < argc) { + log_error("This program takes no positional arguments"); + return -EINVAL; + } + + return 1 /* work to do */; +} + +int main(int argc, char **argv) { + RemoteServer s = {}; + int r, r2; + + log_set_max_level(LOG_DEBUG); + log_show_color(true); + log_parse_environment(); + + r = parse_argv(argc, argv); + if (r <= 0) + return r == 0 ? EXIT_SUCCESS : EXIT_FAILURE; + + if (remoteserver_init(&s) < 0) + return EXIT_FAILURE; + + log_debug("%s running as pid %lu", + program_invocation_short_name, (unsigned long) getpid()); + sd_notify(false, + "READY=1\n" + "STATUS=Processing requests..."); + + while (s.active) { + r = sd_event_get_state(s.events); + if (r < 0) + break; + if (r == SD_EVENT_FINISHED) + break; + + r = sd_event_run(s.events, -1); + if (r < 0) { + log_error("Failed to run event loop: %s", strerror(-r)); + break; + } + } + + log_info("Finishing after writing %" PRIu64 " entries", s.writer.seqnum); + r2 = server_destroy(&s); + + sd_notify(false, "STATUS=Shutting down..."); + + return r >= 0 && r2 >= 0 ? EXIT_SUCCESS : EXIT_FAILURE; +} |