Skip to content

Commit d8aa38f

Browse files
bergquistbergquist
committed
tech(influxdb): refactor and cleanup
1 parent ab87517 commit d8aa38f

File tree

13 files changed

+323
-233
lines changed

13 files changed

+323
-233
lines changed

‎pkg/services/alerting/conditions/reducer_test.go‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,8 @@ package conditions
33
import (
44
"testing"
55

6+
"gopkg.in/guregu/null.v3"
7+
68
"github.com/grafana/grafana/pkg/tsdb"
79
. "github.com/smartystreets/goconvey/convey"
810
)
@@ -43,7 +45,7 @@ func testReducer(typ string, datapoints ...float64) float64 {
4345
}
4446

4547
for idx := range datapoints {
46-
series.Points = append(series.Points, tsdb.NewTimePoint(datapoints[idx], 1234134))
48+
series.Points = append(series.Points, tsdb.NewTimePoint(null.FloatFrom(datapoints[idx]), 1234134))
4749
}
4850

4951
return reducer.Reduce(series).Float64

‎pkg/tsdb/influxdb/influxdb.go‎

Lines changed: 28 additions & 89 deletions
Original file line numberDiff line numberDiff line change
@@ -10,8 +10,6 @@ import (
1010
"path"
1111
"time"
1212

13-
"gopkg.in/guregu/null.v3"
14-
1513
"golang.org/x/net/context/ctxhttp"
1614

1715
"github.com/grafana/grafana/pkg/log"
@@ -21,14 +19,14 @@ import (
2119
type InfluxDBExecutor struct {
2220
*tsdb.DataSourceInfo
2321
QueryParser *InfluxdbQueryParser
24-
QueryBuilder *QueryBuild
22+
QueryBuilder *QueryBuilder
2523
}
2624

2725
func NewInfluxDBExecutor(dsInfo *tsdb.DataSourceInfo) tsdb.Executor {
2826
return &InfluxDBExecutor{
2927
DataSourceInfo: dsInfo,
3028
QueryParser: &InfluxdbQueryParser{},
31-
QueryBuilder: &QueryBuild{},
29+
QueryBuilder: &QueryBuilder{},
3230
}
3331
}
3432

@@ -66,126 +64,67 @@ func (e *InfluxDBExecutor) getQuery(queries tsdb.QuerySlice, context *tsdb.Query
6664
return rawQuery, nil
6765
}
6866

69-
return "", fmt.Errorf("Tsdb request contains no queries")
67+
return "", fmt.Errorf("query request contains no queries")
7068
}
7169

72-
func (e *InfluxDBExecutor) Execute(ctx context.Context, queries tsdb.QuerySlice, context *tsdb.QueryContext) *tsdb.BatchResult {
73-
result := &tsdb.BatchResult{}
74-
75-
query, err := e.getQuery(queries, context)
76-
if err != nil {
77-
result.Error = err
78-
return result
79-
}
80-
81-
glog.Info("Influxdb", "query", query)
82-
70+
func (e *InfluxDBExecutor) createRequest(query string) (*http.Request, error) {
8371
u, _ := url.Parse(e.Url)
8472
u.Path = path.Join(u.Path, "query")
8573

8674
req, err := http.NewRequest(http.MethodGet, u.String(), nil)
8775
if err != nil {
88-
result.Error = err
89-
return result
76+
return nil, err
9077
}
9178

9279
params := req.URL.Query()
9380
params.Set("q", query)
9481
params.Set("db", e.Database)
9582
params.Set("epoch", "s")
96-
9783
req.URL.RawQuery = params.Encode()
9884

99-
req.Header.Set("Content-Type", "")
10085
req.Header.Set("User-Agent", "Grafana")
10186
if e.BasicAuth {
10287
req.SetBasicAuth(e.BasicAuthUser, e.BasicAuthPassword)
10388
}
10489

105-
glog.Info("influxdb request", "url", req.URL.String())
90+
glog.Debug("influxdb request", "url", req.URL.String())
91+
return req, nil
92+
}
93+
94+
func (e *InfluxDBExecutor) Execute(ctx context.Context, queries tsdb.QuerySlice, context *tsdb.QueryContext) *tsdb.BatchResult {
95+
result := &tsdb.BatchResult{}
96+
97+
query, err := e.getQuery(queries, context)
98+
if err != nil {
99+
return result.WithError(err)
100+
}
101+
102+
glog.Debug("Influxdb query", "raw query", query)
103+
104+
req, err := e.createRequest(query)
105+
if err != nil {
106+
return result.WithError(err)
107+
}
108+
106109
resp, err := ctxhttp.Do(ctx, HttpClient, req)
107110
if err != nil {
108-
result.Error = err
109-
return result
111+
return result.WithError(err)
110112
}
111113

112114
if resp.StatusCode/100 != 2 {
113-
result.Error = fmt.Errorf("Influxdb returned statuscode %v body %v", resp.Status)
114-
return result
115+
return result.WithError(fmt.Errorf("Influxdb returned statuscode invalid status code: %v", resp.Status))
115116
}
116117

117118
var response Response
118119
dec := json.NewDecoder(resp.Body)
119120
dec.UseNumber()
120121
err = dec.Decode(&response)
121122
if err != nil {
122-
glog.Error("Influxdb decode failed", "err", err)
123-
result.Error = err
124-
return result
123+
return result.WithError(err)
125124
}
126125

127126
result.QueryResults = make(map[string]*tsdb.QueryResult)
128-
queryRes := tsdb.NewQueryResult()
129-
130-
for _, v := range response.Results {
131-
for _, r := range v.Series {
132-
serie := tsdb.TimeSeries{Name: r.Name}
133-
var points tsdb.TimeSeriesPoints
134-
135-
for _, k := range r.Values {
136-
var value null.Float
137-
var err error
138-
num, ok := k[1].(json.Number)
139-
if !ok {
140-
value = null.FloatFromPtr(nil)
141-
} else {
142-
fvalue, err := num.Float64()
143-
if err == nil {
144-
value = null.FloatFrom(fvalue)
145-
}
146-
}
147-
148-
pos0, ok := k[0].(json.Number)
149-
timestamp, err := pos0.Float64()
150-
if err == nil && ok {
151-
points = append(points, tsdb.NewTimePoint(value, timestamp))
152-
} else {
153-
glog.Error("Failed to convert response", "err1", err, "ok", ok, "timestamp", timestamp, "value", value.Float64)
154-
}
155-
serie.Points = points
156-
}
157-
queryRes.Series = append(queryRes.Series, &serie)
158-
}
159-
}
160-
161-
for _, v := range queryRes.Series {
162-
glog.Info("result", "name", v.Name, "points", v.Points)
163-
}
164-
165-
result.QueryResults["A"] = queryRes
127+
result.QueryResults["A"] = ParseQueryResult(&response)
166128

167129
return result
168130
}
169-
170-
type Response struct {
171-
Results []Result
172-
Err error
173-
}
174-
175-
type Result struct {
176-
Series []Row
177-
Messages []*Message
178-
Err error
179-
}
180-
181-
type Message struct {
182-
Level string `json:"level,omitempty"`
183-
Text string `json:"text,omitempty"`
184-
}
185-
186-
type Row struct {
187-
Name string `json:"name,omitempty"`
188-
Tags map[string]string `json:"tags,omitempty"`
189-
Columns []string `json:"columns,omitempty"`
190-
Values [][]interface{} `json:"values,omitempty"`
191-
}

‎pkg/tsdb/influxdb/parser.go‎ renamed to ‎pkg/tsdb/influxdb/model_parser.go‎

File renamed without changes.

‎pkg/tsdb/influxdb/model_parser_test.go‎

Lines changed: 115 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,115 @@
1+
package influxdb
2+
3+
import (
4+
"testing"
5+
6+
"github.com/grafana/grafana/pkg/components/simplejson"
7+
. "github.com/smartystreets/goconvey/convey"
8+
)
9+
10+
func TestInfluxdbQueryParser(t *testing.T) {
11+
Convey("Influxdb query parser", t, func() {
12+
13+
parser := &InfluxdbQueryParser{}
14+
15+
Convey("can parse influxdb json model", func() {
16+
json := `
17+
{
18+
"dsType": "influxdb",
19+
"groupBy": [
20+
{
21+
"params": [
22+
"$interval"
23+
],
24+
"type": "time"
25+
},
26+
{
27+
"params": [
28+
"datacenter"
29+
],
30+
"type": "tag"
31+
},
32+
{
33+
"params": [
34+
"none"
35+
],
36+
"type": "fill"
37+
}
38+
],
39+
"measurement": "logins.count",
40+
"policy": "default",
41+
"refId": "B",
42+
"resultFormat": "time_series",
43+
"select": [
44+
[
45+
{
46+
"type": "field",
47+
"params": [
48+
"value"
49+
]
50+
},
51+
{
52+
"type": "count",
53+
"params": []
54+
}
55+
],
56+
[
57+
{
58+
"type": "field",
59+
"params": [
60+
"value"
61+
]
62+
},
63+
{
64+
"type": "bottom",
65+
"params": [
66+
3
67+
]
68+
}
69+
],
70+
[
71+
{
72+
"type": "field",
73+
"params": [
74+
"value"
75+
]
76+
},
77+
{
78+
"type": "mean",
79+
"params": []
80+
},
81+
{
82+
"type": "math",
83+
"params": [
84+
" / 100"
85+
]
86+
}
87+
]
88+
],
89+
"tags": [
90+
{
91+
"key": "datacenter",
92+
"operator": "=",
93+
"value": "America"
94+
},
95+
{
96+
"condition": "OR",
97+
"key": "hostname",
98+
"operator": "=",
99+
"value": "server1"
100+
}
101+
]
102+
}
103+
`
104+
105+
modelJson, err := simplejson.NewJson([]byte(json))
106+
So(err, ShouldBeNil)
107+
108+
res, err := parser.Parse(modelJson)
109+
So(err, ShouldBeNil)
110+
So(len(res.GroupBy), ShouldEqual, 3)
111+
So(len(res.Selects), ShouldEqual, 3)
112+
So(len(res.Tags), ShouldEqual, 2)
113+
})
114+
})
115+
}

‎pkg/tsdb/influxdb/models.go‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,3 +23,26 @@ type Select []QueryPart
2323
type InfluxDbSelect struct {
2424
Type string
2525
}
26+
27+
type Response struct {
28+
Results []Result
29+
Err error
30+
}
31+
32+
type Result struct {
33+
Series []Row
34+
Messages []*Message
35+
Err error
36+
}
37+
38+
type Message struct {
39+
Level string `json:"level,omitempty"`
40+
Text string `json:"text,omitempty"`
41+
}
42+
43+
type Row struct {
44+
Name string `json:"name,omitempty"`
45+
Tags map[string]string `json:"tags,omitempty"`
46+
Columns []string `json:"columns,omitempty"`
47+
Values [][]interface{} `json:"values,omitempty"`
48+
}

0 commit comments

Comments
 (0)