chore: support migrate data to other database (#18)

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

README.md

@@ -51,10 +51,10 @@ drwx------  4 root  root   128B 12  6 14:59 db0
 drwx------  4 root  root   128B 12  8 09:01 db1
 ```
 
-We migrate `internal` db
+We migrate all database
 
 ```bash
-> ./dataMigrate run --from /var/lib/influxdb/data --to ip:port --database _internal
+> ./dataMigrate run --from /var/lib/influxdb/data --to ip:port
 
 2023/12/08 14:17:48 Data migrate tool starting
 2023/12/08 14:17:48 Debug mode is enabled
@@ -103,6 +103,21 @@ We migrate `internal` db
 2023/12/08 14:31:47 Total: takes 48.502792ms to migrate, with 1 tags, 2 fields, 2 rows read.
 ```
 
+### example 4: Migrate the specified database and destDatabase
+
+```
+> ./dataMigrate run --from /var/lib/influxdb/data --to ip:port --database db0 --destDatabase db3
+
+2023/12/08 14:31:47 Data migrate tool starting
+2023/12/08 14:31:47 Debug mode is enabled
+2023/12/08 14:31:47 Searching for tsm files to migrate
+2023/12/08 14:31:47 Writing out data from shard db0/autogen/2, [1/1]...
+2023/12/08 14:31:47 Dealing file: /Users/shilinlee/.influxdb/data/db0/autogen/2/000000001-000000001.tsm
+2023/12/08 14:31:47 Shard db0/autogen/2 takes 45.883209ms to migrate, with 1 tags, 2 fields, 2 rows read
+2023/12/08 14:31:47 Total: takes 48.502792ms to migrate, with 1 tags, 2 fields, 2 rows read.
+```
+
+
 
 ## For more help
 
@@ -113,19 +128,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)
+      --database string       Optional: The Source database to read
+      --dest_database string  Optional: the destination database to write, default use --database 
+      --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.Database, "database", "", "", "Optional: the source 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

@@ -138,13 +138,18 @@ func (cmd *DataMigrateCommand) Run() error {
 	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 \"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")
+	db := cmd.opt.Database
+	if cmd.opt.DestDatabase != "" {
+		db = cmd.opt.DestDatabase
+	}
+	shardGroupDuration, err := gs.GetShardGroupDuration(db)
 	if err != nil {
 		return err
 	}
@@ -175,6 +180,9 @@ func (cmd *DataMigrateCommand) validate() error {
 	if cmd.opt.RetentionPolicy != "" && cmd.opt.Database == "" {
 		return fmt.Errorf("dataMigrate: must specify a db")
 	}
+	if cmd.opt.DestDatabase == "" {
+		cmd.opt.DestDatabase = cmd.opt.Database
+	}
 	if cmd.opt.StartTime != 0 && cmd.opt.EndTime != 0 && cmd.opt.EndTime < cmd.opt.StartTime {
 		return fmt.Errorf("dataMigrate: end time before start time")
 	}

src/geminiservice.go

@@ -7,7 +7,7 @@ import (
 )
 
 type GeminiService interface {
-	GetShardGroupDuration(database, rp string) (time.Duration, error)
+	GetShardGroupDuration(database string) (time.Duration, error)
 }
 
 var _ GeminiService = (*geminiService)(nil)
@@ -22,7 +22,7 @@ func NewGeminiService(cmd *DataMigrateCommand) *geminiService {
 	}
 }
 
-func (g *geminiService) GetShardGroupDuration(database, rp string) (time.Duration, error) {
+func (g *geminiService) GetShardGroupDuration(database string) (time.Duration, error) {
 	c, err := client.NewHTTPClient(client.HTTPConfig{
 		Addr: "http://" + g.out,
 	})
@@ -48,7 +48,7 @@ func (g *geminiService) GetShardGroupDuration(database, rp string) (time.Duratio
 	for _, item := range resp.Results {
 		for _, item1 := range item.Series {
 			for _, row := range item1.Values {
-				if row[0] == rp {
+				if row[7] == "true" {
 					shardGroupDuration, _ = time.ParseDuration(row[2].(string))
 					break
 				}

src/migrator.go

@@ -105,9 +105,15 @@ func (m *migrator) getStat() *statInfo {
 }
 
 func NewMigrator(cmd *DataMigrateCommand, info *shardGroupInfo) *migrator {
+	var db string
+	if cmd.opt.DestDatabase == "" {
+		db = info.db
+	} else {
+		db = cmd.opt.DestDatabase
+	}
 	mig := &migrator{
 		out:             cmd.opt.Out,
-		database:        info.db,
+		database:        db,
 		retentionPolicy: info.rp,
 		startTime:       cmd.opt.StartTime,
 		endTime:         cmd.opt.EndTime,

src/options.go

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