diff options
Diffstat (limited to '')
| -rw-r--r-- | streaming/compression.c | 780 |
1 files changed, 653 insertions, 127 deletions
diff --git a/streaming/compression.c b/streaming/compression.c index 6d4a128b0..a94c8a0a6 100644 --- a/streaming/compression.c +++ b/streaming/compression.c @@ -1,181 +1,707 @@ -#include "rrdpush.h" +// SPDX-License-Identifier: GPL-3.0-or-later -#ifdef ENABLE_RRDPUSH_COMPRESSION -#include "lz4.h" +#include "compression.h" -#define STREAM_COMPRESSION_MSG "STREAM_COMPRESSION" +#include "compression_gzip.h" -/* - * Reset compressor state for a new stream - */ -void rrdpush_compressor_reset(struct compressor_state *state) { - if(!state->initialized) { - state->initialized = true; +#ifdef ENABLE_LZ4 +#include "compression_lz4.h" +#endif + +#ifdef ENABLE_ZSTD +#include "compression_zstd.h" +#endif - state->stream.lz4_stream = LZ4_createStream(); - state->stream.input_ring_buffer_size = LZ4_DECODER_RING_BUFFER_SIZE(COMPRESSION_MAX_MSG_SIZE * 2); - state->stream.input_ring_buffer = callocz(1, state->stream.input_ring_buffer_size); - state->compression_result_buffer_size = 0; +#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; + } } - LZ4_resetStream_fast(state->stream.lz4_stream); + 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; - state->stream.input_ring_buffer_pos = 0; + rpt->capabilities &= ~exclude; + break; + } + } + } } -/* - * Destroy compressor state and all related data - */ -void rrdpush_compressor_destroy(struct compressor_state *state) { - if (state->stream.lz4_stream) { - LZ4_freeStream(state->stream.lz4_stream); - state->stream.lz4_stream = NULL; +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; } - freez(state->stream.input_ring_buffer); - state->stream.input_ring_buffer = NULL; + return false; +} - freez(state->compression_result_buffer); - state->compression_result_buffer = NULL; +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; + } - state->initialized = false; + return false; } /* - * 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(struct compressor_state *state, const char *data, size_t size, char **out) { - if(unlikely(!state || !size || !out)) - return 0; - - if(unlikely(size > COMPRESSION_MAX_MSG_SIZE)) { - netdata_log_error("RRDPUSH COMPRESS: Compression Failed - Message size %lu above compression buffer limit: %d", - (long unsigned int)size, COMPRESSION_MAX_MSG_SIZE); - return 0; +* 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; } +} - size_t max_dst_size = LZ4_COMPRESSBOUND(size); - size_t data_size = max_dst_size + RRDPUSH_COMPRESSION_SIGNATURE_SIZE; +// ---------------------------------------------------------------------------- +// compressor public API - if (!state->compression_result_buffer) { - state->compression_result_buffer = mallocz(data_size); - state->compression_result_buffer_size = data_size; +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; } - else if(unlikely(state->compression_result_buffer_size < data_size)) { - state->compression_result_buffer = reallocz(state->compression_result_buffer, data_size); - state->compression_result_buffer_size = data_size; + + 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; } - // the ring buffer always has space for LZ4_MAX_MSG_SIZE - memcpy(state->stream.input_ring_buffer + state->stream.input_ring_buffer_pos, data, size); + state->initialized = false; - // this call needs the last 64K of our previous data - // they are available in the ring buffer - long int compressed_data_size = LZ4_compress_fast_continue( - state->stream.lz4_stream, - state->stream.input_ring_buffer + state->stream.input_ring_buffer_pos, - state->compression_result_buffer + RRDPUSH_COMPRESSION_SIGNATURE_SIZE, - (int)size, - (int)max_dst_size, - 1); + 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 - if (compressed_data_size < 0) { - netdata_log_error("Data compression error: %ld", compressed_data_size); +#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; } - // update the next writing position of the ring buffer - state->stream.input_ring_buffer_pos += size; - if(unlikely(state->stream.input_ring_buffer_pos >= state->stream.input_ring_buffer_size - COMPRESSION_MAX_MSG_SIZE)) - state->stream.input_ring_buffer_pos = 0; + return ret; +} - // update the signature header - uint32_t len = ((compressed_data_size & 0x7f) | 0x80 | (((compressed_data_size & (0x7f << 7)) << 1) | 0x8000)) << 8; - *(uint32_t *)state->compression_result_buffer = len | RRDPUSH_COMPRESSION_SIGNATURE; - *out = state->compression_result_buffer; - netdata_log_debug(D_STREAM, "%s: Compressed data header: %ld", STREAM_COMPRESSION_MSG, compressed_data_size); - return compressed_data_size + RRDPUSH_COMPRESSION_SIGNATURE_SIZE; +// ---------------------------------------------------------------------------- +// 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; } -/* - * Decompress the compressed data in the internal buffer - * Return the size of uncompressed data or 0 for error - */ -size_t rrdpush_decompress(struct decompressor_state *state, const char *compressed_data, size_t compressed_size) { - if (unlikely(!state || !compressed_data || !compressed_size)) - return 0; +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 - if(unlikely(state->stream.read_at != state->stream.write_at)) - fatal("RRDPUSH_DECOMPRESS: asked to decompress new data, while there are unread data in the decompression buffer!"); +#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 - if (unlikely(state->stream.write_at >= state->stream.size / 2)) { - state->stream.write_at = 0; - state->stream.read_at = 0; + default: + case COMPRESSION_ALGORITHM_GZIP: + rrdpush_decompressor_init_gzip(state); + break; } - long int decompressed_size = LZ4_decompress_safe_continue( - state->stream.lz4_stream - , compressed_data - , state->stream.buffer + state->stream.write_at - , (int)compressed_size - , (int)(state->stream.size - state->stream.write_at) - ); + 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; + } - if (unlikely(decompressed_size < 0)) { - netdata_log_error("RRDPUSH DECOMPRESS: decompressor returned negative decompressed bytes: %ld", decompressed_size); + // 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; } - if(unlikely(decompressed_size + state->stream.write_at > state->stream.size)) - fatal("RRDPUSH DECOMPRESS: decompressor overflown the stream_buffer. size: %zu, pos: %zu, added: %ld, " - "exceeding the buffer by %zu" - , state->stream.size - , state->stream.write_at - , decompressed_size - , (size_t)(state->stream.write_at + decompressed_size - state->stream.size) - ); + return ret; +} + +// ---------------------------------------------------------------------------- +// unit test + +static inline long int my_random (void) { + return random(); +} + +void unittest_generate_random_name(char *dst, size_t size) { + if(size < 7) + size = 7; - state->stream.write_at += decompressed_size; + size_t len = 5 + my_random() % (size - 6); - // statistics - state->total_compressed += compressed_size + RRDPUSH_COMPRESSION_SIGNATURE_SIZE; - state->total_uncompressed += decompressed_size; - state->packet_count++; + for(size_t i = 0; i < len ; i++) { + if(my_random() % 2 == 0) + dst[i] = 'A' + my_random() % 26; + else + dst[i] = 'a' + my_random() % 26; + } - return decompressed_size; + dst[len] = '\0'; } -void rrdpush_decompressor_reset(struct decompressor_state *state) { - if(!state->initialized) { - state->initialized = true; - state->stream.lz4_stream = LZ4_createStreamDecode(); - state->stream.size = LZ4_decoderRingBufferSize(COMPRESSION_MAX_MSG_SIZE) * 2; - state->stream.buffer = mallocz(state->stream.size); +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 + my_random() % 5; + char chart[RRD_ID_LENGTH_MAX + 1] = "name"; + unittest_generate_random_name(chart, 5 + my_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); } - LZ4_setStreamDecode(state->stream.lz4_stream, NULL, 0); + 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 + my_random() % 20); + int64_t last_collected_value = (my_random() % 2 == 0) ? (int64_t)(counter + d) : (int64_t)my_random(); + NETDATA_DOUBLE value = (my_random() % 2 == 0) ? (NETDATA_DOUBLE)my_random() / ((NETDATA_DOUBLE)my_random() + 1) : (NETDATA_DOUBLE)last_collected_value; + SN_FLAGS flags = (my_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); + } - state->signature_size = RRDPUSH_COMPRESSION_SIGNATURE_SIZE; - state->stream.write_at = 0; - state->stream.read_at = 0; + buffer_fast_strcat(wb, PLUGINSD_KEYWORD_END_V2 "\n", sizeof(PLUGINSD_KEYWORD_END_V2) - 1 + 1); } -void rrdpush_decompressor_destroy(struct decompressor_state *state) { - if(unlikely(!state->initialized)) - return; +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); - if (state->stream.lz4_stream) { - LZ4_freeStreamDecode(state->stream.lz4_stream); - state->stream.lz4_stream = NULL; + 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); } - freez(state->stream.buffer); - state->stream.buffer = NULL; +cleanup: + rrdpush_compressor_destroy(&cctx); + rrdpush_decompressor_destroy(&dctx); - state->initialized = false; + if(errors) + fprintf(stderr, "Compression with %s: FAILED (%d errors)\n", name, errors); + else + fprintf(stderr, "Compression with %s: OK\n", name); + + return errors; } -#endif +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; +} |