fixed a bug in the interpolation algorithm that was showing unrealistic values under certain conditions; fixes #981

1 files changed, 62 insertions, 55 deletions

diff --git a/src/rrd.c b/src/rrd.c
index acb98021be..e72f495d15 100644
--- a/src/rrd.c
+++ b/src/rrd.c
@@ -988,7 +988,7 @@ unsigned long long rrdset_done(RRDSET *st)
 
     RRDDIM *rd, *last;
     int oldstate, store_this_entry = 1, first_entry = 0;
-    unsigned long long last_ut, now_ut, next_ut, stored_entries = 0;
+    unsigned long long last_stored_ut, now_collect_ut, last_collect_ut, next_store_ut, stored_entries = 0;
 
     if(unlikely(pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &oldstate) != 0))
         error("Cannot set pthread cancel state to DISABLE.");
@@ -1014,6 +1014,7 @@ unsigned long long rrdset_done(RRDSET *st)
         // it is the first entry
         // set the last_collected_time to now
         gettimeofday(&st->last_collected_time, NULL);
+        last_collect_ut = st->last_collected_time.tv_sec * 1000000ULL + st->last_collected_time.tv_usec - (st->update_every * 1000000);
 
         // the first entry should not be stored
         store_this_entry = 0;
@@ -1024,7 +1025,8 @@ unsigned long long rrdset_done(RRDSET *st)
     else {
         // it is not the first entry
         // calculate the proper last_collected_time, using usec_since_last_update
-        unsigned long long ut = st->last_collected_time.tv_sec * 1000000ULL + st->last_collected_time.tv_usec + st->usec_since_last_update;
+        last_collect_ut = st->last_collected_time.tv_sec * 1000000ULL + st->last_collected_time.tv_usec;
+        unsigned long long ut = last_collect_ut + st->usec_since_last_update;
         st->last_collected_time.tv_sec = (time_t) (ut / 1000000ULL);
         st->last_collected_time.tv_usec = (suseconds_t) (ut % 1000000ULL);
     }
@@ -1064,21 +1066,18 @@ unsigned long long rrdset_done(RRDSET *st)
     }
 
     // these are the 3 variables that will help us in interpolation
-    // last_ut = the last time we added a value to the storage
-    //  now_ut = the time the current value is taken at
-    // next_ut = the time of the next interpolation point
-    last_ut = st->last_updated.tv_sec * 1000000ULL + st->last_updated.tv_usec;
-    now_ut  = st->last_collected_time.tv_sec * 1000000ULL + st->last_collected_time.tv_usec;
-    next_ut = (st->last_updated.tv_sec + st->update_every) * 1000000ULL;
-
-    if(unlikely(!first_entry && now_ut < next_ut)) {
-        if(unlikely(st->debug)) debug(D_RRD_STATS, "%s: THIS IS IN THE SAME INTERPOLATION POINT", st->name);
-    }
+    // 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 * 1000000ULL + st->last_updated.tv_usec;
+    now_collect_ut = st->last_collected_time.tv_sec * 1000000ULL + st->last_collected_time.tv_usec;
+    next_store_ut  = (st->last_updated.tv_sec + st->update_every) * 1000000ULL;
 
     if(unlikely(st->debug)) {
-        debug(D_RRD_STATS, "%s: last ut = %0.3Lf (last updated time)", st->name, (long double)last_ut/1000000.0);
-        debug(D_RRD_STATS, "%s: now  ut = %0.3Lf (current update time)", st->name, (long double)now_ut/1000000.0);
-        debug(D_RRD_STATS, "%s: next ut = %0.3Lf (next interpolation point)", st->name, (long double)next_ut/1000000.0);
+        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)) {
@@ -1090,7 +1089,7 @@ unsigned long long rrdset_done(RRDSET *st)
     // calculate totals and count the dimensions
     int dimensions;
     st->collected_total = 0;
-    for( rd = st->dimensions, dimensions = 0 ; likely(rd) ; rd = rd->next, dimensions++ )
+    for( rd = st->dimensions, dimensions = 0 ; rd ; rd = rd->next, dimensions++ )
         if(likely(rd->updated)) st->collected_total += rd->collected_value;
 
     uint32_t storage_flags = SN_EXISTS;
@@ -1098,7 +1097,7 @@ unsigned long long rrdset_done(RRDSET *st)
     // process all dimensions to calculate their values
     // based on the collected figures only
     // at this stage we do not interpolate anything
-    for( rd = st->dimensions ; likely(rd) ; rd = rd->next ) {
+    for( rd = st->dimensions ; rd ; rd = rd->next ) {
 
         if(unlikely(!rd->updated)) {
             rd->calculated_value = 0;
@@ -1177,7 +1176,7 @@ unsigned long long rrdset_done(RRDSET *st)
                     rd->last_collected_value = rd->collected_value;
                 }
 
-                rd->calculated_value =
+                rd->calculated_value +=
                       (calculated_number)(rd->collected_value - rd->last_collected_value)
                     * (calculated_number)rd->multiplier
                     / (calculated_number)rd->divisor;
@@ -1267,21 +1266,26 @@ unsigned long long rrdset_done(RRDSET *st)
     // at this point we have all the calculated values ready
     // it is now time to interpolate values on a second boundary
 
-    unsigned long long first_ut = last_ut;
-    long long iterations = (now_ut - last_ut) / (st->update_every * 1000000ULL);
-    if((now_ut % (st->update_every * 1000000ULL)) == 0) iterations++;
+    if(unlikely(now_collect_ut < next_store_ut)) {
+        // this is collected in the same interpolation point
+        if(unlikely(st->debug)) debug(D_RRD_STATS, "%s: THIS IS IN THE SAME INTERPOLATION POINT", st->name);
+    }
+
+    unsigned long long first_ut = last_stored_ut;
+    long long iterations = (now_collect_ut - last_stored_ut) / (st->update_every * 1000000ULL);
+    if((now_collect_ut % (st->update_every * 1000000ULL)) == 0) iterations++;
 
-    for( ; likely(next_ut <= now_ut) ; next_ut += st->update_every * 1000000ULL, iterations-- ) {
+    for( ; next_store_ut <= now_collect_ut ; next_store_ut += st->update_every * 1000000ULL, iterations-- ) {
 #ifdef NETDATA_INTERNAL_CHECKS
-        if(iterations < 0) { error("%s: iterations calculation wrapped! first_ut = %llu, last_ut = %llu, next_ut = %llu, now_ut = %llu", st->name, first_ut, last_ut, next_ut, now_ut); }
+        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(st->debug)) {
-            debug(D_RRD_STATS, "%s: last ut = %0.3Lf (last updated time)", st->name, (long double)last_ut/1000000.0);
-            debug(D_RRD_STATS, "%s: next ut = %0.3Lf (next interpolation point)", st->name, (long double)next_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);
         }
 
-        st->last_updated.tv_sec = (time_t) (next_ut / 1000000ULL);
+        st->last_updated.tv_sec = (time_t) (next_store_ut / 1000000ULL);
         st->last_updated.tv_usec = 0;
 
         for( rd = st->dimensions ; likely(rd) ; rd = rd->next ) {
@@ -1291,8 +1295,8 @@ unsigned long long rrdset_done(RRDSET *st)
                 case RRDDIM_INCREMENTAL:
                     new_value = (calculated_number)
                         (      rd->calculated_value
-                            * (calculated_number)(next_ut - last_ut)
-                            / (calculated_number)(now_ut - last_ut)
+                            * (calculated_number)(next_store_ut - last_collect_ut)
+                            / (calculated_number)(now_collect_ut - last_collect_ut)
                         );
 
                     if(unlikely(st->debug))
@@ -1304,10 +1308,11 @@ unsigned long long rrdset_done(RRDSET *st)
                             , st->id, rd->name
                             , new_value
                             , rd->calculated_value
-                            , (next_ut - last_ut)
-                            , (now_ut - last_ut)
+                            , (next_store_ut - last_stored_ut)
+                            , (now_collect_ut - last_stored_ut)
                             );
 
+                    last_collect_ut = next_store_ut;
                     rd->calculated_value -= new_value;
                     new_value += rd->last_calculated_value;
                     rd->last_calculated_value = 0;
@@ -1331,8 +1336,8 @@ unsigned long long rrdset_done(RRDSET *st)
 
                         new_value = (calculated_number)
                             (   (     (rd->calculated_value - rd->last_calculated_value)
-                                    * (calculated_number)(next_ut - first_ut)
-                                    / (calculated_number)(now_ut - first_ut)
+                                    * (calculated_number)(next_store_ut - last_collect_ut)
+                                    / (calculated_number)(now_collect_ut - last_collect_ut)
                                 )
                                 +  rd->last_calculated_value
                             );
@@ -1346,22 +1351,15 @@ unsigned long long rrdset_done(RRDSET *st)
                                 , st->id, rd->name
                                 , new_value
                                 , rd->calculated_value, rd->last_calculated_value
-                                , (next_ut - first_ut)
-                                , (now_ut - first_ut), rd->last_calculated_value
+                                , (next_store_ut - first_ut)
+                                , (now_collect_ut - first_ut), rd->last_calculated_value
                                 );
-
-                        // this is wrong
-                        // it fades the value towards the target
-                        // while we know the calculated value is different
-                        // if(likely(next_ut + st->update_every * 1000000ULL > now_ut)) rd->calculated_value = new_value;
                     }
                     break;
             }
 
-            if(unlikely(!store_this_entry)) {
-                // store_this_entry = 1;
+            if(unlikely(!store_this_entry))
                 continue;
-            }
 
             if(likely(rd->updated && rd->counter > 1 && iterations < st->gap_when_lost_iterations_above)) {
                 rd->values[st->current_entry] = pack_storage_number(new_value, storage_flags );
@@ -1419,25 +1417,34 @@ unsigned long long rrdset_done(RRDSET *st)
 
         st->counter++;
         st->current_entry = ((st->current_entry + 1) >= st->entries) ? 0 : st->current_entry + 1;
-        last_ut = next_ut;
+        last_stored_ut = next_store_ut;
     }
 
-    // align next interpolation to last collection point
-    if(likely(stored_entries || !store_this_entry)) {
-        st->last_updated.tv_sec = st->last_collected_time.tv_sec;
-        st->last_updated.tv_usec = st->last_collected_time.tv_usec;
-        st->last_collected_total  = st->collected_total;
-    }
+    st->last_collected_total  = st->collected_total;
 
-    for( rd = st->dimensions; likely(rd) ; rd = rd->next ) {
+    for( rd = st->dimensions; rd ; rd = rd->next ) {
         if(unlikely(!rd->updated)) continue;
 
-        if(likely(stored_entries || !store_this_entry)) {
-            if(unlikely(st->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);
-            rd->last_collected_value = rd->collected_value;
+        if(unlikely(st->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);
+        rd->last_collected_value = rd->collected_value;
+
+        switch(rd->algorithm) {
+            case RRDDIM_INCREMENTAL:
+                if(unlikely(!first_entry)) {
+                    if(unlikely(st->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);
+                    rd->last_calculated_value += rd->calculated_value;
+                }
+                else {
+                    if(unlikely(st->debug)) debug(D_RRD_STATS, "%s: THIS IS THE FIRST POINT", st->name);
+                }
+                break;
 
-            if(unlikely(st->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->last_calculated_value = rd->calculated_value;
+            case RRDDIM_ABSOLUTE:
+            case RRDDIM_PCENT_OVER_ROW_TOTAL:
+            case RRDDIM_PCENT_OVER_DIFF_TOTAL:
+                if(unlikely(st->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->last_calculated_value = rd->calculated_value;
+                break;
         }
 
         rd->calculated_value = 0;