Repository: VictoriaMetrics/VictoriaMetrics
Pull request #2744 (2022-06-17 15:44:30)

Vmalert notifiers
See commits description

codecov[bot]: # [Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/2744?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) Report
> Merging [#2744](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/2744?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) (8244828) into [master](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/commit/10454d17358f5153187962a610b57bfd56a2af89?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) (10454d1) will **increase** coverage by `0.01%`.
> The diff coverage is `80.00%`.

```diff
@@            Coverage Diff             @@
##           master    #2744      +/-   ##
==========================================
+ Coverage   59.55%   59.56%   +0.01%     
==========================================
  Files         257      257              
  Lines       43824    43839      +15     
==========================================
+ Hits        26098    26114      +16     
  Misses      15984    15984              
+ Partials     1742     1741       -1     
```


| [Impacted Files](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/2744?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) | Coverage Δ | |
|---|---|---|
| [app/vmalert/notifier/init.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/2744/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3ZtYWxlcnQvbm90aWZpZXIvaW5pdC5nbw==) | `10.34% <0.00%> (ø)` | |
| [app/vmalert/utils/err\_group.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/2744/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3ZtYWxlcnQvdXRpbHMvZXJyX2dyb3VwLmdv) | `79.31% <72.72%> (-4.91%)` | :arrow_down: |
| [app/vmalert/group.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/2744/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3ZtYWxlcnQvZ3JvdXAuZ28=) | `78.19% <100.00%> (+1.20%)` | :arrow_up: |
| [lib/storage/storage.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/2744/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-bGliL3N0b3JhZ2Uvc3RvcmFnZS5nbw==) | `67.81% <0.00%> (-0.17%)` | :arrow_down: |
| [app/vmselect/promql/aggr\_incremental.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/2744/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3Ztc2VsZWN0L3Byb21xbC9hZ2dyX2luY3JlbWVudGFsLmdv) | `85.24% <0.00%> (+1.20%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/2744?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/2744?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics). Last update [00831e0...8244828](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/2744?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics).


Patch:
diff --git a/app/vmalert/group.go b/app/vmalert/group.go
index 747a4ca54..ac8e712bd 100644
--- a/app/vmalert/group.go
+++ b/app/vmalert/group.go
@@ -438,11 +438,17 @@ func (e *executor) exec(ctx context.Context, rule Rule, ts time.Time, resolveDur
 		return nil
 	}
 
+	wg := sync.WaitGroup{}
 	for _, nt := range e.notifiers() {
-		if err := nt.Send(ctx, alerts); err != nil {
-			errGr.Add(fmt.Errorf("rule %q: failed to send alerts to addr %q: %w", rule, nt.Addr(), err))
-		}
+		wg.Add(1)
+		go func(nt notifier.Notifier) {
+			if err := nt.Send(ctx, alerts); err != nil {
+				errGr.Add(fmt.Errorf("rule %q: failed to send alerts to addr %q: %w", rule, nt.Addr(), err))
+			}
+			wg.Done()
+		}(nt)
 	}
+	wg.Wait()
 	return errGr.Err()
 }
 
diff --git a/app/vmalert/notifier/init.go b/app/vmalert/notifier/init.go
index fdc49af1a..a1b8bcac6 100644
--- a/app/vmalert/notifier/init.go
+++ b/app/vmalert/notifier/init.go
@@ -145,7 +145,7 @@ func notifiersFromFlags(gen AlertURLGenerator) ([]Notifier, error) {
 		}
 
 		addr = strings.TrimSuffix(addr, "/")
-		am, err := NewAlertManager(addr+alertManagerPath, gen, authCfg, nil, time.Minute)
+		am, err := NewAlertManager(addr+alertManagerPath, gen, authCfg, nil, time.Second*10)
 		if err != nil {
 			return nil, err
 		}
diff --git a/app/vmalert/utils/err_group.go b/app/vmalert/utils/err_group.go
index a2f5750e8..c8ce3a0d8 100644
--- a/app/vmalert/utils/err_group.go
+++ b/app/vmalert/utils/err_group.go
@@ -3,24 +3,34 @@ package utils
 import (
 	"fmt"
 	"strings"
+	"sync"
 )
 
 // ErrGroup accumulates multiple errors
 // and produces single error message.
 type ErrGroup struct {
+	mu   sync.Mutex
 	errs []error
 }
 
 // Add adds a new error to group.
-// Isn't thread-safe.
+// Is thread-safe.
 func (eg *ErrGroup) Add(err error) {
+	eg.mu.Lock()
 	eg.errs = append(eg.errs, err)
+	eg.mu.Unlock()
 }
 
 // Err checks if group contains at least
 // one error.
 func (eg *ErrGroup) Err() error {
-	if eg == nil || len(eg.errs) == 0 {
+	if eg == nil {
+		return nil
+	}
+
+	eg.mu.Lock()
+	defer eg.mu.Unlock()
+	if len(eg.errs) == 0 {
 		return nil
 	}
 	return eg
@@ -28,6 +38,9 @@ func (eg *ErrGroup) Err() error {
 
 // Error satisfies Error interface
 func (eg *ErrGroup) Error() string {
+	eg.mu.Lock()
+	defer eg.mu.Unlock()
+
 	if len(eg.errs) == 0 {
 		return ""
 	}


Test patch:
diff --git a/app/vmalert/group_test.go b/app/vmalert/group_test.go
index 95e50c43e..046acd2b0 100644
--- a/app/vmalert/group_test.go
+++ b/app/vmalert/group_test.go
@@ -413,3 +413,42 @@ func TestPurgeStaleSeries(t *testing.T) {
 		[]Rule{&AlertingRule{RuleID: 1}, &AlertingRule{RuleID: 2}},
 	)
 }
+
+func TestFaultyNotifier(t *testing.T) {
+	fq := &fakeQuerier{}
+	fq.add(metricWithValueAndLabels(t, 1, "__name__", "foo", "job", "bar"))
+
+	r := newTestAlertingRule("instant", 0)
+	r.q = fq
+
+	fn := &fakeNotifier{}
+	e := &executor{
+		notifiers: func() []notifier.Notifier {
+			return []notifier.Notifier{
+				&faultyNotifier{},
+				fn,
+			}
+		},
+	}
+	delay := 5 * time.Second
+	ctx, cancel := context.WithTimeout(context.Background(), delay)
+	defer cancel()
+
+	go func() {
+		_ = e.exec(ctx, r, time.Now(), 0, 10)
+	}()
+
+	tn := time.Now()
+	deadline := tn.Add(delay / 2)
+	for {
+		if fn.getCounter() > 0 {
+			return
+		}
+		if tn.After(deadline) {
+			break
+		}
+		tn = time.Now()
+		time.Sleep(time.Millisecond * 100)
+	}
+	t.Fatalf("alive notifier didn't receive notification by %v", deadline)
+}
diff --git a/app/vmalert/helpers_test.go b/app/vmalert/helpers_test.go
index 4524ba449..96bd2b1e2 100644
--- a/app/vmalert/helpers_test.go
+++ b/app/vmalert/helpers_test.go
@@ -87,6 +87,18 @@ func (fn *fakeNotifier) getAlerts() []notifier.Alert {
 	return fn.alerts
 }
 
+type faultyNotifier struct {
+	fakeNotifier
+}
+
+func (fn *faultyNotifier) Send(ctx context.Context, _ []notifier.Alert) error {
+	d, ok := ctx.Deadline()
+	if ok {
+		time.Sleep(time.Until(d))
+	}
+	return fmt.Errorf("send failed")
+}
+
 func metricWithValueAndLabels(t *testing.T, value float64, labels ...string) datasource.Metric {
 	return metricWithValuesAndLabels(t, []float64{value}, labels...)
 }
diff --git a/app/vmalert/utils/err_group_test.go b/app/vmalert/utils/err_group_test.go
index ae409bcc4..de57ee32d 100644
--- a/app/vmalert/utils/err_group_test.go
+++ b/app/vmalert/utils/err_group_test.go
@@ -2,6 +2,7 @@ package utils
 
 import (
 	"errors"
+	"fmt"
 	"testing"
 )
 
@@ -36,3 +37,29 @@ func TestErrGroup(t *testing.T) {
 		}
 	}
 }
+
+// TestErrGroupConcurrent supposed to test concurrent
+// use of error group.
+// Should be executed with -race flag
+func TestErrGroupConcurrent(t *testing.T) {
+	eg := new(ErrGroup)
+
+	const writersN = 4
+	payload := make(chan error, writersN)
+	for i := 0; i < writersN; i++ {
+		go func() {
+			for err := range payload {
+				eg.Add(err)
+			}
+		}()
+	}
+
+	const iterations = 500
+	for i := 0; i < iterations; i++ {
+		payload <- fmt.Errorf("error %d", i)
+		if i%10 == 0 {
+			_ = eg.Err()
+		}
+	}
+	close(payload)
+}