[cherry-pick] Support backup and restore for redis clusters (#164) (#166)

/cherry-pick

Signed-off-by: Tamal Saha <tamal@appscode.com>
Co-authored-by: Tamal Saha <tamal@appscode.com>

Author: 1gtm

Makefile

@@ -43,7 +43,7 @@ else
 endif
 
 RESTIC_VER       := 0.13.1
-REDIS_DUMP_VER   := 0.7.2-ac
+REDIS_DUMP_VER   := 0.8.0-ac
 
 ###
 ### These variables should not need tweaking.
@@ -59,8 +59,8 @@ BIN_PLATFORMS    := $(DOCKER_PLATFORMS)
 OS   := $(if $(GOOS),$(GOOS),$(shell go env GOOS))
 ARCH := $(if $(GOARCH),$(GOARCH),$(shell go env GOARCH))
 
-BASEIMAGE_PROD   ?= redis:6.2.5
-BASEIMAGE_DBG    ?= redis:6.2.5
+BASEIMAGE_PROD   ?= redis:7.0.5
+BASEIMAGE_DBG    ?= redis:7.0.5
 
 IMAGE            := $(REGISTRY)/$(BIN)
 VERSION_PROD     := $(VERSION)

go.mod

@@ -3,7 +3,9 @@ module stash.appscode.dev/redis
 go 1.18
 
 require (
+	github.com/mediocregopher/radix/v3 v3.8.0
 	github.com/spf13/cobra v1.6.0
+	github.com/yannh/redis-dump-go v0.0.0-00010101000000-000000000000
 	go.bytebuilders.dev/license-verifier/kubernetes v0.12.0
 	gomodules.xyz/flags v0.1.3
 	gomodules.xyz/go-sh v0.1.0
@@ -17,7 +19,7 @@ require (
 	kmodules.xyz/custom-resources v0.25.0
 	kmodules.xyz/offshoot-api v0.25.0
 	kubedb.dev/apimachinery v0.28.4-0.20220918021210-a0b96812228b
-	stash.appscode.dev/apimachinery v0.28.0
+	stash.appscode.dev/apimachinery v0.28.1-0.20230429131740-425734e18c7c
 )
 
 require (
@@ -69,6 +71,7 @@ require (
 	golang.org/x/term v0.5.0 // indirect
 	golang.org/x/text v0.7.0 // indirect
 	golang.org/x/time v0.1.0 // indirect
+	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
 	gomodules.xyz/clock v0.0.0-20200817085942-06523dba733f // indirect
 	gomodules.xyz/jsonpatch/v2 v2.2.0 // indirect
 	gomodules.xyz/mergo v0.3.13 // indirect
@@ -93,3 +96,5 @@ require (
 	sigs.k8s.io/structured-merge-diff/v4 v4.2.3 // indirect
 	sigs.k8s.io/yaml v1.3.0 // indirect
 )
+
+replace github.com/yannh/redis-dump-go => github.com/kubedb/redis-dump-go v0.8.1-0.20230429151509-2f2a7ce60763

go.sum

@@ -293,6 +293,8 @@ github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfn
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kubedb/redis-dump-go v0.8.1-0.20230429151509-2f2a7ce60763 h1:HyNjcmSJSLEPXN+y6wNNLtNPfEcZiPotGPHbnkhj1g0=
+github.com/kubedb/redis-dump-go v0.8.1-0.20230429151509-2f2a7ce60763/go.mod h1:u6sFg98XPtTAaIyUv5oq+4D8D6krErkijf78cV30VOA=
 github.com/magiconair/properties v1.8.1/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
 github.com/mailru/easyjson v0.0.0-20190614124828-94de47d64c63/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
 github.com/mailru/easyjson v0.0.0-20190626092158-b2ccc519800e/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
@@ -304,6 +306,8 @@ github.com/mattn/go-isatty v0.0.3/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNx
 github.com/mattn/go-isatty v0.0.14 h1:yVuAays6BHfxijgZPzw+3Zlu5yQgKGP2/hcQbHb7S9Y=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/matttproud/golang_protobuf_extensions v1.0.2-0.20181231171920-c182affec369 h1:I0XW9+e1XWDxdcEniV4rQAIOPUGDq67JSCiRCgGCZLI=
+github.com/mediocregopher/radix/v3 v3.8.0 h1:HI8EgkaM7WzsrFpYAkOXIgUKbjNonb2Ne7K6Le61Pmg=
+github.com/mediocregopher/radix/v3 v3.8.0/go.mod h1:8FL3F6UQRXHXIBSPUs5h0RybMF8i4n7wVopoX3x7Bv8=
 github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
 github.com/mitchellh/cli v1.0.0/go.mod h1:hNIlj7HEI86fIcpObd7a0FcrxTWetlwJDGcceTlRvqc=
 github.com/mitchellh/go-homedir v1.0.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
@@ -673,6 +677,7 @@ golang.org/x/tools v0.1.5/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE=
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 gomodules.xyz/clock v0.0.0-20200817085942-06523dba733f h1:hTyhR4r+tj1Uq7/PpFxLTzbeA0LhMVp7bEYfhkzFjdY=
 gomodules.xyz/clock v0.0.0-20200817085942-06523dba733f/go.mod h1:K3m7N+nBOlf91/tpv8REUGwsAgaKFwElQCuiLhm12AQ=
@@ -909,5 +914,5 @@ sigs.k8s.io/structured-merge-diff/v4 v4.2.3 h1:PRbqxJClWWYMNV1dhaG4NsibJbArud9kF
 sigs.k8s.io/structured-merge-diff/v4 v4.2.3/go.mod h1:qjx8mGObPmV2aSZepjQjbmb2ihdVs8cGKBraizNC69E=
 sigs.k8s.io/yaml v1.3.0 h1:a2VclLzOGrwOHDiV8EfBGhvjHvP46CtW5j6POvhYGGo=
 sigs.k8s.io/yaml v1.3.0/go.mod h1:GeOyir5tyXNByN85N/dRIT9es5UQNerPYEKK56eTBm8=
-stash.appscode.dev/apimachinery v0.28.0 h1:Wy/u6/m1O7qNgNSJs/4T2Eeg1drODv2mkQJndu9dtaQ=
-stash.appscode.dev/apimachinery v0.28.0/go.mod h1:7ao2U0Jgntc10uRf9KRmzXuabO12Nm7+2WpcRE/ZXWo=
+stash.appscode.dev/apimachinery v0.28.1-0.20230429131740-425734e18c7c h1:X5nxy4AApyos2sHrIzoZpEcsT8vMdzqdusz2U+BhlKY=
+stash.appscode.dev/apimachinery v0.28.1-0.20230429131740-425734e18c7c/go.mod h1:7ao2U0Jgntc10uRf9KRmzXuabO12Nm7+2WpcRE/ZXWo=

pkg/backup.go

@@ -27,6 +27,7 @@ import (
 	api_util "stash.appscode.dev/apimachinery/pkg/util"
 
 	"github.com/spf13/cobra"
+	"github.com/yannh/redis-dump-go/pkg/redisdump"
 	license "go.bytebuilders.dev/license-verifier/kubernetes"
 	"gomodules.xyz/flags"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -192,39 +193,46 @@ func (opt *redisOptions) backupRedis(targetRef api_v1beta1.TargetRef) (*restic.B
 		return nil, err
 	}
 
-	session := opt.newSessionWrapper(RedisDumpCMD)
-
-	err = session.setDatabaseCredentials(opt.kubeClient, appBinding)
+	hostname, err := appBinding.Hostname()
 	if err != nil {
 		return nil, err
 	}
-
-	err = opt.setTLSParameters(appBinding, session.cmd)
+	port, err := appBinding.Port()
 	if err != nil {
 		return nil, err
 	}
-
-	err = session.waitForDBReady(appBinding)
+	username, password, err := getDatabaseCredentials(opt.kubeClient, appBinding)
 	if err != nil {
 		return nil, err
 	}
 
-	hostname, err := appBinding.Hostname()
+	s := redisdump.Host{
+		Host:       hostname,
+		Port:       int(port),
+		Username:   username,
+		Password:   password,
+		TlsHandler: nil, // TODO(Shaad7): Add support for tls protected redis
+	}
+
+	session := opt.newSessionWrapper(RedisDumpCMD)
+	session.setDatabaseCredentials(password)
+
+	err = opt.setTLSParameters(appBinding, session.cmd)
 	if err != nil {
 		return nil, err
 	}
 
-	session.cmd.Args = append(session.cmd.Args, "-host", hostname)
-
-	port, err := appBinding.Port()
+	err = session.waitForDBReady(s)
 	if err != nil {
 		return nil, err
 	}
 
+	session.cmd.Args = append(session.cmd.Args, "-host", s.Host)
 	// if port is specified, append port in the arguments
-	if port != 0 {
-		session.cmd.Args = append(session.cmd.Args, "-port", strconv.Itoa(int(port)))
+	if s.Port != 0 {
+		session.cmd.Args = append(session.cmd.Args, "-port", strconv.Itoa(s.Port))
 	}
+	session.cmd.Args = append(session.cmd.Args, "--set-total-keys")
 
 	session.setUserArgs(opt.redisArgs)
 	// add backup command in the pipeline

pkg/restore.go

@@ -18,13 +18,18 @@ package pkg
 
 import (
 	"context"
+	"fmt"
 	"path/filepath"
 	"strconv"
+	"time"
 
 	api_v1beta1 "stash.appscode.dev/apimachinery/apis/stash/v1beta1"
 	"stash.appscode.dev/apimachinery/pkg/restic"
 
+	"github.com/mediocregopher/radix/v3"
 	"github.com/spf13/cobra"
+	"github.com/yannh/redis-dump-go/pkg/config"
+	"github.com/yannh/redis-dump-go/pkg/redisdump"
 	license "go.bytebuilders.dev/license-verifier/kubernetes"
 	"gomodules.xyz/flags"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -134,6 +139,8 @@ func NewCmdRestore() *cobra.Command {
 
 	cmd.Flags().StringVar(&opt.outputDir, "output-dir", opt.outputDir, "Directory where output.json file will be written (keep empty if you don't need to write output in file)")
 
+	cmd.Flags().IntVar(&opt.NWorkers, "n", 10, "Parallel workers")
+
 	return cmd
 }
 
@@ -164,48 +171,157 @@ func (opt *redisOptions) restoreRedis(targetRef api_v1beta1.TargetRef) (*restic.
 		return nil, err
 	}
 
-	session := opt.newSessionWrapper(RedisRestoreCMD)
-
-	err = session.setDatabaseCredentials(opt.kubeClient, appBinding)
+	hostname, err := appBinding.Hostname()
 	if err != nil {
 		return nil, err
 	}
-
-	err = opt.setTLSParameters(appBinding, session.cmd)
+	port, err := appBinding.Port()
 	if err != nil {
 		return nil, err
 	}
-
-	err = session.waitForDBReady(appBinding)
+	username, password, err := getDatabaseCredentials(opt.kubeClient, appBinding)
 	if err != nil {
 		return nil, err
 	}
 
-	session.cmd.Args = append(session.cmd.Args, "--pipe")
-
-	hostname, err := appBinding.Hostname()
-	if err != nil {
-		return nil, err
+	s := redisdump.Host{
+		Host:       hostname,
+		Port:       int(port),
+		Username:   username,
+		Password:   password,
+		TlsHandler: nil, // TODO(Shaad7): Add support for tls protected redis
 	}
-	session.cmd.Args = append(session.cmd.Args, "-h", hostname)
 
-	port, err := appBinding.Port()
-	if err != nil {
+	if hosts, err := redisdump.GetHosts(s, opt.NWorkers); err != nil {
 		return nil, err
-	}
-	// if port is specified, append port in the arguments
-	if port != 0 {
-		session.cmd.Args = append(session.cmd.Args, "-p", strconv.Itoa(int(port)))
-	}
+	} else {
+		redisCluster := len(hosts) > 1
+		// Start clock to measure total restore duration
+		startTime := time.Now()
+		beforeKeys := 0
+		afterKeys := 0
 
-	session.setUserArgs(opt.redisArgs)
+		for _, host := range hosts {
+			session := opt.newSessionWrapper(RedisRestoreCMD)
 
-	// append the restore command to the pipeline
-	opt.dumpOptions.StdoutPipeCommands = append(opt.dumpOptions.StdoutPipeCommands, *session.cmd)
-	resticWrapper, err := restic.NewResticWrapperFromShell(opt.setupOptions, session.sh)
-	if err != nil {
-		return nil, err
+			session.setDatabaseCredentials(host.Password)
+			if err != nil {
+				return nil, err
+			}
+
+			err = opt.setTLSParameters(appBinding, session.cmd)
+			if err != nil {
+				return nil, err
+			}
+
+			err = session.waitForDBReady(host)
+			if err != nil {
+				return nil, err
+			}
+
+			session.cmd.Args = append(session.cmd.Args, "--pipe")
+
+			session.cmd.Args = append(session.cmd.Args, "-h", host.Host)
+
+			// if port is specified, append port in the arguments
+			if host.Port != 0 {
+				session.cmd.Args = append(session.cmd.Args, "-p", strconv.Itoa(host.Port))
+			}
+
+			session.setUserArgs(opt.redisArgs)
+
+			// append the restore command to the pipeline
+			opt.dumpOptions.StdoutPipeCommands = []restic.Command{*session.cmd}
+			resticWrapper, err := restic.NewResticWrapperFromShell(opt.setupOptions, session.sh)
+			if err != nil {
+				return nil, err
+			}
+
+			// if source host is not specified then use current host as source host
+			if opt.dumpOptions.SourceHost == "" {
+				opt.dumpOptions.SourceHost = opt.dumpOptions.Host
+			}
+
+			var client *radix.Pool
+			if redisCluster {
+				client, err = redisdump.NewClient(host, nil, opt.NWorkers)
+				if err != nil {
+					return nil, err
+				}
+				if size, err := DBSize(client); err != nil {
+					return nil, err
+				} else {
+					beforeKeys += size
+				}
+			}
+
+			// Run dump
+			// Redis cluster restore will always return error. So, ignore error for redis clusters
+			_, err = resticWrapper.DumpOnce(opt.dumpOptions)
+			if !redisCluster && err != nil {
+				return nil, err
+			}
+
+			if redisCluster {
+				if size, err := DBSize(client); err != nil {
+					return nil, err
+				} else {
+					afterKeys += size
+				}
+				client.Close()
+			}
+		}
+
+		if redisCluster {
+			client, err := redisdump.NewCluster(hosts)
+			if err != nil {
+				return nil, err
+			}
+			defer client.Close()
+
+			var strBackedupKeys string
+			err = client.Do(radix.Cmd(&strBackedupKeys, "GET", config.KeyTotalKeys))
+			if err != nil {
+				return nil, err
+			}
+			backedupKeys, err := strconv.Atoi(strBackedupKeys)
+			if err != nil {
+				return nil, err
+			}
+			fmt.Printf("Total keys found in backuped data: %d\n", backedupKeys)
+			fmt.Printf("Total keys in redis before restore: %d, after restore: %d\n", beforeKeys, afterKeys)
+
+			_ = client.Do(radix.Cmd(nil, "DEL", config.KeyTotalKeys))
+		} else {
+			client, err := redisdump.NewClient(s, nil, opt.NWorkers)
+			if err != nil {
+				return nil, err
+			}
+			defer client.Close()
+
+			_ = client.Do(radix.Cmd(nil, "DEL", config.KeyTotalKeys))
+		}
+
+		restoreStats := api_v1beta1.HostRestoreStats{
+			Hostname: opt.dumpOptions.Host,
+		}
+		// Dump successful. Now, calculate total session duration.
+		restoreStats.Duration = time.Since(startTime).String()
+		restoreStats.Phase = api_v1beta1.HostRestoreSucceeded
+		restoreOutput := &restic.RestoreOutput{
+			RestoreTargetStatus: api_v1beta1.RestoreMemberStatus{
+				Ref:   targetRef,
+				Stats: []api_v1beta1.HostRestoreStats{restoreStats},
+			},
+		}
+		return restoreOutput, nil
+	}
+}
+
+func DBSize(client *radix.Pool) (int, error) {
+	var dbSize string
+	if err := client.Do(radix.Cmd(&dbSize, "dbsize")); err != nil {
+		return 0, err
 	}
-	// Run dump
-	return resticWrapper.Dump(opt.dumpOptions, targetRef)
+	return strconv.Atoi(dbSize)
 }