Fix DataUploadDeleteAction creating CMs for foreign DataUploads

When a backup tarball incidentally contains DataUpload CRs that belong to
a different backup (common when a schedule includes the velero namespace
where DataUploads live), DataUploadDeleteAction.Execute used to create a
"<du-name>-info" ConfigMap labeled with the *executing* backup's name
instead of the DataUpload's true owning backup. The ConfigMap is
created with Create-only semantics, so the wrong label is never
corrected.

deleteMovedSnapshots in the backup-deletion controller looks up these
ConfigMaps by velero.io/backup-name to discover which Kopia snapshots
to delete. With the wrong label, the real owning backup's expiry pass
finds no ConfigMaps for its DataUploads and silently leaves their Kopia
snapshots in object storage, leaking data over time.

Fix: in DataUploadDeleteAction.Execute, compare the DataUpload's
velero.io/backup-name label against input.Backup.Name (using
label.GetValidName to handle DNS-1035 truncation for long backup names).
If the label is present and differs, skip the DataUpload entirely; this
prevents the over-eager creation of misnamed ConfigMaps without changing
behavior for DataUploads that legitimately belong to the executing
backup, or for legacy DataUploads with no backup-name label.

Refs: #9472

Signed-off-by: Christian Schlichtherle <cs@bsure-analytics.de>

changelogs/unreleased/9472-christian-schlichtherle

@@ -0,0 +1 @@
+Fix DataUploadDeleteAction creating snapshot-info ConfigMaps labeled with the wrong backup name when a DataUpload CR from another backup is incidentally captured in the backup tarball, which caused Kopia snapshots to be leaked in object storage on expiry of the real owning backup.

pkg/datamover/dataupload_delete_action.go

@@ -14,6 +14,7 @@ import (
 
 	velerov1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
 	velerov2alpha1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
+	"github.com/vmware-tanzu/velero/pkg/label"
 	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
 	repotypes "github.com/vmware-tanzu/velero/pkg/repository/types"
 )
@@ -35,6 +36,19 @@ func (d *DataUploadDeleteAction) Execute(input *velero.DeleteItemActionExecuteIn
 	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(input.Item.UnstructuredContent(), &du); err != nil {
 		return errors.WithStack(errors.Wrapf(err, "failed to convert input.Item from unstructured"))
 	}
+	// Skip DataUploads that do not belong to the backup being deleted. The
+	// backup tarball may incidentally include DataUpload CRs from the velero
+	// namespace that belong to a different backup (e.g. when an hourly
+	// schedule with snapshotMoveData=false captures the velero namespace
+	// containing a daily schedule's DataUploads). Creating a snapshot-info
+	// ConfigMap labeled with the wrong backup name causes the real owning
+	// backup's deleteMovedSnapshots query to miss it, leaking the Kopia
+	// snapshot in the object store.
+	if owner := du.Labels[velerov1.BackupNameLabel]; owner != "" && owner != label.GetValidName(input.Backup.Name) {
+		d.logger.Infof("Skipping DataUpload %s/%s: belongs to backup %q, not %q",
+			du.Namespace, du.Name, owner, input.Backup.Name)
+		return nil
+	}
 	cm := genConfigmap(input.Backup, *du)
 	if cm == nil {
 		// will not fail the backup deletion

pkg/datamover/dataupload_delete_action_test.go

@@ -0,0 +1,144 @@
+/*
+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 datamover
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	corev1api "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+	crclient "sigs.k8s.io/controller-runtime/pkg/client"
+
+	velerov1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	velerov2alpha1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v2alpha1"
+	"github.com/vmware-tanzu/velero/pkg/builder"
+	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
+	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+)
+
+func toUnstructured(t *testing.T, du *velerov2alpha1.DataUpload) runtime.Unstructured {
+	t.Helper()
+	m, err := runtime.DefaultUnstructuredConverter.ToUnstructured(du)
+	require.NoError(t, err)
+	return &unstructured.Unstructured{Object: m}
+}
+
+func newCompletedDataUpload(name, ownerBackup string) *velerov2alpha1.DataUpload {
+	du := &velerov2alpha1.DataUpload{
+		TypeMeta: metav1.TypeMeta{
+			APIVersion: velerov2alpha1.SchemeGroupVersion.String(),
+			Kind:       "DataUpload",
+		},
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: "velero",
+			Name:      name,
+		},
+		Spec: velerov2alpha1.DataUploadSpec{
+			SnapshotType:          velerov2alpha1.SnapshotTypeCSI,
+			SourcePVC:             "my-pvc",
+			SourceNamespace:       "app",
+			BackupStorageLocation: "default",
+			DataMover:             "velero",
+		},
+		Status: velerov2alpha1.DataUploadStatus{
+			Phase:      velerov2alpha1.DataUploadPhaseCompleted,
+			SnapshotID: "kopia-snapshot-id",
+		},
+	}
+	if ownerBackup != "" {
+		du.Labels = map[string]string{velerov1.BackupNameLabel: ownerBackup}
+	}
+	return du
+}
+
+func TestDataUploadDeleteActionAppliesTo(t *testing.T) {
+	a := NewDataUploadDeleteAction(logrus.StandardLogger(), nil)
+	selector, err := a.AppliesTo()
+	require.NoError(t, err)
+	require.Equal(t, velero.ResourceSelector{IncludedResources: []string{"datauploads.velero.io"}}, selector)
+}
+
+func TestDataUploadDeleteActionExecute(t *testing.T) {
+	tests := []struct {
+		name             string
+		duName           string
+		duOwnerBackup    string // value placed in velero.io/backup-name label on the DataUpload
+		executingBackup  string // name of the Backup being deleted (input.Backup.Name)
+		wantConfigMap    bool
+	}{
+		{
+			name:            "DataUpload owned by the executing backup creates a snapshot-info ConfigMap",
+			duName:          "daily-backup-abcde",
+			duOwnerBackup:   "daily-backup",
+			executingBackup: "daily-backup",
+			wantConfigMap:   true,
+		},
+		{
+			name:            "DataUpload owned by a different backup is skipped (no ConfigMap created)",
+			duName:          "daily-backup-abcde",
+			duOwnerBackup:   "daily-backup",
+			executingBackup: "hourly-backup",
+			wantConfigMap:   false,
+		},
+		{
+			name:            "DataUpload with no backup-name label falls through (legacy behavior preserved)",
+			duName:          "legacy-du",
+			duOwnerBackup:   "",
+			executingBackup: "some-backup",
+			wantConfigMap:   true,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			crClient := velerotest.NewFakeControllerRuntimeClient(t)
+			action := NewDataUploadDeleteAction(logrus.StandardLogger(), crClient)
+
+			du := newCompletedDataUpload(tc.duName, tc.duOwnerBackup)
+			backup := builder.ForBackup("velero", tc.executingBackup).StorageLocation("default").Result()
+
+			err := action.Execute(&velero.DeleteItemActionExecuteInput{
+				Item:   toUnstructured(t, du),
+				Backup: backup,
+			})
+			require.NoError(t, err)
+
+			cm := &corev1api.ConfigMap{}
+			getErr := crClient.Get(t.Context(), crclient.ObjectKey{
+				Namespace: backup.Namespace,
+				Name:      fmt.Sprintf("%s-info", du.Name),
+			}, cm)
+
+			if tc.wantConfigMap {
+				require.NoError(t, getErr, "expected snapshot-info ConfigMap to be created")
+				assert.Equal(t, tc.executingBackup, cm.Labels[velerov1.BackupNameLabel])
+				assert.Equal(t, "true", cm.Labels[velerov1.DataUploadSnapshotInfoLabel])
+			} else {
+				require.Error(t, getErr)
+				assert.True(t, apierrors.IsNotFound(getErr),
+					"expected no ConfigMap to be created for foreign DataUpload, but got: %v", getErr)
+			}
+		})
+	}
+}