public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Alerting: support alerting on data.Frame (that can be time series) (#22812)

Browse Source 
data.Frame (that can be series) are converted to as tsdb.TimeSeriesSlice - so new backend plugins can be shimmed into existing alerting

use sdk v0.31.0
This commit is contained in:
Kyle Brandt
2020-03-18 10:30:07 -04:00
committed by GitHub
parent 44b7f3ea1c
commit 0e09a3fe15
15 changed files with 808 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/services/alerting/conditions/query.go
+24 -2
View File
@@ -13,6 +13,7 @@ import (
"github.com/grafana/grafana/pkg/models"
"github.com/grafana/grafana/pkg/services/alerting"
"github.com/grafana/grafana/pkg/tsdb"
"github.com/grafana/grafana/pkg/util/errutil"
)
func init() {
@@ -168,12 +169,30 @@ func (c *QueryCondition) executeQuery(context *alerting.EvalContext, timeRange *
return nil, fmt.Errorf("tsdb.HandleRequest() response error %v", v)
}
result = append(result, v.Series...)
// If there are dataframes but no series on the result
useDataframes := v.Dataframes != nil && (v.Series == nil || len(v.Series) == 0)
if useDataframes { // convert the dataframes to tsdb.TimeSeries
frames, err := tsdb.FramesFromBytes(v.Dataframes)
if err != nil {
return nil, errutil.Wrap("tsdb.HandleRequest() failed to unmarshal arrow dataframes from bytes", err)
}
for _, frame := range frames {
ss, err := tsdb.FrameToSeriesSlice(frame)
if err != nil {
return nil, errutil.Wrapf(err, `tsdb.HandleRequest() failed to convert dataframe "%v" to tsdb.TimeSeriesSlice`, frame.Name)
}
result = append(result, ss...)
}
} else {
result = append(result, v.Series...)
}
queryResultData := map[string]interface{}{}
if context.IsTestRun {
queryResultData["series"] = v.Series
queryResultData["series"] = result
}
if context.IsDebug && v.Meta != nil {
@@ -181,6 +200,9 @@ func (c *QueryCondition) executeQuery(context *alerting.EvalContext, timeRange *
}
if context.IsTestRun || context.IsDebug {
if useDataframes {
queryResultData["fromDataframe"] = true
}
context.Logs = append(context.Logs, &alerting.ResultLogEntry{
Message: fmt.Sprintf("Condition[%d]: Query Result", c.Index),
Data: simplejson.NewFromAny(queryResultData),
pkg/services/alerting/conditions/query_test.go
+40 -1
View File
@@ -3,7 +3,9 @@ package conditions
import (
"context"
"testing"
"time"
"github.com/grafana/grafana-plugin-sdk-go/data"
"github.com/grafana/grafana/pkg/bus"
"github.com/grafana/grafana/pkg/components/null"
"github.com/grafana/grafana/pkg/components/simplejson"
@@ -51,6 +53,17 @@ func TestQueryCondition(t *testing.T) {
So(cr.Firing, ShouldBeTrue)
})
Convey("should fire when avg is above 100 on dataframe", func() {
ctx.frame = data.NewFrame("",
data.NewField("time", nil, []time.Time{time.Now()}),
data.NewField("val", nil, []int64{120, 150}),
)
cr, err := ctx.exec()
So(err, ShouldBeNil)
So(cr.Firing, ShouldBeTrue)
})
Convey("Should not fire when avg is below 100", func() {
points := tsdb.NewTimeSeriesPointsFromArgs(90, 0)
ctx.series = tsdb.TimeSeriesSlice{tsdb.NewTimeSeries("test1", points)}
@@ -60,6 +73,17 @@ func TestQueryCondition(t *testing.T) {
So(cr.Firing, ShouldBeFalse)
})
Convey("Should not fire when avg is below 100 on dataframe", func() {
ctx.frame = data.NewFrame("",
data.NewField("time", nil, []time.Time{time.Now()}),
data.NewField("val", nil, []int64{12, 47}),
)
cr, err := ctx.exec()
So(err, ShouldBeNil)
So(cr.Firing, ShouldBeFalse)
})
Convey("Should fire if only first serie matches", func() {
ctx.series = tsdb.TimeSeriesSlice{
tsdb.NewTimeSeries("test1", tsdb.NewTimeSeriesPointsFromArgs(120, 0)),
@@ -144,6 +168,7 @@ type queryConditionTestContext struct {
reducer string
evaluator string
series tsdb.TimeSeriesSlice
frame *data.Frame
result *alerting.EvalContext
condition *QueryCondition
}
@@ -168,10 +193,24 @@ func (ctx *queryConditionTestContext) exec() (*alerting.ConditionResult, error)
ctx.condition = condition
qr := &tsdb.QueryResult{
Series: ctx.series,
}
if ctx.frame != nil {
bFrame, err := data.MarshalArrow(ctx.frame)
if err != nil {
return nil, err
}
qr = &tsdb.QueryResult{
Dataframes: [][]byte{bFrame},
}
}
condition.HandleRequest = func(context context.Context, dsInfo *models.DataSource, req *tsdb.TsdbQuery) (*tsdb.Response, error) {
return &tsdb.Response{
Results: map[string]*tsdb.QueryResult{
"A": {Series: ctx.series},
"A": qr,
},
}, nil
}
pkg/tsdb/frame_util.go
+72
View File
@@ -1,9 +1,12 @@
package tsdb
import (
"fmt"
"time"
"github.com/grafana/grafana-plugin-sdk-go/data"
"github.com/grafana/grafana/pkg/components/null"
"github.com/grafana/grafana/pkg/util/errutil"
)
// SeriesToFrame converts a TimeSeries to a sdk Frame
@@ -35,3 +38,72 @@ func convertTSDBTimePoint(point TimePoint) (t *time.Time, f *float64) {
}
return
}
// FrameToSeriesSlice converts a frame that is a valid time series as per data.TimeSeriesSchema()
// to a TimeSeriesSlice.
func FrameToSeriesSlice(frame *data.Frame) (TimeSeriesSlice, error) {
tsSchema := frame.TimeSeriesSchema()
if tsSchema.Type == data.TimeSeriesTypeNot {
return nil, fmt.Errorf("input frame is not recognized as a time series")
}
// If Long, make wide
if tsSchema.Type == data.TimeSeriesTypeLong {
var err error
frame, err = data.LongToWide(frame)
if err != nil {
return nil, errutil.Wrap("failed to convert long to wide series when converting from dataframe", err)
}
tsSchema = frame.TimeSeriesSchema()
}
seriesCount := len(tsSchema.ValueIndices)
seriesSlice := make(TimeSeriesSlice, 0, seriesCount)
timeField := frame.Fields[tsSchema.TimeIndex]
timeNullFloatSlice := make([]null.Float, timeField.Len())
for i := 0; i < timeField.Len(); i++ { // built slice of time as epoch ms in null floats
tStamp, err := timeField.FloatAt(i)
if err != nil {
return nil, err
}
timeNullFloatSlice[i] = null.FloatFrom(tStamp)
}
for _, fieldIdx := range tsSchema.ValueIndices { // create a TimeSeries for each value Field
field := frame.Fields[fieldIdx]
ts := &TimeSeries{
Name: field.Name,
Tags: field.Labels.Copy(),
Points: make(TimeSeriesPoints, field.Len()),
}
for rowIdx := 0; rowIdx < field.Len(); rowIdx++ { // for each value in the field, make a TimePoint
val, err := field.FloatAt(rowIdx)
if err != nil {
return nil, errutil.Wrapf(err, "failed to convert frame to tsdb.series, can not convert value %v to float", field.At(rowIdx))
}
ts.Points[rowIdx] = TimePoint{
null.FloatFrom(val),
timeNullFloatSlice[rowIdx],
}
}
seriesSlice = append(seriesSlice, ts)
}
return seriesSlice, nil
}
// FramesFromBytes returns a data.Frame slice from marshalled arrow dataframes.
func FramesFromBytes(bFrames [][]byte) ([]*data.Frame, error) {
frames := make([]*data.Frame, len(bFrames))
for i, bFrame := range bFrames {
var err error
frames[i], err = data.UnmarshalArrow(bFrame)
if err != nil {
return nil, err
}
}
return frames, nil
}
pkg/tsdb/frame_util_test.go
+139
View File
@@ -0,0 +1,139 @@
package tsdb
import (
"math"
"testing"
"time"
"github.com/google/go-cmp/cmp"
"github.com/google/go-cmp/cmp/cmpopts"
"github.com/grafana/grafana-plugin-sdk-go/data"
"github.com/grafana/grafana/pkg/components/null"
"github.com/stretchr/testify/require"
"github.com/xorcare/pointer"
)
func TestFrameToSeriesSlice(t *testing.T) {
tests := []struct {
name string
frame *data.Frame
seriesSlice TimeSeriesSlice
Err require.ErrorAssertionFunc
}{
{
name: "a wide series",
frame: data.NewFrame("",
data.NewField("Time", nil, []time.Time{
time.Date(2020, 1, 2, 3, 4, 0, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 30, 0, time.UTC),
}),
data.NewField(`Values Int64s`, data.Labels{"Animal Factor": "cat"}, []*int64{
nil,
pointer.Int64(3),
}),
data.NewField(`Values Floats`, data.Labels{"Animal Factor": "sloth"}, []float64{
2.0,
4.0,
})),
seriesSlice: TimeSeriesSlice{
&TimeSeries{
Name: "Values Int64s",
Tags: map[string]string{"Animal Factor": "cat"},
Points: TimeSeriesPoints{
TimePoint{null.FloatFrom(math.NaN()), null.FloatFrom(1577934240000)},
TimePoint{null.FloatFrom(3), null.FloatFrom(1577934270000)},
},
},
&TimeSeries{
Name: "Values Floats",
Tags: map[string]string{"Animal Factor": "sloth"},
Points: TimeSeriesPoints{
TimePoint{null.FloatFrom(2), null.FloatFrom(1577934240000)},
TimePoint{null.FloatFrom(4), null.FloatFrom(1577934270000)},
},
},
},
Err: require.NoError,
},
{
name: "a long series",
frame: data.NewFrame("",
data.NewField("Time", nil, []time.Time{
time.Date(2020, 1, 2, 3, 4, 0, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 0, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 30, 0, time.UTC),
time.Date(2020, 1, 2, 3, 4, 30, 0, time.UTC),
}),
data.NewField("Values Floats", nil, []float64{
1.0,
2.0,
3.0,
4.0,
}),
data.NewField("Values Int64", nil, []int64{
1,
2,
3,
4,
}),
data.NewField("Animal Factor", nil, []string{
"cat",
"sloth",
"cat",
"sloth",
}),
data.NewField("Location", nil, []string{
"Florida",
"Central & South America",
"Florida",
"Central & South America",
})),
seriesSlice: TimeSeriesSlice{
&TimeSeries{
Name: "Values Floats",
Tags: map[string]string{"Animal Factor": "cat", "Location": "Florida"},
Points: TimeSeriesPoints{
TimePoint{null.FloatFrom(1), null.FloatFrom(1577934240000)},
TimePoint{null.FloatFrom(3), null.FloatFrom(1577934270000)},
},
},
&TimeSeries{
Name: "Values Int64",
Tags: map[string]string{"Animal Factor": "cat", "Location": "Florida"},
Points: TimeSeriesPoints{
TimePoint{null.FloatFrom(1), null.FloatFrom(1577934240000)},
TimePoint{null.FloatFrom(3), null.FloatFrom(1577934270000)},
},
},
&TimeSeries{
Name: "Values Floats",
Tags: map[string]string{"Animal Factor": "sloth", "Location": "Central & South America"},
Points: TimeSeriesPoints{
TimePoint{null.FloatFrom(2), null.FloatFrom(1577934240000)},
TimePoint{null.FloatFrom(4), null.FloatFrom(1577934270000)},
},
},
&TimeSeries{
Name: "Values Int64",
Tags: map[string]string{"Animal Factor": "sloth", "Location": "Central & South America"},
Points: TimeSeriesPoints{
TimePoint{null.FloatFrom(2), null.FloatFrom(1577934240000)},
TimePoint{null.FloatFrom(4), null.FloatFrom(1577934270000)},
},
},
},
Err: require.NoError,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
seriesSlice, err := FrameToSeriesSlice(tt.frame)
tt.Err(t, err)
if diff := cmp.Diff(tt.seriesSlice, seriesSlice, cmpopts.EquateNaNs()); diff != "" {
t.Errorf("Result mismatch (-want +got):\n%s", diff)
}
})
}
}