out_kinesis_streams: add compression_level option

Allow tuning the record compression level (honored for zstd; other
codecs keep their built-in default). The plugin previously compressed
zstd at a hardcoded level 1 with no way to trade CPU for ratio, which
matters for high-volume streams where per-record compression is the
dominant cost lever. Defaults preserve existing behavior; setting
compression_level without compression is a configuration error.

Signed-off-by: Ivan Bushmarinov <ivan.bushmarinov@perplexity.ai>

Author: ivanbushmarinov-pplx

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,