Update lib/promscrape/discovery/http/http_test.go

Co-authored-by: cubic-dev-ai[bot] <191113872+cubic-dev-ai[bot]@users.noreply.github.com>
Signed-off-by: Nikolay <nik@victoriametrics.com>

docs/victoriametrics/changelog/CHANGELOG.md

@@ -26,6 +26,8 @@ See also [LTS releases](https://docs.victoriametrics.com/victoriametrics/lts-rel
 
 ## tip
 
+* FEATURE: [vmagent](https://docs.victoriametrics.com/victoriametrics/vmagent/) and [vmsingle](https://docs.victoriametrics.com/victoriametrics/single-server-victoriametrics/): Improve background discovery performance for [http_sd](https://docs.victoriametrics.com/victoriametrics/sd_configs/#http_sd_configs) discovery. See [#8838](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/8838).
+
 ## [v1.147.0](https://github.com/VictoriaMetrics/VictoriaMetrics/releases/tag/v1.147.0)
 
 Release candidate

lib/promscrape/discovery/http/api.go

@@ -1,25 +1,43 @@
 package http
 
 import (
+	"bytes"
 	"encoding/json"
 	"fmt"
 	"net/http"
 	"net/url"
 	"strconv"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/VictoriaMetrics/metrics"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutil"
-	"github.com/VictoriaMetrics/metrics"
 )
 
 var configMap = discoveryutil.NewConfigMap()
 
 type apiConfig struct {
-	client *discoveryutil.Client
-	path   string
+	client        *discoveryutil.Client
+	path          string
+	sourceURL     string
+	checkInterval time.Duration
 
 	fetchErrors *metrics.Counter
 	parseErrors *metrics.Counter
+
+	initOnce        sync.Once
+	prevAPIResponse atomic.Pointer[[]byte]
+	targetLabels    atomic.Pointer[targetLabelsResult]
+
+	wg sync.WaitGroup
+}
+
+type targetLabelsResult struct {
+	labels []*promutil.Labels
+	err    error
 }
 
 // httpGroupTarget represent prometheus GroupTarget
@@ -49,11 +67,16 @@ func newAPIConfig(sdc *SDConfig, baseDir string) (*apiConfig, error) {
 		return nil, fmt.Errorf("cannot create HTTP client for %q: %w", apiServer, err)
 	}
 	cfg := &apiConfig{
-		client:      client,
-		path:        parsedURL.RequestURI(),
-		fetchErrors: metrics.GetOrCreateCounter(fmt.Sprintf(`promscrape_discovery_http_errors_total{type="fetch",url=%q}`, sdc.URL)),
-		parseErrors: metrics.GetOrCreateCounter(fmt.Sprintf(`promscrape_discovery_http_errors_total{type="parse",url=%q}`, sdc.URL)),
+		client:        client,
+		path:          parsedURL.RequestURI(),
+		sourceURL:     sdc.URL,
+		checkInterval: max(*SDCheckInterval/2, time.Second),
+		fetchErrors:   metrics.GetOrCreateCounter(fmt.Sprintf(`promscrape_discovery_http_errors_total{type="fetch",url=%q}`, sdc.URL)),
+		parseErrors:   metrics.GetOrCreateCounter(fmt.Sprintf(`promscrape_discovery_http_errors_total{type="parse",url=%q}`, sdc.URL)),
 	}
+	cfg.wg.Go(func() {
+		cfg.run()
+	})
 	return cfg, nil
 }
 
@@ -65,21 +88,76 @@ func getAPIConfig(sdc *SDConfig, baseDir string) (*apiConfig, error) {
 	return v.(*apiConfig), nil
 }
 
-func getHTTPTargets(cfg *apiConfig) ([]httpGroupTarget, error) {
+func (cfg *apiConfig) init() {
+	cfg.initOnce.Do(func() {
+		cfg.refreshTargetsIfNeeded()
+	})
+}
+
+func (cfg *apiConfig) run() {
+	cfg.init()
+
+	ticker := time.NewTicker(cfg.checkInterval)
+	defer ticker.Stop()
+	stopCh := cfg.client.Context().Done()
+	for {
+		select {
+		case <-ticker.C:
+			cfg.refreshTargetsIfNeeded()
+		case <-stopCh:
+			return
+		}
+	}
+}
+
+func (cfg *apiConfig) refreshTargetsIfNeeded() {
+	apiResponse, err := cfg.getAPIResponseData()
+	if err != nil {
+		cfg.targetLabels.Store(&targetLabelsResult{err: err})
+		cfg.prevAPIResponse.Store(nil)
+		return
+	}
+	prevAPIResponse := cfg.prevAPIResponse.Load()
+	if prevAPIResponse != nil && bytes.Equal(apiResponse, *prevAPIResponse) {
+		return
+	}
+	hts, err := parseAPIResponse(apiResponse, cfg.path)
+	if err != nil {
+		cfg.prevAPIResponse.Store(nil)
+		cfg.parseErrors.Inc()
+		cfg.targetLabels.Store(&targetLabelsResult{err: err})
+		return
+	}
+	newTargets := addHTTPTargetLabels(hts, cfg.sourceURL)
+	cfg.targetLabels.Store(&targetLabelsResult{labels: newTargets})
+	cfg.prevAPIResponse.Store(&apiResponse)
+}
+
+func (cfg *apiConfig) getAPIResponseData() ([]byte, error) {
 	data, err := cfg.client.GetAPIResponseWithReqParams(cfg.path, func(request *http.Request) {
-		request.Header.Set("X-Prometheus-Refresh-Interval-Seconds", strconv.FormatFloat(SDCheckInterval.Seconds(), 'f', 0, 64))
+		request.Header.Set("X-Prometheus-Refresh-Interval-Seconds", strconv.FormatFloat(cfg.checkInterval.Seconds(), 'f', 0, 64))
 		request.Header.Set("Accept", "application/json")
 	})
 	if err != nil {
 		cfg.fetchErrors.Inc()
 		return nil, fmt.Errorf("cannot read http_sd api response: %w", err)
 	}
-	tg, err := parseAPIResponse(data, cfg.path)
-	if err != nil {
-		cfg.parseErrors.Inc()
-		return nil, err
+	return data, nil
+}
+
+func (cfg *apiConfig) getLabels() ([]*promutil.Labels, error) {
+	cfg.init()
+
+	tlr := cfg.targetLabels.Load()
+	if tlr.err != nil {
+		return nil, tlr.err
 	}
-	return tg, nil
+	return tlr.labels, nil
+}
+
+func (cfg *apiConfig) mustStop() {
+	cfg.client.Stop()
+	cfg.wg.Wait()
 }
 
 func parseAPIResponse(data []byte, path string) ([]httpGroupTarget, error) {

lib/promscrape/discovery/http/http.go

@@ -31,11 +31,16 @@ func (sdc *SDConfig) GetLabels(baseDir string) ([]*promutil.Labels, error) {
 	if err != nil {
 		return nil, fmt.Errorf("cannot get API config: %w", err)
 	}
-	hts, err := getHTTPTargets(cfg)
-	if err != nil {
-		return nil, err
+	return cfg.getLabels()
+}
+
+// MustStop stops further usage for sdc.
+func (sdc *SDConfig) MustStop() {
+	v := configMap.Delete(sdc)
+	if v != nil {
+		cfg := v.(*apiConfig)
+		cfg.mustStop()
 	}
-	return addHTTPTargetLabels(hts, sdc.URL), nil
 }
 
 func addHTTPTargetLabels(src []httpGroupTarget, sourceURL string) []*promutil.Labels {
@@ -54,12 +59,3 @@ func addHTTPTargetLabels(src []httpGroupTarget, sourceURL string) []*promutil.La
 	}
 	return ms
 }
-
-// MustStop stops further usage for sdc.
-func (sdc *SDConfig) MustStop() {
-	v := configMap.Delete(sdc)
-	if v != nil {
-		cfg := v.(*apiConfig)
-		cfg.client.Stop()
-	}
-}

lib/promscrape/discovery/http/http_test.go

@@ -1,8 +1,15 @@
 package http
 
 import (
+	"net/http"
+	"net/http/httptest"
+	"sort"
+	"strconv"
+	"sync/atomic"
 	"testing"
 
+	"github.com/google/go-cmp/cmp"
+
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutil"
 )
@@ -38,3 +45,150 @@ func TestAddHTTPTargetLabels(t *testing.T) {
 	}
 	f(src, labelssExpected)
 }
+
+func TestSDConfigGetLabels(t *testing.T) {
+
+	type apiResponse struct {
+		statusCode int
+		body       string
+	}
+	var currentResponse atomic.Pointer[apiResponse]
+
+	// add initial non-empty response
+	currentResponse.Store(&apiResponse{
+		body:       `[{"targets":["10.0.0.2:9100"],"labels":{"job":"node"}}]`,
+		statusCode: http.StatusOK,
+	})
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		resp := currentResponse.Load()
+		w.WriteHeader(resp.statusCode)
+		_, _ = w.Write([]byte(resp.body))
+	}))
+	defer srv.Close()
+
+	sdc := &SDConfig{
+		URL: srv.URL,
+	}
+	defer sdc.MustStop()
+
+	assertLabelss := func(expectedLabelss []*promutil.Labels) {
+		t.Helper()
+		got, err := sdc.GetLabels(".")
+		if err != nil {
+			t.Fatalf("unexpected GetLabels error: %s", err)
+		}
+		compareLabelss(t, expectedLabelss, got)
+	}
+
+	// check initial state, it must be non-empty
+	// it also inits apiConfig below
+	assertLabelss([]*promutil.Labels{
+		promutil.NewLabelsFromMap(map[string]string{
+			"__address__": "10.0.0.2:9100",
+			"job":         "node",
+			"__meta_url":  srv.URL,
+		}),
+	})
+
+	cfg, err := getAPIConfig(sdc, ".")
+	if err != nil {
+		t.Fatalf("unexpected get apiConfig error: %s", err)
+	}
+
+	updateAPIResponse := func(response apiResponse) {
+		currentResponse.Store(&response)
+		cfg.refreshTargetsIfNeeded()
+
+	}
+
+	// change response to empty
+	updateAPIResponse(apiResponse{
+		statusCode: http.StatusOK,
+		body:       `[]`,
+	})
+	assertLabelss([]*promutil.Labels{})
+
+	// change response to non-empty
+	updateAPIResponse(apiResponse{
+		statusCode: http.StatusOK,
+		body:       `[{"targets":["10.0.0.1:9100"],"labels":{"job":"node"}},{"targets":["10.0.0.5:8429"],"labels":{"job":"vmagent"}}]`,
+	})
+	assertLabelss([]*promutil.Labels{
+		promutil.NewLabelsFromMap(map[string]string{
+			"__address__": "10.0.0.1:9100",
+			"job":         "node",
+			"__meta_url":  srv.URL,
+		}),
+		promutil.NewLabelsFromMap(map[string]string{
+			"__address__": "10.0.0.5:8429",
+			"job":         "vmagent",
+			"__meta_url":  srv.URL,
+		}),
+	})
+
+	// change response to error
+	updateAPIResponse(apiResponse{
+		statusCode: http.StatusServiceUnavailable,
+		body:       `Internal Server Error`,
+	})
+	_, err = sdc.GetLabels(".")
+	if err == nil {
+		t.Fatalf("unexpected empty error")
+	}
+
+	// transit back to correct api response
+	updateAPIResponse(apiResponse{
+		statusCode: http.StatusOK,
+		body:       `[{"targets":["10.0.0.1:9100"],"labels":{"job":"node"}},{"targets":["10.0.0.5:8429"],"labels":{"job":"vmagent"}}]`,
+	})
+	assertLabelss([]*promutil.Labels{
+		promutil.NewLabelsFromMap(map[string]string{
+			"__address__": "10.0.0.1:9100",
+			"job":         "node",
+			"__meta_url":  srv.URL,
+		}),
+		promutil.NewLabelsFromMap(map[string]string{
+			"__address__": "10.0.0.5:8429",
+			"job":         "vmagent",
+			"__meta_url":  srv.URL,
+		}),
+	})
+
+	// make sure that api response is properly cached
+	before := cfg.targetLabels.Load()
+	updateAPIResponse(apiResponse{statusCode: http.StatusOK,
+		body: `[{"targets":["10.0.0.1:9100"],"labels":{"job":"node"}},{"targets":["10.0.0.5:8429"],"labels":{"job":"vmagent"}}]`})
+
+	if cfg.targetLabels.Load() != before {
+		t.Fatalf("expected identical response to be deduplicated")
+	}
+}
+
+func compareLabelss(t *testing.T, want, got []*promutil.Labels) {
+	t.Helper()
+	sortLabelss(want)
+	got = append([]*promutil.Labels(nil), got...)
+	sortLabelss(got)
+
+	if diff := cmp.Diff(want, got); len(diff) > 0 {
+		t.Fatalf("unexpected labelss (-want, +got):\n%s", diff)
+	}
+}
+
+func sortLabelss(a []*promutil.Labels) {
+	sort.Slice(a, func(i, j int) bool {
+		return marshalLabels(a[i]) < marshalLabels(a[j])
+	})
+}
+
+func marshalLabels(a *promutil.Labels) string {
+	var b []byte
+	for _, label := range a.Labels {
+		b = strconv.AppendQuote(b, label.Name)
+		b = append(b, ':')
+		b = strconv.AppendQuote(b, label.Value)
+		b = append(b, ',')
+	}
+	return string(b)
+}