@cryptotaxi247 / netdata-1 / commits / 57eab742c

DBENGINE v2 - improvements part 10 (#14332)

* replication cancels pending queries on exit * log when waiting for inflight queries * when there are collected and not-collected metrics, use the context priority from the collected only * Write metadata with a faster pace * Remove journal file size limit and sync mode to 0 / Drop wal checkpoint for now * Wrap in a big transaction remaining metadata writes (test 1) * fix higher tiers when tiering iterations = 2 * dbengine always returns db-aligned points; query engine expands the queries by 2 points in every direction to have enough data for interpolation * Wrap in a big transaction metadata writes (test 2) * replication cancelling fix * do not first and last entry in replication when the db has no retention * fix internal check condition * Increase metadata write batch size * always apply error limit to dbengine logs * Remove code that processes the obsolete health.db files * cleanup in query.c * do not allow queries to go beyond db boundaries * prevent internal log for +1 delta in timestamp * detect gap pages in conflicts * double protection for gap injection in main cache * Add checkpoint to prevent large WAL while running Remove unused and duplicate functions * do not allocate chart cache dir if not needed * add more info to unittests * revert query expansion to satisfy unittests Co-authored-by: Stelios Fragkakis <52996999+stelfrag@users.noreply.github.com>

Costa Tsaousis committed Jan 27, 2023 at 01:32 UTC 57eab742c88093c89d5d46deb495558ad726e6f0
21 files changed +305 -703
daemon/unit_test.c
+21 -24
@@ -1858,7 +1858,7 @@ static void test_dbengine_create_charts(RRDHOST *host, RRDSET *st[CHARTS], RRDDI
1858 now_realtime_timeval(&now);
1859 rrdset_timed_done(st[i], now, false);
1860 }
1861 - // Fluh pages for subsequent real values
1861 + // Flush pages for subsequent real values
1862 for (i = 0 ; i < CHARTS ; ++i) {
1863 for (j = 0; j < DIMS; ++j) {
1864 rrdeng_store_metric_flush_current_page((rd[i][j])->tiers[0].db_collection_handle);
@@ -1978,10 +1978,11 @@ static int test_dbengine_check_rrdr(RRDSET *st[CHARTS], RRDDIM *rd[CHARTS][DIMS]
1978 int current_region, time_t time_start, time_t time_end)
1979 {
1980 int update_every = REGION_UPDATE_EVERY[current_region];
1981 - fprintf(stderr, "%s() running on region %d, start time %lld, end time %lld, update every %d...\n", __FUNCTION__, current_region, (long long)time_start, (long long)time_end, update_every);
1981 + fprintf(stderr, "%s() running on region %d, start time %lld, end time %lld, update every %d, on %d dimensions...\n",
1982 + __FUNCTION__, current_region, (long long)time_start, (long long)time_end, update_every, CHARTS * DIMS);
1983 uint8_t same;
1984 time_t time_now, time_retrieved;
1984 - int i, j, errors, value_errors = 0, time_errors = 0;
1985 + int i, j, errors, value_errors = 0, time_errors = 0, value_right = 0, time_right = 0;
1986 long c;
1987 collected_number last;
1988 NETDATA_DOUBLE value, expected;
@@ -2020,17 +2021,22 @@ static int test_dbengine_check_rrdr(RRDSET *st[CHARTS], RRDDIM *rd[CHARTS][DIMS]
2021 same = (roundndd(value) == roundndd(expected)) ? 1 : 0;
2022 if(!same) {
2023 if(value_errors < 20)
2023 - fprintf(stderr, " DB-engine unittest %s/%s: at %lu secs, expecting value " NETDATA_DOUBLE_FORMAT
2024 + fprintf(stderr, " DB-engine unittest %s/%s: point #%ld, at %lu secs, expecting value " NETDATA_DOUBLE_FORMAT
2025 ", RRDR found " NETDATA_DOUBLE_FORMAT ", ### E R R O R ###\n",
2025 - rrdset_name(st[i]), rrddim_name(rd[i][j]), (unsigned long)time_now, expected, value);
2026 + rrdset_name(st[i]), rrddim_name(rd[i][j]), (long) c+1, (unsigned long)time_now, expected, value);
2027 value_errors++;
2028 }
2029 + else
2030 + value_right++;
2031 +
2032 if(time_retrieved != time_now) {
2033 if(time_errors < 20)
2030 - fprintf(stderr, " DB-engine unittest %s/%s: at %lu secs, found RRDR timestamp %lu ### E R R O R ###\n",
2031 - rrdset_name(st[i]), rrddim_name(rd[i][j]), (unsigned long)time_now, (unsigned long)time_retrieved);
2034 + fprintf(stderr, " DB-engine unittest %s/%s: point #%ld at %lu secs, found RRDR timestamp %lu ### E R R O R ###\n",
2035 + rrdset_name(st[i]), rrddim_name(rd[i][j]), (long)c+1, (unsigned long)time_now, (unsigned long)time_retrieved);
2036 time_errors++;
2037 }
2038 + else
2039 + time_right++;
2040 }
2041 rrddim_foreach_done(d);
2042 }
@@ -2040,10 +2046,10 @@ static int test_dbengine_check_rrdr(RRDSET *st[CHARTS], RRDDIM *rd[CHARTS][DIMS]
2046 }
2047
2048 if(value_errors)
2043 - fprintf(stderr, "%d value errors encountered\n", value_errors);
2049 + fprintf(stderr, "%d value errors encountered (%d were ok)\n", value_errors, value_right);
2050
2051 if(time_errors)
2046 - fprintf(stderr, "%d time errors encountered\n", time_errors);
2052 + fprintf(stderr, "%d time errors encountered (%d were ok)\n", time_errors, value_right);
2053
2054 return errors + value_errors + time_errors;
2055 }
@@ -2051,7 +2057,7 @@ static int test_dbengine_check_rrdr(RRDSET *st[CHARTS], RRDDIM *rd[CHARTS][DIMS]
2057 int test_dbengine(void)
2058 {
2059 fprintf(stderr, "%s() running...\n", __FUNCTION__ );
2054 - int i, j, errors, value_errors = 0, time_errors = 0, update_every, current_region;
2060 + int i, j, errors = 0, value_errors = 0, time_errors = 0, update_every, current_region;
2061 RRDHOST *host = NULL;
2062 RRDSET *st[CHARTS];
2063 RRDDIM *rd[CHARTS][DIMS];
@@ -2074,9 +2080,7 @@ int test_dbengine(void)
2080 time_start[current_region] = 2 * API_RELATIVE_TIME_MAX;
2081 time_end[current_region] = test_dbengine_create_metrics(st,rd, current_region, time_start[current_region]);
2082
2077 - errors = test_dbengine_check_metrics(st, rd, current_region, time_start[current_region]);
2078 - if (errors)
2079 - goto error_out;
2083 + errors += test_dbengine_check_metrics(st, rd, current_region, time_start[current_region]);
2084
2085 current_region = 1; //this is the second region of data
2086 update_every = REGION_UPDATE_EVERY[current_region]; // set data collection frequency to 3 seconds
@@ -2093,9 +2097,7 @@ int test_dbengine(void)
2097 time_start[current_region] += update_every - time_start[current_region] % update_every;
2098 time_end[current_region] = test_dbengine_create_metrics(st,rd, current_region, time_start[current_region]);
2099
2096 - errors = test_dbengine_check_metrics(st, rd, current_region, time_start[current_region]);
2097 - if (errors)
2098 - goto error_out;
2100 + errors += test_dbengine_check_metrics(st, rd, current_region, time_start[current_region]);
2101
2102 current_region = 2; //this is the third region of data
2103 update_every = REGION_UPDATE_EVERY[current_region]; // set data collection frequency to 1 seconds
@@ -2112,19 +2114,14 @@ int test_dbengine(void)
2114 time_start[current_region] += update_every - time_start[current_region] % update_every;
2115 time_end[current_region] = test_dbengine_create_metrics(st,rd, current_region, time_start[current_region]);
2116
2115 - errors = test_dbengine_check_metrics(st, rd, current_region, time_start[current_region]);
2116 - if (errors)
2117 - goto error_out;
2117 + errors += test_dbengine_check_metrics(st, rd, current_region, time_start[current_region]);
2118
2119 for (current_region = 0 ; current_region < REGIONS ; ++current_region) {
2120 - errors = test_dbengine_check_rrdr(st, rd, current_region, time_start[current_region], time_end[current_region]);
2121 - if (errors)
2122 - goto error_out;
2120 + errors += test_dbengine_check_rrdr(st, rd, current_region, time_start[current_region], time_end[current_region]);
2121 }
2122
2123 current_region = 1;
2124 update_every = REGION_UPDATE_EVERY[current_region]; // use the maximum update_every = 3
2127 - errors = 0;
2125 long points = (time_end[REGIONS - 1] - time_start[0]) / update_every; // cover all time regions with RRDR
2126 long point_offset = (time_start[current_region] - time_start[0]) / update_every;
2127 for (i = 0 ; i < CHARTS ; ++i) {
@@ -2181,7 +2178,7 @@ int test_dbengine(void)
2178 }
2179 onewayalloc_destroy(owa);
2180 }
2184 -error_out:
2181 +
2182 rrd_wrlock();
2183 rrdeng_prepare_exit((struct rrdengine_instance *)host->db[0].instance);
2184 rrdhost_delete_charts(host);
database/engine/metric.c
+24 -15
@@ -327,33 +327,42 @@ bool mrg_metric_set_first_time_s_if_bigger(MRG *mrg __maybe_unused, METRIC *metr
327 return ret;
328 }
329
330 -bool mrg_metric_set_first_time_s_if_zero(MRG *mrg __maybe_unused, METRIC *metric, time_t first_time_s) {
331 - bool ret = false;
330 +time_t mrg_metric_get_first_time_s(MRG *mrg __maybe_unused, METRIC *metric) {
331 + time_t first_time_s;
332
333 netdata_spinlock_lock(&metric->spinlock);
334 - if(!metric->first_time_s) {
335 - metric->first_time_s = first_time_s;
336 - ret = true;
334 +
335 + if(unlikely(!metric->first_time_s)) {
336 + if(metric->latest_time_s_clean)
337 + metric->first_time_s = metric->latest_time_s_clean;
338 +
339 + else if(metric->latest_time_s_hot)
340 + metric->first_time_s = metric->latest_time_s_hot;
341 }
342 +
343 + first_time_s = metric->first_time_s;
344 +
345 netdata_spinlock_unlock(&metric->spinlock);
346
340 - return ret;
347 + return first_time_s;
348 }
349
343 -time_t mrg_metric_get_first_time_s(MRG *mrg __maybe_unused, METRIC *metric) {
344 - time_t first_time_s;
350 +void mrg_metric_get_retention(MRG *mrg __maybe_unused, METRIC *metric, time_t *first_time_s, time_t *last_time_s, time_t *update_every_s) {
351 netdata_spinlock_lock(&metric->spinlock);
346 - first_time_s = metric->first_time_s;
347 - if(!first_time_s) {
352 +
353 + if(unlikely(!metric->first_time_s)) {
354 if(metric->latest_time_s_clean)
349 - first_time_s = metric->latest_time_s_clean;
355 + metric->first_time_s = metric->latest_time_s_clean;
356
351 - if(!first_time_s || metric->latest_time_s_hot < metric->latest_time_s_clean)
352 - first_time_s = metric->latest_time_s_hot;
357 + else if(metric->latest_time_s_hot)
358 + metric->first_time_s = metric->latest_time_s_hot;
359 }
354 - netdata_spinlock_unlock(&metric->spinlock);
360
356 - return first_time_s;
361 + *first_time_s = metric->first_time_s;
362 + *last_time_s = MAX(metric->latest_time_s_clean, metric->latest_time_s_hot);
363 + *update_every_s = metric->latest_update_every_s;
364 +
365 + netdata_spinlock_unlock(&metric->spinlock);
366 }
367
368 bool mrg_metric_set_clean_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric, time_t latest_time_s) {
database/engine/metric.h
+3 -3
@@ -46,18 +46,18 @@ Word_t mrg_metric_section(MRG *mrg, METRIC *metric);
46
47 bool mrg_metric_set_first_time_s(MRG *mrg, METRIC *metric, time_t first_time_s);
48 bool mrg_metric_set_first_time_s_if_bigger(MRG *mrg, METRIC *metric, time_t first_time_s);
49 -bool mrg_metric_set_first_time_s_if_zero(MRG *mrg, METRIC *metric, time_t first_time_s);
49 time_t mrg_metric_get_first_time_s(MRG *mrg, METRIC *metric);
51 -void mrg_metric_expand_retention(MRG *mrg __maybe_unused, METRIC *metric, time_t first_time_s, time_t last_time_s, time_t update_every_s);
50
51 bool mrg_metric_set_clean_latest_time_s(MRG *mrg, METRIC *metric, time_t latest_time_s);
52 bool mrg_metric_set_hot_latest_time_s(MRG *mrg, METRIC *metric, time_t latest_time_s);
53 time_t mrg_metric_get_latest_time_s(MRG *mrg, METRIC *metric);
54
55 bool mrg_metric_set_update_every(MRG *mrg, METRIC *metric, time_t update_every_s);
56 +bool mrg_metric_set_update_every_s_if_zero(MRG *mrg, METRIC *metric, time_t update_every_s);
57 time_t mrg_metric_get_update_every_s(MRG *mrg, METRIC *metric);
58
60 -bool mrg_metric_set_update_every_s_if_zero(MRG *mrg, METRIC *metric, time_t update_every_s);
59 +void mrg_metric_expand_retention(MRG *mrg, METRIC *metric, time_t first_time_s, time_t last_time_s, time_t update_every_s);
60 +void mrg_metric_get_retention(MRG *mrg, METRIC *metric, time_t *first_time_s, time_t *last_time_s, time_t *update_every_s);
61
62 bool mrg_metric_writer_acquire(MRG *mrg, METRIC *metric);
63 bool mrg_metric_writer_release(MRG *mrg, METRIC *metric);
database/engine/pagecache.c
+12 -2
@@ -356,17 +356,27 @@ static size_t get_page_list_from_pgc(PGC *cache, METRIC *metric, struct rrdengin
356 }
357
358 static void pgc_inject_gap(struct rrdengine_instance *ctx, METRIC *metric, time_t start_time_s, time_t end_time_s) {
359 +
360 + time_t db_first_time_s, db_last_time_s, db_update_every_s;
361 + mrg_metric_get_retention(main_mrg, metric, &db_first_time_s, &db_last_time_s, &db_update_every_s);
362 +
363 + if(is_page_in_time_range(start_time_s, end_time_s, db_first_time_s, db_last_time_s) != PAGE_IS_IN_RANGE)
364 + return;
365 +
366 PGC_ENTRY page_entry = {
367 .hot = false,
368 .section = (Word_t)ctx,
369 .metric_id = (Word_t)metric,
363 - .start_time_s = start_time_s,
364 - .end_time_s = end_time_s,
370 + .start_time_s = MAX(start_time_s, db_first_time_s),
371 + .end_time_s = MIN(end_time_s, db_last_time_s),
372 .update_every_s = 0,
373 .size = 0,
374 .data = DBENGINE_EMPTY_PAGE,
375 };
376
377 + if(page_entry.start_time_s >= page_entry.end_time_s)
378 + return;
379 +
380 PGC_PAGE *page = pgc_page_add_and_acquire(main_cache, page_entry, NULL);
381 pgc_page_release(main_cache, page);
382 }
database/engine/pdc.c
+3
@@ -606,6 +606,9 @@ void pdc_acquire(PDC *pdc) {
606 }
607
608 bool pdc_release_and_destroy_if_unreferenced(PDC *pdc, bool worker, bool router __maybe_unused) {
609 + if(unlikely(!pdc))
610 + return true;
611 +
612 netdata_spinlock_lock(&pdc->refcount_spinlock);
613
614 if(pdc->refcount <= 0)
database/engine/rrdengine.c
+9 -1
@@ -1620,9 +1620,17 @@ static void *ctx_shutdown_tp_worker(struct rrdengine_instance *ctx __maybe_unuse
1620 completion_wait_for(&ctx->quiesce.completion);
1621 completion_destroy(&ctx->quiesce.completion);
1622
1623 + bool logged = false;
1624 while(__atomic_load_n(&ctx->atomic.extents_currently_being_flushed, __ATOMIC_RELAXED) ||
1624 - __atomic_load_n(&ctx->atomic.inflight_queries, __ATOMIC_RELAXED))
1625 + __atomic_load_n(&ctx->atomic.inflight_queries, __ATOMIC_RELAXED)) {
1626 + if(!logged) {
1627 + logged = true;
1628 + info("DBENGINE: waiting for %zu inflight queries to finish to shutdown tier %d...",
1629 + __atomic_load_n(&ctx->atomic.inflight_queries, __ATOMIC_RELAXED),
1630 + (ctx->config.legacy) ? -1 : ctx->config.tier);
1631 + }
1632 sleep_usec(1 * USEC_PER_MS);
1633 + }
1634
1635 completion_mark_complete(completion);
1636
database/engine/rrdengineapi.c
+67 -31
@@ -266,16 +266,19 @@ STORAGE_COLLECT_HANDLE *rrdeng_store_metric_init(STORAGE_METRIC_HANDLE *db_metri
266 if(!is_1st_metric_writer)
267 __atomic_add_fetch(&ctx->atomic.collectors_running_duplicate, 1, __ATOMIC_RELAXED);
268
269 + mrg_metric_set_update_every(main_mrg, metric, update_every);
270 +
271 + handle->alignment = (struct pg_alignment *)smg;
272 + rrdeng_page_alignment_acquire(handle->alignment);
273 +
274 // this is important!
275 // if we don't set the page_end_time_ut during the first collection
276 // data collection may be able to go back in time and during the addition of new pages
277 // clean pages may be found matching ours!
273 - handle->page_end_time_ut = (usec_t)mrg_metric_get_latest_time_s(main_mrg, metric) * USEC_PER_SEC;
278
275 - mrg_metric_set_update_every(main_mrg, metric, update_every);
276 -
277 - handle->alignment = (struct pg_alignment *)smg;
278 - rrdeng_page_alignment_acquire(handle->alignment);
279 + time_t db_first_time_s, db_last_time_s, db_update_every_s;
280 + mrg_metric_get_retention(main_mrg, metric, &db_first_time_s, &db_last_time_s, &db_update_every_s);
281 + handle->page_end_time_ut = (usec_t)db_last_time_s * USEC_PER_SEC;
282
283 return (STORAGE_COLLECT_HANDLE *)handle;
284 }
@@ -382,11 +385,12 @@ static void rrdeng_store_metric_create_new_page(struct rrdeng_collect_handle *ha
385 error_limit(&erl,
386 #endif
387 "DBENGINE: metric '%s' new page from %ld to %ld, update every %ld, has a conflict in main cache "
385 - "with existing %s page from %ld to %ld, update every %ld - "
388 + "with existing %s%s page from %ld to %ld, update every %ld - "
389 "is it collected more than once?",
390 uuid,
391 page_entry.start_time_s, page_entry.end_time_s, (time_t)page_entry.update_every_s,
392 pgc_is_page_hot(page) ? "hot" : "not-hot",
393 + pgc_page_data(page) == DBENGINE_EMPTY_PAGE ? " gap" : "",
394 pgc_page_start_time_s(page), pgc_page_end_time_s(page), pgc_page_update_every_s(page)
395 );
396
@@ -580,12 +584,8 @@ static void store_metric_next_error_log(struct rrdeng_collect_handle *handle, us
584 collect_page_flags_to_buffer(wb, handle->page_flags);
585 }
586
583 -#ifdef NETDATA_INTERNAL_CHECKS
584 - internal_error(true,
585 -#else
587 error_limit_static_global_var(erl, 1, 0);
588 error_limit(&erl,
588 -#endif
589 "DBENGINE: metric '%s' collected point at %ld, %s last collection at %ld, "
590 "update every %ld, %s page from %ld to %ld, position %u (of %u), flags: %s",
591 uuid,
@@ -699,8 +699,8 @@ int rrdeng_store_metric_finalize(STORAGE_COLLECT_HANDLE *collection_handle) {
699 if((handle->options & RRDENG_1ST_METRIC_WRITER) && !mrg_metric_writer_release(main_mrg, handle->metric))
700 internal_fatal(true, "DBENGINE: metric is already released");
701
702 - time_t first_time_s = mrg_metric_get_first_time_s(main_mrg, handle->metric);
703 - time_t last_time_s = mrg_metric_get_latest_time_s(main_mrg, handle->metric);
702 + time_t first_time_s, last_time_s, update_every_s;
703 + mrg_metric_get_retention(main_mrg, handle->metric, &first_time_s, &last_time_s, &update_every_s);
704
705 mrg_metric_release(main_mrg, handle->metric);
706 freez(handle);
@@ -759,7 +759,11 @@ static void unregister_query_handle(struct rrdeng_query_handle *handle __maybe_u
759 * Gets a handle for loading metrics from the database.
760 * The handle must be released with rrdeng_load_metric_final().
761 */
762 -void rrdeng_load_metric_init(STORAGE_METRIC_HANDLE *db_metric_handle, struct storage_engine_query_handle *rrddim_handle, time_t start_time_s, time_t end_time_s, STORAGE_PRIORITY priority)
762 +void rrdeng_load_metric_init(STORAGE_METRIC_HANDLE *db_metric_handle,
763 + struct storage_engine_query_handle *rrddim_handle,
764 + time_t start_time_s,
765 + time_t end_time_s,
766 + STORAGE_PRIORITY priority)
767 {
768 usec_t started_ut = now_monotonic_usec();
769
@@ -769,8 +773,6 @@ void rrdeng_load_metric_init(STORAGE_METRIC_HANDLE *db_metric_handle, struct sto
773 struct rrdengine_instance *ctx = mrg_metric_ctx(metric);
774 struct rrdeng_query_handle *handle;
775
772 - mrg_metric_set_update_every_s_if_zero(main_mrg, metric, default_rrd_update_every);
773 -
776 handle = rrdeng_query_handle_get();
777 register_query_handle(handle);
778
@@ -781,23 +783,48 @@ void rrdeng_load_metric_init(STORAGE_METRIC_HANDLE *db_metric_handle, struct sto
783
784 handle->ctx = ctx;
785 handle->metric = metric;
784 - handle->start_time_s = start_time_s;
785 - handle->end_time_s = end_time_s;
786 handle->priority = priority;
787 - handle->now_s = start_time_s;
787
789 - handle->dt_s = mrg_metric_get_update_every_s(main_mrg, metric);
790 - if(!handle->dt_s)
791 - handle->dt_s = default_rrd_update_every;
788 + // IMPORTANT!
789 + // It is crucial not to exceed the db boundaries, because dbengine
790 + // now has gap caching, so when a gap is detected a negative page
791 + // is inserted into the main cache, to avoid scanning the journals
792 + // again for pages matching the gap.
793
793 - rrddim_handle->handle = (STORAGE_QUERY_HANDLE *)handle;
794 - rrddim_handle->start_time_s = start_time_s;
795 - rrddim_handle->end_time_s = end_time_s;
796 - rrddim_handle->priority = priority;
794 + time_t db_first_time_s, db_last_time_s, db_update_every_s;
795 + mrg_metric_get_retention(main_mrg, metric, &db_first_time_s, &db_last_time_s, &db_update_every_s);
796
798 - pg_cache_preload(handle);
797 + if(is_page_in_time_range(start_time_s, end_time_s, db_first_time_s, db_last_time_s) == PAGE_IS_IN_RANGE) {
798 + handle->start_time_s = MAX(start_time_s, db_first_time_s);
799 + handle->end_time_s = MIN(end_time_s, db_last_time_s);
800 + handle->now_s = handle->start_time_s;
801
800 - __atomic_add_fetch(&rrdeng_cache_efficiency_stats.query_time_init, now_monotonic_usec() - started_ut, __ATOMIC_RELAXED);
802 + handle->dt_s = db_update_every_s;
803 + if (!handle->dt_s) {
804 + handle->dt_s = default_rrd_update_every;
805 + mrg_metric_set_update_every_s_if_zero(main_mrg, metric, default_rrd_update_every);
806 + }
807 +
808 + rrddim_handle->handle = (STORAGE_QUERY_HANDLE *) handle;
809 + rrddim_handle->start_time_s = handle->start_time_s;
810 + rrddim_handle->end_time_s = handle->end_time_s;
811 + rrddim_handle->priority = priority;
812 +
813 + pg_cache_preload(handle);
814 +
815 + __atomic_add_fetch(&rrdeng_cache_efficiency_stats.query_time_init, now_monotonic_usec() - started_ut, __ATOMIC_RELAXED);
816 + }
817 + else {
818 + handle->start_time_s = start_time_s;
819 + handle->end_time_s = end_time_s;
820 + handle->now_s = start_time_s;
821 + handle->dt_s = db_update_every_s;
822 +
823 + rrddim_handle->handle = (STORAGE_QUERY_HANDLE *) handle;
824 + rrddim_handle->start_time_s = handle->start_time_s;
825 + rrddim_handle->end_time_s = 0;
826 + rrddim_handle->priority = priority;
827 + }
828 }
829
830 static bool rrdeng_load_page_next(struct storage_engine_query_handle *rrddim_handle, bool debug_this __maybe_unused) {
@@ -827,10 +854,19 @@ static bool rrdeng_load_page_next(struct storage_engine_query_handle *rrddim_han
854 unsigned position;
855 if(likely(handle->now_s >= page_start_time_s && handle->now_s <= page_end_time_s)) {
856
830 - if(unlikely(entries == 1 || page_start_time_s == page_end_time_s))
857 + if(unlikely(entries == 1 || page_start_time_s == page_end_time_s || !page_update_every_s)) {
858 position = 0;
832 - else
859 + handle->now_s = page_start_time_s;
860 + }
861 + else {
862 position = (handle->now_s - page_start_time_s) * (entries - 1) / (page_end_time_s - page_start_time_s);
863 + time_t point_end_time_s = page_start_time_s + position * page_update_every_s;
864 + if(point_end_time_s < handle->now_s && position + 1 < entries) {
865 + position++;
866 + point_end_time_s = page_start_time_s + position * page_update_every_s;
867 + }
868 + handle->now_s = point_end_time_s;
869 + }
870
871 internal_fatal(position >= entries, "DBENGINE: wrong page position calculation");
872 }
@@ -986,8 +1022,8 @@ bool rrdeng_metric_retention_by_uuid(STORAGE_INSTANCE *db_instance, uuid_t *dim_
1022 if (unlikely(!metric))
1023 return false;
1024
989 - *first_entry_s = mrg_metric_get_first_time_s(main_mrg, metric);
990 - *last_entry_s = mrg_metric_get_latest_time_s(main_mrg, metric);
1025 + time_t update_every_s;
1026 + mrg_metric_get_retention(main_mrg, metric, first_entry_s, last_entry_s, &update_every_s);
1027
1028 mrg_metric_release(main_mrg, metric);
1029
database/rrd.c
+1 -1
@@ -135,7 +135,7 @@ const char *rrdset_type_name(RRDSET_TYPE chart_type) {
135 // ----------------------------------------------------------------------------
136 // RRD - cache directory
137
138 -char *rrdset_cache_dir(RRDHOST *host, const char *id) {
138 +char *rrdhost_cache_dir_for_rrdset_alloc(RRDHOST *host, const char *id) {
139 char *ret = NULL;
140
141 char b[FILENAME_MAX + 1];
database/rrd.h
+3 -4
@@ -314,7 +314,7 @@ typedef struct storage_collect_handle STORAGE_COLLECT_HANDLE;
314 struct rrddim_tier {
315 STORAGE_POINT virtual_point;
316 size_t tier_grouping;
317 - time_t next_point_time_s;
317 + time_t next_point_end_time_s;
318 STORAGE_METRIC_HANDLE *db_metric_handle; // the metric handle inside the database
319 STORAGE_COLLECT_HANDLE *db_collection_handle; // the data collection handle
320 struct storage_engine_collect_ops *collect_ops;
@@ -905,9 +905,7 @@ typedef struct health {
905 time_t health_delay_up_to; // a timestamp to delay alarms processing up to
906 STRING *health_default_exec; // the full path of the alarms notifications program
907 STRING *health_default_recipient; // the default recipient for all alarms
908 - char *health_log_filename; // the alarms event log filename
908 size_t health_log_entries_written; // the number of alarm events written to the alarms event log
910 - FILE *health_log_fp; // the FILE pointer to the open alarms event log file
909 uint32_t health_default_warn_repeat_every; // the default value for the interval between repeating warning notifications
910 uint32_t health_default_crit_repeat_every; // the default value for the interval between repeating critical notifications
911 } HEALTH;
@@ -1340,7 +1338,8 @@ void rrdset_free(RRDSET *st);
1338
1339 #ifdef NETDATA_RRD_INTERNALS
1340
1343 -char *rrdset_cache_dir(RRDHOST *host, const char *id);
1341 +char *rrdhost_cache_dir_for_rrdset_alloc(RRDHOST *host, const char *id);
1342 +const char *rrdset_cache_dir(RRDSET *st);
1343
1344 void rrddim_free(RRDSET *st, RRDDIM *rd);
1345
database/rrdcontext.c
+19 -2
@@ -3446,6 +3446,8 @@ static void rrdcontext_post_process_updates(RRDCONTEXT *rc, bool force, RRD_FLAG
3446 if(worker_jobs)
3447 worker_is_busy(WORKER_JOB_PP_CONTEXT);
3448
3449 + size_t min_priority_collected = LONG_MAX;
3450 + size_t min_priority_not_collected = LONG_MAX;
3451 size_t min_priority = LONG_MAX;
3452 time_t min_first_time_t = LONG_MAX, max_last_time_t = 0;
3453 size_t instances_active = 0, instances_deleted = 0;
@@ -3482,8 +3484,16 @@ static void rrdcontext_post_process_updates(RRDCONTEXT *rc, bool force, RRD_FLAG
3484
3485 instances_active++;
3486
3485 - if (ri->priority >= RRDCONTEXT_MINIMUM_ALLOWED_PRIORITY && ri->priority < min_priority)
3486 - min_priority = ri->priority;
3487 + if (ri->priority >= RRDCONTEXT_MINIMUM_ALLOWED_PRIORITY) {
3488 + if(rrd_flag_check(ri, RRD_FLAG_COLLECTED)) {
3489 + if(ri->priority < min_priority_collected)
3490 + min_priority_collected = ri->priority;
3491 + }
3492 + else {
3493 + if(ri->priority < min_priority_not_collected)
3494 + min_priority_not_collected = ri->priority;
3495 + }
3496 + }
3497
3498 if (ri->first_time_s && ri->first_time_s < min_first_time_t)
3499 min_first_time_t = ri->first_time_s;
@@ -3492,6 +3502,13 @@ static void rrdcontext_post_process_updates(RRDCONTEXT *rc, bool force, RRD_FLAG
3502 max_last_time_t = ri->last_time_s;
3503 }
3504 dfe_done(ri);
3505 +
3506 + if(min_priority_collected != LONG_MAX)
3507 + // use the collected priority
3508 + min_priority = min_priority_collected;
3509 + else
3510 + // use the non-collected priority
3511 + min_priority = min_priority_not_collected;
3512 }
3513
3514 {
database/rrddim.c
+1 -1
@@ -686,7 +686,7 @@ bool rrddim_memory_load_or_create_map_save(RRDSET *st, RRDDIM *rd, RRD_MEMORY_MO
686 char filename[FILENAME_MAX + 1];
687 char fullfilename[FILENAME_MAX + 1];
688 rrdset_strncpyz_name(filename, rrddim_id(rd), FILENAME_MAX);
689 - snprintfz(fullfilename, FILENAME_MAX, "%s/%s.db", st->cache_dir, filename);
689 + snprintfz(fullfilename, FILENAME_MAX, "%s/%s.db", rrdset_cache_dir(st), filename);
690
691 rd_on_file = (struct rrddim_map_save_v019 *)netdata_mmap(
692 fullfilename, size, ((memory_mode == RRD_MEMORY_MODE_MAP) ? MAP_SHARED : MAP_PRIVATE), 1, false, NULL);
database/rrdhost.c
+15 -18
@@ -499,7 +499,6 @@ int is_legacy = 1;
499 ", health %s"
500 ", cache_dir '%s'"
501 ", varlib_dir '%s'"
502 - ", health_log '%s'"
502 ", alarms default handler '%s'"
503 ", alarms default recipient '%s'"
504 , rrdhost_hostname(host)
@@ -519,7 +518,6 @@ int is_legacy = 1;
518 , host->health.health_enabled?"enabled":"disabled"
519 , host->cache_dir
520 , host->varlib_dir
522 - , host->health.health_log_filename
521 , string2str(host->health.health_default_exec)
522 , string2str(host->health.health_default_recipient)
523 );
@@ -1085,7 +1083,7 @@ void rrdhost_free___while_having_rrd_wrlock(RRDHOST *host, bool force) {
1083 if(!host) return;
1084
1085 if (netdata_exit || force) {
1088 - info("Freeing all memory for host '%s'...", rrdhost_hostname(host));
1086 + info("RRD: 'host:%s' freeing memory...", rrdhost_hostname(host));
1087
1088 // ------------------------------------------------------------------------
1089 // first remove it from the indexes, so that it will not be discoverable
@@ -1146,7 +1144,7 @@ void rrdhost_free___while_having_rrd_wrlock(RRDHOST *host, bool force) {
1144 #endif
1145
1146 if (!netdata_exit && !force) {
1149 - info("Setting archive mode for host '%s'...", rrdhost_hostname(host));
1147 + info("RRD: 'host:%s' is now in archive mode...", rrdhost_hostname(host));
1148 rrdhost_flag_set(host, RRDHOST_FLAG_ARCHIVED | RRDHOST_FLAG_ORPHAN);
1149 return;
1150 }
@@ -1187,7 +1185,6 @@ void rrdhost_free___while_having_rrd_wrlock(RRDHOST *host, bool force) {
1185 rrdpush_destinations_free(host);
1186 string_freez(host->health.health_default_exec);
1187 string_freez(host->health.health_default_recipient);
1190 - freez(host->health.health_log_filename);
1188 string_freez(host->registry_hostname);
1189 simple_pattern_free(host->rrdpush_send_charts_matching);
1190 netdata_rwlock_destroy(&host->health_log.alarm_log_rwlock);
@@ -1236,7 +1233,7 @@ void rrd_finalize_collection_for_all_hosts(void) {
1233 void rrdhost_save_charts(RRDHOST *host) {
1234 if(!host) return;
1235
1239 - info("Saving/Closing database of host '%s'...", rrdhost_hostname(host));
1236 + info("RRD: 'host:%s' saving / closing database...", rrdhost_hostname(host));
1237
1238 RRDSET *st;
1239
@@ -1393,13 +1390,11 @@ void reload_host_labels(void) {
1390
1391 rrdhost_flag_set(localhost,RRDHOST_FLAG_METADATA_LABELS | RRDHOST_FLAG_METADATA_UPDATE);
1392
1396 - health_label_log_save(localhost);
1397 -
1393 rrdpush_send_host_labels(localhost);
1394 }
1395
1396 void rrdhost_finalize_collection(RRDHOST *host) {
1402 - info("Stopping data collection for host '%s'...", rrdhost_hostname(host));
1397 + info("RRD: 'host:%s' stopping data collection...", rrdhost_hostname(host));
1398
1399 RRDSET *st;
1400 rrdset_foreach_write(st, host)
@@ -1413,16 +1408,18 @@ void rrdhost_finalize_collection(RRDHOST *host) {
1408 void rrdhost_delete_charts(RRDHOST *host) {
1409 if(!host) return;
1410
1416 - info("Deleting database of host '%s'...", rrdhost_hostname(host));
1411 + info("RRD: 'host:%s' deleting disk files...", rrdhost_hostname(host));
1412
1413 RRDSET *st;
1414
1420 - // we get a write lock
1421 - // to ensure only one thread is saving the database
1422 - rrdset_foreach_write(st, host) {
1423 - rrdset_delete_files(st);
1415 + if(host->rrd_memory_mode == RRD_MEMORY_MODE_SAVE || host->rrd_memory_mode == RRD_MEMORY_MODE_MAP) {
1416 + // we get a write lock
1417 + // to ensure only one thread is saving the database
1418 + rrdset_foreach_write(st, host){
1419 + rrdset_delete_files(st);
1420 + }
1421 + rrdset_foreach_done(st);
1422 }
1425 - rrdset_foreach_done(st);
1423
1424 recursively_delete_dir(host->cache_dir, "left over host");
1425 }
@@ -1433,7 +1430,7 @@ void rrdhost_delete_charts(RRDHOST *host) {
1430 void rrdhost_cleanup_charts(RRDHOST *host) {
1431 if(!host) return;
1432
1436 - info("Cleaning up database of host '%s'...", rrdhost_hostname(host));
1433 + info("RRD: 'host:%s' cleaning up disk files...", rrdhost_hostname(host));
1434
1435 RRDSET *st;
1436 uint32_t rrdhost_delete_obsolete_charts = rrdhost_option_check(host, RRDHOST_OPTION_DELETE_OBSOLETE_CHARTS);
@@ -1460,7 +1457,7 @@ void rrdhost_cleanup_charts(RRDHOST *host) {
1457 // RRDHOST - save all hosts to disk
1458
1459 void rrdhost_save_all(void) {
1463 - info("Saving database [%zu hosts(s)]...", rrdhost_hosts_available());
1460 + info("RRD: saving databases [%zu hosts(s)]...", rrdhost_hosts_available());
1461
1462 rrd_rdlock();
1463
@@ -1475,7 +1472,7 @@ void rrdhost_save_all(void) {
1472 // RRDHOST - save or delete all hosts from disk
1473
1474 void rrdhost_cleanup_all(void) {
1478 - info("Cleaning up database [%zu hosts(s)]...", rrdhost_hosts_available());
1475 + info("RRD: cleaning up database [%zu hosts(s)]...", rrdhost_hosts_available());
1476
1477 rrd_rdlock();
1478
database/rrdset.c
+23 -12
@@ -128,7 +128,6 @@ static void rrdset_insert_callback(const DICTIONARY_ITEM *item __maybe_unused, v
128 st->module_name = rrd_string_strdupz(ctr->module);
129 st->priority = ctr->priority;
130
131 - st->cache_dir = rrdset_cache_dir(host, chart_full_id);
131 st->entries = (ctr->memory_mode != RRD_MEMORY_MODE_DBENGINE) ? align_entries_to_pagesize(ctr->memory_mode, ctr->history_entries) : 5;
132 st->update_every = ctr->update_every;
133 st->rrd_memory_mode = ctr->memory_mode;
@@ -601,13 +600,15 @@ void rrdset_get_retention_of_tier_for_collected_chart(RRDSET *st, time_t *first_
600 if(unlikely(!db_last_entry_s)) {
601 db_last_entry_s = rrdset_last_entry_s_of_tier(st, tier);
602
604 - if (unlikely(!db_last_entry_s))
603 + if (unlikely(!db_last_entry_s)) {
604 // we assume this is a collected RRDSET
606 - db_last_entry_s = now_s;
605 + db_first_entry_s = 0;
606 + db_last_entry_s = 0;
607 + }
608 }
609
610 if(unlikely(db_last_entry_s > now_s)) {
610 - internal_error(true,
611 + internal_error(db_last_entry_s > now_s + 1,
612 "RRDSET: 'host:%s/chart:%s' latest db time %ld is in the future, adjusting it to now %ld",
613 rrdhost_hostname(st->rrdhost), rrdset_id(st),
614 db_last_entry_s, now_s);
@@ -831,7 +832,8 @@ void rrdset_delete_files(RRDSET *st) {
832 }
833 rrddim_foreach_done(rd);
834
834 - recursively_delete_dir(st->cache_dir, "left-over chart");
835 + if(st->cache_dir)
836 + recursively_delete_dir(st->cache_dir, "left-over chart");
837 }
838
839 void rrdset_delete_obsolete_dimensions(RRDSET *st) {
@@ -1105,15 +1107,17 @@ static inline time_t tier_next_point_time_s(RRDDIM *rd, struct rrddim_tier *t, t
1107 }
1108
1109 void store_metric_at_tier(RRDDIM *rd, size_t tier, struct rrddim_tier *t, STORAGE_POINT sp, usec_t now_ut __maybe_unused) {
1108 - if (unlikely(!t->next_point_time_s))
1109 - t->next_point_time_s = tier_next_point_time_s(rd, t, sp.end_time_s);
1110 + if (unlikely(!t->next_point_end_time_s))
1111 + t->next_point_end_time_s = tier_next_point_time_s(rd, t, sp.end_time_s);
1112 +
1113 + if(unlikely(sp.start_time_s >= t->next_point_end_time_s)) {
1114 + // flush the virtual point, it is done
1115
1111 - if(unlikely(sp.start_time_s > t->next_point_time_s)) {
1116 if (likely(!storage_point_is_unset(t->virtual_point))) {
1117
1118 t->collect_ops->store_metric(
1119 t->db_collection_handle,
1116 - t->next_point_time_s * USEC_PER_SEC,
1120 + t->next_point_end_time_s * USEC_PER_SEC,
1121 t->virtual_point.sum,
1122 t->virtual_point.min,
1123 t->virtual_point.max,
@@ -1124,7 +1128,7 @@ void store_metric_at_tier(RRDDIM *rd, size_t tier, struct rrddim_tier *t, STORAG
1128 else {
1129 t->collect_ops->store_metric(
1130 t->db_collection_handle,
1127 - t->next_point_time_s * USEC_PER_SEC,
1131 + t->next_point_end_time_s * USEC_PER_SEC,
1132 NAN,
1133 NAN,
1134 NAN,
@@ -1134,7 +1138,7 @@ void store_metric_at_tier(RRDDIM *rd, size_t tier, struct rrddim_tier *t, STORAG
1138
1139 rrdset_done_statistics_points_stored_per_tier[tier]++;
1140 t->virtual_point.count = 0; // make the point unset
1137 - t->next_point_time_s = tier_next_point_time_s(rd, t, sp.end_time_s);
1141 + t->next_point_end_time_s = tier_next_point_time_s(rd, t, sp.end_time_s);
1142 }
1143
1144 // merge the dates into our virtual point
@@ -2073,6 +2077,13 @@ const char *rrdset_cache_filename(RRDSET *st) {
2077 return st_on_file->cache_filename;
2078 }
2079
2080 +const char *rrdset_cache_dir(RRDSET *st) {
2081 + if(!st->cache_dir)
2082 + st->cache_dir = rrdhost_cache_dir_for_rrdset_alloc(st->rrdhost, rrdset_id(st));
2083 +
2084 + return st->cache_dir;
2085 +}
2086 +
2087 void rrdset_memory_file_free(RRDSET *st) {
2088 if(!st->st_on_file) return;
2089
@@ -2103,7 +2114,7 @@ bool rrdset_memory_load_or_create_map_save(RRDSET *st, RRD_MEMORY_MODE memory_mo
2114 return false;
2115
2116 char fullfilename[FILENAME_MAX + 1];
2106 - snprintfz(fullfilename, FILENAME_MAX, "%s/main.db", st->cache_dir);
2117 + snprintfz(fullfilename, FILENAME_MAX, "%s/main.db", rrdset_cache_dir(st));
2118
2119 unsigned long size = sizeof(struct rrdset_map_save_v019);
2120 struct rrdset_map_save_v019 *st_on_file = (struct rrdset_map_save_v019 *)netdata_mmap(
database/sqlite/sqlite_functions.c
-110
@@ -529,116 +529,6 @@ void db_execute(const char *cmd)
529 }
530 }
531
532 -#define SELECT_MIGRATED_FILE "select 1 from metadata_migration where filename = @path;"
533 -
534 -int file_is_migrated(char *path)
535 -{
536 - sqlite3_stmt *res = NULL;
537 - int rc;
538 -
539 - rc = sqlite3_prepare_v2(db_meta, SELECT_MIGRATED_FILE, -1, &res, 0);
540 - if (unlikely(rc != SQLITE_OK)) {
541 - error_report("Failed to prepare statement to fetch host");
542 - return 0;
543 - }
544 -
545 - rc = sqlite3_bind_text(res, 1, path, -1, SQLITE_STATIC);
546 - if (unlikely(rc != SQLITE_OK)) {
547 - error_report("Failed to bind filename parameter to check migration");
548 - return 0;
549 - }
550 -
551 - rc = sqlite3_step_monitored(res);
552 -
553 - if (unlikely(sqlite3_finalize(res) != SQLITE_OK))
554 - error_report("Failed to finalize the prepared statement when checking if metadata file is migrated");
555 -
556 - return (rc == SQLITE_ROW);
557 -}
558 -
559 -#define STORE_MIGRATED_FILE "insert or replace into metadata_migration (filename, file_size, date_created) " \
560 - "values (@file, @size, unixepoch());"
561 -
562 -void add_migrated_file(char *path, uint64_t file_size)
563 -{
564 - sqlite3_stmt *res = NULL;
565 - int rc;
566 -
567 - rc = sqlite3_prepare_v2(db_meta, STORE_MIGRATED_FILE, -1, &res, 0);
568 - if (unlikely(rc != SQLITE_OK)) {
569 - error_report("Failed to prepare statement to fetch host");
570 - return;
571 - }
572 -
573 - rc = sqlite3_bind_text(res, 1, path, -1, SQLITE_STATIC);
574 - if (unlikely(rc != SQLITE_OK)) {
575 - error_report("Failed to bind filename parameter to store migration information");
576 - return;
577 - }
578 -
579 - rc = sqlite3_bind_int64(res, 2, (sqlite_int64) file_size);
580 - if (unlikely(rc != SQLITE_OK)) {
581 - error_report("Failed to bind size parameter to store migration information");
582 - return;
583 - }
584 -
585 - rc = execute_insert(res);
586 - if (unlikely(rc != SQLITE_DONE))
587 - error_report("Failed to store migrated file, rc = %d", rc);
588 -
589 - if (unlikely(sqlite3_finalize(res) != SQLITE_OK))
590 - error_report("Failed to finalize the prepared statement when checking if metadata file is migrated");
591 -}
592 -
593 -
594 -
595 -#define SQL_STORE_CLAIM_ID "insert into node_instance " \
596 - "(host_id, claim_id, date_created) values (@host_id, @claim_id, unixepoch()) " \
597 - "on conflict(host_id) do update set claim_id = excluded.claim_id;"
598 -
599 -void store_claim_id(uuid_t *host_id, uuid_t *claim_id)
600 -{
601 - sqlite3_stmt *res = NULL;
602 - int rc;
603 -
604 - if (unlikely(!db_meta)) {
605 - if (default_rrd_memory_mode == RRD_MEMORY_MODE_DBENGINE)
606 - error_report("Database has not been initialized");
607 - return;
608 - }
609 -
610 - rc = sqlite3_prepare_v2(db_meta, SQL_STORE_CLAIM_ID, -1, &res, 0);
611 - if (unlikely(rc != SQLITE_OK)) {
612 - error_report("Failed to prepare statement store chart labels");
613 - return;
614 - }
615 -
616 - rc = sqlite3_bind_blob(res, 1, host_id, sizeof(*host_id), SQLITE_STATIC);
617 - if (unlikely(rc != SQLITE_OK)) {
618 - error_report("Failed to bind host_id parameter to store node instance information");
619 - goto failed;
620 - }
621 -
622 - if (claim_id)
623 - rc = sqlite3_bind_blob(res, 2, claim_id, sizeof(*claim_id), SQLITE_STATIC);
624 - else
625 - rc = sqlite3_bind_null(res, 2);
626 - if (unlikely(rc != SQLITE_OK)) {
627 - error_report("Failed to bind claim_id parameter to store node instance information");
628 - goto failed;
629 - }
630 -
631 - rc = execute_insert(res);
632 - if (unlikely(rc != SQLITE_DONE))
633 - error_report("Failed to store node instance information, rc = %d", rc);
634 -
635 -failed:
636 - if (unlikely(sqlite3_finalize(res) != SQLITE_OK))
637 - error_report("Failed to finalize the prepared statement when storing node instance information");
638 -
639 - return;
640 -}
641 -
532 static inline void set_host_node_id(RRDHOST *host, uuid_t *node_id)
533 {
534 if (unlikely(!host))
database/sqlite/sqlite_functions.h
-2
@@ -54,9 +54,7 @@ void sql_close_database(void);
54 int bind_text_null(sqlite3_stmt *res, int position, const char *text, bool can_be_null);
55 int prepare_statement(sqlite3 *database, const char *query, sqlite3_stmt **statement);
56 int execute_insert(sqlite3_stmt *res);
57 -int file_is_migrated(char *path);
57 int exec_statement_with_uuid(const char *sql, uuid_t *uuid);
59 -void add_migrated_file(char *path, uint64_t file_size);
58 void db_execute(const char *cmd);
59
60 // Look up functions
database/sqlite/sqlite_health.c
-6
@@ -107,8 +107,6 @@ void sql_health_alarm_log_update(RRDHOST *host, ALARM_ENTRY *ae) {
107 failed:
108 if (unlikely(sqlite3_finalize(res) != SQLITE_OK))
109 error_report("HEALTH [%s]: Failed to finalize the prepared statement for updating health log.", rrdhost_hostname(host));
110 -
111 - return;
110 }
111
112 /* Health related SQL queries
@@ -350,8 +348,6 @@ void sql_health_alarm_log_insert(RRDHOST *host, ALARM_ENTRY *ae) {
348 failed:
349 if (unlikely(sqlite3_finalize(res) != SQLITE_OK))
350 error_report("HEALTH [%s]: Failed to finalize the prepared statement for inserting to health log.", rrdhost_hostname(host));
353 -
354 - return;
351 }
352
353 void sql_health_alarm_log_save(RRDHOST *host, ALARM_ENTRY *ae)
@@ -545,8 +541,6 @@ void sql_inject_removed_status(char *uuid_str, uint32_t alarm_id, uint32_t alarm
541 failed:
542 if (unlikely(sqlite3_finalize(res) != SQLITE_OK))
543 error_report("HEALTH [N/A]: Failed to finalize the prepared statement for injecting removed event.");
548 - return;
549 -
544 }
545
546 #define SQL_SELECT_MAX_UNIQUE_ID(guid) "SELECT MAX(unique_id) from health_log_%s", guid
database/sqlite/sqlite_metadata.c
+10 -1
@@ -972,6 +972,9 @@ static void start_metadata_hosts(uv_work_t *req __maybe_unused)
972
973 bool run_again = false;
974 worker_is_busy(UV_EVENT_METADATA_STORE);
975 +
976 + if (!data->max_count)
977 + db_execute("BEGIN TRANSACTION;");
978 dfe_start_reentrant(rrdhost_root_index, host) {
979 if (rrdhost_flag_check(host, RRDHOST_FLAG_ARCHIVED) || !rrdhost_flag_check(host, RRDHOST_FLAG_METADATA_UPDATE))
980 continue;
@@ -1024,11 +1027,15 @@ static void start_metadata_hosts(uv_work_t *req __maybe_unused)
1027 query_counter++;
1028 }
1029
1030 + if (data->max_count)
1031 + db_execute("BEGIN TRANSACTION;");
1032 if (unlikely(metadata_scan_host(host, data->max_count, &query_counter))) {
1033 run_again = true;
1034 rrdhost_flag_set(host,RRDHOST_FLAG_METADATA_UPDATE);
1035 internal_error(true,"METADATA: 'host:%s': scheduling another run, more charts to store", rrdhost_hostname(host));
1036 }
1037 + if (data->max_count)
1038 + db_execute("COMMIT TRANSACTION;");
1039
1040 usec_t ended_ut = now_monotonic_usec(); (void)ended_ut;
1041 internal_error(true, "METADATA: 'host:%s': saved metadata with %zu SQL statements, in %0.2f ms",
@@ -1036,6 +1043,8 @@ static void start_metadata_hosts(uv_work_t *req __maybe_unused)
1043 (double)(ended_ut - started_ut) / USEC_PER_MS);
1044 }
1045 dfe_done(host);
1046 + if (!data->max_count)
1047 + db_execute("COMMIT TRANSACTION;");
1048
1049 usec_t all_ended_ut = now_monotonic_usec(); (void)all_ended_ut;
1050 internal_error(true, "METADATA: checking all hosts completed in %0.2f ms",
@@ -1170,7 +1179,7 @@ static void metadata_event_loop(void *arg)
1179 cmd.completion = NULL; // Do not complete after launching worker (worker will do)
1180 }
1181 else
1173 - data->max_count = 1000;
1182 + data->max_count = 5000;
1183
1184 metadata_flag_set(wc, METADATA_FLAG_SCANNING_HOSTS);
1185 if (unlikely(
health/health.c
+4 -19
@@ -797,30 +797,15 @@ static void initialize_health(RRDHOST *host, int is_localhost) {
797 if(r != 0 && errno != EEXIST)
798 error("Host '%s': cannot create directory '%s'", rrdhost_hostname(host), filename);
799 }
800 - snprintfz(filename, FILENAME_MAX, "%s/health/health-log.db", host->varlib_dir);
801 - host->health.health_log_filename = strdupz(filename);
800
801 snprintfz(filename, FILENAME_MAX, "%s/alarm-notify.sh", netdata_configured_primary_plugins_dir);
802 host->health.health_default_exec = string_strdupz(config_get(CONFIG_SECTION_HEALTH, "script to execute on alarm", filename));
803 host->health.health_default_recipient = string_strdupz("root");
804
807 - if (!file_is_migrated(host->health.health_log_filename)) {
808 - int rc = sql_create_health_log_table(host);
809 - if (unlikely(rc)) {
810 - log_health("[%s]: Failed to create health log table in the database", rrdhost_hostname(host));
811 - health_alarm_log_load(host);
812 - health_alarm_log_open(host);
813 - }
814 - else {
815 - health_alarm_log_load(host);
816 - add_migrated_file(host->health.health_log_filename, 0);
817 - }
818 - } else {
819 - // TODO: This needs to go to the metadata thread
820 - // Health should wait before accessing the table (needs to be created by the metadata thread)
821 - sql_create_health_log_table(host);
822 - sql_health_alarm_log_load(host);
823 - }
805 + // TODO: This needs to go to the metadata thread
806 + // Health should wait before accessing the table (needs to be created by the metadata thread)
807 + sql_create_health_log_table(host);
808 + sql_health_alarm_log_load(host);
809
810 // ------------------------------------------------------------------------
811 // load health configuration
health/health_log.c
+1 -427
@@ -3,149 +3,10 @@
3 #include "health.h"
4
5 // ----------------------------------------------------------------------------
6 -// health alarm log load/save
7 -// no need for locking - only one thread is reading / writing the alarms log
8 -
9 -inline int health_alarm_log_open(RRDHOST *host) {
10 - if(host->health.health_log_fp)
11 - fclose(host->health.health_log_fp);
12 -
13 - host->health.health_log_fp = fopen(host->health.health_log_filename, "a");
14 -
15 - if(host->health.health_log_fp) {
16 - if (setvbuf(host->health.health_log_fp, NULL, _IOLBF, 0) != 0)
17 - error("HEALTH [%s]: cannot set line buffering on health log file '%s'.", rrdhost_hostname(host), host->health.health_log_filename);
18 - return 0;
19 - }
20 -
21 - error("HEALTH [%s]: cannot open health log file '%s'. Health data will be lost in case of netdata or server crash.", rrdhost_hostname(host), host->health.health_log_filename);
22 - return -1;
23 -}
24 -
25 -static inline void health_alarm_log_close(RRDHOST *host) {
26 - if(host->health.health_log_fp) {
27 - fclose(host->health.health_log_fp);
28 - host->health.health_log_fp = NULL;
29 - }
30 -}
31 -
32 -static inline void health_log_rotate(RRDHOST *host) {
33 - static size_t rotate_every = 0;
34 -
35 - if(unlikely(rotate_every == 0)) {
36 - rotate_every = (size_t)config_get_number(CONFIG_SECTION_HEALTH, "rotate log every lines", 2000);
37 - if(rotate_every < 100) rotate_every = 100;
38 - }
39 -
40 - if(unlikely(host->health.health_log_entries_written > rotate_every)) {
41 - if(unlikely(host->health.health_log_fp)) {
42 - health_alarm_log_close(host);
43 -
44 - char old_filename[FILENAME_MAX + 1];
45 - snprintfz(old_filename, FILENAME_MAX, "%s.old", host->health.health_log_filename);
46 -
47 - if(unlink(old_filename) == -1 && errno != ENOENT)
48 - error("HEALTH [%s]: cannot remove old alarms log file '%s'", rrdhost_hostname(host), old_filename);
49 -
50 - if(link(host->health.health_log_filename, old_filename) == -1 && errno != ENOENT)
51 - error("HEALTH [%s]: cannot move file '%s' to '%s'.", rrdhost_hostname(host), host->health.health_log_filename, old_filename);
52 -
53 - if(unlink(host->health.health_log_filename) == -1 && errno != ENOENT)
54 - error("HEALTH [%s]: cannot remove old alarms log file '%s'", rrdhost_hostname(host), host->health.health_log_filename);
55 -
56 - // open it with truncate
57 - host->health.health_log_fp = fopen(host->health.health_log_filename, "w");
58 -
59 - if(host->health.health_log_fp)
60 - fclose(host->health.health_log_fp);
61 - else
62 - error("HEALTH [%s]: cannot truncate health log '%s'", rrdhost_hostname(host), host->health.health_log_filename);
63 -
64 - host->health.health_log_fp = NULL;
65 -
66 - host->health.health_log_entries_written = 0;
67 - health_alarm_log_open(host);
68 - }
69 - }
70 -}
71 -
72 -inline void health_label_log_save(RRDHOST *host) {
73 - health_log_rotate(host);
74 -
75 - if(unlikely(host->health.health_log_fp)) {
76 - BUFFER *wb = buffer_create(1024, &netdata_buffers_statistics.buffers_health);
77 -
78 - rrdlabels_to_buffer(localhost->rrdlabels, wb, "", "=", "", "\t ", NULL, NULL, NULL, NULL);
79 - char *write = (char *) buffer_tostring(wb);
80 -
81 - if (unlikely(fprintf(host->health.health_log_fp, "L\t%s", write) < 0))
82 - error("HEALTH [%s]: failed to save alarm log entry to '%s'. Health data may be lost in case of abnormal restart.",
83 - rrdhost_hostname(host), host->health.health_log_filename);
84 - else
85 - host->health.health_log_entries_written++;
86 -
87 - buffer_free(wb);
88 - }
89 -}
6
7 inline void health_alarm_log_save(RRDHOST *host, ALARM_ENTRY *ae) {
92 - health_log_rotate(host);
93 - if(unlikely(host->health.health_log_fp)) {
94 - if(unlikely(fprintf(host->health.health_log_fp
95 - , "%c\t%s"
96 - "\t%08x\t%08x\t%08x\t%08x\t%08x"
97 - "\t%08x\t%08x\t%08x"
98 - "\t%08x\t%08x\t%08x"
99 - "\t%s\t%s\t%s\t%s\t%s\t%s\t%s\t%s"
100 - "\t%d\t%d\t%d\t%d"
101 - "\t" NETDATA_DOUBLE_FORMAT_AUTO "\t" NETDATA_DOUBLE_FORMAT_AUTO
102 - "\t%016"PRIx64""
103 - "\t%s\t%s\t%s"
104 - "\n"
105 - , (ae->flags & HEALTH_ENTRY_FLAG_SAVED)?'U':'A'
106 - , rrdhost_hostname(host)
107 -
108 - , ae->unique_id
109 - , ae->alarm_id
110 - , ae->alarm_event_id
111 - , ae->updated_by_id
112 - , ae->updates_id
113 -
114 - , (uint32_t)ae->when
115 - , (uint32_t)ae->duration
116 - , (uint32_t)ae->non_clear_duration
117 - , (uint32_t)ae->flags
118 - , (uint32_t)ae->exec_run_timestamp
119 - , (uint32_t)ae->delay_up_to_timestamp
120 -
121 - , ae_name(ae)
122 - , ae_chart_name(ae)
123 - , ae_family(ae)
124 - , ae_exec(ae)
125 - , ae_recipient(ae)
126 - , ae_source(ae)
127 - , ae_units(ae)
128 - , ae_info(ae)
8
130 - , ae->exec_code
131 - , ae->new_status
132 - , ae->old_status
133 - , ae->delay
134 -
135 - , ae->new_value
136 - , ae->old_value
137 - , (uint64_t)ae->last_repeat
138 - , (ae->classification)?ae_classification(ae):"Unknown"
139 - , (ae->component)?ae_component(ae):"Unknown"
140 - , (ae->type)?ae_type(ae):"Unknown"
141 - ) < 0))
142 - error("HEALTH [%s]: failed to save alarm log entry to '%s'. Health data may be lost in case of abnormal restart.", rrdhost_hostname(host), host->health.health_log_filename);
143 - else {
144 - ae->flags |= HEALTH_ENTRY_FLAG_SAVED;
145 - host->health.health_log_entries_written++;
146 - }
147 - }else
148 - sql_health_alarm_log_save(host, ae);
9 + sql_health_alarm_log_save(host, ae);
10
11 #ifdef ENABLE_ACLK
12 if (netdata_cloud_setting) {
@@ -154,293 +15,6 @@ inline void health_alarm_log_save(RRDHOST *host, ALARM_ENTRY *ae) {
15 #endif
16 }
17
157 -static uint32_t is_valid_alarm_id(RRDHOST *host, const char *chart, const char *name, uint32_t alarm_id)
158 -{
159 - STRING *chart_string = string_strdupz(chart);
160 - STRING *name_string = string_strdupz(name);
161 -
162 - uint32_t ret = 1;
163 -
164 - ALARM_ENTRY *ae;
165 - for(ae = host->health_log.alarms; ae ;ae = ae->next) {
166 - if (unlikely(ae->alarm_id == alarm_id && (!(chart_string == ae->chart && name_string == ae->name)))) {
167 - ret = 0;
168 - break;
169 - }
170 - }
171 -
172 - string_freez(chart_string);
173 - string_freez(name_string);
174 -
175 - return ret;
176 -}
177 -
178 -static inline ssize_t health_alarm_log_read(RRDHOST *host, FILE *fp, const char *filename) {
179 - errno = 0;
180 -
181 - char *s, *buf = mallocz(65536 + 1);
182 - size_t line = 0, len = 0;
183 - ssize_t loaded = 0, updated = 0, errored = 0, duplicate = 0;
184 -
185 - DICTIONARY *all_rrdcalcs = dictionary_create_advanced(
186 - DICT_OPTION_NAME_LINK_DONT_CLONE | DICT_OPTION_VALUE_LINK_DONT_CLONE | DICT_OPTION_DONT_OVERWRITE_VALUE,
187 - &dictionary_stats_category_rrdhealth);
188 -
189 - RRDCALC *rc;
190 - foreach_rrdcalc_in_rrdhost_read(host, rc) {
191 - dictionary_set(all_rrdcalcs, rrdcalc_name(rc), rc, sizeof(*rc));
192 - }
193 - foreach_rrdcalc_in_rrdhost_done(rc);
194 -
195 - netdata_rwlock_rdlock(&host->health_log.alarm_log_rwlock);
196 -
197 - while((s = fgets_trim_len(buf, 65536, fp, &len))) {
198 - host->health.health_log_entries_written++;
199 - line++;
200 -
201 - int max_entries = 33, entries = 0;
202 - char *pointers[max_entries];
203 -
204 - pointers[entries++] = s++;
205 - while(*s) {
206 - if(unlikely(*s == '\t')) {
207 - *s = '\0';
208 - pointers[entries++] = ++s;
209 - if(entries >= max_entries) {
210 - error("HEALTH [%s]: line %zu of file '%s' has more than %d entries. Ignoring excessive entries.", rrdhost_hostname(host), line, filename, max_entries);
211 - break;
212 - }
213 - }
214 - else s++;
215 - }
216 -
217 - if(likely(*pointers[0] == 'L'))
218 - continue;
219 -
220 - if(likely(*pointers[0] == 'U' || *pointers[0] == 'A')) {
221 - ALARM_ENTRY *ae = NULL;
222 -
223 - if(entries < 27) {
224 - error("HEALTH [%s]: line %zu of file '%s' should have at least 27 entries, but it has %d. Ignoring it.", rrdhost_hostname(host), line, filename, entries);
225 - errored++;
226 - continue;
227 - }
228 -
229 - // check that we have valid ids
230 - uint32_t unique_id = (uint32_t)strtoul(pointers[2], NULL, 16);
231 - if(!unique_id) {
232 - error("HEALTH [%s]: line %zu of file '%s' states alarm entry with invalid unique id %u (%s). Ignoring it.", rrdhost_hostname(host), line, filename, unique_id, pointers[2]);
233 - errored++;
234 - continue;
235 - }
236 -
237 - uint32_t alarm_id = (uint32_t)strtoul(pointers[3], NULL, 16);
238 - if(!alarm_id) {
239 - error("HEALTH [%s]: line %zu of file '%s' states alarm entry for invalid alarm id %u (%s). Ignoring it.", rrdhost_hostname(host), line, filename, alarm_id, pointers[3]);
240 - errored++;
241 - continue;
242 - }
243 -
244 - // Check if we got last_repeat field
245 - time_t last_repeat = 0;
246 - if(entries > 27) {
247 - char* alarm_name = pointers[13];
248 - last_repeat = (time_t)strtoul(pointers[27], NULL, 16);
249 -
250 - rc = dictionary_get(all_rrdcalcs, alarm_name);
251 - if(unlikely(rc)) {
252 - if (rrdcalc_isrepeating(rc)) {
253 - rc->last_repeat = last_repeat;
254 - // We iterate through repeating alarm entries only to
255 - // find the latest last_repeat timestamp. Otherwise,
256 - // there is no need to keep them in memory.
257 - continue;
258 - }
259 - }
260 - }
261 -
262 - if(unlikely(*pointers[0] == 'A')) {
263 - // make sure it is properly numbered
264 - if(unlikely(host->health_log.alarms && unique_id < host->health_log.alarms->unique_id)) {
265 - error( "HEALTH [%s]: line %zu of file '%s' has alarm log entry %u in wrong order. Ignoring it."
266 - , rrdhost_hostname(host), line, filename, unique_id);
267 - errored++;
268 - continue;
269 - }
270 -
271 - ae = callocz(1, sizeof(ALARM_ENTRY));
272 - }
273 - else if(unlikely(*pointers[0] == 'U')) {
274 - // find the original
275 - for(ae = host->health_log.alarms; ae ; ae = ae->next) {
276 - if(unlikely(unique_id == ae->unique_id)) {
277 - if(unlikely(*pointers[0] == 'A')) {
278 - error("HEALTH [%s]: line %zu of file '%s' adds duplicate alarm log entry %u. Using the later."
279 - , rrdhost_hostname(host), line, filename, unique_id);
280 - *pointers[0] = 'U';
281 - duplicate++;
282 - }
283 - break;
284 - }
285 - else if(unlikely(unique_id > ae->unique_id)) {
286 - // no need to continue
287 - // the linked list is sorted
288 - ae = NULL;
289 - break;
290 - }
291 - }
292 - }
293 -
294 - // if not found, skip this line
295 - if(unlikely(!ae)) {
296 - // error("HEALTH [%s]: line %zu of file '%s' updates alarm log entry with unique id %u, but it is not found.", host->hostname, line, filename, unique_id);
297 - continue;
298 - }
299 -
300 - // check for a possible host mismatch
301 - //if(strcmp(pointers[1], host->hostname))
302 - // error("HEALTH [%s]: line %zu of file '%s' provides an alarm for host '%s' but this is named '%s'.", host->hostname, line, filename, pointers[1], host->hostname);
303 -
304 - ae->unique_id = unique_id;
305 - if (!is_valid_alarm_id(host, pointers[14], pointers[13], alarm_id)) {
306 - STRING *chart = string_strdupz(pointers[14]);
307 - STRING *name = string_strdupz(pointers[13]);
308 - alarm_id = rrdcalc_get_unique_id(host, chart, name, NULL);
309 - string_freez(chart);
310 - string_freez(name);
311 - }
312 - ae->alarm_id = alarm_id;
313 - ae->alarm_event_id = (uint32_t)strtoul(pointers[4], NULL, 16);
314 - ae->updated_by_id = (uint32_t)strtoul(pointers[5], NULL, 16);
315 - ae->updates_id = (uint32_t)strtoul(pointers[6], NULL, 16);
316 -
317 - ae->when = (uint32_t)strtoul(pointers[7], NULL, 16);
318 - ae->duration = (uint32_t)strtoul(pointers[8], NULL, 16);
319 - ae->non_clear_duration = (uint32_t)strtoul(pointers[9], NULL, 16);
320 -
321 - ae->flags = (uint32_t)strtoul(pointers[10], NULL, 16);
322 - ae->flags |= HEALTH_ENTRY_FLAG_SAVED;
323 -
324 - ae->exec_run_timestamp = (uint32_t)strtoul(pointers[11], NULL, 16);
325 - ae->delay_up_to_timestamp = (uint32_t)strtoul(pointers[12], NULL, 16);
326 -
327 - string_freez(ae->name);
328 - ae->name = string_strdupz(pointers[13]);
329 -
330 - string_freez(ae->chart);
331 - ae->chart = string_strdupz(pointers[14]);
332 -
333 - string_freez(ae->family);
334 - ae->family = string_strdupz(pointers[15]);
335 -
336 - string_freez(ae->exec);
337 - ae->exec = string_strdupz(pointers[16]);
338 -
339 - string_freez(ae->recipient);
340 - ae->recipient = string_strdupz(pointers[17]);
341 -
342 - string_freez(ae->source);
343 - ae->source = string_strdupz(pointers[18]);
344 -
345 - string_freez(ae->units);
346 - ae->units = string_strdupz(pointers[19]);
347 -
348 - string_freez(ae->info);
349 - ae->info = string_strdupz(pointers[20]);
350 -
351 - ae->exec_code = str2i(pointers[21]);
352 - ae->new_status = str2i(pointers[22]);
353 - ae->old_status = str2i(pointers[23]);
354 - ae->delay = str2i(pointers[24]);
355 -
356 - ae->new_value = str2l(pointers[25]);
357 - ae->old_value = str2l(pointers[26]);
358 -
359 - ae->last_repeat = last_repeat;
360 -
361 - if (likely(entries > 30)) {
362 - string_freez(ae->classification);
363 - ae->classification = string_strdupz(pointers[28]);
364 -
365 - string_freez(ae->component);
366 - ae->component = string_strdupz(pointers[29]);
367 -
368 - string_freez(ae->type);
369 - ae->type = string_strdupz(pointers[30]);
370 - }
371 -
372 - char value_string[100 + 1];
373 - string_freez(ae->old_value_string);
374 - string_freez(ae->new_value_string);
375 - ae->old_value_string = string_strdupz(format_value_and_unit(value_string, 100, ae->old_value, ae_units(ae), -1));
376 - ae->new_value_string = string_strdupz(format_value_and_unit(value_string, 100, ae->new_value, ae_units(ae), -1));
377 -
378 - // add it to host if not already there
379 - if(unlikely(*pointers[0] == 'A')) {
380 - ae->next = host->health_log.alarms;
381 - host->health_log.alarms = ae;
382 - sql_health_alarm_log_insert(host, ae);
383 - loaded++;
384 - }
385 - else {
386 - sql_health_alarm_log_update(host, ae);
387 - updated++;
388 - }
389 -
390 - if(unlikely(ae->unique_id > host->health_max_unique_id))
391 - host->health_max_unique_id = ae->unique_id;
392 -
393 - if(unlikely(ae->alarm_id >= host->health_max_alarm_id))
394 - host->health_max_alarm_id = ae->alarm_id;
395 - }
396 - else {
397 - error("HEALTH [%s]: line %zu of file '%s' is invalid (unrecognized entry type '%s').", rrdhost_hostname(host), line, filename, pointers[0]);
398 - errored++;
399 - }
400 - }
401 -
402 - netdata_rwlock_unlock(&host->health_log.alarm_log_rwlock);
403 -
404 - dictionary_destroy(all_rrdcalcs);
405 - all_rrdcalcs = NULL;
406 -
407 - freez(buf);
408 -
409 - if(!host->health_max_unique_id) host->health_max_unique_id = (uint32_t)now_realtime_sec();
410 - if(!host->health_max_alarm_id) host->health_max_alarm_id = (uint32_t)now_realtime_sec();
411 -
412 - host->health_log.next_log_id = host->health_max_unique_id + 1;
413 - if (unlikely(!host->health_log.next_alarm_id || host->health_log.next_alarm_id <= host->health_max_alarm_id))
414 - host->health_log.next_alarm_id = host->health_max_alarm_id + 1;
415 -
416 - debug(D_HEALTH, "HEALTH [%s]: loaded file '%s' with %zd new alarm entries, updated %zd alarms, errors %zd entries, duplicate %zd", rrdhost_hostname(host), filename, loaded, updated, errored, duplicate);
417 - return loaded;
418 -}
419 -
420 -inline void health_alarm_log_load(RRDHOST *host) {
421 - health_alarm_log_close(host);
422 -
423 - char filename[FILENAME_MAX + 1];
424 - snprintfz(filename, FILENAME_MAX, "%s.old", host->health.health_log_filename);
425 - FILE *fp = fopen(filename, "r");
426 - if(!fp)
427 - error("HEALTH [%s]: cannot open health file: %s", rrdhost_hostname(host), filename);
428 - else {
429 - health_alarm_log_read(host, fp, filename);
430 - fclose(fp);
431 - }
432 -
433 - host->health.health_log_entries_written = 0;
434 - fp = fopen(host->health.health_log_filename, "r");
435 - if(!fp)
436 - error("HEALTH [%s]: cannot open health file: %s", rrdhost_hostname(host), host->health.health_log_filename);
437 - else {
438 - health_alarm_log_read(host, fp, host->health.health_log_filename);
439 - fclose(fp);
440 - }
441 -}
442 -
443 -
18 // ----------------------------------------------------------------------------
19 // health alarm log management
20
streaming/replication.c
+58 -7
@@ -1328,6 +1328,7 @@ static bool replication_execute_request(struct replication_request *rq, bool wor
1328 replication_response_execute_and_finalize(
1329 rq->q, (size_t)((unsigned long long)rq->sender->host->sender->buffer->max_size * MAX_REPLICATION_MESSAGE_PERCENT_SENDER_BUFFER / 100ULL));
1330
1331 + rq->q = NULL;
1332 netdata_thread_enable_cancelability();
1333
1334 __atomic_add_fetch(&replication_globals.atomic.executed, 1, __ATOMIC_RELAXED);
@@ -1335,6 +1336,11 @@ static bool replication_execute_request(struct replication_request *rq, bool wor
1336 ret = true;
1337
1338 cleanup:
1339 + if(rq->q) {
1340 + replication_response_cancel_and_finalize(rq->q);
1341 + rq->q = NULL;
1342 + }
1343 +
1344 string_freez(rq->chart_id);
1345 worker_is_idle();
1346 return ret;
@@ -1518,13 +1524,41 @@ static void replication_initialize_workers(bool master) {
1524 #define REQUEST_QUEUE_EMPTY (-1)
1525 #define REQUEST_CHART_NOT_FOUND (-2)
1526
1521 -static int replication_execute_next_pending_request(void) {
1527 +static int replication_execute_next_pending_request(bool cancel) {
1528 static __thread int max_requests_ahead = 0;
1529 static __thread struct replication_request *rqs = NULL;
1530 static __thread int rqs_last_executed = 0, rqs_last_prepared = 0;
1531 static __thread size_t queue_rounds = 0; (void)queue_rounds;
1532 struct replication_request *rq;
1533
1534 + if(unlikely(cancel)) {
1535 + if(rqs) {
1536 + size_t cancelled = 0;
1537 + do {
1538 + if (++rqs_last_executed >= max_requests_ahead)
1539 + rqs_last_executed = 0;
1540 +
1541 + rq = &rqs[rqs_last_executed];
1542 +
1543 + if (rq->q) {
1544 + internal_fatal(rq->executed, "REPLAY FATAL: query has already been executed!");
1545 + internal_fatal(!rq->found, "REPLAY FATAL: orphan q in rq");
1546 +
1547 + replication_response_cancel_and_finalize(rq->q);
1548 + rq->q = NULL;
1549 + cancelled++;
1550 + }
1551 +
1552 + rq->executed = true;
1553 + rq->found = false;
1554 +
1555 + } while (rqs_last_executed != rqs_last_prepared);
1556 +
1557 + internal_error(true, "REPLICATION: cancelled %zu inflight queries", cancelled);
1558 + }
1559 + return REQUEST_QUEUE_EMPTY;
1560 + }
1561 +
1562 if(unlikely(!rqs)) {
1563 max_requests_ahead = get_system_cpus() / 2;
1564
@@ -1545,8 +1579,8 @@ static int replication_execute_next_pending_request(void) {
1579 queue_rounds++;
1580 }
1581
1548 - internal_fatal(queue_rounds > 1 && !rqs[rqs_last_prepared].executed,
1549 - "REPLAY FATAL: query has not been executed!");
1582 + internal_fatal(rqs[rqs_last_prepared].q,
1583 + "REPLAY FATAL: slot is used by query that has not been executed!");
1584
1585 worker_is_busy(WORKER_JOB_FIND_NEXT);
1586 rqs[rqs_last_prepared] = replication_request_get_first_available();
@@ -1562,6 +1596,8 @@ static int replication_execute_next_pending_request(void) {
1596 worker_is_busy(WORKER_JOB_PREPARE_QUERY);
1597 rq->q = replication_response_prepare(rq->st, rq->start_streaming, rq->after, rq->before);
1598 }
1599 +
1600 + rq->executed = false;
1601 }
1602
1603 } while(rq->found && rqs_last_prepared != rqs_last_executed);
@@ -1572,14 +1608,17 @@ static int replication_execute_next_pending_request(void) {
1608 rqs_last_executed = 0;
1609
1610 rq = &rqs[rqs_last_executed];
1575 - rq->executed = true;
1611
1612 if(rq->found) {
1613 + internal_fatal(rq->executed, "REPLAY FATAL: query has already been executed!");
1614 +
1615 if (rq->sender_last_flush_ut != rrdpush_sender_get_flush_time(rq->sender)) {
1616 // the sender has reconnected since this request was queued,
1617 // we can safely throw it away, since the parent will resend it
1618 replication_response_cancel_and_finalize(rq->q);
1619 + rq->executed = true;
1620 rq->found = false;
1621 + rq->q = NULL;
1622 }
1623 else if (rrdpush_sender_replication_buffer_full_get(rq->sender)) {
1624 // the sender buffer is full, so we can ignore this request,
@@ -1587,7 +1626,9 @@ static int replication_execute_next_pending_request(void) {
1626 // and the sender will put it back in when there is
1627 // enough room in the buffer for processing replication requests
1628 replication_response_cancel_and_finalize(rq->q);
1629 + rq->executed = true;
1630 rq->found = false;
1631 + rq->q = NULL;
1632 }
1633 else {
1634 // we can execute this,
@@ -1596,6 +1637,8 @@ static int replication_execute_next_pending_request(void) {
1637 dictionary_del(rq->sender->replication.requests, string2str(rq->chart_id));
1638 }
1639 }
1640 + else
1641 + internal_fatal(rq->q, "REPLAY FATAL: slot status says slot is empty, but it has a pending query!");
1642
1643 } while(!rq->found && rqs_last_executed != rqs_last_prepared);
1644
@@ -1606,7 +1649,12 @@ static int replication_execute_next_pending_request(void) {
1649
1650 replication_set_latest_first_time(rq->after);
1651
1609 - if(unlikely(!replication_execute_request(rq, true))) {
1652 + bool chart_found = replication_execute_request(rq, true);
1653 + rq->executed = true;
1654 + rq->found = false;
1655 + rq->q = NULL;
1656 +
1657 + if(unlikely(!chart_found)) {
1658 worker_is_idle();
1659 return REQUEST_CHART_NOT_FOUND;
1660 }
@@ -1616,6 +1664,7 @@ static int replication_execute_next_pending_request(void) {
1664 }
1665
1666 static void replication_worker_cleanup(void *ptr __maybe_unused) {
1667 + replication_execute_next_pending_request(true);
1668 worker_unregister();
1669 }
1670
@@ -1625,7 +1674,7 @@ static void *replication_worker_thread(void *ptr) {
1674 netdata_thread_cleanup_push(replication_worker_cleanup, ptr);
1675
1676 while(service_running(SERVICE_REPLICATION)) {
1628 - if(unlikely(replication_execute_next_pending_request() == REQUEST_QUEUE_EMPTY)) {
1677 + if(unlikely(replication_execute_next_pending_request(false) == REQUEST_QUEUE_EMPTY)) {
1678 sender_thread_buffer_free();
1679 worker_is_busy(WORKER_JOB_WAIT);
1680 worker_is_idle();
@@ -1641,6 +1690,8 @@ static void replication_main_cleanup(void *ptr) {
1690 struct netdata_static_thread *static_thread = (struct netdata_static_thread *)ptr;
1691 static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;
1692
1693 + replication_execute_next_pending_request(true);
1694 +
1695 int threads = (int)replication_globals.main_thread.threads;
1696 for(int i = 0; i < threads ;i++) {
1697 netdata_thread_join(*replication_globals.main_thread.threads_ptrs[i], NULL);
@@ -1756,7 +1807,7 @@ void *replication_thread_main(void *ptr __maybe_unused) {
1807 worker_is_idle();
1808 }
1809
1759 - if(unlikely(replication_execute_next_pending_request() == REQUEST_QUEUE_EMPTY)) {
1810 + if(unlikely(replication_execute_next_pending_request(false) == REQUEST_QUEUE_EMPTY)) {
1811
1812 worker_is_busy(WORKER_JOB_WAIT);
1813 replication_recursive_lock();
web/api/queries/query.c
+31 -17
@@ -17,6 +17,8 @@
17 #include "percentile/percentile.h"
18 #include "trimmed_mean/trimmed_mean.h"
19
20 +#define POINTS_TO_EXPAND_QUERY 0
21 +
22 // ----------------------------------------------------------------------------
23
24 static struct {
@@ -957,10 +959,12 @@ static void query_planer_initialize_plans(QUERY_ENGINE_OPS *ops) {
959 QUERY_METRIC *qm = ops->qm;
960
961 for(size_t p = 0; p < qm->plan.used ; p++) {
960 - time_t after = qm->plan.array[p].after;
961 - time_t before = qm->plan.array[p].before;
962 -
962 size_t tier = qm->plan.array[p].tier;
963 + time_t update_every = qm->tiers[tier].db_update_every_s;
964 +
965 + time_t after = qm->plan.array[p].after - (update_every * POINTS_TO_EXPAND_QUERY);
966 + time_t before = qm->plan.array[p].before + (update_every * POINTS_TO_EXPAND_QUERY);
967 +
968 struct query_metric_tier *tier_ptr = &qm->tiers[tier];
969 tier_ptr->eng->api.query_ops.init(
970 tier_ptr->db_metric_handle,
@@ -1180,11 +1184,6 @@ static bool query_plan(QUERY_ENGINE_OPS *ops, time_t after_wanted, time_t before
1184 }
1185 #endif
1186
1183 - for(size_t p = 0; p < qm->plan.used ;p++) {
1184 - size_t tier = qm->plan.array[p].tier;
1185 - qm->plan.array[p].before += qm->tiers[tier].db_update_every_s - 1;
1186 - }
1187 -
1187 query_planer_initialize_plans(ops);
1188 query_planer_activate_plan(ops, 0, 0);
1189
@@ -1361,19 +1360,30 @@ static void rrd2rrdr_query_execute(RRDR *r, size_t dim_id_in_rrdr, QUERY_ENGINE_
1360 }
1361
1362 // check if the db is giving us zero duration points
1364 - if(unlikely(db_points_read_since_plan_switch > 1 && new_point.start_time == new_point.end_time)) {
1365 - internal_error(true, "QUERY: '%s', dimension '%s' next_metric() returned point %zu start time %ld, end time %ld, that are both equal",
1366 - qt->id, string2str(qm->dimension.id), new_point.id, new_point.start_time, new_point.end_time);
1363 + if(unlikely(db_points_read_since_plan_switch > 1 &&
1364 + new_point.start_time == new_point.end_time)) {
1365 +
1366 + internal_error(true, "QUERY: '%s', dimension '%s' next_metric() returned "
1367 + "point %zu from %ld to %ld, that are both equal",
1368 + qt->id, string2str(qm->dimension.id),
1369 + new_point.id, new_point.start_time, new_point.end_time);
1370
1371 new_point.start_time = new_point.end_time - ops->tier_ptr->db_update_every_s;
1372 }
1373
1374 // check if the db is advancing the query
1372 - if(unlikely(db_points_read_since_plan_switch > 1 && new_point.end_time <= last1_point.end_time)) {
1375 + if(unlikely(db_points_read_since_plan_switch > 1 &&
1376 + new_point.end_time <= last1_point.end_time)) {
1377 +
1378 internal_error(true,
1374 - "QUERY: '%s', dimension '%s' next_metric() returned point %zu from %ld to %ld, before the last point %zu from %ld to %ld, now is %ld to %ld",
1375 - qt->id, string2str(qm->dimension.id), new_point.id, new_point.start_time, new_point.end_time,
1376 - last1_point.id, last1_point.start_time, last1_point.end_time, now_start_time, now_end_time);
1379 + "QUERY: '%s', dimension '%s' next_metric() returned "
1380 + "point %zu from %ld to %ld, before the "
1381 + "last point %zu from %ld to %ld, "
1382 + "now is %ld to %ld",
1383 + qt->id, string2str(qm->dimension.id),
1384 + new_point.id, new_point.start_time, new_point.end_time,
1385 + last1_point.id, last1_point.start_time, last1_point.end_time,
1386 + now_start_time, now_end_time);
1387
1388 count_same_end_time++;
1389 continue;
@@ -1398,8 +1408,12 @@ static void rrd2rrdr_query_execute(RRDR *r, size_t dim_id_in_rrdr, QUERY_ENGINE_
1408 // at exactly the time we will want
1409
1410 // we only log if this is not point 1
1401 - internal_error(new_point.end_time < after_wanted && new_point.id > 1,
1402 - "QUERY: '%s', dimension '%s' next_metric() returned point %zu from %ld time %ld, which is entirely before our current timeframe %ld to %ld (and before the entire query, after %ld, before %ld)",
1411 + internal_error(new_point.end_time < after_wanted &&
1412 + new_point.id > POINTS_TO_EXPAND_QUERY + 1,
1413 + "QUERY: '%s', dimension '%s' next_metric() "
1414 + "returned point %zu from %ld time %ld, "
1415 + "which is entirely before our current timeframe %ld to %ld "
1416 + "(and before the entire query, after %ld, before %ld)",
1417 qt->id, string2str(qm->dimension.id),
1418 new_point.id, new_point.start_time, new_point.end_time,
1419 now_start_time, now_end_time,