summaryrefslogtreecommitdiffstats
path: root/src/streaming/stream-compression
diff options
context:
space:
mode:
Diffstat (limited to 'src/streaming/stream-compression')
-rw-r--r--src/streaming/stream-compression/brotli.c142
-rw-r--r--src/streaming/stream-compression/brotli.h15
-rw-r--r--src/streaming/stream-compression/compression.c703
-rw-r--r--src/streaming/stream-compression/compression.h183
-rw-r--r--src/streaming/stream-compression/gzip.c164
-rw-r--r--src/streaming/stream-compression/gzip.h15
-rw-r--r--src/streaming/stream-compression/lz4.c143
-rw-r--r--src/streaming/stream-compression/lz4.h19
-rw-r--r--src/streaming/stream-compression/zstd.c163
-rw-r--r--src/streaming/stream-compression/zstd.h19
10 files changed, 1566 insertions, 0 deletions
diff --git a/src/streaming/stream-compression/brotli.c b/src/streaming/stream-compression/brotli.c
new file mode 100644
index 000000000..c2c09cdc5
--- /dev/null
+++ b/src/streaming/stream-compression/brotli.c
@@ -0,0 +1,142 @@
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#include "brotli.h"
+
+#ifdef ENABLE_BROTLI
+#include <brotli/encode.h>
+#include <brotli/decode.h>
+
+void rrdpush_compressor_init_brotli(struct compressor_state *state) {
+ if (!state->initialized) {
+ state->initialized = true;
+ state->stream = BrotliEncoderCreateInstance(NULL, NULL, NULL);
+
+ if (state->level < BROTLI_MIN_QUALITY) {
+ state->level = BROTLI_MIN_QUALITY;
+ } else if (state->level > BROTLI_MAX_QUALITY) {
+ state->level = BROTLI_MAX_QUALITY;
+ }
+
+ BrotliEncoderSetParameter(state->stream, BROTLI_PARAM_QUALITY, state->level);
+ }
+}
+
+void rrdpush_compressor_destroy_brotli(struct compressor_state *state) {
+ if (state->stream) {
+ BrotliEncoderDestroyInstance(state->stream);
+ state->stream = NULL;
+ }
+}
+
+size_t rrdpush_compress_brotli(struct compressor_state *state, const char *data, size_t size, const char **out) {
+ if (unlikely(!state || !size || !out))
+ return 0;
+
+ simple_ring_buffer_make_room(&state->output, MAX(BrotliEncoderMaxCompressedSize(size), COMPRESSION_MAX_CHUNK));
+
+ size_t available_out = state->output.size;
+
+ size_t available_in = size;
+ const uint8_t *next_in = (const uint8_t *)data;
+ uint8_t *next_out = (uint8_t *)state->output.data;
+
+ if (!BrotliEncoderCompressStream(state->stream, BROTLI_OPERATION_FLUSH, &available_in, &next_in, &available_out, &next_out, NULL)) {
+ netdata_log_error("STREAM: Brotli compression failed.");
+ return 0;
+ }
+
+ if(available_in != 0) {
+ netdata_log_error("STREAM: BrotliEncoderCompressStream() did not use all the input buffer, %zu bytes out of %zu remain",
+ available_in, size);
+ return 0;
+ }
+
+ size_t compressed_size = state->output.size - available_out;
+ if(available_out == 0) {
+ netdata_log_error("STREAM: BrotliEncoderCompressStream() needs a bigger output buffer than the one we provided "
+ "(output buffer %zu bytes, compressed payload %zu bytes)",
+ state->output.size, size);
+ return 0;
+ }
+
+ if(compressed_size == 0) {
+ netdata_log_error("STREAM: BrotliEncoderCompressStream() did not produce any output from the input provided "
+ "(input buffer %zu bytes)",
+ size);
+ return 0;
+ }
+
+ state->sender_locked.total_compressions++;
+ state->sender_locked.total_uncompressed += size - available_in;
+ state->sender_locked.total_compressed += compressed_size;
+
+ *out = state->output.data;
+ return compressed_size;
+}
+
+void rrdpush_decompressor_init_brotli(struct decompressor_state *state) {
+ if (!state->initialized) {
+ state->initialized = true;
+ state->stream = BrotliDecoderCreateInstance(NULL, NULL, NULL);
+
+ simple_ring_buffer_make_room(&state->output, COMPRESSION_MAX_CHUNK);
+ }
+}
+
+void rrdpush_decompressor_destroy_brotli(struct decompressor_state *state) {
+ if (state->stream) {
+ BrotliDecoderDestroyInstance(state->stream);
+ state->stream = NULL;
+ }
+}
+
+size_t rrdpush_decompress_brotli(struct decompressor_state *state, const char *compressed_data, size_t compressed_size) {
+ if (unlikely(!state || !compressed_data || !compressed_size))
+ return 0;
+
+ // The state.output ring buffer is always EMPTY at this point,
+ // meaning that (state->output.read_pos == state->output.write_pos)
+ // However, THEY ARE NOT ZERO.
+
+ size_t available_out = state->output.size;
+ size_t available_in = compressed_size;
+ const uint8_t *next_in = (const uint8_t *)compressed_data;
+ uint8_t *next_out = (uint8_t *)state->output.data;
+
+ if (BrotliDecoderDecompressStream(state->stream, &available_in, &next_in, &available_out, &next_out, NULL) == BROTLI_DECODER_RESULT_ERROR) {
+ netdata_log_error("STREAM: Brotli decompression failed.");
+ return 0;
+ }
+
+ if(available_in != 0) {
+ netdata_log_error("STREAM: BrotliDecoderDecompressStream() did not use all the input buffer, %zu bytes out of %zu remain",
+ available_in, compressed_size);
+ return 0;
+ }
+
+ size_t decompressed_size = state->output.size - available_out;
+ if(available_out == 0) {
+ netdata_log_error("STREAM: BrotliDecoderDecompressStream() needs a bigger output buffer than the one we provided "
+ "(output buffer %zu bytes, compressed payload %zu bytes)",
+ state->output.size, compressed_size);
+ return 0;
+ }
+
+ if(decompressed_size == 0) {
+ netdata_log_error("STREAM: BrotliDecoderDecompressStream() did not produce any output from the input provided "
+ "(input buffer %zu bytes)",
+ compressed_size);
+ return 0;
+ }
+
+ state->output.read_pos = 0;
+ state->output.write_pos = decompressed_size;
+
+ state->total_compressed += compressed_size - available_in;
+ state->total_uncompressed += decompressed_size;
+ state->total_compressions++;
+
+ return decompressed_size;
+}
+
+#endif // ENABLE_BROTLI
diff --git a/src/streaming/stream-compression/brotli.h b/src/streaming/stream-compression/brotli.h
new file mode 100644
index 000000000..4955e5a82
--- /dev/null
+++ b/src/streaming/stream-compression/brotli.h
@@ -0,0 +1,15 @@
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#include "compression.h"
+
+#ifndef NETDATA_STREAMING_COMPRESSION_BROTLI_H
+#define NETDATA_STREAMING_COMPRESSION_BROTLI_H
+
+void rrdpush_compressor_init_brotli(struct compressor_state *state);
+void rrdpush_compressor_destroy_brotli(struct compressor_state *state);
+size_t rrdpush_compress_brotli(struct compressor_state *state, const char *data, size_t size, const char **out);
+size_t rrdpush_decompress_brotli(struct decompressor_state *state, const char *compressed_data, size_t compressed_size);
+void rrdpush_decompressor_init_brotli(struct decompressor_state *state);
+void rrdpush_decompressor_destroy_brotli(struct decompressor_state *state);
+
+#endif //NETDATA_STREAMING_COMPRESSION_BROTLI_H
diff --git a/src/streaming/stream-compression/compression.c b/src/streaming/stream-compression/compression.c
new file mode 100644
index 000000000..3c9930656
--- /dev/null
+++ b/src/streaming/stream-compression/compression.c
@@ -0,0 +1,703 @@
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#include "compression.h"
+
+#include "gzip.h"
+
+#ifdef ENABLE_LZ4
+#include "lz4.h"
+#endif
+
+#ifdef ENABLE_ZSTD
+#include "zstd.h"
+#endif
+
+#ifdef ENABLE_BROTLI
+#include "brotli.h"
+#endif
+
+int rrdpush_compression_levels[COMPRESSION_ALGORITHM_MAX] = {
+ [COMPRESSION_ALGORITHM_NONE] = 0,
+ [COMPRESSION_ALGORITHM_ZSTD] = 3, // 1 (faster) - 22 (smaller)
+ [COMPRESSION_ALGORITHM_LZ4] = 1, // 1 (smaller) - 9 (faster)
+ [COMPRESSION_ALGORITHM_BROTLI] = 3, // 0 (faster) - 11 (smaller)
+ [COMPRESSION_ALGORITHM_GZIP] = 1, // 1 (faster) - 9 (smaller)
+};
+
+void rrdpush_parse_compression_order(struct receiver_state *rpt, const char *order) {
+ // empty all slots
+ for(size_t i = 0; i < COMPRESSION_ALGORITHM_MAX ;i++)
+ rpt->config.compression_priorities[i] = STREAM_CAP_NONE;
+
+ char *s = strdupz(order);
+
+ char *words[COMPRESSION_ALGORITHM_MAX + 100] = { NULL };
+ size_t num_words = quoted_strings_splitter_whitespace(s, words, COMPRESSION_ALGORITHM_MAX + 100);
+ size_t slot = 0;
+ STREAM_CAPABILITIES added = STREAM_CAP_NONE;
+ for(size_t i = 0; i < num_words && slot < COMPRESSION_ALGORITHM_MAX ;i++) {
+ if((STREAM_CAP_ZSTD_AVAILABLE) && strcasecmp(words[i], "zstd") == 0 && !(added & STREAM_CAP_ZSTD)) {
+ rpt->config.compression_priorities[slot++] = STREAM_CAP_ZSTD;
+ added |= STREAM_CAP_ZSTD;
+ }
+ else if((STREAM_CAP_LZ4_AVAILABLE) && strcasecmp(words[i], "lz4") == 0 && !(added & STREAM_CAP_LZ4)) {
+ rpt->config.compression_priorities[slot++] = STREAM_CAP_LZ4;
+ added |= STREAM_CAP_LZ4;
+ }
+ else if((STREAM_CAP_BROTLI_AVAILABLE) && strcasecmp(words[i], "brotli") == 0 && !(added & STREAM_CAP_BROTLI)) {
+ rpt->config.compression_priorities[slot++] = STREAM_CAP_BROTLI;
+ added |= STREAM_CAP_BROTLI;
+ }
+ else if(strcasecmp(words[i], "gzip") == 0 && !(added & STREAM_CAP_GZIP)) {
+ rpt->config.compression_priorities[slot++] = STREAM_CAP_GZIP;
+ added |= STREAM_CAP_GZIP;
+ }
+ }
+
+ freez(s);
+
+ // make sure all participate
+ if((STREAM_CAP_ZSTD_AVAILABLE) && slot < COMPRESSION_ALGORITHM_MAX && !(added & STREAM_CAP_ZSTD))
+ rpt->config.compression_priorities[slot++] = STREAM_CAP_ZSTD;
+ if((STREAM_CAP_LZ4_AVAILABLE) && slot < COMPRESSION_ALGORITHM_MAX && !(added & STREAM_CAP_LZ4))
+ rpt->config.compression_priorities[slot++] = STREAM_CAP_LZ4;
+ if((STREAM_CAP_BROTLI_AVAILABLE) && slot < COMPRESSION_ALGORITHM_MAX && !(added & STREAM_CAP_BROTLI))
+ rpt->config.compression_priorities[slot++] = STREAM_CAP_BROTLI;
+ if(slot < COMPRESSION_ALGORITHM_MAX && !(added & STREAM_CAP_GZIP))
+ rpt->config.compression_priorities[slot++] = STREAM_CAP_GZIP;
+}
+
+void rrdpush_select_receiver_compression_algorithm(struct receiver_state *rpt) {
+ if (!rpt->config.rrdpush_compression)
+ rpt->capabilities &= ~STREAM_CAP_COMPRESSIONS_AVAILABLE;
+
+ // select the right compression before sending our capabilities to the child
+ if(stream_has_more_than_one_capability_of(rpt->capabilities, STREAM_CAP_COMPRESSIONS_AVAILABLE)) {
+ STREAM_CAPABILITIES compressions = rpt->capabilities & STREAM_CAP_COMPRESSIONS_AVAILABLE;
+ for(int i = 0; i < COMPRESSION_ALGORITHM_MAX; i++) {
+ STREAM_CAPABILITIES c = rpt->config.compression_priorities[i];
+
+ if(!(c & STREAM_CAP_COMPRESSIONS_AVAILABLE))
+ continue;
+
+ if(compressions & c) {
+ STREAM_CAPABILITIES exclude = compressions;
+ exclude &= ~c;
+
+ rpt->capabilities &= ~exclude;
+ break;
+ }
+ }
+ }
+}
+
+bool rrdpush_compression_initialize(struct sender_state *s) {
+ rrdpush_compressor_destroy(&s->compressor);
+
+ // IMPORTANT
+ // KEEP THE SAME ORDER IN DECOMPRESSION
+
+ if(stream_has_capability(s, STREAM_CAP_ZSTD))
+ s->compressor.algorithm = COMPRESSION_ALGORITHM_ZSTD;
+ else if(stream_has_capability(s, STREAM_CAP_LZ4))
+ s->compressor.algorithm = COMPRESSION_ALGORITHM_LZ4;
+ else if(stream_has_capability(s, STREAM_CAP_BROTLI))
+ s->compressor.algorithm = COMPRESSION_ALGORITHM_BROTLI;
+ else if(stream_has_capability(s, STREAM_CAP_GZIP))
+ s->compressor.algorithm = COMPRESSION_ALGORITHM_GZIP;
+ else
+ s->compressor.algorithm = COMPRESSION_ALGORITHM_NONE;
+
+ if(s->compressor.algorithm != COMPRESSION_ALGORITHM_NONE) {
+ s->compressor.level = rrdpush_compression_levels[s->compressor.algorithm];
+ rrdpush_compressor_init(&s->compressor);
+ return true;
+ }
+
+ return false;
+}
+
+bool rrdpush_decompression_initialize(struct receiver_state *rpt) {
+ rrdpush_decompressor_destroy(&rpt->decompressor);
+
+ // IMPORTANT
+ // KEEP THE SAME ORDER IN COMPRESSION
+
+ if(stream_has_capability(rpt, STREAM_CAP_ZSTD))
+ rpt->decompressor.algorithm = COMPRESSION_ALGORITHM_ZSTD;
+ else if(stream_has_capability(rpt, STREAM_CAP_LZ4))
+ rpt->decompressor.algorithm = COMPRESSION_ALGORITHM_LZ4;
+ else if(stream_has_capability(rpt, STREAM_CAP_BROTLI))
+ rpt->decompressor.algorithm = COMPRESSION_ALGORITHM_BROTLI;
+ else if(stream_has_capability(rpt, STREAM_CAP_GZIP))
+ rpt->decompressor.algorithm = COMPRESSION_ALGORITHM_GZIP;
+ else
+ rpt->decompressor.algorithm = COMPRESSION_ALGORITHM_NONE;
+
+ if(rpt->decompressor.algorithm != COMPRESSION_ALGORITHM_NONE) {
+ rrdpush_decompressor_init(&rpt->decompressor);
+ return true;
+ }
+
+ return false;
+}
+
+/*
+* In case of stream compression buffer overflow
+* Inform the user through the error log file and
+* deactivate compression by downgrading the stream protocol.
+*/
+void rrdpush_compression_deactivate(struct sender_state *s) {
+ switch(s->compressor.algorithm) {
+ case COMPRESSION_ALGORITHM_MAX:
+ case COMPRESSION_ALGORITHM_NONE:
+ netdata_log_error("STREAM_COMPRESSION: compression error on 'host:%s' without any compression enabled. Ignoring error.",
+ rrdhost_hostname(s->host));
+ break;
+
+ case COMPRESSION_ALGORITHM_GZIP:
+ netdata_log_error("STREAM_COMPRESSION: GZIP compression error on 'host:%s'. Disabling GZIP for this node.",
+ rrdhost_hostname(s->host));
+ s->disabled_capabilities |= STREAM_CAP_GZIP;
+ break;
+
+ case COMPRESSION_ALGORITHM_LZ4:
+ netdata_log_error("STREAM_COMPRESSION: LZ4 compression error on 'host:%s'. Disabling ZSTD for this node.",
+ rrdhost_hostname(s->host));
+ s->disabled_capabilities |= STREAM_CAP_LZ4;
+ break;
+
+ case COMPRESSION_ALGORITHM_ZSTD:
+ netdata_log_error("STREAM_COMPRESSION: ZSTD compression error on 'host:%s'. Disabling ZSTD for this node.",
+ rrdhost_hostname(s->host));
+ s->disabled_capabilities |= STREAM_CAP_ZSTD;
+ break;
+
+ case COMPRESSION_ALGORITHM_BROTLI:
+ netdata_log_error("STREAM_COMPRESSION: BROTLI compression error on 'host:%s'. Disabling BROTLI for this node.",
+ rrdhost_hostname(s->host));
+ s->disabled_capabilities |= STREAM_CAP_BROTLI;
+ break;
+ }
+}
+
+// ----------------------------------------------------------------------------
+// compressor public API
+
+void rrdpush_compressor_init(struct compressor_state *state) {
+ switch(state->algorithm) {
+#ifdef ENABLE_ZSTD
+ case COMPRESSION_ALGORITHM_ZSTD:
+ rrdpush_compressor_init_zstd(state);
+ break;
+#endif
+
+#ifdef ENABLE_LZ4
+ case COMPRESSION_ALGORITHM_LZ4:
+ rrdpush_compressor_init_lz4(state);
+ break;
+#endif
+
+#ifdef ENABLE_BROTLI
+ case COMPRESSION_ALGORITHM_BROTLI:
+ rrdpush_compressor_init_brotli(state);
+ break;
+#endif
+
+ default:
+ case COMPRESSION_ALGORITHM_GZIP:
+ rrdpush_compressor_init_gzip(state);
+ break;
+ }
+
+ simple_ring_buffer_reset(&state->input);
+ simple_ring_buffer_reset(&state->output);
+}
+
+void rrdpush_compressor_destroy(struct compressor_state *state) {
+ switch(state->algorithm) {
+#ifdef ENABLE_ZSTD
+ case COMPRESSION_ALGORITHM_ZSTD:
+ rrdpush_compressor_destroy_zstd(state);
+ break;
+#endif
+
+#ifdef ENABLE_LZ4
+ case COMPRESSION_ALGORITHM_LZ4:
+ rrdpush_compressor_destroy_lz4(state);
+ break;
+#endif
+
+#ifdef ENABLE_BROTLI
+ case COMPRESSION_ALGORITHM_BROTLI:
+ rrdpush_compressor_destroy_brotli(state);
+ break;
+#endif
+
+ default:
+ case COMPRESSION_ALGORITHM_GZIP:
+ rrdpush_compressor_destroy_gzip(state);
+ break;
+ }
+
+ state->initialized = false;
+
+ simple_ring_buffer_destroy(&state->input);
+ simple_ring_buffer_destroy(&state->output);
+}
+
+size_t rrdpush_compress(struct compressor_state *state, const char *data, size_t size, const char **out) {
+ size_t ret = 0;
+
+ switch(state->algorithm) {
+#ifdef ENABLE_ZSTD
+ case COMPRESSION_ALGORITHM_ZSTD:
+ ret = rrdpush_compress_zstd(state, data, size, out);
+ break;
+#endif
+
+#ifdef ENABLE_LZ4
+ case COMPRESSION_ALGORITHM_LZ4:
+ ret = rrdpush_compress_lz4(state, data, size, out);
+ break;
+#endif
+
+#ifdef ENABLE_BROTLI
+ case COMPRESSION_ALGORITHM_BROTLI:
+ ret = rrdpush_compress_brotli(state, data, size, out);
+ break;
+#endif
+
+ default:
+ case COMPRESSION_ALGORITHM_GZIP:
+ ret = rrdpush_compress_gzip(state, data, size, out);
+ break;
+ }
+
+ if(unlikely(ret >= COMPRESSION_MAX_CHUNK)) {
+ netdata_log_error("RRDPUSH_COMPRESS: compressed data is %zu bytes, which is >= than the max chunk size %d",
+ ret, COMPRESSION_MAX_CHUNK);
+ return 0;
+ }
+
+ return ret;
+}
+
+// ----------------------------------------------------------------------------
+// decompressor public API
+
+void rrdpush_decompressor_destroy(struct decompressor_state *state) {
+ if(unlikely(!state->initialized))
+ return;
+
+ switch(state->algorithm) {
+#ifdef ENABLE_ZSTD
+ case COMPRESSION_ALGORITHM_ZSTD:
+ rrdpush_decompressor_destroy_zstd(state);
+ break;
+#endif
+
+#ifdef ENABLE_LZ4
+ case COMPRESSION_ALGORITHM_LZ4:
+ rrdpush_decompressor_destroy_lz4(state);
+ break;
+#endif
+
+#ifdef ENABLE_BROTLI
+ case COMPRESSION_ALGORITHM_BROTLI:
+ rrdpush_decompressor_destroy_brotli(state);
+ break;
+#endif
+
+ default:
+ case COMPRESSION_ALGORITHM_GZIP:
+ rrdpush_decompressor_destroy_gzip(state);
+ break;
+ }
+
+ simple_ring_buffer_destroy(&state->output);
+
+ state->initialized = false;
+}
+
+void rrdpush_decompressor_init(struct decompressor_state *state) {
+ switch(state->algorithm) {
+#ifdef ENABLE_ZSTD
+ case COMPRESSION_ALGORITHM_ZSTD:
+ rrdpush_decompressor_init_zstd(state);
+ break;
+#endif
+
+#ifdef ENABLE_LZ4
+ case COMPRESSION_ALGORITHM_LZ4:
+ rrdpush_decompressor_init_lz4(state);
+ break;
+#endif
+
+#ifdef ENABLE_BROTLI
+ case COMPRESSION_ALGORITHM_BROTLI:
+ rrdpush_decompressor_init_brotli(state);
+ break;
+#endif
+
+ default:
+ case COMPRESSION_ALGORITHM_GZIP:
+ rrdpush_decompressor_init_gzip(state);
+ break;
+ }
+
+ state->signature_size = RRDPUSH_COMPRESSION_SIGNATURE_SIZE;
+ simple_ring_buffer_reset(&state->output);
+}
+
+size_t rrdpush_decompress(struct decompressor_state *state, const char *compressed_data, size_t compressed_size) {
+ if (unlikely(state->output.read_pos != state->output.write_pos))
+ fatal("RRDPUSH_DECOMPRESS: asked to decompress new data, while there are unread data in the decompression buffer!");
+
+ size_t ret = 0;
+
+ switch(state->algorithm) {
+#ifdef ENABLE_ZSTD
+ case COMPRESSION_ALGORITHM_ZSTD:
+ ret = rrdpush_decompress_zstd(state, compressed_data, compressed_size);
+ break;
+#endif
+
+#ifdef ENABLE_LZ4
+ case COMPRESSION_ALGORITHM_LZ4:
+ ret = rrdpush_decompress_lz4(state, compressed_data, compressed_size);
+ break;
+#endif
+
+#ifdef ENABLE_BROTLI
+ case COMPRESSION_ALGORITHM_BROTLI:
+ ret = rrdpush_decompress_brotli(state, compressed_data, compressed_size);
+ break;
+#endif
+
+ default:
+ case COMPRESSION_ALGORITHM_GZIP:
+ ret = rrdpush_decompress_gzip(state, compressed_data, compressed_size);
+ break;
+ }
+
+ // for backwards compatibility we cannot check for COMPRESSION_MAX_MSG_SIZE,
+ // because old children may send this big payloads.
+ if(unlikely(ret > COMPRESSION_MAX_CHUNK)) {
+ netdata_log_error("RRDPUSH_DECOMPRESS: decompressed data is %zu bytes, which is bigger than the max msg size %d",
+ ret, COMPRESSION_MAX_CHUNK);
+ return 0;
+ }
+
+ return ret;
+}
+
+// ----------------------------------------------------------------------------
+// unit test
+
+void unittest_generate_random_name(char *dst, size_t size) {
+ if(size < 7)
+ size = 7;
+
+ size_t len = 5 + os_random32() % (size - 6);
+
+ for(size_t i = 0; i < len ; i++) {
+ if(os_random8() % 2 == 0)
+ dst[i] = 'A' + os_random8() % 26;
+ else
+ dst[i] = 'a' + os_random8() % 26;
+ }
+
+ dst[len] = '\0';
+}
+
+void unittest_generate_message(BUFFER *wb, time_t now_s, size_t counter) {
+ bool with_slots = true;
+ NUMBER_ENCODING integer_encoding = NUMBER_ENCODING_BASE64;
+ NUMBER_ENCODING doubles_encoding = NUMBER_ENCODING_BASE64;
+ time_t update_every = 1;
+ time_t point_end_time_s = now_s;
+ time_t wall_clock_time_s = now_s;
+ size_t chart_slot = counter + 1;
+ size_t dimensions = 2 + os_random8() % 5;
+ char chart[RRD_ID_LENGTH_MAX + 1] = "name";
+ unittest_generate_random_name(chart, 5 + os_random8() % 30);
+
+ buffer_fast_strcat(wb, PLUGINSD_KEYWORD_BEGIN_V2, sizeof(PLUGINSD_KEYWORD_BEGIN_V2) - 1);
+
+ if(with_slots) {
+ buffer_fast_strcat(wb, " "PLUGINSD_KEYWORD_SLOT":", sizeof(PLUGINSD_KEYWORD_SLOT) - 1 + 2);
+ buffer_print_uint64_encoded(wb, integer_encoding, chart_slot);
+ }
+
+ buffer_fast_strcat(wb, " '", 2);
+ buffer_strcat(wb, chart);
+ buffer_fast_strcat(wb, "' ", 2);
+ buffer_print_uint64_encoded(wb, integer_encoding, update_every);
+ buffer_fast_strcat(wb, " ", 1);
+ buffer_print_uint64_encoded(wb, integer_encoding, point_end_time_s);
+ buffer_fast_strcat(wb, " ", 1);
+ if(point_end_time_s == wall_clock_time_s)
+ buffer_fast_strcat(wb, "#", 1);
+ else
+ buffer_print_uint64_encoded(wb, integer_encoding, wall_clock_time_s);
+ buffer_fast_strcat(wb, "\n", 1);
+
+
+ for(size_t d = 0; d < dimensions ;d++) {
+ size_t dim_slot = d + 1;
+ char dim_id[RRD_ID_LENGTH_MAX + 1] = "dimension";
+ unittest_generate_random_name(dim_id, 10 + os_random8() % 20);
+ int64_t last_collected_value = (os_random8() % 2 == 0) ? (int64_t)(counter + d) : (int64_t)os_random32();
+ NETDATA_DOUBLE value = (os_random8() % 2 == 0) ? (NETDATA_DOUBLE)os_random64() / ((NETDATA_DOUBLE)os_random64() + 1) : (NETDATA_DOUBLE)last_collected_value;
+ SN_FLAGS flags = (os_random16() % 1000 == 0) ? SN_FLAG_NONE : SN_FLAG_NOT_ANOMALOUS;
+
+ buffer_fast_strcat(wb, PLUGINSD_KEYWORD_SET_V2, sizeof(PLUGINSD_KEYWORD_SET_V2) - 1);
+
+ if(with_slots) {
+ buffer_fast_strcat(wb, " "PLUGINSD_KEYWORD_SLOT":", sizeof(PLUGINSD_KEYWORD_SLOT) - 1 + 2);
+ buffer_print_uint64_encoded(wb, integer_encoding, dim_slot);
+ }
+
+ buffer_fast_strcat(wb, " '", 2);
+ buffer_strcat(wb, dim_id);
+ buffer_fast_strcat(wb, "' ", 2);
+ buffer_print_int64_encoded(wb, integer_encoding, last_collected_value);
+ buffer_fast_strcat(wb, " ", 1);
+
+ if((NETDATA_DOUBLE)last_collected_value == value)
+ buffer_fast_strcat(wb, "#", 1);
+ else
+ buffer_print_netdata_double_encoded(wb, doubles_encoding, value);
+
+ buffer_fast_strcat(wb, " ", 1);
+ buffer_print_sn_flags(wb, flags, true);
+ buffer_fast_strcat(wb, "\n", 1);
+ }
+
+ buffer_fast_strcat(wb, PLUGINSD_KEYWORD_END_V2 "\n", sizeof(PLUGINSD_KEYWORD_END_V2) - 1 + 1);
+}
+
+int unittest_rrdpush_compression_speed(compression_algorithm_t algorithm, const char *name) {
+ fprintf(stderr, "\nTesting streaming compression speed with %s\n", name);
+
+ struct compressor_state cctx = {
+ .initialized = false,
+ .algorithm = algorithm,
+ };
+ struct decompressor_state dctx = {
+ .initialized = false,
+ .algorithm = algorithm,
+ };
+
+ rrdpush_compressor_init(&cctx);
+ rrdpush_decompressor_init(&dctx);
+
+ int errors = 0;
+
+ BUFFER *wb = buffer_create(COMPRESSION_MAX_MSG_SIZE, NULL);
+ time_t now_s = now_realtime_sec();
+ usec_t compression_ut = 0;
+ usec_t decompression_ut = 0;
+ size_t bytes_compressed = 0;
+ size_t bytes_uncompressed = 0;
+
+ usec_t compression_started_ut = now_monotonic_usec();
+ usec_t decompression_started_ut = compression_started_ut;
+
+ for(int i = 0; i < 10000 ;i++) {
+ compression_started_ut = now_monotonic_usec();
+ decompression_ut += compression_started_ut - decompression_started_ut;
+
+ buffer_flush(wb);
+ while(buffer_strlen(wb) < COMPRESSION_MAX_MSG_SIZE - 1024)
+ unittest_generate_message(wb, now_s, i);
+
+ const char *txt = buffer_tostring(wb);
+ size_t txt_len = buffer_strlen(wb);
+ bytes_uncompressed += txt_len;
+
+ const char *out;
+ size_t size = rrdpush_compress(&cctx, txt, txt_len, &out);
+
+ bytes_compressed += size;
+ decompression_started_ut = now_monotonic_usec();
+ compression_ut += decompression_started_ut - compression_started_ut;
+
+ if(size == 0) {
+ fprintf(stderr, "iteration %d: compressed size %zu is zero\n",
+ i, size);
+ errors++;
+ goto cleanup;
+ }
+ else if(size >= COMPRESSION_MAX_CHUNK) {
+ fprintf(stderr, "iteration %d: compressed size %zu exceeds max allowed size\n",
+ i, size);
+ errors++;
+ goto cleanup;
+ }
+ else {
+ size_t dtxt_len = rrdpush_decompress(&dctx, out, size);
+ char *dtxt = (char *) &dctx.output.data[dctx.output.read_pos];
+
+ if(rrdpush_decompressed_bytes_in_buffer(&dctx) != dtxt_len) {
+ fprintf(stderr, "iteration %d: decompressed size %zu does not rrdpush_decompressed_bytes_in_buffer() %zu\n",
+ i, dtxt_len, rrdpush_decompressed_bytes_in_buffer(&dctx)
+ );
+ errors++;
+ goto cleanup;
+ }
+
+ if(!dtxt_len) {
+ fprintf(stderr, "iteration %d: decompressed size is zero\n", i);
+ errors++;
+ goto cleanup;
+ }
+ else if(dtxt_len != txt_len) {
+ fprintf(stderr, "iteration %d: decompressed size %zu does not match original size %zu\n",
+ i, dtxt_len, txt_len
+ );
+ errors++;
+ goto cleanup;
+ }
+ else {
+ if(memcmp(txt, dtxt, txt_len) != 0) {
+ fprintf(stderr, "iteration %d: decompressed data '%s' do not match original data length %zu\n",
+ i, dtxt, txt_len);
+ errors++;
+ goto cleanup;
+ }
+ }
+ }
+
+ // here we are supposed to copy the data and advance the position
+ dctx.output.read_pos += rrdpush_decompressed_bytes_in_buffer(&dctx);
+ }
+
+cleanup:
+ rrdpush_compressor_destroy(&cctx);
+ rrdpush_decompressor_destroy(&dctx);
+
+ if(errors)
+ fprintf(stderr, "Compression with %s: FAILED (%d errors)\n", name, errors);
+ else
+ fprintf(stderr, "Compression with %s: OK "
+ "(compression %zu usec, decompression %zu usec, bytes raw %zu, compressed %zu, savings ratio %0.2f%%)\n",
+ name, compression_ut, decompression_ut,
+ bytes_uncompressed, bytes_compressed,
+ 100.0 - (double)bytes_compressed * 100.0 / (double)bytes_uncompressed);
+
+ return errors;
+}
+
+int unittest_rrdpush_compression(compression_algorithm_t algorithm, const char *name) {
+ fprintf(stderr, "\nTesting streaming compression with %s\n", name);
+
+ struct compressor_state cctx = {
+ .initialized = false,
+ .algorithm = algorithm,
+ };
+ struct decompressor_state dctx = {
+ .initialized = false,
+ .algorithm = algorithm,
+ };
+
+ char txt[COMPRESSION_MAX_MSG_SIZE];
+
+ rrdpush_compressor_init(&cctx);
+ rrdpush_decompressor_init(&dctx);
+
+ int errors = 0;
+
+ memset(txt, '=', COMPRESSION_MAX_MSG_SIZE);
+
+ for(int i = 0; i < COMPRESSION_MAX_MSG_SIZE ;i++) {
+ txt[i] = 'A' + (i % 26);
+ size_t txt_len = i + 1;
+
+ const char *out;
+ size_t size = rrdpush_compress(&cctx, txt, txt_len, &out);
+
+ if(size == 0) {
+ fprintf(stderr, "iteration %d: compressed size %zu is zero\n",
+ i, size);
+ errors++;
+ goto cleanup;
+ }
+ else if(size >= COMPRESSION_MAX_CHUNK) {
+ fprintf(stderr, "iteration %d: compressed size %zu exceeds max allowed size\n",
+ i, size);
+ errors++;
+ goto cleanup;
+ }
+ else {
+ size_t dtxt_len = rrdpush_decompress(&dctx, out, size);
+ char *dtxt = (char *) &dctx.output.data[dctx.output.read_pos];
+
+ if(rrdpush_decompressed_bytes_in_buffer(&dctx) != dtxt_len) {
+ fprintf(stderr, "iteration %d: decompressed size %zu does not rrdpush_decompressed_bytes_in_buffer() %zu\n",
+ i, dtxt_len, rrdpush_decompressed_bytes_in_buffer(&dctx)
+ );
+ errors++;
+ goto cleanup;
+ }
+
+ if(!dtxt_len) {
+ fprintf(stderr, "iteration %d: decompressed size is zero\n", i);
+ errors++;
+ goto cleanup;
+ }
+ else if(dtxt_len != txt_len) {
+ fprintf(stderr, "iteration %d: decompressed size %zu does not match original size %zu\n",
+ i, dtxt_len, txt_len
+ );
+ errors++;
+ goto cleanup;
+ }
+ else {
+ if(memcmp(txt, dtxt, txt_len) != 0) {
+ txt[txt_len] = '\0';
+ dtxt[txt_len + 5] = '\0';
+
+ fprintf(stderr, "iteration %d: decompressed data '%s' do not match original data '%s' of length %zu\n",
+ i, dtxt, txt, txt_len);
+ errors++;
+ goto cleanup;
+ }
+ }
+ }
+
+ // fill the compressed buffer with garbage
+ memset((void *)out, 'x', size);
+
+ // here we are supposed to copy the data and advance the position
+ dctx.output.read_pos += rrdpush_decompressed_bytes_in_buffer(&dctx);
+ }
+
+cleanup:
+ rrdpush_compressor_destroy(&cctx);
+ rrdpush_decompressor_destroy(&dctx);
+
+ if(errors)
+ fprintf(stderr, "Compression with %s: FAILED (%d errors)\n", name, errors);
+ else
+ fprintf(stderr, "Compression with %s: OK\n", name);
+
+ return errors;
+}
+
+int unittest_rrdpush_compressions(void) {
+ int ret = 0;
+
+ ret += unittest_rrdpush_compression(COMPRESSION_ALGORITHM_ZSTD, "ZSTD");
+ ret += unittest_rrdpush_compression(COMPRESSION_ALGORITHM_LZ4, "LZ4");
+ ret += unittest_rrdpush_compression(COMPRESSION_ALGORITHM_BROTLI, "BROTLI");
+ ret += unittest_rrdpush_compression(COMPRESSION_ALGORITHM_GZIP, "GZIP");
+
+ ret += unittest_rrdpush_compression_speed(COMPRESSION_ALGORITHM_ZSTD, "ZSTD");
+ ret += unittest_rrdpush_compression_speed(COMPRESSION_ALGORITHM_LZ4, "LZ4");
+ ret += unittest_rrdpush_compression_speed(COMPRESSION_ALGORITHM_BROTLI, "BROTLI");
+ ret += unittest_rrdpush_compression_speed(COMPRESSION_ALGORITHM_GZIP, "GZIP");
+
+ return ret;
+}
diff --git a/src/streaming/stream-compression/compression.h b/src/streaming/stream-compression/compression.h
new file mode 100644
index 000000000..37f589b85
--- /dev/null
+++ b/src/streaming/stream-compression/compression.h
@@ -0,0 +1,183 @@
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#ifndef NETDATA_RRDPUSH_COMPRESSION_H
+#define NETDATA_RRDPUSH_COMPRESSION_H 1
+
+#include "libnetdata/libnetdata.h"
+
+// signature MUST end with a newline
+
+#if COMPRESSION_MAX_MSG_SIZE >= (COMPRESSION_MAX_CHUNK - COMPRESSION_MAX_OVERHEAD)
+#error "COMPRESSION_MAX_MSG_SIZE >= (COMPRESSION_MAX_CHUNK - COMPRESSION_MAX_OVERHEAD)"
+#endif
+
+typedef uint32_t rrdpush_signature_t;
+#define RRDPUSH_COMPRESSION_SIGNATURE ((rrdpush_signature_t)('z' | 0x80) | (0x80 << 8) | (0x80 << 16) | ('\n' << 24))
+#define RRDPUSH_COMPRESSION_SIGNATURE_MASK ((rrdpush_signature_t) 0xffU | (0x80U << 8) | (0x80U << 16) | (0xffU << 24))
+#define RRDPUSH_COMPRESSION_SIGNATURE_SIZE sizeof(rrdpush_signature_t)
+
+static inline rrdpush_signature_t rrdpush_compress_encode_signature(size_t compressed_data_size) {
+ rrdpush_signature_t len = ((compressed_data_size & 0x7f) | 0x80 | (((compressed_data_size & (0x7f << 7)) << 1) | 0x8000)) << 8;
+ return len | RRDPUSH_COMPRESSION_SIGNATURE;
+}
+
+typedef enum {
+ COMPRESSION_ALGORITHM_NONE = 0,
+ COMPRESSION_ALGORITHM_ZSTD,
+ COMPRESSION_ALGORITHM_LZ4,
+ COMPRESSION_ALGORITHM_GZIP,
+ COMPRESSION_ALGORITHM_BROTLI,
+
+ // terminator
+ COMPRESSION_ALGORITHM_MAX,
+} compression_algorithm_t;
+
+extern int rrdpush_compression_levels[COMPRESSION_ALGORITHM_MAX];
+
+// this defines the order the algorithms will be selected by the receiver (parent)
+#define RRDPUSH_COMPRESSION_ALGORITHMS_ORDER "zstd lz4 brotli gzip"
+
+// ----------------------------------------------------------------------------
+
+typedef struct simple_ring_buffer {
+ const char *data;
+ size_t size;
+ size_t read_pos;
+ size_t write_pos;
+} SIMPLE_RING_BUFFER;
+
+static inline void simple_ring_buffer_reset(SIMPLE_RING_BUFFER *b) {
+ b->read_pos = b->write_pos = 0;
+}
+
+static inline void simple_ring_buffer_make_room(SIMPLE_RING_BUFFER *b, size_t size) {
+ if(b->write_pos + size > b->size) {
+ if(!b->size)
+ b->size = COMPRESSION_MAX_CHUNK;
+ else
+ b->size *= 2;
+
+ if(b->write_pos + size > b->size)
+ b->size += size;
+
+ b->data = (const char *)reallocz((void *)b->data, b->size);
+ }
+}
+
+static inline void simple_ring_buffer_append_data(SIMPLE_RING_BUFFER *b, const void *data, size_t size) {
+ simple_ring_buffer_make_room(b, size);
+ memcpy((void *)(b->data + b->write_pos), data, size);
+ b->write_pos += size;
+}
+
+static inline void simple_ring_buffer_destroy(SIMPLE_RING_BUFFER *b) {
+ freez((void *)b->data);
+ b->data = NULL;
+ b->read_pos = b->write_pos = b->size = 0;
+}
+
+// ----------------------------------------------------------------------------
+
+struct compressor_state {
+ bool initialized;
+ compression_algorithm_t algorithm;
+
+ SIMPLE_RING_BUFFER input;
+ SIMPLE_RING_BUFFER output;
+
+ int level;
+ void *stream;
+
+ struct {
+ size_t total_compressed;
+ size_t total_uncompressed;
+ size_t total_compressions;
+ } sender_locked;
+};
+
+void rrdpush_compressor_init(struct compressor_state *state);
+void rrdpush_compressor_destroy(struct compressor_state *state);
+size_t rrdpush_compress(struct compressor_state *state, const char *data, size_t size, const char **out);
+
+// ----------------------------------------------------------------------------
+
+struct decompressor_state {
+ bool initialized;
+ compression_algorithm_t algorithm;
+ size_t signature_size;
+
+ size_t total_compressed;
+ size_t total_uncompressed;
+ size_t total_compressions;
+
+ SIMPLE_RING_BUFFER output;
+
+ void *stream;
+};
+
+void rrdpush_decompressor_destroy(struct decompressor_state *state);
+void rrdpush_decompressor_init(struct decompressor_state *state);
+size_t rrdpush_decompress(struct decompressor_state *state, const char *compressed_data, size_t compressed_size);
+
+static inline size_t rrdpush_decompress_decode_signature(const char *data, size_t data_size) {
+ if (unlikely(!data || !data_size))
+ return 0;
+
+ if (unlikely(data_size != RRDPUSH_COMPRESSION_SIGNATURE_SIZE))
+ return 0;
+
+ rrdpush_signature_t sign = *(rrdpush_signature_t *)data;
+ if (unlikely((sign & RRDPUSH_COMPRESSION_SIGNATURE_MASK) != RRDPUSH_COMPRESSION_SIGNATURE))
+ return 0;
+
+ size_t length = ((sign >> 8) & 0x7f) | ((sign >> 9) & (0x7f << 7));
+ return length;
+}
+
+static inline size_t rrdpush_decompressor_start(struct decompressor_state *state, const char *header, size_t header_size) {
+ if(unlikely(state->output.read_pos != state->output.write_pos))
+ fatal("RRDPUSH DECOMPRESS: asked to decompress new data, while there are unread data in the decompression buffer!");
+
+ return rrdpush_decompress_decode_signature(header, header_size);
+}
+
+static inline size_t rrdpush_decompressed_bytes_in_buffer(struct decompressor_state *state) {
+ if(unlikely(state->output.read_pos > state->output.write_pos))
+ fatal("RRDPUSH DECOMPRESS: invalid read/write stream positions");
+
+ return state->output.write_pos - state->output.read_pos;
+}
+
+static inline size_t rrdpush_decompressor_get(struct decompressor_state *state, char *dst, size_t size) {
+ if (unlikely(!state || !size || !dst))
+ return 0;
+
+ size_t remaining = rrdpush_decompressed_bytes_in_buffer(state);
+
+ if(unlikely(!remaining))
+ return 0;
+
+ size_t bytes_to_return = size;
+ if(bytes_to_return > remaining)
+ bytes_to_return = remaining;
+
+ memcpy(dst, state->output.data + state->output.read_pos, bytes_to_return);
+ state->output.read_pos += bytes_to_return;
+
+ if(unlikely(state->output.read_pos > state->output.write_pos))
+ fatal("RRDPUSH DECOMPRESS: invalid read/write stream positions");
+
+ return bytes_to_return;
+}
+
+// ----------------------------------------------------------------------------
+
+#include "../rrdpush.h"
+
+bool rrdpush_compression_initialize(struct sender_state *s);
+bool rrdpush_decompression_initialize(struct receiver_state *rpt);
+void rrdpush_parse_compression_order(struct receiver_state *rpt, const char *order);
+void rrdpush_select_receiver_compression_algorithm(struct receiver_state *rpt);
+void rrdpush_compression_deactivate(struct sender_state *s);
+
+#endif // NETDATA_RRDPUSH_COMPRESSION_H 1
diff --git a/src/streaming/stream-compression/gzip.c b/src/streaming/stream-compression/gzip.c
new file mode 100644
index 000000000..d63e9afbe
--- /dev/null
+++ b/src/streaming/stream-compression/gzip.c
@@ -0,0 +1,164 @@
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#include "gzip.h"
+#include <zlib.h>
+
+void rrdpush_compressor_init_gzip(struct compressor_state *state) {
+ if (!state->initialized) {
+ state->initialized = true;
+
+ // Initialize deflate stream
+ z_stream *strm = state->stream = (z_stream *) mallocz(sizeof(z_stream));
+ strm->zalloc = Z_NULL;
+ strm->zfree = Z_NULL;
+ strm->opaque = Z_NULL;
+
+ if(state->level < Z_BEST_SPEED)
+ state->level = Z_BEST_SPEED;
+
+ if(state->level > Z_BEST_COMPRESSION)
+ state->level = Z_BEST_COMPRESSION;
+
+ // int r = deflateInit2(strm, Z_BEST_COMPRESSION, Z_DEFLATED, 15 + 16, 8, Z_DEFAULT_STRATEGY);
+ int r = deflateInit2(strm, state->level, Z_DEFLATED, 15 + 16, 8, Z_DEFAULT_STRATEGY);
+ if (r != Z_OK) {
+ netdata_log_error("Failed to initialize deflate with error: %d", r);
+ freez(state->stream);
+ state->initialized = false;
+ return;
+ }
+
+ }
+}
+
+void rrdpush_compressor_destroy_gzip(struct compressor_state *state) {
+ if (state->stream) {
+ deflateEnd(state->stream);
+ freez(state->stream);
+ state->stream = NULL;
+ }
+}
+
+size_t rrdpush_compress_gzip(struct compressor_state *state, const char *data, size_t size, const char **out) {
+ if (unlikely(!state || !size || !out))
+ return 0;
+
+ simple_ring_buffer_make_room(&state->output, deflateBound(state->stream, size));
+
+ z_stream *strm = state->stream;
+ strm->avail_in = (uInt)size;
+ strm->next_in = (Bytef *)data;
+ strm->avail_out = (uInt)state->output.size;
+ strm->next_out = (Bytef *)state->output.data;
+
+ int ret = deflate(strm, Z_SYNC_FLUSH);
+ if (ret != Z_OK && ret != Z_STREAM_END) {
+ netdata_log_error("STREAM: deflate() failed with error %d", ret);
+ return 0;
+ }
+
+ if(strm->avail_in != 0) {
+ netdata_log_error("STREAM: deflate() did not use all the input buffer, %u bytes out of %zu remain",
+ strm->avail_in, size);
+ return 0;
+ }
+
+ if(strm->avail_out == 0) {
+ netdata_log_error("STREAM: deflate() needs a bigger output buffer than the one we provided "
+ "(output buffer %zu bytes, compressed payload %zu bytes)",
+ state->output.size, size);
+ return 0;
+ }
+
+ size_t compressed_data_size = state->output.size - strm->avail_out;
+
+ if(compressed_data_size == 0) {
+ netdata_log_error("STREAM: deflate() did not produce any output "
+ "(output buffer %zu bytes, compressed payload %zu bytes)",
+ state->output.size, size);
+ return 0;
+ }
+
+ state->sender_locked.total_compressions++;
+ state->sender_locked.total_uncompressed += size;
+ state->sender_locked.total_compressed += compressed_data_size;
+
+ *out = state->output.data;
+ return compressed_data_size;
+}
+
+void rrdpush_decompressor_init_gzip(struct decompressor_state *state) {
+ if (!state->initialized) {
+ state->initialized = true;
+
+ // Initialize inflate stream
+ z_stream *strm = state->stream = (z_stream *)mallocz(sizeof(z_stream));
+ strm->zalloc = Z_NULL;
+ strm->zfree = Z_NULL;
+ strm->opaque = Z_NULL;
+
+ int r = inflateInit2(strm, 15 + 16);
+ if (r != Z_OK) {
+ netdata_log_error("Failed to initialize inflateInit2() with error: %d", r);
+ freez(state->stream);
+ state->initialized = false;
+ return;
+ }
+
+ simple_ring_buffer_make_room(&state->output, COMPRESSION_MAX_CHUNK);
+ }
+}
+
+void rrdpush_decompressor_destroy_gzip(struct decompressor_state *state) {
+ if (state->stream) {
+ inflateEnd(state->stream);
+ freez(state->stream);
+ state->stream = NULL;
+ }
+}
+
+size_t rrdpush_decompress_gzip(struct decompressor_state *state, const char *compressed_data, size_t compressed_size) {
+ if (unlikely(!state || !compressed_data || !compressed_size))
+ return 0;
+
+ // The state.output ring buffer is always EMPTY at this point,
+ // meaning that (state->output.read_pos == state->output.write_pos)
+ // However, THEY ARE NOT ZERO.
+
+ z_stream *strm = state->stream;
+ strm->avail_in = (uInt)compressed_size;
+ strm->next_in = (Bytef *)compressed_data;
+ strm->avail_out = (uInt)state->output.size;
+ strm->next_out = (Bytef *)state->output.data;
+
+ int ret = inflate(strm, Z_SYNC_FLUSH);
+ if (ret != Z_STREAM_END && ret != Z_OK) {
+ netdata_log_error("RRDPUSH DECOMPRESS: inflate() failed with error %d", ret);
+ return 0;
+ }
+
+ if(strm->avail_in != 0) {
+ netdata_log_error("RRDPUSH DECOMPRESS: inflate() did not use all compressed data we provided "
+ "(compressed payload %zu bytes, remaining to be uncompressed %u)"
+ , compressed_size, strm->avail_in);
+ return 0;
+ }
+
+ if(strm->avail_out == 0) {
+ netdata_log_error("RRDPUSH DECOMPRESS: inflate() needs a bigger output buffer than the one we provided "
+ "(compressed payload %zu bytes, output buffer size %zu bytes)"
+ , compressed_size, state->output.size);
+ return 0;
+ }
+
+ size_t decompressed_size = state->output.size - strm->avail_out;
+
+ state->output.read_pos = 0;
+ state->output.write_pos = decompressed_size;
+
+ state->total_compressed += compressed_size;
+ state->total_uncompressed += decompressed_size;
+ state->total_compressions++;
+
+ return decompressed_size;
+}
diff --git a/src/streaming/stream-compression/gzip.h b/src/streaming/stream-compression/gzip.h
new file mode 100644
index 000000000..85f34bc6d
--- /dev/null
+++ b/src/streaming/stream-compression/gzip.h
@@ -0,0 +1,15 @@
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#include "compression.h"
+
+#ifndef NETDATA_STREAMING_COMPRESSION_GZIP_H
+#define NETDATA_STREAMING_COMPRESSION_GZIP_H
+
+void rrdpush_compressor_init_gzip(struct compressor_state *state);
+void rrdpush_compressor_destroy_gzip(struct compressor_state *state);
+size_t rrdpush_compress_gzip(struct compressor_state *state, const char *data, size_t size, const char **out);
+size_t rrdpush_decompress_gzip(struct decompressor_state *state, const char *compressed_data, size_t compressed_size);
+void rrdpush_decompressor_init_gzip(struct decompressor_state *state);
+void rrdpush_decompressor_destroy_gzip(struct decompressor_state *state);
+
+#endif //NETDATA_STREAMING_COMPRESSION_GZIP_H
diff --git a/src/streaming/stream-compression/lz4.c b/src/streaming/stream-compression/lz4.c
new file mode 100644
index 000000000..284192153
--- /dev/null
+++ b/src/streaming/stream-compression/lz4.c
@@ -0,0 +1,143 @@
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#include "lz4.h"
+
+#ifdef ENABLE_LZ4
+#include "lz4.h"
+
+// ----------------------------------------------------------------------------
+// compress
+
+void rrdpush_compressor_init_lz4(struct compressor_state *state) {
+ if(!state->initialized) {
+ state->initialized = true;
+ state->stream = LZ4_createStream();
+
+ // LZ4 needs access to the last 64KB of source data
+ // so, we keep twice the size of each message
+ simple_ring_buffer_make_room(&state->input, 65536 + COMPRESSION_MAX_CHUNK * 2);
+ }
+}
+
+void rrdpush_compressor_destroy_lz4(struct compressor_state *state) {
+ if (state->stream) {
+ LZ4_freeStream(state->stream);
+ state->stream = NULL;
+ }
+}
+
+/*
+ * Compress the given block of data
+ * Compressed data will remain in the internal buffer until the next invocation
+ * Return the size of compressed data block as result and the pointer to internal buffer using the last argument
+ * or 0 in case of error
+ */
+size_t rrdpush_compress_lz4(struct compressor_state *state, const char *data, size_t size, const char **out) {
+ if(unlikely(!state || !size || !out))
+ return 0;
+
+ // we need to keep the last 64K of our previous source data
+ // as they were in the ring buffer
+
+ simple_ring_buffer_make_room(&state->output, LZ4_COMPRESSBOUND(size));
+
+ if(state->input.write_pos + size > state->input.size)
+ // the input buffer cannot fit out data, restart from zero
+ simple_ring_buffer_reset(&state->input);
+
+ simple_ring_buffer_append_data(&state->input, data, size);
+
+ long int compressed_data_size = LZ4_compress_fast_continue(
+ state->stream,
+ state->input.data + state->input.read_pos,
+ (char *)state->output.data,
+ (int)(state->input.write_pos - state->input.read_pos),
+ (int)state->output.size,
+ state->level);
+
+ if (compressed_data_size <= 0) {
+ netdata_log_error("STREAM: LZ4_compress_fast_continue() returned %ld "
+ "(source is %zu bytes, output buffer can fit %zu bytes)",
+ compressed_data_size, size, state->output.size);
+ return 0;
+ }
+
+ state->input.read_pos = state->input.write_pos;
+
+ state->sender_locked.total_compressions++;
+ state->sender_locked.total_uncompressed += size;
+ state->sender_locked.total_compressed += compressed_data_size;
+
+ *out = state->output.data;
+ return compressed_data_size;
+}
+
+// ----------------------------------------------------------------------------
+// decompress
+
+void rrdpush_decompressor_init_lz4(struct decompressor_state *state) {
+ if(!state->initialized) {
+ state->initialized = true;
+ state->stream = LZ4_createStreamDecode();
+ simple_ring_buffer_make_room(&state->output, 65536 + COMPRESSION_MAX_CHUNK * 2);
+ }
+}
+
+void rrdpush_decompressor_destroy_lz4(struct decompressor_state *state) {
+ if (state->stream) {
+ LZ4_freeStreamDecode(state->stream);
+ state->stream = NULL;
+ }
+}
+
+/*
+ * Decompress the compressed data in the internal buffer
+ * Return the size of uncompressed data or 0 for error
+ */
+size_t rrdpush_decompress_lz4(struct decompressor_state *state, const char *compressed_data, size_t compressed_size) {
+ if (unlikely(!state || !compressed_data || !compressed_size))
+ return 0;
+
+ // The state.output ring buffer is always EMPTY at this point,
+ // meaning that (state->output.read_pos == state->output.write_pos)
+ // However, THEY ARE NOT ZERO.
+
+ if (unlikely(state->output.write_pos + COMPRESSION_MAX_CHUNK > state->output.size))
+ // the input buffer cannot fit out data, restart from zero
+ simple_ring_buffer_reset(&state->output);
+
+ long int decompressed_size = LZ4_decompress_safe_continue(
+ state->stream
+ , compressed_data
+ , (char *)(state->output.data + state->output.write_pos)
+ , (int)compressed_size
+ , (int)(state->output.size - state->output.write_pos)
+ );
+
+ if (unlikely(decompressed_size < 0)) {
+ netdata_log_error("RRDPUSH DECOMPRESS: LZ4_decompress_safe_continue() returned negative value: %ld "
+ "(compressed chunk is %zu bytes)"
+ , decompressed_size, compressed_size);
+ return 0;
+ }
+
+ if(unlikely(decompressed_size + state->output.write_pos > state->output.size))
+ fatal("RRDPUSH DECOMPRESS: LZ4_decompress_safe_continue() overflown the stream_buffer "
+ "(size: %zu, pos: %zu, added: %ld, exceeding the buffer by %zu)"
+ , state->output.size
+ , state->output.write_pos
+ , decompressed_size
+ , (size_t)(state->output.write_pos + decompressed_size - state->output.size)
+ );
+
+ state->output.write_pos += decompressed_size;
+
+ // statistics
+ state->total_compressed += compressed_size;
+ state->total_uncompressed += decompressed_size;
+ state->total_compressions++;
+
+ return decompressed_size;
+}
+
+#endif // ENABLE_LZ4
diff --git a/src/streaming/stream-compression/lz4.h b/src/streaming/stream-compression/lz4.h
new file mode 100644
index 000000000..69f0fadcc
--- /dev/null
+++ b/src/streaming/stream-compression/lz4.h
@@ -0,0 +1,19 @@
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#include "compression.h"
+
+#ifndef NETDATA_STREAMING_COMPRESSION_LZ4_H
+#define NETDATA_STREAMING_COMPRESSION_LZ4_H
+
+#ifdef ENABLE_LZ4
+
+void rrdpush_compressor_init_lz4(struct compressor_state *state);
+void rrdpush_compressor_destroy_lz4(struct compressor_state *state);
+size_t rrdpush_compress_lz4(struct compressor_state *state, const char *data, size_t size, const char **out);
+size_t rrdpush_decompress_lz4(struct decompressor_state *state, const char *compressed_data, size_t compressed_size);
+void rrdpush_decompressor_init_lz4(struct decompressor_state *state);
+void rrdpush_decompressor_destroy_lz4(struct decompressor_state *state);
+
+#endif // ENABLE_LZ4
+
+#endif //NETDATA_STREAMING_COMPRESSION_LZ4_H
diff --git a/src/streaming/stream-compression/zstd.c b/src/streaming/stream-compression/zstd.c
new file mode 100644
index 000000000..0ce27c0d3
--- /dev/null
+++ b/src/streaming/stream-compression/zstd.c
@@ -0,0 +1,163 @@
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#include "zstd.h"
+
+#ifdef ENABLE_ZSTD
+#include <zstd.h>
+
+void rrdpush_compressor_init_zstd(struct compressor_state *state) {
+ if(!state->initialized) {
+ state->initialized = true;
+ state->stream = ZSTD_createCStream();
+
+ if(state->level < 1)
+ state->level = 1;
+
+ if(state->level > ZSTD_maxCLevel())
+ state->level = ZSTD_maxCLevel();
+
+ size_t ret = ZSTD_initCStream(state->stream, state->level);
+ if(ZSTD_isError(ret))
+ netdata_log_error("STREAM: ZSTD_initCStream() returned error: %s", ZSTD_getErrorName(ret));
+
+ // ZSTD_CCtx_setParameter(state->stream, ZSTD_c_compressionLevel, 1);
+ // ZSTD_CCtx_setParameter(state->stream, ZSTD_c_strategy, ZSTD_fast);
+ }
+}
+
+void rrdpush_compressor_destroy_zstd(struct compressor_state *state) {
+ if(state->stream) {
+ ZSTD_freeCStream(state->stream);
+ state->stream = NULL;
+ }
+}
+
+size_t rrdpush_compress_zstd(struct compressor_state *state, const char *data, size_t size, const char **out) {
+ if(unlikely(!state || !size || !out))
+ return 0;
+
+ ZSTD_inBuffer inBuffer = {
+ .pos = 0,
+ .size = size,
+ .src = data,
+ };
+
+ size_t wanted_size = MAX(ZSTD_compressBound(inBuffer.size - inBuffer.pos), ZSTD_CStreamOutSize());
+ simple_ring_buffer_make_room(&state->output, wanted_size);
+
+ ZSTD_outBuffer outBuffer = {
+ .pos = 0,
+ .size = state->output.size,
+ .dst = (void *)state->output.data,
+ };
+
+ // compress
+ size_t ret = ZSTD_compressStream(state->stream, &outBuffer, &inBuffer);
+
+ // error handling
+ if(ZSTD_isError(ret)) {
+ netdata_log_error("STREAM: ZSTD_compressStream() return error: %s", ZSTD_getErrorName(ret));
+ return 0;
+ }
+
+ if(inBuffer.pos < inBuffer.size) {
+ netdata_log_error("STREAM: ZSTD_compressStream() left unprocessed input (source payload %zu bytes, consumed %zu bytes)",
+ inBuffer.size, inBuffer.pos);
+ return 0;
+ }
+
+ if(outBuffer.pos == 0) {
+ // ZSTD needs more input to flush the output, so let's flush it manually
+ ret = ZSTD_flushStream(state->stream, &outBuffer);
+
+ if(ZSTD_isError(ret)) {
+ netdata_log_error("STREAM: ZSTD_flushStream() return error: %s", ZSTD_getErrorName(ret));
+ return 0;
+ }
+
+ if(outBuffer.pos == 0) {
+ netdata_log_error("STREAM: ZSTD_compressStream() returned zero compressed bytes "
+ "(source is %zu bytes, output buffer can fit %zu bytes) "
+ , size, outBuffer.size);
+ return 0;
+ }
+ }
+
+ state->sender_locked.total_compressions++;
+ state->sender_locked.total_uncompressed += size;
+ state->sender_locked.total_compressed += outBuffer.pos;
+
+ // return values
+ *out = state->output.data;
+ return outBuffer.pos;
+}
+
+void rrdpush_decompressor_init_zstd(struct decompressor_state *state) {
+ if(!state->initialized) {
+ state->initialized = true;
+ state->stream = ZSTD_createDStream();
+
+ size_t ret = ZSTD_initDStream(state->stream);
+ if(ZSTD_isError(ret))
+ netdata_log_error("STREAM: ZSTD_initDStream() returned error: %s", ZSTD_getErrorName(ret));
+
+ simple_ring_buffer_make_room(&state->output, MAX(COMPRESSION_MAX_CHUNK, ZSTD_DStreamOutSize()));
+ }
+}
+
+void rrdpush_decompressor_destroy_zstd(struct decompressor_state *state) {
+ if (state->stream) {
+ ZSTD_freeDStream(state->stream);
+ state->stream = NULL;
+ }
+}
+
+size_t rrdpush_decompress_zstd(struct decompressor_state *state, const char *compressed_data, size_t compressed_size) {
+ if (unlikely(!state || !compressed_data || !compressed_size))
+ return 0;
+
+ // The state.output ring buffer is always EMPTY at this point,
+ // meaning that (state->output.read_pos == state->output.write_pos)
+ // However, THEY ARE NOT ZERO.
+
+ ZSTD_inBuffer inBuffer = {
+ .pos = 0,
+ .size = compressed_size,
+ .src = compressed_data,
+ };
+
+ ZSTD_outBuffer outBuffer = {
+ .pos = 0,
+ .dst = (char *)state->output.data,
+ .size = state->output.size,
+ };
+
+ size_t ret = ZSTD_decompressStream(
+ state->stream
+ , &outBuffer
+ , &inBuffer);
+
+ if(ZSTD_isError(ret)) {
+ netdata_log_error("STREAM: ZSTD_decompressStream() return error: %s", ZSTD_getErrorName(ret));
+ return 0;
+ }
+
+ if(inBuffer.pos < inBuffer.size)
+ fatal("RRDPUSH DECOMPRESS: ZSTD ZSTD_decompressStream() decompressed %zu bytes, "
+ "but %zu bytes of compressed data remain",
+ inBuffer.pos, inBuffer.size);
+
+ size_t decompressed_size = outBuffer.pos;
+
+ state->output.read_pos = 0;
+ state->output.write_pos = outBuffer.pos;
+
+ // statistics
+ state->total_compressed += compressed_size;
+ state->total_uncompressed += decompressed_size;
+ state->total_compressions++;
+
+ return decompressed_size;
+}
+
+#endif // ENABLE_ZSTD
diff --git a/src/streaming/stream-compression/zstd.h b/src/streaming/stream-compression/zstd.h
new file mode 100644
index 000000000..bfabbf89d
--- /dev/null
+++ b/src/streaming/stream-compression/zstd.h
@@ -0,0 +1,19 @@
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#include "compression.h"
+
+#ifndef NETDATA_STREAMING_COMPRESSION_ZSTD_H
+#define NETDATA_STREAMING_COMPRESSION_ZSTD_H
+
+#ifdef ENABLE_ZSTD
+
+void rrdpush_compressor_init_zstd(struct compressor_state *state);
+void rrdpush_compressor_destroy_zstd(struct compressor_state *state);
+size_t rrdpush_compress_zstd(struct compressor_state *state, const char *data, size_t size, const char **out);
+size_t rrdpush_decompress_zstd(struct decompressor_state *state, const char *compressed_data, size_t compressed_size);
+void rrdpush_decompressor_init_zstd(struct decompressor_state *state);
+void rrdpush_decompressor_destroy_zstd(struct decompressor_state *state);
+
+#endif // ENABLE_ZSTD
+
+#endif //NETDATA_STREAMING_COMPRESSION_ZSTD_H
generated by cgit v1.2.3 (git 2.39.1) at 2025-01-13 14:14:05 +0000