PQ: Add support for event-level compression using ZStandard (ZSTD) (#18121)

* noop: add pq compression-codec with no-op implementation

* pq: add support for event compression using zstd

Adds non-breaking support for event compression to the persisted queue, as
configured by a new per-pipeline setting `queue.compression`, which supports:

 - `none` (default): no compression is performed, but if compressed events
                     are encountered in the queue they will be decompressed
 - `speed`: compression optimized for speed
 - `balanced`: compression balancing speed against result size
 - `size`: compression optimized for maximum reduction of size
 - `disabled`: compression support entirely disabled; if a pipeline is run
               in this configuration against a PQ that already contains
               unacked compressed events, the pipeline WILL crash.

To accomplish this, we then provide an abstract base implementation of the
CompressionCodec whose decode method is capable of _detecting_ and decoding
zstd-encoded payload while letting other payloads through unmodified.
The detection is done with an operation on the first four bytes of the
payload, so no additional context is needed.

An instance of this zstd-aware compression codec is provided with a
pass-through encode operation when configured with `queue.compression: none`,
which is the default, ensuring that by default logstash is able to decode any
event that had previously been written.

We provide an additional implementation that is capable of _encoding_ events
with a configurable goal: speed, size, or a balance of the two.

* license: add notice for `com.github.luben:zstd-jni`

* pq: log compression encode/decode from the codec

* Apply docs suggestions from code review

Co-authored-by: João Duarte <[email protected]>

* remove CleanerThreadLocal utility

* license: add mapping for com.github.luben:zstd-jni

* Apply suggestions from code review

Co-authored-by: Rob Bavey <[email protected]>

---------

Co-authored-by: João Duarte <[email protected]>
Co-authored-by: Rob Bavey <[email protected]>

Author: 3 people

config/logstash.yml

@@ -229,6 +229,10 @@
 #
 # queue.checkpoint.writes: 1024
 #
+# If using queue.type: persisted, the compression goal. Valid values are `none`, `speed`, `balanced`, and `size`.
+# The default `none` is able to decompress previously-written events, even if they were compressed.
+#
+# queue.compression: none
 #
 # ------------ Dead-Letter Queue Settings --------------
 # Flag to turn on dead-letter queue.

docker/data/logstash/env2yaml/env2yaml.go

@@ -58,6 +58,7 @@ var validSettings = []string{
 	"queue.checkpoint.acks",
 	"queue.checkpoint.writes",
 	"queue.checkpoint.interval", // remove it for #17155
+	"queue.compression",
 	"queue.drain",
 	"dead_letter_queue.enable",
 	"dead_letter_queue.max_bytes",

docs/reference/logstash-settings-file.md

@@ -68,6 +68,7 @@ The `logstash.yml` file includes these settings.
 | `queue.checkpoint.acks` | The maximum number of ACKed events before forcing a checkpoint when persistent queues are enabled (`queue.type: persisted`). Specify `queue.checkpoint.acks: 0` to set this value to unlimited. | 1024 |
 | `queue.checkpoint.writes` | The maximum number of written events before forcing a checkpoint when persistent queues are enabled (`queue.type: persisted`). Specify `queue.checkpoint.writes: 0` to set this value to unlimited. | 1024 |
 | `queue.checkpoint.retry` | When enabled, Logstash will retry four times per attempted checkpoint write for any checkpoint writes that fail. Any subsequent errors are not retried. This is a workaround for failed checkpoint writes that have been seen only on Windows platform, filesystems with non-standard behavior such as SANs and is not recommended except in those specific circumstances. (`queue.type: persisted`) | `true` |
+| `queue.compression` {applies_to}`stack: ga 9.2` | Set a persisted queue compression level, which allows the pipeline to reduce the event size on disk at the cost of CPU usage. Possible values are `speed`, `balanced`, and `size`.  | `none` |
 | `queue.drain` | When enabled, Logstash waits until the persistent queue (`queue.type: persisted`) is drained before shutting down. | `false` |
 | `dead_letter_queue.enable` | Flag to instruct Logstash to enable the DLQ feature supported by plugins. | `false` |
 | `dead_letter_queue.max_bytes` | The maximum size of each dead letter queue. Entries will be dropped if they  would increase the size of the dead letter queue beyond this setting. | `1024mb` |

docs/reference/persistent-queues.md

@@ -84,6 +84,17 @@ If you want to define values for a specific pipeline, use [`pipelines.yml`](/ref
 `queue.checkpoint.interval` {applies_to}`stack: deprecated 9.1`
 :   Sets the interval in milliseconds when a checkpoint is forced on the head page. Default is `1000`. Set to `0` to eliminate periodic checkpoints.
 
+`queue.compression` {applies_to}`stack: ga 9.2`
+:   Sets the event compression level for use with the Persisted Queue. Default is `none`. Possible values are:
+    * `none`: does not perform compression, but reads compressed events
+    * `speed`: optimize for fastest compression operation
+    * `size`: optimize for smallest possible size on disk, spending more CPU
+    * `balanced`: a balance between the `speed` and `size` settings
+:::{important}
+Compression can be enabled for an existing PQ, but once compressed elements have been added to a PQ, that PQ cannot be read by previous Logstash releases that did not support compression.
+If you need to downgrade Logstash after enabling the PQ, you will need to either delete the PQ or run the pipeline with `queue.drain: true` first to ensure that no compressed elements remain.
+:::
+
 ## Configuration notes [pq-config-notes]
 
 Every situation and environment is different, and the "ideal" configuration varies. If you optimize for performance, you may increase your risk of losing data. If you optimize for data protection, you may impact performance.

logstash-core/build.gradle

@@ -239,6 +239,7 @@ dependencies {
     implementation 'commons-codec:commons-codec:1.17.0' // transitively required by httpclient
     // Jackson version moved to versions.yml in the project root (the JrJackson version is there too)
     implementation "com.fasterxml.jackson.core:jackson-core:${jacksonVersion}"
+    implementation "com.github.luben:zstd-jni:1.5.7-4"
     api "com.fasterxml.jackson.core:jackson-databind:${jacksonDatabindVersion}"
     api "com.fasterxml.jackson.core:jackson-annotations:${jacksonVersion}"
     implementation 'org.codehaus.janino:janino:3.1.0'

logstash-core/lib/logstash/environment.rb

@@ -96,6 +96,7 @@ def self.as_java_range(r)
            Setting::NumericSetting.new("queue.checkpoint.writes", 1024), # 0 is unlimited
            Setting::NumericSetting.new("queue.checkpoint.interval", 1000), # remove it for #17155
            Setting::BooleanSetting.new("queue.checkpoint.retry", true),
+            Setting::StringSetting.new("queue.compression", "none", true, %w(none speed balanced size disabled)),
            Setting::BooleanSetting.new("dead_letter_queue.enable", false),
              Setting::Bytes.new("dead_letter_queue.max_bytes", "1024mb"),
            Setting::NumericSetting.new("dead_letter_queue.flush_interval", 5000),

logstash-core/lib/logstash/settings.rb

@@ -70,6 +70,7 @@ def self.included(base)
       "queue.checkpoint.interval", # remove it for #17155
       "queue.checkpoint.writes",
       "queue.checkpoint.retry",
+      "queue.compression",
       "queue.drain",
       "queue.max_bytes",
       "queue.max_events",

logstash-core/spec/logstash/queue_factory_spec.rb

@@ -30,6 +30,7 @@
       LogStash::Setting::NumericSetting.new("queue.checkpoint.acks", 1024),
       LogStash::Setting::NumericSetting.new("queue.checkpoint.writes", 1024),
       LogStash::Setting::BooleanSetting.new("queue.checkpoint.retry", false),
+      LogStash::Setting::StringSetting.new("queue.compression", "none", true, %w(none speed balanced size disabled)),
       LogStash::Setting::StringSetting.new("pipeline.id", pipeline_id),
       LogStash::Setting::StringSetting.new("pipeline.batch.metrics.sampling_mode", "minimal", true, ["disabled", "minimal", "full"]),
       LogStash::Setting::PositiveIntegerSetting.new("pipeline.batch.size", 125),

logstash-core/src/main/java/org/logstash/ackedqueue/AbstractZstdAwareCompressionCodec.java

@@ -0,0 +1,41 @@
+package org.logstash.ackedqueue;
+
+import com.github.luben.zstd.Zstd;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+/**
+ * Subclasses of {@link AbstractZstdAwareCompressionCodec} are {@link CompressionCodec}s that are capable
+ * of detecting and decompressing deflate-compressed events. When decoding byte sequences that are <em>NOT</em>
+ * deflate-compressed, the given bytes are emitted verbatim.
+ */
+abstract class AbstractZstdAwareCompressionCodec implements CompressionCodec {
+    // log from the concrete class
+    protected final Logger logger = LogManager.getLogger(this.getClass());
+
+    @Override
+    public byte[] decode(byte[] data) {
+        if (!isZstd(data)) {
+            return data;
+        }
+        try {
+            final byte[] decoded = Zstd.decompress(data);
+            logger.trace("decoded {} -> {}", data.length, decoded.length);
+            return decoded;
+        } catch (Exception e) {
+            throw new RuntimeException("Exception while decoding", e);
+        }
+    }
+
+    private static final byte[] ZSTD_FRAME_MAGIC = { (byte) 0x28, (byte) 0xB5, (byte) 0x2F, (byte) 0xFD };
+
+    static boolean isZstd(byte[] data) {
+        if (data.length < 4) { return false; }
+
+        for (int i = 0; i < 4; i++) {
+            if (data[i] != ZSTD_FRAME_MAGIC[i]) { return false; }
+        }
+
+        return true;
+    }
+}

logstash-core/src/main/java/org/logstash/ackedqueue/CompressionCodec.java

@@ -0,0 +1,58 @@
+package org.logstash.ackedqueue;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+public interface CompressionCodec {
+    Logger LOGGER = LogManager.getLogger(CompressionCodec.class);
+
+    byte[] encode(byte[] data);
+    byte[] decode(byte[] data);
+
+    /**
+     * The {@link CompressionCodec#NOOP} is a {@link CompressionCodec} that
+     * does nothing when encoding and decoding. It is only meant to be activated
+     * as a safety-latch in the event of compression being broken.
+     */
+    CompressionCodec NOOP = new CompressionCodec() {
+        @Override
+        public byte[] encode(byte[] data) {
+            return data;
+        }
+
+        @Override
+        public byte[] decode(byte[] data) {
+            return data;
+        }
+    };
+
+    static CompressionCodec fromConfigValue(final String configValue) {
+        return fromConfigValue(configValue, LOGGER);
+    }
+
+    static CompressionCodec fromConfigValue(final String configValue, final Logger logger) {
+        return switch (configValue) {
+            case "disabled" -> {
+                logger.warn("compression support has been disabled");
+                yield CompressionCodec.NOOP;
+            }
+            case "none" -> {
+                logger.info("compression support is enabled (read-only)");
+                yield ZstdAwareCompressionCodec.getInstance();
+            }
+            case "speed" -> {
+                logger.info("compression support is enabled (goal: speed)");
+                yield new ZstdEnabledCompressionCodec(ZstdEnabledCompressionCodec.Goal.SPEED);
+            }
+            case "balanced" -> {
+                logger.info("compression support is enabled (goal: balanced)");
+                yield new ZstdEnabledCompressionCodec(ZstdEnabledCompressionCodec.Goal.BALANCED);
+            }
+            case "size" -> {
+                logger.info("compression support is enabled (goal: size)");
+                yield new ZstdEnabledCompressionCodec(ZstdEnabledCompressionCodec.Goal.SIZE);
+            }
+            default -> throw new IllegalArgumentException(String.format("Unsupported compression setting `%s`", configValue));
+        };
+    }
+}