reworked pluginsd caching of RDAs to avoid crashes (#15845)
Costa Tsaousis committed
Aug 21, 2023 at 14:22 UTC
980a6053639a297bb1219b5b6dde0df27cb1f10f
2 files changed
+128
-71
collectors/plugins.d/pluginsd_parser.c
+125
-70
@@ -68,7 +68,7 @@ static ssize_t send_to_plugin(const char *txt, void *data) {
68
return -4;
69
}
70
71
-static inline RRDHOST *pluginsd_require_host_from_parent(PARSER *parser, const char *cmd) {
71
+static inline RRDHOST *pluginsd_require_scope_host(PARSER *parser, const char *cmd) {
72
RRDHOST *host = parser->user.host;
73
74
if(unlikely(!host))
@@ -77,7 +77,7 @@ static inline RRDHOST *pluginsd_require_host_from_parent(PARSER *parser, const c
77
return host;
78
}
79
80
-static inline RRDSET *pluginsd_require_chart_from_parent(PARSER *parser, const char *cmd, const char *parent_cmd) {
80
+static inline RRDSET *pluginsd_require_scope_chart(PARSER *parser, const char *cmd, const char *parent_cmd) {
81
RRDSET *st = parser->user.st;
82
83
if(unlikely(!st))
@@ -86,7 +86,7 @@ static inline RRDSET *pluginsd_require_chart_from_parent(PARSER *parser, const c
86
return st;
87
}
88
89
-static inline RRDSET *pluginsd_get_chart_from_parent(PARSER *parser) {
89
+static inline RRDSET *pluginsd_get_scope_chart(PARSER *parser) {
90
return parser->user.st;
91
}
92
@@ -108,20 +108,23 @@ static inline bool pluginsd_unlock_rrdset_data_collection(PARSER *parser) {
108
}
109
110
void pluginsd_rrdset_cleanup(RRDSET *st) {
111
- for(size_t i = 0; i < st->pluginsd.used ; i++) {
112
- if (st->pluginsd.rda[i]) {
113
- rrddim_acquired_release(st->pluginsd.rda[i]);
114
- st->pluginsd.rda[i] = NULL;
115
- }
111
+ spinlock_lock(&st->pluginsd.spinlock);
112
+
113
+ for(size_t i = 0; i < st->pluginsd.size ; i++) {
114
+ rrddim_acquired_release(st->pluginsd.rda[i]); // can be NULL
115
+ st->pluginsd.rda[i] = NULL;
116
}
117
+
118
freez(st->pluginsd.rda);
119
+ st->pluginsd.collector_tid = 0;
120
st->pluginsd.rda = NULL;
121
st->pluginsd.size = 0;
120
- st->pluginsd.used = 0;
122
st->pluginsd.pos = 0;
123
+
124
+ spinlock_unlock(&st->pluginsd.spinlock);
125
}
126
124
-static inline void pluginsd_unlock_previous_chart(PARSER *parser, const char *keyword, bool stale) {
127
+static inline void pluginsd_unlock_previous_scope_chart(PARSER *parser, const char *keyword, bool stale) {
128
if(unlikely(pluginsd_unlock_rrdset_data_collection(parser))) {
129
if(stale)
130
netdata_log_error("PLUGINSD: 'host:%s/chart:%s/' stale data collection lock found during %s; it has been unlocked",
@@ -142,23 +145,49 @@ static inline void pluginsd_unlock_previous_chart(PARSER *parser, const char *ke
145
}
146
}
147
145
-static inline void pluginsd_set_chart_from_parent(PARSER *parser, RRDSET *st, const char *keyword) {
146
- pluginsd_unlock_previous_chart(parser, keyword, true);
148
+static inline void pluginsd_clear_scope_chart(PARSER *parser, const char *keyword) {
149
+ pluginsd_unlock_previous_scope_chart(parser, keyword, true);
150
+ parser->user.st = NULL;
151
+}
152
+
153
+static inline bool pluginsd_set_scope_chart(PARSER *parser, RRDSET *st, const char *keyword) {
154
+ RRDSET *old_st = parser->user.st;
155
+ pid_t old_collector_tid = (old_st) ? old_st->pluginsd.collector_tid : 0;
156
+ pid_t my_collector_tid = gettid();
157
+
158
+ if(unlikely(old_collector_tid)) {
159
+ if(old_collector_tid != my_collector_tid) {
160
+ error_limit_static_global_var(erl, 1, 0);
161
+ error_limit(&erl, "PLUGINSD: keyword %s: 'host:%s/chart:%s' is collected twice (my tid %d, other collector tid %d)",
162
+ keyword ? keyword : "UNKNOWN",
163
+ rrdhost_hostname(st->rrdhost), rrdset_id(st),
164
+ my_collector_tid, old_collector_tid);
165
148
- if(st) {
149
- size_t dims = dictionary_entries(st->rrddim_root_index);
150
- if(unlikely(st->pluginsd.size < dims)) {
151
- st->pluginsd.rda = reallocz(st->pluginsd.rda, dims * sizeof(RRDDIM_ACQUIRED *));
152
- st->pluginsd.size = dims;
166
+ return false;
167
}
168
155
- if(st->pluginsd.pos > st->pluginsd.used && st->pluginsd.pos <= st->pluginsd.size)
156
- st->pluginsd.used = st->pluginsd.pos;
169
+ old_st->pluginsd.collector_tid = 0;
170
+ }
171
158
- st->pluginsd.pos = 0;
172
+ st->pluginsd.collector_tid = my_collector_tid;
173
+
174
+ pluginsd_clear_scope_chart(parser, keyword);
175
+
176
+ size_t dims = dictionary_entries(st->rrddim_root_index);
177
+ if(unlikely(st->pluginsd.size < dims)) {
178
+ st->pluginsd.rda = reallocz(st->pluginsd.rda, dims * sizeof(RRDDIM_ACQUIRED *));
179
+
180
+ // initialize the empty slots
181
+ for(ssize_t i = (ssize_t)dims - 1; i >= (ssize_t)st->pluginsd.size ;i--)
182
+ st->pluginsd.rda[i] = NULL;
183
+
184
+ st->pluginsd.size = dims;
185
}
186
187
+ st->pluginsd.pos = 0;
188
parser->user.st = st;
189
+
190
+ return true;
191
}
192
193
static inline RRDDIM *pluginsd_acquire_dimension(RRDHOST *host, RRDSET *st, const char *dimension, const char *cmd) {
@@ -168,16 +197,21 @@ static inline RRDDIM *pluginsd_acquire_dimension(RRDHOST *host, RRDSET *st, cons
197
return NULL;
198
}
199
171
- RRDDIM_ACQUIRED *rda;
200
+ if(unlikely(st->pluginsd.pos >= st->pluginsd.size))
201
+ st->pluginsd.pos = 0;
202
173
- if(likely(st->pluginsd.pos < st->pluginsd.used)) {
174
- rda = st->pluginsd.rda[st->pluginsd.pos];
203
+ RRDDIM_ACQUIRED *rda = st->pluginsd.rda[st->pluginsd.pos];
204
+
205
+ if(likely(rda)) {
206
RRDDIM *rd = rrddim_acquired_to_rrddim(rda);
207
if (likely(rd && string_strcmp(rd->id, dimension) == 0)) {
208
+ // we found a cached RDA
209
st->pluginsd.pos++;
210
return rd;
211
}
212
else {
213
+ // the collector is sending dimensions in a different order
214
+ // release the previous one, to reuse this slot
215
rrddim_acquired_release(rda);
216
st->pluginsd.rda[st->pluginsd.pos] = NULL;
217
}
@@ -191,8 +225,7 @@ static inline RRDDIM *pluginsd_acquire_dimension(RRDHOST *host, RRDSET *st, cons
225
return NULL;
226
}
227
194
- if(likely(st->pluginsd.pos < st->pluginsd.size))
195
- st->pluginsd.rda[st->pluginsd.pos++] = rda;
228
+ st->pluginsd.rda[st->pluginsd.pos++] = rda;
229
230
return rrddim_acquired_to_rrddim(rda);
231
}
@@ -227,15 +260,17 @@ static inline PARSER_RC pluginsd_set(char **words, size_t num_words, PARSER *par
260
char *dimension = get_word(words, num_words, 1);
261
char *value = get_word(words, num_words, 2);
262
230
- RRDHOST *host = pluginsd_require_host_from_parent(parser, PLUGINSD_KEYWORD_SET);
263
+ RRDHOST *host = pluginsd_require_scope_host(parser, PLUGINSD_KEYWORD_SET);
264
if(!host) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
265
233
- RRDSET *st = pluginsd_require_chart_from_parent(parser, PLUGINSD_KEYWORD_SET, PLUGINSD_KEYWORD_CHART);
266
+ RRDSET *st = pluginsd_require_scope_chart(parser, PLUGINSD_KEYWORD_SET, PLUGINSD_KEYWORD_CHART);
267
if(!st) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
268
269
RRDDIM *rd = pluginsd_acquire_dimension(host, st, dimension, PLUGINSD_KEYWORD_SET);
270
if(!rd) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
271
272
+ st->pluginsd.set = true;
273
+
274
if (unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
275
netdata_log_debug(D_PLUGINSD, "PLUGINSD: 'host:%s/chart:%s/dim:%s' SET is setting value to '%s'",
276
rrdhost_hostname(host), rrdset_id(st), dimension, value && *value ? value : "UNSET");
@@ -250,13 +285,14 @@ static inline PARSER_RC pluginsd_begin(char **words, size_t num_words, PARSER *p
285
char *id = get_word(words, num_words, 1);
286
char *microseconds_txt = get_word(words, num_words, 2);
287
253
- RRDHOST *host = pluginsd_require_host_from_parent(parser, PLUGINSD_KEYWORD_BEGIN);
288
+ RRDHOST *host = pluginsd_require_scope_host(parser, PLUGINSD_KEYWORD_BEGIN);
289
if(!host) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
290
291
RRDSET *st = pluginsd_find_chart(host, id, PLUGINSD_KEYWORD_BEGIN);
292
if(!st) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
293
259
- pluginsd_set_chart_from_parent(parser, st, PLUGINSD_KEYWORD_BEGIN);
294
+ if(!pluginsd_set_scope_chart(parser, st, PLUGINSD_KEYWORD_BEGIN))
295
+ return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
296
297
usec_t microseconds = 0;
298
if (microseconds_txt && *microseconds_txt) {
@@ -296,16 +332,16 @@ static inline PARSER_RC pluginsd_end(char **words, size_t num_words, PARSER *par
332
UNUSED(words);
333
UNUSED(num_words);
334
299
- RRDHOST *host = pluginsd_require_host_from_parent(parser, PLUGINSD_KEYWORD_END);
335
+ RRDHOST *host = pluginsd_require_scope_host(parser, PLUGINSD_KEYWORD_END);
336
if(!host) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
337
302
- RRDSET *st = pluginsd_require_chart_from_parent(parser, PLUGINSD_KEYWORD_END, PLUGINSD_KEYWORD_BEGIN);
338
+ RRDSET *st = pluginsd_require_scope_chart(parser, PLUGINSD_KEYWORD_END, PLUGINSD_KEYWORD_BEGIN);
339
if(!st) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
340
341
if (unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
342
netdata_log_debug(D_PLUGINSD, "requested an END on chart '%s'", rrdset_id(st));
343
308
- pluginsd_set_chart_from_parent(parser, NULL, PLUGINSD_KEYWORD_END);
344
+ pluginsd_clear_scope_chart(parser, PLUGINSD_KEYWORD_END);
345
parser->user.data_collections_count++;
346
347
struct timeval now;
@@ -418,7 +454,7 @@ static inline PARSER_RC pluginsd_host_define_end(char **words __maybe_unused, si
454
pluginsd_host_define_cleanup(parser);
455
456
parser->user.host = host;
421
- pluginsd_set_chart_from_parent(parser, NULL, PLUGINSD_KEYWORD_HOST_DEFINE_END);
457
+ pluginsd_clear_scope_chart(parser, PLUGINSD_KEYWORD_HOST_DEFINE_END);
458
459
rrdhost_flag_clear(host, RRDHOST_FLAG_ORPHAN);
460
rrdcontext_host_child_connected(host);
@@ -450,7 +486,7 @@ static inline PARSER_RC pluginsd_host(char **words, size_t num_words, PARSER *pa
486
}
487
488
static inline PARSER_RC pluginsd_chart(char **words, size_t num_words, PARSER *parser) {
453
- RRDHOST *host = pluginsd_require_host_from_parent(parser, PLUGINSD_KEYWORD_CHART);
489
+ RRDHOST *host = pluginsd_require_scope_host(parser, PLUGINSD_KEYWORD_CHART);
490
if(!host) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
491
492
char *type = get_word(words, num_words, 1);
@@ -554,13 +590,18 @@ static inline PARSER_RC pluginsd_chart(char **words, size_t num_words, PARSER *p
590
rrdset_flag_set(st, RRDSET_FLAG_STORE_FIRST);
591
else
592
rrdset_flag_clear(st, RRDSET_FLAG_STORE_FIRST);
557
- } else {
593
+ }
594
+ else {
595
rrdset_isnot_obsolete(st);
596
rrdset_flag_clear(st, RRDSET_FLAG_DETAIL);
597
rrdset_flag_clear(st, RRDSET_FLAG_STORE_FIRST);
598
}
599
+
600
+ if(!pluginsd_set_scope_chart(parser, st, PLUGINSD_KEYWORD_CHART))
601
+ return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
602
}
563
- pluginsd_set_chart_from_parent(parser, st, PLUGINSD_KEYWORD_CHART);
603
+ else
604
+ pluginsd_clear_scope_chart(parser, PLUGINSD_KEYWORD_CHART);
605
606
return PARSER_RC_OK;
607
}
@@ -570,10 +611,10 @@ static inline PARSER_RC pluginsd_chart_definition_end(char **words, size_t num_w
611
const char *last_entry_txt = get_word(words, num_words, 2);
612
const char *wall_clock_time_txt = get_word(words, num_words, 3);
613
573
- RRDHOST *host = pluginsd_require_host_from_parent(parser, PLUGINSD_KEYWORD_CHART_DEFINITION_END);
614
+ RRDHOST *host = pluginsd_require_scope_host(parser, PLUGINSD_KEYWORD_CHART_DEFINITION_END);
615
if(!host) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
616
576
- RRDSET *st = pluginsd_require_chart_from_parent(parser, PLUGINSD_KEYWORD_CHART_DEFINITION_END, PLUGINSD_KEYWORD_CHART);
617
+ RRDSET *st = pluginsd_require_scope_chart(parser, PLUGINSD_KEYWORD_CHART_DEFINITION_END, PLUGINSD_KEYWORD_CHART);
618
if(!st) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
619
620
time_t first_entry_child = (first_entry_txt && *first_entry_txt) ? (time_t)str2ul(first_entry_txt) : 0;
@@ -615,10 +656,10 @@ static inline PARSER_RC pluginsd_dimension(char **words, size_t num_words, PARSE
656
char *divisor_s = get_word(words, num_words, 5);
657
char *options = get_word(words, num_words, 6);
658
618
- RRDHOST *host = pluginsd_require_host_from_parent(parser, PLUGINSD_KEYWORD_DIMENSION);
659
+ RRDHOST *host = pluginsd_require_scope_host(parser, PLUGINSD_KEYWORD_DIMENSION);
660
if(!host) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
661
621
- RRDSET *st = pluginsd_require_chart_from_parent(parser, PLUGINSD_KEYWORD_DIMENSION, PLUGINSD_KEYWORD_CHART);
662
+ RRDSET *st = pluginsd_require_scope_chart(parser, PLUGINSD_KEYWORD_DIMENSION, PLUGINSD_KEYWORD_CHART);
663
if(!st) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
664
665
if (unlikely(!id))
@@ -860,10 +901,10 @@ static inline PARSER_RC pluginsd_function(char **words, size_t num_words, PARSER
901
char *timeout_s = get_word(words, num_words, i++);
902
char *help = get_word(words, num_words, i++);
903
863
- RRDHOST *host = pluginsd_require_host_from_parent(parser, PLUGINSD_KEYWORD_FUNCTION);
904
+ RRDHOST *host = pluginsd_require_scope_host(parser, PLUGINSD_KEYWORD_FUNCTION);
905
if(!host) return PARSER_RC_ERROR;
906
866
- RRDSET *st = (global)?NULL:pluginsd_require_chart_from_parent(parser, PLUGINSD_KEYWORD_FUNCTION, PLUGINSD_KEYWORD_CHART);
907
+ RRDSET *st = (global)? NULL: pluginsd_require_scope_chart(parser, PLUGINSD_KEYWORD_FUNCTION, PLUGINSD_KEYWORD_CHART);
908
if(!st) global = true;
909
910
if (unlikely(!timeout_s || !name || !help || (!global && !st))) {
@@ -959,10 +1000,10 @@ static inline PARSER_RC pluginsd_variable(char **words, size_t num_words, PARSER
1000
char *value = get_word(words, num_words, 2);
1001
NETDATA_DOUBLE v;
1002
962
- RRDHOST *host = pluginsd_require_host_from_parent(parser, PLUGINSD_KEYWORD_VARIABLE);
1003
+ RRDHOST *host = pluginsd_require_scope_host(parser, PLUGINSD_KEYWORD_VARIABLE);
1004
if(!host) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
1005
965
- RRDSET *st = pluginsd_get_chart_from_parent(parser);
1006
+ RRDSET *st = pluginsd_get_scope_chart(parser);
1007
1008
int global = (st) ? 0 : 1;
1009
@@ -1040,7 +1081,7 @@ static inline PARSER_RC pluginsd_variable(char **words, size_t num_words, PARSER
1081
1082
static inline PARSER_RC pluginsd_flush(char **words __maybe_unused, size_t num_words __maybe_unused, PARSER *parser) {
1083
netdata_log_debug(D_PLUGINSD, "requested a " PLUGINSD_KEYWORD_FLUSH);
1043
- pluginsd_set_chart_from_parent(parser, NULL, PLUGINSD_KEYWORD_FLUSH);
1084
+ pluginsd_clear_scope_chart(parser, PLUGINSD_KEYWORD_FLUSH);
1085
parser->user.replay.start_time = 0;
1086
parser->user.replay.end_time = 0;
1087
parser->user.replay.start_time_ut = 0;
@@ -1101,7 +1142,7 @@ static inline PARSER_RC pluginsd_label(char **words, size_t num_words, PARSER *p
1142
}
1143
1144
static inline PARSER_RC pluginsd_overwrite(char **words __maybe_unused, size_t num_words __maybe_unused, PARSER *parser) {
1104
- RRDHOST *host = pluginsd_require_host_from_parent(parser, PLUGINSD_KEYWORD_OVERWRITE);
1145
+ RRDHOST *host = pluginsd_require_scope_host(parser, PLUGINSD_KEYWORD_OVERWRITE);
1146
if(!host) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
1147
1148
netdata_log_debug(D_PLUGINSD, "requested to OVERWRITE host labels");
@@ -1128,7 +1169,7 @@ static inline PARSER_RC pluginsd_clabel(char **words, size_t num_words, PARSER *
1169
}
1170
1171
if(unlikely(!parser->user.chart_rrdlabels_linked_temporarily)) {
1131
- RRDSET *st = pluginsd_get_chart_from_parent(parser);
1172
+ RRDSET *st = pluginsd_get_scope_chart(parser);
1173
parser->user.chart_rrdlabels_linked_temporarily = st->rrdlabels;
1174
rrdlabels_unmark_all(parser->user.chart_rrdlabels_linked_temporarily);
1175
}
@@ -1139,10 +1180,10 @@ static inline PARSER_RC pluginsd_clabel(char **words, size_t num_words, PARSER *
1180
}
1181
1182
static inline PARSER_RC pluginsd_clabel_commit(char **words __maybe_unused, size_t num_words __maybe_unused, PARSER *parser) {
1142
- RRDHOST *host = pluginsd_require_host_from_parent(parser, PLUGINSD_KEYWORD_CLABEL_COMMIT);
1183
+ RRDHOST *host = pluginsd_require_scope_host(parser, PLUGINSD_KEYWORD_CLABEL_COMMIT);
1184
if(!host) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
1185
1145
- RRDSET *st = pluginsd_require_chart_from_parent(parser, PLUGINSD_KEYWORD_CLABEL_COMMIT, PLUGINSD_KEYWORD_BEGIN);
1186
+ RRDSET *st = pluginsd_require_scope_chart(parser, PLUGINSD_KEYWORD_CLABEL_COMMIT, PLUGINSD_KEYWORD_BEGIN);
1187
if(!st) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
1188
1189
netdata_log_debug(D_PLUGINSD, "requested to commit chart labels");
@@ -1167,17 +1208,19 @@ static inline PARSER_RC pluginsd_replay_begin(char **words, size_t num_words, PA
1208
char *end_time_str = get_word(words, num_words, 3);
1209
char *child_now_str = get_word(words, num_words, 4);
1210
1170
- RRDHOST *host = pluginsd_require_host_from_parent(parser, PLUGINSD_KEYWORD_REPLAY_BEGIN);
1211
+ RRDHOST *host = pluginsd_require_scope_host(parser, PLUGINSD_KEYWORD_REPLAY_BEGIN);
1212
if(!host) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
1213
1214
RRDSET *st;
1215
if (likely(!id || !*id))
1175
- st = pluginsd_require_chart_from_parent(parser, PLUGINSD_KEYWORD_REPLAY_BEGIN, PLUGINSD_KEYWORD_REPLAY_BEGIN);
1216
+ st = pluginsd_require_scope_chart(parser, PLUGINSD_KEYWORD_REPLAY_BEGIN, PLUGINSD_KEYWORD_REPLAY_BEGIN);
1217
else
1218
st = pluginsd_find_chart(host, id, PLUGINSD_KEYWORD_REPLAY_BEGIN);
1219
1220
if(!st) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
1180
- pluginsd_set_chart_from_parent(parser, st, PLUGINSD_KEYWORD_REPLAY_BEGIN);
1221
+
1222
+ if(!pluginsd_set_scope_chart(parser, st, PLUGINSD_KEYWORD_REPLAY_BEGIN))
1223
+ return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
1224
1225
if(start_time_str && end_time_str) {
1226
time_t start_time = (time_t) str2ull_encoded(start_time_str);
@@ -1291,10 +1334,10 @@ static inline PARSER_RC pluginsd_replay_set(char **words, size_t num_words, PARS
1334
char *value_str = get_word(words, num_words, 2);
1335
char *flags_str = get_word(words, num_words, 3);
1336
1294
- RRDHOST *host = pluginsd_require_host_from_parent(parser, PLUGINSD_KEYWORD_REPLAY_SET);
1337
+ RRDHOST *host = pluginsd_require_scope_host(parser, PLUGINSD_KEYWORD_REPLAY_SET);
1338
if(!host) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
1339
1297
- RRDSET *st = pluginsd_require_chart_from_parent(parser, PLUGINSD_KEYWORD_REPLAY_SET, PLUGINSD_KEYWORD_REPLAY_BEGIN);
1340
+ RRDSET *st = pluginsd_require_scope_chart(parser, PLUGINSD_KEYWORD_REPLAY_SET, PLUGINSD_KEYWORD_REPLAY_BEGIN);
1341
if(!st) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
1342
1343
if(!parser->user.replay.rset_enabled) {
@@ -1309,6 +1352,8 @@ static inline PARSER_RC pluginsd_replay_set(char **words, size_t num_words, PARS
1352
RRDDIM *rd = pluginsd_acquire_dimension(host, st, dimension, PLUGINSD_KEYWORD_REPLAY_SET);
1353
if(!rd) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
1354
1355
+ st->pluginsd.set = true;
1356
+
1357
if (unlikely(!parser->user.replay.start_time || !parser->user.replay.end_time)) {
1358
netdata_log_error("PLUGINSD: 'host:%s/chart:%s/dim:%s' got a %s with invalid timestamps %ld to %ld from a %s. Disabling it.",
1359
rrdhost_hostname(host),
@@ -1364,12 +1409,18 @@ static inline PARSER_RC pluginsd_replay_rrddim_collection_state(char **words, si
1409
char *last_calculated_value_str = get_word(words, num_words, 4);
1410
char *last_stored_value_str = get_word(words, num_words, 5);
1411
1367
- RRDHOST *host = pluginsd_require_host_from_parent(parser, PLUGINSD_KEYWORD_REPLAY_RRDDIM_STATE);
1412
+ RRDHOST *host = pluginsd_require_scope_host(parser, PLUGINSD_KEYWORD_REPLAY_RRDDIM_STATE);
1413
if(!host) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
1414
1370
- RRDSET *st = pluginsd_require_chart_from_parent(parser, PLUGINSD_KEYWORD_REPLAY_RRDDIM_STATE, PLUGINSD_KEYWORD_REPLAY_BEGIN);
1415
+ RRDSET *st = pluginsd_require_scope_chart(parser, PLUGINSD_KEYWORD_REPLAY_RRDDIM_STATE, PLUGINSD_KEYWORD_REPLAY_BEGIN);
1416
if(!st) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
1417
1418
+ if(st->pluginsd.set) {
1419
+ // reset pos to reuse the same RDAs
1420
+ st->pluginsd.pos = 0;
1421
+ st->pluginsd.set = false;
1422
+ }
1423
+
1424
RRDDIM *rd = pluginsd_acquire_dimension(host, st, dimension, PLUGINSD_KEYWORD_REPLAY_RRDDIM_STATE);
1425
if(!rd) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
1426
@@ -1394,10 +1445,11 @@ static inline PARSER_RC pluginsd_replay_rrdset_collection_state(char **words, si
1445
char *last_collected_ut_str = get_word(words, num_words, 1);
1446
char *last_updated_ut_str = get_word(words, num_words, 2);
1447
1397
- RRDHOST *host = pluginsd_require_host_from_parent(parser, PLUGINSD_KEYWORD_REPLAY_RRDSET_STATE);
1448
+ RRDHOST *host = pluginsd_require_scope_host(parser, PLUGINSD_KEYWORD_REPLAY_RRDSET_STATE);
1449
if(!host) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
1450
1400
- RRDSET *st = pluginsd_require_chart_from_parent(parser, PLUGINSD_KEYWORD_REPLAY_RRDSET_STATE, PLUGINSD_KEYWORD_REPLAY_BEGIN);
1451
+ RRDSET *st = pluginsd_require_scope_chart(parser, PLUGINSD_KEYWORD_REPLAY_RRDSET_STATE,
1452
+ PLUGINSD_KEYWORD_REPLAY_BEGIN);
1453
if(!st) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
1454
1455
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;
@@ -1446,10 +1498,10 @@ static inline PARSER_RC pluginsd_replay_end(char **words, size_t num_words, PARS
1498
time_t child_world_time = (child_world_time_txt && *child_world_time_txt) ? (time_t) str2ull_encoded(
1499
child_world_time_txt) : now_realtime_sec();
1500
1449
- RRDHOST *host = pluginsd_require_host_from_parent(parser, PLUGINSD_KEYWORD_REPLAY_END);
1501
+ RRDHOST *host = pluginsd_require_scope_host(parser, PLUGINSD_KEYWORD_REPLAY_END);
1502
if(!host) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
1503
1452
- RRDSET *st = pluginsd_require_chart_from_parent(parser, PLUGINSD_KEYWORD_REPLAY_END, PLUGINSD_KEYWORD_REPLAY_BEGIN);
1504
+ RRDSET *st = pluginsd_require_scope_chart(parser, PLUGINSD_KEYWORD_REPLAY_END, PLUGINSD_KEYWORD_REPLAY_BEGIN);
1505
if(!st) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
1506
1507
#ifdef NETDATA_LOG_REPLICATION_REQUESTS
@@ -1512,7 +1564,7 @@ static inline PARSER_RC pluginsd_replay_end(char **words, size_t num_words, PARS
1564
rrdhost_hostname(host), rrdset_id(st));
1565
#endif
1566
1515
- pluginsd_set_chart_from_parent(parser, NULL, PLUGINSD_KEYWORD_REPLAY_END);
1567
+ pluginsd_clear_scope_chart(parser, PLUGINSD_KEYWORD_REPLAY_END);
1568
1569
host->rrdpush_receiver_replication_percent = 100.0;
1570
worker_set_metric(WORKER_RECEIVER_JOB_REPLICATION_COMPLETION, host->rrdpush_receiver_replication_percent);
@@ -1520,7 +1572,7 @@ static inline PARSER_RC pluginsd_replay_end(char **words, size_t num_words, PARS
1572
return PARSER_RC_OK;
1573
}
1574
1523
- pluginsd_set_chart_from_parent(parser, NULL, PLUGINSD_KEYWORD_REPLAY_END);
1575
+ pluginsd_clear_scope_chart(parser, PLUGINSD_KEYWORD_REPLAY_END);
1576
1577
rrdcontext_updated_retention_rrdset(st);
1578
@@ -1541,7 +1593,7 @@ static inline PARSER_RC pluginsd_begin_v2(char **words, size_t num_words, PARSER
1593
if(unlikely(!id || !update_every_str || !end_time_str || !wall_clock_time_str))
1594
return PLUGINSD_DISABLE_PLUGIN(parser, PLUGINSD_KEYWORD_BEGIN_V2, "missing parameters");
1595
1544
- RRDHOST *host = pluginsd_require_host_from_parent(parser, PLUGINSD_KEYWORD_BEGIN_V2);
1596
+ RRDHOST *host = pluginsd_require_scope_host(parser, PLUGINSD_KEYWORD_BEGIN_V2);
1597
if(unlikely(!host)) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
1598
1599
timing_step(TIMING_STEP_BEGIN2_PREPARE);
@@ -1549,7 +1601,8 @@ static inline PARSER_RC pluginsd_begin_v2(char **words, size_t num_words, PARSER
1601
RRDSET *st = pluginsd_find_chart(host, id, PLUGINSD_KEYWORD_BEGIN_V2);
1602
if(unlikely(!st)) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
1603
1552
- pluginsd_set_chart_from_parent(parser, st, PLUGINSD_KEYWORD_BEGIN_V2);
1604
+ if(!pluginsd_set_scope_chart(parser, st, PLUGINSD_KEYWORD_BEGIN_V2))
1605
+ return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
1606
1607
if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_OBSOLETE | RRDSET_FLAG_ARCHIVED)))
1608
rrdset_isnot_obsolete(st);
@@ -1665,10 +1718,10 @@ static inline PARSER_RC pluginsd_set_v2(char **words, size_t num_words, PARSER *
1718
if(unlikely(!dimension || !collected_str || !value_str || !flags_str))
1719
return PLUGINSD_DISABLE_PLUGIN(parser, PLUGINSD_KEYWORD_SET_V2, "missing parameters");
1720
1668
- RRDHOST *host = pluginsd_require_host_from_parent(parser, PLUGINSD_KEYWORD_SET_V2);
1721
+ RRDHOST *host = pluginsd_require_scope_host(parser, PLUGINSD_KEYWORD_SET_V2);
1722
if(unlikely(!host)) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
1723
1671
- RRDSET *st = pluginsd_require_chart_from_parent(parser, PLUGINSD_KEYWORD_SET_V2, PLUGINSD_KEYWORD_BEGIN_V2);
1724
+ RRDSET *st = pluginsd_require_scope_chart(parser, PLUGINSD_KEYWORD_SET_V2, PLUGINSD_KEYWORD_BEGIN_V2);
1725
if(unlikely(!st)) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
1726
1727
timing_step(TIMING_STEP_SET2_PREPARE);
@@ -1676,6 +1729,8 @@ static inline PARSER_RC pluginsd_set_v2(char **words, size_t num_words, PARSER *
1729
RRDDIM *rd = pluginsd_acquire_dimension(host, st, dimension, PLUGINSD_KEYWORD_SET_V2);
1730
if(unlikely(!rd)) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
1731
1732
+ st->pluginsd.set = true;
1733
+
1734
if(unlikely(rrddim_flag_check(rd, RRDDIM_FLAG_OBSOLETE | RRDDIM_FLAG_ARCHIVED)))
1735
rrddim_isnot_obsolete(st, rd);
1736
@@ -1766,16 +1821,16 @@ static inline PARSER_RC pluginsd_set_v2(char **words, size_t num_words, PARSER *
1821
1822
void pluginsd_cleanup_v2(PARSER *parser) {
1823
// this is called when the thread is stopped while processing
1769
- pluginsd_set_chart_from_parent(parser, NULL, "THREAD CLEANUP");
1824
+ pluginsd_clear_scope_chart(parser, "THREAD CLEANUP");
1825
}
1826
1827
static inline PARSER_RC pluginsd_end_v2(char **words __maybe_unused, size_t num_words __maybe_unused, PARSER *parser) {
1828
timing_init();
1829
1775
- RRDHOST *host = pluginsd_require_host_from_parent(parser, PLUGINSD_KEYWORD_END_V2);
1830
+ RRDHOST *host = pluginsd_require_scope_host(parser, PLUGINSD_KEYWORD_END_V2);
1831
if(unlikely(!host)) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
1832
1778
- RRDSET *st = pluginsd_require_chart_from_parent(parser, PLUGINSD_KEYWORD_END_V2, PLUGINSD_KEYWORD_BEGIN_V2);
1833
+ RRDSET *st = pluginsd_require_scope_chart(parser, PLUGINSD_KEYWORD_END_V2, PLUGINSD_KEYWORD_BEGIN_V2);
1834
if(unlikely(!st)) return PLUGINSD_DISABLE_PLUGIN(parser, NULL, NULL);
1835
1836
parser->user.data_collections_count++;
@@ -1793,7 +1848,7 @@ static inline PARSER_RC pluginsd_end_v2(char **words __maybe_unused, size_t num_
1848
// ------------------------------------------------------------------------
1849
// unblock data collection
1850
1796
- pluginsd_unlock_previous_chart(parser, PLUGINSD_KEYWORD_END_V2, false);
1851
+ pluginsd_unlock_previous_scope_chart(parser, PLUGINSD_KEYWORD_END_V2, false);
1852
rrdcontext_collected_rrdset(st);
1853
store_metric_collection_completed();
1854
database/rrd.h
+3
-1
@@ -891,9 +891,11 @@ struct rrdset {
891
} alerts;
892
893
struct {
894
+ SPINLOCK spinlock; // used only for cleanup
895
+ pid_t collector_tid;
896
+ bool set;
897
uint32_t pos;
898
uint32_t size;
896
- uint32_t used;
899
RRDDIM_ACQUIRED **rda;
900
} pluginsd;
901