Merge branch 'main' into pr-bug4-gap6

This commit is contained in:
Tiger Kaovilai
2026-09-03 01:39:25 -04:00
committed by GitHub
77 changed files with 2224 additions and 181 deletions
+8 -6
View File
@@ -12,11 +12,13 @@ env:
on:
push:
pull_request:
# Do not run when the change only includes these directories.
paths-ignore:
- "site/**"
- "design/**"
- "**/*.md"
# Only run when the change affects Go code or a workflow this job depends on.
paths:
- "**/*.go"
- "go.mod"
- "go.sum"
- ".github/workflows/e2e-test-kind.yaml"
- ".github/workflows/get-go-version.yaml"
jobs:
get-go-version:
uses: ./.github/workflows/get-go-version.yaml
@@ -192,7 +194,7 @@ jobs:
- name: Install MinIO
run: |
docker run -d --rm -p 9000:9000 -e "MINIO_ROOT_USER=minio" -e "MINIO_ROOT_PASSWORD=minio123" -e "MINIO_DEFAULT_BUCKETS=bucket,additional-bucket" bitnami/minio:local
- uses: helm/kind-action@7a97ed793754775518f9db3a8151ee7461dc9c31 # v1 + fix: add curl retry flags (https://github.com/helm/kind-action/pull/165)
- uses: helm/kind-action@c72b4750145dbfb1c71734c3782a4db35a1c65c0 # v1 + fix: add curl retry flags (https://github.com/helm/kind-action/pull/165)
with:
cluster_name: "kind"
version: "v0.32.0"
+1 -1
View File
@@ -31,6 +31,6 @@ jobs:
output: 'trivy-results.sarif'
- name: Upload Trivy scan results to GitHub Security tab
uses: github/codeql-action/upload-sarif@v4.37.7
uses: github/codeql-action/upload-sarif@v4.37.9
with:
sarif_file: 'trivy-results.sarif'
+1
View File
@@ -0,0 +1 @@
Validate backup name format before contacting the API server
+1
View File
@@ -0,0 +1 @@
add test coverage for CleanupVolumeSnapshot
+1
View File
@@ -0,0 +1 @@
Fix context propagation bug in GetDefaultBackupStorageLocations and add missing test coverage for core components
+1
View File
@@ -0,0 +1 @@
Check both node-agent daemonsets before returning a non-NotFound lookup error in IsReady, so a transient error fetching one daemonset no longer masks the other daemonset being ready
+1
View File
@@ -0,0 +1 @@
Add unit tests for pkg/itemblock
+1
View File
@@ -0,0 +1 @@
Add in-place restore pre-flight check: target PVC must not be in use
+1
View File
@@ -0,0 +1 @@
Add "IncrementalBytes" field to status of DataDownload and PVR to indicate data transferred by the incremental restore
+1
View File
@@ -0,0 +1 @@
Enhance the doc for backup deletion
+1
View File
@@ -0,0 +1 @@
test: resolve remaining Ginkgo V2 and Gomega anti-patterns (#10440)
@@ -0,0 +1 @@
Fix issue 10454, enforce resource filter policies in memory in stage 1 to fix wildcard namespace bypassing issue
+1
View File
@@ -0,0 +1 @@
Add operation context to user-facing error messages in CR statuses and CLI output
@@ -21,6 +21,11 @@ spec:
jsonPath: .status.phase
name: Status
type: string
- description: Restore type such as Full/Incremental
jsonPath: .spec.restoreType
name: Restore Type
priority: 10
type: string
- description: Time duration since this PodVolumeRestore was started
jsonPath: .status.startTimestamp
name: Started
@@ -35,6 +40,12 @@ spec:
jsonPath: .status.progress.totalBytes
name: Total Bytes
type: integer
- description: Incremental bytes
format: int64
jsonPath: .status.incrementalBytes
name: Incremental Bytes
priority: 10
type: integer
- description: Name of the Backup Storage Location where the backup data is stored
jsonPath: .spec.backupStorageLocation
name: Storage Location
@@ -193,6 +204,10 @@ spec:
format: date-time
nullable: true
type: string
incrementalBytes:
description: IncrementalBytes holds the number of bytes restored incrementally
format: int64
type: integer
message:
description: Message is a message about the pod volume restore's status.
type: string
@@ -21,6 +21,11 @@ spec:
jsonPath: .status.phase
name: Status
type: string
- description: Restore type such as Full/Incremental
jsonPath: .spec.restoreType
name: Restore Type
priority: 10
type: string
- description: Time duration since this DataDownload was started
jsonPath: .status.startTimestamp
name: Started
@@ -35,6 +40,12 @@ spec:
jsonPath: .status.progress.totalBytes
name: Total Bytes
type: integer
- description: Incremental bytes
format: int64
jsonPath: .status.incrementalBytes
name: Incremental Bytes
priority: 10
type: integer
- description: Name of the Backup Storage Location where the backup data is stored
jsonPath: .spec.backupStorageLocation
name: Storage Location
@@ -202,6 +213,11 @@ spec:
format: date-time
nullable: true
type: string
incrementalBytes:
description: IncrementalBytes holds the number of bytes restored incrementally
since the last snapshot
format: int64
type: integer
message:
description: Message is a message about the DataDownload's status.
type: string
@@ -236,7 +236,13 @@ The key requirements for this approach are:
Before initiating an in-place restore for a volume, Velero performs the following pre-flight checks to ensure the operation is safe and valid:
#### 1. PVC is Not Actively Used by a Running Pod
Velero verifies that the target PVC is not currently mounted or consumed by any running Pods in the cluster. If the PVC is in use, Velero will skip the in-place restore for that volume and log an error. This enforces the prerequisite that users must completely delete consuming workloads prior to the restore, which prevents data corruption and avoids deadlocks caused by the Kubernetes `pvc-protection` finalizer during PVC recreation.
Velero verifies that the target PVC is not currently mounted or consumed by any active Pods in the cluster. If the PVC is in use, Velero will skip the in-place restore for that volume and log an error. This enforces the prerequisite that users must completely delete consuming workloads prior to the restore, which prevents data corruption and avoids deadlocks caused by the Kubernetes `pvc-protection` finalizer during PVC recreation.
The "in use" semantics align with the Kubernetes `pvc-protection` controller: Pods in a terminal phase (`Succeeded`/`Failed`) do not block the restore, all other phases do, and terminating Pods are flagged in the error message so users know to simply wait and retry.
The check runs on both restore paths before any side effect on the existing PVC/PV: in the PVC CSI RIA before deleting the existing PVC, and before creating the `PodVolumeRestore` on the file system path. On the file system path, Pods gated by this restore's `restore-wait` init container (identified by the restore UID in its args, and not yet terminated) are exempted: they must mount the PVC for the node-agent to restore the data, and they cannot write to the volume until this restore's `PodVolumeRestore`s complete. Leftover Pods, controller-recreated Pods, and Pods gated by a different restore still block.
This check is a fail-fast validation, not an atomic guarantee; the `pvc-protection` finalizer remains the actual safety gate for PVC deletion. A residual `VolumeAttachment` check (e.g. a `Failed` Pod imposed by the control plane after a non-graceful node shutdown, where the node never unmounted the volume) may be added as a future enhancement.
#### 2. PVC is Bound to the Original PV
Velero checks whether the existing PVC in the cluster is still bound to the same PersistentVolume (PV) it was bound to at the time of the backup. If the PVC is bound to a different PV, performing an in-place restore (especially an incremental one that relies on Changed Block Tracking) may be unsafe or result in unpredictable behavior. If this check fails, Velero will log an error and skip the in-place restore for that volume.
+1 -1
View File
@@ -148,7 +148,7 @@ require (
github.com/kcp-dev/kcp/cli v0.27.1 // indirect
github.com/kcp-dev/kcp/sdk v0.27.1 // indirect
github.com/kcp-dev/logicalcluster/v3 v3.0.5 // indirect
github.com/klauspost/compress v1.18.6 // indirect
github.com/klauspost/compress v1.19.2 // indirect
github.com/klauspost/cpuid/v2 v2.3.0 // indirect
github.com/klauspost/crc32 v1.3.0 // indirect
github.com/klauspost/pgzip v1.2.6 // indirect
+2 -2
View File
@@ -297,8 +297,8 @@ github.com/keybase/go-keychain v0.0.1 h1:way+bWYa6lDppZoZcgMbYsvC7GxljxrskdNInRt
github.com/keybase/go-keychain v0.0.1/go.mod h1:PdEILRW3i9D8JcdM+FmY6RwkHGnhHxXwkPPMeUgOK1k=
github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao=
github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
github.com/klauspost/compress v1.19.2 h1:hMRETovs/pu/dVWN7zIT1PGG8t509MwT6bO7XSi26R8=
github.com/klauspost/compress v1.19.2/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
github.com/klauspost/cpuid/v2 v2.0.1/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
github.com/klauspost/cpuid/v2 v2.3.0 h1:S4CRMLnYUhGeDFDqkGriYKdfoFlDnMtqTiI/sFzhA9Y=
github.com/klauspost/cpuid/v2 v2.3.0/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0=
+97
View File
@@ -0,0 +1,97 @@
/*
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 credentials
import (
"testing"
. "github.com/onsi/gomega"
corev1api "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"sigs.k8s.io/controller-runtime/pkg/client/fake"
)
func TestNamespacedSecretStore(t *testing.T) {
scheme := runtime.NewScheme()
g := NewWithT(t)
g.Expect(corev1api.AddToScheme(scheme)).To(Succeed())
secret := &corev1api.Secret{
ObjectMeta: metav1.ObjectMeta{
Name: "test-secret",
Namespace: "velero",
},
Data: map[string][]byte{
"creds-key": []byte("my-super-secret-value"),
},
}
client := fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects(secret).Build()
store, err := NewNamespacedSecretStore(client, "velero")
g.Expect(err).ToNot(HaveOccurred())
tests := []struct {
name string
selector *corev1api.SecretKeySelector
expectedVal string
expectErr bool
}{
{
name: "existing secret and key returns the correct value",
selector: &corev1api.SecretKeySelector{
LocalObjectReference: corev1api.LocalObjectReference{Name: "test-secret"},
Key: "creds-key",
},
expectedVal: "my-super-secret-value",
expectErr: false,
},
{
name: "missing secret returns an error",
selector: &corev1api.SecretKeySelector{
LocalObjectReference: corev1api.LocalObjectReference{Name: "missing-secret"},
Key: "creds-key",
},
expectedVal: "",
expectErr: true,
},
{
name: "missing key in existing secret returns an error",
selector: &corev1api.SecretKeySelector{
LocalObjectReference: corev1api.LocalObjectReference{Name: "test-secret"},
Key: "missing-key",
},
expectedVal: "",
expectErr: true,
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
g := NewWithT(t)
val, err := store.Get(tc.selector)
if tc.expectErr {
g.Expect(err).To(HaveOccurred())
} else {
g.Expect(err).ToNot(HaveOccurred())
g.Expect(val).To(Equal(tc.expectedVal))
}
})
}
}
+1 -1
View File
@@ -96,7 +96,7 @@ func ListBackupStorageLocations(ctx context.Context, kbClient client.Client, nam
func GetDefaultBackupStorageLocations(ctx context.Context, kbClient client.Client, namespace string) (*velerov1api.BackupStorageLocationList, error) {
locations := new(velerov1api.BackupStorageLocationList)
defaultLocations := new(velerov1api.BackupStorageLocationList)
if err := kbClient.List(context.Background(), locations, &client.ListOptions{Namespace: namespace}); err != nil {
if err := kbClient.List(ctx, locations, &client.ListOptions{Namespace: namespace}); err != nil {
return defaultLocations, errors.Wrapf(err, "failed to list backup storage locations in namespace %s", namespace)
}
+76
View File
@@ -173,3 +173,79 @@ func TestListBackupStorageLocations(t *testing.T) {
})
}
}
func TestGetDefaultBackupStorageLocations(t *testing.T) {
tests := []struct {
name string
locations *velerov1api.BackupStorageLocationList
expectedDefaults []string
expectedErr bool
}{
{
name: "no default locations",
locations: &velerov1api.BackupStorageLocationList{
Items: []velerov1api.BackupStorageLocation{
*builder.ForBackupStorageLocation("ns-1", "loc-1").Default(false).Result(),
*builder.ForBackupStorageLocation("ns-1", "loc-2").Default(false).Result(),
},
},
expectedDefaults: nil,
expectedErr: false,
},
{
name: "one default location",
locations: &velerov1api.BackupStorageLocationList{
Items: []velerov1api.BackupStorageLocation{
*builder.ForBackupStorageLocation("ns-1", "loc-1").Default(false).Result(),
*builder.ForBackupStorageLocation("ns-1", "loc-2").Default(true).Result(),
},
},
expectedDefaults: []string{"loc-2"},
expectedErr: false,
},
{
name: "multiple default locations",
locations: &velerov1api.BackupStorageLocationList{
Items: []velerov1api.BackupStorageLocation{
*builder.ForBackupStorageLocation("ns-1", "loc-1").Default(true).Result(),
*builder.ForBackupStorageLocation("ns-1", "loc-2").Default(true).Result(),
*builder.ForBackupStorageLocation("ns-1", "loc-3").Default(false).Result(),
},
},
expectedDefaults: []string{"loc-1", "loc-2"},
expectedErr: false,
},
{
name: "empty locations list",
locations: &velerov1api.BackupStorageLocationList{},
expectedDefaults: nil,
expectedErr: false,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
g := NewWithT(t)
client := fake.NewClientBuilder().WithScheme(util.VeleroScheme).WithRuntimeObjects(tt.locations).Build()
defaults, err := GetDefaultBackupStorageLocations(t.Context(), client, "ns-1")
if tt.expectedErr {
g.Expect(err).To(HaveOccurred())
} else {
g.Expect(err).ToNot(HaveOccurred())
var defaultNames []string
for _, loc := range defaults.Items {
defaultNames = append(defaultNames, loc.Name)
}
if tt.expectedDefaults == nil {
g.Expect(defaultNames).To(BeEmpty())
} else {
g.Expect(defaultNames).To(ConsistOf(tt.expectedDefaults))
}
}
})
}
}
@@ -111,6 +111,10 @@ type PodVolumeRestoreStatus struct {
// +optional
Progress shared.DataMoveOperationProgress `json:"progress,omitempty"`
// IncrementalBytes holds the number of bytes restored incrementally
// +optional
IncrementalBytes *int64 `json:"incrementalBytes,omitempty"`
// AcceptedTimestamp records the time the pod volume restore is to be prepared.
// The server's time is used for AcceptedTimestamp
// +optional
@@ -129,9 +133,13 @@ type PodVolumeRestoreStatus struct {
// +kubebuilder:object:root=true
// +kubebuilder:storageversion
// +kubebuilder:printcolumn:name="Status",type="string",JSONPath=".status.phase",description="PodVolumeRestore status such as New/InProgress"
// The "Restore Type" column is hidden by default to align with PVB.
// +kubebuilder:printcolumn:name="Restore Type",type="string",JSONPath=".spec.restoreType",description="Restore type such as Full/Incremental",priority=10
// +kubebuilder:printcolumn:name="Started",type="date",JSONPath=".status.startTimestamp",description="Time duration since this PodVolumeRestore was started"
// +kubebuilder:printcolumn:name="Bytes Done",type="integer",format="int64",JSONPath=".status.progress.bytesDone",description="Completed bytes"
// +kubebuilder:printcolumn:name="Total Bytes",type="integer",format="int64",JSONPath=".status.progress.totalBytes",description="Total bytes"
// The "Incremental Bytes" column is hidden by default to align with PVB.
// +kubebuilder:printcolumn:name="Incremental Bytes",type="integer",format="int64",JSONPath=".status.incrementalBytes",description="Incremental bytes",priority=10
// +kubebuilder:printcolumn:name="Storage Location",type="string",JSONPath=".spec.backupStorageLocation",description="Name of the Backup Storage Location where the backup data is stored"
// +kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp",description="Time duration since this PodVolumeRestore was created"
// +kubebuilder:printcolumn:name="Node",type="string",JSONPath=".status.node",description="Name of the node where the PodVolumeRestore is processed"
@@ -1170,6 +1170,11 @@ func (in *PodVolumeRestoreStatus) DeepCopyInto(out *PodVolumeRestoreStatus) {
*out = (*in).DeepCopy()
}
out.Progress = in.Progress
if in.IncrementalBytes != nil {
in, out := &in.IncrementalBytes, &out.IncrementalBytes
*out = new(int64)
**out = **in
}
if in.AcceptedTimestamp != nil {
in, out := &in.AcceptedTimestamp, &out.AcceptedTimestamp
*out = (*in).DeepCopy()
@@ -132,6 +132,10 @@ type DataDownloadStatus struct {
// +optional
Progress shared.DataMoveOperationProgress `json:"progress,omitempty"`
// IncrementalBytes holds the number of bytes restored incrementally since the last snapshot
// +optional
IncrementalBytes *int64 `json:"incrementalBytes,omitempty"`
// Node is name of the node where the DataDownload is processed.
// +optional
Node string `json:"node,omitempty"`
@@ -154,9 +158,13 @@ type DataDownloadStatus struct {
// +kubebuilder:object:root=true
// +kubebuilder:storageversion
// +kubebuilder:printcolumn:name="Status",type="string",JSONPath=".status.phase",description="DataDownload status such as New/InProgress"
// The "Restore Type" column is hidden by default to align with DataUpload.
// +kubebuilder:printcolumn:name="Restore Type",type="string",JSONPath=".spec.restoreType",description="Restore type such as Full/Incremental",priority=10
// +kubebuilder:printcolumn:name="Started",type="date",JSONPath=".status.startTimestamp",description="Time duration since this DataDownload was started"
// +kubebuilder:printcolumn:name="Bytes Done",type="integer",format="int64",JSONPath=".status.progress.bytesDone",description="Completed bytes"
// +kubebuilder:printcolumn:name="Total Bytes",type="integer",format="int64",JSONPath=".status.progress.totalBytes",description="Total bytes"
// The "Incremental Bytes" column is hidden by default to align with DataUpload.
// +kubebuilder:printcolumn:name="Incremental Bytes",type="integer",format="int64",JSONPath=".status.incrementalBytes",description="Incremental bytes",priority=10
// +kubebuilder:printcolumn:name="Storage Location",type="string",JSONPath=".spec.backupStorageLocation",description="Name of the Backup Storage Location where the backup data is stored"
// +kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp",description="Time duration since this DataDownload was created"
// +kubebuilder:printcolumn:name="Node",type="string",JSONPath=".status.node",description="Name of the node where the DataDownload is processed"
@@ -123,6 +123,11 @@ func (in *DataDownloadStatus) DeepCopyInto(out *DataDownloadStatus) {
*out = (*in).DeepCopy()
}
out.Progress = in.Progress
if in.IncrementalBytes != nil {
in, out := &in.IncrementalBytes, &out.IncrementalBytes
*out = new(int64)
**out = **in
}
if in.AcceptedTimestamp != nil {
in, out := &in.AcceptedTimestamp, &out.AcceptedTimestamp
*out = (*in).DeepCopy()
+61 -1
View File
@@ -478,12 +478,15 @@ func (r *itemCollector) getResourceItems(
namespacesToList = []string{""}
}
grString := gr.String()
hasNamespacedPolicies := len(r.backupRequest.NamespacedFilterMap) > 0
var items []*kubernetesResource
for _, namespace := range namespacesToList {
// Check per-namespace resource type filter from ResourcePolicy
if nsFilter := r.backupRequest.GetNamespaceFilter(namespace); nsFilter != nil {
_, hasSpecific := nsFilter.ResourceFilterMap[gr.String()]
_, hasSpecific := nsFilter.ResourceFilterMap[grString]
if !hasSpecific && nsFilter.CatchAllFilter == nil {
log.Debugf("Skipping resource %s in namespace %s: not in resourceFilters",
gr, namespace)
@@ -497,9 +500,66 @@ func (r *itemCollector) getResourceItems(
continue
}
var lastNS string
var lastNSFilter *ResolvedNamespaceFilter
// Collect items in included Namespaces
for i := range unstructuredItems {
item := &unstructuredItems[i]
itemNS := item.GetNamespace()
// Apply namespace inclusion/exclusion and fine-grained filter policies in-memory for cluster-wide queries.
if itemNS != "" && namespace == "" {
if r.backupRequest.NamespaceIncludesExcludes != nil &&
!r.backupRequest.NamespaceIncludesExcludes.ShouldInclude(itemNS) {
log.Debugf("Skipping resource %s in namespace %s: namespace excluded",
gr, itemNS)
continue
}
if hasNamespacedPolicies {
if itemNS != lastNS {
lastNS = itemNS
lastNSFilter = r.backupRequest.GetNamespaceFilter(itemNS)
}
if lastNSFilter != nil {
rf := lastNSFilter.ResourceFilterMap[grString]
if rf == nil {
rf = lastNSFilter.CatchAllFilter
}
if rf == nil {
log.Debugf("Skipping resource %s in namespace %s: not in resourceFilters",
gr, itemNS)
continue
}
// In-memory label selector checks for fine-grained filters
if rf.LabelSelector != nil || len(rf.OrLabelSelectors) > 0 {
itemLabels := labels.Set(item.GetLabels())
if rf.LabelSelector != nil && !rf.LabelSelector.Matches(itemLabels) {
log.Debugf("Skipping resource %s in namespace %s: does not match labelSelector",
gr, itemNS)
continue
}
if len(rf.OrLabelSelectors) > 0 {
matched := false
for _, s := range rf.OrLabelSelectors {
if s.Matches(itemLabels) {
matched = true
break
}
}
if !matched {
log.Debugf("Skipping resource %s in namespace %s: does not match orLabelSelectors",
gr, itemNS)
continue
}
}
}
}
}
}
path, err := r.writeToFile(item)
if err != nil {
+187
View File
@@ -466,3 +466,190 @@ func TestGetOrderedResourcesForTypeTrimsSpaces(t *testing.T) {
{namespace: "ns1", name: "pod3"},
}, sorted)
}
func TestGetResourceItems_NamespacedFilterPolicies_ClusterWideListing(t *testing.T) {
// Simulate cluster-wide API calls where client is called with namespace=""
prodSA := unstructured.Unstructured{
Object: map[string]any{
"apiVersion": "v1",
"kind": "ServiceAccount",
"metadata": map[string]any{
"name": "default",
"namespace": "production",
},
},
}
defaultSA := unstructured.Unstructured{
Object: map[string]any{
"apiVersion": "v1",
"kind": "ServiceAccount",
"metadata": map[string]any{
"name": "default",
"namespace": "default",
},
},
}
kubeSystemSA := unstructured.Unstructured{
Object: map[string]any{
"apiVersion": "v1",
"kind": "ServiceAccount",
"metadata": map[string]any{
"name": "default",
"namespace": "kube-system",
},
},
}
prodCM1 := unstructured.Unstructured{
Object: map[string]any{
"apiVersion": "v1",
"kind": "ConfigMap",
"metadata": map[string]any{
"name": "app-config",
"namespace": "production",
"labels": map[string]any{
"app": "frontend",
},
},
},
}
prodCM2 := unstructured.Unstructured{
Object: map[string]any{
"apiVersion": "v1",
"kind": "ConfigMap",
"metadata": map[string]any{
"name": "other-config",
"namespace": "production",
"labels": map[string]any{
"app": "backend",
},
},
},
}
defaultCM := unstructured.Unstructured{
Object: map[string]any{
"apiVersion": "v1",
"kind": "ConfigMap",
"metadata": map[string]any{
"name": "my-config",
"namespace": "default",
},
},
}
dcClusterWideSA := &test.FakeDynamicClient{}
dcClusterWideSA.On("List", mock.Anything).Return(&unstructured.UnstructuredList{Items: []unstructured.Unstructured{prodSA, defaultSA, kubeSystemSA}}, nil)
dcClusterWideCM := &test.FakeDynamicClient{}
dcClusterWideCM.On("List", mock.Anything).Return(&unstructured.UnstructuredList{Items: []unstructured.Unstructured{prodCM1, prodCM2, defaultCM}}, nil)
factory := &test.FakeDynamicFactory{}
factory.On("ClientForGroupVersionResource", schema.GroupVersion{Version: "v1"}, metav1.APIResource{Name: "serviceaccounts", Namespaced: true, Kind: "ServiceAccount"}, "").Return(dcClusterWideSA, nil)
factory.On("ClientForGroupVersionResource", schema.GroupVersion{Version: "v1"}, metav1.APIResource{Name: "configmaps", Namespaced: true, Kind: "ConfigMap"}, "").Return(dcClusterWideCM, nil)
frontendSelector, err := metav1.LabelSelectorAsSelector(&metav1.LabelSelector{
MatchLabels: map[string]string{"app": "frontend"},
})
require.NoError(t, err)
req := &Request{
Backup: builder.ForBackup("velero", "backup").Result(),
NamespaceIncludesExcludes: collections.NewNamespaceIncludesExcludes().
Includes("*").
Excludes("kube-system"),
NamespacedFilterMap: map[string]*ResolvedNamespaceFilter{
"production": {
ResourceFilterMap: map[string]*ResolvedResourceFilter{
"configmaps": {
LabelSelector: frontendSelector,
},
},
},
},
ResourceIncludesExcludes: includeAllIE{},
}
tempDir := t.TempDir()
r := &itemCollector{
backupRequest: req,
dynamicFactory: factory,
discoveryHelper: test.NewFakeDiscoveryHelper(true, nil),
log: test.NewLogger(),
dir: tempDir,
}
// 1. ServiceAccounts:
// - production should be skipped because ServiceAccount is not in production's resourceFilters
// - kube-system should be skipped because kube-system is in excludedNamespaces
// - default should be included
saResource := metav1.APIResource{Name: "serviceaccounts", Namespaced: true, Kind: "ServiceAccount"}
saItems, err := r.getResourceItems(test.NewLogger(), schema.GroupVersion{Version: "v1"}, saResource, nil)
require.NoError(t, err)
require.Len(t, saItems, 1)
assert.Equal(t, "default", saItems[0].namespace)
assert.Equal(t, "default", saItems[0].name)
// 2. ConfigMaps:
// - production/app-config matches label app=frontend and should be included
// - production/other-config has label app=backend and should be skipped by labelSelector
// - default/my-config has no filter policy and should be included
cmResource := metav1.APIResource{Name: "configmaps", Namespaced: true, Kind: "ConfigMap"}
cmItems, err := r.getResourceItems(test.NewLogger(), schema.GroupVersion{Version: "v1"}, cmResource, nil)
require.NoError(t, err)
require.Len(t, cmItems, 2)
var cmNames []string
for _, it := range cmItems {
cmNames = append(cmNames, it.namespace+"/"+it.name)
}
assert.ElementsMatch(t, []string{"production/app-config", "default/my-config"}, cmNames)
}
func TestGetResourceItems_NamespacedFilterPolicies_SpecificNamespaces(t *testing.T) {
defaultSA := unstructured.Unstructured{
Object: map[string]any{
"apiVersion": "v1",
"kind": "ServiceAccount",
"metadata": map[string]any{
"name": "default",
"namespace": "default",
},
},
}
dcDefaultSA := &test.FakeDynamicClient{}
dcDefaultSA.On("List", mock.Anything).Return(&unstructured.UnstructuredList{Items: []unstructured.Unstructured{defaultSA}}, nil)
factory := &test.FakeDynamicFactory{}
// Note: production client is never even requested because production skips ServiceAccount at the loop top!
factory.On("ClientForGroupVersionResource", schema.GroupVersion{Version: "v1"}, metav1.APIResource{Name: "serviceaccounts", Namespaced: true, Kind: "ServiceAccount"}, "default").Return(dcDefaultSA, nil)
req := &Request{
Backup: builder.ForBackup("velero", "backup").Result(),
NamespaceIncludesExcludes: collections.NewNamespaceIncludesExcludes().
Includes("production", "default"),
NamespacedFilterMap: map[string]*ResolvedNamespaceFilter{
"production": {
ResourceFilterMap: map[string]*ResolvedResourceFilter{
"configmaps": {},
},
},
},
ResourceIncludesExcludes: includeAllIE{},
}
tempDir := t.TempDir()
r := &itemCollector{
backupRequest: req,
dynamicFactory: factory,
discoveryHelper: test.NewFakeDiscoveryHelper(true, nil),
log: test.NewLogger(),
dir: tempDir,
}
saResource := metav1.APIResource{Name: "serviceaccounts", Namespaced: true, Kind: "ServiceAccount"}
saItems, err := r.getResourceItems(test.NewLogger(), schema.GroupVersion{Version: "v1"}, saResource, nil)
require.NoError(t, err)
require.Len(t, saItems, 1)
assert.Equal(t, "default", saItems[0].namespace)
assert.Equal(t, "default", saItems[0].name)
}
+35 -4
View File
@@ -25,6 +25,7 @@ import (
"github.com/spf13/cobra"
"github.com/spf13/pflag"
kubeerrs "k8s.io/apimachinery/pkg/util/errors"
"k8s.io/apimachinery/pkg/util/validation"
"k8s.io/client-go/tools/cache"
kbclient "sigs.k8s.io/controller-runtime/pkg/client"
@@ -45,7 +46,21 @@ func NewCreateCommand(f client.Factory, use string) *cobra.Command {
c := &cobra.Command{
Use: use + " NAME",
Short: "Create a backup",
Args: cobra.MaximumNArgs(1),
Args: func(c *cobra.Command, args []string) error {
if err := cobra.MaximumNArgs(1)(c, args); err != nil {
return err
}
fromSchedule, _ := c.Flags().GetString("from-schedule")
if fromSchedule == "" && len(args) == 0 {
return fmt.Errorf("a backup name is required, unless you are creating based on a schedule")
}
if len(args) == 1 {
if errs := validation.IsDNS1123Subdomain(args[0]); len(errs) > 0 {
return fmt.Errorf("invalid backup name %q: %s", args[0], strings.Join(errs, "; "))
}
}
return nil
},
Run: func(c *cobra.Command, args []string) {
cmd.CheckError(o.Complete(args, f))
cmd.CheckError(o.Validate(c, args, f))
@@ -191,11 +206,27 @@ func (o *CreateOptions) Validate(c *cobra.Command, args []string, f client.Facto
return err
}
// Ensure that unless FromSchedule is set, args contains a backup name
if o.FromSchedule == "" && len(args) != 1 {
// Ensure that unless FromSchedule is set, a backup name is required
if o.FromSchedule == "" && o.Name == "" {
return fmt.Errorf("a backup name is required, unless you are creating based on a schedule")
}
// Validate the backup name format whenever a name is provided
if o.Name != "" {
if errs := validation.IsDNS1123Subdomain(o.Name); len(errs) > 0 {
return fmt.Errorf("invalid backup name %q: %s", o.Name, strings.Join(errs, "; "))
}
}
// When a backup name will be generated from the schedule (i.e. FromSchedule
// is set and no explicit name was given), ensure the schedule name leaves
// enough room for the generated timestamp suffix ("-" + 14-digit timestamp,
// 15 characters total) within the DNS1123 subdomain length limit.
if o.FromSchedule != "" && o.Name == "" {
const timestampSuffixLen = 15 // "-" + "20060102150405"
maxScheduleNameLen := validation.DNS1123SubdomainMaxLength - timestampSuffixLen
if len(o.FromSchedule) > maxScheduleNameLen {
return fmt.Errorf("schedule name %q is too long: must be %d characters or fewer to leave room for the generated timestamp suffix", o.FromSchedule, maxScheduleNameLen)
}
}
errs := collections.ValidateNamespaceIncludesExcludes(o.IncludeNamespaces, o.ExcludeNamespaces)
if len(errs) > 0 {
return kubeerrs.NewAggregate(errs)
+125 -1
View File
@@ -245,7 +245,7 @@ func TestCreateOptions_OrderedResources(t *testing.T) {
}
func TestCreateCommand(t *testing.T) {
name := "nameToBeCreated"
name := "name-to-be-created"
args := []string{name}
t.Run("create a backup create command with full options except fromSchedule and wait, then run by create option", func(t *testing.T) {
@@ -457,3 +457,127 @@ func TestCreateCommand(t *testing.T) {
assert.NoError(t, e)
})
}
func TestCreateCommand_Args(t *testing.T) {
testCases := []struct {
name string
args []string
fromSchedule string
expectError bool
}{
{
name: "should error when no name and no from-schedule",
args: []string{},
expectError: true,
},
{
name: "should pass when a valid name is provided",
args: []string{"my-backup"},
expectError: false,
},
{
name: "should error when the name is not a valid DNS1123 subdomain",
args: []string{"Invalid_Name!"},
expectError: true,
},
{
name: "should pass with no name when from-schedule is set",
args: []string{},
fromSchedule: "daily-backup",
expectError: false,
},
{
name: "should error when more than one arg is given",
args: []string{"name1", "name2"},
expectError: true,
},
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
f := &factorymocks.Factory{}
cmd := NewCreateCommand(f, "")
if tc.fromSchedule != "" {
err := cmd.Flags().Set("from-schedule", tc.fromSchedule)
require.NoError(t, err)
}
err := cmd.Args(cmd, tc.args)
if tc.expectError {
assert.Error(t, err)
} else {
assert.NoError(t, err)
}
})
}
}
func TestCreateOptions_Validate(t *testing.T) {
testCases := []struct {
name string
optName string
fromSchedule string
args []string
expectError bool
}{
{
name: "should error when no name and no from-schedule",
optName: "",
args: []string{},
expectError: true,
},
{
name: "should pass with a valid name and no from-schedule",
optName: "my-backup",
args: []string{"my-backup"},
expectError: false,
},
{
name: "should error when name is invalid, regardless of from-schedule",
optName: "Invalid_Name!",
fromSchedule: "daily-backup",
args: []string{"Invalid_Name!"},
expectError: true,
},
{
name: "should pass when from-schedule is set and no name given",
optName: "",
fromSchedule: "daily-backup",
args: []string{},
expectError: false,
},
{
name: "should pass when schedule name leaves room for timestamp suffix",
optName: "",
fromSchedule: strings.Repeat("a", 238), // exactly at the 238-char limit
args: []string{},
expectError: false,
},
{
name: "should error when schedule name is too long to leave room for timestamp suffix",
optName: "",
fromSchedule: strings.Repeat("a", 239), // one over the 238-char limit
args: []string{},
expectError: true,
},
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
f := &factorymocks.Factory{}
cmd := NewCreateCommand(f, "")
o := NewCreateOptions()
o.Name = tc.optName
o.FromSchedule = tc.fromSchedule
err := o.Validate(cmd, tc.args, f)
if tc.expectError {
assert.Error(t, err)
} else {
assert.NoError(t, err)
}
})
}
}
+1 -1
View File
@@ -27,7 +27,7 @@ import (
func CheckError(err error) {
if err != nil {
if err != context.Canceled {
fmt.Fprintf(os.Stderr, "An error occurred: %v\n", err)
fmt.Fprintf(os.Stderr, "velero: %v\n", err)
}
os.Exit(1)
}
+2 -2
View File
@@ -362,7 +362,7 @@ func (b *backupReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
// result in the backup being Failed.
log.WithError(err).Error("backup failed")
request.Status.Phase = velerov1api.BackupPhaseFailed
request.Status.FailureReason = err.Error()
request.Status.FailureReason = fmt.Sprintf("backup execution failed: %v", err)
}
switch request.Status.Phase {
@@ -619,7 +619,7 @@ func (b *backupReconciler) prepareBackupRequest(ctx context.Context, backup *vel
resourcePolicies, err := resourcepolicies.GetResourcePoliciesFromBackupWithGlobal(
*request.Backup, b.kbClient, b.globalVolumePoliciesConfigMap, request.Namespace, logger)
if err != nil {
request.Status.ValidationErrors = append(request.Status.ValidationErrors, err.Error())
request.Status.ValidationErrors = append(request.Status.ValidationErrors, fmt.Sprintf("invalid resource policies: %v", err))
} else if b.globalVolumePoliciesConfigMap != "" {
// Record the contributing global volume policies ConfigMap so `velero backup describe` can surface it.
request.Annotations[velerov1api.GlobalBackupVolumePolicyConfigMapAnnotation] = b.globalVolumePoliciesConfigMap
+2 -2
View File
@@ -1205,7 +1205,7 @@ func TestProcessBackupCompletions(t *testing.T) {
},
Status: velerov1api.BackupStatus{
Phase: velerov1api.BackupPhaseFailed,
FailureReason: "backup already exists in object storage",
FailureReason: "backup execution failed: backup already exists in object storage",
Version: 1,
FormatVersion: "1.1.0",
StartTimestamp: &timestamp,
@@ -1250,7 +1250,7 @@ func TestProcessBackupCompletions(t *testing.T) {
},
Status: velerov1api.BackupStatus{
Phase: velerov1api.BackupPhaseFailed,
FailureReason: "error checking if backup already exists in object storage: Backup already exists in object storage",
FailureReason: "backup execution failed: error checking if backup already exists in object storage: Backup already exists in object storage",
Version: 1,
FormatVersion: "1.1.0",
StartTimestamp: &timestamp,
+7 -7
View File
@@ -315,7 +315,7 @@ func (r *backupDeletionReconciler) Reconcile(ctx context.Context, req ctrl.Reque
volumeSnapshotter, ok := volumeSnapshotters[snapshot.Spec.Location]
if !ok {
if volumeSnapshotter, err = r.volumeSnapshottersForVSL(ctx, backup.Namespace, snapshot.Spec.Location, pluginManager); err != nil {
errs = append(errs, err.Error())
errs = append(errs, errors.Wrapf(err, "error getting volume snapshotter for location %q", snapshot.Spec.Location).Error())
continue
}
volumeSnapshotters[snapshot.Spec.Location] = volumeSnapshotter
@@ -334,7 +334,7 @@ func (r *backupDeletionReconciler) Reconcile(ctx context.Context, req ctrl.Reque
log.Info("Removing pod volume snapshots")
if deleteErrs := r.deletePodVolumeSnapshots(ctx, backup); len(deleteErrs) > 0 {
for _, err := range deleteErrs {
errs = append(errs, err.Error())
errs = append(errs, errors.Wrap(err, "error deleting pod volume snapshots").Error())
}
}
@@ -342,7 +342,7 @@ func (r *backupDeletionReconciler) Reconcile(ctx context.Context, req ctrl.Reque
log.Info("Removing snapshot data by data mover")
if deleteErrs := r.deleteMovedSnapshots(ctx, backup); len(deleteErrs) > 0 {
for _, err := range deleteErrs {
errs = append(errs, err.Error())
errs = append(errs, errors.Wrap(err, "error deleting moved snapshot data").Error())
}
}
duList := &velerov2alpha1.DataUploadList{}
@@ -354,12 +354,12 @@ func (r *backupDeletionReconciler) Reconcile(ctx context.Context, req ctrl.Reque
}),
}); err != nil {
log.WithError(err).Error("Error listing datauploads")
errs = append(errs, err.Error())
errs = append(errs, errors.Wrap(err, "error listing datauploads for backup").Error())
} else {
for i := range duList.Items {
du := duList.Items[i]
if err := r.Delete(ctx, &du); err != nil {
errs = append(errs, err.Error())
errs = append(errs, errors.Wrapf(err, "error deleting dataupload %q", du.Name).Error())
}
}
}
@@ -368,7 +368,7 @@ func (r *backupDeletionReconciler) Reconcile(ctx context.Context, req ctrl.Reque
if backupStore != nil && len(errs) == 0 {
log.Info("Removing backup from backup storage")
if err := backupStore.DeleteBackup(backup.Name); err != nil {
errs = append(errs, err.Error())
errs = append(errs, errors.Wrap(err, "error removing backup from backup storage").Error())
}
} else if len(errs) > 0 {
log.Info("Skipping removal of backup from backup storage due to previous errors")
@@ -631,7 +631,7 @@ func (r *backupDeletionReconciler) patchDeleteBackupRequest(ctx context.Context,
func (r *backupDeletionReconciler) patchDeleteBackupRequestWithError(ctx context.Context, req *velerov1api.DeleteBackupRequest, err error) error {
_, err = r.patchDeleteBackupRequest(ctx, req, func(r *velerov1api.DeleteBackupRequest) {
r.Status.Phase = velerov1api.DeleteBackupRequestPhaseProcessed
r.Status.Errors = []string{err.Error()}
r.Status.Errors = []string{errors.WithMessage(err, "backup deletion failed").Error()}
})
return err
}
@@ -137,7 +137,7 @@ func TestBackupDeletionControllerReconcile(t *testing.T) {
td.fakeClient.Get(ctx, td.req.NamespacedName, res)
assert.Equal(t, "Processed", string(res.Status.Phase))
assert.Len(t, res.Status.Errors, 1)
assert.True(t, strings.HasPrefix(res.Status.Errors[0], "error getting the backup store"))
assert.True(t, strings.HasPrefix(res.Status.Errors[0], "backup deletion failed: error getting the backup store"))
})
t.Run("missing spec.backupName", func(t *testing.T) {
@@ -153,7 +153,7 @@ func TestBackupDeletionControllerReconcile(t *testing.T) {
require.NoError(t, err)
assert.Equal(t, "Processed", string(res.Status.Phase))
assert.Len(t, res.Status.Errors, 1)
assert.Equal(t, "spec.backupName is required", res.Status.Errors[0])
assert.Equal(t, "backup deletion failed: spec.backupName is required", res.Status.Errors[0])
})
t.Run("existing deletion requests for the backup are deleted", func(t *testing.T) {
@@ -221,7 +221,7 @@ func TestBackupDeletionControllerReconcile(t *testing.T) {
require.NoError(t, err)
assert.Equal(t, "Processed", string(res.Status.Phase))
assert.Len(t, res.Status.Errors, 1)
assert.Equal(t, "backup is still in progress", res.Status.Errors[0])
assert.Equal(t, "backup deletion failed: backup is still in progress", res.Status.Errors[0])
})
t.Run("unable to find backup", func(t *testing.T) {
@@ -235,7 +235,7 @@ func TestBackupDeletionControllerReconcile(t *testing.T) {
require.NoError(t, err)
assert.Equal(t, "Processed", string(res.Status.Phase))
assert.Len(t, res.Status.Errors, 1)
assert.Equal(t, "backup not found", res.Status.Errors[0])
assert.Equal(t, "backup deletion failed: backup not found", res.Status.Errors[0])
})
t.Run("unable to find backup storage location", func(t *testing.T) {
backup := builder.ForBackup(velerov1api.DefaultNamespace, "foo").StorageLocation("default").Result()
@@ -250,7 +250,7 @@ func TestBackupDeletionControllerReconcile(t *testing.T) {
require.NoError(t, err)
assert.Equal(t, "Processed", string(res.Status.Phase))
assert.Len(t, res.Status.Errors, 1)
assert.Equal(t, "backup storage location default not found", res.Status.Errors[0])
assert.Equal(t, "backup deletion failed: backup storage location default not found", res.Status.Errors[0])
})
t.Run("backup storage location is in read-only mode", func(t *testing.T) {
@@ -267,7 +267,7 @@ func TestBackupDeletionControllerReconcile(t *testing.T) {
require.NoError(t, err)
assert.Equal(t, "Processed", string(res.Status.Phase))
assert.Len(t, res.Status.Errors, 1)
assert.Equal(t, "cannot delete backup because backup storage location default is currently in read-only mode", res.Status.Errors[0])
assert.Equal(t, "backup deletion failed: cannot delete backup because backup storage location default is currently in read-only mode", res.Status.Errors[0])
})
t.Run("backup storage location is in unavailable state", func(t *testing.T) {
@@ -284,7 +284,7 @@ func TestBackupDeletionControllerReconcile(t *testing.T) {
require.NoError(t, err)
assert.Equal(t, "Processed", string(res.Status.Phase))
assert.Len(t, res.Status.Errors, 1)
assert.Equal(t, "cannot delete backup because backup storage location default is currently in Unavailable state", res.Status.Errors[0])
assert.Equal(t, "backup deletion failed: cannot delete backup because backup storage location default is currently in Unavailable state", res.Status.Errors[0])
})
t.Run("full delete, no errors", func(t *testing.T) {
@@ -877,7 +877,7 @@ func TestBackupDeletionControllerReconcile(t *testing.T) {
require.NoError(t, err)
assert.Equal(t, "Processed", string(res.Status.Phase))
assert.Len(t, res.Status.Errors, 1)
assert.Equal(t, "backup not found", res.Status.Errors[0])
assert.Equal(t, "backup deletion failed: backup not found", res.Status.Errors[0])
})
}
+14 -14
View File
@@ -439,7 +439,7 @@ var _ = Describe("Backup Sync Reconciler", func() {
}
if test.location != nil {
Expect(r.client.Create(ctx, test.location)).ShouldNot(HaveOccurred())
Expect(r.client.Create(ctx, test.location)).ToNot(HaveOccurred())
backupStores[test.location.Name] = &persistencemocks.BackupStore{}
backupStore, ok := backupStores[test.location.Name]
@@ -457,12 +457,12 @@ var _ = Describe("Backup Sync Reconciler", func() {
for _, existingBackup := range test.existingBackups {
err := client.Create(context.TODO(), existingBackup, &ctrlClient.CreateOptions{})
Expect(err).ShouldNot(HaveOccurred())
Expect(err).ToNot(HaveOccurred())
}
for _, existingPodVolumeBackup := range test.existingPodVolumeBackups {
err := client.Create(context.TODO(), existingPodVolumeBackup, &ctrlClient.CreateOptions{})
Expect(err).ShouldNot(HaveOccurred())
Expect(err).ToNot(HaveOccurred())
}
actualResult, err := r.Reconcile(ctx, ctrl.Request{
@@ -534,7 +534,7 @@ var _ = Describe("Backup Sync Reconciler", func() {
cloudBackupData.backup.Status.Expiration.After(fakeClock.Now())) {
Expect(apierrors.IsNotFound(err)).To(BeTrue())
} else {
Expect(err).ShouldNot(HaveOccurred())
Expect(err).ToNot(HaveOccurred())
// did this cloud pod volume backup already exist in the cluster?
var existingPodVolumeBackup *velerov1api.PodVolumeBackup
@@ -673,7 +673,7 @@ var _ = Describe("Backup Sync Reconciler", func() {
}
queueScheme := runtime.NewScheme()
Expect(velerov1api.AddToScheme(queueScheme)).ShouldNot(HaveOccurred())
Expect(velerov1api.AddToScheme(queueScheme)).ToNot(HaveOccurred())
for _, test := range tests {
var (
@@ -693,7 +693,7 @@ var _ = Describe("Backup Sync Reconciler", func() {
logger: velerotest.NewLogger(),
}
Expect(client.Create(ctx, location)).ShouldNot(HaveOccurred(), test.name)
Expect(client.Create(ctx, location)).ToNot(HaveOccurred(), test.name)
backupStore := &persistencemocks.BackupStore{}
backupStores[location.Name] = backupStore
backupStore.On("ListBackups").Return([]string{test.cloudBackup.Name}, nil)
@@ -704,7 +704,7 @@ var _ = Describe("Backup Sync Reconciler", func() {
_, err := syncReconciler.Reconcile(ctx, ctrl.Request{
NamespacedName: types.NamespacedName{Namespace: location.Namespace, Name: location.Name},
})
Expect(err).ShouldNot(HaveOccurred(), test.name)
Expect(err).ToNot(HaveOccurred(), test.name)
backupKey := types.NamespacedName{Namespace: "ns-1", Name: test.cloudBackup.Name}
synced := &velerov1api.Backup{}
@@ -714,7 +714,7 @@ var _ = Describe("Backup Sync Reconciler", func() {
Expect(apierrors.IsNotFound(err)).To(BeTrue(), test.name)
continue
}
Expect(err).ShouldNot(HaveOccurred(), test.name)
Expect(err).ToNot(HaveOccurred(), test.name)
// Reconcile the synced backup with the queue controller twice: the first
// reconcile would move a New/empty-phase backup to Queued, the second one
@@ -723,11 +723,11 @@ var _ = Describe("Backup Sync Reconciler", func() {
queueReconciler := NewBackupQueueReconciler(client, queueScheme, velerotest.NewLogger(), 1, NewBackupTracker())
for range 2 {
_, err = queueReconciler.Reconcile(ctx, ctrl.Request{NamespacedName: backupKey})
Expect(err).ShouldNot(HaveOccurred(), test.name)
Expect(err).ToNot(HaveOccurred(), test.name)
}
after := &velerov1api.Backup{}
Expect(client.Get(ctx, backupKey, after)).ShouldNot(HaveOccurred(), test.name)
Expect(client.Get(ctx, backupKey, after)).ToNot(HaveOccurred(), test.name)
Expect(after.Status.Phase).To(BeEquivalentTo(test.expectPhase), test.name)
// Hooks are dropped on sync, so the stored metadata cannot carry a payload
// that a later code path could execute.
@@ -880,7 +880,7 @@ var _ = Describe("Backup Sync Reconciler", func() {
for _, backup := range test.k8sBackups {
// add test backup to client
err := client.Create(context.TODO(), backup, &ctrlClient.CreateOptions{})
Expect(err).ShouldNot(HaveOccurred())
Expect(err).ToNot(HaveOccurred())
}
bslName := "default"
@@ -890,7 +890,7 @@ var _ = Describe("Backup Sync Reconciler", func() {
r.deleteOrphanedBackups(ctx, bslName, test.cloudBackups, velerotest.NewLogger())
numBackups, err := numBackups(client)
Expect(err).ShouldNot(HaveOccurred())
Expect(err).ToNot(HaveOccurred())
fmt.Println("")
@@ -913,7 +913,7 @@ var _ = Describe("Backup Sync Reconciler", func() {
testObjList := backupSyncSourceOrderFunc(locationList)
testObjArray, err := meta.ExtractList(testObjList)
Expect(err).ShouldNot(HaveOccurred())
Expect(err).ToNot(HaveOccurred())
expectLocation := testObjArray[0].(*velerov1api.BackupStorageLocation)
Expect(expectLocation.Spec.Default).To(BeEquivalentTo(true))
@@ -1085,7 +1085,7 @@ var _ = Describe("Backup Sync Reconciler", func() {
//create all required schedules as needed.
for _, creatable := range test.toCreate {
err := b.client.Create(context.Background(), creatable)
Expect(err).ShouldNot(HaveOccurred())
Expect(err).ToNot(HaveOccurred())
}
references := b.filterBackupOwnerReferences(context.Background(), test.backup, logger)
@@ -32,6 +32,7 @@ import (
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/client-go/kubernetes"
"k8s.io/utils/clock"
"k8s.io/utils/ptr"
ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/builder"
"sigs.k8s.io/controller-runtime/pkg/client"
@@ -501,6 +502,7 @@ func (r *DataDownloadReconciler) OnDataDownloadCompleted(ctx context.Context, na
}
dd.Status.Phase = velerov2alpha1api.DataDownloadPhaseCompleted
dd.Status.IncrementalBytes = ptr.To(result.Restore.IncrementalBytes)
dd.Status.CompletionTimestamp = &metav1.Time{Time: r.Clock.Now()}
delete(dd.Labels, exposer.ExposeOnGoingLabel)
@@ -784,10 +784,9 @@ func UpdatePVBStatusToFailed(ctx context.Context, c client.Client, pvb *velerov1
pvb.Status.SnapshotID = dataPathError.GetSnapshotID()
}
if len(strings.TrimSpace(msg)) == 0 {
pvb.Status.Message = errOut.Error()
} else {
pvb.Status.Message = errors.WithMessage(errOut, msg).Error()
msg = "pod volume backup failed"
}
pvb.Status.Message = errors.WithMessage(errOut, msg).Error()
if pvb.Status.StartTimestamp.IsZero() {
pvb.Status.StartTimestamp = &metav1.Time{Time: time}
}
@@ -33,6 +33,7 @@ import (
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/client-go/kubernetes"
clocks "k8s.io/utils/clock"
"k8s.io/utils/ptr"
ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/builder"
"sigs.k8s.io/controller-runtime/pkg/client"
@@ -837,6 +838,7 @@ func (r *PodVolumeRestoreReconciler) OnDataPathCompleted(ctx context.Context, na
pvr.Status.Phase = velerov1api.PodVolumeRestorePhaseCompleted
pvr.Status.CompletionTimestamp = &metav1.Time{Time: r.clock.Now()}
pvr.Status.IncrementalBytes = ptr.To(result.Restore.IncrementalBytes)
delete(pvr.Labels, exposer.ExposeOnGoingLabel)
+4 -4
View File
@@ -275,7 +275,7 @@ func (r *restoreReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
if err := r.runValidatedRestore(restore, info, resourceModifiers, restoreResPolicies); err != nil {
log.WithError(err).Debug("Restore failed")
restore.Status.Phase = api.RestorePhaseFailed
restore.Status.FailureReason = err.Error()
restore.Status.FailureReason = fmt.Sprintf("restore execution failed: %v", err)
r.metrics.RegisterRestoreFailed(backupScheduleName)
}
@@ -349,7 +349,7 @@ func (r *restoreReconciler) validateAndComplete(ctx context.Context, restore *ap
// validate Restore Init Hook's InitContainers
restoreHooks, err := hook.GetRestoreHooksFromSpec(&restore.Spec.Hooks)
if err != nil {
restore.Status.ValidationErrors = append(restore.Status.ValidationErrors, err.Error())
restore.Status.ValidationErrors = append(restore.Status.ValidationErrors, fmt.Sprintf("invalid restore hooks: %v", err))
}
for _, resource := range restoreHooks {
for _, h := range resource.RestoreHooks {
@@ -357,7 +357,7 @@ func (r *restoreReconciler) validateAndComplete(ctx context.Context, restore *ap
for _, container := range h.Init.InitContainers {
err = hook.ValidateContainer(container.Raw)
if err != nil {
restore.Status.ValidationErrors = append(restore.Status.ValidationErrors, err.Error())
restore.Status.ValidationErrors = append(restore.Status.ValidationErrors, fmt.Sprintf("invalid init container in restore hook %q: %v", resource.Name, err))
}
}
}
@@ -433,7 +433,7 @@ func (r *restoreReconciler) validateAndComplete(ctx context.Context, restore *ap
)
if err != nil {
restore.Status.ValidationErrors = append(
restore.Status.ValidationErrors, err.Error(),
restore.Status.ValidationErrors, fmt.Sprintf("invalid restore resource policies: %v", err),
)
return backupInfo{}, nil, nil
}
@@ -92,7 +92,7 @@ var _ = Describe("Server Status Request Reconciler", func() {
Expect(apierrors.IsNotFound(err)).To(BeTrue())
} else {
Expect(err).ToNot(HaveOccurred())
Eventually(instance.Status.Phase == test.expected.Status.Phase, timeout).Should(BeTrue())
Expect(instance.Status.Phase).To(Equal(test.expected.Status.Phase))
}
},
Entry("with phase=empty will be processed and phased successfully patched", request{
+1 -1
View File
@@ -102,7 +102,7 @@ func TestOnDataDownloadCompleted(t *testing.T) {
{
name: "marshal fail",
marshalErr: errors.New("fake-marshal-error"),
expectedErr: "Failed to marshal restore result {{ } 0}: fake-marshal-error",
expectedErr: "Failed to marshal restore result {{ } 0 0}: fake-marshal-error",
},
{
name: "succeed",
+2 -2
View File
@@ -250,7 +250,7 @@ func (dp *generalDataPath) StartRestore(snapshotID string, target AccessPoint, u
dp.wgDataPath.Done()
}()
totalBytes, err := dp.uploaderProv.RunRestore(dp.ctx, snapshotID, target.ByPath, restoreParam.Incremental,
incrementalBytes, totalBytes, err := dp.uploaderProv.RunRestore(dp.ctx, snapshotID, target.ByPath, restoreParam.Incremental,
provider.CBTParam{
Source: cbtservice.SourceInfo{
Snapshot: restoreParam.VolumeSnapshotName,
@@ -268,7 +268,7 @@ func (dp *generalDataPath) StartRestore(snapshotID string, target AccessPoint, u
}
dp.callbacks.OnFailed(context.Background(), dp.namespace, dp.jobName, dataPathErr)
} else {
dp.callbacks.OnCompleted(context.Background(), dp.namespace, dp.jobName, Result{Restore: RestoreResult{Target: target, TotalBytes: totalBytes}})
dp.callbacks.OnCompleted(context.Background(), dp.namespace, dp.jobName, Result{Restore: RestoreResult{Target: target, TotalBytes: totalBytes, IncrementalBytes: incrementalBytes}})
}
}()
+1 -1
View File
@@ -190,7 +190,7 @@ func TestAsyncRestore(t *testing.T) {
t.Run(test.name, func(t *testing.T) {
dp := newGeneralDataPath("job-1", "test", nil, "velero", Callbacks{}, velerotest.NewLogger()).(*generalDataPath)
mockProvider := providerMock.NewProvider(t)
mockProvider.On("RunRestore", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(test.result.Restore.TotalBytes, test.err)
mockProvider.On("RunRestore", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(test.result.Restore.IncrementalBytes, test.result.Restore.TotalBytes, test.err)
mockProvider.On("Close", mock.Anything).Return(nil)
dp.uploaderProv = mockProvider
dp.initialized = true
+3 -2
View File
@@ -43,8 +43,9 @@ type BackupResult struct {
// RestoreResult represents the result of a restore
type RestoreResult struct {
Target AccessPoint `json:"target,omitempty"`
TotalBytes int64 `json:"totalBytes,omitempty"`
Target AccessPoint `json:"target,omitempty"`
TotalBytes int64 `json:"totalBytes,omitempty"`
IncrementalBytes int64 `json:"incrementalBytes,omitempty"`
}
// Callbacks defines the collection of callbacks during backup/restore
+70 -22
View File
@@ -252,6 +252,32 @@ func (e *genericRestoreExposer) Expose(ctx context.Context, ownerObject corev1ap
}
}()
// Get volumeID before creating the restore pod because the existingPV may be deleted when creating the PVC if the volume policy is different
var volumeID string
if param.CSI != nil && param.CSI.Snapshot != nil {
vs := &snapshotv1api.VolumeSnapshot{}
if err := e.ctrlClient.Get(ctx, client.ObjectKey{
Namespace: param.CSI.Snapshot.VolumeSnapshotNamespace,
Name: param.CSI.Snapshot.VolumeSnapshot,
}, vs); err != nil {
return errors.Wrapf(err, "error to get volume snapshot %s/%s", param.CSI.Snapshot.VolumeSnapshotNamespace, param.CSI.Snapshot.VolumeSnapshot)
}
var vsc *snapshotv1api.VolumeSnapshotContent
vsc, err = csi.GetVSCForVS(ctx, vs, e.ctrlClient)
if err != nil {
return errors.Wrapf(err, "error to get volume snapshot content for volume snapshot %s/%s", vs.Namespace, vs.Name)
}
var cbtInfo csi.CBTInfo
cbtInfo, err = csi.GetCBTInfo(ctx, e.kubeClient, e.log, vs, vsc, param.TargetPVName)
if err != nil {
return errors.Wrap(err, "error to get CBT info")
}
curLog.Debugf("CBT info: %+v", cbtInfo)
volumeID = cbtInfo.VolumeID
}
curLog.Info("Creating restore PVC")
var targetPV *corev1api.PersistentVolume
@@ -280,30 +306,26 @@ func (e *genericRestoreExposer) Expose(ctx context.Context, ownerObject corev1ap
}
}()
curLog.Info("Creating restore pod")
var volumeID string
if param.CSI != nil && param.CSI.Snapshot != nil {
vs := &snapshotv1api.VolumeSnapshot{}
if err := e.ctrlClient.Get(ctx, client.ObjectKey{
Namespace: param.CSI.Snapshot.VolumeSnapshotNamespace,
Name: param.CSI.Snapshot.VolumeSnapshot,
}, vs); err != nil {
return errors.Wrapf(err, "error to get volume snapshot %s/%s", param.CSI.Snapshot.VolumeSnapshotNamespace, param.CSI.Snapshot.VolumeSnapshot)
var volumeTopology *corev1api.NodeSelector
if !e.validateSelectedNode(ctx, selectedNode, param.DataMover, curLog) {
curLog.WithField("pvc name", restorePVC.Name).Infof("Getting volume topology and ignore selected node %s", selectedNode)
selectedNode = ""
var restorePV *corev1api.PersistentVolume
restorePV, err = kube.WaitPVCBound(ctx, e.kubeClient.CoreV1(), e.kubeClient.CoreV1(), restorePVC.Name, restorePVC.Namespace, param.ExposeTimeout)
if err != nil {
return errors.Wrap(err, "error waiting for restore PVC bound")
}
vsc, err := csi.GetVSCForVS(ctx, vs, e.ctrlClient)
if err != nil {
return errors.Wrapf(err, "error to get volume snapshot content for volume snapshot %s/%s", vs.Namespace, vs.Name)
if tp, err := kube.GetVolumeTopology(ctx, e.kubeClient.CoreV1(), e.kubeClient.StorageV1(), restorePV.Name, restorePV.Spec.StorageClassName); err != nil {
return errors.Wrapf(err, "error getting volume topology for PV %s, storage class %s", restorePV.Name, restorePV.Spec.StorageClassName)
} else {
volumeTopology = tp
}
var cbtInfo csi.CBTInfo
cbtInfo, err = csi.GetCBTInfo(ctx, e.kubeClient, e.log, vs, vsc, param.TargetPVName)
if err != nil {
return errors.Wrap(err, "error to get CBT info")
}
curLog.Debugf("CBT info: %+v", cbtInfo)
volumeID = cbtInfo.VolumeID
}
curLog.Info("Creating restore pod")
var csiSnapshotMetadataServiceConfigs *velerotypes.CSISnapshotMetadataService
if param.CSI != nil {
csiSnapshotMetadataServiceConfigs = param.CSI.SnapshotMetadataServiceConfigs
@@ -325,6 +347,7 @@ func (e *genericRestoreExposer) Expose(ctx context.Context, ownerObject corev1ap
param.TargetNamespace,
volumeID,
csiSnapshotMetadataServiceConfigs,
volumeTopology,
)
if err != nil {
return errors.Wrapf(err, "error to create restore pod")
@@ -725,6 +748,7 @@ func (e *genericRestoreExposer) createRestorePod(
volumeSnapshotNamespace string,
volumeID string,
csiSnapshotMetadataServiceConfigs *velerotypes.CSISnapshotMetadataService,
volumeTopology *corev1api.NodeSelector,
) (*corev1api.Pod, error) {
restorePodName := ownerObject.Name
restorePVCName := ownerObject.Name
@@ -863,7 +887,7 @@ func (e *genericRestoreExposer) createRestorePod(
})
}
podAffinity := kube.ToSystemAffinity(affinity, nil)
podAffinity := kube.ToSystemAffinity(affinity, volumeTopology)
pod := &corev1api.Pod{
ObjectMeta: metav1.ObjectMeta{
@@ -1004,7 +1028,7 @@ func (e *genericRestoreExposer) createRestorePVC(ctx context.Context, ownerObjec
Spec: *targetPV.Spec.DeepCopy(),
}
tmpPV.Spec.VolumeMode = restorePVC.Spec.VolumeMode
e.log.Infof("the volume mode is different, creating temporary PV %s with volume mode %s", tmpPV.Name, tmpPV.Spec.VolumeMode)
e.log.Infof("the volume mode is different, creating temporary PV %s with volume mode %v", tmpPV.Name, tmpPV.Spec.VolumeMode)
tmpPV, err = e.kubeClient.CoreV1().PersistentVolumes().Create(ctx, tmpPV, metav1.CreateOptions{})
if err != nil {
return nil, errors.Wrapf(err, "fail to create the temporary PV %s", volumeName)
@@ -1034,3 +1058,27 @@ func (e *genericRestoreExposer) createRestorePVC(ctx context.Context, ownerObjec
return restorePVC, nil
}
func (e *genericRestoreExposer) validateSelectedNode(ctx context.Context, node string, dataMover string, log logrus.FieldLogger) bool {
if node == "" {
return true
}
os, err := kube.GetNodeOS(ctx, node, e.kubeClient.CoreV1())
if err != nil {
log.WithError(err).Warnf("Unable to get OS for selected node %s", node)
return false
}
if os != kube.NodeOSLinux && os != kube.NodeOSWindows {
log.Warnf("Unsupported OS for selected node %s", node)
return false
}
if dataMover == datamover.DataMoverTypeVeleroBlock && os != kube.NodeOSLinux {
log.Infof("Block data mover will not use selected node %s because its OS %s is not supported", node, os)
return false
}
return true
}
@@ -152,6 +152,7 @@ func TestCreateRestorePodWithPriorityClass(t *testing.T) {
"", // volumeSnapshotNamespace
"", // volumeID
nil,
nil, // volumeTopology
)
require.NoError(t, err, tc.description)
@@ -235,6 +236,7 @@ func TestCreateRestorePodWithMissingConfigMap(t *testing.T) {
"", // volumeSnapshotNamespace
"", // volumeID
nil,
nil, // volumeTopology
)
// Should succeed even when config map is missing
+241 -15
View File
@@ -106,6 +106,65 @@ func TestRestoreExpose(t *testing.T) {
},
}
targetPVCObjWithNode := &corev1api.PersistentVolumeClaim{
ObjectMeta: metav1.ObjectMeta{
Namespace: "fake-ns",
Name: "fake-target-pvc-with-node",
Annotations: map[string]string{
"volume.kubernetes.io/selected-node": "fake-node",
},
},
Spec: corev1api.PersistentVolumeClaimSpec{
StorageClassName: &scName,
},
}
volumeBindingMode := storagev1api.VolumeBindingWaitForFirstConsumer
storageClassWaitForFirstConsumer := &storagev1api.StorageClass{
ObjectMeta: metav1.ObjectMeta{
Name: "fake-sc",
},
VolumeBindingMode: &volumeBindingMode,
}
restorePVCObjBound := &corev1api.PersistentVolumeClaim{
ObjectMeta: metav1.ObjectMeta{
Namespace: velerov1.DefaultNamespace,
Name: "fake-restore",
},
Spec: corev1api.PersistentVolumeClaimSpec{
VolumeName: "fake-restore-pv",
StorageClassName: &scName,
},
Status: corev1api.PersistentVolumeClaimStatus{
Phase: corev1api.ClaimBound,
},
}
restorePVObjWithTopology := &corev1api.PersistentVolume{
ObjectMeta: metav1.ObjectMeta{
Name: "fake-restore-pv",
},
Spec: corev1api.PersistentVolumeSpec{
StorageClassName: "fake-sc",
NodeAffinity: &corev1api.VolumeNodeAffinity{
Required: &corev1api.NodeSelector{
NodeSelectorTerms: []corev1api.NodeSelectorTerm{
{
MatchExpressions: []corev1api.NodeSelectorRequirement{
{
Key: "topology.kubernetes.io/zone",
Operator: corev1api.NodeSelectorOpIn,
Values: []string{"zone-1"},
},
},
},
},
},
},
},
}
daemonSet := &appsv1api.DaemonSet{
ObjectMeta: metav1.ObjectMeta{
Namespace: "velero",
@@ -129,20 +188,22 @@ func TestRestoreExpose(t *testing.T) {
}
tests := []struct {
name string
kubeClientObj []runtime.Object
ownerRestore *velerov1.Restore
targetPVCName string
targetNamespace string
targetPVName string
kubeReactors []reactor
cacheVolume *CacheConfigs
dataMover string
expectBackupPod bool
expectBackupPVC bool
expectCachePVC bool
expectBackupPV bool
err string
name string
kubeClientObj []runtime.Object
ownerRestore *velerov1.Restore
targetPVCName string
targetNamespace string
targetPVName string
kubeReactors []reactor
cacheVolume *CacheConfigs
dataMover string
expectBackupPod bool
expectBackupPVC bool
expectCachePVC bool
expectBackupPV bool
expectedNodeSelector map[string]string
expectedNodeAffinity *corev1api.NodeAffinity
err string
}{
{
name: "wait target pvc consumed fail",
@@ -256,6 +317,54 @@ func TestRestoreExpose(t *testing.T) {
expectBackupPod: true,
expectBackupPVC: true,
},
{
name: "succeed with invalid selected node and volume topology",
targetPVCName: "fake-target-pvc-with-node",
targetNamespace: "fake-ns",
ownerRestore: restore,
kubeClientObj: []runtime.Object{
targetPVCObjWithNode,
daemonSet,
storageClassWaitForFirstConsumer,
restorePVObjWithTopology,
},
kubeReactors: []reactor{
{
verb: "get",
resource: "persistentvolumeclaims",
reactorFunc: func(action clientTesting.Action) (handled bool, ret runtime.Object, err error) {
getAction := action.(clientTesting.GetAction)
if getAction.GetName() == "fake-restore" {
return true, restorePVCObjBound, nil
}
return false, nil, nil
},
},
},
expectBackupPod: true,
expectBackupPVC: true,
expectedNodeSelector: map[string]string{},
expectedNodeAffinity: &corev1api.NodeAffinity{
RequiredDuringSchedulingIgnoredDuringExecution: &corev1api.NodeSelector{
NodeSelectorTerms: []corev1api.NodeSelectorTerm{
{
MatchExpressions: []corev1api.NodeSelectorRequirement{
{
Key: "topology.kubernetes.io/zone",
Operator: corev1api.NodeSelectorOpIn,
Values: []string{"zone-1"},
},
{
Key: "kubernetes.io/os",
Operator: corev1api.NodeSelectorOpNotIn,
Values: []string{"windows"},
},
},
},
},
},
},
},
{
name: "create temporary PV fail",
targetPVCName: "fake-target-pvc",
@@ -473,9 +582,16 @@ func TestRestoreExpose(t *testing.T) {
require.NoError(t, err)
}
_, err = exposer.kubeClient.CoreV1().Pods(ownerObject.Namespace).Get(t.Context(), ownerObject.Name, metav1.GetOptions{})
pod, err := exposer.kubeClient.CoreV1().Pods(ownerObject.Namespace).Get(t.Context(), ownerObject.Name, metav1.GetOptions{})
if test.expectBackupPod {
require.NoError(t, err)
if test.expectedNodeSelector != nil {
assert.Equal(t, test.expectedNodeSelector, pod.Spec.NodeSelector)
}
if test.expectedNodeAffinity != nil {
require.NotNil(t, pod.Spec.Affinity)
assert.Equal(t, test.expectedNodeAffinity, pod.Spec.Affinity.NodeAffinity)
}
} else {
require.True(t, apierrors.IsNotFound(err), "expected IsNotFound, got %v", err)
}
@@ -1522,6 +1638,115 @@ end diagnose restore exposer`,
}
}
func TestValidateSelectedNode(t *testing.T) {
tests := []struct {
name string
node string
dataMover string
kubeClientObj []runtime.Object
expected bool
}{
{
name: "empty node",
node: "",
expected: true,
},
{
name: "node os is linux",
node: "fake-node",
kubeClientObj: []runtime.Object{
&corev1api.Node{
ObjectMeta: metav1.ObjectMeta{
Name: "fake-node",
Labels: map[string]string{
corev1api.LabelOSStable: kube.NodeOSLinux,
},
},
},
},
expected: true,
},
{
name: "node os is windows",
node: "fake-node",
kubeClientObj: []runtime.Object{
&corev1api.Node{
ObjectMeta: metav1.ObjectMeta{
Name: "fake-node",
Labels: map[string]string{
corev1api.LabelOSStable: kube.NodeOSWindows,
},
},
},
},
expected: true,
},
{
name: "node without os label",
node: "fake-node",
kubeClientObj: []runtime.Object{
&corev1api.Node{
ObjectMeta: metav1.ObjectMeta{
Name: "fake-node",
},
},
},
expected: false,
},
{
name: "node not found",
node: "fake-node",
expected: false,
},
{
name: "block data mover with linux node",
node: "fake-node",
dataMover: datamover.DataMoverTypeVeleroBlock,
kubeClientObj: []runtime.Object{
&corev1api.Node{
ObjectMeta: metav1.ObjectMeta{
Name: "fake-node",
Labels: map[string]string{
corev1api.LabelOSStable: kube.NodeOSLinux,
},
},
},
},
expected: true,
},
{
name: "block data mover with windows node",
node: "fake-node",
dataMover: datamover.DataMoverTypeVeleroBlock,
kubeClientObj: []runtime.Object{
&corev1api.Node{
ObjectMeta: metav1.ObjectMeta{
Name: "fake-node",
Labels: map[string]string{
corev1api.LabelOSStable: kube.NodeOSWindows,
},
},
},
},
expected: false,
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
fakeKubeClient := fake.NewSimpleClientset(test.kubeClientObj...)
exposer := genericRestoreExposer{
kubeClient: fakeKubeClient,
log: velerotest.NewLogger(),
}
actual := exposer.validateSelectedNode(t.Context(), test.node, test.dataMover, exposer.log)
assert.Equal(t, test.expected, actual)
})
}
}
func TestCreateRestorePod(t *testing.T) {
scName := "storage-class-01"
@@ -1678,6 +1903,7 @@ func TestCreateRestorePod(t *testing.T) {
"", // volumeSnapshotNamespace
"", // volumeID
nil,
nil, // volumeTopology
)
require.NoError(t, err)
+357
View File
@@ -0,0 +1,357 @@
/*
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 itemblock
import (
"testing"
"github.com/sirupsen/logrus"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/runtime/schema"
)
// newUnstructuredItem creates an unstructured object with the given apiVersion,
// kind, namespace, and name for use in tests.
func newUnstructuredItem(apiVersion, kind, namespace, name string) *unstructured.Unstructured {
return &unstructured.Unstructured{
Object: map[string]any{
"apiVersion": apiVersion,
"kind": kind,
"metadata": map[string]any{
"namespace": namespace,
"name": name,
},
},
}
}
func TestAddUnstructured(t *testing.T) {
tests := []struct {
name string
items []ItemBlockItem
expectedLen int
}{
{
name: "add single item to empty block",
items: []ItemBlockItem{
{
Gr: schema.GroupResource{Group: "apps", Resource: "deployments"},
Item: newUnstructuredItem("apps/v1", "Deployment", "default", "nginx"),
PreferredGVR: schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "deployments"},
},
},
expectedLen: 1,
},
{
name: "add multiple items",
items: []ItemBlockItem{
{
Gr: schema.GroupResource{Group: "", Resource: "pods"},
Item: newUnstructuredItem("v1", "Pod", "default", "pod-1"),
PreferredGVR: schema.GroupVersionResource{Group: "", Version: "v1", Resource: "pods"},
},
{
Gr: schema.GroupResource{Group: "", Resource: "services"},
Item: newUnstructuredItem("v1", "Service", "kube-system", "kube-dns"),
PreferredGVR: schema.GroupVersionResource{Group: "", Version: "v1", Resource: "services"},
},
{
Gr: schema.GroupResource{Group: "apps", Resource: "deployments"},
Item: newUnstructuredItem("apps/v1", "Deployment", "default", "web"),
PreferredGVR: schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "deployments"},
},
},
expectedLen: 3,
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
ib := &ItemBlock{
Log: logrus.New(),
}
for _, item := range tc.items {
ib.AddUnstructured(item.Gr, item.Item, item.PreferredGVR)
}
require.Len(t, ib.Items, tc.expectedLen)
// Verify each added item matches what was provided
for i, item := range tc.items {
assert.Equal(t, item.Gr, ib.Items[i].Gr)
assert.Equal(t, item.Item, ib.Items[i].Item)
assert.Equal(t, item.PreferredGVR, ib.Items[i].PreferredGVR)
}
})
}
}
func TestFindItem(t *testing.T) {
podsGR := schema.GroupResource{Group: "", Resource: "pods"}
deploymentsGR := schema.GroupResource{Group: "apps", Resource: "deployments"}
servicesGR := schema.GroupResource{Group: "", Resource: "services"}
podsGVR := schema.GroupVersionResource{Group: "", Version: "v1", Resource: "pods"}
deploymentsGVR := schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "deployments"}
tests := []struct {
name string
existingItems []ItemBlockItem
searchGR schema.GroupResource
searchNamespace string
searchName string
expectedCount int
// If set, verifies the names in the returned items are in this exact order
expectedNames []string
}{
{
name: "find item in empty block returns nil",
existingItems: nil,
searchGR: podsGR,
searchNamespace: "default",
searchName: "pod-1",
expectedCount: 0,
},
{
name: "find matching item by GR, namespace, and name",
existingItems: []ItemBlockItem{
{
Gr: podsGR,
Item: newUnstructuredItem("v1", "Pod", "default", "pod-1"),
PreferredGVR: podsGVR,
},
},
searchGR: podsGR,
searchNamespace: "default",
searchName: "pod-1",
expectedCount: 1,
expectedNames: []string{"pod-1"},
},
{
name: "no match when GR differs",
existingItems: []ItemBlockItem{
{
Gr: podsGR,
Item: newUnstructuredItem("v1", "Pod", "default", "pod-1"),
PreferredGVR: podsGVR,
},
},
searchGR: deploymentsGR,
searchNamespace: "default",
searchName: "pod-1",
expectedCount: 0,
},
{
name: "no match when namespace differs",
existingItems: []ItemBlockItem{
{
Gr: podsGR,
Item: newUnstructuredItem("v1", "Pod", "default", "pod-1"),
PreferredGVR: podsGVR,
},
},
searchGR: podsGR,
searchNamespace: "kube-system",
searchName: "pod-1",
expectedCount: 0,
},
{
name: "no match when name differs",
existingItems: []ItemBlockItem{
{
Gr: podsGR,
Item: newUnstructuredItem("v1", "Pod", "default", "pod-1"),
PreferredGVR: podsGVR,
},
},
searchGR: podsGR,
searchNamespace: "default",
searchName: "pod-2",
expectedCount: 0,
},
{
name: "nil item is skipped",
existingItems: []ItemBlockItem{
{
Gr: podsGR,
Item: nil,
PreferredGVR: podsGVR,
},
},
searchGR: podsGR,
searchNamespace: "default",
searchName: "pod-1",
expectedCount: 0,
},
{
name: "preferred GVR match is returned first",
existingItems: []ItemBlockItem{
{
Gr: deploymentsGR,
Item: newUnstructuredItem("apps/v1beta1", "Deployment", "default", "web"),
PreferredGVR: deploymentsGVR, // preferred is v1, item is v1beta1 → non-preferred
},
{
Gr: deploymentsGR,
Item: newUnstructuredItem("apps/v1", "Deployment", "default", "web"),
PreferredGVR: deploymentsGVR, // preferred is v1, item is v1 → preferred match
},
},
searchGR: deploymentsGR,
searchNamespace: "default",
searchName: "web",
expectedCount: 2,
expectedNames: []string{"web", "web"},
},
{
name: "multiple non-preferred items returned when no preferred match",
existingItems: []ItemBlockItem{
{
Gr: deploymentsGR,
Item: newUnstructuredItem("apps/v1beta1", "Deployment", "default", "web"),
PreferredGVR: deploymentsGVR,
},
{
Gr: deploymentsGR,
Item: newUnstructuredItem("apps/v1beta2", "Deployment", "default", "web"),
PreferredGVR: deploymentsGVR,
},
},
searchGR: deploymentsGR,
searchNamespace: "default",
searchName: "web",
expectedCount: 2,
},
{
name: "item with unparsable apiVersion is treated as non-preferred",
existingItems: []ItemBlockItem{
{
Gr: deploymentsGR,
Item: newUnstructuredItem("not/a/valid/version", "Deployment", "default", "web"),
PreferredGVR: deploymentsGVR,
},
},
searchGR: deploymentsGR,
searchNamespace: "default",
searchName: "web",
expectedCount: 1,
},
{
name: "only matching GR items are returned from mixed block",
existingItems: []ItemBlockItem{
{
Gr: podsGR,
Item: newUnstructuredItem("v1", "Pod", "default", "app"),
PreferredGVR: podsGVR,
},
{
Gr: deploymentsGR,
Item: newUnstructuredItem("apps/v1", "Deployment", "default", "app"),
PreferredGVR: deploymentsGVR,
},
{
Gr: servicesGR,
Item: newUnstructuredItem("v1", "Service", "default", "app"),
PreferredGVR: schema.GroupVersionResource{Group: "", Version: "v1", Resource: "services"},
},
},
searchGR: deploymentsGR,
searchNamespace: "default",
searchName: "app",
expectedCount: 1,
expectedNames: []string{"app"},
},
{
name: "cluster-scoped item found with empty namespace",
existingItems: []ItemBlockItem{
{
Gr: schema.GroupResource{Group: "", Resource: "namespaces"},
Item: newUnstructuredItem("v1", "Namespace", "", "production"),
PreferredGVR: schema.GroupVersionResource{
Group: "", Version: "v1", Resource: "namespaces",
},
},
},
searchGR: schema.GroupResource{Group: "", Resource: "namespaces"},
searchNamespace: "",
searchName: "production",
expectedCount: 1,
expectedNames: []string{"production"},
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
ib := &ItemBlock{
Log: logrus.New(),
Items: tc.existingItems,
}
result := ib.FindItem(tc.searchGR, tc.searchNamespace, tc.searchName)
require.Len(t, result, tc.expectedCount)
if tc.expectedNames != nil {
for i, expectedName := range tc.expectedNames {
assert.Equal(t, expectedName, result[i].Item.GetName())
}
}
})
}
}
func TestFindItemPreferredOrdering(t *testing.T) {
deploymentsGR := schema.GroupResource{Group: "apps", Resource: "deployments"}
deploymentsGVR := schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "deployments"}
// Insert non-preferred first, preferred second to verify ordering
ib := &ItemBlock{
Log: logrus.New(),
Items: []ItemBlockItem{
{
Gr: deploymentsGR,
Item: newUnstructuredItem("apps/v1beta1", "Deployment", "ns", "deploy"),
PreferredGVR: deploymentsGVR,
},
{
Gr: deploymentsGR,
Item: newUnstructuredItem("apps/v1", "Deployment", "ns", "deploy"),
PreferredGVR: deploymentsGVR,
},
{
Gr: deploymentsGR,
Item: newUnstructuredItem("apps/v1beta2", "Deployment", "ns", "deploy"),
PreferredGVR: deploymentsGVR,
},
},
}
result := ib.FindItem(deploymentsGR, "ns", "deploy")
require.Len(t, result, 3)
// The preferred match (apps/v1) should be first, regardless of insertion order
assert.Equal(t, "apps/v1", result[0].Item.GetAPIVersion(),
"preferred GVR match should be returned first")
// Non-preferred items follow in insertion order
assert.Equal(t, "apps/v1beta1", result[1].Item.GetAPIVersion())
assert.Equal(t, "apps/v1beta2", result[2].Item.GetAPIVersion())
}
+11 -3
View File
@@ -82,13 +82,17 @@ func KbClientIsRunningInNode(ctx context.Context, namespace string, nodeName str
}
// IsReady checks whether the node-agent daemonset has at least one ready pod
// by inspecting the DaemonSet status.
// by inspecting the DaemonSet status. Both the linux and windows daemonsets
// are checked before returning any non-NotFound lookup error, so that a
// transient error fetching one daemonset does not mask the other daemonset
// being ready.
func IsReady(ctx context.Context, namespace string, crClient ctrlclient.Client) error {
dsLinux := new(appsv1api.DaemonSet)
var lookupErr error
if err := crClient.Get(ctx, ctrlclient.ObjectKey{Namespace: namespace, Name: daemonSet}, dsLinux); err != nil {
dsLinux = nil
if !apierrors.IsNotFound(err) {
return errors.Wrap(err, "failed to get linux node-agent daemonset")
lookupErr = errors.Wrap(err, "failed to get linux node-agent daemonset")
}
}
@@ -96,7 +100,7 @@ func IsReady(ctx context.Context, namespace string, crClient ctrlclient.Client)
if err := crClient.Get(ctx, ctrlclient.ObjectKey{Namespace: namespace, Name: daemonsetWindows}, dsWindows); err != nil {
dsWindows = nil
if !apierrors.IsNotFound(err) {
return errors.Wrap(err, "failed to get windows node-agent daemonset")
lookupErr = errors.CombineErrors(lookupErr, errors.Wrap(err, "failed to get windows node-agent daemonset"))
}
}
@@ -108,6 +112,10 @@ func IsReady(ctx context.Context, namespace string, crClient ctrlclient.Client)
return nil
}
if lookupErr != nil {
return lookupErr
}
return errors.New("node-agent is not ready: no ready pods found")
}
+75
View File
@@ -18,6 +18,7 @@ package nodeagent
import (
"context"
"fmt"
"testing"
"github.com/cockroachdb/errors"
@@ -275,6 +276,52 @@ func TestIsReady(t *testing.T) {
},
expectErr: "failed to get windows node-agent daemonset: fake-get-error",
},
{
name: "linux daemonset get error but windows ready",
namespace: "fake-ns",
kubeClientObj: []runtime.Object{
dsWindowsReady,
},
interceptor: &interceptor.Funcs{
Get: func(ctx context.Context, c ctrlclient.WithWatch, key ctrlclient.ObjectKey, obj ctrlclient.Object, opts ...ctrlclient.GetOption) error {
if key.Name == "node-agent" {
return errors.New("fake-get-error")
}
return c.Get(ctx, key, obj, opts...)
},
},
},
{
name: "windows daemonset get error but linux ready",
namespace: "fake-ns",
kubeClientObj: []runtime.Object{
dsLinuxReady,
},
interceptor: &interceptor.Funcs{
Get: func(ctx context.Context, c ctrlclient.WithWatch, key ctrlclient.ObjectKey, obj ctrlclient.Object, opts ...ctrlclient.GetOption) error {
if key.Name == "node-agent-windows" {
return errors.New("fake-get-error")
}
return c.Get(ctx, key, obj, opts...)
},
},
},
{
name: "linux daemonset get error and windows not ready",
namespace: "fake-ns",
kubeClientObj: []runtime.Object{
dsWindowsNotReady,
},
interceptor: &interceptor.Funcs{
Get: func(ctx context.Context, c ctrlclient.WithWatch, key ctrlclient.ObjectKey, obj ctrlclient.Object, opts ...ctrlclient.GetOption) error {
if key.Name == "node-agent" {
return errors.New("fake-get-error")
}
return c.Get(ctx, key, obj, opts...)
},
},
expectErr: "failed to get linux node-agent daemonset: fake-get-error",
},
{
name: "linux ds exist but no ready pods",
namespace: "fake-ns",
@@ -362,6 +409,34 @@ func TestIsReady(t *testing.T) {
}
}
// TestIsReadyBothDaemonsetsGetError ensures that when both daemonset lookups
// return a non-NotFound error, the linux error is returned as the primary
// error while the windows error is retained as a secondary/attached error
// rather than being silently discarded.
func TestIsReadyBothDaemonsetsGetError(t *testing.T) {
scheme := runtime.NewScheme()
appsv1api.AddToScheme(scheme)
fakeClient := clientFake.NewClientBuilder().
WithScheme(scheme).
WithInterceptorFuncs(interceptor.Funcs{
Get: func(ctx context.Context, c ctrlclient.WithWatch, key ctrlclient.ObjectKey, obj ctrlclient.Object, opts ...ctrlclient.GetOption) error {
if key.Name == "node-agent" {
return errors.New("fake-linux-get-error")
}
if key.Name == "node-agent-windows" {
return errors.New("fake-windows-get-error")
}
return c.Get(ctx, key, obj, opts...)
},
}).
Build()
err := IsReady(t.Context(), "fake-ns", fakeClient)
require.EqualError(t, err, "failed to get linux node-agent daemonset: fake-linux-get-error")
assert.Contains(t, fmt.Sprintf("%+v", err), "failed to get windows node-agent daemonset: fake-windows-get-error")
}
func TestGetPodSpec(t *testing.T) {
podSpec := corev1api.PodSpec{
NodeName: "fake-node",
+1 -1
View File
@@ -165,7 +165,7 @@ func TestOnPvrCompleted(t *testing.T) {
{
name: "marshal fail",
marshalErr: errors.New("fake-marshal-error"),
expectedErr: "error marshaling restore result {{ } 0}: fake-marshal-error",
expectedErr: "error marshaling restore result {{ } 0 0}: fake-marshal-error",
},
{
name: "succeed",
+13
View File
@@ -38,6 +38,7 @@ import (
"github.com/vmware-tanzu/velero/pkg/label"
"github.com/vmware-tanzu/velero/pkg/nodeagent"
"github.com/vmware-tanzu/velero/pkg/repository"
"github.com/vmware-tanzu/velero/pkg/restore/inplace"
uploaderutil "github.com/vmware-tanzu/velero/pkg/uploader/util"
"github.com/vmware-tanzu/velero/pkg/util/boolptr"
"github.com/vmware-tanzu/velero/pkg/util/kube"
@@ -179,6 +180,18 @@ func (r *restorer) RestorePodVolumes(data RestoreData, tracker *volume.RestoreVo
}
}
// Pre-flight checks for in-place restore. Pods gated by this
// restore's restore-wait init container are excluded: they must mount
// the PVC so the volume gets mounted on the node for the node-agent
// to write into, and they cannot write to it themselves until this
// restore's PodVolumeRestores complete.
if data.Restore.IsVolumeDataInplaceRestore() && pvc != nil {
if err := inplace.CheckPVCNotInUse(r.ctx, r.crClient, pvc, data.Restore.UID); err != nil {
errs = append(errs, err)
continue
}
}
volumeRestore := newPodVolumeRestore(data.Restore, data.Pod, data.BackupLocation, volume, backupInfo.snapshotID, backupInfo.snapshotSize, "", backupInfo.uploaderType, data.SourceNamespace, pvc)
if err := veleroclient.CreateRetryGenerateName(r.crClient, r.ctx, volumeRestore); err != nil {
errs = append(errs, errors.WithStack(err))
+103 -1
View File
@@ -37,6 +37,7 @@ import (
velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
"github.com/vmware-tanzu/velero/pkg/builder"
"github.com/vmware-tanzu/velero/pkg/repository"
"github.com/vmware-tanzu/velero/pkg/restorehelper"
velerotest "github.com/vmware-tanzu/velero/pkg/test"
)
@@ -119,6 +120,21 @@ func TestGetVolumesRepositoryType(t *testing.T) {
}
}
// createGatedPodObj returns the restored pod as it exists in the cluster:
// running with the restore-wait init container injected by the given restore.
func createGatedPodObj(restoreUID string, volumeNum int) *corev1api.Pod {
pod := createPodObj(true, true, true, volumeNum)
pod.Spec.InitContainers = append([]corev1api.Container{{
Name: restorehelper.WaitInitContainer,
Args: []string{restoreUID},
}}, pod.Spec.InitContainers...)
pod.Status.InitContainerStatuses = []corev1api.ContainerStatus{{
Name: restorehelper.WaitInitContainer,
State: corev1api.ContainerState{Running: &corev1api.ContainerStateRunning{}},
}}
return pod
}
func createNodeAgentDaemonset() *appsv1api.DaemonSet {
ds := &appsv1api.DaemonSet{
ObjectMeta: metav1.ObjectMeta{
@@ -181,6 +197,7 @@ func TestRestorePodVolumes(t *testing.T) {
pvbs []*velerov1api.PodVolumeBackup
restoredPod *corev1api.Pod
sourceNamespace string
inplace bool
errs []expectError
}{
{
@@ -340,6 +357,86 @@ func TestRestorePodVolumes(t *testing.T) {
completedPVR,
},
},
{
name: "in-place restore blocked when the PVC is used by another running pod",
pvbs: []*velerov1api.PodVolumeBackup{
createPVBObj(true, true, 1, "kopia"),
},
inplace: true,
kubeClientObj: []runtime.Object{
createNodeAgentDaemonset(),
createPVCObj(1),
func() *corev1api.Pod {
pod := builder.ForPod("fake-ns", "other-pod").
Volumes(builder.ForVolume("fake-volume-1").PersistentVolumeClaimSource("fake-pvc-1").Result()).
Result()
pod.Status.Phase = corev1api.PodRunning
return pod
}(),
},
ctlClientObj: []runtime.Object{
createBackupRepoObj(),
},
restoredPod: createPodObj(true, true, true, 1),
sourceNamespace: "fake-ns",
bsl: "fake-bsl",
runtimeScheme: scheme,
errs: []expectError{
{
err: "in-place restore pre-flight check failed",
prefixOnly: true,
},
},
},
{
name: "in-place restore blocked when the pod is gated by a different restore",
pvbs: []*velerov1api.PodVolumeBackup{
createPVBObj(true, true, 1, "kopia"),
},
inplace: true,
kubeClientObj: []runtime.Object{
createNodeAgentDaemonset(),
createPVCObj(1),
createGatedPodObj("old-restore-uid", 1),
},
ctlClientObj: []runtime.Object{
createBackupRepoObj(),
},
restoredPod: createPodObj(true, true, true, 1),
sourceNamespace: "fake-ns",
bsl: "fake-bsl",
runtimeScheme: scheme,
errs: []expectError{
{
err: "in-place restore pre-flight check failed",
prefixOnly: true,
},
},
},
{
name: "in-place restore proceeds when the PVC is only used by the gated restored pod",
pvbs: []*velerov1api.PodVolumeBackup{
createPVBObj(true, true, 1, "kopia"),
},
inplace: true,
kubeClientObj: []runtime.Object{
createNodeAgentDaemonset(),
createNodeObj(),
createPVCObj(1),
createGatedPodObj("fake-restore-uid", 1),
createNodeAgentPodObj(true),
},
ctlClientObj: []runtime.Object{
createBackupRepoObj(),
},
restoredPod: createPodObj(true, true, true, 1),
sourceNamespace: "fake-ns",
bsl: "fake-bsl",
runtimeScheme: scheme,
retPVRs: []*velerov1api.PodVolumeRestore{
completedPVR,
},
},
}
for _, test := range tests {
@@ -362,7 +459,12 @@ func TestRestorePodVolumes(t *testing.T) {
ensurer := repository.NewEnsurer(fakeCRClient, velerotest.NewLogger(), time.Millisecond)
restoreObj := builder.ForRestore(velerov1api.DefaultNamespace, "fake-restore").Result()
restoreBuilder := builder.ForRestore(velerov1api.DefaultNamespace, "fake-restore").
ObjectMeta(builder.WithUID("fake-restore-uid"))
if test.inplace {
restoreBuilder = restoreBuilder.ExistingVolumeDataPolicy(string(velerov1api.VolumeDataPolicyTypeFull))
}
restoreObj := restoreBuilder.Result()
rs := newRestorer(ctx, repository.NewRepoLocker(), ensurer, pvrInformer, kubeClient, fakeCRClient, restoreObj, velerotest.NewLogger())
+7
View File
@@ -44,6 +44,7 @@ import (
plugincommon "github.com/vmware-tanzu/velero/pkg/plugin/framework/common"
"github.com/vmware-tanzu/velero/pkg/plugin/velero"
riav2 "github.com/vmware-tanzu/velero/pkg/plugin/velero/restoreitemaction/v2"
"github.com/vmware-tanzu/velero/pkg/restore/inplace"
uploaderUtil "github.com/vmware-tanzu/velero/pkg/uploader/util"
"github.com/vmware-tanzu/velero/pkg/util"
"github.com/vmware-tanzu/velero/pkg/util/boolptr"
@@ -236,6 +237,12 @@ func (p *pvcRestoreItemAction) executeWithDataMove(logger *logrus.Entry, input *
if existingPVC.Status.Phase != corev1api.ClaimBound {
return nil, errors.New("ExistingVolumeDataPolicy is in-place restore, but the existing PVC is not bound.")
}
// Pre-flight checks must pass before any side effect on the existing PVC/PV.
if err := inplace.CheckPVCNotInUse(ctx, p.crClient, existingPVC, input.Restore.UID); err != nil {
return nil, errors.WithStack(err)
}
// take a CSI snapshot of the existing PVC as the baseline of CBT
if input.Restore.IsVolumeDataInplaceIncrementalRestore() && datamover.IsVeleroBlockDataMover(dataUploadResult.DataMover) {
logger.Info("ExistingVolumeDataPolicy is in-place incremental restore and data mover is velero-block. Taking a CSI snapshot of the existing PVC as the baseline of CBT...")
@@ -741,6 +741,105 @@ func TestExecuteInplaceRestore(t *testing.T) {
require.Equal(t, "testPV", dataDownloadList.Items[0].Spec.TargetVolume.PV)
}
// TestExecuteInplaceRestorePreflight verifies the RIA fails the item without
// side effects when the pre-flight check fails. The in-use semantics are
// covered by the pkg/restore/inplace unit tests.
func TestExecuteInplaceRestorePreflight(t *testing.T) {
newPodUsingPVC := func(phase corev1api.PodPhase) *corev1api.Pod {
pod := builder.ForPod("velero", "consumer-pod").
Volumes(builder.ForVolume("data").PersistentVolumeClaimSource("testPVC").Result()).
Result()
pod.Status.Phase = phase
return pod
}
tests := []struct {
name string
pod *corev1api.Pod
expectBlock bool
}{
{
name: "no pod, restore proceeds",
},
{
name: "active pod blocks the restore",
pod: newPodUsingPVC(corev1api.PodRunning),
expectBlock: true,
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
existingPVC := builder.ForPersistentVolumeClaim("velero", "testPVC").
VolumeName("testPV").
Phase(corev1api.ClaimBound).Result()
existingPV := builder.ForPersistentVolume("testPV").Result()
backup := builder.ForBackup("velero", "testBackup").SnapshotMoveData(true).Result()
restore := builder.ForRestore("velero", "testRestore").Backup("testBackup").
ObjectMeta(builder.WithUID("uid")).ExistingVolumeDataPolicy("full").Result()
pvcFromBackup := builder.ForPersistentVolumeClaim("velero", "testPVC").
ObjectMeta(builder.WithAnnotations(
velerov1api.VolumeSnapshotLabel, "vsName",
velerov1api.DataUploadNameAnnotation, "velero/testDU",
)).Result()
dataUploadResult := builder.ForConfigMap("velero", "testCM").Data("uid", "{}").
ObjectMeta(builder.WithLabels(
velerov1api.RestoreUIDLabel, "uid",
velerov1api.PVCNamespaceNameLabel, "velero.testPVC",
velerov1api.ResourceUsageLabel, label.GetValidName(string(velerov1api.VeleroResourceUsageDataUploadResult)),
)).Result()
crObjects := []runtime.Object{existingPVC, existingPV, backup, dataUploadResult}
kubeObjects := []runtime.Object{existingPVC, existingPV}
if tc.pod != nil {
crObjects = append(crObjects, tc.pod)
kubeObjects = append(kubeObjects, tc.pod)
}
pvcRIA := pvcRestoreItemAction{
log: logrus.New(),
crClient: velerotest.NewFakeControllerRuntimeClient(t, crObjects...),
kubeClient: fake.NewSimpleClientset(kubeObjects...),
}
pvcMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(pvcFromBackup.DeepCopy())
require.NoError(t, err)
pvcFromBackupMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(pvcFromBackup)
require.NoError(t, err)
_, err = pvcRIA.Execute(&velero.RestoreItemActionExecuteInput{
Item: &unstructured.Unstructured{Object: pvcMap},
ItemFromBackup: &unstructured.Unstructured{Object: pvcFromBackupMap},
Restore: restore,
})
gotPVC, getErr := pvcRIA.kubeClient.CoreV1().PersistentVolumeClaims("velero").Get(t.Context(), "testPVC", metav1.GetOptions{})
dataDownloadList := new(velerov2alpha1.DataDownloadList)
require.NoError(t, pvcRIA.crClient.List(t.Context(), dataDownloadList, &crclient.ListOptions{}))
if tc.expectBlock {
require.Error(t, err)
require.Contains(t, err.Error(), "pre-flight check failed")
require.Contains(t, err.Error(), "consumer-pod")
// No side effects: PVC untouched with the original volumeName,
// PV reclaim policy not patched, no DataDownload created.
require.NoError(t, getErr)
require.Equal(t, "testPV", gotPVC.Spec.VolumeName)
gotPV, pvErr := pvcRIA.kubeClient.CoreV1().PersistentVolumes().Get(t.Context(), "testPV", metav1.GetOptions{})
require.NoError(t, pvErr)
require.Equal(t, existingPV.Spec.PersistentVolumeReclaimPolicy, gotPV.Spec.PersistentVolumeReclaimPolicy)
require.Empty(t, dataDownloadList.Items)
} else {
require.NoError(t, err)
// The in-place restore proceeded: the existing PVC is deleted
// and a DataDownload is created.
require.True(t, apierrors.IsNotFound(getErr))
require.Len(t, dataDownloadList.Items, 1)
}
})
}
}
func TestPVCAppliesTo(t *testing.T) {
p := pvcRestoreItemAction{
log: logrus.StandardLogger(),
+131
View File
@@ -0,0 +1,131 @@
/*
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 inplace holds the pre-flight checks for in-place volume data
// restores. The checks must pass before Velero performs any side effect on
// the existing PVC/PV.
package inplace
import (
"context"
"fmt"
"strings"
"github.com/cockroachdb/errors"
corev1api "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/types"
crclient "sigs.k8s.io/controller-runtime/pkg/client"
"github.com/vmware-tanzu/velero/pkg/restorehelper"
)
// CheckPVCNotInUse verifies the target PVC is not used by any active pod,
// aligned with the pvc-protection controller semantics: terminal-phase pods
// (Succeeded/Failed) don't block; all other phases do, and terminating pods
// are flagged so the message can hint the user to wait.
//
// Pods gated by this restore's restore-wait init container are exempted: on
// the file system restore path the restored pods must mount the PVC for the
// node-agent to restore the data, and an RWX PVC may be mounted by several of
// them. Such a pod cannot write to the volume since its workload containers
// are blocked until this restore's PodVolumeRestores complete (see
// gatedByThisRestore). Any other pod, including one gated by a different
// restore whose release timing is out of our control, still blocks.
func CheckPVCNotInUse(
ctx context.Context,
cli crclient.Client,
pvc *corev1api.PersistentVolumeClaim,
restoreUID types.UID,
) error {
podList := new(corev1api.PodList)
if err := cli.List(ctx, podList, &crclient.ListOptions{Namespace: pvc.Namespace}); err != nil {
return errors.Wrapf(err, "failed to check whether PVC %s/%s is in use: failed to list pods in namespace %s", pvc.Namespace, pvc.Name, pvc.Namespace)
}
podsInUse := []string{}
terminatingOnly := true
for i := range podList.Items {
pod := &podList.Items[i]
if !podUsesPVC(pod, pvc.Name) ||
pod.Status.Phase == corev1api.PodSucceeded || pod.Status.Phase == corev1api.PodFailed ||
gatedByThisRestore(pod, restoreUID) {
continue
}
state := string(pod.Status.Phase)
if pod.DeletionTimestamp != nil {
state += ", terminating"
} else {
terminatingOnly = false
}
podsInUse = append(podsInUse, fmt.Sprintf("%s (%s)", pod.Name, state))
}
if len(podsInUse) == 0 {
return nil
}
hint := "delete the workloads consuming the PVC and retry"
if terminatingOnly {
hint = "the pod(s) are terminating; retry after they are fully removed"
}
return errors.Errorf("in-place restore pre-flight check failed, skipping volume data restore: PVC %s/%s is still in use by pod(s) [%s]: %s",
pvc.Namespace, pvc.Name, strings.Join(podsInUse, ", "), hint)
}
func podUsesPVC(pod *corev1api.Pod, pvcName string) bool {
for _, vol := range pod.Spec.Volumes {
if vol.PersistentVolumeClaim != nil && vol.PersistentVolumeClaim.ClaimName == pvcName {
return true
}
}
return false
}
// gatedByThisRestore reports whether the pod is blocked by the restore-wait
// init container injected by this restore, identified by the restore UID in
// the init container's args. Such a pod cannot write to the volume: its
// workload containers won't start until this restore's PodVolumeRestores
// complete and write the done signal. A terminated init container means the
// gate is already open, so the pod is no longer exempted. Pods gated by a
// different restore are not exempted either, since their release timing is
// unrelated to this restore.
func gatedByThisRestore(pod *corev1api.Pod, restoreUID types.UID) bool {
if restoreUID == "" {
return false
}
idx := -1
for i, c := range pod.Spec.InitContainers {
if c.Name == restorehelper.WaitInitContainer {
if len(c.Args) == 0 || c.Args[0] != string(restoreUID) {
return false
}
idx = i
break
}
}
if idx < 0 {
return false
}
for _, cs := range pod.Status.InitContainerStatuses {
if cs.Name == restorehelper.WaitInitContainer {
return cs.State.Terminated == nil
}
}
// Statuses not populated yet: the init container hasn't run, so the gate
// is still closed.
return true
}
+202
View File
@@ -0,0 +1,202 @@
/*
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 inplace
import (
"testing"
"github.com/stretchr/testify/assert"
"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"
"github.com/vmware-tanzu/velero/pkg/restorehelper"
velerotest "github.com/vmware-tanzu/velero/pkg/test"
)
func podUsingPVC(name, pvcName string, phase corev1api.PodPhase, terminating bool) *corev1api.Pod {
pod := &corev1api.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: name,
Namespace: "default",
UID: types.UID(name + "-uid"),
},
Spec: corev1api.PodSpec{
Volumes: []corev1api.Volume{
{
Name: "data",
VolumeSource: corev1api.VolumeSource{
PersistentVolumeClaim: &corev1api.PersistentVolumeClaimVolumeSource{
ClaimName: pvcName,
},
},
},
},
},
Status: corev1api.PodStatus{Phase: phase},
}
if terminating {
now := metav1.Now()
pod.DeletionTimestamp = &now
pod.Finalizers = []string{"fake-finalizer"}
}
return pod
}
// gatedPod adds the restore-wait init container (as injected by the
// PodVolumeRestoreAction RIA, carrying the restore UID in args) to the pod.
// terminated simulates the gate having already been released.
func gatedPod(pod *corev1api.Pod, restoreUID string, terminated bool) *corev1api.Pod {
pod.Spec.InitContainers = append([]corev1api.Container{{
Name: restorehelper.WaitInitContainer,
Args: []string{restoreUID},
}}, pod.Spec.InitContainers...)
status := corev1api.ContainerStatus{
Name: restorehelper.WaitInitContainer,
State: corev1api.ContainerState{Running: &corev1api.ContainerStateRunning{}},
}
if terminated {
status.State = corev1api.ContainerState{Terminated: &corev1api.ContainerStateTerminated{}}
}
pod.Status.InitContainerStatuses = []corev1api.ContainerStatus{status}
return pod
}
func TestCheckPVCNotInUse(t *testing.T) {
tests := []struct {
name string
pods []*corev1api.Pod
restoreUID types.UID
expectPass bool
expectMessage []string
}{
{
name: "no pods, check passes",
expectPass: true,
},
{
name: "active pod blocks with the delete hint",
pods: []*corev1api.Pod{podUsingPVC("pod-1", "pvc-1", corev1api.PodRunning, false)},
expectMessage: []string{"pod-1 (Running)", "delete the workloads"},
},
{
name: "unknown-phase pod blocks (node may be unreachable)",
pods: []*corev1api.Pod{podUsingPVC("pod-1", "pvc-1", corev1api.PodUnknown, false)},
expectMessage: []string{"pod-1 (Unknown)"},
},
{
name: "terminating pod blocks with the wait hint",
pods: []*corev1api.Pod{podUsingPVC("pod-1", "pvc-1", corev1api.PodRunning, true)},
expectMessage: []string{"pod-1 (Running, terminating)", "retry after they are fully removed"},
},
{
name: "terminal-phase pods do not block",
pods: []*corev1api.Pod{
podUsingPVC("pod-1", "pvc-1", corev1api.PodSucceeded, false),
podUsingPVC("pod-2", "pvc-1", corev1api.PodFailed, false),
},
expectPass: true,
},
{
name: "pod using another PVC does not block",
pods: []*corev1api.Pod{podUsingPVC("pod-1", "other-pvc", corev1api.PodRunning, false)},
expectPass: true,
},
{
name: "pods gated by this restore do not block, other pods still do",
pods: []*corev1api.Pod{
gatedPod(podUsingPVC("restored-pod-1", "pvc-1", corev1api.PodPending, false), "restore-uid", false),
gatedPod(podUsingPVC("restored-pod-2", "pvc-1", corev1api.PodPending, false), "restore-uid", false),
podUsingPVC("other-pod", "pvc-1", corev1api.PodRunning, false),
},
restoreUID: "restore-uid",
expectMessage: []string{"[other-pod (Running)]"},
},
{
name: "multiple pods gated by this restore pass the check",
pods: []*corev1api.Pod{
gatedPod(podUsingPVC("restored-pod-1", "pvc-1", corev1api.PodPending, false), "restore-uid", false),
gatedPod(podUsingPVC("restored-pod-2", "pvc-1", corev1api.PodPending, false), "restore-uid", false),
},
restoreUID: "restore-uid",
expectPass: true,
},
{
name: "pod gated by a different restore still blocks",
pods: []*corev1api.Pod{
gatedPod(podUsingPVC("old-restored-pod", "pvc-1", corev1api.PodPending, false), "old-restore-uid", false),
},
restoreUID: "restore-uid",
expectMessage: []string{"[old-restored-pod (Pending)]"},
},
{
name: "pod whose restore-wait already terminated still blocks",
pods: []*corev1api.Pod{
gatedPod(podUsingPVC("released-pod", "pvc-1", corev1api.PodRunning, false), "restore-uid", true),
},
restoreUID: "restore-uid",
expectMessage: []string{"[released-pod (Running)]"},
},
{
name: "gated pod without init container status yet passes the check",
pods: []*corev1api.Pod{
func() *corev1api.Pod {
pod := gatedPod(podUsingPVC("new-pod", "pvc-1", corev1api.PodPending, false), "restore-uid", false)
pod.Status.InitContainerStatuses = nil
return pod
}(),
},
restoreUID: "restore-uid",
expectPass: true,
},
{
name: "empty restore UID exempts nothing",
pods: []*corev1api.Pod{
gatedPod(podUsingPVC("restored-pod", "pvc-1", corev1api.PodPending, false), "", false),
},
restoreUID: "",
expectMessage: []string{"[restored-pod (Pending)]"},
},
}
for _, tc := range tests {
t.Run(tc.name, func(t *testing.T) {
objs := []runtime.Object{}
for _, pod := range tc.pods {
objs = append(objs, pod)
}
cli := velerotest.NewFakeControllerRuntimeClient(t, objs...)
pvc := &corev1api.PersistentVolumeClaim{
ObjectMeta: metav1.ObjectMeta{Name: "pvc-1", Namespace: "default"},
}
err := CheckPVCNotInUse(t.Context(), cli, pvc, tc.restoreUID)
if tc.expectPass {
require.NoError(t, err)
return
}
require.Error(t, err)
assert.Contains(t, err.Error(), "pre-flight check failed")
for _, fragment := range tc.expectMessage {
assert.Contains(t, err.Error(), fragment)
}
})
}
}
+9 -9
View File
@@ -205,12 +205,12 @@ func getParentBackupInfo(ctx context.Context, rep udmrepo.BackupRepo, forceFull
}
// Restore restore specific sourcePath with given snapshotID and update progress
func Restore(ctx context.Context, blkUp Uploader, rep udmrepo.BackupRepo, snapshotID, dest string, incremental bool, cbtSource cbtservice.SourceInfo, cbtService cbtservice.Service, uploaderCfg map[string]string, log logrus.FieldLogger) (int64, error) {
func Restore(ctx context.Context, blkUp Uploader, rep udmrepo.BackupRepo, snapshotID, dest string, incremental bool, cbtSource cbtservice.SourceInfo, cbtService cbtservice.Service, uploaderCfg map[string]string, log logrus.FieldLogger) (int64, int64, error) {
log.Info("Start to restore...")
snapshot, err := rep.GetSnapshot(ctx, udmrepo.ID(snapshotID))
if err != nil {
return 0, errors.Wrapf(err, "Unable to load snapshot %v", snapshotID)
return 0, 0, errors.Wrapf(err, "Unable to load snapshot %v", snapshotID)
}
log.Infof("Restore from snapshot %s, incremental %v, cbt source %v, description %s, created time %v, tags %v", snapshotID, incremental, cbtSource, snapshot.Description, snapshot.EndTime, snapshot.Tags)
@@ -246,34 +246,34 @@ func Restore(ctx context.Context, blkUp Uploader, rep udmrepo.BackupRepo, snapsh
destPath, err := filepath.Abs(dest)
if err != nil {
return 0, errors.Wrapf(err, "invalid dest path '%s'", dest)
return 0, 0, errors.Wrapf(err, "invalid dest path '%s'", dest)
}
destPath = filepath.Clean(destPath)
destDev, err := openBlockDeviceFunc(destPath, false)
if err != nil {
return 0, errors.Wrapf(err, "error opening block device '%s'", destPath)
return 0, 0, errors.Wrapf(err, "error opening block device '%s'", destPath)
}
defer destDev.Close()
destSize, err := destDev.Seek(0, io.SeekEnd)
if err != nil {
return 0, errors.Wrapf(err, "error getting length of block device %s", dest)
return 0, 0, errors.Wrapf(err, "error getting length of block device %s", dest)
}
_, err = destDev.Seek(0, io.SeekStart)
if err != nil {
return 0, errors.Wrapf(err, "error reset pos of block device %s", dest)
return 0, 0, errors.Wrapf(err, "error reset pos of block device %s", dest)
}
_, totalSize, err := blkUp.Restore(snapshot, destInfo{dev: destDev, path: destPath, size: destSize}, bitmap.Iterator(), uploaderCfg)
incrementalBytes, totalSize, err := blkUp.Restore(snapshot, destInfo{dev: destDev, path: destPath, size: destSize}, bitmap.Iterator(), uploaderCfg)
if err != nil {
return 0, errors.Wrapf(err, "error restoring to block dev %s", destPath)
return 0, 0, errors.Wrapf(err, "error restoring to block dev %s", destPath)
}
return totalSize, nil
return incrementalBytes, totalSize, nil
}
func findPreviousSnapshot(ctx context.Context, rep udmrepo.BackupRepo, path string, snapshotTags map[string]string, noLaterThan *time.Time, log logrus.FieldLogger) (udmrepo.Snapshot, error) {
+1 -1
View File
@@ -842,7 +842,7 @@ func TestRestore(t *testing.T) {
cbtSvc = tc.cbtService(t)
}
size, err := Restore(ctx, mockBlkup, mockRepo, "snap-001", "/dev/sdb", tc.incremental, tc.cbtSource, cbtSvc, map[string]string{}, testLog())
_, size, err := Restore(ctx, mockBlkup, mockRepo, "snap-001", "/dev/sdb", tc.incremental, tc.cbtSource, cbtSvc, map[string]string{}, testLog())
if tc.expectedErrStr != "" {
require.Error(t, err)
+9 -9
View File
@@ -167,9 +167,9 @@ func (bp *blockProvider) RunRestore(
cbtParam CBTParam,
volMode uploader.PersistentVolumeMode,
uploaderCfg map[string]string,
updater uploader.ProgressUpdater) (int64, error) {
updater uploader.ProgressUpdater) (int64, int64, error) {
if updater == nil {
return 0, errors.New("restore progress updater is invalid")
return 0, 0, errors.New("restore progress updater is invalid")
}
log := bp.log.WithFields(logrus.Fields{
@@ -180,24 +180,24 @@ func (bp *blockProvider) RunRestore(
blkUploader := block.NewUploader(ctx, bp.bkRepo, updater, log)
size, err := blockRestoreFunc(ctx, blkUploader, bp.bkRepo, snapshotID, volumePath, incremental, cbtParam.Source, cbtParam.Service, uploaderCfg, log)
incrementalBytes, totalBytes, err := blockRestoreFunc(ctx, blkUploader, bp.bkRepo, snapshotID, volumePath, incremental, cbtParam.Source, cbtParam.Service, uploaderCfg, log)
// errors.Is, not ==: see the equivalent comment on the backup path above.
if errors.Is(err, block.ErrCanceled) {
log.Warn("Block restore is canceled")
return 0, ErrorCanceled
return 0, 0, ErrorCanceled
}
if err != nil {
return 0, errors.Wrapf(err, "Failed to run block restore")
return 0, 0, errors.Wrapf(err, "Failed to run block restore")
}
updater.UpdateProgress(&uploader.Progress{
TotalBytes: size,
BytesDone: size,
TotalBytes: totalBytes,
BytesDone: totalBytes,
})
log.Infof("Block restore finished, restore size %v", size)
log.Infof("Block restore finished, restore incremental size %v, total size %v", incrementalBytes, totalBytes)
return size, nil
return incrementalBytes, totalBytes, nil
}
+6 -6
View File
@@ -412,8 +412,8 @@ func TestBlockProviderCancelThroughWrappedError(t *testing.T) {
t.Run("restore", func(t *testing.T) {
orig := blockRestoreFunc
defer func() { blockRestoreFunc = orig }()
blockRestoreFunc = func(_ context.Context, _ block.Uploader, _ udmrepo.BackupRepo, _ string, _ string, _ bool, _ cbtservice.SourceInfo, _ cbtservice.Service, _ map[string]string, _ logrus.FieldLogger) (int64, error) {
return 0, errors.Wrap(block.ErrCanceled, "error restoring bdev")
blockRestoreFunc = func(_ context.Context, _ block.Uploader, _ udmrepo.BackupRepo, _ string, _ string, _ bool, _ cbtservice.SourceInfo, _ cbtservice.Service, _ map[string]string, _ logrus.FieldLogger) (int64, int64, error) {
return 0, 0, errors.Wrap(block.ErrCanceled, "error restoring bdev")
}
bp := &blockProvider{
@@ -422,7 +422,7 @@ func TestBlockProviderCancelThroughWrappedError(t *testing.T) {
log: logrus.New(),
}
_, err := bp.RunRestore(t.Context(), "snap-1", "/dev/sda", false, CBTParam{},
_, _, err := bp.RunRestore(t.Context(), "snap-1", "/dev/sda", false, CBTParam{},
uploader.PersistentVolumeBlock, map[string]string{}, &blockMockProgressUpdater{})
require.ErrorIs(t, err, ErrorCanceled)
@@ -496,10 +496,10 @@ func TestBlockProviderRunRestore(t *testing.T) {
var capturedSnapshotID string
var capturedVolumePath string
blockRestoreFunc = func(ctx context.Context, blkUp block.Uploader, rep udmrepo.BackupRepo, snapshotID string, dest string, incremental bool, cbtSource cbtservice.SourceInfo, cbtService cbtservice.Service, uploaderCfg map[string]string, log logrus.FieldLogger) (int64, error) {
blockRestoreFunc = func(ctx context.Context, blkUp block.Uploader, rep udmrepo.BackupRepo, snapshotID string, dest string, incremental bool, cbtSource cbtservice.SourceInfo, cbtService cbtservice.Service, uploaderCfg map[string]string, log logrus.FieldLogger) (int64, int64, error) {
capturedSnapshotID = snapshotID
capturedVolumePath = dest
return tc.mockRestoreSize, tc.mockRestoreErr
return tc.mockRestoreSize, tc.mockRestoreSize, tc.mockRestoreErr
}
bp := &blockProvider{
@@ -507,7 +507,7 @@ func TestBlockProviderRunRestore(t *testing.T) {
log: logrus.New(),
}
size, err := bp.RunRestore(
_, size, err := bp.RunRestore(
t.Context(),
tc.snapshotID,
tc.volumePath,
+5 -4
View File
@@ -215,7 +215,7 @@ func (kp *kopiaProvider) RunRestore(
_ CBTParam,
volMode uploader.PersistentVolumeMode,
uploaderCfg map[string]string,
updater uploader.ProgressUpdater) (int64, error) {
updater uploader.ProgressUpdater) (int64, int64, error) {
log := kp.log.WithFields(logrus.Fields{
"snapshotID": snapshotID,
"volumePath": volumePath,
@@ -239,12 +239,12 @@ func (kp *kopiaProvider) RunRestore(
size, fileCount, err := kopiaRestoreFunc(context.Background(), repoWriter, progress, snapshotID, volumePath, incremental, volMode, uploaderCfg, log, restoreCancel)
if err != nil {
return 0, errors.Wrapf(err, "Failed to run kopia restore")
return 0, 0, errors.Wrapf(err, "Failed to run kopia restore")
}
if atomic.LoadInt32(&kp.canceling) == 1 {
log.Error("Kopia restore is canceled")
return 0, ErrorCanceled
return 0, 0, ErrorCanceled
}
// which ensure that the statistic data of TotalBytes equal to BytesDone when finished
@@ -257,5 +257,6 @@ func (kp *kopiaProvider) RunRestore(
log.Info(output)
return size, nil
// the incremental bytes is the same as the total bytes because total bytes is the size of actual data Kopia writes
return size, size, nil
}
+4 -2
View File
@@ -157,8 +157,10 @@ func TestRunRestore(t *testing.T) {
if tc.volMode == "" {
tc.volMode = uploader.PersistentVolumeFilesystem
}
kopiaRestoreFunc = tc.hookRestoreFunc
_, err := kp.RunRestore(t.Context(), "", "/var", tc.incremental, CBTParam{}, tc.volMode, map[string]string{}, &updater)
kopiaRestoreFunc = func(ctx context.Context, rep repo.RepositoryWriter, progress *kopia.Progress, snapshotID, dest string, incremental bool, volMode uploader.PersistentVolumeMode, uploaderCfg map[string]string, log logrus.FieldLogger, cancleCh chan struct{}) (int64, int32, error) {
return tc.hookRestoreFunc(ctx, rep, progress, snapshotID, dest, incremental, volMode, uploaderCfg, log, cancleCh)
}
_, _, err := kp.RunRestore(t.Context(), "", "/var", tc.incremental, CBTParam{}, tc.volMode, map[string]string{}, &updater)
if tc.notError {
assert.NoError(t, err)
} else {
+18 -9
View File
@@ -223,7 +223,7 @@ func (_c *Provider_RunBackup_Call) RunAndReturn(run func(ctx context.Context, pa
}
// RunRestore provides a mock function for the type Provider
func (_mock *Provider) RunRestore(ctx context.Context, snapshotID string, volumePath string, incremental bool, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, error) {
func (_mock *Provider) RunRestore(ctx context.Context, snapshotID string, volumePath string, incremental bool, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, int64, error) {
ret := _mock.Called(ctx, snapshotID, volumePath, incremental, cbtParam, volMode, uploaderConfig, updater)
if len(ret) == 0 {
@@ -231,8 +231,9 @@ func (_mock *Provider) RunRestore(ctx context.Context, snapshotID string, volume
}
var r0 int64
var r1 error
if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (int64, error)); ok {
var r1 int64
var r2 error
if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) (int64, int64, error)); ok {
return returnFunc(ctx, snapshotID, volumePath, incremental, cbtParam, volMode, uploaderConfig, updater)
}
if returnFunc, ok := ret.Get(0).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok {
@@ -240,12 +241,20 @@ func (_mock *Provider) RunRestore(ctx context.Context, snapshotID string, volume
} else {
r0 = ret.Get(0).(int64)
}
if returnFunc, ok := ret.Get(1).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok {
if returnFunc, ok := ret.Get(1).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) int64); ok {
r1 = returnFunc(ctx, snapshotID, volumePath, incremental, cbtParam, volMode, uploaderConfig, updater)
} else {
r1 = ret.Error(1)
r1 = ret.Get(1).(int64)
}
return r0, r1
if returnFunc, ok := ret.Get(2).(func(context.Context, string, string, bool, provider.CBTParam, uploader.PersistentVolumeMode, map[string]string, uploader.ProgressUpdater) error); ok {
r2 = returnFunc(ctx, snapshotID, volumePath, incremental, cbtParam, volMode, uploaderConfig, updater)
} else {
r2 = ret.Error(2)
}
return r0, r1, r2
}
// Provider_RunRestore_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RunRestore'
@@ -314,12 +323,12 @@ func (_c *Provider_RunRestore_Call) Run(run func(ctx context.Context, snapshotID
return _c
}
func (_c *Provider_RunRestore_Call) Return(n int64, err error) *Provider_RunRestore_Call {
_c.Call.Return(n, err)
func (_c *Provider_RunRestore_Call) Return(_a0 int64, _a1 int64, _a2 error) *Provider_RunRestore_Call {
_c.Call.Return(_a0, _a1, _a2)
return _c
}
func (_c *Provider_RunRestore_Call) RunAndReturn(run func(ctx context.Context, snapshotID string, volumePath string, incremental bool, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, error)) *Provider_RunRestore_Call {
func (_c *Provider_RunRestore_Call) RunAndReturn(run func(ctx context.Context, snapshotID string, volumePath string, incremental bool, cbtParam provider.CBTParam, volMode uploader.PersistentVolumeMode, uploaderConfig map[string]string, updater uploader.ProgressUpdater) (int64, int64, error)) *Provider_RunRestore_Call {
_c.Call.Return(run)
return _c
}
+1 -1
View File
@@ -68,7 +68,7 @@ type Provider interface {
cbtParam CBTParam,
volMode uploader.PersistentVolumeMode,
uploaderConfig map[string]string,
updater uploader.ProgressUpdater) (int64, error)
updater uploader.ProgressUpdater) (int64, int64, error)
// Close which will close related repository
Close(ctx context.Context) error
}
+104
View File
@@ -2179,3 +2179,107 @@ func TestGetVSCForVS(t *testing.T) {
})
}
}
func TestCleanupVolumeSnapshot(t *testing.T) {
retainVSCName := "retain-vsc"
testCases := []struct {
name string
volSnap *snapshotv1api.VolumeSnapshot
objs []runtime.Object
expectDeleted bool
// name of the VolumeSnapshotContent expected to have been patched to
// DeletionPolicy=Delete; empty when no VSC should be touched.
expectedVSC string
}{
{
name: "should be a no-op if the VolumeSnapshot no longer exists",
volSnap: &snapshotv1api.VolumeSnapshot{
ObjectMeta: metav1.ObjectMeta{
Name: "does-not-exist",
Namespace: "velero",
},
},
objs: []runtime.Object{},
expectDeleted: false,
},
{
name: "should delete a VolumeSnapshot with no bound VolumeSnapshotContent",
volSnap: &snapshotv1api.VolumeSnapshot{
ObjectMeta: metav1.ObjectMeta{
Name: "vs-no-vsc",
Namespace: "velero",
},
},
objs: []runtime.Object{
&snapshotv1api.VolumeSnapshot{
ObjectMeta: metav1.ObjectMeta{
Name: "vs-no-vsc",
Namespace: "velero",
},
},
},
expectDeleted: true,
},
{
name: "should patch bound VSC DeletionPolicy to Delete and delete the VolumeSnapshot",
volSnap: &snapshotv1api.VolumeSnapshot{
ObjectMeta: metav1.ObjectMeta{
Name: "vs-with-vsc",
Namespace: "velero",
},
},
objs: []runtime.Object{
&snapshotv1api.VolumeSnapshot{
ObjectMeta: metav1.ObjectMeta{
Name: "vs-with-vsc",
Namespace: "velero",
},
Status: &snapshotv1api.VolumeSnapshotStatus{
BoundVolumeSnapshotContentName: &retainVSCName,
},
},
&snapshotv1api.VolumeSnapshotContent{
ObjectMeta: metav1.ObjectMeta{
Name: "retain-vsc",
},
Spec: snapshotv1api.VolumeSnapshotContentSpec{
DeletionPolicy: snapshotv1api.VolumeSnapshotContentRetain,
},
},
},
expectDeleted: true,
expectedVSC: "retain-vsc",
},
}
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
fakeClient := velerotest.NewFakeControllerRuntimeClient(t, tc.objs...)
CleanupVolumeSnapshot(t.Context(), tc.volSnap, fakeClient, velerotest.NewLogger())
actual := new(snapshotv1api.VolumeSnapshot)
err := fakeClient.Get(
t.Context(),
crclient.ObjectKey{Name: tc.volSnap.Name, Namespace: tc.volSnap.Namespace},
actual,
)
if tc.expectDeleted {
assert.True(t, apierrors.IsNotFound(err), "expected VolumeSnapshot to be deleted")
}
if tc.expectedVSC != "" {
actualVSC := new(snapshotv1api.VolumeSnapshotContent)
err := fakeClient.Get(
t.Context(),
crclient.ObjectKey{Name: tc.expectedVSC},
actualVSC,
)
require.NoError(t, err)
assert.Equal(t, snapshotv1api.VolumeSnapshotContentDelete, actualVSC.Spec.DeletionPolicy)
}
})
}
}
+8 -3
View File
@@ -161,7 +161,12 @@ Pagination can be entirely disabled by setting `--client-page-size` to `0`. This
## Deleting Backups
Use the following commands to delete Velero backups and data:
Use the following commands to delete Velero backups:
`velero backup delete <backupName>`: successful run of this command will:
- Immediately delete the resource backup data from the backup storage
- Immediately delete the volume snapshots associates to the backup if any (e.g., volumes are backed up with CSI snapshot backup or native snapshot method)
- Trigger the deletion of volume backup data if any data is persisted to the backup repository (e.g., volumes are backed up with CSI snapshot data movement or fs-backup method). You will not see the backup storage space is released immediately, the backup repository maintenance jobs will GC the data and finally release the storage space
* `kubectl delete backup <backupName> -n <veleroNamespace>` will delete the backup custom resource only and will not delete any associated data from object/block storage
* `velero backup delete <backupName>` will delete the backup resource including all data in object/block storage
Velero backup deletion needs extra spaces in the backup storage, so make sure the backup storage is not full during the backup deletion, otherwise, the backup deletion or the following backup repository maintenance jobs may fail.
`kubectl delete backup <backupName> -n <veleroNamespace>`: this command will delete the backup custom resource only and will not delete any associated data from the backup storage. So it is used for limited purposes only, e.g., all the backup data has been deleted in the backup storage, you just need to clear the orphaned CRs in the cluster.
@@ -19,6 +19,7 @@ package basic
import (
"context"
"fmt"
"slices"
"time"
. "github.com/onsi/ginkgo/v2"
@@ -45,26 +46,22 @@ func APIExtensionsVersionsTest() {
veleroCfg = VeleroCfg
Expect(KubectlConfigUseContext(context.Background(), veleroCfg.DefaultClusterContext)).To(Succeed())
srcVersions, err := GetAPIVersions(veleroCfg.DefaultClient, resourceName)
Expect(err).ShouldNot(HaveOccurred())
Expect(err).ToNot(HaveOccurred())
dstVersions, err := GetAPIVersions(veleroCfg.StandbyClient, resourceName)
Expect(err).ShouldNot(HaveOccurred())
Expect(err).ToNot(HaveOccurred())
Expect(srcVersions).Should(ContainElement("v1"), func() string {
if !slices.Contains(srcVersions, "v1") {
Skip("CRD with apiextension versions srcVersions should have v1")
return ""
})
Expect(srcVersions).Should(ContainElement("v1beta1"), func() string {
Skip("CRD with apiextension versions srcVersions should have v1")
return ""
})
Expect(dstVersions).Should(ContainElement("v1"), func() string {
}
if !slices.Contains(srcVersions, "v1beta1") {
Skip("CRD with apiextension versions srcVersions should have v1beta1")
}
if !slices.Contains(dstVersions, "v1") {
Skip("CRD with apiextension versions dstVersions should have v1")
return ""
})
Expect(len(srcVersions) > 1 && len(dstVersions) == 1).Should(BeTrue(), func() string {
}
if !(len(srcVersions) > 1 && len(dstVersions) == 1) {
Skip("Source cluster should support apiextension v1 and v1beta1, destination cluster should only support apiextension v1")
return ""
})
}
})
AfterEach(func() {
By(fmt.Sprintf("Switch to default kubeconfig context %s", veleroCfg.DefaultClusterContext), func() {
@@ -55,7 +55,7 @@ func (c *CSIDataMoverVolumeInfo) Verify() error {
BackupObjectsPrefix+"/"+c.BackupName,
)
Expect(err).ShouldNot(HaveOccurred(), "Fail to get VolumeInfo metadata in the Backup Repository.")
Expect(err).ToNot(HaveOccurred(), "Fail to get VolumeInfo metadata in the Backup Repository.")
fmt.Printf("The VolumeInfo metadata content: %+v\n", *volumeInfo[0])
Expect(volumeInfo).ToNot(BeEmpty())
@@ -54,7 +54,7 @@ func (c *CSISnapshotVolumeInfo) Verify() error {
BackupObjectsPrefix+"/"+c.BackupName,
)
Expect(err).ShouldNot(HaveOccurred(), "Fail to get VolumeInfo metadata in the Backup Repository.")
Expect(err).ToNot(HaveOccurred(), "Fail to get VolumeInfo metadata in the Backup Repository.")
fmt.Printf("The VolumeInfo metadata content: %+v\n", *volumeInfo[0])
Expect(volumeInfo).ToNot(BeEmpty())
@@ -54,7 +54,7 @@ func (f *FilesystemUploadVolumeInfo) Verify() error {
BackupObjectsPrefix+"/"+f.BackupName,
)
Expect(err).ShouldNot(HaveOccurred(), "Fail to get VolumeInfo metadata in the Backup Repository.")
Expect(err).ToNot(HaveOccurred(), "Fail to get VolumeInfo metadata in the Backup Repository.")
fmt.Printf("The VolumeInfo metadata content: %+v\n", *volumeInfo[0])
Expect(volumeInfo).ToNot(BeEmpty())
@@ -55,7 +55,7 @@ func (n *NativeSnapshotVolumeInfo) Verify() error {
BackupObjectsPrefix+"/"+n.BackupName,
)
Expect(err).ShouldNot(HaveOccurred(), "Fail to get VolumeInfo metadata in the Backup Repository.")
Expect(err).ToNot(HaveOccurred(), "Fail to get VolumeInfo metadata in the Backup Repository.")
fmt.Printf("The VolumeInfo metadata content: %+v\n", *volumeInfo[0])
Expect(volumeInfo).ToNot(BeEmpty())
@@ -54,7 +54,7 @@ func (s *SkippedVolumeInfo) Verify() error {
BackupObjectsPrefix+"/"+s.BackupName,
)
Expect(err).ShouldNot(HaveOccurred(), "Fail to get VolumeInfo metadata in the Backup Repository.")
Expect(err).ToNot(HaveOccurred(), "Fail to get VolumeInfo metadata in the Backup Repository.")
fmt.Printf("The VolumeInfo metadata content: %+v\n", *volumeInfo[0])
Expect(volumeInfo).ToNot(BeEmpty())
+2 -2
View File
@@ -308,7 +308,7 @@ func BslDeletionTest(useVolumeSnapshots bool) {
By(fmt.Sprintf("Get all backups from 2 BSLs %s before deleting one of them", backupLocation1), func() {
backupsBeforeDel, err := GetAllBackups(context.Background(), veleroCfg.VeleroCLI)
Expect(err).To(Succeed())
Expect(cmp.Diff(backupsInBsl1AndBsl2, backupsBeforeDel, cmpopts.SortSlices(less))).Should(BeEmpty())
Expect(cmp.Diff(backupsInBsl1AndBsl2, backupsBeforeDel, cmpopts.SortSlices(less))).To(BeEmpty())
By(fmt.Sprintf("Backup1 %s should exist in cloud object store before bsl deletion", backupName1), func() {
Expect(ObjectsShouldBeInBucket(veleroCfg.ObjectStoreProvider, veleroCfg.CloudCredentialsFile,
@@ -325,7 +325,7 @@ func BslDeletionTest(useVolumeSnapshots bool) {
backupsAfterDel, err := GetAllBackups(context.Background(), veleroCfg.VeleroCLI)
Expect(err).To(Succeed())
// Default BSL is deleted, so backups in additional BSL should be left only
Expect(cmp.Diff(backupsInBSL2, backupsAfterDel, cmpopts.SortSlices(less))).Should(BeEmpty())
Expect(cmp.Diff(backupsInBSL2, backupsAfterDel, cmpopts.SortSlices(less))).To(BeEmpty())
})
})
+5 -5
View File
@@ -127,7 +127,7 @@ func (e *ExcludeFromBackup) CreateResources() error {
}
By(fmt.Sprintf("Checking secret %s should exists in namespaces ...%s\n", secretName, namespace), func() {
_, err = GetSecret(e.Client.ClientGo, namespace, e.CaseBaseName)
Expect(err).ShouldNot(HaveOccurred(), fmt.Sprintf("failed to list deployment in namespace: %q", namespace))
Expect(err).ToNot(HaveOccurred(), fmt.Sprintf("failed to list deployment in namespace: %q", namespace))
})
//Create Configmap: to be included
configmaptName := e.CaseBaseName
@@ -148,21 +148,21 @@ func (e *ExcludeFromBackup) Verify() error {
By(fmt.Sprintf("Checking resources in namespaces ...%s\n", namespace), func() {
//Check namespace
checkNS, err := GetNamespace(e.Ctx, e.Client, namespace)
Expect(err).ShouldNot(HaveOccurred(), fmt.Sprintf("Could not retrieve test namespace %s", namespace))
Expect(err).ToNot(HaveOccurred(), fmt.Sprintf("Could not retrieve test namespace %s", namespace))
Expect(checkNS.Name).To(Equal(namespace), fmt.Sprintf("Retrieved namespace for %s has name %s instead", namespace, checkNS.Name))
//Check deployment: should be included
_, err = GetDeployment(e.Client.ClientGo, namespace, e.CaseBaseName)
Expect(err).ShouldNot(HaveOccurred(), fmt.Sprintf("failed to list deployment in namespace: %q", namespace))
Expect(err).ToNot(HaveOccurred(), fmt.Sprintf("failed to list deployment in namespace: %q", namespace))
//Check secrets: secrets should not be included
_, err = GetSecret(e.Client.ClientGo, namespace, e.CaseBaseName)
Expect(err).Should(HaveOccurred(), fmt.Sprintf("failed to list deployment in namespace: %q", namespace))
Expect(err).To(HaveOccurred(), fmt.Sprintf("failed to list deployment in namespace: %q", namespace))
Expect(apierrors.IsNotFound(err)).To(BeTrue())
//Check configmap: should be included
_, err = GetConfigMap(e.Client.ClientGo, namespace, e.CaseBaseName)
Expect(err).ShouldNot(HaveOccurred(), fmt.Sprintf("failed to list configmap in namespace: %q", namespace))
Expect(err).ToNot(HaveOccurred(), fmt.Sprintf("failed to list configmap in namespace: %q", namespace))
})
return nil
}
+1 -1
View File
@@ -170,7 +170,7 @@ func (s *InProgressCase) Backup() error {
}
// There should be at most one in-progress backup per schedule.
Expect(inProgressBackupCount).Should(BeNumerically("<=", 1))
Expect(inProgressBackupCount).To(BeNumerically("<=", 1))
// Already ensured at most one in-progress backup when schedule triggered 2 backups.
// Succeed.
+1 -1
View File
@@ -184,7 +184,7 @@ func (n *PeriodicalCase) Verify() error {
By("Namespaces were restored", func() {
for _, ns := range *n.NSIncluded {
_, err := k8sutil.GetConfigMap(n.Client.ClientGo, ns, n.CaseBaseName)
Expect(err).ShouldNot(HaveOccurred(), fmt.Sprintf("failed to list CM in namespace: %s\n", ns))
Expect(err).ToNot(HaveOccurred(), fmt.Sprintf("failed to list CM in namespace: %s\n", ns))
}
})
return nil