[query] Implemented the Graphite delay function

src/query/graphite/native/builtin_functions.go

@@ -199,7 +199,6 @@ func limit(_ *common.Context, series singlePathSpec, n int) (ts.SeriesList, erro
 	return r, nil
 }
 
-// timeShift draws the selected metrics shifted in time. If no sign is given, a minus sign ( - ) is
 // implied which will shift the metric back in time. If a plus sign ( + ) is given, the metric will
 // be shifted forward in time
 func timeShift(
@@ -215,6 +214,7 @@ func timeShift(
 	}
 
 	shift, err := common.ParseInterval(timeShiftS)
+
 	if err != nil {
 		return nil, errors.NewInvalidParamsError(fmt.Errorf("invalid timeShift parameter %s: %v", timeShiftS, err))
 	}
@@ -242,6 +242,50 @@ func timeShift(
 	}, nil
 }
 
+// implied which will shift the metric back in time. If a plus sign ( + ) is given, the metric will
+// be shifted forward in time
+func delay(
+	ctx *common.Context,
+	_ singlePathSpec,
+	steps int,
+) (*unaryContextShifter, error) {
+
+	contextShiftingFn := func(c *common.Context) *common.Context {
+		opts := common.NewChildContextOptions()
+		opts.AdjustTimeRange(0, 0, 0, 0)
+		childCtx := c.NewChildContext(opts)
+		return childCtx
+	}
+
+	transformerFn := func(input ts.SeriesList) (ts.SeriesList, error) {
+		output := make([]*ts.Series, input.Len())
+
+		for i, series := range input.Values {
+
+			delayedVals := delayValues(ctx, *series, steps)
+			delayedSeries := ts.NewSeries(ctx, series.Name(), series.StartTime(), delayedVals)
+			renamedSeries := delayedSeries.RenamedTo(fmt.Sprintf("delay(%s, %s)", delayedSeries.Name(), steps))
+			output[i] = renamedSeries
+
+		}
+		input.Values = output
+		return input, nil
+	}
+
+	return &unaryContextShifter{
+		ContextShiftFunc: contextShiftingFn,
+		UnaryTransformer: transformerFn,
+	}, nil
+}
+
+func delayValues(ctx *common.Context, series ts.Series, steps int ) (ts.Values) {
+	output := ts.NewValues(ctx, series.MillisPerStep(), series.Len())
+	for i := steps; i < series.Len(); i++ {
+		output.SetValueAt(i, series.ValueAt(i - steps))
+	}
+	return output
+}
+
 // absolute returns the absolute value of each element in the series.
 func absolute(ctx *common.Context, input singlePathSpec) (ts.SeriesList, error) {
 	return transform(ctx, input,
@@ -1658,6 +1702,83 @@ func movingMedian(ctx *common.Context, _ singlePathSpec, windowSize string) (*bi
 	}, nil
 }
 
+// movingSum takes one metric or a wildcard seriesList followed by a a quoted string
+// with a length of time like '1hour' or '5min'. Graphs the sum of the preceding
+// datapoints for each point on the graph. All previous datapoints are set to None at
+// the beginning of the graph.
+func movingSum(ctx *common.Context, _ singlePathSpec, windowSize string) (*binaryContextShifter, error) {
+	interval, err := common.ParseInterval(windowSize)
+	if err != nil {
+		return nil, err
+	}
+
+	if interval <= 0 {
+		return nil, common.ErrInvalidIntervalFormat
+	}
+
+	contextShiftingFn := func(c *common.Context) *common.Context {
+		opts := common.NewChildContextOptions()
+		opts.AdjustTimeRange(0, 0, interval, 0)
+		childCtx := c.NewChildContext(opts)
+		return childCtx
+	}
+
+	bootstrapStartTime, bootstrapEndTime := ctx.StartTime.Add(-interval), ctx.StartTime
+	transformerFn := func(bootstrapped, original ts.SeriesList) (ts.SeriesList, error) {
+		bootstrapList, err := combineBootstrapWithOriginal(ctx,
+			bootstrapStartTime, bootstrapEndTime,
+			bootstrapped, singlePathSpec(original))
+		if err != nil {
+			return ts.NewSeriesList(), err
+		}
+
+		results := make([]*ts.Series, 0, original.Len())
+
+		for i, bootstrap := range bootstrapList.Values {
+			series := original.Values[i]
+			windowPoints := int(interval / (time.Duration(series.MillisPerStep()) * time.Millisecond))
+			if windowPoints <= 0 {
+				err := errors.NewInvalidParamsError(fmt.Errorf(
+					"non positive window points, windowSize=%s, stepSize=%d",
+					windowSize, series.MillisPerStep()))
+				return ts.NewSeriesList(), err
+			}
+			window := make([]float64, windowPoints)
+			util.Memset(window, math.NaN())
+			numSteps := series.Len()
+			offset := bootstrap.Len() - numSteps
+			vals := ts.NewValues(ctx, series.MillisPerStep(), numSteps)
+			for i := 0; i < numSteps; i++ {
+				for j := i + offset - windowPoints; j < i+offset; j++ {
+					if j < 0 || j >= bootstrap.Len() {
+						continue
+					}
+
+					idx := j - i - offset + windowPoints
+					if idx < 0 || idx > len(window)-1 {
+						continue
+					}
+
+					window[idx] = bootstrap.ValueAt(j)
+				}
+				vals.SetValueAt(i, common.SafeSum(window))
+			}
+			name := fmt.Sprintf("movingSum(%s,%q)", series.Name(), windowSize)
+			newSeries := ts.NewSeries(ctx, name, series.StartTime(), vals)
+			results = append(results, newSeries)
+		}
+
+		original.Values = results
+		return original, nil
+	}
+
+	return &binaryContextShifter{
+		ContextShiftFunc:  contextShiftingFn,
+		BinaryTransformer: transformerFn,
+	}, nil
+}
+
+
 // legendValue takes one metric or a wildcard seriesList and a string in quotes.
 // Appends a value to the metric name in the legend.  Currently one or several of:
 // "last", "avg", "total", "min", "max".
@@ -1872,6 +1993,7 @@ func init() {
 	MustRegisterFunction(dashed).WithDefaultParams(map[uint8]interface{}{
 		2: 5.0, // dashLength
 	})
+	MustRegisterFunction(delay)
 	MustRegisterFunction(derivative)
 	MustRegisterFunction(diffSeries)
 	MustRegisterFunction(divideSeries)
@@ -1906,6 +2028,7 @@ func init() {
 	MustRegisterFunction(mostDeviant)
 	MustRegisterFunction(movingAverage)
 	MustRegisterFunction(movingMedian)
+	MustRegisterFunction(movingSum)
 	MustRegisterFunction(multiplySeries)
 	MustRegisterFunction(nonNegativeDerivative).WithDefaultParams(map[uint8]interface{}{
 		2: math.NaN(), // maxValue

src/query/graphite/native/builtin_functions_test.go

@@ -685,6 +685,22 @@ func TestMovingAverageError(t *testing.T) {
 	testMovingFunctionError(t, "movingAverage(foo.bar.baz, 0)")
 }
 
+func TestMovingSumSuccess(t *testing.T) {
+	values := []float64{12.0, 19.0, -10.0, math.NaN(), 10.0}
+	bootstrap := []float64{3.0, 4.0, 5.0}
+	expected := []float64{12.0, 21.0, 36.0, 21.0, 9.0} // (3+4+5), (4+5+12), (5+12+19), (12+19-10), (19-10+Nan)
+
+	testMovingFunction(t, "movingSum(foo.bar.baz, '30s')", "movingSum(foo.bar.baz,\"30s\")", values, bootstrap, expected)
+	testMovingFunction(t, "movingSum(foo.bar.baz, '30s')", "movingSum(foo.bar.baz,3)", nil, nil, nil)
+
+	bootstrapEntireSeries := []float64{3.0, 4.0, 5.0, 12.0, 19.0, -10.0, math.NaN(), 10.0}
+	testMovingFunction(t, "movingSum(foo.bar.baz, '30s')", "movingSum(foo.bar.baz,\"30s\")", values, bootstrapEntireSeries, expected)
+}
+
+func TestMovingSumError(t *testing.T) {
+	testMovingFunctionError(t, "movingSum(foo.bar.baz, '-30s')")
+}
+
 func TestIsNonNull(t *testing.T) {
 	ctx := common.NewTestContext()
 	defer ctx.Close()
@@ -2547,14 +2563,14 @@ func TestMovingMedianInvalidLimits(t *testing.T) {
 func TestMovingMismatchedLimits(t *testing.T) {
 	// NB: this tests the behavior when query limits do not snap exactly to data
 	// points. When limits do not snap exactly, the first point should be omitted.
-	for _, fn := range []string{"movingAverage", "movingMedian"} {
+	for _, fn := range []string{"movingAverage", "movingMedian", "movingSum"} {
 		for i := time.Duration(0); i < time.Minute; i += time.Second {
-			testMovingAverageInvalidLimits(t, fn, i)
+			testMovingFunctionInvalidLimits(t, fn, i)
 		}
 	}
 }
 
-func testMovingAverageInvalidLimits(t *testing.T, fn string, offset time.Duration) {
+func testMovingFunctionInvalidLimits(t *testing.T, fn string, offset time.Duration) {
 	ctrl := xgomock.NewController(t)
 	defer ctrl.Finish()
 
@@ -2840,6 +2856,66 @@ func TestTimeShift(t *testing.T) {
 		[]common.TestSeries{expected}, res.Values)
 }
 
+func TestDelay(t *testing.T) {
+	var values = [3][]float64{
+		{54.0, 48.0, 92.0, 54.0, 14.0, 1.2},
+		{4.0, 5.0, math.NaN(), 6.4, 7.2, math.NaN()},
+		{math.NaN(), 8.0, 9.0, 10.6, 11.2, 12.2},
+	}
+	expected := [3][]float64{
+		{math.NaN(), math.NaN(), math.NaN(), 54.0, 48.0, 92.0},
+		{math.NaN(), math.NaN(), math.NaN(), 4.0, 5.0, math.NaN()},
+		{math.NaN(), math.NaN(), math.NaN(), math.NaN(), 8.0, 9.0},
+	}
+
+	for index, value := range values {
+		e := expected[index]
+		testDelay(t, "delay(foo.bar.baz, 3)", "delay(foo.bar.baz,3)", value, e)
+	}
+}
+
+var (
+	testDelayBootstrap = testMovingAverageStart.Add(-30 * time.Second)
+	testDelayStart     = time.Now().Truncate(time.Minute)
+	testDelayEnd       = testMovingAverageStart.Add(time.Minute)
+)
+
+func testDelay(t *testing.T, target, expectedName string, values, output []float64) {
+	ctx := common.NewTestContext()
+	defer ctx.Close()
+
+	emptyBootstrap := []float64{}
+	engine := NewEngine(
+		&common.MovingAverageStorage{
+			StepMillis:     10000,
+			Values:         values,
+			Bootstrap:      emptyBootstrap,
+			BootstrapStart: testDelayBootstrap,
+		},
+	)
+	phonyContext := common.NewContext(common.ContextOptions{
+		Start:  testDelayStart,
+		End:    testDelayEnd,
+		Engine: engine,
+	})
+
+	expr, err := phonyContext.Engine.(*Engine).Compile(target)
+	require.NoError(t, err)
+	res, err := expr.Execute(phonyContext)
+	require.NoError(t, err)
+	var expected []common.TestSeries
+
+	if output != nil {
+		expectedSeries := common.TestSeries{
+			Name: expectedName,
+			Data: output,
+		}
+		expected = append(expected, expectedSeries)
+	}
+	common.CompareOutputsAndExpected(t, 10000, testDelayStart,
+		expected, res.Values)
+}
+
 func TestDashed(t *testing.T) {
 	ctx := common.NewTestContext()
 	defer ctx.Close()
@@ -2927,6 +3003,7 @@ func TestFunctionsRegistered(t *testing.T) {
 		"currentAbove",
 		"currentBelow",
 		"dashed",
+		"delay",
 		"derivative",
 		"diffSeries",
 		"divideSeries",
@@ -2960,6 +3037,7 @@ func TestFunctionsRegistered(t *testing.T) {
 		"mostDeviant",
 		"movingAverage",
 		"movingMedian",
+		"movingSum",
 		"multiplySeries",
 		"nonNegativeDerivative",
 		"nPercentile",