MaxDataPoints consolidation improvements: support nudging for consistent bucketing

cmd/carbonapi/config/config.go

@@ -113,6 +113,9 @@ type ConfigType struct {
 	MaxQueryLength              uint64 `mapstructure:"maxQueryLength"`
 	CombineMultipleTargetsInOne bool   `mapstructure:"combineMultipleTargetsInOne"`
 
+	NudgeStartTimeOnAggregation             bool `mapstructure:"nudgeStartTimeOnAggregation"`
+	UseBucketsHighestTimestampOnAggregation bool `mapstructure:"useBucketsHighestTimestampOnAggregation"`
+
 	ResponseCache cache.BytesCache `mapstructure:"-" json:"-"`
 	BackendCache  cache.BytesCache `mapstructure:"-" json:"-"`
 

cmd/carbonapi/config/init.go

@@ -26,6 +26,7 @@ import (
 	fconfig "github.com/go-graphite/carbonapi/expr/functions/config"
 	"github.com/go-graphite/carbonapi/expr/helper"
 	"github.com/go-graphite/carbonapi/expr/rewrite"
+	tconfig "github.com/go-graphite/carbonapi/expr/types/config"
 	"github.com/go-graphite/carbonapi/limiter"
 	"github.com/go-graphite/carbonapi/pkg/parser"
 	zipperTypes "github.com/go-graphite/carbonapi/zipper/types"
@@ -257,6 +258,9 @@ func SetUpConfig(logger *zap.Logger, BuildVersion string) {
 		)
 	}
 
+	tconfig.Config.NudgeStartTimeOnAggregation = Config.NudgeStartTimeOnAggregation
+	tconfig.Config.UseBucketsHighestTimestampOnAggregation = Config.UseBucketsHighestTimestampOnAggregation
+
 	if Config.Listen != "" {
 		listeners := make(map[string]struct{})
 		for _, l := range Config.Listeners {
@@ -401,6 +405,9 @@ func SetUpViper(logger *zap.Logger, configPath *string, exactConfig bool, viperP
 	viper.SetDefault("useCachingDNSResolver", false)
 	viper.SetDefault("logger", map[string]string{})
 	viper.SetDefault("combineMultipleTargetsInOne", false)
+	viper.SetDefault("nudgeStartTimeOnAggregation", false)
+	viper.SetDefault("useBucketsHighestTimestampOnAggregation", false)
+
 	viper.AutomaticEnv()
 
 	var err error

doc/configuration.md

@@ -56,6 +56,8 @@ Table of Contents
       * [For IRONdb](#for-irondb)
   * [expireDelaySec](#expiredelaysec)
     * [Example](#example-21)
+  * [nudgeStartTimeOnAggregation](#nudgestarttimeonaggregation)
+  * [useBucketsHighestTimestampOnAggregation](#usebucketshighesttimestamponaggregation)
 
 # General configuration for carbonapi
 
@@ -940,3 +942,28 @@ Default: 600 (10 minutes)
 ```yaml
 expireDelaySec: 10
 ```
+
+***
+## nudgeStartTimeOnAggregation
+Enables nudging the start time of metrics  when aggregating to honor MaxDataPoints.
+The start time is nudged in such way that timestamps always fall in the same bucket.
+This is done by GraphiteWeb, and is useful to avoid jitter in graphs when refreshing the page.
+
+Default: false
+
+### Example
+```yaml
+nudgeStartTimeOnAggregation: true
+```
+
+***
+## useBucketsHighestTimestampOnAggregation
+Enables using the highest timestamp of the buckets when aggregating to honor MaxDataPoints, instead of the lowest timestamp.
+This prevents results to appear to predict the future.
+
+Default: false
+
+### Example
+```yaml
+useBucketsHighestTimestampOnAggregation: true
+```

expr/types/config/config.go

@@ -0,0 +1,16 @@
+package config
+
+type ConfigType = struct {
+	// NudgeStartTimeOnAggregation enables nudging the start time of metrics
+	// when aggregated. The start time is nudged in such way that timestamps
+	// always fall in the same bucket. This is done by GraphiteWeb, and is
+	// useful to avoid jitter in graphs when refreshing the page.
+	NudgeStartTimeOnAggregation bool
+
+	// UseBucketsHighestTimestampOnAggregation enables using the highest timestamp of the
+	// buckets when aggregating to honor MaxDataPoints, instead of the lowest timestamp.
+	// This prevents results to appear to predict the future.
+	UseBucketsHighestTimestampOnAggregation bool
+}
+
+var Config = ConfigType{}

expr/types/types.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/go-graphite/carbonapi/expr/consolidations"
 	"github.com/go-graphite/carbonapi/expr/tags"
+	"github.com/go-graphite/carbonapi/expr/types/config"
 	pbv2 "github.com/go-graphite/protocol/carbonapi_v2_pb"
 	pb "github.com/go-graphite/protocol/carbonapi_v3_pb"
 	pickle "github.com/lomik/og-rek"
@@ -141,7 +142,7 @@ func MarshalJSON(results []*MetricData, timestampMultiplier int64, noNullPoints
 		b = append(b, `,"datapoints":[`...)
 
 		var innerComma bool
-		t := r.StartTime * timestampMultiplier
+		t := r.AggregatedStartTime() * timestampMultiplier
 		for _, v := range r.AggregatedValues() {
 			if noNullPoints && math.IsNaN(v) {
 				t += r.AggregatedTimeStep() * timestampMultiplier
@@ -330,6 +331,60 @@ func (r *MetricData) AggregatedTimeStep() int64 {
 	return r.StepTime * int64(r.ValuesPerPoint)
 }
 
+// AggregatedStartTime returns the start time of the aggregated series.
+// This can be different from the original start time if NudgeStartTimeOnAggregation
+// or UseBucketsHighestTimestampOnAggregation are enabled.
+func (r *MetricData) AggregatedStartTime() int64 {
+	start := r.StartTime + r.nudgePointsCount()*r.StepTime
+	if config.Config.UseBucketsHighestTimestampOnAggregation {
+		return start + r.AggregatedTimeStep() - r.StepTime
+	}
+	return start
+}
+
+// nudgePointsCount returns the number of points to discard at the beginning of
+// the series when aggregating. This is done if NudgeStartTimeOnAggregation is
+// enabled, and has the purpose of assigning timestamps of a series to buckets
+// consistently across different time ranges. To simplify the aggregation
+// logic, we discard points at the beginning of the series so that a bucket
+// starts right at the beginning. This function calculates how many points to
+// discard.
+func (r *MetricData) nudgePointsCount() int64 {
+	if !config.Config.NudgeStartTimeOnAggregation {
+		return 0
+	}
+
+	if len(r.Values) <= 2*r.ValuesPerPoint {
+		// There would be less than 2 points after aggregation, removing one would be too impactful.
+		return 0
+	}
+
+	// Suppose r.StartTime=4, r.StepTime=3 and aggTimeStep=6.
+	// - ts:                       00 01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 ...
+	// - original buckets:         -- -- --|        |        |        |        |     |   ...
+	// - aggregated buckets:       -- -- --|                 |                 |         ...
+
+	// We start counting our aggTimeStep buckets at absolute time r.StepTime.
+	// Notice the following:
+	// - ts:                       00 01 02 03 04 05 06 07 08 09 10 11 12 13 14 15 16 ...
+	// - bucket #:                  -  -  -  1  1  1  1  1  1  2  2  2  2  2  2  3  3 ...
+	// - (ts-step) % aggTimeStep:   -  -  -  0  1  2  3  4  5  0  1  2  3  4  5  0  1 ...
+
+	// Given a timestamp 'ts', we can calculate how far it is from the beginning
+	// of the nearest bucket to the right by doing:
+	// * aggTimeStep - ((ts-r.StepTime) % aggTimeStep)
+	// Using this, we calculate the 'distance' from r.StartTime to the
+	// nearest bucket to the right. If this distance is less than aggTimeStep,
+	// then r.StartTime is not the beginning of a bucket. We need to discard
+	// dist / r.StepTime points (which could be zero if dist < r.StepTime).
+	aggTimeStep := r.AggregatedTimeStep()
+	dist := aggTimeStep - ((r.StartTime - r.StepTime) % aggTimeStep)
+	if dist < aggTimeStep {
+		return dist / r.StepTime
+	}
+	return 0
+}
+
 // GetAggregateFunction returns MetricData.AggregateFunction and set it, if it's not yet
 func (r *MetricData) GetAggregateFunction() func([]float64) float64 {
 	if r.AggregateFunction == nil {
@@ -363,7 +418,8 @@ func (r *MetricData) AggregateValues() {
 	n := len(r.Values)/r.ValuesPerPoint + 1
 	aggV := make([]float64, 0, n)
 
-	v := r.Values
+	nudgeCount := r.nudgePointsCount()
+	v := r.Values[nudgeCount:]
 
 	for len(v) >= r.ValuesPerPoint {
 		val := aggFunc(v[:r.ValuesPerPoint])