journal: synchronize seqnum across files

7 files changed, 43 insertions, 11 deletions

diff --git a/src/journal/journal-file.c b/src/journal/journal-file.c
index 6c8d712dbd..537978137b 100644
--- a/src/journal/journal-file.c
+++ b/src/journal/journal-file.c
@@ -364,12 +364,24 @@ int journal_file_move_to_object(JournalFile *f, uint64_t offset, int type, Objec
         return 0;
 }
 
-static uint64_t journal_file_seqnum(JournalFile *f) {
+static uint64_t journal_file_seqnum(JournalFile *f, uint64_t *seqnum) {
         uint64_t r;
 
         assert(f);
 
         r = le64toh(f->header->seqnum) + 1;
+
+        if (seqnum) {
+                /* If an external seqno counter was passed, we update
+                 * both the local and the external one, and set it to
+                 * the maximum of both */
+
+                if (*seqnum + 1 > r)
+                        r = *seqnum + 1;
+
+                *seqnum = r;
+        }
+
         f->header->seqnum = htole64(r);
 
         return r;
@@ -733,6 +745,7 @@ static int journal_file_append_entry_internal(
                 const dual_timestamp *ts,
                 uint64_t xor_hash,
                 const EntryItem items[], unsigned n_items,
+                uint64_t *seqno,
                 Object **ret, uint64_t *offset) {
         uint64_t np;
         uint64_t osize;
@@ -749,7 +762,7 @@ static int journal_file_append_entry_internal(
                 return r;
 
         o->object.type = htole64(OBJECT_ENTRY);
-        o->entry.seqnum = htole64(journal_file_seqnum(f));
+        o->entry.seqnum = htole64(journal_file_seqnum(f, seqno));
         memcpy(o->entry.items, items, n_items * sizeof(EntryItem));
         o->entry.realtime = htole64(ts ? ts->realtime : now(CLOCK_REALTIME));
         o->entry.monotonic = htole64(ts ? ts->monotonic : now(CLOCK_MONOTONIC));
@@ -769,7 +782,7 @@ static int journal_file_append_entry_internal(
         return 0;
 }
 
-int journal_file_append_entry(JournalFile *f, const dual_timestamp *ts, const struct iovec iovec[], unsigned n_iovec, Object **ret, uint64_t *offset) {
+int journal_file_append_entry(JournalFile *f, const dual_timestamp *ts, const struct iovec iovec[], unsigned n_iovec, uint64_t *seqno, Object **ret, uint64_t *offset) {
         unsigned i;
         EntryItem *items;
         int r;
@@ -794,7 +807,7 @@ int journal_file_append_entry(JournalFile *f, const dual_timestamp *ts, const st
                 items[i].object_offset = htole64(p);
         }
 
-        r = journal_file_append_entry_internal(f, ts, xor_hash, items, n_iovec, ret, offset);
+        r = journal_file_append_entry_internal(f, ts, xor_hash, items, n_iovec, seqno, ret, offset);
 
 finish:
         free(items);
diff --git a/src/journal/journal-file.h b/src/journal/journal-file.h
index c51504d11b..92f671a756 100644
--- a/src/journal/journal-file.h
+++ b/src/journal/journal-file.h
@@ -63,7 +63,7 @@ int journal_file_move_to_object(JournalFile *f, uint64_t offset, int type, Objec
 
 uint64_t journal_file_entry_n_items(Object *o);
 
-int journal_file_append_entry(JournalFile *f, const dual_timestamp *ts, const struct iovec iovec[], unsigned n_iovec, Object **ret, uint64_t *offset);
+int journal_file_append_entry(JournalFile *f, const dual_timestamp *ts, const struct iovec iovec[], unsigned n_iovec, uint64_t *seqno, Object **ret, uint64_t *offset);
 
 int journal_file_move_to_entry(JournalFile *f, uint64_t seqnum, Object **ret, uint64_t *offset);
 
diff --git a/src/journal/journalctl.c b/src/journal/journalctl.c
index bb1f18a21f..a6b6e0fbd6 100644
--- a/src/journal/journalctl.c
+++ b/src/journal/journalctl.c
@@ -50,6 +50,7 @@ int main(int argc, char *argv[]) {
                 const void *data;
                 size_t length;
                 char *cursor;
+                uint64_t realtime = 0, monotonic = 0;
 
                 r = sd_journal_get_cursor(j, &cursor);
                 if (r < 0) {
@@ -60,6 +61,13 @@ int main(int argc, char *argv[]) {
                 printf("entry: %s\n", cursor);
                 free(cursor);
 
+                sd_journal_get_realtime_usec(j, &realtime);
+                sd_journal_get_monotonic_usec(j, &monotonic);
+                printf("realtime: %llu\n"
+                       "monotonic: %llu\n",
+                       (unsigned long long) realtime,
+                       (unsigned long long) monotonic);
+
                 SD_JOURNAL_FOREACH_FIELD(j, data, length)
                         printf("\t%.*s\n", (int) length, (const char*) data);
         }
diff --git a/src/journal/journald.c b/src/journal/journald.c
index 1143d81abc..ede314a55f 100644
--- a/src/journal/journald.c
+++ b/src/journal/journald.c
@@ -43,6 +43,8 @@ typedef struct Server {
         JournalFile *runtime_journal;
         JournalFile *system_journal;
         Hashmap *user_journals;
+
+        uint64_t seqnum;
 } Server;
 
 static void fix_perms(JournalFile *f, uid_t uid) {
@@ -118,7 +120,7 @@ static JournalFile* find_journal(Server *s, uid_t uid) {
         if (asprintf(&p, "/var/log/journal/%s/user-%lu.journal", sd_id128_to_string(machine, ids), (unsigned long) uid) < 0)
                 return s->system_journal;
 
-        r = journal_file_open(p, O_RDWR|O_CREAT, 0640, NULL, &f);
+        r = journal_file_open(p, O_RDWR|O_CREAT, 0640, s->system_journal, &f);
         free(p);
 
         if (r < 0)
@@ -252,7 +254,7 @@ static void process_message(Server *s, const char *buf, struct ucred *ucred, str
         if (!f)
                 log_warning("Dropping message, as we can't find a place to store the data.");
         else {
-                r = journal_file_append_entry(f, NULL, iovec, n, NULL, NULL);
+                r = journal_file_append_entry(f, NULL, iovec, n, &s->seqnum, NULL, NULL);
 
                 if (r < 0)
                         log_error("Failed to write entry, ignoring: %s", strerror(-r));
diff --git a/src/journal/sd-journal.c b/src/journal/sd-journal.c
index 1f4ad0ff64..1614bbf27f 100644
--- a/src/journal/sd-journal.c
+++ b/src/journal/sd-journal.c
@@ -561,3 +561,13 @@ int sd_journal_iterate_fields(sd_journal *j, const void **data, size_t *size) {
 
         return 1;
 }
+
+int sd_journal_seek_head(sd_journal *j) {
+        assert(j);
+        return -EINVAL;
+}
+
+int sd_journal_seek_tail(sd_journal *j) {
+        assert(j);
+        return -EINVAL;
+}
diff --git a/src/journal/sd-journal.h b/src/journal/sd-journal.h
index bbfcda6f2a..13b5f891d7 100644
--- a/src/journal/sd-journal.h
+++ b/src/journal/sd-journal.h
@@ -56,7 +56,6 @@ void sd_journal_flush_matches(sd_journal *j);
 int sd_journal_seek_head(sd_journal *j);
 int sd_journal_seek_tail(sd_journal *j);
 
-int sd_journal_seek_seqnum(sd_journal *j, uint64_t seqnum);
 int sd_journal_seek_monotonic_usec(sd_journal *j, uint64_t usec);
 int sd_journal_seek_realtime_usec(sd_journal *j, uint64_t usec);
 
diff --git a/src/journal/test-journal.c b/src/journal/test-journal.c
index 8dd26bbc32..3b67f1aa9f 100644
--- a/src/journal/test-journal.c
+++ b/src/journal/test-journal.c
@@ -42,15 +42,15 @@ int main(int argc, char *argv[]) {
 
         iovec.iov_base = (void*) test;
         iovec.iov_len = strlen(test);
-        assert_se(journal_file_append_entry(f, &ts, &iovec, 1, NULL, NULL) == 0);
+        assert_se(journal_file_append_entry(f, &ts, &iovec, 1, NULL, NULL, NULL) == 0);
 
         iovec.iov_base = (void*) test2;
         iovec.iov_len = strlen(test2);
-        assert_se(journal_file_append_entry(f, &ts, &iovec, 1, NULL, NULL) == 0);
+        assert_se(journal_file_append_entry(f, &ts, &iovec, 1, NULL, NULL, NULL) == 0);
 
         iovec.iov_base = (void*) test;
         iovec.iov_len = strlen(test);
-        assert_se(journal_file_append_entry(f, &ts, &iovec, 1, NULL, NULL) == 0);
+        assert_se(journal_file_append_entry(f, &ts, &iovec, 1, NULL, NULL, NULL) == 0);
 
         journal_file_dump(f);