replicating gaps (#14506)
* instead of sending RBEGIN/REND without points, jump to the end of the gap on every replication step * Remove check for memory mode NONE so that replication when INTERPOLATE is negotiated will work * Remove unused label --------- Co-authored-by: Stelios Fragkakis <52996999+stelfrag@users.noreply.github.com>
Costa Tsaousis committed
Feb 14, 2023 at 13:10 UTC
8fdeec2e85cf5b1de742d672c57e2c0f276e8867
3 files changed
+13
-13
database/engine/rrdengineapi.c
+1
@@ -871,6 +871,7 @@ STORAGE_POINT rrdeng_load_metric_next(struct storage_engine_query_handle *rrddim
871
// We need to get a new page
872
873
if (!rrdeng_load_page_next(rrddim_handle, false)) {
874
+ handle->now_s = rrddim_handle->end_time_s;
875
storage_point_empty(sp, handle->now_s - handle->dt_s, handle->now_s);
876
goto prepare_for_next_iteration;
877
}
database/rrdset.c
-8
@@ -1543,9 +1543,6 @@ void rrdset_timed_done(RRDSET *st, struct timeval now, bool pending_rrdset_next)
1543
// calculate the proper last_collected_time, using usec_since_last_update
1544
last_collect_ut = rrdset_update_last_collected_time(st);
1545
}
1546
- if (unlikely(st->rrd_memory_mode == RRD_MEMORY_MODE_NONE)) {
1547
- goto after_first_database_work;
1548
- }
1546
1547
// if this set has not been updated in the past
1548
// we fake the last_update time to be = now - usec_since_last_update
@@ -1608,7 +1605,6 @@ void rrdset_timed_done(RRDSET *st, struct timeval now, bool pending_rrdset_next)
1605
}
1606
}
1607
1611
-after_first_database_work:
1608
st->counter_done++;
1609
1610
if(stream_buffer.wb && !stream_buffer.v2)
@@ -1670,9 +1666,6 @@ after_first_database_work:
1666
rrddim_foreach_done(rd);
1667
rda_slots = dimensions;
1668
1673
- if (unlikely(st->rrd_memory_mode == RRD_MEMORY_MODE_NONE))
1674
- goto after_second_database_work;
1675
-
1669
rrdset_debug(st, "last_collect_ut = %0.3" NETDATA_DOUBLE_MODIFIER " (last collection time)", (NETDATA_DOUBLE)last_collect_ut/USEC_PER_SEC);
1670
rrdset_debug(st, "now_collect_ut = %0.3" NETDATA_DOUBLE_MODIFIER " (current collection time)", (NETDATA_DOUBLE)now_collect_ut/USEC_PER_SEC);
1671
rrdset_debug(st, "last_stored_ut = %0.3" NETDATA_DOUBLE_MODIFIER " (last updated time)", (NETDATA_DOUBLE)last_stored_ut/USEC_PER_SEC);
@@ -1886,7 +1879,6 @@ after_first_database_work:
1879
, has_reset_value
1880
);
1881
1889
-after_second_database_work:
1882
for(dim_id = 0, rda = rda_base ; dim_id < rda_slots ; ++dim_id, ++rda) {
1883
rd = rda->rd;
1884
if(unlikely(!rd)) continue;
streaming/replication.c
+12
-5
@@ -310,7 +310,8 @@ static bool replication_query_execute(BUFFER *wb, struct replication_query *q, s
310
struct storage_engine_query_ops *ops = q->ops;
311
time_t wall_clock_time = q->wall_clock_time;
312
313
- size_t points_read = q->points_read, points_generated = q->points_generated;
313
+ bool finished_with_gap = false;
314
+ size_t points_read = 0, points_generated = 0;
315
316
#ifdef NETDATA_LOG_REPLICATION_REQUESTS
317
time_t actual_after = 0, actual_before = 0;
@@ -469,9 +470,16 @@ static bool replication_query_execute(BUFFER *wb, struct replication_query *q, s
470
else if(unlikely(min_end_time < now))
471
// the query does not progress
472
break;
472
- else
473
+ else {
474
// we have gap - all points are in the future
475
now = min_start_time;
476
+
477
+ if(min_start_time > before && !points_generated) {
478
+ before = q->query.before = min_start_time - 1;
479
+ finished_with_gap = true;
480
+ break;
481
+ }
482
+ }
483
}
484
485
#ifdef NETDATA_LOG_REPLICATION_REQUESTS
@@ -492,10 +500,9 @@ static bool replication_query_execute(BUFFER *wb, struct replication_query *q, s
500
(unsigned long long)after, (unsigned long long)before);
501
#endif // NETDATA_LOG_REPLICATION_REQUESTS
502
495
- q->points_read = points_read;
496
- q->points_generated = points_generated;
503
+ q->points_read += points_read;
504
+ q->points_generated += points_generated;
505
498
- bool finished_with_gap = false;
506
if(last_end_time_in_buffer < before - q->st->update_every)
507
finished_with_gap = true;
508