Repository: VictoriaMetrics/VictoriaMetrics
Pull request #3844 (2023-02-20 18:07:08)

app/vmctl: add retry backoff policy
I moved the retry backoff policy to utils and used it in importer for different processors and native import process.
When we try to migrate data from OpenTSDB, InfluxDB, Remote read protocol, etc., the retries will start if we lose connection to the storage where we transfer data.

Remote read protocol migration process with retries
<img width="1652" alt="217252234-b192976d-b4c9-4d68-ae2c-bcd62db1c00d" src="https://user-images.githubusercontent.com/12189570/220174942-493ad0ec-962e-4896-8897-29fb5c1fdbbe.png">
<img width="1669" alt="217252264-85827f70-61f4-4160-b2dd-272ef0123a65" src="https://user-images.githubusercontent.com/12189570/220174964-7f00cbf2-3008-4593-b255-ce832e62da60.png">

Related issue: https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3600

Signed-off-by: dmitryk-dk [d.kozlov@victoriametrics.com](mailto:d.kozlov@victoriametrics.com)

codecov[bot]: # [Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3844?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: **57.92**% // Decreases project coverage by **`-0.12%`** :warning:
> Coverage data is based on head [(`9d3c20f`)](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3844?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) compared to base [(`2cea923`)](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/commit/2cea923ff742f50d01a84fdbcf9be5737c8bcd8e?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics).
> Patch coverage: 9.87% of modified lines in pull request are covered.

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


```diff
@@            Coverage Diff             @@
##           master    #3844      +/-   ##
==========================================
- Coverage   58.04%   57.92%   -0.12%     
==========================================
  Files         309      311       +2     
  Lines       52234    52354     +120     
==========================================
+ Hits        30319    30327       +8     
- Misses      20011    20122     +111     
- Partials     1904     1905       +1     
```


| [Impacted Files](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3844?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) | Coverage Δ | |
|---|---|---|
| [app/vmctl/flags.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3844?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3ZtY3RsL2ZsYWdzLmdv) | `0.00% <ø> (ø)` | |
| [app/vmctl/main.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3844?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3ZtY3RsL21haW4uZ28=) | `0.00% <0.00%> (ø)` | |
| [app/vmctl/vm/vm.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3844?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3ZtY3RsL3ZtL3ZtLmdv) | `4.56% <0.00%> (+0.05%)` | :arrow_up: |
| [app/vmctl/vm\_native.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3844?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3ZtY3RsL3ZtX25hdGl2ZS5nbw==) | `0.00% <0.00%> (ø)` | |
| [app/vmctl/utils/backoff.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3844?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3ZtY3RsL3V0aWxzL2JhY2tvZmYuZ28=) | `57.14% <57.14%> (ø)` | |
| [app/vmalert/remotewrite/remotewrite.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3844?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3ZtYWxlcnQvcmVtb3Rld3JpdGUvcmVtb3Rld3JpdGUuZ28=) | `62.32% <0.00%> (-3.43%)` | :arrow_down: |
| [lib/fs/reader\_at.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3844?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-bGliL2ZzL3JlYWRlcl9hdC5nbw==) | `41.25% <0.00%> (-2.50%)` | :arrow_down: |
| [lib/bloomfilter/filter.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3844?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-bGliL2Jsb29tZmlsdGVyL2ZpbHRlci5nbw==) | `98.00% <0.00%> (-2.00%)` | :arrow_down: |
| [app/vmselect/promql/aggr\_incremental.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3844?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3Ztc2VsZWN0L3Byb21xbC9hZ2dyX2luY3JlbWVudGFsLmdv) | `86.52% <0.00%> (-1.20%)` | :arrow_down: |
| [lib/mergeset/table.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3844?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: |
| ... and [5 more](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3844?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) | |

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/3844?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..4bf1b06e1 100644
--- a/README.md
+++ b/README.md
@@ -2263,11 +2263,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/app/vmagent/README.md b/app/vmagent/README.md
index 2ab14f218..636cf2120 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
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/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..e5cbd28cf 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -2264,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
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/vmagent.md b/docs/vmagent.md
index 218436ee6..4bf3334b0 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


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)