New upstream version 1.7.0+dfsgupstream/1.7.0+dfsg

1 files changed, 555 insertions, 366 deletions

diff --git a/src/rrdset.c b/src/rrdset.c
index c847b9690..caa427ff6 100644
--- a/src/rrdset.c
+++ b/src/rrdset.c
@@ -168,6 +168,58 @@ void rrdset_set_name(RRDSET *st, const char *name) {
         error("RRDSET: INTERNAL ERROR: attempted to index duplicate chart name '%s'", st->name);
 }
 
+inline void rrdset_is_obsolete(RRDSET *st) {
+    if(unlikely(!(rrdset_flag_check(st, RRDSET_FLAG_OBSOLETE)))) {
+        rrdset_flag_set(st, RRDSET_FLAG_OBSOLETE);
+        rrdset_flag_clear(st, RRDSET_FLAG_EXPOSED_UPSTREAM);
+
+        // the chart will not get more updates (data collection)
+        // so, we have to push its definition now
+        if(unlikely(st->rrdhost->rrdpush_enabled))
+            rrdset_push_chart_definition(st);
+    }
+}
+
+inline void rrdset_isnot_obsolete(RRDSET *st) {
+    if(unlikely((rrdset_flag_check(st, RRDSET_FLAG_OBSOLETE)))) {
+        rrdset_flag_clear(st, RRDSET_FLAG_OBSOLETE);
+        rrdset_flag_clear(st, RRDSET_FLAG_EXPOSED_UPSTREAM);
+
+        // the chart will be pushed upstream automatically
+        // due to data collection
+    }
+}
+
+inline void rrdset_update_heterogeneous_flag(RRDSET *st) {
+    RRDDIM *rd;
+
+    rrdset_flag_clear(st, RRDSET_FLAG_HOMEGENEOUS_CHECK);
+
+    RRD_ALGORITHM algorithm = st->dimensions->algorithm;
+    collected_number multiplier = abs(st->dimensions->multiplier);
+    collected_number divisor = abs(st->dimensions->divisor);
+
+    rrddim_foreach_read(rd, st) {
+        if(algorithm != rd->algorithm || multiplier != abs(rd->multiplier) || divisor != abs(rd->divisor)) {
+            if(!rrdset_flag_check(st, RRDSET_FLAG_HETEROGENEOUS)) {
+                #ifdef NETDATA_INTERNAL_CHECKS
+                info("Dimension '%s' added on chart '%s' of host '%s' is not homogeneous to other dimensions already present (algorithm is '%s' vs '%s', multiplier is " COLLECTED_NUMBER_FORMAT " vs " COLLECTED_NUMBER_FORMAT ", divisor is " COLLECTED_NUMBER_FORMAT " vs " COLLECTED_NUMBER_FORMAT ").",
+                        rd->name,
+                        st->name,
+                        st->rrdhost->hostname,
+                        rrd_algorithm_name(rd->algorithm), rrd_algorithm_name(algorithm),
+                        rd->multiplier, multiplier,
+                        rd->divisor, divisor
+                );
+                #endif
+                rrdset_flag_set(st, RRDSET_FLAG_HETEROGENEOUS);
+            }
+            return;
+        }
+    }
+
+    rrdset_flag_clear(st, RRDSET_FLAG_HETEROGENEOUS);
+}
 
 // ----------------------------------------------------------------------------
 // RRDSET - reset a chart
@@ -188,7 +240,7 @@ void rrdset_reset(RRDSET *st) {
         rd->last_collected_time.tv_sec = 0;
         rd->last_collected_time.tv_usec = 0;
         rd->collections_counter = 0;
-        memset(rd->values, 0, rd->entries * sizeof(storage_number));
+        // memset(rd->values, 0, rd->entries * sizeof(storage_number));
     }
 }
 
@@ -199,7 +251,7 @@ inline long align_entries_to_pagesize(RRD_MEMORY_MODE mode, long entries) {
     if(unlikely(entries < 5)) entries = 5;
     if(unlikely(entries > RRD_HISTORY_ENTRIES_MAX)) entries = RRD_HISTORY_ENTRIES_MAX;
 
-    if(unlikely(mode == RRD_MEMORY_MODE_NONE || mode == RRD_MEMORY_MODE_RAM))
+    if(unlikely(mode == RRD_MEMORY_MODE_NONE || mode == RRD_MEMORY_MODE_ALLOC))
         return entries;
 
     long page = (size_t)sysconf(_SC_PAGESIZE);
@@ -215,14 +267,18 @@ inline long align_entries_to_pagesize(RRD_MEMORY_MODE mode, long entries) {
     return entries;
 }
 
-static inline void last_collected_time_align(struct timeval *tv, int update_every) {
-    tv->tv_sec -= tv->tv_sec % update_every;
-    tv->tv_usec = 500000;
+static inline void last_collected_time_align(RRDSET *st) {
+    st->last_collected_time.tv_sec -= st->last_collected_time.tv_sec % st->update_every;
+
+    if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_STORE_FIRST)))
+        st->last_collected_time.tv_usec = 0;
+    else
+        st->last_collected_time.tv_usec = 500000;
 }
 
-static inline void last_updated_time_align(struct timeval *tv, int update_every) {
-    tv->tv_sec -= tv->tv_sec % update_every;
-    tv->tv_usec = 0;
+static inline void last_updated_time_align(RRDSET *st) {
+    st->last_updated.tv_sec -= st->last_updated.tv_sec % st->update_every;
+    st->last_updated.tv_usec = 0;
 }
 
 // ----------------------------------------------------------------------------
@@ -279,30 +335,36 @@ void rrdset_free(RRDSET *st) {
     // free directly allocated members
     freez(st->config_section);
 
-    if(st->rrd_memory_mode == RRD_MEMORY_MODE_SAVE || st->rrd_memory_mode == RRD_MEMORY_MODE_MAP) {
-        debug(D_RRD_CALLS, "Unmapping stats '%s'.", st->name);
-        munmap(st, st->memsize);
+    switch(st->rrd_memory_mode) {
+        case RRD_MEMORY_MODE_SAVE:
+        case RRD_MEMORY_MODE_MAP:
+        case RRD_MEMORY_MODE_RAM:
+            debug(D_RRD_CALLS, "Unmapping stats '%s'.", st->name);
+            munmap(st, st->memsize);
+            break;
+
+        case RRD_MEMORY_MODE_ALLOC:
+        case RRD_MEMORY_MODE_NONE:
+            freez(st);
+            break;
     }
-    else
-        freez(st);
 }
 
 void rrdset_save(RRDSET *st) {
-    RRDDIM *rd;
-
     rrdset_check_rdlock(st);
 
     // info("Saving chart '%s' ('%s')", st->id, st->name);
 
     if(st->rrd_memory_mode == RRD_MEMORY_MODE_SAVE) {
         debug(D_RRD_STATS, "Saving stats '%s' to '%s'.", st->name, st->cache_filename);
-        savememory(st->cache_filename, st, st->memsize);
+        memory_file_save(st->cache_filename, st, st->memsize);
     }
 
+    RRDDIM *rd;
     rrddim_foreach_read(rd, st) {
         if(likely(rd->rrd_memory_mode == RRD_MEMORY_MODE_SAVE)) {
             debug(D_RRD_STATS, "Saving dimension '%s' to '%s'.", rd->name, rd->cache_filename);
-            savememory(rd->cache_filename, rd, rd->memsize);
+            memory_file_save(rd->cache_filename, rd, rd->memsize);
         }
     }
 }
@@ -312,19 +374,23 @@ void rrdset_delete(RRDSET *st) {
 
     rrdset_check_rdlock(st);
 
-    // info("Deleting chart '%s' ('%s')", st->id, st->name);
+    info("Deleting chart '%s' ('%s') from disk...", st->id, st->name);
 
-    if(st->rrd_memory_mode == RRD_MEMORY_MODE_SAVE) {
-        debug(D_RRD_STATS, "Deleting stats '%s' to '%s'.", st->name, st->cache_filename);
-        unlink(st->cache_filename);
+    if(st->rrd_memory_mode == RRD_MEMORY_MODE_SAVE || st->rrd_memory_mode == RRD_MEMORY_MODE_MAP) {
+        info("Deleting chart header file '%s'.", st->cache_filename);
+        if(unlikely(unlink(st->cache_filename) == -1))
+            error("Cannot delete chart header file '%s'", st->cache_filename);
     }
 
     rrddim_foreach_read(rd, st) {
-        if(likely(rd->rrd_memory_mode == RRD_MEMORY_MODE_SAVE)) {
-            debug(D_RRD_STATS, "Deleting dimension '%s' to '%s'.", rd->name, rd->cache_filename);
-            unlink(rd->cache_filename);
+        if(likely(rd->rrd_memory_mode == RRD_MEMORY_MODE_SAVE || rd->rrd_memory_mode == RRD_MEMORY_MODE_MAP)) {
+            info("Deleting dimension file '%s'.", rd->cache_filename);
+            if(unlikely(unlink(rd->cache_filename) == -1))
+                error("Cannot delete dimension file '%s'", rd->cache_filename);
         }
     }
+
+    recursively_delete_dir(st->cache_dir, "left-over chart");
 }
 
 // ----------------------------------------------------------------------------
@@ -333,7 +399,7 @@ void rrdset_delete(RRDSET *st) {
 static inline RRDSET *rrdset_find_on_create(RRDHOST *host, const char *fullid) {
     RRDSET *st = rrdset_find(host, fullid);
     if(unlikely(st)) {
-        rrdset_flag_clear(st, RRDSET_FLAG_OBSOLETE);
+        rrdset_isnot_obsolete(st);
         debug(D_RRD_CALLS, "RRDSET '%s', already exists.", fullid);
         return st;
     }
@@ -341,7 +407,7 @@ static inline RRDSET *rrdset_find_on_create(RRDHOST *host, const char *fullid) {
     return NULL;
 }
 
-RRDSET *rrdset_create(
+RRDSET *rrdset_create_custom(
           RRDHOST *host
         , const char *type
         , const char *id
@@ -353,6 +419,8 @@ RRDSET *rrdset_create(
         , long priority
         , int update_every
         , RRDSET_TYPE chart_type
+        , RRD_MEMORY_MODE memory_mode
+        , long history_entries
 ) {
     if(!type || !type[0]) {
         fatal("Cannot create rrd stats without a type.");
@@ -395,11 +463,11 @@ RRDSET *rrdset_create(
     // ------------------------------------------------------------------------
     // get the options from the config, we need to create it
 
-    long rentries = config_get_number(config_section, "history", host->rrd_history_entries);
-    long entries = align_entries_to_pagesize(host->rrd_memory_mode, rentries);
+    long rentries = config_get_number(config_section, "history", history_entries);
+    long entries = align_entries_to_pagesize(memory_mode, rentries);
     if(entries != rentries) entries = config_set_number(config_section, "history", entries);
 
-    if(host->rrd_memory_mode == RRD_MEMORY_MODE_NONE && entries != rentries)
+    if(memory_mode == RRD_MEMORY_MODE_NONE && entries != rentries)
         entries = config_set_number(config_section, "history", 10);
 
     int enabled = config_get_boolean(config_section, "enabled", 1);
@@ -416,8 +484,14 @@ RRDSET *rrdset_create(
     debug(D_RRD_CALLS, "Creating RRD_STATS for '%s.%s'.", type, id);
 
     snprintfz(fullfilename, FILENAME_MAX, "%s/main.db", cache_dir);
-    if(host->rrd_memory_mode == RRD_MEMORY_MODE_SAVE || host->rrd_memory_mode == RRD_MEMORY_MODE_MAP) {
-        st = (RRDSET *) mymmap(fullfilename, size, ((host->rrd_memory_mode == RRD_MEMORY_MODE_MAP) ? MAP_SHARED : MAP_PRIVATE), 0);
+    if(memory_mode == RRD_MEMORY_MODE_SAVE || memory_mode == RRD_MEMORY_MODE_MAP || memory_mode == RRD_MEMORY_MODE_RAM) {
+        st = (RRDSET *) mymmap(
+                  (memory_mode == RRD_MEMORY_MODE_RAM)?NULL:fullfilename
+                , size
+                , ((memory_mode == RRD_MEMORY_MODE_MAP) ? MAP_SHARED : MAP_PRIVATE)
+                , 0
+        );
+
         if(st) {
             memset(&st->avl, 0, sizeof(avl));
             memset(&st->avlname, 0, sizeof(avl));
@@ -426,64 +500,68 @@ RRDSET *rrdset_create(
             memset(&st->rrdset_rwlock, 0, sizeof(netdata_rwlock_t));
 
             st->name = NULL;
+            st->config_section = NULL;
             st->type = NULL;
             st->family = NULL;
-            st->context = NULL;
             st->title = NULL;
             st->units = NULL;
+            st->context = NULL;
+            st->cache_dir = NULL;
             st->dimensions = NULL;
+            st->rrdfamily = NULL;
+            st->rrdhost = NULL;
             st->next = NULL;
             st->variables = NULL;
             st->alarms = NULL;
             st->flags = 0x00000000;
 
-            if(strcmp(st->magic, RRDSET_MAGIC) != 0) {
-                errno = 0;
-                info("Initializing file %s.", fullfilename);
-                memset(st, 0, size);
-            }
-            else if(strcmp(st->id, fullid) != 0) {
-                errno = 0;
-                error("File %s contents are not for chart %s. Clearing it.", fullfilename, fullid);
-                // munmap(st, size);
-                // st = NULL;
+            if(memory_mode == RRD_MEMORY_MODE_RAM) {
                 memset(st, 0, size);
             }
-            else if(st->memsize != size || st->entries != entries) {
-                errno = 0;
-                error("File %s does not have the desired size. Clearing it.", fullfilename);
-                memset(st, 0, size);
-            }
-            else if(st->update_every != update_every) {
-                errno = 0;
-                error("File %s does not have the desired update frequency. Clearing it.", fullfilename);
-                memset(st, 0, size);
-            }
-            else if((now - st->last_updated.tv_sec) > update_every * entries) {
-                errno = 0;
-                error("File %s is too old. Clearing it.", fullfilename);
-                memset(st, 0, size);
-            }
-            else if(st->last_updated.tv_sec > now + update_every) {
-                errno = 0;
-                error("File %s refers to the future. Clearing it.", fullfilename);
-                memset(st, 0, size);
-            }
-
-            // make sure the database is aligned
-            if(st->last_updated.tv_sec)
-                last_updated_time_align(&st->last_updated, update_every);
+            else {
+                if(strcmp(st->magic, RRDSET_MAGIC) != 0) {
+                    info("Initializing file %s.", fullfilename);
+                    memset(st, 0, size);
+                }
+                else if(strcmp(st->id, fullid) != 0) {
+                    error("File %s contents are not for chart %s. Clearing it.", fullfilename, fullid);
+                    // munmap(st, size);
+                    // st = NULL;
+                    memset(st, 0, size);
+                }
+                else if(st->memsize != size || st->entries != entries) {
+                    error("File %s does not have the desired size. Clearing it.", fullfilename);
+                    memset(st, 0, size);
+                }
+                else if(st->update_every != update_every) {
+                    error("File %s does not have the desired update frequency. Clearing it.", fullfilename);
+                    memset(st, 0, size);
+                }
+                else if((now - st->last_updated.tv_sec) > update_every * entries) {
+                    error("File %s is too old. Clearing it.", fullfilename);
+                    memset(st, 0, size);
+                }
+                else if(st->last_updated.tv_sec > now + update_every) {
+                    error("File %s refers to the future. Clearing it.", fullfilename);
+                    memset(st, 0, size);
+                }
 
+                // make sure the database is aligned
+                if(st->last_updated.tv_sec) {
+                    st->update_every = update_every;
+                    last_updated_time_align(st);
+                }
+            }
 
             // make sure we have the right memory mode
             // even if we cleared the memory
-            st->rrd_memory_mode = host->rrd_memory_mode;
+            st->rrd_memory_mode = memory_mode;
         }
     }
 
     if(unlikely(!st)) {
         st = callocz(1, size);
-        st->rrd_memory_mode = (host->rrd_memory_mode == RRD_MEMORY_MODE_NONE) ? RRD_MEMORY_MODE_NONE : RRD_MEMORY_MODE_RAM;
+        st->rrd_memory_mode = (memory_mode == RRD_MEMORY_MODE_NONE) ? RRD_MEMORY_MODE_NONE : RRD_MEMORY_MODE_ALLOC;
     }
 
     st->config_section = strdup(config_section);
@@ -519,6 +597,7 @@ RRDSET *rrdset_create(
     rrdset_flag_clear(st, RRDSET_FLAG_DETAIL);
     rrdset_flag_clear(st, RRDSET_FLAG_DEBUG);
     rrdset_flag_clear(st, RRDSET_FLAG_OBSOLETE);
+    rrdset_flag_clear(st, RRDSET_FLAG_EXPOSED_UPSTREAM);
 
     // if(!strcmp(st->id, "disk_util.dm-0")) {
     //     st->debug = 1;
@@ -536,6 +615,9 @@ RRDSET *rrdset_create(
     st->gap_when_lost_iterations_above = (int) (
             config_get_number(st->config_section, "gap when lost iterations above", RRD_DEFAULT_GAP_INTERPOLATIONS) + 2);
 
+    st->last_accessed_time = 0;
+    st->upstream_resync_time = 0;
+
     avl_init_lock(&st->dimensions_index, rrddim_compare);
     avl_init_lock(&st->variables_root_index, rrdvar_compare);
 
@@ -544,13 +626,8 @@ RRDSET *rrdset_create(
     if(name && *name) rrdset_set_name(st, name);
     else rrdset_set_name(st, id);
 
-    {
-        char varvalue[CONFIG_MAX_VALUE + 1];
-        char varvalue2[CONFIG_MAX_VALUE + 1];
-        snprintfz(varvalue, CONFIG_MAX_VALUE, "%s (%s)", title?title:"", st->name);
-        json_escape_string(varvalue2, varvalue, sizeof(varvalue2));
-        st->title = config_get(st->config_section, "title", varvalue2);
-    }
+    st->title = config_get(st->config_section, "title", title);
+    json_fix_string(st->title);
 
     st->rrdfamily = rrdfamily_create(host, st->family);
 
@@ -571,7 +648,7 @@ RRDSET *rrdset_create(
     rrdsetcalc_link_matching(st);
     rrdcalctemplate_link_matching(st);
 
-    rrdhost_cleanup_obsolete(host);
+    rrdhost_cleanup_obsolete_charts(host);
 
     rrdhost_unlock(host);
 
@@ -583,16 +660,10 @@ RRDSET *rrdset_create(
 // RRDSET - data collection iteration control
 
 inline void rrdset_next_usec_unfiltered(RRDSET *st, usec_t microseconds) {
-
-    if(unlikely(!st->last_collected_time.tv_sec)) {
-        // the first entry
-        microseconds = st->update_every * USEC_PER_SEC;
-    }
-    else if(unlikely(!microseconds)) {
-        // no dt given by the plugin
-        struct timeval now;
-        now_realtime_timeval(&now);
-        microseconds = dt_usec(&now, &st->last_collected_time);
+    if(unlikely(!st->last_collected_time.tv_sec || !microseconds || (st->counter % remote_clock_resync_iterations) == 0)) {
+        // call the full next_usec() function
+        rrdset_next_usec(st, microseconds);
+        return;
     }
 
     st->usec_since_last_update = microseconds;
@@ -612,52 +683,34 @@ inline void rrdset_next_usec(RRDSET *st, usec_t microseconds) {
     }
     else {
         // microseconds has the time since the last collection
-//#ifdef NETDATA_INTERNAL_CHECKS
-//        usec_t now_usec = timeval_usec(&now);
-//        usec_t last_usec = timeval_usec(&st->last_collected_time);
-//#endif
         susec_t since_last_usec = dt_usec_signed(&now, &st->last_collected_time);
 
         if(unlikely(since_last_usec < 0)) {
             // oops! the database is in the future
-            error("Database for chart '%s' on host '%s' is %lld microseconds in the future. Adjusting it to current time.", st->id, st->rrdhost->hostname, -since_last_usec);
+            info("RRD database for chart '%s' on host '%s' is %0.5Lf secs in the future. Adjusting it to current time.", st->id, st->rrdhost->hostname, (long double)-since_last_usec / USEC_PER_SEC);
 
             st->last_collected_time.tv_sec  = now.tv_sec - st->update_every;
             st->last_collected_time.tv_usec = now.tv_usec;
-            last_collected_time_align(&st->last_collected_time, st->update_every);
+            last_collected_time_align(st);
 
             st->last_updated.tv_sec  = now.tv_sec - st->update_every;
             st->last_updated.tv_usec = now.tv_usec;
-            last_updated_time_align(&st->last_updated, st->update_every);
+            last_updated_time_align(st);
 
             microseconds    = st->update_every * USEC_PER_SEC;
-            since_last_usec = st->update_every * USEC_PER_SEC;
-        }
-
-        // verify the microseconds given is good
-        if(unlikely(microseconds > (usec_t)since_last_usec)) {
-            debug(D_RRD_CALLS, "dt %llu usec given is too big - it leads %llu usec to the future, for chart '%s' (%s).", microseconds, microseconds - (usec_t)since_last_usec, st->name, st->id);
-
-//#ifdef NETDATA_INTERNAL_CHECKS
-//            if(unlikely(last_usec + microseconds > now_usec + 1000))
-//                error("dt %llu usec given is too big - it leads %llu usec to the future, for chart '%s' (%s).", microseconds, microseconds - (usec_t)since_last_usec, st->name, st->id);
-//#endif
-
-            microseconds = (usec_t)since_last_usec;
         }
-        else if(unlikely(microseconds < (usec_t)since_last_usec * 0.8)) {
-            debug(D_RRD_CALLS, "dt %llu usec given is too small - expected %llu usec up to -20%%, for chart '%s' (%s).", microseconds, (usec_t)since_last_usec, st->name, st->id);
+        else if(unlikely((usec_t)since_last_usec > (usec_t)(st->update_every * 10 * USEC_PER_SEC))) {
+            // oops! the database is too far behind
+            info("RRD database for chart '%s' on host '%s' is %0.5Lf secs in the past. Adjusting it to current time.", st->id, st->rrdhost->hostname, (long double)since_last_usec / USEC_PER_SEC);
 
-//#ifdef NETDATA_INTERNAL_CHECKS
-//            error("dt %llu usec given is too small - expected %llu usec up to -20%%, for chart '%s' (%s).", microseconds, (usec_t)since_last_usec, st->name, st->id);
-//#endif
             microseconds = (usec_t)since_last_usec;
         }
     }
-    debug(D_RRD_CALLS, "rrdset_next_usec() for chart %s with microseconds %llu", st->name, microseconds);
 
-    if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
-        debug(D_RRD_STATS, "%s: NEXT: %llu microseconds", st->name, microseconds);
+    #ifdef NETDATA_INTERNAL_CHECKS
+    debug(D_RRD_CALLS, "rrdset_next_usec() for chart %s with microseconds %llu", st->name, microseconds);
+    rrdset_debug(st, "NEXT: %llu microseconds", microseconds);
+    #endif
 
     st->usec_since_last_update = microseconds;
 }
@@ -666,9 +719,17 @@ inline void rrdset_next_usec(RRDSET *st, usec_t microseconds) {
 // ----------------------------------------------------------------------------
 // RRDSET - process the collected values for all dimensions of a chart
 
-static inline void rrdset_init_last_collected_time(RRDSET *st) {
+static inline usec_t rrdset_init_last_collected_time(RRDSET *st) {
     now_realtime_timeval(&st->last_collected_time);
-    last_collected_time_align(&st->last_collected_time, st->update_every);
+    last_collected_time_align(st);
+
+    usec_t last_collect_ut = st->last_collected_time.tv_sec * USEC_PER_SEC + st->last_collected_time.tv_usec;
+
+    #ifdef NETDATA_INTERNAL_CHECKS
+    rrdset_debug(st, "initialized last collected time to %0.3Lf", (long double)last_collect_ut / USEC_PER_SEC);
+    #endif
+
+    return last_collect_ut;
 }
 
 static inline usec_t rrdset_update_last_collected_time(RRDSET *st) {
@@ -676,17 +737,42 @@ static inline usec_t rrdset_update_last_collected_time(RRDSET *st) {
     usec_t ut = last_collect_ut + st->usec_since_last_update;
     st->last_collected_time.tv_sec = (time_t) (ut / USEC_PER_SEC);
     st->last_collected_time.tv_usec = (suseconds_t) (ut % USEC_PER_SEC);
+
+    #ifdef NETDATA_INTERNAL_CHECKS
+    rrdset_debug(st, "updated last collected time to %0.3Lf", (long double)last_collect_ut / USEC_PER_SEC);
+    #endif
+
     return last_collect_ut;
 }
 
-static inline void rrdset_init_last_updated_time(RRDSET *st) {
+static inline usec_t rrdset_init_last_updated_time(RRDSET *st) {
     // copy the last collected time to last updated time
     st->last_updated.tv_sec  = st->last_collected_time.tv_sec;
     st->last_updated.tv_usec = st->last_collected_time.tv_usec;
-    last_updated_time_align(&st->last_updated, st->update_every);
+
+    if(rrdset_flag_check(st, RRDSET_FLAG_STORE_FIRST))
+        st->last_updated.tv_sec -= st->update_every;
+
+    last_updated_time_align(st);
+
+    usec_t last_updated_ut = st->last_updated.tv_sec * USEC_PER_SEC + st->last_updated.tv_usec;
+
+    #ifdef NETDATA_INTERNAL_CHECKS
+    rrdset_debug(st, "initialized last updated time to %0.3Lf", (long double)last_updated_ut / USEC_PER_SEC);
+    #endif
+
+    return last_updated_ut;
 }
 
 static inline void rrdset_done_push_exclusive(RRDSET *st) {
+//    usec_t update_every_ut = st->update_every * USEC_PER_SEC; // st->update_every in microseconds
+//
+//    if(unlikely(st->usec_since_last_update > update_every_ut * remote_clock_resync_iterations)) {
+//        error("Chart '%s' was last collected %llu usec before. Resetting it.", st->id, st->usec_since_last_update);
+//        rrdset_reset(st);
+//        st->usec_since_last_update = update_every_ut;
+//    }
+
     if(unlikely(!st->last_collected_time.tv_sec)) {
         // it is the first entry
         // set the last_collected_time to now
@@ -705,6 +791,235 @@ static inline void rrdset_done_push_exclusive(RRDSET *st) {
     rrdset_unlock(st);
 }
 
+
+static inline size_t rrdset_done_interpolate(
+        RRDSET *st
+        , usec_t update_every_ut
+        , usec_t last_stored_ut
+        , usec_t next_store_ut
+        , usec_t last_collect_ut
+        , usec_t now_collect_ut
+        , char store_this_entry
+        , uint32_t storage_flags
+) {
+    RRDDIM *rd;
+
+    size_t stored_entries = 0;     // the number of entries we have stored in the db, during this call to rrdset_done()
+
+    usec_t first_ut = last_stored_ut, last_ut = 0;
+    ssize_t iterations = (ssize_t)((now_collect_ut - last_stored_ut) / (update_every_ut));
+    if((now_collect_ut % (update_every_ut)) == 0) iterations++;
+
+    size_t counter = st->counter;
+    long current_entry = st->current_entry;
+
+    for( ; next_store_ut <= now_collect_ut ; last_collect_ut = next_store_ut, next_store_ut += update_every_ut, iterations-- ) {
+
+        #ifdef NETDATA_INTERNAL_CHECKS
+        if(iterations < 0) { error("INTERNAL CHECK: %s: iterations calculation wrapped! first_ut = %llu, last_stored_ut = %llu, next_store_ut = %llu, now_collect_ut = %llu", st->name, first_ut, last_stored_ut, next_store_ut, now_collect_ut); }
+        rrdset_debug(st, "last_stored_ut = %0.3Lf (last updated time)", (long double)last_stored_ut/USEC_PER_SEC);
+        rrdset_debug(st, "next_store_ut  = %0.3Lf (next interpolation point)", (long double)next_store_ut/USEC_PER_SEC);
+        #endif
+
+        last_ut = next_store_ut;
+
+        rrddim_foreach_read(rd, st) {
+            calculated_number new_value;
+
+            switch(rd->algorithm) {
+                case RRD_ALGORITHM_INCREMENTAL:
+                    new_value = (calculated_number)
+                            (      rd->calculated_value
+                                   * (calculated_number)(next_store_ut - last_collect_ut)
+                                   / (calculated_number)(now_collect_ut - last_collect_ut)
+                            );
+
+                    #ifdef NETDATA_INTERNAL_CHECKS
+                    rrdset_debug(st, "%s: CALC2 INC "
+                                CALCULATED_NUMBER_FORMAT " = "
+                                CALCULATED_NUMBER_FORMAT
+                                " * (%llu - %llu)"
+                                " / (%llu - %llu)"
+                              , rd->name
+                              , new_value
+                              , rd->calculated_value
+                              , next_store_ut, last_collect_ut
+                              , now_collect_ut, last_collect_ut
+                    );
+                    #endif
+
+                    rd->calculated_value -= new_value;
+                    new_value += rd->last_calculated_value;
+                    rd->last_calculated_value = 0;
+                    new_value /= (calculated_number)st->update_every;
+
+                    if(unlikely(next_store_ut - last_stored_ut < update_every_ut)) {
+
+                        #ifdef NETDATA_INTERNAL_CHECKS
+                        rrdset_debug(st, "%s: COLLECTION POINT IS SHORT " CALCULATED_NUMBER_FORMAT " - EXTRAPOLATING",
+                                    rd->name
+                                  , (calculated_number)(next_store_ut - last_stored_ut)
+                        );
+                        #endif
+
+                        new_value = new_value * (calculated_number)(st->update_every * USEC_PER_SEC) / (calculated_number)(next_store_ut - last_stored_ut);
+                    }
+                    break;
+
+                case RRD_ALGORITHM_ABSOLUTE:
+                case RRD_ALGORITHM_PCENT_OVER_ROW_TOTAL:
+                case RRD_ALGORITHM_PCENT_OVER_DIFF_TOTAL:
+                default:
+                    if(iterations == 1) {
+                        // this is the last iteration
+                        // do not interpolate
+                        // just show the calculated value
+
+                        new_value = rd->calculated_value;
+                    }
+                    else {
+                        // we have missed an update
+                        // interpolate in the middle values
+
+                        new_value = (calculated_number)
+                                (   (     (rd->calculated_value - rd->last_calculated_value)
+                                          * (calculated_number)(next_store_ut - last_collect_ut)
+                                          / (calculated_number)(now_collect_ut - last_collect_ut)
+                                    )
+                                    +  rd->last_calculated_value
+                                );
+
+                        #ifdef NETDATA_INTERNAL_CHECKS
+                        rrdset_debug(st, "%s: CALC2 DEF "
+                                    CALCULATED_NUMBER_FORMAT " = ((("
+                                            "(" CALCULATED_NUMBER_FORMAT " - " CALCULATED_NUMBER_FORMAT ")"
+                                            " * %llu"
+                                            " / %llu) + " CALCULATED_NUMBER_FORMAT
+                                  , rd->name
+                                  , new_value
+                                  , rd->calculated_value, rd->last_calculated_value
+                                  , (next_store_ut - first_ut)
+                                  , (now_collect_ut - first_ut), rd->last_calculated_value
+                        );
+                        #endif
+                    }
+                    break;
+            }
+
+            if(unlikely(!store_this_entry)) {
+                rd->values[current_entry] = SN_EMPTY_SLOT; //pack_storage_number(0, SN_NOT_EXISTS);
+                continue;
+            }
+
+            if(likely(rd->updated && rd->collections_counter > 1 && iterations < st->gap_when_lost_iterations_above)) {
+                rd->values[current_entry] = pack_storage_number(new_value, storage_flags );
+                rd->last_stored_value = new_value;
+
+                #ifdef NETDATA_INTERNAL_CHECKS
+                rrdset_debug(st, "%s: STORE[%ld] "
+                            CALCULATED_NUMBER_FORMAT " = " CALCULATED_NUMBER_FORMAT
+                          , rd->name
+                          , current_entry
+                          , unpack_storage_number(rd->values[current_entry]), new_value
+                );
+                #endif
+
+            }
+            else {
+
+                #ifdef NETDATA_INTERNAL_CHECKS
+                rrdset_debug(st, "%s: STORE[%ld] = NON EXISTING "
+                          , rd->name
+                          , current_entry
+                );
+                #endif
+
+                rd->values[current_entry] = SN_EMPTY_SLOT; // pack_storage_number(0, SN_NOT_EXISTS);
+                rd->last_stored_value = NAN;
+            }
+
+            stored_entries++;
+
+            #ifdef NETDATA_INTERNAL_CHECKS
+            if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG))) {
+                calculated_number t1 = new_value * (calculated_number)rd->multiplier / (calculated_number)rd->divisor;
+                calculated_number t2 = unpack_storage_number(rd->values[current_entry]);
+
+                calculated_number accuracy = accuracy_loss(t1, t2);
+                debug(D_RRD_STATS, "%s/%s: UNPACK[%ld] = " CALCULATED_NUMBER_FORMAT " FLAGS=0x%08x (original = " CALCULATED_NUMBER_FORMAT ", accuracy loss = " CALCULATED_NUMBER_FORMAT "%%%s)"
+                      , st->id, rd->name
+                      , current_entry
+                      , t2
+                      , get_storage_number_flags(rd->values[current_entry])
+                      , t1
+                      , accuracy
+                      , (accuracy > ACCURACY_LOSS) ? " **TOO BIG** " : ""
+                );
+
+                rd->collected_volume += t1;
+                rd->stored_volume += t2;
+
+                accuracy = accuracy_loss(rd->collected_volume, rd->stored_volume);
+                debug(D_RRD_STATS, "%s/%s: VOLUME[%ld] = " CALCULATED_NUMBER_FORMAT ", calculated  = " CALCULATED_NUMBER_FORMAT ", accuracy loss = " CALCULATED_NUMBER_FORMAT "%%%s"
+                      , st->id, rd->name
+                      , current_entry
+                      , rd->stored_volume
+                      , rd->collected_volume
+                      , accuracy
+                      , (accuracy > ACCURACY_LOSS) ? " **TOO BIG** " : ""
+                );
+            }
+            #endif
+        }
+        // reset the storage flags for the next point, if any;
+        storage_flags = SN_EXISTS;
+
+        counter++;
+        current_entry = ((current_entry + 1) >= st->entries) ? 0 : current_entry + 1;
+        last_stored_ut = next_store_ut;
+    }
+
+    st->counter = counter;
+    st->current_entry = current_entry;
+
+    if(likely(last_ut)) {
+        st->last_updated.tv_sec = (time_t) (last_ut / USEC_PER_SEC);
+        st->last_updated.tv_usec = 0;
+    }
+
+    return stored_entries;
+}
+
+static inline void rrdset_done_fill_the_gap(RRDSET *st) {
+    usec_t update_every_ut = st->update_every * USEC_PER_SEC;
+    usec_t now_collect_ut  = st->last_collected_time.tv_sec * USEC_PER_SEC + st->last_collected_time.tv_usec;
+
+    long c = 0, entries = st->entries;
+    RRDDIM *rd;
+    rrddim_foreach_read(rd, st) {
+        usec_t next_store_ut = (st->last_updated.tv_sec + st->update_every) * USEC_PER_SEC;
+        long current_entry = st->current_entry;
+
+        for(c = 0; c < entries && next_store_ut <= now_collect_ut ; next_store_ut += update_every_ut, c++) {
+            rd->values[current_entry] = SN_EMPTY_SLOT;
+            current_entry = ((current_entry + 1) >= entries) ? 0 : current_entry + 1;
+
+            #ifdef NETDATA_INTERNAL_CHECKS
+            rrdset_debug(st, "%s: STORE[%ld] = NON EXISTING (FILLED THE GAP)", rd->name, current_entry);
+            #endif
+        }
+    }
+
+    if(c > 0) {
+        c--;
+        st->last_updated.tv_sec += c * st->update_every;
+
+        st->current_entry += c;
+        if(st->current_entry >= st->entries)
+            st->current_entry -= st->entries;
+    }
+}
+
 void rrdset_done(RRDSET *st) {
     if(unlikely(netdata_exit)) return;
 
@@ -726,9 +1041,6 @@ void rrdset_done(RRDSET *st) {
             store_this_entry = 1,   // boolean: 1 = store this entry, 0 = don't store this entry
             first_entry = 0;        // boolean: 1 = this is the first entry seen for this chart, 0 = all other entries
 
-    unsigned int
-            stored_entries = 0;     // the number of entries we have stored in the db, during this call to rrdset_done()
-
     usec_t
             last_collect_ut,        // the timestamp in microseconds, of the last collected value
             now_collect_ut,         // the timestamp in microseconds, of this collected value (this is NOW)
@@ -742,42 +1054,33 @@ void rrdset_done(RRDSET *st) {
     // a read lock is OK here
     rrdset_rdlock(st);
 
-/*
-    // enable the chart, if it was disabled
-    if(unlikely(rrd_delete_unupdated_dimensions) && !st->enabled)
-        st->enabled = 1;
-*/
-
     if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_OBSOLETE))) {
         error("Chart '%s' has the OBSOLETE flag set, but it is collected.", st->id);
-        rrdset_flag_clear(st, RRDSET_FLAG_OBSOLETE);
+        rrdset_isnot_obsolete(st);
     }
 
     // check if the chart has a long time to be updated
     if(unlikely(st->usec_since_last_update > st->entries * update_every_ut)) {
-        info("%s: took too long to be updated (%0.3Lf secs). Resetting it.", st->name, (long double)(st->usec_since_last_update / 1000000.0));
+        info("host '%s', chart %s: took too long to be updated (%0.3Lf secs). Resetting it.", st->rrdhost->hostname, st->name, (long double)st->usec_since_last_update / USEC_PER_SEC);
         rrdset_reset(st);
         st->usec_since_last_update = update_every_ut;
+        store_this_entry = 0;
         first_entry = 1;
     }
 
-    if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
-        debug(D_RRD_STATS, "%s: microseconds since last update: %llu", st->name, st->usec_since_last_update);
+    #ifdef NETDATA_INTERNAL_CHECKS
+    rrdset_debug(st, "microseconds since last update: %llu", st->usec_since_last_update);
+    #endif
 
     // set last_collected_time
     if(unlikely(!st->last_collected_time.tv_sec)) {
         // it is the first entry
         // set the last_collected_time to now
-        rrdset_init_last_collected_time(st);
-
-        last_collect_ut = st->last_collected_time.tv_sec * USEC_PER_SEC + st->last_collected_time.tv_usec - update_every_ut;
+        last_collect_ut = rrdset_init_last_collected_time(st) - update_every_ut;
 
         // the first entry should not be stored
         store_this_entry = 0;
         first_entry = 1;
-
-        if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
-            debug(D_RRD_STATS, "%s: has not set last_collected_time. Setting it now. Will not store the next entry.", st->name);
     }
     else {
         // it is not the first entry
@@ -795,9 +1098,6 @@ void rrdset_done(RRDSET *st) {
         // the first entry should not be stored
         store_this_entry = 0;
         first_entry = 1;
-
-        if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
-            debug(D_RRD_STATS, "%s: initializing last_updated to last_collected_time - %llu microseconds. Will not store the next entry.", st->name, st->usec_since_last_update);
     }
 
     // check if we will re-write the entire data set
@@ -817,27 +1117,48 @@ void rrdset_done(RRDSET *st) {
     // last_stored_ut = the last time we added a value to the storage
     // now_collect_ut = the time the current value has been collected
     // next_store_ut  = the time of the next interpolation point
-    last_stored_ut = st->last_updated.tv_sec * USEC_PER_SEC + st->last_updated.tv_usec;
     now_collect_ut = st->last_collected_time.tv_sec * USEC_PER_SEC + st->last_collected_time.tv_usec;
+    last_stored_ut = st->last_updated.tv_sec * USEC_PER_SEC + st->last_updated.tv_usec;
     next_store_ut  = (st->last_updated.tv_sec + st->update_every) * USEC_PER_SEC;
 
-    if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG))) {
-        debug(D_RRD_STATS, "%s: last_collect_ut = %0.3Lf (last collection time)", st->name, (long double)last_collect_ut/1000000.0);
-        debug(D_RRD_STATS, "%s: now_collect_ut  = %0.3Lf (current collection time)", st->name, (long double)now_collect_ut/1000000.0);
-        debug(D_RRD_STATS, "%s: last_stored_ut  = %0.3Lf (last updated time)", st->name, (long double)last_stored_ut/1000000.0);
-        debug(D_RRD_STATS, "%s: next_store_ut   = %0.3Lf (next interpolation point)", st->name, (long double)next_store_ut/1000000.0);
-    }
-
     if(unlikely(!st->counter_done)) {
-        store_this_entry = 0;
-        if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
-            debug(D_RRD_STATS, "%s: Will not store the next entry.", st->name);
+        // if we have not collected metrics this session (st->counter_done == 0)
+        // and we have collected metrics for this chart in the past (st->counter != 0)
+        // fill the gap (the chart has been just loaded from disk)
+        if(unlikely(st->counter)) {
+            rrdset_done_fill_the_gap(st);
+            last_stored_ut = st->last_updated.tv_sec * USEC_PER_SEC + st->last_updated.tv_usec;
+            next_store_ut  = (st->last_updated.tv_sec + st->update_every) * USEC_PER_SEC;
+        }
+
+        if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_STORE_FIRST))) {
+            store_this_entry = 1;
+            last_collect_ut = next_store_ut - update_every_ut;
+
+            #ifdef NETDATA_INTERNAL_CHECKS
+            rrdset_debug(st, "Fixed first entry.");
+            #endif
+        }
+        else {
+            store_this_entry = 0;
+
+            #ifdef NETDATA_INTERNAL_CHECKS
+            rrdset_debug(st, "Will not store the next entry.");
+            #endif
+        }
     }
     st->counter_done++;
 
     if(unlikely(st->rrdhost->rrdpush_enabled))
         rrdset_done_push(st);
 
+    #ifdef NETDATA_INTERNAL_CHECKS
+    rrdset_debug(st, "last_collect_ut = %0.3Lf (last collection time)", (long double)last_collect_ut/USEC_PER_SEC);
+    rrdset_debug(st, "now_collect_ut  = %0.3Lf (current collection time)", (long double)now_collect_ut/USEC_PER_SEC);
+    rrdset_debug(st, "last_stored_ut  = %0.3Lf (last updated time)", (long double)last_stored_ut/USEC_PER_SEC);
+    rrdset_debug(st, "next_store_ut   = %0.3Lf (next interpolation point)", (long double)next_store_ut/USEC_PER_SEC);
+    #endif
+
     // calculate totals and count the dimensions
     int dimensions = 0;
     st->collected_total = 0;
@@ -859,18 +1180,19 @@ void rrdset_done(RRDSET *st) {
             continue;
         }
 
-        if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
-            debug(D_RRD_STATS, "%s/%s: START "
-                    " last_collected_value = " COLLECTED_NUMBER_FORMAT
-                    " collected_value = " COLLECTED_NUMBER_FORMAT
-                    " last_calculated_value = " CALCULATED_NUMBER_FORMAT
-                    " calculated_value = " CALCULATED_NUMBER_FORMAT
-                                      , st->id, rd->name
+        #ifdef NETDATA_INTERNAL_CHECKS
+        rrdset_debug(st, "%s: START "
+                " last_collected_value = " COLLECTED_NUMBER_FORMAT
+                " collected_value = " COLLECTED_NUMBER_FORMAT
+                " last_calculated_value = " CALCULATED_NUMBER_FORMAT
+                " calculated_value = " CALCULATED_NUMBER_FORMAT
+                                      , rd->name
                                       , rd->last_collected_value
                                       , rd->collected_value
                                       , rd->last_calculated_value
                                       , rd->calculated_value
-            );
+        );
+        #endif
 
         switch(rd->algorithm) {
             case RRD_ALGORITHM_ABSOLUTE:
@@ -878,18 +1200,20 @@ void rrdset_done(RRDSET *st) {
                                        * (calculated_number)rd->multiplier
                                        / (calculated_number)rd->divisor;
 
-                if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
-                    debug(D_RRD_STATS, "%s/%s: CALC ABS/ABS-NO-IN "
+                #ifdef NETDATA_INTERNAL_CHECKS
+                rrdset_debug(st, "%s: CALC ABS/ABS-NO-IN "
                             CALCULATED_NUMBER_FORMAT " = "
                             COLLECTED_NUMBER_FORMAT
                             " * " CALCULATED_NUMBER_FORMAT
                             " / " CALCULATED_NUMBER_FORMAT
-                          , st->id, rd->name
+                          , rd->name
                           , rd->calculated_value
                           , rd->collected_value
                           , (calculated_number)rd->multiplier
                           , (calculated_number)rd->divisor
-                    );
+                );
+                #endif
+
                 break;
 
             case RRD_ALGORITHM_PCENT_OVER_ROW_TOTAL:
@@ -903,16 +1227,18 @@ void rrdset_done(RRDSET *st) {
                             * (calculated_number)rd->collected_value
                             / (calculated_number)st->collected_total;
 
-                if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
-                    debug(D_RRD_STATS, "%s/%s: CALC PCENT-ROW "
+                #ifdef NETDATA_INTERNAL_CHECKS
+                rrdset_debug(st, "%s: CALC PCENT-ROW "
                             CALCULATED_NUMBER_FORMAT " = 100"
-                                    " * " COLLECTED_NUMBER_FORMAT
+                            " * " COLLECTED_NUMBER_FORMAT
                             " / " COLLECTED_NUMBER_FORMAT
-                          , st->id, rd->name
+                          , rd->name
                           , rd->calculated_value
                           , rd->collected_value
                           , st->collected_total
-                    );
+                );
+                #endif
+
                 break;
 
             case RRD_ALGORITHM_INCREMENTAL:
@@ -940,19 +1266,21 @@ void rrdset_done(RRDSET *st) {
                         * (calculated_number)rd->multiplier
                         / (calculated_number)rd->divisor;
 
-                if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
-                    debug(D_RRD_STATS, "%s/%s: CALC INC PRE "
+                #ifdef NETDATA_INTERNAL_CHECKS
+                rrdset_debug(st, "%s: CALC INC PRE "
                             CALCULATED_NUMBER_FORMAT " = ("
                             COLLECTED_NUMBER_FORMAT " - " COLLECTED_NUMBER_FORMAT
                             ")"
                                     " * " CALCULATED_NUMBER_FORMAT
                             " / " CALCULATED_NUMBER_FORMAT
-                          , st->id, rd->name
+                          , rd->name
                           , rd->calculated_value
                           , rd->collected_value, rd->last_collected_value
                           , (calculated_number)rd->multiplier
                           , (calculated_number)rd->divisor
-                    );
+                );
+                #endif
+
                 break;
 
             case RRD_ALGORITHM_PCENT_OVER_DIFF_TOTAL:
@@ -967,7 +1295,8 @@ void rrdset_done(RRDSET *st) {
                     debug(D_RRD_STATS, "%s.%s: RESET or OVERFLOW. Last collected value = " COLLECTED_NUMBER_FORMAT ", current = " COLLECTED_NUMBER_FORMAT
                           , st->name, rd->name
                           , rd->last_collected_value
-                          , rd->collected_value);
+                          , rd->collected_value
+                    );
 
                     if(!(rrddim_flag_check(rd, RRDDIM_FLAG_DONT_DETECT_RESETS_OR_OVERFLOWS)))
                         storage_flags = SN_EXISTS_RESET;
@@ -985,16 +1314,18 @@ void rrdset_done(RRDSET *st) {
                             * (calculated_number)(rd->collected_value - rd->last_collected_value)
                             / (calculated_number)(st->collected_total - st->last_collected_total);
 
-                if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
-                    debug(D_RRD_STATS, "%s/%s: CALC PCENT-DIFF "
+                #ifdef NETDATA_INTERNAL_CHECKS
+                rrdset_debug(st, "%s: CALC PCENT-DIFF "
                             CALCULATED_NUMBER_FORMAT " = 100"
-                                    " * (" COLLECTED_NUMBER_FORMAT " - " COLLECTED_NUMBER_FORMAT ")"
-                                    " / (" COLLECTED_NUMBER_FORMAT " - " COLLECTED_NUMBER_FORMAT ")"
-                          , st->id, rd->name
+                            " * (" COLLECTED_NUMBER_FORMAT " - " COLLECTED_NUMBER_FORMAT ")"
+                            " / (" COLLECTED_NUMBER_FORMAT " - " COLLECTED_NUMBER_FORMAT ")"
+                          , rd->name
                           , rd->calculated_value
                           , rd->collected_value, rd->last_collected_value
                           , st->collected_total, st->last_collected_total
-                    );
+                );
+                #endif
+
                 break;
 
             default:
@@ -1002,203 +1333,53 @@ void rrdset_done(RRDSET *st) {
                 // it gets noticed when we add new types
                 rd->calculated_value = 0;
 
-                if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
-                    debug(D_RRD_STATS, "%s/%s: CALC "
+                #ifdef NETDATA_INTERNAL_CHECKS
+                rrdset_debug(st, "%s: CALC "
                             CALCULATED_NUMBER_FORMAT " = 0"
-                          , st->id, rd->name
+                          , rd->name
                           , rd->calculated_value
-                    );
+                );
+                #endif
+
                 break;
         }
 
-        if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
-            debug(D_RRD_STATS, "%s/%s: PHASE2 "
+        #ifdef NETDATA_INTERNAL_CHECKS
+        rrdset_debug(st, "%s: PHASE2 "
                     " last_collected_value = " COLLECTED_NUMBER_FORMAT
                     " collected_value = " COLLECTED_NUMBER_FORMAT
                     " last_calculated_value = " CALCULATED_NUMBER_FORMAT
                     " calculated_value = " CALCULATED_NUMBER_FORMAT
-                                      , st->id, rd->name
+                                      , rd->name
                                       , rd->last_collected_value
                                       , rd->collected_value
                                       , rd->last_calculated_value
                                       , rd->calculated_value
-            );
+        );
+        #endif
 
     }
 
     // at this point we have all the calculated values ready
     // it is now time to interpolate values on a second boundary
 
+#ifdef NETDATA_INTERNAL_CHECKS
     if(unlikely(now_collect_ut < next_store_ut)) {
         // this is collected in the same interpolation point
-
-        if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
-            debug(D_RRD_STATS, "%s: THIS IS IN THE SAME INTERPOLATION POINT", st->name);
-
-//#ifdef NETDATA_INTERNAL_CHECKS
-//        info("%s is collected in the same interpolation point: short by %llu microseconds", st->name, next_store_ut - now_collect_ut);
-//#endif
-    }
-
-    usec_t first_ut = last_stored_ut;
-    long long iterations = (now_collect_ut - last_stored_ut) / (update_every_ut);
-    if((now_collect_ut % (update_every_ut)) == 0) iterations++;
-
-    for( ; next_store_ut <= now_collect_ut ; last_collect_ut = next_store_ut, next_store_ut += update_every_ut, iterations-- ) {
-//#ifdef NETDATA_INTERNAL_CHECKS
-//        if(iterations < 0) { error("%s: iterations calculation wrapped! first_ut = %llu, last_stored_ut = %llu, next_store_ut = %llu, now_collect_ut = %llu", st->name, first_ut, last_stored_ut, next_store_ut, now_collect_ut); }
-//#endif
-
-        if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG))) {
-            debug(D_RRD_STATS, "%s: last_stored_ut = %0.3Lf (last updated time)", st->name, (long double)last_stored_ut/1000000.0);
-            debug(D_RRD_STATS, "%s: next_store_ut  = %0.3Lf (next interpolation point)", st->name, (long double)next_store_ut/1000000.0);
-        }
-
-        st->last_updated.tv_sec = (time_t) (next_store_ut / USEC_PER_SEC);
-        st->last_updated.tv_usec = 0;
-
-        rrddim_foreach_read(rd, st) {
-            calculated_number new_value;
-
-            switch(rd->algorithm) {
-                case RRD_ALGORITHM_INCREMENTAL:
-                    new_value = (calculated_number)
-                            (      rd->calculated_value
-                                   * (calculated_number)(next_store_ut - last_collect_ut)
-                                   / (calculated_number)(now_collect_ut - last_collect_ut)
-                            );
-
-                    if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
-                        debug(D_RRD_STATS, "%s/%s: CALC2 INC "
-                                CALCULATED_NUMBER_FORMAT " = "
-                                CALCULATED_NUMBER_FORMAT
-                                " * %llu"
-                                        " / %llu"
-                              , st->id, rd->name
-                              , new_value
-                              , rd->calculated_value
-                              , (next_store_ut - last_stored_ut)
-                              , (now_collect_ut - last_stored_ut)
-                        );
-
-                    rd->calculated_value -= new_value;
-                    new_value += rd->last_calculated_value;
-                    rd->last_calculated_value = 0;
-                    new_value /= (calculated_number)st->update_every;
-
-                    if(unlikely(next_store_ut - last_stored_ut < update_every_ut)) {
-                        if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
-                            debug(D_RRD_STATS, "%s/%s: COLLECTION POINT IS SHORT " CALCULATED_NUMBER_FORMAT " - EXTRAPOLATING",
-                                    st->id, rd->name
-                                  , (calculated_number)(next_store_ut - last_stored_ut)
-                            );
-                        new_value = new_value * (calculated_number)(st->update_every * 1000000) / (calculated_number)(next_store_ut - last_stored_ut);
-                    }
-                    break;
-
-                case RRD_ALGORITHM_ABSOLUTE:
-                case RRD_ALGORITHM_PCENT_OVER_ROW_TOTAL:
-                case RRD_ALGORITHM_PCENT_OVER_DIFF_TOTAL:
-                default:
-                    if(iterations == 1) {
-                        // this is the last iteration
-                        // do not interpolate
-                        // just show the calculated value
-
-                        new_value = rd->calculated_value;
-                    }
-                    else {
-                        // we have missed an update
-                        // interpolate in the middle values
-
-                        new_value = (calculated_number)
-                                (   (     (rd->calculated_value - rd->last_calculated_value)
-                                          * (calculated_number)(next_store_ut - last_collect_ut)
-                                          / (calculated_number)(now_collect_ut - last_collect_ut)
-                                    )
-                                    +  rd->last_calculated_value
-                                );
-
-                        if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
-                            debug(D_RRD_STATS, "%s/%s: CALC2 DEF "
-                                    CALCULATED_NUMBER_FORMAT " = ((("
-                                            "(" CALCULATED_NUMBER_FORMAT " - " CALCULATED_NUMBER_FORMAT ")"
-                                            " * %llu"
-                                            " / %llu) + " CALCULATED_NUMBER_FORMAT
-                                  , st->id, rd->name
-                                  , new_value
-                                  , rd->calculated_value, rd->last_calculated_value
-                                  , (next_store_ut - first_ut)
-                                  , (now_collect_ut - first_ut), rd->last_calculated_value
-                            );
-                    }
-                    break;
-            }
-
-            if(unlikely(!store_this_entry)) {
-                rd->values[st->current_entry] = pack_storage_number(0, SN_NOT_EXISTS);
-                continue;
-            }
-
-            if(likely(rd->updated && rd->collections_counter > 1 && iterations < st->gap_when_lost_iterations_above)) {
-                rd->values[st->current_entry] = pack_storage_number(new_value, storage_flags );
-                rd->last_stored_value = new_value;
-
-                if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
-                    debug(D_RRD_STATS, "%s/%s: STORE[%ld] "
-                            CALCULATED_NUMBER_FORMAT " = " CALCULATED_NUMBER_FORMAT
-                          , st->id, rd->name
-                          , st->current_entry
-                          , unpack_storage_number(rd->values[st->current_entry]), new_value
-                    );
-            }
-            else {
-                if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
-                    debug(D_RRD_STATS, "%s/%s: STORE[%ld] = NON EXISTING "
-                          , st->id, rd->name
-                          , st->current_entry
-                    );
-                rd->values[st->current_entry] = pack_storage_number(0, SN_NOT_EXISTS);
-                rd->last_stored_value = NAN;
-            }
-
-            stored_entries++;
-
-            if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG))) {
-                calculated_number t1 = new_value * (calculated_number)rd->multiplier / (calculated_number)rd->divisor;
-                calculated_number t2 = unpack_storage_number(rd->values[st->current_entry]);
-                calculated_number accuracy = accuracy_loss(t1, t2);
-                debug(D_RRD_STATS, "%s/%s: UNPACK[%ld] = " CALCULATED_NUMBER_FORMAT " FLAGS=0x%08x (original = " CALCULATED_NUMBER_FORMAT ", accuracy loss = " CALCULATED_NUMBER_FORMAT "%%%s)"
-                      , st->id, rd->name
-                      , st->current_entry
-                      , t2
-                      , get_storage_number_flags(rd->values[st->current_entry])
-                      , t1
-                      , accuracy
-                      , (accuracy > ACCURACY_LOSS) ? " **TOO BIG** " : ""
-                );
-
-                rd->collected_volume += t1;
-                rd->stored_volume += t2;
-                accuracy = accuracy_loss(rd->collected_volume, rd->stored_volume);
-                debug(D_RRD_STATS, "%s/%s: VOLUME[%ld] = " CALCULATED_NUMBER_FORMAT ", calculated  = " CALCULATED_NUMBER_FORMAT ", accuracy loss = " CALCULATED_NUMBER_FORMAT "%%%s"
-                      , st->id, rd->name
-                      , st->current_entry
-                      , rd->stored_volume
-                      , rd->collected_volume
-                      , accuracy
-                      , (accuracy > ACCURACY_LOSS) ? " **TOO BIG** " : ""
-                );
-
-            }
-        }
-        // reset the storage flags for the next point, if any;
-        storage_flags = SN_EXISTS;
-
-        st->counter++;
-        st->current_entry = ((st->current_entry + 1) >= st->entries) ? 0 : st->current_entry + 1;
-        last_stored_ut = next_store_ut;
+        rrdset_debug(st, "THIS IS IN THE SAME INTERPOLATION POINT");
+        info("INTERNAL CHECK: host '%s', chart '%s' is collected in the same interpolation point: short by %llu microseconds", st->rrdhost->hostname, st->name, next_store_ut - now_collect_ut);
     }
+#endif
+
+    rrdset_done_interpolate(st
+            , update_every_ut
+            , last_stored_ut
+            , next_store_ut
+            , last_collect_ut
+            , now_collect_ut
+            , store_this_entry
+            , storage_flags
+    );
 
     st->last_collected_total  = st->collected_total;
 
@@ -1206,29 +1387,35 @@ void rrdset_done(RRDSET *st) {
         if(unlikely(!rd->updated))
             continue;
 
-        if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
-            debug(D_RRD_STATS, "%s/%s: setting last_collected_value (old: " COLLECTED_NUMBER_FORMAT ") to last_collected_value (new: " COLLECTED_NUMBER_FORMAT ")", st->id, rd->name, rd->last_collected_value, rd->collected_value);
+        #ifdef NETDATA_INTERNAL_CHECKS
+        rrdset_debug(st, "%s: setting last_collected_value (old: " COLLECTED_NUMBER_FORMAT ") to last_collected_value (new: " COLLECTED_NUMBER_FORMAT ")", rd->name, rd->last_collected_value, rd->collected_value);
+        #endif
 
         rd->last_collected_value = rd->collected_value;
 
         switch(rd->algorithm) {
             case RRD_ALGORITHM_INCREMENTAL:
                 if(unlikely(!first_entry)) {
-                    if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
-                        debug(D_RRD_STATS, "%s/%s: setting last_calculated_value (old: " CALCULATED_NUMBER_FORMAT ") to last_calculated_value (new: " CALCULATED_NUMBER_FORMAT ")", st->id, rd->name, rd->last_calculated_value + rd->calculated_value, rd->calculated_value);
+                    #ifdef NETDATA_INTERNAL_CHECKS
+                    rrdset_debug(st, "%s: setting last_calculated_value (old: " CALCULATED_NUMBER_FORMAT ") to last_calculated_value (new: " CALCULATED_NUMBER_FORMAT ")", rd->name, rd->last_calculated_value + rd->calculated_value, rd->calculated_value);
+                    #endif
+
                     rd->last_calculated_value += rd->calculated_value;
                 }
                 else {
-                    if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
-                        debug(D_RRD_STATS, "%s: THIS IS THE FIRST POINT", st->name);
+                    #ifdef NETDATA_INTERNAL_CHECKS
+                    rrdset_debug(st, "THIS IS THE FIRST POINT");
+                    #endif
                 }
                 break;
 
             case RRD_ALGORITHM_ABSOLUTE:
             case RRD_ALGORITHM_PCENT_OVER_ROW_TOTAL:
             case RRD_ALGORITHM_PCENT_OVER_DIFF_TOTAL:
-                if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
-                    debug(D_RRD_STATS, "%s/%s: setting last_calculated_value (old: " CALCULATED_NUMBER_FORMAT ") to last_calculated_value (new: " CALCULATED_NUMBER_FORMAT ")", st->id, rd->name, rd->last_calculated_value, rd->calculated_value);
+                #ifdef NETDATA_INTERNAL_CHECKS
+                rrdset_debug(st, "%s: setting last_calculated_value (old: " CALCULATED_NUMBER_FORMAT ") to last_calculated_value (new: " CALCULATED_NUMBER_FORMAT ")", rd->name, rd->last_calculated_value, rd->calculated_value);
+                #endif
+
                 rd->last_calculated_value = rd->calculated_value;
                 break;
         }
@@ -1237,18 +1424,20 @@ void rrdset_done(RRDSET *st) {
         rd->collected_value = 0;
         rd->updated = 0;
 
-        if(unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
-            debug(D_RRD_STATS, "%s/%s: END "
+        #ifdef NETDATA_INTERNAL_CHECKS
+        rrdset_debug(st, "%s: END "
                     " last_collected_value = " COLLECTED_NUMBER_FORMAT
                     " collected_value = " COLLECTED_NUMBER_FORMAT
                     " last_calculated_value = " CALCULATED_NUMBER_FORMAT
                     " calculated_value = " CALCULATED_NUMBER_FORMAT
-                                      , st->id, rd->name
+                                      , rd->name
                                       , rd->last_collected_value
                                       , rd->collected_value
                                       , rd->last_calculated_value
                                       , rd->calculated_value
-            );
+        );
+        #endif
+
     }
 
     // ALL DONE ABOUT THE DATA UPDATE