// SPDX-License-Identifier: GPL-3.0-or-later #include "replication.h" #include "Judy.h" #define STREAMING_START_MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED 50ULL #define MAX_REPLICATION_MESSAGE_PERCENT_SENDER_BUFFER 25ULL #define MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED 50ULL #define MIN_SENDER_BUFFER_PERCENTAGE_ALLOWED 10ULL #define WORKER_JOB_FIND_NEXT 1 #define WORKER_JOB_QUERYING 2 #define WORKER_JOB_DELETE_ENTRY 3 #define WORKER_JOB_FIND_CHART 4 #define WORKER_JOB_PREPARE_QUERY 5 #define WORKER_JOB_CHECK_CONSISTENCY 6 #define WORKER_JOB_BUFFER_COMMIT 7 #define WORKER_JOB_CLEANUP 8 #define WORKER_JOB_WAIT 9 // master thread worker jobs #define WORKER_JOB_STATISTICS 10 #define WORKER_JOB_CUSTOM_METRIC_PENDING_REQUESTS 11 #define WORKER_JOB_CUSTOM_METRIC_SKIPPED_NO_ROOM 12 #define WORKER_JOB_CUSTOM_METRIC_COMPLETION 13 #define WORKER_JOB_CUSTOM_METRIC_ADDED 14 #define WORKER_JOB_CUSTOM_METRIC_DONE 15 #define WORKER_JOB_CUSTOM_METRIC_SENDER_RESETS 16 #define WORKER_JOB_CUSTOM_METRIC_SENDER_FULL 17 #define ITERATIONS_IDLE_WITHOUT_PENDING_TO_RUN_SENDER_VERIFICATION 30 #define SECONDS_TO_RESET_POINT_IN_TIME 10 static struct replication_query_statistics replication_queries = { .spinlock = NETDATA_SPINLOCK_INITIALIZER, .queries_started = 0, .queries_finished = 0, .points_read = 0, .points_generated = 0, }; struct replication_query_statistics replication_get_query_statistics(void) { spinlock_lock(&replication_queries.spinlock); struct replication_query_statistics ret = replication_queries; spinlock_unlock(&replication_queries.spinlock); return ret; } size_t replication_buffers_allocated = 0; size_t replication_allocated_buffers(void) { return __atomic_load_n(&replication_buffers_allocated, __ATOMIC_RELAXED); } // ---------------------------------------------------------------------------- // sending replication replies struct replication_dimension { STORAGE_POINT sp; struct storage_engine_query_handle handle; bool enabled; bool skip; DICTIONARY *dict; const DICTIONARY_ITEM *rda; RRDDIM *rd; }; struct replication_query { RRDSET *st; struct { time_t first_entry_t; time_t last_entry_t; } db; struct { // what the parent requested time_t after; time_t before; bool enable_streaming; } request; struct { // what the child will do time_t after; time_t before; bool enable_streaming; bool locked_data_collection; bool execute; bool interrupted; STREAM_CAPABILITIES capabilities; } query; time_t wall_clock_time; size_t points_read; size_t points_generated; STORAGE_ENGINE_BACKEND backend; struct replication_request *rq; size_t dimensions; struct replication_dimension data[]; }; static struct replication_query *replication_query_prepare( RRDSET *st, time_t db_first_entry, time_t db_last_entry, time_t requested_after, time_t requested_before, bool requested_enable_streaming, time_t query_after, time_t query_before, bool query_enable_streaming, time_t wall_clock_time, STREAM_CAPABILITIES capabilities ) { size_t dimensions = rrdset_number_of_dimensions(st); struct replication_query *q = callocz(1, sizeof(struct replication_query) + dimensions * sizeof(struct replication_dimension)); __atomic_add_fetch(&replication_buffers_allocated, sizeof(struct replication_query) + dimensions * sizeof(struct replication_dimension), __ATOMIC_RELAXED); q->dimensions = dimensions; q->st = st; q->db.first_entry_t = db_first_entry; q->db.last_entry_t = db_last_entry; q->request.after = requested_after, q->request.before = requested_before, q->request.enable_streaming = requested_enable_streaming, q->query.after = query_after; q->query.before = query_before; q->query.enable_streaming = query_enable_streaming; q->query.capabilities = capabilities; q->wall_clock_time = wall_clock_time; if (!q->dimensions || !q->query.after || !q->query.before) { q->query.execute = false; q->dimensions = 0; return q; } if(q->query.enable_streaming) { spinlock_lock(&st->data_collection_lock); q->query.locked_data_collection = true; if (st->last_updated.tv_sec > q->query.before) { #ifdef NETDATA_LOG_REPLICATION_REQUESTS internal_error(true, "STREAM_SENDER REPLAY: 'host:%s/chart:%s' " "has start_streaming = true, " "adjusting replication before timestamp from %llu to %llu", rrdhost_hostname(st->rrdhost), rrdset_id(st), (unsigned long long) q->query.before, (unsigned long long) st->last_updated.tv_sec ); #endif q->query.before = MIN(st->last_updated.tv_sec, wall_clock_time); } } q->backend = st->rrdhost->db[0].eng->backend; // prepare our array of dimensions size_t count = 0; RRDDIM *rd; rrddim_foreach_read(rd, st) { if (unlikely(!rd || !rd_dfe.item || !rrddim_check_upstream_exposed(rd))) continue; if (unlikely(rd_dfe.counter >= q->dimensions)) { internal_error(true, "STREAM_SENDER REPLAY ERROR: 'host:%s/chart:%s' has more dimensions than the replicated ones", rrdhost_hostname(st->rrdhost), rrdset_id(st)); break; } struct replication_dimension *d = &q->data[rd_dfe.counter]; d->dict = rd_dfe.dict; d->rda = dictionary_acquired_item_dup(rd_dfe.dict, rd_dfe.item); d->rd = rd; storage_engine_query_init(q->backend, rd->tiers[0].db_metric_handle, &d->handle, q->query.after, q->query.before, q->query.locked_data_collection ? STORAGE_PRIORITY_HIGH : STORAGE_PRIORITY_LOW); d->enabled = true; d->skip = false; count++; } rrddim_foreach_done(rd); if(!count) { // no data for this chart q->query.execute = false; if(q->query.locked_data_collection) { spinlock_unlock(&st->data_collection_lock); q->query.locked_data_collection = false; } } else { // we have data for this chart q->query.execute = true; } return q; } static void replication_send_chart_collection_state(BUFFER *wb, RRDSET *st, STREAM_CAPABILITIES capabilities) { bool with_slots = (capabilities & STREAM_CAP_SLOTS) ? true : false; NUMBER_ENCODING integer_encoding = (capabilities & STREAM_CAP_IEEE754) ? NUMBER_ENCODING_BASE64 : NUMBER_ENCODING_DECIMAL; RRDDIM *rd; rrddim_foreach_read(rd, st){ if (!rrddim_check_upstream_exposed(rd)) continue; buffer_fast_strcat(wb, PLUGINSD_KEYWORD_REPLAY_RRDDIM_STATE, sizeof(PLUGINSD_KEYWORD_REPLAY_RRDDIM_STATE) - 1); if(with_slots) { buffer_fast_strcat(wb, " "PLUGINSD_KEYWORD_SLOT":", sizeof(PLUGINSD_KEYWORD_SLOT) - 1 + 2); buffer_print_uint64_encoded(wb, integer_encoding, rd->rrdpush.sender.dim_slot); } buffer_fast_strcat(wb, " '", 2); buffer_fast_strcat(wb, rrddim_id(rd), string_strlen(rd->id)); buffer_fast_strcat(wb, "' ", 2); buffer_print_uint64_encoded(wb, integer_encoding, (usec_t) rd->collector.last_collected_time.tv_sec * USEC_PER_SEC + (usec_t) rd->collector.last_collected_time.tv_usec); buffer_fast_strcat(wb, " ", 1); buffer_print_int64_encoded(wb, integer_encoding, rd->collector.last_collected_value); buffer_fast_strcat(wb, " ", 1); buffer_print_netdata_double_encoded(wb, integer_encoding, rd->collector.last_calculated_value); buffer_fast_strcat(wb, " ", 1); buffer_print_netdata_double_encoded(wb, integer_encoding, rd->collector.last_stored_value); buffer_fast_strcat(wb, "\n", 1); } rrddim_foreach_done(rd); buffer_fast_strcat(wb, PLUGINSD_KEYWORD_REPLAY_RRDSET_STATE " ", sizeof(PLUGINSD_KEYWORD_REPLAY_RRDSET_STATE) - 1 + 1); 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); buffer_fast_strcat(wb, " ", 1); buffer_print_uint64_encoded(wb, integer_encoding, (usec_t) st->last_updated.tv_sec * USEC_PER_SEC + (usec_t) st->last_updated.tv_usec); buffer_fast_strcat(wb, "\n", 1); } static void replication_query_finalize(BUFFER *wb, struct replication_query *q, bool executed) { size_t dimensions = q->dimensions; if(wb && q->query.enable_streaming) replication_send_chart_collection_state(wb, q->st, q->query.capabilities); if(q->query.locked_data_collection) { spinlock_unlock(&q->st->data_collection_lock); q->query.locked_data_collection = false; } // release all the dictionary items acquired // finalize the queries size_t queries = 0; for (size_t i = 0; i < dimensions; i++) { struct replication_dimension *d = &q->data[i]; if (unlikely(!d->enabled)) continue; storage_engine_query_finalize(&d->handle); dictionary_acquired_item_release(d->dict, d->rda); // update global statistics queries++; } if(executed) { spinlock_lock(&replication_queries.spinlock); replication_queries.queries_started += queries; replication_queries.queries_finished += queries; replication_queries.points_read += q->points_read; replication_queries.points_generated += q->points_generated; if(q->st && q->st->rrdhost->sender) { struct sender_state *s = q->st->rrdhost->sender; s->replication.latest_completed_before_t = q->query.before; } spinlock_unlock(&replication_queries.spinlock); } __atomic_sub_fetch(&replication_buffers_allocated, sizeof(struct replication_query) + dimensions * sizeof(struct replication_dimension), __ATOMIC_RELAXED); freez(q); } static void replication_query_align_to_optimal_before(struct replication_query *q) { if(!q->query.execute || q->query.enable_streaming) return; size_t dimensions = q->dimensions; time_t expanded_before = 0; for (size_t i = 0; i < dimensions; i++) { struct replication_dimension *d = &q->data[i]; if(unlikely(!d->enabled)) continue; time_t new_before = storage_engine_align_to_optimal_before(&d->handle); if (!expanded_before || new_before < expanded_before) expanded_before = new_before; } if(expanded_before > q->query.before && // it is later than the original (expanded_before - q->query.before) / q->st->update_every < 1024 && // it is reasonable (up to a page) expanded_before < q->st->last_updated.tv_sec && // it is not the chart's last updated time expanded_before < q->wall_clock_time) // it is not later than the wall clock time q->query.before = expanded_before; } static bool replication_query_execute(BUFFER *wb, struct replication_query *q, size_t max_msg_size) { replication_query_align_to_optimal_before(q); bool with_slots = (q->query.capabilities & STREAM_CAP_SLOTS) ? true : false; NUMBER_ENCODING integer_encoding = (q->query.capabilities & STREAM_CAP_IEEE754) ? NUMBER_ENCODING_BASE64 : NUMBER_ENCODING_DECIMAL; time_t after = q->query.after; time_t before = q->query.before; size_t dimensions = q->dimensions; time_t wall_clock_time = q->wall_clock_time; bool finished_with_gap = false; size_t points_read = 0, points_generated = 0; #ifdef NETDATA_LOG_REPLICATION_REQUESTS time_t actual_after = 0, actual_before = 0; #endif time_t now = after + 1; time_t last_end_time_in_buffer = 0; while(now <= before) { 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; for (size_t i = 0; i < dimensions ;i++) { struct replication_dimension *d = &q->data[i]; if(unlikely(!d->enabled || d->skip)) continue; // fetch the first valid point for the dimension int max_skip = 1000; while(d->sp.end_time_s < now && !storage_engine_query_is_finished(&d->handle) && max_skip-- >= 0) { d->sp = storage_engine_query_next_metric(&d->handle); points_read++; } if(max_skip <= 0) { d->skip = true; nd_log_limit_static_global_var(erl, 1, 0); nd_log_limit(&erl, NDLS_DAEMON, NDLP_ERR, "STREAM_SENDER REPLAY ERROR: 'host:%s/chart:%s/dim:%s': db does not advance the query " "beyond time %llu (tried 1000 times to get the next point and always got back a point in the past)", rrdhost_hostname(q->st->rrdhost), rrdset_id(q->st), rrddim_id(d->rd), (unsigned long long) now); continue; } if(unlikely(d->sp.end_time_s < now || d->sp.end_time_s < d->sp.start_time_s)) // this dimension does not provide any data continue; time_t update_every = d->sp.end_time_s - d->sp.start_time_s; if(unlikely(!update_every)) update_every = q->st->update_every; if(unlikely(!min_update_every)) min_update_every = update_every; if(unlikely(!min_start_time)) min_start_time = d->sp.start_time_s; if(unlikely(!min_end_time)) min_end_time = d->sp.end_time_s; min_update_every = MIN(min_update_every, update_every); max_update_every = MAX(max_update_every, update_every); min_start_time = MIN(min_start_time, d->sp.start_time_s); max_start_time = MAX(max_start_time, d->sp.start_time_s); min_end_time = MIN(min_end_time, d->sp.end_time_s); max_end_time = MAX(max_end_time, d->sp.end_time_s); } if (unlikely(min_update_every != max_update_every || min_start_time != max_start_time)) { time_t fix_min_start_time; if(last_end_time_in_buffer && last_end_time_in_buffer >= min_start_time && last_end_time_in_buffer <= max_start_time) { fix_min_start_time = last_end_time_in_buffer; } else fix_min_start_time = min_end_time - min_update_every; #ifdef NETDATA_INTERNAL_CHECKS nd_log_limit_static_global_var(erl, 1, 0); nd_log_limit(&erl, NDLS_DAEMON, NDLP_WARNING, "REPLAY WARNING: 'host:%s/chart:%s' " "misaligned dimensions, " "update every (min: %ld, max: %ld), " "start time (min: %ld, max: %ld), " "end time (min %ld, max %ld), " "now %ld, last end time sent %ld, " "min start time is fixed to %ld", rrdhost_hostname(q->st->rrdhost), rrdset_id(q->st), min_update_every, max_update_every, min_start_time, max_start_time, min_end_time, max_end_time, now, last_end_time_in_buffer, fix_min_start_time ); #endif min_start_time = fix_min_start_time; } if(likely(min_start_time <= now && min_end_time >= now)) { // we have a valid point if (unlikely(min_end_time == min_start_time)) min_start_time = min_end_time - q->st->update_every; #ifdef NETDATA_LOG_REPLICATION_REQUESTS if (unlikely(!actual_after)) actual_after = min_end_time; actual_before = min_end_time; #endif if(buffer_strlen(wb) > max_msg_size && last_end_time_in_buffer) { q->query.before = last_end_time_in_buffer; q->query.enable_streaming = false; internal_error(true, "REPLICATION: current buffer size %zu is more than the " "max message size %zu for chart '%s' of host '%s'. " "Interrupting replication request (%ld to %ld, %s) at %ld to %ld, %s.", buffer_strlen(wb), max_msg_size, rrdset_id(q->st), rrdhost_hostname(q->st->rrdhost), q->request.after, q->request.before, q->request.enable_streaming?"true":"false", q->query.after, q->query.before, q->query.enable_streaming?"true":"false"); q->query.interrupted = true; break; } last_end_time_in_buffer = min_end_time; buffer_fast_strcat(wb, PLUGINSD_KEYWORD_REPLAY_BEGIN, sizeof(PLUGINSD_KEYWORD_REPLAY_BEGIN) - 1); if(with_slots) { buffer_fast_strcat(wb, " "PLUGINSD_KEYWORD_SLOT":", sizeof(PLUGINSD_KEYWORD_SLOT) - 1 + 2); buffer_print_uint64_encoded(wb, integer_encoding, q->st->rrdpush.sender.chart_slot); } buffer_fast_strcat(wb, " '' ", 4); buffer_print_uint64_encoded(wb, integer_encoding, min_start_time); buffer_fast_strcat(wb, " ", 1); buffer_print_uint64_encoded(wb, integer_encoding, min_end_time); buffer_fast_strcat(wb, " ", 1); buffer_print_uint64_encoded(wb, integer_encoding, wall_clock_time); buffer_fast_strcat(wb, "\n", 1); // output the replay values for this time for (size_t i = 0; i < dimensions; i++) { struct replication_dimension *d = &q->data[i]; if (unlikely(!d->enabled)) continue; if (likely( d->sp.start_time_s <= min_end_time && d->sp.end_time_s >= min_end_time && !storage_point_is_unset(d->sp) && !storage_point_is_gap(d->sp))) { buffer_fast_strcat(wb, PLUGINSD_KEYWORD_REPLAY_SET, sizeof(PLUGINSD_KEYWORD_REPLAY_SET) - 1); if(with_slots) { buffer_fast_strcat(wb, " "PLUGINSD_KEYWORD_SLOT":", sizeof(PLUGINSD_KEYWORD_SLOT) - 1 + 2); buffer_print_uint64_encoded(wb, integer_encoding, d->rd->rrdpush.sender.dim_slot); } buffer_fast_strcat(wb, " \"", 2); buffer_fast_strcat(wb, rrddim_id(d->rd), string_strlen(d->rd->id)); buffer_fast_strcat(wb, "\" ", 2); buffer_print_netdata_double_encoded(wb, integer_encoding, d->sp.sum); buffer_fast_strcat(wb, " ", 1); buffer_print_sn_flags(wb, d->sp.flags, q->query.capabilities & STREAM_CAP_INTERPOLATED); buffer_fast_strcat(wb, "\n", 1); points_generated++; } } now = min_end_time + 1; } else if(unlikely(min_end_time < now)) // the query does not progress break; else { // we have gap - all points are in the future now = min_start_time; if(min_start_time > before && !points_generated) { before = q->query.before = min_start_time - 1; finished_with_gap = true; break; } } } #ifdef NETDATA_LOG_REPLICATION_REQUESTS if(actual_after) { char actual_after_buf[LOG_DATE_LENGTH + 1], actual_before_buf[LOG_DATE_LENGTH + 1]; log_date(actual_after_buf, LOG_DATE_LENGTH, actual_after); log_date(actual_before_buf, LOG_DATE_LENGTH, actual_before); internal_error(true, "STREAM_SENDER REPLAY: 'host:%s/chart:%s': sending data %llu [%s] to %llu [%s] (requested %llu [delta %lld] to %llu [delta %lld])", rrdhost_hostname(q->st->rrdhost), rrdset_id(q->st), (unsigned long long)actual_after, actual_after_buf, (unsigned long long)actual_before, actual_before_buf, (unsigned long long)after, (long long)(actual_after - after), (unsigned long long)before, (long long)(actual_before - before)); } else internal_error(true, "STREAM_SENDER REPLAY: 'host:%s/chart:%s': nothing to send (requested %llu to %llu)", rrdhost_hostname(q->st->rrdhost), rrdset_id(q->st), (unsigned long long)after, (unsigned long long)before); #endif // NETDATA_LOG_REPLICATION_REQUESTS q->points_read += points_read; q->points_generated += points_generated; if(last_end_time_in_buffer < before - q->st->update_every) finished_with_gap = true; return finished_with_gap; } static struct replication_query *replication_response_prepare( RRDSET *st, bool requested_enable_streaming, time_t requested_after, time_t requested_before, STREAM_CAPABILITIES capabilities ) { time_t wall_clock_time = now_realtime_sec(); if(requested_after > requested_before) { // flip them time_t t = requested_before; requested_before = requested_after; requested_after = t; } if(requested_after > wall_clock_time) { requested_after = 0; requested_before = 0; requested_enable_streaming = true; } if(requested_before > wall_clock_time) { requested_before = wall_clock_time; requested_enable_streaming = true; } time_t query_after = requested_after; time_t query_before = requested_before; bool query_enable_streaming = requested_enable_streaming; time_t db_first_entry = 0, db_last_entry = 0; rrdset_get_retention_of_tier_for_collected_chart( st, &db_first_entry, &db_last_entry, wall_clock_time, 0); if(requested_after == 0 && requested_before == 0 && requested_enable_streaming == true) { // no data requested - just enable streaming ; } else { if (query_after < db_first_entry) query_after = db_first_entry; if (query_before > db_last_entry) query_before = db_last_entry; // if the parent asked us to start streaming, then fill the rest with the data that we have if (requested_enable_streaming) query_before = db_last_entry; if (query_after > query_before) { time_t tmp = query_before; query_before = query_after; query_after = tmp; } query_enable_streaming = (requested_enable_streaming || query_before == db_last_entry || !requested_after || !requested_before) ? true : false; } return replication_query_prepare( st, db_first_entry, db_last_entry, requested_after, requested_before, requested_enable_streaming, query_after, query_before, query_enable_streaming, wall_clock_time, capabilities); } void replication_response_cancel_and_finalize(struct replication_query *q) { replication_query_finalize(NULL, q, false); } static bool sender_is_still_connected_for_this_request(struct replication_request *rq); bool replication_response_execute_and_finalize(struct replication_query *q, size_t max_msg_size) { bool with_slots = (q->query.capabilities & STREAM_CAP_SLOTS) ? true : false; NUMBER_ENCODING integer_encoding = (q->query.capabilities & STREAM_CAP_IEEE754) ? NUMBER_ENCODING_BASE64 : NUMBER_ENCODING_DECIMAL; struct replication_request *rq = q->rq; RRDSET *st = q->st; RRDHOST *host = st->rrdhost; // we might want to optimize this by filling a temporary buffer // and copying the result to the host's buffer in order to avoid // holding the host's buffer lock for too long BUFFER *wb = sender_start(host->sender); buffer_fast_strcat(wb, PLUGINSD_KEYWORD_REPLAY_BEGIN, sizeof(PLUGINSD_KEYWORD_REPLAY_BEGIN) - 1); if(with_slots) { buffer_fast_strcat(wb, " "PLUGINSD_KEYWORD_SLOT":", sizeof(PLUGINSD_KEYWORD_SLOT) - 1 + 2); buffer_print_uint64_encoded(wb, integer_encoding, q->st->rrdpush.sender.chart_slot); } buffer_fast_strcat(wb, " '", 2); buffer_fast_strcat(wb, rrdset_id(st), string_strlen(st->id)); buffer_fast_strcat(wb, "'\n", 2); bool locked_data_collection = q->query.locked_data_collection; q->query.locked_data_collection = false; bool finished_with_gap = false; if(q->query.execute) finished_with_gap = replication_query_execute(wb, q, max_msg_size); time_t after = q->request.after; time_t before = q->query.before; bool enable_streaming = q->query.enable_streaming; replication_query_finalize(wb, q, q->query.execute); q = NULL; // IMPORTANT: q is invalid now // get a fresh retention to send to the parent time_t wall_clock_time = now_realtime_sec(); time_t db_first_entry, db_last_entry; rrdset_get_retention_of_tier_for_collected_chart(st, &db_first_entry, &db_last_entry, wall_clock_time, 0); // end with first/last entries we have, and the first start time and // last end time of the data we sent buffer_fast_strcat(wb, PLUGINSD_KEYWORD_REPLAY_END " ", sizeof(PLUGINSD_KEYWORD_REPLAY_END) - 1 + 1); buffer_print_int64_encoded(wb, integer_encoding, st->update_every); buffer_fast_strcat(wb, " ", 1); buffer_print_uint64_encoded(wb, integer_encoding, db_first_entry); buffer_fast_strcat(wb, " ", 1); buffer_print_uint64_encoded(wb, integer_encoding, db_last_entry); buffer_fast_strcat(wb, enable_streaming ? " true " : " false ", 7); buffer_print_uint64_encoded(wb, integer_encoding, after); buffer_fast_strcat(wb, " ", 1); buffer_print_uint64_encoded(wb, integer_encoding, before); buffer_fast_strcat(wb, " ", 1); buffer_print_uint64_encoded(wb, integer_encoding, wall_clock_time); buffer_fast_strcat(wb, "\n", 1); worker_is_busy(WORKER_JOB_BUFFER_COMMIT); sender_commit(host->sender, wb, STREAM_TRAFFIC_TYPE_REPLICATION); worker_is_busy(WORKER_JOB_CLEANUP); if(enable_streaming) { if(sender_is_still_connected_for_this_request(rq)) { // enable normal streaming if we have to // but only if the sender buffer has not been flushed since we started if(rrdset_flag_check(st, RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS)) { rrdset_flag_clear(st, RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS); rrdset_flag_set(st, RRDSET_FLAG_SENDER_REPLICATION_FINISHED); rrdhost_sender_replicating_charts_minus_one(st->rrdhost); if(!finished_with_gap) st->rrdpush.sender.resync_time_s = 0; #ifdef NETDATA_LOG_REPLICATION_REQUESTS internal_error(true, "STREAM_SENDER REPLAY: 'host:%s/chart:%s' streaming starts", rrdhost_hostname(st->rrdhost), rrdset_id(st)); #endif } else internal_error(true, "REPLAY ERROR: 'host:%s/chart:%s' received start streaming command, but the chart is not in progress replicating", rrdhost_hostname(st->rrdhost), rrdset_id(st)); } } if(locked_data_collection) spinlock_unlock(&st->data_collection_lock); return enable_streaming; } // ---------------------------------------------------------------------------- // sending replication requests struct replication_request_details { struct { send_command callback; void *data; } caller; RRDHOST *host; RRDSET *st; struct { time_t first_entry_t; // the first entry time the child has time_t last_entry_t; // the last entry time the child has time_t wall_clock_time; // the current time of the child bool fixed_last_entry; // when set we set the last entry to wall clock time } child_db; struct { time_t first_entry_t; // the first entry time we have time_t last_entry_t; // the last entry time we have time_t wall_clock_time; // the current local world clock time } local_db; struct { time_t from; // the starting time of the entire gap we have time_t to; // the ending time of the entire gap we have } gap; struct { time_t after; // the start time we requested previously from this child time_t before; // the end time we requested previously from this child } last_request; struct { time_t after; // the start time of this replication request - the child will add 1 second time_t before; // the end time of this replication request bool start_streaming; // true when we want the child to send anything remaining and start streaming - the child will overwrite 'before' } wanted; }; static void replicate_log_request(struct replication_request_details *r, const char *msg) { #ifdef NETDATA_INTERNAL_CHECKS internal_error(true, #else nd_log_limit_static_global_var(erl, 1, 0); nd_log_limit(&erl, NDLS_DAEMON, NDLP_ERR, #endif "REPLAY ERROR: 'host:%s/chart:%s' child sent: " "db from %ld to %ld%s, wall clock time %ld, " "last request from %ld to %ld, " "issue: %s - " "sending replication request from %ld to %ld, start streaming %s", rrdhost_hostname(r->st->rrdhost), rrdset_id(r->st), r->child_db.first_entry_t, r->child_db.last_entry_t, r->child_db.fixed_last_entry ? " (fixed)" : "", r->child_db.wall_clock_time, r->last_request.after, r->last_request.before, msg, r->wanted.after, r->wanted.before, r->wanted.start_streaming ? "true" : "false"); } static bool send_replay_chart_cmd(struct replication_request_details *r, const char *msg, bool log) { RRDSET *st = r->st; if(log) replicate_log_request(r, msg); if(st->rrdhost->receiver && (!st->rrdhost->receiver->replication_first_time_t || r->wanted.after < st->rrdhost->receiver->replication_first_time_t)) st->rrdhost->receiver->replication_first_time_t = r->wanted.after; #ifdef NETDATA_LOG_REPLICATION_REQUESTS st->replay.log_next_data_collection = true; char wanted_after_buf[LOG_DATE_LENGTH + 1] = "", wanted_before_buf[LOG_DATE_LENGTH + 1] = ""; if(r->wanted.after) log_date(wanted_after_buf, LOG_DATE_LENGTH, r->wanted.after); if(r->wanted.before) log_date(wanted_before_buf, LOG_DATE_LENGTH, r->wanted.before); internal_error(true, "REPLAY: 'host:%s/chart:%s' sending replication request %ld [%s] to %ld [%s], start streaming '%s': %s: " "last[%ld - %ld] child[%ld - %ld, now %ld %s] local[%ld - %ld, now %ld] gap[%ld - %ld %s] %s" , rrdhost_hostname(r->host), rrdset_id(r->st) , r->wanted.after, wanted_after_buf , r->wanted.before, wanted_before_buf , r->wanted.start_streaming ? "YES" : "NO" , msg , r->last_request.after, r->last_request.before , r->child_db.first_entry_t, r->child_db.last_entry_t , r->child_db.wall_clock_time, (r->child_db.wall_clock_time == r->local_db.wall_clock_time) ? "SAME" : (r->child_db.wall_clock_time < r->local_db.wall_clock_time) ? "BEHIND" : "AHEAD" , r->local_db.first_entry_t, r->local_db.last_entry_t , r->local_db.wall_clock_time , r->gap.from, r->gap.to , (r->gap.from == r->wanted.after) ? "FULL" : "PARTIAL" , (st->replay.after != 0 || st->replay.before != 0) ? "OVERLAPPING" : "" ); st->replay.start_streaming = r->wanted.start_streaming; st->replay.after = r->wanted.after; st->replay.before = r->wanted.before; #endif // NETDATA_LOG_REPLICATION_REQUESTS char buffer[2048 + 1]; snprintfz(buffer, sizeof(buffer) - 1, PLUGINSD_KEYWORD_REPLAY_CHART " \"%s\" \"%s\" %llu %llu\n", rrdset_id(st), r->wanted.start_streaming ? "true" : "false", (unsigned long long)r->wanted.after, (unsigned long long)r->wanted.before); ssize_t ret = r->caller.callback(buffer, r->caller.data); if (ret < 0) { netdata_log_error("REPLAY ERROR: 'host:%s/chart:%s' failed to send replication request to child (error %zd)", rrdhost_hostname(r->host), rrdset_id(r->st), ret); return false; } return true; } bool replicate_chart_request(send_command callback, void *callback_data, RRDHOST *host, RRDSET *st, time_t child_first_entry, time_t child_last_entry, time_t child_wall_clock_time, time_t prev_first_entry_wanted, time_t prev_last_entry_wanted) { struct replication_request_details r = { .caller = { .callback = callback, .data = callback_data, }, .host = host, .st = st, .child_db = { .first_entry_t = child_first_entry, .last_entry_t = child_last_entry, .wall_clock_time = child_wall_clock_time, .fixed_last_entry = false, }, .local_db = { .first_entry_t = 0, .last_entry_t = 0, .wall_clock_time = now_realtime_sec(), }, .last_request = { .after = prev_first_entry_wanted, .before = prev_last_entry_wanted, }, .wanted = { .after = 0, .before = 0, .start_streaming = true, }, }; if(r.child_db.last_entry_t > r.child_db.wall_clock_time) { replicate_log_request(&r, "child's db last entry > child's wall clock time"); r.child_db.last_entry_t = r.child_db.wall_clock_time; r.child_db.fixed_last_entry = true; } rrdset_get_retention_of_tier_for_collected_chart(r.st, &r.local_db.first_entry_t, &r.local_db.last_entry_t, r.local_db.wall_clock_time, 0); // let's find the GAP we have if(!r.last_request.after || !r.last_request.before) { // there is no previous request if(r.local_db.last_entry_t) // we have some data, let's continue from the last point we have r.gap.from = r.local_db.last_entry_t; else // we don't have any data, the gap is the max timeframe we are allowed to replicate r.gap.from = r.local_db.wall_clock_time - r.host->rrdpush_seconds_to_replicate; } else { // we had sent a request - let's continue at the point we left it // for this we don't take into account the actual data in our db // because the child may also have gaps, and we need to get over it r.gap.from = r.last_request.before; } // we want all the data up to now r.gap.to = r.local_db.wall_clock_time; // The gap is now r.gap.from -> r.gap.to if (unlikely(!rrdhost_option_check(host, RRDHOST_OPTION_REPLICATION))) return send_replay_chart_cmd(&r, "empty replication request, replication is disabled", false); if (unlikely(!rrdset_number_of_dimensions(st))) return send_replay_chart_cmd(&r, "empty replication request, chart has no dimensions", false); if (unlikely(!r.child_db.first_entry_t || !r.child_db.last_entry_t)) return send_replay_chart_cmd(&r, "empty replication request, child has no stored data", false); if (unlikely(r.child_db.first_entry_t < 0 || r.child_db.last_entry_t < 0)) return send_replay_chart_cmd(&r, "empty replication request, child db timestamps are invalid", true); if (unlikely(r.child_db.first_entry_t > r.child_db.wall_clock_time)) return send_replay_chart_cmd(&r, "empty replication request, child db first entry is after its wall clock time", true); if (unlikely(r.child_db.first_entry_t > r.child_db.last_entry_t)) return send_replay_chart_cmd(&r, "empty replication request, child timings are invalid (first entry > last entry)", true); if (unlikely(r.local_db.last_entry_t > r.child_db.last_entry_t)) return send_replay_chart_cmd(&r, "empty replication request, local last entry is later than the child one", false); // let's find what the child can provide to fill that gap if(r.child_db.first_entry_t > r.gap.from) // the child does not have all the data - let's get what it has r.wanted.after = r.child_db.first_entry_t; else // ok, the child can fill the entire gap we have r.wanted.after = r.gap.from; if(r.gap.to - r.wanted.after > host->rrdpush_replication_step) // the duration is too big for one request - let's take the first step r.wanted.before = r.wanted.after + host->rrdpush_replication_step; else // wow, we can do it in one request r.wanted.before = r.gap.to; // don't ask from the child more than it has if(r.wanted.before > r.child_db.last_entry_t) r.wanted.before = r.child_db.last_entry_t; if(r.wanted.after > r.wanted.before) { r.wanted.after = 0; r.wanted.before = 0; r.wanted.start_streaming = true; return send_replay_chart_cmd(&r, "empty replication request, wanted after computed bigger than wanted before", true); } // the child should start streaming immediately if the wanted duration is small, or we reached the last entry of the child r.wanted.start_streaming = (r.local_db.wall_clock_time - r.wanted.after <= host->rrdpush_replication_step || r.wanted.before >= r.child_db.last_entry_t || r.wanted.before >= r.child_db.wall_clock_time || r.wanted.before >= r.local_db.wall_clock_time); // the wanted timeframe is now r.wanted.after -> r.wanted.before // send it return send_replay_chart_cmd(&r, "OK", false); } // ---------------------------------------------------------------------------- // replication thread // replication request in sender DICTIONARY // used for de-duplicating the requests struct replication_request { struct sender_state *sender; // the sender we should put the reply at STRING *chart_id; // the chart of the request time_t after; // the start time of the query (maybe zero) key for sorting (JudyL) time_t before; // the end time of the query (maybe zero) usec_t sender_last_flush_ut; // the timestamp of the sender, at the time we indexed this request Word_t unique_id; // auto-increment, later requests have bigger bool start_streaming; // true, when the parent wants to send the rest of the data (before is overwritten) and enable normal streaming bool indexed_in_judy; // true when the request is indexed in judy bool not_indexed_buffer_full; // true when the request is not indexed because the sender is full bool not_indexed_preprocessing; // true when the request is not indexed, but it is pending in preprocessing // prepare ahead members - preprocessing bool found; // used as a result boolean for the find call bool executed; // used to detect if we have skipped requests while preprocessing RRDSET *st; // caching of the chart during preprocessing struct replication_query *q; // the preprocessing query initialization }; // replication sort entry in JudyL array // used for sorting all requests, across all nodes struct replication_sort_entry { struct replication_request *rq; size_t unique_id; // used as a key to identify the sort entry - we never access its contents }; #define MAX_REPLICATION_THREADS 20 // + 1 for the main thread // the global variables for the replication thread static struct replication_thread { ARAL *aral_rse; SPINLOCK spinlock; struct { size_t pending; // number of requests pending in the queue // statistics size_t added; // number of requests added to the queue size_t removed; // number of requests removed from the queue size_t pending_no_room; // number of requests skipped, because the sender has no room for responses size_t senders_full; // number of times a sender reset our last position in the queue size_t sender_resets; // number of times a sender reset our last position in the queue time_t first_time_t; // the minimum 'after' we encountered struct { Word_t after; Word_t unique_id; Pvoid_t JudyL_array; } queue; } unsafe; // protected from replication_recursive_lock() struct { Word_t unique_id; // the last unique id we gave to a request (auto-increment, starting from 1) size_t executed; // the number of replication requests executed size_t latest_first_time; // the 'after' timestamp of the last request we executed size_t memory; // the total memory allocated by replication } atomic; // access should be with atomic operations struct { size_t last_executed; // caching of the atomic.executed to report number of requests executed since last time netdata_thread_t **threads_ptrs; size_t threads; } main_thread; // access is allowed only by the main thread } replication_globals = { .aral_rse = NULL, .spinlock = NETDATA_SPINLOCK_INITIALIZER, .unsafe = { .pending = 0, .added = 0, .removed = 0, .pending_no_room = 0, .sender_resets = 0, .senders_full = 0, .first_time_t = 0, .queue = { .after = 0, .unique_id = 0, .JudyL_array = NULL, }, }, .atomic = { .unique_id = 0, .executed = 0, .latest_first_time = 0, .memory = 0, }, .main_thread = { .last_executed = 0, .threads = 0, .threads_ptrs = NULL, }, }; size_t replication_allocated_memory(void) { return __atomic_load_n(&replication_globals.atomic.memory, __ATOMIC_RELAXED); } #define replication_set_latest_first_time(t) __atomic_store_n(&replication_globals.atomic.latest_first_time, t, __ATOMIC_RELAXED) #define replication_get_latest_first_time() __atomic_load_n(&replication_globals.atomic.latest_first_time, __ATOMIC_RELAXED) static inline bool replication_recursive_lock_mode(char mode) { static __thread int recursions = 0; if(mode == 'L') { // (L)ock if(++recursions == 1) spinlock_lock(&replication_globals.spinlock); } else if(mode == 'U') { // (U)nlock if(--recursions == 0) spinlock_unlock(&replication_globals.spinlock); } else if(mode == 'C') { // (C)heck if(recursions > 0) return true; else return false; } else fatal("REPLICATION: unknown lock mode '%c'", mode); #ifdef NETDATA_INTERNAL_CHECKS if(recursions < 0) fatal("REPLICATION: recursions is %d", recursions); #endif return true; } #define replication_recursive_lock() replication_recursive_lock_mode('L') #define replication_recursive_unlock() replication_recursive_lock_mode('U') #define fatal_when_replication_is_not_locked_for_me() do { \ if(!replication_recursive_lock_mode('C')) \ fatal("REPLICATION: reached %s, but replication is not locked by this thread.", __FUNCTION__); \ } while(0) void replication_set_next_point_in_time(time_t after, size_t unique_id) { replication_recursive_lock(); replication_globals.unsafe.queue.after = after; replication_globals.unsafe.queue.unique_id = unique_id; replication_recursive_unlock(); } // ---------------------------------------------------------------------------- // replication sort entry management static inline struct replication_sort_entry *replication_sort_entry_create(struct replication_request *rq) { struct replication_sort_entry *rse = aral_mallocz(replication_globals.aral_rse); __atomic_add_fetch(&replication_globals.atomic.memory, sizeof(struct replication_sort_entry), __ATOMIC_RELAXED); rrdpush_sender_pending_replication_requests_plus_one(rq->sender); // copy the request rse->rq = rq; rse->unique_id = __atomic_add_fetch(&replication_globals.atomic.unique_id, 1, __ATOMIC_SEQ_CST); // save the unique id into the request, to be able to delete it later rq->unique_id = rse->unique_id; rq->indexed_in_judy = false; rq->not_indexed_buffer_full = false; rq->not_indexed_preprocessing = false; return rse; } static void replication_sort_entry_destroy(struct replication_sort_entry *rse) { aral_freez(replication_globals.aral_rse, rse); __atomic_sub_fetch(&replication_globals.atomic.memory, sizeof(struct replication_sort_entry), __ATOMIC_RELAXED); } static void replication_sort_entry_add(struct replication_request *rq) { if(unlikely(rrdpush_sender_replication_buffer_full_get(rq->sender))) { rq->indexed_in_judy = false; rq->not_indexed_buffer_full = true; rq->not_indexed_preprocessing = false; replication_recursive_lock(); replication_globals.unsafe.pending_no_room++; replication_recursive_unlock(); return; } // cache this, because it will be changed bool decrement_no_room = rq->not_indexed_buffer_full; struct replication_sort_entry *rse = replication_sort_entry_create(rq); replication_recursive_lock(); if(decrement_no_room) replication_globals.unsafe.pending_no_room--; // if(rq->after < (time_t)replication_globals.protected.queue.after && // rq->sender->buffer_used_percentage <= MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED && // !replication_globals.protected.skipped_no_room_since_last_reset) { // // // make it find this request first // replication_set_next_point_in_time(rq->after, rq->unique_id); // } replication_globals.unsafe.added++; replication_globals.unsafe.pending++; Pvoid_t *inner_judy_ptr; // find the outer judy entry, using after as key size_t mem_before_outer_judyl = JudyLMemUsed(replication_globals.unsafe.queue.JudyL_array); inner_judy_ptr = JudyLIns(&replication_globals.unsafe.queue.JudyL_array, (Word_t) rq->after, PJE0); size_t mem_after_outer_judyl = JudyLMemUsed(replication_globals.unsafe.queue.JudyL_array); if(unlikely(!inner_judy_ptr || inner_judy_ptr == PJERR)) fatal("REPLICATION: corrupted outer judyL"); // add it to the inner judy, using unique_id as key size_t mem_before_inner_judyl = JudyLMemUsed(*inner_judy_ptr); Pvoid_t *item = JudyLIns(inner_judy_ptr, rq->unique_id, PJE0); size_t mem_after_inner_judyl = JudyLMemUsed(*inner_judy_ptr); if(unlikely(!item || item == PJERR)) fatal("REPLICATION: corrupted inner judyL"); *item = rse; rq->indexed_in_judy = true; rq->not_indexed_buffer_full = false; rq->not_indexed_preprocessing = false; if(!replication_globals.unsafe.first_time_t || rq->after < replication_globals.unsafe.first_time_t) replication_globals.unsafe.first_time_t = rq->after; replication_recursive_unlock(); __atomic_add_fetch(&replication_globals.atomic.memory, (mem_after_inner_judyl - mem_before_inner_judyl) + (mem_after_outer_judyl - mem_before_outer_judyl), __ATOMIC_RELAXED); } static bool replication_sort_entry_unlink_and_free_unsafe(struct replication_sort_entry *rse, Pvoid_t **inner_judy_ppptr, bool preprocessing) { fatal_when_replication_is_not_locked_for_me(); bool inner_judy_deleted = false; replication_globals.unsafe.removed++; replication_globals.unsafe.pending--; rrdpush_sender_pending_replication_requests_minus_one(rse->rq->sender); rse->rq->indexed_in_judy = false; rse->rq->not_indexed_preprocessing = preprocessing; size_t memory_saved = 0; // delete it from the inner judy size_t mem_before_inner_judyl = JudyLMemUsed(**inner_judy_ppptr); JudyLDel(*inner_judy_ppptr, rse->rq->unique_id, PJE0); size_t mem_after_inner_judyl = JudyLMemUsed(**inner_judy_ppptr); memory_saved = mem_before_inner_judyl - mem_after_inner_judyl; // if no items left, delete it from the outer judy if(**inner_judy_ppptr == NULL) { size_t mem_before_outer_judyl = JudyLMemUsed(replication_globals.unsafe.queue.JudyL_array); JudyLDel(&replication_globals.unsafe.queue.JudyL_array, rse->rq->after, PJE0); size_t mem_after_outer_judyl = JudyLMemUsed(replication_globals.unsafe.queue.JudyL_array); memory_saved += mem_before_outer_judyl - mem_after_outer_judyl; inner_judy_deleted = true; } // free memory replication_sort_entry_destroy(rse); __atomic_sub_fetch(&replication_globals.atomic.memory, memory_saved, __ATOMIC_RELAXED); return inner_judy_deleted; } static void replication_sort_entry_del(struct replication_request *rq, bool buffer_full) { Pvoid_t *inner_judy_pptr; struct replication_sort_entry *rse_to_delete = NULL; replication_recursive_lock(); if(rq->indexed_in_judy) { inner_judy_pptr = JudyLGet(replication_globals.unsafe.queue.JudyL_array, rq->after, PJE0); if (inner_judy_pptr) { Pvoid_t *our_item_pptr = JudyLGet(*inner_judy_pptr, rq->unique_id, PJE0); if (our_item_pptr) { rse_to_delete = *our_item_pptr; replication_sort_entry_unlink_and_free_unsafe(rse_to_delete, &inner_judy_pptr, false); if(buffer_full) { replication_globals.unsafe.pending_no_room++; rq->not_indexed_buffer_full = true; } } } if (!rse_to_delete) fatal("REPLAY: 'host:%s/chart:%s' Cannot find sort entry to delete for time %ld.", rrdhost_hostname(rq->sender->host), string2str(rq->chart_id), rq->after); } replication_recursive_unlock(); } static struct replication_request replication_request_get_first_available() { Pvoid_t *inner_judy_pptr; replication_recursive_lock(); struct replication_request rq_to_return = (struct replication_request){ .found = false }; if(unlikely(!replication_globals.unsafe.queue.after || !replication_globals.unsafe.queue.unique_id)) { replication_globals.unsafe.queue.after = 0; replication_globals.unsafe.queue.unique_id = 0; } Word_t started_after = replication_globals.unsafe.queue.after; size_t round = 0; while(!rq_to_return.found) { round++; if(round > 2) break; if(round == 2) { if(started_after == 0) break; replication_globals.unsafe.queue.after = 0; replication_globals.unsafe.queue.unique_id = 0; } bool find_same_after = true; while (!rq_to_return.found && (inner_judy_pptr = JudyLFirstThenNext(replication_globals.unsafe.queue.JudyL_array, &replication_globals.unsafe.queue.after, &find_same_after))) { Pvoid_t *our_item_pptr; if(unlikely(round == 2 && replication_globals.unsafe.queue.after > started_after)) break; while (!rq_to_return.found && (our_item_pptr = JudyLNext(*inner_judy_pptr, &replication_globals.unsafe.queue.unique_id, PJE0))) { struct replication_sort_entry *rse = *our_item_pptr; struct replication_request *rq = rse->rq; // copy the request to return it rq_to_return = *rq; rq_to_return.chart_id = string_dup(rq_to_return.chart_id); // set the return result to found rq_to_return.found = true; if (replication_sort_entry_unlink_and_free_unsafe(rse, &inner_judy_pptr, true)) // we removed the item from the outer JudyL break; } // prepare for the next iteration on the outer loop replication_globals.unsafe.queue.unique_id = 0; } } replication_recursive_unlock(); return rq_to_return; } // ---------------------------------------------------------------------------- // replication request management static void replication_request_react_callback(const DICTIONARY_ITEM *item __maybe_unused, void *value __maybe_unused, void *sender_state __maybe_unused) { struct sender_state *s = sender_state; (void)s; struct replication_request *rq = value; // IMPORTANT: // We use the react instead of the insert callback // because we want the item to be atomically visible // to our replication thread, immediately after. // If we put this at the insert callback, the item is not guaranteed // to be atomically visible to others, so the replication thread // may see the replication sort entry, but fail to find the dictionary item // related to it. replication_sort_entry_add(rq); // this request is about a unique chart for this sender rrdpush_sender_replicating_charts_plus_one(s); } static bool replication_request_conflict_callback(const DICTIONARY_ITEM *item __maybe_unused, void *old_value, void *new_value, void *sender_state) { struct sender_state *s = sender_state; (void)s; struct replication_request *rq = old_value; (void)rq; struct replication_request *rq_new = new_value; replication_recursive_lock(); if(!rq->indexed_in_judy && rq->not_indexed_buffer_full && !rq->not_indexed_preprocessing) { // we can replace this command internal_error( true, "STREAM %s [send to %s]: REPLAY: 'host:%s/chart:%s' replacing duplicate replication command received (existing from %llu to %llu [%s], new from %llu to %llu [%s])", rrdhost_hostname(s->host), s->connected_to, rrdhost_hostname(s->host), dictionary_acquired_item_name(item), (unsigned long long)rq->after, (unsigned long long)rq->before, rq->start_streaming ? "true" : "false", (unsigned long long)rq_new->after, (unsigned long long)rq_new->before, rq_new->start_streaming ? "true" : "false"); rq->after = rq_new->after; rq->before = rq_new->before; rq->start_streaming = rq_new->start_streaming; } else if(!rq->indexed_in_judy && !rq->not_indexed_preprocessing) { replication_sort_entry_add(rq); internal_error( true, "STREAM %s [send to %s]: REPLAY: 'host:%s/chart:%s' adding duplicate replication command received (existing from %llu to %llu [%s], new from %llu to %llu [%s])", rrdhost_hostname(s->host), s->connected_to, rrdhost_hostname(s->host), dictionary_acquired_item_name(item), (unsigned long long)rq->after, (unsigned long long)rq->before, rq->start_streaming ? "true" : "false", (unsigned long long)rq_new->after, (unsigned long long)rq_new->before, rq_new->start_streaming ? "true" : "false"); } else { internal_error( true, "STREAM %s [send to %s]: REPLAY: 'host:%s/chart:%s' ignoring duplicate replication command received (existing from %llu to %llu [%s], new from %llu to %llu [%s])", rrdhost_hostname(s->host), s->connected_to, rrdhost_hostname(s->host), dictionary_acquired_item_name(item), (unsigned long long) rq->after, (unsigned long long) rq->before, rq->start_streaming ? "true" : "false", (unsigned long long) rq_new->after, (unsigned long long) rq_new->before, rq_new->start_streaming ? "true" : "false"); } replication_recursive_unlock(); string_freez(rq_new->chart_id); return false; } static void replication_request_delete_callback(const DICTIONARY_ITEM *item __maybe_unused, void *value, void *sender_state __maybe_unused) { struct replication_request *rq = value; // this request is about a unique chart for this sender rrdpush_sender_replicating_charts_minus_one(rq->sender); if(rq->indexed_in_judy) replication_sort_entry_del(rq, false); else if(rq->not_indexed_buffer_full) { replication_recursive_lock(); replication_globals.unsafe.pending_no_room--; replication_recursive_unlock(); } string_freez(rq->chart_id); } static bool sender_is_still_connected_for_this_request(struct replication_request *rq) { return rq->sender_last_flush_ut == rrdpush_sender_get_flush_time(rq->sender); }; static bool replication_execute_request(struct replication_request *rq, bool workers) { bool ret = false; if(!rq->st) { if(likely(workers)) worker_is_busy(WORKER_JOB_FIND_CHART); rq->st = rrdset_find(rq->sender->host, string2str(rq->chart_id)); } if(!rq->st) { internal_error(true, "REPLAY ERROR: 'host:%s/chart:%s' not found", rrdhost_hostname(rq->sender->host), string2str(rq->chart_id)); goto cleanup; } netdata_thread_disable_cancelability(); if(!rq->q) { if(likely(workers)) worker_is_busy(WORKER_JOB_PREPARE_QUERY); rq->q = replication_response_prepare( rq->st, rq->start_streaming, rq->after, rq->before, rq->sender->capabilities); } if(likely(workers)) worker_is_busy(WORKER_JOB_QUERYING); // send the replication data rq->q->rq = rq; replication_response_execute_and_finalize( rq->q, (size_t)((unsigned long long)rq->sender->host->sender->buffer->max_size * MAX_REPLICATION_MESSAGE_PERCENT_SENDER_BUFFER / 100ULL)); rq->q = NULL; netdata_thread_enable_cancelability(); __atomic_add_fetch(&replication_globals.atomic.executed, 1, __ATOMIC_RELAXED); ret = true; cleanup: if(rq->q) { replication_response_cancel_and_finalize(rq->q); rq->q = NULL; } string_freez(rq->chart_id); worker_is_idle(); return ret; } // ---------------------------------------------------------------------------- // public API void replication_add_request(struct sender_state *sender, const char *chart_id, time_t after, time_t before, bool start_streaming) { struct replication_request rq = { .sender = sender, .chart_id = string_strdupz(chart_id), .after = after, .before = before, .start_streaming = start_streaming, .sender_last_flush_ut = rrdpush_sender_get_flush_time(sender), .indexed_in_judy = false, .not_indexed_buffer_full = false, .not_indexed_preprocessing = false, }; if(!sender->replication.oldest_request_after_t || rq.after < sender->replication.oldest_request_after_t) sender->replication.oldest_request_after_t = rq.after; if(start_streaming && rrdpush_sender_get_buffer_used_percent(sender) <= STREAMING_START_MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED) replication_execute_request(&rq, false); else dictionary_set(sender->replication.requests, chart_id, &rq, sizeof(struct replication_request)); } void replication_sender_delete_pending_requests(struct sender_state *sender) { // allow the dictionary destructor to go faster on locks dictionary_flush(sender->replication.requests); } void replication_init_sender(struct sender_state *sender) { sender->replication.requests = dictionary_create_advanced(DICT_OPTION_DONT_OVERWRITE_VALUE | DICT_OPTION_FIXED_SIZE, NULL, sizeof(struct replication_request)); dictionary_register_react_callback(sender->replication.requests, replication_request_react_callback, sender); dictionary_register_conflict_callback(sender->replication.requests, replication_request_conflict_callback, sender); dictionary_register_delete_callback(sender->replication.requests, replication_request_delete_callback, sender); } void replication_cleanup_sender(struct sender_state *sender) { // allow the dictionary destructor to go faster on locks replication_recursive_lock(); dictionary_destroy(sender->replication.requests); replication_recursive_unlock(); } void replication_recalculate_buffer_used_ratio_unsafe(struct sender_state *s) { size_t available = cbuffer_available_size_unsafe(s->host->sender->buffer); size_t percentage = (s->buffer->max_size - available) * 100 / s->buffer->max_size; if(unlikely(percentage > MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED && !rrdpush_sender_replication_buffer_full_get(s))) { rrdpush_sender_replication_buffer_full_set(s, true); struct replication_request *rq; dfe_start_read(s->replication.requests, rq) { if(rq->indexed_in_judy) replication_sort_entry_del(rq, true); } dfe_done(rq); replication_recursive_lock(); replication_globals.unsafe.senders_full++; replication_recursive_unlock(); } else if(unlikely(percentage < MIN_SENDER_BUFFER_PERCENTAGE_ALLOWED && rrdpush_sender_replication_buffer_full_get(s))) { rrdpush_sender_replication_buffer_full_set(s, false); struct replication_request *rq; dfe_start_read(s->replication.requests, rq) { if(!rq->indexed_in_judy && (rq->not_indexed_buffer_full || rq->not_indexed_preprocessing)) replication_sort_entry_add(rq); } dfe_done(rq); replication_recursive_lock(); replication_globals.unsafe.senders_full--; replication_globals.unsafe.sender_resets++; // replication_set_next_point_in_time(0, 0); replication_recursive_unlock(); } rrdpush_sender_set_buffer_used_percent(s, percentage); } // ---------------------------------------------------------------------------- // replication thread static size_t verify_host_charts_are_streaming_now(RRDHOST *host) { internal_error( host->sender && !rrdpush_sender_pending_replication_requests(host->sender) && dictionary_entries(host->sender->replication.requests) != 0, "REPLICATION SUMMARY: 'host:%s' reports %zu pending replication requests, but its chart replication index says there are %zu charts pending replication", rrdhost_hostname(host), rrdpush_sender_pending_replication_requests(host->sender), dictionary_entries(host->sender->replication.requests) ); size_t ok = 0; size_t errors = 0; RRDSET *st; rrdset_foreach_read(st, host) { RRDSET_FLAGS flags = rrdset_flag_check(st, RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS | RRDSET_FLAG_SENDER_REPLICATION_FINISHED); bool is_error = false; if(!flags) { internal_error( true, "REPLICATION SUMMARY: 'host:%s/chart:%s' is neither IN PROGRESS nor FINISHED", rrdhost_hostname(host), rrdset_id(st) ); is_error = true; } if(!(flags & RRDSET_FLAG_SENDER_REPLICATION_FINISHED) || (flags & RRDSET_FLAG_SENDER_REPLICATION_IN_PROGRESS)) { internal_error( true, "REPLICATION SUMMARY: 'host:%s/chart:%s' is IN PROGRESS although replication is finished", rrdhost_hostname(host), rrdset_id(st) ); is_error = true; } if(is_error) errors++; else ok++; } rrdset_foreach_done(st); internal_error(errors, "REPLICATION SUMMARY: 'host:%s' finished replicating %zu charts, but %zu charts are still in progress although replication finished", rrdhost_hostname(host), ok, errors); return errors; } static void verify_all_hosts_charts_are_streaming_now(void) { worker_is_busy(WORKER_JOB_CHECK_CONSISTENCY); size_t errors = 0; RRDHOST *host; dfe_start_read(rrdhost_root_index, host) errors += verify_host_charts_are_streaming_now(host); dfe_done(host); size_t executed = __atomic_load_n(&replication_globals.atomic.executed, __ATOMIC_RELAXED); netdata_log_info("REPLICATION SUMMARY: finished, executed %zu replication requests, %zu charts pending replication", executed - replication_globals.main_thread.last_executed, errors); replication_globals.main_thread.last_executed = executed; } static void replication_initialize_workers(bool master) { worker_register("REPLICATION"); worker_register_job_name(WORKER_JOB_FIND_NEXT, "find next"); worker_register_job_name(WORKER_JOB_QUERYING, "querying"); worker_register_job_name(WORKER_JOB_DELETE_ENTRY, "dict delete"); worker_register_job_name(WORKER_JOB_FIND_CHART, "find chart"); worker_register_job_name(WORKER_JOB_PREPARE_QUERY, "prepare query"); worker_register_job_name(WORKER_JOB_CHECK_CONSISTENCY, "check consistency"); worker_register_job_name(WORKER_JOB_BUFFER_COMMIT, "commit"); worker_register_job_name(WORKER_JOB_CLEANUP, "cleanup"); worker_register_job_name(WORKER_JOB_WAIT, "wait"); if(master) { worker_register_job_name(WORKER_JOB_STATISTICS, "statistics"); worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_PENDING_REQUESTS, "pending requests", "requests", WORKER_METRIC_ABSOLUTE); worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_SKIPPED_NO_ROOM, "no room requests", "requests", WORKER_METRIC_ABSOLUTE); worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_COMPLETION, "completion", "%", WORKER_METRIC_ABSOLUTE); worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_ADDED, "added requests", "requests/s", WORKER_METRIC_INCREMENTAL_TOTAL); worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_DONE, "finished requests", "requests/s", WORKER_METRIC_INCREMENTAL_TOTAL); worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_SENDER_RESETS, "sender resets", "resets/s", WORKER_METRIC_INCREMENTAL_TOTAL); worker_register_job_custom_metric(WORKER_JOB_CUSTOM_METRIC_SENDER_FULL, "senders full", "senders", WORKER_METRIC_ABSOLUTE); } } #define REQUEST_OK (0) #define REQUEST_QUEUE_EMPTY (-1) #define REQUEST_CHART_NOT_FOUND (-2) static __thread struct replication_thread_pipeline { int max_requests_ahead; struct replication_request *rqs; int rqs_last_executed, rqs_last_prepared; size_t queue_rounds; } rtp = { .max_requests_ahead = 0, .rqs = NULL, .rqs_last_executed = 0, .rqs_last_prepared = 0, .queue_rounds = 0, }; static void replication_pipeline_cancel_and_cleanup(void) { if(!rtp.rqs) return; struct replication_request *rq; size_t cancelled = 0; do { if (++rtp.rqs_last_executed >= rtp.max_requests_ahead) rtp.rqs_last_executed = 0; rq = &rtp.rqs[rtp.rqs_last_executed]; if (rq->q) { internal_fatal(rq->executed, "REPLAY FATAL: query has already been executed!"); internal_fatal(!rq->found, "REPLAY FATAL: orphan q in rq"); replication_response_cancel_and_finalize(rq->q); rq->q = NULL; cancelled++; } rq->executed = true; rq->found = false; } while (rtp.rqs_last_executed != rtp.rqs_last_prepared); internal_error(true, "REPLICATION: cancelled %zu inflight queries", cancelled); freez(rtp.rqs); rtp.rqs = NULL; rtp.max_requests_ahead = 0; rtp.rqs_last_executed = 0; rtp.rqs_last_prepared = 0; rtp.queue_rounds = 0; } static int replication_pipeline_execute_next(void) { struct replication_request *rq; if(unlikely(!rtp.rqs)) { rtp.max_requests_ahead = (int)get_netdata_cpus() / 2; if(rtp.max_requests_ahead > libuv_worker_threads * 2) rtp.max_requests_ahead = libuv_worker_threads * 2; if(rtp.max_requests_ahead < 2) rtp.max_requests_ahead = 2; rtp.rqs = callocz(rtp.max_requests_ahead, sizeof(struct replication_request)); __atomic_add_fetch(&replication_buffers_allocated, rtp.max_requests_ahead * sizeof(struct replication_request), __ATOMIC_RELAXED); } // fill the queue do { if(++rtp.rqs_last_prepared >= rtp.max_requests_ahead) { rtp.rqs_last_prepared = 0; rtp.queue_rounds++; } internal_fatal(rtp.rqs[rtp.rqs_last_prepared].q, "REPLAY FATAL: slot is used by query that has not been executed!"); worker_is_busy(WORKER_JOB_FIND_NEXT); rtp.rqs[rtp.rqs_last_prepared] = replication_request_get_first_available(); rq = &rtp.rqs[rtp.rqs_last_prepared]; if(rq->found) { if (!rq->st) { worker_is_busy(WORKER_JOB_FIND_CHART); rq->st = rrdset_find(rq->sender->host, string2str(rq->chart_id)); } if (rq->st && !rq->q) { worker_is_busy(WORKER_JOB_PREPARE_QUERY); rq->q = replication_response_prepare( rq->st, rq->start_streaming, rq->after, rq->before, rq->sender->capabilities); } rq->executed = false; } } while(rq->found && rtp.rqs_last_prepared != rtp.rqs_last_executed); // pick the first usable do { if (++rtp.rqs_last_executed >= rtp.max_requests_ahead) rtp.rqs_last_executed = 0; rq = &rtp.rqs[rtp.rqs_last_executed]; if(rq->found) { internal_fatal(rq->executed, "REPLAY FATAL: query has already been executed!"); if (rq->sender_last_flush_ut != rrdpush_sender_get_flush_time(rq->sender)) { // the sender has reconnected since this request was queued, // we can safely throw it away, since the parent will resend it replication_response_cancel_and_finalize(rq->q); rq->executed = true; rq->found = false; rq->q = NULL; } else if (rrdpush_sender_replication_buffer_full_get(rq->sender)) { // the sender buffer is full, so we can ignore this request, // it has already been marked as 'preprocessed' in the dictionary, // and the sender will put it back in when there is // enough room in the buffer for processing replication requests replication_response_cancel_and_finalize(rq->q); rq->executed = true; rq->found = false; rq->q = NULL; } else { // we can execute this, // delete it from the dictionary worker_is_busy(WORKER_JOB_DELETE_ENTRY); dictionary_del(rq->sender->replication.requests, string2str(rq->chart_id)); } } else internal_fatal(rq->q, "REPLAY FATAL: slot status says slot is empty, but it has a pending query!"); } while(!rq->found && rtp.rqs_last_executed != rtp.rqs_last_prepared); if(unlikely(!rq->found)) { worker_is_idle(); return REQUEST_QUEUE_EMPTY; } replication_set_latest_first_time(rq->after); bool chart_found = replication_execute_request(rq, true); rq->executed = true; rq->found = false; rq->q = NULL; if(unlikely(!chart_found)) { worker_is_idle(); return REQUEST_CHART_NOT_FOUND; } worker_is_idle(); return REQUEST_OK; } static void replication_worker_cleanup(void *ptr __maybe_unused) { replication_pipeline_cancel_and_cleanup(); worker_unregister(); } static void *replication_worker_thread(void *ptr) { replication_initialize_workers(false); netdata_thread_cleanup_push(replication_worker_cleanup, ptr); while(service_running(SERVICE_REPLICATION)) { if(unlikely(replication_pipeline_execute_next() == REQUEST_QUEUE_EMPTY)) { sender_thread_buffer_free(); worker_is_busy(WORKER_JOB_WAIT); worker_is_idle(); sleep_usec(1 * USEC_PER_SEC); } } netdata_thread_cleanup_pop(1); return NULL; } static void replication_main_cleanup(void *ptr) { struct netdata_static_thread *static_thread = (struct netdata_static_thread *)ptr; static_thread->enabled = NETDATA_MAIN_THREAD_EXITING; replication_pipeline_cancel_and_cleanup(); int threads = (int)replication_globals.main_thread.threads; for(int i = 0; i < threads ;i++) { netdata_thread_join(*replication_globals.main_thread.threads_ptrs[i], NULL); freez(replication_globals.main_thread.threads_ptrs[i]); __atomic_sub_fetch(&replication_buffers_allocated, sizeof(netdata_thread_t), __ATOMIC_RELAXED); } freez(replication_globals.main_thread.threads_ptrs); replication_globals.main_thread.threads_ptrs = NULL; __atomic_sub_fetch(&replication_buffers_allocated, threads * sizeof(netdata_thread_t), __ATOMIC_RELAXED); aral_destroy(replication_globals.aral_rse); replication_globals.aral_rse = NULL; // custom code worker_unregister(); static_thread->enabled = NETDATA_MAIN_THREAD_EXITED; } void replication_initialize(void) { replication_globals.aral_rse = aral_create("rse", sizeof(struct replication_sort_entry), 0, 65536, aral_by_size_statistics(), NULL, NULL, false, false); } void *replication_thread_main(void *ptr __maybe_unused) { replication_initialize_workers(true); int threads = config_get_number(CONFIG_SECTION_DB, "replication threads", 1); if(threads < 1 || threads > MAX_REPLICATION_THREADS) { netdata_log_error("replication threads given %d is invalid, resetting to 1", threads); threads = 1; } if(--threads) { replication_globals.main_thread.threads = threads; replication_globals.main_thread.threads_ptrs = mallocz(threads * sizeof(netdata_thread_t *)); __atomic_add_fetch(&replication_buffers_allocated, threads * sizeof(netdata_thread_t), __ATOMIC_RELAXED); for(int i = 0; i < threads ;i++) { char tag[NETDATA_THREAD_TAG_MAX + 1]; snprintfz(tag, NETDATA_THREAD_TAG_MAX, "REPLAY[%d]", i + 2); replication_globals.main_thread.threads_ptrs[i] = mallocz(sizeof(netdata_thread_t)); __atomic_add_fetch(&replication_buffers_allocated, sizeof(netdata_thread_t), __ATOMIC_RELAXED); netdata_thread_create(replication_globals.main_thread.threads_ptrs[i], tag, NETDATA_THREAD_OPTION_JOINABLE, replication_worker_thread, NULL); } } netdata_thread_cleanup_push(replication_main_cleanup, ptr); // start from 100% completed worker_set_metric(WORKER_JOB_CUSTOM_METRIC_COMPLETION, 100.0); long run_verification_countdown = LONG_MAX; // LONG_MAX to prevent an initial verification when no replication ever took place bool slow = true; // control the time we sleep - it has to start with true! usec_t last_now_mono_ut = now_monotonic_usec(); time_t replication_reset_next_point_in_time_countdown = SECONDS_TO_RESET_POINT_IN_TIME; // restart from the beginning every 10 seconds size_t last_executed = 0; size_t last_sender_resets = 0; while(service_running(SERVICE_REPLICATION)) { // statistics usec_t now_mono_ut = now_monotonic_usec(); if(unlikely(now_mono_ut - last_now_mono_ut > default_rrd_update_every * USEC_PER_SEC)) { last_now_mono_ut = now_mono_ut; worker_is_busy(WORKER_JOB_STATISTICS); replication_recursive_lock(); size_t current_executed = __atomic_load_n(&replication_globals.atomic.executed, __ATOMIC_RELAXED); if(last_executed != current_executed) { run_verification_countdown = ITERATIONS_IDLE_WITHOUT_PENDING_TO_RUN_SENDER_VERIFICATION; last_executed = current_executed; slow = false; } if(replication_reset_next_point_in_time_countdown-- == 0) { // once per second, make it scan all the pending requests next time replication_set_next_point_in_time(0, 0); // replication_globals.protected.skipped_no_room_since_last_reset = 0; replication_reset_next_point_in_time_countdown = SECONDS_TO_RESET_POINT_IN_TIME; } if(--run_verification_countdown == 0) { if (!replication_globals.unsafe.pending && !replication_globals.unsafe.pending_no_room) { // reset the statistics about completion percentage replication_globals.unsafe.first_time_t = 0; replication_set_latest_first_time(0); verify_all_hosts_charts_are_streaming_now(); run_verification_countdown = LONG_MAX; slow = true; } else run_verification_countdown = ITERATIONS_IDLE_WITHOUT_PENDING_TO_RUN_SENDER_VERIFICATION; } time_t latest_first_time_t = replication_get_latest_first_time(); if(latest_first_time_t && replication_globals.unsafe.pending) { // completion percentage statistics time_t now = now_realtime_sec(); time_t total = now - replication_globals.unsafe.first_time_t; time_t done = latest_first_time_t - replication_globals.unsafe.first_time_t; worker_set_metric(WORKER_JOB_CUSTOM_METRIC_COMPLETION, (NETDATA_DOUBLE) done * 100.0 / (NETDATA_DOUBLE) total); } else worker_set_metric(WORKER_JOB_CUSTOM_METRIC_COMPLETION, 100.0); worker_set_metric(WORKER_JOB_CUSTOM_METRIC_PENDING_REQUESTS, (NETDATA_DOUBLE)replication_globals.unsafe.pending); worker_set_metric(WORKER_JOB_CUSTOM_METRIC_ADDED, (NETDATA_DOUBLE)replication_globals.unsafe.added); worker_set_metric(WORKER_JOB_CUSTOM_METRIC_DONE, (NETDATA_DOUBLE)__atomic_load_n(&replication_globals.atomic.executed, __ATOMIC_RELAXED)); worker_set_metric(WORKER_JOB_CUSTOM_METRIC_SKIPPED_NO_ROOM, (NETDATA_DOUBLE)replication_globals.unsafe.pending_no_room); worker_set_metric(WORKER_JOB_CUSTOM_METRIC_SENDER_RESETS, (NETDATA_DOUBLE)replication_globals.unsafe.sender_resets); worker_set_metric(WORKER_JOB_CUSTOM_METRIC_SENDER_FULL, (NETDATA_DOUBLE)replication_globals.unsafe.senders_full); replication_recursive_unlock(); worker_is_idle(); } if(unlikely(replication_pipeline_execute_next() == REQUEST_QUEUE_EMPTY)) { worker_is_busy(WORKER_JOB_WAIT); replication_recursive_lock(); // the timeout also defines now frequently we will traverse all the pending requests // when the outbound buffers of all senders is full usec_t timeout; if(slow) { // no work to be done, wait for a request to come in timeout = 1000 * USEC_PER_MS; sender_thread_buffer_free(); } else if(replication_globals.unsafe.pending > 0) { if(replication_globals.unsafe.sender_resets == last_sender_resets) timeout = 1000 * USEC_PER_MS; else { // there are pending requests waiting to be executed, // but none could be executed at this time. // try again after this time. timeout = 100 * USEC_PER_MS; } last_sender_resets = replication_globals.unsafe.sender_resets; } else { // no requests pending, but there were requests recently (run_verification_countdown) // so, try in a short time. // if this is big, one chart replicating will be slow to finish (ping - pong just one chart) timeout = 10 * USEC_PER_MS; last_sender_resets = replication_globals.unsafe.sender_resets; } replication_recursive_unlock(); worker_is_idle(); sleep_usec(timeout); // make it scan all the pending requests next time replication_set_next_point_in_time(0, 0); replication_reset_next_point_in_time_countdown = SECONDS_TO_RESET_POINT_IN_TIME; continue; } } netdata_thread_cleanup_pop(1); return NULL; }