diff --git a/cmd/stackgraph/restore.go b/cmd/stackgraph/restore.go index fa82e49..c17fa23 100644 --- a/cmd/stackgraph/restore.go +++ b/cmd/stackgraph/restore.go @@ -35,6 +35,7 @@ var ( background bool skipConfirmation bool skipStackpacks bool + wipePVCs bool ) func restoreCmd(globalFlags *config.CLIGlobalFlags) *cobra.Command { @@ -54,6 +55,7 @@ func restoreCmd(globalFlags *config.CLIGlobalFlags) *cobra.Command { cmd.Flags().BoolVar(&background, "background", false, "Run restore job in background without waiting for completion") cmd.Flags().BoolVarP(&skipConfirmation, "yes", "y", false, "Skip confirmation prompt") cmd.Flags().BoolVar(&skipStackpacks, "skip-stackpacks", false, "Skip restoring stackpacks backup") + cmd.Flags().BoolVar(&wipePVCs, "wipe-pvcs", false, "Wipe HDFS/HBase/Tephra PVCs before restoring") cmd.MarkFlagsMutuallyExclusive("archive", "latest") cmd.MarkFlagsOneRequired("archive", "latest") @@ -103,6 +105,13 @@ func runRestore(appCtx *app.Context) error { return err } + // Scale down hbase/hdfs/tephra and delete all pvcs + if wipePVCs { + if err := restore.WipePVCs(appCtx.Context, appCtx.K8sClient, appCtx.Namespace, appCtx.Config.Stackgraph.Restore.ZookeeperQuorum, appCtx.Logger); err != nil { + return err + } + } + // Ensure deployments are scaled back up and lock released on exit (even if restore fails) defer func() { if len(scaledDeployments) > 0 && !background { diff --git a/cmd/stackgraphv2/restore.go b/cmd/stackgraphv2/restore.go index 9c0fafa..5b2dcd5 100644 --- a/cmd/stackgraphv2/restore.go +++ b/cmd/stackgraphv2/restore.go @@ -37,6 +37,7 @@ var ( useLatest bool skipConfirmation bool skipStackpacks bool + wipePVCs bool ) func restoreCmd(globalFlags *config.CLIGlobalFlags) *cobra.Command { @@ -56,6 +57,7 @@ func restoreCmd(globalFlags *config.CLIGlobalFlags) *cobra.Command { cmd.Flags().BoolVar(&useLatest, "latest", false, "Restore from the most recent backup") cmd.Flags().BoolVarP(&skipConfirmation, "yes", "y", false, "Skip confirmation prompt") cmd.Flags().BoolVar(&skipStackpacks, "skip-stackpacks", false, "Skip restoring stackpacks backup") + cmd.Flags().BoolVar(&wipePVCs, "wipe-pvcs", false, "Wipe HDFS/HBase/Tephra PVCs before restoring") cmd.MarkFlagsMutuallyExclusive("archive", "latest") cmd.MarkFlagsOneRequired("archive", "latest") @@ -128,6 +130,13 @@ func liveRestore(appCtx *app.Context) error { } }() + // Scale down hbase/hdfs/tephra and delete all pvcs + if wipePVCs { + if err := restore.WipePVCs(appCtx.Context, appCtx.K8sClient, appCtx.Namespace, appCtx.Config.Stackgraph.Restore.ZookeeperQuorum, appCtx.Logger); err != nil { + return err + } + } + // Setup Kubernetes resources for restore job appCtx.Logger.Println() if err := restore.EnsureResources(appCtx.K8sClient, appCtx.Namespace, appCtx.Config, appCtx.Logger); err != nil { diff --git a/go.mod b/go.mod index 4a79ca7..79ee58d 100644 --- a/go.mod +++ b/go.mod @@ -11,6 +11,7 @@ require ( github.com/aws/aws-sdk-go-v2/service/s3 v1.114.2 github.com/elastic/go-elasticsearch/v8 v8.19.7 github.com/go-playground/validator/v10 v10.28.0 + github.com/go-zookeeper/zk v1.0.4 github.com/spf13/cobra v1.10.2 github.com/spf13/pflag v1.0.10 github.com/stretchr/testify v1.11.1 diff --git a/go.sum b/go.sum index c5f83a5..a727c58 100644 --- a/go.sum +++ b/go.sum @@ -112,6 +112,8 @@ github.com/go-playground/universal-translator v0.18.2 h1:LCsMLC9RzmbUMNUPVYD15dm github.com/go-playground/universal-translator v0.18.2/go.mod h1:67VZIMp5lQpDWlnStOct22q1bkdJGJqHghbOtmkawxk= github.com/go-playground/validator/v10 v10.28.0 h1:Q7ibns33JjyW48gHkuFT91qX48KG0ktULL6FgHdG688= github.com/go-playground/validator/v10 v10.28.0/go.mod h1:GoI6I1SjPBh9p7ykNE/yj3fFYbyDOpwMn5KXd+m2hUU= +github.com/go-zookeeper/zk v1.0.4 h1:DPzxraQx7OrPyXq2phlGlNSIyWEsAox0RJmjTseMV6I= +github.com/go-zookeeper/zk v1.0.4/go.mod h1:nOB03cncLtlp4t+UAkGSV+9beXP/akpekBwL+UX1Qcw= github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk= github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= diff --git a/internal/orchestration/restore/hbase.go b/internal/orchestration/restore/hbase.go new file mode 100644 index 0000000..359a7e0 --- /dev/null +++ b/internal/orchestration/restore/hbase.go @@ -0,0 +1,125 @@ +package restore + +import ( + "context" + "errors" + "fmt" + "strconv" + "strings" + "time" + + "github.com/go-zookeeper/zk" + "github.com/stackvista/stackstate-backup-cli/internal/clients/k8s" + "github.com/stackvista/stackstate-backup-cli/internal/foundation/logger" + "github.com/stackvista/stackstate-backup-cli/internal/orchestration/portforward" + "github.com/stackvista/stackstate-backup-cli/internal/orchestration/scale" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +const zkSessionTimeout = 2 * time.Second + +func deleteRecursive(zkConn *zk.Conn, path string, logger *logger.Logger) error { + children, _, err := zkConn.Children(path) + switch { + case errors.Is(err, zk.ErrNoNode): + return nil + case err != nil: + logger.Warningf("Unable to list znode for %s: %v", path, err) + default: + for _, child := range children { + if err := deleteRecursive(zkConn, path+"/"+child, logger); err != nil { + return err + } + } + } + if err := zkConn.Delete(path, -1); err != nil && !errors.Is(err, zk.ErrNoNode) { + return err + } + return nil +} + +func scaleDownHBase(k8sClient *k8s.Client, namespace string, logger *logger.Logger) error { + logger.Println() + logger.Infof("Scaling down HBase") + _, err := scale.ScaleDown(k8sClient, namespace, "app.kubernetes.io/name=hbase", logger) + return err +} + +func WipePVCs(ctx context.Context, k8sClient *k8s.Client, namespace, zkQuorum string, logger *logger.Logger) error { + if !strings.Contains(zkQuorum, "-stackgraph") { + zkHost, zkPortStr, portSpecified := strings.Cut(zkQuorum, ":") + port := 2181 + if portSpecified { + if zkPort, err := strconv.Atoi(zkPortStr); err != nil { + logger.Warningf("Unable to parse port %s, falling back to %d", zkPortStr, port) + } else { + port = zkPort + } + } + pf, err := portforward.SetupPortForward(k8sClient, namespace, zkHost, port, logger) + if err != nil { + return err + } + defer close(pf.StopChan) + + zkEndpoint := fmt.Sprintf("localhost:%d", pf.LocalPort) + zkConn, _, err := zk.Connect( + []string{zkEndpoint}, + zkSessionTimeout, + ) + if err != nil { + return err + } + defer zkConn.Close() + + if err = scaleDownHBase(k8sClient, namespace, logger); err != nil { + return err + } + + logger.Println() + logger.Infof("Removing HBase and Tephra nodes from Zookeeper") + + if err := deleteRecursive(zkConn, "/hbase", logger); err != nil { + return err + } + if err := deleteRecursive(zkConn, "/tx.service", logger); err != nil { + return err + } + + logger.Successf("Removed nodes from Zookeeper") + } else { + if err := scaleDownHBase(k8sClient, namespace, logger); err != nil { + return err + } + } + + defer func() { + err := scale.ScaleUpFromAnnotations(k8sClient, namespace, "app.kubernetes.io/name=hbase", logger) + if err != nil { + logger.Warningf("Failed to scale up HBase: %v", err) + } + }() + + // List pvcs in the hbase application + pvcs, err := k8sClient.Clientset().CoreV1().PersistentVolumeClaims(namespace).List(ctx, metav1.ListOptions{ + LabelSelector: "app.kubernetes.io/name=hbase", + }) + if err != nil { + return err + } + + logger.Println() + logger.Infof("Deleting PVCs") + for _, pvc := range pvcs.Items { + if err := k8sClient.DeletePVC(pvc.Namespace, pvc.Name); err != nil { + return err + } + logger.Infof(" - %s", pvc.Name) + } + logger.Successf("Deleted PVCs") + + logger.Println() + logger.Infof("Scaled up HBase after deleting PVCs") + + return nil +}