@cryptotaxi247 / netdata-1 / commits / 57d6c179c

Stream compression - Deactivate compression at runtime in case of a compressor buffer overflow (#12037)

* [Stream compression] Downgrade stream version if compressor buffer overflows * [Stream compression] More user friendly compression messages * [Stream compression] Fix mutex starvation * [Stream compression] enable compression by default * Update streaming/README.md Co-authored-by: Tina Luedtke <kickoke@users.noreply.github.com>

odynik committed Mar 24, 2022 at 12:14 UTC 57d6c179c82a21faba20ff0d2995eee31d071935

5 files changed +80 -36

streaming/README.md

+20 -6

@@ -357,12 +357,26 @@ Note: The `stream-compression` status can be `"enabled" | "disabled" | "N/A"`.

357 A compressed data packet is determined and decompressed on the fly.

358

359 #### Limitations

360 - This limitation will be withdrawn asap and is work-in-progress.

361 -

362 -The current implementation of streaming data compression can support only a few number of dimensions in a chart with names that cannot exceed the size of 16384 bytes. In case you experience stream connection problems or gaps in the charts please disable stream compression in the `stream.conf` file. This limitation can be seen in the error.log file with the sequence of the following messages:

363 -```

364 -Compression error - data discarded

365 -Message size above limit:

360 +This limitation will be withdrawn asap and is work-in-progress.

361 +

362 +The current implementation of streaming data compression can support only a few number of dimensions in a chart with names that cannot exceed the size of 16384 bytes. In case your instance hit this limitation, the agent will deactivate compression during runtime to avoid stream corruption. This limitation can be seen in the error.log file with the sequence of the following messages:

363 +```

364 +netdata INFO : STREAM_SENDER[child01] : STREAM child01 [send to my.parent.IP]: connecting...

365 +netdata INFO : STREAM_SENDER[child01] : STREAM child01 [send to my.parent.IP]: initializing communication...

366 +netdata INFO : STREAM_SENDER[child01] : STREAM child01 [send to my.parent.IP]: waiting response from remote netdata...

367 +netdata INFO : STREAM_SENDER[child01] : STREAM_COMPRESSION: Compressor Reset

368 +netdata INFO : STREAM_SENDER[child01] : STREAM child01 [send to my.parent.IP]: established communication with a parent using protocol version 5 - ready to send metrics...

369 +...

370 +netdata ERROR : PLUGINSD[go.d] : STREAM_COMPRESSION: Compression Failed - Message size 27847 above compression buffer limit: 16384 (errno 9, Bad file descriptor)

371 +netdata ERROR : PLUGINSD[go.d] : STREAM_COMPRESSION: Deactivating compression to avoid stream corruption

372 +netdata ERROR : PLUGINSD[go.d] : STREAM_COMPRESSION child01 [send to my.parent.IP]: Restarting connection without compression

373 +...

374 +netdata INFO : STREAM_SENDER[child01] : STREAM child01 [send to my.parent.IP]: connecting...

375 +netdata INFO : STREAM_SENDER[child01] : STREAM child01 [send to my.parent.IP]: initializing communication...

376 +netdata INFO : STREAM_SENDER[child01] : STREAM child01 [send to my.parent.IP]: waiting response from remote netdata...

377 +netdata INFO : STREAM_SENDER[child01] : Stream is uncompressed! One of the agents (my.parent.IP <-> child01) does not support compression OR compression is disabled.

378 +netdata INFO : STREAM_SENDER[child01] : STREAM child01 [send to my.parent.IP]: established communication with a parent using protocol version 4 - ready to send metrics...

379 +netdata INFO : WEB_SERVER[static4] : STREAM child01 [send]: sending metrics...

380 ```

381

382 #### How to enable stream compression

streaming/compression.c

+16 -14

@@ -3,6 +3,8 @@

3 #ifdef ENABLE_COMPRESSION

4 #include "lz4.h"

6 +#define STREAM_COMPRESSION_MSG "STREAM_COMPRESSION"

7 +

8 #define LZ4_MAX_MSG_SIZE 0x4000

9 #define LZ4_STREAM_BUFFER_SIZE (0x10000 + LZ4_MAX_MSG_SIZE)

@@ -29,7 +31,7 @@ static void lz4_compressor_reset(struct compressor_state *state)

31 if (state->data) {

32 if (state->data->stream) {

33 LZ4_resetStream_fast(state->data->stream);

32 - info("STREAM_COMPRESSION: Compressor resets stream fast!");

34 + info("%s: Compressor Reset", STREAM_COMPRESSION_MSG);

35 }

36 state->data->stream_buffer_pos = 0;

37 }

@@ -50,14 +52,14 @@ static void lz4_compressor_destroy(struct compressor_state **state)

52 freez(s->buffer);

53 freez(s);

54 *state = NULL;

53 - debug(D_STREAM, "STREAM_COMPRESSION: Compressor destroyed!");

55 + debug(D_STREAM, "%s: Compressor Destroyed.", STREAM_COMPRESSION_MSG);

56 }

57 }

59 /*

60 * Compress the given block of data

59 - * Comprecced data will remain in the internal buffer until the next invocation

60 - * Return the size of compressed data block as result and the pointer to internal buffer using the last argument

61 + * Compressed data will remain in the internal buffer until the next invocation

62 + * Return the size of compressed data block as result and the pointer to internal buffer using the last argument

63 * or 0 in case of error

64 */

65 static size_t lz4_compressor_compress(struct compressor_state *state, const char *data, size_t size, char **out)

@@ -65,7 +67,7 @@ static size_t lz4_compressor_compress(struct compressor_state *state, const char

67 if (!state || !size || !out)

68 return 0;

69 if (size > LZ4_MAX_MSG_SIZE) {

68 - error("Message size above limit: %lu", size);

70 + error("%s: Compression Failed - Message size %lu above compression buffer limit: %d", STREAM_COMPRESSION_MSG, size, LZ4_MAX_MSG_SIZE);

71 return 0;

72 }

73 size_t max_dst_size = LZ4_COMPRESSBOUND(size);

@@ -84,7 +86,7 @@ static size_t lz4_compressor_compress(struct compressor_state *state, const char

86 state->data->stream_buffer + state->data->stream_buffer_pos,

87 state->buffer + SIGNATURE_SIZE, size, max_dst_size, 1);

88 if (compressed_data_size < 0) {

87 - error("Date compression error: %ld", compressed_data_size);

89 + error("Data compression error: %ld", compressed_data_size);

90 return 0;

91 }

92 state->data->stream_buffer_pos += size;

@@ -93,7 +95,7 @@ static size_t lz4_compressor_compress(struct compressor_state *state, const char

95 uint32_t len = ((compressed_data_size & 0x7f) | 0x80 | (((compressed_data_size & (0x7f << 7)) << 1) | 0x8000)) << 8;

96 *(uint32_t *)state->buffer = len | SIGNATURE;

97 *out = state->buffer;

96 - debug(D_STREAM, "STREAM: Compressed data header: %ld", compressed_data_size);

98 + debug(D_STREAM, "%s: Compressed data header: %ld", STREAM_COMPRESSION_MSG, compressed_data_size);

99 return compressed_data_size + SIGNATURE_SIZE;

100 }

101

@@ -114,7 +116,7 @@ struct compressor_state *create_compressor()

116 state->data->stream_buffer = callocz(1, LZ4_DECODER_RING_BUFFER_SIZE(LZ4_MAX_MSG_SIZE));

117 state->buffer_size = LZ4_STREAM_BUFFER_SIZE;

118 state->reset(state);

117 - debug(D_STREAM, "STREAM_COMPRESSION: Initialize streaming compression!");

119 + debug(D_STREAM, "%s: Initialize streaming compression!", STREAM_COMPRESSION_MSG);

120 return state;

121 }

122

@@ -150,7 +152,7 @@ static void lz4_decompressor_destroy(struct decompressor_state **state)

152 if (state && *state) {

153 struct decompressor_state *s = *state;

154 if (s->data) {

153 - debug(D_STREAM, "STREAM_COMPRESSION: Destroying decompressor.");

155 + debug(D_STREAM, "%s: Destroying decompressor.", STREAM_COMPRESSION_MSG);

156 if (s->data->stream)

157 LZ4_freeStreamDecode(s->data->stream);

158 freez(s->data->stream_buffer);

@@ -246,7 +248,7 @@ static size_t lz4_decompressor_decompress(struct decompressor_state *state)

248 if (!state)

249 return 0;

250 if (!state->buffer) {

249 - error("STREAM: No decompressor buffer allocated");

251 + error("%s: No decompressor buffer allocated", STREAM_COMPRESSION_MSG);

252 return 0;

253 }

254

@@ -254,7 +256,7 @@ static size_t lz4_decompressor_decompress(struct decompressor_state *state)

256 state->data->stream_buffer + state->data->stream_buffer_pos,

257 state->buffer_len, state->data->stream_buffer_size - state->data->stream_buffer_pos);

258 if (decompressed_size < 0) {

257 - error("STREAM: Decompressor error %ld", decompressed_size);

259 + error("%s: Decompressor error %ld", STREAM_COMPRESSION_MSG, decompressed_size);

260 return 0;

261 }

262

@@ -278,7 +280,7 @@ static size_t lz4_decompressor_decompress(struct decompressor_state *state)

280 size_t avg_size = state->total_uncompressed / state->packet_count;

281

282 if (old_avg_saving != avg_saving || old_avg_size != avg_size){

281 - debug(D_STREAM, "STREAM: Saving: %lu%% (avg. %lu%%), avg.size: %lu", saving, avg_saving, avg_size);

283 + debug(D_STREAM, "%s: Saving: %lu%% (avg. %lu%%), avg.size: %lu", STREAM_COMPRESSION_MSG, saving, avg_saving, avg_size);

284 }

285 return decompressed_size;

286 }

@@ -301,7 +303,7 @@ static size_t lz4_decompressor_get(struct decompressor_state *state, char *data,

303 if (!state || !size || !data)

304 return 0;

305 if (!state->out_buffer)

304 - fatal("STREAM: No decompressor output buffer allocated");

306 + fatal("%s: No decompressor output buffer allocated", STREAM_COMPRESSION_MSG);

307 if (state->out_buffer_pos + size > state->out_buffer_len)

308 size = state->out_buffer_len - state->out_buffer_pos;

309

@@ -339,7 +341,7 @@ struct decompressor_state *create_decompressor()

341 state->data->stream_buffer = mallocz(state->data->stream_buffer_size);

342 fatal_assert(state->data->stream_buffer);

343 state->reset(state);

342 - debug(D_STREAM, "STREAM_COMPRESSION: Initialize streaming decompression!");

344 + debug(D_STREAM, "%s: Initialize streaming decompression!", STREAM_COMPRESSION_MSG);

345 return state;

346 }

347 #endif

streaming/rrdpush.c

+1 -1

@@ -41,7 +41,7 @@ struct config stream_config = {

42 unsigned int default_rrdpush_enabled = 0;

43 #ifdef ENABLE_COMPRESSION

44 -unsigned int default_compression_enabled = 0;

44 +unsigned int default_compression_enabled = 1;

45 #endif

46 char *default_rrdpush_destination = NULL;

47 char *default_rrdpush_api_key = NULL;

streaming/sender.c

+42 -14

@@ -13,24 +13,43 @@ void sender_start(struct sender_state *s) {

13 buffer_flush(s->build);

14 }

16 +static inline void rrdpush_sender_thread_close_socket(RRDHOST *host);

17 +

18 +#ifdef ENABLE_COMPRESSION

19 +/*

20 +* In case of stream compression buffer oveflow

21 +* Inform the user through the error log file and

22 +* deactivate compression by downgrading the stream protocol.

23 +*/

24 +static inline void deactivate_compression(struct sender_state *s)

25 +{

26 + error("STREAM_COMPRESSION: Deactivating compression to avoid stream corruption");

27 + default_compression_enabled = 0;

28 + s->rrdpush_compression = 0;

29 + s->version = STREAM_VERSION_CLABELS;

30 + error("STREAM_COMPRESSION %s [send to %s]: Restarting connection without compression", s->host->hostname, s->connected_to);

31 + rrdpush_sender_thread_close_socket(s->host);

32 +}

33 +#endif

34 +

35 // Collector thread finishing a transmission

36 void sender_commit(struct sender_state *s) {

37 char *src = (char *)buffer_tostring(s->host->sender->build);

38 size_t src_len = s->host->sender->build->len;

39 #ifdef ENABLE_COMPRESSION

21 - do {

22 - if (src && src_len) {

23 - if (s->compressor && s->rrdpush_compression) {

24 - src_len = s->compressor->compress(s->compressor, src, src_len, &src);

25 - if (!src_len) {

26 - error("Compression error - data discarded");

27 - break;

28 - }

40 + if (src && src_len) {

41 + if (s->compressor && s->rrdpush_compression) {

42 + src_len = s->compressor->compress(s->compressor, src, src_len, &src);

43 + if (!src_len) {

44 + deactivate_compression(s);

45 + buffer_flush(s->build);

46 + netdata_mutex_unlock(&s->mutex);

47 + return;

48 }

30 - if(cbuffer_add_unsafe(s->host->sender->buffer, src, src_len))

31 - s->overflow = 1;

49 }

33 - } while (0);

50 + if(cbuffer_add_unsafe(s->host->sender->buffer, src, src_len))

51 + s->overflow = 1;

52 + }

53 #else

54 if(cbuffer_add_unsafe(s->host->sender->buffer, src, src_len))

55 s->overflow = 1;

@@ -252,6 +271,13 @@ static int rrdpush_sender_thread_connect_to_parent(RRDHOST *host, int default_po

271 }

272 #endif

273

274 +#ifdef ENABLE_COMPRESSION

275 +// Negotiate stream VERSION_CLABELS if stream compression is not supported

276 +s->rrdpush_compression = (default_compression_enabled && (s->version >= STREAM_VERSION_COMPRESSION));

277 +if(!s->rrdpush_compression)

278 + s->version = STREAM_VERSION_CLABELS;

279 +#endif //ENABLE_COMPRESSION

280 +

281 /* TODO: During the implementation of #7265 switch the set of variables to HOST_* and CONTAINER_* if the

282 version negotiation resulted in a high enough version.

283 */

@@ -274,7 +300,7 @@ static int rrdpush_sender_thread_connect_to_parent(RRDHOST *host, int default_po

300 "&ml_capable=%d"

301 "&ml_enabled=%d"

302 "&tags=%s"

277 - "&ver=%u"

303 + "&ver=%d"

304 "&NETDATA_SYSTEM_OS_NAME=%s"

305 "&NETDATA_SYSTEM_OS_ID=%s"

306 "&NETDATA_SYSTEM_OS_ID_LIKE=%s"

@@ -316,7 +342,7 @@ static int rrdpush_sender_thread_connect_to_parent(RRDHOST *host, int default_po

342 , host->system_info->ml_capable

343 , host->system_info->ml_enabled

344 , (host->tags) ? host->tags : ""

319 - , STREAMING_PROTOCOL_CURRENT_VERSION

345 + , s->version

346 , se.os_name

347 , se.os_id

348 , (host->system_info->host_os_id_like) ? host->system_info->host_os_id_like : ""

@@ -410,7 +436,7 @@ static int rrdpush_sender_thread_connect_to_parent(RRDHOST *host, int default_po

436 s->version = version;

437

438 #ifdef ENABLE_COMPRESSION

413 - s->rrdpush_compression = (default_compression_enabled && (s->version >= STREAM_VERSION_COMPRESSION));

439 + s->rrdpush_compression = (s->rrdpush_compression && (s->version >= STREAM_VERSION_COMPRESSION));

440 if(s->rrdpush_compression)

441 {

442 // parent supports compression

@@ -420,6 +446,7 @@ static int rrdpush_sender_thread_connect_to_parent(RRDHOST *host, int default_po

446 else {

447 //parent does not support compression or has compression disabled

448 debug(D_STREAM, "Stream is uncompressed! One of the agents (%s <-> %s) does not support compression OR compression is disabled.", s->connected_to, s->host->hostname);

449 + infoerr("Stream is uncompressed! One of the agents (%s <-> %s) does not support compression OR compression is disabled.", s->connected_to, s->host->hostname);

450 s->version = STREAM_VERSION_CLABELS;

451 }

452 #endif //ENABLE_COMPRESSION

@@ -664,6 +691,7 @@ void *rrdpush_sender_thread(void *ptr) {

691 error("STREAM %s [send]: cannot create required pipe. DISABLING STREAMING THREAD", s->host->hostname);

692 return NULL;

693 }

694 + s->version = STREAMING_PROTOCOL_CURRENT_VERSION;

695

696 enum {

697 Collector,

web/api/web_api_v1.c

+1 -1

@@ -1051,7 +1051,7 @@ inline int web_client_api_request_v1_info_fill_buffer(RRDHOST *host, BUFFER *wb)

1051

1052 #ifdef ENABLE_COMPRESSION

1053 buffer_strcat(wb, "\t\"stream-compression\": ");

1054 - buffer_strcat(wb, (default_compression_enabled ? "\"enabled\"" : "\"disabled\""));

1054 + buffer_strcat(wb, (host->sender->rrdpush_compression ? "\"enabled\"" : "\"disabled\""));

1055 buffer_strcat(wb, ",\n");

1056 #else

1057 buffer_strcat(wb, "\t\"stream-compression\": \"N/A\",\n");