Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions cmd/stackgraph/restore.go
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ var (
background bool
skipConfirmation bool
skipStackpacks bool
wipePVCs bool
)

func restoreCmd(globalFlags *config.CLIGlobalFlags) *cobra.Command {
Expand All @@ -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")

Expand Down Expand Up @@ -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 {
Expand Down
9 changes: 9 additions & 0 deletions cmd/stackgraphv2/restore.go
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@ var (
useLatest bool
skipConfirmation bool
skipStackpacks bool
wipePVCs bool
)

func restoreCmd(globalFlags *config.CLIGlobalFlags) *cobra.Command {
Expand All @@ -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")

Expand Down Expand Up @@ -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 {
Expand Down
1 change: 1 addition & 0 deletions go.mod
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 2 additions & 0 deletions go.sum
Original file line number Diff line number Diff line change
Expand Up @@ -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=
Expand Down
125 changes: 125 additions & 0 deletions internal/orchestration/restore/hbase.go
Original file line number Diff line number Diff line change
@@ -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",
Comment thread
fvlankvelt marked this conversation as resolved.
})
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 {
Comment thread
fvlankvelt marked this conversation as resolved.
return err
}
logger.Infof(" - %s", pvc.Name)
}
logger.Successf("Deleted PVCs")

logger.Println()
logger.Infof("Scaled up HBase after deleting PVCs")

return nil
}
Loading