summaryrefslogtreecommitdiffstats
path: root/database/rrdset.c
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--database/rrdset.c108
1 files changed, 70 insertions, 38 deletions
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;
}