feat: implement concurrent workers

.gitignore

@@ -1,2 +1,3 @@
 redis-migrator
 dist
+.test/

README.md

@@ -36,12 +36,17 @@ Supported Platforms:-
 
 For installation on debian and redhat based system, `.deb` and `.rpm` packages can be used.
 
-For installing on MacOS system, use brew:-
+For installing on MacOS system, use brew:
 
 ```shell
 brew install redis-migrator
 ```
 
+Install via `go install`
+```shell
+go install github.com/opstree/redis-migration@latest
+```
+
 ### Configuration
 
 For using redis-migrator, we have to create a configuration file and provide some needful information to it. An example configuration file will look like this:-

client/client.go

@@ -3,7 +3,7 @@ package client
 import (
 	"fmt"
 	"github.com/gomodule/redigo/redis"
-	"redis-migrator/config"
+	"github.com/opstree/redis-migration/config"
 )
 
 func generateClient(connectionURL string, redisPassword string, redisDatabase int) (redis.Conn, error) {

cmd/migrate.go

@@ -1,17 +1,19 @@
 package cmd
 
 import (
+	"github.com/opstree/redis-migration/config"
+	"github.com/opstree/redis-migration/migrator"
 	"github.com/spf13/cobra"
-	"redis-migrator/config"
-	"redis-migrator/migrator"
 )
 
 var migrateCmd = &cobra.Command{
 	Use:   "migrate",
 	Short: "Runs redis-migrator to run migration",
 	Long:  `Runs redis-migrator to run migration`,
 	Run: func(cmd *cobra.Command, args []string) {
-		runMigration()
+		if err := runMigration(); err != nil {
+			panic(err)
+		}
 	},
 }
 
@@ -20,7 +22,7 @@ func init() {
 	rootCmd.AddCommand(migrateCmd)
 }
 
-func runMigration() {
+func runMigration() error {
 	data := config.ParseConfig(configFilePath)
-	migrator.MigrateRedisData(data)
+	return migrator.MigrateRedisData(data)
 }

go.mod

@@ -1,13 +1,17 @@
-module redis-migrator
+module github.com/opstree/redis-migration
 
-go 1.15
+go 1.21
 
 require (
-	github.com/go-redis/redis/v8 v8.4.2
-	github.com/gomodule/redigo v1.8.3
-	github.com/mitchellh/mapstructure v1.1.2
-	github.com/sirupsen/logrus v1.7.0
-	github.com/spf13/cobra v1.1.1
+	github.com/gomodule/redigo v1.8.9
+	github.com/sirupsen/logrus v1.9.3
+	github.com/spf13/cobra v1.7.0
 	github.com/spf13/pflag v1.0.5
-	gopkg.in/yaml.v2 v2.3.0
+	gopkg.in/yaml.v2 v2.4.0
+)
+
+require (
+	github.com/inconshreveable/mousetrap v1.1.0 // indirect
+	github.com/stretchr/testify v1.7.0 // indirect
+	golang.org/x/sys v0.13.0 // indirect
 )

main.go

@@ -2,7 +2,7 @@ package main
 
 import (
 	"fmt"
-	"redis-migrator/cmd"
+	"github.com/opstree/redis-migration/cmd"
 )
 
 var (

migrator/migrator.go

@@ -1,46 +1,92 @@
 package migrator
 
 import (
+	"errors"
+	"fmt"
 	"github.com/gomodule/redigo/redis"
+	"github.com/opstree/redis-migration/client"
+	"github.com/opstree/redis-migration/config"
 	"github.com/sirupsen/logrus"
-	"redis-migrator/client"
-	"redis-migrator/config"
 )
 
 // MigrateRedisData is the function to migrate keys from old to new redis
-func MigrateRedisData(redConfig config.Configuration) {
-	for _, database := range redConfig.Databases {
-		logrus.Debugf("Executing migrator for database: %v", database)
-		oldRedisClient, err := client.OldRedisClient(redConfig, database)
-		if err != nil {
-			logrus.Errorf("Error while connecting with redis %v", err)
+func MigrateRedisData(redConfig config.Configuration) error {
+	concurrentWorkers := max(1, redConfig.ConcurrentWorkers)
+	if concurrentWorkers > len(redConfig.Databases) {
+		concurrentWorkers = len(redConfig.Databases)
+	}
+	logrus.Infof("Migrating with %d concurrent processes", concurrentWorkers)
+	errCh := make(chan error, len(redConfig.Databases))
+	defer close(errCh)
+	databaseCh := make(chan int, concurrentWorkers)
+	// Schedule task for workers
+	go func() {
+		for _, database := range redConfig.Databases {
+			databaseCh <- database
 		}
-		newRedisClient, err := client.NewRedisClient(redConfig, database)
-		if err != nil {
-			logrus.Errorf("Error while connecting with redis %v", err)
+		close(databaseCh)
+	}()
+
+	for i := 0; i < concurrentWorkers; i++ {
+		go func() {
+			var threadErr error
+			defer func() { errCh <- threadErr }()
+
+			for db := range databaseCh {
+				logrus.Infof("Migrating database: %d", db)
+				if err := migrateDB(redConfig, db); err != nil {
+					threadErr = err
+					return
+				}
+				logrus.Infof("Migrated database: %d", db)
+			}
+		}()
+	}
+
+	for i := 0; i < len(redConfig.Databases); i++ {
+		if err := <-errCh; err != nil {
+			return err
 		}
-		keys, err := redis.Strings(oldRedisClient.Do("KEYS", "*"))
+	}
+
+	return nil
+}
+
+func migrateDB(redConfig config.Configuration, db int) error {
+	oldRedisClient, err := client.OldRedisClient(redConfig, db)
+	if err != nil {
+		return fmt.Errorf("[DB %d] Error while connecting with redis %v", db, err)
+	}
+	newRedisClient, err := client.NewRedisClient(redConfig, db)
+	if err != nil {
+		return fmt.Errorf("[DB %d] Error while connecting with redis %v", db, err)
+	}
+	keys, err := redis.Strings(oldRedisClient.Do("KEYS", "*"))
+	if err != nil {
+		return fmt.Errorf("[DB %d] Error while listing redis keys %v", db, err)
+	}
+	logrus.Infof("[DB %d] Migrating %d keys", db, len(keys))
+	for i, key := range keys {
+		keyType, err := redis.String(oldRedisClient.Do("TYPE", key))
 		if err != nil {
-			logrus.Errorf("Error while listing redis keys %v", err)
+			return fmt.Errorf("[DB %d] Not able to get the key type %s: %v", db, key, err)
 		}
-		for _, key := range keys {
-			keyType, err := redis.String(oldRedisClient.Do("TYPE", key))
-			if err != nil {
-				logrus.Errorf("Not able to get the key type %s: %v", key, err)
-			}
-			switch keyType {
-			case "string":
-				migrateStringKeys(oldRedisClient, newRedisClient, key)
-			case "hash":
-				migrateHashKeys(oldRedisClient, newRedisClient, key)
-			case "list":
-				migarteListKeys(oldRedisClient, newRedisClient, key)
-			}
+		switch keyType {
+		case "string":
+			migrateStringKeys(oldRedisClient, newRedisClient, key)
+		case "hash":
+			migrateHashKeys(oldRedisClient, newRedisClient, key)
+		case "list":
+			migrateListKeys(oldRedisClient, newRedisClient, key)
+		default:
+			return errors.New(fmt.Sprintf("[DB %d] key type is not supported: %s", db, keyType))
 		}
+		logrus.Debugf("[DB %d] Migrated %d/%d", db, i, len(keys))
 	}
+	return nil
 }
 
-func migarteListKeys(oldClient redis.Conn, newClient redis.Conn, key string) {
+func migrateListKeys(oldClient redis.Conn, newClient redis.Conn, key string) {
 	value, err := redis.Strings(oldClient.Do("LPOP", key))
 	if err != nil {
 		logrus.Errorf("Not able to get the value for key %s: %v", key, err)