tests: Add Kinesis Streams test cases for compression

Signed-off-by: Shelby Hagman <[email protected]>

Author: ShelbyZ

tests/runtime/out_firehose.c

@@ -375,6 +375,121 @@ void flb_test_firehose_aggregation_with_compression(void)
     flb_destroy(ctx);
 }
 
+void flb_test_firehose_compression_zstd(void)
+{
+    int ret;
+    flb_ctx_t *ctx;
+    int in_ffd;
+    int out_ffd;
+    const char *record1 = "[1, {\"message\":\"zstd_test1\"}]";
+    const char *record2 = "[1, {\"message\":\"zstd_test2\"}]";
+
+    setenv("FLB_FIREHOSE_PLUGIN_UNDER_TEST", "true", 1);
+
+    ctx = flb_create();
+
+    in_ffd = flb_input(ctx, (char *) "lib", NULL);
+    TEST_CHECK(in_ffd >= 0);
+    flb_input_set(ctx, in_ffd, "tag", "test", NULL);
+
+    out_ffd = flb_output(ctx, (char *) "kinesis_firehose", NULL);
+    TEST_CHECK(out_ffd >= 0);
+    flb_output_set(ctx, out_ffd, "match", "*", NULL);
+    flb_output_set(ctx, out_ffd, "region", "us-west-2", NULL);
+    flb_output_set(ctx, out_ffd, "delivery_stream", "fluent", NULL);
+    flb_output_set(ctx, out_ffd, "compression", "zstd", NULL);
+    flb_output_set(ctx, out_ffd, "Retry_Limit", "1", NULL);
+
+    ret = flb_start(ctx);
+    TEST_CHECK(ret == 0);
+
+    /* Push records with ZSTD compression */
+    flb_lib_push(ctx, in_ffd, (char *) record1, strlen(record1));
+    flb_lib_push(ctx, in_ffd, (char *) record2, strlen(record2));
+
+    sleep(2);
+    flb_stop(ctx);
+    flb_destroy(ctx);
+}
+
+void flb_test_firehose_compression_snappy(void)
+{
+    int ret;
+    flb_ctx_t *ctx;
+    int in_ffd;
+    int out_ffd;
+    const char *record1 = "[1, {\"message\":\"snappy_test1\"}]";
+    const char *record2 = "[1, {\"message\":\"snappy_test2\"}]";
+
+    setenv("FLB_FIREHOSE_PLUGIN_UNDER_TEST", "true", 1);
+
+    ctx = flb_create();
+
+    in_ffd = flb_input(ctx, (char *) "lib", NULL);
+    TEST_CHECK(in_ffd >= 0);
+    flb_input_set(ctx, in_ffd, "tag", "test", NULL);
+
+    out_ffd = flb_output(ctx, (char *) "kinesis_firehose", NULL);
+    TEST_CHECK(out_ffd >= 0);
+    flb_output_set(ctx, out_ffd, "match", "*", NULL);
+    flb_output_set(ctx, out_ffd, "region", "us-west-2", NULL);
+    flb_output_set(ctx, out_ffd, "delivery_stream", "fluent", NULL);
+    flb_output_set(ctx, out_ffd, "compression", "snappy", NULL);
+    flb_output_set(ctx, out_ffd, "Retry_Limit", "1", NULL);
+
+    ret = flb_start(ctx);
+    TEST_CHECK(ret == 0);
+
+    /* Push records with Snappy compression */
+    flb_lib_push(ctx, in_ffd, (char *) record1, strlen(record1));
+    flb_lib_push(ctx, in_ffd, (char *) record2, strlen(record2));
+
+    sleep(2);
+    flb_stop(ctx);
+    flb_destroy(ctx);
+}
+
+void flb_test_firehose_compression_snappy_with_aggregation(void)
+{
+    int ret;
+    flb_ctx_t *ctx;
+    int in_ffd;
+    int out_ffd;
+    int i;
+    char record[100];
+
+    setenv("FLB_FIREHOSE_PLUGIN_UNDER_TEST", "true", 1);
+
+    ctx = flb_create();
+
+    in_ffd = flb_input(ctx, (char *) "lib", NULL);
+    TEST_CHECK(in_ffd >= 0);
+    flb_input_set(ctx, in_ffd, "tag", "test", NULL);
+
+    out_ffd = flb_output(ctx, (char *) "kinesis_firehose", NULL);
+    TEST_CHECK(out_ffd >= 0);
+    flb_output_set(ctx, out_ffd, "match", "*", NULL);
+    flb_output_set(ctx, out_ffd, "region", "us-west-2", NULL);
+    flb_output_set(ctx, out_ffd, "delivery_stream", "fluent", NULL);
+    flb_output_set(ctx, out_ffd, "simple_aggregation", "On", NULL);
+    flb_output_set(ctx, out_ffd, "compression", "snappy", NULL);
+    flb_output_set(ctx, out_ffd, "Retry_Limit", "1", NULL);
+
+    ret = flb_start(ctx);
+    TEST_CHECK(ret == 0);
+
+    /* Push many records with Snappy compression and aggregation */
+    for (i = 0; i < 20; i++) {
+        ret = snprintf(record, sizeof(record), "[1, {\"id\":%d,\"msg\":\"snappy_agg\"}]", i);
+        TEST_CHECK(ret < sizeof(record));
+        flb_lib_push(ctx, in_ffd, record, strlen(record));
+    }
+
+    sleep(3);
+    flb_stop(ctx);
+    flb_destroy(ctx);
+}
+
 void flb_test_firehose_aggregation_combined_params(void)
 {
     int ret;
@@ -536,6 +651,9 @@ TEST_LIST = {
     {"aggregation_with_log_key", flb_test_firehose_aggregation_with_log_key },
     {"aggregation_many_records", flb_test_firehose_aggregation_many_records },
     {"aggregation_with_compression", flb_test_firehose_aggregation_with_compression },
+    {"compression_zstd", flb_test_firehose_compression_zstd },
+    {"compression_snappy", flb_test_firehose_compression_snappy },
+    {"compression_snappy_with_aggregation", flb_test_firehose_compression_snappy_with_aggregation },
     {"aggregation_combined_params", flb_test_firehose_aggregation_combined_params },
     {"aggregation_empty_records", flb_test_firehose_aggregation_empty_records },
     {"aggregation_error_handling", flb_test_firehose_aggregation_error_handling },