Keep a count of metrics and samples collected (#17042)

* count metrics and sample points

* Adjust calculation for samples when flushing a hot page

* Restore deleted call to mrg_metric_set_clean_latest_time_s

* Simplify code

* Additional check when removing samples

9 files changed, 105 insertions, 7 deletions

diff --git a/src/database/contexts/api_v2.c b/src/database/contexts/api_v2.c
index 6b36f37945..52a168ae4c 100644
--- a/src/database/contexts/api_v2.c
+++ b/src/database/contexts/api_v2.c
@@ -1131,6 +1131,8 @@ void buffer_json_agents_v2(BUFFER *wb, struct query_timings *timings, time_t now
 
             buffer_json_add_array_item_object(wb);
             buffer_json_member_add_uint64(wb, "tier", tier);
+            buffer_json_member_add_uint64(wb, "metrics", storage_engine_metrics(eng->seb, localhost->db[tier].si));
+            buffer_json_member_add_uint64(wb, "samples", storage_engine_samples(eng->seb, localhost->db[tier].si));
 
             if(used || max) {
                 buffer_json_member_add_uint64(wb, "disk_used", used);
diff --git a/src/database/engine/journalfile.c b/src/database/engine/journalfile.c
index ed128f349f..2caae87a00 100644
--- a/src/database/engine/journalfile.c
+++ b/src/database/engine/journalfile.c
@@ -708,8 +708,14 @@ static void journalfile_restore_extent_metadata(struct rrdengine_instance *ctx,
 
             bool added;
             metric = mrg_metric_add_and_acquire(main_mrg, entry, &added);
-            if(added)
+            if(added) {
+                __atomic_add_fetch(&ctx->atomic.metrics, 1, __ATOMIC_RELAXED);
                 update_metric_time = false;
+            }
+            if (vd.update_every_s) {
+                uint64_t samples = (vd.end_time_s - vd.start_time_s) / vd.update_every_s;
+                __atomic_add_fetch(&ctx->atomic.samples, samples, __ATOMIC_RELAXED);
+            }
         }
         Word_t metric_id = mrg_metric_id(main_mrg, metric);
 
diff --git a/src/database/engine/metric.c b/src/database/engine/metric.c
index c618e22595..97db53efbc 100644
--- a/src/database/engine/metric.c
+++ b/src/database/engine/metric.c
@@ -531,6 +531,11 @@ inline bool mrg_metric_set_hot_latest_time_s(MRG *mrg __maybe_unused, METRIC *me
     return false;
 }
 
+inline time_t mrg_metric_get_latest_clean_time_s(MRG *mrg __maybe_unused, METRIC *metric) {
+    time_t clean = __atomic_load_n(&metric->latest_time_s_clean, __ATOMIC_RELAXED);
+    return clean;
+}
+
 inline time_t mrg_metric_get_latest_time_s(MRG *mrg __maybe_unused, METRIC *metric) {
     time_t clean = __atomic_load_n(&metric->latest_time_s_clean, __ATOMIC_RELAXED);
     time_t hot = __atomic_load_n(&metric->latest_time_s_hot, __ATOMIC_RELAXED);
@@ -641,9 +646,30 @@ inline void mrg_update_metric_retention_and_granularity_by_uuid(
         metric = mrg_metric_add_and_acquire(mrg, entry, &added);
     }
 
-    if (likely(!added))
+    struct rrdengine_instance *ctx = (struct rrdengine_instance *) section;
+    if (likely(!added)) {
+        uint64_t old_samples = 0;
+
+        if (update_every_s && metric->latest_update_every_s && metric->latest_time_s_clean)
+            old_samples = (metric->latest_time_s_clean - metric->first_time_s) / metric->latest_update_every_s;
+
         mrg_metric_expand_retention(mrg, metric, first_time_s, last_time_s, update_every_s);
 
+        uint64_t new_samples = 0;
+        if (update_every_s && metric->latest_update_every_s && metric->latest_time_s_clean)
+            new_samples = (metric->latest_time_s_clean - metric->first_time_s) / metric->latest_update_every_s;
+
+        __atomic_add_fetch(&ctx->atomic.samples, new_samples - old_samples, __ATOMIC_RELAXED);
+    }
+    else {
+        // Newly added
+        if (update_every_s) {
+            uint64_t samples = (last_time_s - first_time_s) / update_every_s;
+            __atomic_add_fetch(&ctx->atomic.samples, samples, __ATOMIC_RELAXED);
+        }
+        __atomic_add_fetch(&ctx->atomic.metrics, 1, __ATOMIC_RELAXED);
+    }
+
     mrg_metric_release(mrg, metric);
 }
 
diff --git a/src/database/engine/metric.h b/src/database/engine/metric.h
index 8fed08abbf..3bace90574 100644
--- a/src/database/engine/metric.h
+++ b/src/database/engine/metric.h
@@ -69,6 +69,7 @@ time_t mrg_metric_get_first_time_s(MRG *mrg, METRIC *metric);
 bool mrg_metric_set_clean_latest_time_s(MRG *mrg, METRIC *metric, time_t latest_time_s);
 bool mrg_metric_set_hot_latest_time_s(MRG *mrg, METRIC *metric, time_t latest_time_s);
 time_t mrg_metric_get_latest_time_s(MRG *mrg, METRIC *metric);
+time_t mrg_metric_get_latest_clean_time_s(MRG *mrg, METRIC *metric);
 
 bool mrg_metric_set_update_every(MRG *mrg, METRIC *metric, uint32_t update_every_s);
 bool mrg_metric_set_update_every_s_if_zero(MRG *mrg, METRIC *metric, uint32_t update_every_s);
diff --git a/src/database/engine/rrdengine.c b/src/database/engine/rrdengine.c
index 9c79435218..bae0fc4650 100644
--- a/src/database/engine/rrdengine.c
+++ b/src/database/engine/rrdengine.c
@@ -1171,7 +1171,17 @@ static void update_metrics_first_time_s(struct rrdengine_instance *ctx, struct r
     for (size_t index = 0; index < added; ++index) {
         uuid_first_t_entry = &uuid_first_entry_list[index];
         if (likely(uuid_first_t_entry->first_time_s != LONG_MAX)) {
-            mrg_metric_set_first_time_s_if_bigger(main_mrg, uuid_first_t_entry->metric, uuid_first_t_entry->first_time_s);
+
+            time_t old_first_time_s = mrg_metric_get_first_time_s(main_mrg, uuid_first_t_entry->metric);
+
+            bool changed = mrg_metric_set_first_time_s_if_bigger(main_mrg, uuid_first_t_entry->metric, uuid_first_t_entry->first_time_s);
+            if (changed) {
+                uint32_t update_every_s = mrg_metric_get_update_every_s(main_mrg, uuid_first_t_entry->metric);
+                if (update_every_s && old_first_time_s && uuid_first_t_entry->first_time_s > old_first_time_s) {
+                    uint64_t remove_samples = (uuid_first_t_entry->first_time_s - old_first_time_s) / update_every_s;
+                    __atomic_sub_fetch(&ctx->atomic.samples, remove_samples, __ATOMIC_RELAXED);
+                }
+            }
             mrg_metric_release(main_mrg, uuid_first_t_entry->metric);
         }
         else {
@@ -1180,6 +1190,14 @@ static void update_metrics_first_time_s(struct rrdengine_instance *ctx, struct r
             // there is no retention for this metric
             bool has_retention = mrg_metric_zero_disk_retention(main_mrg, uuid_first_t_entry->metric);
             if (!has_retention) {
+                time_t first_time_s = mrg_metric_get_first_time_s(main_mrg, uuid_first_t_entry->metric);
+                time_t last_time_s = mrg_metric_get_latest_time_s(main_mrg, uuid_first_t_entry->metric);
+                time_t update_every_s = mrg_metric_get_update_every_s(main_mrg, uuid_first_t_entry->metric);
+                if (update_every_s && first_time_s && last_time_s) {
+                    uint64_t remove_samples = (first_time_s - last_time_s) / update_every_s;
+                    __atomic_sub_fetch(&ctx->atomic.samples, remove_samples, __ATOMIC_RELAXED);
+                }
+
                 bool deleted = mrg_metric_release_and_delete(main_mrg, uuid_first_t_entry->metric);
                 if(deleted)
                     deleted_metrics++;
diff --git a/src/database/engine/rrdengine.h b/src/database/engine/rrdengine.h
index d65ec2c5a8..3047e0c6ac 100644
--- a/src/database/engine/rrdengine.h
+++ b/src/database/engine/rrdengine.h
@@ -387,6 +387,8 @@ struct rrdengine_instance {
         unsigned extents_currently_being_flushed;   // non-zero until we commit data to disk (both datafile and journal file)
 
         time_t first_time_s;
+        uint64_t metrics;
+        uint64_t samples;
     } atomic;
 
     struct {
diff --git a/src/database/engine/rrdengineapi.c b/src/database/engine/rrdengineapi.c
index 9b84e9491b..b9c2b59801 100755
--- a/src/database/engine/rrdengineapi.c
+++ b/src/database/engine/rrdengineapi.c
@@ -145,7 +145,10 @@ static METRIC *rrdeng_metric_create(STORAGE_INSTANCE *si, uuid_t *uuid) {
             .latest_update_every_s = 0,
     };
 
-    METRIC *metric = mrg_metric_add_and_acquire(main_mrg, entry, NULL);
+    bool added;
+    METRIC *metric = mrg_metric_add_and_acquire(main_mrg, entry, &added);
+    if (added)
+        __atomic_add_fetch(&ctx->atomic.metrics, 1, __ATOMIC_RELAXED);
     return metric;
 }
 
@@ -307,6 +310,16 @@ void rrdeng_store_metric_flush_current_page(STORAGE_COLLECT_HANDLE *sch) {
     else {
         check_completed_page_consistency(handle);
         mrg_metric_set_clean_latest_time_s(main_mrg, handle->metric, pgc_page_end_time_s(handle->pgc_page));
+
+        struct rrdengine_instance *ctx = mrg_metric_ctx(handle->metric);
+        time_t start_time_s = pgc_page_start_time_s(handle->pgc_page);
+        time_t end_time_s = pgc_page_end_time_s(handle->pgc_page);
+        uint32_t update_every_s = mrg_metric_get_update_every_s(main_mrg, handle->metric);
+        if (end_time_s && start_time_s && end_time_s > start_time_s && update_every_s) {
+            uint64_t add_samples = (end_time_s - start_time_s) / update_every_s;
+            __atomic_add_fetch(&ctx->atomic.samples, add_samples, __ATOMIC_RELAXED);
+        }
+
         pgc_page_hot_to_dirty_and_release(main_cache, handle->pgc_page);
     }
 
@@ -967,6 +980,16 @@ uint64_t rrdeng_disk_space_used(STORAGE_INSTANCE *si) {
     return __atomic_load_n(&ctx->atomic.current_disk_space, __ATOMIC_RELAXED);
 }
 
+uint64_t rrdeng_metrics(STORAGE_INSTANCE *si) {
+    struct rrdengine_instance *ctx = (struct rrdengine_instance *)si;
+    return __atomic_load_n(&ctx->atomic.metrics, __ATOMIC_RELAXED);
+}
+
+uint64_t rrdeng_samples(STORAGE_INSTANCE *si) {
+    struct rrdengine_instance *ctx = (struct rrdengine_instance *)si;
+    return __atomic_load_n(&ctx->atomic.samples, __ATOMIC_RELAXED);
+}
+
 time_t rrdeng_global_first_time_s(STORAGE_INSTANCE *si) {
     struct rrdengine_instance *ctx = (struct rrdengine_instance *)si;
 
@@ -1154,6 +1177,8 @@ int rrdeng_init(struct rrdengine_instance **ctxp, const char *dbfiles_path,
 
     rw_spinlock_init(&ctx->njfv2idx.spinlock);
     ctx->atomic.first_time_s = LONG_MAX;
+    ctx->atomic.metrics = 0;
+    ctx->atomic.samples = 0;
 
     if (rrdeng_dbengine_spawn(ctx) && !init_rrd_files(ctx)) {
         // success - we run this ctx too
diff --git a/src/database/engine/rrdengineapi.h b/src/database/engine/rrdengineapi.h
index 44b8fcb8ce..fb449cd9ba 100644
--- a/src/database/engine/rrdengineapi.h
+++ b/src/database/engine/rrdengineapi.h
@@ -223,7 +223,4 @@ RRDENG_SIZE_STATS rrdeng_size_statistics(struct rrdengine_instance *ctx);
 size_t rrdeng_collectors_running(struct rrdengine_instance *ctx);
 bool rrdeng_is_legacy(STORAGE_INSTANCE *si);
 
-uint64_t rrdeng_disk_space_max(STORAGE_INSTANCE *si);
-uint64_t rrdeng_disk_space_used(STORAGE_INSTANCE *si);
-
 #endif /* NETDATA_RRDENGINEAPI_H */
diff --git a/src/database/rrd.h b/src/database/rrd.h
index adea384852..a6b1479aaa 100644
--- a/src/database/rrd.h
+++ b/src/database/rrd.h
@@ -446,6 +446,27 @@ static inline uint64_t storage_engine_disk_space_used(STORAGE_ENGINE_BACKEND seb
     return 0;
 }
 
+uint64_t rrdeng_metrics(STORAGE_INSTANCE *si);
+static inline uint64_t storage_engine_metrics(STORAGE_ENGINE_BACKEND seb __maybe_unused, STORAGE_INSTANCE *si __maybe_unused) {
+#ifdef ENABLE_DBENGINE
+    if(likely(seb == STORAGE_ENGINE_BACKEND_DBENGINE))
+        return rrdeng_metrics(si);
+#endif
+
+    // TODO - calculate the total host disk space for memory mode save and map
+    return 0;
+}
+
+uint64_t rrdeng_samples(STORAGE_INSTANCE *si);
+static inline uint64_t storage_engine_samples(STORAGE_ENGINE_BACKEND seb __maybe_unused, STORAGE_INSTANCE *si __maybe_unused) {
+#ifdef ENABLE_DBENGINE
+    if(likely(seb == STORAGE_ENGINE_BACKEND_DBENGINE))
+        return rrdeng_samples(si);
+#endif
+    return 0;
+}
+
+
 time_t rrdeng_global_first_time_s(STORAGE_INSTANCE *si);
 static inline time_t storage_engine_global_first_time_s(STORAGE_ENGINE_BACKEND seb __maybe_unused, STORAGE_INSTANCE *si __maybe_unused) {
 #ifdef ENABLE_DBENGINE