Skip to content

Commit 8b0ca6f

Browse files
ivanbushmarinov-pplxivanbushmarinov-pplx
committed
aws compress: support explicit compression level
Add flb_aws_compression_compress_level() and flb_aws_compression_b64_truncate_compress_level() taking an explicit compression level, with FLB_AWS_COMPRESS_LEVEL_DEFAULT preserving each codec's built-in default. The existing entry points delegate with the default sentinel, so behavior is unchanged for current callers. A non-default level is honored for zstd; codecs without a tunable level fall back to their default with a warning. Covered by a roundtrip internal test. Signed-off-by: Ivan Bushmarinov <ivan.bushmarinov@perplexity.ai>
1 parent bb24120 commit 8b0ca6f

3 files changed

Lines changed: 128 additions & 5 deletions

File tree

‎include/fluent-bit/aws/flb_aws_compress.h‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,25 @@ int flb_aws_compression_get_type(const char *compression_keyword);
4444
int flb_aws_compression_compress(int compression_type, void *in_data, size_t in_len,
4545
void **out_data, size_t *out_len);
4646

47+
/*
48+
* Sentinel for "no explicit compression level": each codec keeps its
49+
* historical built-in default (zstd: 1).
50+
*/
51+
#define FLB_AWS_COMPRESS_LEVEL_DEFAULT -1
52+
53+
/*
54+
* Same as flb_aws_compression_compress, with an explicit compression level.
55+
* Pass FLB_AWS_COMPRESS_LEVEL_DEFAULT to keep the codec's built-in default.
56+
* A non-default level is currently honored for zstd only; other codecs log a
57+
* warning once and compress at their built-in default.
58+
*
59+
* Returns -1 on error
60+
* Returns 0 on success
61+
*/
62+
int flb_aws_compression_compress_level(int compression_type, int compression_level,
63+
void *in_data, size_t in_len,
64+
void **out_data, size_t *out_len);
65+
4766
/*
4867
* Truncate and compress in_data and convert to b64
4968
* 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
6180
void *in_data, size_t in_len,
6281
void **out_data, size_t *out_len);
6382

83+
/*
84+
* Same as flb_aws_compression_b64_truncate_compress, with an explicit
85+
* compression level (see flb_aws_compression_compress_level).
86+
*/
87+
int flb_aws_compression_b64_truncate_compress_level(int compression_type,
88+
int compression_level,
89+
size_t max_out_len,
90+
void *in_data, size_t in_len,
91+
void **out_data, size_t *out_len);
92+
6493
/*
6594
* Columnar output formats for out_s3_compress_columnar(). Compression is
6695
* applied on top of the format via a generic FLB_AWS_COMPRESS_* codec.

‎src/aws/flb_aws_compress.c‎

Lines changed: 43 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,9 @@ struct compression_option {
4949
int compression_type;
5050
char *compression_keyword;
5151
int(*compress)(void *in_data, size_t in_len, void **out_data, size_t *out_len);
52+
/* NULL when the codec has no tunable level */
53+
int(*compress_level)(void *in_data, size_t in_len, void **out_data,
54+
size_t *out_len, int level);
5255
};
5356

5457
/*
@@ -61,17 +64,20 @@ static const struct compression_option compression_options[] = {
6164
{
6265
FLB_AWS_COMPRESS_GZIP,
6366
"gzip",
64-
&flb_gzip_compress
67+
&flb_gzip_compress,
68+
NULL
6569
},
6670
{
6771
FLB_AWS_COMPRESS_ZSTD,
6872
"zstd",
69-
&flb_zstd_compress
73+
&flb_zstd_compress,
74+
&flb_zstd_compress_level
7075
},
7176
{
7277
FLB_AWS_COMPRESS_SNAPPY,
7378
"snappy",
74-
&flb_snappy_compress_wrapper
79+
&flb_snappy_compress_wrapper,
80+
NULL
7581
},
7682
{ 0 }
7783
};
@@ -97,14 +103,33 @@ int flb_aws_compression_get_type(const char *compression_keyword)
97103

98104
int flb_aws_compression_compress(int compression_type, void *in_data, size_t in_len,
99105
void **out_data, size_t *out_len)
106+
{
107+
return flb_aws_compression_compress_level(compression_type,
108+
FLB_AWS_COMPRESS_LEVEL_DEFAULT,
109+
in_data, in_len, out_data, out_len);
110+
}
111+
112+
int flb_aws_compression_compress_level(int compression_type, int compression_level,
113+
void *in_data, size_t in_len,
114+
void **out_data, size_t *out_len)
100115
{
101116
const struct compression_option *o;
102117

103118
o = compression_options;
104119

105120
while (o->compression_type != 0) {
106121
if (o->compression_type == compression_type) {
107-
return o->compress(in_data, in_len, out_data, out_len);
122+
if (compression_level == FLB_AWS_COMPRESS_LEVEL_DEFAULT) {
123+
return o->compress(in_data, in_len, out_data, out_len);
124+
}
125+
if (o->compress_level == NULL) {
126+
flb_warn("[aws_compress] compression level %i ignored: '%s' has "
127+
"no tunable level, using its built-in default",
128+
compression_level, o->compression_keyword);
129+
return o->compress(in_data, in_len, out_data, out_len);
130+
}
131+
return o->compress_level(in_data, in_len, out_data, out_len,
132+
compression_level);
108133
}
109134
++o;
110135
}
@@ -117,6 +142,18 @@ int flb_aws_compression_compress(int compression_type, void *in_data, size_t in_
117142
int flb_aws_compression_b64_truncate_compress(int compression_type, size_t max_out_len,
118143
void *in_data, size_t in_len,
119144
void **out_data, size_t *out_len)
145+
{
146+
return flb_aws_compression_b64_truncate_compress_level(compression_type,
147+
FLB_AWS_COMPRESS_LEVEL_DEFAULT,
148+
max_out_len, in_data, in_len,
149+
out_data, out_len);
150+
}
151+
152+
int flb_aws_compression_b64_truncate_compress_level(int compression_type,
153+
int compression_level,
154+
size_t max_out_len,
155+
void *in_data, size_t in_len,
156+
void **out_data, size_t *out_len)
120157
{
121158
static const void *truncation_suffix = "[Truncated...]";
122159
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
154191
return -1;
155192
}
156193

157-
ret = flb_aws_compression_compress(compression_type, truncated_in_buf,
194+
ret = flb_aws_compression_compress_level(compression_type, compression_level,
195+
truncated_in_buf,
158196
truncated_in_len, &compressed_buf,
159197
&compressed_len);
160198
++compression_attempts;

‎tests/internal/aws_compress.c‎

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,61 @@ void test_compression_zstd()
8585
flb_aws_compress_test_cases(cases);
8686
}
8787

88+
void test_compression_zstd_level()
89+
{
90+
int ret;
91+
void *out_default = NULL;
92+
void *out_high = NULL;
93+
void *roundtrip = NULL;
94+
size_t out_default_len = 0;
95+
size_t out_high_len = 0;
96+
size_t roundtrip_len = 0;
97+
/* repetitive input so higher levels have something to win on */
98+
char in_data[512];
99+
size_t i;
100+
101+
for (i = 0; i < sizeof(in_data); i++) {
102+
in_data[i] = "hello world "[i % 12];
103+
}
104+
105+
/* explicit level compresses and roundtrips */
106+
ret = flb_aws_compression_compress_level(FLB_AWS_COMPRESS_ZSTD, 19,
107+
in_data, sizeof(in_data),
108+
&out_high, &out_high_len);
109+
TEST_CHECK(ret == 0);
110+
ret = flb_zstd_uncompress(out_high, out_high_len, &roundtrip, &roundtrip_len);
111+
TEST_CHECK(ret == 0);
112+
TEST_CHECK(roundtrip_len == sizeof(in_data));
113+
TEST_CHECK(memcmp(roundtrip, in_data, sizeof(in_data)) == 0);
114+
flb_free(roundtrip);
115+
116+
/* sentinel level matches the plain API's built-in default */
117+
ret = flb_aws_compression_compress_level(FLB_AWS_COMPRESS_ZSTD,
118+
FLB_AWS_COMPRESS_LEVEL_DEFAULT,
119+
in_data, sizeof(in_data),
120+
&out_default, &out_default_len);
121+
TEST_CHECK(ret == 0);
122+
123+
/* a higher level must not produce a larger frame on repetitive input */
124+
TEST_CHECK(out_high_len <= out_default_len);
125+
126+
/* out-of-range level fails */
127+
ret = flb_aws_compression_compress_level(FLB_AWS_COMPRESS_ZSTD, 1000000,
128+
in_data, sizeof(in_data),
129+
&roundtrip, &roundtrip_len);
130+
TEST_CHECK(ret == -1);
131+
132+
/* codec without tunable level falls back to its default and succeeds */
133+
ret = flb_aws_compression_compress_level(FLB_AWS_COMPRESS_GZIP, 9,
134+
in_data, sizeof(in_data),
135+
&roundtrip, &roundtrip_len);
136+
TEST_CHECK(ret == 0);
137+
flb_free(roundtrip);
138+
139+
flb_free(out_default);
140+
flb_free(out_high);
141+
}
142+
88143
void test_compression_snappy()
89144
{
90145
struct flb_aws_test_case cases[] =
@@ -525,6 +580,7 @@ void test_arrow_format_gzip_unsupported()
525580
TEST_LIST = {
526581
{ "test_compression_gzip", test_compression_gzip },
527582
{ "test_compression_zstd", test_compression_zstd },
583+
{ "test_compression_zstd_level", test_compression_zstd_level },
528584
{ "test_compression_snappy", test_compression_snappy },
529585
{ "test_compression_snappy_return_value_normalization",
530586
test_compression_snappy_return_value_normalization },

0 commit comments

Comments
 (0)