Update query API to support timeouts (#1014)

* Add timeout parameter for queries

Signed-off-by: Joseph Woodward <joseph.woodward@xeuse.com>

* Update api/prometheus/v1/api.go

Co-authored-by: Kemal Akkoyun <kakkoyun@users.noreply.github.com>
Signed-off-by: Joseph Woodward <joseph.woodward@xeuse.com>

* Update api/prometheus/v1/api.go

Co-authored-by: Kemal Akkoyun <kakkoyun@users.noreply.github.com>
Signed-off-by: Joseph Woodward <joseph.woodward@xeuse.com>

* Pass timeout as stringified time.Duration instead of millisecond value

Signed-off-by: Joseph Woodward <joseph.woodward@xeuse.com>

* Update QueryRange API to support timeouts

Signed-off-by: Joseph Woodward <joseph.woodward@xeuse.com>

* Add timeout to test request params

Signed-off-by: Joseph Woodward <joseph.woodward@xeuse.com>

Co-authored-by: Kemal Akkoyun <kakkoyun@users.noreply.github.com>
This commit is contained in:
Joseph Woodward 2022-04-21 06:23:16 +01:00 committed by GitHub
parent 11ee9add27
commit 48a686a603
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 53 additions and 18 deletions

View File

@ -238,9 +238,9 @@ type API interface {
// LabelValues performs a query for the values of the given label, time range and matchers. // LabelValues performs a query for the values of the given label, time range and matchers.
LabelValues(ctx context.Context, label string, matches []string, startTime time.Time, endTime time.Time) (model.LabelValues, Warnings, error) LabelValues(ctx context.Context, label string, matches []string, startTime time.Time, endTime time.Time) (model.LabelValues, Warnings, error)
// Query performs a query for the given time. // Query performs a query for the given time.
Query(ctx context.Context, query string, ts time.Time) (model.Value, Warnings, error) Query(ctx context.Context, query string, ts time.Time, opts ...Option) (model.Value, Warnings, error)
// QueryRange performs a query for the given range. // QueryRange performs a query for the given range.
QueryRange(ctx context.Context, query string, r Range) (model.Value, Warnings, error) QueryRange(ctx context.Context, query string, r Range, opts ...Option) (model.Value, Warnings, error)
// QueryExemplars performs a query for exemplars by the given query and time range. // QueryExemplars performs a query for exemplars by the given query and time range.
QueryExemplars(ctx context.Context, query string, startTime time.Time, endTime time.Time) ([]ExemplarQueryResult, error) QueryExemplars(ctx context.Context, query string, startTime time.Time, endTime time.Time) ([]ExemplarQueryResult, error)
// Buildinfo returns various build information properties about the Prometheus server // Buildinfo returns various build information properties about the Prometheus server
@ -818,10 +818,33 @@ func (h *httpAPI) LabelValues(ctx context.Context, label string, matches []strin
return labelValues, w, json.Unmarshal(body, &labelValues) return labelValues, w, json.Unmarshal(body, &labelValues)
} }
func (h *httpAPI) Query(ctx context.Context, query string, ts time.Time) (model.Value, Warnings, error) { type apiOptions struct {
timeout time.Duration
}
type Option func(c *apiOptions)
func WithTimeout(timeout time.Duration) Option {
return func(o *apiOptions) {
o.timeout = timeout
}
}
func (h *httpAPI) Query(ctx context.Context, query string, ts time.Time, opts ...Option) (model.Value, Warnings, error) {
u := h.client.URL(epQuery, nil) u := h.client.URL(epQuery, nil)
q := u.Query() q := u.Query()
opt := &apiOptions{}
for _, o := range opts {
o(opt)
}
d := opt.timeout
if d > 0 {
q.Set("timeout", d.String())
}
q.Set("query", query) q.Set("query", query)
if !ts.IsZero() { if !ts.IsZero() {
q.Set("time", formatTime(ts)) q.Set("time", formatTime(ts))
@ -836,7 +859,7 @@ func (h *httpAPI) Query(ctx context.Context, query string, ts time.Time) (model.
return model.Value(qres.v), warnings, json.Unmarshal(body, &qres) return model.Value(qres.v), warnings, json.Unmarshal(body, &qres)
} }
func (h *httpAPI) QueryRange(ctx context.Context, query string, r Range) (model.Value, Warnings, error) { func (h *httpAPI) QueryRange(ctx context.Context, query string, r Range, opts ...Option) (model.Value, Warnings, error) {
u := h.client.URL(epQueryRange, nil) u := h.client.URL(epQueryRange, nil)
q := u.Query() q := u.Query()
@ -845,6 +868,16 @@ func (h *httpAPI) QueryRange(ctx context.Context, query string, r Range) (model.
q.Set("end", formatTime(r.End)) q.Set("end", formatTime(r.End))
q.Set("step", strconv.FormatFloat(r.Step.Seconds(), 'f', -1, 64)) q.Set("step", strconv.FormatFloat(r.Step.Seconds(), 'f', -1, 64))
opt := &apiOptions{}
for _, o := range opts {
o(opt)
}
d := opt.timeout
if d > 0 {
q.Set("timeout", d.String())
}
_, body, warnings, err := h.client.DoGetFallback(ctx, u, q) _, body, warnings, err := h.client.DoGetFallback(ctx, u, q)
if err != nil { if err != nil {
return nil, warnings, err return nil, warnings, err

View File

@ -170,15 +170,15 @@ func TestAPIs(t *testing.T) {
} }
} }
doQuery := func(q string, ts time.Time) func() (interface{}, Warnings, error) { doQuery := func(q string, ts time.Time, opts ...Option) func() (interface{}, Warnings, error) {
return func() (interface{}, Warnings, error) { return func() (interface{}, Warnings, error) {
return promAPI.Query(context.Background(), q, ts) return promAPI.Query(context.Background(), q, ts, opts...)
} }
} }
doQueryRange := func(q string, rng Range) func() (interface{}, Warnings, error) { doQueryRange := func(q string, rng Range, opts ...Option) func() (interface{}, Warnings, error) {
return func() (interface{}, Warnings, error) { return func() (interface{}, Warnings, error) {
return promAPI.QueryRange(context.Background(), q, rng) return promAPI.QueryRange(context.Background(), q, rng, opts...)
} }
} }
@ -246,7 +246,7 @@ func TestAPIs(t *testing.T) {
queryTests := []apiTest{ queryTests := []apiTest{
{ {
do: doQuery("2", testTime), do: doQuery("2", testTime, WithTimeout(5*time.Second)),
inRes: &queryResult{ inRes: &queryResult{
Type: model.ValScalar, Type: model.ValScalar,
Result: &model.Scalar{ Result: &model.Scalar{
@ -260,6 +260,7 @@ func TestAPIs(t *testing.T) {
reqParam: url.Values{ reqParam: url.Values{
"query": []string{"2"}, "query": []string{"2"},
"time": []string{testTime.Format(time.RFC3339Nano)}, "time": []string{testTime.Format(time.RFC3339Nano)},
"timeout": []string{(5 * time.Second).String()},
}, },
res: &model.Scalar{ res: &model.Scalar{
Value: 2, Value: 2,
@ -365,7 +366,7 @@ func TestAPIs(t *testing.T) {
Start: testTime.Add(-time.Minute), Start: testTime.Add(-time.Minute),
End: testTime, End: testTime,
Step: time.Minute, Step: time.Minute,
}), }, WithTimeout(5*time.Second)),
inErr: fmt.Errorf("some error"), inErr: fmt.Errorf("some error"),
reqMethod: "POST", reqMethod: "POST",
@ -375,6 +376,7 @@ func TestAPIs(t *testing.T) {
"start": []string{testTime.Add(-time.Minute).Format(time.RFC3339Nano)}, "start": []string{testTime.Add(-time.Minute).Format(time.RFC3339Nano)},
"end": []string{testTime.Format(time.RFC3339Nano)}, "end": []string{testTime.Format(time.RFC3339Nano)},
"step": []string{time.Minute.String()}, "step": []string{time.Minute.String()},
"timeout": []string{(5 * time.Second).String()},
}, },
err: fmt.Errorf("some error"), err: fmt.Errorf("some error"),
}, },

View File

@ -39,7 +39,7 @@ func ExampleAPI_query() {
v1api := v1.NewAPI(client) v1api := v1.NewAPI(client)
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel() defer cancel()
result, warnings, err := v1api.Query(ctx, "up", time.Now()) result, warnings, err := v1api.Query(ctx, "up", time.Now(), v1.WithTimeout(5*time.Second))
if err != nil { if err != nil {
fmt.Printf("Error querying Prometheus: %v\n", err) fmt.Printf("Error querying Prometheus: %v\n", err)
os.Exit(1) os.Exit(1)
@ -67,7 +67,7 @@ func ExampleAPI_queryRange() {
End: time.Now(), End: time.Now(),
Step: time.Minute, Step: time.Minute,
} }
result, warnings, err := v1api.QueryRange(ctx, "rate(prometheus_tsdb_head_samples_appended_total[5m])", r) result, warnings, err := v1api.QueryRange(ctx, "rate(prometheus_tsdb_head_samples_appended_total[5m])", r, v1.WithTimeout(5*time.Second))
if err != nil { if err != nil {
fmt.Printf("Error querying Prometheus: %v\n", err) fmt.Printf("Error querying Prometheus: %v\n", err)
os.Exit(1) os.Exit(1)