@cryptotaxi247 / netdata-1 / commits / 8e1a99ad7

replication fixes #5 (#14038)

* pluginsd cleanup; replication logic cleanup; fix bug in replication begin * log replication start/stop and change the keyword of NETDATA_LOG_REPLICATION_REQUESTS logs to REPLAY * dont ask for data the child does not have; log fixes * more pluginsd cleanup * count sender dictionary entries * fix dictionary_flush()

Costa Tsaousis committed Nov 24, 2022 at 00:24 UTC 8e1a99ad79a1394cbb0ffcaa24bdde85c7b14d81
10 files changed +773 -499
collectors/plugins.d/pluginsd_parser.c
+252 -241
@@ -53,68 +53,98 @@ static int send_to_plugin(const char *txt, void *data) {
53 return -4;
54 }
55
56 -PARSER_RC pluginsd_set(char **words, size_t num_words, void *user)
57 -{
58 - char *dimension = get_word(words, num_words, 1);
59 - char *value = get_word(words, num_words, 2);
56 +static inline RRDHOST *pluginsd_require_host_from_parent(void *user, const char *cmd) {
57 + RRDHOST *host = ((PARSER_USER_OBJECT *) user)->host;
58
59 + if(unlikely(!host))
60 + error("PLUGINSD: command %s requires a host, but is not set.", cmd);
61 +
62 + return host;
63 +}
64 +
65 +static inline RRDSET *pluginsd_require_chart_from_parent(void *user, const char *cmd, const char *parent_cmd) {
66 RRDSET *st = ((PARSER_USER_OBJECT *) user)->st;
62 - RRDHOST *host = ((PARSER_USER_OBJECT *) user)->host;
67
68 + if(unlikely(!st))
69 + error("PLUGINSD: command %s requires a chart defined via command %s, but is not set.", cmd, parent_cmd);
70 +
71 + return st;
72 +}
73 +
74 +static inline RRDDIM_ACQUIRED *pluginsd_acquire_dimension(RRDHOST *host, RRDSET *st, const char *dimension, const char *cmd) {
75 if (unlikely(!dimension || !*dimension)) {
65 - error("requested a SET on chart '%s' of host '%s', without a dimension. Disabling it.", rrdset_id(st), rrdhost_hostname(host));
66 - goto disable;
76 + error("PLUGINSD: 'host:%s/chart:%s' got a %s, without a dimension.",
77 + rrdhost_hostname(host), rrdset_id(st), cmd);
78 + return NULL;
79 }
80
69 - if (unlikely(!value || !*value))
70 - value = NULL;
81 + RRDDIM_ACQUIRED *rda = rrddim_find_and_acquire(st, dimension);
82
72 - if (unlikely(!st)) {
73 - error(
74 - "requested a SET on dimension %s with value %s on host '%s', without a BEGIN. Disabling it.", dimension,
75 - value ? value : "<nothing>", rrdhost_hostname(host));
76 - goto disable;
77 - }
83 + if (unlikely(!rda))
84 + error("PLUGINSD: 'host:%s/chart:%s/dim:%s' got a %s but dimension does not exist.",
85 + rrdhost_hostname(host), rrdset_id(st), dimension, cmd);
86
79 - if (unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
80 - debug(D_PLUGINSD, "is setting dimension '%s'/'%s' to '%s'", rrdset_id(st), dimension, value ? value : "<nothing>");
87 + return rda;
88 +}
89
82 - if (value) {
83 - RRDDIM_ACQUIRED *rda = rrddim_find_and_acquire(st, dimension);
84 - RRDDIM *rd = rrddim_acquired_to_rrddim(rda);
85 - if (unlikely(!rd)) {
86 - error( "requested a SET to dimension with id '%s' on stats '%s' (%s) on host '%s', which does not exist. Disabling it.",
87 - dimension, rrdset_name(st), rrdset_id(st), rrdhost_hostname(st->rrdhost));
88 - goto disable;
89 - }
90 - rrddim_set_by_pointer(st, rd, strtoll(value, NULL, 0));
91 - rrddim_acquired_release(rda);
90 +static inline RRDSET *pluginsd_find_chart(RRDHOST *host, const char *chart, const char *cmd) {
91 + if (unlikely(!chart || !*chart)) {
92 + error("PLUGINSD: 'host:%s' got a %s without a chart id.",
93 + rrdhost_hostname(host), cmd);
94 + return NULL;
95 }
93 - return PARSER_RC_OK;
96
95 -disable:
97 + RRDSET *st = rrdset_find(host, chart);
98 + if (unlikely(!st))
99 + error("PLUGINSD: 'host:%s/chart:%s' got a %s but chart does not exist.",
100 + rrdhost_hostname(host), chart, cmd);
101 +
102 + return st;
103 +}
104 +
105 +static inline PARSER_RC PLUGINSD_DISABLE_PLUGIN(void *user) {
106 ((PARSER_USER_OBJECT *) user)->enabled = 0;
107 return PARSER_RC_ERROR;
108 }
109
110 +PARSER_RC pluginsd_set(char **words, size_t num_words, void *user)
111 +{
112 + char *dimension = get_word(words, num_words, 1);
113 + char *value = get_word(words, num_words, 2);
114 +
115 + RRDHOST *host = pluginsd_require_host_from_parent(user, PLUGINSD_KEYWORD_SET);
116 + if(!host) return PLUGINSD_DISABLE_PLUGIN(user);
117 +
118 + RRDSET *st = pluginsd_require_chart_from_parent(user, PLUGINSD_KEYWORD_SET, PLUGINSD_KEYWORD_CHART);
119 + if(!st) return PLUGINSD_DISABLE_PLUGIN(user);
120 +
121 + RRDDIM_ACQUIRED *rda = pluginsd_acquire_dimension(host, st, dimension, PLUGINSD_KEYWORD_SET);
122 + if(!rda) return PLUGINSD_DISABLE_PLUGIN(user);
123 +
124 + RRDDIM *rd = rrddim_acquired_to_rrddim(rda);
125 +
126 + if (unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
127 + debug(D_PLUGINSD, "PLUGINSD: 'host:%s/chart:%s/dim:%s' SET is setting value to '%s'",
128 + rrdhost_hostname(host), rrdset_id(st), dimension, value && *value ? value : "UNSET");
129 +
130 + if (value && *value)
131 + rrddim_set_by_pointer(st, rd, strtoll(value, NULL, 0));
132 +
133 + rrddim_acquired_release(rda);
134 + return PARSER_RC_OK;
135 +}
136 +
137 PARSER_RC pluginsd_begin(char **words, size_t num_words, void *user)
138 {
139 char *id = get_word(words, num_words, 1);
140 char *microseconds_txt = get_word(words, num_words, 2);
141
105 - RRDSET *st = NULL;
106 - RRDHOST *host = ((PARSER_USER_OBJECT *)user)->host;
142 + RRDHOST *host = pluginsd_require_host_from_parent(user, PLUGINSD_KEYWORD_BEGIN);
143 + if(!host) return PLUGINSD_DISABLE_PLUGIN(user);
144
108 - if (unlikely(!id)) {
109 - error("requested a BEGIN without a chart id for host '%s'. Disabling it.", rrdhost_hostname(host));
110 - goto disable;
111 - }
145 + RRDSET *st = pluginsd_find_chart(host, id, PLUGINSD_KEYWORD_BEGIN);
146 + if(!st) return PLUGINSD_DISABLE_PLUGIN(user);
147
113 - st = rrdset_find(host, id);
114 - if (unlikely(!st)) {
115 - error("requested a BEGIN on chart '%s', which does not exist on host '%s'. Disabling it.", id, rrdhost_hostname(host));
116 - goto disable;
117 - }
148 ((PARSER_USER_OBJECT *)user)->st = st;
149
150 usec_t microseconds = 0;
@@ -127,13 +157,11 @@ PARSER_RC pluginsd_begin(char **words, size_t num_words, void *user)
157 rrdset_next_usec_unfiltered(st, microseconds);
158 else
159 rrdset_next_usec(st, microseconds);
130 - } else
160 + }
161 + else
162 rrdset_next(st);
163 }
164 return PARSER_RC_OK;
134 -disable:
135 - ((PARSER_USER_OBJECT *)user)->enabled = 0;
136 - return PARSER_RC_ERROR;
165 }
166
167 PARSER_RC pluginsd_end(char **words, size_t num_words, void *user)
@@ -141,14 +169,11 @@ PARSER_RC pluginsd_end(char **words, size_t num_words, void *user)
169 UNUSED(words);
170 UNUSED(num_words);
171
144 - RRDSET *st = ((PARSER_USER_OBJECT *) user)->st;
145 - RRDHOST *host = ((PARSER_USER_OBJECT *) user)->host;
172 + RRDHOST *host = pluginsd_require_host_from_parent(user, PLUGINSD_KEYWORD_END);
173 + if(!host) return PLUGINSD_DISABLE_PLUGIN(user);
174
147 - if (unlikely(!st)) {
148 - error("requested an END, without a BEGIN on host '%s'. Disabling it.", rrdhost_hostname(host));
149 - ((PARSER_USER_OBJECT *) user)->enabled = 0;
150 - return PARSER_RC_ERROR;
151 - }
175 + RRDSET *st = pluginsd_require_chart_from_parent(user, PLUGINSD_KEYWORD_END, PLUGINSD_KEYWORD_BEGIN);
176 + if(!st) return PLUGINSD_DISABLE_PLUGIN(user);
177
178 if (unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
179 debug(D_PLUGINSD, "requested an END on chart '%s'", rrdset_id(st));
@@ -165,11 +190,8 @@ PARSER_RC pluginsd_end(char **words, size_t num_words, void *user)
190
191 PARSER_RC pluginsd_chart(char **words, size_t num_words, void *user)
192 {
168 - RRDHOST *host = ((PARSER_USER_OBJECT *) user)->host;
169 - if (unlikely(!host && !((PARSER_USER_OBJECT *) user)->host_exists)) {
170 - debug(D_PLUGINSD, "Ignoring chart belonging to missing or ignored host.");
171 - return PARSER_RC_OK;
172 - }
193 + RRDHOST *host = pluginsd_require_host_from_parent(user, PLUGINSD_KEYWORD_CHART);
194 + if(!host) return PLUGINSD_DISABLE_PLUGIN(user);
195
196 char *type = get_word(words, num_words, 1);
197 char *name = get_word(words, num_words, 2);
@@ -193,10 +215,9 @@ PARSER_RC pluginsd_chart(char **words, size_t num_words, void *user)
215
216 // make sure we have the required variables
217 if (unlikely((!type || !*type || !id || !*id))) {
196 - if (likely(host))
197 - error("requested a CHART, without a type.id, on host '%s'. Disabling it.", rrdhost_hostname(host));
198 - else
199 - error("requested a CHART, without a type.id. Disabling it.");
218 + error("PLUGINSD: 'host:%s' requested a CHART, without a type.id. Disabling it.",
219 + rrdhost_hostname(host));
220 +
221 ((PARSER_USER_OBJECT *) user)->enabled = 0;
222 return PARSER_RC_ERROR;
223 }
@@ -292,39 +313,33 @@ PARSER_RC pluginsd_chart_definition_end(char **words, size_t num_words, void *us
313 const char *first_entry_txt = get_word(words, num_words, 1);
314 const char *last_entry_txt = get_word(words, num_words, 2);
315
316 + RRDHOST *host = pluginsd_require_host_from_parent(user, PLUGINSD_KEYWORD_CHART_DEFINITION_END);
317 + if(!host) return PLUGINSD_DISABLE_PLUGIN(user);
318 +
319 + RRDSET *st = pluginsd_require_chart_from_parent(user, PLUGINSD_KEYWORD_CHART_DEFINITION_END, PLUGINSD_KEYWORD_CHART);
320 + if(!st) return PLUGINSD_DISABLE_PLUGIN(user);
321 +
322 if(unlikely(!first_entry_txt || !last_entry_txt)) {
296 - error("REPLAY: received " PLUGINSD_KEYWORD_CHART_DEFINITION_END " command without first or last entry. Disabling it.");
297 - return PARSER_RC_ERROR;
323 + error("PLUGINSD: 'host:%s' got a " PLUGINSD_KEYWORD_CHART_DEFINITION_END " without first or last entry. Disabling it.",
324 + rrdhost_hostname(host));
325 + return PLUGINSD_DISABLE_PLUGIN(user);
326 }
327
328 long first_entry_child = str2l(first_entry_txt);
329 long last_entry_child = str2l(last_entry_txt);
330
303 - PARSER_USER_OBJECT *user_object = (PARSER_USER_OBJECT *) user;
304 -
305 - RRDHOST *host = user_object->host;
306 - RRDSET *st = user_object->st;
307 - if(unlikely(!host || !st)) {
308 - error("REPLAY: received " PLUGINSD_KEYWORD_CHART_DEFINITION_END " command without a chart. Disabling it.");
309 - return PARSER_RC_ERROR;
310 - }
311 -
331 internal_error(
313 - (first_entry_child != 0 || last_entry_child != 0)
332 + (first_entry_child != 0 || last_entry_child != 0)
333 && (first_entry_child == 0 || last_entry_child == 0),
315 - "REPLAY: received " PLUGINSD_KEYWORD_CHART_DEFINITION_END " with malformed timings (first time %llu, last time %llu).",
316 - (unsigned long long)first_entry_child, (unsigned long long)last_entry_child);
317 -
318 -// internal_error(
319 -// true,
320 -// "REPLAY host '%s', chart '%s': received " PLUGINSD_KEYWORD_CHART_DEFINITION_END " first time %llu, last time %llu.",
321 -// rrdhost_hostname(host), rrdset_id(st),
322 -// (unsigned long long)first_entry_child, (unsigned long long)last_entry_child);
334 + "PLUGINSD: 'host:%s/chart:%s' got a " PLUGINSD_KEYWORD_CHART_DEFINITION_END " with malformed timings (first time %llu, last time %llu).",
335 + rrdhost_hostname(host), rrdset_id(st),
336 + (unsigned long long)first_entry_child, (unsigned long long)last_entry_child
337 + );
338
339 bool ok = true;
340 if(!rrdset_flag_check(st, RRDSET_FLAG_RECEIVER_REPLICATION_IN_PROGRESS)) {
341
327 -#ifdef NETDATA_INTERNAL_CHECKS
342 +#ifdef NETDATA_LOG_REPLICATION_REQUESTS
343 st->replay.start_streaming = false;
344 st->replay.after = 0;
345 st->replay.before = 0;
@@ -334,12 +349,16 @@ PARSER_RC pluginsd_chart_definition_end(char **words, size_t num_words, void *us
349 rrdset_flag_clear(st, RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED);
350 rrdhost_receiver_replicating_charts_plus_one(st->rrdhost);
351
337 - ok = replicate_chart_request(send_to_plugin, user_object->parser, host, st, first_entry_child,
352 + PARSER *parser = ((PARSER_USER_OBJECT *)user)->parser;
353 + ok = replicate_chart_request(send_to_plugin, parser, host, st, first_entry_child,
354 last_entry_child, 0, 0);
355 }
356 +#ifdef NETDATA_LOG_REPLICATION_REQUESTS
357 else {
341 - internal_error(true, "RRDSET: not sending duplicate replication request for chart '%s'", rrdset_id(st));
358 + internal_error(true, "REPLAY: 'host:%s/chart:%s' not sending duplicate replication request",
359 + rrdhost_hostname(st->rrdhost), rrdset_id(st));
360 }
361 +#endif
362
363 return ok ? PARSER_RC_OK : PARSER_RC_ERROR;
364 }
@@ -353,23 +372,22 @@ PARSER_RC pluginsd_dimension(char **words, size_t num_words, void *user)
372 char *divisor_s = get_word(words, num_words, 5);
373 char *options = get_word(words, num_words, 6);
374
356 - RRDSET *st = ((PARSER_USER_OBJECT *) user)->st;
357 - RRDHOST *host = ((PARSER_USER_OBJECT *) user)->host;
358 - if (unlikely(!host && !((PARSER_USER_OBJECT *) user)->host_exists)) {
359 - debug(D_PLUGINSD, "Ignoring dimension belonging to missing or ignored host.");
360 - return PARSER_RC_OK;
361 - }
375 + RRDHOST *host = pluginsd_require_host_from_parent(user, PLUGINSD_KEYWORD_DIMENSION);
376 + if(!host) return PLUGINSD_DISABLE_PLUGIN(user);
377 +
378 + RRDSET *st = pluginsd_require_chart_from_parent(user, PLUGINSD_KEYWORD_DIMENSION, PLUGINSD_KEYWORD_CHART);
379 + if(!st) return PLUGINSD_DISABLE_PLUGIN(user);
380
381 if (unlikely(!id)) {
364 - error(
365 - "requested a DIMENSION, without an id, host '%s' and chart '%s'. Disabling it.", rrdhost_hostname(host),
366 - st ? rrdset_id(st) : "UNSET");
367 - goto disable;
382 + error("PLUGINSD: 'host:%s/chart:%s' got a DIMENSION, without an id. Disabling it.",
383 + rrdhost_hostname(host), st ? rrdset_id(st) : "UNSET");
384 + return PLUGINSD_DISABLE_PLUGIN(user);
385 }
386
387 if (unlikely(!st && !((PARSER_USER_OBJECT *) user)->st_exists)) {
371 - error("requested a DIMENSION, without a CHART, on host '%s'. Disabling it.", rrdhost_hostname(host));
372 - goto disable;
388 + error("PLUGINSD: 'host:%s' got a DIMENSION, without a CHART. Disabling it.",
389 + rrdhost_hostname(host));
390 + return PLUGINSD_DISABLE_PLUGIN(user);
391 }
392
393 long multiplier = 1;
@@ -421,7 +439,8 @@ PARSER_RC pluginsd_dimension(char **words, size_t num_words, void *user)
439 rrddim_flag_clear(rd, RRDDIM_FLAG_META_HIDDEN);
440 metaqueue_dimension_update_flags(rd);
441 }
424 - } else {
442 + }
443 + else {
444 rrddim_option_set(rd, RRDDIM_OPTION_HIDDEN);
445 if (!rrddim_flag_check(rd, RRDDIM_FLAG_META_HIDDEN)) {
446 rrddim_flag_set(rd, RRDDIM_FLAG_META_HIDDEN);
@@ -430,9 +449,6 @@ PARSER_RC pluginsd_dimension(char **words, size_t num_words, void *user)
449 }
450
451 return PARSER_RC_OK;
433 -disable:
434 - ((PARSER_USER_OBJECT *)user)->enabled = 0;
435 - return PARSER_RC_ERROR;
452 }
453
454 // ----------------------------------------------------------------------------
@@ -491,6 +507,7 @@ static bool inflight_functions_conflict_callback(const DICTIONARY_ITEM *item __m
507
508 return false;
509 }
510 +
511 static void inflight_functions_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, void *func, void *parser_ptr __maybe_unused) {
512 struct inflight_function *pf = func;
513
@@ -587,18 +604,22 @@ PARSER_RC pluginsd_function(char **words, size_t num_words, void *user)
604 char *timeout_s = get_word(words, num_words, i++);
605 char *help = get_word(words, num_words, i++);
606
590 - RRDSET *st = (global)?NULL:((PARSER_USER_OBJECT *) user)->st;
591 - RRDHOST *host = ((PARSER_USER_OBJECT *) user)->host;
607 + RRDHOST *host = pluginsd_require_host_from_parent(user, PLUGINSD_KEYWORD_FUNCTION);
608 + if(!host) return PARSER_RC_ERROR;
609
593 - if (unlikely(!host || !timeout_s || !name || !help || (!global && !st))) {
594 - error("requested a FUNCTION, without providing the required data (global = '%s', name = '%s', timeout = '%s', help = '%s'), host '%s', chart '%s'. Ignoring it.",
610 + RRDSET *st = (global)?NULL:pluginsd_require_chart_from_parent(user, PLUGINSD_KEYWORD_FUNCTION, PLUGINSD_KEYWORD_CHART);
611 + if(!st) global = true;
612 +
613 + if (unlikely(!timeout_s || !name || !help || (!global && !st))) {
614 + error("PLUGINSD: 'host:%s/chart:%s' got a FUNCTION, without providing the required data (global = '%s', name = '%s', timeout = '%s', help = '%s'). Ignoring it.",
615 + rrdhost_hostname(host),
616 + st?rrdset_id(st):"(unset)",
617 global?"yes":"no",
618 name?name:"(unset)",
619 timeout_s?timeout_s:"(unset)",
598 - help?help:"(unset)",
599 - host?rrdhost_hostname(host):"(unset)",
600 - st?rrdset_id(st):"(unset)");
601 - return PARSER_RC_OK;
620 + help?help:"(unset)"
621 + );
622 + return PARSER_RC_ERROR;
623 }
624
625 int timeout = PLUGINS_FUNCTIONS_TIMEOUT_DEFAULT;
@@ -683,8 +704,10 @@ PARSER_RC pluginsd_variable(char **words, size_t num_words, void *user)
704 char *value = get_word(words, num_words, 2);
705 NETDATA_DOUBLE v;
706
707 + RRDHOST *host = pluginsd_require_host_from_parent(user, PLUGINSD_KEYWORD_VARIABLE);
708 + if(!host) return PLUGINSD_DISABLE_PLUGIN(user);
709 +
710 RRDSET *st = ((PARSER_USER_OBJECT *) user)->st;
687 - RRDHOST *host = ((PARSER_USER_OBJECT *) user)->host;
711
712 int global = (st) ? 0 : 1;
713
@@ -701,36 +724,50 @@ PARSER_RC pluginsd_variable(char **words, size_t num_words, void *user)
724 }
725
726 if (unlikely(!name || !*name)) {
704 - error("requested a VARIABLE on host '%s', without a variable name. Disabling it.", rrdhost_hostname(host));
727 + error("PLUGINSD: 'host:%s/chart:%s' got a VARIABLE without a variable name. Disabling it.",
728 + rrdhost_hostname(host), st ? rrdset_id(st):"UNSET");
729 +
730 ((PARSER_USER_OBJECT *)user)->enabled = 0;
706 - return PARSER_RC_ERROR;
731 + return PLUGINSD_DISABLE_PLUGIN(user);
732 }
733
734 if (unlikely(!value || !*value))
735 value = NULL;
736
737 if (unlikely(!value)) {
713 - error("cannot set %s VARIABLE '%s' on host '%s' to an empty value", (global) ? "HOST" : "CHART", name,
714 - rrdhost_hostname(host));
738 + error("PLUGINSD: 'host:%s/chart:%s' cannot set %s VARIABLE '%s' to an empty value",
739 + rrdhost_hostname(host),
740 + st ? rrdset_id(st):"UNSET",
741 + (global) ? "HOST" : "CHART",
742 + name);
743 return PARSER_RC_OK;
744 }
745
746 if (!global && !st) {
719 - error("cannot find/create CHART VARIABLE '%s' on host '%s' without a chart", name, rrdhost_hostname(host));
720 - return PARSER_RC_OK;
747 + error("PLUGINSD: 'host:%s/chart:%s' cannot update CHART VARIABLE '%s' without a chart",
748 + rrdhost_hostname(host),
749 + st ? rrdset_id(st):"UNSET",
750 + name
751 + );
752 + return PLUGINSD_DISABLE_PLUGIN(user);
753 }
754
755 char *endptr = NULL;
756 v = (NETDATA_DOUBLE)str2ndd(value, &endptr);
757 if (unlikely(endptr && *endptr)) {
758 if (endptr == value)
727 - error(
728 - "the value '%s' of VARIABLE '%s' on host '%s' cannot be parsed as a number", value, name,
729 - rrdhost_hostname(host));
759 + error("PLUGINSD: 'host:%s/chart:%s' the value '%s' of VARIABLE '%s' cannot be parsed as a number",
760 + rrdhost_hostname(host),
761 + st ? rrdset_id(st):"UNSET",
762 + value,
763 + name);
764 else
731 - error(
732 - "the value '%s' of VARIABLE '%s' on host '%s' has leftovers: '%s'", value, name, rrdhost_hostname(host),
733 - endptr);
765 + error("PLUGINSD: 'host:%s/chart:%s' the value '%s' of VARIABLE '%s' has leftovers: '%s'",
766 + rrdhost_hostname(host),
767 + st ? rrdset_id(st):"UNSET",
768 + value,
769 + name,
770 + endptr);
771 }
772
773 if (global) {
@@ -740,7 +777,9 @@ PARSER_RC pluginsd_variable(char **words, size_t num_words, void *user)
777 rrdvar_custom_host_variable_release(host, rva);
778 }
779 else
743 - error("cannot find/create HOST VARIABLE '%s' on host '%s'", name, rrdhost_hostname(host));
780 + error("PLUGINSD: 'host:%s' cannot find/create HOST VARIABLE '%s'",
781 + rrdhost_hostname(host),
782 + name);
783 } else {
784 const RRDSETVAR_ACQUIRED *rsa = rrdsetvar_custom_chart_variable_add_and_acquire(st, name);
785 if (rsa) {
@@ -748,7 +787,8 @@ PARSER_RC pluginsd_variable(char **words, size_t num_words, void *user)
787 rrdsetvar_custom_chart_variable_release(st, rsa);
788 }
789 else
751 - error("cannot find/create CHART VARIABLE '%s' on host '%s', chart '%s'", name, rrdhost_hostname(host), rrdset_id(st));
790 + error("PLUGINSD: 'host:%s/chart:%s' cannot find/create CHART VARIABLE '%s'",
791 + rrdhost_hostname(host), rrdset_id(st), name);
792 }
793
794 return PARSER_RC_OK;
@@ -767,7 +807,7 @@ PARSER_RC pluginsd_flush(char **words __maybe_unused, size_t num_words __maybe_u
807
808 PARSER_RC pluginsd_disable(char **words __maybe_unused, size_t num_words __maybe_unused, void *user __maybe_unused)
809 {
770 - info("called DISABLE. Disabling it.");
810 + info("PLUGINSD: plugin called DISABLE. Disabling it.");
811 ((PARSER_USER_OBJECT *) user)->enabled = 0;
812 return PARSER_RC_ERROR;
813 }
@@ -779,8 +819,8 @@ PARSER_RC pluginsd_label(char **words, size_t num_words, void *user)
819 const char *value = get_word(words, num_words, 3);
820
821 if (!name || !label_source || !value) {
782 - error("Ignoring malformed or empty LABEL command.");
783 - return PARSER_RC_OK;
822 + error("PLUGINSD: ignoring malformed or empty LABEL command.");
823 + return PLUGINSD_DISABLE_PLUGIN(user);
824 }
825
826 char *store = (char *)value;
@@ -826,10 +866,12 @@ PARSER_RC pluginsd_label(char **words, size_t num_words, void *user)
866
867 PARSER_RC pluginsd_overwrite(char **words __maybe_unused, size_t num_words __maybe_unused, void *user)
868 {
829 - RRDHOST *host = ((PARSER_USER_OBJECT *) user)->host;
869 + RRDHOST *host = pluginsd_require_host_from_parent(user, PLUGINSD_KEYWORD_OVERWRITE);
870 + if(!host) return PLUGINSD_DISABLE_PLUGIN(user);
871 +
872 debug(D_PLUGINSD, "requested to OVERWRITE host labels");
873
832 - if(!host->rrdlabels)
874 + if(unlikely(!host->rrdlabels))
875 host->rrdlabels = rrdlabels_create();
876
877 rrdlabels_migrate_to_these(host->rrdlabels, (DICTIONARY *) (((PARSER_USER_OBJECT *)user)->new_host_labels));
@@ -849,7 +891,7 @@ PARSER_RC pluginsd_clabel(char **words, size_t num_words, void *user)
891
892 if (!name || !value || !*label_source) {
893 error("Ignoring malformed or empty CHART LABEL command.");
852 - return PARSER_RC_OK;
894 + return PLUGINSD_DISABLE_PLUGIN(user);
895 }
896
897 if(unlikely(!((PARSER_USER_OBJECT *) user)->chart_rrdlabels_linked_temporarily)) {
@@ -865,17 +907,18 @@ PARSER_RC pluginsd_clabel(char **words, size_t num_words, void *user)
907
908 PARSER_RC pluginsd_clabel_commit(char **words __maybe_unused, size_t num_words __maybe_unused, void *user)
909 {
868 - RRDHOST *host = ((PARSER_USER_OBJECT *) user)->host;
869 - RRDSET *st = ((PARSER_USER_OBJECT *)user)->st;
910 + RRDHOST *host = pluginsd_require_host_from_parent(user, PLUGINSD_KEYWORD_CLABEL_COMMIT);
911 + if(!host) return PLUGINSD_DISABLE_PLUGIN(user);
912
871 - if (unlikely(!st))
872 - return PARSER_RC_OK;
913 + RRDSET *st = pluginsd_require_chart_from_parent(user, PLUGINSD_KEYWORD_CLABEL_COMMIT, PLUGINSD_KEYWORD_REPLAY_BEGIN);
914 + if(!st) return PLUGINSD_DISABLE_PLUGIN(user);
915
916 debug(D_PLUGINSD, "requested to commit chart labels");
917
918 if(!((PARSER_USER_OBJECT *)user)->chart_rrdlabels_linked_temporarily) {
877 - error("requested CLABEL_COMMIT on host '%s', without a BEGIN, ignoring it.", rrdhost_hostname(host));
878 - return PARSER_RC_OK;
919 + error("PLUGINSD: 'host:%s' got CLABEL_COMMIT, without a CHART or BEGIN. Ignoring it.",
920 + rrdhost_hostname(host));
921 + return PLUGINSD_DISABLE_PLUGIN(user);
922 }
923
924 rrdlabels_remove_all_unmarked(((PARSER_USER_OBJECT *)user)->chart_rrdlabels_linked_temporarily);
@@ -894,44 +937,42 @@ PARSER_RC pluginsd_replay_rrdset_begin(char **words, size_t num_words, void *use
937 char *end_time_str = get_word(words, num_words, 3);
938 char *child_now_str = get_word(words, num_words, 4);
939
897 - RRDSET *st = ((PARSER_USER_OBJECT *) user)->st;
898 - RRDHOST *host = ((PARSER_USER_OBJECT *)user)->host;
940 + RRDHOST *host = pluginsd_require_host_from_parent(user, PLUGINSD_KEYWORD_REPLAY_BEGIN);
941 + if(!host) return PLUGINSD_DISABLE_PLUGIN(user);
942
900 - if (unlikely(!id || (!st && !*id))) {
901 - error("REPLAY: got a " PLUGINSD_KEYWORD_REPLAY_BEGIN " without a chart id for host '%s'. Disabling it.", rrdhost_hostname(host));
902 - goto disable;
903 - }
904 -
905 - if(*id) {
906 - st = rrdset_find(host, id);
907 - if (unlikely(!st)) {
908 - error("REPLAY: got a " PLUGINSD_KEYWORD_REPLAY_BEGIN " on chart '%s', which does not exist on host '%s'. Disabling it.",
909 - id, rrdhost_hostname(host));
910 - goto disable;
911 - }
943 + RRDSET *st;
944 + if (likely(!id || !*id))
945 + st = pluginsd_require_chart_from_parent(user, PLUGINSD_KEYWORD_REPLAY_BEGIN, PLUGINSD_KEYWORD_REPLAY_BEGIN);
946 + else
947 + st = pluginsd_find_chart(host, id, PLUGINSD_KEYWORD_REPLAY_BEGIN);
948
913 - ((PARSER_USER_OBJECT *) user)->st = st;
914 - }
949 + if(!st) return PLUGINSD_DISABLE_PLUGIN(user);
950 + ((PARSER_USER_OBJECT *) user)->st = st;
951
952 if(start_time_str && end_time_str) {
953 time_t start_time = strtol(start_time_str, NULL, 0);
954 time_t end_time = strtol(end_time_str, NULL, 0);
955
956 time_t wall_clock_time = 0, tolerance;
957 + bool wall_clock_comes_from_child; (void)wall_clock_comes_from_child;
958 if(child_now_str) {
959 wall_clock_time = strtol(child_now_str, NULL, 0);
923 - tolerance = 1;
960 + tolerance = st->update_every + 1;
961 + wall_clock_comes_from_child = true;
962 }
963
964 if(wall_clock_time <= 0) {
965 wall_clock_time = now_realtime_sec();
966 tolerance = st->update_every + 60;
967 + wall_clock_comes_from_child = false;
968 }
969
970 +#ifdef NETDATA_LOG_REPLICATION_REQUESTS
971 internal_error(
972 (!st->replay.start_streaming && (end_time < st->replay.after || start_time > st->replay.before)),
933 - "REPLAY: got a " PLUGINSD_KEYWORD_REPLAY_BEGIN " on chart '%s' ('%s') on host '%s', from %ld to %ld, which does not match our request (%ld to %ld).",
934 - rrdset_name(st), rrdset_id(st), rrdhost_hostname(st->rrdhost), start_time, end_time, st->replay.after, st->replay.before);
973 + "REPLAY: 'host:%s/chart:%s' got a " PLUGINSD_KEYWORD_REPLAY_BEGIN " from %ld to %ld, which does not match our request (%ld to %ld).",
974 + rrdhost_hostname(st->rrdhost), rrdset_id(st), start_time, end_time, st->replay.after, st->replay.before);
975 +#endif
976
977 if(start_time && end_time && start_time < wall_clock_time + tolerance && end_time < wall_clock_time + tolerance && start_time < end_time) {
978 if (unlikely(end_time - start_time != st->update_every))
@@ -962,8 +1003,9 @@ PARSER_RC pluginsd_replay_rrdset_begin(char **words, size_t num_words, void *use
1003 }
1004
1005 internal_error(true,
965 - "REPLAY: got a " PLUGINSD_KEYWORD_REPLAY_BEGIN " on chart '%s' ('%s') on host '%s', from %ld to %ld, but timestamps are invalid (now is %ld).",
966 - rrdset_name(st), rrdset_id(st), rrdhost_hostname(st->rrdhost), start_time, end_time, wall_clock_time);
1006 + "PLUGINSD: 'host:%s/chart:%s' got a " PLUGINSD_KEYWORD_REPLAY_BEGIN " from %ld to %ld, but timestamps are invalid (now is %ld [%s], tolerance %ld).",
1007 + rrdhost_hostname(st->rrdhost), rrdset_id(st), start_time, end_time,
1008 + wall_clock_time, wall_clock_comes_from_child ? "child wall clock" : "parent wall clock", tolerance);
1009 }
1010
1011 // the child sends an RBEGIN without any parameters initially
@@ -976,63 +1018,59 @@ PARSER_RC pluginsd_replay_rrdset_begin(char **words, size_t num_words, void *use
1018 ((PARSER_USER_OBJECT *) user)->replay.wall_clock_time = 0;
1019 ((PARSER_USER_OBJECT *) user)->replay.rset_enabled = false;
1020 return PARSER_RC_OK;
979 -
980 -disable:
981 - ((PARSER_USER_OBJECT *)user)->enabled = 0;
982 - return PARSER_RC_ERROR;
1021 }
1022
1023 PARSER_RC pluginsd_replay_set(char **words, size_t num_words, void *user)
1024 {
987 - if(!((PARSER_USER_OBJECT *) user)->replay.rset_enabled)
988 - return PARSER_RC_OK;
989 -
1025 char *dimension = get_word(words, num_words, 1);
1026 char *value_str = get_word(words, num_words, 2);
1027 char *flags_str = get_word(words, num_words, 3);
1028
994 - RRDSET *st = ((PARSER_USER_OBJECT *) user)->st;
995 - RRDHOST *host = ((PARSER_USER_OBJECT *) user)->host;
1029 + RRDHOST *host = pluginsd_require_host_from_parent(user, PLUGINSD_KEYWORD_REPLAY_SET);
1030 + if(!host) return PLUGINSD_DISABLE_PLUGIN(user);
1031
997 - if (unlikely(!st)) {
998 - error("REPLAY: got a " PLUGINSD_KEYWORD_REPLAY_SET " on dimension '%s' on host '%s', without a " PLUGINSD_KEYWORD_REPLAY_BEGIN ". Disabling it.",
999 - dimension, rrdhost_hostname(host));
1000 - goto disable;
1001 - }
1032 + RRDSET *st = pluginsd_require_chart_from_parent(user, PLUGINSD_KEYWORD_REPLAY_SET, PLUGINSD_KEYWORD_REPLAY_BEGIN);
1033 + if(!st) return PLUGINSD_DISABLE_PLUGIN(user);
1034
1003 - if (unlikely(!dimension || !*dimension)) {
1004 - error("REPLAY: got a " PLUGINSD_KEYWORD_REPLAY_SET " on chart '%s' of host '%s', without a dimension. Disabling it.",
1005 - rrdset_id(st), rrdhost_hostname(host));
1006 - goto disable;
1035 + if(!((PARSER_USER_OBJECT *) user)->replay.rset_enabled) {
1036 + error_limit_static_thread_var(erl, 1, 0);
1037 + error_limit(&erl, "PLUGINSD: 'host:%s/chart:%s' got a " PLUGINSD_KEYWORD_REPLAY_SET " but it is disabled by " PLUGINSD_KEYWORD_REPLAY_BEGIN " errors",
1038 + rrdhost_hostname(host), rrdset_id(st));
1039 +
1040 + // we have to return OK here
1041 + return PARSER_RC_OK;
1042 }
1043
1044 + RRDDIM_ACQUIRED *rda = pluginsd_acquire_dimension(host, st, dimension, PLUGINSD_KEYWORD_REPLAY_SET);
1045 + if(!rda) return PLUGINSD_DISABLE_PLUGIN(user);
1046 +
1047 if (unlikely(!((PARSER_USER_OBJECT *) user)->replay.start_time || !((PARSER_USER_OBJECT *) user)->replay.end_time)) {
1010 - error("REPLAY: got a " PLUGINSD_KEYWORD_REPLAY_SET " on dimension '%s' on host '%s', with invalid timestamps %ld to %ld from a " PLUGINSD_KEYWORD_REPLAY_BEGIN ". Disabling it.",
1011 - dimension, rrdhost_hostname(host),
1048 + error("PLUGINSD: 'host:%s/chart:%s/dim:%s' got a " PLUGINSD_KEYWORD_REPLAY_SET " with invalid timestamps %ld to %ld from a " PLUGINSD_KEYWORD_REPLAY_BEGIN ". Disabling it.",
1049 + rrdhost_hostname(host),
1050 + rrdset_id(st),
1051 + dimension,
1052 ((PARSER_USER_OBJECT *) user)->replay.start_time,
1053 ((PARSER_USER_OBJECT *) user)->replay.end_time);
1014 - goto disable;
1054 + return PARSER_RC_ERROR;
1055 }
1056
1057 if (unlikely(!value_str || !*value_str))
1018 - value_str = "nan";
1058 + value_str = "NAN";
1059
1060 if(unlikely(!flags_str))
1061 flags_str = "";
1062
1063 if (likely(value_str)) {
1024 - RRDDIM_ACQUIRED *rda = rrddim_find_and_acquire(st, dimension);
1064 RRDDIM *rd = rrddim_acquired_to_rrddim(rda);
1026 - if(unlikely(!rd)) {
1027 - error("REPLAY: got a " PLUGINSD_KEYWORD_REPLAY_SET " to dimension '%s' on chart '%s' ('%s') on host '%s', which does not exist. Disabling it.",
1028 - dimension, rrdset_name(st), rrdset_id(st), rrdhost_hostname(st->rrdhost));
1029 - goto disable;
1030 - }
1065
1066 RRDDIM_FLAGS rd_flags = rrddim_flag_check(rd, RRDDIM_FLAG_OBSOLETE | RRDDIM_FLAG_ARCHIVED);
1067
1068 if(unlikely(rd_flags & RRDDIM_FLAG_OBSOLETE)) {
1035 - error("REPLAY: dimension '%s' in chart '%s' has the OBSOLETE flag set, but it is collected.", rrddim_name(rd), rrdset_id(st));
1069 + error("PLUGINSD: 'host:%s/chart:%s/dim:%s' has the OBSOLETE flag set, but it is collected.",
1070 + rrdhost_hostname(st->rrdhost),
1071 + rrdset_id(st),
1072 + rrddim_id(rd)
1073 + );
1074 rrddim_isnot_obsolete(st, rd);
1075 }
1076
@@ -1069,15 +1107,12 @@ PARSER_RC pluginsd_replay_set(char **words, size_t num_words, void *user)
1107 rd->collections_counter++;
1108 }
1109 else
1072 - error("REPLAY: dimension '%s' in chart '%s' has the ARCHIVED flag set, but it is collected. Ignoring data.", rrddim_name(rd), rrdset_id(st));
1073 -
1074 - rrddim_acquired_release(rda);
1110 + error("PLUGINSD: 'host:%s/chart:%s/dim:%s' has the ARCHIVED flag set, but it is collected. Ignoring data.",
1111 + rrdhost_hostname(st->rrdhost), rrdset_id(st), rrddim_name(rd));
1112 }
1076 - return PARSER_RC_OK;
1113
1078 -disable:
1079 - ((PARSER_USER_OBJECT *) user)->enabled = 0;
1080 - return PARSER_RC_ERROR;
1114 + rrddim_acquired_release(rda);
1115 + return PARSER_RC_OK;
1116 }
1117
1118 PARSER_RC pluginsd_replay_rrddim_collection_state(char **words, size_t num_words, void *user)
@@ -1088,29 +1123,16 @@ PARSER_RC pluginsd_replay_rrddim_collection_state(char **words, size_t num_words
1123 char *last_calculated_value_str = get_word(words, num_words, 4);
1124 char *last_stored_value_str = get_word(words, num_words, 5);
1125
1091 - RRDSET *st = ((PARSER_USER_OBJECT *) user)->st;
1092 - RRDHOST *host = ((PARSER_USER_OBJECT *) user)->host;
1126 + RRDHOST *host = pluginsd_require_host_from_parent(user, PLUGINSD_KEYWORD_REPLAY_RRDDIM_STATE);
1127 + if(!host) return PLUGINSD_DISABLE_PLUGIN(user);
1128
1094 - if (unlikely(!st)) {
1095 - error("REPLAY: got a " PLUGINSD_KEYWORD_REPLAY_RRDDIM_STATE " on dimension '%s' on host '%s', without a " PLUGINSD_KEYWORD_REPLAY_BEGIN ". Disabling it.",
1096 - dimension, rrdhost_hostname(host));
1097 - goto disable;
1098 - }
1129 + RRDSET *st = pluginsd_require_chart_from_parent(user, PLUGINSD_KEYWORD_REPLAY_RRDDIM_STATE, PLUGINSD_KEYWORD_REPLAY_BEGIN);
1130 + if(!st) return PLUGINSD_DISABLE_PLUGIN(user);
1131
1100 - if (unlikely(!dimension || !*dimension)) {
1101 - error("REPLAY: got a " PLUGINSD_KEYWORD_REPLAY_RRDDIM_STATE " on chart '%s' of host '%s', without a dimension. Disabling it.",
1102 - rrdset_id(st), rrdhost_hostname(host));
1103 - goto disable;
1104 - }
1132 + RRDDIM_ACQUIRED *rda = pluginsd_acquire_dimension(host, st, dimension, PLUGINSD_KEYWORD_REPLAY_RRDDIM_STATE);
1133 + if(!rda) return PLUGINSD_DISABLE_PLUGIN(user);
1134
1106 - RRDDIM_ACQUIRED *rda = rrddim_find_and_acquire(st, dimension);
1135 RRDDIM *rd = rrddim_acquired_to_rrddim(rda);
1108 - if(unlikely(!rd)) {
1109 - error("REPLAY: got a " PLUGINSD_KEYWORD_REPLAY_RRDDIM_STATE " to dimension with id '%s' on chart '%s' ('%s') on host '%s', which does not exist. Disabling it.",
1110 - dimension, rrdset_name(st), rrdset_id(st), rrdhost_hostname(st->rrdhost));
1111 - goto disable;
1112 - }
1113 -
1136 usec_t dim_last_collected_ut = (usec_t)rd->last_collected_time.tv_sec * USEC_PER_SEC + (usec_t)rd->last_collected_time.tv_usec;
1137 usec_t last_collected_ut = last_collected_ut_str ? str2ull(last_collected_ut_str) : 0;
1138 if(last_collected_ut > dim_last_collected_ut) {
@@ -1123,10 +1145,6 @@ PARSER_RC pluginsd_replay_rrddim_collection_state(char **words, size_t num_words
1145 rd->last_stored_value = last_stored_value_str ? str2ndd(last_stored_value_str, NULL) : 0.0;
1146 rrddim_acquired_release(rda);
1147 return PARSER_RC_OK;
1126 -
1127 -disable:
1128 - ((PARSER_USER_OBJECT *) user)->enabled = 0;
1129 - return PARSER_RC_ERROR;
1148 }
1149
1150 PARSER_RC pluginsd_replay_rrdset_collection_state(char **words, size_t num_words, void *user)
@@ -1134,14 +1152,11 @@ PARSER_RC pluginsd_replay_rrdset_collection_state(char **words, size_t num_words
1152 char *last_collected_ut_str = get_word(words, num_words, 1);
1153 char *last_updated_ut_str = get_word(words, num_words, 2);
1154
1137 - RRDSET *st = ((PARSER_USER_OBJECT *) user)->st;
1138 - RRDHOST *host = ((PARSER_USER_OBJECT *) user)->host;
1155 + RRDHOST *host = pluginsd_require_host_from_parent(user, PLUGINSD_KEYWORD_REPLAY_RRDSET_STATE);
1156 + if(!host) return PLUGINSD_DISABLE_PLUGIN(user);
1157
1140 - if (unlikely(!st)) {
1141 - error("REPLAY: got a " PLUGINSD_KEYWORD_REPLAY_RRDSET_STATE " on host '%s', without a " PLUGINSD_KEYWORD_REPLAY_BEGIN ". Disabling it.",
1142 - rrdhost_hostname(host));
1143 - goto disable;
1144 - }
1158 + RRDSET *st = pluginsd_require_chart_from_parent(user, PLUGINSD_KEYWORD_REPLAY_RRDSET_STATE, PLUGINSD_KEYWORD_REPLAY_BEGIN);
1159 + if(!st) return PLUGINSD_DISABLE_PLUGIN(user);
1160
1161 usec_t chart_last_collected_ut = (usec_t)st->last_collected_time.tv_sec * USEC_PER_SEC + (usec_t)st->last_collected_time.tv_usec;
1162 usec_t last_collected_ut = last_collected_ut_str ? str2ull(last_collected_ut_str) : 0;
@@ -1161,10 +1176,6 @@ PARSER_RC pluginsd_replay_rrdset_collection_state(char **words, size_t num_words
1176 st->counter_done++;
1177
1178 return PARSER_RC_OK;
1164 -
1165 -disable:
1166 - ((PARSER_USER_OBJECT *) user)->enabled = 0;
1167 - return PARSER_RC_ERROR;
1179 }
1180
1181 PARSER_RC pluginsd_replay_end(char **words, size_t num_words, void *user)
@@ -1184,21 +1195,20 @@ PARSER_RC pluginsd_replay_end(char **words, size_t num_words, void *user)
1195
1196 PARSER_USER_OBJECT *user_object = user;
1197
1187 - RRDSET *st = ((PARSER_USER_OBJECT *) user)->st;
1188 - RRDHOST *host = ((PARSER_USER_OBJECT *) user)->host;
1198 + RRDHOST *host = pluginsd_require_host_from_parent(user, PLUGINSD_KEYWORD_REPLAY_END);
1199 + if(!host) return PLUGINSD_DISABLE_PLUGIN(user);
1200
1190 - if (unlikely(!st)) {
1191 - error("REPLAY: got a " PLUGINSD_KEYWORD_REPLAY_END " on host '%s', without a " PLUGINSD_KEYWORD_REPLAY_BEGIN ". Disabling it.",
1192 - rrdhost_hostname(host));
1193 - return PARSER_RC_ERROR;
1194 - }
1201 + RRDSET *st = pluginsd_require_chart_from_parent(user, PLUGINSD_KEYWORD_REPLAY_END, PLUGINSD_KEYWORD_REPLAY_BEGIN);
1202 + if(!st) return PLUGINSD_DISABLE_PLUGIN(user);
1203
1196 -// internal_error(true,
1197 -// "REPLAY: host '%s', chart '%s': received " PLUGINSD_KEYWORD_REPLAY_END " child first_t = %llu, last_t = %llu, start_streaming = %s, requested first_t = %llu, last_t = %llu",
1198 -// rrdhost_hostname(host), rrdset_id(st),
1199 -// (unsigned long long)first_entry_child, (unsigned long long)last_entry_child,
1200 -// start_streaming?"true":"false",
1201 -// (unsigned long long)first_entry_requested, (unsigned long long)last_entry_requested);
1204 +#ifdef NETDATATA_LOG_REPLICATION_REQUESTS
1205 + internal_error(true,
1206 + "PLUGINSD: 'host:%s/chart:%s': received " PLUGINSD_KEYWORD_REPLAY_END " child first_t = %llu, last_t = %llu, start_streaming = %s, requested first_t = %llu, last_t = %llu",
1207 + rrdhost_hostname(host), rrdset_id(st),
1208 + (unsigned long long)first_entry_child, (unsigned long long)last_entry_child,
1209 + start_streaming?"true":"false",
1210 + (unsigned long long)first_entry_requested, (unsigned long long)last_entry_requested);
1211 +#endif
1212
1213 ((PARSER_USER_OBJECT *) user)->st = NULL;
1214 ((PARSER_USER_OBJECT *) user)->count++;
@@ -1222,7 +1232,7 @@ PARSER_RC pluginsd_replay_end(char **words, size_t num_words, void *user)
1232 st->counter++;
1233 st->counter_done++;
1234
1225 -#ifdef NETDATA_INTERNAL_CHECKS
1235 +#ifdef NETDATA_LOG_REPLICATION_REQUESTS
1236 st->replay.start_streaming = false;
1237 st->replay.after = 0;
1238 st->replay.before = 0;
@@ -1238,10 +1248,11 @@ PARSER_RC pluginsd_replay_end(char **words, size_t num_words, void *user)
1248 rrdset_flag_clear(st, RRDSET_FLAG_SYNC_CLOCK);
1249 rrdhost_receiver_replicating_charts_minus_one(st->rrdhost);
1250 }
1251 +#ifdef NETDATA_LOG_REPLICATION_REQUESTS
1252 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.",
1253 + internal_error(true, "REPLAY: 'host:%s/chart:%s' got a " PLUGINSD_KEYWORD_REPLAY_END " with enable_streaming = true, but there is no replication in progress for this chart.",
1254 rrdhost_hostname(host), rrdset_id(st));
1244 -
1255 +#endif
1256 worker_set_metric(WORKER_RECEIVER_JOB_REPLICATION_COMPLETION, 100.0);
1257
1258 return PARSER_RC_OK;
daemon/global_statistics.c
+10 -6
@@ -1225,6 +1225,7 @@ struct dictionary_categories {
1225 RRDDIM *rd_spins_use;
1226 RRDDIM *rd_spins_search;
1227 RRDDIM *rd_spins_insert;
1228 + RRDDIM *rd_spins_delete;
1229
1230 } dictionary_categories[] = {
1231 { .stats = &dictionary_stats_category_other, "dictionaries", "dictionaries", 900000 },
@@ -1481,9 +1482,10 @@ static void update_dictionary_category_charts(struct dictionary_categories *c) {
1482 // ------------------------------------------------------------------------
1483
1484 total = 0;
1484 - load_dictionary_stats_entry(spin_locks.use);
1485 - load_dictionary_stats_entry(spin_locks.search);
1486 - load_dictionary_stats_entry(spin_locks.insert);
1485 + load_dictionary_stats_entry(spin_locks.use_spins);
1486 + load_dictionary_stats_entry(spin_locks.search_spins);
1487 + load_dictionary_stats_entry(spin_locks.insert_spins);
1488 + load_dictionary_stats_entry(spin_locks.delete_spins);
1489
1490 if(c->st_spins || total != 0) {
1491 if (unlikely(!c->st_spins)) {
@@ -1511,13 +1513,15 @@ static void update_dictionary_category_charts(struct dictionary_categories *c) {
1513 c->rd_spins_use = rrddim_add(c->st_spins, "use", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
1514 c->rd_spins_search = rrddim_add(c->st_spins, "search", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
1515 c->rd_spins_insert = rrddim_add(c->st_spins, "insert", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
1516 + c->rd_spins_delete = rrddim_add(c->st_spins, "delete", NULL, 1, 1, RRD_ALGORITHM_INCREMENTAL);
1517
1518 rrdlabels_add(c->st_spins->rrdlabels, "category", stats.name, RRDLABEL_SRC_AUTO);
1519 }
1520
1518 - rrddim_set_by_pointer(c->st_spins, c->rd_spins_use, (collected_number)stats.spin_locks.use);
1519 - rrddim_set_by_pointer(c->st_spins, c->rd_spins_search, (collected_number)stats.spin_locks.search);
1520 - rrddim_set_by_pointer(c->st_spins, c->rd_spins_insert, (collected_number)stats.spin_locks.insert);
1521 + rrddim_set_by_pointer(c->st_spins, c->rd_spins_use, (collected_number)stats.spin_locks.use_spins);
1522 + rrddim_set_by_pointer(c->st_spins, c->rd_spins_search, (collected_number)stats.spin_locks.search_spins);
1523 + rrddim_set_by_pointer(c->st_spins, c->rd_spins_insert, (collected_number)stats.spin_locks.insert_spins);
1524 + rrddim_set_by_pointer(c->st_spins, c->rd_spins_delete, (collected_number)stats.spin_locks.delete_spins);
1525
1526 rrdset_done(c->st_spins);
1527 }
database/rrd.h
+19 -9
@@ -314,6 +314,12 @@ struct rrddim {
314 collected_number collected_value; // the current value, as collected - resets to 0 after being used
315 collected_number last_collected_value; // the last value that was collected, after being processed
316
317 +#ifdef NETDATA_LOG_COLLECTION_ERRORS
318 + usec_t rrddim_store_metric_last_ut; // the timestamp we last called rrddim_store_metric()
319 + size_t rrddim_store_metric_count; // the rrddim_store_metric() counter
320 + const char *rrddim_store_metric_last_caller; // the name of the function that last called rrddim_store_metric()
321 +#endif
322 +
323 // ------------------------------------------------------------------------
324 // db mode RAM, SAVE, MAP, ALLOC, NONE specifics
325 // TODO - they should be managed by storage engine
@@ -532,20 +538,19 @@ typedef enum rrdset_flags {
538
539 RRDSET_FLAG_PENDING_HEALTH_INITIALIZATION = (1 << 21),
540
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
541 + RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS = (1 << 22), // the sending side has replication in progress
542 + RRDSET_FLAG_SENDER_REPLICATION_FINISHED = (1 << 23), // the sending side has completed replication
543 + RRDSET_FLAG_RECEIVER_REPLICATION_IN_PROGRESS = (1 << 24), // the receiving side has replication in progress
544 + RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED = (1 << 25), // the receiving side has completed replication
545
541 - RRDSET_FLAG_UPSTREAM_SEND_VARIABLES = (1 << 27), // a custom variable has been updated and needs to be exposed to parent
546 + RRDSET_FLAG_UPSTREAM_SEND_VARIABLES = (1 << 26), // a custom variable has been updated and needs to be exposed to parent
547 } RRDSET_FLAGS;
548
549 #define rrdset_flag_check(st, flag) (__atomic_load_n(&((st)->flags), __ATOMIC_SEQ_CST) & (flag))
550 #define rrdset_flag_set(st, flag) __atomic_or_fetch(&((st)->flags), flag, __ATOMIC_SEQ_CST)
551 #define rrdset_flag_clear(st, flag) __atomic_and_fetch(&((st)->flags), ~(flag), __ATOMIC_SEQ_CST)
552
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) \
553 +#define rrdset_is_replicating(st) (rrdset_flag_check(st, RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS|RRDSET_FLAG_RECEIVER_REPLICATION_IN_PROGRESS) \
554 && !rrdset_flag_check(st, RRDSET_FLAG_SENDER_REPLICATION_FINISHED|RRDSET_FLAG_RECEIVER_REPLICATION_FINISHED))
555
556 struct rrdset {
@@ -666,13 +671,13 @@ struct rrdset {
671 RRDCALC *base; // double linked list of RRDCALC related to this RRDSET
672 } alerts;
673
669 -#ifdef NETDATA_INTERNAL_CHECKS
674 +#ifdef NETDATA_LOG_REPLICATION_REQUESTS
675 struct {
676 bool start_streaming;
677 time_t after;
678 time_t before;
679 } replay;
675 -#endif
680 +#endif // NETDATA_LOG_REPLICATION_REQUESTS
681 };
682
683 #define rrdset_plugin_name(st) string2str((st)->plugin_name)
@@ -1330,7 +1335,12 @@ time_t calc_dimension_liveness(RRDDIM *rd, time_t now);
1335 #endif
1336 long align_entries_to_pagesize(RRD_MEMORY_MODE mode, long entries);
1337
1338 +#ifdef NETDATA_LOG_COLLECTION_ERRORS
1339 +#define rrddim_store_metric(rd, point_end_time_ut, n, flags) rrddim_store_metric_with_trace(rd, point_end_time_ut, n, flags, __FUNCTION__)
1340 +void rrddim_store_metric_with_trace(RRDDIM *rd, usec_t point_end_time_ut, NETDATA_DOUBLE n, SN_FLAGS flags, const char *function);
1341 +#else
1342 void rrddim_store_metric(RRDDIM *rd, usec_t point_end_time_ut, NETDATA_DOUBLE n, SN_FLAGS flags);
1343 +#endif
1344
1345 // ----------------------------------------------------------------------------
1346 // Miscellaneous functions
database/rrdset.c
+29 -1
@@ -1109,8 +1109,36 @@ void store_metric_at_tier(RRDDIM *rd, struct rrddim_tier *t, STORAGE_POINT sp, u
1109 }
1110 }
1111 }
1112 -
1112 +#ifdef NETDATA_LOG_COLLECTION_ERRORS
1113 +void rrddim_store_metric_with_trace(RRDDIM *rd, usec_t point_end_time_ut, NETDATA_DOUBLE n, SN_FLAGS flags, const char *function) {
1114 +#else // !NETDATA_LOG_COLLECTION_ERRORS
1115 void rrddim_store_metric(RRDDIM *rd, usec_t point_end_time_ut, NETDATA_DOUBLE n, SN_FLAGS flags) {
1116 +#endif // !NETDATA_LOG_COLLECTION_ERRORS
1117 +#ifdef NETDATA_LOG_COLLECTION_ERRORS
1118 + rd->rrddim_store_metric_count++;
1119 +
1120 + if(likely(rd->rrddim_store_metric_count > 1)) {
1121 + usec_t expected = rd->rrddim_store_metric_last_ut + rd->update_every * USEC_PER_SEC;
1122 +
1123 + if(point_end_time_ut != rd->rrddim_store_metric_last_ut) {
1124 + internal_error(true,
1125 + "%s COLLECTION: 'host:%s/chart:%s/dim:%s' granularity %d, collection %zu, expected to store at tier 0 a value at %llu, but it gave %llu [%s%llu usec] (called from %s(), previously by %s())",
1126 + (point_end_time_ut < rd->rrddim_store_metric_last_ut) ? "**PAST**" : "GAP",
1127 + rrdhost_hostname(rd->rrdset->rrdhost), rrdset_id(rd->rrdset), rrddim_id(rd),
1128 + rd->update_every,
1129 + rd->rrddim_store_metric_count,
1130 + expected, point_end_time_ut,
1131 + (point_end_time_ut < rd->rrddim_store_metric_last_ut)?"by -" : "gap ",
1132 + expected - point_end_time_ut,
1133 + function,
1134 + rd->rrddim_store_metric_last_caller?rd->rrddim_store_metric_last_caller:"none");
1135 + }
1136 + }
1137 +
1138 + rd->rrddim_store_metric_last_ut = point_end_time_ut;
1139 + rd->rrddim_store_metric_last_caller = function;
1140 +#endif // NETDATA_LOG_COLLECTION_ERRORS
1141 +
1142 // store the metric on tier 0
1143 rd->tiers[0]->collect_ops->store_metric(rd->tiers[0]->db_collection_handle, point_end_time_ut, n, 0, 0, 1, 0, flags);
1144
libnetdata/dictionary/dictionary.c
+215 -98
@@ -32,7 +32,8 @@ typedef enum item_options {
32
33 typedef enum item_flags {
34 ITEM_FLAG_NONE = 0,
35 - ITEM_FLAG_DELETED = (1 << 0), // this item is deleted, so it is not available for traversal
35 + ITEM_FLAG_DELETED = (1 << 0), // this item is marked deleted, so it is not available for traversal (deleted from the index too)
36 + ITEM_FLAG_BEING_CREATED = (1 << 1), // this item is currently being created - this flag is removed when construction finishes
37
38 // IMPORTANT: This is 8-bit
39 } ITEM_FLAGS;
@@ -175,15 +176,19 @@ static void garbage_collect_pending_deletes(DICTIONARY *dict);
176 static inline void item_linked_list_remove(DICTIONARY *dict, DICTIONARY_ITEM *item);
177 static size_t dict_item_free_with_hooks(DICTIONARY *dict, DICTIONARY_ITEM *item);
178 static inline const char *item_get_name(const DICTIONARY_ITEM *item);
178 -static bool item_is_not_referenced_and_can_be_removed(DICTIONARY *dict, DICTIONARY_ITEM *item);
179 static inline int hashtable_delete_unsafe(DICTIONARY *dict, const char *name, size_t name_len, void *item);
180 static void item_release(DICTIONARY *dict, DICTIONARY_ITEM *item);
181 -
182 -#define ITEM_OK 0
183 -#define ITEM_MARKED_FOR_DELETION (-1) // the item is marked for deletion
184 -#define ITEM_IS_CURRENTLY_BEING_DELETED (-2) // the item is currently being deleted
185 -#define item_check_and_acquire(dict, item) (item_check_and_acquire_advanced(dict, item, false) == ITEM_OK)
181 +static bool dict_item_set_deleted(DICTIONARY *dict, DICTIONARY_ITEM *item);
182 +
183 +#define RC_ITEM_OK ( 0)
184 +#define RC_ITEM_MARKED_FOR_DELETION (-1) // the item is marked for deletion
185 +#define RC_ITEM_IS_CURRENTLY_BEING_DELETED (-2) // the item is currently being deleted
186 +#define RC_ITEM_IS_CURRENTLY_BEING_CREATED (-3) // the item is currently being deleted
187 +#define RC_ITEM_IS_REFERENCED (-4) // the item is currently referenced
188 +#define item_check_and_acquire(dict, item) (item_check_and_acquire_advanced(dict, item, false) == RC_ITEM_OK)
189 static int item_check_and_acquire_advanced(DICTIONARY *dict, DICTIONARY_ITEM *item, bool having_index_lock);
190 +#define item_is_not_referenced_and_can_be_removed(dict, item) (item_is_not_referenced_and_can_be_removed_advanced(dict, item) == RC_ITEM_OK)
191 +static inline int item_is_not_referenced_and_can_be_removed_advanced(DICTIONARY *dict, DICTIONARY_ITEM *item);
192
193 // ----------------------------------------------------------------------------
194 // memory statistics
@@ -345,14 +350,23 @@ static inline void DICTIONARY_ENTRIES_MINUS1(DICTIONARY *dict) {
350 __atomic_fetch_add(&dict->stats->ops.deletes, 1, __ATOMIC_RELAXED);
351 __atomic_fetch_sub(&dict->stats->items.entries, 1, __ATOMIC_RELAXED);
352
353 + size_t entries;
354 if(unlikely(is_dictionary_single_threaded(dict))) {
355 dict->version++;
350 - dict->entries--;
356 + entries = dict->entries++;
357 }
358 else {
359 __atomic_fetch_add(&dict->version, 1, __ATOMIC_SEQ_CST);
354 - __atomic_fetch_sub(&dict->entries, 1, __ATOMIC_SEQ_CST);
360 + entries = __atomic_fetch_sub(&dict->entries, 1, __ATOMIC_SEQ_CST);
361 }
362 +
363 +#ifdef NETDATA_INTERNAL_CHECKS
364 + if(unlikely(entries == 0))
365 + fatal("DICT: negative number of entries in dictionary created from %s() (%zu@%s)",
366 + dict->creation_function,
367 + dict->creation_line,
368 + dict->creation_file);
369 +#endif
370 }
371 static inline void DICTIONARY_VALUE_RESETS_PLUS1(DICTIONARY *dict) {
372 __atomic_fetch_add(&dict->stats->ops.resets, 1, __ATOMIC_RELAXED);
@@ -369,13 +383,16 @@ static inline void DICTIONARY_STATS_WALKTHROUGHS_PLUS1(DICTIONARY *dict) {
383 __atomic_fetch_add(&dict->stats->ops.walkthroughs, 1, __ATOMIC_RELAXED);
384 }
385 static inline void DICTIONARY_STATS_CHECK_SPINS_PLUS(DICTIONARY *dict, size_t count) {
372 - __atomic_fetch_add(&dict->stats->spin_locks.use, count, __ATOMIC_RELAXED);
386 + __atomic_fetch_add(&dict->stats->spin_locks.use_spins, count, __ATOMIC_RELAXED);
387 }
388 static inline void DICTIONARY_STATS_INSERT_SPINS_PLUS(DICTIONARY *dict, size_t count) {
375 - __atomic_fetch_add(&dict->stats->spin_locks.insert, count, __ATOMIC_RELAXED);
389 + __atomic_fetch_add(&dict->stats->spin_locks.insert_spins, count, __ATOMIC_RELAXED);
390 +}
391 +static inline void DICTIONARY_STATS_DELETE_SPINS_PLUS(DICTIONARY *dict, size_t count) {
392 + __atomic_fetch_add(&dict->stats->spin_locks.delete_spins, count, __ATOMIC_RELAXED);
393 }
394 static inline void DICTIONARY_STATS_SEARCH_IGNORES_PLUS1(DICTIONARY *dict) {
378 - __atomic_fetch_add(&dict->stats->spin_locks.search, 1, __ATOMIC_RELAXED);
395 + __atomic_fetch_add(&dict->stats->spin_locks.search_spins, 1, __ATOMIC_RELAXED);
396 }
397 static inline void DICTIONARY_STATS_CALLBACK_INSERTS_PLUS1(DICTIONARY *dict) {
398 __atomic_fetch_add(&dict->stats->callbacks.inserts, 1, __ATOMIC_RELAXED);
@@ -422,10 +439,22 @@ static inline long int DICTIONARY_REFERENCED_ITEMS_PLUS1(DICTIONARY *dict) {
439 static inline long int DICTIONARY_REFERENCED_ITEMS_MINUS1(DICTIONARY *dict) {
440 __atomic_fetch_sub(&dict->stats->items.referenced, 1, __ATOMIC_RELAXED);
441
442 + long int referenced_items;
443 if(unlikely(is_dictionary_single_threaded(dict)))
426 - return --dict->referenced_items;
444 + referenced_items = --dict->referenced_items;
445 else
428 - return __atomic_sub_fetch(&dict->referenced_items, 1, __ATOMIC_SEQ_CST);
446 + referenced_items = __atomic_sub_fetch(&dict->referenced_items, 1, __ATOMIC_SEQ_CST);
447 +
448 +#ifdef NETDATA_INTERNAL_CHECKS
449 + if(unlikely(referenced_items < 0))
450 + fatal("DICT: negative number of referenced items (%ld) in dictionary created from %s() (%zu@%s)",
451 + referenced_items,
452 + dict->creation_function,
453 + dict->creation_line,
454 + dict->creation_file);
455 +#endif
456 +
457 + return referenced_items;
458 }
459
460 static inline long int DICTIONARY_PENDING_DELETES_PLUS1(DICTIONARY *dict) {
@@ -683,8 +712,8 @@ static void garbage_collect_pending_deletes(DICTIONARY *dict) {
712 item_next = item->next;
713 int rc = item_check_and_acquire_advanced(dict, item, is_view);
714
686 - if(rc == ITEM_MARKED_FOR_DELETION) {
687 - // we don't have got a reference
715 + if(rc == RC_ITEM_MARKED_FOR_DELETION) {
716 + // we didn't get a reference
717
718 if(item_is_not_referenced_and_can_be_removed(dict, item)) {
719 DOUBLE_LINKED_LIST_REMOVE_UNSAFE(dict->items.list, item, prev, next);
@@ -696,10 +725,10 @@ static void garbage_collect_pending_deletes(DICTIONARY *dict) {
725 break;
726 }
727 }
699 - else if(rc == ITEM_IS_CURRENTLY_BEING_DELETED)
700 - ; // do not touch this item (we haven't got a reference)
728 + else if(rc == RC_ITEM_IS_CURRENTLY_BEING_DELETED)
729 + ; // do not touch this item (we didn't get a reference)
730
702 - else if(rc == ITEM_OK)
731 + else if(rc == RC_ITEM_OK)
732 item_release(dict, item);
733
734 item = item_next;
@@ -829,6 +858,8 @@ static int item_check_and_acquire_advanced(DICTIONARY *dict, DICTIONARY_ITEM *it
858 size_t spins = 0;
859 REFCOUNT refcount, desired;
860
861 + int ret = RC_ITEM_OK;
862 +
863 do {
864 spins++;
865
@@ -836,12 +867,14 @@ static int item_check_and_acquire_advanced(DICTIONARY *dict, DICTIONARY_ITEM *it
867
868 if(refcount < 0) {
869 // we can't use this item
839 - return ITEM_IS_CURRENTLY_BEING_DELETED;
870 + ret = RC_ITEM_IS_CURRENTLY_BEING_DELETED;
871 + break;
872 }
873
874 if(item_flag_check(item, ITEM_FLAG_DELETED)) {
875 // we can't use this item
844 - return ITEM_MARKED_FOR_DELETION;
876 + ret = RC_ITEM_MARKED_FOR_DELETION;
877 + break;
878 }
879
880 desired = refcount + 1;
@@ -849,63 +882,89 @@ static int item_check_and_acquire_advanced(DICTIONARY *dict, DICTIONARY_ITEM *it
882 } while(!__atomic_compare_exchange_n(&item->refcount, &refcount, desired,
883 false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST));
884
852 - // we acquired the item
885 + // if ret == ITEM_OK, we acquired the item
886
854 - if(is_view_dictionary(dict) && item_shared_flag_check(item, ITEM_FLAG_DELETED) && !item_flag_check(item, ITEM_FLAG_DELETED)) {
855 - // but, we can't use this item
887 + if(ret == RC_ITEM_OK) {
888 + if (is_view_dictionary(dict) &&
889 + item_shared_flag_check(item, ITEM_FLAG_DELETED) &&
890 + !item_flag_check(item, ITEM_FLAG_DELETED)) {
891 + // but, we can't use this item
892
857 - if(having_index_lock) {
858 - // delete it from the hashtable
859 - hashtable_delete_unsafe(dict, item_get_name(item), item->key_len, item);
893 + if (having_index_lock) {
894 + // delete it from the hashtable
895 + hashtable_delete_unsafe(dict, item_get_name(item), item->key_len, item);
896
861 - // mark it in our dictionary as deleted too
862 - // this is safe to be done here, because we have got
863 - // a reference counter on item
864 - item_flag_set(item, ITEM_FLAG_DELETED);
897 + // mark it in our dictionary as deleted too
898 + // this is safe to be done here, because we have got
899 + // a reference counter on item
900 + dict_item_set_deleted(dict, item);
901
866 - DICTIONARY_ENTRIES_MINUS1(dict);
902 + // decrement the refcount we incremented above
903 + if (__atomic_sub_fetch(&item->refcount, 1, __ATOMIC_SEQ_CST) == 0) {
904 + // this is a deleted item, and we are the last one
905 + DICTIONARY_PENDING_DELETES_PLUS1(dict);
906 + }
907
868 - // decrement the refcount we incremented above
869 - if (__atomic_sub_fetch(&item->refcount, 1, __ATOMIC_SEQ_CST) == 0) {
870 - // this is a deleted item, and we are the last one
871 - DICTIONARY_PENDING_DELETES_PLUS1(dict);
908 + // do not touch the item below this point
909 + } else {
910 + // this is traversal / walkthrough
911 + // decrement the refcount we incremented above
912 + __atomic_sub_fetch(&item->refcount, 1, __ATOMIC_SEQ_CST);
913 }
914
874 - // do not touch the item below this point
875 - }
876 - else {
877 - // this is traversal / walkthrough
878 - // decrement the refcount we incremented above
879 - __atomic_sub_fetch(&item->refcount, 1, __ATOMIC_SEQ_CST);
915 + return RC_ITEM_MARKED_FOR_DELETION;
916 }
917
882 - return ITEM_MARKED_FOR_DELETION;
918 + if(desired == 1)
919 + DICTIONARY_REFERENCED_ITEMS_PLUS1(dict);
920 }
921
885 - if(desired == 1)
886 - DICTIONARY_REFERENCED_ITEMS_PLUS1(dict);
922
888 - if(unlikely(spins > 2 && dict->stats))
889 - DICTIONARY_STATS_CHECK_SPINS_PLUS(dict, spins - 2);
923 + if(unlikely(spins > 1 && dict->stats))
924 + DICTIONARY_STATS_CHECK_SPINS_PLUS(dict, spins - 1);
925
891 - return ITEM_OK; // we can use this item
926 + return ret;
927 }
928
929 // if a dictionary item can be deleted, return true, otherwise return false
930 // we use the private reference counter
896 -static inline bool item_is_not_referenced_and_can_be_removed(DICTIONARY *dict, DICTIONARY_ITEM *item) {
931 +static inline int item_is_not_referenced_and_can_be_removed_advanced(DICTIONARY *dict, DICTIONARY_ITEM *item) {
932 // if we can set refcount to REFCOUNT_DELETING, we can delete this item
933
899 - REFCOUNT expected = DICTIONARY_ITEM_REFCOUNT_GET(dict, item);
900 - if(expected == 0 && __atomic_compare_exchange_n(&item->refcount, &expected, REFCOUNT_DELETING,
901 - false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST)) {
934 + size_t spins = 0;
935 + REFCOUNT refcount, desired = REFCOUNT_DELETING;
936
903 - // we are going to delete it
904 - return true;
905 - }
937 + int ret = RC_ITEM_OK;
938
907 - // we can't delete this
908 - return false;
939 + do {
940 + spins++;
941 +
942 + refcount = DICTIONARY_ITEM_REFCOUNT_GET(dict, item);
943 +
944 + if(refcount < 0) {
945 + // we can't use this item
946 + ret = RC_ITEM_IS_CURRENTLY_BEING_DELETED;
947 + break;
948 + }
949 +
950 + if(refcount > 0) {
951 + // we can't delete this
952 + ret = RC_ITEM_IS_REFERENCED;
953 + break;
954 + }
955 +
956 + if(item_flag_check(item, ITEM_FLAG_BEING_CREATED)) {
957 + // we can't use this item
958 + ret = RC_ITEM_IS_CURRENTLY_BEING_CREATED;
959 + break;
960 + }
961 + } while(!__atomic_compare_exchange_n(&item->refcount, &refcount, desired,
962 + false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST));
963 +
964 + if(unlikely(spins > 1 && dict->stats))
965 + DICTIONARY_STATS_DELETE_SPINS_PLUS(dict, spins - 1);
966 +
967 + return ret;
968 }
969
970 // if a dictionary item can be freed, return true, otherwise return false
@@ -954,7 +1013,7 @@ static inline void **hashtable_insert_unsafe(DICTIONARY *dict, const char *name,
1013 JError_t J_Error;
1014 Pvoid_t *Rc = JudyHSIns(&dict->index.JudyHSArray, (void *)name, name_len, &J_Error);
1015 if (unlikely(Rc == PJERR)) {
957 - fatal("DICTIONARY: Cannot insert entry with name '%s' to JudyHS, JU_ERRNO_* == %u, ID == %d",
1016 + error("DICTIONARY: Cannot insert entry with name '%s' to JudyHS, JU_ERRNO_* == %u, ID == %d",
1017 name, JU_ERRNO(&J_Error), JU_ERRID(&J_Error));
1018 }
1019
@@ -1031,6 +1090,10 @@ static inline void item_linked_list_add(DICTIONARY *dict, DICTIONARY_ITEM *item)
1090 else
1091 DOUBLE_LINKED_LIST_APPEND_UNSAFE(dict->items.list, item, prev, next);
1092
1093 + // clear the BEING created flag,
1094 + // after it has been inserted into the linked list
1095 + item_flag_clear(item, ITEM_FLAG_BEING_CREATED);
1096 +
1097 garbage_collect_pending_deletes(dict);
1098 ll_recursive_unlock(dict, DICTIONARY_LOCK_WRITE);
1099 }
@@ -1088,6 +1151,7 @@ static DICTIONARY_ITEM *dict_item_create(DICTIONARY *dict __maybe_unused, size_t
1151 size_t size = sizeof(DICTIONARY_ITEM);
1152 item = callocz(1, size);
1153 item->refcount = 1;
1154 + item->flags = ITEM_FLAG_BEING_CREATED;
1155 *allocated_bytes += size;
1156
1157 if(master_item) {
@@ -1218,6 +1282,9 @@ static void dict_item_reset_value_with_hooks(DICTIONARY *dict, DICTIONARY_ITEM *
1282 static size_t dict_item_free_with_hooks(DICTIONARY *dict, DICTIONARY_ITEM *item) {
1283 debug(D_DICTIONARY, "Destroying name value entry for name '%s'.", item_get_name(item));
1284
1285 + if(!item_flag_check(item, ITEM_FLAG_DELETED))
1286 + DICTIONARY_ENTRIES_MINUS1(dict);
1287 +
1288 size_t item_size = 0, key_size = 0, value_size = 0;
1289
1290 key_size += item->key_len;
@@ -1260,20 +1327,51 @@ static void dict_item_shared_set_deleted(DICTIONARY *dict, DICTIONARY_ITEM *item
1327 }
1328 }
1329
1263 -static inline void dict_item_free_or_mark_deleted(DICTIONARY *dict, DICTIONARY_ITEM *item) {
1264 - if(item_is_not_referenced_and_can_be_removed(dict, item)) {
1265 - dict_item_shared_set_deleted(dict, item);
1266 - item_linked_list_remove(dict, item);
1267 - dict_item_free_with_hooks(dict, item);
1268 - }
1269 - else {
1270 - dict_item_shared_set_deleted(dict, item);
1271 - item_flag_set(item, ITEM_FLAG_DELETED);
1272 - // after this point do not touch the item
1273 - }
1330 +// returns true if we set the deleted flag on this item
1331 +static bool dict_item_set_deleted(DICTIONARY *dict, DICTIONARY_ITEM *item) {
1332 + ITEM_FLAGS expected, desired;
1333 +
1334 + do {
1335 + expected = __atomic_load_n(&item->flags, __ATOMIC_SEQ_CST);
1336 +
1337 + if (expected & ITEM_FLAG_DELETED)
1338 + return false;
1339 +
1340 + desired = expected | ITEM_FLAG_DELETED;
1341 +
1342 + } while(!__atomic_compare_exchange_n(&item->flags, &expected, desired,
1343 + false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST));
1344
1275 - // the item is not available anymore
1345 DICTIONARY_ENTRIES_MINUS1(dict);
1346 + return true;
1347 +}
1348 +
1349 +static inline void dict_item_free_or_mark_deleted(DICTIONARY *dict, DICTIONARY_ITEM *item) {
1350 + int rc = item_is_not_referenced_and_can_be_removed_advanced(dict, item);
1351 + switch(rc) {
1352 + case RC_ITEM_OK:
1353 + // the item is ours, refcount set to -100
1354 + dict_item_shared_set_deleted(dict, item);
1355 + item_linked_list_remove(dict, item);
1356 + dict_item_free_with_hooks(dict, item);
1357 + break;
1358 +
1359 + case RC_ITEM_IS_REFERENCED:
1360 + case RC_ITEM_IS_CURRENTLY_BEING_CREATED:
1361 + // the item is currently referenced by others
1362 + dict_item_shared_set_deleted(dict, item);
1363 + dict_item_set_deleted(dict, item);
1364 + // after this point do not touch the item
1365 + break;
1366 +
1367 + case RC_ITEM_IS_CURRENTLY_BEING_DELETED:
1368 + // an item that is currently being deleted by someone else - don't touch it
1369 + break;
1370 +
1371 + default:
1372 + internal_error(true, "Hey dev! You forgot to add the new condition here!");
1373 + break;
1374 + }
1375 }
1376
1377 // this is used by traversal functions to remove the current item
@@ -1404,10 +1502,11 @@ static DICTIONARY_ITEM *dict_item_add_or_reset_value_and_acquire(DICTIONARY *dic
1502 dictionary_index_lock_unlock(dict);
1503
1504 item_linked_list_add(dict, item);
1505 +
1506 added_or_updated = true;
1507 }
1508 else {
1410 - if(item_check_and_acquire_advanced(dict, *item_pptr, true) != ITEM_OK) {
1509 + if(item_check_and_acquire_advanced(dict, *item_pptr, true) != RC_ITEM_OK) {
1510 spins++;
1511 continue;
1512 }
@@ -1443,6 +1542,7 @@ static DICTIONARY_ITEM *dict_item_add_or_reset_value_and_acquire(DICTIONARY *dic
1542 }
1543
1544 else {
1545 + // conflict callback returned false
1546 // we did really nothing!
1547 ;
1548 }
@@ -1530,11 +1630,10 @@ static bool dictionary_free_all_resources(DICTIONARY *dict, size_t *mem, bool fo
1630 // cache item->next
1631 // because we are going to free item
1632 DICTIONARY_ITEM *item_next = item->next;
1633 +
1634 item_size += dict_item_free_with_hooks(dict, item);
1635 item = item_next;
1636
1536 - DICTIONARY_ENTRIES_MINUS1(dict);
1537 -
1637 // to speed up destruction, we don't
1638 // unlink item from the linked-list here
1639
@@ -1801,22 +1900,11 @@ void dictionary_flush(DICTIONARY *dict) {
1900 if(unlikely(!dict))
1901 return;
1902
1804 -// // delete the index
1805 -// dictionary_index_lock_wrlock(dict);
1806 -// hashtable_destroy_unsafe(dict);
1807 -// dictionary_index_lock_unlock(dict);
1808 -
1809 - // delete all items
1810 - ll_recursive_lock(dict, DICTIONARY_LOCK_WRITE); // get write lock here, to speed it up (it is recursive)
1811 - DICTIONARY_ITEM *item, *item_next;
1812 - for (item = dict->items.list; item; item = item_next) {
1813 - item_next = item->next;
1814 -
1815 -// if(!item_flag_check(item, ITEM_FLAG_DELETED))
1816 -// dict_item_free_or_mark_deleted(dict, item);
1817 - dict_item_del(dict, item_get_name(item), (ssize_t)item_get_name_len(item));
1903 + void *value;
1904 + dfe_start_write(dict, value) {
1905 + dictionary_del_advanced(dict, item_get_name(value_dfe.item), (ssize_t)item_get_name_len(value_dfe.item) + 1);
1906 }
1819 - ll_recursive_unlock(dict, DICTIONARY_LOCK_WRITE);
1907 + dfe_done(value);
1908
1909 DICTIONARY_STATS_DICT_FLUSHES_PLUS1(dict);
1910 }
@@ -2905,8 +2993,29 @@ static void *unittest_dict_thread(void *arg) {
2993 }
2994
2995 dictionary_acquired_item_release(tu->dict, item);
2908 -
2996 dictionary_del(tu->dict, "dict thread checking 1234567890");
2997 +
2998 + // test concurrent deletions and flushes
2999 + {
3000 + if(gettid() % 2) {
3001 + char buf [256 + 1];
3002 +
3003 + for (int i = 0; i < 1000; i++) {
3004 + snprintfz(buf, 256, "del/flush test %d", i);
3005 + dictionary_set(tu->dict, buf, NULL, 0);
3006 + }
3007 +
3008 + for (int i = 0; i < 1000; i++) {
3009 + snprintfz(buf, 256, "del/flush test %d", i);
3010 + dictionary_del(tu->dict, buf);
3011 + }
3012 + }
3013 + else {
3014 + for (int i = 0; i < 10; i++) {
3015 + dictionary_flush(tu->dict);
3016 + }
3017 + }
3018 + }
3019 }
3020
3021 return arg;
@@ -2955,23 +3064,27 @@ static int dictionary_unittest_threads() {
3064 ", deletes %zu"
3065 ", searches %zu"
3066 ", resets %zu"
3067 + ", flushes %zu"
3068 ", entries %ld"
3069 ", referenced_items %ld"
3070 ", pending deletions %ld"
3071 ", check spins %zu"
3072 ", insert spins %zu"
3073 + ", delete spins %zu"
3074 ", search ignores %zu"
3075 "\n",
3076 tu.dict->stats->ops.inserts,
3077 tu.dict->stats->ops.deletes,
3078 tu.dict->stats->ops.searches,
3079 tu.dict->stats->ops.resets,
3080 + tu.dict->stats->ops.flushes,
3081 tu.dict->entries,
3082 tu.dict->referenced_items,
3083 tu.dict->pending_deletion_items,
2972 - tu.dict->stats->spin_locks.use,
2973 - tu.dict->stats->spin_locks.insert,
2974 - tu.dict->stats->spin_locks.search
3084 + tu.dict->stats->spin_locks.use_spins,
3085 + tu.dict->stats->spin_locks.insert_spins,
3086 + tu.dict->stats->spin_locks.delete_spins,
3087 + tu.dict->stats->spin_locks.search_spins
3088 );
3089 dictionary_destroy(tu.dict);
3090 tu.dict = NULL;
@@ -3118,6 +3231,7 @@ static int dictionary_unittest_view_threads() {
3231 ", pending deletions %ld"
3232 ", check spins %zu"
3233 ", insert spins %zu"
3234 + ", delete spins %zu"
3235 ", search ignores %zu"
3236 "\n",
3237 stats_master.ops.inserts,
@@ -3127,9 +3241,10 @@ static int dictionary_unittest_view_threads() {
3241 tv.master->entries,
3242 tv.master->referenced_items,
3243 tv.master->pending_deletion_items,
3130 - stats_master.spin_locks.use,
3131 - stats_master.spin_locks.insert,
3132 - stats_master.spin_locks.search
3244 + stats_master.spin_locks.use_spins,
3245 + stats_master.spin_locks.insert_spins,
3246 + stats_master.spin_locks.delete_spins,
3247 + stats_master.spin_locks.search_spins
3248 );
3249 fprintf(stderr,
3250 "VIEW : inserts %zu"
@@ -3141,6 +3256,7 @@ static int dictionary_unittest_view_threads() {
3256 ", pending deletions %ld"
3257 ", check spins %zu"
3258 ", insert spins %zu"
3259 + ", delete spins %zu"
3260 ", search ignores %zu"
3261 "\n",
3262 stats_view.ops.inserts,
@@ -3150,9 +3266,10 @@ static int dictionary_unittest_view_threads() {
3266 tv.view->entries,
3267 tv.view->referenced_items,
3268 tv.view->pending_deletion_items,
3153 - stats_view.spin_locks.use,
3154 - stats_view.spin_locks.insert,
3155 - stats_view.spin_locks.search
3269 + stats_view.spin_locks.use_spins,
3270 + stats_view.spin_locks.insert_spins,
3271 + stats_view.spin_locks.delete_spins,
3272 + stats_view.spin_locks.search_spins
3273 );
3274 dictionary_destroy(tv.master);
3275 dictionary_destroy(tv.view);
libnetdata/dictionary/dictionary.h
+4 -3
@@ -98,9 +98,10 @@ struct dictionary_stats {
98
99 // spin locks
100 struct {
101 - size_t use; // number of times a reference to item had to spin to acquire it or ignore it
102 - size_t search; // number of times a successful search result had to be thrown away
103 - size_t insert; // number of times an insertion to the hash table had to be repeated
101 + size_t use_spins; // number of times a reference to item had to spin to acquire it or ignore it
102 + size_t search_spins; // number of times a successful search result had to be thrown away
103 + size_t insert_spins; // number of times an insertion to the hash table had to be repeated
104 + size_t delete_spins; // number of times a deletion had to spin to get a decision
105 } spin_locks;
106 };
107
libnetdata/log/log.c
+1 -1
@@ -861,7 +861,7 @@ void error_limit_int(ERROR_LIMIT *erl, const char *prefix, const char *file __ma
861 va_end( args );
862
863 if(erl->count > 1)
864 - fprintf(stderr, " (repeated %zu times in the last %llu secs)", erl->count, (unsigned long long)(erl->last_logged ? now - erl->last_logged : 0));
864 + fprintf(stderr, " (similar messages repeated %zu times in the last %llu secs)", erl->count, (unsigned long long)(erl->last_logged ? now - erl->last_logged : 0));
865
866 if(erl->sleep_ut)
867 fprintf(stderr, " (sleeping for %llu microseconds every time this happens)", erl->sleep_ut);
streaming/replication.c
+219 -132
@@ -3,7 +3,7 @@
3 #include "replication.h"
4 #include "Judy.h"
5
6 -#define MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED 30
6 +#define MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED 20
7 #define MIN_SENDER_BUFFER_PERCENTAGE_ALLOWED 10
8
9 static time_t replicate_chart_timeframe(BUFFER *wb, RRDSET *st, time_t after, time_t before, bool enable_streaming) {
@@ -23,8 +23,8 @@ static time_t replicate_chart_timeframe(BUFFER *wb, RRDSET *st, time_t after, ti
23 memset(data, 0, sizeof(data));
24
25 if(enable_streaming && st->last_updated.tv_sec > before) {
26 - internal_error(true, "REPLAY: '%s' overwriting replication before from %llu to %llu",
27 - rrdset_id(st),
26 + internal_error(true, "REPLAY: 'host:%s/chart:%s' overwriting replication before from %llu to %llu",
27 + rrdhost_hostname(st->rrdhost), rrdset_id(st),
28 (unsigned long long)before,
29 (unsigned long long)st->last_updated.tv_sec
30 );
@@ -65,7 +65,7 @@ static time_t replicate_chart_timeframe(BUFFER *wb, RRDSET *st, time_t after, ti
65 data[i].sp = ops->next_metric(&data[i].handle);
66
67 internal_error(max_skip <= 0,
68 - "REPLAY: host '%s', chart '%s', dimension '%s': db does not advance the query beyond time %llu",
68 + "REPLAY: 'host:%s/chart:%s', dimension '%s': db does not advance the query beyond time %llu",
69 rrdhost_hostname(st->rrdhost), rrdset_id(st), rrddim_id(data[i].rd), (unsigned long long) now);
70
71 if(data[i].sp.end_time < now)
@@ -84,7 +84,7 @@ static time_t replicate_chart_timeframe(BUFFER *wb, RRDSET *st, time_t after, ti
84 time_t wall_clock_time = now_realtime_sec();
85 if(min_start_time > wall_clock_time + 1 || min_end_time > wall_clock_time + 1) {
86 internal_error(true,
87 - "REPLAY: host '%s', chart '%s': db provided future start time %llu or end time %llu (now is %llu)",
87 + "REPLAY: 'host:%s/chart:%s': db provided future start time %llu or end time %llu (now is %llu)",
88 rrdhost_hostname(st->rrdhost), rrdset_id(st),
89 (unsigned long long)min_start_time,
90 (unsigned long long)min_end_time,
@@ -93,9 +93,11 @@ static time_t replicate_chart_timeframe(BUFFER *wb, RRDSET *st, time_t after, ti
93 }
94
95 if(min_end_time < now) {
96 +#ifdef NETDATA_LOG_REPLICATION_REQUESTS
97 internal_error(true,
97 - "REPLAY: host '%s', chart '%s': no data on any dimension beyond time %llu",
98 + "REPLAY: 'host:%s/chart:%s': no data on any dimension beyond time %llu",
99 rrdhost_hostname(st->rrdhost), rrdset_id(st), (unsigned long long)now);
100 +#endif // NETDATA_LOG_REPLICATION_REQUESTS
101 break;
102 }
103
@@ -130,23 +132,23 @@ static time_t replicate_chart_timeframe(BUFFER *wb, RRDSET *st, time_t after, ti
132 now = min_end_time + 1;
133 }
134
133 -#ifdef NETDATA_INTERNAL_CHECKS
135 +#ifdef NETDATA_LOG_REPLICATION_REQUESTS
136 if(actual_after) {
137 char actual_after_buf[LOG_DATE_LENGTH + 1], actual_before_buf[LOG_DATE_LENGTH + 1];
138 log_date(actual_after_buf, LOG_DATE_LENGTH, actual_after);
139 log_date(actual_before_buf, LOG_DATE_LENGTH, actual_before);
140 internal_error(true,
139 - "REPLAY: host '%s', chart '%s': sending data %llu [%s] to %llu [%s] (requested %llu [delta %lld] to %llu [delta %lld])",
141 + "REPLAY: 'host:%s/chart:%s': sending data %llu [%s] to %llu [%s] (requested %llu [delta %lld] to %llu [delta %lld])",
142 rrdhost_hostname(st->rrdhost), rrdset_id(st),
143 (unsigned long long)actual_after, actual_after_buf, (unsigned long long)actual_before, actual_before_buf,
144 (unsigned long long)after, (long long)(actual_after - after), (unsigned long long)before, (long long)(actual_before - before));
145 }
146 else
147 internal_error(true,
146 - "REPLAY: host '%s', chart '%s': nothing to send (requested %llu to %llu)",
148 + "REPLAY: 'host:%s/chart:%s': nothing to send (requested %llu to %llu)",
149 rrdhost_hostname(st->rrdhost), rrdset_id(st),
150 (unsigned long long)after, (unsigned long long)before);
149 -#endif
151 +#endif // NETDATA_LOG_REPLICATION_REQUESTS
152
153 // release all the dictionary items acquired
154 // finalize the queries
@@ -193,8 +195,9 @@ bool replicate_chart_response(RRDHOST *host, RRDSET *st, bool start_streaming, t
195 time_t first_entry_local = rrdset_first_entry_t(st);
196 if(first_entry_local > now + tolerance) {
197 internal_error(true,
196 - "RRDSET: '%s' first time %llu is in the future (now is %llu)",
197 - rrdset_id(st), (unsigned long long)first_entry_local, (unsigned long long)now);
198 + "RRDSET: 'host:%s/chart:%s' first time %llu is in the future (now is %llu)",
199 + rrdhost_hostname(st->rrdhost), rrdset_id(st),
200 + (unsigned long long)first_entry_local, (unsigned long long)now);
201 first_entry_local = now;
202 }
203
@@ -205,15 +208,16 @@ bool replicate_chart_response(RRDHOST *host, RRDSET *st, bool start_streaming, t
208 time_t last_entry_local = st->last_updated.tv_sec;
209 if(!last_entry_local) {
210 internal_error(true,
208 - "RRDSET: '%s' last updated time zero. Querying db for last updated time.",
209 - rrdset_id(st));
211 + "RRDSET: 'host:%s/chart:%s' db reports last updated time zero.",
212 + rrdhost_hostname(st->rrdhost), rrdset_id(st));
213 last_entry_local = rrdset_last_entry_t(st);
214 }
215
216 if(last_entry_local > now + tolerance) {
217 internal_error(true,
215 - "RRDSET: '%s' last updated time %llu is in the future (now is %llu)",
216 - rrdset_id(st), (unsigned long long)last_entry_local, (unsigned long long)now);
218 + "RRDSET: 'host:%s/chart:%s' last updated time %llu is in the future (now is %llu)",
219 + rrdhost_hostname(st->rrdhost), rrdset_id(st),
220 + (unsigned long long)last_entry_local, (unsigned long long)now);
221 last_entry_local = now;
222 }
223
@@ -263,51 +267,90 @@ bool replicate_chart_response(RRDHOST *host, RRDSET *st, bool start_streaming, t
267 return enable_streaming;
268 }
269
266 -static bool send_replay_chart_cmd(send_command callback, void *callback_data, RRDSET *st, bool start_streaming, time_t after, time_t before) {
270 +// ----------------------------------------------------------------------------
271 +// sending replication requests
272
268 - if(st->rrdhost->receiver && (!st->rrdhost->receiver->replication_first_time_t || after < st->rrdhost->receiver->replication_first_time_t))
269 - st->rrdhost->receiver->replication_first_time_t = after;
273 +struct replication_request_details {
274 + struct {
275 + send_command callback;
276 + void *data;
277 + } caller;
278
271 -#ifdef NETDATA_INTERNAL_CHECKS
272 - if(after && before) {
273 - char after_buf[LOG_DATE_LENGTH + 1], before_buf[LOG_DATE_LENGTH + 1];
274 - log_date(after_buf, LOG_DATE_LENGTH, after);
275 - log_date(before_buf, LOG_DATE_LENGTH, before);
276 - internal_error(true,
277 - "REPLAY: host '%s', chart '%s': sending replication request %llu [%s] to %llu [%s], start streaming: %s",
278 - rrdhost_hostname(st->rrdhost), rrdset_id(st),
279 - (unsigned long long)after, after_buf, (unsigned long long)before, before_buf,
280 - start_streaming?"true":"false");
281 - }
282 - else {
283 - internal_error(true,
284 - "REPLAY: host '%s', chart '%s': sending empty replication request, start streaming: %s",
285 - rrdhost_hostname(st->rrdhost), rrdset_id(st),
286 - start_streaming?"true":"false");
287 - }
288 -#endif
279 + RRDHOST *host;
280 + RRDSET *st;
281
290 -#ifdef NETDATA_INTERNAL_CHECKS
291 - internal_error(
292 - st->replay.after != 0 || st->replay.before != 0,
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 -
297 - st->replay.start_streaming = start_streaming;
298 - st->replay.after = after;
299 - st->replay.before = before;
300 -#endif
282 + struct {
283 + time_t first_entry_t; // the first entry time the child has
284 + time_t last_entry_t; // the last entry time the child has
285 + } child_db;
286 +
287 + struct {
288 + time_t first_entry_t; // the first entry time we have
289 + time_t last_entry_t; // the last entry time we have
290 + bool last_entry_t_adjusted_to_now; // true, if the last entry time was in the future and we fixed
291 + } local_db;
292 +
293 + struct {
294 + time_t from; // the starting time of the entire gap we have
295 + time_t to; // the ending time of the entire gap we have
296 + } gap;
297
302 - debug(D_REPLICATION, PLUGINSD_KEYWORD_REPLAY_CHART " \"%s\" \"%s\" %llu %llu\n",
303 - rrdset_id(st), start_streaming ? "true" : "false", (unsigned long long)after, (unsigned long long)before);
298 + struct {
299 + time_t after; // the start time we requested previously from this child
300 + time_t before; // the end time we requested previously from this child
301 + } last_request;
302 +
303 + struct {
304 + time_t after; // the start time of this replication request - the child will add 1 second
305 + time_t before; // the end time of this replication request
306 + bool start_streaming; // true when we want the child to send anything remaining and start streaming - the child will overwrite 'before'
307 + } wanted;
308 +
309 + time_t now; // the current wall clock time
310 +};
311 +
312 +static bool send_replay_chart_cmd(struct replication_request_details *r, const char *msg __maybe_unused) {
313 + RRDSET *st = r->st;
314 +
315 + if(st->rrdhost->receiver && (!st->rrdhost->receiver->replication_first_time_t || r->wanted.after < st->rrdhost->receiver->replication_first_time_t))
316 + st->rrdhost->receiver->replication_first_time_t = r->wanted.after;
317 +
318 +#ifdef NETDATA_LOG_REPLICATION_REQUESTS
319 + char wanted_after_buf[LOG_DATE_LENGTH + 1] = "", wanted_before_buf[LOG_DATE_LENGTH + 1] = "";
320 +
321 + if(r->wanted.after)
322 + log_date(wanted_after_buf, LOG_DATE_LENGTH, r->wanted.after);
323 +
324 + if(r->wanted.before)
325 + log_date(wanted_before_buf, LOG_DATE_LENGTH, r->wanted.before);
326 +
327 + internal_error(true,
328 + "REPLAY: 'host:%s/chart:%s' sending replication request %ld [%s] to %ld [%s], start streaming '%s': %s: "
329 + "last[%ld - %ld] child[%ld - %ld] local[%ld - %ld %s] gap[%ld - %ld %s] %s"
330 + , rrdhost_hostname(r->host), rrdset_id(r->st)
331 + , r->wanted.after, wanted_after_buf
332 + , r->wanted.before, wanted_before_buf
333 + , r->wanted.start_streaming ? "YES" : "NO"
334 + , msg
335 + , r->last_request.after, r->last_request.before
336 + , r->child_db.first_entry_t, r->child_db.last_entry_t
337 + , r->local_db.first_entry_t, r->local_db.last_entry_t, r->local_db.last_entry_t_adjusted_to_now?"FIXED":"RAW"
338 + , r->gap.from, r->gap.to
339 + , (r->gap.from == r->wanted.after) ? "FULL" : "PARTIAL"
340 + , (st->replay.after != 0 || st->replay.before != 0) ? "OVERLAPPING" : ""
341 + );
342 +
343 + st->replay.start_streaming = r->wanted.start_streaming;
344 + st->replay.after = r->wanted.after;
345 + st->replay.before = r->wanted.before;
346 +#endif // NETDATA_LOG_REPLICATION_REQUESTS
347
348 char buffer[2048 + 1];
349 snprintfz(buffer, 2048, PLUGINSD_KEYWORD_REPLAY_CHART " \"%s\" \"%s\" %llu %llu\n",
307 - rrdset_id(st), start_streaming ? "true" : "false",
308 - (unsigned long long)after, (unsigned long long)before);
350 + rrdset_id(st), r->wanted.start_streaming ? "true" : "false",
351 + (unsigned long long)r->wanted.after, (unsigned long long)r->wanted.before);
352
310 - int ret = callback(buffer, callback_data);
353 + int ret = r->caller.callback(buffer, r->caller.data);
354 if (ret < 0) {
355 error("REPLICATION: failed to send replication request to child (error %d)", ret);
356 return false;
@@ -320,81 +363,110 @@ bool replicate_chart_request(send_command callback, void *callback_data, RRDHOST
363 time_t first_entry_child, time_t last_entry_child,
364 time_t prev_first_entry_wanted, time_t prev_last_entry_wanted)
365 {
323 - time_t now = now_realtime_sec();
324 -
325 - // if replication is disabled, send an empty replication request
326 - // asking no data
327 - if (unlikely(!rrdhost_option_check(host, RRDHOST_OPTION_REPLICATION))) {
328 - internal_error(true,
329 - "REPLAY: host '%s', chart '%s': sending empty replication request because replication is disabled",
330 - rrdhost_hostname(host), rrdset_id(st));
366 + struct replication_request_details r = {
367 + .caller = {
368 + .callback = callback,
369 + .data = callback_data,
370 + },
371 +
372 + .host = host,
373 + .st = st,
374 +
375 + .child_db = {
376 + .first_entry_t = first_entry_child,
377 + .last_entry_t = last_entry_child,
378 + },
379 +
380 + .last_request = {
381 + .after = prev_first_entry_wanted,
382 + .before = prev_last_entry_wanted,
383 + },
384 +
385 + .wanted = {
386 + .after = 0,
387 + .before = 0,
388 + .start_streaming = true,
389 + },
390 +
391 + .now = now_realtime_sec(),
392 + };
393
332 - return send_replay_chart_cmd(callback, callback_data, st, true, 0, 0);
394 + // get our local database retention
395 + r.local_db.first_entry_t = rrdset_first_entry_t(st);
396 + r.local_db.last_entry_t = rrdset_last_entry_t(st);
397 + if(r.local_db.last_entry_t > r.now) {
398 + r.local_db.last_entry_t = r.now;
399 + r.local_db.last_entry_t_adjusted_to_now = true;
400 }
401
335 - // Child has no stored data
336 - if (!last_entry_child) {
337 - error("REPLAY: host '%s', chart '%s': sending empty replication request because child has no stored data",
338 - rrdhost_hostname(host), rrdset_id(st));
402 + // let's find the GAP we have
403 + if(!r.last_request.after || !r.last_request.before) {
404 + // there is no previous request
405 +
406 + if(r.local_db.last_entry_t)
407 + // we have some data, let's continue from the last point we have
408 + r.gap.from = r.local_db.last_entry_t;
409 + else
410 + // we don't have any data, the gap is the max timeframe we are allowed to replicate
411 + r.gap.from = r.now - r.host->rrdpush_seconds_to_replicate;
412
340 - return send_replay_chart_cmd(callback, callback_data, st, true, 0, 0);
413 + }
414 + else {
415 + // we had sent a request - let's continue at the point we left it
416 + // for this we don't take into account the actual data in our db
417 + // because the child may also have gaps and we need to get over it
418 + r.gap.from = r.last_request.before;
419 }
420
343 - // Nothing to get if the chart has not dimensions
344 - if (!rrdset_number_of_dimensions(st)) {
345 - error("REPLAY: host '%s', chart '%s': sending empty replication request because chart has no dimensions",
346 - rrdhost_hostname(host), rrdset_id(st));
421 + // we want all the data up to now
422 + r.gap.to = r.now;
423
348 - return send_replay_chart_cmd(callback, callback_data, st, true, 0, 0);
349 - }
424 + // The gap is now r.gap.from -> r.gap.to
425
351 - // if the child's first/last entries are nonsensical, resume streaming
352 - // without asking for any data
353 - if (first_entry_child <= 0) {
354 - error("REPLAY: host '%s', chart '%s': sending empty replication because first entry of the child is invalid (%llu)",
355 - rrdhost_hostname(host), rrdset_id(st), (unsigned long long)first_entry_child);
426 + if (unlikely(!rrdhost_option_check(host, RRDHOST_OPTION_REPLICATION)))
427 + return send_replay_chart_cmd(&r, "empty replication request, replication is disabled");
428
357 - return send_replay_chart_cmd(callback, callback_data, st, true, 0, 0);
358 - }
429 + if (unlikely(!r.child_db.last_entry_t))
430 + return send_replay_chart_cmd(&r, "empty replication request, child has no stored data");
431
360 - if (first_entry_child > last_entry_child) {
361 - error("REPLAY: host '%s', chart '%s': sending empty replication because child timings are invalid (first entry %llu > last entry %llu)",
362 - rrdhost_hostname(host), rrdset_id(st), (unsigned long long)first_entry_child, (unsigned long long)last_entry_child);
432 + if (unlikely(!rrdset_number_of_dimensions(st)))
433 + return send_replay_chart_cmd(&r, "empty replication request, chart has no dimensions");
434
364 - return send_replay_chart_cmd(callback, callback_data, st, true, 0, 0);
365 - }
435 + if (r.child_db.first_entry_t <= 0)
436 + return send_replay_chart_cmd(&r, "empty replication request, first entry of the child db first entry is invalid");
437
367 - time_t last_entry_local = rrdset_last_entry_t(st);
368 - if(last_entry_local > now) {
369 - internal_error(true,
370 - "REPLAY: host '%s', chart '%s': local last entry time %llu is in the future (now is %llu). Adjusting it.",
371 - rrdhost_hostname(host), rrdset_id(st), (unsigned long long)last_entry_local, (unsigned long long)now);
372 - last_entry_local = now;
373 - }
438 + if (r.child_db.first_entry_t > r.child_db.last_entry_t)
439 + return send_replay_chart_cmd(&r, "empty replication request, child timings are invalid (first entry > last entry)");
440
375 - // should never happen but if it does, start streaming without asking for any data
376 - if (last_entry_local > last_entry_child) {
377 - error("REPLAY: host '%s', chart '%s': sending empty replication request because our last entry (%llu) in later than the child one (%llu)",
378 - rrdhost_hostname(host), rrdset_id(st), (unsigned long long)last_entry_local, (unsigned long long)last_entry_child);
441 + if (r.local_db.last_entry_t > r.child_db.last_entry_t)
442 + return send_replay_chart_cmd(&r, "empty replication request, local last entry is later than the child one");
443
380 - return send_replay_chart_cmd(callback, callback_data, st, true, 0, 0);
381 - }
444 + // let's find what the child can provide to fill that gap
445
383 - time_t first_entry_wanted;
384 - if (prev_first_entry_wanted && prev_last_entry_wanted) {
385 - first_entry_wanted = prev_last_entry_wanted;
386 - if ((now - first_entry_wanted) > host->rrdpush_seconds_to_replicate)
387 - first_entry_wanted = now - host->rrdpush_seconds_to_replicate;
388 - }
446 + if(r.child_db.first_entry_t > r.gap.from)
447 + // the child does not have all the data - let's get what it has
448 + r.wanted.after = r.child_db.first_entry_t;
449 + else
450 + // ok, the child can fill the entire gap we have
451 + r.wanted.after = r.gap.from;
452 +
453 + if(r.gap.to - r.wanted.after > host->rrdpush_replication_step)
454 + // the duration is too big for one request - let's take the first step
455 + r.wanted.before = r.wanted.after + host->rrdpush_replication_step;
456 else
390 - first_entry_wanted = MAX(last_entry_local, first_entry_child);
457 + // wow, we can do it in one request
458 + r.wanted.before = r.gap.to;
459
392 - time_t last_entry_wanted = first_entry_wanted + host->rrdpush_replication_step;
393 - last_entry_wanted = MIN(last_entry_wanted, last_entry_child);
460 + // don't ask from the child more than it has
461 + if(r.wanted.before > r.child_db.last_entry_t)
462 + r.wanted.before = r.child_db.last_entry_t;
463
395 - bool start_streaming = (last_entry_wanted == last_entry_child);
464 + // the child should start streaming immediately if the wanted duration is small
465 + r.wanted.start_streaming = (r.wanted.before == r.child_db.last_entry_t);
466
397 - return send_replay_chart_cmd(callback, callback_data, st, start_streaming, first_entry_wanted, last_entry_wanted);
467 + // the wanted timeframe is now r.wanted.after -> r.wanted.before
468 + // send it
469 + return send_replay_chart_cmd(&r, "OK");
470 }
471
472 // ----------------------------------------------------------------------------
@@ -633,6 +705,7 @@ static struct replication_request replication_request_get_first_available() {
705 else if(sender_has_room_to_spare) {
706 // copy the request to return it
707 rq = *rse->rq;
708 + rq.chart_id = string_dup(rq.chart_id);
709
710 // set the return result to found
711 rq.found = true;
@@ -662,14 +735,6 @@ static void replication_request_react_callback(const DICTIONARY_ITEM *item __may
735 struct sender_state *s = sender_state; (void)s;
736 struct replication_request *rq = value;
737
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 -
738 // IMPORTANT:
739 // We use the react instead of the insert callback
740 // because we want the item to be atomically visible
@@ -691,12 +756,26 @@ static bool replication_request_conflict_callback(const DICTIONARY_ITEM *item __
756 struct replication_request *rq = old_value; (void)rq;
757 struct replication_request *rq_new = new_value;
758
694 - internal_error(
695 - true,
696 - "STREAM %s [send to %s]: REPLAY ERROR: ignoring 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");
759 + replication_recursive_lock();
760 +
761 + if(!rq->indexed_in_judy) {
762 + replication_sort_entry_add(rq);
763 + internal_error(
764 + true,
765 + "STREAM %s [send to %s]: REPLAY: 'host:%s/chart:%s' adding duplicate replication command received (existing from %llu to %llu [%s], new from %llu to %llu [%s])",
766 + rrdhost_hostname(s->host), s->connected_to, rrdhost_hostname(s->host), dictionary_acquired_item_name(item),
767 + (unsigned long long)rq->after, (unsigned long long)rq->before, rq->start_streaming ? "true" : "false",
768 + (unsigned long long)rq_new->after, (unsigned long long)rq_new->before, rq_new->start_streaming ? "true" : "false");
769 + }
770 + else
771 + internal_error(
772 + true,
773 + "STREAM %s [send to %s]: REPLAY: 'host:%s/chart:%s' ignoring duplicate replication command received (existing from %llu to %llu [%s], new from %llu to %llu [%s])",
774 + rrdhost_hostname(s->host), s->connected_to, rrdhost_hostname(s->host), dictionary_acquired_item_name(item),
775 + (unsigned long long)rq->after, (unsigned long long)rq->before, rq->start_streaming ? "true" : "false",
776 + (unsigned long long)rq_new->after, (unsigned long long)rq_new->before, rq_new->start_streaming ? "true" : "false");
777 +
778 + replication_recursive_unlock();
779
780 // bool updated_after = false, updated_before = false, updated_start_streaming = false, updated = false;
781 //
@@ -880,7 +959,7 @@ void *replication_thread_main(void *ptr __maybe_unused) {
959 worker_set_metric(WORKER_JOB_CUSTOM_METRIC_COMPLETION, (NETDATA_DOUBLE)done * 100.0 / (NETDATA_DOUBLE)total);
960 }
961
883 - if(!rq.found) {
962 + if(unlikely(!rq.found)) {
963 worker_is_idle();
964
965 if(!rep.requests_count)
@@ -898,7 +977,15 @@ void *replication_thread_main(void *ptr __maybe_unused) {
977 else {
978 // delete the request from the dictionary
979 worker_is_busy(WORKER_JOB_DELETE_ENTRY);
901 - dictionary_del(rq.sender->replication_requests, string2str(rq.chart_id));
980 + if(!dictionary_del(rq.sender->replication_requests, string2str(rq.chart_id)))
981 + error("REPLAY: 'host:%s/chart:%s' failed to be deleted from sender dictionary",
982 + rrdhost_hostname(rq.sender->host), string2str(rq.chart_id));
983 +
984 + if(rq.sender->replication_pending_requests == 0 && dictionary_entries(rq.sender->replication_requests) != 0)
985 + error("REPLAY: 'host:%s/chart:%s' sender dictionary has %zu entries, but sender pending requests are %zu",
986 + rrdhost_hostname(rq.sender->host), string2str(rq.chart_id),
987 + dictionary_entries(rq.sender->replication_requests),
988 + rq.sender->replication_pending_requests);
989 }
990
991 worker_is_busy(WORKER_JOB_FIND_CHART);
@@ -910,13 +997,6 @@ void *replication_thread_main(void *ptr __maybe_unused) {
997 continue;
998 }
999
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 -
1000 worker_is_busy(WORKER_JOB_QUERYING);
1001
1002 latest_first_time_t = rq.after;
@@ -947,11 +1027,18 @@ void *replication_thread_main(void *ptr __maybe_unused) {
1027 rrdset_flag_clear(st, RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS);
1028 rrdset_flag_set(st, RRDSET_FLAG_SENDER_REPLICATION_FINISHED);
1029 rrdhost_sender_replicating_charts_minus_one(st->rrdhost);
1030 +
1031 +#ifdef NETDATA_LOG_REPLICATION_REQUESTS
1032 + internal_error(true, "STREAM_SENDER REPLAY: 'host:%s/chart:%s' streaming starts",
1033 + rrdhost_hostname(st->rrdhost), rrdset_id(st));
1034 +#endif
1035 }
1036 else
1037 internal_error(true, "REPLAY ERROR: received start streaming command for chart '%s' or host '%s', but the chart is not in progress replicating",
1038 string2str(rq.chart_id), rrdhost_hostname(st->rrdhost));
1039 }
1040 +
1041 + string_freez(rq.chart_id);
1042 }
1043
1044 netdata_thread_cleanup_pop(1);
streaming/rrdpush.c
+17 -5
@@ -298,21 +298,32 @@ static inline void rrdpush_send_chart_definition(BUFFER *wb, RRDSET *st) {
298
299 if(!last_entry_local) {
300 internal_error(true,
301 - "RRDSET: '%s' last updated time zero. Querying db for last updated time.",
302 - rrdset_id(st));
301 + "RRDSET: 'host:%s/chart:%s' db reports last updated time zero.",
302 + rrdhost_hostname(st->rrdhost), rrdset_id(st));
303
304 last_entry_local = rrdset_last_entry_t(st);
305 time_t now = now_realtime_sec();
306 +
307 if(last_entry_local > now) {
308 internal_error(true,
308 - "RRDSET: '%s' last updated time %llu is in the future (now is %llu)",
309 - rrdset_id(st), (unsigned long long)last_entry_local, (unsigned long long)now);
309 + "RRDSET: 'host:%s/chart:%s' last updated time %llu is in the future (now is %llu)",
310 + rrdhost_hostname(st->rrdhost), rrdset_id(st),
311 + (unsigned long long)last_entry_local, (unsigned long long)now);
312 last_entry_local = now;
313 }
314 }
315
316 buffer_sprintf(wb, PLUGINSD_KEYWORD_CHART_DEFINITION_END " %llu %llu\n",
317 (unsigned long long)first_entry_local, (unsigned long long)last_entry_local);
318 +
319 + rrdset_flag_set(st, RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS);
320 + rrdset_flag_clear(st, RRDSET_FLAG_SENDER_REPLICATION_FINISHED);
321 + rrdhost_sender_replicating_charts_plus_one(st->rrdhost);
322 +
323 +#ifdef NETDATA_LOG_REPLICATION_REQUESTS
324 + internal_error(true, "REPLAY: 'host:%s/chart:%s' replication starts",
325 + rrdhost_hostname(st->rrdhost), rrdset_id(st));
326 +#endif
327 }
328
329 st->upstream_resync_time = st->last_collected_time.tv_sec + (remote_clock_resync_iterations * st->update_every);
@@ -344,7 +355,8 @@ static void rrdpush_send_chart_metrics(BUFFER *wb, RRDSET *st, struct sender_sta
355 buffer_fast_strcat(wb, "\n", 1);
356 }
357 else {
347 - internal_error(true, "host '%s', chart '%s', dimension '%s' flag 'exposed' is updated but not exposed", rrdhost_hostname(st->rrdhost), rrdset_id(st), rrddim_id(rd));
358 + internal_error(true, "STREAM: 'host:%s/chart:%s/dim:%s' flag 'exposed' is updated but not exposed",
359 + rrdhost_hostname(st->rrdhost), rrdset_id(st), rrddim_id(rd));
360 // we will include it in the next iteration
361 rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED);
362 }
streaming/sender.c
+7 -3
@@ -23,9 +23,10 @@
23 #define WORKER_SENDER_JOB_BYTES_SENT 17
24 #define WORKER_SENDER_JOB_REPLAY_REQUEST 18
25 #define WORKER_SENDER_JOB_FUNCTION_REQUEST 19
26 +#define WORKER_SENDER_JOB_REPLAY_DICT_SIZE 20
27
27 -#if WORKER_UTILIZATION_MAX_JOB_TYPES < 20
28 -#error WORKER_UTILIZATION_MAX_JOB_TYPES has to be at least 20
28 +#if WORKER_UTILIZATION_MAX_JOB_TYPES < 21
29 +#error WORKER_UTILIZATION_MAX_JOB_TYPES has to be at least 21
30 #endif
31
32 extern struct config stream_config;
@@ -225,7 +226,7 @@ static void rrdpush_sender_thread_reset_all_charts(RRDHOST *host) {
226
227 RRDSET *st;
228 rrdset_foreach_read(st, host) {
228 - rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED | RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS | RRDSET_FLAG_SENDER_REPLICATION_QUEUED);
229 + rrdset_flag_clear(st, RRDSET_FLAG_UPSTREAM_EXPOSED | RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS);
230 rrdset_flag_set(st, RRDSET_FLAG_SENDER_REPLICATION_FINISHED);
231
232 st->upstream_resync_time = 0;
@@ -1099,6 +1100,7 @@ void *rrdpush_sender_thread(void *ptr) {
1100 worker_register_job_custom_metric(WORKER_SENDER_JOB_BUFFER_RATIO, "used buffer ratio", "%", WORKER_METRIC_ABSOLUTE);
1101 worker_register_job_custom_metric(WORKER_SENDER_JOB_BYTES_RECEIVED, "bytes received", "bytes/s", WORKER_METRIC_INCREMENT);
1102 worker_register_job_custom_metric(WORKER_SENDER_JOB_BYTES_SENT, "bytes sent", "bytes/s", WORKER_METRIC_INCREMENT);
1103 + worker_register_job_custom_metric(WORKER_SENDER_JOB_REPLAY_DICT_SIZE, "replication dict entries", "entries", WORKER_METRIC_ABSOLUTE);
1104
1105 struct sender_state *s = ptr;
1106 s->tid = gettid();
@@ -1342,6 +1344,8 @@ void *rrdpush_sender_thread(void *ptr) {
1344 rrdhost_hostname(s->host), s->connected_to, s->buffer->size, s->sent_bytes_on_this_connection);
1345 rrdpush_sender_thread_close_socket(s->host);
1346 }
1347 +
1348 + worker_set_metric(WORKER_SENDER_JOB_REPLAY_DICT_SIZE, (NETDATA_DOUBLE) dictionary_entries(s->replication_requests));
1349 }
1350
1351 netdata_thread_cleanup_pop(1);