mirrors/velero
1
0
Fork 0
mirror of https://github.com/vmware-tanzu/velero.git synced 2026-04-20 15:50:33 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

issue 7036: fail early by peek expose

Browse Source 
Signed-off-by: Lyndon-Li <lyonghui@vmware.com>
This commit is contained in:
Lyndon-Li
2024-02-05 16:51:18 +08:00
parent 9a907a21f2
commit 7bf7fb9fc1
13 changed files with 340 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
changelogs/unreleased/7437-Lyndon-Li Normal file
View File

@@ -0,0 +1 @@
Fix issue #7036. Add the implementation of node selection for data mover backups

8
pkg/controller/data_download_controller.go
View File

@@ -215,7 +215,10 @@ func (r *DataDownloadReconciler) Reconcile(ctx context.Context, req ctrl.Request
} else if dd.Status.Phase == velerov2alpha1api.DataDownloadPhaseAccepted {
if dd.Spec.Cancel {
log.Debugf("Data download is been canceled %s in Phase %s", dd.GetName(), dd.Status.Phase)
r.TryCancelDataDownload(ctx, dd)
r.TryCancelDataDownload(ctx, dd, "")
} else if peekErr := r.restoreExposer.PeekExposed(ctx, getDataDownloadOwnerObject(dd)); peekErr != nil {
r.TryCancelDataDownload(ctx, dd, fmt.Sprintf("found a dataupload %s/%s with expose error: %s. mark it as cancel", dd.Namespace, dd.Name, peekErr))
log.Errorf("Cancel dd %s/%s because of expose error %s", dd.Namespace, dd.Name, peekErr)
} else if dd.Status.StartTimestamp != nil {
if time.Since(dd.Status.StartTimestamp.Time) >= r.preparingTimeout {
r.onPrepareTimeout(ctx, dd)
@@ -419,7 +422,7 @@ func (r *DataDownloadReconciler) OnDataDownloadCancelled(ctx context.Context, na
}
}
func (r *DataDownloadReconciler) TryCancelDataDownload(ctx context.Context, dd *velerov2alpha1api.DataDownload) {
func (r *DataDownloadReconciler) TryCancelDataDownload(ctx context.Context, dd *velerov2alpha1api.DataDownload, message string) {
log := r.logger.WithField("datadownload", dd.Name)
log.Warn("Async fs backup data path canceled")
@@ -429,6 +432,7 @@ func (r *DataDownloadReconciler) TryCancelDataDownload(ctx context.Context, dd *
dataDownload.Status.StartTimestamp = &metav1.Time{Time: r.Clock.Now()}
}
dataDownload.Status.CompletionTimestamp = &metav1.Time{Time: r.Clock.Now()}
dataDownload.Status.Message = message
})
if err != nil {

13
pkg/controller/data_download_controller_test.go
View File

@@ -174,6 +174,7 @@ func TestDataDownloadReconcile(t *testing.T) {
needCreateFSBR bool
isExposeErr bool
isGetExposeErr bool
isPeekExposeErr bool
isNilExposer bool
isFSBRInitErr bool
isFSBRRestoreErr bool
@@ -302,6 +303,12 @@ func TestDataDownloadReconcile(t *testing.T) {
dd: dataDownloadBuilder().Phase(velerov2alpha1api.DataDownloadPhaseAccepted).StartTimestamp(&metav1.Time{Time: time.Now().Add(-time.Minute * 5)}).Result(),
expected: dataDownloadBuilder().Phase(velerov2alpha1api.DataDownloadPhaseFailed).Result(),
},
{
name: "peek error",
dd: dataDownloadBuilder().Phase(velerov2alpha1api.DataDownloadPhaseAccepted).Result(),
isPeekExposeErr: true,
expected: dataDownloadBuilder().Phase(velerov2alpha1api.DataDownloadPhaseCanceled).Result(),
},
{
name: "dataDownload with enabled cancel",
dd: func() *velerov2alpha1api.DataDownload {
@@ -369,7 +376,7 @@ func TestDataDownloadReconcile(t *testing.T) {
return fsBR
}
if test.isExposeErr || test.isGetExposeErr || test.isNilExposer || test.notNilExpose {
if test.isExposeErr || test.isGetExposeErr || test.isPeekExposeErr || test.isNilExposer || test.notNilExpose {
if test.isNilExposer {
r.restoreExposer = nil
} else {
@@ -383,6 +390,8 @@ func TestDataDownloadReconcile(t *testing.T) {
ep.On("GetExposed", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(&exposer.ExposeResult{ByPod: exposer.ExposeByPod{HostingPod: hostingPod, VolumeName: "test-pvc"}}, nil)
} else if test.isGetExposeErr {
ep.On("GetExposed", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil, errors.New("Error to get restore exposer"))
} else if test.isPeekExposeErr {
ep.On("PeekExposed", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(errors.New("fake-peek-error"))
}
if !test.notMockCleanUp {
@@ -802,7 +811,7 @@ func TestTryCancelDataDownload(t *testing.T) {
err = r.client.Create(ctx, test.dd)
require.NoError(t, err)
r.TryCancelDataDownload(ctx, test.dd)
r.TryCancelDataDownload(ctx, test.dd, "")
if test.expectedErr == "" {
assert.NoError(t, err)

8
pkg/controller/data_upload_controller.go
View File

@@ -228,7 +228,10 @@ func (r *DataUploadReconciler) Reconcile(ctx context.Context, req ctrl.Request)
// we don't want to update CR into cancel status forcely as it may conflict with CR update in Expose action
// we could retry when the CR requeue in periodcally
log.Debugf("Data upload is been canceled %s in Phase %s", du.GetName(), du.Status.Phase)
r.TryCancelDataUpload(ctx, du)
r.TryCancelDataUpload(ctx, du, "")
} else if peekErr := ep.PeekExposed(ctx, getOwnerObject(du)); peekErr != nil {
r.TryCancelDataUpload(ctx, du, fmt.Sprintf("found a dataupload %s/%s with expose error: %s. mark it as cancel", du.Namespace, du.Name, peekErr))
log.Errorf("Cancel du %s/%s because of expose error %s", du.Namespace, du.Name, peekErr)
} else if du.Status.StartTimestamp != nil {
if time.Since(du.Status.StartTimestamp.Time) >= r.preparingTimeout {
r.onPrepareTimeout(ctx, du)
@@ -444,7 +447,7 @@ func (r *DataUploadReconciler) OnDataUploadCancelled(ctx context.Context, namesp
}
// TryCancelDataUpload clear up resources only when update success
func (r *DataUploadReconciler) TryCancelDataUpload(ctx context.Context, du *velerov2alpha1api.DataUpload) {
func (r *DataUploadReconciler) TryCancelDataUpload(ctx context.Context, du *velerov2alpha1api.DataUpload, message string) {
log := r.logger.WithField("dataupload", du.Name)
log.Warn("Async fs backup data path canceled")
succeeded, err := r.exclusiveUpdateDataUpload(ctx, du, func(dataUpload *velerov2alpha1api.DataUpload) {
@@ -453,6 +456,7 @@ func (r *DataUploadReconciler) TryCancelDataUpload(ctx context.Context, du *vele
dataUpload.Status.StartTimestamp = &metav1.Time{Time: r.Clock.Now()}
}
dataUpload.Status.CompletionTimestamp = &metav1.Time{Time: r.Clock.Now()}
dataUpload.Status.Message = message
})
if err != nil {

17
pkg/controller/data_upload_controller_test.go
View File

@@ -252,6 +252,7 @@ func dataUploadBuilder() *builder.DataUploadBuilder {
type fakeSnapshotExposer struct {
kubeClient kbclient.Client
clock clock.WithTickerAndDelayedExecution
peekErr error
}
func (f *fakeSnapshotExposer) Expose(ctx context.Context, ownerObject corev1.ObjectReference, param interface{}) error {
@@ -283,6 +284,10 @@ func (f *fakeSnapshotExposer) GetExposed(ctx context.Context, du corev1.ObjectRe
return &exposer.ExposeResult{ByPod: exposer.ExposeByPod{HostingPod: pod, VolumeName: dataUploadName}}, nil
}
func (f *fakeSnapshotExposer) PeekExposed(ctx context.Context, ownerObject corev1.ObjectReference) error {
return f.peekErr
}
func (f *fakeSnapshotExposer) CleanUp(context.Context, corev1.ObjectReference, string, string) {
}
@@ -330,6 +335,7 @@ func TestReconcile(t *testing.T) {
expectedRequeue ctrl.Result
expectedErrMsg string
needErrs []bool
peekErr error
}{
{
name: "Dataupload is not initialized",
@@ -420,6 +426,13 @@ func TestReconcile(t *testing.T) {
du: dataUploadBuilder().Phase(velerov2alpha1api.DataUploadPhaseAccepted).SnapshotType(fakeSnapshotType).StartTimestamp(&metav1.Time{Time: time.Now().Add(-time.Minute * 5)}).Result(),
expected: dataUploadBuilder().Phase(velerov2alpha1api.DataUploadPhaseFailed).Result(),
},
{
name: "peek error",
du: dataUploadBuilder().Phase(velerov2alpha1api.DataUploadPhaseAccepted).SnapshotType(fakeSnapshotType).Result(),
peekErr: errors.New("fake-peek-error"),
expectedProcessed: true,
expected: dataUploadBuilder().Phase(velerov2alpha1api.DataUploadPhaseCanceled).Result(),
},
{
name: "Dataupload with enabled cancel",
pod: builder.ForPod(velerov1api.DefaultNamespace, dataUploadName).Volumes(&corev1.Volume{Name: "dataupload-1"}).Result(),
@@ -486,7 +499,7 @@ func TestReconcile(t *testing.T) {
}
if test.du.Spec.SnapshotType == fakeSnapshotType {
r.snapshotExposerList = map[velerov2alpha1api.SnapshotType]exposer.SnapshotExposer{fakeSnapshotType: &fakeSnapshotExposer{r.client, r.Clock}}
r.snapshotExposerList = map[velerov2alpha1api.SnapshotType]exposer.SnapshotExposer{fakeSnapshotType: &fakeSnapshotExposer{r.client, r.Clock, test.peekErr}}
} else if test.du.Spec.SnapshotType == velerov2alpha1api.SnapshotTypeCSI {
r.snapshotExposerList = map[velerov2alpha1api.SnapshotType]exposer.SnapshotExposer{velerov2alpha1api.SnapshotTypeCSI: exposer.NewCSISnapshotExposer(r.kubeClient, r.csiSnapshotClient, velerotest.NewLogger())}
}
@@ -866,7 +879,7 @@ func TestTryCancelDataUpload(t *testing.T) {
err = r.client.Create(ctx, test.dd)
require.NoError(t, err)
r.TryCancelDataUpload(ctx, test.dd)
r.TryCancelDataUpload(ctx, test.dd, "")
if test.expectedErr == "" {
assert.NoError(t, err)

24
pkg/exposer/csi_snapshot.go
View File

@@ -259,6 +259,30 @@ func (e *csiSnapshotExposer) GetExposed(ctx context.Context, ownerObject corev1.
return &ExposeResult{ByPod: ExposeByPod{HostingPod: pod, VolumeName: volumeName}}, nil
}
func (e *csiSnapshotExposer) PeekExposed(ctx context.Context, ownerObject corev1.ObjectReference) error {
backupPodName := ownerObject.Name
curLog := e.log.WithFields(logrus.Fields{
"owner": ownerObject.Name,
})
pod, err := e.kubeClient.CoreV1().Pods(ownerObject.Namespace).Get(ctx, backupPodName, metav1.GetOptions{})
if apierrors.IsNotFound(err) {
return nil
}
if err != nil {
curLog.WithError(err).Warnf("error to peek backup pod %s", backupPodName)
return nil
}
if podFailed, message := kube.IsPodUnrecoverable(pod, curLog); podFailed {
return errors.New(message)
}
return nil
}
func (e *csiSnapshotExposer) CleanUp(ctx context.Context, ownerObject corev1.ObjectReference, vsName string, sourceNamespace string) {
backupPodName := ownerObject.Name
backupPVCName := ownerObject.Name

97
pkg/exposer/csi_snapshot_test.go
View File

@@ -621,3 +621,100 @@ func TestGetExpose(t *testing.T) {
})
}
}
func TestPeekExpose(t *testing.T) {
backup := &velerov1.Backup{
TypeMeta: metav1.TypeMeta{
APIVersion: velerov1.SchemeGroupVersion.String(),
Kind: "Backup",
},
ObjectMeta: metav1.ObjectMeta{
Namespace: velerov1.DefaultNamespace,
Name: "fake-backup",
UID: "fake-uid",
},
}
backupPodUrecoverable := &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{
Namespace: backup.Namespace,
Name: backup.Name,
},
Status: corev1.PodStatus{
Phase: corev1.PodPending,
Conditions: []corev1.PodCondition{
{
Type: corev1.PodScheduled,
Reason: "Unschedulable",
Message: "unrecoverable",
},
},
},
}
backupPod := &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{
Namespace: backup.Namespace,
Name: backup.Name,
},
}
scheme := runtime.NewScheme()
corev1.AddToScheme(scheme)
tests := []struct {
name string
kubeClientObj []runtime.Object
ownerBackup *velerov1.Backup
err string
}{
{
name: "backup pod is not found",
ownerBackup: backup,
},
{
name: "pod is unrecoverable",
ownerBackup: backup,
kubeClientObj: []runtime.Object{
backupPodUrecoverable,
},
err: "Pod is unschedulable: unrecoverable",
},
{
name: "succeed",
ownerBackup: backup,
kubeClientObj: []runtime.Object{
backupPod,
},
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
fakeKubeClient := fake.NewSimpleClientset(test.kubeClientObj...)
exposer := csiSnapshotExposer{
kubeClient: fakeKubeClient,
log: velerotest.NewLogger(),
}
var ownerObject corev1.ObjectReference
if test.ownerBackup != nil {
ownerObject = corev1.ObjectReference{
Kind: test.ownerBackup.Kind,
Namespace: test.ownerBackup.Namespace,
Name: test.ownerBackup.Name,
UID: test.ownerBackup.UID,
APIVersion: test.ownerBackup.APIVersion,
}
}
err := exposer.PeekExposed(context.Background(), ownerObject)
if test.err == "" {
assert.NoError(t, err)
} else {
assert.EqualError(t, err, test.err)
}
})
}
}

29
pkg/exposer/generic_restore.go
View File

@@ -44,6 +44,11 @@ type GenericRestoreExposer interface {
// Otherwise, it returns nil as the expose result without an error.
GetExposed(context.Context, corev1.ObjectReference, client.Client, string, time.Duration) (*ExposeResult, error)
// PeekExposed tests the status of the expose.
// If the expose is incomplete but not recoverable, it returns an error.
// Otherwise, it returns nil immediately.
PeekExposed(context.Context, corev1.ObjectReference) error
// RebindVolume unexposes the restored PV and rebind it to the target PVC
RebindVolume(context.Context, corev1.ObjectReference, string, string, time.Duration) error
@@ -160,6 +165,30 @@ func (e *genericRestoreExposer) GetExposed(ctx context.Context, ownerObject core
return &ExposeResult{ByPod: ExposeByPod{HostingPod: pod, VolumeName: volumeName}}, nil
}
func (e *genericRestoreExposer) PeekExposed(ctx context.Context, ownerObject corev1.ObjectReference) error {
restorePodName := ownerObject.Name
curLog := e.log.WithFields(logrus.Fields{
"owner": ownerObject.Name,
})
pod, err := e.kubeClient.CoreV1().Pods(ownerObject.Namespace).Get(ctx, restorePodName, metav1.GetOptions{})
if apierrors.IsNotFound(err) {
return nil
}
if err != nil {
curLog.WithError(err).Warnf("error to peek restore pod %s", restorePodName)
return nil
}
if podFailed, message := kube.IsPodUnrecoverable(pod, curLog); podFailed {
return errors.New(message)
}
return nil
}
func (e *genericRestoreExposer) CleanUp(ctx context.Context, ownerObject corev1.ObjectReference) {
restorePodName := ownerObject.Name
restorePVCName := ownerObject.Name

94
pkg/exposer/generic_restore_test.go
View File

@@ -409,3 +409,97 @@ func TestRebindVolume(t *testing.T) {
})
}
}
func TestRestorePeekExpose(t *testing.T) {
restore := &velerov1.Restore{
TypeMeta: metav1.TypeMeta{
APIVersion: velerov1.SchemeGroupVersion.String(),
Kind: "Restore",
},
ObjectMeta: metav1.ObjectMeta{
Namespace: velerov1.DefaultNamespace,
Name: "fake-restore",
UID: "fake-uid",
},
}
restorePodUrecoverable := &corev1api.Pod{
ObjectMeta: metav1.ObjectMeta{
Namespace: restore.Namespace,
Name: restore.Name,
},
Status: corev1api.PodStatus{
Phase: corev1api.PodPending,
Conditions: []corev1api.PodCondition{
{
Type: corev1api.PodScheduled,
Reason: "Unschedulable",
Message: "unrecoverable",
},
},
},
}
restorePod := &corev1api.Pod{
ObjectMeta: metav1.ObjectMeta{
Namespace: restore.Namespace,
Name: restore.Name,
},
}
tests := []struct {
name string
kubeClientObj []runtime.Object
ownerRestore *velerov1.Restore
err string
}{
{
name: "restore pod is not found",
ownerRestore: restore,
},
{
name: "pod is unrecoverable",
ownerRestore: restore,
kubeClientObj: []runtime.Object{
restorePodUrecoverable,
},
err: "Pod is unschedulable: unrecoverable",
},
{
name: "succeed",
ownerRestore: restore,
kubeClientObj: []runtime.Object{
restorePod,
},
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
fakeKubeClient := fake.NewSimpleClientset(test.kubeClientObj...)
exposer := genericRestoreExposer{
kubeClient: fakeKubeClient,
log: velerotest.NewLogger(),
}
var ownerObject corev1api.ObjectReference
if test.ownerRestore != nil {
ownerObject = corev1api.ObjectReference{
Kind: test.ownerRestore.Kind,
Namespace: test.ownerRestore.Namespace,
Name: test.ownerRestore.Name,
UID: test.ownerRestore.UID,
APIVersion: test.ownerRestore.APIVersion,
}
}
err := exposer.PeekExposed(context.Background(), ownerObject)
if test.err == "" {
assert.NoError(t, err)
} else {
assert.EqualError(t, err, test.err)
}
})
}
}

41
pkg/exposer/mocks/generic_restore.go
View File

@@ -1,4 +1,4 @@
// Code generated by mockery v2.20.0. DO NOT EDIT.
// Code generated by mockery v2.39.1. DO NOT EDIT.
package mocks
@@ -30,6 +30,10 @@ func (_m *GenericRestoreExposer) CleanUp(_a0 context.Context, _a1 v1.ObjectRefer
func (_m *GenericRestoreExposer) Expose(_a0 context.Context, _a1 v1.ObjectReference, _a2 string, _a3 string, _a4 map[string]string, _a5 time.Duration) error {
ret := _m.Called(_a0, _a1, _a2, _a3, _a4, _a5)
if len(ret) == 0 {
panic("no return value specified for Expose")
}
var r0 error
if rf, ok := ret.Get(0).(func(context.Context, v1.ObjectReference, string, string, map[string]string, time.Duration) error); ok {
r0 = rf(_a0, _a1, _a2, _a3, _a4, _a5)
@@ -44,6 +48,10 @@ func (_m *GenericRestoreExposer) Expose(_a0 context.Context, _a1 v1.ObjectRefere
func (_m *GenericRestoreExposer) GetExposed(_a0 context.Context, _a1 v1.ObjectReference, _a2 client.Client, _a3 string, _a4 time.Duration) (*exposer.ExposeResult, error) {
ret := _m.Called(_a0, _a1, _a2, _a3, _a4)
if len(ret) == 0 {
panic("no return value specified for GetExposed")
}
var r0 *exposer.ExposeResult
var r1 error
if rf, ok := ret.Get(0).(func(context.Context, v1.ObjectReference, client.Client, string, time.Duration) (*exposer.ExposeResult, error)); ok {
@@ -66,10 +74,32 @@ func (_m *GenericRestoreExposer) GetExposed(_a0 context.Context, _a1 v1.ObjectRe
return r0, r1
}
// PeekExposed provides a mock function with given fields: _a0, _a1
func (_m *GenericRestoreExposer) PeekExposed(_a0 context.Context, _a1 v1.ObjectReference) error {
ret := _m.Called(_a0, _a1)
if len(ret) == 0 {
panic("no return value specified for PeekExposed")
}
var r0 error
if rf, ok := ret.Get(0).(func(context.Context, v1.ObjectReference) error); ok {
r0 = rf(_a0, _a1)
} else {
r0 = ret.Error(0)
}
return r0
}
// RebindVolume provides a mock function with given fields: _a0, _a1, _a2, _a3, _a4
func (_m *GenericRestoreExposer) RebindVolume(_a0 context.Context, _a1 v1.ObjectReference, _a2 string, _a3 string, _a4 time.Duration) error {
ret := _m.Called(_a0, _a1, _a2, _a3, _a4)
if len(ret) == 0 {
panic("no return value specified for RebindVolume")
}
var r0 error
if rf, ok := ret.Get(0).(func(context.Context, v1.ObjectReference, string, string, time.Duration) error); ok {
r0 = rf(_a0, _a1, _a2, _a3, _a4)
@@ -80,13 +110,12 @@ func (_m *GenericRestoreExposer) RebindVolume(_a0 context.Context, _a1 v1.Object
return r0
}
type mockConstructorTestingTNewGenericRestoreExposer interface {
// NewGenericRestoreExposer creates a new instance of GenericRestoreExposer. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
// The first argument is typically a *testing.T value.
func NewGenericRestoreExposer(t interface {
mock.TestingT
Cleanup(func())
}
// NewGenericRestoreExposer creates a new instance of GenericRestoreExposer. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
func NewGenericRestoreExposer(t mockConstructorTestingTNewGenericRestoreExposer) *GenericRestoreExposer {
}) *GenericRestoreExposer {
mock := &GenericRestoreExposer{}
mock.Mock.Test(t)

5
pkg/exposer/snapshot.go
View File

@@ -32,6 +32,11 @@ type SnapshotExposer interface {
// Otherwise, it returns nil as the expose result without an error.
GetExposed(context.Context, corev1.ObjectReference, time.Duration, interface{}) (*ExposeResult, error)
// PeekExposed tests the status of the expose.
// If the expose is incomplete but not recoverable, it returns an error.
// Otherwise, it returns nil immediately.
PeekExposed(context.Context, corev1.ObjectReference) error
// CleanUp cleans up any objects generated during the snapshot expose
CleanUp(context.Context, corev1.ObjectReference, string, string)
}

4
pkg/util/kube/pod.go
View File

@@ -123,9 +123,9 @@ func IsPodUnrecoverable(pod *corev1api.Pod, log logrus.FieldLogger) (bool, strin
if pod.Status.Phase == corev1api.PodPending && len(pod.Status.Conditions) > 0 {
for _, condition := range pod.Status.Conditions {
if condition.Type == "PodScheduled" && condition.Reason == "Unschedulable" {
if condition.Type == corev1api.PodScheduled && condition.Reason == "Unschedulable" {
log.Warnf("Pod is unschedulable %s", condition.Message)
return true, fmt.Sprintf("Pod is unschedulable %s", condition.Message)
return true, fmt.Sprintf("Pod is unschedulable: %s", condition.Message)
}
}
}

15
pkg/util/kube/pod_test.go
View File

@@ -401,6 +401,21 @@ func TestIsPodUnrecoverable(t *testing.T) {
},
want: false,
},
{
name: "pod is unschedulable",
pod: &corev1api.Pod{
Status: corev1api.PodStatus{
Phase: corev1api.PodPending,
Conditions: []corev1api.PodCondition{
{
Type: corev1api.PodScheduled,
Reason: "Unschedulable",
},
},
},
},
want: true,
},
{
name: "pod is normal",
pod: &corev1api.Pod{