datasource: testdata - add predicatable csv wave scenario (#18183)

pkg/tsdb/testdata/scenarios.go

@@ -113,6 +113,14 @@ func init() {
 		Description: PredictablePulseDesc,
 	})
 
+	registerScenario(&Scenario{
+		Id:   "predictable_csv_wave",
+		Name: "Predictable CSV Wave",
+		Handler: func(query *tsdb.Query, context *tsdb.TsdbQuery) *tsdb.QueryResult {
+			return getPredictableCSVWave(query, context)
+		},
+	})
+
 	registerScenario(&Scenario{
 		Id:   "random_walk_table",
 		Name: "Random Walk Table",
@@ -385,27 +393,6 @@ func getPredictablePulse(query *tsdb.Query, context *tsdb.TsdbQuery) *tsdb.Query
 		return queryRes
 	}
 
-	fromStringOrNumber := func(val *simplejson.Json) (null.Float, error) {
-		switch v := val.Interface().(type) {
-		case json.Number:
-			fV, err := v.Float64()
-			if err != nil {
-				return null.Float{}, err
-			}
-			return null.FloatFrom(fV), nil
-		case string:
-			if v == "null" {
-				return null.FloatFromPtr(nil), nil
-			}
-			fV, err := strconv.ParseFloat(v, 64)
-			if err != nil {
-				return null.Float{}, err
-			}
-			return null.FloatFrom(fV), nil
-		default:
-			return null.Float{}, fmt.Errorf("failed to extract value")
-		}
-	}
 	onValue, err = fromStringOrNumber(options.Get("onValue"))
 	if err != nil {
 		queryRes.Error = fmt.Errorf("failed to parse onValue value '%v' into float: %v", options.Get("onValue"), err)
@@ -417,35 +404,97 @@ func getPredictablePulse(query *tsdb.Query, context *tsdb.TsdbQuery) *tsdb.Query
 		return queryRes
 	}
 
-	from := context.TimeRange.GetFromAsMsEpoch()
-	to := context.TimeRange.GetToAsMsEpoch()
-
-	series := newSeriesForQuery(query)
-	points := make(tsdb.TimeSeriesPoints, 0)
-
-	timeStep = timeStep * 1000             // Seconds to Milliseconds
-	timeCursor := from - (from % timeStep) // Truncate Start
-	wavePeriod := timeStep * (onCount + offCount)
-	maxPoints := 10000 // Don't return too many points
-
-	onFor := func(mod int64) null.Float { // How many items in the cycle should get the on value
+	timeStep = timeStep * 1000                     // Seconds to Milliseconds
+	onFor := func(mod int64) (null.Float, error) { // How many items in the cycle should get the on value
 		var i int64
 		for i = 0; i < onCount; i++ {
 			if mod == i*timeStep {
-				return onValue
+				return onValue, nil
 			}
 		}
-		return offValue
+		return offValue, nil
 	}
+	points, err := predictableSeries(context.TimeRange, timeStep, onCount+offCount, onFor)
+	if err != nil {
+		queryRes.Error = err
+		return queryRes
+	}
+
+	series := newSeriesForQuery(query)
+	series.Points = *points
+	queryRes.Series = append(queryRes.Series, series)
+	return queryRes
+}
+
+func getPredictableCSVWave(query *tsdb.Query, context *tsdb.TsdbQuery) *tsdb.QueryResult {
+	queryRes := tsdb.NewQueryResult()
+
+	// Process Input
+	var timeStep int64
+
+	options := query.Model.Get("csvWave")
+
+	var err error
+	if timeStep, err = options.Get("timeStep").Int64(); err != nil {
+		queryRes.Error = fmt.Errorf("failed to parse timeStep value '%v' into integer: %v", options.Get("timeStep"), err)
+		return queryRes
+	}
+	rawValues := options.Get("valuesCSV").MustString()
+	rawValues = strings.TrimRight(strings.TrimSpace(rawValues), ",") // Strip Trailing Comma
+	rawValesCSV := strings.Split(rawValues, ",")
+	values := make([]null.Float, len(rawValesCSV))
+	for i, rawValue := range rawValesCSV {
+		val, err := null.FloatFromString(strings.TrimSpace(rawValue), "null")
+		if err != nil {
+			queryRes.Error = fmt.Errorf("failed to parse value '%v' into nullable float: err", rawValue, err)
+			return queryRes
+		}
+		values[i] = val
+	}
+
+	timeStep = timeStep * 1000 // Seconds to Milliseconds
+	valuesLen := int64(len(values))
+	getValue := func(mod int64) (null.Float, error) {
+		var i int64
+		for i = 0; i < valuesLen; i++ {
+			if mod == i*timeStep {
+				return values[i], nil
+			}
+		}
+		return null.Float{}, fmt.Errorf("did not get value at point in waveform - should not be here")
+	}
+	points, err := predictableSeries(context.TimeRange, timeStep, valuesLen, getValue)
+	if err != nil {
+		queryRes.Error = err
+		return queryRes
+	}
+
+	series := newSeriesForQuery(query)
+	series.Points = *points
+	queryRes.Series = append(queryRes.Series, series)
+	return queryRes
+}
+
+func predictableSeries(timeRange *tsdb.TimeRange, timeStep, length int64, getValue func(mod int64) (null.Float, error)) (*tsdb.TimeSeriesPoints, error) {
+	points := make(tsdb.TimeSeriesPoints, 0)
+
+	from := timeRange.GetFromAsMsEpoch()
+	to := timeRange.GetToAsMsEpoch()
+
+	timeCursor := from - (from % timeStep) // Truncate Start
+	wavePeriod := timeStep * length
+	maxPoints := 10000 // Don't return too many points
+
 	for i := 0; i < maxPoints && timeCursor < to; i++ {
-		point := tsdb.NewTimePoint(onFor(timeCursor%wavePeriod), float64(timeCursor))
+		val, err := getValue(timeCursor % wavePeriod)
+		if err != nil {
+			return &points, err
+		}
+		point := tsdb.NewTimePoint(val, float64(timeCursor))
 		points = append(points, point)
 		timeCursor += timeStep
 	}
-
-	series.Points = points
-	queryRes.Series = append(queryRes.Series, series)
-	return queryRes
+	return &points, nil
 }
 
 func getRandomWalk(query *tsdb.Query, tsdbQuery *tsdb.TsdbQuery) *tsdb.QueryResult {
@@ -541,3 +590,18 @@ func newSeriesForQuery(query *tsdb.Query) *tsdb.TimeSeries {
 
 	return &tsdb.TimeSeries{Name: alias}
 }
+
+func fromStringOrNumber(val *simplejson.Json) (null.Float, error) {
+	switch v := val.Interface().(type) {
+	case json.Number:
+		fV, err := v.Float64()
+		if err != nil {
+			return null.Float{}, err
+		}
+		return null.FloatFrom(fV), nil
+	case string:
+		return null.FloatFromString(v, "null")
+	default:
+		return null.Float{}, fmt.Errorf("failed to extract value")
+	}
+}