-
Notifications
You must be signed in to change notification settings - Fork 0
/
migrate.go
80 lines (67 loc) · 2.78 KB
/
migrate.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
package agent
import (
"context"
"fmt"
"sync"
"time"
"github.com/appit-online/redis-dumper/pkg/core/dumper"
"github.com/appit-online/redis-dumper/pkg/core/logger"
"github.com/appit-online/redis-dumper/pkg/core/restore"
"github.com/appit-online/redis-dumper/pkg/core/scanner"
"github.com/spf13/cobra"
)
var parallelDumps, parallelRestores int
var migrateCmd = &cobra.Command{
Use: "migrate <source> <destination>",
Short: "Migrate keys from source redis instance to destination by given pattern",
Long: `Migrate keys from source redis instance to destination by given pattern <source> and <destination>
Can be provided as just ""<host>:<port>" or in Redis URL format: "redis://[:<password>@]<host>:<port>[/<dbIndex>]"`,
Args: cobra.MinimumNArgs(2),
Run: func(cmd *cobra.Command, args []string) {
fmt.Println("Start migration")
ctx := context.Background()
// create redis clients
scannerClient := createClient(ctx, sourcePassword, args[0], sourceUseTLS)
dumperClient := createClient(ctx, sourcePassword, args[0], sourceUseTLS)
restorerClient := createClient(ctx, targetPassword, args[1], targetUseTLS)
// init core services
logger := logger.CreateService()
scanner := scanner.CreateService(
scannerClient,
scanner.Options{
SearchPattern: pattern,
RedisScanCount: scanCount,
},
logger,
)
dumper := dumper.CreateService(
dumperClient,
scanner.GetScanChannel(),
logger,
)
restorer := restore.CreateService(restorerClient, dumper.GetDumpChannel(), logger)
// start processing
wgRestore := new(sync.WaitGroup)
logger.Start(time.Second * time.Duration(logInterval))
restorer.Start(ctx, wgRestore, parallelRestores)
scanner.Start(ctx)
dumper.Start(ctx, parallelDumps)
// wait until all channels are closed
wgRestore.Wait()
logger.Stop()
logger.Report()
fmt.Println("Finish migration")
},
}
func init() {
RootCmd.AddCommand(migrateCmd)
migrateCmd.Flags().StringVar(&pattern, "pattern", "*", "Matching pattern for keys")
migrateCmd.Flags().StringVar(&sourcePassword, "sourcePassword", "", "Password of source redis")
migrateCmd.Flags().BoolVar(&sourceUseTLS, "sourceUseTLS", true, "Enable TLS for source redis- default true")
migrateCmd.Flags().StringVar(&targetPassword, "targetPassword", "", "Password of target redis")
migrateCmd.Flags().BoolVar(&targetUseTLS, "targetUseTLS", true, "Enable TLS for target redis- default true")
migrateCmd.Flags().IntVar(&scanCount, "scanCount", 1000, "COUNT parameter for redis SCAN command")
migrateCmd.Flags().IntVar(&logInterval, "logInterval", 1, "Print current status every N seconds")
migrateCmd.Flags().IntVar(¶llelDumps, "parallelDumps", 100, "Number of parallel dump goroutines")
migrateCmd.Flags().IntVar(¶llelRestores, "parallelRestores", 100, "Number of parallel restore goroutines")
}