avoid checking replication status all the time (#19361)
Costa Tsaousis committed
Jan 9, 2025 at 20:09 UTC
7746c1c1163f4fb7b92e6ca53ee094d7f824e4b8
4 files changed
+51
-25
src/streaming/stream-receiver-internals.h
+1
@@ -81,6 +81,7 @@ struct receiver_state {
81
uint32_t last_counter_in; // copy from the host, to detect progress
82
uint32_t last_counter_out; // copy from the host, to detect progress
83
usec_t last_progress_ut; // last time we found some progress (monotonic)
84
+ usec_t last_checked_ut; // last time we checked for stalled progress (monotonic)
85
86
time_t first_time_s;
87
} replication;
src/streaming/stream-receiver.c
+26
-14
@@ -988,7 +988,7 @@ static bool stream_receiver_did_replication_progress(struct receiver_state *rpt)
988
// we still have requests to execute
989
return true;
990
991
- return (now_monotonic_usec() - rpt->replication.last_progress_ut < 5ULL * 60 * USEC_PER_SEC);
991
+ return (now_monotonic_usec() - rpt->replication.last_progress_ut < 10ULL * 60 * USEC_PER_SEC);
992
}
993
994
void stream_receiver_replication_check_from_poll(struct stream_thread *sth, usec_t now_ut __maybe_unused) {
@@ -1003,39 +1003,51 @@ void stream_receiver_replication_check_from_poll(struct stream_thread *sth, usec
1003
RRDHOST *host = rpt->host;
1004
1005
1006
- if(stream_receiver_did_replication_progress(rpt))
1006
+ if(stream_receiver_did_replication_progress(rpt)) {
1007
+ rpt->replication.last_checked_ut = 0;
1008
+ continue;
1009
+ }
1010
+
1011
+ if(rpt->replication.last_checked_ut == rpt->replication.last_progress_ut)
1012
continue;
1013
1009
- size_t exceptions = 0;
1014
+ size_t stalled = 0, finished = 0;
1015
RRDSET *st;
1016
rrdset_foreach_read(st, rpt->host) {
1017
RRDSET_FLAGS st_flags = rrdset_flag_get(st);
1013
- if(st_flags & (RRDSET_FLAG_OBSOLETE | RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED))
1018
+ if(st_flags & RRDSET_FLAG_OBSOLETE)
1019
continue;
1020
1016
- const char *status = (st_flags & RRDSET_FLAG_RECEIVER_REPLICATION_IN_PROGRESS) ? "has not finished" : "has not started";
1021
+ if(st_flags & RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED) {
1022
+ finished++;
1023
+ continue;
1024
+ }
1025
1018
- nd_log(NDLS_DAEMON, NDLP_WARNING,
1026
+ nd_log(NDLS_DAEMON, NDLP_DEBUG,
1027
"STREAM RCV[%zu] '%s' [from %s]: REPLICATION EXCEPTIONS: instance '%s' %s replication yet.",
1028
sth->id, rrdhost_hostname(host), rpt->remote_ip,
1021
- rrdset_id(st), status);
1029
+ rrdset_id(st),
1030
+ (st_flags & RRDSET_FLAG_RECEIVER_REPLICATION_IN_PROGRESS) ? "has not finished" : "has not started");
1031
1023
- exceptions++;
1032
+ stalled++;
1033
}
1034
rrdset_foreach_done(st);
1035
1027
- if(exceptions && !stream_receiver_did_replication_progress(rpt)) {
1036
+ if(stalled && !stream_receiver_did_replication_progress(rpt)) {
1037
nd_log(NDLS_DAEMON, NDLP_WARNING,
1029
- "STREAM RCV[%zu] '%s' [from %s]: REPLICATION EXCEPTIONS SUMMARY: node has %zu stalled replication requests. "
1030
- "We have received %u and sent %u replication commands. "
1038
+ "STREAM RCV[%zu] '%s' [from %s]: REPLICATION EXCEPTIONS SUMMARY: node has %zu stalled replication requests (%zu finished). "
1039
+ "We have requested %u and got replies for %u replication commands. "
1040
"Disconnecting node to restore streaming.",
1032
- sth->id, rrdhost_hostname(rpt->host), rpt->remote_ip, exceptions,
1033
- __atomic_load_n(&host->stream.rcv.status.replication.counter_in, __ATOMIC_RELAXED),
1034
- __atomic_load_n(&host->stream.rcv.status.replication.counter_out, __ATOMIC_RELAXED));
1041
+ sth->id, rrdhost_hostname(rpt->host), rpt->remote_ip,
1042
+ stalled, finished,
1043
+ __atomic_load_n(&host->stream.rcv.status.replication.counter_out, __ATOMIC_RELAXED),
1044
+ __atomic_load_n(&host->stream.rcv.status.replication.counter_in, __ATOMIC_RELAXED));
1045
1046
receiver_set_exit_reason(rpt, STREAM_HANDSHAKE_REPLICATION_STALLED, false);
1047
stream_receiver_remove(sth, rpt, "replication reception stalled");
1048
}
1049
+
1050
+ rpt->replication.last_checked_ut = rpt->replication.last_progress_ut;
1051
}
1052
}
1053
src/streaming/stream-sender-internals.h
+1
@@ -88,6 +88,7 @@ struct sender_state {
88
uint32_t last_counter_in; // copy from the host, to detect progress
89
uint32_t last_counter_out; // copy from the host, to detect progress
90
usec_t last_progress_ut; // last time we found some progress (monotonic)
91
+ usec_t last_checked_ut; // last time we checked for stalled progress (monotonic)
92
93
DICTIONARY *requests; // de-duplication of replication requests, per chart
94
time_t oldest_request_after_t; // the timestamp of the oldest replication request
src/streaming/stream-sender.c
+23
-11
@@ -530,7 +530,7 @@ static bool stream_sender_did_replication_progress(struct sender_state *s) {
530
// we still have requests to execute
531
return true;
532
533
- return (now_monotonic_usec() - s->replication.last_progress_ut < 5ULL * 60 * USEC_PER_SEC);
533
+ return (now_monotonic_usec() - s->replication.last_progress_ut < 10ULL * 60 * USEC_PER_SEC);
534
}
535
536
void stream_sender_replication_check_from_poll(struct stream_thread *sth, usec_t now_ut __maybe_unused) {
@@ -544,7 +544,12 @@ void stream_sender_replication_check_from_poll(struct stream_thread *sth, usec_t
544
struct sender_state *s = m->s;
545
RRDHOST *host = s->host;
546
547
- if(stream_sender_did_replication_progress(s))
547
+ if(stream_sender_did_replication_progress(s)) {
548
+ s->replication.last_checked_ut = 0;
549
+ continue;
550
+ }
551
+
552
+ if(s->replication.last_checked_ut == s->replication.last_progress_ut)
553
continue;
554
555
ND_LOG_STACK lgs[] = {
@@ -557,35 +562,42 @@ void stream_sender_replication_check_from_poll(struct stream_thread *sth, usec_t
562
};
563
ND_LOG_STACK_PUSH(lgs);
564
560
- size_t exceptions = 0;
565
+ size_t stalled = 0, finished = 0;
566
RRDSET *st;
567
rrdset_foreach_read(st, host) {
568
RRDSET_FLAGS st_flags = rrdset_flag_get(st);
564
- if(st_flags & (RRDSET_FLAG_OBSOLETE | RRDSET_FLAG_UPSTREAM_IGNORE | RRDSET_FLAG_SENDER_REPLICATION_FINISHED))
569
+ if(st_flags & (RRDSET_FLAG_OBSOLETE | RRDSET_FLAG_UPSTREAM_IGNORE))
570
continue;
571
567
- const char *status = (st_flags & RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS) ? "has not finished" : "has not started";
572
+ if(st_flags & RRDSET_FLAG_SENDER_REPLICATION_FINISHED) {
573
+ finished++;
574
+ continue;
575
+ }
576
569
- nd_log(NDLS_DAEMON, NDLP_WARNING,
577
+ nd_log(NDLS_DAEMON, NDLP_DEBUG,
578
"STREAM SND[%zu] '%s' [to %s]: REPLICATION STALLED: instance '%s' %s replication yet.",
579
sth->id, rrdhost_hostname(host), s->remote_ip,
572
- rrdset_id(st), status);
580
+ rrdset_id(st),
581
+ (st_flags & RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS) ? "has not finished" : "has not started");
582
574
- exceptions++;
583
+ stalled++;
584
}
585
rrdset_foreach_done(st);
586
578
- if(exceptions && !stream_sender_did_replication_progress(s)) {
587
+ if(stalled && !stream_sender_did_replication_progress(s)) {
588
nd_log(NDLS_DAEMON, NDLP_ERR,
580
- "STREAM SND[%zu] '%s' [to %s]: REPLICATION EXCEPTIONS SUMMARY: node has %zu stalled replication requests."
589
+ "STREAM SND[%zu] '%s' [to %s]: REPLICATION EXCEPTIONS SUMMARY: node has %zu stalled replication requests (%zu completed)."
590
"We have received %u and sent %u replication commands. "
591
"Disconnecting node to restore streaming.",
583
- sth->id, rrdhost_hostname(s->host), s->remote_ip, exceptions,
592
+ sth->id, rrdhost_hostname(s->host), s->remote_ip,
593
+ stalled, finished,
594
__atomic_load_n(&host->stream.snd.status.replication.counter_in, __ATOMIC_RELAXED),
595
__atomic_load_n(&host->stream.snd.status.replication.counter_out, __ATOMIC_RELAXED));
596
597
stream_sender_move_running_to_connector_or_remove(sth, s, STREAM_HANDSHAKE_REPLICATION_STALLED, true);
598
}
599
+
600
+ s->replication.last_checked_ut = s->replication.last_progress_ut;
601
}
602
}
603