src/journal/journald-stream.c

   1 /* SPDX-License-Identifier: LGPL-2.1+ */
   2 /***
   3   This file is part of systemd.
   4
   5   Copyright 2011 Lennart Poettering
   6
   7   systemd is free software; you can redistribute it and/or modify it
   8   under the terms of the GNU Lesser General Public License as published by
   9   the Free Software Foundation; either version 2.1 of the License, or
  10   (at your option) any later version.
  11
  12   systemd is distributed in the hope that it will be useful, but
  13   WITHOUT ANY WARRANTY; without even the implied warranty of
  14   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  15   Lesser General Public License for more details.
  16
  17   You should have received a copy of the GNU Lesser General Public License
  18   along with systemd; If not, see <http://www.gnu.org/licenses/>.
  19 ***/
  20
  21 #include <stddef.h>
  22 #include <unistd.h>
  23
  24 #if HAVE_SELINUX
  25 #include <selinux/selinux.h>
  26 #endif
  27
  28 #include "sd-daemon.h"
  29 #include "sd-event.h"
  30
  31 #include "alloc-util.h"
  32 #include "dirent-util.h"
  33 #include "escape.h"
  34 #include "fd-util.h"
  35 #include "fileio.h"
  36 #include "io-util.h"
  37 #include "journald-console.h"
  38 #include "journald-context.h"
  39 #include "journald-kmsg.h"
  40 #include "journald-server.h"
  41 #include "journald-stream.h"
  42 #include "journald-syslog.h"
  43 #include "journald-wall.h"
  44 #include "mkdir.h"
  45 #include "parse-util.h"
  46 #include "process-util.h"
  47 #include "selinux-util.h"
  48 #include "socket-util.h"
  49 #include "stdio-util.h"
  50 #include "string-util.h"
  51 #include "syslog-util.h"
  52 #include "unit-name.h"
  53
  54 #define STDOUT_STREAMS_MAX 4096
  55
  56 typedef enum StdoutStreamState {
  57         STDOUT_STREAM_IDENTIFIER,
  58         STDOUT_STREAM_UNIT_ID,
  59         STDOUT_STREAM_PRIORITY,
  60         STDOUT_STREAM_LEVEL_PREFIX,
  61         STDOUT_STREAM_FORWARD_TO_SYSLOG,
  62         STDOUT_STREAM_FORWARD_TO_KMSG,
  63         STDOUT_STREAM_FORWARD_TO_CONSOLE,
  64         STDOUT_STREAM_RUNNING
  65 } StdoutStreamState;
  66
  67 /* The different types of log record terminators: a real \n was read, a NUL character was read, the maximum line length
  68  * was reached, or the end of the stream was reached */
  69
  70 typedef enum LineBreak {
  71         LINE_BREAK_NEWLINE,
  72         LINE_BREAK_NUL,
  73         LINE_BREAK_LINE_MAX,
  74         LINE_BREAK_EOF,
  75 } LineBreak;
  76
  77 struct StdoutStream {
  78         Server *server;
  79         StdoutStreamState state;
  80
  81         int fd;
  82
  83         struct ucred ucred;
  84         char *label;
  85         char *identifier;
  86         char *unit_id;
  87         int priority;
  88         bool level_prefix:1;
  89         bool forward_to_syslog:1;
  90         bool forward_to_kmsg:1;
  91         bool forward_to_console:1;
  92
  93         bool fdstore:1;
  94         bool in_notify_queue:1;
  95
  96         char *buffer;
  97         size_t length;
  98         size_t allocated;
  99
 100         sd_event_source *event_source;
 101
 102         char *state_file;
 103
 104         ClientContext *context;
 105
 106         LIST_FIELDS(StdoutStream, stdout_stream);
 107         LIST_FIELDS(StdoutStream, stdout_stream_notify_queue);
 108
 109         char id_field[sizeof("_STREAM_ID=")-1 + SD_ID128_STRING_MAX];
 110 };
 111
 112 void stdout_stream_free(StdoutStream *s) {
 113         if (!s)
 114                 return;
 115
 116         if (s->server) {
 117
 118                 if (s->context)
 119                         client_context_release(s->server, s->context);
 120
 121                 assert(s->server->n_stdout_streams > 0);
 122                 s->server->n_stdout_streams--;
 123                 LIST_REMOVE(stdout_stream, s->server->stdout_streams, s);
 124
 125                 if (s->in_notify_queue)
 126                         LIST_REMOVE(stdout_stream_notify_queue, s->server->stdout_streams_notify_queue, s);
 127         }
 128
 129         if (s->event_source) {
 130                 sd_event_source_set_enabled(s->event_source, SD_EVENT_OFF);
 131                 s->event_source = sd_event_source_unref(s->event_source);
 132         }
 133
 134         safe_close(s->fd);
 135         free(s->label);
 136         free(s->identifier);
 137         free(s->unit_id);
 138         free(s->state_file);
 139         free(s->buffer);
 140
 141         free(s);
 142 }
 143
 144 DEFINE_TRIVIAL_CLEANUP_FUNC(StdoutStream*, stdout_stream_free);
 145
 146 static void stdout_stream_destroy(StdoutStream *s) {
 147         if (!s)
 148                 return;
 149
 150         if (s->state_file)
 151                 (void) unlink(s->state_file);
 152
 153         stdout_stream_free(s);
 154 }
 155
 156 static int stdout_stream_save(StdoutStream *s) {
 157         _cleanup_free_ char *temp_path = NULL;
 158         _cleanup_fclose_ FILE *f = NULL;
 159         int r;
 160
 161         assert(s);
 162
 163         if (s->state != STDOUT_STREAM_RUNNING)
 164                 return 0;
 165
 166         if (!s->state_file) {
 167                 struct stat st;
 168
 169                 r = fstat(s->fd, &st);
 170                 if (r < 0)
 171                         return log_warning_errno(errno, "Failed to stat connected stream: %m");
 172
 173                 /* We use device and inode numbers as identifier for the stream */
 174                 if (asprintf(&s->state_file, "/run/systemd/journal/streams/%lu:%lu", (unsigned long) st.st_dev, (unsigned long) st.st_ino) < 0)
 175                         return log_oom();
 176         }
 177
 178         mkdir_p("/run/systemd/journal/streams", 0755);
 179
 180         r = fopen_temporary(s->state_file, &f, &temp_path);
 181         if (r < 0)
 182                 goto fail;
 183
 184         fprintf(f,
 185                 "# This is private data. Do not parse\n"
 186                 "PRIORITY=%i\n"
 187                 "LEVEL_PREFIX=%i\n"
 188                 "FORWARD_TO_SYSLOG=%i\n"
 189                 "FORWARD_TO_KMSG=%i\n"
 190                 "FORWARD_TO_CONSOLE=%i\n"
 191                 "STREAM_ID=%s\n",
 192                 s->priority,
 193                 s->level_prefix,
 194                 s->forward_to_syslog,
 195                 s->forward_to_kmsg,
 196                 s->forward_to_console,
 197                 s->id_field + strlen("_STREAM_ID="));
 198
 199         if (!isempty(s->identifier)) {
 200                 _cleanup_free_ char *escaped;
 201
 202                 escaped = cescape(s->identifier);
 203                 if (!escaped) {
 204                         r = -ENOMEM;
 205                         goto fail;
 206                 }
 207
 208                 fprintf(f, "IDENTIFIER=%s\n", escaped);
 209         }
 210
 211         if (!isempty(s->unit_id)) {
 212                 _cleanup_free_ char *escaped;
 213
 214                 escaped = cescape(s->unit_id);
 215                 if (!escaped) {
 216                         r = -ENOMEM;
 217                         goto fail;
 218                 }
 219
 220                 fprintf(f, "UNIT=%s\n", escaped);
 221         }
 222
 223         r = fflush_and_check(f);
 224         if (r < 0)
 225                 goto fail;
 226
 227         if (rename(temp_path, s->state_file) < 0) {
 228                 r = -errno;
 229                 goto fail;
 230         }
 231
 232         if (!s->fdstore && !s->in_notify_queue) {
 233                 LIST_PREPEND(stdout_stream_notify_queue, s->server->stdout_streams_notify_queue, s);
 234                 s->in_notify_queue = true;
 235
 236                 if (s->server->notify_event_source) {
 237                         r = sd_event_source_set_enabled(s->server->notify_event_source, SD_EVENT_ON);
 238                         if (r < 0)
 239                                 log_warning_errno(r, "Failed to enable notify event source: %m");
 240                 }
 241         }
 242
 243         return 0;
 244
 245 fail:
 246         (void) unlink(s->state_file);
 247
 248         if (temp_path)
 249                 (void) unlink(temp_path);
 250
 251         return log_error_errno(r, "Failed to save stream data %s: %m", s->state_file);
 252 }
 253
 254 static int stdout_stream_log(StdoutStream *s, const char *p, LineBreak line_break) {
 255         struct iovec *iovec;
 256         int priority;
 257         char syslog_priority[] = "PRIORITY=\0";
 258         char syslog_facility[sizeof("SYSLOG_FACILITY=")-1 + DECIMAL_STR_MAX(int) + 1];
 259         _cleanup_free_ char *message = NULL, *syslog_identifier = NULL;
 260         size_t n = 0, m;
 261         int r;
 262
 263         assert(s);
 264         assert(p);
 265
 266         if (s->context)
 267                 (void) client_context_maybe_refresh(s->server, s->context, NULL, NULL, 0, NULL, USEC_INFINITY);
 268         else if (pid_is_valid(s->ucred.pid)) {
 269                 r = client_context_acquire(s->server, s->ucred.pid, &s->ucred, s->label, strlen_ptr(s->label), s->unit_id, &s->context);
 270                 if (r < 0)
 271                         log_warning_errno(r, "Failed to acquire client context, ignoring: %m");
 272         }
 273
 274         priority = s->priority;
 275
 276         if (s->level_prefix)
 277                 syslog_parse_priority(&p, &priority, false);
 278
 279         if (!client_context_test_priority(s->context, priority))
 280                 return 0;
 281
 282         if (isempty(p))
 283                 return 0;
 284
 285         if (s->forward_to_syslog || s->server->forward_to_syslog)
 286                 server_forward_syslog(s->server, syslog_fixup_facility(priority), s->identifier, p, &s->ucred, NULL);
 287
 288         if (s->forward_to_kmsg || s->server->forward_to_kmsg)
 289                 server_forward_kmsg(s->server, priority, s->identifier, p, &s->ucred);
 290
 291         if (s->forward_to_console || s->server->forward_to_console)
 292                 server_forward_console(s->server, priority, s->identifier, p, &s->ucred);
 293
 294         if (s->server->forward_to_wall)
 295                 server_forward_wall(s->server, priority, s->identifier, p, &s->ucred);
 296
 297         m = N_IOVEC_META_FIELDS + 7 + client_context_extra_fields_n_iovec(s->context);
 298         iovec = newa(struct iovec, m);
 299
 300         iovec[n++] = IOVEC_MAKE_STRING("_TRANSPORT=stdout");
 301         iovec[n++] = IOVEC_MAKE_STRING(s->id_field);
 302
 303         syslog_priority[strlen("PRIORITY=")] = '0' + LOG_PRI(priority);
 304         iovec[n++] = IOVEC_MAKE_STRING(syslog_priority);
 305
 306         if (priority & LOG_FACMASK) {
 307                 xsprintf(syslog_facility, "SYSLOG_FACILITY=%i", LOG_FAC(priority));
 308                 iovec[n++] = IOVEC_MAKE_STRING(syslog_facility);
 309         }
 310
 311         if (s->identifier) {
 312                 syslog_identifier = strappend("SYSLOG_IDENTIFIER=", s->identifier);
 313                 if (syslog_identifier)
 314                         iovec[n++] = IOVEC_MAKE_STRING(syslog_identifier);
 315         }
 316
 317         if (line_break != LINE_BREAK_NEWLINE) {
 318                 const char *c;
 319
 320                 /* If this log message was generated due to an uncommon line break then mention this in the log
 321                  * entry */
 322
 323                 c =     line_break == LINE_BREAK_NUL ?      "_LINE_BREAK=nul" :
 324                         line_break == LINE_BREAK_LINE_MAX ? "_LINE_BREAK=line-max" :
 325                                                             "_LINE_BREAK=eof";
 326                 iovec[n++] = IOVEC_MAKE_STRING(c);
 327         }
 328
 329         message = strappend("MESSAGE=", p);
 330         if (message)
 331                 iovec[n++] = IOVEC_MAKE_STRING(message);
 332
 333         server_dispatch_message(s->server, iovec, n, m, s->context, NULL, priority, 0);
 334         return 0;
 335 }
 336
 337 static int stdout_stream_line(StdoutStream *s, char *p, LineBreak line_break) {
 338         int r;
 339         char *orig;
 340
 341         assert(s);
 342         assert(p);
 343
 344         orig = p;
 345         p = strstrip(p);
 346
 347         /* line breaks by NUL, line max length or EOF are not permissible during the negotiation part of the protocol */
 348         if (line_break != LINE_BREAK_NEWLINE && s->state != STDOUT_STREAM_RUNNING) {
 349                 log_warning("Control protocol line not properly terminated.");
 350                 return -EINVAL;
 351         }
 352
 353         switch (s->state) {
 354
 355         case STDOUT_STREAM_IDENTIFIER:
 356                 if (!isempty(p)) {
 357                         s->identifier = strdup(p);
 358                         if (!s->identifier)
 359                                 return log_oom();
 360                 }
 361
 362                 s->state = STDOUT_STREAM_UNIT_ID;
 363                 return 0;
 364
 365         case STDOUT_STREAM_UNIT_ID:
 366                 if (s->ucred.uid == 0 &&
 367                     unit_name_is_valid(p, UNIT_NAME_PLAIN|UNIT_NAME_INSTANCE)) {
 368
 369                         s->unit_id = strdup(p);
 370                         if (!s->unit_id)
 371                                 return log_oom();
 372                 }
 373
 374                 s->state = STDOUT_STREAM_PRIORITY;
 375                 return 0;
 376
 377         case STDOUT_STREAM_PRIORITY:
 378                 r = safe_atoi(p, &s->priority);
 379                 if (r < 0 || s->priority < 0 || s->priority > 999) {
 380                         log_warning("Failed to parse log priority line.");
 381                         return -EINVAL;
 382                 }
 383
 384                 s->state = STDOUT_STREAM_LEVEL_PREFIX;
 385                 return 0;
 386
 387         case STDOUT_STREAM_LEVEL_PREFIX:
 388                 r = parse_boolean(p);
 389                 if (r < 0) {
 390                         log_warning("Failed to parse level prefix line.");
 391                         return -EINVAL;
 392                 }
 393
 394                 s->level_prefix = !!r;
 395                 s->state = STDOUT_STREAM_FORWARD_TO_SYSLOG;
 396                 return 0;
 397
 398         case STDOUT_STREAM_FORWARD_TO_SYSLOG:
 399                 r = parse_boolean(p);
 400                 if (r < 0) {
 401                         log_warning("Failed to parse forward to syslog line.");
 402                         return -EINVAL;
 403                 }
 404
 405                 s->forward_to_syslog = !!r;
 406                 s->state = STDOUT_STREAM_FORWARD_TO_KMSG;
 407                 return 0;
 408
 409         case STDOUT_STREAM_FORWARD_TO_KMSG:
 410                 r = parse_boolean(p);
 411                 if (r < 0) {
 412                         log_warning("Failed to parse copy to kmsg line.");
 413                         return -EINVAL;
 414                 }
 415
 416                 s->forward_to_kmsg = !!r;
 417                 s->state = STDOUT_STREAM_FORWARD_TO_CONSOLE;
 418                 return 0;
 419
 420         case STDOUT_STREAM_FORWARD_TO_CONSOLE:
 421                 r = parse_boolean(p);
 422                 if (r < 0) {
 423                         log_warning("Failed to parse copy to console line.");
 424                         return -EINVAL;
 425                 }
 426
 427                 s->forward_to_console = !!r;
 428                 s->state = STDOUT_STREAM_RUNNING;
 429
 430                 /* Try to save the stream, so that journald can be restarted and we can recover */
 431                 (void) stdout_stream_save(s);
 432                 return 0;
 433
 434         case STDOUT_STREAM_RUNNING:
 435                 return stdout_stream_log(s, orig, line_break);
 436         }
 437
 438         assert_not_reached("Unknown stream state");
 439 }
 440
 441 static int stdout_stream_scan(StdoutStream *s, bool force_flush) {
 442         char *p;
 443         size_t remaining;
 444         int r;
 445
 446         assert(s);
 447
 448         p = s->buffer;
 449         remaining = s->length;
 450
 451         /* XXX: This function does nothing if (s->length == 0) */
 452
 453         for (;;) {
 454                 LineBreak line_break;
 455                 size_t skip;
 456                 char *end1, *end2;
 457
 458                 end1 = memchr(p, '\n', remaining);
 459                 end2 = memchr(p, 0, end1 ? (size_t) (end1 - p) : remaining);
 460
 461                 if (end2) {
 462                         /* We found a NUL terminator */
 463                         skip = end2 - p + 1;
 464                         line_break = LINE_BREAK_NUL;
 465                 } else if (end1) {
 466                         /* We found a \n terminator */
 467                         *end1 = 0;
 468                         skip = end1 - p + 1;
 469                         line_break = LINE_BREAK_NEWLINE;
 470                 } else if (remaining >= s->server->line_max) {
 471                         /* Force a line break after the maximum line length */
 472                         *(p + s->server->line_max) = 0;
 473                         skip = remaining;
 474                         line_break = LINE_BREAK_LINE_MAX;
 475                 } else
 476                         break;
 477
 478                 r = stdout_stream_line(s, p, line_break);
 479                 if (r < 0)
 480                         return r;
 481
 482                 remaining -= skip;
 483                 p += skip;
 484         }
 485
 486         if (force_flush && remaining > 0) {
 487                 p[remaining] = 0;
 488                 r = stdout_stream_line(s, p, LINE_BREAK_EOF);
 489                 if (r < 0)
 490                         return r;
 491
 492                 p += remaining;
 493                 remaining = 0;
 494         }
 495
 496         if (p > s->buffer) {
 497                 memmove(s->buffer, p, remaining);
 498                 s->length = remaining;
 499         }
 500
 501         return 0;
 502 }
 503
 504 static int stdout_stream_process(sd_event_source *es, int fd, uint32_t revents, void *userdata) {
 505         StdoutStream *s = userdata;
 506         size_t limit;
 507         ssize_t l;
 508         int r;
 509
 510         assert(s);
 511
 512         if ((revents|EPOLLIN|EPOLLHUP) != (EPOLLIN|EPOLLHUP)) {
 513                 log_error("Got invalid event from epoll for stdout stream: %"PRIx32, revents);
 514                 goto terminate;
 515         }
 516
 517         /* If the buffer is full already (discounting the extra NUL we need), add room for another 1K */
 518         if (s->length + 1 >= s->allocated) {
 519                 if (!GREEDY_REALLOC(s->buffer, s->allocated, s->length + 1 + 1024)) {
 520                         log_oom();
 521                         goto terminate;
 522                 }
 523         }
 524
 525         /* Try to make use of the allocated buffer in full, but never read more than the configured line size. Also,
 526          * always leave room for a terminating NUL we might need to add. */
 527         limit = MIN(s->allocated - 1, s->server->line_max);
 528
 529         l = read(s->fd, s->buffer + s->length, limit - s->length);
 530         if (l < 0) {
 531                 if (errno == EAGAIN)
 532                         return 0;
 533
 534                 log_warning_errno(errno, "Failed to read from stream: %m");
 535                 goto terminate;
 536         }
 537
 538         if (l == 0) {
 539                 stdout_stream_scan(s, true);
 540                 goto terminate;
 541         }
 542
 543         s->length += l;
 544         r = stdout_stream_scan(s, false);
 545         if (r < 0)
 546                 goto terminate;
 547
 548         return 1;
 549
 550 terminate:
 551         stdout_stream_destroy(s);
 552         return 0;
 553 }
 554
 555 static int stdout_stream_install(Server *s, int fd, StdoutStream **ret) {
 556         _cleanup_(stdout_stream_freep) StdoutStream *stream = NULL;
 557         sd_id128_t id;
 558         int r;
 559
 560         assert(s);
 561         assert(fd >= 0);
 562
 563         r = sd_id128_randomize(&id);
 564         if (r < 0)
 565                 return log_error_errno(r, "Failed to generate stream ID: %m");
 566
 567         stream = new0(StdoutStream, 1);
 568         if (!stream)
 569                 return log_oom();
 570
 571         stream->fd = -1;
 572         stream->priority = LOG_INFO;
 573
 574         xsprintf(stream->id_field, "_STREAM_ID=" SD_ID128_FORMAT_STR, SD_ID128_FORMAT_VAL(id));
 575
 576         r = getpeercred(fd, &stream->ucred);
 577         if (r < 0)
 578                 return log_error_errno(r, "Failed to determine peer credentials: %m");
 579
 580         if (mac_selinux_use()) {
 581                 r = getpeersec(fd, &stream->label);
 582                 if (r < 0 && r != -EOPNOTSUPP)
 583                         (void) log_warning_errno(r, "Failed to determine peer security context: %m");
 584         }
 585
 586         (void) shutdown(fd, SHUT_WR);
 587
 588         r = sd_event_add_io(s->event, &stream->event_source, fd, EPOLLIN, stdout_stream_process, stream);
 589         if (r < 0)
 590                 return log_error_errno(r, "Failed to add stream to event loop: %m");
 591
 592         r = sd_event_source_set_priority(stream->event_source, SD_EVENT_PRIORITY_NORMAL+5);
 593         if (r < 0)
 594                 return log_error_errno(r, "Failed to adjust stdout event source priority: %m");
 595
 596         stream->fd = fd;
 597
 598         stream->server = s;
 599         LIST_PREPEND(stdout_stream, s->stdout_streams, stream);
 600         s->n_stdout_streams++;
 601
 602         if (ret)
 603                 *ret = stream;
 604
 605         stream = NULL;
 606
 607         return 0;
 608 }
 609
 610 static int stdout_stream_new(sd_event_source *es, int listen_fd, uint32_t revents, void *userdata) {
 611         _cleanup_close_ int fd = -1;
 612         Server *s = userdata;
 613         int r;
 614
 615         assert(s);
 616
 617         if (revents != EPOLLIN) {
 618                 log_error("Got invalid event from epoll for stdout server fd: %"PRIx32, revents);
 619                 return -EIO;
 620         }
 621
 622         fd = accept4(s->stdout_fd, NULL, NULL, SOCK_NONBLOCK|SOCK_CLOEXEC);
 623         if (fd < 0) {
 624                 if (errno == EAGAIN)
 625                         return 0;
 626
 627                 return log_error_errno(errno, "Failed to accept stdout connection: %m");
 628         }
 629
 630         if (s->n_stdout_streams >= STDOUT_STREAMS_MAX) {
 631                 log_warning("Too many stdout streams, refusing connection.");
 632                 return 0;
 633         }
 634
 635         r = stdout_stream_install(s, fd, NULL);
 636         if (r < 0)
 637                 return r;
 638
 639         fd = -1;
 640         return 0;
 641 }
 642
 643 static int stdout_stream_load(StdoutStream *stream, const char *fname) {
 644         _cleanup_free_ char
 645                 *priority = NULL,
 646                 *level_prefix = NULL,
 647                 *forward_to_syslog = NULL,
 648                 *forward_to_kmsg = NULL,
 649                 *forward_to_console = NULL,
 650                 *stream_id = NULL;
 651         int r;
 652
 653         assert(stream);
 654         assert(fname);
 655
 656         if (!stream->state_file) {
 657                 stream->state_file = strappend("/run/systemd/journal/streams/", fname);
 658                 if (!stream->state_file)
 659                         return log_oom();
 660         }
 661
 662         r = parse_env_file(stream->state_file, NEWLINE,
 663                            "PRIORITY", &priority,
 664                            "LEVEL_PREFIX", &level_prefix,
 665                            "FORWARD_TO_SYSLOG", &forward_to_syslog,
 666                            "FORWARD_TO_KMSG", &forward_to_kmsg,
 667                            "FORWARD_TO_CONSOLE", &forward_to_console,
 668                            "IDENTIFIER", &stream->identifier,
 669                            "UNIT", &stream->unit_id,
 670                            "STREAM_ID", &stream_id,
 671                            NULL);
 672         if (r < 0)
 673                 return log_error_errno(r, "Failed to read: %s", stream->state_file);
 674
 675         if (priority) {
 676                 int p;
 677
 678                 p = log_level_from_string(priority);
 679                 if (p >= 0)
 680                         stream->priority = p;
 681         }
 682
 683         if (level_prefix) {
 684                 r = parse_boolean(level_prefix);
 685                 if (r >= 0)
 686                         stream->level_prefix = r;
 687         }
 688
 689         if (forward_to_syslog) {
 690                 r = parse_boolean(forward_to_syslog);
 691                 if (r >= 0)
 692                         stream->forward_to_syslog = r;
 693         }
 694
 695         if (forward_to_kmsg) {
 696                 r = parse_boolean(forward_to_kmsg);
 697                 if (r >= 0)
 698                         stream->forward_to_kmsg = r;
 699         }
 700
 701         if (forward_to_console) {
 702                 r = parse_boolean(forward_to_console);
 703                 if (r >= 0)
 704                         stream->forward_to_console = r;
 705         }
 706
 707         if (stream_id) {
 708                 sd_id128_t id;
 709
 710                 r = sd_id128_from_string(stream_id, &id);
 711                 if (r >= 0)
 712                         xsprintf(stream->id_field, "_STREAM_ID=" SD_ID128_FORMAT_STR, SD_ID128_FORMAT_VAL(id));
 713         }
 714
 715         return 0;
 716 }
 717
 718 static int stdout_stream_restore(Server *s, const char *fname, int fd) {
 719         StdoutStream *stream;
 720         int r;
 721
 722         assert(s);
 723         assert(fname);
 724         assert(fd >= 0);
 725
 726         if (s->n_stdout_streams >= STDOUT_STREAMS_MAX) {
 727                 log_warning("Too many stdout streams, refusing restoring of stream.");
 728                 return -ENOBUFS;
 729         }
 730
 731         r = stdout_stream_install(s, fd, &stream);
 732         if (r < 0)
 733                 return r;
 734
 735         stream->state = STDOUT_STREAM_RUNNING;
 736         stream->fdstore = true;
 737
 738         /* Ignore all parsing errors */
 739         (void) stdout_stream_load(stream, fname);
 740
 741         return 0;
 742 }
 743
 744 int server_restore_streams(Server *s, FDSet *fds) {
 745         _cleanup_closedir_ DIR *d = NULL;
 746         struct dirent *de;
 747         int r;
 748
 749         d = opendir("/run/systemd/journal/streams");
 750         if (!d) {
 751                 if (errno == ENOENT)
 752                         return 0;
 753
 754                 return log_warning_errno(errno, "Failed to enumerate /run/systemd/journal/streams: %m");
 755         }
 756
 757         FOREACH_DIRENT(de, d, goto fail) {
 758                 unsigned long st_dev, st_ino;
 759                 bool found = false;
 760                 Iterator i;
 761                 int fd;
 762
 763                 if (sscanf(de->d_name, "%lu:%lu", &st_dev, &st_ino) != 2)
 764                         continue;
 765
 766                 FDSET_FOREACH(fd, fds, i) {
 767                         struct stat st;
 768
 769                         if (fstat(fd, &st) < 0)
 770                                 return log_error_errno(errno, "Failed to stat %s: %m", de->d_name);
 771
 772                         if (S_ISSOCK(st.st_mode) && st.st_dev == st_dev && st.st_ino == st_ino) {
 773                                 found = true;
 774                                 break;
 775                         }
 776                 }
 777
 778                 if (!found) {
 779                         /* No file descriptor? Then let's delete the state file */
 780                         log_debug("Cannot restore stream file %s", de->d_name);
 781                         unlinkat(dirfd(d), de->d_name, 0);
 782                         continue;
 783                 }
 784
 785                 fdset_remove(fds, fd);
 786
 787                 r = stdout_stream_restore(s, de->d_name, fd);
 788                 if (r < 0)
 789                         safe_close(fd);
 790         }
 791
 792         return 0;
 793
 794 fail:
 795         return log_error_errno(errno, "Failed to read streams directory: %m");
 796 }
 797
 798 int server_open_stdout_socket(Server *s) {
 799         static const union sockaddr_union sa = {
 800                 .un.sun_family = AF_UNIX,
 801                 .un.sun_path = "/run/systemd/journal/stdout",
 802         };
 803         int r;
 804
 805         assert(s);
 806
 807         if (s->stdout_fd < 0) {
 808                 s->stdout_fd = socket(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC|SOCK_NONBLOCK, 0);
 809                 if (s->stdout_fd < 0)
 810                         return log_error_errno(errno, "socket() failed: %m");
 811
 812                 (void) unlink(sa.un.sun_path);
 813
 814                 r = bind(s->stdout_fd, &sa.sa, SOCKADDR_UN_LEN(sa.un));
 815                 if (r < 0)
 816                         return log_error_errno(errno, "bind(%s) failed: %m", sa.un.sun_path);
 817
 818                 (void) chmod(sa.un.sun_path, 0666);
 819
 820                 if (listen(s->stdout_fd, SOMAXCONN) < 0)
 821                         return log_error_errno(errno, "listen(%s) failed: %m", sa.un.sun_path);
 822         } else
 823                 fd_nonblock(s->stdout_fd, 1);
 824
 825         r = sd_event_add_io(s->event, &s->stdout_event_source, s->stdout_fd, EPOLLIN, stdout_stream_new, s);
 826         if (r < 0)
 827                 return log_error_errno(r, "Failed to add stdout server fd to event source: %m");
 828
 829         r = sd_event_source_set_priority(s->stdout_event_source, SD_EVENT_PRIORITY_NORMAL+5);
 830         if (r < 0)
 831                 return log_error_errno(r, "Failed to adjust priority of stdout server event source: %m");
 832
 833         return 0;
 834 }
 835
 836 void stdout_stream_send_notify(StdoutStream *s) {
 837         struct iovec iovec = {
 838                 .iov_base = (char*) "FDSTORE=1",
 839                 .iov_len = strlen("FDSTORE=1"),
 840         };
 841         struct msghdr msghdr = {
 842                 .msg_iov = &iovec,
 843                 .msg_iovlen = 1,
 844         };
 845         struct cmsghdr *cmsg;
 846         ssize_t l;
 847
 848         assert(s);
 849         assert(!s->fdstore);
 850         assert(s->in_notify_queue);
 851         assert(s->server);
 852         assert(s->server->notify_fd >= 0);
 853
 854         /* Store the connection fd in PID 1, so that we get it passed
 855          * in again on next start */
 856
 857         msghdr.msg_controllen = CMSG_SPACE(sizeof(int));
 858         msghdr.msg_control = alloca0(msghdr.msg_controllen);
 859
 860         cmsg = CMSG_FIRSTHDR(&msghdr);
 861         cmsg->cmsg_level = SOL_SOCKET;
 862         cmsg->cmsg_type = SCM_RIGHTS;
 863         cmsg->cmsg_len = CMSG_LEN(sizeof(int));
 864
 865         memcpy(CMSG_DATA(cmsg), &s->fd, sizeof(int));
 866
 867         l = sendmsg(s->server->notify_fd, &msghdr, MSG_DONTWAIT|MSG_NOSIGNAL);
 868         if (l < 0) {
 869                 if (errno == EAGAIN)
 870                         return;
 871
 872                 log_error_errno(errno, "Failed to send stream file descriptor to service manager: %m");
 873         } else {
 874                 log_debug("Successfully sent stream file descriptor to service manager.");
 875                 s->fdstore = 1;
 876         }
 877
 878         LIST_REMOVE(stdout_stream_notify_queue, s->server->stdout_streams_notify_queue, s);
 879         s->in_notify_queue = false;
 880
 881 }