chore: support migrate data to other database

Signed-off-by: xiangyu5632 <[email protected]>

README.md

@@ -54,7 +54,7 @@ drwx------  4 root  root   128B 12  8 09:01 db1
 We migrate `internal` db
 
 ```bash
-> ./dataMigrate run --from /var/lib/influxdb/data --to ip:port --database _internal
+> ./dataMigrate run --from /var/lib/influxdb/data --to ip:port --src_database _internal --dest_database _internal
 
 2023/12/08 14:17:48 Data migrate tool starting
 2023/12/08 14:17:48 Debug mode is enabled
@@ -77,7 +77,7 @@ We migrate `internal` db
 ### example 2: Migrate the specified database
 
 ```bash
-> ./dataMigrate run --from /var/lib/influxdb/data --to ip:port --database db0
+> ./dataMigrate run --from /var/lib/influxdb/data --to ip:port --src_database db0 --dest_database db3
 
 2023/12/08 14:31:47 Data migrate tool starting
 2023/12/08 14:31:47 Debug mode is enabled
@@ -91,7 +91,7 @@ We migrate `internal` db
 ### example 3: Migrate the specified database with auth and https
 
 ```bash
-> ./dataMigrate run --from /var/lib/influxdb/data --to ip:port --database db0 \
+> ./dataMigrate run --from /var/lib/influxdb/data --to ip:port --src_database db0 --dest_database db0\
     --ssl --unsafeSsl --username rwusr --password This@123
 
 2023/12/08 14:31:47 Data migrate tool starting
@@ -113,19 +113,20 @@ Usage:
   run [flags]
 
 Flags:
-      --batch int          Optional: specify batch size for inserting lines (default 1000)
-      --database string    Optional: the database to read
-      --debug              Optional: whether to enable debug log or not
-      --end string         Optional: the end time to read (RFC3339 format)
-  -f, --from string        Influxdb Data storage path. See your influxdb config item: data.dir (default "/var/lib/influxdb/data")
-  -h, --help               help for run
-  -p, --password string    Optional: The password to connect to the openGemini cluster.
-      --retention string   Optional: the retention policy to read (required -database)
-      --ssl                Optional: Use https for requests.
-      --start string       Optional: the start time to read (RFC3339 format)
-  -t, --to string          Destination host to write data to (default "127.0.0.1:8086")
-      --unsafeSsl          Optional: Set this when connecting to the cluster using https and not use SSL verification.
-  -u, --username string    Optional: The username to connect to the openGemini cluster.
+      --batch int             Optional: specify batch size for inserting lines (default 1000)
+      --src_database string   Optional: the database to read
+      --dest_database string  Optional: the destination database to write
+      --debug                 Optional: whether to enable debug log or not
+      --end string            Optional: the end time to read (RFC3339 format)
+  -f, --from string           Influxdb Data storage path. See your influxdb config item: data.dir (default "/var/lib/influxdb/data")
+  -h, --help                  help for run
+  -p, --password string       Optional: The password to connect to the openGemini cluster.
+      --retention string      Optional: the retention policy to read (required -database)
+      --ssl                   Optional: Use https for requests.
+      --start string          Optional: the start time to read (RFC3339 format)
+  -t, --to string             Destination host to write data to (default "127.0.0.1:8086")
+      --unsafeSsl             Optional: Set this when connecting to the cluster using https and not use SSL verification.
+  -u, --username string       Optional: The username to connect to the openGemini cluster.
 ```
 
 **Welcome to add more features.**

cmd/root.go

@@ -29,7 +29,8 @@ func init() {
 	RootCmd.Flags().StringVarP(&opt.Password, "password", "p", "", "Optional: The password to connect to the openGemini cluster.")
 	RootCmd.Flags().StringVarP(&opt.DataDir, "from", "f", "/var/lib/influxdb/data", "Influxdb Data storage path. See your influxdb config item: data.dir")
 	RootCmd.Flags().StringVarP(&opt.Out, "to", "t", "127.0.0.1:8086", "Destination host to write data to")
-	RootCmd.Flags().StringVarP(&opt.Database, "database", "", "", "Optional: the database to read")
+	RootCmd.Flags().StringVarP(&opt.SrcDatabase, "src_database", "", "", "Optional: the database to read")
+	RootCmd.Flags().StringVarP(&opt.DestDatabase, "dest_database", "", "", "Optional: the database to write")
 	RootCmd.Flags().StringVarP(&opt.RetentionPolicy, "retention", "", "", "Optional: the retention policy to read (required -database)")
 	RootCmd.Flags().StringVarP(&opt.Start, "start", "", "", "Optional: the start time to read (RFC3339 format)")
 	RootCmd.Flags().StringVarP(&opt.End, "end", "", "", "Optional: the end time to read (RFC3339 format)")

src/dataMigrate.go

@@ -137,14 +137,15 @@ func (cmd *DataMigrateCommand) Run() error {
 	// write params to log
 	logger.LogString("Got param \"from\": "+cmd.opt.DataDir, TOLOGFILE, LEVEL_INFO)
 	logger.LogString("Got param \"to\": "+cmd.opt.Out, TOLOGFILE, LEVEL_INFO)
-	logger.LogString("Got param \"database\": "+cmd.opt.Database, TOLOGFILE, LEVEL_INFO)
+	logger.LogString("Got param \"src_database\": "+cmd.opt.SrcDatabase, TOLOGFILE, LEVEL_INFO)
+	logger.LogString("Got param \"dest_database\": "+cmd.opt.DestDatabase, TOLOGFILE, LEVEL_INFO)
 	logger.LogString("Got param \"retention\": "+cmd.opt.RetentionPolicy, TOLOGFILE, LEVEL_INFO)
 	logger.LogString("Got param \"start\": "+cmd.opt.Start, TOLOGFILE, LEVEL_INFO)
 	logger.LogString("Got param \"end\": "+cmd.opt.End, TOLOGFILE, LEVEL_INFO)
 	logger.LogString("Got param \"batch\": "+strconv.Itoa(cmd.opt.BatchSize), TOLOGFILE, LEVEL_INFO)
 
 	gs := NewGeminiService(cmd)
-	shardGroupDuration, err := gs.GetShardGroupDuration(cmd.opt.Database, "autogen")
+	shardGroupDuration, err := gs.GetShardGroupDuration(cmd.opt.DestDatabase, "autogen")
 	if err != nil {
 		return err
 	}
@@ -172,7 +173,7 @@ func (cmd *DataMigrateCommand) setOutput(url string) {
 
 // Check whether the parameters are valid or not.
 func (cmd *DataMigrateCommand) validate() error {
-	if cmd.opt.RetentionPolicy != "" && cmd.opt.Database == "" {
+	if cmd.opt.RetentionPolicy != "" && cmd.opt.SrcDatabase == "" && cmd.opt.DestDatabase == "" {
 		return fmt.Errorf("dataMigrate: must specify a db")
 	}
 	if cmd.opt.StartTime != 0 && cmd.opt.EndTime != 0 && cmd.opt.EndTime < cmd.opt.StartTime {
@@ -227,7 +228,7 @@ func (cmd *DataMigrateCommand) walkTSMFiles() error {
 			return fmt.Errorf("invalid directory structure for %s", path)
 		}
 
-		if (dirs[0] == cmd.opt.Database || cmd.opt.Database == "") &&
+		if (dirs[0] == cmd.opt.SrcDatabase || cmd.opt.SrcDatabase == "") &&
 			(dirs[1] == cmd.opt.RetentionPolicy || cmd.opt.RetentionPolicy == "") {
 			key := filepath.Join(dirs[0], dirs[1], dirs[2])
 			cmd.tsmFiles[key] = append(cmd.tsmFiles[key], path)

src/migrator.go

@@ -107,7 +107,7 @@ func (m *migrator) getStat() *statInfo {
 func NewMigrator(cmd *DataMigrateCommand, info *shardGroupInfo) *migrator {
 	mig := &migrator{
 		out:             cmd.opt.Out,
-		database:        info.db,
+		database:        cmd.opt.DestDatabase,
 		retentionPolicy: info.rp,
 		startTime:       cmd.opt.StartTime,
 		endTime:         cmd.opt.EndTime,

src/options.go

@@ -5,7 +5,8 @@ type DataMigrateOptions struct {
 	Out             string
 	Username        string
 	Password        string
-	Database        string
+	SrcDatabase     string
+	DestDatabase    string
 	RetentionPolicy string
 	Start           string // rfc3339 format
 	End             string // rfc3339 format