picnoir
/
Systemd
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

journal-remote: tool to receive messages over the network

This commit is contained in:
Zbigniew Jędrzejewski-Szmek 2012-11-01 23:08:03 +01:00
parent f12be7e8ca
commit fdfccdbc98
9 changed files with 1693 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Makefile-man.am
View File

@ -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 \

14
Makefile.am
View File

@ -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

289
man/systemd-journal-remote.xml Normal file
View File

@ -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>

1
src/journal/journal-gatewayd.c
View File

@ -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;

415
src/journal/journal-remote-parse.c Normal file
View File

@ -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;
}

60
src/journal/journal-remote-parse.h Normal file
View File

@ -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);

124
src/journal/journal-remote-write.c Normal file
View File

@ -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;
}

51
src/journal/journal-remote-write.h Normal file
View File

@ -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);

738
src/journal/journal-remote.c Normal file
View File

@ -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;
}