journal: implement inotify-based live logging logic

src/journal/journal-file.c

@@ -879,6 +879,18 @@ static int journal_file_append_entry_internal(
         return 0;
 }
 
+static void journal_file_post_change(JournalFile *f) {
+        assert(f);
+
+        /* inotify() does not receive IN_MODIFY events from file
+         * accesses done via mmap(). After each access we hence
+         * trigger IN_MODIFY by truncating the journal file to its
+         * current size which triggers IN_MODIFY. */
+
+        if (ftruncate(f->fd, f->last_stat.st_size) < 0)
+                log_error("Failed to to truncate file to its own size: %m");
+}
+
 int journal_file_append_entry(JournalFile *f, const dual_timestamp *ts, const struct iovec iovec[], unsigned n_iovec, uint64_t *seqnum, Object **ret, uint64_t *offset) {
         unsigned i;
         EntryItem *items;
@@ -923,6 +935,8 @@ int journal_file_append_entry(JournalFile *f, const dual_timestamp *ts, const st
 
         r = journal_file_append_entry_internal(f, ts, xor_hash, items, n_iovec, seqnum, ret, offset);
 
+        journal_file_post_change(f);
+
 finish:
         free(items);
 

src/journal/journalctl.c

@@ -26,12 +26,15 @@
 #include <stdio.h>
 #include <unistd.h>
 #include <stdlib.h>
+#include <sys/poll.h>
 
 #include "sd-journal.h"
 #include "log.h"
 
+static bool arg_follow = true;
+
 int main(int argc, char *argv[]) {
-        int r, i;
+        int r, i, fd;
         sd_journal *j = NULL;
 
         log_set_max_level(LOG_DEBUG);
@@ -54,32 +57,68 @@ int main(int argc, char *argv[]) {
                 }
         }
 
-        SD_JOURNAL_FOREACH(j) {
+        fd = sd_journal_get_fd(j);
+        if (fd < 0) {
+                log_error("Failed to get wakeup fd: %s", strerror(-fd));
+                goto finish;
+        }
 
-                const void *data;
+        r = sd_journal_seek_head(j);
-                size_t length;
+        if (r < 0) {
-                char *cursor;
+                log_error("Failed to seek to head: %s", strerror(-r));
-                uint64_t realtime = 0, monotonic = 0;
+                goto finish;
+        }
 
-                r = sd_journal_get_cursor(j, &cursor);
+        for (;;) {
-                if (r < 0) {
+                struct pollfd pollfd;
-                        log_error("Failed to get cursor: %s", strerror(-r));
+
+                while (sd_journal_next(j) > 0) {
+                        const void *data;
+                        size_t length;
+                        char *cursor;
+                        uint64_t realtime = 0, monotonic = 0;
+
+                        r = sd_journal_get_cursor(j, &cursor);
+                        if (r < 0) {
+                                log_error("Failed to get cursor: %s", strerror(-r));
+                                goto finish;
+                        }
+
+                        printf("entry: %s\n", cursor);
+                        free(cursor);
+
+                        sd_journal_get_realtime_usec(j, &realtime);
+                        sd_journal_get_monotonic_usec(j, &monotonic, NULL);
+                        printf("realtime: %llu\n"
+                               "monotonic: %llu\n",
+                               (unsigned long long) realtime,
+                               (unsigned long long) monotonic);
+
+                        SD_JOURNAL_FOREACH_DATA(j, data, length)
+                                printf("\t%.*s\n", (int) length, (const char*) data);
+                }
+
+                if (!arg_follow)
+                        break;
+
+                zero(pollfd);
+                pollfd.fd = fd;
+                pollfd.events = POLLIN;
+
+                if (poll(&pollfd, 1, -1) < 0) {
+                        if (errno == EINTR)
+                                break;
+
+                        log_error("poll(): %m");
+                        r = -errno;
                         goto finish;
                 }
 
-                printf("entry: %s\n", cursor);
+                r = sd_journal_process(j);
-                free(cursor);
+                if (r < 0) {
-
+                        log_error("Failed to process: %s", strerror(-r));
-                sd_journal_get_realtime_usec(j, &realtime);
+                        goto finish;
-                sd_journal_get_monotonic_usec(j, &monotonic, NULL);
+                }
-                printf("realtime: %llu\n"
-                       "monotonic: %llu\n",
-                       (unsigned long long) realtime,
-                       (unsigned long long) monotonic);
-
-                SD_JOURNAL_FOREACH_DATA(j, data, length)
-                        printf("\t%.*s\n", (int) length, (const char*) data);
-
         }
 
 finish:

src/journal/journald.c

@@ -866,7 +866,7 @@ int main(int argc, char *argv[]) {
         sd_notify(false,
                   "READY=1\n"
                   "STATUS=Processing messages...");
-#
+
         for (;;) {
                 struct epoll_event event;
 

src/journal/sd-journal.c

@@ -22,6 +22,8 @@
 #include <errno.h>
 #include <fcntl.h>
 #include <stddef.h>
+#include <unistd.h>
+#include <sys/inotify.h>
 
 #include "sd-journal.h"
 #include "journal-def.h"
@@ -73,6 +75,10 @@ struct sd_journal {
         JournalFile *current_file;
         uint64_t current_field;
 
+        int inotify_fd;
+        Hashmap *inotify_wd_dirs;
+        Hashmap *inotify_wd_roots;
+
         LIST_HEAD(Match, matches);
         unsigned n_matches;
 };
@@ -934,11 +940,6 @@ static int add_file(sd_journal *j, const char *prefix, const char *dir, const ch
         assert(prefix);
         assert(filename);
 
-        if (hashmap_size(j->files) >= JOURNAL_FILES_MAX) {
-                log_debug("Too many open journal files, ignoring.");
-                return 0;
-        }
-
         if (dir)
                 fn = join(prefix, "/", dir, "/", filename, NULL);
         else
@@ -947,6 +948,17 @@ static int add_file(sd_journal *j, const char *prefix, const char *dir, const ch
         if (!fn)
                 return -ENOMEM;
 
+        if (hashmap_get(j->files, fn)) {
+                free(fn);
+                return 0;
+        }
+
+        if (hashmap_size(j->files) >= JOURNAL_FILES_MAX) {
+                log_debug("Too many open journal files, not adding %s, ignoring.", fn);
+                free(fn);
+                return 0;
+        }
+
         r = journal_file_open(fn, O_RDONLY, 0, NULL, &f);
         free(fn);
 
@@ -965,6 +977,37 @@ static int add_file(sd_journal *j, const char *prefix, const char *dir, const ch
                 return r;
         }
 
+        log_debug("File %s got added.", f->path);
+
+        return 0;
+}
+
+static int remove_file(sd_journal *j, const char *prefix, const char *dir, const char *filename) {
+        char *fn;
+        JournalFile *f;
+
+        assert(j);
+        assert(prefix);
+        assert(filename);
+
+        if (dir)
+                fn = join(prefix, "/", dir, "/", filename, NULL);
+        else
+                fn = join(prefix, "/", filename, NULL);
+
+        if (!fn)
+                return -ENOMEM;
+
+        f = hashmap_get(j->files, fn);
+        free(fn);
+
+        if (!f)
+                return 0;
+
+        hashmap_remove(j->files, f->path);
+        journal_file_close(f);
+
+        log_debug("File %s got removed.", f->path);
         return 0;
 }
 
@@ -972,6 +1015,7 @@ static int add_directory(sd_journal *j, const char *prefix, const char *dir) {
         char *fn;
         int r;
         DIR *d;
+        int wd;
 
         assert(j);
         assert(prefix);
@@ -982,15 +1026,28 @@ static int add_directory(sd_journal *j, const char *prefix, const char *dir) {
                 return -ENOMEM;
 
         d = opendir(fn);
-        free(fn);
 
         if (!d) {
+                free(fn);
                 if (errno == ENOENT)
                         return 0;
 
                 return -errno;
         }
 
+        wd = inotify_add_watch(j->inotify_fd, fn,
+                               IN_CREATE|IN_MOVED_TO|IN_MODIFY|IN_ATTRIB|IN_DELETE|
+                               IN_DELETE_SELF|IN_MOVE_SELF|IN_UNMOUNT|
+                               IN_DONT_FOLLOW|IN_ONLYDIR);
+        if (wd > 0) {
+                if (hashmap_put(j->inotify_wd_dirs, INT_TO_PTR(wd), fn) < 0)
+                        inotify_rm_watch(j->inotify_fd, wd);
+                else
+                        fn = NULL;
+        }
+
+        free(fn);
+
         for (;;) {
                 struct dirent buf, *de;
 
@@ -1008,9 +1065,65 @@ static int add_directory(sd_journal *j, const char *prefix, const char *dir) {
 
         closedir(d);
 
+        log_debug("Directory %s/%s got added.", prefix, dir);
+
         return 0;
 }
 
+static void remove_directory_wd(sd_journal *j, int wd) {
+        char *p;
+
+        assert(j);
+        assert(wd > 0);
+
+        if (j->inotify_fd >= 0)
+                inotify_rm_watch(j->inotify_fd, wd);
+
+        p = hashmap_remove(j->inotify_wd_dirs, INT_TO_PTR(wd));
+
+        if (p) {
+                log_debug("Directory %s got removed.", p);
+                free(p);
+        }
+}
+
+static void add_root_wd(sd_journal *j, const char *p) {
+        int wd;
+        char *k;
+
+        assert(j);
+        assert(p);
+
+        wd = inotify_add_watch(j->inotify_fd, p,
+                               IN_CREATE|IN_MOVED_TO|IN_MODIFY|IN_ATTRIB|IN_DELETE|
+                               IN_DONT_FOLLOW|IN_ONLYDIR);
+        if (wd <= 0)
+                return;
+
+        k = strdup(p);
+        if (!k || hashmap_put(j->inotify_wd_roots, INT_TO_PTR(wd), k) < 0) {
+                inotify_rm_watch(j->inotify_fd, wd);
+                free(k);
+        }
+}
+
+static void remove_root_wd(sd_journal *j, int wd) {
+        char *p;
+
+        assert(j);
+        assert(wd > 0);
+
+        if (j->inotify_fd >= 0)
+                inotify_rm_watch(j->inotify_fd, wd);
+
+        p = hashmap_remove(j->inotify_wd_roots, INT_TO_PTR(wd));
+
+        if (p) {
+                log_debug("Root %s got removed.", p);
+                free(p);
+        }
+}
+
 int sd_journal_open(sd_journal **ret) {
         sd_journal *j;
         const char *p;
@@ -1025,12 +1138,26 @@ int sd_journal_open(sd_journal **ret) {
         if (!j)
                 return -ENOMEM;
 
+        j->inotify_fd = inotify_init1(IN_NONBLOCK|IN_CLOEXEC);
+        if (j->inotify_fd < 0) {
+                r = -errno;
+                goto fail;
+        }
+
         j->files = hashmap_new(string_hash_func, string_compare_func);
         if (!j->files) {
                 r = -ENOMEM;
                 goto fail;
         }
 
+        j->inotify_wd_dirs = hashmap_new(trivial_hash_func, trivial_compare_func);
+        j->inotify_wd_roots = hashmap_new(trivial_hash_func, trivial_compare_func);
+
+        if (!j->inotify_wd_dirs || !j->inotify_wd_roots) {
+                r = -ENOMEM;
+                goto fail;
+        }
+
         /* We ignore most errors here, since the idea is to only open
          * what's actually accessible, and ignore the rest. */
 
@@ -1044,6 +1171,8 @@ int sd_journal_open(sd_journal **ret) {
                         continue;
                 }
 
+                add_root_wd(j, p);
+
                 for (;;) {
                         struct dirent buf, *de;
                         sd_id128_t id;
@@ -1081,6 +1210,24 @@ fail:
 void sd_journal_close(sd_journal *j) {
         assert(j);
 
+        if (j->inotify_wd_dirs) {
+                void *k;
+
+                while ((k = hashmap_first_key(j->inotify_wd_dirs)))
+                        remove_directory_wd(j, PTR_TO_INT(k));
+
+                hashmap_free(j->inotify_wd_dirs);
+        }
+
+        if (j->inotify_wd_roots) {
+                void *k;
+
+                while ((k = hashmap_first_key(j->inotify_wd_roots)))
+                        remove_root_wd(j, PTR_TO_INT(k));
+
+                hashmap_free(j->inotify_wd_roots);
+        }
+
         if (j->files) {
                 JournalFile *f;
 
@@ -1092,6 +1239,9 @@ void sd_journal_close(sd_journal *j) {
 
         sd_journal_flush_matches(j);
 
+        if (j->inotify_fd >= 0)
+                close_nointr_nofail(j->inotify_fd);
+
         free(j);
 }
 
@@ -1275,3 +1425,119 @@ void sd_journal_restart_data(sd_journal *j) {
 
         j->current_field = 0;
 }
+
+int sd_journal_get_fd(sd_journal *j) {
+        assert(j);
+
+        return j->inotify_fd;
+}
+
+static void process_inotify_event(sd_journal *j, struct inotify_event *e) {
+        char *p;
+        int r;
+
+        assert(j);
+        assert(e);
+
+        /* Is this a subdirectory we watch? */
+        p = hashmap_get(j->inotify_wd_dirs, INT_TO_PTR(e->wd));
+        if (p) {
+
+                if (!(e->mask & IN_ISDIR) && e->len > 0 && endswith(e->name, ".journal")) {
+
+                        /* Event for a journal file */
+
+                        if (e->mask & (IN_CREATE|IN_MOVED_TO|IN_MODIFY|IN_ATTRIB)) {
+                                r = add_file(j, p, NULL, e->name);
+                                if (r < 0)
+                                        log_debug("Failed to add file %s/%s: %s", p, e->name, strerror(-r));
+                        } else if (e->mask & (IN_DELETE|IN_UNMOUNT)) {
+
+                                r = remove_file(j, p, NULL, e->name);
+                                if (r < 0)
+                                        log_debug("Failed to remove file %s/%s: %s", p, e->name, strerror(-r));
+                        }
+
+                } else if (e->len == 0) {
+
+                        /* Event for the directory itself */
+
+                        if (e->mask & (IN_DELETE_SELF|IN_MOVE_SELF|IN_UNMOUNT))
+                                remove_directory_wd(j, e->wd);
+                }
+
+                return;
+        }
+
+        /* Must be the root directory then? */
+        p = hashmap_get(j->inotify_wd_roots, INT_TO_PTR(e->wd));
+        if (p) {
+                sd_id128_t id;
+
+                if (!(e->mask & IN_ISDIR) && e->len > 0 && endswith(e->name, ".journal")) {
+
+                        /* Event for a journal file */
+
+                        if (e->mask & (IN_CREATE|IN_MOVED_TO|IN_MODIFY|IN_ATTRIB)) {
+                                r = add_file(j, p, NULL, e->name);
+                                if (r < 0)
+                                        log_debug("Failed to add file %s/%s: %s", p, e->name, strerror(-r));
+                        } else if (e->mask & (IN_DELETE|IN_UNMOUNT)) {
+
+                                r = remove_file(j, p, NULL, e->name);
+                                if (r < 0)
+                                        log_debug("Failed to remove file %s/%s: %s", p, e->name, strerror(-r));
+                        }
+
+                } else if ((e->mask & IN_ISDIR) && e->len > 0 && sd_id128_from_string(e->name, &id) >= 0) {
+
+                        /* Event for subdirectory */
+
+                        if (e->mask & (IN_CREATE|IN_MOVED_TO|IN_MODIFY|IN_ATTRIB)) {
+
+                                r = add_directory(j, p, e->name);
+                                if (r < 0)
+                                        log_debug("Failed to add directory %s/%s: %s", p, e->name, strerror(-r));
+                        }
+                }
+
+                return;
+        }
+
+        if (e->mask & IN_IGNORED)
+                return;
+
+        log_warning("Unknown inotify event.");
+}
+
+int sd_journal_process(sd_journal *j) {
+        uint8_t buffer[sizeof(struct inotify_event) + FILENAME_MAX];
+
+        assert(j);
+
+        for (;;) {
+                struct inotify_event *e;
+                ssize_t l;
+
+                l = read(j->inotify_fd, buffer, sizeof(buffer));
+                if (l < 0) {
+                        if (errno == EINTR || errno == EAGAIN)
+                                return 0;
+
+                        return -errno;
+                }
+
+                e = (struct inotify_event*) buffer;
+                while (l > 0) {
+                        size_t step;
+
+                        process_inotify_event(j, e);
+
+                        step = sizeof(struct inotify_event) + e->len;
+                        assert(step <= (size_t) l);
+
+                        e = (struct inotify_event*) ((uint8_t*) e + step);
+                        l -= step;
+                }
+        }
+}

src/journal/sd-journal.h

@@ -32,7 +32,6 @@
 /* TODO:
  *
  *   - check LE/BE conversion for 8bit, 16bit, 32bit values
- *   - implement inotify usage on client
  *   - implement audit gateway
  *   - implement stdout gateway
  *   - extend hash tables table as we go
@@ -40,7 +39,7 @@
  *   - throttling
  *   - cryptographic hash
  *   - fix space reservation logic
- *   - comm, argv can be manipulated, should it be _COMM=, _CMDLINE= or COMM=, CMDLINE=?
+ *   - compression
  */
 
 /* Write to daemon */
@@ -92,16 +91,16 @@ enum {
         SD_JOURNAL_INVALIDATE_REMOVE
 };
 
-int sd_journal_get_fd(sd_journal *j);                              /* missing */
+int sd_journal_get_fd(sd_journal *j);
-int sd_journal_process(sd_journal *j);                             /* missing */
+int sd_journal_process(sd_journal *j);
 
 #define SD_JOURNAL_FOREACH(j)                                           \
         if (sd_journal_seek_head(j) >= 0)                               \
-                while (sd_journal_next(j) > 0)                          \
+                while (sd_journal_next(j) > 0)
 
 #define SD_JOURNAL_FOREACH_BACKWARDS(j)                                 \
         if (sd_journal_seek_tail(j) >= 0)                               \
-                while (sd_journal_previous(j) > 0)                      \
+                while (sd_journal_previous(j) > 0)
 
 #define SD_JOURNAL_FOREACH_DATA(j, data, l)                             \
         for (sd_journal_restart_data(j); sd_journal_enumerate_data((j), &(data), &(l)) > 0; )

src/util.c

@@ -2674,7 +2674,7 @@ int acquire_terminal(const char *name, bool fail, bool force, bool ignore_tiocst
                         ssize_t l;
                         struct inotify_event *e;
 
-                        if ((l = read(notify, &inotify_buffer, sizeof(inotify_buffer))) < 0) {
+                        if ((l = read(notify, inotify_buffer, sizeof(inotify_buffer))) < 0) {
 
                                 if (errno == EINTR)
                                         continue;