replication to streaming transition when there are gaps (#14434)
fix https://github.com/netdata/netdata/issues/14432
Costa Tsaousis committed
Feb 6, 2023 at 12:43 UTC
d53ac630f8c6abfd7f9b7bf9fd4029a164ca775f
2 files changed
+14
-4
streaming/replication.c
+12
-2
@@ -293,7 +293,7 @@ static void replication_query_align_to_optimal_before(struct replication_query *
293
q->query.before = expanded_before;
294
}
295
296
-static void replication_query_execute(BUFFER *wb, struct replication_query *q, size_t max_msg_size) {
296
+static bool replication_query_execute(BUFFER *wb, struct replication_query *q, size_t max_msg_size) {
297
replication_query_align_to_optimal_before(q);
298
299
time_t after = q->query.after;
@@ -475,6 +475,12 @@ static void replication_query_execute(BUFFER *wb, struct replication_query *q, s
475
476
q->points_read = points_read;
477
q->points_generated = points_generated;
478
+
479
+ bool finished_with_gap = false;
480
+ if(last_end_time_in_buffer < before - q->st->update_every)
481
+ finished_with_gap = true;
482
+
483
+ return finished_with_gap;
484
}
485
486
static struct replication_query *replication_response_prepare(RRDSET *st, bool requested_enable_streaming, time_t requested_after, time_t requested_before) {
@@ -561,8 +567,9 @@ bool replication_response_execute_and_finalize(struct replication_query *q, size
567
bool locked_data_collection = q->query.locked_data_collection;
568
q->query.locked_data_collection = false;
569
570
+ bool finished_with_gap = false;
571
if(q->query.execute)
565
- replication_query_execute(wb, q, max_msg_size);
572
+ finished_with_gap = replication_query_execute(wb, q, max_msg_size);
573
574
time_t after = q->request.after;
575
time_t before = q->query.before;
@@ -610,6 +617,9 @@ bool replication_response_execute_and_finalize(struct replication_query *q, size
617
rrdset_flag_set(st, RRDSET_FLAG_SENDER_REPLICATION_FINISHED);
618
rrdhost_sender_replicating_charts_minus_one(st->rrdhost);
619
620
+ if(!finished_with_gap)
621
+ st->upstream_resync_time_s = 0;
622
+
623
#ifdef NETDATA_LOG_REPLICATION_REQUESTS
624
internal_error(true, "STREAM_SENDER REPLAY: 'host:%s/chart:%s' streaming starts",
625
rrdhost_hostname(st->rrdhost), rrdset_id(st));
streaming/rrdpush.c
+2
-2
@@ -321,12 +321,12 @@ static inline bool rrdpush_send_chart_definition(BUFFER *wb, RRDSET *st) {
321
}
322
323
// sends the current chart dimensions
324
-static void rrdpush_send_chart_metrics(BUFFER *wb, RRDSET *st, struct sender_state *s, RRDSET_FLAGS flags) {
324
+static void rrdpush_send_chart_metrics(BUFFER *wb, RRDSET *st, struct sender_state *s __maybe_unused, RRDSET_FLAGS flags) {
325
buffer_fast_strcat(wb, "BEGIN \"", 7);
326
buffer_fast_strcat(wb, rrdset_id(st), string_strlen(st->id));
327
buffer_fast_strcat(wb, "\" ", 2);
328
329
- if(stream_has_capability(s, STREAM_CAP_REPLICATION) || st->last_collected_time.tv_sec > st->upstream_resync_time_s)
329
+ if(st->last_collected_time.tv_sec > st->upstream_resync_time_s)
330
buffer_print_llu(wb, st->usec_since_last_update);
331
else
332
buffer_fast_strcat(wb, "0", 1);