Repository: VictoriaMetrics/VictoriaMetrics
Pull request #1257 (2021-04-30 06:39:17)

Vmalert: adjust `time` param for datasource queries according to `evaluationInterval`
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.

#1232 


Patch:
diff --git a/app/vmalert/alerting.go b/app/vmalert/alerting.go
index c454d66bc..75c80f2de 100644
--- a/app/vmalert/alerting.go
+++ b/app/vmalert/alerting.go
@@ -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())
diff --git a/app/vmalert/datasource/vm.go b/app/vmalert/datasource/vm.go
index 6e92db2c1..f7a149c53 100644
--- a/app/vmalert/datasource/vm.go
+++ b/app/vmalert/datasource/vm.go
@@ -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)
diff --git a/app/vmalert/recording.go b/app/vmalert/recording.go
index 698c19b86..fbdeb7d59 100644
--- a/app/vmalert/recording.go
+++ b/app/vmalert/recording.go
@@ -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())


Test patch:
diff --git a/app/vmalert/datasource/vm_test.go b/app/vmalert/datasource/vm_test.go
index 9ed26ef74..f9bad36be 100644
--- a/app/vmalert/datasource/vm_test.go
+++ b/app/vmalert/datasource/vm_test.go
@@ -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)
+	}
+}