diff --git a/Makefile.am b/Makefile.am
index 3996c91097..23be6207c3 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -667,6 +667,8 @@ STREAMING_PLUGIN_FILES = \
streaming/rrdpush.c \
streaming/compression.c \
streaming/compression.h \
+ streaming/compression_brotli.c \
+ streaming/compression_brotli.h \
streaming/compression_gzip.c \
streaming/compression_gzip.h \
streaming/compression_lz4.c \
@@ -1151,6 +1153,8 @@ NETDATA_COMMON_LIBS = \
$(OPTIONAL_UV_LIBS) \
$(OPTIONAL_LZ4_LIBS) \
$(OPTIONAL_ZSTD_LIBS) \
+ $(OPTIONAL_BROTLIENC_LIBS) \
+ $(OPTIONAL_BROTLIDEC_LIBS) \
$(OPTIONAL_DATACHANNEL_LIBS) \
libjudy.a \
$(OPTIONAL_SSL_LIBS) \
diff --git a/configure.ac b/configure.ac
index 8a78fbc5a4..a63b8523ad 100644
--- a/configure.ac
+++ b/configure.ac
@@ -577,6 +577,27 @@ if test "x$LIBZSTD_FOUND" = "xyes"; then
OPTIONAL_ZSTD_LIBS="-lzstd"
fi
+# -----------------------------------------------------------------------------
+# brotli
+
+AC_CHECK_LIB([brotlienc], [BrotliEncoderCreateInstance, BrotliEncoderCompressStream],
+ [LIBBROTLIENC_FOUND=yes],
+ [LIBBROTLIENC_FOUND=no])
+
+if test "x$LIBBROTLIENC_FOUND" = "xyes"; then
+ AC_DEFINE([ENABLE_BROTLIENC], [1], [libbrotlienc usability])
+ OPTIONAL_BROTLIENC_LIBS="-lbrotlienc"
+fi
+
+AC_CHECK_LIB([brotlidec], [BrotliDecoderCreateInstance, BrotliDecoderDecompressStream],
+ [LIBBROTLIDEC_FOUND=yes],
+ [LIBBROTLIDEC_FOUND=no])
+
+if test "x$LIBBROTLIDEC_FOUND" = "xyes"; then
+ AC_DEFINE([ENABLE_BROTLIDEC], [1], [libbrotlidec usability])
+ OPTIONAL_BROTLIDEC_LIBS="-lbrotlidec"
+fi
+
# -----------------------------------------------------------------------------
# zlib
@@ -1912,6 +1933,8 @@ AC_SUBST([OPTIONAL_MATH_LIBS])
AC_SUBST([OPTIONAL_DATACHANNEL_LIBS])
AC_SUBST([OPTIONAL_UV_LIBS])
AC_SUBST([OPTIONAL_LZ4_LIBS])
+AC_SUBST([OPTIONAL_BROTLIENC_LIBS])
+AC_SUBST([OPTIONAL_BROTLIDEC_LIBS])
AC_SUBST([OPTIONAL_ZSTD_LIBS])
AC_SUBST([OPTIONAL_SSL_LIBS])
AC_SUBST([OPTIONAL_JSONC_LIBS])
diff --git a/daemon/buildinfo.c b/daemon/buildinfo.c
index bbe646993e..aa043a0fd0 100644
--- a/daemon/buildinfo.c
+++ b/daemon/buildinfo.c
@@ -1115,6 +1115,9 @@ __attribute__((constructor)) void initialize_build_info(void) {
build_info_set_status(BIB_FEATURE_STREAMING_COMPRESSION, true);
+#ifdef ENABLE_BROTLI
+ build_info_append_value(BIB_FEATURE_STREAMING_COMPRESSION, "brotli");
+#endif
#ifdef ENABLE_ZSTD
build_info_append_value(BIB_FEATURE_STREAMING_COMPRESSION, "zstd");
#endif
diff --git a/libnetdata/libnetdata.h b/libnetdata/libnetdata.h
index 7459cd0ed7..daa247f1fb 100644
--- a/libnetdata/libnetdata.h
+++ b/libnetdata/libnetdata.h
@@ -11,6 +11,10 @@ extern "C" {
#include <config.h>
#endif
+#if defined(ENABLE_BROTLIENC) && defined(ENABLE_BROTLIDEC)
+#define ENABLE_BROTLI 1
+#endif
+
#ifdef ENABLE_OPENSSL
#define ENABLE_HTTPS 1
#endif
diff --git a/streaming/compression.c b/streaming/compression.c
index 22f0877fd0..6fdce54735 100644
--- a/streaming/compression.c
+++ b/streaming/compression.c
@@ -12,6 +12,175 @@
#include "compression_zstd.h"
#endif
+#ifdef ENABLE_BROTLI
+#include "compression_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_pluginsd(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
@@ -29,6 +198,12 @@ void rrdpush_compressor_init(struct compressor_state *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);
@@ -53,6 +228,12 @@ void rrdpush_compressor_destroy(struct compressor_state *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);
@@ -81,6 +262,12 @@ size_t rrdpush_compress(struct compressor_state *state, const char *data, size_t
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);
@@ -116,6 +303,12 @@ void rrdpush_decompressor_destroy(struct decompressor_state *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);
@@ -141,6 +334,12 @@ void rrdpush_decompressor_init(struct decompressor_state *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);
@@ -170,6 +369,12 @@ size_t rrdpush_decompress(struct decompressor_state *state, const char *compress
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);
@@ -190,6 +395,201 @@ size_t rrdpush_decompress(struct decompressor_state *state, const char *compress
// ----------------------------------------------------------------------------
// unit test
+void unittest_generate_random_name(char *dst, size_t size) {
+ if(size < 7)
+ size = 7;
+
+ size_t len = 5 + random() % (size - 6);
+
+ for(size_t i = 0; i < len ; i++) {
+ if(random() % 2 == 0)
+ dst[i] = 'A' + random() % 26;
+ else
+ dst[i] = 'a' + random() % 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 + random() % 5;
+ char chart[RRD_ID_LENGTH_MAX + 1] = "name";
+ unittest_generate_random_name(chart, 5 + random() % 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 + random() % 20);
+ int64_t last_collected_value = (random() % 2 == 0) ? (int64_t)(counter + d) : (int64_t)random();
+ NETDATA_DOUBLE value = (random() % 2 == 0) ? (NETDATA_DOUBLE)random() / ((NETDATA_DOUBLE)random() + 1) : (NETDATA_DOUBLE)last_collected_value;
+ SN_FLAGS flags = (random() % 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);
@@ -218,7 +618,13 @@ int unittest_rrdpush_compression(compression_algorithm_t algorithm, const char *
const char *out;
size_t size = rrdpush_compress(&cctx, txt, txt_len, &out);
- if(size >= COMPRESSION_MAX_CHUNK) {
+ 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++;
@@ -236,7 +642,12 @@ int unittest_rrdpush_compression(compression_algorithm_t algorithm, const char *
goto cleanup;
}
- if(dtxt_len != txt_len) {
+ 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
);
@@ -278,9 +689,15 @@ cleanup:
int unittest_rrdpush_compressions(void) {
int ret = 0;
- ret += unittest_rrdpush_compression(COMPRESSION_ALGORITHM_GZIP, "GZIP");
- ret += unittest_rrdpush_compression(COMPRESSION_ALGORITHM_LZ4, "LZ4");
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/streaming/compression.h b/streaming/compression.h
index 011e14eadd..a67f65b83a 100644
--- a/streaming/compression.h
+++ b/streaming/compression.h
@@ -26,6 +26,7 @@ typedef enum {
COMPRESSION_ALGORITHM_ZSTD,
COMPRESSION_ALGORITHM_LZ4,
COMPRESSION_ALGORITHM_GZIP,
+ COMPRESSION_ALGORITHM_BROTLI,
// terminator
COMPRESSION_ALGORITHM_MAX,
@@ -33,6 +34,9 @@ typedef enum {
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 {
diff --git a/streaming/compression_brotli.c b/streaming/compression_brotli.c
new file mode 100644
index 0000000000..5f772d56e1
--- /dev/null
+++ b/streaming/compression_brotli.c
@@ -0,0 +1,142 @@
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+#include "compression_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, %u 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, %u 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/streaming/compression_brotli.h b/streaming/compression_brotli.h
new file mode 100644
index 0000000000..4955e5a821
--- /dev/null
+++ b/streaming/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/streaming/receiver.c b/streaming/receiver.c
index 2399449579..02b2e83a0c 100644
--- a/streaming/receiver.c
+++ b/streaming/receiver.c
@@ -562,29 +562,6 @@ void rrdpush_receive_log_status(struct receiver_state *rpt, const char *msg, con
}
-static void rrdpush_parse_compression_order(struct receiver_state *rpt, const char *order) {
- rpt->config.compression_priorities[0] = STREAM_CAP_ZSTD;
- rpt->config.compression_priorities[1] = STREAM_CAP_LZ4;
- rpt->config.compression_priorities[2] = STREAM_CAP_GZIP;
-
- char *s = strdupz(order);
-
- char *words[COMPRESSION_ALGORITHM_MAX] = { NULL };
- size_t num_words = quoted_strings_splitter_pluginsd(s, words, COMPRESSION_ALGORITHM_MAX);
- for(size_t i = 0; i < num_words ;i++) {
- if(strcasecmp(words[i], "zstd") == 0)
- rpt->config.compression_priorities[i] = STREAM_CAP_ZSTD;
- else if(strcasecmp(words[i], "lz4") == 0)
- rpt->config.compression_priorities[i] = STREAM_CAP_LZ4;
- else if(strcasecmp(words[i], "gzip") == 0)
- rpt->config.compression_priorities[i] = STREAM_CAP_GZIP;
- else
- rpt->config.compression_priorities[i] = 0;
- }
-
- freez(s);
-}
-
static void rrdpush_receive(struct receiver_state *rpt)
{
rpt->config.mode = default_rrd_memory_mode;
@@ -658,7 +635,7 @@ static void rrdpush_receive(struct receiver_state *rpt)
rpt->config.rrdpush_compression = appconfig_get_boolean(&stream_config, rpt->machine_guid, "enable compression", rpt->config.rrdpush_compression);
if(rpt->config.rrdpush_compression) {
- char *order = appconfig_get(&stream_config, rpt->key, "compression algorithms order", "zstd lz4 gzip");
+ char *order = appconfig_get(&stream_config, rpt->key, "compression algorithms order", RRDPUSH_COMPRESSION_ALGORITHMS_ORDER);
order = appconfig_get(&stream_config, rpt->machine_guid, "compression algorithms order", order);
rrdpush_parse_compression_order(rpt, order);
}
@@ -755,27 +732,7 @@ static void rrdpush_receive(struct receiver_state *rpt)
snprintfz(cd.fullfilename, FILENAME_MAX, "%s:%s", rpt->client_ip, rpt->client_port);
snprintfz(cd.cmd, PLUGINSD_CMD_MAX, "%s:%s", rpt->client_ip, rpt->client_port);
- 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;
- }
- }
- }
+ rrdpush_select_receiver_compression_algorithm(rpt);
{
// netdata_log_info("STREAM %s [receive from [%s]:%s]: initializing communication...", rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port);
diff --git a/streaming/rrdpush.c b/streaming/rrdpush.c
index 0878e73686..74af46fe03 100644
--- a/streaming/rrdpush.c
+++ b/streaming/rrdpush.c
@@ -111,6 +111,10 @@ int rrdpush_init() {
default_rrdpush_compression_enabled = (unsigned int)appconfig_get_boolean(&stream_config, CONFIG_SECTION_STREAM,
"enable compression", default_rrdpush_compression_enabled);
+ rrdpush_compression_levels[COMPRESSION_ALGORITHM_BROTLI] = (int)appconfig_get_number(
+ &stream_config, CONFIG_SECTION_STREAM, "brotli compression level",
+ rrdpush_compression_levels[COMPRESSION_ALGORITHM_BROTLI]);
+
rrdpush_compression_levels[COMPRESSION_ALGORITHM_ZSTD] = (int)appconfig_get_number(
&stream_config, CONFIG_SECTION_STREAM, "zstd compression level",
rrdpush_compression_levels[COMPRESSION_ALGORITHM_ZSTD]);
@@ -1440,6 +1444,7 @@ static struct {
{STREAM_CAP_SLOTS, "SLOTS" },
{STREAM_CAP_ZSTD, "ZSTD" },
{STREAM_CAP_GZIP, "GZIP" },
+ {STREAM_CAP_BROTLI, "BROTLI" },
{0 , NULL },
};
@@ -1558,57 +1563,3 @@ int32_t stream_capabilities_to_vn(uint32_t caps) {
if(caps & STREAM_CAP_CLABELS) return STREAM_OLD_VERSION_CLABELS;
return STREAM_OLD_VERSION_CLAIM; // if(caps & STREAM_CAP_CLAIM)
}
-
-int rrdpush_compression_levels[COMPRESSION_ALGORITHM_MAX] = {
- [COMPRESSION_ALGORITHM_NONE] = 0,
- [COMPRESSION_ALGORITHM_ZSTD] = 3, // 1 (faster) - 22 (best compression),
- [COMPRESSION_ALGORITHM_LZ4] = 1, // 1 (best compression) - 9 (faster)
- [COMPRESSION_ALGORITHM_GZIP] = 1, // 1 (faster) - 9 (best compression)
-};
-
-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_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_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;
-}
diff --git a/streaming/rrdpush.h b/streaming/rrdpush.h
index c392af307e..4eab0ccea4 100644
--- a/streaming/rrdpush.h
+++ b/streaming/rrdpush.h
@@ -51,6 +51,7 @@ typedef enum {
STREAM_CAP_SLOTS = (1 << 18), // the sender can appoint a unique slot for each chart
STREAM_CAP_ZSTD = (1 << 19), // ZSTD compression supported
STREAM_CAP_GZIP = (1 << 20), // GZIP compression supported
+ STREAM_CAP_BROTLI = (1 << 21), // BROTLI compression supported
STREAM_CAP_INVALID = (1 << 30), // used as an invalid value for capabilities when this is set
// this must be signed int, so don't use the last bit
@@ -69,7 +70,13 @@ typedef enum {
#define STREAM_CAP_ZSTD_AVAILABLE 0
#endif // ENABLE_ZSTD
-#define STREAM_CAP_COMPRESSIONS_AVAILABLE (STREAM_CAP_LZ4_AVAILABLE|STREAM_CAP_ZSTD_AVAILABLE|STREAM_CAP_GZIP)
+#ifdef ENABLE_BROTLI
+#define STREAM_CAP_BROTLI_AVAILABLE STREAM_CAP_BROTLI
+#else
+#define STREAM_CAP_BROTLI_AVAILABLE 0
+#endif // ENABLE_BROTLI
+
+#define STREAM_CAP_COMPRESSIONS_AVAILABLE (STREAM_CAP_LZ4_AVAILABLE|STREAM_CAP_ZSTD_AVAILABLE|STREAM_CAP_BROTLI_AVAILABLE|STREAM_CAP_GZIP)
extern STREAM_CAPABILITIES globally_disabled_capabilities;
@@ -709,5 +716,8 @@ void rrdpush_send_dyncfg_reset(RRDHOST *host, const char *plugin_name);
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_H
diff --git a/streaming/sender.c b/streaming/sender.c
index d706cde255..d7fd03f135 100644
--- a/streaming/sender.c
+++ b/streaming/sender.c
@@ -69,43 +69,6 @@ BUFFER *sender_start(struct sender_state *s) {
static inline void rrdpush_sender_thread_close_socket(RRDHOST *host);
-/*
-* In case of stream compression buffer overflow
-* Inform the user through the error log file and
-* deactivate compression by downgrading the stream protocol.
-*/
-static inline void deactivate_compression(struct sender_state *s) {
- worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_NO_COMPRESSION);
-
- 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;
- }
-
- rrdpush_sender_thread_close_socket(s->host);
-}
-
#define SENDER_BUFFER_ADAPT_TO_TIMES_MAX_SIZE 3
// Collector thread finishing a transmission
@@ -179,7 +142,9 @@ void sender_commit(struct sender_state *s, BUFFER *wb, STREAM_TRAFFIC_TYPE type)
netdata_log_error("STREAM %s [send to %s]: COMPRESSION failed again. Deactivating compression",
rrdhost_hostname(s->host), s->connected_to);
- deactivate_compression(s);
+ worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_NO_COMPRESSION);
+ rrdpush_compression_deactivate(s);
+ rrdpush_sender_thread_close_socket(s->host);
sender_unlock(s);
return;
}
diff --git a/streaming/stream.conf b/streaming/stream.conf
index 0841953940..11c48fcdac 100644
--- a/streaming/stream.conf
+++ b/streaming/stream.conf
@@ -168,14 +168,14 @@
# Stream Compression
# By default it is enabled.
# You can control stream compression in this parent agent stream with options: yes | no
- enable compression = yes
+ #enable compression = yes
# select the order the compression algorithms will be used, when multiple are offered by the child
- compression algorithms order = zstd lz4 gzip
+ #compression algorithms order = zstd lz4 brotli gzip
# Replication
# Enable replication for all hosts using this api key. Default: enabled
- enable replication = yes
+ #enable replication = yes
# How many seconds to replicate from each child. Default: a day
#seconds to replicate = 86400