Merge pull request #10007 from Lyndon-Li/cancel-pvb-on-timeout

Issue 9997: cancel ongoing PVB on timeout

changelogs/unreleased/10007-Lyndon-Li

@@ -0,0 +1 @@
+Fix issue #9997, cancel ongoing PVB on timeout and wait for all PVBs to terminal state

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,6 +414,24 @@ func (b *backupper) WaitAllPodVolumesProcessed(log logrus.FieldLogger) []*velero
 	select {
 	case <-b.ctx.Done():
 		log.Error("timed out waiting for all PodVolumeBackups to complete")
+
+		for _, obj := range b.pvbIndexer.List() {
+			pvb, ok := obj.(*velerov1api.PodVolumeBackup)
+			if !ok {
+				log.Errorf("expected PVB, but got %T", obj)
+				continue
+			}
+
+			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:
 	}
 
@@ -432,6 +453,32 @@ func (b *backupper) WaitAllPodVolumesProcessed(log logrus.FieldLogger) []*velero
 	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

@@ -733,14 +733,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
 }
 
@@ -808,12 +808,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)
@@ -831,9 +854,18 @@ 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 {