@cryptotaxi247 / netdata-1 / commits / 4e61a4244

Replication fixes #3 (#14035)

* cleanup and additional information about replication * fix deadlock on sender mutex * do not ignore start streaming empty requests; when there duplicate requests, merge them * flipped the flag * final touch * added queued flag on the charts to prevent them from being obsoleted by the service thread

Costa Tsaousis committed Nov 22, 2022 at 22:42 UTC 4e61a4244e2ab45c29de0ddd84bfec8d9339f388
10 files changed +279 -162
collectors/plugins.d/pluginsd_parser.c
+11 -9
@@ -330,8 +330,9 @@ PARSER_RC pluginsd_chart_definition_end(char **words, size_t num_words, void *us
330 st->replay.before = 0;
331 #endif
332
333 - rrdset_flag_clear(st, RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED);
333 rrdset_flag_set(st, RRDSET_FLAG_RECEIVER_REPLICATION_IN_PROGRESS);
334 + rrdset_flag_clear(st, RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED);
335 + rrdhost_receiver_replicating_charts_plus_one(st->rrdhost);
336
337 ok = replicate_chart_request(send_to_plugin, user_object->parser, host, st, first_entry_child,
338 last_entry_child, 0, 0);
@@ -912,11 +913,6 @@ PARSER_RC pluginsd_replay_rrdset_begin(char **words, size_t num_words, void *use
913 ((PARSER_USER_OBJECT *) user)->st = st;
914 }
915
915 - if(rrdset_flag_check(st, RRDSET_FLAG_OBSOLETE) && !rrdset_flag_check(st, RRDSET_FLAG_ARCHIVED)) {
916 - error("REPLAY: chart '%s' on host '%s' has the OBSOLETE flag set, but it is collected.", rrdset_id(st), rrdhost_hostname(host));
917 - rrdset_isnot_obsolete(st);
918 - }
919 -
916 if(start_time_str && end_time_str) {
917 time_t start_time = strtol(start_time_str, NULL, 0);
918 time_t end_time = strtol(end_time_str, NULL, 0);
@@ -1236,9 +1232,15 @@ PARSER_RC pluginsd_replay_end(char **words, size_t num_words, void *user)
1232 if (st->update_every != update_every_child)
1233 rrdset_set_update_every(st, update_every_child);
1234
1239 - rrdset_flag_set(st, RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED);
1240 - rrdset_flag_clear(st, RRDSET_FLAG_RECEIVER_REPLICATION_IN_PROGRESS);
1241 - rrdset_flag_clear(st, RRDSET_FLAG_SYNC_CLOCK);
1235 + if(rrdset_flag_check(st, RRDSET_FLAG_RECEIVER_REPLICATION_IN_PROGRESS)) {
1236 + rrdset_flag_set(st, RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED);
1237 + rrdset_flag_clear(st, RRDSET_FLAG_RECEIVER_REPLICATION_IN_PROGRESS);
1238 + rrdset_flag_clear(st, RRDSET_FLAG_SYNC_CLOCK);
1239 + rrdhost_receiver_replicating_charts_minus_one(st->rrdhost);
1240 + }
1241 + else
1242 + internal_error(true, "REPLAY: got a " PLUGINSD_KEYWORD_REPLAY_END " on host '%s', chart '%s' with enable_streaming = true, but there is no replication in progress for this chart.",
1243 + rrdhost_hostname(host), rrdset_id(st));
1244
1245 worker_set_metric(WORKER_RECEIVER_JOB_REPLICATION_COMPLETION, 100.0);
1246
daemon/service.c
+23 -17
@@ -138,6 +138,9 @@ static void svc_rrdhost_cleanup_obsolete_charts(RRDHOST *host) {
138 time_t now = now_realtime_sec();
139 RRDSET *st;
140 rrdset_foreach_reentrant(st, host) {
141 + if(rrdset_is_replicating(st))
142 + continue;
143 +
144 if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_OBSOLETE)
145 && st->last_accessed_time + rrdset_free_obsolete_time < now
146 && st->last_updated.tv_sec + rrdset_free_obsolete_time < now
@@ -160,7 +163,7 @@ static void svc_rrdset_check_obsoletion(RRDHOST *host) {
163 time_t last_entry_t;
164 RRDSET *st;
165 rrdset_foreach_read(st, host) {
163 - if(!rrdset_flag_check(st, RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED))
166 + if(rrdset_is_replicating(st))
167 continue;
168
169 last_entry_t = rrdset_last_entry_t(st);
@@ -181,6 +184,8 @@ static void svc_rrd_cleanup_obsolete_charts_from_all_hosts() {
184
185 RRDHOST *host;
186 rrdhost_foreach_read(host) {
187 + if(rrdhost_receiver_replicating_charts(host) || rrdhost_sender_replicating_charts(host))
188 + continue;
189
190 if(rrdhost_flag_check(host, RRDHOST_FLAG_PENDING_OBSOLETE_CHARTS|RRDHOST_FLAG_PENDING_OBSOLETE_DIMENSIONS)) {
191 rrdhost_flag_clear(host, RRDHOST_FLAG_PENDING_OBSOLETE_CHARTS|RRDHOST_FLAG_PENDING_OBSOLETE_DIMENSIONS);
@@ -215,25 +220,26 @@ static void svc_rrdhost_cleanup_orphan_hosts(RRDHOST *protected_host) {
220
221 restart_after_removal:
222 rrdhost_foreach_write(host) {
218 - if(rrdhost_should_be_removed(host, protected_host, now)) {
219 - info("Host '%s' with machine guid '%s' is obsolete - cleaning up.", rrdhost_hostname(host), host->machine_guid);
223 + if(!rrdhost_should_be_removed(host, protected_host, now))
224 + continue;
225
221 - if (rrdhost_option_check(host, RRDHOST_OPTION_DELETE_ORPHAN_HOST)
222 - /* don't delete multi-host DB host files */
223 - && !(host->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE && is_storage_engine_shared(host->db[0].instance))
224 - ) {
225 - worker_is_busy(WORKER_JOB_DELETE_HOST_CHARTS);
226 - rrdhost_delete_charts(host);
227 - }
228 - else {
229 - worker_is_busy(WORKER_JOB_SAVE_HOST_CHARTS);
230 - rrdhost_save_charts(host);
231 - }
226 + info("Host '%s' with machine guid '%s' is obsolete - cleaning up.", rrdhost_hostname(host), host->machine_guid);
227
233 - worker_is_busy(WORKER_JOB_FREE_HOST);
234 - rrdhost_free(host, 0);
235 - goto restart_after_removal;
228 + if (rrdhost_option_check(host, RRDHOST_OPTION_DELETE_ORPHAN_HOST)
229 + /* don't delete multi-host DB host files */
230 + && !(host->rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE && is_storage_engine_shared(host->db[0].instance))
231 + ) {
232 + worker_is_busy(WORKER_JOB_DELETE_HOST_CHARTS);
233 + rrdhost_delete_charts(host);
234 }
235 + else {
236 + worker_is_busy(WORKER_JOB_SAVE_HOST_CHARTS);
237 + rrdhost_save_charts(host);
238 + }
239 +
240 + worker_is_busy(WORKER_JOB_FREE_HOST);
241 + rrdhost_free(host, 0);
242 + goto restart_after_removal;
243 }
244
245 rrd_unlock();
database/rrd.h
+49 -32
@@ -505,44 +505,49 @@ void rrdr_fill_tier_gap_from_smaller_tiers(RRDDIM *rd, size_t tier, time_t now);
505 // and may lead to missing information.
506
507 typedef enum rrdset_flags {
508 - RRDSET_FLAG_DETAIL = (1 << 1), // if set, the data set should be considered as a detail of another
509 - // (the master data set should be the one that has the same family and is not detail)
510 - RRDSET_FLAG_DEBUG = (1 << 2), // enables or disables debugging for a chart
511 - RRDSET_FLAG_OBSOLETE = (1 << 3), // this is marked by the collector/module as obsolete
512 - RRDSET_FLAG_EXPORTING_SEND = (1 << 4), // if set, this chart should be sent to Prometheus web API and external databases
513 - RRDSET_FLAG_EXPORTING_IGNORE = (1 << 5), // if set, this chart should not be sent to Prometheus web API and external databases
514 -
515 - RRDSET_FLAG_UPSTREAM_SEND = (1 << 6), // if set, this chart should be sent upstream (streaming)
516 - RRDSET_FLAG_UPSTREAM_IGNORE = (1 << 7), // if set, this chart should not be sent upstream (streaming)
517 - RRDSET_FLAG_UPSTREAM_EXPOSED = (1 << 8), // if set, we have sent this chart definition to netdata parent (streaming)
518 -
519 - RRDSET_FLAG_STORE_FIRST = (1 << 9), // if set, do not eliminate the first collection during interpolation
520 - RRDSET_FLAG_HETEROGENEOUS = (1 << 10), // if set, the chart is not homogeneous (dimensions in it have multiple algorithms, multipliers or dividers)
521 - RRDSET_FLAG_HOMOGENEOUS_CHECK = (1 << 11), // if set, the chart should be checked to determine if the dimensions are homogeneous
522 - RRDSET_FLAG_HIDDEN = (1 << 12), // if set, do not show this chart on the dashboard, but use it for exporting
523 - RRDSET_FLAG_SYNC_CLOCK = (1 << 13), // if set, microseconds on next data collection will be ignored (the chart will be synced to now)
524 - RRDSET_FLAG_OBSOLETE_DIMENSIONS = (1 << 14), // this is marked by the collector/module when a chart has obsolete dimensions
525 - // No new values have been collected for this chart since agent start, or it was marked RRDSET_FLAG_OBSOLETE at
526 - // least rrdset_free_obsolete_time seconds ago.
527 - RRDSET_FLAG_ARCHIVED = (1 << 15),
528 - RRDSET_FLAG_METADATA_UPDATE = (1 << 16), // Mark that metadata needs to be stored
529 - RRDSET_FLAG_ANOMALY_DETECTION = (1 << 18), // flag to identify anomaly detection charts.
530 - RRDSET_FLAG_INDEXED_ID = (1 << 19), // the rrdset is indexed by its id
531 - RRDSET_FLAG_INDEXED_NAME = (1 << 20), // the rrdset is indexed by its name
532 -
533 - RRDSET_FLAG_PENDING_HEALTH_INITIALIZATION = (1 << 21),
534 -
535 - RRDSET_FLAG_SENDER_REPLICATION_FINISHED = (1 << 22), // the sending side has completed replication
536 - RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED = (1 << 23), // the receiving side has completed replication
537 - RRDSET_FLAG_RECEIVER_REPLICATION_IN_PROGRESS = (1 << 24), // the receiving side has replication in progress
538 -
539 - RRDSET_FLAG_UPSTREAM_SEND_VARIABLES = (1 << 25), // a custom variable has been updated and needs to be exposed to parent
508 + RRDSET_FLAG_DETAIL = (1 << 1), // if set, the data set should be considered as a detail of another
509 + // (the master data set should be the one that has the same family and is not detail)
510 + RRDSET_FLAG_DEBUG = (1 << 2), // enables or disables debugging for a chart
511 + RRDSET_FLAG_OBSOLETE = (1 << 3), // this is marked by the collector/module as obsolete
512 + RRDSET_FLAG_EXPORTING_SEND = (1 << 4), // if set, this chart should be sent to Prometheus web API and external databases
513 + RRDSET_FLAG_EXPORTING_IGNORE = (1 << 5), // if set, this chart should not be sent to Prometheus web API and external databases
514 +
515 + RRDSET_FLAG_UPSTREAM_SEND = (1 << 6), // if set, this chart should be sent upstream (streaming)
516 + RRDSET_FLAG_UPSTREAM_IGNORE = (1 << 7), // if set, this chart should not be sent upstream (streaming)
517 + RRDSET_FLAG_UPSTREAM_EXPOSED = (1 << 8), // if set, we have sent this chart definition to netdata parent (streaming)
518 +
519 + RRDSET_FLAG_STORE_FIRST = (1 << 9), // if set, do not eliminate the first collection during interpolation
520 + RRDSET_FLAG_HETEROGENEOUS = (1 << 10), // if set, the chart is not homogeneous (dimensions in it have multiple algorithms, multipliers or dividers)
521 + RRDSET_FLAG_HOMOGENEOUS_CHECK = (1 << 11), // if set, the chart should be checked to determine if the dimensions are homogeneous
522 + RRDSET_FLAG_HIDDEN = (1 << 12), // if set, do not show this chart on the dashboard, but use it for exporting
523 + RRDSET_FLAG_SYNC_CLOCK = (1 << 13), // if set, microseconds on next data collection will be ignored (the chart will be synced to now)
524 + RRDSET_FLAG_OBSOLETE_DIMENSIONS = (1 << 14), // this is marked by the collector/module when a chart has obsolete dimensions
525 + // No new values have been collected for this chart since agent start, or it was marked RRDSET_FLAG_OBSOLETE at
526 + // least rrdset_free_obsolete_time seconds ago.
527 + RRDSET_FLAG_ARCHIVED = (1 << 15),
528 + RRDSET_FLAG_METADATA_UPDATE = (1 << 16), // Mark that metadata needs to be stored
529 + RRDSET_FLAG_ANOMALY_DETECTION = (1 << 18), // flag to identify anomaly detection charts.
530 + RRDSET_FLAG_INDEXED_ID = (1 << 19), // the rrdset is indexed by its id
531 + RRDSET_FLAG_INDEXED_NAME = (1 << 20), // the rrdset is indexed by its name
532 +
533 + RRDSET_FLAG_PENDING_HEALTH_INITIALIZATION = (1 << 21),
534 +
535 + RRDSET_FLAG_SENDER_REPLICATION_QUEUED = (1 << 22), // the sending side has replication in progress
536 + RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS = (1 << 23), // the sending side has replication in progress
537 + RRDSET_FLAG_SENDER_REPLICATION_FINISHED = (1 << 24), // the sending side has completed replication
538 + RRDSET_FLAG_RECEIVER_REPLICATION_IN_PROGRESS = (1 << 25), // the receiving side has replication in progress
539 + RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED = (1 << 26), // the receiving side has completed replication
540 +
541 + RRDSET_FLAG_UPSTREAM_SEND_VARIABLES = (1 << 27), // a custom variable has been updated and needs to be exposed to parent
542 } RRDSET_FLAGS;
543
544 #define rrdset_flag_check(st, flag) (__atomic_load_n(&((st)->flags), __ATOMIC_SEQ_CST) & (flag))
545 #define rrdset_flag_set(st, flag) __atomic_or_fetch(&((st)->flags), flag, __ATOMIC_SEQ_CST)
546 #define rrdset_flag_clear(st, flag) __atomic_and_fetch(&((st)->flags), ~(flag), __ATOMIC_SEQ_CST)
547
548 +#define rrdset_is_replicating(st) (rrdset_flag_check(st, RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS|RRDSET_FLAG_RECEIVER_REPLICATION_IN_PROGRESS|RRDSET_FLAG_SENDER_REPLICATION_QUEUED) \
549 + && !rrdset_flag_check(st, RRDSET_FLAG_SENDER_REPLICATION_FINISHED|RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED))
550 +
551 struct rrdset {
552 uuid_t chart_uuid; // the global UUID for this chart
553
@@ -951,11 +956,13 @@ struct rrdhost {
956
957 time_t rrdpush_seconds_to_replicate; // max time we want to replicate from the child
958 time_t rrdpush_replication_step; // seconds per replication step
959 + size_t rrdpush_receiver_replicating_charts; // the number of charts currently being replicated from a child
960
961 // the following are state information for the threading
962 // streaming metrics from this netdata to an upstream netdata
963 struct sender_state *sender;
964 netdata_thread_t rrdpush_sender_thread; // the sender thread
965 + size_t rrdpush_sender_replicating_charts; // the number of charts currently being replicated to a parent
966 void *dbsync_worker;
967
968 // ------------------------------------------------------------------------
@@ -1056,6 +1063,16 @@ extern RRDHOST *localhost;
1063 #define rrdhost_aclk_state_lock(host) netdata_mutex_lock(&((host)->aclk_state_lock))
1064 #define rrdhost_aclk_state_unlock(host) netdata_mutex_unlock(&((host)->aclk_state_lock))
1065
1066 +#define rrdhost_receiver_replicating_charts(host) (__atomic_load_n(&((host)->rrdpush_receiver_replicating_charts), __ATOMIC_RELAXED))
1067 +#define rrdhost_receiver_replicating_charts_plus_one(host) (__atomic_add_fetch(&((host)->rrdpush_receiver_replicating_charts), 1, __ATOMIC_RELAXED))
1068 +#define rrdhost_receiver_replicating_charts_minus_one(host) (__atomic_sub_fetch(&((host)->rrdpush_receiver_replicating_charts), 1, __ATOMIC_RELAXED))
1069 +#define rrdhost_receiver_replicating_charts_zero(host) (__atomic_store_n(&((host)->rrdpush_receiver_replicating_charts), 0, __ATOMIC_RELAXED))
1070 +
1071 +#define rrdhost_sender_replicating_charts(host) (__atomic_load_n(&((host)->rrdpush_sender_replicating_charts), __ATOMIC_RELAXED))
1072 +#define rrdhost_sender_replicating_charts_plus_one(host) (__atomic_add_fetch(&((host)->rrdpush_sender_replicating_charts), 1, __ATOMIC_RELAXED))
1073 +#define rrdhost_sender_replicating_charts_minus_one(host) (__atomic_sub_fetch(&((host)->rrdpush_sender_replicating_charts), 1, __ATOMIC_RELAXED))
1074 +#define rrdhost_sender_replicating_charts_zero(host) (__atomic_store_n(&((host)->rrdpush_sender_replicating_charts), 0, __ATOMIC_RELAXED))
1075 +
1076 long rrdhost_hosts_available(void);
1077
1078 // ----------------------------------------------------------------------------
database/rrdhost.c
+2
@@ -741,6 +741,8 @@ RRDHOST *rrdhost_find_or_create(
741 inline int rrdhost_should_be_removed(RRDHOST *host, RRDHOST *protected_host, time_t now) {
742 if(host != protected_host
743 && host != localhost
744 + && rrdhost_receiver_replicating_charts(host) == 0
745 + && rrdhost_sender_replicating_charts(host) == 0
746 && rrdhost_flag_check(host, RRDHOST_FLAG_ORPHAN)
747 && !rrdhost_flag_check(host, RRDHOST_FLAG_ARCHIVED)
748 && !host->receiver
database/rrdset.c
+5 -4
@@ -135,10 +135,11 @@ static void rrdset_insert_callback(const DICTIONARY_ITEM *item __maybe_unused, v
135 st->gap_when_lost_iterations_above = (int) (gap_when_lost_iterations_above + 2);
136 st->rrdhost = host;
137
138 - st->flags = RRDSET_FLAG_SYNC_CLOCK | RRDSET_FLAG_INDEXED_ID;
139 -
140 - if(host == localhost || !host->receiver || !stream_has_capability(host->receiver, STREAM_CAP_REPLICATION))
141 - st->flags |= RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED;
138 + st->flags = RRDSET_FLAG_SYNC_CLOCK
139 + | RRDSET_FLAG_INDEXED_ID
140 + | RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED
141 + | RRDSET_FLAG_SENDER_REPLICATION_FINISHED
142 + ;
143
144 netdata_rwlock_init(&st->alerts.rwlock);
145
streaming/receiver.c
+11 -17
@@ -431,6 +431,15 @@ done:
431 return result;
432 }
433
434 +static void rrdpush_receiver_replication_reset(struct receiver_state *rpt) {
435 + RRDSET *st;
436 + rrdset_foreach_read(st, rpt->host) {
437 + rrdset_flag_clear(st, RRDSET_FLAG_RECEIVER_REPLICATION_IN_PROGRESS);
438 + rrdset_flag_set(st, RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED);
439 + }
440 + rrdset_foreach_done(st);
441 + rrdhost_receiver_replicating_charts_zero(rpt->host);
442 +}
443
444 static int rrdpush_receive(struct receiver_state *rpt)
445 {
@@ -721,14 +730,7 @@ static int rrdpush_receive(struct receiver_state *rpt)
730
731 rrdhost_set_is_parent_label(++localhost->senders_count);
732
724 - if(stream_has_capability(rpt->host->receiver, STREAM_CAP_REPLICATION)) {
725 - RRDSET *st;
726 - rrdset_foreach_read(st, rpt->host) {
727 - rrdset_flag_clear(st, RRDSET_FLAG_RECEIVER_REPLICATION_IN_PROGRESS | RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED);
728 - }
729 - rrdset_foreach_done(st);
730 - }
731 -
733 + rrdpush_receiver_replication_reset(rpt);
734 rrdcontext_host_child_connected(rpt->host);
735
736 rrdhost_flag_clear(rpt->host, RRDHOST_FLAG_RRDPUSH_RECEIVER_DISCONNECTED);
@@ -750,16 +752,8 @@ static int rrdpush_receive(struct receiver_state *rpt)
752 error("STREAM %s [receive from [%s]:%s]: disconnected (completed %zu updates).",
753 rpt->hostname, rpt->client_ip, rpt->client_port, count);
754
753 - if(stream_has_capability(rpt->host->receiver, STREAM_CAP_REPLICATION)) {
754 - RRDSET *st;
755 - rrdset_foreach_read(st, rpt->host) {
756 - rrdset_flag_clear(st, RRDSET_FLAG_RECEIVER_REPLICATION_IN_PROGRESS);
757 - rrdset_flag_set(st, RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED);
758 - }
759 - rrdset_foreach_done(st);
760 - }
761 -
755 rrdcontext_host_child_disconnected(rpt->host);
756 + rrdpush_receiver_replication_reset(rpt);
757
758 #ifdef ENABLE_ACLK
759 // in case we have cloud connection we inform cloud
streaming/replication.c
+111 -40
@@ -290,7 +290,7 @@ static bool send_replay_chart_cmd(send_command callback, void *callback_data, RR
290 #ifdef NETDATA_INTERNAL_CHECKS
291 internal_error(
292 st->replay.after != 0 || st->replay.before != 0,
293 - "REPLAY: host '%s', chart '%s': sending replication request, while there is another inflight",
293 + "REPLAY ERROR: host '%s', chart '%s': sending replication request, while there is another inflight",
294 rrdhost_hostname(st->rrdhost), rrdset_id(st)
295 );
296
@@ -403,15 +403,16 @@ bool replicate_chart_request(send_command callback, void *callback_data, RRDHOST
403 // replication request in sender DICTIONARY
404 // used for de-duplicating the requests
405 struct replication_request {
406 - struct sender_state *sender;
407 - usec_t sender_last_flush_ut;
408 - STRING *chart_id;
409 - time_t after; // key for sorting (JudyL)
410 - time_t before;
411 - Word_t unique_id;
412 - bool start_streaming;
413 - bool found;
414 - bool index_in_judy;
406 + struct sender_state *sender; // the sender we should put the reply at
407 + STRING *chart_id; // the chart of the request
408 + time_t after; // the start time of the query (maybe zero) key for sorting (JudyL)
409 + time_t before; // the end time of the query (maybe zero)
410 + bool start_streaming; // true, when the parent wants to send the rest of the data (before is overwritten) and enable normal streaming
411 +
412 + usec_t sender_last_flush_ut; // the timestamp of the sender, at the time we indexed this request
413 + Word_t unique_id; // auto-increment, later requests have bigger
414 + bool found; // used as a result boolean for the find call
415 + bool indexed_in_judy; // true when the request is indexed in judy
416 };
417
418 // replication sort entry in JudyL array
@@ -486,13 +487,15 @@ static void replication_recursive_unlock() {
487 static struct replication_sort_entry *replication_sort_entry_create(struct replication_request *rq) {
488 struct replication_sort_entry *rse = mallocz(sizeof(struct replication_sort_entry));
489
490 + rrdpush_sender_pending_replication_requests_plus_one(rq->sender);
491 +
492 // copy the request
493 rse->rq = rq;
494 rse->unique_id = rep.next_unique_id++;
495
496 // save the unique id into the request, to be able to delete it later
497 rq->unique_id = rse->unique_id;
495 - rq->index_in_judy = false;
498 + rq->indexed_in_judy = false;
499 return rse;
500 }
501
@@ -524,7 +527,7 @@ static struct replication_sort_entry *replication_sort_entry_add(struct replicat
527 // add it to the inner judy, using unique_id as key
528 Pvoid_t *item = JudyLIns(inner_judy_ptr, rq->unique_id, PJE0);
529 *item = rse;
527 - rq->index_in_judy = true;
530 + rq->indexed_in_judy = true;
531
532 if(!rep.first_time_t || rq->after < rep.first_time_t)
533 rep.first_time_t = rq->after;
@@ -540,7 +543,9 @@ static bool replication_sort_entry_unlink_and_free_unsafe(struct replication_sor
543 rep.removed++;
544 rep.requests_count--;
545
543 - rse->rq->index_in_judy = false;
546 + rrdpush_sender_pending_replication_requests_minus_one(rse->rq->sender);
547 +
548 + rse->rq->indexed_in_judy = false;
549
550 // delete it from the inner judy
551 JudyLDel(*inner_judy_ppptr, rse->rq->unique_id, PJE0);
@@ -562,7 +567,7 @@ static void replication_sort_entry_del(struct replication_request *rq) {
567 struct replication_sort_entry *rse_to_delete = NULL;
568
569 replication_recursive_lock();
565 - if(rq->index_in_judy) {
570 + if(rq->indexed_in_judy) {
571
572 inner_judy_pptr = JudyLGet(rep.JudyL_array, rq->after, PJE0);
573 if (inner_judy_pptr) {
@@ -582,6 +587,13 @@ static void replication_sort_entry_del(struct replication_request *rq) {
587 replication_recursive_unlock();
588 }
589
590 +static inline PPvoid_t JudyLFirstOrNext(Pcvoid_t PArray, Word_t * PIndex, bool first) {
591 + if(unlikely(first))
592 + return JudyLFirst(PArray, PIndex, PJE0);
593 +
594 + return JudyLNext(PArray, PIndex, PJE0);
595 +}
596 +
597 static struct replication_request replication_request_get_first_available() {
598 Pvoid_t *inner_judy_pptr;
599
@@ -590,16 +602,13 @@ static struct replication_request replication_request_get_first_available() {
602 struct replication_request rq = (struct replication_request){ .found = false };
603
604
593 - if(rep.last_after && rep.last_unique_id) {
594 - rep.last_after--;
595 - rep.last_unique_id--;
596 - }
597 - else {
605 + if(unlikely(!rep.last_after || !rep.last_unique_id)) {
606 rep.last_after = 0;
607 rep.last_unique_id = 0;
608 }
609
602 - while(!rq.found && (inner_judy_pptr = JudyLNext(rep.JudyL_array, &rep.last_after, PJE0))) {
610 + bool find_same_after = true;
611 + while(!rq.found && (inner_judy_pptr = JudyLFirstOrNext(rep.JudyL_array, &rep.last_after, find_same_after))) {
612 Pvoid_t *our_item_pptr;
613
614 while(!rq.found && (our_item_pptr = JudyLNext(*inner_judy_pptr, &rep.last_unique_id, PJE0))) {
@@ -610,10 +619,10 @@ static struct replication_request replication_request_get_first_available() {
619 rrdhost_flag_check(s->host, RRDHOST_FLAG_RRDPUSH_SENDER_CONNECTED);
620
621 bool sender_has_been_flushed_since_this_request =
613 - rse->rq->sender_last_flush_ut != __atomic_load_n(&s->last_flush_time_ut, __ATOMIC_SEQ_CST);
622 + rse->rq->sender_last_flush_ut != rrdpush_sender_get_flush_time(s);
623
624 bool sender_has_room_to_spare =
616 - s->replication_sender_buffer_percent_used <= MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED;
625 + s->buffer_used_percentage <= MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED;
626
627 if(unlikely(!sender_is_connected || sender_has_been_flushed_since_this_request)) {
628 rep.skipped_not_connected++;
@@ -635,6 +644,9 @@ static struct replication_request replication_request_get_first_available() {
644 rep.skipped_no_room++;
645 }
646
647 + // call JudyLNext from now on
648 + find_same_after = false;
649 +
650 // prepare for the next iteration on the outer loop
651 rep.last_unique_id = 0;
652 }
@@ -650,6 +662,14 @@ static void replication_request_react_callback(const DICTIONARY_ITEM *item __may
662 struct sender_state *s = sender_state; (void)s;
663 struct replication_request *rq = value;
664
665 + RRDSET *st = rrdset_find(rq->sender->host, string2str(rq->chart_id));
666 + if(!st) {
667 + internal_error(true, "REPLAY: chart '%s' not found on host '%s'",
668 + string2str(rq->chart_id), rrdhost_hostname(rq->sender->host));
669 + }
670 + else
671 + rrdset_flag_set(st, RRDSET_FLAG_SENDER_REPLICATION_QUEUED);
672 +
673 // IMPORTANT:
674 // We use the react instead of the insert callback
675 // because we want the item to be atomically visible
@@ -661,7 +681,9 @@ static void replication_request_react_callback(const DICTIONARY_ITEM *item __may
681 // related to it.
682
683 replication_sort_entry_add(rq);
664 - __atomic_fetch_add(&rq->sender->replication_pending_requests, 1, __ATOMIC_SEQ_CST);
684 +
685 + // this request is about a unique chart for this sender
686 + rrdpush_sender_replicating_charts_plus_one(s);
687 }
688
689 static bool replication_request_conflict_callback(const DICTIONARY_ITEM *item __maybe_unused, void *old_value, void *new_value, void *sender_state) {
@@ -671,24 +693,63 @@ static bool replication_request_conflict_callback(const DICTIONARY_ITEM *item __
693
694 internal_error(
695 true,
674 - "STREAM %s [send to %s]: ignoring duplicate replication command received for chart '%s' (existing from %llu to %llu [%s], new from %llu to %llu [%s])",
696 + "STREAM %s [send to %s]: REPLAY ERROR: merging duplicate replication command received for chart '%s' (existing from %llu to %llu [%s], new from %llu to %llu [%s])",
697 rrdhost_hostname(s->host), s->connected_to, dictionary_acquired_item_name(item),
698 (unsigned long long)rq->after, (unsigned long long)rq->before, rq->start_streaming ? "true" : "false",
699 (unsigned long long)rq_new->after, (unsigned long long)rq_new->before, rq_new->start_streaming ? "true" : "false");
700
679 - string_freez(rq_new->chart_id);
701 + bool updated_after = false, updated_before = false, updated_start_streaming = false, updated = false;
702
681 - return false;
703 + if(rq_new->after < rq->after && rq_new->after != 0)
704 + updated_after = true;
705 +
706 + if(rq_new->before > rq->before)
707 + updated_before = true;
708 +
709 + if(rq_new->start_streaming != rq->start_streaming)
710 + updated_start_streaming = true;
711 +
712 + if(updated_after || updated_before || updated_start_streaming) {
713 + replication_recursive_lock();
714 +
715 + if(rq->indexed_in_judy)
716 + replication_sort_entry_del(rq);
717 +
718 + if(rq_new->after < rq->after && rq_new->after != 0)
719 + rq->after = rq_new->after;
720 +
721 + if(rq->after == 0)
722 + rq->before = 0;
723 + else if(rq_new->before > rq->before)
724 + rq->before = rq_new->before;
725 +
726 + rq->start_streaming = rq->start_streaming;
727 + replication_sort_entry_add(rq);
728 +
729 + replication_recursive_unlock();
730 + updated = true;
731 +
732 + internal_error(
733 + true,
734 + "STREAM %s [send to %s]: REPLAY ERROR: updated duplicate replication command for chart '%s' (from %llu to %llu [%s])",
735 + rrdhost_hostname(s->host), s->connected_to, dictionary_acquired_item_name(item),
736 + (unsigned long long)rq->after, (unsigned long long)rq->before, rq->start_streaming ? "true" : "false");
737 + }
738 +
739 + string_freez(rq_new->chart_id);
740 + return updated;
741 }
742
743 static void replication_request_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, void *value, void *sender_state __maybe_unused) {
744 struct replication_request *rq = value;
745
687 - if(rq->index_in_judy)
746 + // this request is about a unique chart for this sender
747 + rrdpush_sender_replicating_charts_minus_one(rq->sender);
748 +
749 + if(rq->indexed_in_judy)
750 replication_sort_entry_del(rq);
751
752 string_freez(rq->chart_id);
691 - __atomic_fetch_sub(&rq->sender->replication_pending_requests, 1, __ATOMIC_SEQ_CST);
753 }
754
755
@@ -702,13 +763,13 @@ void replication_add_request(struct sender_state *sender, const char *chart_id,
763 .after = after,
764 .before = before,
765 .start_streaming = start_streaming,
705 - .sender_last_flush_ut = __atomic_load_n(&sender->last_flush_time_ut, __ATOMIC_SEQ_CST),
766 + .sender_last_flush_ut = rrdpush_sender_get_flush_time(sender),
767 };
768
769 dictionary_set(sender->replication_requests, chart_id, &rq, sizeof(struct replication_request));
770 }
771
711 -void replication_flush_sender(struct sender_state *sender) {
772 +void replication_sender_delete_pending_requests(struct sender_state *sender) {
773 // allow the dictionary destructor to go faster on locks
774 replication_recursive_lock();
775 dictionary_flush(sender->replication_requests);
@@ -746,7 +807,7 @@ void replication_recalculate_buffer_used_ratio_unsafe(struct sender_state *s) {
807 replication_recursive_unlock();
808 }
809
749 - s->replication_sender_buffer_percent_used = percentage;
810 + s->buffer_used_percentage = percentage;
811 }
812
813 // ----------------------------------------------------------------------------
@@ -843,12 +904,19 @@ void *replication_thread_main(void *ptr __maybe_unused) {
904 worker_is_busy(WORKER_JOB_FIND_CHART);
905 RRDSET *st = rrdset_find(rq.sender->host, string2str(rq.chart_id));
906 if(!st) {
846 - internal_error(true, "REPLAY: chart '%s' not found on host '%s'",
907 + internal_error(true, "REPLAY ERROR: chart '%s' not found on host '%s'",
908 string2str(rq.chart_id), rrdhost_hostname(rq.sender->host));
909
910 continue;
911 }
912
913 + if(!rrdset_flag_check(st, RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS)) {
914 + rrdset_flag_set(st, RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS);
915 + rrdset_flag_clear(st, RRDSET_FLAG_SENDER_REPLICATION_FINISHED);
916 + rrdhost_sender_replicating_charts_plus_one(st->rrdhost);
917 + }
918 + rrdset_flag_clear(st, RRDSET_FLAG_SENDER_REPLICATION_QUEUED);
919 +
920 worker_is_busy(WORKER_JOB_QUERYING);
921
922 latest_first_time_t = rq.after;
@@ -856,8 +924,8 @@ void *replication_thread_main(void *ptr __maybe_unused) {
924 if(rq.after < rq.sender->replication_first_time || !rq.sender->replication_first_time)
925 rq.sender->replication_first_time = rq.after;
926
859 - if(rq.before < rq.sender->replication_min_time || !rq.sender->replication_min_time)
860 - rq.sender->replication_min_time = rq.before;
927 + if(rq.before < rq.sender->replication_current_time || !rq.sender->replication_current_time)
928 + rq.sender->replication_current_time = rq.before;
929
930 netdata_thread_disable_cancelability();
931
@@ -869,17 +937,20 @@ void *replication_thread_main(void *ptr __maybe_unused) {
937
938 rep.executed++;
939
872 - if(start_streaming && rq.sender_last_flush_ut == __atomic_load_n(&rq.sender->last_flush_time_ut, __ATOMIC_SEQ_CST)) {
940 + if(start_streaming && rq.sender_last_flush_ut == rrdpush_sender_get_flush_time(rq.sender)) {
941 worker_is_busy(WORKER_JOB_ACTIVATE_ENABLE_STREAMING);
874 - __atomic_fetch_add(&rq.sender->receiving_metrics, 1, __ATOMIC_SEQ_CST);
942
943 // enable normal streaming if we have to
944 // but only if the sender buffer has not been flushed since we started
945
879 - debug(D_REPLICATION, "Enabling metric streaming for chart %s.%s",
880 - rrdhost_hostname(rq.sender->host), rrdset_id(st));
881 -
882 - rrdset_flag_set(st, RRDSET_FLAG_SENDER_REPLICATION_FINISHED);
946 + if(rrdset_flag_check(st, RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS)) {
947 + rrdset_flag_clear(st, RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS);
948 + rrdset_flag_set(st, RRDSET_FLAG_SENDER_REPLICATION_FINISHED);
949 + rrdhost_sender_replicating_charts_minus_one(st->rrdhost);
950 + }
951 + else
952 + internal_error(true, "REPLAY ERROR: received start streaming command for chart '%s' or host '%s', but the chart is not in progress replicating",
953 + string2str(rq.chart_id), rrdhost_hostname(st->rrdhost));
954 }
955 }
956
streaming/replication.h
+1 -1
@@ -16,7 +16,7 @@ bool replicate_chart_request(send_command callback, void *callback_data,
16
17 void replication_init_sender(struct sender_state *sender);
18 void replication_cleanup_sender(struct sender_state *sender);
19 -void replication_flush_sender(struct sender_state *sender);
19 +void replication_sender_delete_pending_requests(struct sender_state *sender);
20 void replication_add_request(struct sender_state *sender, const char *chart_id, time_t after, time_t before, bool start_streaming);
21 void replication_recalculate_buffer_used_ratio_unsafe(struct sender_state *s);
22
streaming/rrdpush.h
+25 -9
@@ -159,20 +159,36 @@ struct sender_state {
159 struct compressor_state *compressor;
160 #endif
161 #ifdef ENABLE_HTTPS
162 - struct netdata_ssl ssl; // Structure used to encrypt the connection
162 + struct netdata_ssl ssl; // structure used to encrypt the connection
163 #endif
164
165 - DICTIONARY *replication_requests;
166 - size_t replication_pending_requests;
167 - time_t replication_first_time;
168 - time_t replication_min_time;
169 - size_t replication_sender_buffer_percent_used;
170 - bool replication_reached_max;
165 + DICTIONARY *replication_requests; // de-duplication of replication requests, per chart
166
172 - usec_t last_flush_time_ut;
173 - size_t receiving_metrics;
167 + size_t replication_pending_requests; // the currently outstanding replication requests
168 + size_t replication_charts_replicating; // the number of unique charts having pending replication requests (on every request one is added and is removed when we finish it - it does not track completion of the replication for this chart)
169 +
170 + time_t replication_first_time; // the oldest time that has been requested to be replicated
171 + time_t replication_current_time; // the minimum(before) of the executed replication requests
172 +
173 + bool replication_reached_max; // used to avoid resetting the replication thread too frequently
174 +
175 + size_t buffer_used_percentage; // the current utilization of the sending buffer
176 + usec_t last_flush_time_ut; // the last time the sender flushed the sending buffer in USEC
177 };
178
179 +#define rrdpush_sender_set_flush_time(sender) __atomic_store_n(&((sender)->last_flush_time_ut), now_realtime_usec(), __ATOMIC_RELAXED);
180 +#define rrdpush_sender_get_flush_time(sender) __atomic_load_n(&((sender)->last_flush_time_ut), __ATOMIC_RELAXED)
181 +
182 +#define rrdpush_sender_replicating_charts(sender) __atomic_load_n(&((sender)->replication_charts_replicating), __ATOMIC_RELAXED)
183 +#define rrdpush_sender_replicating_charts_plus_one(sender) __atomic_add_fetch(&((sender)->replication_charts_replicating), 1, __ATOMIC_RELAXED)
184 +#define rrdpush_sender_replicating_charts_minus_one(sender) __atomic_sub_fetch(&((sender)->replication_charts_replicating), 1, __ATOMIC_RELAXED)
185 +#define rrdpush_sender_replicating_charts_zero(sender) __atomic_store_n(&((sender)->replication_charts_replicating), 0, __ATOMIC_RELAXED)
186 +
187 +#define rrdpush_sender_pending_replication_requests(sender) __atomic_load_n(&((sender)->replication_pending_requests), __ATOMIC_RELAXED)
188 +#define rrdpush_sender_pending_replication_requests_plus_one(sender) __atomic_add_fetch(&((sender)->replication_pending_requests), 1, __ATOMIC_RELAXED)
189 +#define rrdpush_sender_pending_replication_requests_minus_one(sender) __atomic_sub_fetch(&((sender)->replication_pending_requests), 1, __ATOMIC_RELAXED)
190 +#define rrdpush_sender_pending_replication_requests_zero(sender) __atomic_store_n(&((sender)->replication_pending_requests), 0, __ATOMIC_RELAXED)
191 +
192 struct receiver_state {
193 RRDHOST *host;
194 netdata_thread_t thread;
streaming/sender.c
+41 -33
@@ -169,19 +169,6 @@ void sender_commit(struct sender_state *s, BUFFER *wb) {
169 rrdpush_signal_sender_to_wake_up(s);
170 }
171
172 -
173 -static inline void rrdpush_sender_thread_close_socket(RRDHOST *host) {
174 - if(host->sender->rrdpush_sender_socket != -1) {
175 - close(host->sender->rrdpush_sender_socket);
176 - host->sender->rrdpush_sender_socket = -1;
177 - }
178 -
179 - rrdhost_flag_clear(host, RRDHOST_FLAG_RRDPUSH_SENDER_READY_4_METRICS);
180 - rrdhost_flag_clear(host, RRDHOST_FLAG_RRDPUSH_SENDER_CONNECTED);
181 -
182 - replication_flush_sender(host->sender);
183 -}
184 -
172 static inline void rrdpush_sender_add_host_variable_to_buffer(BUFFER *wb, const RRDVAR_ACQUIRED *rva) {
173 buffer_sprintf(
174 wb
@@ -236,21 +223,10 @@ static void rrdpush_sender_thread_send_custom_host_variables(RRDHOST *host) {
223 static void rrdpush_sender_thread_reset_all_charts(RRDHOST *host) {
224 error("Clearing stream_collected_metrics flag in charts of host %s", rrdhost_hostname(host));
225
239 - bool receive_has_replication = host != localhost && host->receiver && stream_has_capability(host->receiver, STREAM_CAP_REPLICATION);
240 - bool send_has_replication = host->sender && stream_has_capability(host->sender, STREAM_CAP_REPLICATION);
241 -
226 RRDSET *st;
227 rrdset_foreach_read(st, host) {
244 - rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
245 -
246 - if(!receive_has_replication)
247 - rrdset_flag_set(st, RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED);
248 -
249 - if(send_has_replication)
250 - // it will be enabled once replication is done on the sending side
251 - rrdset_flag_clear(st, RRDSET_FLAG_SENDER_REPLICATION_FINISHED);
252 - else
253 - rrdset_flag_set(st, RRDSET_FLAG_SENDER_REPLICATION_FINISHED);
228 + rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED | RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS | RRDSET_FLAG_SENDER_REPLICATION_QUEUED);
229 + rrdset_flag_set(st, RRDSET_FLAG_SENDER_REPLICATION_FINISHED);
230
231 st->upstream_resync_time = 0;
232
@@ -260,21 +236,52 @@ static void rrdpush_sender_thread_reset_all_charts(RRDHOST *host) {
236 rrddim_foreach_done(rd);
237 }
238 rrdset_foreach_done(st);
239 +
240 + rrdhost_sender_replicating_charts_zero(host);
241 }
242
265 -static inline void rrdpush_sender_thread_data_flush(RRDHOST *host) {
266 - __atomic_store_n(&host->sender->last_flush_time_ut, now_realtime_usec(), __ATOMIC_SEQ_CST);
243 +static void rrdpush_sender_cbuffer_flush(RRDHOST *host) {
244 + rrdpush_sender_set_flush_time(host->sender);
245
246 netdata_mutex_lock(&host->sender->mutex);
247 +
248 + // flush the output buffer from any data it may have
249 cbuffer_flush(host->sender->buffer);
250 replication_recalculate_buffer_used_ratio_unsafe(host->sender);
251 +
252 netdata_mutex_unlock(&host->sender->mutex);
253 +}
254 +
255 +static void rrdpush_sender_charts_and_replication_reset(RRDHOST *host) {
256 + rrdpush_sender_set_flush_time(host->sender);
257
258 + // stop all replication commands inflight
259 + replication_sender_delete_pending_requests(host->sender);
260 +
261 + // reset the state of all charts
262 rrdpush_sender_thread_reset_all_charts(host);
263 +
264 + rrdpush_sender_replicating_charts_zero(host->sender);
265 +}
266 +
267 +static void rrdpush_sender_on_connect(RRDHOST *host) {
268 + rrdpush_sender_cbuffer_flush(host);
269 + rrdpush_sender_charts_and_replication_reset(host);
270 rrdpush_sender_thread_send_custom_host_variables(host);
275 - replication_flush_sender(host->sender);
271 +}
272 +
273 +static inline void rrdpush_sender_thread_close_socket(RRDHOST *host) {
274 + if(host->sender->rrdpush_sender_socket != -1) {
275 + close(host->sender->rrdpush_sender_socket);
276 + host->sender->rrdpush_sender_socket = -1;
277 + }
278 +
279 + rrdhost_flag_clear(host, RRDHOST_FLAG_RRDPUSH_SENDER_READY_4_METRICS);
280 + rrdhost_flag_clear(host, RRDHOST_FLAG_RRDPUSH_SENDER_CONNECTED);
281
277 - __atomic_store_n(&host->sender->receiving_metrics, 0, __ATOMIC_SEQ_CST);
282 + // do not flush the circular buffer here
283 + // this function is called sometimes with the mutex lock, sometimes without the lock
284 + rrdpush_sender_charts_and_replication_reset(host);
285 }
286
287 void rrdpush_encode_variable(stream_encoded_t *se, RRDHOST *host)
@@ -704,7 +711,7 @@ static bool attempt_to_connect(struct sender_state *state)
711
712 if(rrdpush_sender_thread_connect_to_parent(state->host, state->default_port, state->timeout, state)) {
713 // reset the buffer, to properly send charts and metrics
707 - rrdpush_sender_thread_data_flush(state->host);
714 + rrdpush_sender_on_connect(state->host);
715
716 // send from the beginning
717 state->begin = 0;
@@ -1185,8 +1192,9 @@ void *rrdpush_sender_thread(void *ptr) {
1192
1193 // If the TCP window never opened then something is wrong, restart connection
1194 if(unlikely(now_monotonic_sec() - s->last_traffic_seen_t > s->timeout &&
1188 - __atomic_load_n(&s->replication_pending_requests, __ATOMIC_SEQ_CST) == 0) &&
1189 - __atomic_load_n(&s->receiving_metrics, __ATOMIC_SEQ_CST) != 0) {
1195 + !rrdpush_sender_pending_replication_requests(s) &&
1196 + !rrdpush_sender_replicating_charts(s)
1197 + )) {
1198 worker_is_busy(WORKER_SENDER_JOB_DISCONNECT_TIMEOUT);
1199 error("STREAM %s [send to %s]: could not send metrics for %d seconds - closing connection - we have sent %zu bytes on this connection via %zu send attempts.", rrdhost_hostname(s->host), s->connected_to, s->timeout, s->sent_bytes_on_this_connection, s->send_attempts);
1200 rrdpush_sender_thread_close_socket(s->host);