5
#include "stream-receiver-internals.h"
6
#include "web/server/h2o/http_server.h"
7
8
+#ifdef NETDATA_LOG_STREAM_RECEIVER
9
+void stream_receiver_log_payload(struct receiver_state *rpt, const char *payload, STREAM_TRAFFIC_TYPE type __maybe_unused, bool inbound) {
10
+ if (!rpt || type != STREAM_TRAFFIC_TYPE_REPLICATION) return; // not a streaming parser
11
+
12
+ spinlock_lock(&rpt->log.spinlock);
13
+
14
+ if (!rpt->log.fp) {
15
+ char filename[FILENAME_MAX + 1];
16
+ snprintfz(
17
+ filename, FILENAME_MAX, "/tmp/stream-receiver-%s.txt", rpt->host ? rrdhost_hostname(rpt->host) : "unknown");
18
+
19
+ rpt->log.fp = fopen(filename, "w");
20
+
21
+ // Align first_call to wall clock time
22
+ clock_gettime(CLOCK_REALTIME, &rpt->log.first_call);
23
+ rpt->log.first_call.tv_nsec = 0; // Align to the start of the second
24
+ }
25
+
26
+ if (rpt->log.fp) {
27
+ struct timespec now;
28
+ clock_gettime(CLOCK_REALTIME, &now);
29
+
30
+ time_t elapsed_sec = now.tv_sec - rpt->log.first_call.tv_sec;
31
+ long elapsed_nsec = now.tv_nsec - rpt->log.first_call.tv_nsec;
32
+
33
+ if (elapsed_nsec < 0) {
34
+ elapsed_sec--;
35
+ elapsed_nsec += 1000000000;
36
+ }
37
+
38
+ uint16_t days = elapsed_sec / 86400;
39
+ uint8_t hours = (elapsed_sec % 86400) / 3600;
40
+ uint8_t minutes = (elapsed_sec % 3600) / 60;
41
+ uint8_t seconds = elapsed_sec % 60;
42
+ uint16_t milliseconds = elapsed_nsec / 1000000;
43
+
44
+ char prefix[30];
45
+ snprintf(prefix, sizeof(prefix), "%03ud.%02u:%02u:%02u.%03u ",
46
+ days, hours, minutes, seconds, milliseconds);
47
+
48
+ const char *line_start = payload;
49
+ const char *line_end;
50
+
51
+ while (line_start && *line_start) {
52
+ line_end = strchr(line_start, '\n');
53
+ if (line_end) {
54
+ fprintf(rpt->log.fp, "%s%s%.*s\n", prefix, inbound ? "> " : "< ", (int)(line_end - line_start), line_start);
55
+ line_start = line_end + 1;
56
+ } else {
57
+ fprintf(rpt->log.fp, "%s%s%s\n", prefix, inbound ? "> " : "< ", line_start);
58
+ break;
59
+ }
60
+ }
61
+ }
62
+
63
+ fflush(rpt->log.fp);
64
+ spinlock_unlock(&rpt->log.spinlock);
65
+}
66
+#endif
67
+
68
static void stream_receiver_remove(struct stream_thread *sth, struct receiver_state *rpt, const char *why);
69
70
// When a child disconnects this is the maximum we will wait
157
// --------------------------------------------------------------------------------------------------------------------
158
159
static inline ssize_t receiver_read_uncompressed(struct receiver_state *r) {
100
- internal_fatal(r->reader.read_buffer[r->reader.read_len] != '\0',
160
+ internal_fatal(r->thread.uncompressed.read_buffer[r->thread.uncompressed.read_len] != '\0',
161
"%s: read_buffer does not start with zero #2", __FUNCTION__ );
162
103
- ssize_t bytes = read_stream(r, r->reader.read_buffer + r->reader.read_len, sizeof(r->reader.read_buffer) - r->reader.read_len - 1);
163
+ ssize_t bytes = read_stream(r, r->thread.uncompressed.read_buffer + r->thread.uncompressed.read_len, sizeof(r->thread.uncompressed.read_buffer) - r->thread.uncompressed.read_len - 1);
164
if(bytes > 0) {
165
worker_set_metric(WORKER_RECEIVER_JOB_BYTES_READ, (NETDATA_DOUBLE)bytes);
166
worker_set_metric(WORKER_RECEIVER_JOB_BYTES_UNCOMPRESSED, (NETDATA_DOUBLE)bytes);
167
108
- r->reader.read_len += bytes;
109
- r->reader.read_buffer[r->reader.read_len] = '\0';
168
+ r->thread.uncompressed.read_len += bytes;
169
+ r->thread.uncompressed.read_buffer[r->thread.uncompressed.read_len] = '\0';
170
}
171
172
return bytes;
209
if (unlikely(!compressed_message_size)) {
210
nd_log(NDLS_DAEMON, NDLP_ERR,
211
"STREAM RCV[x] '%s' [from [%s]:%s]: multiplexed uncompressed data in compressed stream!",
152
- rrdhost_hostname(r->host), r->client_ip, r->client_port);
212
+ rrdhost_hostname(r->host), r->remote_ip, r->remote_port);
213
return DECOMPRESS_FAILED;
214
}
215
218
"STREAM RCV[x] '%s' [from [%s]:%s]: received a compressed message of %zu bytes, "
219
"which is bigger than the max compressed message "
220
"size supported of %zu. Ignoring message.",
161
- rrdhost_hostname(r->host), r->client_ip, r->client_port,
221
+ rrdhost_hostname(r->host), r->remote_ip, r->remote_port,
222
compressed_message_size, (size_t)COMPRESSION_MAX_MSG_SIZE);
223
return DECOMPRESS_FAILED;
224
}
235
if (unlikely(!bytes_to_parse)) {
236
nd_log(NDLS_DAEMON, NDLP_ERR,
237
"STREAM RCV[x] '%s' [from [%s]:%s]: no bytes to decompress.",
178
- rrdhost_hostname(r->host), r->client_ip, r->client_port);
238
+ rrdhost_hostname(r->host), r->remote_ip, r->remote_port);
239
return DECOMPRESS_FAILED;
240
}
241
251
if (unlikely(!stream_decompressed_bytes_in_buffer(&r->thread.compressed.decompressor)))
252
return DECOMPRESS_NEED_MORE_DATA;
253
194
- size_t available = sizeof(r->reader.read_buffer) - r->reader.read_len - 1;
254
+ size_t available = sizeof(r->thread.uncompressed.read_buffer) - r->thread.uncompressed.read_len - 1;
255
if (likely(available)) {
256
size_t len = stream_decompressor_get(
197
- &r->thread.compressed.decompressor, r->reader.read_buffer + r->reader.read_len, available);
257
+ &r->thread.compressed.decompressor, r->thread.uncompressed.read_buffer + r->thread.uncompressed.read_len, available);
258
if (unlikely(!len)) {
259
internal_error(true, "decompressor returned zero length #1");
260
return DECOMPRESS_FAILED;
261
}
262
203
- r->reader.read_len += (int)len;
204
- r->reader.read_buffer[r->reader.read_len] = '\0';
263
+ r->thread.uncompressed.read_len += (int)len;
264
+ r->thread.uncompressed.read_buffer[r->thread.uncompressed.read_len] = '\0';
265
}
266
else {
207
- internal_fatal(true, "The line to read is too big! Already have %zd bytes in read_buffer.", r->reader.read_len);
267
+ internal_fatal(true, "The line to read is too big! Already have %zd bytes in read_buffer.", r->thread.uncompressed.read_len);
268
return DECOMPRESS_FAILED;
269
}
270
273
274
static inline ssize_t receiver_read_compressed(struct receiver_state *r) {
275
216
- internal_fatal(r->reader.read_buffer[r->reader.read_len] != '\0',
276
+ internal_fatal(r->thread.uncompressed.read_buffer[r->thread.uncompressed.read_len] != '\0',
277
"%s: read_buffer does not start with zero #2", __FUNCTION__ );
278
279
ssize_t bytes_read = read_stream(r, r->thread.compressed.buf + r->thread.compressed.used,
308
void stream_receiver_handle_op(struct stream_thread *sth, struct receiver_state *rpt, struct stream_opcode *msg) {
309
ND_LOG_STACK lgs[] = {
310
ND_LOG_FIELD_STR(NDF_NIDL_NODE, rpt->host->hostname),
251
- ND_LOG_FIELD_TXT(NDF_SRC_IP, rpt->client_ip),
252
- ND_LOG_FIELD_TXT(NDF_SRC_PORT, rpt->client_port),
311
+ ND_LOG_FIELD_TXT(NDF_SRC_IP, rpt->remote_ip),
312
+ ND_LOG_FIELD_TXT(NDF_SRC_PORT, rpt->remote_port),
313
ND_LOG_FIELD_CB(NDF_SRC_TRANSPORT, stream_receiver_log_transport, rpt),
314
ND_LOG_FIELD_CB(NDF_SRC_CAPABILITIES, stream_receiver_log_capabilities, rpt),
255
- ND_LOG_FIELD_UUID(NDF_MESSAGE_ID, &streaming_to_parent_msgid),
315
ND_LOG_FIELD_END(),
316
};
317
ND_LOG_STACK_PUSH(lgs);
326
nd_log(NDLS_DAEMON, NDLP_ERR,
327
"STREAM RCV[%zu] '%s' [from [%s]:%s]: send buffer is full (buffer size %u, max %u, used %u, available %u). "
328
"Restarting connection.",
270
- sth->id, rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port,
329
+ sth->id, rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port,
330
stats.bytes_size, stats.bytes_max_size, stats.bytes_outstanding, stats.bytes_available);
331
332
stream_receiver_remove(sth, rpt, "receiver send buffer overflow");
356
msg.opcode = STREAM_OPCODE_RECEIVER_BUFFER_OVERFLOW;
357
rc = -1;
358
}
300
- else if(was_empty)
301
- msg.opcode = STREAM_OPCODE_RECEIVER_POLLOUT;
359
+ else {
360
+ stream_receiver_log_payload(rpt, txt, type, false);
361
+
362
+ if(was_empty)
363
+ msg.opcode = STREAM_OPCODE_RECEIVER_POLLOUT;
364
+ }
365
366
spinlock_unlock(&rpt->thread.send_to_child.spinlock);
367
371
return rc;
372
}
373
311
-static void streaming_parser_init(struct receiver_state *rpt) {
312
- rpt->thread.cd = (struct plugind){
313
- .update_every = default_rrd_update_every,
314
- .unsafe = {
315
- .spinlock = SPINLOCK_INITIALIZER,
316
- .running = true,
317
- .enabled = true,
318
- },
319
- .started_t = now_realtime_sec(),
320
- };
321
-
322
- // put the client IP and port into the buffers used by plugins.d
323
- {
324
- char buf[CONFIG_MAX_NAME];
325
- snprintfz(buf, sizeof(buf), "[%s]:%s", rpt->client_ip, rpt->client_port);
326
- string_freez(rpt->thread.cd.id);
327
- rpt->thread.cd.id = string_strdupz(buf);
328
-
329
- string_freez(rpt->thread.cd.filename);
330
- rpt->thread.cd.filename = string_strdupz(buf);
331
-
332
- string_freez(rpt->thread.cd.fullfilename);
333
- rpt->thread.cd.fullfilename = string_strdupz(buf);
334
-
335
- string_freez(rpt->thread.cd.cmd);
336
- rpt->thread.cd.cmd = string_strdupz(buf);
337
- }
338
-
339
- PARSER *parser = NULL;
340
- {
341
- PARSER_USER_OBJECT user = {
342
- .enabled = plugin_is_enabled(&rpt->thread.cd),
343
- .host = rpt->host,
344
- .opaque = rpt,
345
- .cd = &rpt->thread.cd,
346
- .trust_durations = 1,
347
- .capabilities = rpt->capabilities,
348
- };
349
-
350
- parser = parser_init(&user, -1, -1, PARSER_INPUT_SPLIT, &rpt->sock);
351
- parser->send_to_plugin_data = rpt;
352
- parser->send_to_plugin_cb = send_to_child;
353
- }
354
-
355
-#ifdef ENABLE_H2O
356
- parser->h2o_ctx = rpt->h2o_ctx;
357
-#endif
358
-
359
- pluginsd_keywords_init(parser, PARSER_INIT_STREAMING);
360
-
361
- rpt->thread.compressed.start = 0;
362
- rpt->thread.compressed.used = 0;
363
- rpt->thread.compressed.enabled = stream_decompression_initialize(rpt);
364
- buffered_reader_init(&rpt->reader);
365
-
366
-#ifdef NETDATA_LOG_STREAM_RECEIVE
367
- {
368
- char filename[FILENAME_MAX + 1];
369
- snprintfz(filename, FILENAME_MAX, "/tmp/stream-receiver-%s.txt", rpt->host ? rrdhost_hostname(
370
- rpt->host) : "unknown"
371
- );
372
- parser->user.stream_log_fp = fopen(filename, "w");
373
- parser->user.stream_log_repertoire = PARSER_REP_METADATA;
374
- }
375
-#endif
376
-
377
- __atomic_store_n(&rpt->thread.parser, parser, __ATOMIC_RELAXED);
378
- stream_receiver_send_node_and_claim_id_to_child(rpt->host);
379
-
380
- rpt->thread.buffer = buffer_create(sizeof(rpt->reader.read_buffer), NULL);
381
-
382
- // help rrdset_push_metric_initialize() select the right buffer
383
- rpt->host->stream.snd.commit.receiver_tid = gettid_cached();
384
-}
385
-
374
// --------------------------------------------------------------------------------------------------------------------
375
376
static void stream_receive_log_database_gap(struct receiver_state *rpt) {
386
if(!last_db_entry) {
387
nd_log(NDLS_DAEMON, NDLP_NOTICE,
388
"STREAM RCV '%s' [from [%s]:%s]: node connected; for the first time!",
401
- rrdhost_hostname(host), rpt->client_ip, rpt->client_port);
389
+ rrdhost_hostname(host), rpt->remote_ip, rpt->remote_port);
390
}
391
else {
392
char buf[128];
393
duration_snprintf(buf, sizeof(buf), now - last_db_entry, "s", true);
394
nd_log(NDLS_DAEMON, NDLP_NOTICE,
395
"STREAM RCV '%s' [from [%s]:%s]: node connected; last sample in the database %s ago",
408
- rrdhost_hostname(host), rpt->client_ip, rpt->client_port, buf);
396
+ rrdhost_hostname(host), rpt->remote_ip, rpt->remote_port, buf);
397
}
398
}
399
404
405
ND_LOG_STACK lgs[] = {
406
ND_LOG_FIELD_STR(NDF_NIDL_NODE, rpt->host->hostname),
419
- ND_LOG_FIELD_UUID(NDF_MESSAGE_ID, &streaming_to_parent_msgid),
407
+ ND_LOG_FIELD_UUID(NDF_MESSAGE_ID, &streaming_from_child_msgid),
408
ND_LOG_FIELD_END(),
409
};
410
ND_LOG_STACK_PUSH(lgs);
411
412
nd_log(NDLS_DAEMON, NDLP_DEBUG,
413
"STREAM RCV[%zu] '%s' [from [%s]:%s]: moving host from receiver queue to receiver running...",
426
- sth->id, rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port);
414
+ sth->id, rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port);
415
+
416
+ sock_setcloexec(rpt->sock.fd, true);
417
+ sock_enlarge_rcv_buf(rpt->sock.fd);
418
+ sock_enlarge_snd_buf(rpt->sock.fd);
419
+ sock_setcork(rpt->sock.fd, false);
420
+ if(sock_setnonblock(rpt->sock.fd, true) != 1)
421
+ nd_log(NDLS_DAEMON, NDLP_ERR,
422
+ "STREAM RCV '%s' [from [%s]:%s]: failed to set non-blocking mode on socket %d",
423
+ rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port, rpt->sock.fd);
424
425
rpt->host->stream.rcv.status.tid = gettid_cached();
426
rpt->thread.meta.type = POLLFD_TYPE_RECEIVER;
436
internal_fatal(META_GET(&sth->run.meta, (Word_t)&rpt->thread.meta) != NULL, "Receiver to be added is already in the list of receivers");
437
META_SET(&sth->run.meta, (Word_t)&rpt->thread.meta, &rpt->thread.meta);
438
442
- if(sock_setnonblock(rpt->sock.fd) < 0)
443
- nd_log(NDLS_DAEMON, NDLP_ERR,
444
- "STREAM RCV '%s' [from [%s]:%s]: cannot set the non-blocking flag from socket %d",
445
- rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port, rpt->sock.fd);
446
-
447
- if(!nd_poll_add(sth->run.ndpl, rpt->sock.fd, ND_POLL_READ, &rpt->thread.meta))
439
+ rpt->thread.wanted = ND_POLL_READ;
440
+ if(!nd_poll_add(sth->run.ndpl, rpt->sock.fd, rpt->thread.wanted, &rpt->thread.meta))
441
nd_log(NDLS_DAEMON, NDLP_ERR,
442
"STREAM RCV[%zu] '%s' [from [%s]:%s]:"
443
"Failed to add receiver socket to nd_poll()",
451
- sth->id, rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port);
444
+ sth->id, rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port);
445
+
446
+ // put the client IP and port into the buffers used by plugins.d
447
+ {
448
+ char buf[CONFIG_MAX_NAME];
449
+ snprintfz(buf, sizeof(buf), "[%s]:%s", rpt->remote_ip, rpt->remote_port);
450
+ string_freez(rpt->thread.cd.id);
451
+ rpt->thread.cd.id = string_strdupz(buf);
452
+
453
+ string_freez(rpt->thread.cd.filename);
454
+ rpt->thread.cd.filename = string_strdupz(buf);
455
+
456
+ string_freez(rpt->thread.cd.fullfilename);
457
+ rpt->thread.cd.fullfilename = string_strdupz(buf);
458
+
459
+ string_freez(rpt->thread.cd.cmd);
460
+ rpt->thread.cd.cmd = string_strdupz(buf);
461
+ }
462
+
463
+ rpt->thread.compressed.start = 0;
464
+ rpt->thread.compressed.used = 0;
465
+ rpt->thread.compressed.enabled = stream_decompression_initialize(rpt);
466
+ buffered_reader_init(&rpt->thread.uncompressed);
467
+
468
+ rpt->thread.line_buffer = buffer_create(sizeof(rpt->thread.uncompressed.read_buffer), NULL);
469
+
470
+ // help preferred_sender_buffer() select the right buffer
471
+ rpt->host->stream.snd.commit.receiver_tid = gettid_cached();
472
+
473
+ rpt->replication.last_progress_ut = now_monotonic_usec();
474
+
475
+ PARSER *parser = NULL;
476
+ {
477
+ rpt->thread.cd = (struct plugind){
478
+ .update_every = nd_profile.update_every,
479
+ .unsafe = {
480
+ .spinlock = SPINLOCK_INITIALIZER,
481
+ .running = true,
482
+ .enabled = true,
483
+ },
484
+ .started_t = now_realtime_sec(),
485
+ };
486
+
487
+ PARSER_USER_OBJECT user = {
488
+ .enabled = plugin_is_enabled(&rpt->thread.cd),
489
+ .host = rpt->host,
490
+ .opaque = rpt,
491
+ .cd = &rpt->thread.cd,
492
+ .trust_durations = 1,
493
+ .capabilities = rpt->capabilities,
494
+#ifdef NETDATA_LOG_STREAM_RECEIVER
495
+ .rpt = rpt,
496
+#endif
497
+ };
498
+
499
+ parser = parser_init(&user, -1, -1, PARSER_INPUT_SPLIT, &rpt->sock);
500
+ parser->send_to_plugin_data = rpt;
501
+ parser->send_to_plugin_cb = send_to_child;
502
+
503
+ pluginsd_keywords_init(parser, PARSER_INIT_STREAMING);
504
+
505
+ __atomic_store_n(&rpt->thread.parser, parser, __ATOMIC_RELAXED);
506
+ }
507
+
508
+#ifdef ENABLE_H2O
509
+ parser->h2o_ctx = rpt->h2o_ctx;
510
+#endif
511
512
stream_receive_log_database_gap(rpt);
454
- rrdhost_state_connected(rpt->host);
513
456
- // keep this last, since it sends commands back to the child
457
- streaming_parser_init(rpt);
514
+ // keep this last - it needs everything ready since to sends data to the child
515
+ stream_receiver_send_node_and_claim_id_to_child(rpt->host);
516
}
517
518
void stream_receiver_move_entire_queue_to_running_unsafe(struct stream_thread *sth) {
531
static void stream_receiver_remove(struct stream_thread *sth, struct receiver_state *rpt, const char *why) {
532
internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__ );
533
534
+ ND_LOG_STACK lgs[] = {
535
+ ND_LOG_FIELD_STR(NDF_NIDL_NODE, rpt->host->hostname),
536
+ ND_LOG_FIELD_TXT(NDF_SRC_IP, rpt->remote_ip),
537
+ ND_LOG_FIELD_TXT(NDF_SRC_PORT, rpt->remote_port),
538
+ ND_LOG_FIELD_CB(NDF_SRC_TRANSPORT, stream_receiver_log_transport, rpt),
539
+ ND_LOG_FIELD_CB(NDF_SRC_CAPABILITIES, stream_receiver_log_capabilities, rpt),
540
+ ND_LOG_FIELD_UUID(NDF_MESSAGE_ID, &streaming_from_child_msgid),
541
+ ND_LOG_FIELD_END(),
542
+ };
543
+ ND_LOG_STACK_PUSH(lgs);
544
+
545
PARSER *parser = __atomic_load_n(&rpt->thread.parser, __ATOMIC_RELAXED);
546
size_t count = parser ? parser->user.data_collections_count : 0;
547
551
"receiver disconnected (after %zu received messages): %s"
552
, sth->id
553
, rpt->hostname ? rpt->hostname : "-"
485
- , rpt->client_ip ? rpt->client_ip : "-"
486
- , rpt->client_port ? rpt->client_port : "-"
554
+ , rpt->remote_ip ? rpt->remote_ip : "-"
555
+ , rpt->remote_port ? rpt->remote_port : "-"
556
, count
557
, why ? why : "");
558
490
- rrdhost_state_disconnected(rpt->host);
559
+ internal_fatal(META_GET(&sth->run.meta, (Word_t)&rpt->thread.meta) == NULL,
560
+ "Receiver to be removed is not found in the list of receivers");
561
492
- internal_fatal(META_GET(&sth->run.meta, (Word_t)&rpt->thread.meta) == NULL, "Receiver to be removed is not found in the list of receivers");
562
META_DEL(&sth->run.meta, (Word_t)&rpt->thread.meta);
563
495
- if(!nd_poll_del(sth->run.ndpl, rpt->sock.fd))
564
+ rpt->thread.wanted = 0;
565
+ if(!nd_poll_del(sth->run.ndpl, rpt->sock.fd, &rpt->thread.meta))
566
nd_log(NDLS_DAEMON, NDLP_ERR, "Failed to delete receiver socket from nd_poll()");
567
568
rpt->host->stream.rcv.status.tid = 0;
569
500
- spinlock_lock(&rpt->thread.send_to_child.spinlock);
501
- rpt->thread.send_to_child.msg.session = 0;
502
- rpt->thread.send_to_child.msg.meta = NULL;
503
- stream_circular_buffer_destroy(rpt->thread.send_to_child.scb);
504
- rpt->thread.send_to_child.scb = NULL;
505
- spinlock_unlock(&rpt->thread.send_to_child.spinlock);
506
-
507
- stream_thread_node_removed(rpt->host);
508
-
509
- buffer_free(rpt->thread.buffer);
510
- rpt->thread.buffer = NULL;
511
-
570
+ // make sure send_to_plugin() will not write any data to the socket (or wait for it to finish)
571
if(parser) {
513
- parser->user.v2.stream_buffer.wb = NULL;
514
-
515
- // make sure send_to_plugin() will not write any data to the socket
572
spinlock_lock(&parser->writer.spinlock);
573
parser->fd_input = -1;
574
parser->fd_output = -1;
575
parser->sock = NULL;
576
spinlock_unlock(&parser->writer.spinlock);
577
+
578
+ parser->user.v2.stream_buffer.wb = NULL;
579
}
580
523
- // the parser stopped
581
+ stream_thread_node_removed(rpt->host);
582
+
583
+ // set a default exit reason, if not set
584
receiver_set_exit_reason(rpt, STREAM_HANDSHAKE_DISCONNECT_PARSER_EXIT, false);
585
586
// in case we are connected to netdata cloud,
595
// DO NOT USE rpt after this point
596
}
597
598
+static bool stream_receiver_dequeue_senders(struct stream_thread *sth, struct receiver_state *rpt, usec_t now_ut) {
599
+ internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__);
600
+
601
+ // re-check if we need to send data after reading - if we do, try now
602
+ if(rpt->thread.wanted & ND_POLL_WRITE) {
603
+ worker_is_busy(WORKER_STREAM_JOB_SOCKET_SEND);
604
+ if(!stream_receiver_send_data(sth, rpt, now_ut, false))
605
+ return false;
606
+ }
607
+
608
+ if(rpt->host->sender && // the host has a sender
609
+ rpt->host->stream.snd.status.tid == gettid_cached() && // the sender is mine
610
+ (rpt->host->sender->thread.wanted & ND_POLL_WRITE)) // the sender needs to send data
611
+ if(!stream_sender_send_data(sth, rpt->host->sender, now_ut, false))
612
+ return false;
613
+
614
+ return true;
615
+}
616
+
617
static ssize_t
539
-stream_receive_and_process(struct stream_thread *sth, struct receiver_state *rpt, PARSER *parser, bool *removed) {
618
+stream_receive_and_process(struct stream_thread *sth, struct receiver_state *rpt, PARSER *parser, usec_t now_ut __maybe_unused, bool *removed) {
619
internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__);
620
+ *removed = false;
621
622
ssize_t rc;
623
if(rpt->thread.compressed.enabled) {
639
if (likely(decompress_rc == DECOMPRESS_OK)) {
640
// loop through all the complete lines found in the uncompressed buffer
641
562
- while (buffered_reader_next_line(&rpt->reader, rpt->thread.buffer)) {
563
- if (unlikely(parser_action(parser, rpt->thread.buffer->buffer))) {
642
+ while (buffered_reader_next_line(&rpt->thread.uncompressed, rpt->thread.line_buffer)) {
643
+ if (unlikely(parser_action(parser, rpt->thread.line_buffer->buffer))) {
644
receiver_set_exit_reason(rpt, STREAM_HANDSHAKE_DISCONNECT_PARSER_FAILED, false);
645
stream_receiver_remove(sth, rpt, "parser action failed");
646
*removed = true;
647
return -1;
648
}
649
570
- rpt->thread.buffer->len = 0;
571
- rpt->thread.buffer->buffer[0] = '\0';
650
+ rpt->thread.line_buffer->len = 0;
651
+ rpt->thread.line_buffer->buffer[0] = '\0';
652
}
653
}
654
else if (decompress_rc == DECOMPRESS_NEED_MORE_DATA)
681
}
682
else {
683
rc = receiver_read_uncompressed(rpt);
604
- if(rc <= 0) return rc;
684
+ if(rc <= 0)
685
+ return rc;
686
606
- while(buffered_reader_next_line(&rpt->reader, rpt->thread.buffer)) {
607
- if(unlikely(parser_action(parser, rpt->thread.buffer->buffer))) {
687
+ while(buffered_reader_next_line(&rpt->thread.uncompressed, rpt->thread.line_buffer)) {
688
+ if(unlikely(parser_action(parser, rpt->thread.line_buffer->buffer))) {
689
receiver_set_exit_reason(rpt, STREAM_HANDSHAKE_DISCONNECT_PARSER_FAILED, false);
690
stream_receiver_remove(sth, rpt, "parser action failed");
691
*removed = true;
692
return -1;
693
}
694
614
- rpt->thread.buffer->len = 0;
615
- rpt->thread.buffer->buffer[0] = '\0';
695
+ rpt->thread.line_buffer->len = 0;
696
+ rpt->thread.line_buffer->buffer[0] = '\0';
697
}
698
}
699
700
return rc;
701
}
702
703
+bool stream_receiver_send_data(struct stream_thread *sth, struct receiver_state *rpt, usec_t now_ut, bool process_opcodes_and_enable_removal) {
704
+ internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__ );
705
+
706
+ EVLOOP_STATUS status = EVLOOP_STATUS_CONTINUE;
707
+ while(status == EVLOOP_STATUS_CONTINUE) {
708
+ if (!spinlock_trylock(&rpt->thread.send_to_child.spinlock)) {
709
+ status = EVLOOP_STATUS_CANT_GET_LOCK;
710
+ break;
711
+ }
712
+
713
+ char *chunk;
714
+ STREAM_CIRCULAR_BUFFER *scb = rpt->thread.send_to_child.scb;
715
+ STREAM_CIRCULAR_BUFFER_STATS *stats = stream_circular_buffer_stats_unsafe(scb);
716
+ size_t outstanding = stream_circular_buffer_get_unsafe(scb, &chunk);
717
+
718
+ if(!outstanding) {
719
+ status = EVLOOP_STATUS_NO_MORE_DATA;
720
+ spinlock_unlock(&rpt->thread.send_to_child.spinlock);
721
+ continue;
722
+ }
723
+
724
+ ssize_t rc = write_stream(rpt, chunk, outstanding);
725
+ if (likely(rc > 0)) {
726
+ rpt->thread.last_traffic_ut = now_ut;
727
+ stream_circular_buffer_del_unsafe(scb, rc, now_ut);
728
+ if (!stats->bytes_outstanding) {
729
+ rpt->thread.wanted = ND_POLL_READ;
730
+ if (!nd_poll_upd(sth->run.ndpl, rpt->sock.fd, rpt->thread.wanted, &rpt->thread.meta))
731
+ nd_log(NDLS_DAEMON, NDLP_ERR,
732
+ "STREAM RCV[%zu] '%s' [from [%s]:%s]: cannot update nd_poll()",
733
+ sth->id, rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port);
734
+
735
+ // recreate the circular buffer if we have to
736
+ stream_circular_buffer_recreate_timed_unsafe(rpt->thread.send_to_child.scb, now_ut, false);
737
+ status = EVLOOP_STATUS_NO_MORE_DATA;
738
+ }
739
+ }
740
+ else if (rc == 0 || errno == ECONNRESET)
741
+ status = EVLOOP_STATUS_SOCKET_CLOSED;
742
+
743
+ else if (rc < 0) {
744
+ if (errno == EWOULDBLOCK || errno == EAGAIN || errno == EINTR)
745
+ status = EVLOOP_STATUS_SOCKET_FULL;
746
+ else
747
+ status = EVLOOP_STATUS_SOCKET_ERROR;
748
+ }
749
+
750
+ spinlock_unlock(&rpt->thread.send_to_child.spinlock);
751
+
752
+ if (status == EVLOOP_STATUS_SOCKET_ERROR || status == EVLOOP_STATUS_SOCKET_CLOSED) {
753
+ const char *disconnect_reason;
754
+ STREAM_HANDSHAKE reason;
755
+
756
+ if(status == EVLOOP_STATUS_SOCKET_ERROR) {
757
+ worker_is_busy(WORKER_STREAM_JOB_DISCONNECT_SEND_ERROR);
758
+ disconnect_reason = "socket reports error while writing";
759
+ reason = STREAM_HANDSHAKE_DISCONNECT_SOCKET_WRITE_FAILED;
760
+ }
761
+ else /* if(status == EVLOOP_STATUS_SOCKET_CLOSED) */ {
762
+ worker_is_busy(WORKER_STREAM_JOB_DISCONNECT_REMOTE_CLOSED);
763
+ disconnect_reason = "socket reports EOF (closed by child)";
764
+ reason = STREAM_HANDSHAKE_DISCONNECT_SOCKET_CLOSED_BY_REMOTE_END;
765
+ }
766
+
767
+ nd_log(NDLS_DAEMON, NDLP_ERR,
768
+ "STREAM RCV[%zu] '%s' [from [%s]:%s]: %s (%zd, on fd %d) - closing receiver connection - "
769
+ "we have sent %zu bytes in %zu operations.",
770
+ sth->id, rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port,
771
+ disconnect_reason, rc, rpt->sock.fd, stats->bytes_sent, stats->sends);
772
+
773
+ receiver_set_exit_reason(rpt, reason, false);
774
+
775
+ if(process_opcodes_and_enable_removal) {
776
+ // this is not executed from the opcode handling mechanism
777
+ // so we can safely remove the receiver.
778
+ stream_receiver_remove(sth, rpt, disconnect_reason);
779
+ }
780
+ else {
781
+ // protection against this case:
782
+ //
783
+ // 1. receiver gets a replication request
784
+ // 2. parser processes the request
785
+ // 3. parser decides to send back a message to the child (REPLAY_CHART)
786
+ // 4. send_to_child appends the data to the sending circular buffer
787
+ // 5. send_to_child sends opcode to enable sending
788
+ // 6. opcode bypasses the signal and runs this function inline to dispatch immediately
789
+ // 7. sending fails (child disconnected)
790
+ // 8. receiver is removed
791
+ //
792
+ // Point 2 above crashes. The parser is no longer there (freed at point 7)
793
+ // and there is no way for point 2 to know...
794
+ }
795
+ }
796
+ else if(process_opcodes_and_enable_removal &&
797
+ status == EVLOOP_STATUS_CONTINUE &&
798
+ stream_thread_process_opcodes(sth, &rpt->thread.meta))
799
+ status = EVLOOP_STATUS_OPCODE_ON_ME;
800
+ }
801
+
802
+ return EVLOOP_STATUS_STILL_ALIVE(status);
803
+}
804
+
805
+bool stream_receiver_receive_data(struct stream_thread *sth, struct receiver_state *rpt, usec_t now_ut, bool process_opcodes) {
806
+ internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__ );
807
+
808
+ PARSER *parser = __atomic_load_n(&rpt->thread.parser, __ATOMIC_RELAXED);
809
+ ND_LOG_STACK lgs[] = {
810
+ ND_LOG_FIELD_CB(NDF_REQUEST, line_splitter_reconstruct_line, &parser->line),
811
+ ND_LOG_FIELD_CB(NDF_NIDL_NODE, parser_reconstruct_node, parser),
812
+ ND_LOG_FIELD_CB(NDF_NIDL_INSTANCE, parser_reconstruct_instance, parser),
813
+ ND_LOG_FIELD_CB(NDF_NIDL_CONTEXT, parser_reconstruct_context, parser),
814
+ ND_LOG_FIELD_END(),
815
+ };
816
+ ND_LOG_STACK_PUSH(lgs);
817
+
818
+ EVLOOP_STATUS status = EVLOOP_STATUS_CONTINUE;
819
+ while(status == EVLOOP_STATUS_CONTINUE) {
820
+ bool removed = false;
821
+ ssize_t rc = stream_receive_and_process(sth, rpt, parser, now_ut, &removed);
822
+ if(unlikely(removed))
823
+ status = EVLOOP_STATUS_PARSER_FAILED;
824
+
825
+ else if (likely(rc > 0)) {
826
+ rpt->thread.last_traffic_ut = now_ut;
827
+
828
+ if(!stream_receiver_dequeue_senders(sth, rpt, now_ut))
829
+ status = EVLOOP_STATUS_SOCKET_ERROR;
830
+ }
831
+ else if (rc == 0 || errno == ECONNRESET) {
832
+ status = EVLOOP_STATUS_SOCKET_CLOSED;
833
+ }
834
+ else if (rc < 0) {
835
+ if ((errno == EWOULDBLOCK || errno == EAGAIN || errno == EINTR))
836
+ status = EVLOOP_STATUS_SOCKET_FULL;
837
+ else
838
+ status = EVLOOP_STATUS_SOCKET_ERROR;
839
+ }
840
+
841
+ if(status == EVLOOP_STATUS_SOCKET_ERROR || status == EVLOOP_STATUS_SOCKET_CLOSED) {
842
+ const char *disconnect_reason;
843
+ STREAM_HANDSHAKE reason;
844
+
845
+ if(status == EVLOOP_STATUS_SOCKET_ERROR) {
846
+ worker_is_busy(WORKER_STREAM_JOB_DISCONNECT_RECEIVE_ERROR);
847
+ reason = STREAM_HANDSHAKE_DISCONNECT_SOCKET_READ_FAILED;
848
+ disconnect_reason = "error during receive";
849
+ }
850
+ else /* if(status == EVLOOP_STATUS_SOCKET_CLOSED) */ {
851
+ worker_is_busy(WORKER_STREAM_JOB_DISCONNECT_REMOTE_CLOSED);
852
+ reason = STREAM_HANDSHAKE_DISCONNECT_SOCKET_CLOSED_BY_REMOTE_END;
853
+ disconnect_reason = "socket reports EOF (closed by child)";
854
+ }
855
+
856
+ nd_log(NDLS_DAEMON, NDLP_ERR,
857
+ "STREAM RCV[%zu] '%s' [from [%s]:%s]: %s (fd %d) - closing receiver connection.",
858
+ sth->id, rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port, disconnect_reason, rpt->sock.fd);
859
+
860
+ receiver_set_exit_reason(rpt, reason, false);
861
+ stream_receiver_remove(sth, rpt, disconnect_reason);
862
+ }
863
+ else if(status == EVLOOP_STATUS_CONTINUE && process_opcodes && stream_thread_process_opcodes(sth, &rpt->thread.meta))
864
+ status = EVLOOP_STATUS_OPCODE_ON_ME;
865
+ }
866
+
867
+ return EVLOOP_STATUS_STILL_ALIVE(status);
868
+}
869
+
870
// process poll() events for streaming receivers
871
// returns true when the receiver is still there, false if it removed it
624
-bool stream_receive_process_poll_events(struct stream_thread *sth, struct receiver_state *rpt, nd_poll_event_t events, usec_t now_ut)
625
-{
872
+bool stream_receive_process_poll_events(struct stream_thread *sth, struct receiver_state *rpt, nd_poll_event_t events, usec_t now_ut) {
873
internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__);
874
628
- PARSER *parser = __atomic_load_n(&rpt->thread.parser, __ATOMIC_RELAXED);
875
ND_LOG_STACK lgs[] = {
630
- ND_LOG_FIELD_TXT(NDF_SRC_IP, rpt->client_ip),
631
- ND_LOG_FIELD_TXT(NDF_SRC_PORT, rpt->client_port),
876
+ ND_LOG_FIELD_TXT(NDF_SRC_IP, rpt->remote_ip),
877
+ ND_LOG_FIELD_TXT(NDF_SRC_PORT, rpt->remote_port),
878
ND_LOG_FIELD_TXT(NDF_NIDL_NODE, rpt->hostname),
879
ND_LOG_FIELD_CB(NDF_SRC_TRANSPORT, stream_receiver_log_transport, rpt),
880
ND_LOG_FIELD_CB(NDF_SRC_CAPABILITIES, stream_receiver_log_capabilities, rpt),
635
- ND_LOG_FIELD_CB(NDF_REQUEST, line_splitter_reconstruct_line, &parser->line),
636
- ND_LOG_FIELD_CB(NDF_NIDL_NODE, parser_reconstruct_node, parser),
637
- ND_LOG_FIELD_CB(NDF_NIDL_INSTANCE, parser_reconstruct_instance, parser),
638
- ND_LOG_FIELD_CB(NDF_NIDL_CONTEXT, parser_reconstruct_context, parser),
881
ND_LOG_FIELD_END(),
882
};
883
ND_LOG_STACK_PUSH(lgs);
906
907
nd_log(NDLS_DAEMON, NDLP_ERR,
908
"STREAM RCV[%zu] '%s' [from [%s]:%s]: %s - closing connection",
667
- sth->id, rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port, error);
909
+ sth->id, rrdhost_hostname(rpt->host), rpt->remote_ip, rpt->remote_port, error);
910
911
receiver_set_exit_reason(rpt, STREAM_HANDSHAKE_DISCONNECT_SOCKET_ERROR, false);
912
stream_receiver_remove(sth, rpt, error);
915
916
if (events & ND_POLL_WRITE) {
917
worker_is_busy(WORKER_STREAM_JOB_SOCKET_SEND);
918
+ if(!stream_receiver_send_data(sth, rpt, now_ut, true))
919
+ return false;
920
+ }
921
677
- bool stop = false;
678
- while(!stop) {
679
- if (spinlock_trylock(&rpt->thread.send_to_child.spinlock)) {
680
- const char *disconnect_reason = NULL;
681
- STREAM_HANDSHAKE reason;
682
-
683
- char *chunk;
684
- STREAM_CIRCULAR_BUFFER *scb = rpt->thread.send_to_child.scb;
685
- STREAM_CIRCULAR_BUFFER_STATS *stats = stream_circular_buffer_stats_unsafe(scb);
686
- size_t outstanding = stream_circular_buffer_get_unsafe(scb, &chunk);
687
- ssize_t rc = write_stream(rpt, chunk, outstanding);
688
- if (likely(rc > 0)) {
689
- stream_circular_buffer_del_unsafe(scb, rc);
690
- if (!stats->bytes_outstanding) {
691
- if (!nd_poll_upd(sth->run.ndpl, rpt->sock.fd, ND_POLL_READ, &rpt->thread.meta))
692
- nd_log(NDLS_DAEMON, NDLP_ERR,
693
- "STREAM RCV[%zu] '%s' [from [%s]:%s]: cannot update nd_poll()",
694
- sth->id, rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port);
695
-
696
- // recreate the circular buffer if we have to
697
- stream_circular_buffer_recreate_timed_unsafe(rpt->thread.send_to_child.scb, now_ut, false);
698
- stop = true;
699
- }
700
- else if(stream_thread_process_opcodes(sth, &rpt->thread.meta))
701
- stop = true;
702
- }
703
- else if (rc == 0 || errno == ECONNRESET) {
704
- disconnect_reason = "socket reports EOF (closed by child)";
705
- reason = STREAM_HANDSHAKE_DISCONNECT_SOCKET_CLOSED_BY_REMOTE_END;
706
- }
707
- else if (rc < 0) {
708
- if (errno == EWOULDBLOCK || errno == EAGAIN || errno == EINTR)
709
- // will try later
710
- stop = true;
711
- else {
712
- disconnect_reason = "socket reports error while writing";
713
- reason = STREAM_HANDSHAKE_DISCONNECT_SOCKET_WRITE_FAILED;
714
- }
715
- }
716
- spinlock_unlock(&rpt->thread.send_to_child.spinlock);
922
+ if (events & ND_POLL_READ) {
923
+ worker_is_busy(WORKER_STREAM_JOB_SOCKET_RECEIVE);
924
+ if(!stream_receiver_receive_data(sth, rpt, now_ut, true))
925
+ return false;
926
+ }
927
718
- if (disconnect_reason) {
719
- worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_SEND_ERROR);
720
- nd_log(NDLS_DAEMON, NDLP_ERR,
721
- "STREAM RCV[%zu] '%s' [from [%s]:%s]: %s (%zd, on fd %d) - closing connection - "
722
- "we have sent %zu bytes in %zu operations.",
723
- sth->id, rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port,
724
- disconnect_reason, rc, rpt->sock.fd, stats->bytes_sent, stats->sends);
725
-
726
- receiver_set_exit_reason(rpt, reason, false);
727
- stream_receiver_remove(sth, rpt, disconnect_reason);
728
- return false;
729
- }
730
- }
731
- else
732
- break;
928
+ return true;
929
+}
930
+
931
+void stream_receiver_check_all_nodes_from_poll(struct stream_thread *sth, usec_t now_ut) {
932
+ internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__ );
933
+
934
+ NETDATA_DOUBLE overall_buffer_ratio = 0.0;
935
+
936
+ Word_t idx = 0;
937
+ for(struct pollfd_meta *m = META_FIRST(&sth->run.meta, &idx);
938
+ m;
939
+ m = META_NEXT(&sth->run.meta, &idx)) {
940
+ if (m->type != POLLFD_TYPE_RECEIVER) continue;
941
+ struct receiver_state *rpt = m->rpt;
942
+
943
+ spinlock_lock(&rpt->thread.send_to_child.spinlock);
944
+ STREAM_CIRCULAR_BUFFER_STATS stats = *stream_circular_buffer_stats_unsafe(rpt->thread.send_to_child.scb);
945
+ spinlock_unlock(&rpt->thread.send_to_child.spinlock);
946
+
947
+ if (stats.buffer_ratio > overall_buffer_ratio)
948
+ overall_buffer_ratio = stats.buffer_ratio;
949
+
950
+ time_t timeout_s = 600;
951
+ if(unlikely(rpt->thread.last_traffic_ut + timeout_s * USEC_PER_SEC < now_ut &&
952
+ !rrdhost_receiver_replicating_charts(rpt->host))) {
953
+
954
+ ND_LOG_STACK lgs[] = {
955
+ ND_LOG_FIELD_TXT(NDF_SRC_IP, rpt->remote_ip),
956
+ ND_LOG_FIELD_TXT(NDF_SRC_PORT, rpt->remote_port),
957
+ ND_LOG_FIELD_TXT(NDF_NIDL_NODE, rpt->hostname),
958
+ ND_LOG_FIELD_CB(NDF_SRC_TRANSPORT, stream_receiver_log_transport, rpt),
959
+ ND_LOG_FIELD_CB(NDF_SRC_CAPABILITIES, stream_receiver_log_capabilities, rpt),
960
+ ND_LOG_FIELD_END(),
961
+ };
962
+ ND_LOG_STACK_PUSH(lgs);
963
+
964
+ worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_TIMEOUT);
965
+
966
+ char duration[RFC3339_MAX_LENGTH];
967
+ duration_snprintf(duration, sizeof(duration), (int64_t)(now_monotonic_usec() - rpt->thread.last_traffic_ut), "us", true);
968
+
969
+ char pending[64] = "0";
970
+ if(stats.bytes_outstanding)
971
+ size_snprintf(pending, sizeof(pending), stats.bytes_outstanding, "B", false);
972
+
973
+ nd_log(NDLS_DAEMON, NDLP_ERR,
974
+ "STREAM RCV[%zu] '%s' [from %s]: there was not traffic for %ld seconds - closing connection - "
975
+ "we have sent %zu bytes in %zu operations, it is idle for %s, and we have %s pending to send "
976
+ "(buffer is used %.2f%%).",
977
+ sth->id, rrdhost_hostname(rpt->host), rpt->remote_ip, timeout_s,
978
+ stats.bytes_sent, stats.sends, duration, pending, stats.buffer_ratio);
979
+
980
+ receiver_set_exit_reason(rpt, STREAM_HANDSHAKE_DISCONNECT_SOCKET_TIMEOUT, false);
981
+ stream_receiver_remove(sth, rpt, "timeout");
982
+ continue;
983
}
984
+
985
+ rpt->thread.wanted = ND_POLL_READ | (stats.bytes_outstanding ? ND_POLL_WRITE : 0);
986
+ if(!nd_poll_upd(sth->run.ndpl, rpt->sock.fd, rpt->thread.wanted, &rpt->thread.meta))
987
+ nd_log(NDLS_DAEMON, NDLP_ERR,
988
+ "STREAM RCV[%zu] '%s' [from %s]: failed to update nd_poll().",
989
+ sth->id, rrdhost_hostname(rpt->host), rpt->remote_ip);
990
+
991
+ }
992
+}
993
+
994
+static bool stream_receiver_did_replication_progress(struct receiver_state *rpt) {
995
+ RRDHOST *host = rpt->host;
996
+
997
+ size_t my_counter_in = __atomic_load_n(&rpt->replication.last_counter_in, __ATOMIC_RELAXED);
998
+ size_t my_counter_out = __atomic_load_n(&rpt->replication.last_counter_out, __ATOMIC_RELAXED);
999
+ size_t host_counter_in = __atomic_load_n(&host->stream.rcv.status.replication.counter_in, __ATOMIC_RELAXED);
1000
+ size_t host_counter_out = __atomic_load_n(&host->stream.rcv.status.replication.counter_out, __ATOMIC_RELAXED);
1001
+ if(my_counter_in != host_counter_in || my_counter_out != host_counter_out) {
1002
+ // there has been some progress
1003
+ __atomic_store_n(&rpt->replication.last_counter_in, __atomic_load_n(&host->stream.rcv.status.replication.counter_in, __ATOMIC_RELAXED), __ATOMIC_RELAXED);
1004
+ __atomic_store_n(&rpt->replication.last_counter_out, __atomic_load_n(&host->stream.rcv.status.replication.counter_out, __ATOMIC_RELAXED), __ATOMIC_RELAXED);
1005
+ rpt->replication.last_progress_ut = now_monotonic_usec();
1006
+ return true;
1007
}
1008
736
- if (!(events & ND_POLL_READ))
1009
+ if(!my_counter_in || !my_counter_out)
1010
+ // we have not started yet
1011
return true;
1012
739
- // we can receive data from this socket
1013
+ if(__atomic_load_n(&host->stream.rcv.status.replication.backfill_pending, __ATOMIC_RELAXED))
1014
+ // we still have requests to execute
1015
+ return true;
1016
741
- worker_is_busy(WORKER_STREAM_JOB_SOCKET_RECEIVE);
742
- bool removed = false, stop = false;
743
- size_t iterations = 0;
744
- while(!removed && !stop && iterations++ < MAX_IO_ITERATIONS_PER_EVENT) {
745
- ssize_t rc = stream_receive_and_process(sth, rpt, parser, &removed);
746
- if (likely(rc > 0)) {
747
- rpt->last_msg_t = (time_t)(now_ut / USEC_PER_SEC);
1017
+ return (now_monotonic_usec() - rpt->replication.last_progress_ut < 5ULL * 60 * USEC_PER_SEC);
1018
+}
1019
749
- if(stream_thread_process_opcodes(sth, &rpt->thread.meta))
750
- stop = true;
751
- }
752
- else if (rc == 0 || errno == ECONNRESET) {
753
- worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_REMOTE_CLOSED);
754
- nd_log(NDLS_DAEMON, NDLP_ERR,
755
- "STREAM RCV[%zu] '%s' [from [%s]:%s]: socket %d reports EOF (closed by child).",
756
- sth->id, rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port, rpt->sock.fd);
757
- receiver_set_exit_reason(rpt, STREAM_HANDSHAKE_DISCONNECT_SOCKET_CLOSED_BY_REMOTE_END, false);
758
- stream_receiver_remove(sth, rpt, "socket reports EOF (closed by child)");
759
- return false;
760
- }
761
- else if (rc < 0) {
762
- if(removed)
763
- return false;
1020
+void stream_receiver_replication_check_from_poll(struct stream_thread *sth, usec_t now_ut __maybe_unused) {
1021
+ internal_fatal(sth->tid != gettid_cached(), "Function %s() should only be used by the dispatcher thread", __FUNCTION__);
1022
765
- else if ((errno == EWOULDBLOCK || errno == EAGAIN || errno == EINTR))
766
- // will try later
767
- stop = true;
768
- else {
769
- worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_RECEIVE_ERROR);
770
- nd_log(NDLS_DAEMON, NDLP_ERR,
771
- "STREAM RCV[%zu] '%s' [from [%s]:%s]: error during receive (%zd, on fd %d) - closing connection.",
772
- sth->id, rrdhost_hostname(rpt->host), rpt->client_ip, rpt->client_port, rc, rpt->sock.fd);
773
- receiver_set_exit_reason(rpt, STREAM_HANDSHAKE_DISCONNECT_SOCKET_READ_FAILED, false);
774
- stream_receiver_remove(sth, rpt, "error during receive");
775
- return false;
776
- }
1023
+ Word_t idx = 0;
1024
+ for(struct pollfd_meta *m = META_FIRST(&sth->run.meta, &idx);
1025
+ m;
1026
+ m = META_NEXT(&sth->run.meta, &idx)) {
1027
+ if (m->type != POLLFD_TYPE_RECEIVER) continue;
1028
+ struct receiver_state *rpt = m->rpt;
1029
+ RRDHOST *host = rpt->host;
1030
+
1031
+
1032
+ if(stream_receiver_did_replication_progress(rpt))
1033
+ continue;
1034
+
1035
+ size_t exceptions = 0;
1036
+ RRDSET *st;
1037
+ rrdset_foreach_read(st, rpt->host) {
1038
+ RRDSET_FLAGS st_flags = rrdset_flag_get(st);
1039
+ if(st_flags & (RRDSET_FLAG_OBSOLETE | RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED))
1040
+ continue;
1041
+
1042
+ const char *status = (st_flags & RRDSET_FLAG_RECEIVER_REPLICATION_IN_PROGRESS) ? "has not finished" : "has not started";
1043
+
1044
+ nd_log(NDLS_DAEMON, NDLP_WARNING,
1045
+ "STREAM RCV[%zu] '%s' [from %s]: REPLICATION EXCEPTIONS: instance '%s' %s replication yet.",
1046
+ sth->id, rrdhost_hostname(host), rpt->remote_ip,
1047
+ rrdset_id(st), status);
1048
+
1049
+ exceptions++;
1050
+ }
1051
+ rrdset_foreach_done(st);
1052
+
1053
+ if(exceptions && !stream_receiver_did_replication_progress(rpt)) {
1054
+ nd_log(NDLS_DAEMON, NDLP_WARNING,
1055
+ "STREAM RCV[%zu] '%s' [from %s]: REPLICATION EXCEPTIONS SUMMARY: node has %zu stalled replication requests. "
1056
+ "We have received %u and sent %u replication commands. "
1057
+ "Disconnecting node to restore streaming.",
1058
+ sth->id, rrdhost_hostname(rpt->host), rpt->remote_ip, exceptions,
1059
+ __atomic_load_n(&host->stream.rcv.status.replication.counter_in, __ATOMIC_RELAXED),
1060
+ __atomic_load_n(&host->stream.rcv.status.replication.counter_out, __ATOMIC_RELAXED));
1061
+
1062
+ receiver_set_exit_reason(rpt, STREAM_HANDSHAKE_REPLICATION_STALLED, false);
1063
+ stream_receiver_remove(sth, rpt, "replication reception stalled");
1064
}
1065
}
779
-
780
- return !removed;
1066
}
1067
1068
void stream_receiver_cleanup(struct stream_thread *sth) {
1080
static void stream_receiver_replication_reset(RRDHOST *host) {
1081
RRDSET *st;
1082
rrdset_foreach_read(st, host) {
798
- rrdset_flag_clear(st, RRDSET_FLAG_RECEIVER_REPLICATION_IN_PROGRESS);
799
- rrdset_flag_set(st, RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED);
1083
+ RRDSET_FLAGS old = rrdset_flag_set_and_clear(st, RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED, RRDSET_FLAG_RECEIVER_REPLICATION_IN_PROGRESS);
1084
+ if(!(old & RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED))
1085
+ rrdhost_receiver_replicating_charts_minus_one(host);
1086
+
1087
+#ifdef REPLICATION_TRACKING
1088
+ st->stream.rcv.who = REPLAY_WHO_UNKNOWN;
1089
+#endif
1090
}
1091
rrdset_foreach_done(st);
802
- rrdhost_receiver_replicating_charts_zero(host);
1092
+
1093
+ if(rrdhost_receiver_replicating_charts(host) != 0) {
1094
+ nd_log(NDLS_DAEMON, NDLP_WARNING,
1095
+ "STREAM REPLAY ERROR: receiver replication instances counter should be zero, but it is %u"
1096
+ " - resetting it to zero",
1097
+ rrdhost_receiver_replicating_charts(host));
1098
+
1099
+ rrdhost_receiver_replicating_charts_zero(host);
1100
+ }
1101
+
1102
+ __atomic_store_n(&host->stream.rcv.status.replication.counter_in, 0, __ATOMIC_RELAXED);
1103
+ __atomic_store_n(&host->stream.rcv.status.replication.counter_out, 0, __ATOMIC_RELAXED);
1104
}
1105
1106
bool rrdhost_set_receiver(RRDHOST *host, struct receiver_state *rpt) {
1110
rrdhost_receiver_lock(host);
1111
1112
if (!host->receiver) {
1113
+ object_state_activate(&host->state_id);
1114
+
1115
rrdhost_flag_clear(host, RRDHOST_FLAG_ORPHAN);
1116
+ rrdhost_set_health_evloop_iteration(host);
1117
1118
host->stream.rcv.status.connections++;
1119
streaming_receiver_connected();
1133
nd_log(NDLS_DAEMON, NDLP_DEBUG,
1134
"STREAM RCV '%s' [from [%s]:%s]: "
1135
"Postponing health checks for %" PRId64 " seconds, because it was just connected.",
832
- rrdhost_hostname(host), rpt->client_ip, rpt->client_port,
1136
+ rrdhost_hostname(host), rpt->remote_ip, rpt->remote_port,
1137
(int64_t) rpt->config.health.delay);
1138
}
1139
}
1179
1180
rrdhost_receiver_unlock(host);
1181
{
1182
+ // this will wait until all workers finish
1183
+ object_state_deactivate(&host->state_id);
1184
+
1185
// run all these without having the receiver lock
1186
1187
+ rrdhost_set_health_evloop_iteration(host);
1188
ml_host_stop(host);
1189
stream_path_child_disconnected(host);
1190
stream_sender_signal_to_stop_and_wait(host, STREAM_HANDSHAKE_DISCONNECT_RECEIVER_LEFT, false);
883
- stream_receiver_replication_reset(host);
1191
rrdcontext_host_child_disconnected(host);
1192
1193
if (rpt->config.health.enabled)
1199
1200
// now we have the lock again
1201
1202
+ stream_receiver_replication_reset(host);
1203
streaming_receiver_disconnected();
1204
1205
__atomic_store_n(&host->receiver->exit.shutdown, false, __ATOMIC_RELAXED);
1248
netdata_log_error("STREAM RCV[x] '%s' [from [%s]:%s]: "
1249
"streaming thread takes too long to stop, giving up..."
1250
, rrdhost_hostname(host)
943
- , host->receiver->client_ip, host->receiver->client_port);
1251
+ , host->receiver->remote_ip, host->receiver->remote_port);
1252
else
1253
ret = true;
1254