]> git.ipfire.org Git - thirdparty/systemd.git/blob - src/journal/journald.c
projects / thirdparty / systemd.git / blob
? search:
summary | shortlog | log | commit | commitdiff | tree
blame | history | raw | HEAD
journal: move max_use into metrics structure
[thirdparty/systemd.git] / src / journal / journald.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 General Public License as published by
10 the Free Software Foundation; either version 2 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 General Public License for more details.
17
18 You should have received a copy of the GNU General Public License
19 along with systemd; If not, see <http://www.gnu.org/licenses/>.
20 ***/
21
22 #include <sys/epoll.h>
23 #include <sys/socket.h>
24 #include <errno.h>
25 #include <sys/signalfd.h>
26 #include <unistd.h>
27 #include <fcntl.h>
28 #include <sys/acl.h>
29 #include <acl/libacl.h>
30 #include <stddef.h>
31 #include <sys/ioctl.h>
32 #include <linux/sockios.h>
33 #include <sys/statvfs.h>
34
35 #include "hashmap.h"
36 #include "journal-file.h"
37 #include "sd-daemon.h"
38 #include "socket-util.h"
39 #include "acl-util.h"
40 #include "cgroup-util.h"
41 #include "list.h"
42 #include "journal-rate-limit.h"
43 #include "sd-journal.h"
44 #include "journal-internal.h"
45
46 #define USER_JOURNALS_MAX 1024
47 #define STDOUT_STREAMS_MAX 4096
48
49 #define DEFAULT_RATE_LIMIT_INTERVAL (10*USEC_PER_SEC)
50 #define DEFAULT_RATE_LIMIT_BURST 200
51
52 #define RECHECK_AVAILABLE_SPACE_USEC (30*USEC_PER_SEC)
53
54 #define RECHECK_VAR_AVAILABLE_USEC (30*USEC_PER_SEC)
55
56 #define SYSLOG_TIMEOUT_USEC (5*USEC_PER_SEC)
57
58 typedef struct StdoutStream StdoutStream;
59
60 typedef struct Server {
61 int epoll_fd;
62 int signal_fd;
63 int syslog_fd;
64 int native_fd;
65 int stdout_fd;
66
67 JournalFile *runtime_journal;
68 JournalFile *system_journal;
69 Hashmap *user_journals;
70
71 uint64_t seqnum;
72
73 char *buffer;
74 size_t buffer_size;
75
76 JournalRateLimit *rate_limit;
77
78 JournalMetrics metrics;
79 bool compress;
80
81 uint64_t cached_available_space;
82 usec_t cached_available_space_timestamp;
83
84 uint64_t var_available_timestamp;
85
86 LIST_HEAD(StdoutStream, stdout_streams);
87 unsigned n_stdout_streams;
88 } Server;
89
90 typedef enum StdoutStreamState {
91 STDOUT_STREAM_TAG,
92 STDOUT_STREAM_PRIORITY,
93 STDOUT_STREAM_PRIORITY_PREFIX,
94 STDOUT_STREAM_TEE_CONSOLE,
95 STDOUT_STREAM_RUNNING
96 } StdoutStreamState;
97
98 struct StdoutStream {
99 Server *server;
100 StdoutStreamState state;
101
102 int fd;
103
104 struct ucred ucred;
105
106 char *tag;
107 int priority;
108 bool priority_prefix:1;
109 bool tee_console:1;
110
111 char buffer[LINE_MAX+1];
112 size_t length;
113
114 LIST_FIELDS(StdoutStream, stdout_stream);
115 };
116
117 static int server_flush_to_var(Server *s);
118
119 static uint64_t available_space(Server *s) {
120 char ids[33];
121 sd_id128_t machine;
122 char *p;
123 const char *f;
124 struct statvfs ss;
125 uint64_t sum = 0, avail = 0, ss_avail = 0;
126 int r;
127 DIR *d;
128 usec_t ts = now(CLOCK_MONOTONIC);
129
130 if (s->cached_available_space_timestamp + RECHECK_AVAILABLE_SPACE_USEC > ts)
131 return s->cached_available_space;
132
133 r = sd_id128_get_machine(&machine);
134 if (r < 0)
135 return 0;
136
137 if (s->system_journal)
138 f = "/var/log/journal/";
139 else
140 f = "/run/log/journal/";
141
142 p = strappend(f, sd_id128_to_string(machine, ids));
143 if (!p)
144 return 0;
145
146 d = opendir(p);
147 free(p);
148
149 if (!d)
150 return 0;
151
152 if (fstatvfs(dirfd(d), &ss) < 0)
153 goto finish;
154
155 for (;;) {
156 struct stat st;
157 struct dirent buf, *de;
158 int k;
159
160 k = readdir_r(d, &buf, &de);
161 if (k != 0) {
162 r = -k;
163 goto finish;
164 }
165
166 if (!de)
167 break;
168
169 if (!dirent_is_file_with_suffix(de, ".journal"))
170 continue;
171
172 if (fstatat(dirfd(d), de->d_name, &st, AT_SYMLINK_NOFOLLOW) < 0)
173 continue;
174
175 sum += (uint64_t) st.st_blocks * (uint64_t) st.st_blksize;
176 }
177
178 avail = sum >= s->metrics.max_use ? 0 : s->metrics.max_use - sum;
179
180 ss_avail = ss.f_bsize * ss.f_bavail;
181
182 ss_avail = ss_avail < s->metrics.keep_free ? 0 : ss_avail - s->metrics.keep_free;
183
184 if (ss_avail < avail)
185 avail = ss_avail;
186
187 s->cached_available_space = avail;
188 s->cached_available_space_timestamp = ts;
189
190 finish:
191 closedir(d);
192
193 return avail;
194 }
195
196 static void fix_perms(JournalFile *f, uid_t uid) {
197 acl_t acl;
198 acl_entry_t entry;
199 acl_permset_t permset;
200 int r;
201
202 assert(f);
203
204 r = fchmod_and_fchown(f->fd, 0640, 0, 0);
205 if (r < 0)
206 log_warning("Failed to fix access mode/rights on %s, ignoring: %s", f->path, strerror(-r));
207
208 if (uid <= 0)
209 return;
210
211 acl = acl_get_fd(f->fd);
212 if (!acl) {
213 log_warning("Failed to read ACL on %s, ignoring: %m", f->path);
214 return;
215 }
216
217 r = acl_find_uid(acl, uid, &entry);
218 if (r <= 0) {
219
220 if (acl_create_entry(&acl, &entry) < 0 ||
221 acl_set_tag_type(entry, ACL_USER) < 0 ||
222 acl_set_qualifier(entry, &uid) < 0) {
223 log_warning("Failed to patch ACL on %s, ignoring: %m", f->path);
224 goto finish;
225 }
226 }
227
228 if (acl_get_permset(entry, &permset) < 0 ||
229 acl_add_perm(permset, ACL_READ) < 0 ||
230 acl_calc_mask(&acl) < 0) {
231 log_warning("Failed to patch ACL on %s, ignoring: %m", f->path);
232 goto finish;
233 }
234
235 if (acl_set_fd(f->fd, acl) < 0)
236 log_warning("Failed to set ACL on %s, ignoring: %m", f->path);
237
238 finish:
239 acl_free(acl);
240 }
241
242 static JournalFile* find_journal(Server *s, uid_t uid) {
243 char *p;
244 int r;
245 JournalFile *f;
246 char ids[33];
247 sd_id128_t machine;
248
249 assert(s);
250
251 /* We split up user logs only on /var, not on /run. If the
252 * runtime file is open, we write to it exclusively, in order
253 * to guarantee proper order as soon as we flush /run to
254 * /var and close the runtime file. */
255
256 if (s->runtime_journal)
257 return s->runtime_journal;
258
259 if (uid <= 0)
260 return s->system_journal;
261
262 r = sd_id128_get_machine(&machine);
263 if (r < 0)
264 return s->system_journal;
265
266 f = hashmap_get(s->user_journals, UINT32_TO_PTR(uid));
267 if (f)
268 return f;
269
270 if (asprintf(&p, "/var/log/journal/%s/user-%lu.journal", sd_id128_to_string(machine, ids), (unsigned long) uid) < 0)
271 return s->system_journal;
272
273 while (hashmap_size(s->user_journals) >= USER_JOURNALS_MAX) {
274 /* Too many open? Then let's close one */
275 f = hashmap_steal_first(s->user_journals);
276 assert(f);
277 journal_file_close(f);
278 }
279
280 r = journal_file_open(p, O_RDWR|O_CREAT, 0640, s->system_journal, &f);
281 free(p);
282
283 if (r < 0)
284 return s->system_journal;
285
286 fix_perms(f, uid);
287 f->metrics = s->metrics;
288 f->compress = s->compress;
289
290 r = hashmap_put(s->user_journals, UINT32_TO_PTR(uid), f);
291 if (r < 0) {
292 journal_file_close(f);
293 return s->system_journal;
294 }
295
296 return f;
297 }
298
299 static void server_vacuum(Server *s) {
300 Iterator i;
301 void *k;
302 char *p;
303 char ids[33];
304 sd_id128_t machine;
305 int r;
306 JournalFile *f;
307
308 log_info("Rotating...");
309
310 if (s->runtime_journal) {
311 r = journal_file_rotate(&s->runtime_journal);
312 if (r < 0)
313 log_error("Failed to rotate %s: %s", s->runtime_journal->path, strerror(-r));
314 }
315
316 if (s->system_journal) {
317 r = journal_file_rotate(&s->system_journal);
318 if (r < 0)
319 log_error("Failed to rotate %s: %s", s->system_journal->path, strerror(-r));
320 }
321
322 HASHMAP_FOREACH_KEY(f, k, s->user_journals, i) {
323 r = journal_file_rotate(&f);
324 if (r < 0)
325 log_error("Failed to rotate %s: %s", f->path, strerror(-r));
326 else
327 hashmap_replace(s->user_journals, k, f);
328 }
329
330 log_info("Vacuuming...");
331
332 r = sd_id128_get_machine(&machine);
333 if (r < 0) {
334 log_error("Failed to get machine ID: %s", strerror(-r));
335 return;
336 }
337
338 if (asprintf(&p, "/var/log/journal/%s", sd_id128_to_string(machine, ids)) < 0) {
339 log_error("Out of memory.");
340 return;
341 }
342
343 r = journal_directory_vacuum(p, s->metrics.max_use, s->metrics.keep_free);
344 if (r < 0 && r != -ENOENT)
345 log_error("Failed to vacuum %s: %s", p, strerror(-r));
346 free(p);
347
348 if (asprintf(&p, "/run/log/journal/%s", ids) < 0) {
349 log_error("Out of memory.");
350 return;
351 }
352
353 r = journal_directory_vacuum(p, s->metrics.max_use, s->metrics.keep_free);
354 if (r < 0 && r != -ENOENT)
355 log_error("Failed to vacuum %s: %s", p, strerror(-r));
356 free(p);
357
358 s->cached_available_space_timestamp = 0;
359 }
360
361 static char *shortened_cgroup_path(pid_t pid) {
362 int r;
363 char *process_path, *init_path, *path;
364
365 assert(pid > 0);
366
367 r = cg_get_by_pid(SYSTEMD_CGROUP_CONTROLLER, pid, &process_path);
368 if (r < 0)
369 return NULL;
370
371 r = cg_get_by_pid(SYSTEMD_CGROUP_CONTROLLER, 1, &init_path);
372 if (r < 0) {
373 free(process_path);
374 return NULL;
375 }
376
377 if (streq(init_path, "/"))
378 init_path[0] = 0;
379
380 if (startswith(process_path, init_path)) {
381 char *p;
382
383 p = strdup(process_path + strlen(init_path));
384 if (!p) {
385 free(process_path);
386 free(init_path);
387 return NULL;
388 }
389 path = p;
390 } else {
391 path = process_path;
392 process_path = NULL;
393 }
394
395 free(process_path);
396 free(init_path);
397
398 return path;
399 }
400
401 static void dispatch_message_real(Server *s,
402 struct iovec *iovec, unsigned n, unsigned m,
403 struct ucred *ucred,
404 struct timeval *tv) {
405
406 char *pid = NULL, *uid = NULL, *gid = NULL,
407 *source_time = NULL, *boot_id = NULL, *machine_id = NULL,
408 *comm = NULL, *cmdline = NULL, *hostname = NULL,
409 *audit_session = NULL, *audit_loginuid = NULL,
410 *exe = NULL, *cgroup = NULL;
411
412 char idbuf[33];
413 sd_id128_t id;
414 int r;
415 char *t;
416 uid_t loginuid = 0, realuid = 0;
417 JournalFile *f;
418 bool vacuumed = false;
419
420 assert(s);
421 assert(iovec);
422 assert(n > 0);
423 assert(n + 13 <= m);
424
425 if (ucred) {
426 uint32_t session;
427 char *path;
428
429 realuid = ucred->uid;
430
431 if (asprintf(&pid, "_PID=%lu", (unsigned long) ucred->pid) >= 0)
432 IOVEC_SET_STRING(iovec[n++], pid);
433
434 if (asprintf(&uid, "_UID=%lu", (unsigned long) ucred->uid) >= 0)
435 IOVEC_SET_STRING(iovec[n++], uid);
436
437 if (asprintf(&gid, "_GID=%lu", (unsigned long) ucred->gid) >= 0)
438 IOVEC_SET_STRING(iovec[n++], gid);
439
440 r = get_process_comm(ucred->pid, &t);
441 if (r >= 0) {
442 comm = strappend("_COMM=", t);
443 if (comm)
444 IOVEC_SET_STRING(iovec[n++], comm);
445 free(t);
446 }
447
448 r = get_process_exe(ucred->pid, &t);
449 if (r >= 0) {
450 exe = strappend("_EXE=", t);
451 if (comm)
452 IOVEC_SET_STRING(iovec[n++], exe);
453 free(t);
454 }
455
456 r = get_process_cmdline(ucred->pid, LINE_MAX, false, &t);
457 if (r >= 0) {
458 cmdline = strappend("_CMDLINE=", t);
459 if (cmdline)
460 IOVEC_SET_STRING(iovec[n++], cmdline);
461 free(t);
462 }
463
464 r = audit_session_from_pid(ucred->pid, &session);
465 if (r >= 0)
466 if (asprintf(&audit_session, "_AUDIT_SESSION=%lu", (unsigned long) session) >= 0)
467 IOVEC_SET_STRING(iovec[n++], audit_session);
468
469 r = audit_loginuid_from_pid(ucred->pid, &loginuid);
470 if (r >= 0)
471 if (asprintf(&audit_loginuid, "_AUDIT_LOGINUID=%lu", (unsigned long) loginuid) >= 0)
472 IOVEC_SET_STRING(iovec[n++], audit_loginuid);
473
474 path = shortened_cgroup_path(ucred->pid);
475 if (path) {
476 cgroup = strappend("_SYSTEMD_CGROUP=", path);
477 if (cgroup)
478 IOVEC_SET_STRING(iovec[n++], cgroup);
479
480 free(path);
481 }
482 }
483
484 if (tv) {
485 if (asprintf(&source_time, "_SOURCE_REALTIME_TIMESTAMP=%llu",
486 (unsigned long long) timeval_load(tv)) >= 0)
487 IOVEC_SET_STRING(iovec[n++], source_time);
488 }
489
490 /* Note that strictly speaking storing the boot id here is
491 * redundant since the entry includes this in-line
492 * anyway. However, we need this indexed, too. */
493 r = sd_id128_get_boot(&id);
494 if (r >= 0)
495 if (asprintf(&boot_id, "_BOOT_ID=%s", sd_id128_to_string(id, idbuf)) >= 0)
496 IOVEC_SET_STRING(iovec[n++], boot_id);
497
498 r = sd_id128_get_machine(&id);
499 if (r >= 0)
500 if (asprintf(&machine_id, "_MACHINE_ID=%s", sd_id128_to_string(id, idbuf)) >= 0)
501 IOVEC_SET_STRING(iovec[n++], machine_id);
502
503 t = gethostname_malloc();
504 if (t) {
505 hostname = strappend("_HOSTNAME=", t);
506 if (hostname)
507 IOVEC_SET_STRING(iovec[n++], hostname);
508 free(t);
509 }
510
511 assert(n <= m);
512
513 server_flush_to_var(s);
514
515 retry:
516 f = find_journal(s, realuid == 0 ? 0 : loginuid);
517 if (!f)
518 log_warning("Dropping message, as we can't find a place to store the data.");
519 else {
520 r = journal_file_append_entry(f, NULL, iovec, n, &s->seqnum, NULL, NULL);
521
522 if (r == -E2BIG && !vacuumed) {
523 log_info("Allocation limit reached.");
524
525 server_vacuum(s);
526 vacuumed = true;
527
528 log_info("Retrying write.");
529 goto retry;
530 }
531
532 if (r < 0)
533 log_error("Failed to write entry, ignoring: %s", strerror(-r));
534 }
535
536 free(pid);
537 free(uid);
538 free(gid);
539 free(comm);
540 free(exe);
541 free(cmdline);
542 free(source_time);
543 free(boot_id);
544 free(machine_id);
545 free(hostname);
546 free(audit_session);
547 free(audit_loginuid);
548 free(cgroup);
549 }
550
551 static void dispatch_message(Server *s,
552 struct iovec *iovec, unsigned n, unsigned m,
553 struct ucred *ucred,
554 struct timeval *tv,
555 int priority) {
556 int rl;
557 char *path = NULL, *c;
558
559 assert(s);
560 assert(iovec || n == 0);
561
562 if (n == 0)
563 return;
564
565 if (!ucred)
566 goto finish;
567
568 path = shortened_cgroup_path(ucred->pid);
569 if (!path)
570 goto finish;
571
572 /* example: /user/lennart/3/foobar
573 * /system/dbus.service/foobar
574 *
575 * So let's cut of everything past the third /, since that is
576 * wher user directories start */
577
578 c = strchr(path, '/');
579 if (c) {
580 c = strchr(c+1, '/');
581 if (c) {
582 c = strchr(c+1, '/');
583 if (c)
584 *c = 0;
585 }
586 }
587
588 rl = journal_rate_limit_test(s->rate_limit, path, priority, available_space(s));
589
590 if (rl == 0) {
591 free(path);
592 return;
593 }
594
595 if (rl > 1) {
596 int j = 0;
597 char suppress_message[LINE_MAX];
598 struct iovec suppress_iovec[15];
599
600 /* Write a suppression message if we suppressed something */
601
602 snprintf(suppress_message, sizeof(suppress_message), "MESSAGE=Suppressed %u messages from %s", rl - 1, path);
603 char_array_0(suppress_message);
604
605 IOVEC_SET_STRING(suppress_iovec[j++], "PRIORITY=5");
606 IOVEC_SET_STRING(suppress_iovec[j++], suppress_message);
607
608 dispatch_message_real(s, suppress_iovec, j, ELEMENTSOF(suppress_iovec), NULL, NULL);
609 }
610
611 free(path);
612
613 finish:
614 dispatch_message_real(s, iovec, n, m, ucred, tv);
615 }
616
617 static void process_syslog_message(Server *s, const char *buf, struct ucred *ucred, struct timeval *tv) {
618 char *message = NULL, *syslog_priority = NULL, *syslog_facility = NULL;
619 struct iovec iovec[16];
620 unsigned n = 0;
621 int priority = LOG_USER | LOG_INFO;
622
623 assert(s);
624 assert(buf);
625
626 parse_syslog_priority((char**) &buf, &priority);
627 skip_syslog_date((char**) &buf);
628
629 if (asprintf(&syslog_priority, "PRIORITY=%i", priority & LOG_PRIMASK) >= 0)
630 IOVEC_SET_STRING(iovec[n++], syslog_priority);
631
632 if (asprintf(&syslog_facility, "SYSLOG_FACILITY=%i", LOG_FAC(priority)) >= 0)
633 IOVEC_SET_STRING(iovec[n++], syslog_facility);
634
635 message = strappend("MESSAGE=", buf);
636 if (message)
637 IOVEC_SET_STRING(iovec[n++], message);
638
639 dispatch_message(s, iovec, n, ELEMENTSOF(iovec), ucred, tv, priority & LOG_PRIMASK);
640
641 free(message);
642 free(syslog_facility);
643 free(syslog_priority);
644 }
645
646 static bool valid_user_field(const char *p, size_t l) {
647 const char *a;
648
649 /* We kinda enforce POSIX syntax recommendations for
650 environment variables here, but make a couple of additional
651 requirements.
652
653 http://pubs.opengroup.org/onlinepubs/000095399/basedefs/xbd_chap08.html */
654
655 /* No empty field names */
656 if (l <= 0)
657 return false;
658
659 /* Don't allow names longer than 64 chars */
660 if (l > 64)
661 return false;
662
663 /* Variables starting with an underscore are protected */
664 if (p[0] == '_')
665 return false;
666
667 /* Don't allow digits as first character */
668 if (p[0] >= '0' && p[0] <= '9')
669 return false;
670
671 /* Only allow A-Z0-9 and '_' */
672 for (a = p; a < p + l; a++)
673 if (!((*a >= 'A' && *a <= 'Z') ||
674 (*a >= '0' && *a <= '9') ||
675 *a == '_'))
676 return false;
677
678 return true;
679 }
680
681 static void process_native_message(Server *s, const void *buffer, size_t buffer_size, struct ucred *ucred, struct timeval *tv) {
682 struct iovec *iovec = NULL;
683 unsigned n = 0, m = 0, j;
684 const char *p;
685 size_t remaining;
686 int priority = LOG_INFO;
687
688 assert(s);
689 assert(buffer || n == 0);
690
691 p = buffer;
692 remaining = buffer_size;
693
694 while (remaining > 0) {
695 const char *e, *q;
696
697 e = memchr(p, '\n', remaining);
698
699 if (!e) {
700 /* Trailing noise, let's ignore it, and flush what we collected */
701 log_debug("Received message with trailing noise, ignoring.");
702 break;
703 }
704
705 if (e == p) {
706 /* Entry separator */
707 dispatch_message(s, iovec, n, m, ucred, tv, priority);
708 n = 0;
709 priority = LOG_INFO;
710
711 p++;
712 remaining--;
713 continue;
714 }
715
716 if (*p == '.' || *p == '#') {
717 /* Ignore control commands for now, and
718 * comments too. */
719 remaining -= (e - p) + 1;
720 p = e + 1;
721 continue;
722 }
723
724 /* A property follows */
725
726 if (n+13 >= m) {
727 struct iovec *c;
728 unsigned u;
729
730 u = MAX((n+13U) * 2U, 4U);
731 c = realloc(iovec, u * sizeof(struct iovec));
732 if (!c) {
733 log_error("Out of memory");
734 break;
735 }
736
737 iovec = c;
738 m = u;
739 }
740
741 q = memchr(p, '=', e - p);
742 if (q) {
743 if (valid_user_field(p, q - p)) {
744 /* If the field name starts with an
745 * underscore, skip the variable,
746 * since that indidates a trusted
747 * field */
748 iovec[n].iov_base = (char*) p;
749 iovec[n].iov_len = e - p;
750 n++;
751
752 /* We need to determine the priority
753 * of this entry for the rate limiting
754 * logic */
755 if (e - p == 10 &&
756 memcmp(p, "PRIORITY=", 10) == 0 &&
757 p[10] >= '0' &&
758 p[10] <= '9')
759 priority = p[10] - '0';
760 }
761
762 remaining -= (e - p) + 1;
763 p = e + 1;
764 continue;
765 } else {
766 uint64_t l;
767 char *k;
768
769 if (remaining < e - p + 1 + sizeof(uint64_t) + 1) {
770 log_debug("Failed to parse message, ignoring.");
771 break;
772 }
773
774 memcpy(&l, e + 1, sizeof(uint64_t));
775 l = le64toh(l);
776
777 if (remaining < e - p + 1 + sizeof(uint64_t) + l + 1 ||
778 e[1+sizeof(uint64_t)+l] != '\n') {
779 log_debug("Failed to parse message, ignoring.");
780 break;
781 }
782
783 k = malloc((e - p) + 1 + l);
784 if (!k) {
785 log_error("Out of memory");
786 break;
787 }
788
789 memcpy(k, p, e - p);
790 k[e - p] = '=';
791 memcpy(k + (e - p) + 1, e + 1 + sizeof(uint64_t), l);
792
793 if (valid_user_field(p, e - p)) {
794 iovec[n].iov_base = k;
795 iovec[n].iov_len = (e - p) + 1 + l;
796 n++;
797 } else
798 free(k);
799
800 remaining -= (e - p) + 1 + sizeof(uint64_t) + l + 1;
801 p = e + 1 + sizeof(uint64_t) + l + 1;
802 }
803 }
804
805 dispatch_message(s, iovec, n, m, ucred, tv, priority);
806
807 for (j = 0; j < n; j++)
808 if (iovec[j].iov_base < buffer ||
809 (const uint8_t*) iovec[j].iov_base >= (const uint8_t*) buffer + buffer_size)
810 free(iovec[j].iov_base);
811 }
812
813 static int stdout_stream_log(StdoutStream *s, const char *p, size_t l) {
814 struct iovec iovec[15];
815 char *message = NULL, *syslog_priority = NULL;
816 unsigned n = 0;
817 size_t tag_len;
818 int priority;
819
820 assert(s);
821 assert(p);
822
823 priority = s->priority;
824
825 if (s->priority_prefix &&
826 l > 3 &&
827 p[0] == '<' &&
828 p[1] >= '0' && p[1] <= '7' &&
829 p[2] == '>') {
830
831 priority = p[1] - '0';
832 p += 3;
833 l -= 3;
834 }
835
836 if (l <= 0)
837 return 0;
838
839 if (asprintf(&syslog_priority, "PRIORITY=%i", priority) >= 0)
840 IOVEC_SET_STRING(iovec[n++], syslog_priority);
841
842 tag_len = s->tag ? strlen(s->tag) + 2: 0;
843 message = malloc(8 + tag_len + l);
844 if (message) {
845 memcpy(message, "MESSAGE=", 8);
846
847 if (s->tag) {
848 memcpy(message+8, s->tag, tag_len-2);
849 memcpy(message+8+tag_len-2, ": ", 2);
850 }
851
852 memcpy(message+8+tag_len, p, l);
853 iovec[n].iov_base = message;
854 iovec[n].iov_len = 8+tag_len+l;
855 n++;
856 }
857
858 dispatch_message(s->server, iovec, n, ELEMENTSOF(iovec), &s->ucred, NULL, priority);
859
860 if (s->tee_console) {
861 int console;
862
863 console = open_terminal("/dev/console", O_WRONLY|O_NOCTTY|O_CLOEXEC);
864 if (console >= 0) {
865 n = 0;
866 if (s->tag) {
867 IOVEC_SET_STRING(iovec[n++], s->tag);
868 IOVEC_SET_STRING(iovec[n++], ": ");
869 }
870
871 iovec[n].iov_base = (void*) p;
872 iovec[n].iov_len = l;
873 n++;
874
875 IOVEC_SET_STRING(iovec[n++], (char*) "\n");
876
877 writev(console, iovec, n);
878 }
879 }
880
881 free(message);
882 free(syslog_priority);
883
884 return 0;
885 }
886
887 static int stdout_stream_line(StdoutStream *s, const char *p, size_t l) {
888 assert(s);
889 assert(p);
890
891 while (l > 0 && strchr(WHITESPACE, *p)) {
892 l--;
893 p++;
894 }
895
896 while (l > 0 && strchr(WHITESPACE, *(p+l-1)))
897 l--;
898
899 switch (s->state) {
900
901 case STDOUT_STREAM_TAG:
902
903 if (l > 0) {
904 s->tag = strndup(p, l);
905 if (!s->tag) {
906 log_error("Out of memory");
907 return -EINVAL;
908 }
909 }
910
911 s->state = STDOUT_STREAM_PRIORITY;
912 return 0;
913
914 case STDOUT_STREAM_PRIORITY:
915 if (l != 1 || *p < '0' || *p > '7') {
916 log_warning("Failed to parse log priority line.");
917 return -EINVAL;
918 }
919
920 s->priority = *p - '0';
921 s->state = STDOUT_STREAM_PRIORITY_PREFIX;
922 return 0;
923
924 case STDOUT_STREAM_PRIORITY_PREFIX:
925 if (l != 1 || *p < '0' || *p > '1') {
926 log_warning("Failed to parse priority prefix line.");
927 return -EINVAL;
928 }
929
930 s->priority_prefix = *p - '0';
931 s->state = STDOUT_STREAM_TEE_CONSOLE;
932 return 0;
933
934 case STDOUT_STREAM_TEE_CONSOLE:
935 if (l != 1 || *p < '0' || *p > '1') {
936 log_warning("Failed to parse tee to console line.");
937 return -EINVAL;
938 }
939
940 s->tee_console = *p - '0';
941 s->state = STDOUT_STREAM_RUNNING;
942 return 0;
943
944 case STDOUT_STREAM_RUNNING:
945 return stdout_stream_log(s, p, l);
946 }
947
948 assert_not_reached("Unknown stream state");
949 }
950
951 static int stdout_stream_scan(StdoutStream *s, bool force_flush) {
952 char *p;
953 size_t remaining;
954 int r;
955
956 assert(s);
957
958 p = s->buffer;
959 remaining = s->length;
960 for (;;) {
961 char *end;
962 size_t skip;
963
964 end = memchr(p, '\n', remaining);
965 if (!end) {
966 if (remaining >= LINE_MAX) {
967 end = p + LINE_MAX;
968 skip = LINE_MAX;
969 } else
970 break;
971 } else
972 skip = end - p + 1;
973
974 r = stdout_stream_line(s, p, end - p);
975 if (r < 0)
976 return r;
977
978 remaining -= skip;
979 p += skip;
980 }
981
982 if (force_flush && remaining > 0) {
983 r = stdout_stream_line(s, p, remaining);
984 if (r < 0)
985 return r;
986
987 p += remaining;
988 remaining = 0;
989 }
990
991 if (p > s->buffer) {
992 memmove(s->buffer, p, remaining);
993 s->length = remaining;
994 }
995
996 return 0;
997 }
998
999 static int stdout_stream_process(StdoutStream *s) {
1000 ssize_t l;
1001 int r;
1002
1003 assert(s);
1004
1005 l = read(s->fd, s->buffer+s->length, sizeof(s->buffer)-1-s->length);
1006 if (l < 0) {
1007
1008 if (errno == EAGAIN)
1009 return 0;
1010
1011 log_warning("Failed to read from stream: %m");
1012 return -errno;
1013 }
1014
1015 if (l == 0) {
1016 r = stdout_stream_scan(s, true);
1017 if (r < 0)
1018 return r;
1019
1020 return 0;
1021 }
1022
1023 s->length += l;
1024 r = stdout_stream_scan(s, false);
1025 if (r < 0)
1026 return r;
1027
1028 return 1;
1029
1030 }
1031
1032 static void stdout_stream_free(StdoutStream *s) {
1033 assert(s);
1034
1035 if (s->server) {
1036 assert(s->server->n_stdout_streams > 0);
1037 s->server->n_stdout_streams --;
1038 LIST_REMOVE(StdoutStream, stdout_stream, s->server->stdout_streams, s);
1039 }
1040
1041 if (s->fd >= 0) {
1042 if (s->server)
1043 epoll_ctl(s->server->epoll_fd, EPOLL_CTL_DEL, s->fd, NULL);
1044
1045 close_nointr_nofail(s->fd);
1046 }
1047
1048 free(s->tag);
1049 free(s);
1050 }
1051
1052 static int stdout_stream_new(Server *s) {
1053 StdoutStream *stream;
1054 int fd, r;
1055 socklen_t len;
1056 struct epoll_event ev;
1057
1058 assert(s);
1059
1060 fd = accept4(s->stdout_fd, NULL, NULL, SOCK_NONBLOCK|SOCK_CLOEXEC);
1061 if (fd < 0) {
1062 if (errno == EAGAIN)
1063 return 0;
1064
1065 log_error("Failed to accept stdout connection: %m");
1066 return -errno;
1067 }
1068
1069 if (s->n_stdout_streams >= STDOUT_STREAMS_MAX) {
1070 log_warning("Too many stdout streams, refusing connection.");
1071 close_nointr_nofail(fd);
1072 return 0;
1073 }
1074
1075 stream = new0(StdoutStream, 1);
1076 if (!stream) {
1077 log_error("Out of memory.");
1078 close_nointr_nofail(fd);
1079 return -ENOMEM;
1080 }
1081
1082 stream->fd = fd;
1083
1084 len = sizeof(stream->ucred);
1085 if (getsockopt(fd, SOL_SOCKET, SO_PEERCRED, &stream->ucred, &len) < 0) {
1086 log_error("Failed to determine peer credentials: %m");
1087 r = -errno;
1088 goto fail;
1089 }
1090
1091 if (shutdown(fd, SHUT_WR) < 0) {
1092 log_error("Failed to shutdown writing side of socket: %m");
1093 r = -errno;
1094 goto fail;
1095 }
1096
1097 zero(ev);
1098 ev.data.ptr = stream;
1099 ev.events = EPOLLIN;
1100 if (epoll_ctl(s->epoll_fd, EPOLL_CTL_ADD, fd, &ev) < 0) {
1101 log_error("Failed to add stream to event loop: %m");
1102 r = -errno;
1103 goto fail;
1104 }
1105
1106 stream->server = s;
1107 LIST_PREPEND(StdoutStream, stdout_stream, s->stdout_streams, stream);
1108 s->n_stdout_streams ++;
1109
1110 return 0;
1111
1112 fail:
1113 stdout_stream_free(stream);
1114 return r;
1115 }
1116
1117 static int system_journal_open(Server *s) {
1118 int r;
1119 char *fn;
1120 sd_id128_t machine;
1121 char ids[33];
1122
1123 r = sd_id128_get_machine(&machine);
1124 if (r < 0)
1125 return r;
1126
1127 sd_id128_to_string(machine, ids);
1128
1129 if (!s->system_journal) {
1130
1131 /* First try to create the machine path, but not the prefix */
1132 fn = strappend("/var/log/journal/", ids);
1133 if (!fn)
1134 return -ENOMEM;
1135 (void) mkdir(fn, 0755);
1136 free(fn);
1137
1138 /* The create the system journal file */
1139 fn = join("/var/log/journal/", ids, "/system.journal", NULL);
1140 if (!fn)
1141 return -ENOMEM;
1142
1143 r = journal_file_open(fn, O_RDWR|O_CREAT, 0640, NULL, &s->system_journal);
1144 free(fn);
1145
1146 if (r >= 0) {
1147 s->system_journal->metrics = s->metrics;
1148 s->system_journal->compress = s->compress;
1149
1150 fix_perms(s->system_journal, 0);
1151 } else if (r < 0) {
1152
1153 if (r == -ENOENT)
1154 r = 0;
1155 else {
1156 log_error("Failed to open system journal: %s", strerror(-r));
1157 return r;
1158 }
1159 }
1160 }
1161
1162 if (!s->runtime_journal) {
1163
1164 fn = join("/run/log/journal/", ids, "/system.journal", NULL);
1165 if (!fn)
1166 return -ENOMEM;
1167
1168 if (s->system_journal) {
1169
1170 /* Try to open the runtime journal, but only
1171 * if it already exists, so that we can flush
1172 * it into the system journal */
1173
1174 r = journal_file_open(fn, O_RDWR, 0640, NULL, &s->runtime_journal);
1175 free(fn);
1176
1177 if (r < 0) {
1178
1179 if (r == -ENOENT)
1180 r = 0;
1181 else {
1182 log_error("Failed to open runtime journal: %s", strerror(-r));
1183 return r;
1184 }
1185 }
1186
1187 } else {
1188
1189 /* OK, we really need the runtime journal, so create
1190 * it if necessary. */
1191
1192 (void) mkdir_parents(fn, 0755);
1193 r = journal_file_open(fn, O_RDWR|O_CREAT, 0640, NULL, &s->runtime_journal);
1194 free(fn);
1195
1196 if (r < 0) {
1197 log_error("Failed to open runtime journal: %s", strerror(-r));
1198 return r;
1199 }
1200 }
1201
1202 if (s->runtime_journal) {
1203 s->runtime_journal->metrics = s->metrics;
1204 s->runtime_journal->compress = s->compress;
1205
1206 fix_perms(s->runtime_journal, 0);
1207 }
1208 }
1209
1210 return r;
1211 }
1212
1213 static int server_flush_to_var(Server *s) {
1214 char path[] = "/run/log/journal/xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx";
1215 Object *o = NULL;
1216 int r;
1217 sd_id128_t machine;
1218 sd_journal *j;
1219 usec_t ts;
1220
1221 assert(s);
1222
1223 if (!s->runtime_journal)
1224 return 0;
1225
1226 ts = now(CLOCK_MONOTONIC);
1227 if (s->var_available_timestamp + RECHECK_VAR_AVAILABLE_USEC > ts)
1228 return 0;
1229
1230 s->var_available_timestamp = ts;
1231
1232 system_journal_open(s);
1233
1234 if (!s->system_journal)
1235 return 0;
1236
1237 r = sd_id128_get_machine(&machine);
1238 if (r < 0) {
1239 log_error("Failed to get machine id: %s", strerror(-r));
1240 return r;
1241 }
1242
1243 r = sd_journal_open(&j, SD_JOURNAL_RUNTIME_ONLY);
1244 if (r < 0) {
1245 log_error("Failed to read runtime journal: %s", strerror(-r));
1246 return r;
1247 }
1248
1249 SD_JOURNAL_FOREACH(j) {
1250 JournalFile *f;
1251
1252 f = j->current_file;
1253 assert(f && f->current_offset > 0);
1254
1255 r = journal_file_move_to_object(f, OBJECT_ENTRY, f->current_offset, &o);
1256 if (r < 0) {
1257 log_error("Can't read entry: %s", strerror(-r));
1258 goto finish;
1259 }
1260
1261 r = journal_file_copy_entry(f, s->system_journal, o, f->current_offset, NULL, NULL, NULL);
1262 if (r == -E2BIG) {
1263 log_info("Allocation limit reached.");
1264
1265 journal_file_post_change(s->system_journal);
1266 server_vacuum(s);
1267
1268 r = journal_file_copy_entry(f, s->system_journal, o, f->current_offset, NULL, NULL, NULL);
1269 }
1270
1271 if (r < 0) {
1272 log_error("Can't write entry: %s", strerror(-r));
1273 goto finish;
1274 }
1275 }
1276
1277 finish:
1278 journal_file_post_change(s->system_journal);
1279
1280 journal_file_close(s->runtime_journal);
1281 s->runtime_journal = NULL;
1282
1283 if (r >= 0) {
1284 sd_id128_to_string(machine, path + 17);
1285 rm_rf(path, false, true, false);
1286 }
1287
1288 return r;
1289 }
1290
1291 static void forward_syslog(Server *s, const void *buffer, size_t length, struct ucred *ucred, struct timeval *tv) {
1292 struct msghdr msghdr;
1293 struct iovec iovec;
1294 struct cmsghdr *cmsg;
1295 union {
1296 struct cmsghdr cmsghdr;
1297 uint8_t buf[CMSG_SPACE(sizeof(struct ucred)) +
1298 CMSG_SPACE(sizeof(struct timeval))];
1299 } control;
1300 union sockaddr_union sa;
1301
1302 assert(s);
1303
1304 zero(msghdr);
1305
1306 zero(iovec);
1307 iovec.iov_base = (void*) buffer;
1308 iovec.iov_len = length;
1309 msghdr.msg_iov = &iovec;
1310 msghdr.msg_iovlen = 1;
1311
1312 zero(sa);
1313 sa.un.sun_family = AF_UNIX;
1314 strncpy(sa.un.sun_path, "/run/systemd/syslog", sizeof(sa.un.sun_path));
1315 msghdr.msg_name = &sa;
1316 msghdr.msg_namelen = offsetof(union sockaddr_union, un.sun_path) + strlen(sa.un.sun_path);
1317
1318 zero(control);
1319 msghdr.msg_control = &control;
1320 msghdr.msg_controllen = sizeof(control);
1321
1322 cmsg = CMSG_FIRSTHDR(&msghdr);
1323 cmsg->cmsg_level = SOL_SOCKET;
1324 cmsg->cmsg_type = SCM_CREDENTIALS;
1325 cmsg->cmsg_len = CMSG_LEN(sizeof(struct ucred));
1326 memcpy(CMSG_DATA(cmsg), ucred, sizeof(struct ucred));
1327 msghdr.msg_controllen = cmsg->cmsg_len;
1328
1329 /* Forward the syslog message we received via /dev/log to
1330 * /run/systemd/syslog. Unfortunately we currently can't set
1331 * the SO_TIMESTAMP auxiliary data, and hence we don't. */
1332
1333 if (sendmsg(s->syslog_fd, &msghdr, MSG_NOSIGNAL) >= 0)
1334 return;
1335
1336 if (errno == ESRCH) {
1337 struct ucred u;
1338
1339 /* Hmm, presumably the sender process vanished
1340 * by now, so let's fix it as good as we
1341 * can, and retry */
1342
1343 u = *ucred;
1344 u.pid = getpid();
1345 memcpy(CMSG_DATA(cmsg), &u, sizeof(struct ucred));
1346
1347 if (sendmsg(s->syslog_fd, &msghdr, MSG_NOSIGNAL) >= 0)
1348 return;
1349 }
1350
1351 log_debug("Failed to forward syslog message: %m");
1352 }
1353
1354 static int process_event(Server *s, struct epoll_event *ev) {
1355 assert(s);
1356
1357 if (ev->data.fd == s->signal_fd) {
1358 struct signalfd_siginfo sfsi;
1359 ssize_t n;
1360
1361 if (ev->events != EPOLLIN) {
1362 log_info("Got invalid event from epoll.");
1363 return -EIO;
1364 }
1365
1366 n = read(s->signal_fd, &sfsi, sizeof(sfsi));
1367 if (n != sizeof(sfsi)) {
1368
1369 if (n >= 0)
1370 return -EIO;
1371
1372 if (errno == EINTR || errno == EAGAIN)
1373 return 0;
1374
1375 return -errno;
1376 }
1377
1378 if (sfsi.ssi_signo == SIGUSR1) {
1379 server_flush_to_var(s);
1380 return 0;
1381 }
1382
1383 log_debug("Received SIG%s", signal_to_string(sfsi.ssi_signo));
1384 return 0;
1385
1386 } else if (ev->data.fd == s->native_fd ||
1387 ev->data.fd == s->syslog_fd) {
1388
1389 if (ev->events != EPOLLIN) {
1390 log_info("Got invalid event from epoll.");
1391 return -EIO;
1392 }
1393
1394 for (;;) {
1395 struct msghdr msghdr;
1396 struct iovec iovec;
1397 struct ucred *ucred = NULL;
1398 struct timeval *tv = NULL;
1399 struct cmsghdr *cmsg;
1400 union {
1401 struct cmsghdr cmsghdr;
1402 uint8_t buf[CMSG_SPACE(sizeof(struct ucred)) +
1403 CMSG_SPACE(sizeof(struct timeval))];
1404 } control;
1405 ssize_t n;
1406 int v;
1407
1408 if (ioctl(ev->data.fd, SIOCINQ, &v) < 0) {
1409 log_error("SIOCINQ failed: %m");
1410 return -errno;
1411 }
1412
1413 if (v <= 0)
1414 return 1;
1415
1416 if (s->buffer_size < (size_t) v) {
1417 void *b;
1418 size_t l;
1419
1420 l = MAX(LINE_MAX + (size_t) v, s->buffer_size * 2);
1421 b = realloc(s->buffer, l+1);
1422
1423 if (!b) {
1424 log_error("Couldn't increase buffer.");
1425 return -ENOMEM;
1426 }
1427
1428 s->buffer_size = l;
1429 s->buffer = b;
1430 }
1431
1432 zero(iovec);
1433 iovec.iov_base = s->buffer;
1434 iovec.iov_len = s->buffer_size;
1435
1436 zero(control);
1437 zero(msghdr);
1438 msghdr.msg_iov = &iovec;
1439 msghdr.msg_iovlen = 1;
1440 msghdr.msg_control = &control;
1441 msghdr.msg_controllen = sizeof(control);
1442
1443 n = recvmsg(ev->data.fd, &msghdr, MSG_DONTWAIT);
1444 if (n < 0) {
1445
1446 if (errno == EINTR || errno == EAGAIN)
1447 return 1;
1448
1449 log_error("recvmsg() failed: %m");
1450 return -errno;
1451 }
1452
1453 for (cmsg = CMSG_FIRSTHDR(&msghdr); cmsg; cmsg = CMSG_NXTHDR(&msghdr, cmsg)) {
1454
1455 if (cmsg->cmsg_level == SOL_SOCKET &&
1456 cmsg->cmsg_type == SCM_CREDENTIALS &&
1457 cmsg->cmsg_len == CMSG_LEN(sizeof(struct ucred)))
1458 ucred = (struct ucred*) CMSG_DATA(cmsg);
1459 else if (cmsg->cmsg_level == SOL_SOCKET &&
1460 cmsg->cmsg_type == SO_TIMESTAMP &&
1461 cmsg->cmsg_len == CMSG_LEN(sizeof(struct timeval)))
1462 tv = (struct timeval*) CMSG_DATA(cmsg);
1463 }
1464
1465 if (ev->data.fd == s->syslog_fd) {
1466 char *e;
1467
1468 e = memchr(s->buffer, '\n', n);
1469 if (e)
1470 *e = 0;
1471 else
1472 s->buffer[n] = 0;
1473
1474 forward_syslog(s, s->buffer, n, ucred, tv);
1475 process_syslog_message(s, strstrip(s->buffer), ucred, tv);
1476 } else
1477 process_native_message(s, s->buffer, n, ucred, tv);
1478 }
1479
1480 return 1;
1481
1482 } else if (ev->data.fd == s->stdout_fd) {
1483
1484 if (ev->events != EPOLLIN) {
1485 log_info("Got invalid event from epoll.");
1486 return -EIO;
1487 }
1488
1489 stdout_stream_new(s);
1490 return 1;
1491
1492 } else {
1493 StdoutStream *stream;
1494
1495 if ((ev->events|EPOLLIN|EPOLLHUP) != (EPOLLIN|EPOLLHUP)) {
1496 log_info("Got invalid event from epoll.");
1497 return -EIO;
1498 }
1499
1500 /* If it is none of the well-known fds, it must be an
1501 * stdout stream fd. Note that this is a bit ugly here
1502 * (since we rely that none of the well-known fds
1503 * could be interpreted as pointer), but nonetheless
1504 * safe, since the well-known fds would never get an
1505 * fd > 4096, i.e. beyond the first memory page */
1506
1507 stream = ev->data.ptr;
1508
1509 if (stdout_stream_process(stream) <= 0)
1510 stdout_stream_free(stream);
1511
1512 return 1;
1513 }
1514
1515 log_error("Unknown event.");
1516 return 0;
1517 }
1518
1519 static int open_syslog_socket(Server *s) {
1520 union sockaddr_union sa;
1521 int one, r;
1522 struct epoll_event ev;
1523 struct timeval tv;
1524
1525 assert(s);
1526
1527 if (s->syslog_fd < 0) {
1528
1529 s->syslog_fd = socket(AF_UNIX, SOCK_DGRAM|SOCK_CLOEXEC, 0);
1530 if (s->syslog_fd < 0) {
1531 log_error("socket() failed: %m");
1532 return -errno;
1533 }
1534
1535 zero(sa);
1536 sa.un.sun_family = AF_UNIX;
1537 strncpy(sa.un.sun_path, "/dev/log", sizeof(sa.un.sun_path));
1538
1539 unlink(sa.un.sun_path);
1540
1541 r = bind(s->syslog_fd, &sa.sa, offsetof(union sockaddr_union, un.sun_path) + strlen(sa.un.sun_path));
1542 if (r < 0) {
1543 log_error("bind() failed: %m");
1544 return -errno;
1545 }
1546
1547 chmod(sa.un.sun_path, 0666);
1548 }
1549
1550 one = 1;
1551 r = setsockopt(s->syslog_fd, SOL_SOCKET, SO_PASSCRED, &one, sizeof(one));
1552 if (r < 0) {
1553 log_error("SO_PASSCRED failed: %m");
1554 return -errno;
1555 }
1556
1557 one = 1;
1558 r = setsockopt(s->syslog_fd, SOL_SOCKET, SO_TIMESTAMP, &one, sizeof(one));
1559 if (r < 0) {
1560 log_error("SO_TIMESTAMP failed: %m");
1561 return -errno;
1562 }
1563
1564 /* Since we use the same socket for forwarding this to some
1565 * other syslog implementation, make sure we don't hang
1566 * forever */
1567 timeval_store(&tv, SYSLOG_TIMEOUT_USEC);
1568 if (setsockopt(s->syslog_fd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv)) < 0) {
1569 log_error("SO_SNDTIMEO failed: %m");
1570 return -errno;
1571 }
1572
1573 zero(ev);
1574 ev.events = EPOLLIN;
1575 ev.data.fd = s->syslog_fd;
1576 if (epoll_ctl(s->epoll_fd, EPOLL_CTL_ADD, s->syslog_fd, &ev) < 0) {
1577 log_error("Failed to add syslog server fd to epoll object: %m");
1578 return -errno;
1579 }
1580
1581 return 0;
1582 }
1583
1584 static int open_native_socket(Server*s) {
1585 union sockaddr_union sa;
1586 int one, r;
1587 struct epoll_event ev;
1588
1589 assert(s);
1590
1591 if (s->native_fd < 0) {
1592
1593 s->native_fd = socket(AF_UNIX, SOCK_DGRAM|SOCK_CLOEXEC, 0);
1594 if (s->native_fd < 0) {
1595 log_error("socket() failed: %m");
1596 return -errno;
1597 }
1598
1599 zero(sa);
1600 sa.un.sun_family = AF_UNIX;
1601 strncpy(sa.un.sun_path, "/run/systemd/journal", sizeof(sa.un.sun_path));
1602
1603 unlink(sa.un.sun_path);
1604
1605 r = bind(s->native_fd, &sa.sa, offsetof(union sockaddr_union, un.sun_path) + strlen(sa.un.sun_path));
1606 if (r < 0) {
1607 log_error("bind() failed: %m");
1608 return -errno;
1609 }
1610
1611 chmod(sa.un.sun_path, 0666);
1612 }
1613
1614 one = 1;
1615 r = setsockopt(s->native_fd, SOL_SOCKET, SO_PASSCRED, &one, sizeof(one));
1616 if (r < 0) {
1617 log_error("SO_PASSCRED failed: %m");
1618 return -errno;
1619 }
1620
1621 one = 1;
1622 r = setsockopt(s->native_fd, SOL_SOCKET, SO_TIMESTAMP, &one, sizeof(one));
1623 if (r < 0) {
1624 log_error("SO_TIMESTAMP failed: %m");
1625 return -errno;
1626 }
1627
1628 zero(ev);
1629 ev.events = EPOLLIN;
1630 ev.data.fd = s->native_fd;
1631 if (epoll_ctl(s->epoll_fd, EPOLL_CTL_ADD, s->native_fd, &ev) < 0) {
1632 log_error("Failed to add native server fd to epoll object: %m");
1633 return -errno;
1634 }
1635
1636 return 0;
1637 }
1638
1639 static int open_stdout_socket(Server *s) {
1640 union sockaddr_union sa;
1641 int r;
1642 struct epoll_event ev;
1643
1644 assert(s);
1645
1646 if (s->stdout_fd < 0) {
1647
1648 s->stdout_fd = socket(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC, 0);
1649 if (s->stdout_fd < 0) {
1650 log_error("socket() failed: %m");
1651 return -errno;
1652 }
1653
1654 zero(sa);
1655 sa.un.sun_family = AF_UNIX;
1656 strncpy(sa.un.sun_path, "/run/systemd/stdout", sizeof(sa.un.sun_path));
1657
1658 unlink(sa.un.sun_path);
1659
1660 r = bind(s->stdout_fd, &sa.sa, offsetof(union sockaddr_union, un.sun_path) + strlen(sa.un.sun_path));
1661 if (r < 0) {
1662 log_error("bind() failed: %m");
1663 return -errno;
1664 }
1665
1666 chmod(sa.un.sun_path, 0666);
1667
1668 if (listen(s->stdout_fd, SOMAXCONN) < 0) {
1669 log_error("liste() failed: %m");
1670 return -errno;
1671 }
1672 }
1673
1674 zero(ev);
1675 ev.events = EPOLLIN;
1676 ev.data.fd = s->stdout_fd;
1677 if (epoll_ctl(s->epoll_fd, EPOLL_CTL_ADD, s->stdout_fd, &ev) < 0) {
1678 log_error("Failed to add stdout server fd to epoll object: %m");
1679 return -errno;
1680 }
1681
1682 return 0;
1683 }
1684
1685 static int open_signalfd(Server *s) {
1686 sigset_t mask;
1687 struct epoll_event ev;
1688
1689 assert(s);
1690
1691 assert_se(sigemptyset(&mask) == 0);
1692 sigset_add_many(&mask, SIGINT, SIGTERM, SIGUSR1, -1);
1693 assert_se(sigprocmask(SIG_SETMASK, &mask, NULL) == 0);
1694
1695 s->signal_fd = signalfd(-1, &mask, SFD_NONBLOCK|SFD_CLOEXEC);
1696 if (s->signal_fd < 0) {
1697 log_error("signalfd(): %m");
1698 return -errno;
1699 }
1700
1701 zero(ev);
1702 ev.events = EPOLLIN;
1703 ev.data.fd = s->signal_fd;
1704
1705 if (epoll_ctl(s->epoll_fd, EPOLL_CTL_ADD, s->signal_fd, &ev) < 0) {
1706 log_error("epoll_ctl(): %m");
1707 return -errno;
1708 }
1709
1710 return 0;
1711 }
1712
1713 static int server_init(Server *s) {
1714 int n, r, fd;
1715
1716 assert(s);
1717
1718 zero(*s);
1719 s->syslog_fd = s->native_fd = s->stdout_fd = s->signal_fd = s->epoll_fd = -1;
1720 s->metrics.max_size = DEFAULT_MAX_SIZE;
1721 s->metrics.min_size = DEFAULT_MIN_SIZE;
1722 s->metrics.keep_free = DEFAULT_KEEP_FREE;
1723 s->metrics.max_use = DEFAULT_MAX_USE;
1724 s->compress = true;
1725
1726 s->user_journals = hashmap_new(trivial_hash_func, trivial_compare_func);
1727 if (!s->user_journals) {
1728 log_error("Out of memory.");
1729 return -ENOMEM;
1730 }
1731
1732 s->epoll_fd = epoll_create1(EPOLL_CLOEXEC);
1733 if (s->epoll_fd < 0) {
1734 log_error("Failed to create epoll object: %m");
1735 return -errno;
1736 }
1737
1738 n = sd_listen_fds(true);
1739 if (n < 0) {
1740 log_error("Failed to read listening file descriptors from environment: %s", strerror(-n));
1741 return n;
1742 }
1743
1744 for (fd = SD_LISTEN_FDS_START; fd < SD_LISTEN_FDS_START + n; fd++) {
1745
1746 if (sd_is_socket_unix(fd, SOCK_DGRAM, -1, "/run/systemd/native", 0) > 0) {
1747
1748 if (s->native_fd >= 0) {
1749 log_error("Too many native sockets passed.");
1750 return -EINVAL;
1751 }
1752
1753 s->native_fd = fd;
1754
1755 } else if (sd_is_socket_unix(fd, SOCK_STREAM, 1, "/run/systemd/stdout", 0) > 0) {
1756
1757 if (s->stdout_fd >= 0) {
1758 log_error("Too many stdout sockets passed.");
1759 return -EINVAL;
1760 }
1761
1762 s->stdout_fd = fd;
1763
1764 } else if (sd_is_socket_unix(fd, SOCK_DGRAM, -1, "/dev/log", 0) > 0) {
1765
1766 if (s->syslog_fd >= 0) {
1767 log_error("Too many /dev/log sockets passed.");
1768 return -EINVAL;
1769 }
1770
1771 s->syslog_fd = fd;
1772
1773 } else {
1774 log_error("Unknown socket passed.");
1775 return -EINVAL;
1776 }
1777 }
1778
1779 r = open_syslog_socket(s);
1780 if (r < 0)
1781 return r;
1782
1783 r = open_native_socket(s);
1784 if (r < 0)
1785 return r;
1786
1787 r = open_stdout_socket(s);
1788 if (r < 0)
1789 return r;
1790
1791 r = system_journal_open(s);
1792 if (r < 0)
1793 return r;
1794
1795 r = open_signalfd(s);
1796 if (r < 0)
1797 return r;
1798
1799 s->rate_limit = journal_rate_limit_new(DEFAULT_RATE_LIMIT_INTERVAL, DEFAULT_RATE_LIMIT_BURST);
1800 if (!s->rate_limit)
1801 return -ENOMEM;
1802
1803 return 0;
1804 }
1805
1806 static void server_done(Server *s) {
1807 JournalFile *f;
1808 assert(s);
1809
1810 while (s->stdout_streams)
1811 stdout_stream_free(s->stdout_streams);
1812
1813 if (s->system_journal)
1814 journal_file_close(s->system_journal);
1815
1816 if (s->runtime_journal)
1817 journal_file_close(s->runtime_journal);
1818
1819 while ((f = hashmap_steal_first(s->user_journals)))
1820 journal_file_close(f);
1821
1822 hashmap_free(s->user_journals);
1823
1824 if (s->epoll_fd >= 0)
1825 close_nointr_nofail(s->epoll_fd);
1826
1827 if (s->signal_fd >= 0)
1828 close_nointr_nofail(s->signal_fd);
1829
1830 if (s->syslog_fd >= 0)
1831 close_nointr_nofail(s->syslog_fd);
1832
1833 if (s->native_fd >= 0)
1834 close_nointr_nofail(s->native_fd);
1835
1836 if (s->stdout_fd >= 0)
1837 close_nointr_nofail(s->stdout_fd);
1838
1839 if (s->rate_limit)
1840 journal_rate_limit_free(s->rate_limit);
1841
1842 free(s->buffer);
1843 }
1844
1845 int main(int argc, char *argv[]) {
1846 Server server;
1847 int r;
1848
1849 /* if (getppid() != 1) { */
1850 /* log_error("This program should be invoked by init only."); */
1851 /* return EXIT_FAILURE; */
1852 /* } */
1853
1854 if (argc > 1) {
1855 log_error("This program does not take arguments.");
1856 return EXIT_FAILURE;
1857 }
1858
1859 log_set_target(LOG_TARGET_CONSOLE);
1860 log_parse_environment();
1861 log_open();
1862
1863 umask(0022);
1864
1865 r = server_init(&server);
1866 if (r < 0)
1867 goto finish;
1868
1869 log_debug("systemd-journald running as pid %lu", (unsigned long) getpid());
1870
1871 sd_notify(false,
1872 "READY=1\n"
1873 "STATUS=Processing requests...");
1874
1875 server_vacuum(&server);
1876 server_flush_to_var(&server);
1877
1878 for (;;) {
1879 struct epoll_event event;
1880
1881 r = epoll_wait(server.epoll_fd, &event, 1, -1);
1882 if (r < 0) {
1883
1884 if (errno == EINTR)
1885 continue;
1886
1887 log_error("epoll_wait() failed: %m");
1888 r = -errno;
1889 goto finish;
1890 } else if (r == 0)
1891 break;
1892
1893 r = process_event(&server, &event);
1894 if (r < 0)
1895 goto finish;
1896 else if (r == 0)
1897 break;
1898 }
1899
1900 log_debug("systemd-journald stopped as pid %lu", (unsigned long) getpid());
1901
1902 finish:
1903 sd_notify(false,
1904 "STATUS=Shutting down...");
1905
1906 server_done(&server);
1907
1908 return r < 0 ? EXIT_FAILURE : EXIT_SUCCESS;
1909 }
Unnamed repository; edit this file 'description' to name the repository.