journal: add inline compression support with XZ

9 files changed, 187 insertions, 13 deletions

diff --git a/Makefile.am b/Makefile.am
index f9093b6d8e..7d551a9dcf 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -1028,6 +1028,15 @@ test_journal_CFLAGS = \
 test_journal_LDADD = \
 	libsystemd-basic.la
 
+if HAVE_XZ
+test_journal_SOURCES += \
+        src/journal/compress.c
+test_journal_CFLAGS += \
+        $(XZ_CFLAGS)
+test_journal_LDADD += \
+        $(XZ_LIBS)
+endif
+
 systemd_journald_SOURCES = \
 	src/journal/journald.c \
 	src/journal/sd-journal.c \
@@ -1046,6 +1055,15 @@ systemd_journald_LDADD = \
         libsystemd-daemon.la \
         $(ACL_LIBS)
 
+if HAVE_XZ
+systemd_journald_SOURCES += \
+        src/journal/compress.c
+systemd_journald_CFLAGS += \
+        $(XZ_CFLAGS)
+systemd_journald_LDADD += \
+        $(XZ_LIBS)
+endif
+
 systemd_journalctl_SOURCES = \
 	src/journal/journalctl.c \
 	src/journal/sd-journal.c \
@@ -1059,6 +1077,15 @@ systemd_journalctl_CFLAGS = \
 systemd_journalctl_LDADD = \
 	libsystemd-basic.la
 
+if HAVE_XZ
+systemd_journalctl_SOURCES += \
+        src/journal/compress.c
+systemd_journalctl_CFLAGS += \
+        $(XZ_CFLAGS)
+systemd_journalctl_LDADD += \
+        $(XZ_LIBS)
+endif
+
 systemd_stdout_syslog_bridge_SOURCES = \
 	src/stdout-syslog-bridge.c \
 	src/tcpwrap.c
diff --git a/TODO b/TODO
index 5b64f048e5..8daf79a7f0 100644
--- a/TODO
+++ b/TODO
@@ -21,6 +21,8 @@ Bugfixes:
 
 Features:
 
+* logind: sends SessionNew on Lock()?
+
 * logind: allow showing logout dialog from system
 
 * document that %% can be used to write % in a string that is specifier extended
diff --git a/configure.ac b/configure.ac
index 596a32ac23..a2e9c7d65c 100644
--- a/configure.ac
+++ b/configure.ac
@@ -149,6 +149,19 @@ if test "x$enable_selinux" != "xno"; then
 fi
 AM_CONDITIONAL(HAVE_SELINUX, [test "$have_selinux" = "yes"])
 
+have_xz=no
+AC_ARG_ENABLE(xz, AS_HELP_STRING([--disable-xz], [Disable optional XZ support]))
+if test "x$enable_xz" != "xno"; then
+        PKG_CHECK_MODULES(XZ, [ liblzma ],
+                [AC_DEFINE(HAVE_XZ, 1, [Define if XZ is available]) have_xz=yes], have_xz=no)
+        AC_SUBST(XZ_CFLAGS)
+        AC_SUBST(XZ_LIBS)
+        if test "x$have_xz" = xno -a "x$enable_xz" = xyes; then
+                AC_MSG_ERROR([*** Xz support requested but libraries not found])
+        fi
+fi
+AM_CONDITIONAL(HAVE_XZ, [test "$have_xz" = "yes"])
+
 AC_ARG_ENABLE([tcpwrap],
         AS_HELP_STRING([--disable-tcpwrap],[Disable optional TCP wrappers support]),
                 [case "${enableval}" in
@@ -591,6 +604,7 @@ AC_MSG_RESULT([
         PAM:                     ${have_pam}
         AUDIT:                   ${have_audit}
         SELinux:                 ${have_selinux}
+        XZ:                      ${have_xz}
         ACL:                     ${have_acl}
         binfmt:                  ${have_binfmt}
         hostnamed:               ${have_hostnamed}
diff --git a/src/journal/journal-def.h b/src/journal/journal-def.h
index 5f026ee0f6..ef0cb6dae6 100644
--- a/src/journal/journal-def.h
+++ b/src/journal/journal-def.h
@@ -50,9 +50,15 @@ enum {
         _OBJECT_TYPE_MAX
 };
 
+/* Object flags */
+enum {
+        OBJECT_COMPRESSED = 1
+};
+
 _packed_ struct ObjectHeader {
         uint8_t type;
-        uint8_t reserved[7];
+        uint8_t flags;
+        uint8_t reserved[6];
         uint64_t size;
         uint8_t payload[];
 };
@@ -123,6 +129,11 @@ enum {
         STATE_ARCHIVED
 };
 
+/* Header flags */
+enum {
+        HEADER_INCOMPATIBLE_COMPRESSED = 1
+};
+
 _packed_ struct Header {
         uint8_t signature[8]; /* "LPKSHHRH" */
         uint32_t compatible_flags;
diff --git a/src/journal/journal-file.c b/src/journal/journal-file.c
index 8f9b61bc2f..a0c479fc67 100644
--- a/src/journal/journal-file.c
+++ b/src/journal/journal-file.c
@@ -30,12 +30,15 @@
 #include "journal-def.h"
 #include "journal-file.h"
 #include "lookup3.h"
+#include "compress.h"
 
 #define DEFAULT_DATA_HASH_TABLE_SIZE (2047ULL*16ULL)
 #define DEFAULT_FIELD_HASH_TABLE_SIZE (2047ULL*16ULL)
 
 #define DEFAULT_WINDOW_SIZE (128ULL*1024ULL*1024ULL)
 
+#define COMPRESSION_SIZE_THRESHOLD (64ULL)
+
 static const char signature[] = { 'L', 'P', 'K', 'S', 'H', 'H', 'R', 'H' };
 
 #define ALIGN64(x) (((x) + 7ULL) & ~7ULL)
@@ -57,6 +60,11 @@ void journal_file_close(JournalFile *f) {
                 close_nointr_nofail(f->fd);
 
         free(f->path);
+
+#ifdef HAVE_XZ
+        free(f->compress_buffer);
+#endif
+
         free(f);
 }
 
@@ -120,8 +128,13 @@ static int journal_file_verify_header(JournalFile *f) {
         if (memcmp(f->header, signature, 8))
                 return -EBADMSG;
 
+#ifdef HAVE_XZ
+        if ((le64toh(f->header->incompatible_flags) & ~HEADER_INCOMPATIBLE_COMPRESSED) != 0)
+                return -EPROTONOSUPPORT;
+#else
         if (f->header->incompatible_flags != 0)
                 return -EPROTONOSUPPORT;
+#endif
 
         if ((uint64_t) f->last_stat.st_size < (le64toh(f->header->arena_offset) + le64toh(f->header->arena_size)))
                 return -ENODATA;
@@ -309,7 +322,7 @@ static bool verify_hash(Object *o) {
 
         assert(o);
 
-        if (o->object.type == OBJECT_DATA) {
+        if (o->object.type == OBJECT_DATA && !(o->object.flags & OBJECT_COMPRESSED)) {
                 h1 = le64toh(o->data.hash);
                 h2 = hash64(o->data.payload, le64toh(o->object.size) - offsetof(Object, data.payload));
         } else if (o->object.type == OBJECT_FIELD) {
@@ -581,12 +594,40 @@ int journal_file_find_data_object_with_hash(
                 if (r < 0)
                         return r;
 
-                if (le64toh(o->object.size) == osize &&
-                    memcmp(o->data.payload, data, size) == 0) {
+                if (le64toh(o->data.hash) != hash)
+                        return -EBADMSG;
+
+                if (o->object.flags & OBJECT_COMPRESSED) {
+#ifdef HAVE_XZ
+                        uint64_t l, rsize;
 
-                        if (le64toh(o->data.hash) != hash)
+                        l = le64toh(o->object.size);
+                        if (l <= offsetof(Object, data.payload))
                                 return -EBADMSG;
 
+                        l -= offsetof(Object, data.payload);
+
+                        if (!uncompress_blob(o->data.payload, l, &f->compress_buffer, &f->compress_buffer_size, &rsize))
+                                return -EBADMSG;
+
+                        if (rsize == size &&
+                            memcmp(f->compress_buffer, data, size) == 0) {
+
+                                if (ret)
+                                        *ret = o;
+
+                                if (offset)
+                                        *offset = p;
+
+                                return 1;
+                        }
+#else
+                        return -EPROTONOSUPPORT;
+#endif
+
+                } else if (le64toh(o->object.size) == osize &&
+                           memcmp(o->data.payload, data, size) == 0) {
+
                         if (ret)
                                 *ret = o;
 
@@ -624,6 +665,7 @@ static int journal_file_append_data(JournalFile *f, const void *data, uint64_t s
         uint64_t osize;
         Object *o;
         int r;
+        bool compressed = false;
 
         assert(f);
         assert(data || size == 0);
@@ -650,7 +692,27 @@ static int journal_file_append_data(JournalFile *f, const void *data, uint64_t s
                 return r;
 
         o->data.hash = htole64(hash);
-        memcpy(o->data.payload, data, size);
+
+#ifdef HAVE_XZ
+        if (f->compress &&
+            size >= COMPRESSION_SIZE_THRESHOLD) {
+                uint64_t rsize;
+
+                compressed = compress_blob(data, size, o->data.payload, &rsize);
+
+                if (compressed) {
+                        o->object.size = htole64(offsetof(Object, data.payload) + rsize);
+                        o->object.flags |= OBJECT_COMPRESSED;
+
+                        f->header->incompatible_flags = htole32(le32toh(f->header->incompatible_flags) | HEADER_INCOMPATIBLE_COMPRESSED);
+
+                        log_debug("Compressed data object %lu -> %lu", (unsigned long) size, (unsigned long) rsize);
+                }
+        }
+#endif
+
+        if (!compressed)
+                memcpy(o->data.payload, data, size);
 
         r = journal_file_link_data(f, o, p, hash);
         if (r < 0)
@@ -1585,6 +1647,9 @@ void journal_file_dump(JournalFile *f) {
                         break;
                 }
 
+                if (o->object.flags & OBJECT_COMPRESSED)
+                        printf("Flags: COMPRESSED\n");
+
                 if (p == le64toh(f->header->tail_object_offset))
                         p = 0;
                 else
diff --git a/src/journal/journal-file.h b/src/journal/journal-file.h
index 20712b5dcc..421dfa6766 100644
--- a/src/journal/journal-file.h
+++ b/src/journal/journal-file.h
@@ -75,6 +75,13 @@ typedef struct JournalFile {
         uint64_t current_offset;
 
         JournalMetrics metrics;
+
+        bool compress;
+
+#ifdef HAVE_XZ
+        void *compress_buffer;
+        size_t compress_buffer_size;
+#endif
 } JournalFile;
 
 typedef enum direction {
diff --git a/src/journal/journald.c b/src/journal/journald.c
index 37f8f16754..ca274ee44a 100644
--- a/src/journal/journald.c
+++ b/src/journal/journald.c
@@ -57,6 +57,7 @@ typedef struct Server {
 
         JournalMetrics metrics;
         uint64_t max_use;
+        bool compress;
 } Server;
 
 static void fix_perms(JournalFile *f, uid_t uid) {
@@ -146,6 +147,8 @@ static JournalFile* find_journal(Server *s, uid_t uid) {
                 return s->system_journal;
 
         fix_perms(f, uid);
+        f->metrics = s->metrics;
+        f->compress = s->compress;
 
         r = hashmap_put(s->user_journals, UINT32_TO_PTR(uid), f);
         if (r < 0) {
@@ -661,6 +664,9 @@ static int system_journal_open(Server *s) {
         free(fn);
 
         if (r >= 0) {
+                s->system_journal->metrics = s->metrics;
+                s->system_journal->compress = s->compress;
+
                 fix_perms(s->system_journal, 0);
                 return r;
         }
@@ -685,6 +691,9 @@ static int system_journal_open(Server *s) {
                 return r;
         }
 
+        s->runtime_journal->metrics = s->metrics;
+        s->runtime_journal->compress = s->compress;
+
         fix_perms(s->runtime_journal, 0);
         return r;
 }
@@ -794,6 +803,7 @@ static int server_init(Server *s) {
         s->metrics.min_size = DEFAULT_MIN_SIZE;
         s->metrics.keep_free = DEFAULT_KEEP_FREE;
         s->max_use = DEFAULT_MAX_USE;
+        s->compress = true;
 
         s->epoll_fd = epoll_create1(EPOLL_CLOEXEC);
         if (s->epoll_fd < 0) {
@@ -931,6 +941,7 @@ int main(int argc, char *argv[]) {
         }
 
         log_set_target(LOG_TARGET_CONSOLE);
+        log_set_max_level(LOG_DEBUG);
         log_parse_environment();
         log_open();
 
diff --git a/src/journal/sd-journal.c b/src/journal/sd-journal.c
index 9dff72429b..bc575b43ef 100644
--- a/src/journal/sd-journal.c
+++ b/src/journal/sd-journal.c
@@ -31,6 +31,7 @@
 #include "hashmap.h"
 #include "list.h"
 #include "lookup3.h"
+#include "compress.h"
 
 #define JOURNAL_FILES_MAX 1024
 
@@ -1344,7 +1345,7 @@ int sd_journal_get_data(sd_journal *j, const char *field, const void **data, siz
                 size_t t;
 
                 p = le64toh(o->entry.items[i].object_offset);
-                le_hash = o->entry.items[j->current_field].hash;
+                le_hash = o->entry.items[i].hash;
                 r = journal_file_move_to_object(f, OBJECT_DATA, p, &o);
                 if (r < 0)
                         return r;
@@ -1354,9 +1355,31 @@ int sd_journal_get_data(sd_journal *j, const char *field, const void **data, siz
 
                 l = le64toh(o->object.size) - offsetof(Object, data.payload);
 
-                if (l >= field_length+1 &&
-                    memcmp(o->data.payload, field, field_length) == 0 &&
-                    o->data.payload[field_length] == '=') {
+                if (o->object.flags & OBJECT_COMPRESSED) {
+
+#ifdef HAVE_XZ
+                        if (uncompress_startswith(o->data.payload, l,
+                                                  &f->compress_buffer, &f->compress_buffer_size,
+                                                  field, field_length, '=')) {
+
+                                uint64_t rsize;
+
+                                if (!uncompress_blob(o->data.payload, l,
+                                                     &f->compress_buffer, &f->compress_buffer_size, &rsize))
+                                        return -EBADMSG;
+
+                                *data = f->compress_buffer;
+                                *size = (size_t) rsize;
+
+                                return 0;
+                        }
+#else
+                        return -EPROTONOSUPPORT;
+#endif
+
+                } else if (l >= field_length+1 &&
+                           memcmp(o->data.payload, field, field_length) == 0 &&
+                           o->data.payload[field_length] == '=') {
 
                         t = (size_t) l;
 
@@ -1419,8 +1442,22 @@ int sd_journal_enumerate_data(sd_journal *j, const void **data, size_t *size) {
         if ((uint64_t) t != l)
                 return -E2BIG;
 
-        *data = o->data.payload;
-        *size = t;
+        if (o->object.flags & OBJECT_COMPRESSED) {
+#ifdef HAVE_XZ
+                uint64_t rsize;
+
+                if (!uncompress_blob(o->data.payload, l, &f->compress_buffer, &f->compress_buffer_size, &rsize))
+                        return -EBADMSG;
+
+                *data = f->compress_buffer;
+                *size = (size_t) rsize;
+#else
+                return -EPROTONOSUPPORT;
+#endif
+        } else {
+                *data = o->data.payload;
+                *size = t;
+        }
 
         j->current_field ++;
 
diff --git a/src/journal/sd-journal.h b/src/journal/sd-journal.h
index ee9813f28c..b29680b3a2 100644
--- a/src/journal/sd-journal.h
+++ b/src/journal/sd-journal.h
@@ -38,7 +38,7 @@
  *   - accelerate looking for "all hostnames" and suchlike.
  *   - throttling
  *   - cryptographic hash
- *   - compression
+ *   - never access beyond fle size check
  */
 
 /* Write to daemon */