Add PVC as storage Support for Snapshot Download

go.mod

@@ -12,7 +12,7 @@ require (
 	gomodules.xyz/go-sh v0.3.0
 	gomodules.xyz/logs v0.0.7
 	gomodules.xyz/pointer v0.1.0
-	gomodules.xyz/restic v0.5.0
+	gomodules.xyz/restic v0.5.1
 	gomodules.xyz/runtime v0.3.0
 	gomodules.xyz/x v0.0.17
 	k8s.io/api v0.34.3
@@ -28,7 +28,7 @@ require (
 	kmodules.xyz/offshoot-api v0.34.0
 	kmodules.xyz/prober v0.34.0
 	kubedb.dev/apimachinery v0.63.0
-	kubestash.dev/apimachinery v0.29.0
+	kubestash.dev/apimachinery v0.29.1-0.20260727105629-5cb309a9aa2b
 	sigs.k8s.io/controller-runtime v0.22.4
 	sigs.k8s.io/yaml v1.6.0
 	stash.appscode.dev/apimachinery v0.42.1

go.sum

@@ -778,8 +778,8 @@ gomodules.xyz/mergo v0.3.13 h1:q6cL/MMXZH/MrR2+yjSihFFq6UifXqjwaqI48B6cMEM=
 gomodules.xyz/mergo v0.3.13/go.mod h1:F/2rKC7j0URTnHUKDiTiLcGdLMhdv8jK2Za3cRTUVmc=
 gomodules.xyz/pointer v0.1.0 h1:sG2UKrYVSo6E3r4itAjXfPfe4fuXMi0KdyTHpR3vGCg=
 gomodules.xyz/pointer v0.1.0/go.mod h1:sPLsC0+yLTRecUiC5yVlyvXhZ6LAGojNCRWNNqoplvo=
-gomodules.xyz/restic v0.5.0 h1:wW0U53kZzdv4V1SZgl/x6N4uov17O5RXkV94e/sNT8E=
-gomodules.xyz/restic v0.5.0/go.mod h1:/2xLB6rUUyS2o2c/wD06Q5/3e04vW5ScyCLRnbEGXcs=
+gomodules.xyz/restic v0.5.1 h1:kJXBmT/9yqqYmn9MyG6lgCsJKniQFrJvtEaYQnJx0fQ=
+gomodules.xyz/restic v0.5.1/go.mod h1:/2xLB6rUUyS2o2c/wD06Q5/3e04vW5ScyCLRnbEGXcs=
 gomodules.xyz/runtime v0.3.0 h1:Fgf3fjIE3xY/sswO73iRBeR3mundZAjlY42fQPigPR0=
 gomodules.xyz/runtime v0.3.0/go.mod h1:lJuiayVYjz8LWDwKhbDqFzUrXqr1btLbJS5/lKDz1YU=
 gomodules.xyz/sets v0.2.0/go.mod h1:jKgNp01/iDs+svOWXaPk5cKP3VXy0mWUoTF/ore+aMc=
@@ -912,8 +912,8 @@ kubeops.dev/petset v0.0.15 h1:iwTRFAp0RNw0A87sw2c97UZ6WIA9H/nhJBpDhXLa7fk=
 kubeops.dev/petset v0.0.15/go.mod h1:sw96WiXfzhpmKpXj4a5AdmEHs0Bx4QMhf+iW15zY4Gg=
 kubeops.dev/sidekick v0.0.12 h1:pmUjQLZDKxgREiM6z0PogLR1aDbgvkE9jRjbxG6dEt0=
 kubeops.dev/sidekick v0.0.12/go.mod h1:RU7QH3E8DOLw15rBYlOOJSyczuwAnVVtYyZjJb00UB8=
-kubestash.dev/apimachinery v0.29.0 h1:yNurTDPthRm6Z5f+DboT5jdjfm9VkE0oMuTemLF0s24=
-kubestash.dev/apimachinery v0.29.0/go.mod h1:Gp3D0Ah1meSf+KAB+Nwh3sKTOO5lhswqk9KDNhBM0Eg=
+kubestash.dev/apimachinery v0.29.1-0.20260727105629-5cb309a9aa2b h1:KGrytfzwF7bKXDPqHb2Mzi7q4q9PNFvegAZKGfr73wU=
+kubestash.dev/apimachinery v0.29.1-0.20260727105629-5cb309a9aa2b/go.mod h1:xRL87YFpBMKgpHOZXh2gSSNgxW1qLNaNPHI9FrbK3y0=
 open-cluster-management.io/api v1.2.0 h1:+yeQgJiErrur5S4s205UM37EcZ2XbC9pFSm0xgV5/hU=
 open-cluster-management.io/api v1.2.0/go.mod h1:YcmA6SpGEekIMxdoeVIIyOaBhMA6ImWRLXP4g8n8T+4=
 rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=

pkg/add_key.go

@@ -64,7 +64,7 @@ func NewCmdAddKey(opt *keyOptions) *cobra.Command {
 			}
 
 			if backupStorage.Spec.Storage.Local != nil {
-				if !backupStorage.LocalNetworkVolume() {
+				if !backupStorage.LocalNetworkVolume() && !backupStorage.LocalBackendPVC() {
 					return fmt.Errorf("unsupported type of local backend provided")
 				}
 

pkg/constants.go

@@ -22,6 +22,7 @@ import "time"
 const (
 	CmdKubectl = "kubectl"
 	CmdDocker  = "docker"
+	CmdTar     = "tar"
 
 	ScratchDir          = "/tmp/scratch"
 	DestinationDir      = "/tmp/destination"
@@ -33,6 +34,12 @@ const (
 	ResticImage   = "ghcr.io/appscode-images/restic:0.18.1"
 	EnvHttpProxy  = "HTTP_PROXY"
 	EnvHttpsProxy = "HTTPS_PROXY"
+
+	// EnvCopyMode selects how downloaded data is copied from the pod to the
+	// local destination: "tar" (default) streams a tarball through the exec
+	// API, "cp" uses `kubectl cp`.
+	EnvCopyMode = "KUBESTASH_COPY_MODE"
+	CopyModeCP  = "cp"
 )
 
 // Constants for debugging

pkg/download.go

@@ -103,9 +103,20 @@ func NewCmdDownload(clientGetter genericclioptions.RESTClientGetter) *cobra.Comm
 				return err
 			}
 
+			klog.Infof("Resolved Snapshot %s/%s -> Repository %s -> BackupStorage %s/%s (provider: %s)",
+				srcNamespace, snapshotName, repository.Name,
+				backupStorage.Namespace, backupStorage.Name, backupStorage.Spec.Storage.Provider)
+
 			if backupStorage.Spec.Storage.Local != nil {
-				if !backupStorage.LocalNetworkVolume() {
-					return fmt.Errorf("unsupported type of local backend provided")
+				switch {
+				case backupStorage.LocalNetworkVolume():
+					klog.Infof("Local backend type: NFS (server: %s)", backupStorage.Spec.Storage.Local.NFS.Server)
+				case backupStorage.LocalBackendPVC():
+					klog.Infof("Local backend type: PersistentVolumeClaim (claim: %s, mountPath: %s)",
+						backupStorage.Spec.Storage.Local.PersistentVolumeClaim.ClaimName, backupStorage.Spec.Storage.Local.MountPath)
+				default:
+					return fmt.Errorf("unsupported type of local backend provided: BackupStorage %s/%s uses a local volume source that is neither NFS nor PersistentVolumeClaim",
+						backupStorage.Namespace, backupStorage.Name)
 				}
 
 				accessorPod, err := getLocalBackendAccessorPod(repository.Spec.StorageRef)
@@ -127,6 +138,7 @@ func NewCmdDownload(clientGetter genericclioptions.RESTClientGetter) *cobra.Comm
 			}
 
 			if yes {
+				klog.Infof("Workload identity detected on operator pod %s/%s; downloading via operator pod", operatorPod.Namespace, operatorPod.Name)
 				return downloadOpt.runRestoreViaPod(&operatorPod, snapshotName)
 			}
 
@@ -226,15 +238,15 @@ func NewCmdDownload(clientGetter genericclioptions.RESTClientGetter) *cobra.Comm
 
 func (opt *downloadOptions) runRestoreViaPod(pod *core.Pod, snapshotName string) error {
 	if err := opt.runCmdViaPod(pod, snapshotName); err != nil {
-		return err
+		return fmt.Errorf("failed to run download inside pod %s/%s: %w", pod.Namespace, pod.Name, err)
 	}
 
 	if err := opt.copyDownloadedDataToDestination(pod); err != nil {
-		return err
+		return fmt.Errorf("failed to copy downloaded data from pod %s/%s to %s: %w", pod.Namespace, pod.Name, opt.destinationDir, err)
 	}
 
 	if err := opt.clearDataFromPod(pod); err != nil {
-		return err
+		return fmt.Errorf("failed to clean up temporary data from pod %s/%s: %w", pod.Namespace, pod.Name, err)
 	}
 
 	klog.Infof("Snapshot %s/%s restored in path %s", srcNamespace, snapshotName, opt.destinationDir)
@@ -265,16 +277,26 @@ func (opt *downloadOptions) runCmdViaPod(pod *core.Pod, snapshotName string) err
 	if err != nil {
 		return err
 	}
-	klog.Infoln("Output:", out)
+	if out != "" {
+		klog.Infoln("Output:", out)
+	}
 	return nil
 }
 
 func (opt *downloadOptions) copyDownloadedDataToDestination(pod *core.Pod) error {
-	_, err := exec.Command(CmdKubectl, "cp", fmt.Sprintf("%s/%s:%s", pod.Namespace, pod.Name, SnapshotDownloadDir), opt.destinationDir).CombinedOutput()
-	if err != nil {
-		return err
+	if strings.EqualFold(os.Getenv(EnvCopyMode), CopyModeCP) {
+		cmd := exec.Command(CmdKubectl, "cp", fmt.Sprintf("%s/%s:%s", pod.Namespace, pod.Name, SnapshotDownloadDir), opt.destinationDir)
+		klog.Infof("Copying downloaded data with: %v", cmd.Args)
+		out, err := cmd.CombinedOutput()
+		if len(out) > 0 {
+			klog.Infoln("kubectl cp output:", string(out))
+		}
+		if err != nil {
+			return fmt.Errorf("kubectl cp failed: %w, output: %q", err, string(out))
+		}
+		return nil
 	}
-	return nil
+	return copyDataFromPodViaTar(opt.restConfig, pod, SnapshotDownloadDir, opt.destinationDir)
 }
 
 func (opt *downloadOptions) clearDataFromPod(pod *core.Pod) error {

pkg/list_key.go

@@ -64,7 +64,7 @@ func NewCmdListKey(opt *keyOptions) *cobra.Command {
 			}
 
 			if backupStorage.Spec.Storage.Local != nil {
-				if !backupStorage.LocalNetworkVolume() {
+				if !backupStorage.LocalNetworkVolume() && !backupStorage.LocalBackendPVC() {
 					return fmt.Errorf("unsupported type of local backend provided")
 				}
 

pkg/remove_key.go

@@ -71,7 +71,7 @@ func NewCmdRemoveKey(opt *keyOptions) *cobra.Command {
 				opt.paths = idWithPath[1:]
 
 				if backupStorage.Spec.Storage.Local != nil {
-					if !backupStorage.LocalNetworkVolume() {
+					if !backupStorage.LocalNetworkVolume() && !backupStorage.LocalBackendPVC() {
 						return fmt.Errorf("unsupported type of local backend provided")
 					}
 

pkg/restore.go

@@ -112,7 +112,7 @@ func NewCmdManifestRestore(clientGetter genericclioptions.RESTClientGetter) *cob
 			}()
 
 			if backupStorage.Spec.Storage.Local != nil {
-				if !backupStorage.LocalNetworkVolume() {
+				if !backupStorage.LocalNetworkVolume() && !backupStorage.LocalBackendPVC() {
 					return fmt.Errorf("unsupported type of local backend provided")
 				}
 				accessorPod, err := getLocalBackendAccessorPod(repository.Spec.StorageRef)
@@ -353,11 +353,11 @@ func (opt *options) runCmdViaPod(pod *core.Pod, snapshotName string) error {
 }
 
 func (opt *options) copyDownloadedDataToDestination(pod *core.Pod) error {
-	_, err := exec.Command(CmdKubectl, "cp", fmt.Sprintf("%s/%s:%s", pod.Namespace, pod.Name, SnapshotDownloadDir), opt.DataDir).CombinedOutput()
-	if err != nil {
+	if strings.EqualFold(os.Getenv(EnvCopyMode), CopyModeCP) {
+		_, err := exec.Command(CmdKubectl, "cp", fmt.Sprintf("%s/%s:%s", pod.Namespace, pod.Name, SnapshotDownloadDir), opt.DataDir).CombinedOutput()
 		return err
 	}
-	return nil
+	return copyDataFromPodViaTar(opt.Config, pod, SnapshotDownloadDir, opt.DataDir)
 }
 
 func (opt *options) clearDataFromPod(pod *core.Pod) error {

pkg/unlock.go

@@ -90,7 +90,7 @@ func NewCmdUnlockRepository(clientGetter genericclioptions.RESTClientGetter) *co
 			}
 
 			if backupStorage.Spec.Storage.Local != nil {
-				if !backupStorage.LocalNetworkVolume() {
+				if !backupStorage.LocalNetworkVolume() && !backupStorage.LocalBackendPVC() {
 					return fmt.Errorf("unsupported type of local backend provided")
 				}
 

pkg/update_key.go

@@ -61,7 +61,7 @@ func NewCmdUpdateKey(opt *keyOptions) *cobra.Command {
 			}
 
 			if backupStorage.Spec.Storage.Local != nil {
-				if !backupStorage.LocalNetworkVolume() {
+				if !backupStorage.LocalNetworkVolume() && !backupStorage.LocalBackendPVC() {
 					return fmt.Errorf("unsupported type of local backend provided")
 				}
 

pkg/util.go

@@ -20,7 +20,9 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"io"
 	"os"
+	"os/exec"
 	"strconv"
 	"strings"
 	"time"
@@ -302,17 +304,30 @@ func getLocalBackendAccessorPod(obj kmapi.ObjectReference) (*core.Pod, error) {
 		return nil, err
 	}
 
+	var skipped []string
 	for i := range pods.Items {
+		if pods.Items[i].Status.Phase != core.PodRunning ||
+			pods.Items[i].DeletionTimestamp != nil {
+			skipped = append(skipped, fmt.Sprintf("%s (phase=%s, terminating=%t)",
+				pods.Items[i].Name, pods.Items[i].Status.Phase, pods.Items[i].DeletionTimestamp != nil))
+			continue
+		}
 		if hasVolume(pods.Items[i].Spec.Volumes, obj.Name) {
 			for _, c := range pods.Items[i].Spec.Containers {
 				if hasVolumeMount(c.VolumeMounts, obj.Name) {
+					klog.Infof("Using local backend accessor pod %s/%s (volume: %s)",
+						pods.Items[i].Namespace, pods.Items[i].Name, obj.Name)
 					return &pods.Items[i], nil
 				}
 			}
 		}
+		skipped = append(skipped, fmt.Sprintf("%s (running, but does not mount volume %q)", pods.Items[i].Name, obj.Name))
 	}
 
-	return nil, fmt.Errorf("no local backend accessor pod found for BackupStorage: %s/%s", obj.Namespace, obj.Name)
+	return nil, fmt.Errorf("no usable local backend accessor pod found for BackupStorage %s/%s: "+
+		"listed %d pod(s) with label %s=%s in namespace %s, skipped: %v. "+
+		"Make sure the KubeStash operator has created the accessor Deployment and its pod is Running",
+		obj.Namespace, obj.Name, len(pods.Items), apis.KubeStashApp, apis.KubeStashNetVolAccessor, obj.Namespace, skipped)
 }
 
 func hasVolume(volumes []core.Volume, name string) bool {
@@ -369,7 +384,7 @@ func execOnPod(config *rest.Config, pod *core.Pod, command []string) (string, er
 		execErr bytes.Buffer
 	)
 
-	klog.V(2).Infof("Executing command %v on pod %s/%s", command, pod.Namespace, pod.Name)
+	klog.Infof("Executing command %v on pod %s/%s (container: %s)", command, pod.Namespace, pod.Name, getContainerName(pod))
 
 	kubeClient, err := kubernetes.NewForConfig(config)
 	if err != nil {
@@ -394,13 +409,19 @@ func execOnPod(config *rest.Config, pod *core.Pod, command []string) (string, er
 		return "", fmt.Errorf("failed to init executor: %v", err)
 	}
 
+	// Tty must stay disabled: a tty merges stderr into stdout and can drop
+	// output of a fast-exiting process, hiding the actual error.
 	err = executor.StreamWithContext(context.Background(), remotecommand.StreamOptions{
 		Stdout: &execOut,
 		Stderr: &execErr,
-		Tty:    true,
 	})
 	if err != nil {
-		return "", fmt.Errorf("could not execute: %v, reason: %s", err, execErr.String())
+		return "", fmt.Errorf("failed to execute command %v on pod %s/%s: %w, output: %q, stderr: %q",
+			command, pod.Namespace, pod.Name, err, execOut.String(), execErr.String())
+	}
+
+	if execErr.Len() > 0 {
+		klog.Warningf("command %v on pod %s/%s wrote to stderr: %s", command, pod.Namespace, pod.Name, execErr.String())
 	}
 
 	return execOut.String(), nil
@@ -412,3 +433,81 @@ func getContainerName(pod *core.Pod) string {
 	}
 	return apis.KubeStashContainer
 }
+
+func execOnPodWithStreams(config *rest.Config, pod *core.Pod, command []string, stdout, stderr io.Writer) error {
+	klog.Infof("Executing command %v on pod %s/%s", command, pod.Namespace, pod.Name)
+
+	kubeClient, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		return err
+	}
+
+	req := kubeClient.CoreV1().RESTClient().Post().
+		Resource("pods").
+		Name(pod.Name).
+		Namespace(pod.Namespace).
+		SubResource("exec").
+		Timeout(5 * time.Minute)
+	req.VersionedParams(&core.PodExecOptions{
+		Container: getContainerName(pod),
+		Command:   command,
+		Stdout:    true,
+		Stderr:    true,
+	}, scheme.ParameterCodec)
+
+	executor, err := remotecommand.NewSPDYExecutor(config, "POST", req.URL())
+	if err != nil {
+		return fmt.Errorf("failed to init executor: %v", err)
+	}
+
+	// Tty must stay disabled; a tty would mangle the binary stream.
+	return executor.StreamWithContext(context.Background(), remotecommand.StreamOptions{
+		Stdout: stdout,
+		Stderr: stderr,
+	})
+}
+
+// copyDataFromPodViaTar streams srcDir from the pod as a gzipped tarball
+// through the exec API and extracts it into destDir, avoiding the extra
+// kubectl binary dependency and temp files of `kubectl cp`.
+func copyDataFromPodViaTar(config *rest.Config, pod *core.Pod, srcDir, destDir string) error {
+	podCmd := []string{CmdTar, "-C", srcDir, "-czf", "-", "."}
+	klog.Infof("Streaming %s from pod %s/%s into %s (pod command: %v)", srcDir, pod.Namespace, pod.Name, destDir, podCmd)
+
+	localTar := exec.Command(CmdTar, "-C", destDir, "-xzf", "-")
+	var localTarStderr bytes.Buffer
+	localTar.Stderr = &localTarStderr
+	localTarStdin, err := localTar.StdinPipe()
+	if err != nil {
+		return fmt.Errorf("failed to pipe local tar stdin: %w", err)
+	}
+	if err := localTar.Start(); err != nil {
+		return fmt.Errorf("failed to start local tar %v: %w", localTar.Args, err)
+	}
+
+	var execStderr bytes.Buffer
+	execErr := execOnPodWithStreams(config, pod, podCmd, localTarStdin, &execStderr)
+	if closeErr := localTarStdin.Close(); closeErr != nil && execErr == nil {
+		execErr = closeErr
+	}
+	tarErr := localTar.Wait()
+
+	if execStderr.Len() > 0 {
+		klog.Warningf("tar on pod %s/%s wrote to stderr: %s", pod.Namespace, pod.Name, execStderr.String())
+	}
+	if localTarStderr.Len() > 0 {
+		klog.Warningf("local tar wrote to stderr: %s", localTarStderr.String())
+	}
+
+	if execErr != nil && tarErr != nil {
+		return fmt.Errorf("failed to stream data from pod %s/%s: %w (pod stderr: %q); local tar also failed: %v (local stderr: %q)",
+			pod.Namespace, pod.Name, execErr, execStderr.String(), tarErr, localTarStderr.String())
+	}
+	if execErr != nil {
+		return fmt.Errorf("failed to stream data from pod %s/%s: %w, stderr: %q", pod.Namespace, pod.Name, execErr, execStderr.String())
+	}
+	if tarErr != nil {
+		return fmt.Errorf("local tar failed: %w, stderr: %q", tarErr, localTarStderr.String())
+	}
+	return nil
+}

pkg/view.go

@@ -120,7 +120,7 @@ func NewCmdManifestView(clientGetter genericclioptions.RESTClientGetter) *cobra.
 			viewOpt.initialize()
 
 			if backupStorage.Spec.Storage.Local != nil {
-				if !backupStorage.LocalNetworkVolume() {
+				if !backupStorage.LocalNetworkVolume() && !backupStorage.LocalBackendPVC() {
 					return fmt.Errorf("unsupported type of local backend provided")
 				}
 				accessorPod, err := getLocalBackendAccessorPod(repository.Spec.StorageRef)

vendor/gomodules.xyz/restic/setup.go

@@ -81,7 +81,7 @@ func (w *ResticWrapper) setupEnvsForBackend(b *Backend) error {
 
 	switch b.Provider {
 	case storage.ProviderLocal:
-		b.Envs[RESTIC_REPOSITORY] = fmt.Sprintf("%s/%s", b.Bucket, b.Directory)
+		b.Envs[RESTIC_REPOSITORY] = filepath.Join(b.Bucket, b.Directory)
 
 	case storage.ProviderS3:
 		b.Envs[RESTIC_REPOSITORY] = fmt.Sprintf("s3:%s/%s", b.Endpoint, filepath.Join(b.Bucket, b.Prefix, b.Directory))

vendor/kubestash.dev/apimachinery/pkg/blob/blob.go

@@ -490,7 +490,8 @@ func (b *Blob) getS3Config(ctx context.Context, debug bool) (aws2.Config, error)
 
 	if debug {
 		loadOptions = append(loadOptions, config.WithClientLogMode(
-			aws2.LogRetries|aws2.LogRequestWithBody|aws2.LogResponseWithBody))
+			aws2.LogRetries|aws2.LogRequestWithBody|aws2.LogResponseWithBody,
+		))
 	}
 
 	if b.backupStorage.Spec.Storage.S3.SecretName != "" {