@cryptotaxi247 / netdata-1
Code Commits Issues Pulls Sessions CI/CD
master
master
c 342 lines 15 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "stream-thread.h"
4 #include "stream-replication-receiver.h"
5 #include "stream-replication-sender.h"
6
7 struct inflight_stream_function {
8 struct sender_state *sender;
9 STRING *transaction;
10 usec_t received_ut;
11 };
12
13 static void stream_execute_function_callback(BUFFER *func_wb, int code, void *data) {
14 struct inflight_stream_function *tmp = data;
15 struct sender_state *s = tmp->sender;
16
17 if(rrdhost_can_stream_metadata_to_parent(s->host)) {
18 // for functions we use a new buffer, to avoid keeping a big buffer in memory
19 CLEAN_BUFFER *wb = buffer_create(0, NULL);
20
21 pluginsd_function_result_begin_to_buffer(
22 wb, string2str(tmp->transaction), code,
23 content_type_id2string(func_wb->content_type), func_wb->expires);
24
25 buffer_fast_strcat(wb, buffer_tostring(func_wb), buffer_strlen(func_wb));
26 pluginsd_function_result_end_to_buffer(wb);
27
28 sender_commit_clean_buffer(s, wb, STREAM_TRAFFIC_TYPE_FUNCTIONS);
29
30 internal_error(true, "STREAM SND '%s' [to %s]: FUNCTION transaction %s sending back response (%zu bytes, %"PRIu64" usec).",
31 rrdhost_hostname(s->host), s->remote_ip,
32 string2str(tmp->transaction),
33 buffer_strlen(func_wb),
34 now_realtime_usec() - tmp->received_ut);
35 }
36
37 string_freez(tmp->transaction);
38 buffer_free(func_wb);
39 freez(tmp);
40 }
41
42 static void stream_execute_function_progress_callback(nd_uuid_t *transaction, void *data, size_t done, size_t all) {
43 struct inflight_stream_function *tmp = data;
44 struct sender_state *s = tmp->sender;
45
46 if(rrdhost_can_stream_metadata_to_parent(s->host)) {
47 CLEAN_BUFFER *wb = buffer_create(0, NULL);
48
49 char transaction_str[UUID_COMPACT_STR_LEN];
50 uuid_unparse_lower_compact(*transaction, transaction_str);
51
52 buffer_sprintf(wb, PLUGINSD_KEYWORD_FUNCTION_PROGRESS " '%s' %zu %zu\n",
53 transaction_str, done, all);
54
55 sender_commit_clean_buffer(s, wb, STREAM_TRAFFIC_TYPE_FUNCTIONS);
56 }
57 }
58
59 static void execute_commands_function(struct sender_state *s, const char *command, const char *transaction, const char *timeout_s, const char *function, BUFFER *payload, const char *access, const char *source) {
60 worker_is_busy(WORKER_SENDER_JOB_EXECUTE_FUNCTION);
61 nd_log(NDLS_ACCESS, NDLP_INFO, NULL);
62
63 if(!transaction || !*transaction || !timeout_s || !*timeout_s || !function || !*function) {
64 netdata_log_error("STREAM SND '%s' [to %s]: %s execution command is incomplete (transaction = '%s', timeout = '%s', function = '%s'). Ignoring it.",
65 rrdhost_hostname(s->host), s->remote_ip,
66 command,
67 transaction?transaction:"(unset)",
68 timeout_s?timeout_s:"(unset)",
69 function?function:"(unset)");
70 }
71 else {
72 int timeout = str2i(timeout_s);
73 if(timeout <= 0) timeout = PLUGINS_FUNCTIONS_TIMEOUT_DEFAULT;
74
75 struct inflight_stream_function *tmp = callocz(1, sizeof(struct inflight_stream_function));
76 tmp->received_ut = now_realtime_usec();
77 tmp->sender = s;
78 tmp->transaction = string_strdupz(transaction);
79 BUFFER *wb = buffer_create(1024, &netdata_buffers_statistics.buffers_functions);
80
81 rrd_function_run(s->host, wb, timeout,
82 http_access_from_hex_mapping_old_roles(access), function, false, transaction,
83 stream_execute_function_callback, tmp,
84 stream_has_capability(s, STREAM_CAP_PROGRESS) ? stream_execute_function_progress_callback : NULL,
85 stream_has_capability(s, STREAM_CAP_PROGRESS) ? tmp : NULL,
86 NULL, NULL, payload, source, true);
87 }
88 }
89
90 struct deferred_function {
91 const char *transaction;
92 const char *timeout_s;
93 const char *function;
94 const char *access;
95 const char *source;
96 };
97
98 static void execute_deferred_function(struct sender_state *s, void *data) {
99 struct deferred_function *dfd = data;
100 execute_commands_function(s, s->thread.defer.end_keyword,
101 dfd->transaction, dfd->timeout_s,
102 dfd->function, s->thread.defer.payload,
103 dfd->access, dfd->source);
104 }
105
106 static void execute_deferred_json(struct sender_state *s, void *data) {
107 const char *keyword = data;
108
109 if(strcmp(keyword, PLUGINSD_KEYWORD_JSON_CMD_STREAM_PATH) == 0)
110 stream_path_set_from_json(s->host, buffer_tostring(s->thread.defer.payload), true);
111 else
112 nd_log(NDLS_DAEMON, NDLP_ERR,
113 "STREAM SND '%s' [to %s]: unknown JSON keyword '%s' with payload: %s",
114 rrdhost_hostname(s->host), s->remote_ip,
115 keyword, buffer_tostring(s->thread.defer.payload));
116 }
117
118 static void cleanup_deferred_json(struct sender_state *s __maybe_unused, void *data) {
119 const char *keyword = data;
120 freez((void *)keyword);
121 }
122
123 static void cleanup_deferred_function(struct sender_state *s __maybe_unused, void *data) {
124 struct deferred_function *dfd = data;
125 freez((void *)dfd->transaction);
126 freez((void *)dfd->timeout_s);
127 freez((void *)dfd->function);
128 freez((void *)dfd->access);
129 freez((void *)dfd->source);
130 freez(dfd);
131 }
132
133 static void cleanup_deferred_data(struct sender_state *s) {
134 if(s->thread.defer.cleanup)
135 s->thread.defer.cleanup(s, s->thread.defer.action_data);
136
137 buffer_free(s->thread.defer.payload);
138 s->thread.defer.payload = NULL;
139 s->thread.defer.end_keyword = NULL;
140 s->thread.defer.action = NULL;
141 s->thread.defer.cleanup = NULL;
142 s->thread.defer.action_data = NULL;
143 }
144
145 void stream_sender_execute_commands_cleanup(struct sender_state *s) {
146 cleanup_deferred_data(s);
147 }
148
149 // This is just a placeholder until the gap filling state machine is inserted
150 void stream_sender_execute_commands(struct sender_state *s) {
151 ND_LOG_STACK lgs[] = {
152 ND_LOG_FIELD_CB(NDF_REQUEST, line_splitter_reconstruct_line, &s->thread.rbuf.line),
153 ND_LOG_FIELD_END(),
154 };
155 ND_LOG_STACK_PUSH(lgs);
156
157 #ifdef NETDATA_LOG_STREAM_SENDER
158 if(!s->log.received)
159 s->log.received = buffer_create(0, NULL);
160 #endif
161
162 char *start = s->thread.rbuf.b, *end = &s->thread.rbuf.b[s->thread.rbuf.read_len], *newline;
163 *end = '\0';
164 for( ; start < end ; start = newline + 1) {
165 newline = strchr(start, '\n');
166
167 if(!newline) {
168 if(s->thread.defer.end_keyword) {
169 buffer_strcat(s->thread.defer.payload, start);
170 start = end;
171 }
172 break;
173 }
174
175 *newline = '\0';
176 s->thread.rbuf.line.count++;
177
178 if(s->thread.defer.end_keyword) {
179 if(strcmp(start, s->thread.defer.end_keyword) == 0) {
180 #ifdef NETDATA_LOG_STREAM_SENDER
181 buffer_strcat(s->log.received, buffer_tostring(s->thread.defer.payload));
182 buffer_strcat(s->log.received, "\n");
183 buffer_strcat(s->log.received, s->thread.defer.end_keyword);
184 buffer_strcat(s->log.received, "\n");
185 stream_sender_log_payload(s, s->log.received, STREAM_TRAFFIC_TYPE_METADATA, true);
186 #endif
187 s->thread.defer.action(s, s->thread.defer.action_data);
188 cleanup_deferred_data(s);
189 }
190 else {
191 buffer_strcat(s->thread.defer.payload, start);
192 buffer_putc(s->thread.defer.payload, '\n');
193 }
194
195 continue;
196 }
197
198 #ifdef NETDATA_LOG_STREAM_SENDER
199 buffer_reset(s->log.received);
200 buffer_strcat(s->log.received, start);
201 buffer_strcat(s->log.received, "\n");
202 #endif
203
204 s->thread.rbuf.line.num_words = quoted_strings_splitter_whitespace(start, s->thread.rbuf.line.words, PLUGINSD_MAX_WORDS);
205 const char *command = get_word(s->thread.rbuf.line.words, s->thread.rbuf.line.num_words, 0);
206
207 if(command && strcmp(command, PLUGINSD_CALL_FUNCTION) == 0) {
208 #ifdef NETDATA_LOG_STREAM_SENDER
209 stream_sender_log_payload(s, s->log.received, STREAM_TRAFFIC_TYPE_FUNCTIONS, true);
210 #endif
211 char *transaction = get_word(s->thread.rbuf.line.words, s->thread.rbuf.line.num_words, 1);
212 char *timeout_s = get_word(s->thread.rbuf.line.words, s->thread.rbuf.line.num_words, 2);
213 char *function = get_word(s->thread.rbuf.line.words, s->thread.rbuf.line.num_words, 3);
214 char *access = get_word(s->thread.rbuf.line.words, s->thread.rbuf.line.num_words, 4);
215 char *source = get_word(s->thread.rbuf.line.words, s->thread.rbuf.line.num_words, 5);
216
217 execute_commands_function(s, command, transaction, timeout_s, function, NULL, access, source);
218 }
219 else if(command && strcmp(command, PLUGINSD_CALL_FUNCTION_PAYLOAD_BEGIN) == 0) {
220 char *transaction = get_word(s->thread.rbuf.line.words, s->thread.rbuf.line.num_words, 1);
221 char *timeout_s = get_word(s->thread.rbuf.line.words, s->thread.rbuf.line.num_words, 2);
222 char *function = get_word(s->thread.rbuf.line.words, s->thread.rbuf.line.num_words, 3);
223 char *access = get_word(s->thread.rbuf.line.words, s->thread.rbuf.line.num_words, 4);
224 char *source = get_word(s->thread.rbuf.line.words, s->thread.rbuf.line.num_words, 5);
225 char *content_type = get_word(s->thread.rbuf.line.words, s->thread.rbuf.line.num_words, 6);
226
227 s->thread.defer.end_keyword = PLUGINSD_CALL_FUNCTION_PAYLOAD_END;
228 s->thread.defer.payload = buffer_create(0, NULL);
229 s->thread.defer.payload->content_type = content_type_string2id(content_type);
230 s->thread.defer.action = execute_deferred_function;
231 s->thread.defer.cleanup = cleanup_deferred_function;
232
233 struct deferred_function *dfd = callocz(1, sizeof(*dfd));
234 dfd->transaction = strdupz(transaction ? transaction : "");
235 dfd->timeout_s = strdupz(timeout_s ? timeout_s : "");
236 dfd->function = strdupz(function ? function : "");
237 dfd->access = strdupz(access ? access : "");
238 dfd->source = strdupz(source ? source : "");
239
240 s->thread.defer.action_data = dfd;
241 }
242 else if(command && strcmp(command, PLUGINSD_CALL_FUNCTION_CANCEL) == 0) {
243 worker_is_busy(WORKER_SENDER_JOB_EXECUTE_FUNCTION);
244 #ifdef NETDATA_LOG_STREAM_SENDER
245 stream_sender_log_payload(s, s->log.received, STREAM_TRAFFIC_TYPE_FUNCTIONS, true);
246 #endif
247 nd_log(NDLS_ACCESS, NDLP_DEBUG, NULL);
248
249 char *transaction = get_word(s->thread.rbuf.line.words, s->thread.rbuf.line.num_words, 1);
250 if(transaction && *transaction)
251 rrd_function_cancel(transaction);
252 }
253 else if(command && strcmp(command, PLUGINSD_CALL_FUNCTION_PROGRESS) == 0) {
254 worker_is_busy(WORKER_SENDER_JOB_EXECUTE_FUNCTION);
255 #ifdef NETDATA_LOG_STREAM_SENDER
256 stream_sender_log_payload(s, s->log.received, STREAM_TRAFFIC_TYPE_FUNCTIONS, true);
257 #endif
258 nd_log(NDLS_ACCESS, NDLP_DEBUG, NULL);
259
260 char *transaction = get_word(s->thread.rbuf.line.words, s->thread.rbuf.line.num_words, 1);
261 if(transaction && *transaction)
262 rrd_function_progress(transaction);
263 }
264 else if (command && strcmp(command, PLUGINSD_KEYWORD_REPLAY_CHART) == 0) {
265 worker_is_busy(WORKER_SENDER_JOB_EXECUTE_REPLAY);
266 #ifdef NETDATA_LOG_STREAM_SENDER
267 stream_sender_log_payload(s, s->log.received, STREAM_TRAFFIC_TYPE_REPLICATION, true);
268 #endif
269
270 __atomic_add_fetch(&s->host->stream.snd.status.replication.counter_in, 1, __ATOMIC_RELAXED);
271
272 // do not log replication commands received - way too many!
273 // nd_log(NDLS_ACCESS, NDLP_DEBUG, NULL);
274
275 const char *chart_id = get_word(s->thread.rbuf.line.words, s->thread.rbuf.line.num_words, 1);
276 const char *start_streaming = get_word(s->thread.rbuf.line.words, s->thread.rbuf.line.num_words, 2);
277 const char *after = get_word(s->thread.rbuf.line.words, s->thread.rbuf.line.num_words, 3);
278 const char *before = get_word(s->thread.rbuf.line.words, s->thread.rbuf.line.num_words, 4);
279
280 if (!chart_id || !start_streaming || !after || !before) {
281 netdata_log_error("STREAM REPLAY ERROR '%s' [send to %s] %s command is incomplete"
282 " (chart=%s, start_streaming=%s, after=%s, before=%s)",
283 rrdhost_hostname(s->host), s->remote_ip,
284 command,
285 chart_id ? chart_id : "(unset)",
286 start_streaming ? start_streaming : "(unset)",
287 after ? after : "(unset)",
288 before ? before : "(unset)");
289 }
290 else {
291 #ifdef REPLICATION_TRACKING
292 RRDSET *st = rrdset_find(s->host, chart_id, true);
293 if(st)
294 st->stream.snd.who = REPLAY_WHO_ME;
295 #endif
296
297 replication_sender_request_add(
298 s, chart_id,
299 strtoll(after, NULL, 0),
300 strtoll(before, NULL, 0),
301 stream_parse_enable_streaming(start_streaming));
302 }
303 }
304 else if(command && strcmp(command, PLUGINSD_KEYWORD_NODE_ID) == 0) {
305 worker_is_busy(WORKER_SENDER_JOB_EXECUTE_META);
306 #ifdef NETDATA_LOG_STREAM_SENDER
307 stream_sender_log_payload(s, s->log.received, STREAM_TRAFFIC_TYPE_METADATA, true);
308 #endif
309 char *claim_id_str = get_word(s->thread.rbuf.line.words, s->thread.rbuf.line.num_words, 1);
310 char *node_id_str = get_word(s->thread.rbuf.line.words, s->thread.rbuf.line.num_words, 2);
311 char *url = get_word(s->thread.rbuf.line.words, s->thread.rbuf.line.num_words, 3);
312
313 stream_sender_get_node_and_claim_id_from_parent(s, claim_id_str, node_id_str, url);
314 }
315 else if(command && strcmp(command, PLUGINSD_KEYWORD_JSON) == 0) {
316 worker_is_busy(WORKER_SENDER_JOB_EXECUTE_META);
317
318 char *keyword = get_word(s->thread.rbuf.line.words, s->thread.rbuf.line.num_words, 1);
319
320 s->thread.defer.end_keyword = PLUGINSD_KEYWORD_JSON_END;
321 s->thread.defer.payload = buffer_create(0, NULL);
322 s->thread.defer.action = execute_deferred_json;
323 s->thread.defer.cleanup = cleanup_deferred_json;
324 s->thread.defer.action_data = strdupz(keyword);
325 }
326 else {
327 netdata_log_error("STREAM SND '%s' [to %s] received unknown command over connection: %s",
328 rrdhost_hostname(s->host), s->remote_ip, s->thread.rbuf.line.words[0]?s->thread.rbuf.line.words[0]:"(unset)");
329 }
330
331 line_splitter_reset(&s->thread.rbuf.line);
332 }
333
334 if (start < end) {
335 memmove(s->thread.rbuf.b, start, end-start);
336 s->thread.rbuf.read_len = end - start;
337 }
338 else {
339 s->thread.rbuf.b[0] = '\0';
340 s->thread.rbuf.read_len = 0;
341 }
342 }