Skip to content

Commit 63bf332

Browse files
bergquistbergquist
committed
Merge branch 'alerting_mqe'
2 parents b620c74 + e6c557b commit 63bf332

File tree

16 files changed

+2326
-0
lines changed

16 files changed

+2326
-0
lines changed

‎pkg/cmd/grafana-server/main.go

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ import (
2121
_ "github.com/grafana/grafana/pkg/services/alerting/notifiers"
2222
_ "github.com/grafana/grafana/pkg/tsdb/graphite"
2323
_ "github.com/grafana/grafana/pkg/tsdb/influxdb"
24+
_ "github.com/grafana/grafana/pkg/tsdb/mqe"
2425
_ "github.com/grafana/grafana/pkg/tsdb/opentsdb"
2526
_ "github.com/grafana/grafana/pkg/tsdb/prometheus"
2627
_ "github.com/grafana/grafana/pkg/tsdb/testdata"

‎pkg/tsdb/mqe/model_parser.go

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
package mqe
2+
3+
import (
4+
"github.com/grafana/grafana/pkg/components/simplejson"
5+
"github.com/grafana/grafana/pkg/models"
6+
"github.com/grafana/grafana/pkg/tsdb"
7+
)
8+
9+
func NewQueryParser() *QueryParser {
10+
return &QueryParser{}
11+
}
12+
13+
type QueryParser struct{}
14+
15+
func (qp *QueryParser) Parse(model *simplejson.Json, dsInfo *models.DataSource, queryContext *tsdb.QueryContext) (*Query, error) {
16+
query := &Query{TimeRange: queryContext.TimeRange}
17+
query.AddAppToAlias = model.Get("addAppToAlias").MustBool(false)
18+
query.AddHostToAlias = model.Get("addHostToAlias").MustBool(false)
19+
query.UseRawQuery = model.Get("rawQuery").MustBool(false)
20+
query.RawQuery = model.Get("query").MustString("")
21+
22+
query.Apps = model.Get("apps").MustStringArray([]string{})
23+
query.Hosts = model.Get("hosts").MustStringArray([]string{})
24+
25+
var metrics []Metric
26+
var err error
27+
for _, metricsObj := range model.Get("metrics").MustArray() {
28+
metricJson := simplejson.NewFromAny(metricsObj)
29+
var m Metric
30+
31+
m.Alias = metricJson.Get("alias").MustString("")
32+
m.Metric, err = metricJson.Get("metric").String()
33+
if err != nil {
34+
return nil, err
35+
}
36+
37+
metrics = append(metrics, m)
38+
}
39+
40+
query.Metrics = metrics
41+
42+
return query, nil
43+
}

‎pkg/tsdb/mqe/model_parser_test.go

Lines changed: 119 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,119 @@
1+
package mqe
2+
3+
import (
4+
"testing"
5+
6+
"github.com/grafana/grafana/pkg/components/simplejson"
7+
"github.com/grafana/grafana/pkg/models"
8+
"github.com/grafana/grafana/pkg/tsdb"
9+
. "github.com/smartystreets/goconvey/convey"
10+
)
11+
12+
func TestMQEQueryParser(t *testing.T) {
13+
Convey("MQE query parser", t, func() {
14+
parser := &QueryParser{}
15+
16+
dsInfo := &models.DataSource{JsonData: simplejson.New()}
17+
queryContext := &tsdb.QueryContext{}
18+
19+
Convey("can parse simple mqe model", func() {
20+
json := `
21+
{
22+
"apps": [],
23+
"hosts": [
24+
"staples-lab-1"
25+
],
26+
"metrics": [
27+
{
28+
"metric": "os.cpu.all*"
29+
}
30+
],
31+
"rawQuery": "",
32+
"refId": "A"
33+
}
34+
`
35+
modelJson, err := simplejson.NewJson([]byte(json))
36+
So(err, ShouldBeNil)
37+
38+
query, err := parser.Parse(modelJson, dsInfo, queryContext)
39+
So(err, ShouldBeNil)
40+
So(query.UseRawQuery, ShouldBeFalse)
41+
42+
So(len(query.Apps), ShouldEqual, 0)
43+
So(query.Hosts[0], ShouldEqual, "staples-lab-1")
44+
So(query.Metrics[0].Metric, ShouldEqual, "os.cpu.all*")
45+
})
46+
47+
Convey("can parse multi serie mqe model", func() {
48+
json := `
49+
{
50+
"apps": [
51+
"demoapp"
52+
],
53+
"hosts": [
54+
"staples-lab-1"
55+
],
56+
"metrics": [
57+
{
58+
"metric": "os.cpu.all.active_percentage"
59+
},
60+
{
61+
"metric": "os.disk.sda.io_time"
62+
}
63+
],
64+
"rawQuery": "",
65+
"refId": "A",
66+
"addAppToAlias": true,
67+
"addHostToAlias": true
68+
}
69+
`
70+
modelJson, err := simplejson.NewJson([]byte(json))
71+
So(err, ShouldBeNil)
72+
73+
query, err := parser.Parse(modelJson, dsInfo, queryContext)
74+
So(err, ShouldBeNil)
75+
So(query.UseRawQuery, ShouldBeFalse)
76+
So(query.Apps[0], ShouldEqual, "demoapp")
77+
So(query.Metrics[0].Metric, ShouldEqual, "os.cpu.all.active_percentage")
78+
So(query.Metrics[1].Metric, ShouldEqual, "os.disk.sda.io_time")
79+
})
80+
81+
Convey("can parse raw query", func() {
82+
json := `
83+
{
84+
"addAppToAlias": true,
85+
"addHostToAlias": true,
86+
"apps": [],
87+
"hosts": [
88+
"staples-lab-1"
89+
],
90+
"metrics": [
91+
{
92+
"alias": "cpu active",
93+
"metric": "os.cpu.all.active_percentage"
94+
},
95+
{
96+
"alias": "disk sda time",
97+
"metric": "os.disk.sda.io_time"
98+
}
99+
],
100+
"rawQuery": true,
101+
"query": "raw-query",
102+
"refId": "A",
103+
"addAppToAlias": true,
104+
"addHostToAlias": true
105+
}
106+
`
107+
modelJson, err := simplejson.NewJson([]byte(json))
108+
So(err, ShouldBeNil)
109+
110+
query, err := parser.Parse(modelJson, dsInfo, queryContext)
111+
So(err, ShouldBeNil)
112+
113+
So(query.UseRawQuery, ShouldBeTrue)
114+
So(query.RawQuery, ShouldEqual, "raw-query")
115+
So(query.AddAppToAlias, ShouldBeTrue)
116+
So(query.AddHostToAlias, ShouldBeTrue)
117+
})
118+
})
119+
}

‎pkg/tsdb/mqe/mqe.go

Lines changed: 144 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,144 @@
1+
package mqe
2+
3+
import (
4+
"context"
5+
"net/http"
6+
"net/url"
7+
"path"
8+
"strings"
9+
10+
"golang.org/x/net/context/ctxhttp"
11+
12+
"github.com/grafana/grafana/pkg/components/simplejson"
13+
"github.com/grafana/grafana/pkg/log"
14+
"github.com/grafana/grafana/pkg/models"
15+
"github.com/grafana/grafana/pkg/setting"
16+
"github.com/grafana/grafana/pkg/tsdb"
17+
)
18+
19+
/*
20+
TODO:
21+
* performance. outgoing requests in pararell.
22+
* frontend plugin. targetContainsTemplates
23+
*/
24+
25+
type MQEExecutor struct {
26+
*models.DataSource
27+
queryParser *QueryParser
28+
responseParser *ResponseParser
29+
httpClient *http.Client
30+
log log.Logger
31+
tokenClient *TokenClient
32+
}
33+
34+
func NewMQEExecutor(dsInfo *models.DataSource) (tsdb.Executor, error) {
35+
httpclient, err := dsInfo.GetHttpClient()
36+
if err != nil {
37+
return nil, err
38+
}
39+
40+
return &MQEExecutor{
41+
DataSource: dsInfo,
42+
httpClient: httpclient,
43+
log: log.New("tsdb.mqe"),
44+
queryParser: NewQueryParser(),
45+
responseParser: NewResponseParser(),
46+
tokenClient: NewTokenClient(dsInfo),
47+
}, nil
48+
}
49+
50+
func init() {
51+
tsdb.RegisterExecutor("mqe-datasource", NewMQEExecutor)
52+
}
53+
54+
type QueryToSend struct {
55+
RawQuery string
56+
QueryRef *Query
57+
}
58+
59+
func (e *MQEExecutor) Execute(ctx context.Context, queries tsdb.QuerySlice, queryContext *tsdb.QueryContext) *tsdb.BatchResult {
60+
result := &tsdb.BatchResult{}
61+
62+
availableSeries, err := e.tokenClient.GetTokenData(ctx)
63+
if err != nil {
64+
return result.WithError(err)
65+
}
66+
67+
var mqeQueries []*Query
68+
for _, v := range queries {
69+
q, err := e.queryParser.Parse(v.Model, e.DataSource, queryContext)
70+
if err != nil {
71+
return result.WithError(err)
72+
}
73+
mqeQueries = append(mqeQueries, q)
74+
}
75+
76+
var rawQueries []QueryToSend
77+
for _, v := range mqeQueries {
78+
queries, err := v.Build(availableSeries.Metrics)
79+
if err != nil {
80+
return result.WithError(err)
81+
}
82+
83+
rawQueries = append(rawQueries, queries...)
84+
}
85+
86+
e.log.Debug("Sending request", "url", e.DataSource.Url)
87+
88+
queryResult := &tsdb.QueryResult{}
89+
for _, v := range rawQueries {
90+
if setting.Env == setting.DEV {
91+
e.log.Debug("Executing", "query", v)
92+
}
93+
94+
req, err := e.createRequest(v.RawQuery)
95+
96+
resp, err := ctxhttp.Do(ctx, e.httpClient, req)
97+
if err != nil {
98+
return result.WithError(err)
99+
}
100+
101+
series, err := e.responseParser.Parse(resp, v.QueryRef)
102+
if err != nil {
103+
return result.WithError(err)
104+
}
105+
106+
queryResult.Series = append(queryResult.Series, series.Series...)
107+
}
108+
109+
result.QueryResults = make(map[string]*tsdb.QueryResult)
110+
result.QueryResults["A"] = queryResult
111+
112+
return result
113+
}
114+
115+
func (e *MQEExecutor) createRequest(query string) (*http.Request, error) {
116+
u, err := url.Parse(e.Url)
117+
if err != nil {
118+
return nil, err
119+
}
120+
121+
u.Path = path.Join(u.Path, "query")
122+
123+
payload := simplejson.New()
124+
payload.Set("query", query)
125+
126+
jsonPayload, err := payload.MarshalJSON()
127+
if err != nil {
128+
return nil, err
129+
}
130+
131+
req, err := http.NewRequest(http.MethodPost, u.String(), strings.NewReader(string(jsonPayload)))
132+
if err != nil {
133+
return nil, err
134+
}
135+
136+
req.Header.Set("User-Agent", "Grafana")
137+
req.Header.Set("Content-Type", "application/json")
138+
139+
if e.BasicAuth {
140+
req.SetBasicAuth(e.BasicAuthUser, e.BasicAuthPassword)
141+
}
142+
143+
return req, nil
144+
}

0 commit comments

Comments
 (0)