Repository: VictoriaMetrics/VictoriaMetrics
Pull request #325 (2020-02-20 20:39:36)

[vmalert] integration with AlertManager
addresses #119 


Patch:
diff --git a/Makefile b/Makefile
index 6913b99f4..350daf7b9 100644
--- a/Makefile
+++ b/Makefile
@@ -72,6 +72,7 @@ errcheck: install-errcheck
 	errcheck -exclude=errcheck_excludes.txt ./app/vmstorage/...
 	errcheck -exclude=errcheck_excludes.txt ./app/vmbackup/...
 	errcheck -exclude=errcheck_excludes.txt ./app/vmrestore/...
+	errcheck -exclude=errcheck_excludes.txt ./app/vmalert/...
 
 install-errcheck:
 	which errcheck || GO111MODULE=off go get -u github.com/kisielk/errcheck
diff --git a/README.md b/README.md
index 2cac9d126..cc615c1ed 100644
--- a/README.md
+++ b/README.md
@@ -848,6 +848,10 @@ The most interesting metrics are:
   of data loss stored in the broken parts. In the future, `vmrecover` tool will be created
   for automatic recovering from such errors.
 
+* If you see gaps on the graphs, try resetting the cache by sending request to `/internal/resetRollupResultCache`.
+  If this removes gaps on the graphs, then it is likely data with timestamps older than `-search.cacheTimestampOffset`
+  is ingested into VictoriaMetrics. Make sure that data sources have synchronized time with VictoriaMetrics.
+
 
 ### Backfilling
 
@@ -858,6 +862,9 @@ It is recommended disabling query cache with `-search.disableCache` command-line
 historical data with timestamps from the past, since the cache assumes that the data is written with
 the current timestamps. Query cache can be enabled after the backfilling is complete.
 
+An alternative solution is to query `/internal/resetRollupResultCache` url after backfilling is complete. This will reset
+the query cache, which could contain incomplete data cached during the backfilling.
+
 
 ### Profiling
 
diff --git a/app/vmalert/config/parser.go b/app/vmalert/config/parser.go
index 3f1a8081d..5b547d8f2 100644
--- a/app/vmalert/config/parser.go
+++ b/app/vmalert/config/parser.go
@@ -2,11 +2,6 @@ package config
 
 import "time"
 
-// Labels basic struct of different labels
-type Labels struct {
-	Severity string
-}
-
 // Annotations basic annotation for alert rule
 type Annotations struct {
 	Summary     string
@@ -18,8 +13,11 @@ type Alert struct {
 	Name        string
 	Expr        string
 	For         time.Duration
-	Labels      Labels
+	Labels      map[string]string
 	Annotations Annotations
+
+	Start time.Time
+	End   time.Time
 }
 
 // Group grouping array of alert
diff --git a/app/vmalert/main.go b/app/vmalert/main.go
index 9b3da62eb..b613e6f32 100644
--- a/app/vmalert/main.go
+++ b/app/vmalert/main.go
@@ -41,7 +41,9 @@ func main() {
 	}()
 	sig := procutil.WaitForSigterm()
 	logger.Infof("service received signal %s", sig)
-	httpserver.Stop(*httpListenAddr)
+	if err := httpserver.Stop(*httpListenAddr); err != nil {
+		logger.Fatalf("cannot stop the webservice: %s", err)
+	}
 	w.stop()
 }
 
diff --git a/app/vmalert/provider/alert_manager_request.qtpl b/app/vmalert/provider/alert_manager_request.qtpl
new file mode 100644
index 000000000..edcec0268
--- /dev/null
+++ b/app/vmalert/provider/alert_manager_request.qtpl
@@ -0,0 +1,26 @@
+{% import (
+    "time"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
+) %}
+{% stripspace %}
+
+{% func amRequest(alert *config.Alert, generatorURL string) %}
+{
+	"startsAt":{%q= alert.Start.Format(time.RFC3339Nano) %},
+	"generatorURL": {%q= generatorURL %},
+    {% if !alert.End.IsZero() %}
+    "endsAt":{%q= alert.End.Format(time.RFC3339Nano) %},
+    {% endif %}
+    "labels": {
+        "alertname":{%q= alert.Name %}
+        {% for k,v := range alert.Labels %}
+            ,{%q= k %}:{%q= v %}
+        {% endfor %}
+    },
+    "annotations": {
+        "summary": {%q= alert.Annotations.Summary %},
+        "description": {%q= alert.Annotations.Description %}
+    }
+}
+{% endfunc %}
+{% endstripspace %}
diff --git a/app/vmalert/provider/alert_manager_request.qtpl.go b/app/vmalert/provider/alert_manager_request.qtpl.go
new file mode 100644
index 000000000..4fea9441a
--- /dev/null
+++ b/app/vmalert/provider/alert_manager_request.qtpl.go
@@ -0,0 +1,101 @@
+// Code generated by qtc from "alert_manager_request.qtpl". DO NOT EDIT.
+// See https://github.com/valyala/quicktemplate for details.
+
+//line app/vmalert/provider/alert_manager_request.qtpl:1
+package provider
+
+//line app/vmalert/provider/alert_manager_request.qtpl:1
+import (
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
+	"time"
+)
+
+//line app/vmalert/provider/alert_manager_request.qtpl:7
+import (
+	qtio422016 "io"
+
+	qt422016 "github.com/valyala/quicktemplate"
+)
+
+//line app/vmalert/provider/alert_manager_request.qtpl:7
+var (
+	_ = qtio422016.Copy
+	_ = qt422016.AcquireByteBuffer
+)
+
+//line app/vmalert/provider/alert_manager_request.qtpl:7
+func streamamRequest(qw422016 *qt422016.Writer, alert *config.Alert, generatorURL string) {
+//line app/vmalert/provider/alert_manager_request.qtpl:7
+	qw422016.N().S(`{"startsAt":`)
+//line app/vmalert/provider/alert_manager_request.qtpl:9
+	qw422016.N().Q(alert.Start.Format(time.RFC3339Nano))
+//line app/vmalert/provider/alert_manager_request.qtpl:9
+	qw422016.N().S(`,"generatorURL":`)
+//line app/vmalert/provider/alert_manager_request.qtpl:10
+	qw422016.N().Q(generatorURL)
+//line app/vmalert/provider/alert_manager_request.qtpl:10
+	qw422016.N().S(`,`)
+//line app/vmalert/provider/alert_manager_request.qtpl:11
+	if !alert.End.IsZero() {
+//line app/vmalert/provider/alert_manager_request.qtpl:11
+		qw422016.N().S(`"endsAt":`)
+//line app/vmalert/provider/alert_manager_request.qtpl:12
+		qw422016.N().Q(alert.End.Format(time.RFC3339Nano))
+//line app/vmalert/provider/alert_manager_request.qtpl:12
+		qw422016.N().S(`,`)
+//line app/vmalert/provider/alert_manager_request.qtpl:13
+	}
+//line app/vmalert/provider/alert_manager_request.qtpl:13
+	qw422016.N().S(`"labels": {"alertname":`)
+//line app/vmalert/provider/alert_manager_request.qtpl:15
+	qw422016.N().Q(alert.Name)
+//line app/vmalert/provider/alert_manager_request.qtpl:16
+	for k, v := range alert.Labels {
+//line app/vmalert/provider/alert_manager_request.qtpl:16
+		qw422016.N().S(`,`)
+//line app/vmalert/provider/alert_manager_request.qtpl:17
+		qw422016.N().Q(k)
+//line app/vmalert/provider/alert_manager_request.qtpl:17
+		qw422016.N().S(`:`)
+//line app/vmalert/provider/alert_manager_request.qtpl:17
+		qw422016.N().Q(v)
+//line app/vmalert/provider/alert_manager_request.qtpl:18
+	}
+//line app/vmalert/provider/alert_manager_request.qtpl:18
+	qw422016.N().S(`},"annotations": {"summary":`)
+//line app/vmalert/provider/alert_manager_request.qtpl:21
+	qw422016.N().Q(alert.Annotations.Summary)
+//line app/vmalert/provider/alert_manager_request.qtpl:21
+	qw422016.N().S(`,"description":`)
+//line app/vmalert/provider/alert_manager_request.qtpl:22
+	qw422016.N().Q(alert.Annotations.Description)
+//line app/vmalert/provider/alert_manager_request.qtpl:22
+	qw422016.N().S(`}}`)
+//line app/vmalert/provider/alert_manager_request.qtpl:25
+}
+
+//line app/vmalert/provider/alert_manager_request.qtpl:25
+func writeamRequest(qq422016 qtio422016.Writer, alert *config.Alert, generatorURL string) {
+//line app/vmalert/provider/alert_manager_request.qtpl:25
+	qw422016 := qt422016.AcquireWriter(qq422016)
+//line app/vmalert/provider/alert_manager_request.qtpl:25
+	streamamRequest(qw422016, alert, generatorURL)
+//line app/vmalert/provider/alert_manager_request.qtpl:25
+	qt422016.ReleaseWriter(qw422016)
+//line app/vmalert/provider/alert_manager_request.qtpl:25
+}
+
+//line app/vmalert/provider/alert_manager_request.qtpl:25
+func amRequest(alert *config.Alert, generatorURL string) string {
+//line app/vmalert/provider/alert_manager_request.qtpl:25
+	qb422016 := qt422016.AcquireByteBuffer()
+//line app/vmalert/provider/alert_manager_request.qtpl:25
+	writeamRequest(qb422016, alert, generatorURL)
+//line app/vmalert/provider/alert_manager_request.qtpl:25
+	qs422016 := string(qb422016.B)
+//line app/vmalert/provider/alert_manager_request.qtpl:25
+	qt422016.ReleaseByteBuffer(qb422016)
+//line app/vmalert/provider/alert_manager_request.qtpl:25
+	return qs422016
+//line app/vmalert/provider/alert_manager_request.qtpl:25
+}
diff --git a/app/vmalert/provider/alertmanager.go b/app/vmalert/provider/alertmanager.go
index 47f229cff..1589c2c24 100644
--- a/app/vmalert/provider/alertmanager.go
+++ b/app/vmalert/provider/alertmanager.go
@@ -1,11 +1,66 @@
 package provider
 
-import "github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"sync"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+)
+
+const alertsPath = "/api/v2/alerts"
+
+var pool = sync.Pool{New: func() interface{} {
+	return &bytes.Buffer{}
+}}
 
 // AlertManager represents integration provider with Prometheus alert manager
-type AlertManager struct{}
+type AlertManager struct {
+	alertURL string
+	argFunc  AlertURLGenerator
+	client   *http.Client
+}
+
+// AlertURLGenerator returns URL to single alert by given name
+type AlertURLGenerator func(name string) string
+
+// NewAlertManager is a constructor for AlertManager
+func NewAlertManager(alertManagerURL string, fn AlertURLGenerator, c *http.Client) *AlertManager {
+	return &AlertManager{
+		alertURL: strings.TrimSuffix(alertManagerURL, "/") + alertsPath,
+		argFunc:  fn,
+		client:   c,
+	}
+}
+
+const (
+	jsonArrayOpen  byte = 91 // [
+	jsonArrayClose byte = 93 // ]
+)
 
 // Send an alert or resolve message
-func (a *AlertManager) Send(rule config.Alert) error {
+func (am *AlertManager) Send(alert *config.Alert) error {
+	b := pool.Get().(*bytes.Buffer)
+	b.Reset()
+	defer pool.Put(b)
+	b.WriteByte(jsonArrayOpen)
+	writeamRequest(b, alert, am.argFunc(alert.Name))
+	b.WriteByte(jsonArrayClose)
+	resp, err := am.client.Post(am.alertURL, "application/json", b)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = resp.Body.Close() }()
+	if resp.StatusCode != http.StatusOK {
+		b.Reset()
+		if _, err := io.Copy(b, resp.Body); err != nil {
+			logger.Errorf("unable to copy error response body to buffer %s", err)
+		}
+		return fmt.Errorf("invalid response from alertmanager %s", b)
+	}
 	return nil
 }
diff --git a/app/vmselect/main.go b/app/vmselect/main.go
index f0935f633..7c8b827a2 100644
--- a/app/vmselect/main.go
+++ b/app/vmselect/main.go
@@ -23,7 +23,8 @@ var (
 	deleteAuthKey         = flag.String("deleteAuthKey", "", "authKey for metrics' deletion via /api/v1/admin/tsdb/delete_series")
 	maxConcurrentRequests = flag.Int("search.maxConcurrentRequests", getDefaultMaxConcurrentRequests(), "The maximum number of concurrent search requests. "+
 		"It shouldn't be high, since a single request can saturate all the CPU cores. See also -search.maxQueueDuration")
-	maxQueueDuration = flag.Duration("search.maxQueueDuration", 10*time.Second, "The maximum time the request waits for execution when -search.maxConcurrentRequests limit is reached")
+	maxQueueDuration  = flag.Duration("search.maxQueueDuration", 10*time.Second, "The maximum time the request waits for execution when -search.maxConcurrentRequests limit is reached")
+	resetCacheAuthKey = flag.String("search.resetCacheAuthKey", "", "Optional authKey for resetting rollup cache via /internal/resetCache call")
 )
 
 func getDefaultMaxConcurrentRequests() int {
@@ -99,6 +100,15 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 	}
 
 	path := strings.Replace(r.URL.Path, "//", "/", -1)
+	if path == "/internal/resetRollupResultCache" {
+		if len(*resetCacheAuthKey) > 0 && r.FormValue("authKey") != *resetCacheAuthKey {
+			sendPrometheusError(w, r, fmt.Errorf("invalid authKey=%q for %q", r.FormValue("authKey"), path))
+			return true
+		}
+		promql.ResetRollupResultCache()
+		return true
+	}
+
 	if strings.HasPrefix(path, "/api/v1/label/") {
 		s := r.URL.Path[len("/api/v1/label/"):]
 		if strings.HasSuffix(s, "/values") {
diff --git a/app/vmselect/promql/rollup_result_cache.go b/app/vmselect/promql/rollup_result_cache.go
index 8a807a802..6769bd15d 100644
--- a/app/vmselect/promql/rollup_result_cache.go
+++ b/app/vmselect/promql/rollup_result_cache.go
@@ -18,7 +18,12 @@ import (
 	"github.com/VictoriaMetrics/metrics"
 )
 
-var disableCache = flag.Bool("search.disableCache", false, "Whether to disable response caching. This may be useful during data backfilling")
+var (
+	disableCache         = flag.Bool("search.disableCache", false, "Whether to disable response caching. This may be useful during data backfilling")
+	cacheTimestampOffset = flag.Duration("search.cacheTimestampOffset", 5*time.Minute, "The maximum duration since the current time for response data, "+
+		"which is always queried from the original raw data, without using the response cache. Increase this value if you see gaps in responses "+
+		"due to time synchronization issues between VictoriaMetrics and data sources")
+)
 
 var rollupResultCacheV = &rollupResultCache{
 	c: workingsetcache.New(1024*1024, time.Hour), // This is a cache for testing.
@@ -127,6 +132,7 @@ var rollupResultCacheResets = metrics.NewCounter(`vm_cache_resets_total{type="pr
 func ResetRollupResultCache() {
 	rollupResultCacheResets.Inc()
 	rollupResultCacheV.c.Reset()
+	logger.Infof("rollupResult cache has been cleared")
 }
 
 func (rrc *rollupResultCache) Get(ec *EvalConfig, expr metricsql.Expr, window int64) (tss []*timeseries, newStart int64) {
@@ -215,10 +221,10 @@ func (rrc *rollupResultCache) Put(ec *EvalConfig, expr metricsql.Expr, window in
 		return
 	}
 
-	// Remove values up to currentTime - step - maxSilenceInterval,
+	// Remove values up to currentTime - step - cacheTimestampOffset,
 	// since these values may be added later.
 	timestamps := tss[0].Timestamps
-	deadline := (time.Now().UnixNano() / 1e6) - ec.Step - maxSilenceInterval
+	deadline := (time.Now().UnixNano() / 1e6) - ec.Step - cacheTimestampOffset.Milliseconds()
 	i := len(timestamps) - 1
 	for i >= 0 && timestamps[i] > deadline {
 		i--
diff --git a/lib/storage/dedup.go b/lib/storage/dedup.go
index 3d4dc3c74..cc92b8c85 100644
--- a/lib/storage/dedup.go
+++ b/lib/storage/dedup.go
@@ -58,7 +58,7 @@ func DeduplicateSamples(srcTimestamps []int64, srcValues []float64) ([]int64, []
 	return dstTimestamps, dstValues
 }
 
-var dedupsDuringSelect = metrics.NewCounter(`deduplicated_samples_total{type="select"}`)
+var dedupsDuringSelect = metrics.NewCounter(`vm_deduplicated_samples_total{type="select"}`)
 
 func deduplicateSamplesDuringMerge(srcTimestamps []int64, srcValues []int64) ([]int64, []int64) {
 	if minScrapeInterval <= 0 {
@@ -94,7 +94,7 @@ func deduplicateSamplesDuringMerge(srcTimestamps []int64, srcValues []int64) ([]
 	return dstTimestamps, dstValues
 }
 
-var dedupsDuringMerge = metrics.NewCounter(`deduplicated_samples_total{type="merge"}`)
+var dedupsDuringMerge = metrics.NewCounter(`vm_deduplicated_samples_total{type="merge"}`)
 
 func needsDedup(timestamps []int64, minDelta int64) bool {
 	if len(timestamps) == 0 {


Test patch:
diff --git a/app/vmalert/provider/alertmanager_test.go b/app/vmalert/provider/alertmanager_test.go
new file mode 100644
index 000000000..cc4f32388
--- /dev/null
+++ b/app/vmalert/provider/alertmanager_test.go
@@ -0,0 +1,80 @@
+package provider
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmalert/config"
+)
+
+func TestAlertManager_Send(t *testing.T) {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/", func(_ http.ResponseWriter, _ *http.Request) {
+		t.Errorf("should not be called")
+	})
+	c := -1
+	mux.HandleFunc(alertsPath, func(w http.ResponseWriter, r *http.Request) {
+		c++
+		if r.Method != http.MethodPost {
+			t.Errorf("expected POST method got %s", r.Method)
+		}
+		switch c {
+		case 0:
+			conn, _, _ := w.(http.Hijacker).Hijack()
+			_ = conn.Close()
+		case 1:
+			w.WriteHeader(500)
+		case 2:
+			var a []struct {
+				Labels       map[string]string `json:"labels"`
+				StartsAt     time.Time         `json:"startsAt"`
+				EndAt        time.Time         `json:"endsAt"`
+				Annotations  map[string]string `json:"annotations"`
+				GeneratorURL string            `json:"generatorURL"`
+			}
+			if err := json.NewDecoder(r.Body).Decode(&a); err != nil {
+				t.Errorf("can not unmarshal data into alert %s", err)
+				t.FailNow()
+			}
+			if len(a) != 1 {
+				t.Errorf("expected 1 alert in array got %d", len(a))
+			}
+			if a[0].GeneratorURL != "alert0" {
+				t.Errorf("exptected alert0 as generatorURL got %s", a[0].GeneratorURL)
+			}
+			if a[0].Labels["alertname"] != "alert0" {
+				t.Errorf("exptected alert0 as alert name got %s", a[0].Labels["alertname"])
+			}
+			if a[0].StartsAt.IsZero() {
+				t.Errorf("exptected non-zero start time")
+			}
+			if a[0].EndAt.IsZero() {
+				t.Errorf("exptected non-zero end time")
+			}
+		}
+	})
+	srv := httptest.NewServer(mux)
+	defer srv.Close()
+	am := NewAlertManager(srv.URL, func(name string) string {
+		return name
+	}, srv.Client())
+	if err := am.Send(&config.Alert{}); err == nil {
+		t.Error("expected connection error got nil")
+	}
+	if err := am.Send(&config.Alert{}); err == nil {
+		t.Error("expected wrong http code error got nil")
+	}
+	if err := am.Send(&config.Alert{
+		Name:  "alert0",
+		Start: time.Now().UTC(),
+		End:   time.Now().UTC(),
+	}); err != nil {
+		t.Errorf("unexpected error %s", err)
+	}
+	if c != 2 {
+		t.Errorf("expected 2 calls(count from zero) to server got %d", c)
+	}
+}