diff --git a/include/fluent-bit/aws/flb_aws_compress.h b/include/fluent-bit/aws/flb_aws_compress.h index 86805790075..5c5b87ebe36 100644 --- a/include/fluent-bit/aws/flb_aws_compress.h +++ b/include/fluent-bit/aws/flb_aws_compress.h @@ -62,8 +62,9 @@ int flb_aws_compression_b64_truncate_compress(int compression_type, size_t max_o 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. + * Columnar output formats for flb_aws_compression_compress_columnar(). + * Compression is applied on top of the format via a generic + * FLB_AWS_COMPRESS_* codec. */ #define FLB_AWS_COMPRESS_FORMAT_ARROW 0 #define FLB_AWS_COMPRESS_FORMAT_PARQUET 1 @@ -78,6 +79,12 @@ int flb_aws_compression_b64_truncate_compress(int compression_type, size_t max_o * * Returns 0 on success, -1 on failure. */ +int flb_aws_compression_compress_columnar(int columnar_format, + void *json, size_t size, + void **out_buf, size_t *out_size, + int compression_type); + +/* Backward-compatible name retained for existing out-of-tree consumers. */ int out_s3_compress_columnar(int columnar_format, void *json, size_t size, void **out_buf, size_t *out_size, int compression_type); diff --git a/plugins/out_gcs/gcs.c b/plugins/out_gcs/gcs.c index e9062359d13..11fa758cbc3 100644 --- a/plugins/out_gcs/gcs.c +++ b/plugins/out_gcs/gcs.c @@ -19,7 +19,6 @@ #include #include -#include #include #include #include @@ -28,15 +27,51 @@ #include #include #include +#include #include "gcs.h" #include "gcs_store.h" +#include #include #include static int gcs_ctx_destroy(void *data, struct flb_config *config); +static int enable_parquet_format(struct flb_gcs *ctx) +{ +#ifdef FLB_HAVE_ARROW_PARQUET + ctx->gcs_format = FLB_GCS_FORMAT_PARQUET; + return 0; +#else + flb_plg_error(ctx->ins, + "parquet format requires parquet-glib at compile time"); + return -1; +#endif +} + +static int parse_output_format(const char *format) +{ + if (strcasecmp(format, "parquet") == 0) { + return FLB_GCS_FORMAT_PARQUET; + } + + return flb_pack_to_json_format_type(format); +} + +static int validate_parquet_compression(int compression_type) +{ + switch (compression_type) { + case FLB_AWS_COMPRESS_NONE: + case FLB_AWS_COMPRESS_SNAPPY: + case FLB_AWS_COMPRESS_GZIP: + case FLB_AWS_COMPRESS_ZSTD: + return 0; + default: + return -1; + } +} + static const char *get_predefined_acl(const char *canned_acl) { if (!canned_acl) { @@ -782,9 +817,20 @@ static int gcs_upload_object(struct flb_gcs *ctx, }; char final_body_md5[25]; + if (ctx->gcs_format == FLB_GCS_FORMAT_PARQUET && + flb_output_get_property("content_type", ctx->ins) == NULL) { + content_type_header.val = "application/vnd.apache.parquet"; + content_type_header.val_len = 30; + } + else { + content_type_header.val = ctx->content_type; + content_type_header.val_len = flb_sds_len(ctx->content_type); + } + if (gcs_under_test_mode() == FLB_TRUE) { mock_gcs_call_increment_counter("UploadObject"); gcs_setenv("TEST_GCS_LAST_URI", uri); + gcs_setenv("TEST_GCS_LAST_CONTENT_TYPE", content_type_header.val); if (body_size >= 2 && (unsigned char) body[0] == 0x1f && (unsigned char) body[1] == 0x8b) { @@ -793,6 +839,14 @@ static int gcs_upload_object(struct flb_gcs *ctx, else { gcs_setenv("TEST_GCS_LAST_BODY_GZIP", "false"); } + if (body_size >= 8 && + memcmp(body, "PAR1", 4) == 0 && + memcmp(body + body_size - 4, "PAR1", 4) == 0) { + gcs_setenv("TEST_GCS_LAST_BODY_PARQUET", "true"); + } + else { + gcs_setenv("TEST_GCS_LAST_BODY_PARQUET", "false"); + } if (getenv("TEST_GCS_UPLOAD_ERROR") != NULL) { return -1; @@ -813,8 +867,6 @@ static int gcs_upload_object(struct flb_gcs *ctx, return -1; } - content_type_header.val = ctx->content_type; - content_type_header.val_len = flb_sds_len(ctx->content_type); flb_http_add_header(c, content_type_header.key, content_type_header.key_len, content_type_header.val, content_type_header.val_len); flb_http_add_header(c, "Authorization", 13, auth, flb_sds_len(auth)); @@ -859,8 +911,8 @@ static int upload_data(struct flb_gcs *ctx, flb_sds_t gcs_key_encoded; flb_sds_t uri; flb_sds_t tmp; - void *gz_data = NULL; - size_t gz_size = 0; + void *compressed_data = NULL; + size_t compressed_size = 0; char *upload_body; size_t upload_size; char random_hex[9]; @@ -955,7 +1007,8 @@ static int upload_data(struct flb_gcs *ctx, } uri = tmp; - if (ctx->compression_type == FLB_GCS_COMPRESSION_GZIP) { + if (ctx->gcs_format != FLB_GCS_FORMAT_PARQUET && + ctx->compression_type == FLB_AWS_COMPRESS_GZIP) { tmp = flb_sds_cat(uri, "&contentEncoding=gzip", 21); if (!tmp) { flb_sds_destroy(uri); @@ -978,10 +1031,35 @@ static int upload_data(struct flb_gcs *ctx, upload_body = buffer; upload_size = buffer_size; - if (ctx->compression_type == FLB_GCS_COMPRESSION_GZIP) { - ret = flb_gzip_compress(buffer, buffer_size, &gz_data, &gz_size); - if (ret != 0 || !gz_data) { - flb_plg_error(ctx->ins, "could not gzip buffered data"); +#ifdef FLB_HAVE_ARROW_PARQUET + if (ctx->gcs_format == FLB_GCS_FORMAT_PARQUET) { + ret = flb_aws_compression_compress_columnar( + FLB_AWS_COMPRESS_FORMAT_PARQUET, + buffer, buffer_size, + &compressed_data, &compressed_size, + ctx->compression_type); + if (ret != 0 || !compressed_data) { + flb_plg_error(ctx->ins, "could not convert buffered data to parquet"); + flb_sds_destroy(auth); + flb_sds_destroy(uri); + if (ctx->key_fmt_has_seq_index && ctx->seq_index > 0) { + ctx->seq_index--; + write_seq_index(ctx->seq_index_file, ctx->seq_index); + } + return -1; + } + + upload_body = compressed_data; + upload_size = compressed_size; + } + else +#endif + if (ctx->compression_type != FLB_AWS_COMPRESS_NONE) { + ret = flb_aws_compression_compress(ctx->compression_type, + buffer, buffer_size, + &compressed_data, &compressed_size); + if (ret != 0 || !compressed_data) { + flb_plg_error(ctx->ins, "could not compress buffered data"); flb_sds_destroy(auth); flb_sds_destroy(uri); if (ctx->key_fmt_has_seq_index && ctx->seq_index > 0) { @@ -991,16 +1069,16 @@ static int upload_data(struct flb_gcs *ctx, return -1; } - upload_body = gz_data; - upload_size = gz_size; + upload_body = compressed_data; + upload_size = compressed_size; flb_plg_debug(ctx->ins, "Pre-compression chunk size is %zu, After compression, chunk is %zu bytes", - buffer_size, gz_size); + buffer_size, compressed_size); } ret = gcs_upload_object(ctx, auth, uri, upload_body, upload_size); - if (gz_data) { - flb_free(gz_data); + if (compressed_data) { + flb_free(compressed_data); } flb_sds_destroy(auth); flb_sds_destroy(uri); @@ -1218,6 +1296,8 @@ static int flush_init(struct flb_gcs *ctx) static int cb_gcs_init(struct flb_output_instance *ins, struct flb_config *config, void *data) { int ret; + size_t index; + flb_sds_t normalized_compression; struct flb_gcs *ctx; const char *tmp; (void) data; @@ -1312,6 +1392,7 @@ static int cb_gcs_init(struct flb_output_instance *ins, struct flb_config *confi goto error; } ctx->out_format = FLB_PACK_JSON_FORMAT_LINES; + ctx->gcs_format = FLB_GCS_FORMAT_JSON_LINES; ctx->json_date_format = FLB_PACK_JSON_DATE_DOUBLE; if (ctx->content_type == NULL) { ctx->content_type = flb_sds_create("application/json"); @@ -1320,11 +1401,65 @@ static int cb_gcs_init(struct flb_output_instance *ins, struct flb_config *confi } } + tmp = flb_output_get_property("format", ins); + if (tmp) { + ret = parse_output_format(tmp); + if (ret == FLB_GCS_FORMAT_PARQUET) { + if (enable_parquet_format(ctx) == -1) { + goto error; + } + } + else if (ret == FLB_PACK_JSON_FORMAT_JSON) { + flb_plg_warn(ctx->ins, + "'json' format is interpreted as 'json_lines'"); + } + else if (ret != FLB_PACK_JSON_FORMAT_LINES) { + flb_plg_error(ctx->ins, "unsupported format '%s'", tmp); + goto error; + } + } + tmp = flb_output_get_property("compression", ins); - if (tmp && strcasecmp(tmp, "gzip") == 0) { - ctx->compression_type = FLB_GCS_COMPRESSION_GZIP; + if (!tmp) { + ctx->compression_type = FLB_AWS_COMPRESS_NONE; + } + else { + normalized_compression = flb_sds_create(tmp); + if (!normalized_compression) { + flb_errno(); + goto error; + } + + for (index = 0; index < flb_sds_len(normalized_compression); index++) { + normalized_compression[index] = + tolower((unsigned char) normalized_compression[index]); + } + + if (strcmp(normalized_compression, "none") == 0) { + ret = FLB_AWS_COMPRESS_NONE; + } + else { + ret = flb_aws_compression_get_type(normalized_compression); + } + + flb_sds_destroy(normalized_compression); + if (ret == -1) { + flb_plg_error(ins, "unsupported compression type '%s'", tmp); + goto error; + } + ctx->compression_type = ret; + } + + if (ctx->gcs_format == FLB_GCS_FORMAT_PARQUET) { + if (validate_parquet_compression(ctx->compression_type) != 0) { + flb_plg_error(ins, + "'%s' is not a supported parquet compression codec", + tmp); + goto error; + } } - else if (tmp && strcasecmp(tmp, "none") != 0) { + else if (ctx->compression_type != FLB_AWS_COMPRESS_NONE && + ctx->compression_type != FLB_AWS_COMPRESS_GZIP) { flb_plg_error(ins, "unsupported compression type '%s'", tmp); goto error; } @@ -1462,6 +1597,12 @@ static int cb_gcs_exit(void *data, struct flb_config *config) } static struct flb_config_map config_map[] = { + { + FLB_CONFIG_MAP_STR, "format", "json_lines", + 0, FLB_FALSE, 0, + "Output format. Supported values: json_lines and parquet. When format is " + "parquet, compression selects the page-level codec." + }, { FLB_CONFIG_MAP_STR, "bucket", NULL, 0, FLB_TRUE, offsetof(struct flb_gcs, bucket), @@ -1515,7 +1656,8 @@ static struct flb_config_map config_map[] = { { FLB_CONFIG_MAP_STR, "content_type", "application/json", 0, FLB_TRUE, offsetof(struct flb_gcs, content_type), - "Content type." + "Content type. Defaults to application/json for JSON lines and " + "application/vnd.apache.parquet for Parquet." }, { FLB_CONFIG_MAP_STR, "google_service_credentials", NULL, @@ -1530,7 +1672,8 @@ static struct flb_config_map config_map[] = { { FLB_CONFIG_MAP_STR, "compression", "none", 0, FLB_FALSE, 0, - "Compression: none or gzip." + "Compression type. JSON lines support none and gzip. Parquet supports " + "none, snappy, gzip, and zstd." }, {0} }; diff --git a/plugins/out_gcs/gcs.h b/plugins/out_gcs/gcs.h index 57f9de845f9..f6f375e7201 100644 --- a/plugins/out_gcs/gcs.h +++ b/plugins/out_gcs/gcs.h @@ -32,8 +32,8 @@ #define FLB_GCS_AUTH_URL "https://oauth2.googleapis.com/token" #define FLB_GCS_TOKEN_REFRESH 3000 -#define FLB_GCS_COMPRESSION_NONE 0 -#define FLB_GCS_COMPRESSION_GZIP 1 +#define FLB_GCS_FORMAT_JSON_LINES 0 +#define FLB_GCS_FORMAT_PARQUET 100 struct upload_queue { struct gcs_file *upload_file; @@ -83,6 +83,7 @@ struct flb_gcs { int static_file_path; int out_format; + int gcs_format; int json_date_format; flb_sds_t json_date_key; int compression_type; diff --git a/plugins/out_s3/s3.c b/plugins/out_s3/s3.c index fad4b931568..3a0f1273b91 100644 --- a/plugins/out_s3/s3.c +++ b/plugins/out_s3/s3.c @@ -208,9 +208,9 @@ static int s3_format_is_columnar(int s3_format) * aws-compress columnar format identifier * * Translates FLB_S3_FORMAT_* to the FLB_AWS_COMPRESS_FORMAT_* identifier - * consumed by out_s3_compress_columnar(), keeping the compression layer - * decoupled from the plugin's format enum. Returns -1 for any format that is - * not a known columnar format, so a future format added to + * consumed by flb_aws_compression_compress_columnar(), keeping the + * compression layer decoupled from the plugin's format enum. Returns -1 for + * any format that is not a known columnar format, so a future format added to * s3_format_is_columnar() but not mapped here fails loudly instead of being * silently emitted as Arrow. */ @@ -1527,9 +1527,10 @@ static int upload_data(struct flb_s3 *ctx, struct s3_file *chunk, #ifdef FLB_HAVE_ARROW if (s3_format_is_columnar(ctx->s3_format)) { - ret = out_s3_compress_columnar(s3_format_to_aws_compress_format(ctx->s3_format), - body, body_size, &payload_buf, - &payload_size, ctx->compression); + ret = flb_aws_compression_compress_columnar( + s3_format_to_aws_compress_format(ctx->s3_format), + body, body_size, &payload_buf, + &payload_size, ctx->compression); if (ret == -1) { flb_plg_error(ctx->ins, "Failed to convert data to columnar " "format"); @@ -1777,7 +1778,7 @@ static int put_all_chunks(struct flb_s3 *ctx) #ifdef FLB_HAVE_ARROW if (s3_format_is_columnar(ctx->s3_format)) { - ret = out_s3_compress_columnar( + ret = flb_aws_compression_compress_columnar( s3_format_to_aws_compress_format(ctx->s3_format), buffer, buffer_size, &payload_buf, &payload_size, diff --git a/src/aws/compression/arrow/compress.c b/src/aws/compression/arrow/compress.c index b57ec04ca8e..16eb39d3df6 100644 --- a/src/aws/compression/arrow/compress.c +++ b/src/aws/compression/arrow/compress.c @@ -239,9 +239,10 @@ static GArrowResizableBuffer* table_to_parquet_buffer(GArrowTable *table, } #endif -int out_s3_compress_columnar(int columnar_format, void *json, size_t size, - void **out_buf, size_t *out_size, - int compression_type) +int flb_aws_compression_compress_columnar(int columnar_format, + void *json, size_t size, + void **out_buf, size_t *out_size, + int compression_type) { GArrowTable *table; GArrowResizableBuffer *buffer; @@ -320,3 +321,12 @@ int out_s3_compress_columnar(int columnar_format, void *json, size_t size, g_bytes_unref(bytes); return 0; } + +int out_s3_compress_columnar(int columnar_format, void *json, size_t size, + void **out_buf, size_t *out_size, + int compression_type) +{ + return flb_aws_compression_compress_columnar( + columnar_format, json, size, + out_buf, out_size, compression_type); +} diff --git a/tests/internal/aws_compress.c b/tests/internal/aws_compress.c index 0bb4414c419..e081a7f433a 100644 --- a/tests/internal/aws_compress.c +++ b/tests/internal/aws_compress.c @@ -344,7 +344,7 @@ void test_parquet_format_snappy() "{\"key\":\"other\",\"num\":99}\n"; size_t json_len = strlen(json); - ret = out_s3_compress_columnar(FLB_AWS_COMPRESS_FORMAT_PARQUET, + ret = flb_aws_compression_compress_columnar(FLB_AWS_COMPRESS_FORMAT_PARQUET, json, json_len, &out_buf, &out_size, FLB_AWS_COMPRESS_SNAPPY); if (!TEST_CHECK(ret == 0 && out_buf != NULL && out_size >= 8)) { @@ -365,7 +365,7 @@ void test_parquet_format_zstd() "{\"key\":\"other\",\"num\":99}\n"; size_t json_len = strlen(json); - ret = out_s3_compress_columnar(FLB_AWS_COMPRESS_FORMAT_PARQUET, + ret = flb_aws_compression_compress_columnar(FLB_AWS_COMPRESS_FORMAT_PARQUET, json, json_len, &out_buf, &out_size, FLB_AWS_COMPRESS_ZSTD); if (!TEST_CHECK(ret == 0 && out_buf != NULL && out_size >= 8)) { @@ -386,7 +386,7 @@ void test_parquet_format_gzip() "{\"key\":\"other\",\"num\":99}\n"; size_t json_len = strlen(json); - ret = out_s3_compress_columnar(FLB_AWS_COMPRESS_FORMAT_PARQUET, + ret = flb_aws_compression_compress_columnar(FLB_AWS_COMPRESS_FORMAT_PARQUET, json, json_len, &out_buf, &out_size, FLB_AWS_COMPRESS_GZIP); if (!TEST_CHECK(ret == 0 && out_buf != NULL && out_size >= 8)) { @@ -407,7 +407,7 @@ void test_parquet_format_uncompressed() "{\"key\":\"other\",\"num\":99}\n"; size_t json_len = strlen(json); - ret = out_s3_compress_columnar(FLB_AWS_COMPRESS_FORMAT_PARQUET, + ret = flb_aws_compression_compress_columnar(FLB_AWS_COMPRESS_FORMAT_PARQUET, json, json_len, &out_buf, &out_size, FLB_AWS_COMPRESS_NONE); if (!TEST_CHECK(ret == 0 && out_buf != NULL && out_size >= 8)) { @@ -433,7 +433,7 @@ void test_parquet_compression_reduces_size() "{\"msg\":\"hello hello hello hello hello hello\"}\n"; size_t json_len = strlen(json); - ret = out_s3_compress_columnar(FLB_AWS_COMPRESS_FORMAT_PARQUET, + ret = flb_aws_compression_compress_columnar(FLB_AWS_COMPRESS_FORMAT_PARQUET, json, json_len, &buf_none, &size_none, FLB_AWS_COMPRESS_NONE); if (!TEST_CHECK(ret == 0 && buf_none != NULL)) { @@ -441,7 +441,7 @@ void test_parquet_compression_reduces_size() return; } - ret = out_s3_compress_columnar(FLB_AWS_COMPRESS_FORMAT_PARQUET, + ret = flb_aws_compression_compress_columnar(FLB_AWS_COMPRESS_FORMAT_PARQUET, json, json_len, &buf_snappy, &size_snappy, FLB_AWS_COMPRESS_SNAPPY); if (!TEST_CHECK(ret == 0 && buf_snappy != NULL)) { @@ -466,7 +466,7 @@ void test_arrow_format_uncompressed() "{\"key\":\"other\",\"num\":99}\n"; size_t json_len = strlen(json); - ret = out_s3_compress_columnar(FLB_AWS_COMPRESS_FORMAT_ARROW, + ret = flb_aws_compression_compress_columnar(FLB_AWS_COMPRESS_FORMAT_ARROW, json, json_len, &out_buf, &out_size, FLB_AWS_COMPRESS_NONE); if (!TEST_CHECK(ret == 0 && out_buf != NULL && out_size >= 8)) { @@ -487,7 +487,7 @@ void test_arrow_format_zstd() "{\"key\":\"other\",\"num\":99}\n"; size_t json_len = strlen(json); - ret = out_s3_compress_columnar(FLB_AWS_COMPRESS_FORMAT_ARROW, + ret = flb_aws_compression_compress_columnar(FLB_AWS_COMPRESS_FORMAT_ARROW, json, json_len, &out_buf, &out_size, FLB_AWS_COMPRESS_ZSTD); if (!TEST_CHECK(ret == 0 && out_buf != NULL && out_size >= 8)) { @@ -512,7 +512,7 @@ void test_arrow_format_gzip_unsupported() "{\"key\":\"other\",\"num\":99}\n"; size_t json_len = strlen(json); - ret = out_s3_compress_columnar(FLB_AWS_COMPRESS_FORMAT_ARROW, + ret = flb_aws_compression_compress_columnar(FLB_AWS_COMPRESS_FORMAT_ARROW, json, json_len, &out_buf, &out_size, FLB_AWS_COMPRESS_GZIP); TEST_CHECK(ret == -1); diff --git a/tests/runtime/out_gcs.c b/tests/runtime/out_gcs.c index f58a25bf262..1cdda73df17 100644 --- a/tests/runtime/out_gcs.c +++ b/tests/runtime/out_gcs.c @@ -109,7 +109,7 @@ void flb_test_gcs_upload_success(void) flb_output_set(ctx, out_ffd, "store_dir", store_dir, NULL); flb_output_set(ctx, out_ffd, "gcs_key_format", "logs/$TAG", NULL); flb_output_set(ctx, out_ffd, "static_file_path", "true", NULL); - flb_output_set(ctx, out_ffd, "compression", "gzip", NULL); + flb_output_set(ctx, out_ffd, "compression", "GZIP", NULL); flb_output_set(ctx, out_ffd, "canned_acl", "public-read", NULL); ret = flb_start(ctx); @@ -143,9 +143,142 @@ void flb_test_gcs_upload_success(void) unsetenv("TEST_GCS_UploadObject_CALL_COUNT"); unsetenv("TEST_GCS_LAST_URI"); unsetenv("TEST_GCS_LAST_BODY_GZIP"); + unsetenv("TEST_GCS_LAST_BODY_PARQUET"); + unsetenv("TEST_GCS_LAST_CONTENT_TYPE"); flb_free(store_dir); } +#ifdef FLB_HAVE_ARROW_PARQUET +static void test_gcs_upload_parquet(const char *compression, + const char *store_directory_postfix) +{ + int ret; + int call_count; + int in_ffd; + int out_ffd; + char *call_count_str; + char *store_dir; + flb_ctx_t *ctx; + + store_dir = create_test_store_directory(store_directory_postfix); + TEST_CHECK(store_dir != NULL); + if (!store_dir) { + return; + } + + setenv("FLB_GCS_PLUGIN_UNDER_TEST", "true", 1); + unsetenv("TEST_GCS_UploadObject_CALL_COUNT"); + + 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 *) "gcs", NULL); + TEST_CHECK(out_ffd >= 0); + flb_output_set(ctx, out_ffd, "match", "*", NULL); + flb_output_set(ctx, out_ffd, "bucket", "fluent", NULL); + flb_output_set(ctx, out_ffd, "google_service_credentials", SERVICE_CREDENTIALS, NULL); + flb_output_set(ctx, out_ffd, "upload_timeout", "3s", NULL); + flb_output_set(ctx, out_ffd, "store_dir", store_dir, NULL); + flb_output_set(ctx, out_ffd, "gcs_key_format", "logs/$TAG", NULL); + flb_output_set(ctx, out_ffd, "static_file_path", "true", NULL); + flb_output_set(ctx, out_ffd, "format", "parquet", NULL); + flb_output_set(ctx, out_ffd, "compression", compression, NULL); + + ret = flb_start(ctx); + TEST_CHECK(ret == 0); + + flb_lib_push(ctx, in_ffd, (char *) JSON_TD, (int) sizeof(JSON_TD) - 1); + sleep(5); + + call_count_str = getenv("TEST_GCS_UploadObject_CALL_COUNT"); + call_count = call_count_str ? atoi(call_count_str) : 0; + TEST_CHECK_(call_count == 1, + "Expected 1 UploadObject call, got %d", call_count); + TEST_CHECK_(getenv("TEST_GCS_LAST_URI") != NULL, + "Expected the mock upload URI to be captured"); + if (getenv("TEST_GCS_LAST_URI")) { + TEST_CHECK(strcmp(getenv("TEST_GCS_LAST_URI"), + "/upload/storage/v1/b/fluent/o?uploadType=media&" + "name=logs%2Ftest") == 0); + } + TEST_CHECK_(getenv("TEST_GCS_LAST_BODY_PARQUET") != NULL, + "Expected the mock upload body format to be captured"); + if (getenv("TEST_GCS_LAST_BODY_PARQUET")) { + TEST_CHECK(strcmp(getenv("TEST_GCS_LAST_BODY_PARQUET"), "true") == 0); + } + TEST_CHECK_(getenv("TEST_GCS_LAST_BODY_GZIP") != NULL, + "Expected the mock upload body encoding to be captured"); + if (getenv("TEST_GCS_LAST_BODY_GZIP")) { + TEST_CHECK(strcmp(getenv("TEST_GCS_LAST_BODY_GZIP"), "false") == 0); + } + TEST_CHECK_(getenv("TEST_GCS_LAST_CONTENT_TYPE") != NULL, + "Expected the mock upload content type to be captured"); + if (getenv("TEST_GCS_LAST_CONTENT_TYPE")) { + TEST_CHECK(strcmp(getenv("TEST_GCS_LAST_CONTENT_TYPE"), + "application/vnd.apache.parquet") == 0); + } + + flb_stop(ctx); + flb_destroy(ctx); + + unsetenv("FLB_GCS_PLUGIN_UNDER_TEST"); + unsetenv("TEST_GCS_UploadObject_CALL_COUNT"); + unsetenv("TEST_GCS_LAST_URI"); + unsetenv("TEST_GCS_LAST_BODY_GZIP"); + unsetenv("TEST_GCS_LAST_BODY_PARQUET"); + unsetenv("TEST_GCS_LAST_CONTENT_TYPE"); + flb_free(store_dir); +} + +void flb_test_gcs_upload_parquet_zstd(void) +{ + test_gcs_upload_parquet("ZSTD", "/flb-gcs-test-parquet-zstd-XXXXXX"); +} + +void flb_test_gcs_upload_parquet_snappy(void) +{ + test_gcs_upload_parquet("SNAPPY", "/flb-gcs-test-parquet-snappy-XXXXXX"); +} +#endif + +#ifndef FLB_HAVE_ARROW_PARQUET +void flb_test_gcs_rejects_parquet_without_support(void) +{ + int ret; + int in_ffd; + int out_ffd; + char *store_dir; + flb_ctx_t *ctx; + + store_dir = create_test_store_directory("/flb-gcs-test-no-parquet-XXXXXX"); + TEST_CHECK(store_dir != NULL); + if (!store_dir) { + return; + } + + 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 *) "gcs", NULL); + TEST_CHECK(out_ffd >= 0); + flb_output_set(ctx, out_ffd, "match", "*", NULL); + flb_output_set(ctx, out_ffd, "bucket", "fluent", NULL); + flb_output_set(ctx, out_ffd, "google_service_credentials", SERVICE_CREDENTIALS, NULL); + flb_output_set(ctx, out_ffd, "store_dir", store_dir, NULL); + flb_output_set(ctx, out_ffd, "format", "parquet", NULL); + + ret = flb_start(ctx); + TEST_CHECK(ret != 0); + + flb_destroy(ctx); + flb_free(store_dir); +} +#endif + void flb_test_gcs_rejects_invalid_configuration(void) { int ret; @@ -300,6 +433,8 @@ void flb_test_gcs_upload_error(void) unsetenv("TEST_GCS_UploadObject_CALL_COUNT"); unsetenv("TEST_GCS_LAST_URI"); unsetenv("TEST_GCS_LAST_BODY_GZIP"); + unsetenv("TEST_GCS_LAST_BODY_PARQUET"); + unsetenv("TEST_GCS_LAST_CONTENT_TYPE"); flb_free(store_dir); } @@ -387,6 +522,8 @@ void flb_test_gcs_shutdown_preserves_pending_upload(void) unsetenv("TEST_GCS_UploadObject_CALL_COUNT"); unsetenv("TEST_GCS_LAST_URI"); unsetenv("TEST_GCS_LAST_BODY_GZIP"); + unsetenv("TEST_GCS_LAST_BODY_PARQUET"); + unsetenv("TEST_GCS_LAST_CONTENT_TYPE"); flb_free(store_dir); } @@ -394,6 +531,12 @@ TEST_LIST = { {"jwt_signing", flb_test_gcs_jwt_signing}, {"uri_encode_object_name", flb_test_gcs_uri_encode_object_name}, {"upload_success", flb_test_gcs_upload_success}, +#ifdef FLB_HAVE_ARROW_PARQUET + {"upload_parquet_zstd", flb_test_gcs_upload_parquet_zstd}, + {"upload_parquet_snappy", flb_test_gcs_upload_parquet_snappy}, +#else + {"rejects_parquet_without_support", flb_test_gcs_rejects_parquet_without_support}, +#endif {"rejects_invalid_configuration", flb_test_gcs_rejects_invalid_configuration}, {"rejects_invalid_compression", flb_test_gcs_rejects_invalid_compression}, {"accepts_extra_credential_fields", flb_test_gcs_accepts_extra_credential_fields},