diff --git a/go.mod b/go.mod index b579b51b..1cce979f 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 809c5192..19d3f771 100644 --- a/go.sum +++ b/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= diff --git a/pkg/add_key.go b/pkg/add_key.go index 5986b18e..7099bce1 100644 --- a/pkg/add_key.go +++ b/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") } diff --git a/pkg/constants.go b/pkg/constants.go index c2df3f0a..4c5bac7a 100644 --- a/pkg/constants.go +++ b/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 diff --git a/pkg/download.go b/pkg/download.go index fd47ca7c..ac542736 100644 --- a/pkg/download.go +++ b/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 { diff --git a/pkg/list_key.go b/pkg/list_key.go index 4ef9e1f9..bfb30dc5 100644 --- a/pkg/list_key.go +++ b/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") } diff --git a/pkg/remove_key.go b/pkg/remove_key.go index 51c6876e..6d9f0614 100644 --- a/pkg/remove_key.go +++ b/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") } diff --git a/pkg/restore.go b/pkg/restore.go index 703382c2..010eb6b5 100644 --- a/pkg/restore.go +++ b/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 { diff --git a/pkg/unlock.go b/pkg/unlock.go index 871c4039..2beb1759 100644 --- a/pkg/unlock.go +++ b/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") } diff --git a/pkg/update_key.go b/pkg/update_key.go index 031ac353..90fa8795 100644 --- a/pkg/update_key.go +++ b/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") } diff --git a/pkg/util.go b/pkg/util.go index 44a5ede4..2c6e4fbc 100644 --- a/pkg/util.go +++ b/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 +} diff --git a/pkg/view.go b/pkg/view.go index a82857a8..00f1af27 100644 --- a/pkg/view.go +++ b/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) diff --git a/vendor/gomodules.xyz/restic/setup.go b/vendor/gomodules.xyz/restic/setup.go index 41e3896b..00a511b6 100644 --- a/vendor/gomodules.xyz/restic/setup.go +++ b/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)) diff --git a/vendor/kubestash.dev/apimachinery/pkg/blob/blob.go b/vendor/kubestash.dev/apimachinery/pkg/blob/blob.go index b8a329f3..7f7bf70b 100644 --- a/vendor/kubestash.dev/apimachinery/pkg/blob/blob.go +++ b/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 != "" { diff --git a/vendor/modules.txt b/vendor/modules.txt index 85421c24..bd5572b8 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -997,7 +997,7 @@ gomodules.xyz/mergo # gomodules.xyz/pointer v0.1.0 ## explicit; go 1.15 gomodules.xyz/pointer -# gomodules.xyz/restic v0.5.0 +# gomodules.xyz/restic v0.5.1 ## explicit; go 1.25.0 gomodules.xyz/restic # gomodules.xyz/runtime v0.3.0 @@ -1880,7 +1880,7 @@ kubeops.dev/petset/crds kubeops.dev/sidekick/apis/apps kubeops.dev/sidekick/apis/apps/v1alpha1 kubeops.dev/sidekick/crds -# kubestash.dev/apimachinery v0.29.0 +# kubestash.dev/apimachinery v0.29.1-0.20260727105629-5cb309a9aa2b ## explicit; go 1.25.0 kubestash.dev/apimachinery/apis kubestash.dev/apimachinery/apis/addons/v1alpha1