src/journal/journald-stream.c

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