mirrors/velero
1
0
Fork 0
mirror of https://github.com/vmware-tanzu/velero.git synced 2026-07-28 19:12:43 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge branch 'main' into block-data-mover-backup-expose

Browse Source
This commit is contained in:
Lyndon-Li
2026-06-02 16:56:40 +08:00
parent e4ecf26b33 5cc0398662
commit 2e1ab5ab62
19 changed files with 327 additions and 1217 deletions
Download Patch File Download Diff File Expand all files Collapse all files
changelogs/unreleased/9840-adam-jian-zhang
+1
View File
@@ -0,0 +1 @@
Fix issue #9812, validate ClusterScopedFilterPolicy and NamespacedFilterPolicy incompatible with legacy filters
changelogs/unreleased/9861-Lyndon-Li
+1
View File
@@ -0,0 +1 @@
Remove restic command package
changelogs/unreleased/9864-Lyndon-Li
+1
View File
@@ -0,0 +1 @@
Add cbt service parameters to node-agent-config for block data mover
pkg/cmd/cli/nodeagent/server.go
+7
View File
@@ -385,6 +385,12 @@ func (s *nodeAgentServer) run() {
s.logger.Info("Backup repo config is not provided, using default values for cache volume configs")
}
var csiSnapshotMetadataServiceConfigs *velerotypes.CSISnapshotMetadataService
if s.dataPathConfigs != nil && s.dataPathConfigs.CSISnapshotMetadataServiceConfigs != nil {
csiSnapshotMetadataServiceConfigs = s.dataPathConfigs.CSISnapshotMetadataServiceConfigs
s.logger.Infof("Using CSI snapshot metadata service config %v", s.dataPathConfigs.CSISnapshotMetadataServiceConfigs)
}
pvbReconciler := controller.NewPodVolumeBackupReconciler(
s.mgr.GetClient(),
s.mgr,
@@ -447,6 +453,7 @@ func (s *nodeAgentServer) run() {
dataMovePriorityClass,
podLabels,
podAnnotations,
csiSnapshotMetadataServiceConfigs,
)
if err := dataUploadReconciler.SetupWithManager(s.mgr); err != nil {
s.logger.WithError(err).Fatal("Unable to create the data upload controller")
pkg/controller/backup_controller.go
+7
View File
@@ -595,6 +595,13 @@ func (b *backupReconciler) prepareBackupRequest(ctx context.Context, backup *vel
request.Status.ValidationErrors = append(request.Status.ValidationErrors, "include-resources, exclude-resources and include-cluster-resources are old filter parameters.\n"+
"They cannot be used with include-exclude policies.")
}
// namespacedFilterPolicies and clusterScopedFilterPolicy incompatible with old-style filters
if resourcePolicies != nil &&
(len(resourcePolicies.GetNamespacedFilterPolicies()) > 0 || resourcePolicies.GetClusterScopedFilterPolicy() != nil) &&
collections.UseOldResourceFilters(request.Spec) {
request.Status.ValidationErrors = append(request.Status.ValidationErrors, "include-resources, exclude-resources and include-cluster-resources are old filter parameters.\n"+
"They cannot be used with namespace-scoped or fine-grained global filter policies.")
}
request.ResPolicies = resourcePolicies
return request
}
pkg/controller/backup_controller_test.go
+236
View File
@@ -21,6 +21,7 @@ import (
"fmt"
"io"
"reflect"
"slices"
"sort"
"strings"
"testing"
@@ -34,6 +35,7 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
corev1api "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
@@ -2020,3 +2022,237 @@ func TestPatchResourceWorksWithStatus(t *testing.T) {
})
}
}
// TestPrepareBackupRequest_NamespacedFilterPoliciesIncompatibleWithOldFilters verifies
// that a backup referencing a ResourcePolicy ConfigMap with namespacedFilterPolicies
// produces a validation error when old-style resource filters are also set on the spec.
func TestPrepareBackupRequest_NamespacedFilterPoliciesIncompatibleWithOldFilters(t *testing.T) {
formatFlag := logging.FormatText
logger := logging.DefaultLogger(logrus.DebugLevel, formatFlag)
policyYAML := `version: v1
namespacedFilterPolicies:
- namespaces: ["production"]
resourceFilters:
- kinds: ["Deployment"]
names: ["api-server"]
`
policyConfigMap := &corev1api.ConfigMap{
ObjectMeta: metav1.ObjectMeta{
Name: "my-filter-policy",
Namespace: velerov1api.DefaultNamespace,
},
Data: map[string]string{"policy": policyYAML},
}
backup := defaultBackup().IncludedResources("deployments").Result()
backup.Spec.ResourcePolicy = &corev1api.TypedLocalObjectReference{
Kind: "configmap",
Name: "my-filter-policy",
}
fakeClient := velerotest.NewFakeControllerRuntimeClient(t, policyConfigMap)
apiServer := velerotest.NewAPIServer(t)
discoveryHelper, err := discovery.NewHelper(apiServer.DiscoveryClient, logger)
require.NoError(t, err)
c := &backupReconciler{
logger: logger,
discoveryHelper: discoveryHelper,
kbClient: fakeClient,
clock: &clock.RealClock{},
formatFlag: formatFlag,
}
res := c.prepareBackupRequest(ctx, backup, logger)
require.NotEmpty(t, res.Status.ValidationErrors)
hasTargetError := slices.ContainsFunc(res.Status.ValidationErrors, func(e string) bool {
return strings.Contains(e, "namespace-scoped or fine-grained global filter policies")
})
assert.True(t, hasTargetError, "expected validation error about namespacedFilterPolicies incompatibility with old-style filters, got: %v", res.Status.ValidationErrors)
}
// TestPrepareBackupRequest_ClusterScopedFilterPolicyIncompatibleWithOldFilters verifies
// that a backup referencing a ResourcePolicy ConfigMap with clusterScopedFilterPolicy
// produces a validation error when old-style resource filters are also set on the spec.
func TestPrepareBackupRequest_ClusterScopedFilterPolicyIncompatibleWithOldFilters(t *testing.T) {
formatFlag := logging.FormatText
logger := logging.DefaultLogger(logrus.DebugLevel, formatFlag)
policyYAML := `version: v1
clusterScopedFilterPolicy:
resourceFilters:
- kinds: ["ClusterRole"]
names: ["my-app-*"]
`
policyConfigMap := &corev1api.ConfigMap{
ObjectMeta: metav1.ObjectMeta{
Name: "my-cluster-filter-policy",
Namespace: velerov1api.DefaultNamespace,
},
Data: map[string]string{"policy": policyYAML},
}
backup := defaultBackup().IncludedResources("clusterroles").Result()
backup.Spec.ResourcePolicy = &corev1api.TypedLocalObjectReference{
Kind: "configmap",
Name: "my-cluster-filter-policy",
}
fakeClient := velerotest.NewFakeControllerRuntimeClient(t, policyConfigMap)
apiServer := velerotest.NewAPIServer(t)
discoveryHelper, err := discovery.NewHelper(apiServer.DiscoveryClient, logger)
require.NoError(t, err)
c := &backupReconciler{
logger: logger,
discoveryHelper: discoveryHelper,
kbClient: fakeClient,
clock: &clock.RealClock{},
formatFlag: formatFlag,
}
res := c.prepareBackupRequest(ctx, backup, logger)
require.NotEmpty(t, res.Status.ValidationErrors)
hasClusterError := slices.ContainsFunc(res.Status.ValidationErrors, func(e string) bool {
return strings.Contains(e, "namespace-scoped or fine-grained global filter policies")
})
assert.True(t, hasClusterError, "expected validation error about clusterScopedFilterPolicy incompatibility with old-style filters, got: %v", res.Status.ValidationErrors)
}
const (
namespacedFilterPolicyYAML = `version: v1
namespacedFilterPolicies:
- namespaces: ["production"]
resourceFilters:
- kinds: ["Deployment"]
names: ["api-server"]
`
clusterScopedFilterPolicyYAML = `version: v1
clusterScopedFilterPolicy:
resourceFilters:
- kinds: ["ClusterRole"]
names: ["my-app-*"]
`
bothFilterPoliciesYAML = `version: v1
namespacedFilterPolicies:
- namespaces: ["production"]
resourceFilters:
- kinds: ["Deployment"]
names: ["api-server"]
clusterScopedFilterPolicy:
resourceFilters:
- kinds: ["ClusterRole"]
names: ["my-app-*"]
`
)
// TestPrepareBackupRequest_FilterPoliciesWithNewFilters verifies that backups referencing
// a ResourcePolicy ConfigMap with namespacedFilterPolicies and/or clusterScopedFilterPolicy
// succeed when old-style resource filters are not set on the spec.
func TestPrepareBackupRequest_FilterPoliciesWithNewFilters(t *testing.T) {
tests := []struct {
name string
policyYAML string
policyConfigMapName string
backup *velerov1api.Backup
expectNamespacedPolicies int
expectClusterScopedPolicy bool
}{
{
name: "namespacedFilterPolicies only",
policyYAML: namespacedFilterPolicyYAML,
policyConfigMapName: "my-filter-policy",
backup: defaultBackup().StorageLocation("loc-1").Result(),
expectNamespacedPolicies: 1,
},
{
name: "clusterScopedFilterPolicy only",
policyYAML: clusterScopedFilterPolicyYAML,
policyConfigMapName: "my-cluster-filter-policy",
backup: defaultBackup().StorageLocation("loc-1").Result(),
expectClusterScopedPolicy: true,
},
{
name: "both filter policies",
policyYAML: bothFilterPoliciesYAML,
policyConfigMapName: "my-combined-filter-policy",
backup: defaultBackup().StorageLocation("loc-1").Result(),
expectNamespacedPolicies: 1,
expectClusterScopedPolicy: true,
},
{
name: "with new-style spec filters",
policyYAML: bothFilterPoliciesYAML,
policyConfigMapName: "my-combined-filter-policy",
backup: defaultBackup().
StorageLocation("loc-1").
IncludedNamespaceScopedResources("deployments").
IncludedClusterScopedResources("clusterroles").
Result(),
expectNamespacedPolicies: 1,
expectClusterScopedPolicy: true,
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
formatFlag := logging.FormatText
logger := logging.DefaultLogger(logrus.DebugLevel, formatFlag)
policyConfigMap := &corev1api.ConfigMap{
ObjectMeta: metav1.ObjectMeta{
Name: test.policyConfigMapName,
Namespace: velerov1api.DefaultNamespace,
},
Data: map[string]string{"policy": test.policyYAML},
}
test.backup.Spec.ResourcePolicy = &corev1api.TypedLocalObjectReference{
Kind: "configmap",
Name: test.policyConfigMapName,
}
backupLocation := builder.ForBackupStorageLocation(velerov1api.DefaultNamespace, "loc-1").
Phase(velerov1api.BackupStorageLocationPhaseAvailable).Result()
fakeClient := velerotest.NewFakeControllerRuntimeClient(t, backupLocation, policyConfigMap)
apiServer := velerotest.NewAPIServer(t)
discoveryHelper, err := discovery.NewHelper(apiServer.DiscoveryClient, logger)
require.NoError(t, err)
c := &backupReconciler{
logger: logger,
discoveryHelper: discoveryHelper,
kbClient: fakeClient,
clock: &clock.RealClock{},
formatFlag: formatFlag,
}
res := c.prepareBackupRequest(ctx, test.backup, logger)
defer res.WorkerPool.Stop()
assert.Empty(t, res.Status.ValidationErrors)
hasIncompatibilityError := slices.ContainsFunc(res.Status.ValidationErrors, func(e string) bool {
return strings.Contains(e, "namespace-scoped or fine-grained global filter policies")
})
assert.False(t, hasIncompatibilityError)
require.NotNil(t, res.ResPolicies)
assert.Len(t, res.ResPolicies.GetNamespacedFilterPolicies(), test.expectNamespacedPolicies)
if test.expectClusterScopedPolicy {
assert.NotNil(t, res.ResPolicies.GetClusterScopedFilterPolicy())
} else {
assert.Nil(t, res.ResPolicies.GetClusterScopedFilterPolicy())
}
})
}
}
pkg/controller/data_upload_controller.go
+52 -48
View File
@@ -66,25 +66,26 @@ const (
// DataUploadReconciler reconciles a DataUpload object
type DataUploadReconciler struct {
client client.Client
kubeClient kubernetes.Interface
csiSnapshotClient snapshotter.SnapshotV1Interface
mgr manager.Manager
Clock clocks.WithTickerAndDelayedExecution
nodeName string
logger logrus.FieldLogger
snapshotExposerList map[velerov2alpha1api.SnapshotType]exposer.SnapshotExposer
dataPathMgr *datapath.Manager
vgdpCounter *exposer.VgdpCounter
loadAffinity []*kube.LoadAffinity
backupPVCConfig map[string]velerotypes.BackupPVC
podResources corev1api.ResourceRequirements
preparingTimeout time.Duration
metrics *metrics.ServerMetrics
cancelledDataUpload map[string]time.Time
dataMovePriorityClass string
podLabels map[string]string
podAnnotations map[string]string
client client.Client
kubeClient kubernetes.Interface
csiSnapshotClient snapshotter.SnapshotV1Interface
mgr manager.Manager
Clock clocks.WithTickerAndDelayedExecution
nodeName string
logger logrus.FieldLogger
snapshotExposerList map[velerov2alpha1api.SnapshotType]exposer.SnapshotExposer
dataPathMgr *datapath.Manager
vgdpCounter *exposer.VgdpCounter
loadAffinity []*kube.LoadAffinity
backupPVCConfig map[string]velerotypes.BackupPVC
podResources corev1api.ResourceRequirements
preparingTimeout time.Duration
metrics *metrics.ServerMetrics
cancelledDataUpload map[string]time.Time
dataMovePriorityClass string
podLabels map[string]string
podAnnotations map[string]string
snapshotMetadataServiceConfigs *velerotypes.CSISnapshotMetadataService
}
func NewDataUploadReconciler(
@@ -105,6 +106,7 @@ func NewDataUploadReconciler(
dataMovePriorityClass string,
podLabels map[string]string,
podAnnotations map[string]string,
snapshotMetadataServiceConfigs *velerotypes.CSISnapshotMetadataService,
) *DataUploadReconciler {
return &DataUploadReconciler{
client: client,
@@ -121,17 +123,18 @@ func NewDataUploadReconciler(
log,
),
},
dataPathMgr: dataPathMgr,
vgdpCounter: counter,
loadAffinity: loadAffinity,
backupPVCConfig: backupPVCConfig,
podResources: podResources,
preparingTimeout: preparingTimeout,
metrics: metrics,
cancelledDataUpload: make(map[string]time.Time),
dataMovePriorityClass: dataMovePriorityClass,
podLabels: podLabels,
podAnnotations: podAnnotations,
dataPathMgr: dataPathMgr,
vgdpCounter: counter,
loadAffinity: loadAffinity,
backupPVCConfig: backupPVCConfig,
podResources: podResources,
preparingTimeout: preparingTimeout,
metrics: metrics,
cancelledDataUpload: make(map[string]time.Time),
dataMovePriorityClass: dataMovePriorityClass,
podLabels: podLabels,
podAnnotations: podAnnotations,
snapshotMetadataServiceConfigs: snapshotMetadataServiceConfigs,
}
}
@@ -998,24 +1001,25 @@ func (r *DataUploadReconciler) setupExposeParam(du *velerov2alpha1api.DataUpload
}
return &exposer.CSISnapshotExposeParam{
SnapshotName: du.Spec.CSISnapshot.VolumeSnapshot,
SourceNamespace: du.Spec.SourceNamespace,
SourcePVCName: pvc.Name,
SourcePVName: pv.Name,
StorageClass: du.Spec.CSISnapshot.StorageClass,
HostingPodLabels: hostingPodLabels,
HostingPodAnnotations: hostingPodAnnotation,
HostingPodTolerations: hostingPodTolerations,
AccessMode: accessMode,
OperationTimeout: du.Spec.OperationTimeout.Duration,
ExposeTimeout: r.preparingTimeout,
VolumeSize: pvc.Spec.Resources.Requests[corev1api.ResourceStorage],
Affinity: r.loadAffinity,
BackupPVCConfig: r.backupPVCConfig,
Resources: r.podResources,
NodeOS: nodeOS,
PriorityClassName: r.dataMovePriorityClass,
DataMover: du.Spec.DataMover,
SnapshotName: du.Spec.CSISnapshot.VolumeSnapshot,
SourceNamespace: du.Spec.SourceNamespace,
SourcePVCName: pvc.Name,
SourcePVName: pv.Name,
StorageClass: du.Spec.CSISnapshot.StorageClass,
HostingPodLabels: hostingPodLabels,
HostingPodAnnotations: hostingPodAnnotation,
HostingPodTolerations: hostingPodTolerations,
AccessMode: accessMode,
OperationTimeout: du.Spec.OperationTimeout.Duration,
ExposeTimeout: r.preparingTimeout,
VolumeSize: pvc.Spec.Resources.Requests[corev1api.ResourceStorage],
Affinity: r.loadAffinity,
BackupPVCConfig: r.backupPVCConfig,
Resources: r.podResources,
NodeOS: nodeOS,
PriorityClassName: r.dataMovePriorityClass,
DataMover: du.Spec.DataMover,
SnapshotMetadataServiceConfigs: r.snapshotMetadataServiceConfigs,
}, nil
}
pkg/controller/data_upload_controller_test.go
+2
View File
@@ -251,6 +251,7 @@ func initDataUploaderReconcilerWithError(needError ...error) (*DataUploadReconci
"", // dataMovePriorityClass
nil, // podLabels
nil, // podAnnotations
nil,
), nil
}
@@ -1513,6 +1514,7 @@ func TestDataUploadSetupExposeParam(t *testing.T) {
"upload-priority",
tt.args.customLabels,
tt.args.customAnnotations,
nil,
)
// Act
pkg/exposer/csi_snapshot.go
+11
View File
@@ -98,6 +98,9 @@ type CSISnapshotExposeParam struct {
// DataMover is the data mover type, e.g., velero-fs, velero-block
DataMover string
// SnapshotMetadataServiceConfigs is the config for CSI snapshot metadata service
SnapshotMetadataServiceConfigs *velerotypes.CSISnapshotMetadataService
}
// CSISnapshotExposeWaitParam define the input param for WaitExposed of CSI snapshots
@@ -269,6 +272,7 @@ func (e *csiSnapshotExposer) Expose(ctx context.Context, ownerObject corev1api.O
csiExposeParam.PriorityClassName,
intoleratableNodes,
volumeTopology,
csiExposeParam.SnapshotMetadataServiceConfigs,
)
if err != nil {
return errors.Wrap(err, "error to create backup pod")
@@ -613,6 +617,7 @@ func (e *csiSnapshotExposer) createBackupPod(
priorityClassName string,
intoleratableNodes []string,
volumeTopology *corev1api.NodeSelector,
csiSnapshotMetadataServiceConfigs *velerotypes.CSISnapshotMetadataService,
) (*corev1api.Pod, error) {
podName := ownerObject.Name
@@ -668,6 +673,12 @@ func (e *csiSnapshotExposer) createBackupPod(
args = append(args, podInfo.logFormatArgs...)
args = append(args, podInfo.logLevelArgs...)
if csiSnapshotMetadataServiceConfigs != nil {
if csiSnapshotMetadataServiceConfigs.SAName != "" {
args = append(args, fmt.Sprintf("--csi-snapshot-metadata-service-sa=%s", csiSnapshotMetadataServiceConfigs.SAName))
}
}
if affinity == nil {
affinity = &kube.LoadAffinity{}
}
pkg/exposer/csi_snapshot_priority_test.go
+2
View File
@@ -155,6 +155,7 @@ func TestCreateBackupPodWithPriorityClass(t *testing.T) {
tc.expectedPriorityClass,
nil,
nil,
nil,
)
require.NoError(t, err, tc.description)
@@ -241,6 +242,7 @@ func TestCreateBackupPodWithMissingConfigMap(t *testing.T) {
"", // empty priority class since config map is missing
nil,
nil,
nil,
)
// Should succeed even when config map is missing
pkg/restic/command.go
-104
View File
@@ -1,104 +0,0 @@
/*
Copyright 2020 the Velero contributors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package restic
import (
"fmt"
"os"
"os/exec"
"path/filepath"
"strings"
)
// Command represents a restic command.
type Command struct {
Command string
RepoIdentifier string
PasswordFile string
CACertFile string
Dir string
Args []string
ExtraFlags []string
Env []string
}
func (c *Command) RepoName() string {
if c.RepoIdentifier == "" {
return ""
}
return c.RepoIdentifier[strings.LastIndex(c.RepoIdentifier, "/")+1:]
}
// StringSlice returns the command as a slice of strings.
func (c *Command) StringSlice() []string {
res := []string{"restic"}
res = append(res, c.Command, repoFlag(c.RepoIdentifier))
if c.PasswordFile != "" {
res = append(res, passwordFlag(c.PasswordFile))
}
if c.CACertFile != "" {
res = append(res, cacertFlag(c.CACertFile))
}
// If VELERO_SCRATCH_DIR is defined, put the restic cache within it. If not,
// allow restic to choose the location. This makes running either in-cluster
// or local (dev) work properly.
if scratch := os.Getenv("VELERO_SCRATCH_DIR"); scratch != "" {
res = append(res, cacheDirFlag(filepath.Join(scratch, ".cache", "restic")))
}
res = append(res, c.Args...)
res = append(res, c.ExtraFlags...)
return res
}
// String returns the command as a string.
func (c *Command) String() string {
return strings.Join(c.StringSlice(), " ")
}
// Cmd returns an exec.Cmd for the command.
func (c *Command) Cmd() *exec.Cmd {
parts := c.StringSlice()
cmd := exec.Command(parts[0], parts[1:]...) //nolint:gosec,noctx // Internal call. No need to check the parameter. No to add context for deprecated Restic.
cmd.Dir = c.Dir
if len(c.Env) > 0 {
cmd.Env = c.Env
}
return cmd
}
func repoFlag(repoIdentifier string) string {
return fmt.Sprintf("--repo=%s", repoIdentifier)
}
func passwordFlag(file string) string {
return fmt.Sprintf("--password-file=%s", file)
}
func cacheDirFlag(dir string) string {
return fmt.Sprintf("--cache-dir=%s", dir)
}
func cacertFlag(path string) string {
return fmt.Sprintf("--cacert=%s", path)
}
pkg/restic/command_factory.go
-125
View File
@@ -1,125 +0,0 @@
/*
Copyright 2018, 2019 the Velero contributors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package restic
import (
"fmt"
"strings"
)
// BackupCommand returns a Command for running a restic backup.
func BackupCommand(repoIdentifier, passwordFile, path string, tags map[string]string) *Command {
// --host flag is provided with a generic value because restic uses the host
// to find a parent snapshot, and by default it will be the name of the daemonset pod
// where the `restic backup` command is run. If this pod is recreated, we want to continue
// taking incremental backups rather than triggering a full one due to a new pod name.
return &Command{
Command: "backup",
RepoIdentifier: repoIdentifier,
PasswordFile: passwordFile,
Dir: path,
Args: []string{"."},
ExtraFlags: append(backupTagFlags(tags), "--host=velero", "--json"),
}
}
func backupTagFlags(tags map[string]string) []string {
var flags []string
for k, v := range tags {
flags = append(flags, fmt.Sprintf("--tag=%s=%s", k, v))
}
return flags
}
// RestoreCommand returns a Command for running a restic restore.
func RestoreCommand(repoIdentifier, passwordFile, snapshotID, target string) *Command {
return &Command{
Command: "restore",
RepoIdentifier: repoIdentifier,
PasswordFile: passwordFile,
Dir: target,
Args: []string{snapshotID},
ExtraFlags: []string{"--target=."},
}
}
// GetSnapshotCommand returns a Command for running a restic (get) snapshots.
func GetSnapshotCommand(repoIdentifier, passwordFile string, tags map[string]string) *Command {
return &Command{
Command: "snapshots",
RepoIdentifier: repoIdentifier,
PasswordFile: passwordFile,
// "--last" is replaced by "--latest=1" in restic v0.12.1
ExtraFlags: []string{"--json", "--latest=1", getSnapshotTagFlag(tags)},
}
}
func getSnapshotTagFlag(tags map[string]string) string {
var tagFilters []string
for k, v := range tags {
tagFilters = append(tagFilters, fmt.Sprintf("%s=%s", k, v))
}
return fmt.Sprintf("--tag=%s", strings.Join(tagFilters, ","))
}
func InitCommand(repoIdentifier string) *Command {
return &Command{
Command: "init",
RepoIdentifier: repoIdentifier,
}
}
func SnapshotsCommand(repoIdentifier string) *Command {
return &Command{
Command: "snapshots",
RepoIdentifier: repoIdentifier,
}
}
func PruneCommand(repoIdentifier string) *Command {
return &Command{
Command: "prune",
RepoIdentifier: repoIdentifier,
}
}
func ForgetCommand(repoIdentifier, snapshotID string) *Command {
return &Command{
Command: "forget",
RepoIdentifier: repoIdentifier,
Args: []string{snapshotID},
}
}
func UnlockCommand(repoIdentifier string) *Command {
return &Command{
Command: "unlock",
RepoIdentifier: repoIdentifier,
}
}
func StatsCommand(repoIdentifier, passwordFile, snapshotID string) *Command {
return &Command{
Command: "stats",
RepoIdentifier: repoIdentifier,
PasswordFile: passwordFile,
Args: []string{snapshotID},
ExtraFlags: []string{"--json"},
}
}
pkg/restic/command_factory_test.go
-131
View File
@@ -1,131 +0,0 @@
/*
Copyright 2018 the Velero contributors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package restic
import (
"sort"
"strings"
"testing"
"github.com/stretchr/testify/assert"
)
func TestBackupCommand(t *testing.T) {
c := BackupCommand("repo-id", "password-file", "path", map[string]string{"foo": "bar", "c": "d"})
assert.Equal(t, "backup", c.Command)
assert.Equal(t, "repo-id", c.RepoIdentifier)
assert.Equal(t, "password-file", c.PasswordFile)
assert.Equal(t, "path", c.Dir)
assert.Equal(t, []string{"."}, c.Args)
expected := []string{"--tag=foo=bar", "--tag=c=d", "--host=velero", "--json"}
sort.Strings(expected)
sort.Strings(c.ExtraFlags)
assert.Equal(t, expected, c.ExtraFlags)
}
func TestRestoreCommand(t *testing.T) {
c := RestoreCommand("repo-id", "password-file", "snapshot-id", "target")
assert.Equal(t, "restore", c.Command)
assert.Equal(t, "repo-id", c.RepoIdentifier)
assert.Equal(t, "password-file", c.PasswordFile)
assert.Equal(t, "target", c.Dir)
assert.Equal(t, []string{"snapshot-id"}, c.Args)
assert.Equal(t, []string{"--target=."}, c.ExtraFlags)
}
func TestGetSnapshotCommand(t *testing.T) {
expectedTags := map[string]string{"foo": "bar", "c": "d"}
c := GetSnapshotCommand("repo-id", "password-file", expectedTags)
assert.Equal(t, "snapshots", c.Command)
assert.Equal(t, "repo-id", c.RepoIdentifier)
assert.Equal(t, "password-file", c.PasswordFile)
// set up expected flag names
expectedFlags := []string{"--json", "--latest=1", "--tag"}
// for tracking actual flag names
actualFlags := []string{}
// for tracking actual --tag values as a map
actualTags := make(map[string]string)
// loop through actual flags
for _, flag := range c.ExtraFlags {
// split into 2 parts from the first = sign (if any)
parts := strings.SplitN(flag, "=", 2)
// convert --tag data to a map
if parts[0] == "--tag" {
actualFlags = append(actualFlags, parts[0])
// split based on ,
tags := strings.Split(parts[1], ",")
// loop through each key-value tag pair
for _, tag := range tags {
// split the pair on =
kvs := strings.Split(tag, "=")
// record actual key & value
actualTags[kvs[0]] = kvs[1]
}
} else {
actualFlags = append(actualFlags, flag)
}
}
assert.Equal(t, expectedFlags, actualFlags)
assert.Equal(t, expectedTags, actualTags)
}
func TestInitCommand(t *testing.T) {
c := InitCommand("repo-id")
assert.Equal(t, "init", c.Command)
assert.Equal(t, "repo-id", c.RepoIdentifier)
}
func TestSnapshotsCommand(t *testing.T) {
c := SnapshotsCommand("repo-id")
assert.Equal(t, "snapshots", c.Command)
assert.Equal(t, "repo-id", c.RepoIdentifier)
}
func TestPruneCommand(t *testing.T) {
c := PruneCommand("repo-id")
assert.Equal(t, "prune", c.Command)
assert.Equal(t, "repo-id", c.RepoIdentifier)
}
func TestForgetCommand(t *testing.T) {
c := ForgetCommand("repo-id", "snapshot-id")
assert.Equal(t, "forget", c.Command)
assert.Equal(t, "repo-id", c.RepoIdentifier)
assert.Equal(t, []string{"snapshot-id"}, c.Args)
}
func TestStatsCommand(t *testing.T) {
c := StatsCommand("repo-id", "password-file", "snapshot-id")
assert.Equal(t, "stats", c.Command)
assert.Equal(t, "repo-id", c.RepoIdentifier)
assert.Equal(t, "password-file", c.PasswordFile)
assert.Equal(t, []string{"snapshot-id"}, c.Args)
assert.Equal(t, []string{"--json"}, c.ExtraFlags)
}
pkg/restic/command_test.go
-106
View File
@@ -1,106 +0,0 @@
/*
Copyright 2020 the Velero contributors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package restic
import (
"os"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestRepoName(t *testing.T) {
c := &Command{RepoIdentifier: ""}
assert.Empty(t, c.RepoName())
c.RepoIdentifier = "s3:s3.amazonaws.com/bucket/prefix/repo"
assert.Equal(t, "repo", c.RepoName())
c.RepoIdentifier = "azure:bucket:/repo"
assert.Equal(t, "repo", c.RepoName())
c.RepoIdentifier = "gs:bucket:/prefix/repo"
assert.Equal(t, "repo", c.RepoName())
}
func TestStringSlice(t *testing.T) {
c := &Command{
Command: "cmd",
RepoIdentifier: "repo-id",
PasswordFile: "/path/to/password-file",
Dir: "/some/pwd",
Args: []string{"arg-1", "arg-2"},
ExtraFlags: []string{"--foo=bar"},
}
require.NoError(t, os.Unsetenv("VELERO_SCRATCH_DIR"))
assert.Equal(t, []string{
"restic",
"cmd",
"--repo=repo-id",
"--password-file=/path/to/password-file",
"arg-1",
"arg-2",
"--foo=bar",
}, c.StringSlice())
os.Setenv("VELERO_SCRATCH_DIR", "/foo")
assert.Equal(t, []string{
"restic",
"cmd",
"--repo=repo-id",
"--password-file=/path/to/password-file",
"--cache-dir=/foo/.cache/restic",
"arg-1",
"arg-2",
"--foo=bar",
}, c.StringSlice())
require.NoError(t, os.Unsetenv("VELERO_SCRATCH_DIR"))
}
func TestString(t *testing.T) {
c := &Command{
Command: "cmd",
RepoIdentifier: "repo-id",
PasswordFile: "/path/to/password-file",
Dir: "/some/pwd",
Args: []string{"arg-1", "arg-2"},
ExtraFlags: []string{"--foo=bar"},
}
require.NoError(t, os.Unsetenv("VELERO_SCRATCH_DIR"))
assert.Equal(t, "restic cmd --repo=repo-id --password-file=/path/to/password-file arg-1 arg-2 --foo=bar", c.String())
}
func TestCmd(t *testing.T) {
c := &Command{
Command: "cmd",
RepoIdentifier: "repo-id",
PasswordFile: "/path/to/password-file",
Dir: "/some/pwd",
Args: []string{"arg-1", "arg-2"},
ExtraFlags: []string{"--foo=bar"},
}
require.NoError(t, os.Unsetenv("VELERO_SCRATCH_DIR"))
execCmd := c.Cmd()
assert.Equal(t, c.StringSlice(), execCmd.Args)
assert.Equal(t, c.Dir, execCmd.Dir)
}
pkg/restic/common.go
-159
View File
@@ -1,159 +0,0 @@
/*
Copyright the Velero contributors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package restic
import (
"fmt"
"os"
"strconv"
"strings"
"time"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"github.com/vmware-tanzu/velero/internal/credentials"
velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
repoconfig "github.com/vmware-tanzu/velero/pkg/repository/config"
"github.com/vmware-tanzu/velero/pkg/util/filesystem"
)
const (
// DefaultMaintenanceFrequency is the default time interval
// at which restic prune is run.
DefaultMaintenanceFrequency = 7 * 24 * time.Hour
// insecureSkipTLSVerifyKey is the flag in BackupStorageLocation's config
// to indicate whether to skip TLS verify to setup insecure HTTPS connection.
insecureSkipTLSVerifyKey = "insecureSkipTLSVerify"
// resticInsecureTLSFlag is the flag for Restic command line to indicate
// skip TLS verify on https connection.
resticInsecureTLSFlag = "--insecure-tls"
)
// TempCACertFile creates a temp file containing a CA bundle
// and returns its path. The caller should generally call os.Remove()
// to remove the file when done with it.
func TempCACertFile(caCert []byte, bsl string, fs filesystem.Interface) (string, error) {
file, err := fs.TempFile("", fmt.Sprintf("cacert-%s", bsl))
if err != nil {
return "", errors.WithStack(err)
}
if _, err := file.Write(caCert); err != nil {
// nothing we can do about an error closing the file here, and we're
// already returning an error about the write failing.
file.Close()
return "", errors.WithStack(err)
}
name := file.Name()
if err := file.Close(); err != nil {
return "", errors.WithStack(err)
}
return name, nil
}
// environ is a slice of strings representing the environment, in the form "key=value".
type environ []string
// Unset a single environment variable.
func (e *environ) Unset(key string) {
for i := range *e {
if strings.HasPrefix((*e)[i], key+"=") {
(*e)[i] = (*e)[len(*e)-1]
*e = (*e)[:len(*e)-1]
break
}
}
}
// CmdEnv returns a list of environment variables (in the format var=val) that
// should be used when running a restic command for a particular backend provider.
// This list is the current environment, plus any provider-specific variables restic needs.
func CmdEnv(backupLocation *velerov1api.BackupStorageLocation, credentialFileStore credentials.FileStore) ([]string, error) {
var env environ
env = os.Environ()
customEnv := map[string]string{}
var err error
config := backupLocation.Spec.Config
if config == nil {
config = map[string]string{}
}
if backupLocation.Spec.Credential != nil {
credsFile, err := credentialFileStore.Path(backupLocation.Spec.Credential)
if err != nil {
return []string{}, errors.WithStack(err)
}
config[repoconfig.CredentialsFileKey] = credsFile
}
backendType := repoconfig.GetBackendType(backupLocation.Spec.Provider, backupLocation.Spec.Config)
switch backendType {
case repoconfig.AWSBackend:
customEnv, err = repoconfig.GetS3ResticEnvVars(config)
if err != nil {
return []string{}, err
}
case repoconfig.AzureBackend:
customEnv, err = repoconfig.GetAzureResticEnvVars(config)
if err != nil {
return []string{}, err
}
case repoconfig.GCPBackend:
customEnv, err = repoconfig.GetGCPResticEnvVars(config)
if err != nil {
return []string{}, err
}
}
for k, v := range customEnv {
env.Unset(k)
if v == "" {
continue
}
env = append(env, fmt.Sprintf("%s=%s", k, v))
}
return env, nil
}
// GetInsecureSkipTLSVerifyFromBSL get insecureSkipTLSVerify flag from BSL configuration,
// Then return --insecure-tls flag with boolean value as result.
func GetInsecureSkipTLSVerifyFromBSL(backupLocation *velerov1api.BackupStorageLocation, logger logrus.FieldLogger) string {
result := ""
if backupLocation == nil {
logger.Info("bsl is nil. return empty.")
return result
}
if insecure, _ := strconv.ParseBool(backupLocation.Spec.Config[insecureSkipTLSVerifyKey]); insecure {
logger.Debugf("set --insecure-tls=true for Restic command according to BSL %s config", backupLocation.Name)
result = resticInsecureTLSFlag + "=true"
return result
}
return result
}
pkg/restic/common_test.go
-141
View File
@@ -1,141 +0,0 @@
/*
Copyright the Velero contributors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package restic
import (
"os"
"testing"
"github.com/sirupsen/logrus"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
velerotest "github.com/vmware-tanzu/velero/pkg/test"
)
func TestTempCACertFile(t *testing.T) {
var (
fs = velerotest.NewFakeFileSystem()
caCertData = []byte("cacert")
)
fileName, err := TempCACertFile(caCertData, "default", fs)
require.NoError(t, err)
contents, err := fs.ReadFile(fileName)
require.NoError(t, err)
assert.Equal(t, string(caCertData), string(contents))
os.Remove(fileName)
}
func TestGetInsecureSkipTLSVerifyFromBSL(t *testing.T) {
log := logrus.StandardLogger()
tests := []struct {
name string
backupLocation *velerov1api.BackupStorageLocation
logger logrus.FieldLogger
expected string
}{
{
"Test with nil BSL. Should return empty string.",
nil,
log,
"",
},
{
"Test BSL with no configuration. Should return empty string.",
&velerov1api.BackupStorageLocation{
Spec: velerov1api.BackupStorageLocationSpec{
Provider: "azure",
},
},
log,
"",
},
{
"Test with AWS BSL's insecureSkipTLSVerify set to false.",
&velerov1api.BackupStorageLocation{
Spec: velerov1api.BackupStorageLocationSpec{
Provider: "aws",
Config: map[string]string{
"insecureSkipTLSVerify": "false",
},
},
},
log,
"",
},
{
"Test with AWS BSL's insecureSkipTLSVerify set to true.",
&velerov1api.BackupStorageLocation{
Spec: velerov1api.BackupStorageLocationSpec{
Provider: "aws",
Config: map[string]string{
"insecureSkipTLSVerify": "true",
},
},
},
log,
"--insecure-tls=true",
},
{
"Test with Azure BSL's insecureSkipTLSVerify set to invalid.",
&velerov1api.BackupStorageLocation{
Spec: velerov1api.BackupStorageLocationSpec{
Provider: "azure",
Config: map[string]string{
"insecureSkipTLSVerify": "invalid",
},
},
},
log,
"",
},
{
"Test with GCP without insecureSkipTLSVerify.",
&velerov1api.BackupStorageLocation{
Spec: velerov1api.BackupStorageLocationSpec{
Provider: "gcp",
Config: map[string]string{},
},
},
log,
"",
},
{
"Test with AWS without config.",
&velerov1api.BackupStorageLocation{
Spec: velerov1api.BackupStorageLocationSpec{
Provider: "aws",
},
},
log,
"",
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
res := GetInsecureSkipTLSVerifyFromBSL(test.backupLocation, test.logger)
assert.Equal(t, test.expected, res)
})
}
}
pkg/restic/exec_commands.go
-292
View File
@@ -1,292 +0,0 @@
/*
Copyright The Velero Contributors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package restic
import (
"bytes"
"encoding/json"
"fmt"
"strings"
"time"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"github.com/vmware-tanzu/velero/pkg/uploader"
"github.com/vmware-tanzu/velero/pkg/util/exec"
"github.com/vmware-tanzu/velero/pkg/util/filesystem"
)
const restoreProgressCheckInterval = 10 * time.Second
const backupProgressCheckInterval = 10 * time.Second
var fileSystem = filesystem.NewFileSystem()
type backupStatusLine struct {
MessageType string `json:"message_type"`
// seen in status lines
TotalBytes int64 `json:"total_bytes"`
BytesDone int64 `json:"bytes_done"`
// seen in summary line at the end
TotalBytesProcessed int64 `json:"total_bytes_processed"`
}
// GetSnapshotID runs provided 'restic snapshots' command to get the ID of a snapshot
// and an error if a unique snapshot cannot be identified.
func GetSnapshotID(snapshotIDCmd *Command) (string, error) {
stdout, stderr, err := exec.RunCommand(snapshotIDCmd.Cmd())
if err != nil {
return "", errors.Wrapf(err, "error running command, stderr=%s", stderr)
}
type snapshotID struct {
ShortID string `json:"short_id"`
}
var snapshots []snapshotID
if err := json.Unmarshal([]byte(stdout), &snapshots); err != nil {
return "", errors.Wrap(err, "error unmarshaling restic snapshots result")
}
if len(snapshots) != 1 {
return "", errors.Errorf("expected one matching snapshot by command: %s, got %d", snapshotIDCmd.String(), len(snapshots))
}
return snapshots[0].ShortID, nil
}
// RunBackup runs a `restic backup` command and watches the output to provide
// progress updates to the caller.
func RunBackup(backupCmd *Command, log logrus.FieldLogger, updater uploader.ProgressUpdater) (string, string, error) {
// buffers for copying command stdout/err output into
stdoutBuf := new(bytes.Buffer)
stderrBuf := new(bytes.Buffer)
// create a channel to signal when to end the goroutine scanning for progress
// updates
quit := make(chan struct{})
cmd := backupCmd.Cmd()
cmd.Stdout = stdoutBuf
cmd.Stderr = stderrBuf
err := cmd.Start()
if err != nil {
exec.LogErrorAsExitCode(err, log)
return stdoutBuf.String(), stderrBuf.String(), err
}
go func() {
ticker := time.NewTicker(backupProgressCheckInterval)
for {
select {
case <-ticker.C:
lastLine := getLastLine(stdoutBuf.Bytes())
if len(lastLine) > 0 {
stat, err := decodeBackupStatusLine(lastLine)
if err != nil {
log.WithError(err).Errorf("error getting restic backup progress")
}
// if the line contains a non-empty bytes_done field, we can update the
// caller with the progress
if stat.BytesDone != 0 {
updater.UpdateProgress(&uploader.Progress{
TotalBytes: stat.TotalBytes,
BytesDone: stat.BytesDone,
})
}
}
case <-quit:
ticker.Stop()
return
}
}
}()
err = cmd.Wait()
if err != nil {
exec.LogErrorAsExitCode(err, log)
return stdoutBuf.String(), stderrBuf.String(), err
}
quit <- struct{}{}
summary, err := getSummaryLine(stdoutBuf.Bytes())
if err != nil {
return stdoutBuf.String(), stderrBuf.String(), err
}
stat, err := decodeBackupStatusLine(summary)
if err != nil {
return stdoutBuf.String(), stderrBuf.String(), err
}
if stat.MessageType != "summary" {
return stdoutBuf.String(), stderrBuf.String(), errors.WithStack(fmt.Errorf("error getting restic backup summary: %s", string(summary)))
}
// update progress to 100%
updater.UpdateProgress(&uploader.Progress{
TotalBytes: stat.TotalBytesProcessed,
BytesDone: stat.TotalBytesProcessed,
})
return string(summary), stderrBuf.String(), nil
}
func decodeBackupStatusLine(lastLine []byte) (backupStatusLine, error) {
var stat backupStatusLine
if err := json.Unmarshal(lastLine, &stat); err != nil {
return stat, errors.Wrapf(err, "unable to decode backup JSON line: %s", string(lastLine))
}
return stat, nil
}
// getLastLine returns the last line of a byte array. The string is assumed to
// have a newline at the end of it, so this returns the substring between the
// last two newlines.
func getLastLine(b []byte) []byte {
if len(b) == 0 {
return []byte("")
}
// subslice the byte array to ignore the newline at the end of the string
lastNewLineIdx := bytes.LastIndex(b[:len(b)-1], []byte("\n"))
return b[lastNewLineIdx+1 : len(b)-1]
}
// getSummaryLine looks for the summary JSON line
// (`{"message_type:"summary",...`) in the restic backup command output. Due to
// an issue in Restic, this might not always be the last line
// (https://github.com/restic/restic/issues/2389). It returns an error if it
// can't be found.
func getSummaryLine(b []byte) ([]byte, error) {
summaryLineIdx := bytes.LastIndex(b, []byte(`{"message_type":"summary"`))
if summaryLineIdx < 0 {
return nil, errors.New("unable to find summary in restic backup command output")
}
// find the end of the summary line
newLineIdx := bytes.Index(b[summaryLineIdx:], []byte("\n"))
if newLineIdx < 0 {
return nil, errors.New("unable to get summary line from restic backup command output")
}
return b[summaryLineIdx : summaryLineIdx+newLineIdx], nil
}
// RunRestore runs a `restic restore` command and monitors the volume size to
// provide progress updates to the caller.
func RunRestore(restoreCmd *Command, log logrus.FieldLogger, updater uploader.ProgressUpdater) (string, string, error) {
insecureTLSFlag := ""
for _, extraFlag := range restoreCmd.ExtraFlags {
if strings.Contains(extraFlag, resticInsecureTLSFlag) {
insecureTLSFlag = extraFlag
}
}
snapshotSize, err := getSnapshotSize(restoreCmd.RepoIdentifier, restoreCmd.PasswordFile, restoreCmd.CACertFile, restoreCmd.Args[0], restoreCmd.Env, insecureTLSFlag)
if err != nil {
return "", "", errors.Wrap(err, "error getting snapshot size")
}
updater.UpdateProgress(&uploader.Progress{
TotalBytes: snapshotSize,
})
// create a channel to signal when to end the goroutine scanning for progress
// updates
quit := make(chan struct{})
go func() {
ticker := time.NewTicker(restoreProgressCheckInterval)
for {
select {
case <-ticker.C:
volumeSize, err := getVolumeSize(restoreCmd.Dir)
if err != nil {
log.WithError(err).Errorf("error getting restic restore progress")
}
if volumeSize != 0 {
updater.UpdateProgress(&uploader.Progress{
TotalBytes: snapshotSize,
BytesDone: volumeSize,
})
}
case <-quit:
ticker.Stop()
return
}
}
}()
stdout, stderr, err := exec.RunCommandWithLog(restoreCmd.Cmd(), log)
quit <- struct{}{}
// update progress to 100%
updater.UpdateProgress(&uploader.Progress{
TotalBytes: snapshotSize,
BytesDone: snapshotSize,
})
return stdout, stderr, err
}
func getSnapshotSize(repoIdentifier, passwordFile, caCertFile, snapshotID string, env []string, insecureTLS string) (int64, error) {
cmd := StatsCommand(repoIdentifier, passwordFile, snapshotID)
cmd.Env = env
cmd.CACertFile = caCertFile
if len(insecureTLS) > 0 {
cmd.ExtraFlags = append(cmd.ExtraFlags, insecureTLS)
}
stdout, stderr, err := exec.RunCommand(cmd.Cmd())
if err != nil {
return 0, errors.Wrapf(err, "error running command, stderr=%s", stderr)
}
var snapshotStats struct {
TotalSize int64 `json:"total_size"`
}
if err := json.Unmarshal([]byte(stdout), &snapshotStats); err != nil {
return 0, errors.Wrapf(err, "error unmarshaling restic stats result, stdout=%s", stdout)
}
return snapshotStats.TotalSize, nil
}
func getVolumeSize(path string) (int64, error) {
var size int64
files, err := fileSystem.ReadDir(path)
if err != nil {
return 0, errors.Wrapf(err, "error reading directory %s", path)
}
for _, file := range files {
if file.IsDir() {
s, err := getVolumeSize(fmt.Sprintf("%s/%s", path, file.Name()))
if err != nil {
return 0, err
}
size += s
} else {
size += file.Size()
}
}
return size, nil
}
pkg/restic/exec_commands_test.go
-111
View File
@@ -1,111 +0,0 @@
/*
Copyright 2019 the Velero contributors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package restic
import (
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/vmware-tanzu/velero/pkg/test"
"github.com/vmware-tanzu/velero/pkg/util/filesystem"
)
func Test_getSummaryLine(t *testing.T) {
summaryLine := `{"message_type":"summary","files_new":0,"files_changed":0,"files_unmodified":3,"dirs_new":0,"dirs_changed":0,"dirs_unmodified":0,"data_blobs":0,"tree_blobs":0,"data_added":0,"total_files_processed":3,"total_bytes_processed":13238272000,"total_duration":0.319265105,"snapshot_id":"38515bb5"}`
tests := []struct {
name string
output string
wantErr bool
}{
{"no summary", `{"message_type":"status","percent_done":0,"total_files":1,"total_bytes":10485760000}
{"message_type":"status","percent_done":0,"total_files":3,"files_done":1,"total_bytes":13238272000}
`, true},
{"no newline after summary", `{"message_type":"status","percent_done":0,"total_files":1,"total_bytes":10485760000}
{"message_type":"status","percent_done":0,"total_files":3,"files_done":1,"total_bytes":13238272000}
{"message_type":"summary","files_new":0,"files_changed":0,"files_unmodified":3,"dirs_new":0`, true},
{"summary at end", `{"message_type":"status","percent_done":0,"total_files":1,"total_bytes":10485760000}
{"message_type":"status","percent_done":0,"total_files":3,"files_done":1,"total_bytes":13238272000}
{"message_type":"status","percent_done":1,"total_files":3,"files_done":3,"total_bytes":13238272000,"bytes_done":13238272000}
{"message_type":"summary","files_new":0,"files_changed":0,"files_unmodified":3,"dirs_new":0,"dirs_changed":0,"dirs_unmodified":0,"data_blobs":0,"tree_blobs":0,"data_added":0,"total_files_processed":3,"total_bytes_processed":13238272000,"total_duration":0.319265105,"snapshot_id":"38515bb5"}
`, false},
{"summary before status", `{"message_type":"status","percent_done":0,"total_files":1,"total_bytes":10485760000}
{"message_type":"status","percent_done":0,"total_files":3,"files_done":1,"total_bytes":13238272000}
{"message_type":"summary","files_new":0,"files_changed":0,"files_unmodified":3,"dirs_new":0,"dirs_changed":0,"dirs_unmodified":0,"data_blobs":0,"tree_blobs":0,"data_added":0,"total_files_processed":3,"total_bytes_processed":13238272000,"total_duration":0.319265105,"snapshot_id":"38515bb5"}
{"message_type":"status","percent_done":1,"total_files":3,"files_done":3,"total_bytes":13238272000,"bytes_done":13238272000}
`, false},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
summary, err := getSummaryLine([]byte(tt.output))
if tt.wantErr {
assert.Error(t, err)
} else {
require.NoError(t, err)
assert.Equal(t, summaryLine, string(summary))
}
})
}
}
func Test_getLastLine(t *testing.T) {
tests := []struct {
output []byte
want string
}{
{[]byte(`last line
`), "last line"},
{[]byte(`first line
second line
third line
`), "third line"},
{[]byte(""), ""},
{nil, ""},
}
for _, tt := range tests {
t.Run(tt.want, func(t *testing.T) {
assert.Equal(t, []byte(tt.want), getLastLine(tt.output))
})
}
}
func Test_getVolumeSize(t *testing.T) {
files := map[string][]byte{
"/file1.txt": []byte("file1"),
"/file2.txt": []byte("file2"),
"/file3.txt": []byte("file3"),
"/files/file4.txt": []byte("file4"),
"/files/nested/file5.txt": []byte("file5"),
}
fakefs := test.NewFakeFileSystem()
var expectedSize int64
for path, content := range files {
fakefs.WithFile(path, content)
expectedSize += int64(len(content))
}
fileSystem = fakefs
defer func() { fileSystem = filesystem.NewFileSystem() }()
actualSize, err := getVolumeSize("/")
require.NoError(t, err)
assert.Equal(t, expectedSize, actualSize)
}
pkg/types/node_agent.go
+7
View File
@@ -74,6 +74,10 @@ type CachePVC struct {
ResidentThresholdInMB int64 `json:"residentThresholdInMB,omitempty"`
}
type CSISnapshotMetadataService struct {
SAName string `json:"saName,omitempty"`
}
type NodeAgentConfigs struct {
// LoadConcurrency is the config for data path load concurrency per node.
LoadConcurrency *LoadConcurrency `json:"loadConcurrency,omitempty"`
@@ -104,4 +108,7 @@ type NodeAgentConfigs struct {
// PodLabels are labels to be added to pods created by node-agent, i.e., data mover pods.
PodLabels map[string]string `json:"podLabels,omitempty"`
// CSISnapshotMetadataServiceConfigs is the config for CSI snapshot metadata service
CSISnapshotMetadataServiceConfigs *CSISnapshotMetadataService `json:"csiSnapshotMetadataServiceConfigs,omitempty"`
}