[cherry-pick] Fix component status updating for shard backup and restore (#1806) (#1833)

Signed-off-by: hmsayem <hmsayem@appscode.com>

Author: 1gtm

go.mod

@@ -4,33 +4,33 @@ go 1.18
 
 require (
 	github.com/pkg/errors v0.9.1
-	github.com/spf13/cobra v1.6.0
+	github.com/spf13/cobra v1.7.0
 	go.bytebuilders.dev/license-verifier/kubernetes v0.12.0
 	gomodules.xyz/flags v0.1.3
 	gomodules.xyz/go-sh v0.1.0
 	gomodules.xyz/logs v0.0.6
 	gomodules.xyz/pointer v0.1.0
-	gomodules.xyz/x v0.0.14
-	k8s.io/api v0.25.1
+	gomodules.xyz/x v0.0.15
+	k8s.io/api v0.25.3
 	k8s.io/apimachinery v0.25.3
-	k8s.io/client-go v0.25.1
+	k8s.io/client-go v0.25.3
 	k8s.io/klog/v2 v2.80.1
-	kmodules.xyz/client-go v0.25.19
-	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.29.0
+	kmodules.xyz/client-go v0.25.23
+	kmodules.xyz/custom-resources v0.25.1
+	kmodules.xyz/offshoot-api v0.25.3
+	kubedb.dev/apimachinery v0.33.1
+	stash.appscode.dev/apimachinery v0.29.1-0.20230529131221-1e979c48da10
 )
 
 require (
 	cloud.google.com/go v0.99.0 // indirect
 	github.com/Azure/go-autorest v14.2.0+incompatible // indirect
-	github.com/Azure/go-autorest/autorest v0.11.27 // indirect
-	github.com/Azure/go-autorest/autorest/adal v0.9.20 // indirect
+	github.com/Azure/go-autorest/autorest v0.11.28 // indirect
+	github.com/Azure/go-autorest/autorest/adal v0.9.21 // indirect
 	github.com/Azure/go-autorest/autorest/date v0.3.0 // indirect
 	github.com/Azure/go-autorest/logger v0.2.1 // indirect
 	github.com/Azure/go-autorest/tracing v0.6.0 // indirect
-	github.com/Masterminds/semver/v3 v3.1.1 // indirect
+	github.com/Masterminds/semver/v3 v3.2.1 // indirect
 	github.com/armon/circbuf v0.0.0-20190214190532-5111143e8da2 // indirect
 	github.com/codegangsta/inject v0.0.0-20150114235600-33e0aa1cb7c0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
@@ -49,7 +49,7 @@ require (
 	github.com/google/go-cmp v0.5.9 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
 	github.com/imdario/mergo v0.3.13 // indirect
-	github.com/inconshreveable/mousetrap v1.0.1 // indirect
+	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect
@@ -62,16 +62,16 @@ require (
 	github.com/yudai/gojsondiff v1.0.0 // indirect
 	github.com/yudai/golcs v0.0.0-20170316035057-ecda9a501e82 // indirect
 	go.bytebuilders.dev/license-proxyserver v0.0.3 // indirect
-	go.bytebuilders.dev/license-verifier v0.12.1-0.20221113063237-6eb88040dd50 // indirect
-	golang.org/x/crypto v0.6.0 // indirect
-	golang.org/x/net v0.7.0 // indirect
+	go.bytebuilders.dev/license-verifier v0.12.1 // indirect
+	golang.org/x/crypto v0.9.0 // indirect
+	golang.org/x/net v0.10.0 // indirect
 	golang.org/x/oauth2 v0.5.0 // indirect
-	golang.org/x/sys v0.5.0 // indirect
-	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/sys v0.8.0 // indirect
+	golang.org/x/term v0.8.0 // indirect
+	golang.org/x/text v0.9.0 // indirect
+	golang.org/x/time v0.3.0 // indirect
 	gomodules.xyz/clock v0.0.0-20200817085942-06523dba733f // indirect
-	gomodules.xyz/jsonpatch/v2 v2.2.0 // indirect
+	gomodules.xyz/jsonpatch/v2 v2.3.0 // indirect
 	gomodules.xyz/mergo v0.3.13 // indirect
 	gomodules.xyz/sets v0.2.1 // indirect
 	gomodules.xyz/wait v0.2.0 // indirect
@@ -80,11 +80,11 @@ require (
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
-	k8s.io/apiextensions-apiserver v0.25.1 // indirect
-	k8s.io/apiserver v0.25.1 // indirect
+	k8s.io/apiextensions-apiserver v0.25.3 // indirect
+	k8s.io/apiserver v0.25.3 // indirect
 	k8s.io/kube-aggregator v0.25.1 // indirect
-	k8s.io/kube-openapi v0.0.0-20220803164354-a70c9af30aea // indirect
-	k8s.io/utils v0.0.0-20220823124924-e9cbc92d1a73 // indirect
+	k8s.io/kube-openapi v0.0.0-20221012153701-172d655c2280 // indirect
+	k8s.io/utils v0.0.0-20221012122500-cfd413dd9e85 // indirect
 	kmodules.xyz/apiversion v0.2.0 // indirect
 	kmodules.xyz/objectstore-api v0.25.1-0.20221104003322-f0289b5b6ca2 // indirect
 	kmodules.xyz/prober v0.25.0 // indirect

go.sum

@@ -149,14 +149,8 @@ func NewCmdBackup() *cobra.Command {
 			if err != nil {
 				backupOutput = &restic.BackupOutput{
 					BackupTargetStatus: api_v1beta1.BackupTargetStatus{
-						Ref: targetRef,
-						Stats: []api_v1beta1.HostBackupStats{
-							{
-								Hostname: opt.defaultBackupOptions.Host,
-								Phase:    api_v1beta1.HostBackupFailed,
-								Error:    err.Error(),
-							},
-						},
+						Ref:   targetRef,
+						Stats: opt.getHostBackupStats(err),
 					},
 				}
 			}
@@ -301,8 +295,10 @@ func (opt *mongoOptions) backupMongoDB(targetRef api_v1beta1.TargetRef) (*restic
 	// So, for stand-alone MongoDB and MongoDB ReplicaSet, we don't have to do anything.
 	// We only need to update totalHosts field for sharded MongoDB
 
+	opt.totalHosts = 1
 	// For sharded MongoDB, parameter.ConfigServer will not be empty
 	if parameters.ConfigServer != "" {
+		opt.totalHosts = len(parameters.ReplicaSets) + 1 // for each shard there will be one key in parameters.ReplicaSet
 		backupSession, err := opt.stashClient.StashV1beta1().BackupSessions(opt.namespace).Get(context.TODO(), opt.backupSessionName, metav1.GetOptions{})
 		if err != nil {
 			return nil, err
@@ -314,7 +310,7 @@ func (opt *mongoOptions) backupMongoDB(targetRef api_v1beta1.TargetRef) (*restic
 					opt.stashClient.StashV1beta1(),
 					backupSession.ObjectMeta,
 					func(status *api_v1beta1.BackupSessionStatus) (types.UID, *api_v1beta1.BackupSessionStatus) {
-						status.Targets[i].TotalHosts = pointer.Int32P(int32(len(parameters.ReplicaSets) + 1)) // for each shard there will be one key in parameters.ReplicaSet
+						status.Targets[i].TotalHosts = pointer.Int32P(int32(opt.totalHosts))
 						return backupSession.UID, status
 					},
 					metav1.UpdateOptions{},
@@ -571,7 +567,12 @@ func (opt *mongoOptions) backupMongoDB(targetRef api_v1beta1.TargetRef) (*restic
 	// hide password, don't print cmd
 	resticWrapper.HideCMD()
 
-	return resticWrapper.RunParallelBackup(opt.backupOptions, targetRef, opt.maxConcurrency)
+	out, err := resticWrapper.RunParallelBackup(opt.backupOptions, targetRef, opt.maxConcurrency)
+	if err != nil {
+		klog.Warningln("backup failed!", err.Error())
+	}
+	// error not returned, error is encoded into output
+	return out, nil
 }
 
 // cleanup usually unlocks the locked servers
@@ -583,6 +584,32 @@ func cleanup() {
 	}
 }
 
+func (opt *mongoOptions) getHostBackupStats(err error) []api_v1beta1.HostBackupStats {
+	var backupStats []api_v1beta1.HostBackupStats
+
+	errMsg := fmt.Sprintf("failed to start backup: %s", err.Error())
+	for _, backupOpt := range opt.backupOptions {
+		backupStats = append(backupStats, api_v1beta1.HostBackupStats{
+			Hostname: backupOpt.Host,
+			Phase:    api_v1beta1.HostBackupFailed,
+			Error:    errMsg,
+		})
+	}
+
+	if opt.totalHosts > len(backupStats) {
+		rem := opt.totalHosts - len(backupStats)
+		for i := 0; i < rem; i++ {
+			backupStats = append(backupStats, api_v1beta1.HostBackupStats{
+				Hostname: fmt.Sprintf("unknown-%s", strconv.Itoa(i)),
+				Phase:    api_v1beta1.HostBackupFailed,
+				Error:    errMsg,
+			})
+		}
+	}
+
+	return backupStats
+}
+
 func getSSLUser(path string) (string, error) {
 	data, err := sh.Command(OpenSSLCMD, "x509", "-in", path, "-inform", "PEM", "-subject", "-nameopt", "RFC2253", "-noout").Output()
 	if err != nil {

pkg/backup.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"strconv"
 	"strings"
 
 	api_v1beta1 "stash.appscode.dev/apimachinery/apis/stash/v1beta1"
@@ -33,7 +34,6 @@ import (
 	"github.com/spf13/cobra"
 	license "go.bytebuilders.dev/license-verifier/kubernetes"
 	"gomodules.xyz/flags"
-	"gomodules.xyz/go-sh"
 	"gomodules.xyz/pointer"
 	core "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -106,14 +106,8 @@ func NewCmdRestore() *cobra.Command {
 			if err != nil {
 				restoreOutput = &restic.RestoreOutput{
 					RestoreTargetStatus: api_v1beta1.RestoreMemberStatus{
-						Ref: targetRef,
-						Stats: []api_v1beta1.HostRestoreStats{
-							{
-								Hostname: opt.defaultDumpOptions.Host,
-								Phase:    api_v1beta1.HostRestoreFailed,
-								Error:    err.Error(),
-							},
-						},
+						Ref:   targetRef,
+						Stats: opt.getHostRestoreStats(err),
 					},
 				}
 			}
@@ -232,8 +226,10 @@ func (opt *mongoOptions) restoreMongoDB(targetRef api_v1beta1.TargetRef) (*resti
 	// So, for stand-alone MongoDB and MongoDB ReplicaSet, we don't have to do anything.
 	// We only need to update totalHosts field for sharded MongoDB
 
+	opt.totalHosts = 1
 	// For sharded MongoDB, parameter.ConfigServer will not be empty
 	if parameters.ConfigServer != "" {
+		opt.totalHosts = len(parameters.ReplicaSets) + 1 // for each shard there will be one key in parameters.ReplicaSet
 		restoreSession, err := opt.stashClient.StashV1beta1().RestoreSessions(opt.namespace).Get(context.TODO(), opt.restoreSessionName, metav1.GetOptions{})
 		if err != nil {
 			return nil, err
@@ -243,7 +239,7 @@ func (opt *mongoOptions) restoreMongoDB(targetRef api_v1beta1.TargetRef) (*resti
 			opt.stashClient.StashV1beta1(),
 			restoreSession.ObjectMeta,
 			func(status *api_v1beta1.RestoreSessionStatus) (types.UID, *api_v1beta1.RestoreSessionStatus) {
-				status.TotalHosts = pointer.Int32P(int32(len(parameters.ReplicaSets) + 1)) // for each shard there will be one key in parameters.ReplicaSet
+				status.TotalHosts = pointer.Int32P(int32(opt.totalHosts))
 				return restoreSession.UID, status
 			},
 			metav1.UpdateOptions{},
@@ -397,29 +393,36 @@ func (opt *mongoOptions) restoreMongoDB(targetRef api_v1beta1.TargetRef) (*resti
 	resticWrapper.HideCMD()
 
 	// Run dump
-	restoreOutput, err := resticWrapper.ParallelDump(opt.dumpOptions, targetRef, opt.maxConcurrency)
+	out, err := resticWrapper.ParallelDump(opt.dumpOptions, targetRef, opt.maxConcurrency)
 	if err != nil {
-		return nil, err
+		klog.Warningln("restore failed!", err.Error())
 	}
+	// error not returned, error is encoded into output
+	return out, nil
+}
 
-	if parameters.ConfigServer != "" {
-		err = dropTempReshardCollection(parameters.ConfigServer)
-		if err != nil {
-			klog.Errorf("error while deleting temporary reshard collection for %v. error: %v", parameters.ConfigServer, err)
-			return nil, err
-		}
-	}
+func (opt *mongoOptions) getHostRestoreStats(err error) []api_v1beta1.HostRestoreStats {
+	var restoreStats []api_v1beta1.HostRestoreStats
 
-	return restoreOutput, nil
-}
+	errMsg := fmt.Sprintf("failed to start data restoration: %s", err.Error())
+	for _, dumpOpt := range opt.dumpOptions {
+		restoreStats = append(restoreStats, api_v1beta1.HostRestoreStats{
+			Hostname: dumpOpt.Host,
+			Phase:    api_v1beta1.HostRestoreFailed,
+			Error:    errMsg,
+		})
+	}
 
-func dropTempReshardCollection(configsvrDSN string) error {
-	args := append([]interface{}{
-		"config",
-		"--host", configsvrDSN,
-		"--quiet",
-		"--eval", `db.reshardingOperations_temp.drop()`,
-	}, mongoCreds...)
+	if opt.totalHosts > len(restoreStats) {
+		rem := opt.totalHosts - len(restoreStats)
+		for i := 0; i < rem; i++ {
+			restoreStats = append(restoreStats, api_v1beta1.HostRestoreStats{
+				Hostname: fmt.Sprintf("unknown-%s", strconv.Itoa(i)),
+				Phase:    api_v1beta1.HostRestoreFailed,
+				Error:    errMsg,
+			})
+		}
+	}
 
-	return sh.Command(MongoCMD, args...).Command("/usr/bin/tail", "-1").Run()
+	return restoreStats
 }

pkg/restore.go

@@ -68,6 +68,7 @@ type mongoOptions struct {
 	dumpOptions          []restic.DumpOptions
 	defaultDumpOptions   restic.DumpOptions
 	config               *restclient.Config
+	totalHosts           int
 }
 
 func waitForDBReady(host string, port, waitTimeout int32) {