Repository: VictoriaMetrics/VictoriaMetrics
Pull request #2202 (2022-02-16 08:09:01)

Consul SD - update services on the watcher's start
See commits description

codecov[bot]: # [Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/2202?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) Report
> Merging [#2202](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/2202?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) (1ef8de6) into [master](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/commit/445edcc6ac8e20ba021a5df6d626e2521c0be76b?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) (445edcc) will **decrease** coverage by `0.03%`.
> The diff coverage is `1.63%`.

[![Impacted file tree graph](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/2202/graphs/tree.svg?width=650&height=150&src=pr&token=YND8dYVcXn&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics)](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/2202?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics)

```diff
@@            Coverage Diff             @@
##           master    #2202      +/-   ##
==========================================
- Coverage   60.52%   60.49%   -0.04%     
==========================================
  Files         240      240              
  Lines       40680    40691      +11     
==========================================
- Hits        24622    24615       -7     
- Misses      14420    14437      +17     
- Partials     1638     1639       +1     
```


| [Impacted Files](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/2202?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) | Coverage Δ | |
|---|---|---|
| [lib/promscrape/discovery/consul/watch.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/2202/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-bGliL3Byb21zY3JhcGUvZGlzY292ZXJ5L2NvbnN1bC93YXRjaC5nbw==) | `0.00% <0.00%> (ø)` | |
| [app/vmalert/notifier/config\_watcher.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/2202/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3ZtYWxlcnQvbm90aWZpZXIvY29uZmlnX3dhdGNoZXIuZ28=) | `70.98% <33.33%> (-3.57%)` | :arrow_down: |
| [app/vmselect/promql/aggr\_incremental.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/2202/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3Ztc2VsZWN0L3Byb21xbC9hZ2dyX2luY3JlbWVudGFsLmdv) | `82.83% <0.00%> (-1.21%)` | :arrow_down: |
| [lib/storage/storage.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/2202/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-bGliL3N0b3JhZ2Uvc3RvcmFnZS5nbw==) | `67.64% <0.00%> (+0.05%)` | :arrow_up: |
| [lib/mergeset/table.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/2202/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-bGliL21lcmdlc2V0L3RhYmxlLmdv) | `65.68% <0.00%> (+0.48%)` | :arrow_up: |
| [app/vmalert/remotewrite/remotewrite.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/2202/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3ZtYWxlcnQvcmVtb3Rld3JpdGUvcmVtb3Rld3JpdGUuZ28=) | `59.31% <0.00%> (+1.37%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/2202?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/2202?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics). Last update [88a2659...1ef8de6](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/2202?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/notifier/config_watcher.go b/app/vmalert/notifier/config_watcher.go
index 5cba98bd5..a49990151 100644
--- a/app/vmalert/notifier/config_watcher.go
+++ b/app/vmalert/notifier/config_watcher.go
@@ -73,35 +73,18 @@ func (cw *configWatcher) reload(path string) error {
 	}
 
 	// stop existing discovery
-	close(cw.syncCh)
-	cw.wg.Wait()
+	cw.mustStop()
 
 	// re-start cw with new config
 	cw.syncCh = make(chan struct{})
 	cw.cfg = cfg
-
-	cw.resetTargets()
 	return cw.start()
 }
 
-const (
-	addRetryBackoff = time.Millisecond * 100
-	addRetryCount   = 2
-)
-
 func (cw *configWatcher) add(typeK TargetType, interval time.Duration, labelsFn getLabels) error {
-	var targets []Target
-	var errors []error
-	var count int
-	for { // retry addRetryCount times if first discovery attempts gave no results
-		targets, errors = targetsFromLabels(labelsFn, cw.cfg, cw.genFn)
-		for _, err := range errors {
-			return fmt.Errorf("failed to init notifier for %q: %s", typeK, err)
-		}
-		if len(targets) > 0 || count >= addRetryCount {
-			break
-		}
-		time.Sleep(addRetryBackoff)
+	targets, errors := targetsFromLabels(labelsFn, cw.cfg, cw.genFn)
+	for _, err := range errors {
+		return fmt.Errorf("failed to init notifier for %q: %s", typeK, err)
 	}
 
 	cw.setTargets(typeK, targets)
@@ -215,7 +198,10 @@ func (cw *configWatcher) start() error {
 	return nil
 }
 
-func (cw *configWatcher) resetTargets() {
+func (cw *configWatcher) mustStop() {
+	close(cw.syncCh)
+	cw.wg.Wait()
+
 	cw.targetsMu.Lock()
 	for _, targets := range cw.targets {
 		for _, t := range targets {
@@ -224,6 +210,11 @@ func (cw *configWatcher) resetTargets() {
 	}
 	cw.targets = make(map[TargetType][]Target)
 	cw.targetsMu.Unlock()
+
+	for i := range cw.cfg.ConsulSDConfigs {
+		cw.cfg.ConsulSDConfigs[i].MustStop()
+	}
+	cw.cfg = nil
 }
 
 func (cw *configWatcher) setTargets(key TargetType, targets []Target) {
diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index f1cae3d22..cdbaba37e 100644
--- a/docs/CHANGELOG.md
+++ b/docs/CHANGELOG.md
@@ -25,6 +25,7 @@ The following tip changes can be tested by building VictoriaMetrics components f
 * BUGFIX: reduce memory usage during the first three hours after the upgrade from versions older than v1.73.0. The memory usage spike was related to the need of in-memory caches' re-population after the upgrade because of the fix for [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/1401). Now cache size limits are reduced in order to occupy less memory during the upgrade.
 * BUGFIX: fix a bug, which could significantly slow down requests to `/api/v1/labels` and `/api/v1/label/<label_name>/values`. These APIs are used by Grafana for auto-completion of label names and label values. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/2200).
 * BUGFIX: vmalert: add support for `$externalLabels` and `$externalURL` template vars in the same way as Prometheus does. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/2193).
+* BUGFIX: vmalert: make sure notifiers are discovered during initialization if they are configured via `consul_sd_configs`. Previously they could be discovered in 30 seconds (the default value for `-promscrape.consulSDCheckInterval` command-line flag) after the initialization. See [this pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/2202).
 * BUGFIX: update default value for `-promscrape.fileSDCheckInterval`, so it matches default duration used by Prometheus for checking for updates in `file_sd_configs`. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/2187). Thanks to @corporate-gadfly for the fix.
 
 
diff --git a/lib/promscrape/discovery/consul/watch.go b/lib/promscrape/discovery/consul/watch.go
index f5d60c4ea..8162998c9 100644
--- a/lib/promscrape/discovery/consul/watch.go
+++ b/lib/promscrape/discovery/consul/watch.go
@@ -42,7 +42,7 @@ type serviceWatcher struct {
 	stopCh       chan struct{}
 }
 
-// newConsulWatcher creates new watcher and start background service discovery for Consul.
+// newConsulWatcher creates new watcher and starts background service discovery for Consul.
 func newConsulWatcher(client *discoveryutils.Client, sdc *SDConfig, datacenter, namespace string) *consulWatcher {
 	baseQueryArgs := "?dc=" + url.QueryEscape(datacenter)
 	if sdc.AllowStale {
@@ -67,7 +67,10 @@ func newConsulWatcher(client *discoveryutils.Client, sdc *SDConfig, datacenter,
 		services:              make(map[string]*serviceWatcher),
 		stopCh:                make(chan struct{}),
 	}
-	go cw.watchForServicesUpdates()
+	initCh := make(chan struct{})
+	go cw.watchForServicesUpdates(initCh)
+	// wait for initialization to complete
+	<-initCh
 	return cw
 }
 
@@ -78,11 +81,57 @@ func (cw *consulWatcher) mustStop() {
 	// TODO: add ability to cancel blocking requests.
 }
 
+func (cw *consulWatcher) updateServices(serviceNames []string) {
+	var initWG sync.WaitGroup
+	// Start watchers for new services.
+	cw.servicesLock.Lock()
+	for _, serviceName := range serviceNames {
+		if _, ok := cw.services[serviceName]; ok {
+			// The watcher for serviceName already exists.
+			continue
+		}
+		sw := &serviceWatcher{
+			serviceName: serviceName,
+			stopCh:      make(chan struct{}),
+		}
+		cw.services[serviceName] = sw
+		cw.wg.Add(1)
+		serviceWatchersCreated.Inc()
+		initWG.Add(1)
+		go func() {
+			serviceWatchersCount.Inc()
+			sw.watchForServiceNodesUpdates(cw, &initWG)
+			serviceWatchersCount.Dec()
+			cw.wg.Done()
+		}()
+	}
+
+	// Stop watchers for removed services.
+	newServiceNamesMap := make(map[string]struct{}, len(serviceNames))
+	for _, serviceName := range serviceNames {
+		newServiceNamesMap[serviceName] = struct{}{}
+	}
+	for serviceName, sw := range cw.services {
+		if _, ok := newServiceNamesMap[serviceName]; ok {
+			continue
+		}
+		close(sw.stopCh)
+		delete(cw.services, serviceName)
+		serviceWatchersStopped.Inc()
+
+		// Do not wait for the watcher goroutine to exit, since this may take for up to maxWaitTime
+		// if it is blocked in Consul API request.
+	}
+	cw.servicesLock.Unlock()
+
+	// Wait for initialization to complete.
+	initWG.Wait()
+}
+
 // watchForServicesUpdates watches for new services and updates it in cw.
-func (cw *consulWatcher) watchForServicesUpdates() {
-	checkInterval := getCheckInterval()
-	ticker := time.NewTicker(checkInterval / 2)
-	defer ticker.Stop()
+//
+// watchForServicesUpdates closes the initCh once the initialization is complete and first discovery iteration is done.
+func (cw *consulWatcher) watchForServicesUpdates(initCh chan struct{}) {
 	index := int64(0)
 	clientAddr := cw.client.Addr()
 	f := func() {
@@ -95,51 +144,19 @@ func (cw *consulWatcher) watchForServicesUpdates() {
 			// Nothing changed.
 			return
 		}
-
-		cw.servicesLock.Lock()
-		// Start watchers for new services.
-		for _, serviceName := range serviceNames {
-			if _, ok := cw.services[serviceName]; ok {
-				// The watcher for serviceName already exists.
-				continue
-			}
-			sw := &serviceWatcher{
-				serviceName: serviceName,
-				stopCh:      make(chan struct{}),
-			}
-			cw.services[serviceName] = sw
-			cw.wg.Add(1)
-			serviceWatchersCreated.Inc()
-			go func() {
-				serviceWatchersCount.Inc()
-				sw.watchForServiceNodesUpdates(cw)
-				serviceWatchersCount.Dec()
-				cw.wg.Done()
-			}()
-		}
-		// Stop watchers for removed services.
-		newServiceNamesMap := make(map[string]struct{}, len(serviceNames))
-		for _, serviceName := range serviceNames {
-			newServiceNamesMap[serviceName] = struct{}{}
-		}
-		for serviceName, sw := range cw.services {
-			if _, ok := newServiceNamesMap[serviceName]; ok {
-				continue
-			}
-			close(sw.stopCh)
-			delete(cw.services, serviceName)
-			serviceWatchersStopped.Inc()
-
-			// Do not wait for the watcher goroutine to exit, since this may take for up to maxWaitTime
-			// if it is blocked in Consul API request.
-		}
-		cw.servicesLock.Unlock()
-
+		cw.updateServices(serviceNames)
 		index = newIndex
 	}
 
 	logger.Infof("started Consul service watcher for %q", clientAddr)
 	f()
+
+	// send signal that initialization is complete
+	close(initCh)
+
+	checkInterval := getCheckInterval()
+	ticker := time.NewTicker(checkInterval / 2)
+	defer ticker.Stop()
 	for {
 		select {
 		case <-ticker.C:
@@ -196,10 +213,9 @@ func (cw *consulWatcher) getBlockingServiceNames(index int64) ([]string, int64,
 }
 
 // watchForServiceNodesUpdates watches for Consul serviceNode changes for the given serviceName.
-func (sw *serviceWatcher) watchForServiceNodesUpdates(cw *consulWatcher) {
-	checkInterval := getCheckInterval()
-	ticker := time.NewTicker(checkInterval / 2)
-	defer ticker.Stop()
+//
+// watchForServiceNodesUpdates calls initWG.Done() once the initialization is complete and the first discovery iteration is done.
+func (sw *serviceWatcher) watchForServiceNodesUpdates(cw *consulWatcher, initWG *sync.WaitGroup) {
 	clientAddr := cw.client.Addr()
 	index := int64(0)
 	path := "/v1/health/service/" + sw.serviceName + cw.serviceNodesQueryArgs
@@ -227,6 +243,12 @@ func (sw *serviceWatcher) watchForServiceNodesUpdates(cw *consulWatcher) {
 	}
 
 	f()
+	// Notify caller that initialization is complete
+	initWG.Done()
+
+	checkInterval := getCheckInterval()
+	ticker := time.NewTicker(checkInterval / 2)
+	defer ticker.Stop()
 	for {
 		select {
 		case <-ticker.C:


Test patch:
diff --git a/app/vmalert/notifier/config_watcher_test.go b/app/vmalert/notifier/config_watcher_test.go
index 9107157b4..4120e75b2 100644
--- a/app/vmalert/notifier/config_watcher_test.go
+++ b/app/vmalert/notifier/config_watcher_test.go
@@ -9,9 +9,6 @@ import (
 	"os"
 	"sync"
 	"testing"
-	"time"
-
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/consul"
 )
 
 func TestConfigWatcherReload(t *testing.T) {
@@ -31,6 +28,7 @@ static_configs:
 	if err != nil {
 		t.Fatalf("failed to start config watcher: %s", err)
 	}
+	defer cw.mustStop()
 	ns := cw.notifiers()
 	if len(ns) != 2 {
 		t.Fatalf("expected to have 2 notifiers; got %d %#v", len(ns), ns)
@@ -78,16 +76,11 @@ consul_sd_configs:
       - alertmanager
 `, consulSDServer.URL))
 
-	prevCheckInterval := *consul.SDCheckInterval
-	defer func() { *consul.SDCheckInterval = prevCheckInterval }()
-
-	*consul.SDCheckInterval = time.Millisecond * 100
-
 	cw, err := newWatcher(consulSDFile.Name(), nil)
 	if err != nil {
 		t.Fatalf("failed to start config watcher: %s", err)
 	}
-	time.Sleep(*consul.SDCheckInterval * 2)
+	defer cw.mustStop()
 
 	if len(cw.notifiers()) != 2 {
 		t.Fatalf("expected to get 2 notifiers; got %d", len(cw.notifiers()))
@@ -161,6 +154,7 @@ consul_sd_configs:
 	if err != nil {
 		t.Fatalf("failed to start config watcher: %s", err)
 	}
+	defer cw.mustStop()
 
 	const workers = 500
 	const iterations = 10