forked from grafana/grafana
-
Notifications
You must be signed in to change notification settings - Fork 0
/
httpClient.go
129 lines (105 loc) · 2.79 KB
/
httpClient.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
package mqe
import (
"context"
"net/http"
"net/url"
"path"
"strings"
"github.com/grafana/grafana/pkg/components/simplejson"
"github.com/grafana/grafana/pkg/log"
"github.com/grafana/grafana/pkg/models"
"github.com/grafana/grafana/pkg/setting"
"github.com/grafana/grafana/pkg/tsdb"
"golang.org/x/net/context/ctxhttp"
)
var (
MaxWorker int = 4
)
type apiClient struct {
*models.DataSource
log log.Logger
httpClient *http.Client
responseParser *ResponseParser
}
func NewApiClient(httpClient *http.Client, datasource *models.DataSource) *apiClient {
return &apiClient{
DataSource: datasource,
log: log.New("tsdb.mqe"),
httpClient: httpClient,
responseParser: NewResponseParser(),
}
}
func (e *apiClient) PerformRequests(ctx context.Context, queries []QueryToSend) (*tsdb.QueryResult, error) {
queryResult := &tsdb.QueryResult{}
queryCount := len(queries)
jobsChan := make(chan QueryToSend, queryCount)
resultChan := make(chan []*tsdb.TimeSeries, queryCount)
errorsChan := make(chan error, 1)
for w := 1; w <= MaxWorker; w++ {
go e.spawnWorker(ctx, w, jobsChan, resultChan, errorsChan)
}
for _, v := range queries {
jobsChan <- v
}
close(jobsChan)
resultCounter := 0
for {
select {
case timeseries := <-resultChan:
queryResult.Series = append(queryResult.Series, timeseries...)
resultCounter++
if resultCounter == queryCount {
close(resultChan)
return queryResult, nil
}
case err := <-errorsChan:
return nil, err
case <-ctx.Done():
return nil, ctx.Err()
}
}
}
func (e *apiClient) spawnWorker(ctx context.Context, id int, jobs chan QueryToSend, results chan []*tsdb.TimeSeries, errors chan error) {
e.log.Debug("Spawning worker", "id", id)
for query := range jobs {
if setting.Env == setting.DEV {
e.log.Debug("Sending request", "query", query.RawQuery)
}
req, err := e.createRequest(query.RawQuery)
resp, err := ctxhttp.Do(ctx, e.httpClient, req)
if err != nil {
errors <- err
return
}
series, err := e.responseParser.Parse(resp, query)
if err != nil {
errors <- err
return
}
results <- series
}
e.log.Debug("Worker is complete", "id", id)
}
func (e *apiClient) createRequest(query string) (*http.Request, error) {
u, err := url.Parse(e.Url)
if err != nil {
return nil, err
}
u.Path = path.Join(u.Path, "query")
payload := simplejson.New()
payload.Set("query", query)
jsonPayload, err := payload.MarshalJSON()
if err != nil {
return nil, err
}
req, err := http.NewRequest(http.MethodPost, u.String(), strings.NewReader(string(jsonPayload)))
if err != nil {
return nil, err
}
req.Header.Set("User-Agent", "Grafana")
req.Header.Set("Content-Type", "application/json")
if e.BasicAuth {
req.SetBasicAuth(e.BasicAuthUser, e.BasicAuthPassword)
}
return req, nil
}