mirror of
https://github.com/VictoriaMetrics/VictoriaMetrics.git
synced 2025-03-11 15:34:56 +00:00
Vmalert: adjust time
param for datasource queries according to evaluationInterval
(#1257)
* Simplify arguments list for fn `queryDataSource` to improve readbility * vmalert: adjust `time` param according to rule evaluation interval With this change, vmalert will start to use rule's evaluation interval for truncating the `time` param. This is mostly needed to produce consistent time series with timestamps unaffected by vmalert start time. Now, timestamp becomes predictable. Additionally, adjustment is similar to what Grafana does for plotting range graphs. Hence, recording rule series and recording rule expression plotted in grafana suppose to become similar in most of cases.
This commit is contained in:
parent
8be1cb297b
commit
b55677e93d
4 changed files with 209 additions and 48 deletions
|
@ -62,9 +62,12 @@ func newAlertingRule(qb datasource.QuerierBuilder, group *Group, cfg config.Rule
|
|||
Annotations: cfg.Annotations,
|
||||
GroupID: group.ID(),
|
||||
GroupName: group.Name,
|
||||
q: qb.BuildWithParams(datasource.QuerierParams{DataSourceType: &cfg.Type}),
|
||||
alerts: make(map[uint64]*notifier.Alert),
|
||||
metrics: &alertingRuleMetrics{},
|
||||
q: qb.BuildWithParams(datasource.QuerierParams{
|
||||
DataSourceType: &cfg.Type,
|
||||
EvaluationInterval: group.Interval,
|
||||
}),
|
||||
alerts: make(map[uint64]*notifier.Alert),
|
||||
metrics: &alertingRuleMetrics{},
|
||||
}
|
||||
|
||||
labels := fmt.Sprintf(`alertname=%q, group=%q, id="%d"`, ar.Name, group.Name, ar.ID())
|
||||
|
|
|
@ -81,7 +81,9 @@ type VMStorage struct {
|
|||
appendTypePrefix bool
|
||||
lookBack time.Duration
|
||||
queryStep time.Duration
|
||||
dataSourceType Type
|
||||
|
||||
dataSourceType Type
|
||||
evaluationInterval time.Duration
|
||||
}
|
||||
|
||||
const queryPath = "/api/v1/query"
|
||||
|
@ -92,7 +94,8 @@ const graphitePrefix = "/graphite"
|
|||
|
||||
// QuerierParams params for Querier.
|
||||
type QuerierParams struct {
|
||||
DataSourceType *Type
|
||||
DataSourceType *Type
|
||||
EvaluationInterval time.Duration
|
||||
}
|
||||
|
||||
// Clone makes clone of VMStorage, shares http client.
|
||||
|
@ -113,6 +116,7 @@ func (s *VMStorage) Clone() *VMStorage {
|
|||
func (s *VMStorage) ApplyParams(params QuerierParams) *VMStorage {
|
||||
if params.DataSourceType != nil {
|
||||
s.dataSourceType = *params.DataSourceType
|
||||
s.evaluationInterval = params.EvaluationInterval
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
@ -136,24 +140,27 @@ func NewVMStorage(baseURL, basicAuthUser, basicAuthPass string, lookBack time.Du
|
|||
}
|
||||
}
|
||||
|
||||
// Query reads metrics from datasource by given query and type
|
||||
// Query executes the given query and returns parsed response
|
||||
func (s *VMStorage) Query(ctx context.Context, query string) ([]Metric, error) {
|
||||
switch s.dataSourceType.name {
|
||||
case "", prometheusType:
|
||||
return s.queryDataSource(ctx, query, s.setPrometheusReqParams, parsePrometheusResponse)
|
||||
case graphiteType:
|
||||
return s.queryDataSource(ctx, query, s.setGraphiteReqParams, parseGraphiteResponse)
|
||||
default:
|
||||
return nil, fmt.Errorf("engine not found: %q", s.dataSourceType.name)
|
||||
req, err := s.prepareReq(query, time.Now())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
resp, err := s.do(ctx, req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
parseFn := parsePrometheusResponse
|
||||
if s.dataSourceType.name != prometheusType {
|
||||
parseFn = parseGraphiteResponse
|
||||
}
|
||||
return parseFn(req, resp)
|
||||
}
|
||||
|
||||
func (s *VMStorage) queryDataSource(
|
||||
ctx context.Context,
|
||||
query string,
|
||||
setReqParams func(r *http.Request, query string),
|
||||
processResponse func(r *http.Request, resp *http.Response,
|
||||
) ([]Metric, error)) ([]Metric, error) {
|
||||
func (s *VMStorage) prepareReq(query string, timestamp time.Time) (*http.Request, error) {
|
||||
req, err := http.NewRequest("POST", s.datasourceURL, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
@ -162,20 +169,32 @@ func (s *VMStorage) queryDataSource(
|
|||
if s.basicAuthPass != "" {
|
||||
req.SetBasicAuth(s.basicAuthUser, s.basicAuthPass)
|
||||
}
|
||||
setReqParams(req, query)
|
||||
|
||||
switch s.dataSourceType.name {
|
||||
case "", prometheusType:
|
||||
s.setPrometheusReqParams(req, query, timestamp)
|
||||
case graphiteType:
|
||||
s.setGraphiteReqParams(req, query, timestamp)
|
||||
default:
|
||||
return nil, fmt.Errorf("engine not found: %q", s.dataSourceType.name)
|
||||
}
|
||||
return req, nil
|
||||
}
|
||||
|
||||
func (s *VMStorage) do(ctx context.Context, req *http.Request) (*http.Response, error) {
|
||||
resp, err := s.c.Do(req.WithContext(ctx))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("error getting response from %s: %w", req.URL, err)
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
body, _ := ioutil.ReadAll(resp.Body)
|
||||
return nil, fmt.Errorf("datasource returns unexpected response code %d for %s. Response body %s", resp.StatusCode, req.URL, body)
|
||||
_ = resp.Body.Close()
|
||||
return nil, fmt.Errorf("unexpected response code %d for %s. Response body %s", resp.StatusCode, req.URL, body)
|
||||
}
|
||||
return processResponse(req, resp)
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
func (s *VMStorage) setPrometheusReqParams(r *http.Request, query string) {
|
||||
func (s *VMStorage) setPrometheusReqParams(r *http.Request, query string, timestamp time.Time) {
|
||||
if s.appendTypePrefix {
|
||||
r.URL.Path += prometheusPrefix
|
||||
}
|
||||
|
@ -183,16 +202,20 @@ func (s *VMStorage) setPrometheusReqParams(r *http.Request, query string) {
|
|||
q := r.URL.Query()
|
||||
q.Set("query", query)
|
||||
if s.lookBack > 0 {
|
||||
lookBack := time.Now().Add(-s.lookBack)
|
||||
q.Set("time", fmt.Sprintf("%d", lookBack.Unix()))
|
||||
timestamp = timestamp.Add(-s.lookBack)
|
||||
}
|
||||
if s.evaluationInterval > 0 {
|
||||
// see https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1232
|
||||
timestamp = timestamp.Truncate(s.evaluationInterval)
|
||||
}
|
||||
q.Set("time", fmt.Sprintf("%d", timestamp.Unix()))
|
||||
if s.queryStep > 0 {
|
||||
q.Set("step", s.queryStep.String())
|
||||
}
|
||||
r.URL.RawQuery = q.Encode()
|
||||
}
|
||||
|
||||
func (s *VMStorage) setGraphiteReqParams(r *http.Request, query string) {
|
||||
func (s *VMStorage) setGraphiteReqParams(r *http.Request, query string, timestamp time.Time) {
|
||||
if s.appendTypePrefix {
|
||||
r.URL.Path += graphitePrefix
|
||||
}
|
||||
|
@ -202,7 +225,7 @@ func (s *VMStorage) setGraphiteReqParams(r *http.Request, query string) {
|
|||
q.Set("target", query)
|
||||
from := "-5min"
|
||||
if s.lookBack > 0 {
|
||||
lookBack := time.Now().Add(-s.lookBack)
|
||||
lookBack := timestamp.Add(-s.lookBack)
|
||||
from = strconv.FormatInt(lookBack.Unix(), 10)
|
||||
}
|
||||
q.Set("from", from)
|
||||
|
|
|
@ -2,8 +2,10 @@ package datasource
|
|||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"reflect"
|
||||
"strconv"
|
||||
"testing"
|
||||
"time"
|
||||
|
@ -69,26 +71,31 @@ func TestVMSelectQuery(t *testing.T) {
|
|||
|
||||
srv := httptest.NewServer(mux)
|
||||
defer srv.Close()
|
||||
am := NewVMStorage(srv.URL, basicAuthName, basicAuthPass, time.Minute, 0, false, srv.Client())
|
||||
if _, err := am.Query(ctx, query); err == nil {
|
||||
|
||||
s := NewVMStorage(srv.URL, basicAuthName, basicAuthPass, time.Minute, 0, false, srv.Client())
|
||||
|
||||
p := NewPrometheusType()
|
||||
pq := s.BuildWithParams(QuerierParams{DataSourceType: &p, EvaluationInterval: 15 * time.Second})
|
||||
|
||||
if _, err := pq.Query(ctx, query); err == nil {
|
||||
t.Fatalf("expected connection error got nil")
|
||||
}
|
||||
if _, err := am.Query(ctx, query); err == nil {
|
||||
if _, err := pq.Query(ctx, query); err == nil {
|
||||
t.Fatalf("expected invalid response status error got nil")
|
||||
}
|
||||
if _, err := am.Query(ctx, query); err == nil {
|
||||
if _, err := pq.Query(ctx, query); err == nil {
|
||||
t.Fatalf("expected response body error got nil")
|
||||
}
|
||||
if _, err := am.Query(ctx, query); err == nil {
|
||||
if _, err := pq.Query(ctx, query); err == nil {
|
||||
t.Fatalf("expected error status got nil")
|
||||
}
|
||||
if _, err := am.Query(ctx, query); err == nil {
|
||||
if _, err := pq.Query(ctx, query); err == nil {
|
||||
t.Fatalf("expected unknown status got nil")
|
||||
}
|
||||
if _, err := am.Query(ctx, query); err == nil {
|
||||
if _, err := pq.Query(ctx, query); err == nil {
|
||||
t.Fatalf("expected non-vector resultType error got nil")
|
||||
}
|
||||
m, err := am.Query(ctx, query)
|
||||
m, err := pq.Query(ctx, query)
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected %s", err)
|
||||
}
|
||||
|
@ -100,16 +107,14 @@ func TestVMSelectQuery(t *testing.T) {
|
|||
Timestamp: 1583786142,
|
||||
Value: 13763,
|
||||
}
|
||||
if m[0].Timestamp != expected.Timestamp &&
|
||||
m[0].Value != expected.Value &&
|
||||
m[0].Labels[0].Value != expected.Labels[0].Value &&
|
||||
m[0].Labels[0].Name != expected.Labels[0].Name {
|
||||
if !reflect.DeepEqual(m[0], expected) {
|
||||
t.Fatalf("unexpected metric %+v want %+v", m[0], expected)
|
||||
}
|
||||
|
||||
dst := NewGraphiteType()
|
||||
q := am.BuildWithParams(QuerierParams{&dst})
|
||||
m, err = q.Query(ctx, queryRender)
|
||||
g := NewGraphiteType()
|
||||
gq := s.BuildWithParams(QuerierParams{DataSourceType: &g})
|
||||
|
||||
m, err = gq.Query(ctx, queryRender)
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected %s", err)
|
||||
}
|
||||
|
@ -121,10 +126,137 @@ func TestVMSelectQuery(t *testing.T) {
|
|||
Timestamp: 1611758403,
|
||||
Value: 10,
|
||||
}
|
||||
if m[0].Timestamp != expected.Timestamp &&
|
||||
m[0].Value != expected.Value &&
|
||||
m[0].Labels[0].Value != expected.Labels[0].Value &&
|
||||
m[0].Labels[0].Name != expected.Labels[0].Name {
|
||||
if !reflect.DeepEqual(m[0], expected) {
|
||||
t.Fatalf("unexpected metric %+v want %+v", m[0], expected)
|
||||
}
|
||||
}
|
||||
|
||||
func TestPrepareReq(t *testing.T) {
|
||||
query := "up"
|
||||
timestamp := time.Date(2001, 2, 3, 4, 5, 6, 0, time.UTC)
|
||||
testCases := []struct {
|
||||
name string
|
||||
vm *VMStorage
|
||||
checkFn func(t *testing.T, r *http.Request)
|
||||
}{
|
||||
{
|
||||
"prometheus path",
|
||||
&VMStorage{
|
||||
dataSourceType: NewPrometheusType(),
|
||||
},
|
||||
func(t *testing.T, r *http.Request) {
|
||||
checkEqualString(t, queryPath, r.URL.Path)
|
||||
},
|
||||
},
|
||||
{
|
||||
"prometheus prefix",
|
||||
&VMStorage{
|
||||
dataSourceType: NewPrometheusType(),
|
||||
appendTypePrefix: true,
|
||||
},
|
||||
func(t *testing.T, r *http.Request) {
|
||||
checkEqualString(t, prometheusPrefix+queryPath, r.URL.Path)
|
||||
},
|
||||
},
|
||||
{
|
||||
"graphite path",
|
||||
&VMStorage{
|
||||
dataSourceType: NewGraphiteType(),
|
||||
},
|
||||
func(t *testing.T, r *http.Request) {
|
||||
checkEqualString(t, graphitePath, r.URL.Path)
|
||||
},
|
||||
},
|
||||
{
|
||||
"graphite prefix",
|
||||
&VMStorage{
|
||||
dataSourceType: NewGraphiteType(),
|
||||
appendTypePrefix: true,
|
||||
},
|
||||
func(t *testing.T, r *http.Request) {
|
||||
checkEqualString(t, graphitePrefix+graphitePath, r.URL.Path)
|
||||
},
|
||||
},
|
||||
{
|
||||
"default params",
|
||||
&VMStorage{},
|
||||
func(t *testing.T, r *http.Request) {
|
||||
exp := fmt.Sprintf("query=%s&time=%d", query, timestamp.Unix())
|
||||
checkEqualString(t, exp, r.URL.RawQuery)
|
||||
},
|
||||
},
|
||||
{
|
||||
"basic auth",
|
||||
&VMStorage{
|
||||
basicAuthUser: "foo",
|
||||
basicAuthPass: "bar",
|
||||
},
|
||||
func(t *testing.T, r *http.Request) {
|
||||
u, p, _ := r.BasicAuth()
|
||||
checkEqualString(t, "foo", u)
|
||||
checkEqualString(t, "bar", p)
|
||||
},
|
||||
},
|
||||
{
|
||||
"lookback",
|
||||
&VMStorage{
|
||||
lookBack: time.Minute,
|
||||
},
|
||||
func(t *testing.T, r *http.Request) {
|
||||
exp := fmt.Sprintf("query=%s&time=%d", query, timestamp.Add(-time.Minute).Unix())
|
||||
checkEqualString(t, exp, r.URL.RawQuery)
|
||||
},
|
||||
},
|
||||
{
|
||||
"evaluation interval",
|
||||
&VMStorage{
|
||||
evaluationInterval: 15 * time.Second,
|
||||
},
|
||||
func(t *testing.T, r *http.Request) {
|
||||
tt := timestamp.Truncate(15 * time.Second)
|
||||
exp := fmt.Sprintf("query=%s&time=%d", query, tt.Unix())
|
||||
checkEqualString(t, exp, r.URL.RawQuery)
|
||||
},
|
||||
},
|
||||
{
|
||||
"lookback + evaluation interval",
|
||||
&VMStorage{
|
||||
lookBack: time.Minute,
|
||||
evaluationInterval: 15 * time.Second,
|
||||
},
|
||||
func(t *testing.T, r *http.Request) {
|
||||
tt := timestamp.Add(-time.Minute)
|
||||
tt = tt.Truncate(15 * time.Second)
|
||||
exp := fmt.Sprintf("query=%s&time=%d", query, tt.Unix())
|
||||
checkEqualString(t, exp, r.URL.RawQuery)
|
||||
},
|
||||
},
|
||||
{
|
||||
"step",
|
||||
&VMStorage{
|
||||
queryStep: time.Minute,
|
||||
},
|
||||
func(t *testing.T, r *http.Request) {
|
||||
exp := fmt.Sprintf("query=%s&step=%v&time=%d", query, time.Minute, timestamp.Unix())
|
||||
checkEqualString(t, exp, r.URL.RawQuery)
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range testCases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
req, err := tc.vm.prepareReq(query, timestamp)
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error: %s", err)
|
||||
}
|
||||
tc.checkFn(t, req)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func checkEqualString(t *testing.T, exp, got string) {
|
||||
t.Helper()
|
||||
if got != exp {
|
||||
t.Errorf("expected to get %q; got %q", exp, got)
|
||||
}
|
||||
}
|
||||
|
|
|
@ -63,7 +63,10 @@ func newRecordingRule(qb datasource.QuerierBuilder, group *Group, cfg config.Rul
|
|||
Labels: cfg.Labels,
|
||||
GroupID: group.ID(),
|
||||
metrics: &recordingRuleMetrics{},
|
||||
q: qb.BuildWithParams(datasource.QuerierParams{DataSourceType: &cfg.Type}),
|
||||
q: qb.BuildWithParams(datasource.QuerierParams{
|
||||
DataSourceType: &cfg.Type,
|
||||
EvaluationInterval: group.Interval,
|
||||
}),
|
||||
}
|
||||
|
||||
labels := fmt.Sprintf(`recording=%q, group=%q, id="%d"`, rr.Name, group.Name, rr.ID())
|
||||
|
|
Loading…
Reference in a new issue