vmalert: add support for datasource.lookback flag (#779)

New datasource flag `datasource.lookback` defines how far to look into
past when evaluating queries.

Address https://github.com/VictoriaMetrics/VictoriaMetrics/issues/668

app/vmalert/README.md

@@ -166,6 +166,8 @@ The shortlist of configuration flags is the following:
     	Optional basic auth password for -datasource.url
   -datasource.basicAuth.username string
     	Optional basic auth username for -datasource.url
+ -datasource.lookback duration
+        Lookback defines how far to look into past when evaluating queries. For example, if datasource.lookback=5m then param "time" with value now()-5m will be added to every query.
   -datasource.tlsCAFile string
     	Optional path to TLS CA file to use for verifying connections to -datasource.url. By default system CA is used
   -datasource.tlsCertFile string

app/vmalert/datasource/init.go

@@ -21,6 +21,9 @@ var (
 		"By default system CA is used")
 	tlsServerName = flag.String("datasource.tlsServerName", "", "Optional TLS server name to use for connections to -datasource.url. "+
 		"By default the server name from -datasource.url is used")
+
+	lookBack = flag.Duration("datasource.lookback", 0, "Lookback defines how far to look into past when evaluating queries. "+
+		"For example, if datasource.lookback=5m then param \"time\" with value now()-5m will be added to every query.")
 )
 
 // Init creates a Querier from provided flag values.
@@ -34,5 +37,5 @@ func Init() (Querier, error) {
 		return nil, fmt.Errorf("failed to create transport: %w", err)
 	}
 	c := &http.Client{Transport: tr}
-	return NewVMStorage(*addr, *basicAuthUsername, *basicAuthPassword, c), nil
+	return NewVMStorage(*addr, *basicAuthUsername, *basicAuthPassword, *lookBack, c), nil
 }

app/vmalert/datasource/vm.go

@@ -9,6 +9,7 @@ import (
 	"net/url"
 	"strconv"
 	"strings"
+	"time"
 )
 
 type response struct {
@@ -45,23 +46,25 @@ func (r response) metrics() ([]Metric, error) {
 	return ms, nil
 }
 
-const queryPath = "/api/v1/query?query="
-
 // VMStorage represents vmstorage entity with ability to read and write metrics
 type VMStorage struct {
 	c             *http.Client
 	queryURL      string
 	basicAuthUser string
 	basicAuthPass string
+	lookBack      time.Duration
 }
 
+const queryPath = "/api/v1/query?query="
+
 // NewVMStorage is a constructor for VMStorage
-func NewVMStorage(baseURL, basicAuthUser, basicAuthPass string, c *http.Client) *VMStorage {
+func NewVMStorage(baseURL, basicAuthUser, basicAuthPass string, lookBack time.Duration, c *http.Client) *VMStorage {
 	return &VMStorage{
 		c:             c,
 		basicAuthUser: basicAuthUser,
 		basicAuthPass: basicAuthPass,
 		queryURL:      strings.TrimSuffix(baseURL, "/") + queryPath,
+		lookBack:      lookBack,
 	}
 }
 
@@ -70,7 +73,12 @@ func (s *VMStorage) Query(ctx context.Context, query string) ([]Metric, error) {
 	const (
 		statusSuccess, statusError, rtVector = "success", "error", "vector"
 	)
-	req, err := http.NewRequest("POST", s.queryURL+url.QueryEscape(query), nil)
+	q := s.queryURL + url.QueryEscape(query)
+	if s.lookBack > 0 {
+		lookBack := time.Now().UTC().Add(-s.lookBack)
+		q += fmt.Sprintf("&time=%d", lookBack.Unix())
+	}
+	req, err := http.NewRequest("POST", q, nil)
 	if err != nil {
 		return nil, err
 	}

app/vmalert/datasource/vm_test.go

@@ -4,7 +4,9 @@ import (
 	"context"
 	"net/http"
 	"net/http/httptest"
+	"strconv"
 	"testing"
+	"time"
 )
 
 var (
@@ -31,6 +33,13 @@ func TestVMSelectQuery(t *testing.T) {
 		if r.URL.Query().Get("query") != query {
 			t.Errorf("expected %s in query param, got %s", query, r.URL.Query().Get("query"))
 		}
+		timeParam := r.URL.Query().Get("time")
+		if timeParam == "" {
+			t.Errorf("expected 'time' in query param, got nil instead")
+		}
+		if _, err := strconv.ParseInt(timeParam, 10, 64); err != nil {
+			t.Errorf("failed to parse 'time' query param: %s", err)
+		}
 		switch c {
 		case 0:
 			conn, _, _ := w.(http.Hijacker).Hijack()
@@ -52,7 +61,7 @@ func TestVMSelectQuery(t *testing.T) {
 
 	srv := httptest.NewServer(mux)
 	defer srv.Close()
-	am := NewVMStorage(srv.URL, basicAuthName, basicAuthPass, srv.Client())
+	am := NewVMStorage(srv.URL, basicAuthName, basicAuthPass, time.Minute, srv.Client())
 	if _, err := am.Query(ctx, query); err == nil {
 		t.Fatalf("expected connection error got nil")
 	}
@@ -89,5 +98,4 @@ func TestVMSelectQuery(t *testing.T) {
 		m[0].Labels[0].Name != expected.Labels[0].Name {
 		t.Fatalf("unexpected metric %+v want %+v", m[0], expected)
 	}
-
 }

app/vmalert/remoteread/init.go

@@ -35,5 +35,5 @@ func Init() (datasource.Querier, error) {
 		return nil, fmt.Errorf("failed to create transport: %w", err)
 	}
 	c := &http.Client{Transport: tr}
-	return datasource.NewVMStorage(*addr, *basicAuthUsername, *basicAuthPassword, c), nil
+	return datasource.NewVMStorage(*addr, *basicAuthUsername, *basicAuthPassword, 0, c), nil
 }