out_kinesis_streams: add compression_level option

include/fluent-bit/aws/flb_aws_compress.h

@@ -44,6 +44,25 @@ int flb_aws_compression_get_type(const char *compression_keyword);
 int flb_aws_compression_compress(int compression_type, void *in_data, size_t in_len,
                                 void **out_data, size_t *out_len);
 
+/*
+ * Sentinel for "no explicit compression level": each codec keeps its
+ * historical built-in default (zstd: 1).
+ */
+#define FLB_AWS_COMPRESS_LEVEL_DEFAULT -1
+
+/*
+ * Same as flb_aws_compression_compress, with an explicit compression level.
+ * Pass FLB_AWS_COMPRESS_LEVEL_DEFAULT to keep the codec's built-in default.
+ * A non-default level is currently honored for zstd only; other codecs log a
+ * warning once and compress at their built-in default.
+ *
+ * Returns -1 on error
+ * Returns 0 on success
+ */
+int flb_aws_compression_compress_level(int compression_type, int compression_level,
+                                       void *in_data, size_t in_len,
+                                       void **out_data, size_t *out_len);
+
 /*
  * Truncate and compress in_data and convert to b64
  * If b64 output data is larger than max_out_len, the input is truncated with a
@@ -61,6 +80,16 @@ int flb_aws_compression_b64_truncate_compress(int compression_type, size_t max_o
                                              void *in_data, size_t in_len,
                                              void **out_data, size_t *out_len);
 
+/*
+ * Same as flb_aws_compression_b64_truncate_compress, with an explicit
+ * compression level (see flb_aws_compression_compress_level).
+ */
+int flb_aws_compression_b64_truncate_compress_level(int compression_type,
+                                                    int compression_level,
+                                                    size_t max_out_len,
+                                                    void *in_data, size_t in_len,
+                                                    void **out_data, size_t *out_len);
+
 /*
  * Columnar output formats for out_s3_compress_columnar(). Compression is
  * applied on top of the format via a generic FLB_AWS_COMPRESS_* codec.

include/fluent-bit/flb_zstd.h

@@ -27,6 +27,8 @@
 struct flb_decompression_context;
 
 int flb_zstd_compress(void *in_data, size_t in_len, void **out_data, size_t *out_len);
+int flb_zstd_compress_level(void *in_data, size_t in_len, void **out_data,
+                            size_t *out_len, int level);
 int flb_zstd_uncompress(void *in_data, size_t in_len, void **out_data, size_t *out_len);
 
 int flb_zstd_decompressor_dispatch(struct flb_decompression_context *context,

plugins/out_kinesis_streams/kinesis.c

@@ -160,6 +160,17 @@ static int cb_kinesis_init(struct flb_output_instance *ins,
         ctx->compression = FLB_AWS_COMPRESS_NONE;
     }
 
+    ctx->compression_level = FLB_AWS_COMPRESS_LEVEL_DEFAULT;
+    tmp = flb_output_get_property("compression_level", ins);
+    if (tmp) {
+        if (ctx->compression == FLB_AWS_COMPRESS_NONE) {
+            flb_plg_error(ctx->ins,
+                          "compression_level requires compression to be set");
+            goto error;
+        }
+        ctx->compression_level = atoi(tmp);
+    }
+
     tmp = flb_output_get_property("region", ins);
     if (tmp) {
         ctx->region = tmp;
@@ -544,6 +555,15 @@ static struct flb_config_map config_map[] = {
     "Defaults to no compression."
     },
 
+    {
+     FLB_CONFIG_MAP_INT, "compression_level", "-1",
+     0, FLB_FALSE, 0,
+    "Compression level for the configured 'compression' type. Currently honored "
+    "for 'zstd' only (valid levels follow the linked zstd library, typically "
+    "-131072..22); other codecs use their built-in default and log a warning. "
+    "Defaults to the codec's built-in default (zstd: 1)."
+    },
+
     /* EOF */
     {0}
 };

plugins/out_kinesis_streams/kinesis.h

@@ -99,6 +99,8 @@ struct flb_kinesis {
     int retry_requests;
     int simple_aggregation;
     int compression;
+    /* FLB_AWS_COMPRESS_LEVEL_DEFAULT unless compression_level is set */
+    int compression_level;
     char *sts_endpoint;
     int custom_endpoint;
     uint16_t port;

plugins/out_kinesis_streams/kinesis_api.c

@@ -390,7 +390,8 @@ static int process_event(struct flb_kinesis *ctx, struct flush *buf,
         void *compressed_buf = NULL;
         size_t compressed_size = 0;
 
-        ret = flb_aws_compression_b64_truncate_compress(ctx->compression,
+        ret = flb_aws_compression_b64_truncate_compress_level(ctx->compression,
+                                                       ctx->compression_level,
                                                        MAX_B64_EVENT_SIZE,
                                                        tmp_buf_ptr,
                                                        written,
@@ -464,7 +465,8 @@ static int send_aggregated_record(struct flb_kinesis *ctx, struct flush *buf) {
         void *compressed_buf = NULL;
         size_t compressed_size = 0;
 
-        ret = flb_aws_compression_b64_truncate_compress(ctx->compression,
+        ret = flb_aws_compression_b64_truncate_compress_level(ctx->compression,
+                                                       ctx->compression_level,
                                                        MAX_B64_EVENT_SIZE,
                                                        buf->agg_buf.agg_buf,
                                                        agg_size,

src/aws/flb_aws_compress.c

@@ -49,6 +49,9 @@ struct compression_option {
     int compression_type;
     char *compression_keyword;
     int(*compress)(void *in_data, size_t in_len, void **out_data, size_t *out_len);
+    /* NULL when the codec has no tunable level */
+    int(*compress_level)(void *in_data, size_t in_len, void **out_data,
+                         size_t *out_len, int level);
 };
 
 /*
@@ -61,17 +64,20 @@ static const struct compression_option compression_options[] = {
     {
         FLB_AWS_COMPRESS_GZIP,
         "gzip",
-        &flb_gzip_compress
+        &flb_gzip_compress,
+        NULL
     },
     {
         FLB_AWS_COMPRESS_ZSTD,
         "zstd",
-        &flb_zstd_compress
+        &flb_zstd_compress,
+        &flb_zstd_compress_level
     },
     {
         FLB_AWS_COMPRESS_SNAPPY,
         "snappy",
-        &flb_snappy_compress_wrapper
+        &flb_snappy_compress_wrapper,
+        NULL
     },
     { 0 }
 };
@@ -97,14 +103,33 @@ int flb_aws_compression_get_type(const char *compression_keyword)
 
 int flb_aws_compression_compress(int compression_type, void *in_data, size_t in_len,
                                 void **out_data, size_t *out_len)
+{
+    return flb_aws_compression_compress_level(compression_type,
+                                              FLB_AWS_COMPRESS_LEVEL_DEFAULT,
+                                              in_data, in_len, out_data, out_len);
+}
+
+int flb_aws_compression_compress_level(int compression_type, int compression_level,
+                                       void *in_data, size_t in_len,
+                                       void **out_data, size_t *out_len)
 {
     const struct compression_option *o;
 
     o = compression_options;
 
     while (o->compression_type != 0) {
         if (o->compression_type == compression_type) {
-            return o->compress(in_data, in_len, out_data, out_len);
+            if (compression_level == FLB_AWS_COMPRESS_LEVEL_DEFAULT) {
+                return o->compress(in_data, in_len, out_data, out_len);
+            }
+            if (o->compress_level == NULL) {
+                flb_warn("[aws_compress] compression level %i ignored: '%s' has "
+                         "no tunable level, using its built-in default",
+                         compression_level, o->compression_keyword);
+                return o->compress(in_data, in_len, out_data, out_len);
+            }
+            return o->compress_level(in_data, in_len, out_data, out_len,
+                                     compression_level);
         }
         ++o;
     }
@@ -117,6 +142,18 @@ int flb_aws_compression_compress(int compression_type, void *in_data, size_t in_
 int flb_aws_compression_b64_truncate_compress(int compression_type, size_t max_out_len,
                                              void *in_data, size_t in_len,
                                              void **out_data, size_t *out_len)
+{
+    return flb_aws_compression_b64_truncate_compress_level(compression_type,
+                                                    FLB_AWS_COMPRESS_LEVEL_DEFAULT,
+                                                    max_out_len, in_data, in_len,
+                                                    out_data, out_len);
+}
+
+int flb_aws_compression_b64_truncate_compress_level(int compression_type,
+                                                    int compression_level,
+                                                    size_t max_out_len,
+                                                    void *in_data, size_t in_len,
+                                                    void **out_data, size_t *out_len)
 {
     static const void *truncation_suffix = "[Truncated...]";
     static const size_t truncation_suffix_len = 14;
@@ -154,7 +191,8 @@ int flb_aws_compression_b64_truncate_compress(int compression_type, size_t max_o
             return -1;
         }
 
-        ret = flb_aws_compression_compress(compression_type, truncated_in_buf,
+        ret = flb_aws_compression_compress_level(compression_type, compression_level,
+                                          truncated_in_buf,
                                           truncated_in_len, &compressed_buf,
                                           &compressed_len);
         ++compression_attempts;

src/flb_zstd.c

@@ -29,22 +29,37 @@ struct flb_zstd_decompression_context {
 };
 
 #define FLB_ZSTD_DEFAULT_CHUNK      (64 * 1024)       /* 64 KB buffer */
+/* Historical default used by flb_zstd_compress(); kept for compatibility */
+#define FLB_ZSTD_DEFAULT_COMPRESSION_LEVEL 1
 #define FLB_ZSTD_DECOMPRESS_MAX     (100 * 1024 * 1024)  /* 100 MB limit */
 
 int flb_zstd_compress(void *in_data, size_t in_len, void **out_data, size_t *out_len)
+{
+    return flb_zstd_compress_level(in_data, in_len, out_data, out_len,
+                                   FLB_ZSTD_DEFAULT_COMPRESSION_LEVEL);
+}
+
+int flb_zstd_compress_level(void *in_data, size_t in_len, void **out_data,
+                            size_t *out_len, int level)
 {
     void *buf;
     size_t size;
     size_t bound;
 
+    if (level < ZSTD_minCLevel() || level > ZSTD_maxCLevel()) {
+        flb_error("[zstd] invalid compression level %i (valid range: %i..%i)",
+                  level, ZSTD_minCLevel(), ZSTD_maxCLevel());
+        return -1;
+    }
+
     bound = ZSTD_compressBound(in_len);
     buf = flb_malloc(bound);
     if (!buf) {
         flb_errno();
         return -1;
     }
 
-    size = ZSTD_compress(buf, bound, in_data, in_len, 1);
+    size = ZSTD_compress(buf, bound, in_data, in_len, level);
     if (ZSTD_isError(size)) {
         flb_error("[zstd] compression failed: %s", ZSTD_getErrorName(size));
         flb_free(buf);

tests/internal/aws_compress.c

@@ -85,6 +85,61 @@ void test_compression_zstd()
     flb_aws_compress_test_cases(cases);
 }
 
+void test_compression_zstd_level()
+{
+    int ret;
+    void *out_default = NULL;
+    void *out_high = NULL;
+    void *roundtrip = NULL;
+    size_t out_default_len = 0;
+    size_t out_high_len = 0;
+    size_t roundtrip_len = 0;
+    /* repetitive input so higher levels have something to win on */
+    char in_data[512];
+    size_t i;
+
+    for (i = 0; i < sizeof(in_data); i++) {
+        in_data[i] = "hello world "[i % 12];
+    }
+
+    /* explicit level compresses and roundtrips */
+    ret = flb_aws_compression_compress_level(FLB_AWS_COMPRESS_ZSTD, 19,
+                                             in_data, sizeof(in_data),
+                                             &out_high, &out_high_len);
+    TEST_CHECK(ret == 0);
+    ret = flb_zstd_uncompress(out_high, out_high_len, &roundtrip, &roundtrip_len);
+    TEST_CHECK(ret == 0);
+    TEST_CHECK(roundtrip_len == sizeof(in_data));
+    TEST_CHECK(memcmp(roundtrip, in_data, sizeof(in_data)) == 0);
+    flb_free(roundtrip);
+
+    /* sentinel level matches the plain API's built-in default */
+    ret = flb_aws_compression_compress_level(FLB_AWS_COMPRESS_ZSTD,
+                                             FLB_AWS_COMPRESS_LEVEL_DEFAULT,
+                                             in_data, sizeof(in_data),
+                                             &out_default, &out_default_len);
+    TEST_CHECK(ret == 0);
+
+    /* a higher level must not produce a larger frame on repetitive input */
+    TEST_CHECK(out_high_len <= out_default_len);
+
+    /* out-of-range level fails */
+    ret = flb_aws_compression_compress_level(FLB_AWS_COMPRESS_ZSTD, 1000000,
+                                             in_data, sizeof(in_data),
+                                             &roundtrip, &roundtrip_len);
+    TEST_CHECK(ret == -1);
+
+    /* codec without tunable level falls back to its default and succeeds */
+    ret = flb_aws_compression_compress_level(FLB_AWS_COMPRESS_GZIP, 9,
+                                             in_data, sizeof(in_data),
+                                             &roundtrip, &roundtrip_len);
+    TEST_CHECK(ret == 0);
+    flb_free(roundtrip);
+
+    flb_free(out_default);
+    flb_free(out_high);
+}
+
 void test_compression_snappy()
 {
     struct flb_aws_test_case cases[] =
@@ -525,6 +580,7 @@ void test_arrow_format_gzip_unsupported()
 TEST_LIST = {
     { "test_compression_gzip", test_compression_gzip },
     { "test_compression_zstd", test_compression_zstd },
+    { "test_compression_zstd_level", test_compression_zstd_level },
     { "test_compression_snappy", test_compression_snappy },
     { "test_compression_snappy_return_value_normalization",
       test_compression_snappy_return_value_normalization },