Merge branch 'main' into pvr-restorer-could-run-concurrently

pkg/podvolume/backupper.go

@@ -20,12 +20,15 @@ import (
 	"context"
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/cockroachdb/errors"
 	"github.com/sirupsen/logrus"
 	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/util/sets"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/tools/cache"
 	ctrlcache "sigs.k8s.io/controller-runtime/pkg/cache"
 	ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
@@ -181,7 +184,7 @@ func newBackupper(
 					// the PVB in the indexer is already in final status, no need to call WaitGroup.Done()
 					if ok && (existPVB.Status.Phase == velerov1api.PodVolumeBackupPhaseCompleted ||
 						existPVB.Status.Phase == velerov1api.PodVolumeBackupPhaseFailed ||
-						pvb.Status.Phase == velerov1api.PodVolumeBackupPhaseCanceled) {
+						existPVB.Status.Phase == velerov1api.PodVolumeBackupPhaseCanceled) {
 						statusChangedToFinal = false
 					}
 				}
@@ -411,24 +414,71 @@ func (b *backupper) WaitAllPodVolumesProcessed(log logrus.FieldLogger) []*velero
 	select {
 	case <-b.ctx.Done():
 		log.Error("timed out waiting for all PodVolumeBackups to complete")
-	case <-done:
+
 		for _, obj := range b.pvbIndexer.List() {
 			pvb, ok := obj.(*velerov1api.PodVolumeBackup)
 			if !ok {
-				log.Errorf("expected PodVolumeBackup, but got %T", obj)
+				log.Errorf("expected PVB, but got %T", obj)
 				continue
 			}
-			podVolumeBackups = append(podVolumeBackups, pvb)
-			if pvb.Status.Phase == velerov1api.PodVolumeBackupPhaseFailed {
-				log.Errorf("pod volume backup failed: %s", pvb.Status.Message)
-			} else if pvb.Status.Phase == velerov1api.PodVolumeBackupPhaseCanceled {
-				log.Errorf("pod volume backup canceled: %s", pvb.Status.Message)
+
+			if pvb.Status.Phase != velerov1api.PodVolumeBackupPhaseCompleted &&
+				pvb.Status.Phase != velerov1api.PodVolumeBackupPhaseFailed &&
+				pvb.Status.Phase != velerov1api.PodVolumeBackupPhaseCanceled {
+				log.Infof("Setting cancel flag for ongoing PVB %s/%s", pvb.Namespace, pvb.Name)
+				if err := updatePVBWithRetry(context.Background(), b.crClient, pvb.Namespace, pvb.Name); err != nil {
+					log.WithError(err).Errorf("Failed to set cancel flag for PVB %s/%s", pvb.Namespace, pvb.Name)
+				}
 			}
 		}
+		<-done
+	case <-done:
+	}
+
+	// Collect tracked PVBs regardless of whether we timed out or completed normally.
+	// On timeout, already-completed PVBs must still be persisted so their data remains restorable.
+	for _, obj := range b.pvbIndexer.List() {
+		pvb, ok := obj.(*velerov1api.PodVolumeBackup)
+		if !ok {
+			log.Errorf("expected PodVolumeBackup, but got %T", obj)
+			continue
+		}
+		podVolumeBackups = append(podVolumeBackups, pvb)
+		if pvb.Status.Phase == velerov1api.PodVolumeBackupPhaseFailed {
+			log.Errorf("pod volume backup failed: %s", pvb.Status.Message)
+		} else if pvb.Status.Phase == velerov1api.PodVolumeBackupPhaseCanceled {
+			log.Errorf("pod volume backup canceled: %s", pvb.Status.Message)
+		}
 	}
 	return podVolumeBackups
 }
 
+func updatePVBWithRetry(ctx context.Context, client ctrlclient.Client, namespace, name string) error {
+	return wait.PollUntilContextCancel(ctx, 100*time.Millisecond, true, func(ctx context.Context) (bool, error) {
+		pvb := &velerov1api.PodVolumeBackup{}
+		if err := client.Get(ctx, ctrlclient.ObjectKey{Namespace: namespace, Name: name}, pvb); err != nil {
+			return false, errors.Wrap(err, "getting PVB")
+		}
+
+		if pvb.Spec.Cancel {
+			return true, nil
+		}
+
+		pvb.Spec.Cancel = true
+		pvb.Status.Message = "Cancel PVB on pod volume timeout"
+
+		err := client.Update(ctx, pvb)
+		if err != nil {
+			if apierrors.IsConflict(err) {
+				return false, nil
+			}
+			return false, errors.Wrapf(err, "error updating PVB %s/%s", pvb.Namespace, pvb.Name)
+		}
+
+		return true, nil
+	})
+}
+
 func (b *backupper) GetPodVolumeBackupByPodAndVolume(podNamespace, podName, volume string) (*velerov1api.PodVolumeBackup, error) {
 	obj, exist, err := b.pvbIndexer.GetByKey(fmt.Sprintf(pvbKeyPattern, podNamespace, podName, volume))
 	if err != nil {

pkg/podvolume/backupper_test.go

@@ -380,6 +380,8 @@ func TestBackupPodVolumes(t *testing.T) {
 		pvbs                  int
 		mockGetRepositoryType bool
 		errs                  []string
+		expectedBackedup      []string
+		expectedSkipped       map[string]string
 	}{
 		{
 			name: "empty volume list",
@@ -573,6 +575,10 @@ func TestBackupPodVolumes(t *testing.T) {
 			uploaderType:  "kopia",
 			bsl:           "fake-bsl",
 			errs:          []string{},
+			expectedSkipped: map[string]string{
+				"fake-volume-1": "volume fake-volume-1 is declared in pod fake-ns/fake-pod but not mounted by any container, skipping",
+				"fake-volume-2": "volume fake-volume-2 is declared in pod fake-ns/fake-pod but not mounted by any container, skipping",
+			},
 		},
 		{
 			name: "return completed pvbs",
@@ -589,14 +595,14 @@ func TestBackupPodVolumes(t *testing.T) {
 			ctlClientObj: []runtime.Object{
 				createBackupRepoObj(),
 			},
-			runtimeScheme: scheme,
-			uploaderType:  "kopia",
-			bsl:           "fake-bsl",
-			pvbs:          1,
-			errs:          []string{},
+			runtimeScheme:    scheme,
+			uploaderType:     "kopia",
+			bsl:              "fake-bsl",
+			pvbs:             1,
+			errs:             []string{},
+			expectedBackedup: []string{"fake-volume-1"},
 		},
 	}
-	// TODO add more verification around PVCBackupSummary returned by "BackupPodVolumes"
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			ctx := t.Context()
@@ -627,7 +633,7 @@ func TestBackupPodVolumes(t *testing.T) {
 				funcGetRepositoryType = getRepositoryType
 			}
 
-			pvbs, _, errs := bp.BackupPodVolumes(backupObj, test.sourcePod, test.volumes, nil, velerotest.NewLogger())
+			pvbs, summary, errs := bp.BackupPodVolumes(backupObj, test.sourcePod, test.volumes, nil, velerotest.NewLogger())
 
 			if test.errs != nil {
 				for i := 0; i < len(errs); i++ {
@@ -636,6 +642,22 @@ func TestBackupPodVolumes(t *testing.T) {
 			}
 
 			assert.Len(t, pvbs, test.pvbs)
+
+			if summary != nil {
+				assert.Len(t, summary.Backedup, len(test.expectedBackedup))
+				for _, vol := range test.expectedBackedup {
+					assert.Contains(t, summary.Backedup, vol)
+				}
+
+				assert.Len(t, summary.Skipped, len(test.expectedSkipped))
+				for vol, reason := range test.expectedSkipped {
+					require.Contains(t, summary.Skipped, vol)
+					assert.Equal(t, reason, summary.Skipped[vol].Reason)
+				}
+			} else {
+				assert.Empty(t, test.expectedBackedup)
+				assert.Empty(t, test.expectedSkipped)
+			}
 		})
 	}
 }
@@ -733,14 +755,14 @@ func TestListPodVolumeBackupsByPodp(t *testing.T) {
 }
 
 type logHook struct {
-	entry *logrus.Entry
+	entries []*logrus.Entry
 }
 
 func (l *logHook) Levels() []logrus.Level {
 	return []logrus.Level{logrus.ErrorLevel}
 }
 func (l *logHook) Fire(entry *logrus.Entry) error {
-	l.entry = entry
+	l.entries = append(l.entries, entry)
 	return nil
 }
 
@@ -757,16 +779,18 @@ func TestWaitAllPodVolumesProcessed(t *testing.T) {
 		statusToBeUpdated *velerov1api.PodVolumeBackupStatus
 		expectedErr       string
 		expectedPVBPhase  velerov1api.PodVolumeBackupPhase
+		expectedPVBCount  int
 	}{
 		{
 			name: "contains no pvb should report no error",
 			ctx:  timeoutCtx,
 		},
 		{
-			name:        "context canceled",
-			ctx:         timeoutCtx,
-			pvb:         pvb,
-			expectedErr: "timed out waiting for all PodVolumeBackups to complete",
+			name:             "context canceled should still return tracked pvbs",
+			ctx:              timeoutCtx,
+			pvb:              pvb,
+			expectedErr:      "timed out waiting for all PodVolumeBackups to complete",
+			expectedPVBCount: 1,
 		},
 		{
 			name: "failed pvbs",
@@ -806,12 +830,35 @@ func TestWaitAllPodVolumesProcessed(t *testing.T) {
 		logHook := &logHook{}
 		logger.Hooks.Add(logHook)
 
-		backuper := newBackupper(c.ctx, log, nil, nil, informer, nil, "", &velerov1api.Backup{})
+		backuper := newBackupper(c.ctx, log, nil, nil, informer, client, "", &velerov1api.Backup{})
 		if c.pvb != nil {
 			require.NoError(t, backuper.pvbIndexer.Add(c.pvb))
 			backuper.wg.Add(1)
 		}
 
+		if c.ctx == timeoutCtx && c.pvb != nil {
+			// Start a goroutine to simulate the controller's cancellation behavior
+			go func() {
+				// Wait a short time for the cancel flag to be set
+				ticker := time.NewTicker(10 * time.Millisecond)
+				defer ticker.Stop()
+				for range ticker.C {
+					pvb := &velerov1api.PodVolumeBackup{}
+					err := client.Get(t.Context(), ctrlclient.ObjectKey{Namespace: c.pvb.Namespace, Name: c.pvb.Name}, pvb)
+					if err == nil && pvb.Spec.Cancel {
+						oldPVB := pvb.DeepCopy()
+						pvb.Status.Phase = velerov1api.PodVolumeBackupPhaseCanceled
+						pvb.Status.Message = "canceled"
+						_ = client.Update(t.Context(), pvb)
+						if informer.handler != nil {
+							informer.handler.OnUpdate(oldPVB, pvb)
+						}
+						return
+					}
+				}
+			}()
+		}
+
 		if c.statusToBeUpdated != nil {
 			pvb := &velerov1api.PodVolumeBackup{}
 			err := client.Get(t.Context(), ctrlclient.ObjectKey{Namespace: c.pvb.Namespace, Name: c.pvb.Name}, pvb)
@@ -829,9 +876,22 @@ func TestWaitAllPodVolumesProcessed(t *testing.T) {
 		pvbs := backuper.WaitAllPodVolumesProcessed(logger)
 
 		if c.expectedErr != "" {
-			assert.Equal(t, c.expectedErr, logHook.entry.Message)
+			found := false
+			var loggedMsgs []string
+			for _, entry := range logHook.entries {
+				loggedMsgs = append(loggedMsgs, entry.Message)
+				if entry.Message == c.expectedErr {
+					found = true
+					break
+				}
+			}
+			assert.True(t, found, "Expected error %q to be logged, but got %v", c.expectedErr, loggedMsgs)
 		} else {
-			assert.Nil(t, logHook.entry)
+			assert.Empty(t, logHook.entries)
+		}
+
+		if c.expectedPVBCount > 0 {
+			require.Len(t, pvbs, c.expectedPVBCount)
 		}
 
 		if c.expectedPVBPhase != "" {

pkg/podvolume/restore_micro_service.go

@@ -184,7 +184,7 @@ func (r *RestoreMicroService) RunCancelableDataPath(ctx context.Context) (string
 
 	log.Info("Async fs br init")
 
-	if err := fsRestore.StartRestore(pvr.Spec.SnapshotID, r.sourceTargetPath, pvr.Spec.UploaderSettings); err != nil {
+	if err := fsRestore.StartRestore(pvr.Spec.SnapshotID, r.sourceTargetPath, pvr.Spec.UploaderSettings, &datapath.RestoreStartParam{}); err != nil {
 		return "", errors.Wrap(err, "error starting data path restore")
 	}
 

pkg/podvolume/restore_micro_service_test.go

@@ -436,12 +436,12 @@ func TestRunCancelableDataPathRestore(t *testing.T) {
 
 				if test.startErr != nil {
 					fsBR.On("Init", mock.Anything, mock.Anything).Return(nil)
-					fsBR.On("StartRestore", mock.Anything, mock.Anything, mock.Anything).Return(test.startErr)
+					fsBR.On("StartRestore", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(test.startErr)
 				}
 
 				if test.dataPathStarted {
 					fsBR.On("Init", mock.Anything, mock.Anything).Return(nil)
-					fsBR.On("StartRestore", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+					fsBR.On("StartRestore", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
 				}
 
 				return fsBR