Merge branch 'release-1.12' into finalizer-optimize

pkg/backup/item_collector.go

@@ -196,9 +196,8 @@ func (r *itemCollector) getResourceItems(log logrus.FieldLogger, gv schema.Group
 	log.Info("Getting items for resource")
 
 	var (
-		gvr           = gv.WithResource(resource.Name)
-		gr            = gvr.GroupResource()
-		clusterScoped = !resource.Namespaced
+		gvr = gv.WithResource(resource.Name)
+		gr  = gvr.GroupResource()
 	)
 
 	orders := getOrderedResourcesForType(r.backupRequest.Backup.Spec.OrderedResources, resource.Name)
@@ -272,8 +271,6 @@ func (r *itemCollector) getResourceItems(log logrus.FieldLogger, gv schema.Group
 		}
 	}
 
-	namespacesToList := getNamespacesToList(r.backupRequest.NamespaceIncludesExcludes)
-
 	// Handle namespace resource here.
 	// Namespace are only filtered by namespace include/exclude filters.
 	// Label selectors are not checked.
@@ -289,10 +286,12 @@ func (r *itemCollector) getResourceItems(log logrus.FieldLogger, gv schema.Group
 			return nil, errors.WithStack(err)
 		}
 
-		items := r.backupNamespaces(unstructuredList, namespacesToList, gr, preferredGVR, log)
+		items := r.backupNamespaces(unstructuredList, r.backupRequest.NamespaceIncludesExcludes, gr, preferredGVR, log)
 
 		return items, nil
 	}
+	clusterScoped := !resource.Namespaced
+	namespacesToList := getNamespacesToList(r.backupRequest.NamespaceIncludesExcludes)
 
 	// If we get here, we're backing up something other than namespaces
 	if clusterScoped {
@@ -533,31 +532,13 @@ func (r *itemCollector) listItemsForLabel(unstructuredItems []unstructured.Unstr
 
 // backupNamespaces process namespace resource according to namespace filters.
 func (r *itemCollector) backupNamespaces(unstructuredList *unstructured.UnstructuredList,
-	namespacesToList []string, gr schema.GroupResource, preferredGVR schema.GroupVersionResource,
+	ie *collections.IncludesExcludes, gr schema.GroupResource, preferredGVR schema.GroupVersionResource,
 	log logrus.FieldLogger) []*kubernetesResource {
 	var items []*kubernetesResource
 	for index, unstructured := range unstructuredList.Items {
-		found := false
-		if len(namespacesToList) == 0 {
-			// No namespace found. By far, this condition cannot be triggered. Either way,
-			// namespacesToList is not empty.
-			log.Debug("Skip namespace resource, because no item found by namespace filters.")
-			break
-		} else if len(namespacesToList) == 1 && namespacesToList[0] == "" {
-			// All namespaces are included.
-			log.Debugf("Backup namespace %s due to full cluster backup.", unstructured.GetName())
-			found = true
-		} else {
-			for _, ns := range namespacesToList {
-				if unstructured.GetName() == ns {
-					log.Debugf("Backup namespace %s due to namespace filters setting.", unstructured.GetName())
-					found = true
-					break
-				}
-			}
-		}
+		if ie.ShouldInclude(unstructured.GetName()) {
+			log.Debugf("Backup namespace %s due to namespace filters setting.", unstructured.GetName())
 
-		if found {
 			path, err := r.writeToFile(&unstructuredList.Items[index])
 			if err != nil {
 				log.WithError(err).Error("Error writing item to file")

pkg/controller/restore_controller.go

@@ -671,14 +671,13 @@ func (r *restoreReconciler) deleteExternalResources(restore *api.Restore) error
 
 	backupInfo, err := r.fetchBackupInfo(restore.Spec.BackupName)
 	if err != nil {
+		if apierrors.IsNotFound(err) {
+			r.logger.Errorf("got not found error: %v, skip deleting the restore files in object storage", err)
+			return nil
+		}
 		return errors.Wrap(err, fmt.Sprintf("can't get backup info, backup: %s", restore.Spec.BackupName))
 	}
 
-	// if storage locations is read-only, skip deletion
-	if backupInfo.location.Spec.AccessMode == api.BackupStorageLocationAccessModeReadOnly {
-		return nil
-	}
-
 	// delete restore files in object storage
 	pluginManager := r.newPluginManager(r.logger)
 	defer pluginManager.CleanupClients()

pkg/controller/restore_operations_controller.go

@@ -139,18 +139,6 @@ func (r *restoreOperationsReconciler) Reconcile(ctx context.Context, req ctrl.Re
 		return ctrl.Result{}, errors.Wrap(err, "error getting backup info")
 	}
 
-	if info.location.Spec.AccessMode == velerov1api.BackupStorageLocationAccessModeReadOnly {
-		log.Infof("Cannot check progress on Restore operations because backup storage location %s is currently in read-only mode; marking restore PartiallyFailed", info.location.Name)
-		restore.Status.Phase = velerov1api.RestorePhasePartiallyFailed
-		restore.Status.CompletionTimestamp = &metav1.Time{Time: r.clock.Now()}
-		r.metrics.RegisterRestorePartialFailure(restore.Spec.ScheduleName)
-		err := r.updateRestoreAndOperationsJSON(ctx, original, restore, nil, &itemoperationmap.OperationsForRestore{ErrsSinceUpdate: []string{"BSL is read-only"}}, false, false)
-		if err != nil {
-			log.WithError(err).Error("error updating Restore")
-		}
-		return ctrl.Result{}, nil
-	}
-
 	pluginManager := r.newPluginManager(r.logger)
 	defer pluginManager.CleanupClients()
 	backupStore, err := r.backupStoreGetter.Get(info.location, pluginManager, r.logger)

pkg/exposer/csi_snapshot.go

@@ -246,8 +246,9 @@ func (e *csiSnapshotExposer) createBackupVS(ctx context.Context, ownerObject cor
 
 	vs := &snapshotv1api.VolumeSnapshot{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      backupVSName,
-			Namespace: ownerObject.Namespace,
+			Name:        backupVSName,
+			Namespace:   ownerObject.Namespace,
+			Annotations: snapshotVS.Annotations,
 			// Don't add ownerReference to SnapshotBackup.
 			// The backupPVC should be deleted before backupVS, otherwise, the deletion of backupVS will fail since
 			// backupPVC has its dataSource referring to it
@@ -268,7 +269,8 @@ func (e *csiSnapshotExposer) createBackupVSC(ctx context.Context, ownerObject co
 
 	vsc := &snapshotv1api.VolumeSnapshotContent{
 		ObjectMeta: metav1.ObjectMeta{
-			Name: backupVSCName,
+			Name:        backupVSCName,
+			Annotations: snapshotVSC.Annotations,
 		},
 		Spec: snapshotv1api.VolumeSnapshotContentSpec{
 			VolumeSnapshotRef: corev1.ObjectReference{

pkg/exposer/csi_snapshot_test.go

@@ -58,10 +58,22 @@ func TestExpose(t *testing.T) {
 			UID:       "fake-uid",
 		},
 	}
+
+	snapshotClass := "fake-snapshot-class"
 	vsObject := &snapshotv1api.VolumeSnapshot{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "fake-vs",
 			Namespace: "fake-ns",
+			Annotations: map[string]string{
+				"fake-key-1": "fake-value-1",
+				"fake-key-2": "fake-value-2",
+			},
+		},
+		Spec: snapshotv1api.VolumeSnapshotSpec{
+			Source: snapshotv1api.VolumeSnapshotSource{
+				VolumeSnapshotContentName: &vscName,
+			},
+			VolumeSnapshotClassName: &snapshotClass,
 		},
 		Status: &snapshotv1api.VolumeSnapshotStatus{
 			BoundVolumeSnapshotContentName: &vscName,
@@ -71,15 +83,23 @@ func TestExpose(t *testing.T) {
 	}
 
 	var restoreSize int64
+	snapshotHandle := "fake-handle"
 	vscObj := &snapshotv1api.VolumeSnapshotContent{
 		ObjectMeta: metav1.ObjectMeta{
-			Name: "fake-vsc",
+			Name: vscName,
+			Annotations: map[string]string{
+				"fake-key-3": "fake-value-3",
+				"fake-key-4": "fake-value-4",
+			},
 		},
 		Spec: snapshotv1api.VolumeSnapshotContentSpec{
-			DeletionPolicy: snapshotv1api.VolumeSnapshotContentDelete,
+			DeletionPolicy:          snapshotv1api.VolumeSnapshotContentDelete,
+			Driver:                  "fake-driver",
+			VolumeSnapshotClassName: &snapshotClass,
 		},
 		Status: &snapshotv1api.VolumeSnapshotContentStatus{
-			RestoreSize: &restoreSize,
+			RestoreSize:    &restoreSize,
+			SnapshotHandle: &snapshotHandle,
 		},
 	}
 
@@ -284,6 +304,23 @@ func TestExpose(t *testing.T) {
 			},
 			err: "error to create backup pod: fake-create-error",
 		},
+		{
+			name:        "success",
+			ownerBackup: backup,
+			exposeParam: CSISnapshotExposeParam{
+				SnapshotName:    "fake-vs",
+				SourceNamespace: "fake-ns",
+				AccessMode:      AccessModeFileSystem,
+				Timeout:         time.Millisecond,
+			},
+			snapshotClientObj: []runtime.Object{
+				vsObject,
+				vscObj,
+			},
+			kubeClientObj: []runtime.Object{
+				daemonSet,
+			},
+		},
 	}
 
 	for _, test := range tests {
@@ -317,7 +354,33 @@ func TestExpose(t *testing.T) {
 			}
 
 			err := exposer.Expose(context.Background(), ownerObject, &test.exposeParam)
-			assert.EqualError(t, err, test.err)
+			if err == nil {
+				assert.NoError(t, err)
+
+				_, err = exposer.kubeClient.CoreV1().Pods(ownerObject.Namespace).Get(context.Background(), ownerObject.Name, metav1.GetOptions{})
+				assert.NoError(t, err)
+
+				_, err = exposer.kubeClient.CoreV1().PersistentVolumeClaims(ownerObject.Namespace).Get(context.Background(), ownerObject.Name, metav1.GetOptions{})
+				assert.NoError(t, err)
+
+				expectedVS, err := exposer.csiSnapshotClient.VolumeSnapshots(ownerObject.Namespace).Get(context.Background(), ownerObject.Name, metav1.GetOptions{})
+				assert.NoError(t, err)
+
+				expectedVSC, err := exposer.csiSnapshotClient.VolumeSnapshotContents().Get(context.Background(), ownerObject.Name, metav1.GetOptions{})
+				assert.NoError(t, err)
+
+				assert.Equal(t, expectedVS.Annotations, vsObject.Annotations)
+				assert.Equal(t, *expectedVS.Spec.VolumeSnapshotClassName, *vsObject.Spec.VolumeSnapshotClassName)
+				assert.Equal(t, *expectedVS.Spec.Source.VolumeSnapshotContentName, expectedVSC.Name)
+
+				assert.Equal(t, expectedVSC.Annotations, vscObj.Annotations)
+				assert.Equal(t, expectedVSC.Spec.DeletionPolicy, vscObj.Spec.DeletionPolicy)
+				assert.Equal(t, expectedVSC.Spec.Driver, vscObj.Spec.Driver)
+				assert.Equal(t, *expectedVSC.Spec.VolumeSnapshotClassName, *vscObj.Spec.VolumeSnapshotClassName)
+			} else {
+				assert.EqualError(t, err, test.err)
+			}
+
 		})
 	}
 }

pkg/install/resources.go

@@ -32,7 +32,11 @@ import (
 	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 )
 
-const defaultServiceAccountName = "velero"
+const (
+	defaultServiceAccountName = "velero"
+	podSecurityLevel          = "privileged"
+	podSecurityVersion        = "latest"
+)
 
 var (
 	DefaultVeleroPodCPURequest    = "500m"
@@ -148,8 +152,12 @@ func Namespace(namespace string) *corev1.Namespace {
 		},
 	}
 
-	ns.Labels["pod-security.kubernetes.io/enforce"] = "privileged"
-	ns.Labels["pod-security.kubernetes.io/enforce-version"] = "latest"
+	ns.Labels["pod-security.kubernetes.io/enforce"] = podSecurityLevel
+	ns.Labels["pod-security.kubernetes.io/enforce-version"] = podSecurityVersion
+	ns.Labels["pod-security.kubernetes.io/audit"] = podSecurityLevel
+	ns.Labels["pod-security.kubernetes.io/audit-version"] = podSecurityVersion
+	ns.Labels["pod-security.kubernetes.io/warn"] = podSecurityLevel
+	ns.Labels["pod-security.kubernetes.io/warn-version"] = podSecurityVersion
 
 	return ns
 }

pkg/install/resources_test.go

@@ -47,6 +47,10 @@ func TestResources(t *testing.T) {
 	// PSA(Pod Security Admission) and PSS(Pod Security Standards).
 	assert.Equal(t, ns.Labels["pod-security.kubernetes.io/enforce"], "privileged")
 	assert.Equal(t, ns.Labels["pod-security.kubernetes.io/enforce-version"], "latest")
+	assert.Equal(t, ns.Labels["pod-security.kubernetes.io/audit"], "privileged")
+	assert.Equal(t, ns.Labels["pod-security.kubernetes.io/audit-version"], "latest")
+	assert.Equal(t, ns.Labels["pod-security.kubernetes.io/warn"], "privileged")
+	assert.Equal(t, ns.Labels["pod-security.kubernetes.io/warn-version"], "latest")
 
 	crb := ClusterRoleBinding(DefaultVeleroNamespace)
 	// The CRB is a cluster-scoped resource

pkg/uploader/kopia/shim.go

@@ -210,7 +210,25 @@ func (sr *shimRepository) DeleteManifest(ctx context.Context, id manifest.ID) er
 }
 
 func (sr *shimRepository) ReplaceManifests(ctx context.Context, labels map[string]string, payload interface{}) (manifest.ID, error) {
-	return manifest.ID(""), errors.New("ReplaceManifests is not supported")
+	const minReplaceManifestTimeDelta = 100 * time.Millisecond
+
+	md, err := sr.FindManifests(ctx, labels)
+	if err != nil {
+		return "", errors.Wrap(err, "unable to load manifests")
+	}
+
+	for _, em := range md {
+		age := sr.Time().Sub(em.ModTime)
+		if age < minReplaceManifestTimeDelta {
+			time.Sleep(minReplaceManifestTimeDelta)
+		}
+
+		if err := sr.DeleteManifest(ctx, em.ID); err != nil {
+			return "", errors.Wrapf(err, "unable to delete previous manifest %v", em.ID)
+		}
+	}
+
+	return sr.PutManifest(ctx, labels, payload)
 }
 
 // Flush all the unifited repository data

pkg/uploader/kopia/shim_test.go

@@ -49,7 +49,6 @@ func TestShimRepo(t *testing.T) {
 	shim.PrefetchObjects(ctx, []object.ID{}, "hint")
 	shim.UpdateDescription("desc")
 	shim.NewWriter(ctx, repo.WriteSessionOptions{})
-	shim.ReplaceManifests(ctx, map[string]string{}, nil)
 	shim.OnSuccessfulFlush(func(ctx context.Context, w repo.RepositoryWriter) error { return nil })
 
 	backupRepo.On("Close", mock.Anything).Return(nil)
@@ -202,3 +201,92 @@ func TestShimObjWriter(t *testing.T) {
 	objWriter.On("Close").Return(nil)
 	writer.Close()
 }
+
+func TestReplaceManifests(t *testing.T) {
+	meta1 := udmrepo.ManifestEntryMetadata{
+		ID: "mani-1",
+	}
+
+	meta2 := udmrepo.ManifestEntryMetadata{
+		ID: "mani-2",
+	}
+
+	tests := []struct {
+		name               string
+		backupRepo         *mocks.BackupRepo
+		isGetManifestError bool
+		expectedError      string
+		expectedID         manifest.ID
+	}{
+		{
+			name:               "Failed to find manifest",
+			isGetManifestError: true,
+			backupRepo: func() *mocks.BackupRepo {
+				backupRepo := &mocks.BackupRepo{}
+				backupRepo.On("FindManifests", mock.Anything, mock.Anything).Return([]*udmrepo.ManifestEntryMetadata{},
+					errors.New("fake-find-error"))
+				return backupRepo
+			}(),
+			expectedError: "unable to load manifests: failed to get manifests with labels map[]: fake-find-error",
+		},
+		{
+			name:               "Failed to delete manifest",
+			isGetManifestError: true,
+			backupRepo: func() *mocks.BackupRepo {
+				backupRepo := &mocks.BackupRepo{}
+				backupRepo.On("FindManifests", mock.Anything, mock.Anything).Return([]*udmrepo.ManifestEntryMetadata{
+					&meta1,
+					&meta2,
+				}, nil)
+				backupRepo.On("Time").Return(time.Now())
+				backupRepo.On("DeleteManifest", mock.Anything, mock.Anything).Return(errors.New("fake-delete-error"))
+				return backupRepo
+			}(),
+			expectedError: "unable to delete previous manifest mani-1: fake-delete-error",
+		},
+		{
+			name: "Failed to put manifest",
+			backupRepo: func() *mocks.BackupRepo {
+				backupRepo := &mocks.BackupRepo{}
+				backupRepo.On("FindManifests", mock.Anything, mock.Anything).Return([]*udmrepo.ManifestEntryMetadata{
+					&meta1,
+					&meta2,
+				}, nil)
+				backupRepo.On("Time").Return(time.Now())
+				backupRepo.On("DeleteManifest", mock.Anything, mock.Anything).Return(nil)
+				backupRepo.On("PutManifest", mock.Anything, mock.Anything).Return(udmrepo.ID(""), errors.New("fake-put-error"))
+				return backupRepo
+			}(),
+			expectedError: "fake-put-error",
+		},
+		{
+			name: "Success",
+			backupRepo: func() *mocks.BackupRepo {
+				backupRepo := &mocks.BackupRepo{}
+				backupRepo.On("FindManifests", mock.Anything, mock.Anything).Return([]*udmrepo.ManifestEntryMetadata{
+					&meta1,
+					&meta2,
+				}, nil)
+				backupRepo.On("Time").Return(time.Now())
+				backupRepo.On("DeleteManifest", mock.Anything, mock.Anything).Return(nil)
+				backupRepo.On("PutManifest", mock.Anything, mock.Anything).Return(udmrepo.ID("fake-id"), nil)
+				return backupRepo
+			}(),
+			expectedID: manifest.ID("fake-id"),
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := context.Background()
+			id, err := NewShimRepo(tc.backupRepo).ReplaceManifests(ctx, map[string]string{}, nil)
+
+			if tc.expectedError != "" {
+				assert.EqualError(t, err, tc.expectedError)
+			} else {
+				assert.NoError(t, err)
+			}
+
+			assert.Equal(t, tc.expectedID, id)
+		})
+	}
+}

pkg/uploader/kopia/snapshot.go

@@ -44,7 +44,8 @@ import (
 )
 
 // All function mainly used to make testing more convenient
-var applyRetentionPolicyFunc = policy.ApplyRetentionPolicy
+var treeForSourceFunc = policy.TreeForSource
+var setPolicyFunc = policy.SetPolicy
 var saveSnapshotFunc = snapshot.SaveSnapshot
 var loadSnapshotFunc = snapshot.LoadSnapshot
 var listSnapshotsFunc = snapshot.ListSnapshots
@@ -72,18 +73,46 @@ func newOptionalBool(b bool) *policy.OptionalBool {
 	return &ob
 }
 
-// setupDefaultPolicy set default policy for kopia
-func setupDefaultPolicy() *policy.Tree {
-	defaultPolicy := *policy.DefaultPolicy
+func getDefaultPolicy() *policy.Policy {
+	return &policy.Policy{
+		RetentionPolicy: policy.RetentionPolicy{
+			KeepLatest:  newOptionalInt(math.MaxInt32),
+			KeepAnnual:  newOptionalInt(math.MaxInt32),
+			KeepDaily:   newOptionalInt(math.MaxInt32),
+			KeepHourly:  newOptionalInt(math.MaxInt32),
+			KeepMonthly: newOptionalInt(math.MaxInt32),
+			KeepWeekly:  newOptionalInt(math.MaxInt32),
+		},
+		CompressionPolicy: policy.CompressionPolicy{
+			CompressorName: "none",
+		},
+		UploadPolicy: policy.UploadPolicy{
+			MaxParallelFileReads:    newOptionalInt(runtime.NumCPU()),
+			ParallelUploadAboveSize: nil,
+		},
+		SchedulingPolicy: policy.SchedulingPolicy{
+			Manual: true,
+		},
+		ErrorHandlingPolicy: policy.ErrorHandlingPolicy{
+			IgnoreUnknownTypes: newOptionalBool(true),
+		},
+	}
+}
 
-	defaultPolicy.RetentionPolicy.KeepLatest = newOptionalInt(math.MaxInt32)
-	defaultPolicy.CompressionPolicy.CompressorName = "none"
-	defaultPolicy.UploadPolicy.MaxParallelFileReads = newOptionalInt(runtime.NumCPU())
-	defaultPolicy.UploadPolicy.ParallelUploadAboveSize = nil
-	defaultPolicy.SchedulingPolicy.Manual = true
-	defaultPolicy.ErrorHandlingPolicy.IgnoreUnknownTypes = newOptionalBool(true)
+func setupDefaultPolicy(ctx context.Context, rep repo.RepositoryWriter, sourceInfo snapshot.SourceInfo) (*policy.Tree, error) {
+	// some internal operations from Kopia code retrieves policies from repo directly, so we need to persist the policy to repo
+	err := setPolicyFunc(ctx, rep, sourceInfo, getDefaultPolicy())
+	if err != nil {
+		return nil, errors.Wrap(err, "error to set policy")
+	}
 
-	return policy.BuildTree(nil, &defaultPolicy)
+	// retrieve policy from repo
+	policyTree, err := treeForSourceFunc(ctx, rep, sourceInfo)
+	if err != nil {
+		return nil, errors.Wrap(err, "error to retrieve policy")
+	}
+
+	return policyTree, nil
 }
 
 // Backup backup specific sourcePath and update progress
@@ -117,7 +146,7 @@ func Backup(ctx context.Context, fsUploader SnapshotUploader, repoWriter repo.Re
 		Host:     udmrepo.GetRepoDomain(),
 		Path:     filepath.Clean(realSource),
 	}
-	if sourceInfo.Path == "" {
+	if realSource == "" {
 		sourceInfo.Path = dir
 	}
 
@@ -213,7 +242,10 @@ func SnapshotSource(
 		log.Infof("Using parent snapshot %s, start time %v, end time %v, description %s", previous[i].ID, previous[i].StartTime.ToTime(), previous[i].EndTime.ToTime(), previous[i].Description)
 	}
 
-	policyTree := setupDefaultPolicy()
+	policyTree, err := setupDefaultPolicy(ctx, rep, sourceInfo)
+	if err != nil {
+		return "", 0, errors.Wrapf(err, "unable to set policy for si %v", sourceInfo)
+	}
 
 	manifest, err := u.Upload(ctx, rootDir, policyTree, sourceInfo, previous...)
 	if err != nil {
@@ -223,14 +255,12 @@ func SnapshotSource(
 	manifest.Tags = snapshotTags
 
 	manifest.Description = description
+	manifest.Pins = []string{"velero-pin"}
 
 	if _, err = saveSnapshotFunc(ctx, rep, manifest); err != nil {
 		return "", 0, errors.Wrapf(err, "Failed to save kopia manifest %v", manifest.ID)
 	}
-	_, err = applyRetentionPolicyFunc(ctx, rep, sourceInfo, true)
-	if err != nil {
-		return "", 0, errors.Wrapf(err, "Failed to apply kopia retention policy for si %v", sourceInfo)
-	}
+
 	if err = rep.Flush(ctx); err != nil {
 		return "", 0, errors.Wrapf(err, "Failed to flush kopia repository")
 	}

pkg/uploader/kopia/snapshot_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/kopia/kopia/repo"
 	"github.com/kopia/kopia/repo/manifest"
 	"github.com/kopia/kopia/snapshot"
+	"github.com/kopia/kopia/snapshot/policy"
 	"github.com/kopia/kopia/snapshot/restore"
 	"github.com/kopia/kopia/snapshot/snapshotfs"
 	"github.com/pkg/errors"
@@ -58,7 +59,8 @@ func injectSnapshotFuncs() *snapshotMockes {
 		repoWriterMock: &repomocks.RepositoryWriter{},
 	}
 
-	applyRetentionPolicyFunc = s.policyMock.ApplyRetentionPolicy
+	setPolicyFunc = s.policyMock.SetPolicy
+	treeForSourceFunc = s.policyMock.TreeForSource
 	loadSnapshotFunc = s.snapshotMock.LoadSnapshot
 	saveSnapshotFunc = s.snapshotMock.SaveSnapshot
 	return s
@@ -135,13 +137,13 @@ func TestSnapshotSource(t *testing.T) {
 			notError: false,
 		},
 		{
-			name: "failed to apply policy",
+			name: "failed to set policy",
 			args: []mockArgs{
 				{methodName: "LoadSnapshot", returns: []interface{}{manifest, nil}},
 				{methodName: "SaveSnapshot", returns: []interface{}{manifest.ID, nil}},
 				{methodName: "TreeForSource", returns: []interface{}{nil, nil}},
-				{methodName: "ApplyRetentionPolicy", returns: []interface{}{nil, errors.New("failed to save snapshot")}},
-				{methodName: "SetPolicy", returns: []interface{}{nil}},
+				{methodName: "ApplyRetentionPolicy", returns: []interface{}{nil, nil}},
+				{methodName: "SetPolicy", returns: []interface{}{errors.New("failed to set policy")}},
 				{methodName: "Upload", returns: []interface{}{manifest, nil}},
 				{methodName: "Flush", returns: []interface{}{nil}},
 			},
@@ -232,7 +234,7 @@ func TestReportSnapshotStatus(t *testing.T) {
 			},
 		}
 
-		result, size, err := reportSnapshotStatus(manifest, setupDefaultPolicy())
+		result, size, err := reportSnapshotStatus(manifest, policy.BuildTree(nil, getDefaultPolicy()))
 
 		switch {
 		case tc.shouldError && err == nil:

pkg/uploader/provider/kopia.go

@@ -159,6 +159,10 @@ func (kp *kopiaProvider) RunBackup(
 	tags[uploader.SnapshotRequesterTag] = kp.requestorType
 	tags[uploader.SnapshotUploaderTag] = uploader.KopiaType
 
+	if realSource != "" {
+		realSource = fmt.Sprintf("%s/%s/%s", kp.requestorType, uploader.KopiaType, realSource)
+	}
+
 	snapshotInfo, isSnapshotEmpty, err := BackupFunc(ctx, kpUploader, repoWriter, path, realSource, forceFull, parentSnapshot, volMode, tags, log)
 	if err != nil {
 		if kpUploader.IsCanceled() {