github-mirrors/VictoriaMetrics
1
0
Fork 0
mirror of https://github.com/VictoriaMetrics/VictoriaMetrics.git synced 2025-01-10 15:14:09 +00:00
Code Issues Projects Releases Packages Wiki Activity

vmctl: implement support of chunking data for vm-native export (#3044)

Browse source 
vmctl: implement support of chunking data for vm-native export process

See https://github.com/VictoriaMetrics/VictoriaMetrics/issues/2733
This commit is contained in:
Zakhar Bessarab 2022-09-06 10:09:34 +03:00 committed by Aliaksandr Valialkin
parent d02f09700f
commit ef01f01b6c
No known key found for this signature in database
GPG key ID: A72BEC6CD3D0DED1
9 changed files with 447 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

65
app/vmctl/README.md
View file

@ -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

7
app/vmctl/flags.go
View file

@ -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" +

4
app/vmctl/main.go
View file

@ -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), "/"),

63
app/vmctl/stepper/split.go Normal file
View file

@ -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
}

152
app/vmctl/stepper/split_test.go Normal file
View file

@ -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)
}
})
}
}

79
app/vmctl/vm_native.go
View file

@ -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()

23
app/vmctl/vm_native_test.go
View file

@ -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) {

2
docs/CHANGELOG.md
View file

@ -21,6 +21,8 @@ The following tip changes can be tested by building VictoriaMetrics components f
* FEATURE: check the correctess of raw sample timestamps stored on disk when reading them. This reduces the probability of possible silent corruption of the data stored on disk. This should help [this](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/2998) and [this issue](https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3011).
* 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). See [this pull request](https://github.com/VictoriaMetrics/VictoriaMetrics/pull/3052).
* 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).

65
docs/vmctl.md
View file

@ -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