@cryptotaxi247 / netdata-1
Code Commits Issues Pulls Sessions CI/CD
master
master
c 1,932 lines 78.1 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "stream-sender-internals.h"
4 #include "stream-replication-sender.h"
5
6 #define MAX_REPLICATION_MESSAGE_PERCENT_SENDER_BUFFER 25ULL
7 #define MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED 50ULL
8 #define MIN_SENDER_BUFFER_PERCENTAGE_ALLOWED 10ULL
9
10 #define WORKER_JOB_FIND_NEXT 1
11 #define WORKER_JOB_QUERYING 2
12 #define WORKER_JOB_DELETE_ENTRY 3
13 #define WORKER_JOB_FIND_CHART 4
14 #define WORKER_JOB_PREPARE_QUERY 5
15 #define WORKER_JOB_CHECK_CONSISTENCY 6
16 #define WORKER_JOB_BUFFER_COMMIT 7
17 #define WORKER_JOB_CLEANUP 8
18 #define WORKER_JOB_WAIT 9
19
20 // master thread worker jobs
21 #define WORKER_JOB_STATISTICS 10
22 #define WORKER_JOB_CUSTOM_METRIC_PENDING_REQUESTS 11
23 #define WORKER_JOB_CUSTOM_METRIC_SKIPPED_NO_ROOM 12
24 #define WORKER_JOB_CUSTOM_METRIC_COMPLETION 13
25 #define WORKER_JOB_CUSTOM_METRIC_ADDED 14
26 #define WORKER_JOB_CUSTOM_METRIC_DONE 15
27 #define WORKER_JOB_CUSTOM_METRIC_SENDER_RESETS 16
28 #define WORKER_JOB_CUSTOM_METRIC_SENDER_FULL 17
29
30 #define ITERATIONS_IDLE_WITHOUT_PENDING_TO_RUN_SENDER_VERIFICATION 30
31 #define SECONDS_TO_RESET_POINT_IN_TIME 10
32
33 static struct replication_query_statistics replication_queries = {
34 .spinlock = SPINLOCK_INITIALIZER,
35 .queries_started = 0,
36 .queries_finished = 0,
37 .points_read = 0,
38 .points_generated = 0,
39 };
40
41 struct replication_query_statistics replication_get_query_statistics(void) {
42 spinlock_lock(&replication_queries.spinlock);
43 struct replication_query_statistics ret = replication_queries;
44 spinlock_unlock(&replication_queries.spinlock);
45 return ret;
46 }
47
48 static size_t replication_buffers_allocated = 0;
49
50 size_t replication_sender_allocated_buffers(void) {
51 return __atomic_load_n(&replication_buffers_allocated, __ATOMIC_RELAXED);
52 }
53
54 // ----------------------------------------------------------------------------
55 // sending replication replies
56
57 struct replication_dimension {
58 STORAGE_POINT sp;
59 struct storage_engine_query_handle handle;
60 bool enabled;
61 bool skip;
62
63 DICTIONARY *dict;
64 const DICTIONARY_ITEM *rda;
65 RRDDIM *rd;
66 };
67
68 struct replication_query {
69 RRDSET *st;
70
71 struct {
72 time_t first_entry_t;
73 time_t last_entry_t;
74 } db;
75
76 struct { // what the parent requested
77 time_t after;
78 time_t before;
79 bool enable_streaming;
80 } request;
81
82 struct { // what the child will do
83 time_t after;
84 time_t before;
85 bool enable_streaming;
86
87 bool locked_data_collection;
88 bool execute;
89 bool interrupted;
90 STREAM_CAPABILITIES capabilities;
91 } query;
92
93 time_t wall_clock_time;
94
95 size_t points_read;
96 size_t points_generated;
97
98 STORAGE_ENGINE_BACKEND backend;
99 struct replication_request *rq;
100
101 size_t dimensions;
102 struct replication_dimension data[];
103 };
104
105 ALWAYS_INLINE
106 static struct replication_query *replication_query_prepare(
107 RRDSET *st,
108 time_t db_first_entry,
109 time_t db_last_entry,
110 time_t requested_after,
111 time_t requested_before,
112 bool requested_enable_streaming,
113 time_t query_after,
114 time_t query_before,
115 bool query_enable_streaming,
116 time_t wall_clock_time,
117 STREAM_CAPABILITIES capabilities,
118 bool synchronous
119 ) {
120 size_t dimensions = rrdset_number_of_dimensions(st);
121 struct replication_query *q = callocz(1, sizeof(struct replication_query) + dimensions * sizeof(struct replication_dimension));
122 __atomic_add_fetch(&replication_buffers_allocated, sizeof(struct replication_query) + dimensions * sizeof(struct replication_dimension), __ATOMIC_RELAXED);
123
124 q->dimensions = dimensions;
125 q->st = st;
126
127 q->db.first_entry_t = db_first_entry;
128 q->db.last_entry_t = db_last_entry;
129
130 q->request.after = requested_after,
131 q->request.before = requested_before,
132 q->request.enable_streaming = requested_enable_streaming,
133
134 q->query.after = query_after;
135 q->query.before = query_before;
136 q->query.enable_streaming = query_enable_streaming;
137 q->query.capabilities = capabilities;
138
139 q->wall_clock_time = wall_clock_time;
140
141 if (!q->dimensions || !q->query.after || !q->query.before) {
142 q->query.execute = false;
143 q->dimensions = 0;
144 return q;
145 }
146
147 if(q->query.enable_streaming) {
148 spinlock_lock(&st->data_collection_lock);
149 q->query.locked_data_collection = true;
150
151 if (st->last_updated.tv_sec > q->query.before) {
152 #ifdef NETDATA_LOG_REPLICATION_REQUESTS
153 internal_error(true,
154 "STREAM SND REPLAY: 'host:%s/chart:%s' "
155 "has start_streaming = true, "
156 "adjusting replication before timestamp from %llu to %llu",
157 rrdhost_hostname(st->rrdhost), rrdset_id(st),
158 (unsigned long long) q->query.before,
159 (unsigned long long) st->last_updated.tv_sec
160 );
161 #endif
162 q->query.before = MIN(st->last_updated.tv_sec, wall_clock_time);
163 }
164 }
165
166 q->backend = st->rrdhost->db[0].eng->seb;
167
168 // prepare our array of dimensions
169 size_t count = 0;
170 RRDDIM *rd;
171 rrddim_foreach_read(rd, st) {
172 if (unlikely(!rd || !rd_dfe.item || !rrddim_check_upstream_exposed(rd)))
173 continue;
174
175 if (unlikely(rd_dfe.counter >= q->dimensions)) {
176 internal_error(true,
177 "STREAM SND REPLAY ERROR: 'host:%s/chart:%s' has more dimensions than the replicated ones",
178 rrdhost_hostname(st->rrdhost), rrdset_id(st));
179 break;
180 }
181
182 struct replication_dimension *d = &q->data[rd_dfe.counter];
183
184 d->dict = rd_dfe.dict;
185 d->rda = dictionary_acquired_item_dup(rd_dfe.dict, rd_dfe.item);
186 d->rd = rd;
187
188 STORAGE_PRIORITY priority = (synchronous) ? STORAGE_PRIORITY_SYNCHRONOUS_FIRST : STORAGE_PRIORITY_LOW;
189
190 stream_control_replication_query_started();
191 storage_engine_query_init(q->backend, rd->tiers[0].smh, &d->handle,
192 q->query.after, q->query.before, priority);
193 d->enabled = true;
194 d->skip = false;
195 count++;
196 }
197 rrddim_foreach_done(rd);
198
199 if(!count) {
200 // no data for this chart
201
202 q->query.execute = false;
203
204 if(q->query.locked_data_collection) {
205 spinlock_unlock(&st->data_collection_lock);
206 q->query.locked_data_collection = false;
207 }
208
209 }
210 else {
211 // we have data for this chart
212
213 q->query.execute = true;
214 }
215
216 return q;
217 }
218
219 static void replication_send_chart_collection_state(BUFFER *wb, RRDSET *st, STREAM_CAPABILITIES capabilities) {
220 bool with_slots = (capabilities & STREAM_CAP_SLOTS) ? true : false;
221 NUMBER_ENCODING integer_encoding = (capabilities & STREAM_CAP_IEEE754) ? NUMBER_ENCODING_BASE64 : NUMBER_ENCODING_DECIMAL;
222 RRDDIM *rd;
223 rrddim_foreach_read(rd, st){
224 if (!rrddim_check_upstream_exposed(rd)) continue;
225
226 buffer_fast_strcat(wb, PLUGINSD_KEYWORD_REPLAY_RRDDIM_STATE, sizeof(PLUGINSD_KEYWORD_REPLAY_RRDDIM_STATE) - 1);
227
228 if(with_slots) {
229 buffer_fast_strcat(wb, " "PLUGINSD_KEYWORD_SLOT":", sizeof(PLUGINSD_KEYWORD_SLOT) - 1 + 2);
230 buffer_print_uint64_encoded(wb, integer_encoding, rd->stream.snd.dim_slot);
231 }
232
233 buffer_fast_strcat(wb, " '", 2);
234 buffer_fast_strcat(wb, rrddim_id(rd), string_strlen(rd->id));
235 buffer_fast_strcat(wb, "' ", 2);
236 buffer_print_uint64_encoded(wb, integer_encoding, (usec_t) rd->collector.last_collected_time.tv_sec * USEC_PER_SEC +
237 (usec_t) rd->collector.last_collected_time.tv_usec);
238 buffer_fast_strcat(wb, " ", 1);
239
240 bool send_double_baseline = rrddim_is_float(rd) && (capabilities & STREAM_CAP_FLOAT_BASELINE);
241 if(send_double_baseline)
242 buffer_print_netdata_double_encoded(wb, integer_encoding, rrddim_last_collected_as_double(rd));
243 else
244 buffer_print_int64_encoded(wb, integer_encoding, rrddim_last_collected_raw_int(rd));
245
246 buffer_fast_strcat(wb, " ", 1);
247 buffer_print_netdata_double_encoded(wb, integer_encoding, rd->collector.last_calculated_value);
248 buffer_fast_strcat(wb, " ", 1);
249 buffer_print_netdata_double_encoded(wb, integer_encoding, rd->collector.last_stored_value);
250 buffer_fast_strcat(wb, "\n", 1);
251 }
252 rrddim_foreach_done(rd);
253
254 buffer_fast_strcat(wb, PLUGINSD_KEYWORD_REPLAY_RRDSET_STATE " ", sizeof(PLUGINSD_KEYWORD_REPLAY_RRDSET_STATE) - 1 + 1);
255 buffer_print_uint64_encoded(wb, integer_encoding, (usec_t) st->last_collected_time.tv_sec * USEC_PER_SEC + (usec_t) st->last_collected_time.tv_usec);
256 buffer_fast_strcat(wb, " ", 1);
257 buffer_print_uint64_encoded(wb, integer_encoding, (usec_t) st->last_updated.tv_sec * USEC_PER_SEC + (usec_t) st->last_updated.tv_usec);
258 buffer_fast_strcat(wb, "\n", 1);
259 }
260
261 static void replication_query_finalize(BUFFER *wb, struct replication_query *q, bool executed) {
262 size_t dimensions = q->dimensions;
263
264 if(wb && q->query.enable_streaming)
265 replication_send_chart_collection_state(wb, q->st, q->query.capabilities);
266
267 if(q->query.locked_data_collection) {
268 spinlock_unlock(&q->st->data_collection_lock);
269 q->query.locked_data_collection = false;
270 }
271
272 // release all the dictionary items acquired
273 // finalize the queries
274 size_t queries = 0;
275
276 for (size_t i = 0; i < dimensions; i++) {
277 struct replication_dimension *d = &q->data[i];
278 if (unlikely(!d->enabled)) continue;
279
280 storage_engine_query_finalize(&d->handle);
281 stream_control_replication_query_finished();
282
283 dictionary_acquired_item_release(d->dict, d->rda);
284
285 // update global statistics
286 queries++;
287 }
288
289 if(executed) {
290 spinlock_lock(&replication_queries.spinlock);
291 replication_queries.queries_started += queries;
292 replication_queries.queries_finished += queries;
293 replication_queries.points_read += q->points_read;
294 replication_queries.points_generated += q->points_generated;
295
296 if(q->st && q->st->rrdhost->sender) {
297 struct sender_state *s = q->st->rrdhost->sender;
298 s->replication.latest_completed_before_t = q->query.before;
299 }
300
301 spinlock_unlock(&replication_queries.spinlock);
302 }
303
304 __atomic_sub_fetch(&replication_buffers_allocated, sizeof(struct replication_query) + dimensions * sizeof(struct replication_dimension), __ATOMIC_RELAXED);
305 freez(q);
306 }
307
308 static void replication_query_align_to_optimal_before(struct replication_query *q) {
309 if(!q->query.execute || q->query.enable_streaming)
310 return;
311
312 size_t dimensions = q->dimensions;
313 time_t expanded_before = 0;
314
315 for (size_t i = 0; i < dimensions; i++) {
316 struct replication_dimension *d = &q->data[i];
317 if(unlikely(!d->enabled)) continue;
318
319 time_t new_before = storage_engine_align_to_optimal_before(&d->handle);
320 if (!expanded_before || new_before < expanded_before)
321 expanded_before = new_before;
322 }
323
324 if(expanded_before > q->query.before && // it is later than the original
325 (expanded_before - q->query.before) / q->st->update_every < 1024 && // it is reasonable (up to a page)
326 expanded_before < q->st->last_updated.tv_sec && // it is not the chart's last updated time
327 expanded_before < q->wall_clock_time) // it is not later than the wall clock time
328 q->query.before = expanded_before;
329 }
330
331 static bool replication_query_execute(BUFFER *wb, struct replication_query *q, size_t max_msg_size) {
332 replication_query_align_to_optimal_before(q);
333
334 bool with_slots = (q->query.capabilities & STREAM_CAP_SLOTS) ? true : false;
335 NUMBER_ENCODING integer_encoding = (q->query.capabilities & STREAM_CAP_IEEE754) ? NUMBER_ENCODING_BASE64 : NUMBER_ENCODING_DECIMAL;
336 time_t after = q->query.after;
337 time_t before = q->query.before;
338 size_t dimensions = q->dimensions;
339 time_t wall_clock_time = q->wall_clock_time;
340
341 bool finished_with_gap = false;
342 size_t points_read = 0, points_generated = 0;
343
344 #ifdef NETDATA_LOG_REPLICATION_REQUESTS
345 time_t actual_after = 0, actual_before = 0;
346 #endif
347
348 time_t now = after + 1;
349 time_t last_end_time_in_buffer = 0;
350 while(now <= before) {
351 time_t min_start_time = 0, max_start_time = 0, min_end_time = 0, max_end_time = 0, min_update_every = 0, max_update_every = 0;
352 for (size_t i = 0; i < dimensions ;i++) {
353 struct replication_dimension *d = &q->data[i];
354 if(unlikely(!d->enabled || d->skip)) continue;
355
356 // fetch the first valid point for the dimension
357 int max_skip = 1000;
358 while(d->sp.end_time_s < now && !storage_engine_query_is_finished(&d->handle) && max_skip-- >= 0) {
359 d->sp = storage_engine_query_next_metric(&d->handle);
360 points_read++;
361 }
362
363 if(max_skip <= 0) {
364 d->skip = true;
365
366 nd_log_limit_static_global_var(erl, 1, 0);
367 nd_log_limit(&erl, NDLS_DAEMON, NDLP_ERR,
368 "STREAM SND REPLAY: 'host:%s/chart:%s/dim:%s': db does not advance the query "
369 "beyond time %llu (tried 1000 times to get the next point and always got back a point in the past)",
370 rrdhost_hostname(q->st->rrdhost), rrdset_id(q->st), rrddim_id(d->rd),
371 (unsigned long long) now);
372
373 continue;
374 }
375
376 if(unlikely(d->sp.end_time_s < now || d->sp.end_time_s < d->sp.start_time_s))
377 // this dimension does not provide any data
378 continue;
379
380 time_t update_every = d->sp.end_time_s - d->sp.start_time_s;
381 if(unlikely(!update_every))
382 update_every = q->st->update_every;
383
384 if(unlikely(!min_update_every))
385 min_update_every = update_every;
386
387 if(unlikely(!min_start_time))
388 min_start_time = d->sp.start_time_s;
389
390 if(unlikely(!min_end_time))
391 min_end_time = d->sp.end_time_s;
392
393 min_update_every = MIN(min_update_every, update_every);
394 max_update_every = MAX(max_update_every, update_every);
395
396 min_start_time = MIN(min_start_time, d->sp.start_time_s);
397 max_start_time = MAX(max_start_time, d->sp.start_time_s);
398
399 min_end_time = MIN(min_end_time, d->sp.end_time_s);
400 max_end_time = MAX(max_end_time, d->sp.end_time_s);
401 }
402
403 if (unlikely(min_update_every != max_update_every ||
404 min_start_time != max_start_time)) {
405
406 time_t fix_min_start_time;
407 if(last_end_time_in_buffer &&
408 last_end_time_in_buffer >= min_start_time &&
409 last_end_time_in_buffer <= max_start_time) {
410 fix_min_start_time = last_end_time_in_buffer;
411 }
412 else
413 fix_min_start_time = min_end_time - min_update_every;
414
415 #ifdef NETDATA_INTERNAL_CHECKS
416 nd_log_limit_static_global_var(erl, 1, 0);
417 nd_log_limit(&erl, NDLS_DAEMON, NDLP_WARNING,
418 "STREAM SND REPLAY WARNING: 'host:%s/chart:%s' misaligned dimensions, "
419 "update every (min: %ld, max: %ld), "
420 "start time (min: %ld, max: %ld), "
421 "end time (min %ld, max %ld), "
422 "now %ld, last end time sent %ld, "
423 "min start time is fixed to %ld",
424 rrdhost_hostname(q->st->rrdhost), rrdset_id(q->st),
425 min_update_every, max_update_every,
426 min_start_time, max_start_time,
427 min_end_time, max_end_time,
428 now, last_end_time_in_buffer,
429 fix_min_start_time
430 );
431 #endif
432
433 min_start_time = fix_min_start_time;
434 }
435
436 if(likely(min_start_time <= now && min_end_time >= now)) {
437 // we have a valid point
438
439 if (unlikely(min_end_time == min_start_time))
440 min_start_time = min_end_time - q->st->update_every;
441
442 #ifdef NETDATA_LOG_REPLICATION_REQUESTS
443 if (unlikely(!actual_after))
444 actual_after = min_end_time;
445
446 actual_before = min_end_time;
447 #endif
448
449 if(buffer_strlen(wb) > max_msg_size && last_end_time_in_buffer) {
450 q->query.before = last_end_time_in_buffer;
451
452 // CRITICAL: When splitting a response due to buffer overflow, we MUST set
453 // enable_streaming=false to prevent data loss. If we send start_streaming=true
454 // with partial data, the parent will think replication is complete when we've
455 // actually truncated the requested interval, causing permanent data loss.
456 //
457 // The parent-side stuck detection will handle infinite loops differently by:
458 // 1. Detecting when no progress is made after multiple rounds
459 // 2. Explicitly marking replication as FINISHED before sending final request
460 // 3. Handling the child's response appropriately whether it says true or false
461 q->query.enable_streaming = false;
462
463 internal_error(
464 true,
465 "STREAM SND REPLAY: current remaining sender buffer of %zu bytes cannot fit the "
466 "message size %zu bytes for chart '%s' of host '%s'. "
467 "Sending partial replication response %ld to %ld, %s (original: %ld to %ld, %s).",
468 buffer_strlen(wb), max_msg_size, rrdset_id(q->st), rrdhost_hostname(q->st->rrdhost),
469 q->query.after, q->query.before, q->query.enable_streaming?"true":"false",
470 q->request.after, q->request.before, q->request.enable_streaming?"true":"false");
471
472 q->query.interrupted = true;
473
474 break;
475 }
476 last_end_time_in_buffer = min_end_time;
477
478 buffer_fast_strcat(wb, PLUGINSD_KEYWORD_REPLAY_BEGIN, sizeof(PLUGINSD_KEYWORD_REPLAY_BEGIN) - 1);
479
480 if(with_slots) {
481 buffer_fast_strcat(wb, " "PLUGINSD_KEYWORD_SLOT":", sizeof(PLUGINSD_KEYWORD_SLOT) - 1 + 2);
482 buffer_print_uint64_encoded(wb, integer_encoding, q->st->stream.snd.chart_slot);
483 }
484
485 buffer_fast_strcat(wb, " '' ", 4);
486 buffer_print_uint64_encoded(wb, integer_encoding, min_start_time);
487 buffer_fast_strcat(wb, " ", 1);
488 buffer_print_uint64_encoded(wb, integer_encoding, min_end_time);
489 buffer_fast_strcat(wb, " ", 1);
490 buffer_print_uint64_encoded(wb, integer_encoding, wall_clock_time);
491 buffer_fast_strcat(wb, "\n", 1);
492
493 // output the replay values for this time
494 for (size_t i = 0; i < dimensions; i++) {
495 struct replication_dimension *d = &q->data[i];
496 if (unlikely(!d->enabled)) continue;
497
498 if (likely( d->sp.start_time_s <= min_end_time &&
499 d->sp.end_time_s >= min_end_time &&
500 !storage_point_is_unset(d->sp) &&
501 !storage_point_is_gap(d->sp))) {
502
503 buffer_fast_strcat(wb, PLUGINSD_KEYWORD_REPLAY_SET, sizeof(PLUGINSD_KEYWORD_REPLAY_SET) - 1);
504
505 if(with_slots) {
506 buffer_fast_strcat(wb, " "PLUGINSD_KEYWORD_SLOT":", sizeof(PLUGINSD_KEYWORD_SLOT) - 1 + 2);
507 buffer_print_uint64_encoded(wb, integer_encoding, d->rd->stream.snd.dim_slot);
508 }
509
510 buffer_fast_strcat(wb, " \"", 2);
511 buffer_fast_strcat(wb, rrddim_id(d->rd), string_strlen(d->rd->id));
512 buffer_fast_strcat(wb, "\" ", 2);
513 buffer_print_netdata_double_encoded(wb, integer_encoding, d->sp.sum);
514 buffer_fast_strcat(wb, " ", 1);
515 buffer_print_sn_flags(wb, d->sp.flags, q->query.capabilities & STREAM_CAP_INTERPOLATED);
516 buffer_fast_strcat(wb, "\n", 1);
517
518 points_generated++;
519 }
520 }
521
522 now = min_end_time + 1;
523 }
524 else if(unlikely(min_end_time < now))
525 // the query does not progress
526 break;
527 else {
528 // we have gap - all points are in the future
529 now = min_start_time;
530
531 if(min_start_time > before && !points_generated) {
532 before = q->query.before = min_start_time - 1;
533 finished_with_gap = true;
534 break;
535 }
536 }
537 }
538
539 #ifdef NETDATA_LOG_REPLICATION_REQUESTS
540 if(actual_after) {
541 char actual_after_buf[LOG_DATE_LENGTH + 1], actual_before_buf[LOG_DATE_LENGTH + 1];
542 log_date(actual_after_buf, LOG_DATE_LENGTH, actual_after);
543 log_date(actual_before_buf, LOG_DATE_LENGTH, actual_before);
544 internal_error(true,
545 "STREAM SND REPLAY: 'host:%s/chart:%s': sending data %llu [%s] to %llu [%s] (requested %llu [delta %lld] to %llu [delta %lld])",
546 rrdhost_hostname(q->st->rrdhost), rrdset_id(q->st),
547 (unsigned long long)actual_after, actual_after_buf, (unsigned long long)actual_before, actual_before_buf,
548 (unsigned long long)after, (long long)(actual_after - after), (unsigned long long)before, (long long)(actual_before - before));
549 }
550 else
551 internal_error(true,
552 "STREAM SND REPLAY: 'host:%s/chart:%s': nothing to send (requested %llu to %llu)",
553 rrdhost_hostname(q->st->rrdhost), rrdset_id(q->st),
554 (unsigned long long)after, (unsigned long long)before);
555 #endif // NETDATA_LOG_REPLICATION_REQUESTS
556
557 q->points_read += points_read;
558 q->points_generated += points_generated;
559
560 if(last_end_time_in_buffer < before - q->st->update_every)
561 finished_with_gap = true;
562
563 return finished_with_gap;
564 }
565
566 ALWAYS_INLINE
567 static struct replication_query *replication_response_prepare(
568 RRDSET *st,
569 bool requested_enable_streaming,
570 time_t requested_after,
571 time_t requested_before,
572 STREAM_CAPABILITIES capabilities,
573 bool synchronous
574 ) {
575
576 bool query_enable_streaming = requested_enable_streaming;
577 time_t query_after = requested_after;
578 time_t query_before = requested_before;
579
580 time_t wall_clock_time = now_realtime_sec();
581
582 if(query_after > query_before)
583 SWAP(query_before, query_after);
584
585 if(!query_after || !query_before || query_after > wall_clock_time) {
586 query_after = 0;
587 query_before = 0;
588 query_enable_streaming = true;
589 }
590 else if(query_before >= (wall_clock_time - (st->update_every * 100))) {
591 query_before = wall_clock_time;
592 query_enable_streaming = true;
593 }
594
595 time_t db_first_entry = 0, db_last_entry = 0;
596 rrdset_get_retention_of_tier_for_collected_chart(
597 st, &db_first_entry, &db_last_entry, wall_clock_time, 0);
598
599 if(query_after && query_before) {
600 if (query_after < db_first_entry)
601 query_after = db_first_entry;
602
603 if (query_before > db_last_entry)
604 query_before = db_last_entry;
605
606 if (query_after > query_before)
607 SWAP(query_after, query_before);
608
609 if (query_enable_streaming || query_before >= db_last_entry) {
610 query_before = db_last_entry;
611 query_enable_streaming = true;
612 }
613 }
614
615 return replication_query_prepare(
616 st,
617 db_first_entry, db_last_entry,
618 requested_after, requested_before, requested_enable_streaming,
619 query_after, query_before, query_enable_streaming,
620 wall_clock_time, capabilities, synchronous);
621 }
622
623 static inline void replication_response_cancel_and_finalize(struct replication_query *q) {
624 if(!q) return;
625 replication_query_finalize(NULL, q, false);
626 }
627
628 static bool sender_is_still_connected_for_this_request(struct replication_request *rq);
629 static void replication_replied_add(void);
630
631 bool replication_response_execute_finalize_and_send(struct replication_query *q, size_t max_msg_size, bool workers) {
632 bool with_slots = (q->query.capabilities & STREAM_CAP_SLOTS) ? true : false;
633 NUMBER_ENCODING integer_encoding = (q->query.capabilities & STREAM_CAP_IEEE754) ? NUMBER_ENCODING_BASE64 : NUMBER_ENCODING_DECIMAL;
634 struct replication_request *rq = q->rq;
635 RRDSET *st = q->st;
636 RRDHOST *host = st->rrdhost;
637
638 // we might want to optimize this by filling a temporary buffer
639 // and copying the result to the host's buffer in order to avoid
640 // holding the host's buffer lock for too long
641 BUFFER *wb = sender_thread_buffer(host->sender, REPLICATION_THREAD_BUFFER_INITIAL_SIZE);
642
643 buffer_fast_strcat(wb, PLUGINSD_KEYWORD_REPLAY_BEGIN, sizeof(PLUGINSD_KEYWORD_REPLAY_BEGIN) - 1);
644
645 if(with_slots) {
646 buffer_fast_strcat(wb, " "PLUGINSD_KEYWORD_SLOT":", sizeof(PLUGINSD_KEYWORD_SLOT) - 1 + 2);
647 buffer_print_uint64_encoded(wb, integer_encoding, q->st->stream.snd.chart_slot);
648 }
649
650 buffer_fast_strcat(wb, " '", 2);
651 buffer_fast_strcat(wb, rrdset_id(st), string_strlen(st->id));
652 buffer_fast_strcat(wb, "'\n", 2);
653
654 bool locked_data_collection = q->query.locked_data_collection;
655 q->query.locked_data_collection = false;
656
657 bool finished_with_gap = false;
658 if(q->query.execute)
659 finished_with_gap = replication_query_execute(wb, q, max_msg_size);
660
661 time_t after = q->query.after;
662 time_t before = q->query.before;
663 bool enable_streaming = q->query.enable_streaming;
664
665 replication_query_finalize(wb, q, q->query.execute);
666 q = NULL; // IMPORTANT: q is invalid now
667
668 // get a fresh retention to send to the parent
669 time_t wall_clock_time = now_realtime_sec();
670 time_t db_first_entry, db_last_entry;
671 rrdset_get_retention_of_tier_for_collected_chart(st, &db_first_entry, &db_last_entry, wall_clock_time, 0);
672
673 // end with first/last entries we have, and the first start time and
674 // last end time of the data we sent
675
676 buffer_fast_strcat(wb, PLUGINSD_KEYWORD_REPLAY_END " ", sizeof(PLUGINSD_KEYWORD_REPLAY_END) - 1 + 1);
677 buffer_print_int64_encoded(wb, integer_encoding, st->update_every);
678 buffer_fast_strcat(wb, " ", 1);
679 buffer_print_uint64_encoded(wb, integer_encoding, db_first_entry);
680 buffer_fast_strcat(wb, " ", 1);
681 buffer_print_uint64_encoded(wb, integer_encoding, db_last_entry);
682
683 buffer_fast_strcat(wb, enable_streaming ? " true " : " false ", 7);
684
685 buffer_print_uint64_encoded(wb, integer_encoding, after);
686 buffer_fast_strcat(wb, " ", 1);
687 buffer_print_uint64_encoded(wb, integer_encoding, before);
688 buffer_fast_strcat(wb, " ", 1);
689 buffer_print_uint64_encoded(wb, integer_encoding, wall_clock_time);
690 buffer_fast_strcat(wb, "\n", 1);
691
692 if(workers) worker_is_busy(WORKER_JOB_BUFFER_COMMIT);
693 sender_commit(host->sender, wb, STREAM_TRAFFIC_TYPE_REPLICATION);
694 if(workers) worker_is_busy(WORKER_JOB_CLEANUP);
695 __atomic_add_fetch(&host->stream.snd.status.replication.counter_out, 1, __ATOMIC_RELAXED);
696 replication_replied_add();
697
698 if(enable_streaming) {
699 #ifdef REPLICATION_TRACKING
700 st->stream.snd.who = REPLAY_WHO_FINISHED;
701 #endif
702
703 if(sender_is_still_connected_for_this_request(rq)) {
704 // enable normal streaming if we have to
705 // but only if the sender buffer has not been flushed since we started
706
707 RRDSET_FLAGS old = rrdset_flag_set_and_clear(st, RRDSET_FLAG_SENDER_REPLICATION_FINISHED, RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS);
708 if(!(old & RRDSET_FLAG_SENDER_REPLICATION_FINISHED)) {
709 if(rrdhost_sender_replicating_charts_minus_one(st->rrdhost) == 0)
710 pulse_host_status(st->rrdhost, PULSE_HOST_STATUS_SND_RUNNING, 0);
711
712 if(!finished_with_gap)
713 st->stream.snd.resync_time_s = 0;
714
715 #ifdef NETDATA_LOG_REPLICATION_REQUESTS
716 internal_error(true, "STREAM SND REPLAY: 'host:%s/chart:%s' streaming starts",
717 rrdhost_hostname(st->rrdhost), rrdset_id(st));
718 #endif
719 }
720 else
721 internal_error(
722 true,
723 "STREAM SND REPLAY ERROR: 'host:%s/chart:%s' "
724 "received start streaming command, but the chart was not in progress replicating",
725 rrdhost_hostname(st->rrdhost), rrdset_id(st));
726 }
727 }
728 else {
729 #ifdef REPLICATION_TRACKING
730 st->stream.snd.who = REPLAY_WHO_THEM;
731 #endif
732 }
733
734 if(locked_data_collection)
735 spinlock_unlock(&st->data_collection_lock);
736
737 return enable_streaming;
738 }
739
740 // ----------------------------------------------------------------------------
741 // replication thread
742
743 // replication request in sender DICTIONARY
744 // used for de-duplicating the requests
745 struct replication_request {
746 struct sender_state *sender; // the sender we should put the reply at
747 STRING *chart_id; // the chart of the request
748 time_t after; // the start time of the query (maybe zero) key for sorting (JudyL)
749 time_t before; // the end time of the query (maybe zero)
750
751 usec_t sender_circular_buffer_last_flush_ut; // the timestamp of the sender, at the time we indexed this request
752 Word_t unique_id; // auto-increment, later requests have bigger
753
754 bool start_streaming; // true, when the parent wants to send the rest of the data (before is overwritten) and enable normal streaming
755 bool indexed_in_judy; // true when the request is indexed in judy
756 bool not_indexed_buffer_full; // true when the request is not indexed because the sender is full
757 bool not_indexed_preprocessing; // true when the request is not indexed, but it is pending in preprocessing
758
759 // prepare ahead members - preprocessing
760 bool found; // used as a result boolean for the find call
761 bool executed; // used to detect if we have skipped requests while preprocessing
762 RRDSET *st; // caching of the chart during preprocessing
763 struct replication_query *q; // the preprocessing query initialization
764 };
765
766 // replication sort entry in JudyL array
767 // used for sorting all requests, across all nodes
768 struct replication_sort_entry {
769 struct replication_request *rq;
770
771 size_t unique_id; // used as a key to identify the sort entry - we never access its contents
772 };
773
774 // the global variables for the replication thread
775 static struct replication_thread {
776 ARAL *aral_rse;
777
778 SPINLOCK spinlock;
779
780 struct {
781 size_t pending; // number of requests pending in the queue
782
783 // statistics
784 size_t added; // number of requests added to the queue
785 size_t removed; // number of requests removed from the queue
786 size_t pending_no_room; // number of requests skipped, because the sender has no room for responses
787 size_t senders_full; // number of times a sender reset our last position in the queue
788 size_t sender_resets; // number of times a sender reset our last position in the queue
789 time_t first_time_t; // the minimum 'after' we encountered
790
791 struct {
792 Word_t after;
793 Word_t unique_id;
794 Pvoid_t JudyL_array;
795 } queue;
796
797 } unsafe; // protected from replication_recursive_lock()
798
799 struct {
800 Word_t unique_id; // the last unique id we gave to a request (auto-increment, starting from 1)
801 size_t received; // the number of replication requests received
802 size_t executed; // the number of replication requests executed
803 size_t replied;
804 size_t error_not_found; // the number of replication requests ignored because the chart was not found
805 size_t error_duplicate; // the number of replication requests found duplicate (same chart)
806 size_t error_flushed; // the number of replication requests deleted due to disconnections
807 size_t latest_first_time; // the 'after' timestamp of the last request we executed
808 int64_t memory; // the total memory allocated by replication
809 } atomic; // access should be with atomic operations
810
811 struct {
812 // same as the atomic versions, for finding the delta over time
813 size_t last_received;
814 size_t last_executed;
815 size_t last_replied;
816 size_t last_error_flushed;
817 size_t last_error_duplicate;
818 size_t last_error_not_found;
819
820 ND_THREAD **threads_ptrs;
821 size_t threads;
822 } main_thread; // access is allowed only by the main thread
823
824 } replication_globals = {
825 .aral_rse = NULL,
826 .spinlock = SPINLOCK_INITIALIZER,
827 .unsafe = {
828 .pending = 0,
829
830 .added = 0,
831 .removed = 0,
832 .pending_no_room = 0,
833 .sender_resets = 0,
834 .senders_full = 0,
835
836 .first_time_t = 0,
837
838 .queue = {
839 .after = 0,
840 .unique_id = 0,
841 .JudyL_array = NULL,
842 },
843 },
844 .atomic = {
845 .unique_id = 0,
846 .executed = 0,
847 .latest_first_time = 0,
848 .memory = 0,
849 },
850 .main_thread = {
851 .last_executed = 0,
852 .threads = 0,
853 .threads_ptrs = NULL,
854 },
855 };
856
857 int64_t replication_sender_allocated_memory(void) {
858 return __atomic_load_n(&replication_globals.atomic.memory, __ATOMIC_RELAXED);
859 }
860
861 #define replication_set_latest_first_time(t) __atomic_store_n(&replication_globals.atomic.latest_first_time, t, __ATOMIC_RELAXED)
862 #define replication_get_latest_first_time() __atomic_load_n(&replication_globals.atomic.latest_first_time, __ATOMIC_RELAXED)
863
864 static inline bool replication_recursive_lock_mode(char mode) {
865 static __thread int recursions = 0;
866
867 if(mode == 'L') { // (L)ock
868 if(++recursions == 1)
869 spinlock_lock(&replication_globals.spinlock);
870 }
871 else if(mode == 'U') { // (U)nlock
872 if(--recursions == 0)
873 spinlock_unlock(&replication_globals.spinlock);
874 }
875 else if(mode == 'C') { // (C)heck
876 if(recursions > 0)
877 return true;
878 else
879 return false;
880 }
881 else
882 fatal("REPLICATION: unknown lock mode '%c'", mode);
883
884 #ifdef NETDATA_INTERNAL_CHECKS
885 if(recursions < 0)
886 fatal("REPLICATION: recursions is %d", recursions);
887 #endif
888
889 return true;
890 }
891
892 #define replication_recursive_lock() replication_recursive_lock_mode('L')
893 #define replication_recursive_unlock() replication_recursive_lock_mode('U')
894 #define fatal_when_replication_is_not_locked_for_me() do { \
895 if(!replication_recursive_lock_mode('C')) \
896 fatal("REPLICATION: reached %s, but replication is not locked by this thread.", __FUNCTION__); \
897 } while(0)
898
899 void replication_set_next_point_in_time(time_t after, size_t unique_id) {
900 replication_recursive_lock();
901 replication_globals.unsafe.queue.after = after;
902 replication_globals.unsafe.queue.unique_id = unique_id;
903 replication_recursive_unlock();
904 }
905
906 // ----------------------------------------------------------------------------
907 // replication sort entry management
908
909 static inline struct replication_sort_entry *replication_sort_entry_create(struct replication_request *rq) {
910 struct replication_sort_entry *rse = aral_mallocz(replication_globals.aral_rse);
911 __atomic_add_fetch(&replication_globals.atomic.memory, sizeof(struct replication_sort_entry), __ATOMIC_RELAXED);
912
913 stream_sender_pending_replication_requests_plus_one(rq->sender);
914
915 // copy the request
916 rse->rq = rq;
917 rse->unique_id = __atomic_add_fetch(&replication_globals.atomic.unique_id, 1, __ATOMIC_SEQ_CST);
918
919 // save the unique id into the request, to be able to delete it later
920 rq->unique_id = rse->unique_id;
921 rq->indexed_in_judy = false;
922 rq->not_indexed_buffer_full = false;
923 rq->not_indexed_preprocessing = false;
924 return rse;
925 }
926
927 static void replication_sort_entry_destroy(struct replication_sort_entry *rse) {
928 aral_freez(replication_globals.aral_rse, rse);
929 __atomic_sub_fetch(&replication_globals.atomic.memory, sizeof(struct replication_sort_entry), __ATOMIC_RELAXED);
930 }
931
932 static void replication_sort_entry_add(struct replication_request *rq) {
933 if(unlikely(stream_sender_replication_buffer_full_get(rq->sender))) {
934 rq->indexed_in_judy = false;
935 rq->not_indexed_buffer_full = true;
936 rq->not_indexed_preprocessing = false;
937 replication_recursive_lock();
938 replication_globals.unsafe.pending_no_room++;
939 replication_recursive_unlock();
940 return;
941 }
942
943 // cache this, because it will be changed
944 bool decrement_no_room = rq->not_indexed_buffer_full;
945
946 struct replication_sort_entry *rse = replication_sort_entry_create(rq);
947
948 replication_recursive_lock();
949
950 if(decrement_no_room)
951 replication_globals.unsafe.pending_no_room--;
952
953 // if(rq->after < (time_t)replication_globals.protected.queue.after &&
954 // rq->sender->buffer_used_percentage <= MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED &&
955 // !replication_globals.protected.skipped_no_room_since_last_reset) {
956 //
957 // // make it find this request first
958 // replication_set_next_point_in_time(rq->after, rq->unique_id);
959 // }
960
961 replication_globals.unsafe.added++;
962 replication_globals.unsafe.pending++;
963
964 Pvoid_t *inner_judy_ptr;
965
966 JudyAllocThreadPulseReset();
967
968 // find the outer judy entry, using after as key
969 inner_judy_ptr = JudyLIns(&replication_globals.unsafe.queue.JudyL_array, (Word_t) rq->after, PJE0);
970 if(unlikely(!inner_judy_ptr || inner_judy_ptr == PJERR))
971 fatal("REPLICATION: corrupted outer judyL");
972
973 // add it to the inner judy, using unique_id as key
974 Pvoid_t *item = JudyLIns(inner_judy_ptr, rq->unique_id, PJE0);
975 if(unlikely(!item || item == PJERR))
976 fatal("REPLICATION: corrupted inner judyL");
977
978 *item = rse;
979 rq->indexed_in_judy = true;
980 rq->not_indexed_buffer_full = false;
981 rq->not_indexed_preprocessing = false;
982
983 if(!replication_globals.unsafe.first_time_t || rq->after < replication_globals.unsafe.first_time_t)
984 replication_globals.unsafe.first_time_t = rq->after;
985
986 replication_recursive_unlock();
987
988 __atomic_add_fetch(&replication_globals.atomic.memory, JudyAllocThreadPulseGetAndReset(), __ATOMIC_RELAXED);
989 }
990
991 static bool replication_sort_entry_unlink_and_free_unsafe(struct replication_sort_entry *rse, Pvoid_t **inner_judy_ppptr, bool preprocessing) {
992 fatal_when_replication_is_not_locked_for_me();
993
994 bool inner_judy_deleted = false;
995
996 replication_globals.unsafe.removed++;
997 replication_globals.unsafe.pending--;
998
999 stream_sender_pending_replication_requests_minus_one(rse->rq->sender);
1000
1001 rse->rq->indexed_in_judy = false;
1002 rse->rq->not_indexed_preprocessing = preprocessing;
1003
1004 JudyAllocThreadPulseReset();
1005
1006 // delete it from the inner judy
1007 JudyLDel(*inner_judy_ppptr, rse->rq->unique_id, PJE0);
1008
1009 // if no items left, delete it from the outer judy
1010 if(**inner_judy_ppptr == NULL) {
1011 JudyLDel(&replication_globals.unsafe.queue.JudyL_array, rse->rq->after, PJE0);
1012 inner_judy_deleted = true;
1013 }
1014
1015 // free memory
1016 replication_sort_entry_destroy(rse);
1017
1018 __atomic_add_fetch(&replication_globals.atomic.memory, JudyAllocThreadPulseGetAndReset(), __ATOMIC_RELAXED);
1019
1020 return inner_judy_deleted;
1021 }
1022
1023 static void replication_sort_entry_del(struct replication_request *rq, bool buffer_full) {
1024 Pvoid_t *inner_judy_pptr;
1025 struct replication_sort_entry *rse_to_delete = NULL;
1026
1027 replication_recursive_lock();
1028 if(rq->indexed_in_judy) {
1029
1030 inner_judy_pptr = JudyLGet(replication_globals.unsafe.queue.JudyL_array, rq->after, PJE0);
1031 if (inner_judy_pptr) {
1032 Pvoid_t *our_item_pptr = JudyLGet(*inner_judy_pptr, rq->unique_id, PJE0);
1033 if (our_item_pptr) {
1034 rse_to_delete = *our_item_pptr;
1035 replication_sort_entry_unlink_and_free_unsafe(rse_to_delete, &inner_judy_pptr, false);
1036
1037 if(buffer_full) {
1038 replication_globals.unsafe.pending_no_room++;
1039 rq->not_indexed_buffer_full = true;
1040 }
1041 }
1042 }
1043
1044 if (!rse_to_delete)
1045 fatal("STREAM SND REPLAY: 'host:%s/chart:%s' Cannot find sort entry to delete for time %ld.",
1046 rrdhost_hostname(rq->sender->host), string2str(rq->chart_id), rq->after);
1047
1048 }
1049
1050 replication_recursive_unlock();
1051 }
1052
1053 ALWAYS_INLINE_HOT
1054 static struct replication_request replication_request_get_first_available() {
1055 Pvoid_t *inner_judy_pptr;
1056
1057 replication_recursive_lock();
1058
1059 struct replication_request rq_to_return = (struct replication_request){ .found = false };
1060
1061 if(unlikely(!replication_globals.unsafe.queue.after || !replication_globals.unsafe.queue.unique_id)) {
1062 replication_globals.unsafe.queue.after = 0;
1063 replication_globals.unsafe.queue.unique_id = 0;
1064 }
1065
1066 Word_t started_after = replication_globals.unsafe.queue.after;
1067
1068 size_t round = 0;
1069 while(!rq_to_return.found) {
1070 round++;
1071
1072 if(round > 2)
1073 break;
1074
1075 if(round == 2) {
1076 if(started_after == 0)
1077 break;
1078
1079 replication_globals.unsafe.queue.after = 0;
1080 replication_globals.unsafe.queue.unique_id = 0;
1081 }
1082
1083 bool find_same_after = true;
1084 while (!rq_to_return.found && (inner_judy_pptr = JudyLFirstThenNext(replication_globals.unsafe.queue.JudyL_array, &replication_globals.unsafe.queue.after, &find_same_after))) {
1085 Pvoid_t *our_item_pptr;
1086
1087 if(unlikely(round == 2 && replication_globals.unsafe.queue.after > started_after))
1088 break;
1089
1090 while (!rq_to_return.found && (our_item_pptr = JudyLNext(*inner_judy_pptr, &replication_globals.unsafe.queue.unique_id, PJE0))) {
1091 struct replication_sort_entry *rse = *our_item_pptr;
1092 struct replication_request *rq = rse->rq;
1093
1094 // copy the request to return it
1095 rq_to_return = *rq;
1096 rq_to_return.chart_id = string_dup(rq_to_return.chart_id);
1097
1098 // set the return result to found
1099 rq_to_return.found = true;
1100
1101 if (replication_sort_entry_unlink_and_free_unsafe(rse, &inner_judy_pptr, true))
1102 // we removed the item from the outer JudyL
1103 break;
1104 }
1105
1106 // prepare for the next iteration on the outer loop
1107 replication_globals.unsafe.queue.unique_id = 0;
1108 }
1109 }
1110
1111 replication_recursive_unlock();
1112 return rq_to_return;
1113 }
1114
1115 // ----------------------------------------------------------------------------
1116 // replication request management
1117
1118 static void replication_request_react_callback(const DICTIONARY_ITEM *item __maybe_unused, void *value __maybe_unused, void *sender_state __maybe_unused) {
1119 struct sender_state *s = sender_state; (void)s;
1120 struct replication_request *rq = value;
1121
1122 // IMPORTANT:
1123 // We use the react instead of the insert callback
1124 // because we want the item to be atomically visible
1125 // to our replication thread, immediately after.
1126
1127 // If we put this at the insert callback, the item is not guaranteed
1128 // to be atomically visible to others, so the replication thread
1129 // may see the replication sort entry, but fail to find the dictionary item
1130 // related to it.
1131
1132 replication_sort_entry_add(rq);
1133
1134 // this request is about a unique chart for this sender
1135 stream_sender_replicating_charts_plus_one(s);
1136 }
1137
1138 static bool replication_request_conflict_callback(const DICTIONARY_ITEM *item __maybe_unused, void *old_value, void *new_value, void *sender_state) {
1139 struct sender_state *s = sender_state; (void)s;
1140 struct replication_request *rq = old_value; (void)rq;
1141 struct replication_request *rq_new = new_value;
1142
1143 __atomic_add_fetch(&replication_globals.atomic.error_duplicate, 1, __ATOMIC_RELAXED);
1144
1145 replication_recursive_lock();
1146
1147 if(!rq->indexed_in_judy && rq->not_indexed_buffer_full && !rq->not_indexed_preprocessing) {
1148 // we can replace this command
1149 internal_error(
1150 true,
1151 "STREAM SND '%s' [to %s]: REPLAY: 'host:%s/chart:%s' replacing duplicate replication command received (existing from %llu to %llu [%s], new from %llu to %llu [%s])",
1152 rrdhost_hostname(s->host), s->remote_ip, rrdhost_hostname(s->host), dictionary_acquired_item_name(item),
1153 (unsigned long long)rq->after, (unsigned long long)rq->before, rq->start_streaming ? "true" : "false",
1154 (unsigned long long)rq_new->after, (unsigned long long)rq_new->before, rq_new->start_streaming ? "true" : "false");
1155
1156 rq->after = rq_new->after;
1157 rq->before = rq_new->before;
1158 rq->start_streaming = rq_new->start_streaming;
1159 }
1160 else if(!rq->indexed_in_judy && !rq->not_indexed_preprocessing) {
1161 replication_sort_entry_add(rq);
1162 internal_error(
1163 true,
1164 "STREAM SND '%s' [to %s]: REPLAY: 'host:%s/chart:%s' adding duplicate replication command received (existing from %llu to %llu [%s], new from %llu to %llu [%s])",
1165 rrdhost_hostname(s->host), s->remote_ip, rrdhost_hostname(s->host), dictionary_acquired_item_name(item),
1166 (unsigned long long)rq->after, (unsigned long long)rq->before, rq->start_streaming ? "true" : "false",
1167 (unsigned long long)rq_new->after, (unsigned long long)rq_new->before, rq_new->start_streaming ? "true" : "false");
1168 }
1169 else {
1170 internal_error(
1171 true,
1172 "STREAM SND '%s' [to %s]: REPLAY: 'host:%s/chart:%s' ignoring duplicate replication command received (existing from %llu to %llu [%s], new from %llu to %llu [%s])",
1173 rrdhost_hostname(s->host), s->remote_ip, rrdhost_hostname(s->host),
1174 dictionary_acquired_item_name(item),
1175 (unsigned long long) rq->after, (unsigned long long) rq->before, rq->start_streaming ? "true" : "false",
1176 (unsigned long long) rq_new->after, (unsigned long long) rq_new->before, rq_new->start_streaming ? "true" : "false");
1177 }
1178
1179 replication_recursive_unlock();
1180
1181 string_freez(rq_new->chart_id);
1182 return false;
1183 }
1184
1185 static void replication_request_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, void *value, void *sender_state __maybe_unused) {
1186 struct replication_request *rq = value;
1187
1188 // this request is about a unique chart for this sender
1189 stream_sender_replicating_charts_minus_one(rq->sender);
1190
1191 if(rq->indexed_in_judy)
1192 replication_sort_entry_del(rq, false);
1193
1194 else if(rq->not_indexed_buffer_full) {
1195 replication_recursive_lock();
1196 replication_globals.unsafe.pending_no_room--;
1197 replication_recursive_unlock();
1198 }
1199
1200 string_freez(rq->chart_id);
1201 }
1202
1203 static bool sender_is_still_connected_for_this_request(struct replication_request *rq) {
1204 return rq->sender_circular_buffer_last_flush_ut == stream_circular_buffer_last_flush_ut(rq->sender->scb);
1205 }
1206
1207 ALWAYS_INLINE_HOT
1208 static bool replication_execute_request(struct replication_request *rq, bool workers) {
1209 bool ret = false;
1210
1211 if(!rq->st) {
1212 if(likely(workers)) worker_is_busy(WORKER_JOB_FIND_CHART);
1213 rq->st = rrdset_find(rq->sender->host, string2str(rq->chart_id), true);
1214 if(!rq->st) {
1215 __atomic_add_fetch(&replication_globals.atomic.error_not_found, 1, __ATOMIC_RELAXED);
1216 nd_log(NDLS_DAEMON, NDLP_ERR,
1217 "STREAM SND REPLAY ERROR: 'host:%s/chart:%s' not found, sending empty response to unblock parent",
1218 rrdhost_hostname(rq->sender->host), string2str(rq->chart_id));
1219
1220 // CRITICAL: Parent is waiting for a response! We MUST send REPLAY_END even if chart not found
1221 // Otherwise parent will wait forever with chart stuck in replicating state.
1222 // Send empty response with start_streaming=true to finish replication for this non-existent chart.
1223 BUFFER *wb = sender_thread_buffer(rq->sender, REPLICATION_THREAD_BUFFER_INITIAL_SIZE);
1224
1225 bool with_slots = (rq->sender->capabilities & STREAM_CAP_SLOTS) ? true : false;
1226 NUMBER_ENCODING integer_encoding = (rq->sender->capabilities & STREAM_CAP_IEEE754) ?
1227 NUMBER_ENCODING_BASE64 : NUMBER_ENCODING_DECIMAL;
1228
1229 buffer_fast_strcat(wb, PLUGINSD_KEYWORD_REPLAY_BEGIN, sizeof(PLUGINSD_KEYWORD_REPLAY_BEGIN) - 1);
1230 if(with_slots) {
1231 buffer_fast_strcat(wb, " "PLUGINSD_KEYWORD_SLOT":", sizeof(PLUGINSD_KEYWORD_SLOT) - 1 + 2);
1232 buffer_print_uint64_encoded(wb, integer_encoding, 0); // slot 0 for unknown chart
1233 }
1234 buffer_fast_strcat(wb, " '", 2);
1235 buffer_fast_strcat(wb, string2str(rq->chart_id), string_strlen(rq->chart_id));
1236 buffer_fast_strcat(wb, "'\n", 2);
1237
1238 // Send REPLAY_END with empty data and start_streaming=true to unblock parent
1239 buffer_fast_strcat(wb, PLUGINSD_KEYWORD_REPLAY_END " ", sizeof(PLUGINSD_KEYWORD_REPLAY_END) - 1 + 1);
1240 buffer_print_int64_encoded(wb, integer_encoding, 0); // update_every
1241 buffer_fast_strcat(wb, " ", 1);
1242 buffer_print_uint64_encoded(wb, integer_encoding, 0); // db_first_entry
1243 buffer_fast_strcat(wb, " ", 1);
1244 buffer_print_uint64_encoded(wb, integer_encoding, 0); // db_last_entry
1245 buffer_fast_strcat(wb, " true ", 7); // start_streaming=true (force finish)
1246 buffer_print_uint64_encoded(wb, integer_encoding, 0); // after
1247 buffer_fast_strcat(wb, " ", 1);
1248 buffer_print_uint64_encoded(wb, integer_encoding, 0); // before
1249 buffer_fast_strcat(wb, " ", 1);
1250 buffer_print_uint64_encoded(wb, integer_encoding, now_realtime_sec()); // wall_clock_time
1251 buffer_fast_strcat(wb, "\n", 1);
1252
1253 sender_commit(rq->sender, wb, STREAM_TRAFFIC_TYPE_REPLICATION);
1254 __atomic_add_fetch(&rq->sender->host->stream.snd.status.replication.counter_out, 1, __ATOMIC_RELAXED);
1255 replication_replied_add();
1256
1257 ret = true; // Consider this a successful response
1258 goto cleanup;
1259 }
1260 }
1261
1262 if(!rq->q) {
1263 if(likely(workers)) worker_is_busy(WORKER_JOB_PREPARE_QUERY);
1264 rq->q = replication_response_prepare(
1265 rq->st,
1266 rq->start_streaming,
1267 rq->after,
1268 rq->before,
1269 rq->sender->capabilities, true);
1270 }
1271
1272 if(likely(workers)) worker_is_busy(WORKER_JOB_QUERYING);
1273
1274 // send the replication data
1275 size_t max_msg_size = (size_t)((unsigned long long)stream_circular_buffer_get_max_size(rq->sender->scb) * MAX_REPLICATION_MESSAGE_PERCENT_SENDER_BUFFER / 100ULL);
1276 rq->q->rq = rq;
1277 replication_response_execute_finalize_and_send(rq->q, max_msg_size, workers);
1278 rq->q = NULL;
1279
1280 __atomic_add_fetch(&replication_globals.atomic.executed, 1, __ATOMIC_RELAXED);
1281 ret = true;
1282
1283 cleanup:
1284 replication_response_cancel_and_finalize(rq->q);
1285 rq->q = NULL;
1286
1287 string_freez(rq->chart_id);
1288 worker_is_idle();
1289 return ret;
1290 }
1291
1292 // ----------------------------------------------------------------------------
1293 // public API
1294
1295 void replication_sender_request_add(struct sender_state *sender, const char *chart_id, time_t after, time_t before, bool start_streaming) {
1296 struct replication_request rq = {
1297 .sender = sender,
1298 .chart_id = string_strdupz(chart_id),
1299 .after = after,
1300 .before = before,
1301 .start_streaming = start_streaming,
1302 .sender_circular_buffer_last_flush_ut = stream_circular_buffer_last_flush_ut(sender->scb),
1303 .indexed_in_judy = false,
1304 .not_indexed_buffer_full = false,
1305 .not_indexed_preprocessing = false,
1306 };
1307
1308 if(!sender->replication.oldest_request_after_t || rq.after < sender->replication.oldest_request_after_t)
1309 sender->replication.oldest_request_after_t = rq.after;
1310
1311 dictionary_set(sender->replication.requests, chart_id, &rq, sizeof(struct replication_request));
1312 __atomic_add_fetch(&replication_globals.atomic.received, 1, __ATOMIC_RELAXED);
1313 }
1314
1315 void replication_sender_delete_pending_requests(struct sender_state *sender) {
1316 // allow the dictionary destructor to go faster on locks
1317 __atomic_add_fetch(&replication_globals.atomic.error_flushed, dictionary_entries(sender->replication.requests), __ATOMIC_RELAXED);
1318 dictionary_flush(sender->replication.requests);
1319 sender->replication.oldest_request_after_t = 0;
1320 }
1321
1322 void replication_sender_init(struct sender_state *sender) {
1323 sender->replication.requests = dictionary_create_advanced(DICT_OPTION_DONT_OVERWRITE_VALUE | DICT_OPTION_FIXED_SIZE,
1324 &dictionary_stats_category_replication, sizeof(struct replication_request));
1325
1326 dictionary_register_react_callback(sender->replication.requests, replication_request_react_callback, sender);
1327 dictionary_register_conflict_callback(sender->replication.requests, replication_request_conflict_callback, sender);
1328 dictionary_register_delete_callback(sender->replication.requests, replication_request_delete_callback, sender);
1329 }
1330
1331 void replication_sender_cleanup(struct sender_state *sender) {
1332 // allow the dictionary destructor to go faster on locks
1333 replication_recursive_lock();
1334 dictionary_destroy(sender->replication.requests);
1335 replication_recursive_unlock();
1336 }
1337
1338 static void replication_replied_add(void) {
1339 __atomic_add_fetch(&replication_globals.atomic.replied, 1, __ATOMIC_RELAXED);
1340 }
1341
1342 void replication_sender_recalculate_buffer_used_ratio_unsafe(struct sender_state *s) {
1343 size_t percentage = stream_sender_get_buffer_used_percent(s->scb);
1344
1345 if(unlikely(percentage > MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED && !stream_sender_replication_buffer_full_get(s))) {
1346 stream_sender_replication_buffer_full_set(s, true);
1347
1348 struct replication_request *rq;
1349 dfe_start_read(s->replication.requests, rq) {
1350 if(rq->indexed_in_judy)
1351 replication_sort_entry_del(rq, true);
1352 }
1353 dfe_done(rq);
1354
1355 replication_recursive_lock();
1356 replication_globals.unsafe.senders_full++;
1357 replication_recursive_unlock();
1358 }
1359 else if(unlikely(percentage < MIN_SENDER_BUFFER_PERCENTAGE_ALLOWED && stream_sender_replication_buffer_full_get(s))) {
1360 stream_sender_replication_buffer_full_set(s, false);
1361
1362 struct replication_request *rq;
1363 dfe_start_read(s->replication.requests, rq) {
1364 if(!rq->indexed_in_judy && (rq->not_indexed_buffer_full || rq->not_indexed_preprocessing))
1365 replication_sort_entry_add(rq);
1366 }
1367 dfe_done(rq);
1368
1369 replication_recursive_lock();
1370 replication_globals.unsafe.senders_full--;
1371 replication_globals.unsafe.sender_resets++;
1372 // replication_set_next_point_in_time(0, 0);
1373 replication_recursive_unlock();
1374 }
1375 }
1376
1377 // ----------------------------------------------------------------------------
1378 // replication thread
1379
1380 static size_t verify_host_charts_are_streaming_now(RRDHOST *host) {
1381 internal_error(
1382 host->sender &&
1383 !stream_sender_pending_replication_requests(host->sender) &&
1384 dictionary_entries(host->sender->replication.requests) != 0,
1385 "STREAM SND REPLAY SUMMARY: 'host:%s' reports %zu pending replication requests, "
1386 "but its chart replication index says there are %zu charts pending replication",
1387 rrdhost_hostname(host),
1388 stream_sender_pending_replication_requests(host->sender),
1389 dictionary_entries(host->sender->replication.requests)
1390 );
1391
1392 size_t ok = 0;
1393 size_t errors = 0;
1394 size_t ignored = 0;
1395
1396 RRDSET *st;
1397 rrdset_foreach_read(st, host) {
1398 RRDSET_FLAGS flags = rrdset_flag_check(
1399 st, RRDSET_FLAG_OBSOLETE | RRDSET_FLAG_UPSTREAM_IGNORE |
1400 RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS | RRDSET_FLAG_SENDER_REPLICATION_FINISHED);
1401
1402 if(flags & (RRDSET_FLAG_OBSOLETE | RRDSET_FLAG_UPSTREAM_IGNORE)) {
1403 ignored++;
1404 continue;
1405 }
1406
1407 bool is_error = false;
1408
1409 if(!flags) {
1410 internal_error(
1411 true,
1412 "STREAM SND REPLAY SUMMARY: 'host:%s/chart:%s' is neither IN PROGRESS nor FINISHED",
1413 rrdhost_hostname(host), rrdset_id(st)
1414 );
1415 is_error = true;
1416 }
1417
1418 if(!(flags & RRDSET_FLAG_SENDER_REPLICATION_FINISHED) || (flags & RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS)) {
1419 internal_error(
1420 true,
1421 "STREAM SND REPLAY SUMMARY: 'host:%s/chart:%s' is IN PROGRESS although replication is finished",
1422 rrdhost_hostname(host), rrdset_id(st)
1423 );
1424 is_error = true;
1425 }
1426
1427 if(is_error)
1428 errors++;
1429 else
1430 ok++;
1431 }
1432 rrdset_foreach_done(st);
1433
1434 internal_error(errors,
1435 "STREAM SND REPLAY SUMMARY: 'host:%s' finished replicating %zu charts, "
1436 "but %zu charts are still in progress although replication finished "
1437 "(%zu charts are not streamed - obsolete or excluded)",
1438 rrdhost_hostname(host), ok, errors, ignored);
1439
1440 return errors;
1441 }
1442
1443 static void verify_all_hosts_charts_are_streaming_now(void) {
1444 worker_is_busy(WORKER_JOB_CHECK_CONSISTENCY);
1445
1446 size_t charts_flagged_pending = 0, entries_in_dictionaries = 0;
1447 RRDHOST *host;
1448 dfe_start_read(rrdhost_root_index, host) {
1449 charts_flagged_pending += verify_host_charts_are_streaming_now(host);
1450
1451 if(host->sender)
1452 entries_in_dictionaries += dictionary_entries(host->sender->replication.requests);
1453 }
1454 dfe_done(host);
1455
1456 size_t flushed = __atomic_load_n(&replication_globals.atomic.error_flushed, __ATOMIC_RELAXED);
1457 size_t duplicate = __atomic_load_n(&replication_globals.atomic.error_duplicate, __ATOMIC_RELAXED);
1458 size_t not_found = __atomic_load_n(&replication_globals.atomic.error_not_found, __ATOMIC_RELAXED);
1459 size_t received = __atomic_load_n(&replication_globals.atomic.received, __ATOMIC_RELAXED);
1460 size_t executed = __atomic_load_n(&replication_globals.atomic.executed, __ATOMIC_RELAXED);
1461 size_t replied = __atomic_load_n(&replication_globals.atomic.replied, __ATOMIC_RELAXED);
1462
1463 CLEAN_BUFFER *wb = buffer_create(0, NULL);
1464
1465 if(entries_in_dictionaries) {
1466 if(buffer_strlen(wb)) buffer_strcat(wb, ", ");
1467 buffer_sprintf(wb, "%zu requests pending",
1468 entries_in_dictionaries);
1469 }
1470 if(charts_flagged_pending) {
1471 if(buffer_strlen(wb)) buffer_strcat(wb, ", ");
1472 buffer_sprintf(wb, "%zu instances waiting parent",
1473 charts_flagged_pending);
1474 }
1475 if(not_found - replication_globals.main_thread.last_error_not_found) {
1476 if(buffer_strlen(wb)) buffer_strcat(wb, ", ");
1477 buffer_sprintf(wb, "%zu ignored-not-found",
1478 not_found - replication_globals.main_thread.last_error_not_found);
1479 }
1480 if(duplicate - replication_globals.main_thread.last_error_duplicate) {
1481 if(buffer_strlen(wb)) buffer_strcat(wb, ", ");
1482 buffer_sprintf(wb, "%zu ignored-merged",
1483 duplicate - replication_globals.main_thread.last_error_duplicate);
1484 }
1485 if(flushed - replication_globals.main_thread.last_error_flushed) {
1486 if(buffer_strlen(wb)) buffer_strcat(wb, ", ");
1487 buffer_sprintf(wb, "%zu were flushed",
1488 flushed - replication_globals.main_thread.last_error_flushed);
1489 }
1490
1491 nd_log(NDLS_DAEMON, NDLP_NOTICE,
1492 "REPLICATION SEND SUMMARY: all senders finished replication. "
1493 "Received %zu, executed %zu and replied to %zu requests. %s",
1494 received - replication_globals.main_thread.last_received,
1495 executed - replication_globals.main_thread.last_executed,
1496 replied - replication_globals.main_thread.last_replied,
1497 buffer_tostring(wb));
1498
1499 replication_globals.main_thread.last_error_flushed = flushed;
1500 replication_globals.main_thread.last_error_duplicate = duplicate;
1501 replication_globals.main_thread.last_error_not_found = not_found;
1502 replication_globals.main_thread.last_received = received;
1503 replication_globals.main_thread.last_executed = executed;
1504 replication_globals.main_thread.last_replied = replied;
1505 }
1506
1507 static void replication_initialize_workers(bool master) {
1508 worker_register("REPLICATION");
1509 worker_register_job_name(WORKER_JOB_FIND_NEXT, "find next");
1510 worker_register_job_name(WORKER_JOB_QUERYING, "querying");
1511 worker_register_job_name(WORKER_JOB_DELETE_ENTRY, "dict delete");
1512 worker_register_job_name(WORKER_JOB_FIND_CHART, "find chart");
1513 worker_register_job_name(WORKER_JOB_PREPARE_QUERY, "prepare query");
1514 worker_register_job_name(WORKER_JOB_CHECK_CONSISTENCY, "check consistency");
1515 worker_register_job_name(WORKER_JOB_BUFFER_COMMIT, "commit");
1516 worker_register_job_name(WORKER_JOB_CLEANUP, "cleanup");
1517 worker_register_job_name(WORKER_JOB_WAIT, "wait");
1518
1519 if(master) {
1520 worker_register_job_name(WORKER_JOB_STATISTICS, "statistics");
1521 worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_PENDING_REQUESTS, "pending requests", "requests", WORKER_METRIC_ABSOLUTE);
1522 worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_SKIPPED_NO_ROOM, "no room requests", "requests", WORKER_METRIC_ABSOLUTE);
1523 worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_COMPLETION, "completion", "%", WORKER_METRIC_ABSOLUTE);
1524 worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_ADDED, "added requests", "requests/s", WORKER_METRIC_INCREMENTAL_TOTAL);
1525 worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_DONE, "finished requests", "requests/s", WORKER_METRIC_INCREMENTAL_TOTAL);
1526 worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_SENDER_RESETS, "sender resets", "resets/s", WORKER_METRIC_INCREMENTAL_TOTAL);
1527 worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_SENDER_FULL, "senders full", "senders", WORKER_METRIC_ABSOLUTE);
1528 }
1529 }
1530
1531 #define REQUEST_OK (0)
1532 #define REQUEST_QUEUE_EMPTY (-1)
1533 #define REQUEST_CHART_NOT_FOUND (-2)
1534
1535 static __thread struct replication_thread_pipeline {
1536 int max_requests_ahead;
1537 struct replication_request *rqs;
1538 int rqs_last_executed, rqs_last_prepared;
1539 size_t queue_rounds;
1540 } rtp = {
1541 .max_requests_ahead = 0,
1542 .rqs = NULL,
1543 .rqs_last_executed = 0,
1544 .rqs_last_prepared = 0,
1545 .queue_rounds = 0,
1546 };
1547
1548 static void replication_pipeline_cancel_and_cleanup(void) {
1549 if(!rtp.rqs)
1550 return;
1551
1552 struct replication_request *rq;
1553 size_t cancelled = 0;
1554
1555 do {
1556 if (++rtp.rqs_last_executed >= rtp.max_requests_ahead)
1557 rtp.rqs_last_executed = 0;
1558
1559 rq = &rtp.rqs[rtp.rqs_last_executed];
1560
1561 if (rq->q) {
1562 internal_fatal(rq->executed, "REPLAY FATAL: query has already been executed!");
1563 internal_fatal(!rq->found, "REPLAY FATAL: orphan q in rq");
1564
1565 replication_response_cancel_and_finalize(rq->q);
1566 rq->q = NULL;
1567 cancelled++;
1568 }
1569
1570 rq->executed = true;
1571 rq->found = false;
1572
1573 } while (rtp.rqs_last_executed != rtp.rqs_last_prepared);
1574
1575 internal_error(true, "REPLICATION: cancelled %zu inflight queries", cancelled);
1576
1577 freez(rtp.rqs);
1578 rtp.rqs = NULL;
1579 rtp.max_requests_ahead = 0;
1580 rtp.rqs_last_executed = 0;
1581 rtp.rqs_last_prepared = 0;
1582 rtp.queue_rounds = 0;
1583 }
1584
1585 static int replication_pipeline_execute_next(void) {
1586 struct replication_request *rq;
1587
1588 if(unlikely(!rtp.rqs)) {
1589 rtp.max_requests_ahead = stream_send.replication.prefetch;
1590 rtp.rqs = callocz(rtp.max_requests_ahead, sizeof(struct replication_request));
1591 __atomic_add_fetch(&replication_buffers_allocated, rtp.max_requests_ahead * sizeof(struct replication_request), __ATOMIC_RELAXED);
1592 }
1593
1594 // fill the queue
1595 do {
1596 if(++rtp.rqs_last_prepared >= rtp.max_requests_ahead) {
1597 rtp.rqs_last_prepared = 0;
1598 rtp.queue_rounds++;
1599 }
1600
1601 internal_fatal(rtp.rqs[rtp.rqs_last_prepared].q,
1602 "REPLAY FATAL: slot is used by query that has not been executed!");
1603
1604 worker_is_busy(WORKER_JOB_FIND_NEXT);
1605 rtp.rqs[rtp.rqs_last_prepared] = replication_request_get_first_available();
1606 rq = &rtp.rqs[rtp.rqs_last_prepared];
1607
1608 if(rq->found) {
1609 if(!rq->start_streaming) {
1610 if (!rq->st) {
1611 worker_is_busy(WORKER_JOB_FIND_CHART);
1612 rq->st = rrdset_find(rq->sender->host, string2str(rq->chart_id), true);
1613 }
1614
1615 if (rq->st && !rq->q) {
1616 worker_is_busy(WORKER_JOB_PREPARE_QUERY);
1617 rq->q = replication_response_prepare(
1618 rq->st,
1619 rq->start_streaming,
1620 rq->after,
1621 rq->before,
1622 rq->sender->capabilities,
1623 rtp.max_requests_ahead == 1);
1624 }
1625 }
1626
1627 rq->executed = false;
1628 }
1629
1630 } while(rq->found && rtp.rqs_last_prepared != rtp.rqs_last_executed);
1631
1632 // pick the first usable
1633 do {
1634 if (++rtp.rqs_last_executed >= rtp.max_requests_ahead)
1635 rtp.rqs_last_executed = 0;
1636
1637 rq = &rtp.rqs[rtp.rqs_last_executed];
1638
1639 if(rq->found) {
1640 internal_fatal(rq->executed, "REPLAY FATAL: query has already been executed!");
1641
1642 if (rq->sender_circular_buffer_last_flush_ut != stream_circular_buffer_last_flush_ut(rq->sender->scb)) {
1643 // the sender has reconnected since this request was queued,
1644 // we can safely throw it away, since the parent will resend it
1645 replication_response_cancel_and_finalize(rq->q);
1646 rq->executed = true;
1647 rq->found = false;
1648 rq->q = NULL;
1649 }
1650 else if (stream_sender_replication_buffer_full_get(rq->sender)) {
1651 // the sender buffer is full, so we can ignore this request,
1652 // it has already been marked as 'preprocessed' in the dictionary,
1653 // and the sender will put it back in when there is
1654 // enough room in the buffer for processing replication requests
1655 replication_response_cancel_and_finalize(rq->q);
1656 rq->executed = true;
1657 rq->found = false;
1658 rq->q = NULL;
1659 }
1660 else {
1661 // we can execute this,
1662 // delete it from the dictionary
1663 worker_is_busy(WORKER_JOB_DELETE_ENTRY);
1664 dictionary_del(rq->sender->replication.requests, string2str(rq->chart_id));
1665 }
1666 }
1667 else
1668 internal_fatal(rq->q, "REPLAY FATAL: slot status says slot is empty, but it has a pending query!");
1669
1670 } while(!rq->found && rtp.rqs_last_executed != rtp.rqs_last_prepared);
1671
1672 if(unlikely(!rq->found)) {
1673 worker_is_idle();
1674 return REQUEST_QUEUE_EMPTY;
1675 }
1676
1677 replication_set_latest_first_time(rq->after);
1678
1679 bool chart_found = replication_execute_request(rq, true);
1680 rq->executed = true;
1681 rq->found = false;
1682 rq->q = NULL;
1683
1684 if(unlikely(!chart_found)) {
1685 worker_is_idle();
1686 return REQUEST_CHART_NOT_FOUND;
1687 }
1688
1689 worker_is_idle();
1690 return REQUEST_OK;
1691 }
1692
1693 static void replication_worker_cleanup(void *pptr) {
1694 if(CLEANUP_FUNCTION_GET_PTR(pptr) != (void *)0x01) return;
1695 replication_pipeline_cancel_and_cleanup();
1696 worker_unregister();
1697 }
1698
1699 static void replication_worker_thread(void *ptr __maybe_unused) {
1700 CLEANUP_FUNCTION_REGISTER(replication_worker_cleanup) cleanup_ptr = (void *)0x1;
1701 replication_initialize_workers(false);
1702
1703 while (service_running(SERVICE_REPLICATION)) {
1704 if(!stream_control_replication_should_be_running()) {
1705 worker_is_idle();
1706 stream_control_throttle();
1707 continue;
1708 }
1709
1710 if (unlikely(replication_pipeline_execute_next() == REQUEST_QUEUE_EMPTY)) {
1711 sender_thread_buffer_free();
1712 worker_is_busy(WORKER_JOB_WAIT);
1713 worker_is_idle();
1714 sleep_usec(1 * USEC_PER_SEC);
1715 }
1716 }
1717 }
1718
1719 static void replication_main_cleanup(void *pptr) {
1720 struct netdata_static_thread *static_thread = CLEANUP_FUNCTION_GET_PTR(pptr);
1721 if(!static_thread) return;
1722
1723 static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;
1724
1725 replication_pipeline_cancel_and_cleanup();
1726
1727 int threads = (int)replication_globals.main_thread.threads;
1728 for(int i = 0; i < threads ;i++) {
1729 nd_thread_join(replication_globals.main_thread.threads_ptrs[i]);
1730 __atomic_sub_fetch(&replication_buffers_allocated, sizeof(ND_THREAD *), __ATOMIC_RELAXED);
1731 }
1732 freez(replication_globals.main_thread.threads_ptrs);
1733 replication_globals.main_thread.threads_ptrs = NULL;
1734 __atomic_sub_fetch(&replication_buffers_allocated, threads * sizeof(ND_THREAD *), __ATOMIC_RELAXED);
1735
1736 // we should not destroy aral on exit
1737 // the sender threads may still be working on flushing senders replication requests
1738 //aral_destroy(replication_globals.aral_rse);
1739 //replication_globals.aral_rse = NULL;
1740
1741 // custom code
1742 worker_unregister();
1743
1744 static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;
1745 }
1746
1747 struct aral_statistics aral_replication_stats = { 0 };
1748 void replication_initialize(void) {
1749 replication_globals.aral_rse = aral_create(
1750 "replication",
1751 sizeof(struct replication_sort_entry),
1752 0,
1753 128 * 1024, // limit it so that when replication finishes, we will not have a lot of memory lost
1754 &aral_replication_stats,
1755 NULL, NULL, false, false, false);
1756
1757 pulse_aral_register_statistics(&aral_replication_stats, "replication");
1758 }
1759
1760 void *replication_thread_main(void *ptr) {
1761 CLEANUP_FUNCTION_REGISTER(replication_main_cleanup) cleanup_ptr = ptr;
1762
1763 replication_initialize_workers(true);
1764
1765 size_t threads = stream_send.replication.threads;
1766 if(--threads) {
1767 replication_globals.main_thread.threads = threads;
1768 replication_globals.main_thread.threads_ptrs = mallocz(threads * sizeof(ND_THREAD *));
1769 __atomic_add_fetch(&replication_buffers_allocated, threads * sizeof(ND_THREAD *), __ATOMIC_RELAXED);
1770
1771 for(size_t i = 0; i < threads ;i++) {
1772 char tag[NETDATA_THREAD_TAG_MAX + 1];
1773 snprintfz(tag, NETDATA_THREAD_TAG_MAX, "REPLAY[%zu]", i + 2);
1774 __atomic_add_fetch(&replication_buffers_allocated, sizeof(ND_THREAD *), __ATOMIC_RELAXED);
1775 replication_globals.main_thread.threads_ptrs[i] = nd_thread_create(tag, NETDATA_THREAD_OPTION_DEFAULT,
1776 replication_worker_thread, NULL);
1777 }
1778 }
1779
1780 // start from 100% completed
1781 worker_set_metric(WORKER_JOB_CUSTOM_METRIC_COMPLETION, 100.0);
1782
1783 long run_verification_countdown = LONG_MAX; // LONG_MAX to prevent an initial verification when no replication ever took place
1784 bool slow = true; // control the time we sleep - it has to start with true!
1785 usec_t last_now_mono_ut = now_monotonic_usec();
1786 time_t replication_reset_next_point_in_time_countdown = SECONDS_TO_RESET_POINT_IN_TIME; // restart from the beginning every 10 seconds
1787
1788 size_t last_executed = 0;
1789 size_t last_sender_resets = 0;
1790
1791 while(service_running(SERVICE_REPLICATION)) {
1792
1793 if(!stream_control_replication_should_be_running()) {
1794 worker_is_idle();
1795 stream_control_throttle();
1796 continue;
1797 }
1798
1799 // statistics
1800 usec_t now_mono_ut = now_monotonic_usec();
1801 if(unlikely(now_mono_ut - last_now_mono_ut > nd_profile.update_every * USEC_PER_SEC)) {
1802 last_now_mono_ut = now_mono_ut;
1803
1804 worker_is_busy(WORKER_JOB_STATISTICS);
1805 replication_recursive_lock();
1806
1807 size_t current_executed = __atomic_load_n(&replication_globals.atomic.executed, __ATOMIC_RELAXED);
1808 if(last_executed != current_executed) {
1809 run_verification_countdown = ITERATIONS_IDLE_WITHOUT_PENDING_TO_RUN_SENDER_VERIFICATION;
1810 last_executed = current_executed;
1811 slow = false;
1812 }
1813
1814 if(replication_reset_next_point_in_time_countdown-- == 0) {
1815 // once per second, make it scan all the pending requests next time
1816 replication_set_next_point_in_time(0, 0);
1817 // replication_globals.protected.skipped_no_room_since_last_reset = 0;
1818 replication_reset_next_point_in_time_countdown = SECONDS_TO_RESET_POINT_IN_TIME;
1819 }
1820
1821 if(--run_verification_countdown == 0) {
1822 if (!replication_globals.unsafe.pending && !replication_globals.unsafe.pending_no_room) {
1823 // reset the statistics about completion percentage
1824 replication_globals.unsafe.first_time_t = 0;
1825 replication_set_latest_first_time(0);
1826
1827 verify_all_hosts_charts_are_streaming_now();
1828
1829 run_verification_countdown = LONG_MAX;
1830 slow = true;
1831 }
1832 else
1833 run_verification_countdown = ITERATIONS_IDLE_WITHOUT_PENDING_TO_RUN_SENDER_VERIFICATION;
1834 }
1835
1836 time_t current_s = replication_get_latest_first_time();
1837 if(current_s && replication_globals.unsafe.pending) {
1838 // completion percentage statistics
1839 time_t now_s = now_realtime_sec();
1840 if(current_s > now_s)
1841 current_s = now_s;
1842
1843 time_t started_s = replication_globals.unsafe.first_time_t;
1844 if(current_s < started_s)
1845 replication_globals.unsafe.first_time_t = started_s = current_s;
1846
1847 time_t total = now_s - started_s;
1848 time_t done = current_s - started_s;
1849
1850 worker_set_metric(WORKER_JOB_CUSTOM_METRIC_COMPLETION, total == 0 ? 0.0 :
1851 (NETDATA_DOUBLE) done * 100.0 / (NETDATA_DOUBLE) total);
1852 }
1853 else
1854 worker_set_metric(WORKER_JOB_CUSTOM_METRIC_COMPLETION, 100.0);
1855
1856 worker_set_metric(WORKER_JOB_CUSTOM_METRIC_PENDING_REQUESTS, (NETDATA_DOUBLE)replication_globals.unsafe.pending);
1857 worker_set_metric(WORKER_JOB_CUSTOM_METRIC_ADDED, (NETDATA_DOUBLE)replication_globals.unsafe.added);
1858 worker_set_metric(WORKER_JOB_CUSTOM_METRIC_DONE, (NETDATA_DOUBLE)__atomic_load_n(&replication_globals.atomic.executed, __ATOMIC_RELAXED));
1859 worker_set_metric(WORKER_JOB_CUSTOM_METRIC_SKIPPED_NO_ROOM, (NETDATA_DOUBLE)replication_globals.unsafe.pending_no_room);
1860 worker_set_metric(WORKER_JOB_CUSTOM_METRIC_SENDER_RESETS, (NETDATA_DOUBLE)replication_globals.unsafe.sender_resets);
1861 worker_set_metric(WORKER_JOB_CUSTOM_METRIC_SENDER_FULL, (NETDATA_DOUBLE)replication_globals.unsafe.senders_full);
1862
1863 replication_recursive_unlock();
1864 worker_is_idle();
1865 }
1866
1867 if(unlikely(replication_pipeline_execute_next() == REQUEST_QUEUE_EMPTY)) {
1868 worker_is_busy(WORKER_JOB_WAIT);
1869 replication_recursive_lock();
1870
1871 // the timeout also defines now frequently we will traverse all the pending requests
1872 // when the outbound buffers of all senders is full
1873 usec_t timeout;
1874 if(slow) {
1875 // no work to be done, wait for a request to come in
1876 timeout = 1000 * USEC_PER_MS;
1877 sender_thread_buffer_free();
1878 }
1879
1880 else if(replication_globals.unsafe.pending > 0) {
1881 if(replication_globals.unsafe.sender_resets == last_sender_resets)
1882 timeout = 1000 * USEC_PER_MS;
1883
1884 else {
1885 // there are pending requests waiting to be executed,
1886 // but none could be executed at this time.
1887 // try again after this time.
1888 timeout = 100 * USEC_PER_MS;
1889 }
1890
1891 last_sender_resets = replication_globals.unsafe.sender_resets;
1892 }
1893 else {
1894 // no requests pending, but there were requests recently (run_verification_countdown)
1895 // so, try in a short time.
1896 // if this is big, one chart replicating will be slow to finish (ping - pong just one chart)
1897 timeout = 10 * USEC_PER_MS;
1898 last_sender_resets = replication_globals.unsafe.sender_resets;
1899 }
1900
1901 replication_recursive_unlock();
1902
1903 worker_is_idle();
1904 sleep_usec(timeout);
1905
1906 // make it scan all the pending requests next time
1907 replication_set_next_point_in_time(0, 0);
1908 replication_reset_next_point_in_time_countdown = SECONDS_TO_RESET_POINT_IN_TIME;
1909
1910 continue;
1911 }
1912 }
1913
1914 return NULL;
1915 }
1916
1917 int replication_threads_default(void) {
1918 int threads = netdata_conf_is_parent() ? (int)MAX(netdata_conf_cpus() / 3, 4) : 1;
1919 threads = FIT_IN_RANGE(threads, 1, MAX_REPLICATION_THREADS);
1920 return threads;
1921 }
1922
1923 int replication_prefetch_default(void) {
1924 // Our goal is to feed the pipeline with enough requests,
1925 // since this will allow dbengine to merge the requests that load the same extents,
1926 // providing the best performance and minimizing disk I/O.
1927 int target = MAX(libuv_worker_threads / 2, (int)stream_send.replication.threads * 10);
1928
1929 int prefetch = (int)HOWMANY(target, stream_send.replication.threads);
1930 prefetch = FIT_IN_RANGE(prefetch, 1, MAX_REPLICATION_PREFETCH);
1931 return prefetch;
1932 }