forked from grafana/grafana
-
Notifications
You must be signed in to change notification settings - Fork 0
/
influxdb.go
142 lines (110 loc) · 3.01 KB
/
influxdb.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
package influxdb
import (
"context"
"encoding/json"
"fmt"
"net/http"
"net/url"
"path"
"golang.org/x/net/context/ctxhttp"
"github.com/grafana/grafana/pkg/log"
"github.com/grafana/grafana/pkg/models"
"github.com/grafana/grafana/pkg/setting"
"github.com/grafana/grafana/pkg/tsdb"
)
type InfluxDBExecutor struct {
*models.DataSource
QueryParser *InfluxdbQueryParser
ResponseParser *ResponseParser
HttpClient *http.Client
}
func NewInfluxDBExecutor(datasource *models.DataSource) (tsdb.Executor, error) {
httpClient, err := datasource.GetHttpClient()
if err != nil {
return nil, err
}
return &InfluxDBExecutor{
DataSource: datasource,
QueryParser: &InfluxdbQueryParser{},
ResponseParser: &ResponseParser{},
HttpClient: httpClient,
}, nil
}
var (
glog log.Logger
)
func init() {
glog = log.New("tsdb.influxdb")
tsdb.RegisterExecutor("influxdb", NewInfluxDBExecutor)
}
func (e *InfluxDBExecutor) Execute(ctx context.Context, queries tsdb.QuerySlice, context *tsdb.QueryContext) *tsdb.BatchResult {
result := &tsdb.BatchResult{}
query, err := e.getQuery(queries, context)
if err != nil {
return result.WithError(err)
}
rawQuery, err := query.Build(context)
if err != nil {
return result.WithError(err)
}
if setting.Env == setting.DEV {
glog.Debug("Influxdb query", "raw query", rawQuery)
}
req, err := e.createRequest(rawQuery)
if err != nil {
return result.WithError(err)
}
resp, err := ctxhttp.Do(ctx, e.HttpClient, req)
if err != nil {
return result.WithError(err)
}
if resp.StatusCode/100 != 2 {
return result.WithError(fmt.Errorf("Influxdb returned statuscode invalid status code: %v", resp.Status))
}
var response Response
dec := json.NewDecoder(resp.Body)
defer resp.Body.Close()
dec.UseNumber()
err = dec.Decode(&response)
if err != nil {
return result.WithError(err)
}
if response.Err != nil {
return result.WithError(response.Err)
}
result.QueryResults = make(map[string]*tsdb.QueryResult)
result.QueryResults["A"] = e.ResponseParser.Parse(&response, query)
return result
}
func (e *InfluxDBExecutor) getQuery(queries tsdb.QuerySlice, context *tsdb.QueryContext) (*Query, error) {
for _, v := range queries {
query, err := e.QueryParser.Parse(v.Model, e.DataSource)
if err != nil {
return nil, err
}
return query, nil
}
return nil, fmt.Errorf("query request contains no queries")
}
func (e *InfluxDBExecutor) createRequest(query string) (*http.Request, error) {
u, _ := url.Parse(e.Url)
u.Path = path.Join(u.Path, "query")
req, err := http.NewRequest(http.MethodGet, u.String(), nil)
if err != nil {
return nil, err
}
params := req.URL.Query()
params.Set("q", query)
params.Set("db", e.Database)
params.Set("epoch", "s")
req.URL.RawQuery = params.Encode()
req.Header.Set("User-Agent", "Grafana")
if e.BasicAuth {
req.SetBasicAuth(e.BasicAuthUser, e.BasicAuthPassword)
}
if !e.BasicAuth && e.User != "" {
req.SetBasicAuth(e.User, e.Password)
}
glog.Debug("Influxdb request", "url", req.URL.String())
return req, nil
}