From 81581f9719bc56f01d5aa08952671d65fda9867a Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Mon, 8 May 2023 18:27:08 +0200 Subject: Merging upstream version 1.39.0. Signed-off-by: Daniel Baumann --- database/rrdset.c | 108 +++++++++++++++++++++++++++++++++++------------------- 1 file changed, 70 insertions(+), 38 deletions(-) (limited to 'database/rrdset.c') diff --git a/database/rrdset.c b/database/rrdset.c index 57f962cd6..2843bb330 100644 --- a/database/rrdset.c +++ b/database/rrdset.c @@ -158,7 +158,7 @@ static void rrdset_insert_callback(const DICTIONARY_ITEM *item __maybe_unused, v STORAGE_ENGINE *eng = st->rrdhost->db[tier].eng; if(!eng) continue; - st->storage_metrics_groups[tier] = eng->api.collect_ops.metrics_group_get(host->db[tier].instance, &st->chart_uuid); + st->storage_metrics_groups[tier] = storage_engine_metrics_group_get(eng->backend, host->db[tier].instance, &st->chart_uuid); } } @@ -184,6 +184,8 @@ static void rrdset_insert_callback(const DICTIONARY_ITEM *item __maybe_unused, v ml_chart_new(st); } +void pluginsd_rrdset_cleanup(RRDSET *st); + void rrdset_finalize_collection(RRDSET *st, bool dimensions_too) { RRDHOST *host = st->rrdhost; @@ -201,10 +203,12 @@ void rrdset_finalize_collection(RRDSET *st, bool dimensions_too) { if(!eng) continue; if(st->storage_metrics_groups[tier]) { - eng->api.collect_ops.metrics_group_release(host->db[tier].instance, st->storage_metrics_groups[tier]); + storage_engine_metrics_group_release(eng->backend, host->db[tier].instance, st->storage_metrics_groups[tier]); st->storage_metrics_groups[tier] = NULL; } } + + pluginsd_rrdset_cleanup(st); } // the destructor - the dictionary is write locked while this runs @@ -475,6 +479,27 @@ inline RRDSET *rrdset_find_byname(RRDHOST *host, const char *name) { return(st); } +RRDSET_ACQUIRED *rrdset_find_and_acquire(RRDHOST *host, const char *id) { + debug(D_RRD_CALLS, "rrdset_find_and_acquire() for host %s, chart %s", rrdhost_hostname(host), id); + + return (RRDSET_ACQUIRED *)dictionary_get_and_acquire_item(host->rrdset_root_index, id); +} + +RRDSET *rrdset_acquired_to_rrdset(RRDSET_ACQUIRED *rsa) { + if(unlikely(!rsa)) + return NULL; + + return (RRDSET *) dictionary_acquired_item_value((const DICTIONARY_ITEM *)rsa); +} + +void rrdset_acquired_release(RRDSET_ACQUIRED *rsa) { + if(unlikely(!rsa)) + return; + + RRDSET *rs = rrdset_acquired_to_rrdset(rsa); + dictionary_acquired_item_release(rs->rrdhost->rrdset_root_index, (const DICTIONARY_ITEM *)rsa); +} + // ---------------------------------------------------------------------------- // RRDSET - rename charts @@ -737,10 +762,8 @@ void rrdset_reset(RRDSET *st) { rd->collections_counter = 0; if(!rrddim_flag_check(rd, RRDDIM_FLAG_ARCHIVED)) { - for(size_t tier = 0; tier < storage_tiers ;tier++) { - if(rd->tiers[tier].db_collection_handle) - rd->tiers[tier].collect_ops->flush(rd->tiers[tier].db_collection_handle); - } + for(size_t tier = 0; tier < storage_tiers ;tier++) + storage_engine_store_flush(rd->tiers[tier].db_collection_handle); } } rrddim_foreach_done(rd); @@ -1116,7 +1139,7 @@ void store_metric_at_tier(RRDDIM *rd, size_t tier, struct rrddim_tier *t, STORAG if (likely(!storage_point_is_unset(t->virtual_point))) { - t->collect_ops->store_metric( + storage_engine_store_metric( t->db_collection_handle, t->next_point_end_time_s * USEC_PER_SEC, t->virtual_point.sum, @@ -1127,7 +1150,7 @@ void store_metric_at_tier(RRDDIM *rd, size_t tier, struct rrddim_tier *t, STORAG t->virtual_point.flags); } else { - t->collect_ops->store_metric( + storage_engine_store_metric( t->db_collection_handle, t->next_point_end_time_s * USEC_PER_SEC, NAN, @@ -1199,7 +1222,10 @@ void rrddim_store_metric(RRDDIM *rd, usec_t point_end_time_ut, NETDATA_DOUBLE n, #endif // NETDATA_LOG_COLLECTION_ERRORS // store the metric on tier 0 - rd->tiers[0].collect_ops->store_metric(rd->tiers[0].db_collection_handle, point_end_time_ut, n, 0, 0, 1, 0, flags); + storage_engine_store_metric(rd->tiers[0].db_collection_handle, point_end_time_ut, + n, 0, 0, + 1, 0, flags); + rrdset_done_statistics_points_stored_per_tier[0]++; time_t now_s = (time_t)(point_end_time_ut / USEC_PER_SEC); @@ -1229,6 +1255,8 @@ void rrddim_store_metric(RRDDIM *rd, usec_t point_end_time_ut, NETDATA_DOUBLE n, store_metric_at_tier(rd, tier, t, sp, point_end_time_ut); } + + rrdcontext_collected_rrddim(rd); } void store_metric_collection_completed() { @@ -1269,7 +1297,8 @@ void rrdset_thread_rda_free(void) { } static inline size_t rrdset_done_interpolate( - RRDSET *st + RRDSET_STREAM_BUFFER *rsb + , RRDSET *st , struct rda_item *rda_base , size_t rda_slots , usec_t update_every_ut @@ -1399,32 +1428,38 @@ static inline size_t rrdset_done_interpolate( time_t current_time_s = (time_t) (next_store_ut / USEC_PER_SEC); if(unlikely(!store_this_entry)) { - (void) ml_is_anomalous(rd, current_time_s, 0, false); + (void) ml_dimension_is_anomalous(rd, current_time_s, 0, false); + + if(rsb->wb && rsb->v2) + rrddim_push_metrics_v2(rsb, rd, next_store_ut, NAN, SN_FLAG_NONE); rrddim_store_metric(rd, next_store_ut, NAN, SN_FLAG_NONE); - rrdcontext_collected_rrddim(rd); continue; } if(likely(rd->updated && rd->collections_counter > 1 && iterations < gap_when_lost_iterations_above)) { uint32_t dim_storage_flags = storage_flags; - if (ml_is_anomalous(rd, current_time_s, new_value, true)) { + if (ml_dimension_is_anomalous(rd, current_time_s, new_value, true)) { // clear anomaly bit: 0 -> is anomalous, 1 -> not anomalous dim_storage_flags &= ~((storage_number)SN_FLAG_NOT_ANOMALOUS); } + if(rsb->wb && rsb->v2) + rrddim_push_metrics_v2(rsb, rd, next_store_ut, new_value, dim_storage_flags); + rrddim_store_metric(rd, next_store_ut, new_value, dim_storage_flags); - rrdcontext_collected_rrddim(rd); rd->last_stored_value = new_value; } else { - (void) ml_is_anomalous(rd, current_time_s, 0, false); + (void) ml_dimension_is_anomalous(rd, current_time_s, 0, false); rrdset_debug(st, "%s: STORE[%ld] = NON EXISTING ", rrddim_name(rd), current_entry); + if(rsb->wb && rsb->v2) + rrddim_push_metrics_v2(rsb, rd, next_store_ut, NAN, SN_FLAG_NONE); + rrddim_store_metric(rd, next_store_ut, NAN, SN_FLAG_NONE); - rrdcontext_collected_rrddim(rd); rd->last_stored_value = NAN; } @@ -1468,6 +1503,10 @@ void rrdset_done(RRDSET *st) { void rrdset_timed_done(RRDSET *st, struct timeval now, bool pending_rrdset_next) { if(unlikely(!service_running(SERVICE_COLLECTORS))) return; + RRDSET_STREAM_BUFFER stream_buffer = { .wb = NULL, }; + if(unlikely(rrdhost_has_rrdpush_sender_enabled(st->rrdhost))) + stream_buffer = rrdset_push_metric_initialize(st, now.tv_sec); + netdata_spinlock_lock(&st->data_collection_lock); if (pending_rrdset_next) @@ -1489,10 +1528,10 @@ void rrdset_timed_done(RRDSET *st, struct timeval now, bool pending_rrdset_next) update_every_ut = st->update_every * USEC_PER_SEC; // st->update_every in microseconds RRDSET_FLAGS rrdset_flags = rrdset_flag_check(st, ~0); - if(unlikely(rrdset_flags & RRDSET_FLAG_COLLECTION_FINISHED)) + if(unlikely(rrdset_flags & RRDSET_FLAG_COLLECTION_FINISHED)) { + netdata_spinlock_unlock(&st->data_collection_lock); return; - - netdata_thread_disable_cancelability(); + } if (unlikely(rrdset_flags & RRDSET_FLAG_OBSOLETE)) { error("Chart '%s' has the OBSOLETE flag set, but it is collected.", rrdset_id(st)); @@ -1500,8 +1539,7 @@ void rrdset_timed_done(RRDSET *st, struct timeval now, bool pending_rrdset_next) } // check if the chart has a long time to be updated - if(unlikely(st->usec_since_last_update > st->entries * update_every_ut && - st->rrd_memory_mode != RRD_MEMORY_MODE_DBENGINE && st->rrd_memory_mode != RRD_MEMORY_MODE_NONE)) { + if(unlikely(st->usec_since_last_update > MAX(st->entries, 60) * update_every_ut)) { info("host '%s', chart '%s': took too long to be updated (counter #%zu, update #%zu, %0.3" NETDATA_DOUBLE_MODIFIER " secs). Resetting it.", rrdhost_hostname(st->rrdhost), rrdset_id(st), st->counter, st->counter_done, (NETDATA_DOUBLE)st->usec_since_last_update / USEC_PER_SEC); rrdset_reset(st); @@ -1527,9 +1565,6 @@ void rrdset_timed_done(RRDSET *st, struct timeval now, bool pending_rrdset_next) // calculate the proper last_collected_time, using usec_since_last_update last_collect_ut = rrdset_update_last_collected_time(st); } - if (unlikely(st->rrd_memory_mode == RRD_MEMORY_MODE_NONE)) { - goto after_first_database_work; - } // if this set has not been updated in the past // we fake the last_update time to be = now - usec_since_last_update @@ -1592,11 +1627,10 @@ void rrdset_timed_done(RRDSET *st, struct timeval now, bool pending_rrdset_next) } } -after_first_database_work: st->counter_done++; - if(unlikely(rrdhost_has_rrdpush_sender_enabled(st->rrdhost))) - rrdset_done_push(st); + if(stream_buffer.wb && !stream_buffer.v2) + rrdset_push_metrics_v1(&stream_buffer, st); uint32_t has_reset_value = 0; @@ -1654,9 +1688,6 @@ after_first_database_work: rrddim_foreach_done(rd); rda_slots = dimensions; - if (unlikely(st->rrd_memory_mode == RRD_MEMORY_MODE_NONE)) - goto after_second_database_work; - rrdset_debug(st, "last_collect_ut = %0.3" NETDATA_DOUBLE_MODIFIER " (last collection time)", (NETDATA_DOUBLE)last_collect_ut/USEC_PER_SEC); rrdset_debug(st, "now_collect_ut = %0.3" NETDATA_DOUBLE_MODIFIER " (current collection time)", (NETDATA_DOUBLE)now_collect_ut/USEC_PER_SEC); rrdset_debug(st, "last_stored_ut = %0.3" NETDATA_DOUBLE_MODIFIER " (last updated time)", (NETDATA_DOUBLE)last_stored_ut/USEC_PER_SEC); @@ -1857,7 +1888,8 @@ after_first_database_work: // #endif rrdset_done_interpolate( - st + &stream_buffer + , st , rda_base , rda_slots , update_every_ut @@ -1869,7 +1901,6 @@ after_first_database_work: , has_reset_value ); -after_second_database_work: for(dim_id = 0, rda = rda_base ; dim_id < rda_slots ; ++dim_id, ++rda) { rd = rda->rd; if(unlikely(!rd)) continue; @@ -1928,6 +1959,7 @@ after_second_database_work: } netdata_spinlock_unlock(&st->data_collection_lock); + rrdset_push_metrics_finished(&stream_buffer, st); // ALL DONE ABOUT THE DATA UPDATE // -------------------------------------------------------------------- @@ -1955,8 +1987,6 @@ after_second_database_work: rrdcontext_collected_rrdset(st); - netdata_thread_enable_cancelability(); - store_metric_collection_completed(); } @@ -1965,18 +1995,20 @@ time_t rrdset_set_update_every_s(RRDSET *st, time_t update_every_s) { internal_error(true, "RRDSET '%s' switching update every from %d to %d", rrdset_id(st), (int)st->update_every, (int)update_every_s); - time_t prev_update_every_s = st->update_every; - st->update_every = update_every_s; + time_t prev_update_every_s = (time_t) st->update_every; + st->update_every = (int) update_every_s; // switch update every to the storage engine RRDDIM *rd; rrddim_foreach_read(rd, st) { for (size_t tier = 0; tier < storage_tiers; tier++) { if (rd->tiers[tier].db_collection_handle) - rd->tiers[tier].collect_ops->change_collection_frequency(rd->tiers[tier].db_collection_handle, (int)(st->rrdhost->db[tier].tier_grouping * st->update_every)); + storage_engine_store_change_collection_frequency( + rd->tiers[tier].db_collection_handle, + (int)(st->rrdhost->db[tier].tier_grouping * st->update_every)); } - assert(rd->update_every == prev_update_every_s && + assert(rd->update_every == (int) prev_update_every_s && "chart's update every differs from the update every of its dimensions"); rd->update_every = st->update_every; } -- cgit v1.2.3