1 /* SPDX-License-Identifier: LGPL-2.1-or-later */
12 #include "alloc-util.h"
13 #include "constants.h"
14 #include "errno-util.h"
17 #include "journal-file-util.h"
18 #include "journal-remote-write.h"
19 #include "journal-remote.h"
20 #include "journald-native.h"
22 #include "parse-util.h"
23 #include "parse-helpers.h"
24 #include "process-util.h"
25 #include "socket-util.h"
26 #include "stdio-util.h"
27 #include "string-util.h"
30 #define REMOTE_JOURNAL_PATH "/var/log/journal/remote"
32 #define filename_escape(s) xescape((s), "/ ")
35 MHDDaemonWrapper
*MHDDaemonWrapper_free(MHDDaemonWrapper
*d
) {
40 MHD_stop_daemon(d
->daemon
);
41 sd_event_source_unref(d
->io_event
);
42 sd_event_source_unref(d
->timer_event
);
48 static int open_output(RemoteServer
*s
, Writer
*w
, const char* host
) {
49 _cleanup_free_
char *_filename
= NULL
;
56 switch (s
->split_mode
) {
57 case JOURNAL_WRITE_SPLIT_NONE
:
61 case JOURNAL_WRITE_SPLIT_HOST
: {
62 _cleanup_free_
char *name
= NULL
;
66 name
= filename_escape(host
);
70 r
= asprintf(&_filename
, "%s/remote-%s.journal", s
->output
, name
);
82 r
= journal_file_open_reliably(
93 return log_error_errno(r
, "Failed to open output journal %s: %m", filename
);
95 log_debug("Opened output file %s", w
->journal
->path
);
99 /**********************************************************************
100 **********************************************************************
101 **********************************************************************/
103 static int init_writer_hashmap(RemoteServer
*s
) {
104 static const struct hash_ops
* const hash_ops
[] = {
105 [JOURNAL_WRITE_SPLIT_NONE
] = NULL
,
106 [JOURNAL_WRITE_SPLIT_HOST
] = &string_hash_ops
,
110 assert(s
->split_mode
>= 0 && s
->split_mode
< (int) ELEMENTSOF(hash_ops
));
112 s
->writers
= hashmap_new(hash_ops
[s
->split_mode
]);
119 int journal_remote_get_writer(RemoteServer
*s
, const char *host
, Writer
**writer
) {
120 _cleanup_(writer_unrefp
) Writer
*w
= NULL
;
127 switch (s
->split_mode
) {
128 case JOURNAL_WRITE_SPLIT_NONE
:
129 key
= "one and only";
132 case JOURNAL_WRITE_SPLIT_HOST
:
138 assert_not_reached();
141 w
= hashmap_get(s
->writers
, key
);
145 r
= writer_new(s
, &w
);
149 if (s
->split_mode
== JOURNAL_WRITE_SPLIT_HOST
) {
150 w
->hashmap_key
= strdup(key
);
155 r
= open_output(s
, w
, host
);
159 r
= hashmap_put(s
->writers
, w
->hashmap_key
?: key
, w
);
164 *writer
= TAKE_PTR(w
);
168 /**********************************************************************
169 **********************************************************************
170 **********************************************************************/
172 /* This should go away as soon as μhttpd allows state to be passed around. */
173 RemoteServer
*journal_remote_server_global
;
175 static int dispatch_raw_source_event(sd_event_source
*event
,
179 static int dispatch_raw_source_until_block(sd_event_source
*event
,
181 static int dispatch_blocking_source_event(sd_event_source
*event
,
183 static int dispatch_raw_connection_event(sd_event_source
*event
,
188 static int get_source_for_fd(RemoteServer
*s
,
189 int fd
, char *name
, RemoteSource
**source
) {
193 /* This takes ownership of name, but only on success. */
199 if (!GREEDY_REALLOC0(s
->sources
, fd
+ 1))
202 r
= journal_remote_get_writer(s
, name
, &writer
);
204 return log_warning_errno(r
, "Failed to get writer for source %s: %m",
207 if (!s
->sources
[fd
]) {
208 s
->sources
[fd
] = source_new(fd
, false, name
, writer
);
209 if (!s
->sources
[fd
]) {
210 writer_unref(writer
);
217 *source
= s
->sources
[fd
];
221 static int remove_source(RemoteServer
*s
, int fd
) {
222 RemoteSource
*source
;
225 assert(fd
>= 0 && fd
< (ssize_t
) MALLOC_ELEMENTSOF(s
->sources
));
227 source
= s
->sources
[fd
];
229 /* this closes fd too */
231 s
->sources
[fd
] = NULL
;
238 int journal_remote_add_source(RemoteServer
*s
, int fd
, char* name
, bool own_name
) {
239 RemoteSource
*source
= NULL
;
242 /* This takes ownership of name, even on failure, if own_name is true. */
254 r
= get_source_for_fd(s
, fd
, name
, &source
);
256 log_error_errno(r
, "Failed to create source for fd:%d (%s): %m",
262 r
= sd_event_add_io(s
->events
, &source
->event
,
263 fd
, EPOLLIN
|EPOLLRDHUP
|EPOLLPRI
,
264 dispatch_raw_source_event
, source
);
266 /* Add additional source for buffer processing. It will be
268 r
= sd_event_add_defer(s
->events
, &source
->buffer_event
,
269 dispatch_raw_source_until_block
, source
);
271 r
= sd_event_source_set_enabled(source
->buffer_event
, SD_EVENT_OFF
);
272 } else if (r
== -EPERM
) {
273 log_debug("Falling back to sd_event_add_defer for fd:%d (%s)", fd
, name
);
274 r
= sd_event_add_defer(s
->events
, &source
->event
,
275 dispatch_blocking_source_event
, source
);
277 r
= sd_event_source_set_enabled(source
->event
, SD_EVENT_ON
);
280 log_error_errno(r
, "Failed to register event source for fd:%d: %m",
285 r
= sd_event_source_set_description(source
->event
, name
);
287 log_error_errno(r
, "Failed to set source name for fd:%d: %m", fd
);
291 return 1; /* work to do */
294 remove_source(s
, fd
);
298 int journal_remote_add_raw_socket(RemoteServer
*s
, int fd
) {
299 _unused_ _cleanup_close_
int fd_
= fd
;
300 char name
[STRLEN("raw-socket-") + DECIMAL_STR_MAX(int) + 1];
306 r
= sd_event_add_io(s
->events
, &s
->listen_event
,
308 dispatch_raw_connection_event
, s
);
312 xsprintf(name
, "raw-socket-%d", fd
);
314 r
= sd_event_source_set_description(s
->listen_event
, name
);
323 /**********************************************************************
324 **********************************************************************
325 **********************************************************************/
327 int journal_remote_server_init(
330 JournalWriteSplitMode split_mode
,
331 JournalFileFlags file_flags
) {
337 assert(journal_remote_server_global
== NULL
);
338 journal_remote_server_global
= s
;
340 s
->split_mode
= split_mode
;
341 s
->file_flags
= file_flags
;
345 else if (split_mode
== JOURNAL_WRITE_SPLIT_NONE
)
346 s
->output
= REMOTE_JOURNAL_PATH
"/remote.journal";
347 else if (split_mode
== JOURNAL_WRITE_SPLIT_HOST
)
348 s
->output
= REMOTE_JOURNAL_PATH
;
350 assert_not_reached();
352 r
= sd_event_default(&s
->events
);
354 return log_error_errno(r
, "Failed to allocate event loop: %m");
356 r
= init_writer_hashmap(s
);
363 void journal_remote_server_destroy(RemoteServer
*s
) {
370 hashmap_free_with_destructor(s
->daemons
, MHDDaemonWrapper_free
);
373 for (i
= 0; i
< MALLOC_ELEMENTSOF(s
->sources
); i
++)
377 writer_unref(s
->_single_writer
);
378 hashmap_free(s
->writers
);
380 sd_event_source_unref(s
->sigterm_event
);
381 sd_event_source_unref(s
->sigint_event
);
382 sd_event_source_unref(s
->listen_event
);
383 sd_event_unref(s
->events
);
385 if (s
== journal_remote_server_global
)
386 journal_remote_server_global
= NULL
;
388 /* fds that we're listening on remain open... */
391 /**********************************************************************
392 **********************************************************************
393 **********************************************************************/
395 int journal_remote_handle_raw_source(
396 sd_event_source
*event
,
401 RemoteSource
*source
;
404 /* Returns 1 if there might be more data pending,
405 * 0 if data is currently exhausted, negative on error.
409 assert(fd
>= 0 && fd
< (ssize_t
) MALLOC_ELEMENTSOF(s
->sources
));
410 source
= s
->sources
[fd
];
411 assert(source
->importer
.fd
== fd
);
413 r
= process_source(source
, s
->file_flags
);
414 if (journal_importer_eof(&source
->importer
)) {
417 log_debug("EOF reached with source %s (fd=%d)",
418 source
->importer
.name
, source
->importer
.fd
);
420 remaining
= journal_importer_bytes_remaining(&source
->importer
);
422 log_notice("Premature EOF. %zu bytes lost.", remaining
);
423 remove_source(s
, source
->importer
.fd
);
424 log_debug("%zu active sources remaining", s
->active
);
426 } else if (r
== -E2BIG
) {
427 log_notice("Entry with too many fields, skipped");
429 } else if (r
== -ENOBUFS
) {
430 log_notice("Entry too big, skipped");
432 } else if (r
== -EAGAIN
) {
435 log_debug_errno(r
, "Closing connection: %m");
436 remove_source(s
, fd
);
442 static int dispatch_raw_source_until_block(sd_event_source
*event
,
444 RemoteSource
*source
= ASSERT_PTR(userdata
);
449 /* Make sure event stays around even if source is destroyed */
450 sd_event_source_ref(event
);
452 r
= journal_remote_handle_raw_source(event
, source
->importer
.fd
, EPOLLIN
, journal_remote_server_global
);
456 /* No more data for now */
457 k
= sd_event_source_set_enabled(event
, SD_EVENT_OFF
);
462 sd_event_source_unref(event
);
467 static int dispatch_raw_source_event(sd_event_source
*event
,
471 RemoteSource
*source
= ASSERT_PTR(userdata
);
474 assert(source
->event
);
475 assert(source
->buffer_event
);
477 r
= journal_remote_handle_raw_source(event
, fd
, EPOLLIN
, journal_remote_server_global
);
481 /* Might have more data. We need to rerun the handler
482 * until we are sure the buffer is exhausted. */
483 k
= sd_event_source_set_enabled(source
->buffer_event
, SD_EVENT_ON
);
491 static int dispatch_blocking_source_event(sd_event_source
*event
,
493 RemoteSource
*source
= ASSERT_PTR(userdata
);
495 return journal_remote_handle_raw_source(event
, source
->importer
.fd
, EPOLLIN
, journal_remote_server_global
);
498 static int accept_connection(
504 _cleanup_close_
int fd2
= -EBADF
;
510 log_debug("Accepting new %s connection on fd:%d", type
, fd
);
511 fd2
= accept4(fd
, &addr
->sockaddr
.sa
, &addr
->size
, SOCK_NONBLOCK
|SOCK_CLOEXEC
);
513 if (ERRNO_IS_ACCEPT_AGAIN(errno
))
516 return log_error_errno(errno
, "accept() on fd:%d failed: %m", fd
);
519 switch (socket_address_family(addr
)) {
522 _cleanup_free_
char *a
= NULL
;
525 r
= socket_address_print(addr
, &a
);
527 return log_error_errno(r
, "socket_address_print(): %m");
529 r
= socknameinfo_pretty(&addr
->sockaddr
, addr
->size
, &b
);
531 return log_error_errno(r
, "Resolving hostname failed: %m");
533 log_debug("Accepted %s %s connection from %s",
535 af_to_ipv4_ipv6(socket_address_family(addr
)),
543 return log_error_errno(SYNTHETIC_ERRNO(EINVAL
),
544 "Rejected %s connection with unsupported family %d",
545 type
, socket_address_family(addr
));
549 static int dispatch_raw_connection_event(
550 sd_event_source
*event
,
555 RemoteServer
*s
= ASSERT_PTR(userdata
);
557 SocketAddress addr
= {
558 .size
= sizeof(union sockaddr_union
),
561 char *hostname
= NULL
;
563 fd2
= accept_connection("raw", fd
, &addr
, &hostname
);
569 return journal_remote_add_source(s
, fd2
, hostname
, true);