Brotli streaming compression (#16287)
* initial brotli implementation * configuration for brotli * default brotli compression to 6 * ordering of compression algorithms, always enables all compressions * brotli is the default, with compression level 3 * updated stream.conf * final brotli configuration * core re-org * more practical speed test
Costa Tsaousis committed
Oct 27, 2023 at 23:08 UTC
2bc7004ba81ed916341a9d87cb45c65c3f85683f
13 files changed
+640
-145
Makefile.am
+4
@@ -667,6 +667,8 @@ STREAMING_PLUGIN_FILES = \
667
streaming/rrdpush.c \
668
streaming/compression.c \
669
streaming/compression.h \
670
+ streaming/compression_brotli.c \
671
+ streaming/compression_brotli.h \
672
streaming/compression_gzip.c \
673
streaming/compression_gzip.h \
674
streaming/compression_lz4.c \
@@ -1151,6 +1153,8 @@ NETDATA_COMMON_LIBS = \
1153
$(OPTIONAL_UV_LIBS) \
1154
$(OPTIONAL_LZ4_LIBS) \
1155
$(OPTIONAL_ZSTD_LIBS) \
1156
+ $(OPTIONAL_BROTLIENC_LIBS) \
1157
+ $(OPTIONAL_BROTLIDEC_LIBS) \
1158
$(OPTIONAL_DATACHANNEL_LIBS) \
1159
libjudy.a \
1160
$(OPTIONAL_SSL_LIBS) \
configure.ac
+23
@@ -577,6 +577,27 @@ if test "x$LIBZSTD_FOUND" = "xyes"; then
577
OPTIONAL_ZSTD_LIBS="-lzstd"
578
fi
579
580
+# -----------------------------------------------------------------------------
581
+# brotli
582
+
583
+AC_CHECK_LIB([brotlienc], [BrotliEncoderCreateInstance, BrotliEncoderCompressStream],
584
+ [LIBBROTLIENC_FOUND=yes],
585
+ [LIBBROTLIENC_FOUND=no])
586
+
587
+if test "x$LIBBROTLIENC_FOUND" = "xyes"; then
588
+ AC_DEFINE([ENABLE_BROTLIENC], [1], [libbrotlienc usability])
589
+ OPTIONAL_BROTLIENC_LIBS="-lbrotlienc"
590
+fi
591
+
592
+AC_CHECK_LIB([brotlidec], [BrotliDecoderCreateInstance, BrotliDecoderDecompressStream],
593
+ [LIBBROTLIDEC_FOUND=yes],
594
+ [LIBBROTLIDEC_FOUND=no])
595
+
596
+if test "x$LIBBROTLIDEC_FOUND" = "xyes"; then
597
+ AC_DEFINE([ENABLE_BROTLIDEC], [1], [libbrotlidec usability])
598
+ OPTIONAL_BROTLIDEC_LIBS="-lbrotlidec"
599
+fi
600
+
601
# -----------------------------------------------------------------------------
602
# zlib
603
@@ -1912,6 +1933,8 @@ AC_SUBST([OPTIONAL_MATH_LIBS])
1933
AC_SUBST([OPTIONAL_DATACHANNEL_LIBS])
1934
AC_SUBST([OPTIONAL_UV_LIBS])
1935
AC_SUBST([OPTIONAL_LZ4_LIBS])
1936
+AC_SUBST([OPTIONAL_BROTLIENC_LIBS])
1937
+AC_SUBST([OPTIONAL_BROTLIDEC_LIBS])
1938
AC_SUBST([OPTIONAL_ZSTD_LIBS])
1939
AC_SUBST([OPTIONAL_SSL_LIBS])
1940
AC_SUBST([OPTIONAL_JSONC_LIBS])
daemon/buildinfo.c
+3
@@ -1115,6 +1115,9 @@ __attribute__((constructor)) void initialize_build_info(void) {
1115
1116
build_info_set_status(BIB_FEATURE_STREAMING_COMPRESSION, true);
1117
1118
+#ifdef ENABLE_BROTLI
1119
+ build_info_append_value(BIB_FEATURE_STREAMING_COMPRESSION, "brotli");
1120
+#endif
1121
#ifdef ENABLE_ZSTD
1122
build_info_append_value(BIB_FEATURE_STREAMING_COMPRESSION, "zstd");
1123
#endif
libnetdata/libnetdata.h
+4
@@ -11,6 +11,10 @@ extern "C" {
11
#include <config.h>
12
#endif
13
14
+#if defined(ENABLE_BROTLIENC) && defined(ENABLE_BROTLIDEC)
15
+#define ENABLE_BROTLI 1
16
+#endif
17
+
18
#ifdef ENABLE_OPENSSL
19
#define ENABLE_HTTPS 1
20
#endif
streaming/compression.c
+421
-4
@@ -12,6 +12,175 @@
12
#include "compression_zstd.h"
13
#endif
14
15
+#ifdef ENABLE_BROTLI
16
+#include "compression_brotli.h"
17
+#endif
18
+
19
+int rrdpush_compression_levels[COMPRESSION_ALGORITHM_MAX] = {
20
+ [COMPRESSION_ALGORITHM_NONE] = 0,
21
+ [COMPRESSION_ALGORITHM_ZSTD] = 3, // 1 (faster) - 22 (smaller)
22
+ [COMPRESSION_ALGORITHM_LZ4] = 1, // 1 (smaller) - 9 (faster)
23
+ [COMPRESSION_ALGORITHM_BROTLI] = 3, // 0 (faster) - 11 (smaller)
24
+ [COMPRESSION_ALGORITHM_GZIP] = 1, // 1 (faster) - 9 (smaller)
25
+};
26
+
27
+void rrdpush_parse_compression_order(struct receiver_state *rpt, const char *order) {
28
+ // empty all slots
29
+ for(size_t i = 0; i < COMPRESSION_ALGORITHM_MAX ;i++)
30
+ rpt->config.compression_priorities[i] = STREAM_CAP_NONE;
31
+
32
+ char *s = strdupz(order);
33
+
34
+ char *words[COMPRESSION_ALGORITHM_MAX + 100] = { NULL };
35
+ size_t num_words = quoted_strings_splitter_pluginsd(s, words, COMPRESSION_ALGORITHM_MAX + 100);
36
+ size_t slot = 0;
37
+ STREAM_CAPABILITIES added = STREAM_CAP_NONE;
38
+ for(size_t i = 0; i < num_words && slot < COMPRESSION_ALGORITHM_MAX ;i++) {
39
+ if((STREAM_CAP_ZSTD_AVAILABLE) && strcasecmp(words[i], "zstd") == 0 && !(added & STREAM_CAP_ZSTD)) {
40
+ rpt->config.compression_priorities[slot++] = STREAM_CAP_ZSTD;
41
+ added |= STREAM_CAP_ZSTD;
42
+ }
43
+ else if((STREAM_CAP_LZ4_AVAILABLE) && strcasecmp(words[i], "lz4") == 0 && !(added & STREAM_CAP_LZ4)) {
44
+ rpt->config.compression_priorities[slot++] = STREAM_CAP_LZ4;
45
+ added |= STREAM_CAP_LZ4;
46
+ }
47
+ else if((STREAM_CAP_BROTLI_AVAILABLE) && strcasecmp(words[i], "brotli") == 0 && !(added & STREAM_CAP_BROTLI)) {
48
+ rpt->config.compression_priorities[slot++] = STREAM_CAP_BROTLI;
49
+ added |= STREAM_CAP_BROTLI;
50
+ }
51
+ else if(strcasecmp(words[i], "gzip") == 0 && !(added & STREAM_CAP_GZIP)) {
52
+ rpt->config.compression_priorities[slot++] = STREAM_CAP_GZIP;
53
+ added |= STREAM_CAP_GZIP;
54
+ }
55
+ }
56
+
57
+ freez(s);
58
+
59
+ // make sure all participate
60
+ if((STREAM_CAP_ZSTD_AVAILABLE) && slot < COMPRESSION_ALGORITHM_MAX && !(added & STREAM_CAP_ZSTD))
61
+ rpt->config.compression_priorities[slot++] = STREAM_CAP_ZSTD;
62
+ if((STREAM_CAP_LZ4_AVAILABLE) && slot < COMPRESSION_ALGORITHM_MAX && !(added & STREAM_CAP_LZ4))
63
+ rpt->config.compression_priorities[slot++] = STREAM_CAP_LZ4;
64
+ if((STREAM_CAP_BROTLI_AVAILABLE) && slot < COMPRESSION_ALGORITHM_MAX && !(added & STREAM_CAP_BROTLI))
65
+ rpt->config.compression_priorities[slot++] = STREAM_CAP_BROTLI;
66
+ if(slot < COMPRESSION_ALGORITHM_MAX && !(added & STREAM_CAP_GZIP))
67
+ rpt->config.compression_priorities[slot++] = STREAM_CAP_GZIP;
68
+}
69
+
70
+void rrdpush_select_receiver_compression_algorithm(struct receiver_state *rpt) {
71
+ if (!rpt->config.rrdpush_compression)
72
+ rpt->capabilities &= ~STREAM_CAP_COMPRESSIONS_AVAILABLE;
73
+
74
+ // select the right compression before sending our capabilities to the child
75
+ if(stream_has_more_than_one_capability_of(rpt->capabilities, STREAM_CAP_COMPRESSIONS_AVAILABLE)) {
76
+ STREAM_CAPABILITIES compressions = rpt->capabilities & STREAM_CAP_COMPRESSIONS_AVAILABLE;
77
+ for(int i = 0; i < COMPRESSION_ALGORITHM_MAX; i++) {
78
+ STREAM_CAPABILITIES c = rpt->config.compression_priorities[i];
79
+
80
+ if(!(c & STREAM_CAP_COMPRESSIONS_AVAILABLE))
81
+ continue;
82
+
83
+ if(compressions & c) {
84
+ STREAM_CAPABILITIES exclude = compressions;
85
+ exclude &= ~c;
86
+
87
+ rpt->capabilities &= ~exclude;
88
+ break;
89
+ }
90
+ }
91
+ }
92
+}
93
+
94
+bool rrdpush_compression_initialize(struct sender_state *s) {
95
+ rrdpush_compressor_destroy(&s->compressor);
96
+
97
+ // IMPORTANT
98
+ // KEEP THE SAME ORDER IN DECOMPRESSION
99
+
100
+ if(stream_has_capability(s, STREAM_CAP_ZSTD))
101
+ s->compressor.algorithm = COMPRESSION_ALGORITHM_ZSTD;
102
+ else if(stream_has_capability(s, STREAM_CAP_LZ4))
103
+ s->compressor.algorithm = COMPRESSION_ALGORITHM_LZ4;
104
+ else if(stream_has_capability(s, STREAM_CAP_BROTLI))
105
+ s->compressor.algorithm = COMPRESSION_ALGORITHM_BROTLI;
106
+ else if(stream_has_capability(s, STREAM_CAP_GZIP))
107
+ s->compressor.algorithm = COMPRESSION_ALGORITHM_GZIP;
108
+ else
109
+ s->compressor.algorithm = COMPRESSION_ALGORITHM_NONE;
110
+
111
+ if(s->compressor.algorithm != COMPRESSION_ALGORITHM_NONE) {
112
+ s->compressor.level = rrdpush_compression_levels[s->compressor.algorithm];
113
+ rrdpush_compressor_init(&s->compressor);
114
+ return true;
115
+ }
116
+
117
+ return false;
118
+}
119
+
120
+bool rrdpush_decompression_initialize(struct receiver_state *rpt) {
121
+ rrdpush_decompressor_destroy(&rpt->decompressor);
122
+
123
+ // IMPORTANT
124
+ // KEEP THE SAME ORDER IN COMPRESSION
125
+
126
+ if(stream_has_capability(rpt, STREAM_CAP_ZSTD))
127
+ rpt->decompressor.algorithm = COMPRESSION_ALGORITHM_ZSTD;
128
+ else if(stream_has_capability(rpt, STREAM_CAP_LZ4))
129
+ rpt->decompressor.algorithm = COMPRESSION_ALGORITHM_LZ4;
130
+ else if(stream_has_capability(rpt, STREAM_CAP_BROTLI))
131
+ rpt->decompressor.algorithm = COMPRESSION_ALGORITHM_BROTLI;
132
+ else if(stream_has_capability(rpt, STREAM_CAP_GZIP))
133
+ rpt->decompressor.algorithm = COMPRESSION_ALGORITHM_GZIP;
134
+ else
135
+ rpt->decompressor.algorithm = COMPRESSION_ALGORITHM_NONE;
136
+
137
+ if(rpt->decompressor.algorithm != COMPRESSION_ALGORITHM_NONE) {
138
+ rrdpush_decompressor_init(&rpt->decompressor);
139
+ return true;
140
+ }
141
+
142
+ return false;
143
+}
144
+
145
+/*
146
+* In case of stream compression buffer overflow
147
+* Inform the user through the error log file and
148
+* deactivate compression by downgrading the stream protocol.
149
+*/
150
+void rrdpush_compression_deactivate(struct sender_state *s) {
151
+ switch(s->compressor.algorithm) {
152
+ case COMPRESSION_ALGORITHM_MAX:
153
+ case COMPRESSION_ALGORITHM_NONE:
154
+ netdata_log_error("STREAM_COMPRESSION: compression error on 'host:%s' without any compression enabled. Ignoring error.",
155
+ rrdhost_hostname(s->host));
156
+ break;
157
+
158
+ case COMPRESSION_ALGORITHM_GZIP:
159
+ netdata_log_error("STREAM_COMPRESSION: GZIP compression error on 'host:%s'. Disabling GZIP for this node.",
160
+ rrdhost_hostname(s->host));
161
+ s->disabled_capabilities |= STREAM_CAP_GZIP;
162
+ break;
163
+
164
+ case COMPRESSION_ALGORITHM_LZ4:
165
+ netdata_log_error("STREAM_COMPRESSION: LZ4 compression error on 'host:%s'. Disabling ZSTD for this node.",
166
+ rrdhost_hostname(s->host));
167
+ s->disabled_capabilities |= STREAM_CAP_LZ4;
168
+ break;
169
+
170
+ case COMPRESSION_ALGORITHM_ZSTD:
171
+ netdata_log_error("STREAM_COMPRESSION: ZSTD compression error on 'host:%s'. Disabling ZSTD for this node.",
172
+ rrdhost_hostname(s->host));
173
+ s->disabled_capabilities |= STREAM_CAP_ZSTD;
174
+ break;
175
+
176
+ case COMPRESSION_ALGORITHM_BROTLI:
177
+ netdata_log_error("STREAM_COMPRESSION: BROTLI compression error on 'host:%s'. Disabling BROTLI for this node.",
178
+ rrdhost_hostname(s->host));
179
+ s->disabled_capabilities |= STREAM_CAP_BROTLI;
180
+ break;
181
+ }
182
+}
183
+
184
// ----------------------------------------------------------------------------
185
// compressor public API
186
@@ -29,6 +198,12 @@ void rrdpush_compressor_init(struct compressor_state *state) {
198
break;
199
#endif
200
201
+#ifdef ENABLE_BROTLI
202
+ case COMPRESSION_ALGORITHM_BROTLI:
203
+ rrdpush_compressor_init_brotli(state);
204
+ break;
205
+#endif
206
+
207
default:
208
case COMPRESSION_ALGORITHM_GZIP:
209
rrdpush_compressor_init_gzip(state);
@@ -53,6 +228,12 @@ void rrdpush_compressor_destroy(struct compressor_state *state) {
228
break;
229
#endif
230
231
+#ifdef ENABLE_BROTLI
232
+ case COMPRESSION_ALGORITHM_BROTLI:
233
+ rrdpush_compressor_destroy_brotli(state);
234
+ break;
235
+#endif
236
+
237
default:
238
case COMPRESSION_ALGORITHM_GZIP:
239
rrdpush_compressor_destroy_gzip(state);
@@ -81,6 +262,12 @@ size_t rrdpush_compress(struct compressor_state *state, const char *data, size_t
262
break;
263
#endif
264
265
+#ifdef ENABLE_BROTLI
266
+ case COMPRESSION_ALGORITHM_BROTLI:
267
+ ret = rrdpush_compress_brotli(state, data, size, out);
268
+ break;
269
+#endif
270
+
271
default:
272
case COMPRESSION_ALGORITHM_GZIP:
273
ret = rrdpush_compress_gzip(state, data, size, out);
@@ -116,6 +303,12 @@ void rrdpush_decompressor_destroy(struct decompressor_state *state) {
303
break;
304
#endif
305
306
+#ifdef ENABLE_BROTLI
307
+ case COMPRESSION_ALGORITHM_BROTLI:
308
+ rrdpush_decompressor_destroy_brotli(state);
309
+ break;
310
+#endif
311
+
312
default:
313
case COMPRESSION_ALGORITHM_GZIP:
314
rrdpush_decompressor_destroy_gzip(state);
@@ -141,6 +334,12 @@ void rrdpush_decompressor_init(struct decompressor_state *state) {
334
break;
335
#endif
336
337
+#ifdef ENABLE_BROTLI
338
+ case COMPRESSION_ALGORITHM_BROTLI:
339
+ rrdpush_decompressor_init_brotli(state);
340
+ break;
341
+#endif
342
+
343
default:
344
case COMPRESSION_ALGORITHM_GZIP:
345
rrdpush_decompressor_init_gzip(state);
@@ -170,6 +369,12 @@ size_t rrdpush_decompress(struct decompressor_state *state, const char *compress
369
break;
370
#endif
371
372
+#ifdef ENABLE_BROTLI
373
+ case COMPRESSION_ALGORITHM_BROTLI:
374
+ ret = rrdpush_decompress_brotli(state, compressed_data, compressed_size);
375
+ break;
376
+#endif
377
+
378
default:
379
case COMPRESSION_ALGORITHM_GZIP:
380
ret = rrdpush_decompress_gzip(state, compressed_data, compressed_size);
@@ -190,6 +395,201 @@ size_t rrdpush_decompress(struct decompressor_state *state, const char *compress
395
// ----------------------------------------------------------------------------
396
// unit test
397
398
+void unittest_generate_random_name(char *dst, size_t size) {
399
+ if(size < 7)
400
+ size = 7;
401
+
402
+ size_t len = 5 + random() % (size - 6);
403
+
404
+ for(size_t i = 0; i < len ; i++) {
405
+ if(random() % 2 == 0)
406
+ dst[i] = 'A' + random() % 26;
407
+ else
408
+ dst[i] = 'a' + random() % 26;
409
+ }
410
+
411
+ dst[len] = '\0';
412
+}
413
+
414
+void unittest_generate_message(BUFFER *wb, time_t now_s, size_t counter) {
415
+ bool with_slots = true;
416
+ NUMBER_ENCODING integer_encoding = NUMBER_ENCODING_BASE64;
417
+ NUMBER_ENCODING doubles_encoding = NUMBER_ENCODING_BASE64;
418
+ time_t update_every = 1;
419
+ time_t point_end_time_s = now_s;
420
+ time_t wall_clock_time_s = now_s;
421
+ size_t chart_slot = counter + 1;
422
+ size_t dimensions = 2 + random() % 5;
423
+ char chart[RRD_ID_LENGTH_MAX + 1] = "name";
424
+ unittest_generate_random_name(chart, 5 + random() % 30);
425
+
426
+ buffer_fast_strcat(wb, PLUGINSD_KEYWORD_BEGIN_V2, sizeof(PLUGINSD_KEYWORD_BEGIN_V2) - 1);
427
+
428
+ if(with_slots) {
429
+ buffer_fast_strcat(wb, " "PLUGINSD_KEYWORD_SLOT":", sizeof(PLUGINSD_KEYWORD_SLOT) - 1 + 2);
430
+ buffer_print_uint64_encoded(wb, integer_encoding, chart_slot);
431
+ }
432
+
433
+ buffer_fast_strcat(wb, " '", 2);
434
+ buffer_strcat(wb, chart);
435
+ buffer_fast_strcat(wb, "' ", 2);
436
+ buffer_print_uint64_encoded(wb, integer_encoding, update_every);
437
+ buffer_fast_strcat(wb, " ", 1);
438
+ buffer_print_uint64_encoded(wb, integer_encoding, point_end_time_s);
439
+ buffer_fast_strcat(wb, " ", 1);
440
+ if(point_end_time_s == wall_clock_time_s)
441
+ buffer_fast_strcat(wb, "#", 1);
442
+ else
443
+ buffer_print_uint64_encoded(wb, integer_encoding, wall_clock_time_s);
444
+ buffer_fast_strcat(wb, "\n", 1);
445
+
446
+
447
+ for(size_t d = 0; d < dimensions ;d++) {
448
+ size_t dim_slot = d + 1;
449
+ char dim_id[RRD_ID_LENGTH_MAX + 1] = "dimension";
450
+ unittest_generate_random_name(dim_id, 10 + random() % 20);
451
+ int64_t last_collected_value = (random() % 2 == 0) ? (int64_t)(counter + d) : (int64_t)random();
452
+ NETDATA_DOUBLE value = (random() % 2 == 0) ? (NETDATA_DOUBLE)random() / ((NETDATA_DOUBLE)random() + 1) : (NETDATA_DOUBLE)last_collected_value;
453
+ SN_FLAGS flags = (random() % 1000 == 0) ? SN_FLAG_NONE : SN_FLAG_NOT_ANOMALOUS;
454
+
455
+ buffer_fast_strcat(wb, PLUGINSD_KEYWORD_SET_V2, sizeof(PLUGINSD_KEYWORD_SET_V2) - 1);
456
+
457
+ if(with_slots) {
458
+ buffer_fast_strcat(wb, " "PLUGINSD_KEYWORD_SLOT":", sizeof(PLUGINSD_KEYWORD_SLOT) - 1 + 2);
459
+ buffer_print_uint64_encoded(wb, integer_encoding, dim_slot);
460
+ }
461
+
462
+ buffer_fast_strcat(wb, " '", 2);
463
+ buffer_strcat(wb, dim_id);
464
+ buffer_fast_strcat(wb, "' ", 2);
465
+ buffer_print_int64_encoded(wb, integer_encoding, last_collected_value);
466
+ buffer_fast_strcat(wb, " ", 1);
467
+
468
+ if((NETDATA_DOUBLE)last_collected_value == value)
469
+ buffer_fast_strcat(wb, "#", 1);
470
+ else
471
+ buffer_print_netdata_double_encoded(wb, doubles_encoding, value);
472
+
473
+ buffer_fast_strcat(wb, " ", 1);
474
+ buffer_print_sn_flags(wb, flags, true);
475
+ buffer_fast_strcat(wb, "\n", 1);
476
+ }
477
+
478
+ buffer_fast_strcat(wb, PLUGINSD_KEYWORD_END_V2 "\n", sizeof(PLUGINSD_KEYWORD_END_V2) - 1 + 1);
479
+}
480
+
481
+int unittest_rrdpush_compression_speed(compression_algorithm_t algorithm, const char *name) {
482
+ fprintf(stderr, "\nTesting streaming compression speed with %s\n", name);
483
+
484
+ struct compressor_state cctx = {
485
+ .initialized = false,
486
+ .algorithm = algorithm,
487
+ };
488
+ struct decompressor_state dctx = {
489
+ .initialized = false,
490
+ .algorithm = algorithm,
491
+ };
492
+
493
+ rrdpush_compressor_init(&cctx);
494
+ rrdpush_decompressor_init(&dctx);
495
+
496
+ int errors = 0;
497
+
498
+ BUFFER *wb = buffer_create(COMPRESSION_MAX_MSG_SIZE, NULL);
499
+ time_t now_s = now_realtime_sec();
500
+ usec_t compression_ut = 0;
501
+ usec_t decompression_ut = 0;
502
+ size_t bytes_compressed = 0;
503
+ size_t bytes_uncompressed = 0;
504
+
505
+ usec_t compression_started_ut = now_monotonic_usec();
506
+ usec_t decompression_started_ut = compression_started_ut;
507
+
508
+ for(int i = 0; i < 10000 ;i++) {
509
+ compression_started_ut = now_monotonic_usec();
510
+ decompression_ut += compression_started_ut - decompression_started_ut;
511
+
512
+ buffer_flush(wb);
513
+ while(buffer_strlen(wb) < COMPRESSION_MAX_MSG_SIZE - 1024)
514
+ unittest_generate_message(wb, now_s, i);
515
+
516
+ const char *txt = buffer_tostring(wb);
517
+ size_t txt_len = buffer_strlen(wb);
518
+ bytes_uncompressed += txt_len;
519
+
520
+ const char *out;
521
+ size_t size = rrdpush_compress(&cctx, txt, txt_len, &out);
522
+
523
+ bytes_compressed += size;
524
+ decompression_started_ut = now_monotonic_usec();
525
+ compression_ut += decompression_started_ut - compression_started_ut;
526
+
527
+ if(size == 0) {
528
+ fprintf(stderr, "iteration %d: compressed size %zu is zero\n",
529
+ i, size);
530
+ errors++;
531
+ goto cleanup;
532
+ }
533
+ else if(size >= COMPRESSION_MAX_CHUNK) {
534
+ fprintf(stderr, "iteration %d: compressed size %zu exceeds max allowed size\n",
535
+ i, size);
536
+ errors++;
537
+ goto cleanup;
538
+ }
539
+ else {
540
+ size_t dtxt_len = rrdpush_decompress(&dctx, out, size);
541
+ char *dtxt = (char *) &dctx.output.data[dctx.output.read_pos];
542
+
543
+ if(rrdpush_decompressed_bytes_in_buffer(&dctx) != dtxt_len) {
544
+ fprintf(stderr, "iteration %d: decompressed size %zu does not rrdpush_decompressed_bytes_in_buffer() %zu\n",
545
+ i, dtxt_len, rrdpush_decompressed_bytes_in_buffer(&dctx)
546
+ );
547
+ errors++;
548
+ goto cleanup;
549
+ }
550
+
551
+ if(!dtxt_len) {
552
+ fprintf(stderr, "iteration %d: decompressed size is zero\n", i);
553
+ errors++;
554
+ goto cleanup;
555
+ }
556
+ else if(dtxt_len != txt_len) {
557
+ fprintf(stderr, "iteration %d: decompressed size %zu does not match original size %zu\n",
558
+ i, dtxt_len, txt_len
559
+ );
560
+ errors++;
561
+ goto cleanup;
562
+ }
563
+ else {
564
+ if(memcmp(txt, dtxt, txt_len) != 0) {
565
+ fprintf(stderr, "iteration %d: decompressed data '%s' do not match original data length %zu\n",
566
+ i, dtxt, txt_len);
567
+ errors++;
568
+ goto cleanup;
569
+ }
570
+ }
571
+ }
572
+
573
+ // here we are supposed to copy the data and advance the position
574
+ dctx.output.read_pos += rrdpush_decompressed_bytes_in_buffer(&dctx);
575
+ }
576
+
577
+cleanup:
578
+ rrdpush_compressor_destroy(&cctx);
579
+ rrdpush_decompressor_destroy(&dctx);
580
+
581
+ if(errors)
582
+ fprintf(stderr, "Compression with %s: FAILED (%d errors)\n", name, errors);
583
+ else
584
+ fprintf(stderr, "Compression with %s: OK "
585
+ "(compression %zu usec, decompression %zu usec, bytes raw %zu, compressed %zu, savings ratio %0.2f%%)\n",
586
+ name, compression_ut, decompression_ut,
587
+ bytes_uncompressed, bytes_compressed,
588
+ 100.0 - (double)bytes_compressed * 100.0 / (double)bytes_uncompressed);
589
+
590
+ return errors;
591
+}
592
+
593
int unittest_rrdpush_compression(compression_algorithm_t algorithm, const char *name) {
594
fprintf(stderr, "\nTesting streaming compression with %s\n", name);
595
@@ -218,7 +618,13 @@ int unittest_rrdpush_compression(compression_algorithm_t algorithm, const char *
618
const char *out;
619
size_t size = rrdpush_compress(&cctx, txt, txt_len, &out);
620
221
- if(size >= COMPRESSION_MAX_CHUNK) {
621
+ if(size == 0) {
622
+ fprintf(stderr, "iteration %d: compressed size %zu is zero\n",
623
+ i, size);
624
+ errors++;
625
+ goto cleanup;
626
+ }
627
+ else if(size >= COMPRESSION_MAX_CHUNK) {
628
fprintf(stderr, "iteration %d: compressed size %zu exceeds max allowed size\n",
629
i, size);
630
errors++;
@@ -236,7 +642,12 @@ int unittest_rrdpush_compression(compression_algorithm_t algorithm, const char *
642
goto cleanup;
643
}
644
239
- if(dtxt_len != txt_len) {
645
+ if(!dtxt_len) {
646
+ fprintf(stderr, "iteration %d: decompressed size is zero\n", i);
647
+ errors++;
648
+ goto cleanup;
649
+ }
650
+ else if(dtxt_len != txt_len) {
651
fprintf(stderr, "iteration %d: decompressed size %zu does not match original size %zu\n",
652
i, dtxt_len, txt_len
653
);
@@ -278,9 +689,15 @@ cleanup:
689
int unittest_rrdpush_compressions(void) {
690
int ret = 0;
691
281
- ret += unittest_rrdpush_compression(COMPRESSION_ALGORITHM_GZIP, "GZIP");
282
- ret += unittest_rrdpush_compression(COMPRESSION_ALGORITHM_LZ4, "LZ4");
692
ret += unittest_rrdpush_compression(COMPRESSION_ALGORITHM_ZSTD, "ZSTD");
693
+ ret += unittest_rrdpush_compression(COMPRESSION_ALGORITHM_LZ4, "LZ4");
694
+ ret += unittest_rrdpush_compression(COMPRESSION_ALGORITHM_BROTLI, "BROTLI");
695
+ ret += unittest_rrdpush_compression(COMPRESSION_ALGORITHM_GZIP, "GZIP");
696
+
697
+ ret += unittest_rrdpush_compression_speed(COMPRESSION_ALGORITHM_ZSTD, "ZSTD");
698
+ ret += unittest_rrdpush_compression_speed(COMPRESSION_ALGORITHM_LZ4, "LZ4");
699
+ ret += unittest_rrdpush_compression_speed(COMPRESSION_ALGORITHM_BROTLI, "BROTLI");
700
+ ret += unittest_rrdpush_compression_speed(COMPRESSION_ALGORITHM_GZIP, "GZIP");
701
702
return ret;
703
}
streaming/compression.h
+4
@@ -26,6 +26,7 @@ typedef enum {
26
COMPRESSION_ALGORITHM_ZSTD,
27
COMPRESSION_ALGORITHM_LZ4,
28
COMPRESSION_ALGORITHM_GZIP,
29
+ COMPRESSION_ALGORITHM_BROTLI,
30
31
// terminator
32
COMPRESSION_ALGORITHM_MAX,
@@ -33,6 +34,9 @@ typedef enum {
34
35
extern int rrdpush_compression_levels[COMPRESSION_ALGORITHM_MAX];
36
37
+// this defines the order the algorithms will be selected by the receiver (parent)
38
+#define RRDPUSH_COMPRESSION_ALGORITHMS_ORDER "zstd lz4 brotli gzip"
39
+
40
// ----------------------------------------------------------------------------
41
42
typedef struct simple_ring_buffer {
streaming/compression_brotli.c
new
+142
@@ -0,0 +1,142 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+#include "compression_brotli.h"
4
+
5
+#ifdef ENABLE_BROTLI
6
+#include <brotli/encode.h>
7
+#include <brotli/decode.h>
8
+
9
+void rrdpush_compressor_init_brotli(struct compressor_state *state) {
10
+ if (!state->initialized) {
11
+ state->initialized = true;
12
+ state->stream = BrotliEncoderCreateInstance(NULL, NULL, NULL);
13
+
14
+ if (state->level < BROTLI_MIN_QUALITY) {
15
+ state->level = BROTLI_MIN_QUALITY;
16
+ } else if (state->level > BROTLI_MAX_QUALITY) {
17
+ state->level = BROTLI_MAX_QUALITY;
18
+ }
19
+
20
+ BrotliEncoderSetParameter(state->stream, BROTLI_PARAM_QUALITY, state->level);
21
+ }
22
+}
23
+
24
+void rrdpush_compressor_destroy_brotli(struct compressor_state *state) {
25
+ if (state->stream) {
26
+ BrotliEncoderDestroyInstance(state->stream);
27
+ state->stream = NULL;
28
+ }
29
+}
30
+
31
+size_t rrdpush_compress_brotli(struct compressor_state *state, const char *data, size_t size, const char **out) {
32
+ if (unlikely(!state || !size || !out))
33
+ return 0;
34
+
35
+ simple_ring_buffer_make_room(&state->output, MAX(BrotliEncoderMaxCompressedSize(size), COMPRESSION_MAX_CHUNK));
36
+
37
+ size_t available_out = state->output.size;
38
+
39
+ size_t available_in = size;
40
+ const uint8_t *next_in = (const uint8_t *)data;
41
+ uint8_t *next_out = (uint8_t *)state->output.data;
42
+
43
+ if (!BrotliEncoderCompressStream(state->stream, BROTLI_OPERATION_FLUSH, &available_in, &next_in, &available_out, &next_out, NULL)) {
44
+ netdata_log_error("STREAM: Brotli compression failed.");
45
+ return 0;
46
+ }
47
+
48
+ if(available_in != 0) {
49
+ netdata_log_error("STREAM: BrotliEncoderCompressStream() did not use all the input buffer, %u bytes out of %zu remain",
50
+ available_in, size);
51
+ return 0;
52
+ }
53
+
54
+ size_t compressed_size = state->output.size - available_out;
55
+ if(available_out == 0) {
56
+ netdata_log_error("STREAM: BrotliEncoderCompressStream() needs a bigger output buffer than the one we provided "
57
+ "(output buffer %zu bytes, compressed payload %zu bytes)",
58
+ state->output.size, size);
59
+ return 0;
60
+ }
61
+
62
+ if(compressed_size == 0) {
63
+ netdata_log_error("STREAM: BrotliEncoderCompressStream() did not produce any output from the input provided "
64
+ "(input buffer %zu bytes)",
65
+ size);
66
+ return 0;
67
+ }
68
+
69
+ state->sender_locked.total_compressions++;
70
+ state->sender_locked.total_uncompressed += size - available_in;
71
+ state->sender_locked.total_compressed += compressed_size;
72
+
73
+ *out = state->output.data;
74
+ return compressed_size;
75
+}
76
+
77
+void rrdpush_decompressor_init_brotli(struct decompressor_state *state) {
78
+ if (!state->initialized) {
79
+ state->initialized = true;
80
+ state->stream = BrotliDecoderCreateInstance(NULL, NULL, NULL);
81
+
82
+ simple_ring_buffer_make_room(&state->output, COMPRESSION_MAX_CHUNK);
83
+ }
84
+}
85
+
86
+void rrdpush_decompressor_destroy_brotli(struct decompressor_state *state) {
87
+ if (state->stream) {
88
+ BrotliDecoderDestroyInstance(state->stream);
89
+ state->stream = NULL;
90
+ }
91
+}
92
+
93
+size_t rrdpush_decompress_brotli(struct decompressor_state *state, const char *compressed_data, size_t compressed_size) {
94
+ if (unlikely(!state || !compressed_data || !compressed_size))
95
+ return 0;
96
+
97
+ // The state.output ring buffer is always EMPTY at this point,
98
+ // meaning that (state->output.read_pos == state->output.write_pos)
99
+ // However, THEY ARE NOT ZERO.
100
+
101
+ size_t available_out = state->output.size;
102
+ size_t available_in = compressed_size;
103
+ const uint8_t *next_in = (const uint8_t *)compressed_data;
104
+ uint8_t *next_out = (uint8_t *)state->output.data;
105
+
106
+ if (BrotliDecoderDecompressStream(state->stream, &available_in, &next_in, &available_out, &next_out, NULL) == BROTLI_DECODER_RESULT_ERROR) {
107
+ netdata_log_error("STREAM: Brotli decompression failed.");
108
+ return 0;
109
+ }
110
+
111
+ if(available_in != 0) {
112
+ netdata_log_error("STREAM: BrotliDecoderDecompressStream() did not use all the input buffer, %u bytes out of %zu remain",
113
+ available_in, compressed_size);
114
+ return 0;
115
+ }
116
+
117
+ size_t decompressed_size = state->output.size - available_out;
118
+ if(available_out == 0) {
119
+ netdata_log_error("STREAM: BrotliDecoderDecompressStream() needs a bigger output buffer than the one we provided "
120
+ "(output buffer %zu bytes, compressed payload %zu bytes)",
121
+ state->output.size, compressed_size);
122
+ return 0;
123
+ }
124
+
125
+ if(decompressed_size == 0) {
126
+ netdata_log_error("STREAM: BrotliDecoderDecompressStream() did not produce any output from the input provided "
127
+ "(input buffer %zu bytes)",
128
+ compressed_size);
129
+ return 0;
130
+ }
131
+
132
+ state->output.read_pos = 0;
133
+ state->output.write_pos = decompressed_size;
134
+
135
+ state->total_compressed += compressed_size - available_in;
136
+ state->total_uncompressed += decompressed_size;
137
+ state->total_compressions++;
138
+
139
+ return decompressed_size;
140
+}
141
+
142
+#endif // ENABLE_BROTLI
streaming/compression_brotli.h
new
+15
@@ -0,0 +1,15 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+#include "compression.h"
4
+
5
+#ifndef NETDATA_STREAMING_COMPRESSION_BROTLI_H
6
+#define NETDATA_STREAMING_COMPRESSION_BROTLI_H
7
+
8
+void rrdpush_compressor_init_brotli(struct compressor_state *state);
9
+void rrdpush_compressor_destroy_brotli(struct compressor_state *state);
10
+size_t rrdpush_compress_brotli(struct compressor_state *state, const char *data, size_t size, const char **out);
11
+size_t rrdpush_decompress_brotli(struct decompressor_state *state, const char *compressed_data, size_t compressed_size);
12
+void rrdpush_decompressor_init_brotli(struct decompressor_state *state);
13
+void rrdpush_decompressor_destroy_brotli(struct decompressor_state *state);
14
+
15
+#endif //NETDATA_STREAMING_COMPRESSION_BROTLI_H
streaming/receiver.c
+2
-45
@@ -562,29 +562,6 @@ void rrdpush_receive_log_status(struct receiver_state *rpt, const char *msg, con
562
563
}
564
565
-static void rrdpush_parse_compression_order(struct receiver_state *rpt, const char *order) {
566
- rpt->config.compression_priorities[0] = STREAM_CAP_ZSTD;
567
- rpt->config.compression_priorities[1] = STREAM_CAP_LZ4;
568
- rpt->config.compression_priorities[2] = STREAM_CAP_GZIP;
569
-
570
- char *s = strdupz(order);
571
-
572
- char *words[COMPRESSION_ALGORITHM_MAX] = { NULL };
573
- size_t num_words = quoted_strings_splitter_pluginsd(s, words, COMPRESSION_ALGORITHM_MAX);
574
- for(size_t i = 0; i < num_words ;i++) {
575
- if(strcasecmp(words[i], "zstd") == 0)
576
- rpt->config.compression_priorities[i] = STREAM_CAP_ZSTD;
577
- else if(strcasecmp(words[i], "lz4") == 0)
578
- rpt->config.compression_priorities[i] = STREAM_CAP_LZ4;
579
- else if(strcasecmp(words[i], "gzip") == 0)
580
- rpt->config.compression_priorities[i] = STREAM_CAP_GZIP;
581
- else
582
- rpt->config.compression_priorities[i] = 0;
583
- }
584
-
585
- freez(s);
586
-}
587
-
565
static void rrdpush_receive(struct receiver_state *rpt)
566
{
567
rpt->config.mode = default_rrd_memory_mode;
@@ -658,7 +635,7 @@ static void rrdpush_receive(struct receiver_state *rpt)
635
rpt->config.rrdpush_compression = appconfig_get_boolean(&stream_config, rpt->machine_guid, "enable compression", rpt->config.rrdpush_compression);
636
637
if(rpt->config.rrdpush_compression) {
661
- char *order = appconfig_get(&stream_config, rpt->key, "compression algorithms order", "zstd lz4 gzip");
638
+ char *order = appconfig_get(&stream_config, rpt->key, "compression algorithms order", RRDPUSH_COMPRESSION_ALGORITHMS_ORDER);
639
order = appconfig_get(&stream_config, rpt->machine_guid, "compression algorithms order", order);
640
rrdpush_parse_compression_order(rpt, order);
641
}
@@ -755,27 +732,7 @@ static void rrdpush_receive(struct receiver_state *rpt)
732
snprintfz(cd.fullfilename, FILENAME_MAX, "%s:%s", rpt->client_ip, rpt->client_port);
733
snprintfz(cd.cmd, PLUGINSD_CMD_MAX, "%s:%s", rpt->client_ip, rpt->client_port);
734
758
- if (!rpt->config.rrdpush_compression)
759
- rpt->capabilities &= ~STREAM_CAP_COMPRESSIONS_AVAILABLE;
760
-
761
- // select the right compression before sending our capabilities to the child
762
- if(stream_has_more_than_one_capability_of(rpt->capabilities, STREAM_CAP_COMPRESSIONS_AVAILABLE)) {
763
- STREAM_CAPABILITIES compressions = rpt->capabilities & STREAM_CAP_COMPRESSIONS_AVAILABLE;
764
- for(int i = 0; i < COMPRESSION_ALGORITHM_MAX; i++) {
765
- STREAM_CAPABILITIES c = rpt->config.compression_priorities[i];
766
-
767
- if(!(c & STREAM_CAP_COMPRESSIONS_AVAILABLE))
768
- continue;
769
-
770
- if(compressions & c) {
771
- STREAM_CAPABILITIES exclude = compressions;
772
- exclude &= ~c;
773
-
774
- rpt->capabilities &= ~exclude;
775
- break;
776
- }
777
- }
778
- }
735
+ rrdpush_select_receiver_compression_algorithm(rpt);
736
737
{
738
// netdata_log_info("STREAM %s [receive from [%s]:%s]: initializing communication...", rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port);
streaming/rrdpush.c
+5
-54
@@ -111,6 +111,10 @@ int rrdpush_init() {
111
default_rrdpush_compression_enabled = (unsigned int)appconfig_get_boolean(&stream_config, CONFIG_SECTION_STREAM,
112
"enable compression", default_rrdpush_compression_enabled);
113
114
+ rrdpush_compression_levels[COMPRESSION_ALGORITHM_BROTLI] = (int)appconfig_get_number(
115
+ &stream_config, CONFIG_SECTION_STREAM, "brotli compression level",
116
+ rrdpush_compression_levels[COMPRESSION_ALGORITHM_BROTLI]);
117
+
118
rrdpush_compression_levels[COMPRESSION_ALGORITHM_ZSTD] = (int)appconfig_get_number(
119
&stream_config, CONFIG_SECTION_STREAM, "zstd compression level",
120
rrdpush_compression_levels[COMPRESSION_ALGORITHM_ZSTD]);
@@ -1440,6 +1444,7 @@ static struct {
1444
{STREAM_CAP_SLOTS, "SLOTS" },
1445
{STREAM_CAP_ZSTD, "ZSTD" },
1446
{STREAM_CAP_GZIP, "GZIP" },
1447
+ {STREAM_CAP_BROTLI, "BROTLI" },
1448
{0 , NULL },
1449
};
1450
@@ -1558,57 +1563,3 @@ int32_t stream_capabilities_to_vn(uint32_t caps) {
1563
if(caps & STREAM_CAP_CLABELS) return STREAM_OLD_VERSION_CLABELS;
1564
return STREAM_OLD_VERSION_CLAIM; // if(caps & STREAM_CAP_CLAIM)
1565
}
1561
-
1562
-int rrdpush_compression_levels[COMPRESSION_ALGORITHM_MAX] = {
1563
- [COMPRESSION_ALGORITHM_NONE] = 0,
1564
- [COMPRESSION_ALGORITHM_ZSTD] = 3, // 1 (faster) - 22 (best compression),
1565
- [COMPRESSION_ALGORITHM_LZ4] = 1, // 1 (best compression) - 9 (faster)
1566
- [COMPRESSION_ALGORITHM_GZIP] = 1, // 1 (faster) - 9 (best compression)
1567
-};
1568
-
1569
-bool rrdpush_compression_initialize(struct sender_state *s) {
1570
- rrdpush_compressor_destroy(&s->compressor);
1571
-
1572
- // IMPORTANT
1573
- // KEEP THE SAME ORDER IN DECOMPRESSION
1574
-
1575
- if(stream_has_capability(s, STREAM_CAP_ZSTD))
1576
- s->compressor.algorithm = COMPRESSION_ALGORITHM_ZSTD;
1577
- else if(stream_has_capability(s, STREAM_CAP_LZ4))
1578
- s->compressor.algorithm = COMPRESSION_ALGORITHM_LZ4;
1579
- else if(stream_has_capability(s, STREAM_CAP_GZIP))
1580
- s->compressor.algorithm = COMPRESSION_ALGORITHM_GZIP;
1581
- else
1582
- s->compressor.algorithm = COMPRESSION_ALGORITHM_NONE;
1583
-
1584
- if(s->compressor.algorithm != COMPRESSION_ALGORITHM_NONE) {
1585
- s->compressor.level = rrdpush_compression_levels[s->compressor.algorithm];
1586
- rrdpush_compressor_init(&s->compressor);
1587
- return true;
1588
- }
1589
-
1590
- return false;
1591
-}
1592
-
1593
-bool rrdpush_decompression_initialize(struct receiver_state *rpt) {
1594
- rrdpush_decompressor_destroy(&rpt->decompressor);
1595
-
1596
- // IMPORTANT
1597
- // KEEP THE SAME ORDER IN COMPRESSION
1598
-
1599
- if(stream_has_capability(rpt, STREAM_CAP_ZSTD))
1600
- rpt->decompressor.algorithm = COMPRESSION_ALGORITHM_ZSTD;
1601
- else if(stream_has_capability(rpt, STREAM_CAP_LZ4))
1602
- rpt->decompressor.algorithm = COMPRESSION_ALGORITHM_LZ4;
1603
- else if(stream_has_capability(rpt, STREAM_CAP_GZIP))
1604
- rpt->decompressor.algorithm = COMPRESSION_ALGORITHM_GZIP;
1605
- else
1606
- rpt->decompressor.algorithm = COMPRESSION_ALGORITHM_NONE;
1607
-
1608
- if(rpt->decompressor.algorithm != COMPRESSION_ALGORITHM_NONE) {
1609
- rrdpush_decompressor_init(&rpt->decompressor);
1610
- return true;
1611
- }
1612
-
1613
- return false;
1614
-}
streaming/rrdpush.h
+11
-1
@@ -51,6 +51,7 @@ typedef enum {
51
STREAM_CAP_SLOTS = (1 << 18), // the sender can appoint a unique slot for each chart
52
STREAM_CAP_ZSTD = (1 << 19), // ZSTD compression supported
53
STREAM_CAP_GZIP = (1 << 20), // GZIP compression supported
54
+ STREAM_CAP_BROTLI = (1 << 21), // BROTLI compression supported
55
56
STREAM_CAP_INVALID = (1 << 30), // used as an invalid value for capabilities when this is set
57
// this must be signed int, so don't use the last bit
@@ -69,7 +70,13 @@ typedef enum {
70
#define STREAM_CAP_ZSTD_AVAILABLE 0
71
#endif // ENABLE_ZSTD
72
72
-#define STREAM_CAP_COMPRESSIONS_AVAILABLE (STREAM_CAP_LZ4_AVAILABLE|STREAM_CAP_ZSTD_AVAILABLE|STREAM_CAP_GZIP)
73
+#ifdef ENABLE_BROTLI
74
+#define STREAM_CAP_BROTLI_AVAILABLE STREAM_CAP_BROTLI
75
+#else
76
+#define STREAM_CAP_BROTLI_AVAILABLE 0
77
+#endif // ENABLE_BROTLI
78
+
79
+#define STREAM_CAP_COMPRESSIONS_AVAILABLE (STREAM_CAP_LZ4_AVAILABLE|STREAM_CAP_ZSTD_AVAILABLE|STREAM_CAP_BROTLI_AVAILABLE|STREAM_CAP_GZIP)
80
81
extern STREAM_CAPABILITIES globally_disabled_capabilities;
82
@@ -709,5 +716,8 @@ void rrdpush_send_dyncfg_reset(RRDHOST *host, const char *plugin_name);
716
717
bool rrdpush_compression_initialize(struct sender_state *s);
718
bool rrdpush_decompression_initialize(struct receiver_state *rpt);
719
+void rrdpush_parse_compression_order(struct receiver_state *rpt, const char *order);
720
+void rrdpush_select_receiver_compression_algorithm(struct receiver_state *rpt);
721
+void rrdpush_compression_deactivate(struct sender_state *s);
722
723
#endif //NETDATA_RRDPUSH_H
streaming/sender.c
+3
-38
@@ -69,43 +69,6 @@ BUFFER *sender_start(struct sender_state *s) {
69
70
static inline void rrdpush_sender_thread_close_socket(RRDHOST *host);
71
72
-/*
73
-* In case of stream compression buffer overflow
74
-* Inform the user through the error log file and
75
-* deactivate compression by downgrading the stream protocol.
76
-*/
77
-static inline void deactivate_compression(struct sender_state *s) {
78
- worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_NO_COMPRESSION);
79
-
80
- switch(s->compressor.algorithm) {
81
- case COMPRESSION_ALGORITHM_MAX:
82
- case COMPRESSION_ALGORITHM_NONE:
83
- netdata_log_error("STREAM_COMPRESSION: compression error on 'host:%s' without any compression enabled. Ignoring error.",
84
- rrdhost_hostname(s->host));
85
- break;
86
-
87
- case COMPRESSION_ALGORITHM_GZIP:
88
- netdata_log_error("STREAM_COMPRESSION: GZIP compression error on 'host:%s'. Disabling GZIP for this node.",
89
- rrdhost_hostname(s->host));
90
- s->disabled_capabilities |= STREAM_CAP_GZIP;
91
- break;
92
-
93
- case COMPRESSION_ALGORITHM_LZ4:
94
- netdata_log_error("STREAM_COMPRESSION: LZ4 compression error on 'host:%s'. Disabling ZSTD for this node.",
95
- rrdhost_hostname(s->host));
96
- s->disabled_capabilities |= STREAM_CAP_LZ4;
97
- break;
98
-
99
- case COMPRESSION_ALGORITHM_ZSTD:
100
- netdata_log_error("STREAM_COMPRESSION: ZSTD compression error on 'host:%s'. Disabling ZSTD for this node.",
101
- rrdhost_hostname(s->host));
102
- s->disabled_capabilities |= STREAM_CAP_ZSTD;
103
- break;
104
- }
105
-
106
- rrdpush_sender_thread_close_socket(s->host);
107
-}
108
-
72
#define SENDER_BUFFER_ADAPT_TO_TIMES_MAX_SIZE 3
73
74
// Collector thread finishing a transmission
@@ -179,7 +142,9 @@ void sender_commit(struct sender_state *s, BUFFER *wb, STREAM_TRAFFIC_TYPE type)
142
netdata_log_error("STREAM %s [send to %s]: COMPRESSION failed again. Deactivating compression",
143
rrdhost_hostname(s->host), s->connected_to);
144
182
- deactivate_compression(s);
145
+ worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_NO_COMPRESSION);
146
+ rrdpush_compression_deactivate(s);
147
+ rrdpush_sender_thread_close_socket(s->host);
148
sender_unlock(s);
149
return;
150
}
streaming/stream.conf
+3
-3
@@ -168,14 +168,14 @@
168
# Stream Compression
169
# By default it is enabled.
170
# You can control stream compression in this parent agent stream with options: yes | no
171
- enable compression = yes
171
+ #enable compression = yes
172
173
# select the order the compression algorithms will be used, when multiple are offered by the child
174
- compression algorithms order = zstd lz4 gzip
174
+ #compression algorithms order = zstd lz4 brotli gzip
175
176
# Replication
177
# Enable replication for all hosts using this api key. Default: enabled
178
- enable replication = yes
178
+ #enable replication = yes
179
180
# How many seconds to replicate from each child. Default: a day
181
#seconds to replicate = 86400