Merge pull request #27126 from yuwata/journal-compress

sd-journal: allow to specify compression algorithm through env

docs/ENVIRONMENT.md

@@ -515,6 +515,14 @@ SYSTEMD_HOME_DEBUG_SUFFIX=foo \
   journal. Note that journal files in compact mode are limited to 4G to allow use of
   32-bit offsets. Enabled by default.
 
+* `$SYSTEMD_JOURNAL_COMPRESS` – Takes a boolean, or one of the compression
+  algorithms "XZ", "LZ4", and "ZSTD". If enabled, the default compression
+  algorithm set at compile time will be used when opening a new journal file.
+  If disabled, the journal file compression will be disabled. Note that the
+  compression mode of existing journal files are not changed. To make the
+  specified algorithm takes an effect immediately, you need to explicitly run
+  `journalctl --rotate`.
+
 `systemd-pcrphase`, `systemd-cryptsetup`:
 
 * `$SYSTEMD_FORCE_MEASURE=1` — If set, force measuring of resources (which are

src/basic/compress.c

@@ -65,6 +65,16 @@ static const char* const compression_table[_COMPRESSION_MAX] = {
 
 DEFINE_STRING_TABLE_LOOKUP(compression, Compression);
 
+bool compression_supported(Compression c) {
+        static const unsigned supported =
+                (1U << COMPRESSION_NONE) |
+                (1U << COMPRESSION_XZ) * HAVE_XZ |
+                (1U << COMPRESSION_LZ4) * HAVE_LZ4 |
+                (1U << COMPRESSION_ZSTD) * HAVE_ZSTD;
+
+        return c >= 0 && c < _COMPRESSION_MAX && FLAGS_SET(supported, 1U << c);
+}
+
 int compress_blob_xz(const void *src, uint64_t src_size,
                      void *dst, size_t dst_alloc_size, size_t *dst_size) {
 #if HAVE_XZ

src/basic/compress.h

@@ -2,6 +2,7 @@
 #pragma once
 
 #include <errno.h>
+#include <stdbool.h>
 #include <stdint.h>
 #include <unistd.h>
 
@@ -17,6 +18,8 @@ typedef enum Compression {
 const char* compression_to_string(Compression compression);
 Compression compression_from_string(const char *compression);
 
+bool compression_supported(Compression c);
+
 int compress_blob_xz(const void *src, uint64_t src_size,
                      void *dst, size_t dst_alloc_size, size_t *dst_size);
 int compress_blob_lz4(const void *src, uint64_t src_size,

src/journal/journald-server.c

@@ -2124,7 +2124,7 @@ static int vl_method_synchronize(Varlink *link, JsonVariant *parameters, Varlink
         if (json_variant_elements(parameters) > 0)
                 return varlink_error_invalid_parameter(link, parameters);
 
-        log_info("Received client request to rotate journal.");
+        log_info("Received client request to sync journal.");
 
         /* We don't do the main work now, but instead enqueue a deferred event loop job which will do
          * it. That job is scheduled at low priority, so that we return from this method call only after all

src/libsystemd/sd-journal/journal-file.c

@@ -27,6 +27,7 @@
 #include "journal-internal.h"
 #include "lookup3.h"
 #include "memory-util.h"
+#include "missing_threads.h"
 #include "path-util.h"
 #include "prioq.h"
 #include "random-util.h"
@@ -313,27 +314,79 @@ JournalFile* journal_file_close(JournalFile *f) {
 }
 
 static bool keyed_hash_requested(void) {
+        static thread_local int cached = -1;
         int r;
 
-        r = getenv_bool("SYSTEMD_JOURNAL_KEYED_HASH");
+        if (cached < 0) {
-        if (r >= 0)
+                r = getenv_bool("SYSTEMD_JOURNAL_KEYED_HASH");
-                return r;
+                if (r < 0) {
-        if (r != -ENXIO)
+                        if (r != -ENXIO)
-                log_debug_errno(r, "Failed to parse $SYSTEMD_JOURNAL_KEYED_HASH environment variable, ignoring: %m");
+                                log_debug_errno(r, "Failed to parse $SYSTEMD_JOURNAL_KEYED_HASH environment variable, ignoring: %m");
+                        cached = true;
+                } else
+                        cached = r;
+        }
 
-        return true;
+        return cached;
 }
 
 static bool compact_mode_requested(void) {
+        static thread_local int cached = -1;
         int r;
 
-        r = getenv_bool("SYSTEMD_JOURNAL_COMPACT");
+        if (cached < 0) {
-        if (r >= 0)
+                r = getenv_bool("SYSTEMD_JOURNAL_COMPACT");
-                return r;
+                if (r < 0) {
-        if (r != -ENXIO)
+                        if (r != -ENXIO)
-                log_debug_errno(r, "Failed to parse $SYSTEMD_JOURNAL_COMPACT environment variable, ignoring: %m");
+                                log_debug_errno(r, "Failed to parse $SYSTEMD_JOURNAL_COMPACT environment variable, ignoring: %m");
+                        cached = true;
+                } else
+                        cached = r;
+        }
 
-        return true;
+        return cached;
+}
+
+#if HAVE_COMPRESSION
+static Compression getenv_compression(void) {
+        Compression c;
+        const char *e;
+        int r;
+
+        e = getenv("SYSTEMD_JOURNAL_COMPRESS");
+        if (!e)
+                return DEFAULT_COMPRESSION;
+
+        r = parse_boolean(e);
+        if (r >= 0)
+                return r ? DEFAULT_COMPRESSION : COMPRESSION_NONE;
+
+        c = compression_from_string(e);
+        if (c < 0) {
+                log_debug_errno(c, "Failed to parse SYSTEMD_JOURNAL_COMPRESS value, ignoring: %s", e);
+                return DEFAULT_COMPRESSION;
+        }
+
+        if (!compression_supported(c)) {
+                log_debug("Unsupported compression algorithm specified, ignoring: %s", e);
+                return DEFAULT_COMPRESSION;
+        }
+
+        return c;
+}
+#endif
+
+static Compression compression_requested(void) {
+#if HAVE_COMPRESSION
+        static thread_local Compression cached = _COMPRESSION_INVALID;
+
+        if (cached < 0)
+                cached = getenv_compression();
+
+        return cached;
+#else
+        return COMPRESSION_NONE;
+#endif
 }
 
 static int journal_file_init_header(
@@ -355,7 +408,7 @@ static int journal_file_init_header(
         Header h = {
                 .header_size = htole64(ALIGN64(sizeof(h))),
                 .incompatible_flags = htole32(
-                                FLAGS_SET(file_flags, JOURNAL_COMPRESS) * COMPRESSION_TO_HEADER_INCOMPATIBLE_FLAG(DEFAULT_COMPRESSION) |
+                                FLAGS_SET(file_flags, JOURNAL_COMPRESS) * COMPRESSION_TO_HEADER_INCOMPATIBLE_FLAG(compression_requested()) |
                                 keyed_hash_requested() * HEADER_INCOMPATIBLE_KEYED_HASH |
                                 compact_mode_requested() * HEADER_INCOMPATIBLE_COMPACT),
                 .compatible_flags = htole32(
@@ -1619,24 +1672,31 @@ static int journal_file_append_field(
 }
 
 static Compression maybe_compress_payload(JournalFile *f, uint8_t *dst, const uint8_t *src, uint64_t size, size_t *rsize) {
-        Compression compression = COMPRESSION_NONE;
-
         assert(f);
         assert(f->header);
 
 #if HAVE_COMPRESSION
-        if (JOURNAL_FILE_COMPRESS(f) && size >= f->compress_threshold_bytes) {
+        Compression c;
-                compression = compress_blob(src, size, dst, size - 1, rsize);
+        int r;
-                if (compression > 0)
-                        log_debug("Compressed data object %"PRIu64" -> %zu using %s",
-                                  size, *rsize, compression_to_string(compression));
-                else
-                        /* Compression didn't work, we don't really care why, let's continue without compression */
-                        compression = COMPRESSION_NONE;
-        }
-#endif
 
-        return compression;
+        c = JOURNAL_FILE_COMPRESSION(f);
+        if (c == COMPRESSION_NONE || size < f->compress_threshold_bytes)
+                return COMPRESSION_NONE;
+
+        r = compress_blob_explicit(c, src, size, dst, size - 1, rsize);
+        if (r < 0) {
+                log_debug_errno(r, "Failed to compress data object using %s, ignoring: %m", compression_to_string(c));
+                /* Compression didn't work, we don't really care why, let's continue without compression */
+                return COMPRESSION_NONE;
+        }
+
+        assert(r == c);
+        log_debug("Compressed data object %"PRIu64" -> %zu using %s", size, *rsize, compression_to_string(c));
+
+        return c;
+#else
+        return COMPRESSION_NONE;
+#endif
 }
 
 static int journal_file_append_data(
@@ -3956,20 +4016,21 @@ int journal_file_open(
         f->close_fd = true;
 
         if (DEBUG_LOGGING) {
-                static int last_seal = -1, last_compress = -1, last_keyed_hash = -1;
+                static int last_seal = -1, last_keyed_hash = -1;
+                static Compression last_compression = _COMPRESSION_INVALID;
                 static uint64_t last_bytes = UINT64_MAX;
 
                 if (last_seal != JOURNAL_HEADER_SEALED(f->header) ||
                     last_keyed_hash != JOURNAL_HEADER_KEYED_HASH(f->header) ||
-                    last_compress != JOURNAL_FILE_COMPRESS(f) ||
+                    last_compression != JOURNAL_FILE_COMPRESSION(f) ||
                     last_bytes != f->compress_threshold_bytes) {
 
                         log_debug("Journal effective settings seal=%s keyed_hash=%s compress=%s compress_threshold_bytes=%s",
                                   yes_no(JOURNAL_HEADER_SEALED(f->header)), yes_no(JOURNAL_HEADER_KEYED_HASH(f->header)),
-                                  yes_no(JOURNAL_FILE_COMPRESS(f)), FORMAT_BYTES(f->compress_threshold_bytes));
+                                  compression_to_string(JOURNAL_FILE_COMPRESSION(f)), FORMAT_BYTES(f->compress_threshold_bytes));
                         last_seal = JOURNAL_HEADER_SEALED(f->header);
                         last_keyed_hash = JOURNAL_HEADER_KEYED_HASH(f->header);
-                        last_compress = JOURNAL_FILE_COMPRESS(f);
+                        last_compression = JOURNAL_FILE_COMPRESSION(f);
                         last_bytes = f->compress_threshold_bytes;
                 }
         }

src/libsystemd/sd-journal/journal-file.h

@@ -321,10 +321,16 @@ bool journal_file_rotate_suggested(JournalFile *f, usec_t max_file_usec, int log
 int journal_file_map_data_hash_table(JournalFile *f);
 int journal_file_map_field_hash_table(JournalFile *f);
 
-static inline bool JOURNAL_FILE_COMPRESS(JournalFile *f) {
+static inline Compression JOURNAL_FILE_COMPRESSION(JournalFile *f) {
         assert(f);
-        return JOURNAL_HEADER_COMPRESSED_XZ(f->header) || JOURNAL_HEADER_COMPRESSED_LZ4(f->header) ||
+
-                        JOURNAL_HEADER_COMPRESSED_ZSTD(f->header);
+        if (JOURNAL_HEADER_COMPRESSED_XZ(f->header))
+                return COMPRESSION_XZ;
+        if (JOURNAL_HEADER_COMPRESSED_LZ4(f->header))
+                return COMPRESSION_LZ4;
+        if (JOURNAL_HEADER_COMPRESSED_ZSTD(f->header))
+                return COMPRESSION_ZSTD;
+        return COMPRESSION_NONE;
 }
 
 uint64_t journal_file_hash_data(JournalFile *f, const void *data, size_t sz);

test/units/testsuite-04.sh

@@ -290,4 +290,36 @@ rm -rf "$JTMP"
 
 rm /tmp/lb1
 
+# https://bugzilla.redhat.com/show_bug.cgi?id=2183546
+mkdir /run/systemd/system/systemd-journald.service.d
+MID=$(cat /etc/machine-id)
+for c in "NONE" "XZ" "LZ4" "ZSTD"; do
+    cat >/run/systemd/system/systemd-journald.service.d/compress.conf <<EOF
+[Service]
+Environment=SYSTEMD_JOURNAL_COMPRESS=${c}
+EOF
+    systemctl daemon-reload
+    systemctl restart systemd-journald.service
+    journalctl --rotate
+
+    ID=$(systemd-id128 new)
+    systemd-cat -t "$ID" /bin/bash -c "for ((i=0;i<100;i++)); do echo -n hoge with ${c}; done; echo"
+    journalctl --sync
+    timeout 10 bash -c "while ! SYSTEMD_LOG_LEVEL=debug journalctl --verify --quiet --file /var/log/journal/$MID/system.journal 2>&1 | grep -q -F 'compress=${c}'; do sleep .5; done"
+
+    # $SYSTEMD_JOURNAL_COMPRESS= also works for journal-remote
+    if [[ -x /usr/lib/systemd/systemd-journal-remote ]]; then
+        for cc in "NONE" "XZ" "LZ4" "ZSTD"; do
+            rm -f /tmp/foo.journal
+            SYSTEMD_JOURNAL_COMPRESS="${cc}" /usr/lib/systemd/systemd-journal-remote --split-mode=none -o /tmp/foo.journal --getter="journalctl -b -o export -t $ID"
+            SYSTEMD_LOG_LEVEL=debug journalctl --verify --quiet --file /tmp/foo.journal 2>&1 | grep -q -F "compress=${cc}"
+            journalctl -t "$ID" -o cat --file /tmp/foo.journal | grep -q -F "hoge with ${c}"
+        done
+    fi
+done
+rm /run/systemd/system/systemd-journald.service.d/compress.conf
+systemctl daemon-reload
+systemctl restart systemd-journald.service
+journalctl --rotate
+
 touch /testok