Repository: VictoriaMetrics/VictoriaMetrics
Pull request #498 (2020-05-17 13:03:28)

vmalert: avoid sending resolves for pending alerts
Before the change we were sending notifications to notifier if:
* alert is in Fire state
* alert is in Inactive state

We were sending Inactive notifications with End set to
last execution time to resolve alert ASAP. Unfortunately,
we were sending resolves for Pending alerts that become
Inactive, which is wrong.
In this change we delete alert from the active list if
it was Pending and become Inactive. In this way we now
have Inactive alerts only if they were in state Fire before.
See test change for example.

#483 

codecov[bot]: # [Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/498?src=pr&el=h1) Report
> Merging [#498](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/498?src=pr&el=desc) into [master](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/commit/d61bac9fd9ff7c30673fe006703d130fbce60614&el=desc) will **increase** coverage by `0.15%`.
> The diff coverage is `66.66%`.

[![Impacted file tree graph](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/498/graphs/tree.svg?width=650&height=150&src=pr&token=YND8dYVcXn)](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/498?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master     #498      +/-   ##
==========================================
+ Coverage   67.83%   67.99%   +0.15%     
==========================================
  Files         149      149              
  Lines       20544    20554      +10     
==========================================
+ Hits        13936    13975      +39     
+ Misses       5498     5462      -36     
- Partials     1110     1117       +7     
```


| [Impacted Files](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/498?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [app/vmalert/group.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/498/diff?src=pr&el=tree#diff-YXBwL3ZtYWxlcnQvZ3JvdXAuZ28=) | `76.59% <62.50%> (+2.78%)` | :arrow_up: |
| [app/vmalert/rule.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/498/diff?src=pr&el=tree#diff-YXBwL3ZtYWxlcnQvcnVsZS5nbw==) | `87.35% <100.00%> (ø)` | |
| [lib/fasttime/fasttime.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/498/diff?src=pr&el=tree#diff-bGliL2Zhc3R0aW1lL2Zhc3R0aW1lLmdv) | `53.84% <0.00%> (-23.08%)` | :arrow_down: |
| [lib/storage/merge.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/498/diff?src=pr&el=tree#diff-bGliL3N0b3JhZ2UvbWVyZ2UuZ28=) | `82.64% <0.00%> (-1.66%)` | :arrow_down: |
| [lib/encoding/int.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/498/diff?src=pr&el=tree#diff-bGliL2VuY29kaW5nL2ludC5nbw==) | `79.48% <0.00%> (-1.29%)` | :arrow_down: |
| [lib/storage/partition.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/498/diff?src=pr&el=tree#diff-bGliL3N0b3JhZ2UvcGFydGl0aW9uLmdv) | `72.11% <0.00%> (ø)` | |
| [app/vmselect/promql/aggr\_incremental.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/498/diff?src=pr&el=tree#diff-YXBwL3Ztc2VsZWN0L3Byb21xbC9hZ2dyX2luY3JlbWVudGFsLmdv) | `92.09% <0.00%> (ø)` | |
| [lib/storage/index\_db.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/498/diff?src=pr&el=tree#diff-bGliL3N0b3JhZ2UvaW5kZXhfZGIuZ28=) | `75.40% <0.00%> (+0.11%)` | :arrow_up: |
| [lib/mergeset/table.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/498/diff?src=pr&el=tree#diff-bGliL21lcmdlc2V0L3RhYmxlLmdv) | `68.01% <0.00%> (+0.27%)` | :arrow_up: |
| [lib/mergeset/block\_stream\_reader.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/498/diff?src=pr&el=tree#diff-bGliL21lcmdlc2V0L2Jsb2NrX3N0cmVhbV9yZWFkZXIuZ28=) | `62.58% <0.00%> (+11.61%)` | :arrow_up: |
| ... and [1 more](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/498/diff?src=pr&el=tree-more) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/498?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/498?src=pr&el=footer). Last update [7a8ef51...a576f7a](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/498?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

tenmozes: looks good, thank you

Patch:
diff --git a/app/vmalert/Makefile b/app/vmalert/Makefile
index d49984ddf..0d9439398 100644
--- a/app/vmalert/Makefile
+++ b/app/vmalert/Makefile
@@ -52,7 +52,9 @@ publish-vmalert:
 	APP_NAME=vmalert $(MAKE) publish-via-docker
 
 test-vmalert:
-	go test -race -cover ./app/vmalert
+	go test -v -race -cover ./app/vmalert -loggerLevel=ERROR
+	go test -v -race -cover ./app/vmalert/datasource
+	go test -v -race -cover ./app/vmalert/notifier
 
 run-vmalert: vmalert
 	./bin/vmalert -rule=app/vmalert/testdata/rules0-good.rules \
diff --git a/app/vmalert/config.go b/app/vmalert/config.go
index e46c0c497..b84e7044d 100644
--- a/app/vmalert/config.go
+++ b/app/vmalert/config.go
@@ -32,8 +32,9 @@ func Parse(pathPatterns []string, validateAnnotations bool) ([]Group, error) {
 				return nil, fmt.Errorf("one file can not contain groups with the same name %s, filepath:%s", g.Name, file)
 			}
 			g.File = file
-			g.done = make(chan struct{})
-			g.finished = make(chan struct{})
+			g.doneCh = make(chan struct{})
+			g.finishedCh = make(chan struct{})
+			g.updateCh = make(chan Group)
 
 			groupsNames[g.Name] = struct{}{}
 			for _, rule := range g.Rules {
diff --git a/app/vmalert/group.go b/app/vmalert/group.go
index d4713496f..4d5e5a318 100644
--- a/app/vmalert/group.go
+++ b/app/vmalert/group.go
@@ -19,13 +19,16 @@ type Group struct {
 	File  string
 	Rules []*Rule
 
-	done     chan struct{}
-	finished chan struct{}
+	doneCh     chan struct{}
+	finishedCh chan struct{}
+	// channel accepts new Group obj
+	// which supposed to update current group
+	updateCh chan Group
 }
 
 // ID return unique group ID that consists of
 // rules file and group name
-func (g Group) ID() uint64 {
+func (g *Group) ID() uint64 {
 	hash := fnv.New64a()
 	hash.Write([]byte(g.File))
 	hash.Write([]byte("\xff"))
@@ -47,8 +50,8 @@ func (g *Group) Restore(ctx context.Context, q datasource.Querier, lookback time
 }
 
 // updateWith updates existing group with
-// passed group object. Must be called
-// under mutex lock.
+// passed group object.
+// Not thread-safe.
 func (g *Group) updateWith(newGroup Group) {
 	rulesRegistry := make(map[string]*Rule)
 	for _, nr := range newGroup.Rules {
@@ -66,7 +69,7 @@ func (g *Group) updateWith(newGroup Group) {
 
 		// copy all significant fields.
 		// alerts state isn't copied since
-		// it should be updated in next 2 Evals
+		// it should be updated in next 2 Execs
 		or.For = nr.For
 		or.Expr = nr.Expr
 		or.Labels = nr.Labels
@@ -106,11 +109,11 @@ var (
 )
 
 func (g *Group) close() {
-	if g.done == nil {
+	if g.doneCh == nil {
 		return
 	}
-	close(g.done)
-	<-g.finished
+	close(g.doneCh)
+	<-g.finishedCh
 }
 
 func (g *Group) start(ctx context.Context, interval time.Duration,
@@ -122,12 +125,14 @@ func (g *Group) start(ctx context.Context, interval time.Duration,
 		select {
 		case <-ctx.Done():
 			logger.Infof("group %q: context cancelled", g.Name)
-			close(g.finished)
+			close(g.finishedCh)
 			return
-		case <-g.done:
+		case <-g.doneCh:
 			logger.Infof("group %q: received stop signal", g.Name)
-			close(g.finished)
+			close(g.finishedCh)
 			return
+		case ng := <-g.updateCh:
+			g.updateWith(ng)
 		case <-t.C:
 			iterationTotal.Inc()
 			iterationStart := time.Now()
@@ -146,30 +151,47 @@ func (g *Group) start(ctx context.Context, interval time.Duration,
 
 				var alertsToSend []notifier.Alert
 				for _, a := range rule.alerts {
-					if a.State != notifier.StatePending {
+					switch a.State {
+					case notifier.StateFiring:
+						// set End to execStart + 3 intervals
+						// so notifier can resolve it automatically if `vmalert`
+						// won't be able to send resolve for some reason
+						a.End = execStart.Add(3 * interval)
+						alertsToSend = append(alertsToSend, *a)
+						pushToRW(rw, rule, a, execStart)
+					case notifier.StatePending:
+						pushToRW(rw, rule, a, execStart)
+					case notifier.StateInactive:
+						// set End to execStart to notify
+						// that it was just resolved
+						a.End = execStart
 						alertsToSend = append(alertsToSend, *a)
-					}
-					if a.State == notifier.StateInactive || rw == nil {
-						continue
-					}
-					tss := rule.AlertToTimeSeries(a, execStart)
-					for _, ts := range tss {
-						remoteWriteSent.Inc()
-						if err := rw.Push(ts); err != nil {
-							remoteWriteErrors.Inc()
-							logger.Errorf("failed to push timeseries to remotewrite: %s", err)
-						}
 					}
 				}
-				if len(alertsToSend) > 0 {
-					alertsSent.Add(len(alertsToSend))
-					if err := nr.Send(ctx, alertsToSend); err != nil {
-						alertsSendErrors.Inc()
-						logger.Errorf("failed to send alert for rule %q.%q: %s", g.Name, rule.Name, err)
-					}
+				if len(alertsToSend) == 0 {
+					continue
+				}
+				alertsSent.Add(len(alertsToSend))
+				if err := nr.Send(ctx, alertsToSend); err != nil {
+					alertsSendErrors.Inc()
+					logger.Errorf("failed to send alert for rule %q.%q: %s", g.Name, rule.Name, err)
 				}
 			}
 			iterationDuration.UpdateDuration(iterationStart)
 		}
 	}
 }
+
+func pushToRW(rw *remotewrite.Client, rule *Rule, a *notifier.Alert, timestamp time.Time) {
+	if rw == nil {
+		return
+	}
+	tss := rule.AlertToTimeSeries(a, timestamp)
+	remoteWriteSent.Add(len(tss))
+	for _, ts := range tss {
+		if err := rw.Push(ts); err != nil {
+			remoteWriteErrors.Inc()
+			logger.Errorf("failed to push timeseries to remotewrite: %s", err)
+		}
+	}
+}
diff --git a/app/vmalert/manager.go b/app/vmalert/manager.go
index 38504fd52..4429ca8d3 100644
--- a/app/vmalert/manager.go
+++ b/app/vmalert/manager.go
@@ -87,8 +87,7 @@ func (m *manager) update(ctx context.Context, path []string, validate, restore b
 
 	m.groupsMu.Lock()
 	for _, og := range m.groups {
-		id := og.ID()
-		ng, ok := groupsRegistry[id]
+		ng, ok := groupsRegistry[og.ID()]
 		if !ok {
 			// old group is not present in new list
 			// and must be stopped and deleted
@@ -97,7 +96,7 @@ func (m *manager) update(ctx context.Context, path []string, validate, restore b
 			og = nil
 			continue
 		}
-		og.updateWith(ng)
+		og.updateCh <- ng
 		delete(groupsRegistry, ng.ID())
 	}
 
diff --git a/app/vmalert/rule.go b/app/vmalert/rule.go
index 08d1d64d1..bdb0d9134 100644
--- a/app/vmalert/rule.go
+++ b/app/vmalert/rule.go
@@ -71,7 +71,7 @@ func (r *Rule) Exec(ctx context.Context, q datasource.Querier) error {
 	}
 
 	for h, a := range r.alerts {
-		// cleanup inactive alerts from previous Eval
+		// cleanup inactive alerts from previous Exec
 		if a.State == notifier.StateInactive {
 			delete(r.alerts, h)
 		}
@@ -109,19 +109,19 @@ func (r *Rule) Exec(ctx context.Context, q datasource.Querier) error {
 		// if alert wasn't updated in this iteration
 		// means it is resolved already
 		if _, ok := updated[h]; !ok {
+			if a.State == notifier.StatePending {
+				// alert was in Pending state - it is not
+				// active anymore
+				delete(r.alerts, h)
+				continue
+			}
 			a.State = notifier.StateInactive
-			// set endTime to last execution time
-			// so it can be sent by notifier on next step
-			a.End = r.lastExecTime
 			continue
 		}
 		if a.State == notifier.StatePending && time.Since(a.Start) >= r.For {
 			a.State = notifier.StateFiring
 			alertsFired.Inc()
 		}
-		if a.State == notifier.StateFiring {
-			a.End = r.lastExecTime.Add(3 * *evaluationInterval)
-		}
 	}
 	return nil
 }
@@ -293,7 +293,7 @@ func newTimeSeries(value float64, labels map[string]string, timestamp time.Time)
 
 // Restore restores the state of active alerts basing on previously written timeseries.
 // Restore restores only Start field. Field State will be always Pending and supposed
-// to be updated on next Eval, as well as Value field.
+// to be updated on next Exec, as well as Value field.
 func (r *Rule) Restore(ctx context.Context, q datasource.Querier, lookback time.Duration) error {
 	if q == nil {
 		return fmt.Errorf("querier is nil")
@@ -313,7 +313,7 @@ func (r *Rule) Restore(ctx context.Context, q datasource.Querier, lookback time.
 		labels := m.Labels
 		m.Labels = make([]datasource.Label, 0)
 		// drop all extra labels, so hash key will
-		// be identical to timeseries received in Eval
+		// be identical to timeseries received in Exec
 		for _, l := range labels {
 			if l.Name == alertNameLabel {
 				continue


Test patch:
diff --git a/app/vmalert/manager_test.go b/app/vmalert/manager_test.go
index afe30d5b7..1e96e6b49 100644
--- a/app/vmalert/manager_test.go
+++ b/app/vmalert/manager_test.go
@@ -26,24 +26,35 @@ func TestManagerUpdateError(t *testing.T) {
 // execution of configuration update.
 // Should be executed with -race flag
 func TestManagerUpdateConcurrent(t *testing.T) {
-	m := &manager{groups: make(map[uint64]*Group)}
+	m := &manager{
+		groups:   make(map[uint64]*Group),
+		storage:  &fakeQuerier{},
+		notifier: &fakeNotifier{},
+	}
 	paths := []string{
 		"testdata/dir/rules0-good.rules",
 		"testdata/dir/rules1-good.rules",
 		"testdata/rules0-good.rules",
 	}
+	*evaluationInterval = time.Millisecond
+	if err := m.start(context.Background(), []string{paths[0]}, true); err != nil {
+		t.Fatalf("failed to start: %s", err)
+	}
 
-	const n = 500
+	const workers = 500
+	const iterations = 10
 	wg := sync.WaitGroup{}
-	wg.Add(n)
-	for i := 0; i < n; i++ {
+	wg.Add(workers)
+	for i := 0; i < workers; i++ {
 		go func() {
 			defer wg.Done()
-			rnd := rand.Intn(len(paths))
-			path := []string{paths[rnd]}
-			err := m.update(context.Background(), path, true, false)
-			if err != nil {
-				t.Errorf("update error: %s", err)
+			for i := 0; i < iterations; i++ {
+				rnd := rand.Intn(len(paths))
+				path := []string{paths[rnd]}
+				err := m.update(context.Background(), path, true, false)
+				if err != nil {
+					t.Errorf("update error: %s", err)
+				}
 			}
 		}()
 	}
@@ -111,7 +122,7 @@ func TestManagerUpdate(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			ctx, cancel := context.WithCancel(context.TODO())
-			m := &manager{groups: make(map[uint64]*Group)}
+			m := &manager{groups: make(map[uint64]*Group), storage: &fakeQuerier{}}
 			path := []string{tc.initPath}
 			if err := m.update(ctx, path, true, false); err != nil {
 				t.Fatalf("failed to complete initial rules update: %s", err)
diff --git a/app/vmalert/rule_test.go b/app/vmalert/rule_test.go
index ae11105ba..73227b476 100644
--- a/app/vmalert/rule_test.go
+++ b/app/vmalert/rule_test.go
@@ -296,16 +296,14 @@ func TestRule_Exec(t *testing.T) {
 			},
 		},
 		{
-			newTestRule("for-pending=>inactive", time.Millisecond),
+			newTestRule("for-pending=>empty", time.Second),
 			[][]datasource.Metric{
 				{metricWithLabels(t, "name", "foo")},
 				{metricWithLabels(t, "name", "foo")},
-				// empty step to reset pending alerts
+				// empty step to reset and delete pending alerts
 				{},
 			},
-			map[uint64]*notifier.Alert{
-				hash(metricWithLabels(t, "name", "foo")): {State: notifier.StateInactive},
-			},
+			map[uint64]*notifier.Alert{},
 		},
 		{
 			newTestRule("for-pending=>firing=>inactive", time.Millisecond),