diff --git a/changelogs/unreleased/10007-Lyndon-Li b/changelogs/unreleased/10007-Lyndon-Li new file mode 100644 index 000000000..808c5920b --- /dev/null +++ b/changelogs/unreleased/10007-Lyndon-Li @@ -0,0 +1 @@ +Fix issue #9997, cancel ongoing PVB on timeout and wait for all PVBs to terminal state \ No newline at end of file diff --git a/pkg/podvolume/backupper.go b/pkg/podvolume/backupper.go index 5864a2090..261f227f8 100644 --- a/pkg/podvolume/backupper.go +++ b/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 { diff --git a/pkg/podvolume/backupper_test.go b/pkg/podvolume/backupper_test.go index 66ad9e5ae..59466e02a 100644 --- a/pkg/podvolume/backupper_test.go +++ b/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 {