Repository: VictoriaMetrics/VictoriaMetrics
Pull request #3044 (2022-08-30 15:53:03)

vmctl: implement support of chunking data for vm-native export
Add support of data chunking during VM-native exports(addresses #2733).

This PR adds a flag: `--vm-native-filter-chunk` which supports the following values: `month`, `day`, `hour`.
Flag requires `--vm-native-filter-time-start` to be set to calculate time ranges for export process.
When both flags are present time from start to end is splitted into timeframes and import is executed for each chunk.
Note that `month` chunk period enforces exporting only single month per export process by cutting time window to be exactly within month. 

Logs output for this is:
```
./vmctl vm-native 
    --vm-native-filter-time-start 2022-06-17T00:07:00Z \
    --vm-native-filter-time-end 2022-10-03T00:07:00Z \
    --vm-native-src-addr http://localhost:8428 \
    --vm-native-dst-addr http://localhost:8528 \
    --vm-native-filter-chunk=month
VictoriaMetrics Native import mode
2022/08/30 19:48:24 Processing range 1/5: 2022-06-17T00:07:00Z - 2022-06-30T23:59:59Z 
2022/08/30 19:48:24 Initing export pipe from "http://localhost:8428" with filters: 
        filter: match[]={__name__!=""}
        start: 2022-06-17T00:07:00Z
        end: 2022-06-30T23:59:59Z
Initing import process to "http://localhost:8428":
2022/08/30 19:48:24 Import finished!
Total: 16 B ↗ Speed: 28.89 KiB p/s 
2022/08/30 19:48:24 Processing range 2/5: 2022-07-01T00:00:00Z - 2022-07-31T23:59:59Z 
2022/08/30 19:48:24 Initing export pipe from "http://localhost:8428" with filters: 
        filter: match[]={__name__!=""}
        start: 2022-07-01T00:00:00Z
        end: 2022-07-31T23:59:59Z
Initing import process to "http://localhost:8428":
2022/08/30 19:48:24 Import finished!
Total: 16 B ↗ Speed: 164.35 KiB p/s 
2022/08/30 19:48:24 Processing range 3/5: 2022-08-01T00:00:00Z - 2022-08-31T23:59:59Z 
2022/08/30 19:48:24 Initing export pipe from "http://localhost:8428" with filters: 
        filter: match[]={__name__!=""}
        start: 2022-08-01T00:00:00Z
        end: 2022-08-31T23:59:59Z
Initing import process to "http://localhost:8428":
2022/08/30 19:48:24 Import finished!
Total: 16 B ↗ Speed: 191.42 KiB p/s 
2022/08/30 19:48:24 Processing range 4/5: 2022-09-01T00:00:00Z - 2022-09-30T23:59:59Z 
2022/08/30 19:48:24 Initing export pipe from "http://localhost:8428" with filters: 
        filter: match[]={__name__!=""}
        start: 2022-09-01T00:00:00Z
        end: 2022-09-30T23:59:59Z
Initing import process to "http://localhost:8428":
2022/08/30 19:48:24 Import finished!
Total: 16 B ↗ Speed: 141.04 KiB p/s 
2022/08/30 19:48:24 Processing range 5/5: 2022-10-01T00:00:00Z - 2022-10-03T00:07:00Z 
2022/08/30 19:48:24 Initing export pipe from "http://localhost:8428" with filters: 
        filter: match[]={__name__!=""}
        start: 2022-10-01T00:00:00Z
        end: 2022-10-03T00:07:00Z
Initing import process to "http://localhost:8428":
2022/08/30 19:48:24 Import finished!
Total: 16 B ↗ Speed: 186.32 KiB p/s 
2022/08/30 19:48:24 Total time: 12.680582ms

```
Note the time ranges:
```
2022-06-17T00:07:00Z - 2022-06-30T23:59:59Z  # Cut to end of current month to transfer data block only once
2022-07-01T00:00:00Z - 2022-07-31T23:59:59Z # Strictly one month
2022-08-01T00:00:00Z - 2022-08-31T23:59:59Z
2022-09-01T00:00:00Z - 2022-09-30T23:59:59Z 
2022-10-01T00:00:00Z - 2022-10-03T00:07:00Z # Last piece till the required filter `end`
```



codecov[bot]: # [Codecov](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3044?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) Report
> Merging [#3044](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3044?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) (8792711) into [master](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/commit/6d81584d2a3f6ff3ea49296808ec917eb864cd3d?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics) (6d81584) will **decrease** coverage by `0.03%`.
> The diff coverage is `46.46%`.

```diff
@@            Coverage Diff             @@
##           master    #3044      +/-   ##
==========================================
- Coverage   58.00%   57.97%   -0.04%     
==========================================
  Files         275      275              
  Lines       46510    46597      +87     
==========================================
+ Hits        26979    27014      +35     
- Misses      17760    17809      +49     
- Partials     1771     1774       +3     
```


| [Impacted Files](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3044?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/3044/diff?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/3044/diff?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\_native.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3044/diff?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.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3044/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3ZtY3RsL3V0aWxzLmdv) | `56.79% <100.00%> (+56.79%)` | :arrow_up: |
| [app/vmalert/remotewrite/remotewrite.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3044/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3ZtYWxlcnQvcmVtb3Rld3JpdGUvcmVtb3Rld3JpdGUuZ28=) | `64.42% <0.00%> (-2.02%)` | :arrow_down: |
| [lib/bloomfilter/filter.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3044/diff?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/vmalert/notifier/config\_watcher.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3044/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3ZtYWxlcnQvbm90aWZpZXIvY29uZmlnX3dhdGNoZXIuZ28=) | `64.64% <0.00%> (-1.66%)` | :arrow_down: |
| [app/vmselect/promql/aggr\_incremental.go](https://codecov.io/gh/VictoriaMetrics/VictoriaMetrics/pull/3044/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=VictoriaMetrics#diff-YXBwL3Ztc2VsZWN0L3Byb21xbC9hZ2dyX2luY3JlbWVudGFsLmdv) | `84.03% <0.00%> (-1.21%)` | :arrow_down: |

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)

hagen1778: Sorry, one more thing.
Could you please mention this flag and its usage in docs https://docs.victoriametrics.com/vmctl.html#migrating-data-from-victoriametrics (update only vmctl/README and then do `make sync-docs`)
and add a change description to the https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/docs/CHANGELOG.md
Thanks!
zekker6: > Sorry, one more thing. Could you please mention this new flag and its usage in docs https://docs.victoriametrics.com/vmctl.html#migrating-data-from-victoriametrics (update only vmctl/README and then do `make sync-docs`) and add a change description to the https://github.com/VictoriaMetrics/VictoriaMetrics/blob/master/docs/CHANGELOG.md Thanks!

@hagen1778 Added description and changelog entry, please have a look.

Patch:
diff --git a/app/vmctl/README.md b/app/vmctl/README.md
index cb5b23e8e..595546705 100644
--- a/app/vmctl/README.md
+++ b/app/vmctl/README.md
@@ -521,6 +521,71 @@ To avoid such situation try to filter out VM process metrics via `--vm-native-fi
 Instead, use [relabeling in VictoriaMetrics](https://github.com/VictoriaMetrics/vmctl/issues/4#issuecomment-683424375).
 5. When importing in or from cluster version remember to use correct [URL format](https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html#url-format)
 and specify `accountID` param.
+6. When migrating large volumes of data it might be useful to use `--vm-native-step-interval` flag to split single process into smaller steps.
+
+#### Using time-based chunking of migration
+
+It is possible split migration process into set of smaller batches based on time. This is especially useful when migrating large volumes of data as this adds indication of progress and ability to restore process from certain point in case of failure. 
+
+To use this you need to specify `--vm-native-step-interval` flag. Supported values are: `month`, `day`, `hour`.
+Note that in order to use this it is required `--vm-native-filter-time-start` to be set to calculate time ranges for export process.
+
+Every range is being processed independently, which means that:
+- after range processing is finished all data within range is migrated
+- if process fails on one of stages it is guaranteed that data of prior stages is already written, so it is possible to restart process starting from failed range
+
+Usage example:
+```console
+./vmctl vm-native 
+    --vm-native-filter-time-start 2022-06-17T00:07:00Z \
+    --vm-native-filter-time-end 2022-10-03T00:07:00Z \
+    --vm-native-src-addr http://localhost:8428 \
+    --vm-native-dst-addr http://localhost:8528 \
+    --vm-native-step-interval=month
+VictoriaMetrics Native import mode
+2022/08/30 19:48:24 Processing range 1/5: 2022-06-17T00:07:00Z - 2022-06-30T23:59:59Z 
+2022/08/30 19:48:24 Initing export pipe from "http://localhost:8428" with filters: 
+        filter: match[]={__name__!=""}
+        start: 2022-06-17T00:07:00Z
+        end: 2022-06-30T23:59:59Z
+Initing import process to "http://localhost:8428":
+2022/08/30 19:48:24 Import finished!
+Total: 16 B ↗ Speed: 28.89 KiB p/s 
+2022/08/30 19:48:24 Processing range 2/5: 2022-07-01T00:00:00Z - 2022-07-31T23:59:59Z 
+2022/08/30 19:48:24 Initing export pipe from "http://localhost:8428" with filters: 
+        filter: match[]={__name__!=""}
+        start: 2022-07-01T00:00:00Z
+        end: 2022-07-31T23:59:59Z
+Initing import process to "http://localhost:8428":
+2022/08/30 19:48:24 Import finished!
+Total: 16 B ↗ Speed: 164.35 KiB p/s 
+2022/08/30 19:48:24 Processing range 3/5: 2022-08-01T00:00:00Z - 2022-08-31T23:59:59Z 
+2022/08/30 19:48:24 Initing export pipe from "http://localhost:8428" with filters: 
+        filter: match[]={__name__!=""}
+        start: 2022-08-01T00:00:00Z
+        end: 2022-08-31T23:59:59Z
+Initing import process to "http://localhost:8428":
+2022/08/30 19:48:24 Import finished!
+Total: 16 B ↗ Speed: 191.42 KiB p/s 
+2022/08/30 19:48:24 Processing range 4/5: 2022-09-01T00:00:00Z - 2022-09-30T23:59:59Z 
+2022/08/30 19:48:24 Initing export pipe from "http://localhost:8428" with filters: 
+        filter: match[]={__name__!=""}
+        start: 2022-09-01T00:00:00Z
+        end: 2022-09-30T23:59:59Z
+Initing import process to "http://localhost:8428":
+2022/08/30 19:48:24 Import finished!
+Total: 16 B ↗ Speed: 141.04 KiB p/s 
+2022/08/30 19:48:24 Processing range 5/5: 2022-10-01T00:00:00Z - 2022-10-03T00:07:00Z 
+2022/08/30 19:48:24 Initing export pipe from "http://localhost:8428" with filters: 
+        filter: match[]={__name__!=""}
+        start: 2022-10-01T00:00:00Z
+        end: 2022-10-03T00:07:00Z
+Initing import process to "http://localhost:8428":
+2022/08/30 19:48:24 Import finished!
+Total: 16 B ↗ Speed: 186.32 KiB p/s 
+2022/08/30 19:48:24 Total time: 12.680582ms
+```
+
 
 ## Verifying exported blocks from VictoriaMetrics
 
diff --git a/app/vmctl/flags.go b/app/vmctl/flags.go
index 348d983b8..3a8dd0c4e 100644
--- a/app/vmctl/flags.go
+++ b/app/vmctl/flags.go
@@ -4,6 +4,8 @@ import (
 	"fmt"
 
 	"github.com/urfave/cli/v2"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/stepper"
 )
 
 const (
@@ -318,6 +320,7 @@ const (
 	vmNativeFilterMatch     = "vm-native-filter-match"
 	vmNativeFilterTimeStart = "vm-native-filter-time-start"
 	vmNativeFilterTimeEnd   = "vm-native-filter-time-end"
+	vmNativeStepInterval    = "vm-native-step-interval"
 
 	vmNativeSrcAddr     = "vm-native-src-addr"
 	vmNativeSrcUser     = "vm-native-src-user"
@@ -345,6 +348,10 @@ var (
 			Name:  vmNativeFilterTimeEnd,
 			Usage: "The time filter may contain either unix timestamp in seconds or RFC3339 values. E.g. '2020-01-01T20:07:00Z'",
 		},
+		&cli.StringFlag{
+			Name:  vmNativeStepInterval,
+			Usage: fmt.Sprintf("Split export data into chunks. Requires setting --%s. Valid values are '%s','%s','%s'.", vmNativeFilterTimeStart, stepper.StepMonth, stepper.StepDay, stepper.StepHour),
+		},
 		&cli.StringFlag{
 			Name: vmNativeSrcAddr,
 			Usage: "VictoriaMetrics address to perform export from. \n" +
diff --git a/app/vmctl/main.go b/app/vmctl/main.go
index 3c37fdd24..891ff0c78 100644
--- a/app/vmctl/main.go
+++ b/app/vmctl/main.go
@@ -11,6 +11,8 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/urfave/cli/v2"
+
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/influx"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/opentsdb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/prometheus"
@@ -18,7 +20,6 @@ import (
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/buildinfo"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/common"
 	parser "github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/native"
-	"github.com/urfave/cli/v2"
 )
 
 func main() {
@@ -161,6 +162,7 @@ func main() {
 							match:     c.String(vmNativeFilterMatch),
 							timeStart: c.String(vmNativeFilterTimeStart),
 							timeEnd:   c.String(vmNativeFilterTimeEnd),
+							chunk:     c.String(vmNativeStepInterval),
 						},
 						src: &vmNativeClient{
 							addr:     strings.Trim(c.String(vmNativeSrcAddr), "/"),
diff --git a/app/vmctl/stepper/split.go b/app/vmctl/stepper/split.go
new file mode 100644
index 000000000..15c176b09
--- /dev/null
+++ b/app/vmctl/stepper/split.go
@@ -0,0 +1,63 @@
+package stepper
+
+import (
+	"fmt"
+	"time"
+)
+
+const (
+	// StepMonth represents a one month interval
+	StepMonth string = "month"
+	// StepDay represents a one day interval
+	StepDay string = "day"
+	// StepHour represents a one hour interval
+	StepHour string = "hour"
+)
+
+// SplitDateRange splits start-end range in a subset of ranges respecting the given step
+// Ranges with granularity of StepMonth are aligned to 1st of each month in order to improve export efficiency at block transfer level
+func SplitDateRange(start, end time.Time, step string) ([][]time.Time, error) {
+
+	if start.After(end) {
+		return nil, fmt.Errorf("start time %q should come before end time %q", start.Format(time.RFC3339), end.Format(time.RFC3339))
+	}
+
+	var nextStep func(time.Time) (time.Time, time.Time)
+
+	switch step {
+	case StepMonth:
+		nextStep = func(t time.Time) (time.Time, time.Time) {
+			endOfMonth := time.Date(t.Year(), t.Month()+1, 1, 0, 0, 0, 0, t.Location()).Add(-1 * time.Nanosecond)
+			if t == endOfMonth {
+				endOfMonth = time.Date(t.Year(), t.Month()+2, 1, 0, 0, 0, 0, t.Location()).Add(-1 * time.Nanosecond)
+				t = time.Date(t.Year(), t.Month()+1, 1, 0, 0, 0, 0, t.Location())
+			}
+			return t, endOfMonth
+		}
+	case StepDay:
+		nextStep = func(t time.Time) (time.Time, time.Time) {
+			return t, t.AddDate(0, 0, 1)
+		}
+	case StepHour:
+		nextStep = func(t time.Time) (time.Time, time.Time) {
+			return t, t.Add(time.Hour * 1)
+		}
+	default:
+		return nil, fmt.Errorf("failed to parse step value, valid values are: '%s', '%s', '%s'. provided: '%s'", StepMonth, StepDay, StepHour, step)
+	}
+
+	currentStep := start
+
+	ranges := make([][]time.Time, 0)
+
+	for end.After(currentStep) {
+		s, e := nextStep(currentStep)
+		if e.After(end) {
+			e = end
+		}
+		ranges = append(ranges, []time.Time{s, e})
+		currentStep = e
+	}
+
+	return ranges, nil
+}
diff --git a/app/vmctl/vm_native.go b/app/vmctl/vm_native.go
index 85b0a9539..0f6becfaa 100644
--- a/app/vmctl/vm_native.go
+++ b/app/vmctl/vm_native.go
@@ -6,9 +6,12 @@ import (
 	"io"
 	"log"
 	"net/http"
+	"time"
+
+	"github.com/dmitryk-dk/pb/v3"
 
-	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/barpool"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/limiter"
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/stepper"
 	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/vm"
 )
 
@@ -31,6 +34,7 @@ type filter struct {
 	match     string
 	timeStart string
 	timeEnd   string
+	chunk     string
 }
 
 func (f filter) String() string {
@@ -52,10 +56,54 @@ const (
 )
 
 func (p *vmNativeProcessor) run(ctx context.Context) error {
+	if p.filter.chunk == "" {
+		return p.runSingle(ctx, p.filter)
+	}
+
+	startOfRange, err := time.Parse(time.RFC3339, p.filter.timeStart)
+	if err != nil {
+		return fmt.Errorf("failed to parse %s, provided: %s, expected format: %s, error: %v", vmNativeFilterTimeStart, p.filter.timeStart, time.RFC3339, err)
+	}
+
+	var endOfRange time.Time
+	if p.filter.timeEnd != "" {
+		endOfRange, err = time.Parse(time.RFC3339, p.filter.timeEnd)
+		if err != nil {
+			return fmt.Errorf("failed to parse %s, provided: %s, expected format: %s, error: %v", vmNativeFilterTimeEnd, p.filter.timeEnd, time.RFC3339, err)
+		}
+	} else {
+		endOfRange = time.Now()
+	}
+
+	ranges, err := stepper.SplitDateRange(startOfRange, endOfRange, p.filter.chunk)
+	if err != nil {
+		return fmt.Errorf("failed to create date ranges for the given time filters: %v", err)
+	}
+
+	for rangeIdx, r := range ranges {
+		formattedStartTime := r[0].Format(time.RFC3339)
+		formattedEndTime := r[1].Format(time.RFC3339)
+		log.Printf("Processing range %d/%d: %s - %s \n", rangeIdx+1, len(ranges), formattedStartTime, formattedEndTime)
+		f := filter{
+			match:     p.filter.match,
+			timeStart: formattedStartTime,
+			timeEnd:   formattedEndTime,
+		}
+		err := p.runSingle(ctx, f)
+
+		if err != nil {
+			log.Printf("processing failed for range %d/%d: %s - %s \n", rangeIdx+1, len(ranges), formattedStartTime, formattedEndTime)
+			return err
+		}
+	}
+	return nil
+}
+
+func (p *vmNativeProcessor) runSingle(ctx context.Context, f filter) error {
 	pr, pw := io.Pipe()
 
-	fmt.Printf("Initing export pipe from %q with filters: %s\n", p.src.addr, p.filter)
-	exportReader, err := p.exportPipe(ctx)
+	log.Printf("Initing export pipe from %q with filters: %s\n", p.src.addr, f)
+	exportReader, err := p.exportPipe(ctx, f)
 	if err != nil {
 		return fmt.Errorf("failed to init export pipe: %s", err)
 	}
@@ -83,13 +131,20 @@ func (p *vmNativeProcessor) run(ctx context.Context) error {
 	}()
 
 	fmt.Printf("Initing import process to %q:\n", p.dst.addr)
-	bar := barpool.AddWithTemplate(nativeBarTpl, 0)
+	pool := pb.NewPool()
+	bar := pb.ProgressBarTemplate(nativeBarTpl).New(0)
+	pool.Add(bar)
 	barReader := bar.NewProxyReader(exportReader)
-	if err := barpool.Start(); err != nil {
+	if err := pool.Start(); err != nil {
 		log.Printf("error start process bars pool: %s", err)
 		return err
 	}
-	defer barpool.Stop()
+	defer func() {
+		bar.Finish()
+		if err := pool.Stop(); err != nil {
+			fmt.Printf("failed to stop barpool: %+v\n", err)
+		}
+	}()
 
 	w := io.Writer(pw)
 	if p.rateLimit > 0 {
@@ -111,7 +166,7 @@ func (p *vmNativeProcessor) run(ctx context.Context) error {
 	return nil
 }
 
-func (p *vmNativeProcessor) exportPipe(ctx context.Context) (io.ReadCloser, error) {
+func (p *vmNativeProcessor) exportPipe(ctx context.Context, f filter) (io.ReadCloser, error) {
 	u := fmt.Sprintf("%s/%s", p.src.addr, nativeExportAddr)
 	req, err := http.NewRequestWithContext(ctx, "GET", u, nil)
 	if err != nil {
@@ -119,12 +174,12 @@ func (p *vmNativeProcessor) exportPipe(ctx context.Context) (io.ReadCloser, erro
 	}
 
 	params := req.URL.Query()
-	params.Set("match[]", p.filter.match)
-	if p.filter.timeStart != "" {
-		params.Set("start", p.filter.timeStart)
+	params.Set("match[]", f.match)
+	if f.timeStart != "" {
+		params.Set("start", f.timeStart)
 	}
-	if p.filter.timeEnd != "" {
-		params.Set("end", p.filter.timeEnd)
+	if f.timeEnd != "" {
+		params.Set("end", f.timeEnd)
 	}
 	req.URL.RawQuery = params.Encode()
 
diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index e9225ebf7..606f74b19 100644
--- a/docs/CHANGELOG.md
+++ b/docs/CHANGELOG.md
@@ -17,6 +17,8 @@ The following tip changes can be tested by building VictoriaMetrics components f
 
 * FEATURE: set the `start` arg to `end - 5 minutes` if isn't passed explicitly to [/api/v1/labels](https://docs.victoriametrics.com/url-examples.html#apiv1labels) and [/api/v1/label/.../values](https://docs.victoriametrics.com/url-examples.html#apiv1labelvalues).
 
+* FEATURE: [vmctl](https://docs.victoriametrics.com/vmctl.html): add `vm-native-step-interval` command line flag for `vm-native` mode. New option allows splitting the import process into chunks by time interval. This helps migrating data sets with high churn rate and provides better control over the process. See [feature request](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/2733).
+
 * BUGFIX: [MetricsQL](https://docs.victoriametrics.com/MetricsQL.html): properly calculate `rate_over_sum(m[d])` as `sum_over_time(m[d])/d`. Previously the `sum_over_time(m[d])` could be improperly divided by smaller than `d` time range. See [rate_over_sum() docs](https://docs.victoriametrics.com/MetricsQL.html#rate_over_sum) and [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3045).
 
 
diff --git a/docs/vmctl.md b/docs/vmctl.md
index fa7e71e12..7f9ea647b 100644
--- a/docs/vmctl.md
+++ b/docs/vmctl.md
@@ -525,6 +525,71 @@ To avoid such situation try to filter out VM process metrics via `--vm-native-fi
 Instead, use [relabeling in VictoriaMetrics](https://github.com/VictoriaMetrics/vmctl/issues/4#issuecomment-683424375).
 5. When importing in or from cluster version remember to use correct [URL format](https://docs.victoriametrics.com/Cluster-VictoriaMetrics.html#url-format)
 and specify `accountID` param.
+6. When migrating large volumes of data it might be useful to use `--vm-native-step-interval` flag to split single process into smaller steps.
+
+#### Using time-based chunking of migration
+
+It is possible to split migration process by time intervals. This is especially useful when migrating large volumes of data with high churn rate, adds better indication of a progress and ability to restore process from certain point of time in case of failure. 
+
+To split migration process by time, specify `--vm-native-step-interval` flag to one of the supported values: `month`, `day` or `hour`.
+Please note, command-line flag `--vm-native-filter-time-start` is required when using `--vm-native-step-interval`.
+
+Every time range (step) is processed independently, which means the following:
+- if specific time range (step) is finished, then all the data within the range has been successfully migrated;
+- if process fails on one of steps, then it is recommend to restart the process from the failed step. As all previous steps were successfully completed.
+
+Usage example:
+```console
+./vmctl vm-native 
+    --vm-native-filter-time-start 2022-06-17T00:07:00Z \
+    --vm-native-filter-time-end 2022-10-03T00:07:00Z \
+    --vm-native-src-addr http://localhost:8428 \
+    --vm-native-dst-addr http://localhost:8528 \
+    --vm-native-step-interval=month
+VictoriaMetrics Native import mode
+2022/08/30 19:48:24 Processing range 1/5: 2022-06-17T00:07:00Z - 2022-06-30T23:59:59Z 
+2022/08/30 19:48:24 Initing export pipe from "http://localhost:8428" with filters: 
+        filter: match[]={__name__!=""}
+        start: 2022-06-17T00:07:00Z
+        end: 2022-06-30T23:59:59Z
+Initing import process to "http://localhost:8428":
+2022/08/30 19:48:24 Import finished!
+Total: 16 B ↗ Speed: 28.89 KiB p/s 
+2022/08/30 19:48:24 Processing range 2/5: 2022-07-01T00:00:00Z - 2022-07-31T23:59:59Z 
+2022/08/30 19:48:24 Initing export pipe from "http://localhost:8428" with filters: 
+        filter: match[]={__name__!=""}
+        start: 2022-07-01T00:00:00Z
+        end: 2022-07-31T23:59:59Z
+Initing import process to "http://localhost:8428":
+2022/08/30 19:48:24 Import finished!
+Total: 16 B ↗ Speed: 164.35 KiB p/s 
+2022/08/30 19:48:24 Processing range 3/5: 2022-08-01T00:00:00Z - 2022-08-31T23:59:59Z 
+2022/08/30 19:48:24 Initing export pipe from "http://localhost:8428" with filters: 
+        filter: match[]={__name__!=""}
+        start: 2022-08-01T00:00:00Z
+        end: 2022-08-31T23:59:59Z
+Initing import process to "http://localhost:8428":
+2022/08/30 19:48:24 Import finished!
+Total: 16 B ↗ Speed: 191.42 KiB p/s 
+2022/08/30 19:48:24 Processing range 4/5: 2022-09-01T00:00:00Z - 2022-09-30T23:59:59Z 
+2022/08/30 19:48:24 Initing export pipe from "http://localhost:8428" with filters: 
+        filter: match[]={__name__!=""}
+        start: 2022-09-01T00:00:00Z
+        end: 2022-09-30T23:59:59Z
+Initing import process to "http://localhost:8428":
+2022/08/30 19:48:24 Import finished!
+Total: 16 B ↗ Speed: 141.04 KiB p/s 
+2022/08/30 19:48:24 Processing range 5/5: 2022-10-01T00:00:00Z - 2022-10-03T00:07:00Z 
+2022/08/30 19:48:24 Initing export pipe from "http://localhost:8428" with filters: 
+        filter: match[]={__name__!=""}
+        start: 2022-10-01T00:00:00Z
+        end: 2022-10-03T00:07:00Z
+Initing import process to "http://localhost:8428":
+2022/08/30 19:48:24 Import finished!
+Total: 16 B ↗ Speed: 186.32 KiB p/s 
+2022/08/30 19:48:24 Total time: 12.680582ms
+```
+
 
 ## Verifying exported blocks from VictoriaMetrics
 


Test patch:
diff --git a/app/vmctl/stepper/split_test.go b/app/vmctl/stepper/split_test.go
new file mode 100644
index 000000000..dec06b18f
--- /dev/null
+++ b/app/vmctl/stepper/split_test.go
@@ -0,0 +1,152 @@
+package stepper
+
+import (
+	"reflect"
+	"testing"
+	"time"
+)
+
+type testTimeRange []string
+
+func mustParseDatetime(t string) time.Time {
+	result, err := time.Parse(time.RFC3339, t)
+	if err != nil {
+		panic(err)
+	}
+	return result
+}
+
+func Test_splitDateRange(t *testing.T) {
+	type args struct {
+		start       string
+		end         string
+		granularity string
+	}
+	tests := []struct {
+		name    string
+		args    args
+		want    []testTimeRange
+		wantErr bool
+	}{
+		{
+			name: "validates start is before end",
+			args: args{
+				start:       "2022-02-01T00:00:00Z",
+				end:         "2022-01-01T00:00:00Z",
+				granularity: StepMonth,
+			},
+			want:    nil,
+			wantErr: true,
+		},
+		{
+			name: "validates granularity value",
+			args: args{
+				start:       "2022-01-01T00:00:00Z",
+				end:         "2022-02-01T00:00:00Z",
+				granularity: "non-existent-format",
+			},
+			want:    nil,
+			wantErr: true,
+		},
+		{
+			name: "month chunking",
+			args: args{
+				start:       "2022-01-03T11:11:11Z",
+				end:         "2022-03-03T12:12:12Z",
+				granularity: StepMonth,
+			},
+			want: []testTimeRange{
+				{
+					"2022-01-03T11:11:11Z",
+					"2022-01-31T23:59:59.999999999Z",
+				},
+				{
+					"2022-02-01T00:00:00Z",
+					"2022-02-28T23:59:59.999999999Z",
+				},
+				{
+					"2022-03-01T00:00:00Z",
+					"2022-03-03T12:12:12Z",
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "daily chunking",
+			args: args{
+				start:       "2022-01-03T11:11:11Z",
+				end:         "2022-01-05T12:12:12Z",
+				granularity: StepDay,
+			},
+			want: []testTimeRange{
+				{
+					"2022-01-03T11:11:11Z",
+					"2022-01-04T11:11:11Z",
+				},
+				{
+					"2022-01-04T11:11:11Z",
+					"2022-01-05T11:11:11Z",
+				},
+				{
+					"2022-01-05T11:11:11Z",
+					"2022-01-05T12:12:12Z",
+				},
+			},
+			wantErr: false,
+		},
+		{
+			name: "hourly chunking",
+			args: args{
+				start:       "2022-01-03T11:11:11Z",
+				end:         "2022-01-03T14:14:14Z",
+				granularity: StepHour,
+			},
+			want: []testTimeRange{
+				{
+					"2022-01-03T11:11:11Z",
+					"2022-01-03T12:11:11Z",
+				},
+				{
+					"2022-01-03T12:11:11Z",
+					"2022-01-03T13:11:11Z",
+				},
+				{
+					"2022-01-03T13:11:11Z",
+					"2022-01-03T14:11:11Z",
+				},
+				{
+					"2022-01-03T14:11:11Z",
+					"2022-01-03T14:14:14Z",
+				},
+			},
+			wantErr: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			start := mustParseDatetime(tt.args.start)
+			end := mustParseDatetime(tt.args.end)
+
+			got, err := SplitDateRange(start, end, tt.args.granularity)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("splitDateRange() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+
+			var testExpectedResults [][]time.Time
+			if tt.want != nil {
+				testExpectedResults = make([][]time.Time, 0)
+				for _, dr := range tt.want {
+					testExpectedResults = append(testExpectedResults, []time.Time{
+						mustParseDatetime(dr[0]),
+						mustParseDatetime(dr[1]),
+					})
+				}
+			}
+
+			if !reflect.DeepEqual(got, testExpectedResults) {
+				t.Errorf("splitDateRange() got = %v, want %v", got, testExpectedResults)
+			}
+		})
+	}
+}
diff --git a/app/vmctl/vm_native_test.go b/app/vmctl/vm_native_test.go
index 3e229bcbc..b254dc33c 100644
--- a/app/vmctl/vm_native_test.go
+++ b/app/vmctl/vm_native_test.go
@@ -4,6 +4,8 @@ import (
 	"context"
 	"testing"
 	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/stepper"
 )
 
 // If you want to run this test:
@@ -16,6 +18,7 @@ import (
 const (
 	matchFilter     = `{job="avalanche"}`
 	timeStartFilter = "2020-01-01T20:07:00Z"
+	timeEndFilter   = "2020-08-01T20:07:00Z"
 	srcAddr         = "http://127.0.0.1:8428"
 	dstAddr         = "http://127.0.0.1:8528"
 )
@@ -74,6 +77,26 @@ func Test_vmNativeProcessor_run(t *testing.T) {
 			closer:  func(cancelFunc context.CancelFunc) {},
 			wantErr: false,
 		},
+		{
+			name: "simulate correct work with chunking",
+			fields: fields{
+				filter: filter{
+					match:     matchFilter,
+					timeStart: timeStartFilter,
+					timeEnd:   timeEndFilter,
+					chunk:     stepper.StepMonth,
+				},
+				rateLimit: 0,
+				dst: &vmNativeClient{
+					addr: dstAddr,
+				},
+				src: &vmNativeClient{
+					addr: srcAddr,
+				},
+			},
+			closer:  func(cancelFunc context.CancelFunc) {},
+			wantErr: false,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {