Merge branch 'main' into optimize-sub-object-description

.github/workflows/push.yml

@@ -8,6 +8,9 @@ on:
     tags:
       - '*'
 
+permissions:
+  contents: read
+
 jobs:
   get-go-version:
     uses: ./.github/workflows/get-go-version.yaml
@@ -20,21 +23,21 @@ jobs:
     needs: get-go-version
     steps:
       - name: Check out the code
-        uses: actions/checkout@v6
+        uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803 # v6
 
       - name: Set up Go version
-        uses: actions/setup-go@v6
+        uses: actions/setup-go@924ae3a1cded613372ab5595356fb5720e22ba16 # v6
         with:
           go-version: ${{ needs.get-go-version.outputs.version }}
 
       - name: Set up QEMU
         id: qemu
-        uses: docker/setup-qemu-action@v4
+        uses: docker/setup-qemu-action@96fe6ef7f33517b61c61be40b68a1882f3264fb8 # v4
         with:
           platforms: all
       - name: Set up Docker Buildx
         id: buildx
-        uses: docker/setup-buildx-action@v3
+        uses: docker/setup-buildx-action@8d2750c68a42422c14e847fe6c8ac0403b4cbd6f # v3
         with:
           version: latest
       - name: Build
@@ -45,7 +48,7 @@ jobs:
       - name: Test
         run: make test
       - name: Upload test coverage
-        uses: codecov/codecov-action@v7
+        uses: codecov/codecov-action@fb8b3582c8e4def4969c97caa2f19720cb33a72f # v7
         with:
           token: ${{ secrets.CODECOV_TOKEN }}
           files: coverage.out

changelogs/unreleased/10045-lubronzhan

@@ -0,0 +1 @@
+Fix issue #5836, respect schedule.spec.template.metadata.annotations to override annotations copied from the Schedule to Backup objects, matching the existing behavior for labels

changelogs/unreleased/10087-adam-jian-zhang

@@ -0,0 +1 @@
+Stop force-including VolumeSnapshotContents via resourceMustHave on every restore; CSI VolumeSnapshot/PVC RestoreItemActions now set restore.velero.io/must-include-additional-items so bound snapshot dependencies are restored only when their parent is restored (fixes #9957)

changelogs/unreleased/10091-Lyndon-Li

@@ -0,0 +1 @@
+Refactor block uploader thread module for better thread safety and code reading

config/crd/v1/bases/velero.io_backups.yaml

@@ -393,6 +393,11 @@ spec:
                 x-kubernetes-map-type: atomic
               metadata:
                 properties:
+                  annotations:
+                    additionalProperties:
+                      type: string
+                    nullable: true
+                    type: object
                   labels:
                     additionalProperties:
                       type: string

config/crd/v1/bases/velero.io_schedules.yaml

@@ -434,6 +434,11 @@ spec:
                     x-kubernetes-map-type: atomic
                   metadata:
                     properties:
+                      annotations:
+                        additionalProperties:
+                          type: string
+                        nullable: true
+                        type: object
                       labels:
                         additionalProperties:
                           type: string

pkg/apis/velero/v1/backup_types.go

@@ -23,6 +23,9 @@ import (
 
 type Metadata struct {
 	Labels map[string]string `json:"labels,omitempty"`
+	// +optional
+	// +nullable
+	Annotations map[string]string `json:"annotations,omitempty"`
 }
 
 // BackupSpec defines the specification for a Velero backup.

pkg/apis/velero/v1/zz_generated.deepcopy.go

@@ -895,6 +895,13 @@ func (in *Metadata) DeepCopyInto(out *Metadata) {
 			(*out)[key] = val
 		}
 	}
+	if in.Annotations != nil {
+		in, out := &in.Annotations, &out.Annotations
+		*out = make(map[string]string, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
+		}
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Metadata.

pkg/builder/backup_builder.go

@@ -109,8 +109,23 @@ func (b *BackupBuilder) FromSchedule(schedule *velerov1api.Schedule) *BackupBuil
 	b.object.Spec = schedule.Spec.Template
 	b.ObjectMeta(WithLabelsMap(labels))
 
-	if schedule.Annotations != nil {
-		b.ObjectMeta(WithAnnotationsMap(schedule.Annotations))
+	var annotations map[string]string
+
+	// Check if there's explicit Annotations defined in the Schedule object template
+	// and if present then copy it to the backup object.
+	if schedule.Spec.Template.Metadata.Annotations != nil {
+		logger := logging.DefaultLogger(logging.LogLevelFlag(logrus.InfoLevel).Parse(), logging.NewFormatFlag().Parse())
+		annotations = schedule.Spec.Template.Metadata.Annotations
+		logger.WithFields(logrus.Fields{
+			"backup":      fmt.Sprintf("%s/%s", b.object.GetNamespace(), b.object.GetName()),
+			"annotations": schedule.Spec.Template.Metadata.Annotations,
+		}).Info("Schedule.template.metadata.annotations set - using those annotations instead of schedule.annotations for backup object")
+	} else {
+		annotations = schedule.Annotations
+	}
+
+	if annotations != nil {
+		b.ObjectMeta(WithAnnotationsMap(annotations))
 	}
 
 	if boolptr.IsSetToTrue(schedule.Spec.UseOwnerReferencesInBackup) {

pkg/builder/backup_builder_test.go

@@ -0,0 +1,84 @@
+/*
+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 builder
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+)
+
+func TestBackupFromSchedule(t *testing.T) {
+	tests := []struct {
+		name                string
+		schedule            *velerov1api.Schedule
+		expectedLabels      map[string]string
+		expectedAnnotations map[string]string
+	}{
+		{
+			name: "no schedule labels/annotations and no template overrides",
+			schedule: ForSchedule("velero", "test").
+				Result(),
+			expectedLabels:      map[string]string{velerov1api.ScheduleNameLabel: "test"},
+			expectedAnnotations: nil,
+		},
+		{
+			name: "schedule labels/annotations are copied when no template override is set",
+			schedule: ForSchedule("velero", "test").
+				ObjectMeta(
+					WithLabels("schedule-label", "schedule-value"),
+					WithAnnotations("schedule-annotation", "schedule-value"),
+				).
+				Result(),
+			expectedLabels: map[string]string{
+				"schedule-label":              "schedule-value",
+				velerov1api.ScheduleNameLabel: "test",
+			},
+			expectedAnnotations: map[string]string{"schedule-annotation": "schedule-value"},
+		},
+		{
+			name: "template.metadata.labels/annotations override schedule labels/annotations",
+			schedule: ForSchedule("velero", "test").
+				ObjectMeta(
+					WithLabels("schedule-label", "schedule-value"),
+					WithAnnotations("schedule-annotation", "schedule-value"),
+				).
+				Template(velerov1api.BackupSpec{
+					Metadata: velerov1api.Metadata{
+						Labels:      map[string]string{"template-label": "template-value"},
+						Annotations: map[string]string{"template-annotation": "template-value"},
+					},
+				}).
+				Result(),
+			expectedLabels: map[string]string{
+				"template-label":              "template-value",
+				velerov1api.ScheduleNameLabel: "test",
+			},
+			expectedAnnotations: map[string]string{"template-annotation": "template-value"},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			backup := ForBackup("velero", "test-backup").FromSchedule(test.schedule).Result()
+			assert.Equal(t, test.expectedLabels, backup.GetLabels())
+			assert.Equal(t, test.expectedAnnotations, backup.GetAnnotations())
+		})
+	}
+}

pkg/restore/actions/csi/pvc_action.go

@@ -175,6 +175,15 @@ func (p *pvcRestoreItemAction) Execute(
 				Name:          vsName,
 				Namespace:     pvc.Namespace,
 			})
+
+			// Force-restore the VolumeSnapshot even when restore resource filters
+			// would otherwise exclude it (mirrors backup-side must-include).
+			annotations := pvc.GetAnnotations()
+			if annotations == nil {
+				annotations = map[string]string{}
+			}
+			annotations[velerov1api.MustIncludeAdditionalItemRestoreAnnotation] = "true"
+			pvc.SetAnnotations(annotations)
 		}
 	}
 

pkg/restore/actions/csi/pvc_action_test.go

@@ -371,6 +371,7 @@ func TestExecute(t *testing.T) {
 		backup               *velerov1api.Backup
 		restore              *velerov1api.Restore
 		pvc                  *corev1api.PersistentVolumeClaim
+		pvcFromBackup        *corev1api.PersistentVolumeClaim
 		vs                   *snapshotv1api.VolumeSnapshot
 		dataUploadResult     *corev1api.ConfigMap
 		expectedErr          string
@@ -402,15 +403,40 @@ func TestExecute(t *testing.T) {
 			vs: builder.ForVolumeSnapshot("velero", vsName).ObjectMeta(
 				builder.WithAnnotations(velerov1api.VolumeSnapshotRestoreSize, "10Gi"),
 			).Result(),
-			expectedPVC: builder.ForPersistentVolumeClaim("velero", "testPVC").ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName")).Result(),
+			expectedPVC: builder.ForPersistentVolumeClaim("velero", "testPVC").ObjectMeta(builder.WithAnnotations(
+				velerov1api.VolumeSnapshotLabel, "vsName",
+				velerov1api.MustIncludeAdditionalItemRestoreAnnotation, "true",
+			)).Result(),
 		},
 		{
-			name:        "Restore from VolumeSnapshot without volume-snapshot-name annotation",
-			backup:      builder.ForBackup("velero", "testBackup").Result(),
-			restore:     builder.ForRestore("velero", "testRestore").Backup("testBackup").Result(),
-			pvc:         builder.ForPersistentVolumeClaim("velero", "testPVC").ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName", AnnSelectedNode, "node1")).Result(),
-			vs:          builder.ForVolumeSnapshot("velero", "testVS").ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotRestoreSize, "10Gi")).Result(),
-			expectedPVC: builder.ForPersistentVolumeClaim("velero", "testPVC").ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName", AnnSelectedNode, "node1")).Result(),
+			name:    "Restore from VolumeSnapshot with nil PVC annotations",
+			backup:  builder.ForBackup("velero", "testBackup").Result(),
+			restore: builder.ForRestore("velero", "testRestore").ObjectMeta(builder.WithUID("restoreUID")).Backup("testBackup").Result(),
+			pvc: &corev1api.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "testPVC",
+					Namespace: "velero",
+				},
+			},
+			pvcFromBackup: builder.ForPersistentVolumeClaim("velero", "testPVC").ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName")).Result(),
+			vs: builder.ForVolumeSnapshot("velero", vsName).ObjectMeta(
+				builder.WithAnnotations(velerov1api.VolumeSnapshotRestoreSize, "10Gi"),
+			).Result(),
+			expectedPVC: builder.ForPersistentVolumeClaim("velero", "testPVC").ObjectMeta(builder.WithAnnotations(
+				velerov1api.MustIncludeAdditionalItemRestoreAnnotation, "true",
+			)).Result(),
+		},
+		{
+			name:    "Restore from VolumeSnapshot without volume-snapshot-name annotation",
+			backup:  builder.ForBackup("velero", "testBackup").Result(),
+			restore: builder.ForRestore("velero", "testRestore").Backup("testBackup").Result(),
+			pvc:     builder.ForPersistentVolumeClaim("velero", "testPVC").ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotLabel, "vsName", AnnSelectedNode, "node1")).Result(),
+			vs:      builder.ForVolumeSnapshot("velero", "testVS").ObjectMeta(builder.WithAnnotations(velerov1api.VolumeSnapshotRestoreSize, "10Gi")).Result(),
+			expectedPVC: builder.ForPersistentVolumeClaim("velero", "testPVC").ObjectMeta(builder.WithAnnotations(
+				velerov1api.VolumeSnapshotLabel, "vsName",
+				AnnSelectedNode, "node1",
+				velerov1api.MustIncludeAdditionalItemRestoreAnnotation, "true",
+			)).Result(),
 		},
 		{
 			name:        "DataUploadResult cannot be found",
@@ -480,7 +506,13 @@ func TestExecute(t *testing.T) {
 				require.NoError(t, err)
 
 				input.Item = &unstructured.Unstructured{Object: pvcMap}
-				input.ItemFromBackup = &unstructured.Unstructured{Object: pvcMap}
+				if tc.pvcFromBackup != nil {
+					pvcFromBackupMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(tc.pvcFromBackup)
+					require.NoError(t, err)
+					input.ItemFromBackup = &unstructured.Unstructured{Object: pvcFromBackupMap}
+				} else {
+					input.ItemFromBackup = &unstructured.Unstructured{Object: pvcMap}
+				}
 				input.Restore = tc.restore
 			}
 			if tc.preCreatePVC {
@@ -508,6 +540,12 @@ func TestExecute(t *testing.T) {
 				err := runtime.DefaultUnstructuredConverter.FromUnstructured(output.UpdatedItem.UnstructuredContent(), pvc)
 				require.NoError(t, err)
 				require.Equal(t, tc.expectedPVC.GetObjectMeta(), pvc.GetObjectMeta())
+				if tc.name == "Restore from VolumeSnapshot" {
+					require.Equal(t, "true", pvc.GetAnnotations()[velerov1api.MustIncludeAdditionalItemRestoreAnnotation])
+					require.Len(t, output.AdditionalItems, 1)
+					require.Equal(t, "volumesnapshots.snapshot.storage.k8s.io", output.AdditionalItems[0].GroupResource.String())
+					require.Equal(t, "vsName", output.AdditionalItems[0].Name)
+				}
 				if pvc.Spec.Selector != nil && pvc.Spec.Selector.MatchLabels != nil {
 					// This is used for long name and namespace case.
 					if len(tc.pvc.Namespace+"."+tc.pvc.Name) >= validation.DNS1035LabelMaxLength {

pkg/restore/actions/csi/volumesnapshot_action.go

@@ -66,6 +66,9 @@ func resetVolumeSnapshotSpecForRestore(vs *snapshotv1api.VolumeSnapshot, vscName
 }
 
 func resetVolumeSnapshotAnnotation(vs *snapshotv1api.VolumeSnapshot) {
+	if vs.ObjectMeta.Annotations == nil {
+		vs.ObjectMeta.Annotations = make(map[string]string)
+	}
 	vs.ObjectMeta.Annotations[velerov1api.VSCDeletionPolicyAnnotation] =
 		string(snapshotv1api.VolumeSnapshotContentRetain)
 }
@@ -282,12 +285,6 @@ func (p *volumeSnapshotRestoreItemAction) Execute(
 			vs.Namespace, vs.Name)
 	}
 
-	vsMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(&vs)
-	if err != nil {
-		p.log.Errorf("Fail to convert VS %s to unstructured", vs.Namespace+"/"+vs.Name)
-		return nil, errors.WithStack(err)
-	}
-
 	if vsFromBackup.Status == nil ||
 		vsFromBackup.Status.BoundVolumeSnapshotContentName == nil {
 		p.log.Errorf("VS %s doesn't have bound VSC", vsFromBackup.Name)
@@ -299,6 +296,21 @@ func (p *volumeSnapshotRestoreItemAction) Execute(
 		Name:          *vsFromBackup.Status.BoundVolumeSnapshotContentName,
 	}
 
+	// Force-restore the bound VSC even when restore resource filters would
+	// otherwise exclude it (mirrors backup-side must-include for CSI deps).
+	annotations := vs.GetAnnotations()
+	if annotations == nil {
+		annotations = map[string]string{}
+	}
+	annotations[velerov1api.MustIncludeAdditionalItemRestoreAnnotation] = "true"
+	vs.SetAnnotations(annotations)
+
+	vsMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(&vs)
+	if err != nil {
+		p.log.Errorf("Fail to convert VS %s to unstructured", vs.Namespace+"/"+vs.Name)
+		return nil, errors.WithStack(err)
+	}
+
 	p.log.Infof(`Returning from VolumeSnapshotRestoreItemAction with 
 		VolumeSnapshotContent in additionalItems`)
 

pkg/restore/actions/csi/volumesnapshot_action_test.go

@@ -103,6 +103,26 @@ func TestResetVolumeSnapshotSpecForRestore(t *testing.T) {
 	}
 }
 
+func TestResetVolumeSnapshotAnnotation(t *testing.T) {
+	t.Run("should set deletion policy annotation when annotations is nil", func(t *testing.T) {
+		vs := snapshotv1api.VolumeSnapshot{}
+		resetVolumeSnapshotAnnotation(&vs)
+		assert.NotNil(t, vs.ObjectMeta.Annotations)
+		assert.Equal(t, string(snapshotv1api.VolumeSnapshotContentRetain), vs.ObjectMeta.Annotations[velerov1api.VSCDeletionPolicyAnnotation])
+	})
+
+	t.Run("should preserve existing annotations and set deletion policy annotation", func(t *testing.T) {
+		vs := snapshotv1api.VolumeSnapshot{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: map[string]string{"foo": "bar"},
+			},
+		}
+		resetVolumeSnapshotAnnotation(&vs)
+		assert.Equal(t, "bar", vs.ObjectMeta.Annotations["foo"])
+		assert.Equal(t, string(snapshotv1api.VolumeSnapshotContentRetain), vs.ObjectMeta.Annotations[velerov1api.VSCDeletionPolicyAnnotation])
+	})
+}
+
 func TestVSExecute(t *testing.T) {
 	newVscName := util.GenerateSha256FromRestoreUIDAndVsName("restoreUID", "vsName")
 	tests := []struct {
@@ -145,6 +165,18 @@ func TestVSExecute(t *testing.T) {
 			expectErr:  false,
 			expectedVS: builder.ForVolumeSnapshot("ns", "test").SourceVolumeSnapshotContentName(newVscName).Result(),
 		},
+		{
+			name: "Normal case with nil VS annotations, VSC should be created",
+			vs: builder.ForVolumeSnapshot("ns", "vsName").
+				SourceVolumeSnapshotContentName(newVscName).
+				VolumeSnapshotClass("vscClass").
+				Status().
+				BoundVolumeSnapshotContentName("vscName").
+				Result(),
+			restore:    builder.ForRestore("velero", "restore").ObjectMeta(builder.WithUID("restoreUID")).Result(),
+			expectErr:  false,
+			expectedVS: builder.ForVolumeSnapshot("ns", "test").SourceVolumeSnapshotContentName(newVscName).Result(),
+		},
 	}
 
 	for _, test := range tests {
@@ -184,6 +216,10 @@ func TestVSExecute(t *testing.T) {
 				require.NoError(t, runtime.DefaultUnstructuredConverter.FromUnstructured(
 					result.UpdatedItem.UnstructuredContent(), &vs))
 				require.Equal(t, test.expectedVS.Spec, vs.Spec)
+				require.Equal(t, "true", vs.GetAnnotations()[velerov1api.MustIncludeAdditionalItemRestoreAnnotation])
+				require.Len(t, result.AdditionalItems, 1)
+				require.Equal(t, "volumesnapshotcontents.snapshot.storage.k8s.io", result.AdditionalItems[0].GroupResource.String())
+				require.Equal(t, "vscName", result.AdditionalItems[0].Name)
 			}
 		})
 	}

pkg/restore/restore.go

@@ -87,7 +87,6 @@ const ObjectStatusRestoreAnnotationKey = "velero.io/restore-status"
 
 var resourceMustHave = []string{
 	"datauploads.velero.io",
-	"volumesnapshotcontents.snapshot.storage.k8s.io",
 }
 
 type VolumeSnapshotterGetter interface {

pkg/restore/restore_test.go

@@ -754,6 +754,29 @@ func TestRestoreResourceFiltering(t *testing.T) {
 			apiResources: []*test.APIResource{test.ServiceAccounts()},
 			want:         map[*test.APIResource][]string{test.ServiceAccounts(): {"ns-1/sa-1"}},
 		},
+		{
+			// Regression for #9957: VSC must not be force-included via resourceMustHave
+			// when the restore only selects unrelated resource types.
+			name:    "volumesnapshotcontents are not force-included for selective resource restores",
+			restore: defaultRestore().IncludedResources("storageclasses").IncludeClusterResources(true).Result(),
+			backup:  defaultBackup().Result(),
+			tarball: test.NewTarWriter(t).
+				AddItems("storageclasses.storage.k8s.io",
+					builder.ForStorageClass("sc-1").Result(),
+				).
+				AddItems("volumesnapshotcontents.snapshot.storage.k8s.io",
+					builder.ForVolumeSnapshotContent("vsc-1").Result(),
+				).
+				Done(),
+			apiResources: []*test.APIResource{
+				test.StorageClasses(),
+				test.VolumeSnapshotContents(),
+			},
+			want: map[*test.APIResource][]string{
+				test.StorageClasses():         {"/sc-1"},
+				test.VolumeSnapshotContents(): nil,
+			},
+		},
 	}
 
 	for _, tc := range tests {
@@ -2592,6 +2615,52 @@ func TestRestoreMustIncludeAdditionalItems(t *testing.T) {
 			test.PVCs(): nil,
 		})
 	})
+
+	t.Run("VS must-include restores excluded VolumeSnapshotContent additional item", func(t *testing.T) {
+		h := newHarness(t)
+		h.AddItems(t, test.VolumeSnapshots())
+		h.AddItems(t, test.VolumeSnapshotContents())
+
+		data := &Request{
+			Log:     h.log,
+			Restore: defaultRestore().IncludedResources("volumesnapshots.snapshot.storage.k8s.io").IncludeClusterResources(true).Result(),
+			Backup:  defaultBackup().Result(),
+			BackupReader: test.NewTarWriter(t).
+				AddItems("volumesnapshots.snapshot.storage.k8s.io", builder.ForVolumeSnapshot("ns-1", "vs-1").Result()).
+				AddItems("volumesnapshotcontents.snapshot.storage.k8s.io", builder.ForVolumeSnapshotContent("vsc-1").Result()).
+				Done(),
+		}
+		warnings, errs := h.restorer.Restore(
+			data,
+			[]riav2.RestoreItemAction{
+				&pluggableAction{
+					selector: velero.ResourceSelector{IncludedResources: []string{"volumesnapshots.snapshot.storage.k8s.io"}},
+					executeFunc: func(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+						item := input.Item.(*unstructured.Unstructured)
+						annotations := item.GetAnnotations()
+						if annotations == nil {
+							annotations = map[string]string{}
+						}
+						annotations[velerov1api.MustIncludeAdditionalItemRestoreAnnotation] = "true"
+						item.SetAnnotations(annotations)
+						return &velero.RestoreItemActionExecuteOutput{
+							UpdatedItem: item,
+							AdditionalItems: []velero.ResourceIdentifier{
+								{GroupResource: kuberesource.VolumeSnapshotContents, Name: "vsc-1"},
+							},
+						}, nil
+					},
+				},
+			},
+			nil,
+		)
+
+		assertEmptyResults(t, warnings, errs)
+		assertAPIContents(t, h, map[*test.APIResource][]string{
+			test.VolumeSnapshots():        {"ns-1/vs-1"},
+			test.VolumeSnapshotContents(): {"/vsc-1"},
+		})
+	})
 }
 
 // TestShouldRestore runs the ShouldRestore function for various permutations of

pkg/test/api_server.go

@@ -58,6 +58,9 @@ func NewAPIServer(t *testing.T) *APIServer {
 				{Group: "velero.io", Version: "v2alpha1", Resource: "datauploads"}:                         "DataUploadsList",
 				{Group: "mygroup.io", Version: "v1", Resource: "mycustomkinds"}:                            "MyCustomKindList",
 				{Group: "mygroup.io", Version: "v1", Resource: "myclustercustomkinds"}:                     "MyClusterCustomKindList",
+				{Group: "storage.k8s.io", Version: "v1", Resource: "storageclasses"}:                       "StorageClassList",
+				{Group: "snapshot.storage.k8s.io", Version: "v1", Resource: "volumesnapshots"}:             "VolumeSnapshotList",
+				{Group: "snapshot.storage.k8s.io", Version: "v1", Resource: "volumesnapshotcontents"}:      "VolumeSnapshotContentList",
 			})
 		discoveryClient = &DiscoveryClient{FakeDiscovery: kubeClient.Discovery().(*discoveryfake.FakeDiscovery)}
 	)

pkg/test/resources.go

@@ -220,3 +220,37 @@ func DataUploads(items ...metav1.Object) *APIResource {
 		Items:      items,
 	}
 }
+
+func StorageClasses(items ...metav1.Object) *APIResource {
+	return &APIResource{
+		Group:      "storage.k8s.io",
+		Version:    "v1",
+		Name:       "storageclasses",
+		ShortName:  "sc",
+		Kind:       "StorageClass",
+		Namespaced: false,
+		Items:      items,
+	}
+}
+
+func VolumeSnapshotContents(items ...metav1.Object) *APIResource {
+	return &APIResource{
+		Group:      "snapshot.storage.k8s.io",
+		Version:    "v1",
+		Name:       "volumesnapshotcontents",
+		Kind:       "VolumeSnapshotContent",
+		Namespaced: false,
+		Items:      items,
+	}
+}
+
+func VolumeSnapshots(items ...metav1.Object) *APIResource {
+	return &APIResource{
+		Group:      "snapshot.storage.k8s.io",
+		Version:    "v1",
+		Name:       "volumesnapshots",
+		Kind:       "VolumeSnapshot",
+		Namespaced: true,
+		Items:      items,
+	}
+}

pkg/uploader/block/uploader.go

@@ -25,6 +25,7 @@ import (
 	"runtime"
 	"strconv"
 	"strings"
+	"sync"
 
 	"github.com/cockroachdb/errors"
 	"github.com/sirupsen/logrus"
@@ -212,112 +213,33 @@ func (r *readResult) resetBuffer(list *freelist.FreeList) {
 
 func (blkup *blockUploader) backupData(reader io.ReaderAt, writer udmrepo.ObjectWriter, bitmap cbt.Iterator, totalLength int64) (int64, int64, error) {
 	blockSize := bitmap.BlockSize()
+	totalCount := int64(bitmap.Count())
 	list := freelist.New(bufferSize, int(blockSize))
 	resultChan := make(chan readResult, list.Capacity())
-	totalCount := bitmap.Count()
-	aligned := (totalLength + int64(blockSize) - 1) / int64(blockSize) * int64(blockSize)
-
 	quit := make(chan struct{})
-	defer close(quit)
+	aligned := (totalLength + int64(blockSize) - 1) / int64(blockSize) * int64(blockSize)
+	wg := &sync.WaitGroup{}
+	var writeErr error
+	var written int64
+	var lastPos int64
+
+	wg.Add(2)
 
 	go func() {
-		defer close(resultChan)
-
-		offset, valid := bitmap.Next()
-		var buffer []byte
-		for valid {
-			select {
-			case <-blkup.ctx.Done():
-				return
-			case <-quit:
-				return
-			case buffer = <-list.Chunks():
-			}
-
-			length := blockSize
-			if offset+uint64(length) > uint64(totalLength) {
-				length = uint(uint64(totalLength) - offset)
-				clear(buffer)
-			}
-
-			readBytes, err := reader.ReadAt(buffer[:length], int64(offset))
-			if err == nil && readBytes <= 0 {
-				err = io.ErrUnexpectedEOF
-			}
-
-			r := readResult{
-				buffer: buffer,
-				offset: int64(offset),
-				err:    err,
-			}
-
-			if r.err != nil {
-				r.resetBuffer(list)
-			}
-
-			resultChan <- r
-
-			if r.err != nil {
-				return
-			}
-
-			offset, valid = bitmap.Next()
-		}
+		defer wg.Done()
+		backupReadProc(blkup.ctx, reader, resultChan, quit, bitmap, list, totalLength)
 	}()
 
-	var lastPos int64
-	var result readResult
-	var written int64
-	var curCount int64
-	var writeErr error
-	var readerRunning bool
+	go func() {
+		defer wg.Done()
+		defer close(quit)
+		written, lastPos, writeErr = backupWriteProc(blkup.ctx, writer, resultChan, list, aligned, totalCount, int(blockSize), blkup.progress)
+	}()
 
-	for curCount < int64(totalCount) {
-		select {
-		case <-blkup.ctx.Done():
-			writeErr = ErrCanceled
-		case result, readerRunning = <-resultChan:
-			if !readerRunning {
-				if blkup.ctx.Err() != nil {
-					writeErr = ErrCanceled
-				} else {
-					writeErr = io.ErrUnexpectedEOF
-				}
-			}
-		}
-
-		if writeErr != nil {
-			break
-		}
-
-		if result.err != nil {
-			writeErr = result.err
-			break
-		}
-
-		n, err := writer.WriteAt(result.buffer, result.offset)
-		if err != nil {
-			writeErr = err
-			break
-		}
-
-		if blockSize != uint(n) {
-			writeErr = io.ErrShortWrite
-			break
-		}
-
-		written += int64(blockSize)
-		lastPos = result.offset + int64(blockSize)
-		result.resetBuffer(list)
-		curCount++
-
-		blkup.progress.UpdateProgress(&uploader.Progress{BytesDone: lastPos, TotalBytes: aligned})
-	}
-
-	result.resetBuffer(list)
+	wg.Wait()
 
 	if writeErr != nil {
-		return written, aligned, writeErr
+		return written, aligned, errors.Wrap(writeErr, "error writing data")
 	}
 
 	if lastPos < aligned {
@@ -334,6 +256,119 @@ func (blkup *blockUploader) backupData(reader io.ReaderAt, writer udmrepo.Object
 	return written, aligned, nil
 }
 
+func backupReadProc(ctx context.Context, reader io.ReaderAt, resultChan chan readResult, quit chan struct{}, bitmap cbt.Iterator, list *freelist.FreeList, totalLength int64) {
+	defer close(resultChan)
+
+	blockSize := bitmap.BlockSize()
+	offset, valid := bitmap.Next()
+	var buffer []byte
+	for valid {
+		select {
+		case <-ctx.Done():
+			return
+		case <-quit:
+			return
+		case buffer = <-list.Chunks():
+		}
+
+		length := blockSize
+		if offset+uint64(length) > uint64(totalLength) {
+			length = uint(uint64(totalLength) - offset)
+			clear(buffer)
+		}
+
+		readBytes, err := reader.ReadAt(buffer[:length], int64(offset))
+		if err == nil && readBytes <= 0 {
+			err = io.ErrUnexpectedEOF
+		}
+
+		r := readResult{
+			buffer: buffer,
+			offset: int64(offset),
+			err:    err,
+		}
+
+		if r.err != nil {
+			r.resetBuffer(list)
+		}
+
+		resultChan <- r
+
+		if r.err != nil {
+			return
+		}
+
+		offset, valid = bitmap.Next()
+	}
+}
+
+func backupWriteProc(ctx context.Context, writer udmrepo.ObjectWriter, resultChan chan readResult, list *freelist.FreeList, totalLength int64,
+	totalCount int64, blockSize int, progress uploader.ProgressUpdater) (int64, int64, error) {
+	var lastPos int64
+	var result readResult
+	var written int64
+	var curCount int64
+	var writeErr error
+
+	for {
+		select {
+		case <-ctx.Done():
+			writeErr = ErrCanceled
+		case r, ok := <-resultChan:
+			if !ok {
+				if ctx.Err() != nil {
+					writeErr = ErrCanceled
+				}
+			} else {
+				result = r
+			}
+		}
+
+		if writeErr != nil {
+			break
+		}
+
+		if result.err != nil {
+			writeErr = result.err
+			break
+		}
+
+		if result.buffer == nil {
+			break
+		}
+
+		n, err := writer.WriteAt(result.buffer, result.offset)
+		if err != nil {
+			writeErr = err
+			break
+		}
+
+		if blockSize != n {
+			writeErr = io.ErrShortWrite
+			break
+		}
+
+		written += int64(blockSize)
+		lastPos = result.offset + int64(blockSize)
+		result.resetBuffer(list)
+		curCount++
+
+		progress.UpdateProgress(&uploader.Progress{BytesDone: lastPos, TotalBytes: totalLength})
+	}
+
+	result.resetBuffer(list)
+
+	if writeErr != nil {
+		return written, lastPos, writeErr
+	}
+
+	if curCount < totalCount {
+		return written, lastPos, io.ErrUnexpectedEOF
+	}
+
+	return written, lastPos, nil
+}
+
 func copyTailData(source io.ReaderAt, writer udmrepo.ObjectWriter, totalLength int64, blockSize int64) (int64, error) {
 	roundUp := (totalLength + blockSize - 1) / blockSize * blockSize
 	roundDown := totalLength / blockSize * blockSize
@@ -364,83 +399,111 @@ func getObjectName(source string) string {
 }
 
 func (blkup *blockUploader) restoreData(reader io.ReadSeeker, dest *os.File, bitmap cbt.Iterator, totalLength int64, destPath string) (int64, error) {
-	list := freelist.New(bufferSize, blockSize)
+	blockSize := bitmap.BlockSize()
+	totalCount := int64(bitmap.Count())
+	list := freelist.New(bufferSize, int(blockSize))
 	resultChan := make(chan readResult, list.Capacity())
-	zeroBlock := make([]byte, blockSize)
-	totalCount := bitmap.Count()
-
 	quit := make(chan struct{})
-	defer close(quit)
+	var writeErr error
+	var written int64
+
+	wg := &sync.WaitGroup{}
+
+	wg.Add(2)
 
 	go func() {
-		defer close(resultChan)
-
-		offset, valid := bitmap.Next()
-		var buffer []byte
-		var nextPos = uint64(0)
-		for valid {
-			select {
-			case <-blkup.ctx.Done():
-				return
-			case <-quit:
-				return
-			case buffer = <-list.Chunks():
-			}
-
-			var err error
-
-			if nextPos != offset {
-				_, err = reader.Seek(int64(offset), io.SeekStart)
-			}
-
-			if err == nil {
-				var length int
-				length, err = io.ReadFull(reader, buffer)
-				if err == nil && length <= 0 {
-					err = io.ErrUnexpectedEOF
-				}
-			}
-
-			r := readResult{
-				buffer: buffer,
-				offset: int64(offset),
-				err:    err,
-			}
-
-			if r.err != nil {
-				r.resetBuffer(list)
-			}
-
-			resultChan <- r
-
-			if r.err != nil {
-				return
-			}
-
-			nextPos = offset + uint64(blockSize)
-			offset, valid = bitmap.Next()
-		}
+		defer wg.Done()
+		restoreReadProc(blkup.ctx, reader, resultChan, quit, bitmap, list)
 	}()
 
+	go func() {
+		defer wg.Done()
+		defer close(quit)
+		written, writeErr = restoreWriteProc(blkup.ctx, dest, resultChan, list, totalLength, totalCount, int(blockSize), destPath, blkup.progress, blkup.log)
+	}()
+
+	wg.Wait()
+
+	if writeErr != nil {
+		return written, errors.Wrap(writeErr, "error writing data")
+	}
+
+	return written, nil
+}
+
+func restoreReadProc(ctx context.Context, reader io.ReadSeeker, resultChan chan readResult, quit chan struct{}, bitmap cbt.Iterator, list *freelist.FreeList) {
+	defer close(resultChan)
+
+	blockSize := bitmap.BlockSize()
+	offset, valid := bitmap.Next()
+	var buffer []byte
+	var nextPos = uint64(0)
+	for valid {
+		select {
+		case <-ctx.Done():
+			return
+		case <-quit:
+			return
+		case buffer = <-list.Chunks():
+		}
+
+		var err error
+
+		if nextPos != offset {
+			_, err = reader.Seek(int64(offset), io.SeekStart)
+		}
+
+		if err == nil {
+			var length int
+			length, err = io.ReadFull(reader, buffer)
+			if err == nil && length <= 0 {
+				err = io.ErrUnexpectedEOF
+			}
+		}
+
+		r := readResult{
+			buffer: buffer,
+			offset: int64(offset),
+			err:    err,
+		}
+
+		if r.err != nil {
+			r.resetBuffer(list)
+		}
+
+		resultChan <- r
+
+		if r.err != nil {
+			return
+		}
+
+		nextPos = offset + uint64(blockSize)
+		offset, valid = bitmap.Next()
+	}
+}
+
+func restoreWriteProc(ctx context.Context, dest *os.File, resultChan chan readResult, list *freelist.FreeList, totalLength int64, totalCount int64,
+	blockSize int, destPath string, progress uploader.ProgressUpdater, log logrus.FieldLogger) (int64, error) {
+	zeroBlock := make([]byte, blockSize)
+
 	var written int64
 	var result readResult
 	var writeErr error
-	var readerRunning bool
 	var zeroStart int64 = -1
 	var zeroLength int64
 	var curCount int64
 
-	for curCount < int64(totalCount) {
+	for {
 		select {
-		case <-blkup.ctx.Done():
+		case <-ctx.Done():
 			writeErr = ErrCanceled
-		case result, readerRunning = <-resultChan:
-			if !readerRunning {
-				if blkup.ctx.Err() != nil {
+		case r, ok := <-resultChan:
+			if !ok {
+				if ctx.Err() != nil {
 					writeErr = ErrCanceled
-				} else {
-					writeErr = io.ErrUnexpectedEOF
 				}
+			} else {
+				result = r
 			}
 		}
 
@@ -453,6 +516,10 @@ func (blkup *blockUploader) restoreData(reader io.ReadSeeker, dest *os.File, bit
 			break
 		}
 
+		if result.buffer == nil {
+			break
+		}
+
 		length := min(int64(blockSize), totalLength-result.offset)
 		if bytes.Equal(result.buffer, zeroBlock) {
 			if zeroStart == -1 {
@@ -461,7 +528,7 @@ func (blkup *blockUploader) restoreData(reader io.ReadSeeker, dest *os.File, bit
 			} else if result.offset == zeroStart+zeroLength {
 				zeroLength += length
 			} else {
-				if err := blkup.flushZeroBlocks(dest, zeroStart, zeroLength, zeroBlock, destPath); err != nil {
+				if err := flushZeroBlocks(dest, zeroStart, zeroLength, zeroBlock, destPath, log); err != nil {
 					writeErr = errors.Wrapf(err, "error flushing zero blocks from %v, length %v", zeroStart, zeroLength)
 					break
 				}
@@ -470,7 +537,7 @@ func (blkup *blockUploader) restoreData(reader io.ReadSeeker, dest *os.File, bit
 			}
 		} else {
 			if zeroStart != -1 {
-				if err := blkup.flushZeroBlocks(dest, zeroStart, zeroLength, zeroBlock, destPath); err != nil {
+				if err := flushZeroBlocks(dest, zeroStart, zeroLength, zeroBlock, destPath, log); err != nil {
 					writeErr = errors.Wrapf(err, "error flushing zero blocks from %v, length %v", zeroStart, zeroLength)
 					break
 				}
@@ -496,7 +563,7 @@ func (blkup *blockUploader) restoreData(reader io.ReadSeeker, dest *os.File, bit
 
 		result.resetBuffer(list)
 
-		blkup.progress.UpdateProgress(&uploader.Progress{BytesDone: written, TotalBytes: totalLength})
+		progress.UpdateProgress(&uploader.Progress{BytesDone: written, TotalBytes: totalLength})
 	}
 
 	result.resetBuffer(list)
@@ -505,8 +572,12 @@ func (blkup *blockUploader) restoreData(reader io.ReadSeeker, dest *os.File, bit
 		return written, writeErr
 	}
 
+	if curCount < totalCount {
+		return written, io.ErrUnexpectedEOF
+	}
+
 	if zeroStart != -1 {
-		if err := blkup.flushZeroBlocks(dest, zeroStart, zeroLength, zeroBlock, destPath); err != nil {
+		if err := flushZeroBlocks(dest, zeroStart, zeroLength, zeroBlock, destPath, log); err != nil {
 			return written, errors.Wrapf(err, "error flushing zero blocks from %v, length %v", zeroStart, zeroLength)
 		}
 	}
@@ -514,13 +585,13 @@ func (blkup *blockUploader) restoreData(reader io.ReadSeeker, dest *os.File, bit
 	return written, nil
 }
 
-func (blkup *blockUploader) flushZeroBlocks(dest *os.File, start int64, length int64, zeroBlock []byte, destPath string) error {
+func flushZeroBlocks(dest *os.File, start int64, length int64, zeroBlock []byte, destPath string, log logrus.FieldLogger) error {
 	err := blkZeroOut(dest, start, length)
 	if err == nil {
 		return nil
 	}
 
-	blkup.log.WithError(err).Warnf("Failed to call zero out from dev %s, start %v, length %v. Fallback to conservative way", destPath, start, length)
+	log.WithError(err).Warnf("Failed to call zero out from dev %s, start %v, length %v. Fallback to conservative way", destPath, start, length)
 
 	var written int64
 	for written < length {

pkg/uploader/block/uploader_test.go

@@ -197,7 +197,7 @@ func TestBlockUploaderBackup(t *testing.T) {
 			name:             "canceled in progress",
 			cancelInProgress: true,
 			expectErr:        true,
-			expectErrStr:     "error backing up bdev /data/volume1: uploader is canceled",
+			expectErrStr:     "error backing up bdev /data/volume1: error writing data: uploader is canceled",
 		},
 		{
 			name:         "create object writer err",
@@ -523,13 +523,11 @@ func TestFlushZeroBlocks(t *testing.T) {
 
 		require.NoError(t, f.Truncate(2048))
 
-		blkup := &blockUploader{
-			log: logrus.New(),
-		}
-		blkup.log.(*logrus.Logger).Out = io.Discard
+		log := logrus.New()
+		log.Out = io.Discard
 
 		zeroBlock := make([]byte, 1024)
-		err = blkup.flushZeroBlocks(f, 0, 2048, zeroBlock, f.Name())
+		err = flushZeroBlocks(f, 0, 2048, zeroBlock, f.Name(), log)
 
 		require.NoError(t, err)
 
@@ -577,6 +575,7 @@ func TestRestoreData(t *testing.T) {
 		iterMock.On("Count").Return(uint64(1))
 		iterMock.On("Next").Return(uint64(0), true).Once()
 		iterMock.On("Next").Return(uint64(0), false)
+		iterMock.On("BlockSize").Return(uint(1048576))
 
 		written, err := blkup.restoreData(reader, f, iterMock, 1048576, f.Name())
 		require.NoError(t, err)
@@ -606,6 +605,7 @@ func TestRestoreData(t *testing.T) {
 		iterMock.On("Count").Return(uint64(1))
 		iterMock.On("Next").Return(uint64(0), true).Once()
 		iterMock.On("Next").Return(uint64(0), false)
+		iterMock.On("BlockSize").Return(uint(1048576))
 
 		_, err = blkup.restoreData(reader, f, iterMock, 1048576, f.Name())
 		require.Error(t, err)
@@ -682,6 +682,7 @@ func TestBlockUploaderRestore(t *testing.T) {
 		iterMock.On("Count").Return(uint64(1))
 		iterMock.On("Next").Return(uint64(0), true).Once()
 		iterMock.On("Next").Return(uint64(0), false)
+		iterMock.On("BlockSize").Return(uint(1048576))
 
 		written, err := blkup.Restore(snap, dest, iterMock, nil)
 		require.NoError(t, err)

site/content/docs/main/api-types/schedule.md

@@ -155,11 +155,13 @@ spec:
     uploaderConfig:
         # ParallelFilesUpload is the number of files parallel uploads to perform when using the uploader.
         parallelFilesUpload: 10
-    # The labels you want on backup objects, created from this schedule (instead of copying the labels you have on schedule object itself).
-    # When this field is set, the labels from the Schedule resource are not copied to the Backup resource.
+    # The labels/annotations you want on backup objects, created from this schedule (instead of copying the labels/annotations you have on schedule object itself).
+    # When this field is set, the labels/annotations from the Schedule resource are not copied to the Backup resource.
     metadata:
       labels:
         labelname: somelabelvalue
+      annotations:
+        annotationname: someannotationvalue
     # Actions to perform at different times during a backup. The only hook supported is
     # executing a command in a container in a pod using the pod exec API. Optional.
     hooks: