Repository: VictoriaMetrics/VictoriaMetrics
Pull request #3853 (2023-02-22 08:53:58)

lib/promscrape: fix cancelling in-flight scrape requests during configuration reload
Implements cancelling in-flight requests for `fasthttp.HostClient`.

Related #3747

codecov[bot]: # [Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3853?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) Report
Base: **58.04**% // Head: **58.02**% // Decreases project coverage by **`-0.03%`** :warning:
> Coverage data is based on head [(`ae3af74`)](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3853?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) compared to base [(`10cf6c9`)](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/commit/10cf6c9781d18cd48e8ef991b3252c575db498a9?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics).
> Patch coverage: 0.00% of modified lines in pull request are covered.

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master    #3853      +/-   ##
==========================================
- Coverage   58.04%   58.02%   -0.03%     
==========================================
  Files         309      309              
  Lines       52248    52255       +7     
==========================================
- Hits        30329    30320       -9     
- Misses      20020    20031      +11     
- Partials     1899     1904       +5     
```


| [Impacted Files](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3853?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) | Coverage Δ | |
|---|---|---|
| [lib/promscrape/client.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3853?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-bGliL3Byb21zY3JhcGUvY2xpZW50Lmdv) | `5.82% <0.00%> (-0.15%)` | :arrow_down: |
| [lib/storage/partition.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3853?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-bGliL3N0b3JhZ2UvcGFydGl0aW9uLmdv) | `60.95% <0.00%> (-1.18%)` | :arrow_down: |
| [lib/storage/block\_stream\_reader.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3853?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-bGliL3N0b3JhZ2UvYmxvY2tfc3RyZWFtX3JlYWRlci5nbw==) | `56.80% <0.00%> (-1.17%)` | :arrow_down: |
| [lib/mergeset/table.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3853?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-bGliL21lcmdlc2V0L3RhYmxlLmdv) | `63.06% <0.00%> (-0.32%)` | :arrow_down: |
| [lib/storage/storage.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3853?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-bGliL3N0b3JhZ2Uvc3RvcmFnZS5nbw==) | `71.15% <0.00%> (-0.12%)` | :arrow_down: |
| [app/vmselect/promql/transform.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3853?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3Ztc2VsZWN0L3Byb21xbC90cmFuc2Zvcm0uZ28=) | `88.39% <0.00%> (+0.14%)` | :arrow_up: |
| [lib/bloomfilter/filter.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3853?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-bGliL2Jsb29tZmlsdGVyL2ZpbHRlci5nbw==) | `100.00% <0.00%> (+2.00%)` | :arrow_up: |
| [app/vmalert/remotewrite/remotewrite.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3853?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3ZtYWxlcnQvcmVtb3Rld3JpdGUvcmVtb3Rld3JpdGUuZ28=) | `65.75% <0.00%> (+2.05%)` | :arrow_up: |
| [app/vmselect/promql/aggr\_incremental.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3853?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3Ztc2VsZWN0L3Byb21xbC9hZ2dyX2luY3JlbWVudGFsLmdv) | `87.72% <0.00%> (+2.39%)` | :arrow_up: |
| [lib/fasttime/fasttime.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3853?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-bGliL2Zhc3R0aW1lL2Zhc3R0aW1lLmdv) | `71.42% <0.00%> (+21.42%)` | :arrow_up: |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics)

</details>

[:umbrella: View full report at Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3853?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics).   
:loudspeaker: Do you have feedback about the report comment? [Let us know in this issue](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics).


Patch:
diff --git a/README.md b/README.md
index 80f092913..827583273 100644
--- a/README.md
+++ b/README.md
@@ -105,6 +105,7 @@ Case studies:
 * [Brandwatch](https://docs.victoriametrics.com/CaseStudies.html#brandwatch)
 * [CERN](https://docs.victoriametrics.com/CaseStudies.html#cern)
 * [COLOPL](https://docs.victoriametrics.com/CaseStudies.html#colopl)
+* [Dig Security](https://docs.victoriametrics.com/CaseStudies.html#dig-security)
 * [Fly.io](https://docs.victoriametrics.com/CaseStudies.html#flyio)
 * [German Research Center for Artificial Intelligence](https://docs.victoriametrics.com/CaseStudies.html#german-research-center-for-artificial-intelligence)
 * [Grammarly](https://docs.victoriametrics.com/CaseStudies.html#grammarly)
@@ -2263,11 +2264,11 @@ Pass `-help` to VictoriaMetrics in order to see the list of supported command-li
   -metricsAuthKey string
      Auth key for /metrics endpoint. It must be passed via authKey query arg. It overrides httpAuth.* settings
   -opentsdbHTTPListenAddr string
-     TCP address to listen for OpentTSDB HTTP put requests. Usually :4242 must be set. Doesn't work if empty. See also -opentsdbHTTPListenAddr.useProxyProtocol
+     TCP address to listen for OpenTSDB HTTP put requests. Usually :4242 must be set. Doesn't work if empty. See also -opentsdbHTTPListenAddr.useProxyProtocol
   -opentsdbHTTPListenAddr.useProxyProtocol
      Whether to use proxy protocol for connections accepted at -opentsdbHTTPListenAddr . See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt
   -opentsdbListenAddr string
-     TCP and UDP address to listen for OpentTSDB metrics. Telnet put messages and HTTP /api/put messages are simultaneously served on TCP port. Usually :4242 must be set. Doesn't work if empty. See also -opentsdbListenAddr.useProxyProtocol
+     TCP and UDP address to listen for OpenTSDB metrics. Telnet put messages and HTTP /api/put messages are simultaneously served on TCP port. Usually :4242 must be set. Doesn't work if empty. See also -opentsdbListenAddr.useProxyProtocol
   -opentsdbListenAddr.useProxyProtocol
      Whether to use proxy protocol for connections accepted at -opentsdbListenAddr . See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt
   -opentsdbTrimTimestamp duration
@@ -2337,6 +2338,10 @@ Pass `-help` to VictoriaMetrics in order to see the list of supported command-li
      How frequently to reload the full state from Kubernetes API server (default 30m0s)
   -promscrape.kubernetesSDCheckInterval duration
      Interval for checking for changes in Kubernetes API server. This works only if kubernetes_sd_configs is configured in '-promscrape.config' file. See https://docs.victoriametrics.com/sd_configs.html#kubernetes_sd_configs for details (default 30s)
+  -promscrape.kumaSDCheckInterval duration
+     Interval for checking for changes in Kuma Service Mesh API service discovery. This works only if kuma_sd_configs is configured in '-promscrape.config' file. See https://docs.victoriametrics.com/sd_configs.html#kuma_sd_configs for details (default 1m0s)
+  -promscrape.kuma.waitTime duration
+     Wait time used by Kuma service discovery. Default value is used if not set
   -promscrape.maxDroppedTargets int
      The maximum number of droppedTargets to show at /api/v1/targets page. Increase this value if your setup drops more scrape targets during relabeling and you need investigating labels for all the dropped targets. Note that the increased number of tracked dropped targets may result in increased memory usage (default 1000)
   -promscrape.maxResponseHeadersSize size
diff --git a/app/vmagent/README.md b/app/vmagent/README.md
index 2ab14f218..486055ecf 100644
--- a/app/vmagent/README.md
+++ b/app/vmagent/README.md
@@ -1299,11 +1299,11 @@ See the docs at https://docs.victoriametrics.com/vmagent.html .
   -metricsAuthKey string
      Auth key for /metrics endpoint. It must be passed via authKey query arg. It overrides httpAuth.* settings
   -opentsdbHTTPListenAddr string
-     TCP address to listen for OpentTSDB HTTP put requests. Usually :4242 must be set. Doesn't work if empty. See also -opentsdbHTTPListenAddr.useProxyProtocol
+     TCP address to listen for OpenTSDB HTTP put requests. Usually :4242 must be set. Doesn't work if empty. See also -opentsdbHTTPListenAddr.useProxyProtocol
   -opentsdbHTTPListenAddr.useProxyProtocol
      Whether to use proxy protocol for connections accepted at -opentsdbHTTPListenAddr . See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt
   -opentsdbListenAddr string
-     TCP and UDP address to listen for OpentTSDB metrics. Telnet put messages and HTTP /api/put messages are simultaneously served on TCP port. Usually :4242 must be set. Doesn't work if empty. See also -opentsdbListenAddr.useProxyProtocol
+     TCP and UDP address to listen for OpenTSDB metrics. Telnet put messages and HTTP /api/put messages are simultaneously served on TCP port. Usually :4242 must be set. Doesn't work if empty. See also -opentsdbListenAddr.useProxyProtocol
   -opentsdbListenAddr.useProxyProtocol
      Whether to use proxy protocol for connections accepted at -opentsdbListenAddr . See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt
   -opentsdbTrimTimestamp duration
@@ -1371,6 +1371,10 @@ See the docs at https://docs.victoriametrics.com/vmagent.html .
      How frequently to reload the full state from Kubernetes API server (default 30m0s)
   -promscrape.kubernetesSDCheckInterval duration
      Interval for checking for changes in Kubernetes API server. This works only if kubernetes_sd_configs is configured in '-promscrape.config' file. See https://docs.victoriametrics.com/sd_configs.html#kubernetes_sd_configs for details (default 30s)
+  -promscrape.kumaSDCheckInterval duration
+     Interval for checking for changes in Kuma Service Mesh API service discovery. This works only if kuma_sd_configs is configured in '-promscrape.config' file. See https://docs.victoriametrics.com/sd_configs.html#kuma_sd_configs for details (default 1m0s)
+  -promscrape.kuma.waitTime duration
+     Wait time used by Kuma service discovery. Default value is used if not set
   -promscrape.maxDroppedTargets int
      The maximum number of droppedTargets to show at /api/v1/targets page. Increase this value if your setup drops more scrape targets during relabeling and you need investigating labels for all the dropped targets. Note that the increased number of tracked dropped targets may result in increased memory usage (default 1000)
   -promscrape.maxResponseHeadersSize size
diff --git a/app/vmagent/main.go b/app/vmagent/main.go
index 7b01e734c..9b78b13f3 100644
--- a/app/vmagent/main.go
+++ b/app/vmagent/main.go
@@ -56,12 +56,12 @@ var (
 		"See also -graphiteListenAddr.useProxyProtocol")
 	graphiteUseProxyProtocol = flag.Bool("graphiteListenAddr.useProxyProtocol", false, "Whether to use proxy protocol for connections accepted at -graphiteListenAddr . "+
 		"See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt")
-	opentsdbListenAddr = flag.String("opentsdbListenAddr", "", "TCP and UDP address to listen for OpentTSDB metrics. "+
+	opentsdbListenAddr = flag.String("opentsdbListenAddr", "", "TCP and UDP address to listen for OpenTSDB metrics. "+
 		"Telnet put messages and HTTP /api/put messages are simultaneously served on TCP port. "+
 		"Usually :4242 must be set. Doesn't work if empty. See also -opentsdbListenAddr.useProxyProtocol")
 	opentsdbUseProxyProtocol = flag.Bool("opentsdbListenAddr.useProxyProtocol", false, "Whether to use proxy protocol for connections accepted at -opentsdbListenAddr . "+
 		"See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt")
-	opentsdbHTTPListenAddr = flag.String("opentsdbHTTPListenAddr", "", "TCP address to listen for OpentTSDB HTTP put requests. Usually :4242 must be set. Doesn't work if empty. "+
+	opentsdbHTTPListenAddr = flag.String("opentsdbHTTPListenAddr", "", "TCP address to listen for OpenTSDB HTTP put requests. Usually :4242 must be set. Doesn't work if empty. "+
 		"See also -opentsdbHTTPListenAddr.useProxyProtocol")
 	opentsdbHTTPUseProxyProtocol = flag.Bool("opentsdbHTTPListenAddr.useProxyProtocol", false, "Whether to use proxy protocol for connections accepted "+
 		"at -opentsdbHTTPListenAddr . See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt")
diff --git a/app/vmctl/backoff/backoff.go b/app/vmctl/backoff/backoff.go
new file mode 100644
index 000000000..31cbf0017
--- /dev/null
+++ b/app/vmctl/backoff/backoff.go
@@ -0,0 +1,60 @@
+package backoff
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"math"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+)
+
+const (
+	backoffRetries     = 5
+	backoffFactor      = 1.7
+	backoffMinDuration = time.Second
+)
+
+// retryableFunc describes call back which will repeat on errors
+type retryableFunc func() error
+
+var ErrBadRequest = errors.New("bad request")
+
+// Backoff describes object with backoff policy params
+type Backoff struct {
+	retries     int
+	factor      float64
+	minDuration time.Duration
+}
+
+// New initialize backoff object
+func New() *Backoff {
+	return &Backoff{
+		retries:     backoffRetries,
+		factor:      backoffFactor,
+		minDuration: backoffMinDuration,
+	}
+}
+
+// Retry process retries until all attempts are completed
+func (b *Backoff) Retry(ctx context.Context, cb retryableFunc) (uint64, error) {
+	var attempt uint64
+	for i := 0; i < b.retries; i++ {
+		// @TODO we should use context to cancel retries
+		err := cb()
+		if err == nil {
+			return attempt, nil
+		}
+		if errors.Is(err, ErrBadRequest) {
+			logger.Errorf("unrecoverable error: %s", err)
+			return attempt, err // fail fast if not recoverable
+		}
+		attempt++
+		backoff := float64(b.minDuration) * math.Pow(b.factor, float64(i))
+		dur := time.Duration(backoff)
+		logger.Errorf("got error: %s on attempt: %d; will retry in %v", err, attempt, dur)
+		time.Sleep(time.Duration(backoff))
+	}
+	return attempt, fmt.Errorf("execution failed after %d retry attempts", b.retries)
+}
diff --git a/app/vmctl/main.go b/app/vmctl/main.go
index df9a21165..d5039a6be 100644
--- a/app/vmctl/main.go
+++ b/app/vmctl/main.go
@@ -65,7 +65,7 @@ func main() {
 					// disable progress bars since openTSDB implementation
 					// does not use progress bar pool
 					vmCfg.DisableProgressBar = true
-					importer, err := vm.NewImporter(vmCfg)
+					importer, err := vm.NewImporter(ctx, vmCfg)
 					if err != nil {
 						return fmt.Errorf("failed to create VM importer: %s", err)
 					}
@@ -100,7 +100,7 @@ func main() {
 					}
 
 					vmCfg := initConfigVM(c)
-					importer, err = vm.NewImporter(vmCfg)
+					importer, err = vm.NewImporter(ctx, vmCfg)
 					if err != nil {
 						return fmt.Errorf("failed to create VM importer: %s", err)
 					}
@@ -137,7 +137,7 @@ func main() {
 
 					vmCfg := initConfigVM(c)
 
-					importer, err := vm.NewImporter(vmCfg)
+					importer, err := vm.NewImporter(ctx, vmCfg)
 					if err != nil {
 						return fmt.Errorf("failed to create VM importer: %s", err)
 					}
@@ -163,7 +163,7 @@ func main() {
 					fmt.Println("Prometheus import mode")
 
 					vmCfg := initConfigVM(c)
-					importer, err = vm.NewImporter(vmCfg)
+					importer, err = vm.NewImporter(ctx, vmCfg)
 					if err != nil {
 						return fmt.Errorf("failed to create VM importer: %s", err)
 					}
diff --git a/app/vmctl/vm/vm.go b/app/vmctl/vm/vm.go
index 96e7c968a..487414d3c 100644
--- a/app/vmctl/vm/vm.go
+++ b/app/vmctl/vm/vm.go
@@ -3,15 +3,16 @@ package vm
 import (
 	"bufio"
 	"compress/gzip"
+	"context"
 	"errors"
 	"fmt"
 	"io"
-	"math"
 	"net/http"
 	"strings"
 	"sync"
 	"time"
 
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/backoff"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/barpool"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/limiter"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/decimal"
@@ -75,7 +76,8 @@ type Importer struct {
 	wg   sync.WaitGroup
 	once sync.Once
 
-	s *stats
+	s       *stats
+	backoff *backoff.Backoff
 }
 
 // ResetStats resets im stats.
@@ -107,7 +109,7 @@ func AddExtraLabelsToImportPath(path string, extraLabels []string) (string, erro
 }
 
 // NewImporter creates new Importer for the given cfg.
-func NewImporter(cfg Config) (*Importer, error) {
+func NewImporter(ctx context.Context, cfg Config) (*Importer, error) {
 	if cfg.Concurrency < 1 {
 		return nil, fmt.Errorf("concurrency can't be lower than 1")
 	}
@@ -136,6 +138,7 @@ func NewImporter(cfg Config) (*Importer, error) {
 		close:      make(chan struct{}),
 		input:      make(chan *TimeSeries, cfg.Concurrency*4),
 		errors:     make(chan *ImportError, cfg.Concurrency),
+		backoff:    backoff.New(),
 	}
 	if err := im.Ping(); err != nil {
 		return nil, fmt.Errorf("ping to %q failed: %s", addr, err)
@@ -154,7 +157,7 @@ func NewImporter(cfg Config) (*Importer, error) {
 		}
 		go func(bar *pb.ProgressBar) {
 			defer im.wg.Done()
-			im.startWorker(bar, cfg.BatchSize, cfg.SignificantFigures, cfg.RoundDigits)
+			im.startWorker(ctx, bar, cfg.BatchSize, cfg.SignificantFigures, cfg.RoundDigits)
 		}(bar)
 	}
 	im.ResetStats()
@@ -205,7 +208,7 @@ func (im *Importer) Close() {
 	})
 }
 
-func (im *Importer) startWorker(bar *pb.ProgressBar, batchSize, significantFigures, roundDigits int) {
+func (im *Importer) startWorker(ctx context.Context, bar *pb.ProgressBar, batchSize, significantFigures, roundDigits int) {
 	var batch []*TimeSeries
 	var dataPoints int
 	var waitForBatch time.Time
@@ -219,7 +222,9 @@ func (im *Importer) startWorker(bar *pb.ProgressBar, batchSize, significantFigur
 			exitErr := &ImportError{
 				Batch: batch,
 			}
-			if err := im.Import(batch); err != nil {
+			retryableFunc := func() error { return im.Import(batch) }
+			_, err := im.backoff.Retry(ctx, retryableFunc)
+			if err != nil {
 				exitErr.Err = err
 			}
 			im.errors <- exitErr
@@ -249,7 +254,7 @@ func (im *Importer) startWorker(bar *pb.ProgressBar, batchSize, significantFigur
 			im.s.idleDuration += time.Since(waitForBatch)
 			im.s.Unlock()
 
-			if err := im.flush(batch); err != nil {
+			if err := im.flush(ctx, batch); err != nil {
 				im.errors <- &ImportError{
 					Batch: batch,
 					Err:   err,
@@ -264,30 +269,16 @@ func (im *Importer) startWorker(bar *pb.ProgressBar, batchSize, significantFigur
 	}
 }
 
-const (
-	// TODO: make configurable
-	backoffRetries     = 5
-	backoffFactor      = 1.7
-	backoffMinDuration = time.Second
-)
-
-func (im *Importer) flush(b []*TimeSeries) error {
-	var err error
-	for i := 0; i < backoffRetries; i++ {
-		err = im.Import(b)
-		if err == nil {
-			return nil
-		}
-		if errors.Is(err, ErrBadRequest) {
-			return err // fail fast if not recoverable
-		}
-		im.s.Lock()
-		im.s.retries++
-		im.s.Unlock()
-		backoff := float64(backoffMinDuration) * math.Pow(backoffFactor, float64(i))
-		time.Sleep(time.Duration(backoff))
+func (im *Importer) flush(ctx context.Context, b []*TimeSeries) error {
+	retryableFunc := func() error { return im.Import(b) }
+	attempts, err := im.backoff.Retry(ctx, retryableFunc)
+	if err != nil {
+		return fmt.Errorf("import failed with %d retries: %s", attempts, err)
 	}
-	return fmt.Errorf("import failed with %d retries: %s", backoffRetries, err)
+	im.s.Lock()
+	im.s.retries = attempts
+	im.s.Unlock()
+	return nil
 }
 
 // Ping sends a ping to im.addr.
diff --git a/app/vminsert/main.go b/app/vminsert/main.go
index 358e120ad..20b86f2cb 100644
--- a/app/vminsert/main.go
+++ b/app/vminsert/main.go
@@ -48,13 +48,13 @@ var (
 		"See also -influxListenAddr.useProxyProtocol")
 	influxUseProxyProtocol = flag.Bool("influxListenAddr.useProxyProtocol", false, "Whether to use proxy protocol for connections accepted at -influxListenAddr . "+
 		"See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt")
-	opentsdbListenAddr = flag.String("opentsdbListenAddr", "", "TCP and UDP address to listen for OpentTSDB metrics. "+
+	opentsdbListenAddr = flag.String("opentsdbListenAddr", "", "TCP and UDP address to listen for OpenTSDB metrics. "+
 		"Telnet put messages and HTTP /api/put messages are simultaneously served on TCP port. "+
 		"Usually :4242 must be set. Doesn't work if empty. "+
 		"See also -opentsdbListenAddr.useProxyProtocol")
 	opentsdbUseProxyProtocol = flag.Bool("opentsdbListenAddr.useProxyProtocol", false, "Whether to use proxy protocol for connections accepted at -opentsdbListenAddr . "+
 		"See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt")
-	opentsdbHTTPListenAddr = flag.String("opentsdbHTTPListenAddr", "", "TCP address to listen for OpentTSDB HTTP put requests. Usually :4242 must be set. Doesn't work if empty. "+
+	opentsdbHTTPListenAddr = flag.String("opentsdbHTTPListenAddr", "", "TCP address to listen for OpenTSDB HTTP put requests. Usually :4242 must be set. Doesn't work if empty. "+
 		"See also -opentsdbHTTPListenAddr.useProxyProtocol")
 	opentsdbHTTPUseProxyProtocol = flag.Bool("opentsdbHTTPListenAddr.useProxyProtocol", false, "Whether to use proxy protocol for connections accepted "+
 		"at -opentsdbHTTPListenAddr . See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt")
diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index d724c4f68..3613c7b7c 100644
--- a/docs/CHANGELOG.md
+++ b/docs/CHANGELOG.md
@@ -32,6 +32,7 @@ The following tip changes can be tested by building VictoriaMetrics components f
 * FEATURE: [MetricsQL](https://docs.victoriametrics.com/MetricsQL.html): add `range_trim_zscore(z, q)` function for dropping outliers located farther than `z*range_stddev(q)` from `range_avg(q)`. This should help removing outliers during query time at [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3759).
 * FEATURE: [vmui](https://docs.victoriametrics.com/#vmui): show `median` instead of `avg` in graph tooltip and line legend, since `median` is more tolerant against spikes. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3706).
 * FEATURE: allow setting zero value for `-search.latencyOffset` command-line flag. This may be needed in [some cases](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/2061#issuecomment-1299109836). Previously the minimum supported value for `-search.latencyOffset` command-line flag was `1s`.
+* FEATURE: [vmagent](https://docs.victoriametrics.com/vmagent.html): add `kuma_sd_config` for [Kuma](http://kuma.io/) Control Plane targets discovery. See [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3389).
 
 * BUGFIX: prevent from possible data ingestion slowdown and query performance slowdown during [background merges of big parts](https://docs.victoriametrics.com/#storage) on systems with small number of CPU cores (1 or 2 CPU cores). The issue has been introduced in [v1.85.0](https://docs.victoriametrics.com/CHANGELOG.html#v1850) when implementing [this feature](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3337). See also [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3790).
 * BUGFIX: properly parse timestamps in milliseconds when [ingesting data via OpenTSDB telnet put protocol](https://docs.victoriametrics.com/#sending-data-via-telnet-put-protocol). Previously timestamps in milliseconds were mistakenly multiplied by 1000. Thanks to @Droxenator for the [pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/3810).
diff --git a/docs/CaseStudies.md b/docs/CaseStudies.md
index 73e7aef80..a42926ec7 100644
--- a/docs/CaseStudies.md
+++ b/docs/CaseStudies.md
@@ -15,6 +15,7 @@ where you can chat with VictoriaMetrics users to get additional references, revi
   - [Brandwatch](#brandwatch)
   - [CERN](#cern)
   - [COLOPL](#colopl)
+  - [Dig Security](#dig-security)
   - [Fly.io](#flyio)
   - [German Research Center for Artificial Intelligence](#german-research-center-for-artificial-intelligence)
   - [Grammarly](#grammarly)
@@ -231,6 +232,49 @@ after evaulating the following remote storage solutions for Prometheus:
 See [slides](https://speakerdeck.com/inletorder/monitoring-platform-with-victoria-metrics) and [video](https://www.youtube.com/watch?v=hUpHIluxw80)
 from `Large-scale, super-load system monitoring platform built with VictoriaMetrics` talk at [Prometheus Meetup Tokyo #3](https://prometheus.connpass.com/event/157721/).
 
+## Dig Security
+
+[Dig Security](https://www.dig.security) is a cloud data security startup with 50+ employees that provides real-time visibility, control, and protection of data assets.
+
+### How we came across VictoriaMetrics
+
+We started with a Prometheus server on EKS. That worked until it didn't. We then spent time scaling it, maintaining it, throwing more $ at it, until we stumbled across VictoriaMetrics.
+
+### What VictoriaMetrics means for us
+
+* Easy to use and maintain
+* Cost effective
+* The ability to handle billions of time series events at any point of time
+* Multiple K8s clusters to monitor
+* Consistent monitoring infra for each cluster across multiple Regions and clouds
+* Secure communication and data storage
+* Easy Retention
+
+### Some of our initial challenges prior to moving to VictoriaMetrics
+
+* Reducing cost by not using a managed solution of one of the clouds
+* Support HA and recover fast
+* No downtimes
+* Having our main prometheus using too much Ram and restarts.
+
+### Some of the reasons we chose VictoriaMetrics
+
+* The API is compatible with Prometheus and all standard PromQL queries work well out of the box
+* Handles storage well
+* Available to use in Grafana easily
+* Single and small executable
+* Easy and fast backups
+* Better benchmarks than all the competitors
+* Open Source and maintained with good community
+
+### Some of the benefits we experienced since working with VictoriaMetrics
+
+* We saved around &#x24;5K USD per month
+* It’s seamless and doesn’t cause any override complications on the Infrastructure team
+* It doesn’t use lots of storage
+* It can serve us in the future in even bigger scales
+* It has support with a great community.
+
 ## Fly.io
 
 [Fly.io](https://fly.io/about/) is a platform for running full stack apps and databases close to your users.
diff --git a/docs/Cluster-VictoriaMetrics.md b/docs/Cluster-VictoriaMetrics.md
index fc0177733..dd833969f 100644
--- a/docs/Cluster-VictoriaMetrics.md
+++ b/docs/Cluster-VictoriaMetrics.md
@@ -928,11 +928,11 @@ Below is the output for `/path/to/vminsert -help`:
   -metricsAuthKey string
      Auth key for /metrics endpoint. It must be passed via authKey query arg. It overrides httpAuth.* settings
   -opentsdbHTTPListenAddr string
-     TCP address to listen for OpentTSDB HTTP put requests. Usually :4242 must be set. Doesn't work if empty. See also -opentsdbHTTPListenAddr.useProxyProtocol
+     TCP address to listen for OpenTSDB HTTP put requests. Usually :4242 must be set. Doesn't work if empty. See also -opentsdbHTTPListenAddr.useProxyProtocol
   -opentsdbHTTPListenAddr.useProxyProtocol
      Whether to use proxy protocol for connections accepted at -opentsdbHTTPListenAddr . See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt
   -opentsdbListenAddr string
-     TCP and UDP address to listen for OpentTSDB metrics. Telnet put messages and HTTP /api/put messages are simultaneously served on TCP port. Usually :4242 must be set. Doesn't work if empty. See also -opentsdbListenAddr.useProxyProtocol
+     TCP and UDP address to listen for OpenTSDB metrics. Telnet put messages and HTTP /api/put messages are simultaneously served on TCP port. Usually :4242 must be set. Doesn't work if empty. See also -opentsdbListenAddr.useProxyProtocol
   -opentsdbListenAddr.useProxyProtocol
      Whether to use proxy protocol for connections accepted at -opentsdbListenAddr . See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt
   -opentsdbTrimTimestamp duration
diff --git a/docs/README.md b/docs/README.md
index 371683916..27f3f0917 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -106,6 +106,7 @@ Case studies:
 * [Brandwatch](https://docs.victoriametrics.com/CaseStudies.html#brandwatch)
 * [CERN](https://docs.victoriametrics.com/CaseStudies.html#cern)
 * [COLOPL](https://docs.victoriametrics.com/CaseStudies.html#colopl)
+* [Dig Security](https://docs.victoriametrics.com/CaseStudies.html#dig-security)
 * [Fly.io](https://docs.victoriametrics.com/CaseStudies.html#flyio)
 * [German Research Center for Artificial Intelligence](https://docs.victoriametrics.com/CaseStudies.html#german-research-center-for-artificial-intelligence)
 * [Grammarly](https://docs.victoriametrics.com/CaseStudies.html#grammarly)
@@ -2264,11 +2265,11 @@ Pass `-help` to VictoriaMetrics in order to see the list of supported command-li
   -metricsAuthKey string
      Auth key for /metrics endpoint. It must be passed via authKey query arg. It overrides httpAuth.* settings
   -opentsdbHTTPListenAddr string
-     TCP address to listen for OpentTSDB HTTP put requests. Usually :4242 must be set. Doesn't work if empty. See also -opentsdbHTTPListenAddr.useProxyProtocol
+     TCP address to listen for OpenTSDB HTTP put requests. Usually :4242 must be set. Doesn't work if empty. See also -opentsdbHTTPListenAddr.useProxyProtocol
   -opentsdbHTTPListenAddr.useProxyProtocol
      Whether to use proxy protocol for connections accepted at -opentsdbHTTPListenAddr . See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt
   -opentsdbListenAddr string
-     TCP and UDP address to listen for OpentTSDB metrics. Telnet put messages and HTTP /api/put messages are simultaneously served on TCP port. Usually :4242 must be set. Doesn't work if empty. See also -opentsdbListenAddr.useProxyProtocol
+     TCP and UDP address to listen for OpenTSDB metrics. Telnet put messages and HTTP /api/put messages are simultaneously served on TCP port. Usually :4242 must be set. Doesn't work if empty. See also -opentsdbListenAddr.useProxyProtocol
   -opentsdbListenAddr.useProxyProtocol
      Whether to use proxy protocol for connections accepted at -opentsdbListenAddr . See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt
   -opentsdbTrimTimestamp duration
diff --git a/docs/Single-server-VictoriaMetrics.md b/docs/Single-server-VictoriaMetrics.md
index d72a42314..7a3a2a046 100644
--- a/docs/Single-server-VictoriaMetrics.md
+++ b/docs/Single-server-VictoriaMetrics.md
@@ -2267,11 +2267,11 @@ Pass `-help` to VictoriaMetrics in order to see the list of supported command-li
   -metricsAuthKey string
      Auth key for /metrics endpoint. It must be passed via authKey query arg. It overrides httpAuth.* settings
   -opentsdbHTTPListenAddr string
-     TCP address to listen for OpentTSDB HTTP put requests. Usually :4242 must be set. Doesn't work if empty. See also -opentsdbHTTPListenAddr.useProxyProtocol
+     TCP address to listen for OpenTSDB HTTP put requests. Usually :4242 must be set. Doesn't work if empty. See also -opentsdbHTTPListenAddr.useProxyProtocol
   -opentsdbHTTPListenAddr.useProxyProtocol
      Whether to use proxy protocol for connections accepted at -opentsdbHTTPListenAddr . See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt
   -opentsdbListenAddr string
-     TCP and UDP address to listen for OpentTSDB metrics. Telnet put messages and HTTP /api/put messages are simultaneously served on TCP port. Usually :4242 must be set. Doesn't work if empty. See also -opentsdbListenAddr.useProxyProtocol
+     TCP and UDP address to listen for OpenTSDB metrics. Telnet put messages and HTTP /api/put messages are simultaneously served on TCP port. Usually :4242 must be set. Doesn't work if empty. See also -opentsdbListenAddr.useProxyProtocol
   -opentsdbListenAddr.useProxyProtocol
      Whether to use proxy protocol for connections accepted at -opentsdbListenAddr . See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt
   -opentsdbTrimTimestamp duration
diff --git a/docs/sd_configs.md b/docs/sd_configs.md
index 7b43a4e24..19dd2057c 100644
--- a/docs/sd_configs.md
+++ b/docs/sd_configs.md
@@ -20,6 +20,7 @@ sort: 24
 * `gce_sd_configs` is for discovering and scraping [Google Compute Engine](https://cloud.google.com/compute) targets. See [these docs](#gce_sd_configs).
 * `http_sd_configs` is for discovering and scraping targerts provided by external http-based service discovery. See [these docs](#http_sd_configs).
 * `kubernetes_sd_configs` is for discovering and scraping [Kubernetes](https://kubernetes.io/) targets. See [these docs](#kubernetes_sd_configs).
+* `kuma_sd_configs` is for discovering and scraping [Kuma](https://kuma.io) targets. See [these docs](#kuma_sd_configs).
 * `nomad_sd_configs` is for discovering and scraping targets registered in [HashiCorp Nomad](https://www.nomadproject.io/). See [these docs](#nomad_sd_configs).
 * `openstack_sd_configs` is for discovering and scraping OpenStack targets. See [these docs](#openstack_sd_configs).
 * `static_configs` is for scraping statically defined targets. See [these docs](#static_configs).
@@ -944,6 +945,42 @@ One of the following `role` types can be configured to discover targets:
   * `__meta_kubernetes_ingress_scheme`: Protocol scheme of ingress, https if TLS config is set. Defaults to http.
   * `__meta_kubernetes_ingress_path`: Path from ingress spec. Defaults to `/`.
 
+## kuma_sd_configs
+
+Kuma service discovery config allows to fetch targets from the specified control plane `server` of [Kuma Service Mesh](https://kuma.io).
+
+It discovers "monitoring assignments" based on Kuma Dataplane Proxies, 
+via the [MADS (Monitoring Assignment Discovery Service)](https://kuma.io/docs/2.1.x/policies/traffic-metrics/#traffic-metrics) 
+[xDS RESP API](http://envoyproxy.io/docs/envoy/latest/api-docs/xds_protocol).
+
+Configuration example:
+
+```yaml
+scrape_configs:
+- job_name: kuma
+  kuma_sd_configs:
+    # server must contain the URL of Kuma Control Plane's MADS xDS server.
+  - server: "http://localhost:5676"
+
+    # Additional HTTP API client options can be specified here.
+    # See https://docs.victoriametrics.com/sd_configs.html#http-api-client-options
+```
+
+The `server` is queried periodically with the interval specified in `-promscrape.kumaSDCheckInterval` command-line flag.
+Discovery errors are tracked in `promscrape_discovery_kuma_errors_total` metric.
+
+Each discovered target has an [`__address__`](https://docs.victoriametrics.com/relabeling.html#how-to-modify-scrape-urls-in-targets) label set
+to one of the targets returned by the http service.
+
+The following meta labels are available on discovered targets during [relabeling](https://docs.victoriametrics.com/vmagent.html#relabeling):
+
+* `__meta_server`: the URL of Kuma Control Plane's MADS xDS server from which the target was extracted
+* `__meta_kuma_mesh`: the name of the mesh
+* `__meta_kuma_service`: the name of the service associated with the proxy 
+* `__meta_kuma_dataplane`: the name of the proxy
+* `__meta_kuma_label_<label_name>`: each label of target given from Kuma Control Plane
+* `__metrics_path__`: the path by which the service metrics are scraped
+
 ## nomad_sd_configs
 
 Nomad SD configuration allows retrieving scrape targets from [HashiCorp Nomad Services](https://www.hashicorp.com/blog/nomad-service-discovery).
diff --git a/docs/vmagent.md b/docs/vmagent.md
index 218436ee6..9cd1419aa 100644
--- a/docs/vmagent.md
+++ b/docs/vmagent.md
@@ -1303,11 +1303,11 @@ See the docs at https://docs.victoriametrics.com/vmagent.html .
   -metricsAuthKey string
      Auth key for /metrics endpoint. It must be passed via authKey query arg. It overrides httpAuth.* settings
   -opentsdbHTTPListenAddr string
-     TCP address to listen for OpentTSDB HTTP put requests. Usually :4242 must be set. Doesn't work if empty. See also -opentsdbHTTPListenAddr.useProxyProtocol
+     TCP address to listen for OpenTSDB HTTP put requests. Usually :4242 must be set. Doesn't work if empty. See also -opentsdbHTTPListenAddr.useProxyProtocol
   -opentsdbHTTPListenAddr.useProxyProtocol
      Whether to use proxy protocol for connections accepted at -opentsdbHTTPListenAddr . See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt
   -opentsdbListenAddr string
-     TCP and UDP address to listen for OpentTSDB metrics. Telnet put messages and HTTP /api/put messages are simultaneously served on TCP port. Usually :4242 must be set. Doesn't work if empty. See also -opentsdbListenAddr.useProxyProtocol
+     TCP and UDP address to listen for OpenTSDB metrics. Telnet put messages and HTTP /api/put messages are simultaneously served on TCP port. Usually :4242 must be set. Doesn't work if empty. See also -opentsdbListenAddr.useProxyProtocol
   -opentsdbListenAddr.useProxyProtocol
      Whether to use proxy protocol for connections accepted at -opentsdbListenAddr . See https://www.haproxy.org/download/1.8/doc/proxy-protocol.txt
   -opentsdbTrimTimestamp duration
@@ -1373,6 +1373,10 @@ See the docs at https://docs.victoriametrics.com/vmagent.html .
      Interval for checking for changes in http endpoint service discovery. This works only if http_sd_configs is configured in '-promscrape.config' file. See https://docs.victoriametrics.com/sd_configs.html#http_sd_configs for details (default 1m0s)
   -promscrape.kubernetes.apiServerTimeout duration
      How frequently to reload the full state from Kubernetes API server (default 30m0s)
+  -promscrape.kumaSDCheckInterval duration
+     Interval for checking for changes in Kuma Service Mesh API service discovery. This works only if kuma_sd_configs is configured in '-promscrape.config' file. See https://docs.victoriametrics.com/sd_configs.html#kuma_sd_configs for details (default 1m0s)
+  -promscrape.kuma.waitTime duration
+     Wait time used by Kuma service discovery. Default value is used if not set
   -promscrape.kubernetesSDCheckInterval duration
      Interval for checking for changes in Kubernetes API server. This works only if kubernetes_sd_configs is configured in '-promscrape.config' file. See https://docs.victoriametrics.com/sd_configs.html#kubernetes_sd_configs for details (default 30s)
   -promscrape.maxDroppedTargets int
diff --git a/lib/promscrape/client.go b/lib/promscrape/client.go
index 52fa157b5..bf30e0369 100644
--- a/lib/promscrape/client.go
+++ b/lib/promscrape/client.go
@@ -262,7 +262,8 @@ func (c *client) ReadData(dst []byte) ([]byte, error) {
 		// This should reduce memory uage when scraping big targets.
 		dst = resp.SwapBody(dst)
 	}
-	err := doRequestWithPossibleRetry(c.hc, req, resp, deadline)
+
+	err := doRequestWithPossibleRetry(c.ctx, c.hc, req, resp, deadline)
 	statusCode := resp.StatusCode()
 	redirectsCount := 0
 	for err == nil && isStatusRedirect(statusCode) {
@@ -282,7 +283,7 @@ func (c *client) ReadData(dst []byte) ([]byte, error) {
 			break
 		}
 		req.URI().UpdateBytes(location)
-		err = doRequestWithPossibleRetry(c.hc, req, resp, deadline)
+		err = doRequestWithPossibleRetry(c.ctx, c.hc, req, resp, deadline)
 		statusCode = resp.StatusCode()
 		redirectsCount++
 	}
@@ -349,13 +350,14 @@ var (
 	scrapeRetries         = metrics.NewCounter(`vm_promscrape_scrape_retries_total`)
 )
 
-func doRequestWithPossibleRetry(hc *fasthttp.HostClient, req *fasthttp.Request, resp *fasthttp.Response, deadline time.Time) error {
+func doRequestWithPossibleRetry(ctx context.Context, hc *fasthttp.HostClient, req *fasthttp.Request, resp *fasthttp.Response, deadline time.Time) error {
 	sleepTime := time.Second
 	scrapeRequests.Inc()
+	reqCtx, cancel := context.WithDeadline(ctx, deadline)
+	defer cancel()
 	for {
-		// Use DoDeadline instead of Do even if hc.ReadTimeout is already set in order to guarantee the given deadline
-		// across multiple retries.
-		err := hc.DoDeadline(req, resp, deadline)
+		// Use DoCtx instead of Do in order to support context cancellation
+		err := hc.DoCtx(reqCtx, req, resp)
 		if err == nil {
 			statusCode := resp.StatusCode()
 			if statusCode != fasthttp.StatusTooManyRequests {
@@ -364,6 +366,7 @@ func doRequestWithPossibleRetry(hc *fasthttp.HostClient, req *fasthttp.Request,
 		} else if err != fasthttp.ErrConnectionClosed && !strings.Contains(err.Error(), "broken pipe") {
 			return err
 		}
+
 		// Retry request after exponentially increased sleep.
 		maxSleepTime := time.Until(deadline)
 		if sleepTime > maxSleepTime {
diff --git a/lib/promscrape/config.go b/lib/promscrape/config.go
index dd64733cc..4ae3369e9 100644
--- a/lib/promscrape/config.go
+++ b/lib/promscrape/config.go
@@ -30,6 +30,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/gce"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/http"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/kubernetes"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/kuma"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/nomad"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/openstack"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/yandexcloud"
@@ -264,6 +265,7 @@ type ScrapeConfig struct {
 	GCESDConfigs          []gce.SDConfig          `yaml:"gce_sd_configs,omitempty"`
 	HTTPSDConfigs         []http.SDConfig         `yaml:"http_sd_configs,omitempty"`
 	KubernetesSDConfigs   []kubernetes.SDConfig   `yaml:"kubernetes_sd_configs,omitempty"`
+	KumaSDConfigs         []kuma.SDConfig         `yaml:"kuma_sd_configs,omitempty"`
 	NomadSDConfigs        []nomad.SDConfig        `yaml:"nomad_sd_configs,omitempty"`
 	OpenStackSDConfigs    []openstack.SDConfig    `yaml:"openstack_sd_configs,omitempty"`
 	StaticConfigs         []StaticConfig          `yaml:"static_configs,omitempty"`
@@ -332,6 +334,9 @@ func (sc *ScrapeConfig) mustStop() {
 	for i := range sc.KubernetesSDConfigs {
 		sc.KubernetesSDConfigs[i].MustStop()
 	}
+	for i := range sc.KumaSDConfigs {
+		sc.KumaSDConfigs[i].MustStop()
+	}
 	for i := range sc.NomadSDConfigs {
 		sc.NomadSDConfigs[i].MustStop()
 	}
@@ -812,6 +817,33 @@ func (cfg *Config) getKubernetesSDScrapeWork(prev []*ScrapeWork) []*ScrapeWork {
 	return dst
 }
 
+// getKumaSDScrapeWork returns `kuma_sd_configs` ScrapeWork from cfg.
+func (cfg *Config) getKumaSDScrapeWork(prev []*ScrapeWork) []*ScrapeWork {
+	swsPrevByJob := getSWSByJob(prev)
+	dst := make([]*ScrapeWork, 0, len(prev))
+	for _, sc := range cfg.ScrapeConfigs {
+		dstLen := len(dst)
+		ok := true
+		for j := range sc.KumaSDConfigs {
+			sdc := &sc.KumaSDConfigs[j]
+			var okLocal bool
+			dst, okLocal = appendSDScrapeWork(dst, sdc, cfg.baseDir, sc.swc, "kuma_sd_config")
+			if ok {
+				ok = okLocal
+			}
+		}
+		if ok {
+			continue
+		}
+		swsPrev := swsPrevByJob[sc.swc.jobName]
+		if len(swsPrev) > 0 {
+			logger.Errorf("there were errors when discovering kuma targets for job %q, so preserving the previous targets", sc.swc.jobName)
+			dst = append(dst[:dstLen], swsPrev...)
+		}
+	}
+	return dst
+}
+
 // getNomadSDScrapeWork returns `nomad_sd_configs` ScrapeWork from cfg.
 func (cfg *Config) getNomadSDScrapeWork(prev []*ScrapeWork) []*ScrapeWork {
 	swsPrevByJob := getSWSByJob(prev)
@@ -893,7 +925,7 @@ func (cfg *Config) getYandexCloudSDScrapeWork(prev []*ScrapeWork) []*ScrapeWork
 	return dst
 }
 
-// getStaticScrapeWork returns `static_configs` ScrapeWork from from cfg.
+// getStaticScrapeWork returns `static_configs` ScrapeWork from cfg.
 func (cfg *Config) getStaticScrapeWork() []*ScrapeWork {
 	var dst []*ScrapeWork
 	for _, sc := range cfg.ScrapeConfigs {
diff --git a/lib/promscrape/discovery/kuma/api.go b/lib/promscrape/discovery/kuma/api.go
new file mode 100644
index 000000000..2e682a475
--- /dev/null
+++ b/lib/promscrape/discovery/kuma/api.go
@@ -0,0 +1,207 @@
+package kuma
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"path"
+	"sync"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutils"
+	"github.com/VictoriaMetrics/metrics"
+)
+
+var configMap = discoveryutils.NewConfigMap()
+
+type apiConfig struct {
+	client *discoveryutils.Client
+	path   string
+
+	cancel        context.CancelFunc
+	wg            sync.WaitGroup
+	mu            sync.Mutex // protects targets
+	targets       []kumaTarget
+	latestVersion string
+	latestNonce   string
+
+	fetchErrors *metrics.Counter
+	parseErrors *metrics.Counter
+}
+
+const (
+	discoveryNode      = "victoria-metrics"
+	xdsApiVersion      = "v3"
+	xdsRequestType     = "discovery"
+	xdsResourceType    = "monitoringassignments"
+	xdsResourceTypeUrl = "type.googleapis.com/kuma.observability.v1.MonitoringAssignment"
+)
+
+var waitTime = flag.Duration("promscrape.kuma.waitTime", 0, "Wait time used by Kuma service discovery. Default value is used if not set")
+
+func getAPIConfig(sdc *SDConfig, baseDir string) (*apiConfig, error) {
+	v, err := configMap.Get(sdc, func() (interface{}, error) { return newAPIConfig(sdc, baseDir) })
+	if err != nil {
+		return nil, err
+	}
+	return v.(*apiConfig), nil
+}
+
+func newAPIConfig(sdc *SDConfig, baseDir string) (*apiConfig, error) {
+	ac, err := sdc.HTTPClientConfig.NewConfig(baseDir)
+	if err != nil {
+		return nil, fmt.Errorf("cannot parse auth config: %w", err)
+	}
+	parsedURL, err := url.Parse(sdc.Server)
+	if err != nil {
+		return nil, fmt.Errorf("cannot parse kuma_sd server URL: %w", err)
+	}
+	apiServer := fmt.Sprintf("%s://%s", parsedURL.Scheme, parsedURL.Host)
+
+	proxyAC, err := sdc.ProxyClientConfig.NewConfig(baseDir)
+	if err != nil {
+		return nil, fmt.Errorf("cannot parse proxy auth config: %w", err)
+	}
+	client, err := discoveryutils.NewClient(apiServer, ac, sdc.ProxyURL, proxyAC)
+	if err != nil {
+		return nil, fmt.Errorf("cannot create HTTP client for %q: %w", apiServer, err)
+	}
+
+	apiPath := path.Join(
+		parsedURL.RequestURI(),
+		xdsApiVersion,
+		xdsRequestType+":"+xdsResourceType,
+	)
+
+	cfg := &apiConfig{
+		client:      client,
+		path:        apiPath,
+		fetchErrors: metrics.GetOrCreateCounter(fmt.Sprintf(`promscrape_discovery_kuma_errors_total{type="fetch",url=%q}`, sdc.Server)),
+		parseErrors: metrics.GetOrCreateCounter(fmt.Sprintf(`promscrape_discovery_kuma_errors_total{type="parse",url=%q}`, sdc.Server)),
+	}
+
+	// initialize targets synchronously and start updating them in background
+	cfg.startWatcher()
+
+	return cfg, nil
+}
+
+func (cfg *apiConfig) startWatcher() func() {
+	ctx, cancel := context.WithCancel(context.Background())
+	cfg.cancel = cancel
+
+	// blocking initial targets update
+	if err := cfg.updateTargets(ctx); err != nil {
+		logger.Errorf("there were errors when discovering kuma targets, so preserving the previous targets. error: %v", err)
+	}
+
+	// start updating targets with a long polling in background
+	cfg.wg.Add(1)
+	go func() {
+		ticker := time.NewTicker(*SDCheckInterval)
+		defer cfg.wg.Done()
+		defer ticker.Stop()
+
+		for {
+			select {
+			case <-ticker.C:
+				// we are constantly waiting for targets updates in long polling requests
+				err := cfg.updateTargets(ctx)
+				if err != nil {
+					logger.Errorf("there were errors when discovering kuma targets, so preserving the previous targets. error: %v", err)
+				}
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	return cancel
+}
+
+func (cfg *apiConfig) stopWatcher() {
+	cfg.cancel()
+	cfg.wg.Wait()
+}
+
+func (cfg *apiConfig) getTargets() ([]kumaTarget, error) {
+	cfg.mu.Lock()
+	defer cfg.mu.Unlock()
+	return cfg.targets, nil
+}
+
+func (cfg *apiConfig) updateTargets(ctx context.Context) error {
+	requestBody, err := json.Marshal(discoveryRequest{
+		VersionInfo:   cfg.latestVersion,
+		Node:          discoveryRequestNode{Id: discoveryNode},
+		TypeUrl:       xdsResourceTypeUrl,
+		ResponseNonce: cfg.latestNonce,
+	})
+	if err != nil {
+		return fmt.Errorf("cannot marshal request body for kuma_sd api: %w", err)
+	}
+
+	var statusCode int
+	data, err := cfg.client.GetBlockingAPIResponseWithParamsCtx(
+		ctx,
+		cfg.path,
+		func(request *http.Request) {
+			request.Method = http.MethodPost
+			request.Body = io.NopCloser(bytes.NewReader(requestBody))
+
+			// set max duration for long polling request
+			query := request.URL.Query()
+			query.Add("fetch-timeout", cfg.getWaitTime().String())
+			request.URL.RawQuery = query.Encode()
+
+			request.Header.Set("Accept", "application/json")
+			request.Header.Set("Content-Type", "application/json")
+		},
+		func(response *http.Response) {
+			statusCode = response.StatusCode
+		},
+	)
+
+	if statusCode == http.StatusNotModified {
+		return nil
+	}
+	if err != nil {
+		cfg.fetchErrors.Inc()
+		return fmt.Errorf("cannot read kuma_sd api response: %w", err)
+	}
+
+	response, err := parseDiscoveryResponse(data)
+	if err != nil {
+		cfg.parseErrors.Inc()
+		return fmt.Errorf("cannot parse kuma_sd api response: %w", err)
+	}
+
+	cfg.mu.Lock()
+	defer cfg.mu.Unlock()
+	cfg.targets = parseKumaTargets(response)
+	cfg.latestVersion = response.VersionInfo
+	cfg.latestNonce = response.Nonce
+
+	return nil
+}
+
+func (cfg *apiConfig) getWaitTime() time.Duration {
+	d := discoveryutils.BlockingClientReadTimeout
+	// Reduce wait time to avoid timeouts (request execution time should be less than the read timeout)
+	d -= d / 8
+	if *waitTime > time.Second && *waitTime < d {
+		d = *waitTime
+	}
+	return d
+}
+
+func (cfg *apiConfig) mustStop() {
+	cfg.stopWatcher()
+	cfg.client.Stop()
+}
diff --git a/lib/promscrape/discovery/kuma/kuma.go b/lib/promscrape/discovery/kuma/kuma.go
new file mode 100644
index 000000000..a3c62fc61
--- /dev/null
+++ b/lib/promscrape/discovery/kuma/kuma.go
@@ -0,0 +1,113 @@
+package kuma
+
+import (
+	"flag"
+	"fmt"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/proxy"
+)
+
+// SDCheckInterval defines interval for targets refresh.
+var SDCheckInterval = flag.Duration("promscrape.kumaSDCheckInterval", time.Minute, "Interval for checking for changes in kuma service discovery. "+
+	"This works only if kuma_sd_configs is configured in '-promscrape.config' file. "+
+	"See https://docs.victoriametrics.com/sd_configs.html#kuma_sd_configs for details")
+
+// SDConfig represents service discovery config for Kuma Service Mesh.
+//
+// See https://prometheus.io/docs/prometheus/latest/configuration/configuration/#kuma_sd_config
+type SDConfig struct {
+	Server            string                     `yaml:"server"`
+	HTTPClientConfig  promauth.HTTPClientConfig  `yaml:",inline"`
+	ProxyURL          *proxy.URL                 `yaml:"proxy_url,omitempty"`
+	ProxyClientConfig promauth.ProxyClientConfig `yaml:",inline"`
+}
+
+type kumaTarget struct {
+	Mesh         string            `json:"mesh"`
+	ControlPlane string            `json:"controlplane"`
+	Service      string            `json:"service"`
+	DataPlane    string            `json:"dataplane"`
+	Instance     string            `json:"instance"`
+	Scheme       string            `json:"scheme"`
+	Address      string            `json:"address"`
+	MetricsPath  string            `json:"metrics_path"`
+	Labels       map[string]string `json:"labels"`
+}
+
+// GetLabels returns kuma service discovery labels according to sdc.
+func (sdc *SDConfig) GetLabels(baseDir string) ([]*promutils.Labels, error) {
+	cfg, err := getAPIConfig(sdc, baseDir)
+	if err != nil {
+		return nil, fmt.Errorf("cannot get API config for kuma_sd: %w", err)
+	}
+	targets, err := cfg.getTargets()
+	if err != nil {
+		return nil, err
+	}
+	return kumaTargetsToLabels(targets, sdc.Server), nil
+}
+
+// MustStop stops further usage for sdc.
+func (sdc *SDConfig) MustStop() {
+	v := configMap.Delete(sdc)
+	if v != nil {
+		// v can be nil if GetLabels wasn't called yet.
+		cfg := v.(*apiConfig)
+		cfg.mustStop()
+	}
+}
+
+func kumaTargetsToLabels(src []kumaTarget, sourceURL string) []*promutils.Labels {
+	ms := make([]*promutils.Labels, 0, len(src))
+	for _, target := range src {
+		m := promutils.NewLabels(8 + len(target.Labels))
+
+		m.Add("instance", target.Instance)
+		m.Add("__address__", target.Address)
+		m.Add("__scheme__", target.Scheme)
+		m.Add("__metrics_path__", target.MetricsPath)
+		m.Add("__meta_server", sourceURL)
+		m.Add("__meta_kuma_mesh", target.Mesh)
+		m.Add("__meta_kuma_service", target.Service)
+		m.Add("__meta_kuma_dataplane", target.DataPlane)
+		for k, v := range target.Labels {
+			m.Add("__meta_kuma_label_"+k, v)
+		}
+
+		m.RemoveDuplicates()
+		ms = append(ms, m)
+	}
+	return ms
+}
+
+func parseKumaTargets(response discoveryResponse) []kumaTarget {
+	result := make([]kumaTarget, 0, len(response.Resources))
+
+	for _, resource := range response.Resources {
+		for _, target := range resource.Targets {
+			labels := make(map[string]string)
+			for label, value := range resource.Labels {
+				labels[label] = value
+			}
+			for label, value := range target.Labels {
+				labels[label] = value
+			}
+			result = append(result, kumaTarget{
+				Mesh:         resource.Mesh,
+				ControlPlane: response.ControlPlane.Identifier,
+				Service:      resource.Service,
+				DataPlane:    target.Name,
+				Instance:     target.Name,
+				Scheme:       target.Scheme,
+				Address:      target.Address,
+				MetricsPath:  target.MetricsPath,
+				Labels:       labels,
+			})
+		}
+	}
+
+	return result
+}
diff --git a/lib/promscrape/discovery/kuma/xds.go b/lib/promscrape/discovery/kuma/xds.go
new file mode 100644
index 000000000..32fcdafa4
--- /dev/null
+++ b/lib/promscrape/discovery/kuma/xds.go
@@ -0,0 +1,56 @@
+package kuma
+
+import (
+	"encoding/json"
+	"fmt"
+)
+
+// discoveryRequest represent xDS-requests for Kuma Service Mesh
+// https://www.envoyproxy.io/docs/envoy/latest/api-v3/service/discovery/v3/discovery.proto#envoy-v3-api-msg-service-discovery-v3-discoveryrequest
+type discoveryRequest struct {
+	VersionInfo   string               `json:"version_info,omitempty"`
+	Node          discoveryRequestNode `json:"node,omitempty"`
+	ResourceNames []string             `json:"resource_names,omitempty"`
+	TypeUrl       string               `json:"type_url,omitempty"`
+	ResponseNonce string               `json:"response_nonce,omitempty"`
+}
+
+type discoveryRequestNode struct {
+	Id string `json:"id,omitempty"`
+}
+
+// discoveryResponse represent xDS-requests for Kuma Service Mesh
+// https://www.envoyproxy.io/docs/envoy/latest/api-v3/service/discovery/v3/discovery.proto#envoy-v3-api-msg-service-discovery-v3-discoveryresponse
+type discoveryResponse struct {
+	VersionInfo string `json:"version_info,omitempty"`
+	Resources   []struct {
+		Mesh    string `json:"mesh,omitempty"`
+		Service string `json:"service,omitempty"`
+		Targets []struct {
+			Name        string            `json:"name,omitempty"`
+			Scheme      string            `json:"scheme,omitempty"`
+			Address     string            `json:"address,omitempty"`
+			MetricsPath string            `json:"metrics_path,omitempty"`
+			Labels      map[string]string `json:"labels,omitempty"`
+		} `json:"targets,omitempty"`
+		Labels map[string]string `json:"labels,omitempty"`
+	} `json:"resources,omitempty"`
+	TypeUrl      string `json:"type_url,omitempty"`
+	Nonce        string `json:"nonce,omitempty"`
+	ControlPlane struct {
+		Identifier string `json:"identifier,omitempty"`
+	} `json:"control_plane,omitempty"`
+}
+
+func parseDiscoveryResponse(data []byte) (discoveryResponse, error) {
+	response := discoveryResponse{}
+	err := json.Unmarshal(data, &response)
+	if err != nil {
+		return discoveryResponse{}, fmt.Errorf("cannot parse kuma_sd api response, err:  %w", err)
+	}
+	if response.TypeUrl != xdsResourceTypeUrl {
+		return discoveryResponse{}, fmt.Errorf("unexpected type_url in kuma_sd api response, expected: %s, got: %s", xdsResourceTypeUrl, response.TypeUrl)
+	}
+
+	return response, nil
+}
diff --git a/lib/promscrape/discoveryutils/client.go b/lib/promscrape/discoveryutils/client.go
index 319c1f2f9..714c4b16f 100644
--- a/lib/promscrape/discoveryutils/client.go
+++ b/lib/promscrape/discoveryutils/client.go
@@ -42,6 +42,9 @@ const (
 	DefaultClientReadTimeout = time.Minute
 )
 
+type RequestCallback func(request *http.Request)
+type ResponseCallback func(response *http.Response)
+
 func concurrencyLimitChInit() {
 	concurrencyLimitCh = make(chan struct{}, *maxConcurrency)
 }
@@ -165,7 +168,7 @@ func (c *Client) Context() context.Context {
 }
 
 // GetAPIResponseWithReqParams returns response for given absolute path with optional callback for request.
-func (c *Client) GetAPIResponseWithReqParams(path string, modifyRequest func(request *http.Request)) ([]byte, error) {
+func (c *Client) GetAPIResponseWithReqParams(path string, modifyRequest RequestCallback) ([]byte, error) {
 	return c.getAPIResponse(path, modifyRequest)
 }
 
@@ -175,7 +178,7 @@ func (c *Client) GetAPIResponse(path string) ([]byte, error) {
 }
 
 // GetAPIResponse returns response for the given absolute path with optional callback for request.
-func (c *Client) getAPIResponse(path string, modifyRequest func(request *http.Request)) ([]byte, error) {
+func (c *Client) getAPIResponse(path string, modifyRequest RequestCallback) ([]byte, error) {
 	// Limit the number of concurrent API requests.
 	concurrencyLimitChOnce.Do(concurrencyLimitChInit)
 	t := timerpool.Get(*maxWaitTime)
@@ -194,17 +197,22 @@ func (c *Client) getAPIResponse(path string, modifyRequest func(request *http.Re
 }
 
 // GetBlockingAPIResponse returns response for given absolute path with blocking client and optional callback for api response,
-func (c *Client) GetBlockingAPIResponse(path string, inspectResponse func(resp *http.Response)) ([]byte, error) {
+func (c *Client) GetBlockingAPIResponse(path string, inspectResponse ResponseCallback) ([]byte, error) {
 	return c.getAPIResponseWithParamsAndClientCtx(c.clientCtx, c.blockingClient, path, nil, inspectResponse)
 }
 
 // GetBlockingAPIResponseCtx returns response for given absolute path with blocking client and optional callback for api response,
-func (c *Client) GetBlockingAPIResponseCtx(ctx context.Context, path string, inspectResponse func(resp *http.Response)) ([]byte, error) {
+func (c *Client) GetBlockingAPIResponseCtx(ctx context.Context, path string, inspectResponse ResponseCallback) ([]byte, error) {
 	return c.getAPIResponseWithParamsAndClientCtx(ctx, c.blockingClient, path, nil, inspectResponse)
 }
 
+// GetBlockingAPIResponseWithParamsCtx returns response for given absolute path with blocking client and optional callback for api response,
+func (c *Client) GetBlockingAPIResponseWithParamsCtx(ctx context.Context, path string, modifyRequest RequestCallback, inspectResponse ResponseCallback) ([]byte, error) {
+	return c.getAPIResponseWithParamsAndClientCtx(ctx, c.blockingClient, path, modifyRequest, inspectResponse)
+}
+
 // getAPIResponseWithParamsAndClient returns response for the given absolute path with optional callback for request and for response.
-func (c *Client) getAPIResponseWithParamsAndClientCtx(ctx context.Context, client *HTTPClient, path string, modifyRequest func(req *http.Request), inspectResponse func(resp *http.Response)) ([]byte, error) {
+func (c *Client) getAPIResponseWithParamsAndClientCtx(ctx context.Context, client *HTTPClient, path string, modifyRequest RequestCallback, inspectResponse ResponseCallback) ([]byte, error) {
 	requestURL := c.apiServer + path
 	u, err := url.Parse(requestURL)
 	if err != nil {
diff --git a/lib/promscrape/scraper.go b/lib/promscrape/scraper.go
index 0686e47c2..8118fe719 100644
--- a/lib/promscrape/scraper.go
+++ b/lib/promscrape/scraper.go
@@ -26,6 +26,7 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/gce"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/http"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/kubernetes"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/kuma"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/nomad"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/openstack"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discovery/yandexcloud"
@@ -132,6 +133,7 @@ func runScraper(configFile string, pushData func(at *auth.Token, wr *prompbmarsh
 	scs.add("gce_sd_configs", *gce.SDCheckInterval, func(cfg *Config, swsPrev []*ScrapeWork) []*ScrapeWork { return cfg.getGCESDScrapeWork(swsPrev) })
 	scs.add("http_sd_configs", *http.SDCheckInterval, func(cfg *Config, swsPrev []*ScrapeWork) []*ScrapeWork { return cfg.getHTTPDScrapeWork(swsPrev) })
 	scs.add("kubernetes_sd_configs", *kubernetes.SDCheckInterval, func(cfg *Config, swsPrev []*ScrapeWork) []*ScrapeWork { return cfg.getKubernetesSDScrapeWork(swsPrev) })
+	scs.add("kuma_sd_configs", *kuma.SDCheckInterval, func(cfg *Config, swsPrev []*ScrapeWork) []*ScrapeWork { return cfg.getKumaSDScrapeWork(swsPrev) })
 	scs.add("nomad_sd_configs", *nomad.SDCheckInterval, func(cfg *Config, swsPrev []*ScrapeWork) []*ScrapeWork { return cfg.getNomadSDScrapeWork(swsPrev) })
 	scs.add("openstack_sd_configs", *openstack.SDCheckInterval, func(cfg *Config, swsPrev []*ScrapeWork) []*ScrapeWork { return cfg.getOpenStackSDScrapeWork(swsPrev) })
 	scs.add("yandexcloud_sd_configs", *yandexcloud.SDCheckInterval, func(cfg *Config, swsPrev []*ScrapeWork) []*ScrapeWork { return cfg.getYandexCloudSDScrapeWork(swsPrev) })


Test patch:
diff --git a/app/vmagent/remotewrite/pendingseries_test.go b/app/vmagent/remotewrite/pendingseries_test.go
index 2535cbb8c..ef4ab11ee 100644
--- a/app/vmagent/remotewrite/pendingseries_test.go
+++ b/app/vmagent/remotewrite/pendingseries_test.go
@@ -2,6 +2,7 @@ package remotewrite
 
 import (
 	"fmt"
+	"math"
 	"testing"
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/prompbmarshal"
@@ -21,7 +22,7 @@ func TestPushWriteRequest(t *testing.T) {
 }
 
 func testPushWriteRequest(t *testing.T, rowsCount, expectedBlockLenProm, expectedBlockLenVM int) {
-	f := func(isVMRemoteWrite bool, expectedBlockLen int) {
+	f := func(isVMRemoteWrite bool, expectedBlockLen int, tolerancePrc float64) {
 		t.Helper()
 		wr := newTestWriteRequest(rowsCount, 20)
 		pushBlockLen := 0
@@ -32,17 +33,27 @@ func testPushWriteRequest(t *testing.T, rowsCount, expectedBlockLenProm, expecte
 			pushBlockLen = len(block)
 		}
 		pushWriteRequest(wr, pushBlock, isVMRemoteWrite)
-		if pushBlockLen != expectedBlockLen {
+
+		if pushBlockLen == expectedBlockLen {
+			return
+		}
+
+		if tolerancePrc == 0 {
 			t.Fatalf("unexpected block len for rowsCount=%d, isVMRemoteWrite=%v; got %d bytes; expecting %d bytes",
 				rowsCount, isVMRemoteWrite, pushBlockLen, expectedBlockLen)
 		}
+
+		if math.Abs(float64(pushBlockLen-expectedBlockLen)/float64(expectedBlockLen)*100) > tolerancePrc {
+			t.Fatalf("unexpected block len for rowsCount=%d, isVMRemoteWrite=%v; got %d bytes; expecting %d bytes +- %.0f%%",
+				rowsCount, isVMRemoteWrite, pushBlockLen, expectedBlockLen, tolerancePrc)
+		}
 	}
 
 	// Check Prometheus remote write
-	f(false, expectedBlockLenProm)
+	f(false, expectedBlockLenProm, 0)
 
 	// Check VictoriaMetrics remote write
-	f(true, expectedBlockLenVM)
+	f(true, expectedBlockLenVM, 15)
 }
 
 func newTestWriteRequest(seriesCount, labelsCount int) *prompbmarshal.WriteRequest {
diff --git a/app/vmctl/backoff/backoff_test.go b/app/vmctl/backoff/backoff_test.go
new file mode 100644
index 000000000..e9972a27e
--- /dev/null
+++ b/app/vmctl/backoff/backoff_test.go
@@ -0,0 +1,96 @@
+package backoff
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+)
+
+func TestRetry_Do(t *testing.T) {
+	counter := 0
+	tests := []struct {
+		name               string
+		backoffRetries     int
+		backoffFactor      float64
+		backoffMinDuration time.Duration
+		retryableFunc      retryableFunc
+		ctx                context.Context
+		withCancel         bool
+		want               uint64
+		wantErr            bool
+	}{
+		{
+			name: "return bad request",
+			retryableFunc: func() error {
+				return ErrBadRequest
+			},
+			ctx:     context.Background(),
+			want:    0,
+			wantErr: true,
+		},
+		{
+			name: "empty retries values",
+			retryableFunc: func() error {
+				time.Sleep(time.Millisecond * 100)
+				return nil
+			},
+			ctx:     context.Background(),
+			want:    0,
+			wantErr: false,
+		},
+		{
+			name:               "only one retry test",
+			backoffRetries:     5,
+			backoffFactor:      1.7,
+			backoffMinDuration: time.Millisecond * 10,
+			retryableFunc: func() error {
+				t := time.NewTicker(time.Millisecond * 5)
+				defer t.Stop()
+				for range t.C {
+					counter++
+					if counter%2 == 0 {
+						return fmt.Errorf("got some error")
+					}
+					if counter%3 == 0 {
+						return nil
+					}
+				}
+				return nil
+			},
+			ctx:     context.Background(),
+			want:    1,
+			wantErr: false,
+		},
+		{
+			name:               "all retries failed test",
+			backoffRetries:     5,
+			backoffFactor:      0.1,
+			backoffMinDuration: time.Millisecond * 10,
+			retryableFunc: func() error {
+				t := time.NewTicker(time.Millisecond * 5)
+				defer t.Stop()
+				for range t.C {
+					return fmt.Errorf("got some error")
+				}
+				return nil
+			},
+			ctx:     context.Background(),
+			want:    5,
+			wantErr: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			r := New()
+			got, err := r.Retry(tt.ctx, tt.retryableFunc)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("Retry() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if got != tt.want {
+				t.Errorf("Retry() got = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
diff --git a/app/vmctl/prometheus_test.go b/app/vmctl/prometheus_test.go
index cf515ee10..dbf74eda5 100644
--- a/app/vmctl/prometheus_test.go
+++ b/app/vmctl/prometheus_test.go
@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"os"
 	"testing"
 	"time"
@@ -58,7 +59,7 @@ func Test_prometheusProcessor_run(t *testing.T) {
 					return client
 				},
 				im: func(vmCfg vm.Config) *vm.Importer {
-					importer, err := vm.NewImporter(vmCfg)
+					importer, err := vm.NewImporter(context.Background(), vmCfg)
 					if err != nil {
 						t.Fatalf("error init importer: %s", err)
 					}
@@ -95,7 +96,7 @@ func Test_prometheusProcessor_run(t *testing.T) {
 					return client
 				},
 				im: func(vmCfg vm.Config) *vm.Importer {
-					importer, err := vm.NewImporter(vmCfg)
+					importer, err := vm.NewImporter(context.Background(), vmCfg)
 					if err != nil {
 						t.Fatalf("error init importer: %s", err)
 					}
diff --git a/app/vmctl/remote_read_test.go b/app/vmctl/remote_read_test.go
index 1fa0c4b5d..bfb19b2cd 100644
--- a/app/vmctl/remote_read_test.go
+++ b/app/vmctl/remote_read_test.go
@@ -110,6 +110,7 @@ func TestRemoteRead(t *testing.T) {
 
 	for _, tt := range testCases {
 		t.Run(tt.name, func(t *testing.T) {
+			ctx := context.Background()
 			remoteReadServer := remote_read_integration.NewRemoteReadServer(t)
 			defer remoteReadServer.Close()
 			remoteWriteServer := remote_read_integration.NewRemoteWriteServer(t)
@@ -139,7 +140,7 @@ func TestRemoteRead(t *testing.T) {
 
 			tt.vmCfg.Addr = remoteWriteServer.URL()
 
-			importer, err := vm.NewImporter(tt.vmCfg)
+			importer, err := vm.NewImporter(ctx, tt.vmCfg)
 			if err != nil {
 				t.Fatalf("failed to create VM importer: %s", err)
 			}
@@ -156,7 +157,6 @@ func TestRemoteRead(t *testing.T) {
 				cc: 1,
 			}
 
-			ctx := context.Background()
 			err = rmp.run(ctx, true, false)
 			if err != nil {
 				t.Fatalf("failed to run remote read processor: %s", err)
@@ -263,6 +263,7 @@ func TestSteamRemoteRead(t *testing.T) {
 
 	for _, tt := range testCases {
 		t.Run(tt.name, func(t *testing.T) {
+			ctx := context.Background()
 			remoteReadServer := remote_read_integration.NewRemoteReadStreamServer(t)
 			defer remoteReadServer.Close()
 			remoteWriteServer := remote_read_integration.NewRemoteWriteServer(t)
@@ -292,7 +293,7 @@ func TestSteamRemoteRead(t *testing.T) {
 
 			tt.vmCfg.Addr = remoteWriteServer.URL()
 
-			importer, err := vm.NewImporter(tt.vmCfg)
+			importer, err := vm.NewImporter(ctx, tt.vmCfg)
 			if err != nil {
 				t.Fatalf("failed to create VM importer: %s", err)
 			}
@@ -309,7 +310,6 @@ func TestSteamRemoteRead(t *testing.T) {
 				cc: 1,
 			}
 
-			ctx := context.Background()
 			err = rmp.run(ctx, true, false)
 			if err != nil {
 				t.Fatalf("failed to run remote read processor: %s", err)
diff --git a/lib/promscrape/discovery/kuma/api_test.go b/lib/promscrape/discovery/kuma/api_test.go
new file mode 100644
index 000000000..ee11ac69a
--- /dev/null
+++ b/lib/promscrape/discovery/kuma/api_test.go
@@ -0,0 +1,176 @@
+package kuma
+
+import (
+	"reflect"
+	"testing"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
+)
+
+func Test_buildAPIPath(t *testing.T) {
+	type args struct {
+		server string
+	}
+	tests := []struct {
+		name    string
+		args    args
+		want    string
+		wantErr bool
+	}{
+		{
+			name: "get api path ok",
+			args: args{server: "http://localhost:5676"},
+			want: "/v3/discovery:monitoringassignments",
+		},
+		{
+			name:    "get api path incorrect server URL",
+			args:    args{server: ":"},
+			wantErr: true,
+		},
+		{
+			name:    "get api path incorrect server URL err",
+			args:    args{server: "api"},
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			sdConf := &SDConfig{
+				Server:            tt.args.server,
+				HTTPClientConfig:  promauth.HTTPClientConfig{},
+				ProxyClientConfig: promauth.ProxyClientConfig{},
+			}
+			apiConf, err := getAPIConfig(sdConf, ".")
+
+			if tt.wantErr {
+				if err == nil {
+					t.Errorf("buildAPIPath() error = %v, wantErr %v", err, tt.wantErr)
+				}
+				return
+			}
+
+			if apiConf.path != tt.want {
+				t.Errorf("buildAPIPath() got = %v, want = %v", apiConf.path, tt.want)
+			}
+
+			sdConf.MustStop()
+		})
+	}
+
+}
+
+func Test_parseAPIResponse(t *testing.T) {
+	type args struct {
+		data []byte
+	}
+	tests := []struct {
+		name    string
+		args    args
+		want    []kumaTarget
+		wantErr bool
+	}{
+
+		{
+			name: "parse ok",
+			args: args{
+				data: []byte(`{
+    "version_info":"5dc9a5dd-2091-4426-a886-dfdc24fc99d7",
+    "resources":[
+       {
+          "@type":"type.googleapis.com/kuma.observability.v1.MonitoringAssignment",
+          "mesh":"default",
+          "service":"redis",
+          "labels":{ "test":"test1" },
+          "targets":[
+             {
+                "name":"redis",
+                "scheme":"http",
+                "address":"127.0.0.1:5670",
+                "metrics_path":"/metrics",
+                "labels":{ "kuma_io_protocol":"tcp" }
+             }
+          ]
+       },
+       {
+          "@type":"type.googleapis.com/kuma.observability.v1.MonitoringAssignment",
+          "mesh":"default",
+          "service":"app",
+          "labels":{ "test":"test2" },
+          "targets":[
+             {
+                "name":"app",
+                "scheme":"http",
+                "address":"127.0.0.1:5671",
+                "metrics_path":"/metrics",
+                "labels":{ "kuma_io_protocol":"http" }
+             }
+          ]
+       }
+    ],
+    "type_url":"type.googleapis.com/kuma.observability.v1.MonitoringAssignment"
+ }`),
+			},
+			want: []kumaTarget{
+				{
+					Mesh:        "default",
+					Service:     "redis",
+					DataPlane:   "redis",
+					Instance:    "redis",
+					Scheme:      "http",
+					Address:     "127.0.0.1:5670",
+					MetricsPath: "/metrics",
+					Labels:      map[string]string{"kuma_io_protocol": "tcp", "test": "test1"},
+				},
+				{
+					Mesh:        "default",
+					Service:     "app",
+					DataPlane:   "app",
+					Instance:    "app",
+					Scheme:      "http",
+					Address:     "127.0.0.1:5671",
+					MetricsPath: "/metrics",
+					Labels:      map[string]string{"kuma_io_protocol": "http", "test": "test2"},
+				},
+			},
+		},
+		{
+			name:    "parse err",
+			args:    args{data: []byte(`[]`)},
+			wantErr: true,
+		},
+		{
+			name: "api version err",
+			args: args{
+				data: []byte(`{
+    "resources":[
+       {
+          "@type":"type.googleapis.com/kuma.observability.v2.MonitoringAssignment",
+          "mesh":"default",
+          "service":"redis",
+          "targets":[]
+       }
+    ],
+    "type_url":"type.googleapis.com/kuma.observability.v2.MonitoringAssignment"
+ }`),
+			},
+			wantErr: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			resp, err := parseDiscoveryResponse(tt.args.data)
+			if tt.wantErr {
+				if err == nil {
+					t.Errorf("parseDiscoveryResponse() error = %v, wantErr %v", err, tt.wantErr)
+				}
+				return
+			}
+
+			got := parseKumaTargets(resp)
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("parseDiscoveryResponse() got = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
diff --git a/lib/promscrape/discovery/kuma/kuma_test.go b/lib/promscrape/discovery/kuma/kuma_test.go
new file mode 100644
index 000000000..683e4b545
--- /dev/null
+++ b/lib/promscrape/discovery/kuma/kuma_test.go
@@ -0,0 +1,79 @@
+package kuma
+
+import (
+	"testing"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutils"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promutils"
+)
+
+func Test_kumaTargetsToLabels(t *testing.T) {
+	type args struct {
+		src []kumaTarget
+	}
+	tests := []struct {
+		name string
+		args args
+		want []*promutils.Labels
+	}{
+		{
+			name: "convert to labels ok",
+			args: args{
+				src: []kumaTarget{
+					{
+						Mesh:        "default",
+						Service:     "redis",
+						DataPlane:   "redis",
+						Instance:    "redis",
+						Scheme:      "http",
+						Address:     "127.0.0.1:5670",
+						MetricsPath: "/metrics",
+						Labels:      map[string]string{"kuma_io_protocol": "tcp", "kuma_io_service": "redis"},
+					},
+					{
+						Mesh:        "default",
+						Service:     "app",
+						DataPlane:   "app",
+						Instance:    "app",
+						Scheme:      "http",
+						Address:     "127.0.0.1:5671",
+						MetricsPath: "/vm/metrics",
+						Labels:      map[string]string{"kuma_io_protocol": "http", "kuma_io_service": "app"},
+					},
+				},
+			},
+			want: []*promutils.Labels{
+				promutils.NewLabelsFromMap(map[string]string{
+					"instance":                           "redis",
+					"__address__":                        "127.0.0.1:5670",
+					"__scheme__":                         "http",
+					"__metrics_path__":                   "/metrics",
+					"__meta_server":                      "http://localhost:5676",
+					"__meta_kuma_mesh":                   "default",
+					"__meta_kuma_service":                "redis",
+					"__meta_kuma_dataplane":              "redis",
+					"__meta_kuma_label_kuma_io_protocol": "tcp",
+					"__meta_kuma_label_kuma_io_service":  "redis",
+				}),
+				promutils.NewLabelsFromMap(map[string]string{
+					"instance":                           "app",
+					"__address__":                        "127.0.0.1:5671",
+					"__scheme__":                         "http",
+					"__metrics_path__":                   "/vm/metrics",
+					"__meta_server":                      "http://localhost:5676",
+					"__meta_kuma_mesh":                   "default",
+					"__meta_kuma_service":                "app",
+					"__meta_kuma_dataplane":              "app",
+					"__meta_kuma_label_kuma_io_protocol": "http",
+					"__meta_kuma_label_kuma_io_service":  "app",
+				}),
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := kumaTargetsToLabels(tt.args.src, "http://localhost:5676")
+			discoveryutils.TestEqualLabelss(t, got, tt.want)
+		})
+	}
+}