From 4172e381aea16fdc160fd67a5a921380da30dc0c Mon Sep 17 00:00:00 2001 From: Frank van Lankvelt Date: Wed, 14 Oct 2026 15:38:54 +0200 Subject: [PATCH 1/4] STS 537: wipe pvcs of hbase when requested --- cmd/stackgraph/restore.go | 12 +++ cmd/stackgraphv2/restore.go | 12 +++ go.mod | 1 + go.sum | 2 + internal/orchestration/restore/hbase.go | 103 ++++++++++++++++++++++++ 5 files changed, 130 insertions(+) create mode 100644 internal/orchestration/restore/hbase.go diff --git a/cmd/stackgraph/restore.go b/cmd/stackgraph/restore.go index fa82e49..73da211 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,16 @@ func runRestore(appCtx *app.Context) error { return err } + // Scale down hbase/hdfs/tephra and delete all pvcs + if wipePVCs { + err = restore.WipePVCs(appCtx.Context, appCtx.K8sClient, appCtx.Namespace, appCtx.Config.Stackgraph.Restore.ZookeeperQuorum, appCtx.Logger) + if err != nil { + // Create restore job + appCtx.Logger.Println() + appCtx.Logger.Warningf("Wiping PVCs failed: %s", 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 be2f1ab..d20973a 100644 --- a/cmd/stackgraphv2/restore.go +++ b/cmd/stackgraphv2/restore.go @@ -34,6 +34,7 @@ var ( useLatest bool skipConfirmation bool skipStackpacks bool + wipePVCs bool ) func restoreCmd(globalFlags *config.CLIGlobalFlags) *cobra.Command { @@ -53,6 +54,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") @@ -125,6 +127,16 @@ func liveRestore(appCtx *app.Context) error { } }() + // Scale down hbase/hdfs/tephra and delete all pvcs + if wipePVCs { + err = restore.WipePVCs(appCtx.Context, appCtx.K8sClient, appCtx.Namespace, appCtx.Config.Stackgraph.Restore.ZookeeperQuorum, appCtx.Logger) + if err != nil { + // Create restore job + appCtx.Logger.Println() + appCtx.Logger.Warningf("Wiping PVCs failed: %s", 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 04f2e61..3d0ca0f 100644 --- a/go.mod +++ b/go.mod @@ -11,6 +11,7 @@ require ( github.com/aws/aws-sdk-go-v2/service/s3 v1.97.3 github.com/elastic/go-elasticsearch/v8 v8.19.0 github.com/go-playground/validator/v10 v10.28.0 + github.com/go-zookeeper/zk v1.0.4 github.com/spf13/cobra v1.10.1 github.com/stretchr/testify v1.11.1 gopkg.in/yaml.v3 v3.0.1 diff --git a/go.sum b/go.sum index 9cb0984..49089b1 100644 --- a/go.sum +++ b/go.sum @@ -89,6 +89,8 @@ github.com/go-playground/validator/v10 v10.28.0 h1:Q7ibns33JjyW48gHkuFT91qX48KG0 github.com/go-playground/validator/v10 v10.28.0/go.mod h1:GoI6I1SjPBh9p7ykNE/yj3fFYbyDOpwMn5KXd+m2hUU= github.com/go-task/slim-sprig/v3 v3.0.0 h1:sUs3vkvUymDpBKi3qH1YSqBQk9+9D/8M2mN1vB6EwHI= github.com/go-task/slim-sprig/v3 v3.0.0/go.mod h1:W848ghGpv3Qj3dhTPRyJypKRiqCdHZiAzKg9hl15HA8= +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 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q= github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk= diff --git a/internal/orchestration/restore/hbase.go b/internal/orchestration/restore/hbase.go new file mode 100644 index 0000000..ecfc02a --- /dev/null +++ b/internal/orchestration/restore/hbase.go @@ -0,0 +1,103 @@ +package restore + +import ( + "context" + "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) { + children, _, err := zkConn.Children(path) + if err != nil { + logger.Warningf("Unable to list znode for %s: %v", path, err) + } else { + for _, child := range children { + deleteRecursive(zkConn, path+"/"+child, logger) + } + } + err = zkConn.Delete(path, -1) + if err != nil { + logger.Warningf("Failed to delete znode %s: %v", path, err) + } +} + +func WipePVCs(ctx context.Context, k8sClient *k8s.Client, namespace, zkQuorum string, logger *logger.Logger) error { + logger.Println() + logger.Infof("Scaling down HBase and deleting PVCs") + + 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() + + _, err = scale.ScaleDown(k8sClient, namespace, "app.kubernetes.io/name=hbase", logger) + if err != nil { + return err + } + + deleteRecursive(zkConn, "/hbase", logger) + deleteRecursive(zkConn, "/tx.service", logger) + } else { + _, err := scale.ScaleDown(k8sClient, namespace, "app.kubernetes.io/name=hbase", logger) + if err != nil { + return 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 + } + for _, pvc := range pvcs.Items { + if err := k8sClient.DeletePVC(pvc.Namespace, pvc.Name); err != nil { + logger.Warningf("Failed to delete PVC: %v", err) + } else { + logger.Successf("PVC deleted: %s", pvc.Name) + } + } + + err = scale.ScaleUpFromAnnotations(k8sClient, namespace, "app.kubernetes.io/name=hbase", logger) + if err != nil { + return err + } + + logger.Println() + logger.Infof("Scaled up HBase after deleting PVCs") + + return nil +} From 05a47eb29ae3f56af80de583560fa56f18269a80 Mon Sep 17 00:00:00 2001 From: Frank van Lankvelt Date: Fri, 16 Oct 2026 10:54:53 +0200 Subject: [PATCH 2/4] STS 537: abort restore when deleting pvcs fails, add some logging --- internal/orchestration/restore/hbase.go | 39 ++++++++++++++++--------- 1 file changed, 26 insertions(+), 13 deletions(-) diff --git a/internal/orchestration/restore/hbase.go b/internal/orchestration/restore/hbase.go index ecfc02a..8ed1b97 100644 --- a/internal/orchestration/restore/hbase.go +++ b/internal/orchestration/restore/hbase.go @@ -32,10 +32,14 @@ func deleteRecursive(zkConn *zk.Conn, path string, logger *logger.Logger) { } } -func WipePVCs(ctx context.Context, k8sClient *k8s.Client, namespace, zkQuorum string, logger *logger.Logger) error { +func scaleDownHBase(k8sClient *k8s.Client, namespace string, logger *logger.Logger) error { logger.Println() - logger.Infof("Scaling down HBase and deleting PVCs") + 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 @@ -62,20 +66,30 @@ func WipePVCs(ctx context.Context, k8sClient *k8s.Client, namespace, zkQuorum st } defer zkConn.Close() - _, err = scale.ScaleDown(k8sClient, namespace, "app.kubernetes.io/name=hbase", logger) - if err != nil { + if err = scaleDownHBase(k8sClient, namespace, logger); err != nil { return err } + logger.Println() + logger.Infof("Removing HBase and Tephra nodes from Zookeeper") + deleteRecursive(zkConn, "/hbase", logger) deleteRecursive(zkConn, "/tx.service", logger) + + logger.Successf("Removed nodes from Zookeeper") } else { - _, err := scale.ScaleDown(k8sClient, namespace, "app.kubernetes.io/name=hbase", logger) - if err != nil { + 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", @@ -83,18 +97,17 @@ func WipePVCs(ctx context.Context, k8sClient *k8s.Client, namespace, zkQuorum st 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 { - logger.Warningf("Failed to delete PVC: %v", err) + return err } else { - logger.Successf("PVC deleted: %s", pvc.Name) + logger.Infof(" - %s", pvc.Name) } } - - err = scale.ScaleUpFromAnnotations(k8sClient, namespace, "app.kubernetes.io/name=hbase", logger) - if err != nil { - return err - } + logger.Successf("Deleted PVCs") logger.Println() logger.Infof("Scaled up HBase after deleting PVCs") From 7d118ea50cbfc3c18d80dbba6f5254f48e7968a4 Mon Sep 17 00:00:00 2001 From: Frank van Lankvelt Date: Fri, 16 Oct 2026 12:09:54 +0200 Subject: [PATCH 3/4] STS 537: no else needed --- internal/orchestration/restore/hbase.go | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/internal/orchestration/restore/hbase.go b/internal/orchestration/restore/hbase.go index 8ed1b97..75c5376 100644 --- a/internal/orchestration/restore/hbase.go +++ b/internal/orchestration/restore/hbase.go @@ -103,9 +103,8 @@ func WipePVCs(ctx context.Context, k8sClient *k8s.Client, namespace, zkQuorum st for _, pvc := range pvcs.Items { if err := k8sClient.DeletePVC(pvc.Namespace, pvc.Name); err != nil { return err - } else { - logger.Infof(" - %s", pvc.Name) } + logger.Infof(" - %s", pvc.Name) } logger.Successf("Deleted PVCs") From e1787957654011481eca9cb04e43b73b4f7e3b0c Mon Sep 17 00:00:00 2001 From: Frank van Lankvelt Date: Fri, 16 Oct 2026 14:57:19 +0200 Subject: [PATCH 4/4] STS 537: bow to astra --- cmd/stackgraph/restore.go | 7 ++----- cmd/stackgraphv2/restore.go | 7 ++----- internal/orchestration/restore/hbase.go | 28 +++++++++++++++++-------- 3 files changed, 23 insertions(+), 19 deletions(-) diff --git a/cmd/stackgraph/restore.go b/cmd/stackgraph/restore.go index 73da211..c17fa23 100644 --- a/cmd/stackgraph/restore.go +++ b/cmd/stackgraph/restore.go @@ -107,11 +107,8 @@ func runRestore(appCtx *app.Context) error { // Scale down hbase/hdfs/tephra and delete all pvcs if wipePVCs { - err = restore.WipePVCs(appCtx.Context, appCtx.K8sClient, appCtx.Namespace, appCtx.Config.Stackgraph.Restore.ZookeeperQuorum, appCtx.Logger) - if err != nil { - // Create restore job - appCtx.Logger.Println() - appCtx.Logger.Warningf("Wiping PVCs failed: %s", err) + if err := restore.WipePVCs(appCtx.Context, appCtx.K8sClient, appCtx.Namespace, appCtx.Config.Stackgraph.Restore.ZookeeperQuorum, appCtx.Logger); err != nil { + return err } } diff --git a/cmd/stackgraphv2/restore.go b/cmd/stackgraphv2/restore.go index 64e92d6..5b2dcd5 100644 --- a/cmd/stackgraphv2/restore.go +++ b/cmd/stackgraphv2/restore.go @@ -132,11 +132,8 @@ func liveRestore(appCtx *app.Context) error { // Scale down hbase/hdfs/tephra and delete all pvcs if wipePVCs { - err = restore.WipePVCs(appCtx.Context, appCtx.K8sClient, appCtx.Namespace, appCtx.Config.Stackgraph.Restore.ZookeeperQuorum, appCtx.Logger) - if err != nil { - // Create restore job - appCtx.Logger.Println() - appCtx.Logger.Warningf("Wiping PVCs failed: %s", err) + if err := restore.WipePVCs(appCtx.Context, appCtx.K8sClient, appCtx.Namespace, appCtx.Config.Stackgraph.Restore.ZookeeperQuorum, appCtx.Logger); err != nil { + return err } } diff --git a/internal/orchestration/restore/hbase.go b/internal/orchestration/restore/hbase.go index 75c5376..359a7e0 100644 --- a/internal/orchestration/restore/hbase.go +++ b/internal/orchestration/restore/hbase.go @@ -2,6 +2,7 @@ package restore import ( "context" + "errors" "fmt" "strconv" "strings" @@ -17,19 +18,24 @@ import ( const zkSessionTimeout = 2 * time.Second -func deleteRecursive(zkConn *zk.Conn, path string, logger *logger.Logger) { +func deleteRecursive(zkConn *zk.Conn, path string, logger *logger.Logger) error { children, _, err := zkConn.Children(path) - if err != nil { + switch { + case errors.Is(err, zk.ErrNoNode): + return nil + case err != nil: logger.Warningf("Unable to list znode for %s: %v", path, err) - } else { + default: for _, child := range children { - deleteRecursive(zkConn, path+"/"+child, logger) + if err := deleteRecursive(zkConn, path+"/"+child, logger); err != nil { + return err + } } } - err = zkConn.Delete(path, -1) - if err != nil { - logger.Warningf("Failed to delete znode %s: %v", path, 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 { @@ -73,8 +79,12 @@ func WipePVCs(ctx context.Context, k8sClient *k8s.Client, namespace, zkQuorum st logger.Println() logger.Infof("Removing HBase and Tephra nodes from Zookeeper") - deleteRecursive(zkConn, "/hbase", logger) - deleteRecursive(zkConn, "/tx.service", logger) + 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 {