@cryptotaxi247 / netdata-1 / commits / 9d8440cb4

Remove VLA (variable-length arrays) (Streaming and Web) (#22276)

thiagoftsm committed May 1, 2026 at 20:38 UTC 9d8440cb4f5a184971d81996d275253ae52578d4

14 files changed +223 -80

src/streaming/stream-parents.c

+11 -6

@@ -602,15 +602,16 @@ bool stream_parent_connect_to_one_unsafe(

602 return false;

603 }

604

605 - STREAM_PARENT *array[size];

605 + STREAM_PARENT **array = callocz(size, sizeof(*array));

606 usec_t now_ut = now_realtime_usec();

607 + bool rc = false;

608

609 // fetch stream info for all of them and put them in the array

610 size_t count = 0, skipped_but_useful = 0, skipped_not_useful = 0, potential = 0;

611 for (STREAM_PARENT *d = host->stream.snd.parents.all; d && count < size ; d = d->next) {

612 if (nd_thread_signaled_to_cancel()) {

613 sender_sock->error = ND_SOCK_ERR_THREAD_CANCELLED;

613 - return false;

614 + goto cleanup;

615 }

616

617 // make sure they all have a random number

@@ -740,7 +741,7 @@ bool stream_parent_connect_to_one_unsafe(

741 pulse_host_status(host, PULSE_HOST_STATUS_SND_NO_DST, 0);

742 }

743

743 - return false;

744 + goto cleanup;

745 }

746

747 // order the parents in the array the way we want to connect

@@ -831,7 +832,7 @@ bool stream_parent_connect_to_one_unsafe(

832 sender_sock->error = ND_SOCK_ERR_THREAD_CANCELLED;

833 host->stream.snd.status.reason = STREAM_HANDSHAKE_DISCONNECT_SIGNALED_TO_STOP;

834 pulse_host_status(host, PULSE_HOST_STATUS_SND_OFFLINE, host->stream.snd.status.reason);

834 - return false;

835 + goto cleanup;

836 }

837

838 nd_log(NDLS_DAEMON, NDLP_DEBUG,

@@ -873,7 +874,8 @@ bool stream_parent_connect_to_one_unsafe(

874 sender_sock->error = ND_SOCK_ERR_NONE;

875 host->stream.snd.status.reason = STREAM_HANDSHAKE_SP_CONNECTED;

876 pulse_host_status(host, PULSE_HOST_STATUS_SND_CONNECTING, host->stream.snd.status.reason);

876 - return true;

877 + rc = true;

878 + goto cleanup;

879 }

880 else {

881 stream_parent_nd_sock_error_to_reason(d, sender_sock);

@@ -889,7 +891,10 @@ bool stream_parent_connect_to_one_unsafe(

891 }

892

893 pulse_host_status(host, PULSE_HOST_STATUS_SND_OFFLINE, 0);

892 - return false;

894 +

895 +cleanup:

896 + freez(array);

897 + return rc;

898 }

899

900 bool stream_parent_connect_to_one(

src/web/api/functions/function-metrics-cardinality.c

+3 -2

@@ -69,8 +69,7 @@ int function_metrics_cardinality(BUFFER *wb, const char *function __maybe_unused

69 // Parse function parameters

70 bool by_node = false;

71 {

72 - char function_copy[strlen(function) + 1];

73 - memcpy(function_copy, function, sizeof(function_copy));

72 + char *function_copy = strdupz(function);

73 char *words[1024];

74 size_t num_words = quoted_strings_splitter_whitespace(function_copy, words, 1024);

75 for (size_t i = 1; i < num_words; i++) {

@@ -81,9 +80,11 @@ int function_metrics_cardinality(BUFFER *wb, const char *function __maybe_unused

80 by_node = false;

81 } else if (strcmp(param, "info") == 0) {

82 buffer_json_finalize(wb);

83 + freez(function_copy);

84 return HTTP_RESP_OK;

85 }

86 }

87 + freez(function_copy);

88 }

90 DICTIONARY *contexts_dict = dictionary_create(DICT_OPTION_SINGLE_THREADED|DICT_OPTION_DONT_OVERWRITE_VALUE);

src/web/api/maps/rrdr_options.c

+48 -8

@@ -74,17 +74,57 @@ RRDR_OPTIONS rrdr_options_parse_one(const char *o) {

74 return ret;

75 }

77 -RRDR_OPTIONS rrdr_options_parse(const char *options_str) {

78 - char src[strlen(options_str) + 1];

79 - strcatz(src, 0, options_str, sizeof(src));

80 - char *o = src;

77 +static inline bool rrdr_options_is_separator(char c) {

78 + return c == ',' || c == ' ' || c == '|';

79 +}

80 +

81 +static inline bool rrdr_option_token_matches(const char *token, size_t len, const char *name) {

82 + if(!name)

83 + return false;

84 +

85 + for(size_t i = 0; i < len; i++) {

86 + if(!name[i] || token[i] != name[i])

87 + return false;

88 + }

89 +

90 + return name[len] == '\0';

91 +}

92 +

93 +static RRDR_OPTIONS rrdr_options_parse_one_n(const char *o, size_t len) {

94 + RRDR_OPTIONS ret = 0;

95 +

96 + if(!o || !len) return ret;

97 +

98 + for(int i = 0; ; i++) {

99 + const char *name = rrdr_options[i].name;

100 + if(!name)

101 + break;

102

103 + if (rrdr_option_token_matches(o, len, name)) {

104 + ret |= rrdr_options[i].value;

105 + break;

106 + }

107 + }

108 +

109 + return ret;

110 +}

111 +

112 +RRDR_OPTIONS rrdr_options_parse(const char *options_str) {

113 RRDR_OPTIONS ret = 0;

83 - char *tok;

114

85 - while(o && *o && (tok = strsep_skip_consecutive_separators(&o, ", |"))) {

86 - if(!*tok) continue;

87 - ret |= rrdr_options_parse_one(tok);

115 + if(!options_str || !*options_str)

116 + return ret;

117 +

118 + const char *s = options_str;

119 + while(*s) {

120 + while(*s && rrdr_options_is_separator(*s))

121 + s++;

122 +

123 + const char *tok = s;

124 + while(*s && !rrdr_options_is_separator(*s))

125 + s++;

126 +

127 + ret |= rrdr_options_parse_one_n(tok, (size_t)(s - tok));

128 }

129

130 return ret;

src/web/api/queries/backfill.c

+4 -4

@@ -42,11 +42,11 @@ static struct {

42 bool backfill_request_add(RRDSET *st, backfill_callback_t cb, struct backfill_request_data *data) {

43 bool rc = false;

44 size_t dimensions = dictionary_entries(st->rrddim_root_index);

45 - if(!dimensions || dimensions > 200)

45 + if(!dimensions)

46 return rc;

48 size_t added = 0;

49 - struct backfill_dim_work *array[dimensions];

49 + struct backfill_dim_work **array = mallocz(dimensions * sizeof(*array));

51 if(backfill_globals.running) {

52 struct backfill_request *br = aral_callocz(backfill_globals.ar_br);

@@ -96,6 +96,7 @@ bool backfill_request_add(RRDSET *st, backfill_callback_t cb, struct backfill_re

96 }

97 }

99 + freez(array);

100 return rc;

101 }

102

@@ -225,7 +226,7 @@ void backfill_thread(void *ptr) {

226 size_t threads = netdata_conf_cpus() / 2;

227 if(threads < 2) threads = 2;

228 if(threads > 16) threads = 16;

228 - ND_THREAD *th[threads - 1];

229 + ND_THREAD *th[15];

230

231 for(size_t t = 0; t < threads - 1 ;t++) {

232 char tag[15];

@@ -258,4 +259,3 @@ void backfill_thread(void *ptr) {

259

260 static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;

261 }

261 -

src/web/api/queries/query-group-by.c

+33 -16

@@ -2,39 +2,57 @@

3 #include "query-internal.h"

5 -RRDR_GROUP_BY group_by_parse(const char *group_by_txt) {

6 - char src[strlen(group_by_txt) + 1];

7 - strcatz(src, 0, group_by_txt, sizeof(src));

8 - char *s = src;

5 +static inline bool group_by_is_separator(char c) {

6 + return c == ',' || c == '|' || c == ' ';

7 +}

8 +

9 +static inline bool group_by_token_matches(const char *token, size_t len, const char *name, size_t name_len) {

10 + if(!name)

11 + return false;

12 +

13 + return name_len == len && !strncmp(token, name, len);

14 +}

16 +RRDR_GROUP_BY group_by_parse(const char *group_by_txt) {

17 RRDR_GROUP_BY group_by = RRDR_GROUP_BY_NONE;

12 - while(s) {

13 - char *key = strsep_skip_consecutive_separators(&s, ",| ");

14 - if (!key || !*key) continue;

19 + if(!group_by_txt || !*group_by_txt)

20 + return group_by;

21 +

22 + const char *s = group_by_txt;

23 + while(*s) {

24 + while(*s && group_by_is_separator(*s))

25 + s++;

26 +

27 + const char *key = s;

28 + while(*s && !group_by_is_separator(*s))

29 + s++;

16 - if (strcmp(key, "selected") == 0)

31 + size_t len = (size_t)(s - key);

32 + if(!len) continue;

33 +

34 + if (group_by_token_matches(key, len, "selected", sizeof("selected") - 1))

35 group_by |= RRDR_GROUP_BY_SELECTED;

19 - if (strcmp(key, "dimension") == 0)

37 + if (group_by_token_matches(key, len, "dimension", sizeof("dimension") - 1))

38 group_by |= RRDR_GROUP_BY_DIMENSION;

22 - if (strcmp(key, "instance") == 0)

40 + if (group_by_token_matches(key, len, "instance", sizeof("instance") - 1))

41 group_by |= RRDR_GROUP_BY_INSTANCE;

25 - if (strcmp(key, "percentage-of-instance") == 0)

43 + if (group_by_token_matches(key, len, "percentage-of-instance", sizeof("percentage-of-instance") - 1))

44 group_by |= RRDR_GROUP_BY_PERCENTAGE_OF_INSTANCE;

28 - if (strcmp(key, "label") == 0)

46 + if (group_by_token_matches(key, len, "label", sizeof("label") - 1))

47 group_by |= RRDR_GROUP_BY_LABEL;

31 - if (strcmp(key, "node") == 0)

49 + if (group_by_token_matches(key, len, "node", sizeof("node") - 1))

50 group_by |= RRDR_GROUP_BY_NODE;

34 - if (strcmp(key, "context") == 0)

52 + if (group_by_token_matches(key, len, "context", sizeof("context") - 1))

53 group_by |= RRDR_GROUP_BY_CONTEXT;

37 - if (strcmp(key, "units") == 0)

55 + if (group_by_token_matches(key, len, "units", sizeof("units") - 1))

56 group_by |= RRDR_GROUP_BY_UNITS;

57 }

@@ -224,4 +242,3 @@ void rrd2rrdr_set_timestamps(RRDR *r) {

242 "QUERY: wrong last timestamp in the query, expected %ld, found %ld",

243 before_wanted, r->t[points_wanted - 1]);

244 }

227 -

src/web/api/queries/query-plan.c

+6 -3

@@ -123,9 +123,12 @@ static size_t rrddim_find_best_tier_for_timeframe(QUERY_TARGET *qt, time_t after

123 return 0;

124 }

125

126 - long weight[nd_profile.storage_tiers];

126 + long weight[RRD_STORAGE_TIERS];

127

128 - for(size_t tier = 0; tier < nd_profile.storage_tiers; tier++) {

128 + // cap at the compile-time maximum to guard the fixed-size weight[] array

129 + size_t tiers = MIN(nd_profile.storage_tiers, RRD_STORAGE_TIERS);

130 +

131 + for(size_t tier = 0; tier < tiers; tier++) {

132

133 time_t common_first_time_s = 0;

134 time_t common_last_time_s = 0;

@@ -162,7 +165,7 @@ static size_t rrddim_find_best_tier_for_timeframe(QUERY_TARGET *qt, time_t after

165 }

166

167 size_t best_tier = 0;

165 - for(size_t tier = 1; tier < nd_profile.storage_tiers; tier++) {

168 + for(size_t tier = 1; tier < tiers; tier++) {

169 if(weight[tier] >= weight[best_tier])

170 best_tier = tier;

171 }

src/web/api/queries/weights.c

+13 -6

@@ -1405,13 +1405,17 @@ static double ks_2samp(

1405 }

1406

1407 static double kstwo(

1408 + ONEWAYALLOC *owa,

1409 NETDATA_DOUBLE baseline[], int baseline_points,

1410 NETDATA_DOUBLE highlight[], int highlight_points,

1411 uint32_t base_shifts) {

1412

1413 + if(unlikely(baseline_points <= 1 || highlight_points <= 1))

1414 + return NAN;

1415 +

1416 // -1 in size, since the calculate_pairs_diffs() returns one less point

1413 - DIFFS_NUMBERS baseline_diffs[baseline_points - 1];

1414 - DIFFS_NUMBERS highlight_diffs[highlight_points - 1];

1417 + DIFFS_NUMBERS *baseline_diffs = onewayalloc_mallocz(owa, (size_t)(baseline_points - 1) * sizeof(*baseline_diffs));

1418 + DIFFS_NUMBERS *highlight_diffs = onewayalloc_mallocz(owa, (size_t)(highlight_points - 1) * sizeof(*highlight_diffs));

1419

1420 int base_size = (int)calculate_pairs_diff(baseline_diffs, baseline, baseline_points);

1421 int high_size = (int)calculate_pairs_diff(highlight_diffs, highlight, highlight_points);

@@ -1549,7 +1553,7 @@ static void rrdset_metric_correlations_ks2(

1553

1554 stats->binary_searches += 2 * (base_points - 1) + 2 * (high_points - 1);

1555

1552 - double prob = kstwo(baseline, (int)base_points, highlight, (int)high_points, shifts);

1556 + double prob = kstwo(owa, baseline, (int)base_points, highlight, (int)high_points, shifts);

1557 if(!isnan(prob) && !isinf(prob)) {

1558

1559 // these conditions should never happen, but still let's check

@@ -1805,7 +1809,7 @@ static size_t spread_results_evenly(DICTIONARY *results, WEIGHTS_STATS *stats) {

1809 stats->max_base_high_ratio = 1.0;

1810

1811 // create an array of the right size and copy all the values in it

1808 - NETDATA_DOUBLE slots[dimensions];

1812 + NETDATA_DOUBLE *slots = mallocz(dimensions * sizeof(*slots));

1813 dimensions = 0;

1814 dfe_start_read(results, t) {

1815 if(t->flags & RESULT_IS_PERCENTAGE_OF_TIME)

@@ -1815,7 +1819,10 @@ static size_t spread_results_evenly(DICTIONARY *results, WEIGHTS_STATS *stats) {

1819 }

1820 dfe_done(t);

1821

1818 - if(!dimensions) return 0; // Coverity fix

1822 + if(!dimensions) {

1823 + freez(slots);

1824 + return 0; // Coverity fix

1825 + }

1826

1827 // sort the array with the values of all dimensions

1828 qsort(slots, dimensions, sizeof(NETDATA_DOUBLE), compare_netdata_doubles);

@@ -1844,6 +1851,7 @@ static size_t spread_results_evenly(DICTIONARY *results, WEIGHTS_STATS *stats) {

1851 }

1852 dfe_done(t);

1853

1854 + freez(slots);

1855 return dimensions;

1856 }

1857

@@ -2691,4 +2699,3 @@ int mc_unittest(void) {

2699

2700 return errors;

2701 }

2694 -

src/web/api/v1/api_v1_config.c

+8 -4

@@ -40,18 +40,20 @@ int api_v1_config(RRDHOST *host, struct web_client *w, char *url __maybe_unused)

41 size_t len = strlen(action) + (id ? strlen(id) : 0) + strlen(path) + (add_name ? strlen(add_name) : 0) + 100;

43 - char cmd[len];

43 + char *cmd = mallocz(len);

44 if(strcmp(action, "tree") == 0)

45 - snprintfz(cmd, sizeof(cmd), PLUGINSD_FUNCTION_CONFIG " tree '%s' '%s'", path, id?id:"");

45 + snprintfz(cmd, len, PLUGINSD_FUNCTION_CONFIG " tree '%s' '%s'", path, id?id:"");

46 else {

47 DYNCFG_CMDS c = dyncfg_cmds2id(action);

48 if(!id || !*id || !dyncfg_is_valid_id(id)) {

49 rrd_call_function_error(w->response.data, "Invalid id", HTTP_RESP_BAD_REQUEST);

50 + freez(cmd);

51 return HTTP_RESP_BAD_REQUEST;

52 }

54 if(c == DYNCFG_CMD_NONE) {

55 rrd_call_function_error(w->response.data, "Invalid action", HTTP_RESP_BAD_REQUEST);

56 + freez(cmd);

57 return HTTP_RESP_BAD_REQUEST;

58 }

@@ -69,12 +71,13 @@ int api_v1_config(RRDHOST *host, struct web_client *w, char *url __maybe_unused)

72 if(!add_name || !*add_name || !dyncfg_is_valid_id(add_name)) {

73 rrd_call_function_error(w->response.data, "Invalid name", HTTP_RESP_BAD_REQUEST);

74 + freez(cmd);

75 return HTTP_RESP_BAD_REQUEST;

76 }

74 - snprintfz(cmd, sizeof(cmd), PLUGINSD_FUNCTION_CONFIG " %s %s %s", id, dyncfg_id2cmd_one(c), add_name);

77 + snprintfz(cmd, len, PLUGINSD_FUNCTION_CONFIG " %s %s %s", id, dyncfg_id2cmd_one(c), add_name);

78 }

79 else

77 - snprintfz(cmd, sizeof(cmd), PLUGINSD_FUNCTION_CONFIG " %s %s", id, dyncfg_id2cmd_one(c));

80 + snprintfz(cmd, len, PLUGINSD_FUNCTION_CONFIG " %s %s", id, dyncfg_id2cmd_one(c));

81 }

83 CLEAN_BUFFER *source = buffer_create(100, NULL);

@@ -88,5 +91,6 @@ int api_v1_config(RRDHOST *host, struct web_client *w, char *url __maybe_unused)

91 web_client_interrupt_callback, w,

92 w->payload, buffer_tostring(source), false);

94 + freez(cmd);

95 return code;

96 }

src/web/server/web_client.c

+6 -8

@@ -1070,15 +1070,13 @@ static inline int web_client_switch_host(RRDHOST *host, struct web_client *w, ch

1070 //no delim found

1071 return append_slash_to_url_and_redirect(w);

1072

1073 - size_t len = strlen(url) + 2;

1074 - char buf[len];

1075 - buf[0] = '/';

1076 - strcpy(&buf[1], url);

1077 - buf[len - 1] = '\0';

1078 -

1073 buffer_flush(w->url_path_decoded);

1080 - buffer_strcat(w->url_path_decoded, buf);

1081 - return func(host, w, buf);

1074 + buffer_strcat(w->url_path_decoded, "/");

1075 + buffer_strcat(w->url_path_decoded, url);

1076 + char *mutable_path = strdupz(buffer_tostring(w->url_path_decoded));

1077 + int rc = func(host, w, mutable_path);

1078 + freez(mutable_path);

1079 + return rc;

1080 }

1081 }

1082

src/web/websocket/websocket-handshake.c

@@ -23,6 +23,8 @@ void websocket_threads_init(void) {

23 websocket_threads[i].ndpl = NULL;

24 websocket_threads[i].cmd.pipe[PIPE_READ] = -1;

25 websocket_threads[i].cmd.pipe[PIPE_WRITE] = -1;

26 + websocket_threads[i].cmd.buffer = NULL;

27 + websocket_threads[i].cmd.buffer_size = 0;

28 }

29 }

src/web/websocket/websocket-internal.h

+4 -1

@@ -47,6 +47,7 @@ struct websocket_thread;

47 #define WS_MAX_INCOMING_FRAME_SIZE (20ULL * 1024 * 1024) // 20MB max incoming frame size (browsers have ~16MiB)

48 #define WS_MAX_OUTGOING_FRAME_SIZE (4ULL * 1024 * 1024) // 4MB max outgoing frame size for browser compatibility

49 #define WS_MAX_DECOMPRESSED_SIZE (200ULL * 1024 * 1024) // 200MB max inbound uncompressed message

50 +#define WS_DEBUG_DUMP_BYTES 32 // max payload bytes captured in debug hex dumps

52 // WebSocket timeout constants (in seconds)

53 #define WS_PERIODIC_PING_INTERVAL 60 // Send periodic ping every 60 seconds

@@ -170,6 +171,8 @@ typedef struct websocket_thread {

171

172 struct {

173 int pipe[2]; // Command pipe [0] = read, [1] = write

174 + char *buffer; // Reusable scratch buffer for command payloads

175 + size_t buffer_size;

176 } cmd;

177

178 } WEBSOCKET_THREAD;

@@ -259,4 +262,4 @@ int websocket_broadcast_message(const char *message, WEBSOCKET_OPCODE opcode);

262

263 bool websocket_protocol_parse_header_from_buffer(const char *buffer, size_t length,

264 WEBSOCKET_FRAME_HEADER *header);

262 -#endif // NETDATA_WEBSOCKET_INTERNAL_H

\ No newline at end of file

265 +#endif // NETDATA_WEBSOCKET_INTERNAL_H

src/web/websocket/websocket-send.c

+2 -2

@@ -376,9 +376,9 @@ int websocket_protocol_send_close(WS_CLIENT *wsc, WEBSOCKET_CLOSE_CODE code, con

376 reason_len = 123; // Truncate reason to fit

377 }

378

379 - // Use stack buffer for close frame payload (max 125 bytes per RFC 6455)

379 + // Control frames are capped at 125 bytes, so a fixed stack buffer is sufficient.

380 size_t payload_len = 2 + reason_len;

381 - char payload[payload_len];

381 + char payload[125];

382

383 // Set status code in network byte order (big-endian)

384 uint16_t code_value = (uint16_t)code;

src/web/websocket/websocket-thread.c

+80 -17

@@ -105,6 +105,29 @@ struct pipe_header {

105 };

106 };

107

108 +static ssize_t write_pipe_block(int fd, const void *buffer, size_t size) {

109 + const char *buf = buffer;

110 + ssize_t total_written = 0;

111 +

112 + while (total_written < (ssize_t) size) {

113 + ssize_t bytes = write(fd, buf + total_written, size - total_written);

114 +

115 + if (bytes < 0) {

116 + if (errno == EINTR)

117 + continue;

118 + if (errno == EAGAIN || errno == EWOULDBLOCK)

119 + return total_written;

120 + return -1;

121 + }

122 + else if (bytes == 0)

123 + return total_written;

124 +

125 + total_written += bytes;

126 + }

127 +

128 + return total_written;

129 +}

130 +

131 // Send command to a thread

132 bool websocket_thread_send_command(WEBSOCKET_THREAD *wth, uint8_t cmd, uint32_t id) {

133 if(!wth || wth->cmd.pipe[PIPE_WRITE] == -1) {

@@ -122,8 +145,8 @@ bool websocket_thread_send_command(WEBSOCKET_THREAD *wth, uint8_t cmd, uint32_t

145 spinlock_lock(&wth->spinlock);

146

147 // Write command header

125 - ssize_t bytes = write(wth->cmd.pipe[PIPE_WRITE], &header, sizeof(header));

126 - if(bytes != sizeof(header)) {

148 + ssize_t bytes = write_pipe_block(wth->cmd.pipe[PIPE_WRITE], &header, sizeof(header));

149 + if(bytes != (ssize_t)sizeof(header)) {

150 netdata_log_error("WEBSOCKET[%zu]: Failed to write command header to pipe", wth->id);

151 spinlock_unlock(&wth->spinlock);

152 return false;

@@ -136,12 +159,17 @@ bool websocket_thread_send_command(WEBSOCKET_THREAD *wth, uint8_t cmd, uint32_t

159 }

160

161 bool websocket_thread_send_broadcast(WEBSOCKET_THREAD *wth, WEBSOCKET_OPCODE opcode, const char *message) {

139 - if(!wth || wth->cmd.pipe[PIPE_WRITE] == -1) {

162 + if(!wth || !message || wth->cmd.pipe[PIPE_WRITE] == -1) {

163 netdata_log_error("WEBSOCKET[%zu]: Failed to send command - pipe is not initialized", wth ? wth->id : 0);

164 return false;

165 }

166

144 - uint32_t message_len = strlen(message);

167 + size_t message_len_sz = strlen(message);

168 + if(message_len_sz > UINT32_MAX || message_len_sz > WS_MAX_OUTGOING_FRAME_SIZE) {

169 + netdata_log_error("WEBSOCKET[%zu]: Broadcast message too large: %zu bytes", wth ? wth->id : 0, message_len_sz);

170 + return false;

171 + }

172 + uint32_t message_len = (uint32_t)message_len_sz;

173

174 // Prepare command

175 struct pipe_header header = {

@@ -153,24 +181,24 @@ bool websocket_thread_send_broadcast(WEBSOCKET_THREAD *wth, WEBSOCKET_OPCODE opc

181 spinlock_lock(&wth->spinlock);

182

183 // Write command header

156 - ssize_t bytes = write(wth->cmd.pipe[PIPE_WRITE], &header, sizeof(header.cmd));

157 - if(bytes != sizeof(header.cmd)) {

184 + ssize_t bytes = write_pipe_block(wth->cmd.pipe[PIPE_WRITE], &header, sizeof(header));

185 + if(bytes != (ssize_t)sizeof(header)) {

186 netdata_log_error("WEBSOCKET[%zu]: Failed to write command header to pipe", wth->id);

187 spinlock_unlock(&wth->spinlock);

188 return false;

189 }

190

191 // Write the opcode

164 - bytes = write(wth->cmd.pipe[PIPE_WRITE], &opcode, sizeof(opcode));

165 - if(bytes != sizeof(opcode)) {

192 + bytes = write_pipe_block(wth->cmd.pipe[PIPE_WRITE], &opcode, sizeof(opcode));

193 + if(bytes != (ssize_t)sizeof(opcode)) {

194 netdata_log_error("WEBSOCKET[%zu]: Failed to write broadcast opcode to pipe", wth->id);

195 spinlock_unlock(&wth->spinlock);

196 return false;

197 }

198

199 // Write the message

172 - bytes = write(wth->cmd.pipe[PIPE_WRITE], message, message_len);;

173 - if(bytes != message_len) {

200 + bytes = write_pipe_block(wth->cmd.pipe[PIPE_WRITE], message, message_len);

201 + if(bytes != (ssize_t)message_len) {

202 netdata_log_error("WEBSOCKET[%zu]: Failed to write broadcast message to pipe", wth->id);

203 spinlock_unlock(&wth->spinlock);

204 return false;

@@ -281,19 +309,50 @@ static void websocket_thread_process_commands(WEBSOCKET_THREAD *wth) {

309 continue;

310 }

311

312 + if(header.len < sizeof(opcode)) {

313 + netdata_log_error("WEBSOCKET[%zu]: Broadcast command header.len %u is too small", wth->id, header.len);

314 + continue;

315 + }

316 uint32_t message_len = header.len - sizeof(opcode);

285 - char message[message_len + 1];

286 - bytes = read_pipe_block(wth->cmd.pipe[PIPE_READ], message, message_len);

287 - if(bytes != message_len) {

288 - netdata_log_error("WEBSOCKET[%zu]: Failed to read broadcast message from pipe", wth->id);

317 + if(message_len > WS_MAX_OUTGOING_FRAME_SIZE) {

318 + netdata_log_error("WEBSOCKET[%zu]: Broadcast message too large: %u bytes", wth->id, message_len);

319 + // Drain the payload to keep the pipe synchronized for subsequent commands.

320 + char drain_buf[4096];

321 + uint32_t remaining = message_len;

322 + while(remaining > 0) {

323 + size_t chunk = (remaining < sizeof(drain_buf)) ? remaining : sizeof(drain_buf);

324 + ssize_t drained = read_pipe_block(wth->cmd.pipe[PIPE_READ], drain_buf, chunk);

325 + if(drained <= 0) {

326 + // Cannot complete drain: close both pipe ends so the next poll cycle does not

327 + // try to parse the leftover payload bytes as a new pipe_header, and so that

328 + // future write attempts fail the FD guard rather than hitting EPIPE.

329 + netdata_log_error("WEBSOCKET[%zu]: Failed to fully drain oversized broadcast payload, closing command pipe to avoid desynchronization", wth->id);

330 + if(wth->cmd.pipe[PIPE_READ] != -1) {

331 + close(wth->cmd.pipe[PIPE_READ]);

332 + wth->cmd.pipe[PIPE_READ] = -1;

333 + }

334 + if(wth->cmd.pipe[PIPE_WRITE] != -1) {

335 + close(wth->cmd.pipe[PIPE_WRITE]);

336 + wth->cmd.pipe[PIPE_WRITE] = -1;

337 + }

338 + return;

339 + }

340 + remaining -= (uint32_t)drained;

341 + }

342 continue;

343 }

344 + if(message_len + 1 > wth->cmd.buffer_size) {

345 + wth->cmd.buffer = reallocz(wth->cmd.buffer, message_len + 1);

346 + wth->cmd.buffer_size = message_len + 1;

347 + }

348

292 - // Ensure we have the complete message

293 - if(header.len != sizeof(WEBSOCKET_OPCODE) + message_len) {

294 - netdata_log_error("WEBSOCKET[%zu]: Broadcast command size mismatch", wth->id);

349 + char *message = wth->cmd.buffer;

350 + bytes = read_pipe_block(wth->cmd.pipe[PIPE_READ], message, message_len);

351 + if(bytes != message_len) {

352 + netdata_log_error("WEBSOCKET[%zu]: Failed to read broadcast message from pipe", wth->id);

353 continue;

354 }

355 + message[message_len] = '\0';

356

357 // Send to all clients in this thread

358 spinlock_lock(&wth->clients_spinlock);

@@ -568,6 +627,10 @@ void websocket_thread(void *ptr) {

627 wth->cmd.pipe[PIPE_WRITE] = -1;

628 }

629

630 + freez(wth->cmd.buffer);

631 + wth->cmd.buffer = NULL;

632 + wth->cmd.buffer_size = 0;

633 +

634 // Mark thread as not running

635 spinlock_lock(&wth->spinlock);

636 wth->running = false;

src/web/websocket/websocket-utils.c

+3 -3

@@ -75,10 +75,10 @@ void websocket_dump_debug(WS_CLIENT *wsc __maybe_unused, const char *payload __m

76 // If payload is provided and not empty, create and log a hex dump

77 if (payload && payload_length > 0) {

78 - size_t bytes_to_dump = (payload_length < 32) ? payload_length : 32;

78 + size_t bytes_to_dump = (payload_length < WS_DEBUG_DUMP_BYTES) ? payload_length : WS_DEBUG_DUMP_BYTES;

80 - char hex_dump[bytes_to_dump * 2 + 1];

81 - char ascii_dump[bytes_to_dump + 1];

80 + char hex_dump[(WS_DEBUG_DUMP_BYTES * 2) + 1];

81 + char ascii_dump[WS_DEBUG_DUMP_BYTES + 1];

83 // Payload check is redundant as we already have it in the outer if