mirrors/velero
1
0
Fork 0
mirror of https://github.com/vmware-tanzu/velero.git synced 2026-08-19 13:46:08 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Bump up the versions of severel Kubernetes-related libs

Browse Source 
Bump up the versions of severel Kubernetes-related libs

Signed-off-by: Wenkai Yin(尹文开) <yinw@vmware.com>
This commit is contained in:
Wenkai Yin(尹文开)
2024-03-05 13:09:38 +08:00
parent e727d29bcd
commit 8752c3a820
70 changed files with 335 additions and 403 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/apis/velero/v1/register.go
+2 -2
View File
@@ -61,8 +61,8 @@ func CustomResources() map[string]typeInfo {
}
// CustomResourceKinds returns a list of all custom resources kinds within the Velero
func CustomResourceKinds() sets.String {
kinds := sets.NewString()
func CustomResourceKinds() sets.Set[string] {
kinds := sets.New[string]()
resources := CustomResources()
for kind := range resources {
pkg/apis/velero/v2alpha1/register.go
+2 -2
View File
@@ -52,8 +52,8 @@ func CustomResources() map[string]typeInfo {
}
// CustomResourceKinds returns a list of all custom resources kinds within the Velero
func CustomResourceKinds() sets.String {
kinds := sets.NewString()
func CustomResourceKinds() sets.Set[string] {
kinds := sets.New[string]()
resources := CustomResources()
for kind := range resources {
pkg/backup/backup_test.go
+1 -1
View File
@@ -2756,7 +2756,7 @@ func TestBackupWithInvalidHooks(t *testing.T) {
builder.ForPod("foo", "bar").Result(),
),
},
want: errors.New("\"nonexistent-operator\" is not a valid pod selector operator"),
want: errors.New("\"nonexistent-operator\" is not a valid label selector operator"),
},
}
pkg/backup/snapshots.go
+1 -1
View File
@@ -19,7 +19,7 @@ package backup
import (
"context"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
"github.com/sirupsen/logrus"
"k8s.io/apimachinery/pkg/util/sets"
kbclient "sigs.k8s.io/controller-runtime/pkg/client"
pkg/builder/persistent_volume_claim_builder.go
+1 -1
View File
@@ -100,7 +100,7 @@ func (b *PersistentVolumeClaimBuilder) DataSource(dataSource *corev1api.TypedLoc
}
// DataSourceRef sets the PersistentVolumeClaim's spec.DataSourceRef.
func (b *PersistentVolumeClaimBuilder) DataSourceRef(dataSourceRef *corev1api.TypedLocalObjectReference) *PersistentVolumeClaimBuilder {
func (b *PersistentVolumeClaimBuilder) DataSourceRef(dataSourceRef *corev1api.TypedObjectReference) *PersistentVolumeClaimBuilder {
b.object.Spec.DataSourceRef = dataSourceRef
return b
}
pkg/builder/volume_snapshot_builder.go
+1 -1
View File
@@ -17,7 +17,7 @@ limitations under the License.
package builder
import (
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
pkg/builder/volume_snapshot_class_builder.go
+1 -1
View File
@@ -17,7 +17,7 @@ limitations under the License.
package builder
import (
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
pkg/builder/volume_snapshot_content_builder.go
+1 -1
View File
@@ -17,7 +17,7 @@ limitations under the License.
package builder
import (
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
v1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
pkg/client/factory.go
+1 -1
View File
@@ -24,7 +24,7 @@ import (
k8scheme "k8s.io/client-go/kubernetes/scheme"
kbclient "sigs.k8s.io/controller-runtime/pkg/client"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
"github.com/pkg/errors"
"github.com/spf13/pflag"
"k8s.io/apimachinery/pkg/runtime"
pkg/client/factory_test.go
+4 -4
View File
@@ -134,12 +134,12 @@ func TestFactory(t *testing.T) {
assert.NotNil(t, dynamicClient)
kubebuilderClient, e := f.KubebuilderClient()
assert.Contains(t, e.Error(), fmt.Sprintf("Get \"%s/api?timeout=", test.expectedHost))
assert.Nil(t, kubebuilderClient)
assert.Nil(t, e)
assert.NotNil(t, kubebuilderClient)
kbClientWithWatch, e := f.KubebuilderWatchClient()
assert.Contains(t, e.Error(), fmt.Sprintf("Get \"%s/api?timeout=", test.expectedHost))
assert.Nil(t, kbClientWithWatch)
assert.Nil(t, e)
assert.NotNil(t, kbClientWithWatch)
})
}
}
pkg/cmd/cli/backup/create.go
+1 -1
View File
@@ -252,7 +252,7 @@ func (o *CreateOptions) Run(c *cobra.Command, f client.Factory) error {
ObjectList: new(velerov1api.BackupList),
}
backupInformer := cache.NewSharedInformer(&lw, &velerov1api.Backup{}, time.Second)
backupInformer.AddEventHandler(
_, _ = backupInformer.AddEventHandler(
cache.FilteringResourceEventHandler{
FilterFunc: func(obj interface{}) bool {
backup, ok := obj.(*velerov1api.Backup)
pkg/cmd/cli/nodeagent/server.go
+4 -4
View File
@@ -45,7 +45,7 @@ import (
"sigs.k8s.io/controller-runtime/pkg/log/zap"
"sigs.k8s.io/controller-runtime/pkg/manager"
snapshotv1client "github.com/kubernetes-csi/external-snapshotter/client/v4/clientset/versioned"
snapshotv1client "github.com/kubernetes-csi/external-snapshotter/client/v7/clientset/versioned"
"github.com/vmware-tanzu/velero/internal/credentials"
velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
@@ -172,7 +172,7 @@ func newNodeAgentServer(logger logrus.FieldLogger, factory client.Factory, confi
// use a field selector to filter to only pods scheduled on this node.
cacheOption := cache.Options{
SelectorsByObject: cache.SelectorsByObject{
ByObject: map[ctrlclient.Object]cache.ByObject{
&v1.Pod{}: {
Field: fields.Set{"spec.nodeName": nodeName}.AsSelector(),
},
@@ -191,8 +191,8 @@ func newNodeAgentServer(logger logrus.FieldLogger, factory client.Factory, confi
},
}
mgr, err := ctrl.NewManager(clientConfig, ctrl.Options{
Scheme: scheme,
NewCache: cache.BuilderWithOptions(cacheOption),
Scheme: scheme,
Cache: cacheOption,
})
if err != nil {
cancelFunc()
pkg/cmd/cli/restore/create.go
+1 -1
View File
@@ -354,7 +354,7 @@ func (o *CreateOptions) Run(c *cobra.Command, f client.Factory) error {
}
restoreInformer := cache.NewSharedInformer(&lw, &api.Restore{}, time.Second)
restoreInformer.AddEventHandler(
_, _ = restoreInformer.AddEventHandler(
cache.FilteringResourceEventHandler{
FilterFunc: func(obj interface{}) bool {
restore, ok := obj.(*api.Restore)
pkg/cmd/cli/uninstall/uninstall.go
+3 -3
View File
@@ -273,11 +273,11 @@ func deleteResources(ctx context.Context, kbClient kbclient.Client, namespace st
// First attempt to gracefully delete all the resources within a specified time frame, If the process exceeds the timeout limit,
// it is likely that there may be errors during the finalization of restores. In such cases, we should proceed with forcefully deleting the restores.
err = gracefullyDeleteResources(ctx, kbClient, namespace)
if err != nil && err != wait.ErrWaitTimeout {
if err != nil && !wait.Interrupted(err) {
return errors.Wrap(err, "Error deleting resources")
}
if err == wait.ErrWaitTimeout {
if wait.Interrupted(err) {
err = forcedlyDeleteResources(ctx, kbClient, namespace)
if err != nil {
return errors.Wrap(err, "Error deleting resources forcedly")
@@ -351,7 +351,7 @@ func gracefullyDeleteResource(ctx context.Context, kbClient kbclient.Client, nam
func waitDeletingResources(ctx context.Context, kbClient kbclient.Client, namespace string) error {
// Wait for the deletion of all the restores within a specified time frame
err := wait.PollImmediate(time.Second, gracefulDeletionMaximumDuration, func() (bool, error) {
err := wait.PollUntilContextTimeout(ctx, time.Second, gracefulDeletionMaximumDuration, true, func(ctx context.Context) (bool, error) {
itemsCount := 0
for i := range resToDelete {
if errList := kbClient.List(ctx, resToDelete[i], &kbclient.ListOptions{Namespace: namespace}); errList != nil {
pkg/cmd/server/server.go
+9 -4
View File
@@ -27,7 +27,7 @@ import (
"time"
logrusr "github.com/bombsimon/logrusr/v3"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
"github.com/pkg/errors"
"github.com/prometheus/client_golang/prometheus/promhttp"
"github.com/sirupsen/logrus"
@@ -47,6 +47,7 @@ import (
"k8s.io/client-go/rest"
"k8s.io/utils/clock"
ctrl "sigs.k8s.io/controller-runtime"
"sigs.k8s.io/controller-runtime/pkg/cache"
ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/manager"
@@ -350,8 +351,12 @@ func newServer(f client.Factory, config serverConfig, logger *logrus.Logger) (*s
ctrl.SetLogger(logrusr.New(logger))
mgr, err := ctrl.NewManager(clientConfig, ctrl.Options{
Scheme: scheme,
Namespace: f.Namespace(),
Scheme: scheme,
Cache: cache.Options{
DefaultNamespaces: map[string]cache.Config{
f.Namespace(): {},
},
},
})
if err != nil {
cancelFunc()
@@ -529,7 +534,7 @@ func (s *server) veleroResourcesExist() error {
s.logger.Info("Checking existence of Velero custom resource definitions")
// add more group versions whenever available
gvResources := map[string]sets.String{
gvResources := map[string]sets.Set[string]{
velerov1api.SchemeGroupVersion.String(): velerov1api.CustomResourceKinds(),
velerov2alpha1api.SchemeGroupVersion.String(): velerov2alpha1api.CustomResourceKinds(),
}
pkg/cmd/util/output/backup_describer.go
+1 -1
View File
@@ -28,7 +28,7 @@ import (
v1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
"github.com/pkg/errors"
"github.com/fatih/color"
pkg/controller/backup_controller.go
+1 -1
View File
@@ -24,7 +24,7 @@ import (
"strings"
"time"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
corev1api "k8s.io/api/core/v1"
pkg/controller/backup_controller_test.go
+1 -1
View File
@@ -28,7 +28,7 @@ import (
"time"
"github.com/google/go-cmp/cmp"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"github.com/stretchr/testify/assert"
pkg/controller/backup_deletion_controller.go
+2 -2
View File
@@ -101,7 +101,7 @@ func (r *backupDeletionReconciler) SetupWithManager(mgr ctrl.Manager) error {
s := kube.NewPeriodicalEnqueueSource(r.logger, mgr.GetClient(), &velerov1api.DeleteBackupRequestList{}, time.Hour, kube.PeriodicalEnqueueSourceOption{})
return ctrl.NewControllerManagedBy(mgr).
For(&velerov1api.DeleteBackupRequest{}).
Watches(s, nil).
WatchesRawSource(s, nil).
Complete(r)
}
@@ -375,7 +375,7 @@ func (r *backupDeletionReconciler) Reconcile(ctx context.Context, req ctrl.Reque
// Wait for the deletion of restores within certain amount of time.
// Notice that there could be potential errors during the finalization process, which may result in the failure to delete the restore.
// Therefore, it is advisable to set a timeout period for waiting.
err := wait.PollImmediate(time.Second, time.Minute, func() (bool, error) {
err := wait.PollUntilContextTimeout(ctx, time.Second, time.Minute, true, func(ctx context.Context) (bool, error) {
restoreList := &velerov1api.RestoreList{}
if err := r.List(ctx, restoreList, &client.ListOptions{Namespace: backup.Namespace, LabelSelector: selector}); err != nil {
return false, err
pkg/controller/backup_operations_controller.go
+1 -1
View File
@@ -92,7 +92,7 @@ func (c *backupOperationsReconciler) SetupWithManager(mgr ctrl.Manager) error {
})
return ctrl.NewControllerManagedBy(mgr).
For(&velerov1api.Backup{}, builder.WithPredicates(kube.FalsePredicate{})).
Watches(s, nil, builder.WithPredicates(gp)).
WatchesRawSource(s, nil, builder.WithPredicates(gp)).
Complete(c)
}
pkg/controller/backup_repository_controller.go
+3 -4
View File
@@ -32,7 +32,6 @@ import (
"sigs.k8s.io/controller-runtime/pkg/builder"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
"sigs.k8s.io/controller-runtime/pkg/source"
velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
"github.com/vmware-tanzu/velero/pkg/label"
@@ -74,8 +73,8 @@ func (r *BackupRepoReconciler) SetupWithManager(mgr ctrl.Manager) error {
return ctrl.NewControllerManagedBy(mgr).
For(&velerov1api.BackupRepository{}).
Watches(s, nil).
Watches(&source.Kind{Type: &velerov1api.BackupStorageLocation{}}, kube.EnqueueRequestsFromMapUpdateFunc(r.invalidateBackupReposForBSL),
WatchesRawSource(s, nil).
Watches(&velerov1api.BackupStorageLocation{}, kube.EnqueueRequestsFromMapUpdateFunc(r.invalidateBackupReposForBSL),
builder.WithPredicates(
// When BSL updates, check if the backup repositories need to be invalidated
kube.NewUpdateEventPredicate(r.needInvalidBackupRepo),
@@ -84,7 +83,7 @@ func (r *BackupRepoReconciler) SetupWithManager(mgr ctrl.Manager) error {
Complete(r)
}
func (r *BackupRepoReconciler) invalidateBackupReposForBSL(bslObj client.Object) []reconcile.Request {
func (r *BackupRepoReconciler) invalidateBackupReposForBSL(ctx context.Context, bslObj client.Object) []reconcile.Request {
bsl := bslObj.(*velerov1api.BackupStorageLocation)
list := &velerov1api.BackupRepositoryList{}
pkg/controller/backup_storage_location_controller.go
+1 -1
View File
@@ -203,7 +203,7 @@ func (r *backupStorageLocationReconciler) SetupWithManager(mgr ctrl.Manager) err
return ctrl.NewControllerManagedBy(mgr).
// As the "status.LastValidationTime" field is always updated, this triggers new reconciling process, skip the update event that include no spec change to avoid the reconcile loop
For(&velerov1api.BackupStorageLocation{}, builder.WithPredicates(kube.SpecChangePredicate{})).
Watches(g, nil, builder.WithPredicates(gp)).
WatchesRawSource(g, nil, builder.WithPredicates(gp)).
Complete(r)
}
pkg/controller/backup_sync_controller.go
+5 -5
View File
@@ -21,7 +21,7 @@ import (
"fmt"
"time"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -108,7 +108,7 @@ func (b *backupSyncReconciler) Reconcile(ctx context.Context, req ctrl.Request)
log.WithError(err).Error("Error listing backups in backup store")
return ctrl.Result{}, nil
}
backupStoreBackups := sets.NewString(res...)
backupStoreBackups := sets.New[string](res...)
log.WithField("backupCount", len(backupStoreBackups)).Debug("Got backups from backup store")
// get a list of all the backups that exist as custom resources in the cluster
@@ -126,7 +126,7 @@ func (b *backupSyncReconciler) Reconcile(ctx context.Context, req ctrl.Request)
}
// get a list of backups that *are* in the backup storage location and *aren't* in the cluster
clusterBackupsSet := sets.NewString()
clusterBackupsSet := sets.New[string]()
for _, b := range clusterBackupList.Items {
clusterBackupsSet.Insert(b.Name)
}
@@ -349,13 +349,13 @@ func (b *backupSyncReconciler) SetupWithManager(mgr ctrl.Manager) error {
return ctrl.NewControllerManagedBy(mgr).
// Filter all BSL events, because this controller is supposed to run periodically, not by event.
For(&velerov1api.BackupStorageLocation{}, builder.WithPredicates(kube.FalsePredicate{})).
Watches(backupSyncSource, nil, builder.WithPredicates(gp)).
WatchesRawSource(backupSyncSource, nil, builder.WithPredicates(gp)).
Complete(b)
}
// deleteOrphanedBackups deletes backup objects (CRDs) from Kubernetes that have the specified location
// and a phase of Completed, but no corresponding backup in object storage.
func (b *backupSyncReconciler) deleteOrphanedBackups(ctx context.Context, locationName string, backupStoreBackups sets.String, log logrus.FieldLogger) {
func (b *backupSyncReconciler) deleteOrphanedBackups(ctx context.Context, locationName string, backupStoreBackups sets.Set[string], log logrus.FieldLogger) {
var backupList velerov1api.BackupList
listOption := client.ListOptions{
LabelSelector: labels.Set(map[string]string{
pkg/controller/backup_sync_controller_test.go
+17 -17
View File
@@ -24,7 +24,7 @@ import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
"github.com/sirupsen/logrus"
apierrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/api/meta"
@@ -547,50 +547,50 @@ var _ = Describe("Backup Sync Reconciler", func() {
tests := []struct {
name string
cloudBackups sets.String
cloudBackups sets.Set[string]
k8sBackups []*velerov1api.Backup
namespace string
expectedDeletes sets.String
expectedDeletes sets.Set[string]
useLongBSLName bool
}{
{
name: "no overlapping backups",
namespace: "ns-1",
cloudBackups: sets.NewString("backup-1", "backup-2", "backup-3"),
cloudBackups: sets.New[string]("backup-1", "backup-2", "backup-3"),
k8sBackups: []*velerov1api.Backup{
baseBuilder("backupA").Phase(velerov1api.BackupPhaseCompleted).Result(),
baseBuilder("backupB").Phase(velerov1api.BackupPhaseCompleted).Result(),
baseBuilder("backupC").Phase(velerov1api.BackupPhasePartiallyFailed).Result(),
},
expectedDeletes: sets.NewString("backupA", "backupB", "backupC"),
expectedDeletes: sets.New[string]("backupA", "backupB", "backupC"),
},
{
name: "some overlapping backups",
namespace: "ns-1",
cloudBackups: sets.NewString("backup-1", "backup-2", "backup-3"),
cloudBackups: sets.New[string]("backup-1", "backup-2", "backup-3"),
k8sBackups: []*velerov1api.Backup{
baseBuilder("backup-1").Phase(velerov1api.BackupPhaseCompleted).Result(),
baseBuilder("backup-2").Phase(velerov1api.BackupPhaseCompleted).Result(),
baseBuilder("backup-B").Phase(velerov1api.BackupPhaseCompleted).Result(),
baseBuilder("backup-C").Phase(velerov1api.BackupPhasePartiallyFailed).Result(),
},
expectedDeletes: sets.NewString("backup-B", "backup-C"),
expectedDeletes: sets.New[string]("backup-B", "backup-C"),
},
{
name: "all overlapping backups",
namespace: "ns-1",
cloudBackups: sets.NewString("backup-1", "backup-2", "backup-3"),
cloudBackups: sets.New[string]("backup-1", "backup-2", "backup-3"),
k8sBackups: []*velerov1api.Backup{
baseBuilder("backup-1").Phase(velerov1api.BackupPhaseCompleted).Result(),
baseBuilder("backup-2").Phase(velerov1api.BackupPhaseCompleted).Result(),
baseBuilder("backup-3").Phase(velerov1api.BackupPhasePartiallyFailed).Result(),
},
expectedDeletes: sets.NewString(),
expectedDeletes: sets.New[string](),
},
{
name: "no overlapping backups but including backups that are not complete",
namespace: "ns-1",
cloudBackups: sets.NewString("backup-1", "backup-2", "backup-3"),
cloudBackups: sets.New[string]("backup-1", "backup-2", "backup-3"),
k8sBackups: []*velerov1api.Backup{
baseBuilder("backupA").Phase(velerov1api.BackupPhaseCompleted).Result(),
baseBuilder("backupB").Phase(velerov1api.BackupPhasePartiallyFailed).Result(),
@@ -600,23 +600,23 @@ var _ = Describe("Backup Sync Reconciler", func() {
baseBuilder("InProgress").Phase(velerov1api.BackupPhaseInProgress).Result(),
baseBuilder("New").Phase(velerov1api.BackupPhaseNew).Result(),
},
expectedDeletes: sets.NewString("backupA", "backupB"),
expectedDeletes: sets.New[string]("backupA", "backupB"),
},
{
name: "all overlapping backups and all backups that are not complete",
namespace: "ns-1",
cloudBackups: sets.NewString("backup-1", "backup-2", "backup-3"),
cloudBackups: sets.New[string]("backup-1", "backup-2", "backup-3"),
k8sBackups: []*velerov1api.Backup{
baseBuilder("backup-1").Phase(velerov1api.BackupPhaseFailed).Result(),
baseBuilder("backup-2").Phase(velerov1api.BackupPhaseFailedValidation).Result(),
baseBuilder("backup-3").Phase(velerov1api.BackupPhaseInProgress).Result(),
},
expectedDeletes: sets.NewString(),
expectedDeletes: sets.New[string](),
},
{
name: "no completed backups in other locations are deleted",
namespace: "ns-1",
cloudBackups: sets.NewString("backup-1", "backup-2", "backup-3"),
cloudBackups: sets.New[string]("backup-1", "backup-2", "backup-3"),
k8sBackups: []*velerov1api.Backup{
baseBuilder("backup-1").Phase(velerov1api.BackupPhaseCompleted).Result(),
baseBuilder("backup-2").Phase(velerov1api.BackupPhaseCompleted).Result(),
@@ -627,12 +627,12 @@ var _ = Describe("Backup Sync Reconciler", func() {
baseBuilder("backup-5").ObjectMeta(builder.WithLabels(velerov1api.StorageLocationLabel, "alternate")).Phase(velerov1api.BackupPhaseCompleted).Result(),
baseBuilder("backup-6").ObjectMeta(builder.WithLabels(velerov1api.StorageLocationLabel, "alternate")).Phase(velerov1api.BackupPhasePartiallyFailed).Result(),
},
expectedDeletes: sets.NewString("backup-C", "backup-D"),
expectedDeletes: sets.New[string]("backup-C", "backup-D"),
},
{
name: "some overlapping backups",
namespace: "ns-1",
cloudBackups: sets.NewString("backup-1", "backup-2", "backup-3"),
cloudBackups: sets.New[string]("backup-1", "backup-2", "backup-3"),
k8sBackups: []*velerov1api.Backup{
builder.ForBackup("ns-1", "backup-1").
ObjectMeta(
@@ -659,7 +659,7 @@ var _ = Describe("Backup Sync Reconciler", func() {
Phase(velerov1api.BackupPhasePartiallyFailed).
Result(),
},
expectedDeletes: sets.NewString("backup-C", "backup-D"),
expectedDeletes: sets.New[string]("backup-C", "backup-D"),
useLongBSLName: true,
},
}
pkg/controller/backup_tracker.go
+2 -2
View File
@@ -35,13 +35,13 @@ type BackupTracker interface {
type backupTracker struct {
lock sync.RWMutex
backups sets.String
backups sets.Set[string]
}
// NewBackupTracker returns a new BackupTracker.
func NewBackupTracker() BackupTracker {
return &backupTracker{
backups: sets.NewString(),
backups: sets.New[string](),
}
}
pkg/controller/data_download_controller.go
+4 -5
View File
@@ -37,7 +37,6 @@ import (
"sigs.k8s.io/controller-runtime/pkg/event"
"sigs.k8s.io/controller-runtime/pkg/predicate"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
"sigs.k8s.io/controller-runtime/pkg/source"
"github.com/vmware-tanzu/velero/internal/credentials"
"github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
@@ -476,8 +475,8 @@ func (r *DataDownloadReconciler) SetupWithManager(mgr ctrl.Manager) error {
return ctrl.NewControllerManagedBy(mgr).
For(&velerov2alpha1api.DataDownload{}).
Watches(s, nil, builder.WithPredicates(gp)).
Watches(&source.Kind{Type: &v1.Pod{}}, kube.EnqueueRequestsFromMapUpdateFunc(r.findSnapshotRestoreForPod),
WatchesRawSource(s, nil, builder.WithPredicates(gp)).
Watches(&v1.Pod{}, kube.EnqueueRequestsFromMapUpdateFunc(r.findSnapshotRestoreForPod),
builder.WithPredicates(predicate.Funcs{
UpdateFunc: func(ue event.UpdateEvent) bool {
newObj := ue.ObjectNew.(*v1.Pod)
@@ -505,7 +504,7 @@ func (r *DataDownloadReconciler) SetupWithManager(mgr ctrl.Manager) error {
Complete(r)
}
func (r *DataDownloadReconciler) findSnapshotRestoreForPod(podObj client.Object) []reconcile.Request {
func (r *DataDownloadReconciler) findSnapshotRestoreForPod(ctx context.Context, podObj client.Object) []reconcile.Request {
pod := podObj.(*v1.Pod)
dd, err := findDataDownloadByPod(r.client, *pod)
@@ -780,7 +779,7 @@ func isDataDownloadInFinalState(dd *velerov2alpha1api.DataDownload) bool {
}
func UpdateDataDownloadWithRetry(ctx context.Context, client client.Client, namespacedName types.NamespacedName, log *logrus.Entry, updateFunc func(dataDownload *velerov2alpha1api.DataDownload)) error {
return wait.PollUntilWithContext(ctx, time.Second, func(ctx context.Context) (done bool, err error) {
return wait.PollUntilContextCancel(ctx, time.Second, true, func(ctx context.Context) (done bool, err error) {
dd := &velerov2alpha1api.DataDownload{}
if err := client.Get(ctx, namespacedName, dd); err != nil {
return false, errors.Wrap(err, "getting DataDownload")
pkg/controller/data_download_controller_test.go
+1 -1
View File
@@ -670,7 +670,7 @@ func TestFindDataDownloadForPod(t *testing.T) {
assert.NoError(t, r.client.Create(ctx, test.pod))
assert.NoError(t, r.client.Create(ctx, test.du))
// Call the findSnapshotRestoreForPod function
requests := r.findSnapshotRestoreForPod(test.pod)
requests := r.findSnapshotRestoreForPod(context.Background(), test.pod)
test.checkFunc(test.du, requests)
r.client.Delete(ctx, test.du, &kbclient.DeleteOptions{})
if test.pod != nil {
pkg/controller/data_upload_controller.go
+5 -6
View File
@@ -37,9 +37,8 @@ import (
"sigs.k8s.io/controller-runtime/pkg/event"
"sigs.k8s.io/controller-runtime/pkg/predicate"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
"sigs.k8s.io/controller-runtime/pkg/source"
snapshotter "github.com/kubernetes-csi/external-snapshotter/client/v4/clientset/versioned/typed/volumesnapshot/v1"
snapshotter "github.com/kubernetes-csi/external-snapshotter/client/v7/clientset/versioned/typed/volumesnapshot/v1"
"github.com/vmware-tanzu/velero/internal/credentials"
"github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
@@ -512,8 +511,8 @@ func (r *DataUploadReconciler) SetupWithManager(mgr ctrl.Manager) error {
return ctrl.NewControllerManagedBy(mgr).
For(&velerov2alpha1api.DataUpload{}).
Watches(s, nil, builder.WithPredicates(gp)).
Watches(&source.Kind{Type: &corev1.Pod{}}, kube.EnqueueRequestsFromMapUpdateFunc(r.findDataUploadForPod),
WatchesRawSource(s, nil, builder.WithPredicates(gp)).
Watches(&corev1.Pod{}, kube.EnqueueRequestsFromMapUpdateFunc(r.findDataUploadForPod),
builder.WithPredicates(predicate.Funcs{
UpdateFunc: func(ue event.UpdateEvent) bool {
newObj := ue.ObjectNew.(*corev1.Pod)
@@ -541,7 +540,7 @@ func (r *DataUploadReconciler) SetupWithManager(mgr ctrl.Manager) error {
Complete(r)
}
func (r *DataUploadReconciler) findDataUploadForPod(podObj client.Object) []reconcile.Request {
func (r *DataUploadReconciler) findDataUploadForPod(ctx context.Context, podObj client.Object) []reconcile.Request {
pod := podObj.(*corev1.Pod)
du, err := findDataUploadByPod(r.client, *pod)
log := r.logger.WithFields(logrus.Fields{
@@ -874,7 +873,7 @@ func isDataUploadInFinalState(du *velerov2alpha1api.DataUpload) bool {
}
func UpdateDataUploadWithRetry(ctx context.Context, client client.Client, namespacedName types.NamespacedName, log *logrus.Entry, updateFunc func(dataUpload *velerov2alpha1api.DataUpload)) error {
return wait.PollUntilWithContext(ctx, time.Second, func(ctx context.Context) (done bool, err error) {
return wait.PollUntilContextCancel(ctx, time.Second, true, func(ctx context.Context) (done bool, err error) {
du := &velerov2alpha1api.DataUpload{}
if err := client.Get(ctx, namespacedName, du); err != nil {
return false, errors.Wrap(err, "getting DataUpload")
pkg/controller/data_upload_controller_test.go
+12 -4
View File
@@ -22,8 +22,8 @@ import (
"testing"
"time"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
snapshotFake "github.com/kubernetes-csi/external-snapshotter/client/v4/clientset/versioned/fake"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
snapshotFake "github.com/kubernetes-csi/external-snapshotter/client/v7/clientset/versioned/fake"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"github.com/stretchr/testify/assert"
@@ -71,7 +71,7 @@ type FakeClient struct {
listError error
}
func (c *FakeClient) Get(ctx context.Context, key kbclient.ObjectKey, obj kbclient.Object) error {
func (c *FakeClient) Get(ctx context.Context, key kbclient.ObjectKey, obj kbclient.Object, opts ...kbclient.GetOption) error {
if c.getError != nil {
return c.getError
}
@@ -465,8 +465,16 @@ func TestReconcile(t *testing.T) {
}()
ctx := context.Background()
if test.du.Namespace == velerov1api.DefaultNamespace {
isDeletionTimestampSet := test.du.DeletionTimestamp != nil
err = r.client.Create(ctx, test.du)
require.NoError(t, err)
// because of the changes introduced by https://github.com/kubernetes-sigs/controller-runtime/commit/7a66d580c0c53504f5b509b45e9300cc18a1cc30
// the fake client ignores the DeletionTimestamp when calling the Create(),
// so call Delete() here
if isDeletionTimestampSet {
err = r.client.Delete(ctx, test.du)
require.NoError(t, err)
}
}
if test.pod != nil {
@@ -720,7 +728,7 @@ func TestFindDataUploadForPod(t *testing.T) {
assert.NoError(t, r.client.Create(ctx, test.pod))
assert.NoError(t, r.client.Create(ctx, test.du))
// Call the findDataUploadForPod function
requests := r.findDataUploadForPod(test.pod)
requests := r.findDataUploadForPod(context.Background(), test.pod)
test.checkFunc(test.du, requests)
r.client.Delete(ctx, test.du, &kbclient.DeleteOptions{})
if test.pod != nil {
pkg/controller/download_request_controller.go
+1 -1
View File
@@ -229,6 +229,6 @@ func (r *downloadRequestReconciler) SetupWithManager(mgr ctrl.Manager) error {
return ctrl.NewControllerManagedBy(mgr).
For(&velerov1api.DownloadRequest{}).
Watches(downloadRequestSource, nil, builder.WithPredicates(downloadRequestPredicates)).
WatchesRawSource(downloadRequestSource, nil, builder.WithPredicates(downloadRequestPredicates)).
Complete(r)
}
pkg/controller/gc_controller.go
+1 -1
View File
@@ -88,7 +88,7 @@ func (c *gcReconciler) SetupWithManager(mgr ctrl.Manager) error {
return false
},
})).
Watches(s, nil).
WatchesRawSource(s, nil).
Complete(c)
}
pkg/controller/pod_volume_restore_controller.go
+2 -3
View File
@@ -35,7 +35,6 @@ import (
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/handler"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
"sigs.k8s.io/controller-runtime/pkg/source"
"github.com/vmware-tanzu/velero/internal/credentials"
veleroapishared "github.com/vmware-tanzu/velero/pkg/apis/velero/shared"
@@ -207,11 +206,11 @@ func (c *PodVolumeRestoreReconciler) SetupWithManager(mgr ctrl.Manager) error {
// By watching the pods, we can trigger the PVR reconciliation again once the pod is finally scheduled on the node.
return ctrl.NewControllerManagedBy(mgr).
For(&velerov1api.PodVolumeRestore{}).
Watches(&source.Kind{Type: &corev1api.Pod{}}, handler.EnqueueRequestsFromMapFunc(c.findVolumeRestoresForPod)).
Watches(&corev1api.Pod{}, handler.EnqueueRequestsFromMapFunc(c.findVolumeRestoresForPod)).
Complete(c)
}
func (c *PodVolumeRestoreReconciler) findVolumeRestoresForPod(pod client.Object) []reconcile.Request {
func (c *PodVolumeRestoreReconciler) findVolumeRestoresForPod(ctx context.Context, pod client.Object) []reconcile.Request {
list := &velerov1api.PodVolumeRestoreList{}
options := &client.ListOptions{
LabelSelector: labels.Set(map[string]string{
pkg/controller/pod_volume_restore_controller_test.go
+2 -2
View File
@@ -491,7 +491,7 @@ func TestFindVolumeRestoresForPod(t *testing.T) {
Client: clientBuilder.Build(),
logger: logrus.New(),
}
requests := reconciler.findVolumeRestoresForPod(pod)
requests := reconciler.findVolumeRestoresForPod(context.Background(), pod)
assert.Len(t, requests, 0)
// contain one matching PVR
@@ -515,6 +515,6 @@ func TestFindVolumeRestoresForPod(t *testing.T) {
},
},
}).Build()
requests = reconciler.findVolumeRestoresForPod(pod)
requests = reconciler.findVolumeRestoresForPod(context.Background(), pod)
assert.Len(t, requests, 1)
}
pkg/controller/restore_controller_test.go
+9 -1
View File
@@ -23,7 +23,7 @@ import (
"testing"
"time"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"github.com/stretchr/testify/assert"
@@ -473,7 +473,15 @@ func TestRestoreReconcile(t *testing.T) {
}
if test.restore != nil {
isDeletionTimestampSet := test.restore.DeletionTimestamp != nil
require.NoError(t, r.kbClient.Create(context.Background(), test.restore))
// because of the changes introduced by https://github.com/kubernetes-sigs/controller-runtime/commit/7a66d580c0c53504f5b509b45e9300cc18a1cc30
// the fake client ignores the DeletionTimestamp when calling the Create(),
// so call Delete() here
if isDeletionTimestampSet {
err = r.kbClient.Delete(ctx, test.restore)
require.NoError(t, err)
}
}
var warnings, errors results.Result
pkg/controller/restore_operations_controller.go
+1 -1
View File
@@ -90,7 +90,7 @@ func (r *restoreOperationsReconciler) SetupWithManager(mgr ctrl.Manager) error {
})
return ctrl.NewControllerManagedBy(mgr).
For(&velerov1api.Restore{}, builder.WithPredicates(kube.FalsePredicate{})).
Watches(s, nil, builder.WithPredicates(gp)).
WatchesRawSource(s, nil, builder.WithPredicates(gp)).
Complete(r)
}
pkg/controller/schedule_controller.go
+1 -1
View File
@@ -81,7 +81,7 @@ func (c *scheduleReconciler) SetupWithManager(mgr ctrl.Manager) error {
return true
})).
For(&velerov1.Schedule{}, bld.WithPredicates(kube.SpecChangePredicate{})).
Watches(s, nil).
WatchesRawSource(s, nil).
Complete(c)
}
pkg/exposer/csi_snapshot.go
+4 -4
View File
@@ -31,12 +31,12 @@ import (
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/kubernetes"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
"github.com/vmware-tanzu/velero/pkg/util/csi"
"github.com/vmware-tanzu/velero/pkg/util/kube"
snapshotter "github.com/kubernetes-csi/external-snapshotter/client/v4/clientset/versioned/typed/volumesnapshot/v1"
snapshotter "github.com/kubernetes-csi/external-snapshotter/client/v7/clientset/versioned/typed/volumesnapshot/v1"
"sigs.k8s.io/controller-runtime/pkg/client"
apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -364,9 +364,9 @@ func (e *csiSnapshotExposer) createBackupPVC(ctx context.Context, ownerObject co
StorageClassName: &storageClass,
VolumeMode: &volumeMode,
DataSource: dataSource,
DataSourceRef: dataSource,
DataSourceRef: nil,
Resources: corev1.ResourceRequirements{
Resources: corev1.VolumeResourceRequirements{
Requests: corev1.ResourceList{
corev1.ResourceStorage: resource,
},
pkg/exposer/csi_snapshot_test.go
+2 -2
View File
@@ -21,8 +21,8 @@ import (
"testing"
"time"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
snapshotFake "github.com/kubernetes-csi/external-snapshotter/client/v4/clientset/versioned/fake"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
snapshotFake "github.com/kubernetes-csi/external-snapshotter/client/v7/clientset/versioned/fake"
"github.com/pkg/errors"
"github.com/stretchr/testify/assert"
"k8s.io/apimachinery/pkg/api/resource"
pkg/exposer/generic_restore_test.go
+1 -1
View File
@@ -374,7 +374,7 @@ func TestRebindVolume(t *testing.T) {
restorePVObj,
restorePod,
},
err: "error to wait restore PV bound, restore PV fake-restore-pv: error to wait for bound of PV: timed out waiting for the condition",
err: "error to wait restore PV bound, restore PV fake-restore-pv: error to wait for bound of PV: context deadline exceeded",
},
}
pkg/features/feature_flags.go
+3 -3
View File
@@ -23,7 +23,7 @@ import (
)
type featureFlagSet struct {
set sets.String
set sets.Set[string]
}
// featureFlags will store all the flags for this process until NewFeatureFlagSet is called.
@@ -51,7 +51,7 @@ func Disable(names ...string) {
// All returns enabled features as a slice of strings.
func All() []string {
return featureFlags.set.List()
return sets.List[string](featureFlags.set)
}
// Serialize returns all features as a comma-separated string.
@@ -64,6 +64,6 @@ func Serialize() string {
// It is also useful for selectively controlling flags during tests.
func NewFeatureFlagSet(flags ...string) {
featureFlags = featureFlagSet{
set: sets.NewString(flags...),
set: sets.New[string](flags...),
}
}
pkg/install/install.go
+11 -11
View File
@@ -63,15 +63,15 @@ type ResourceGroup struct {
// crdV1Beta1ReadinessFn returns a function that can be used for polling to check
// if the provided unstructured v1beta1 CRDs are ready for use in the cluster.
func crdV1Beta1ReadinessFn(kbClient kbclient.Client, unstructuredCrds []*unstructured.Unstructured) func() (bool, error) {
func crdV1Beta1ReadinessFn(kbClient kbclient.Client, unstructuredCrds []*unstructured.Unstructured) func(context.Context) (bool, error) {
// Track all the CRDs that have been found and in ready state.
// len should be equal to len(unstructuredCrds) in the happy path.
return func() (bool, error) {
return func(ctx context.Context) (bool, error) {
foundCRDs := make([]*apiextv1beta1.CustomResourceDefinition, 0)
for _, unstructuredCrd := range unstructuredCrds {
crd := &apiextv1beta1.CustomResourceDefinition{}
key := kbclient.ObjectKey{Name: unstructuredCrd.GetName()}
err := kbClient.Get(context.Background(), key, crd)
err := kbClient.Get(ctx, key, crd)
if apierrors.IsNotFound(err) {
return false, nil
} else if err != nil {
@@ -96,13 +96,13 @@ func crdV1Beta1ReadinessFn(kbClient kbclient.Client, unstructuredCrds []*unstruc
// crdV1ReadinessFn returns a function that can be used for polling to check
// if the provided unstructured v1 CRDs are ready for use in the cluster.
func crdV1ReadinessFn(kbClient kbclient.Client, unstructuredCrds []*unstructured.Unstructured) func() (bool, error) {
return func() (bool, error) {
func crdV1ReadinessFn(kbClient kbclient.Client, unstructuredCrds []*unstructured.Unstructured) func(context.Context) (bool, error) {
return func(ctx context.Context) (bool, error) {
foundCRDs := make([]*apiextv1.CustomResourceDefinition, 0)
for _, unstructuredCrd := range unstructuredCrds {
crd := &apiextv1.CustomResourceDefinition{}
key := kbclient.ObjectKey{Name: unstructuredCrd.GetName()}
err := kbClient.Get(context.Background(), key, crd)
err := kbClient.Get(ctx, key, crd)
if apierrors.IsNotFound(err) {
return false, nil
} else if err != nil {
@@ -136,7 +136,7 @@ func crdsAreReady(kbClient kbclient.Client, crds []*unstructured.Unstructured) (
// first CRD to determine whether to use the v1beta1 or v1 API during polling.
gvk := crds[0].GroupVersionKind()
var crdReadinessFn func() (bool, error)
var crdReadinessFn func(context.Context) (bool, error)
if gvk.Version == "v1beta1" {
crdReadinessFn = crdV1Beta1ReadinessFn(kbClient, crds)
} else if gvk.Version == "v1" {
@@ -145,7 +145,7 @@ func crdsAreReady(kbClient kbclient.Client, crds []*unstructured.Unstructured) (
return false, fmt.Errorf("unsupported CRD version %q", gvk.Version)
}
err := wait.PollImmediate(time.Second, time.Minute, crdReadinessFn)
err := wait.PollUntilContextTimeout(context.Background(), time.Second, time.Minute, true, crdReadinessFn)
if err != nil {
return false, errors.Wrap(err, "Error polling for CRDs")
}
@@ -178,7 +178,7 @@ func DeploymentIsReady(factory client.DynamicFactory, namespace string) (bool, e
// declare this variable out of scope so we can return it
var isReady bool
var readyObservations int32
err = wait.PollImmediate(time.Second, 3*time.Minute, func() (bool, error) {
err = wait.PollUntilContextTimeout(context.Background(), time.Second, 3*time.Minute, true, func(ctx context.Context) (bool, error) {
unstructuredDeployment, err := c.Get("velero", metav1.GetOptions{})
if apierrors.IsNotFound(err) {
return false, nil
@@ -224,7 +224,7 @@ func DaemonSetIsReady(factory client.DynamicFactory, namespace string) (bool, er
var isReady bool
var readyObservations int32
err = wait.PollImmediate(time.Second, time.Minute, func() (bool, error) {
err = wait.PollUntilContextTimeout(context.Background(), time.Second, time.Minute, true, func(ctx context.Context) (bool, error) {
unstructuredDaemonSet, err := c.Get("node-agent", metav1.GetOptions{})
if apierrors.IsNotFound(err) {
return false, nil
@@ -339,7 +339,7 @@ func Install(dynamicFactory client.DynamicFactory, kbClient kbclient.Client, res
// Wait for CRDs to be ready before proceeding
fmt.Fprint(w, "Waiting for resources to be ready in cluster...\n")
_, err := crdsAreReady(kbClient, rg.CRDResources)
if err == wait.ErrWaitTimeout {
if wait.Interrupted(err) {
return errors.Errorf("timeout reached, CRDs not ready")
} else if err != nil {
return err
pkg/persistence/mocks/backup_store.go
+1 -1
View File
@@ -21,7 +21,7 @@ import (
io "io"
mock "github.com/stretchr/testify/mock"
volumesnapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
volumesnapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
internalVolume "github.com/vmware-tanzu/velero/internal/volume"
itemoperation "github.com/vmware-tanzu/velero/pkg/itemoperation"
pkg/persistence/object_store.go
+1 -1
View File
@@ -23,7 +23,7 @@ import (
"strings"
"time"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
pkg/persistence/object_store_test.go
+1 -1
View File
@@ -27,7 +27,7 @@ import (
"strings"
"testing"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
pkg/podexec/pod_command_executor.go
+2 -1
View File
@@ -18,6 +18,7 @@ package podexec
import (
"bytes"
"context"
"net/url"
"time"
@@ -159,7 +160,7 @@ func (e *defaultPodCommandExecutor) ExecutePodCommand(log logrus.FieldLogger, it
errCh := make(chan error)
go func() {
err = executor.Stream(streamOptions)
err = executor.StreamWithContext(context.Background(), streamOptions)
errCh <- err
}()
pkg/podexec/pod_command_executor_test.go
+4 -3
View File
@@ -18,6 +18,7 @@ package podexec
import (
"bytes"
"context"
"fmt"
"net/url"
"strings"
@@ -231,7 +232,7 @@ func TestExecutePodCommand(t *testing.T) {
Stdout: &stdout,
Stderr: &stderr,
}
streamExecutor.On("Stream", expectedStreamOptions).Return(test.hookError)
streamExecutor.On("StreamWithContext", mock.Anything, expectedStreamOptions).Return(test.hookError)
err = podCommandExecutor.ExecutePodCommand(velerotest.NewLogger(), pod, "namespace", "name", "hookName", &hook)
if test.expectedError != "" {
@@ -307,8 +308,8 @@ type mockStreamExecutor struct {
remotecommand.Executor
}
func (e *mockStreamExecutor) Stream(options remotecommand.StreamOptions) error {
args := e.Called(options)
func (e *mockStreamExecutor) StreamWithContext(ctx context.Context, options remotecommand.StreamOptions) error {
args := e.Called(ctx, options)
return args.Error(0)
}
pkg/podvolume/backupper.go
+3 -3
View File
@@ -117,7 +117,7 @@ func newBackupper(
results: make(map[string]chan *velerov1api.PodVolumeBackup),
}
pvbInformer.AddEventHandler(
_, _ = pvbInformer.AddEventHandler(
cache.ResourceEventHandlerFuncs{
UpdateFunc: func(_, obj interface{}) {
pvb := obj.(*velerov1api.PodVolumeBackup)
@@ -225,8 +225,8 @@ func (b *backupper) BackupPodVolumes(backup *velerov1api.Backup, pod *corev1api.
var (
podVolumeBackups []*velerov1api.PodVolumeBackup
mountedPodVolumes = sets.String{}
attachedPodDevices = sets.String{}
mountedPodVolumes = sets.Set[string]{}
attachedPodDevices = sets.Set[string]{}
)
for _, container := range pod.Spec.Containers {
pkg/podvolume/restorer.go
+3 -3
View File
@@ -88,7 +88,7 @@ func newRestorer(
log: log,
}
pvrInformer.AddEventHandler(
_, _ = pvrInformer.AddEventHandler(
cache.ResourceEventHandlerFuncs{
UpdateFunc: func(_, obj interface{}) {
pvr := obj.(*velerov1api.PodVolumeRestore)
@@ -205,8 +205,8 @@ func (r *restorer) RestorePodVolumes(data RestoreData) []error {
return true, nil
}
err := wait.PollWithContext(checkCtx, time.Millisecond*500, time.Minute*10, checkFunc)
if err == wait.ErrWaitTimeout {
err := wait.PollUntilContextTimeout(checkCtx, time.Millisecond*500, time.Minute*10, true, checkFunc)
if wait.Interrupted(err) {
r.log.WithError(err).Error("Restoring pod is not scheduled until timeout or cancel, disengage")
} else if err != nil {
r.log.WithError(err).Error("Failed to check node-agent pod status, disengage")
pkg/repository/ensurer.go
+1 -1
View File
@@ -129,7 +129,7 @@ func (r *Ensurer) waitBackupRepository(ctx context.Context, namespace string, ba
}
}
err := wait.PollWithContext(ctx, time.Millisecond*500, r.resourceTimeout, checkFunc)
err := wait.PollUntilContextTimeout(ctx, time.Millisecond*500, r.resourceTimeout, true, checkFunc)
if err != nil {
return nil, errors.Wrap(err, "failed to wait BackupRepository")
}
pkg/repository/ensurer_test.go
+3 -3
View File
@@ -102,7 +102,7 @@ func TestEnsureRepo(t *testing.T) {
bkRepoObjNotReady,
},
runtimeScheme: scheme,
err: "failed to wait BackupRepository: timed out waiting for the condition",
err: "failed to wait BackupRepository: context deadline exceeded",
},
{
name: "create fail",
@@ -110,7 +110,7 @@ func TestEnsureRepo(t *testing.T) {
bsl: "fake-bsl",
repositoryType: "fake-repo-type",
runtimeScheme: scheme,
err: "failed to wait BackupRepository: timed out waiting for the condition",
err: "failed to wait BackupRepository: context deadline exceeded",
},
}
@@ -183,7 +183,7 @@ func TestCreateBackupRepositoryAndWait(t *testing.T) {
bsl: "fake-bsl",
repositoryType: "fake-repo-type",
runtimeScheme: scheme,
err: "failed to wait BackupRepository: timed out waiting for the condition",
err: "failed to wait BackupRepository: context deadline exceeded",
},
}
pkg/restore/request.go
+1 -1
View File
@@ -21,7 +21,7 @@ import (
"io"
"sort"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
"github.com/sirupsen/logrus"
"k8s.io/apimachinery/pkg/runtime"
pkg/restore/restore.go
+16 -16
View File
@@ -30,7 +30,7 @@ import (
"time"
"github.com/google/uuid"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
v1 "k8s.io/api/core/v1"
@@ -288,7 +288,7 @@ func (kr *kubernetesRestorer) RestoreWithResolvers(
resourceIncludesExcludes: resourceIncludesExcludes,
resourceStatusIncludesExcludes: restoreStatusIncludesExcludes,
namespaceIncludesExcludes: namespaceIncludesExcludes,
resourceMustHave: sets.NewString(resourceMustHave...),
resourceMustHave: sets.New[string](resourceMustHave...),
chosenGrpVersToRestore: make(map[string]ChosenGroupVersion),
selector: selector,
OrSelectors: OrSelectors,
@@ -300,7 +300,7 @@ func (kr *kubernetesRestorer) RestoreWithResolvers(
volumeSnapshotterGetter: volumeSnapshotterGetter,
podVolumeRestorer: podVolumeRestorer,
podVolumeErrs: make(chan error),
pvsToProvision: sets.NewString(),
pvsToProvision: sets.New[string](),
pvRestorer: pvRestorer,
volumeSnapshots: req.VolumeSnapshots,
csiVolumeSnapshots: req.CSIVolumeSnapshots,
@@ -338,7 +338,7 @@ type restoreContext struct {
resourceIncludesExcludes *collections.IncludesExcludes
resourceStatusIncludesExcludes *collections.IncludesExcludes
namespaceIncludesExcludes *collections.IncludesExcludes
resourceMustHave sets.String
resourceMustHave sets.Set[string]
chosenGrpVersToRestore map[string]ChosenGroupVersion
selector labels.Selector
OrSelectors []labels.Selector
@@ -351,7 +351,7 @@ type restoreContext struct {
podVolumeRestorer podvolume.Restorer
podVolumeWaitGroup sync.WaitGroup
podVolumeErrs chan error
pvsToProvision sets.String
pvsToProvision sets.Set[string]
pvRestorer PVRestorer
volumeSnapshots []*volume.Snapshot
csiVolumeSnapshots []*snapshotv1api.VolumeSnapshot
@@ -516,7 +516,7 @@ func (ctx *restoreContext) execute() (results.Result, results.Result) {
}()
// totalItems: previously discovered items, i: iteration counter.
totalItems, processedItems, existingNamespaces := 0, 0, sets.NewString()
totalItems, processedItems, existingNamespaces := 0, 0, sets.New[string]()
// First restore CRDs. This is needed so that they are available in the cluster
// when getOrderedResourceCollection is called again on the whole backup and
@@ -524,7 +524,7 @@ func (ctx *restoreContext) execute() (results.Result, results.Result) {
crdResourceCollection, processedResources, w, e := ctx.getOrderedResourceCollection(
backupResources,
make([]restoreableResource, 0),
sets.NewString(),
sets.New[string](),
Priorities{HighPriorities: []string{"customresourcedefinitions"}},
false,
)
@@ -710,7 +710,7 @@ func (ctx *restoreContext) processSelectedResource(
selectedResource restoreableResource,
totalItems int,
processedItems int,
existingNamespaces sets.String,
existingNamespaces sets.Set[string],
update chan progressUpdate,
) (int, results.Result, results.Result) {
warnings, errs := results.Result{}, results.Result{}
@@ -873,7 +873,7 @@ func (ctx *restoreContext) shouldRestore(name string, pvClient client.Dynamic) (
pvLogger := ctx.log.WithField("pvName", name)
var shouldRestore bool
err := wait.PollImmediate(time.Second, ctx.resourceTerminatingTimeout, func() (bool, error) {
err := wait.PollUntilContextTimeout(go_context.Background(), time.Second, ctx.resourceTerminatingTimeout, true, func(go_context.Context) (bool, error) {
unstructuredPV, err := pvClient.Get(name, metav1.GetOptions{})
if apierrors.IsNotFound(err) {
pvLogger.Debug("PV not found, safe to restore")
@@ -956,7 +956,7 @@ func (ctx *restoreContext) shouldRestore(name string, pvClient client.Dynamic) (
return true, nil
})
if err == wait.ErrWaitTimeout {
if wait.Interrupted(err) {
pvLogger.Warn("timeout reached waiting for persistent volume to delete")
}
@@ -970,7 +970,7 @@ func (ctx *restoreContext) crdAvailable(name string, crdClient client.Dynamic) (
var available bool
err := wait.PollImmediate(time.Second, ctx.resourceTimeout, func() (bool, error) {
err := wait.PollUntilContextTimeout(go_context.Background(), time.Second, ctx.resourceTimeout, true, func(ctx go_context.Context) (bool, error) {
unstructuredCRD, err := crdClient.Get(name, metav1.GetOptions{})
if err != nil {
return true, err
@@ -989,7 +989,7 @@ func (ctx *restoreContext) crdAvailable(name string, crdClient client.Dynamic) (
return available, nil
})
if err == wait.ErrWaitTimeout {
if wait.Interrupted(err) {
crdLogger.Debug("timeout reached waiting for custom resource definition to be ready")
}
@@ -1008,7 +1008,7 @@ func (ctx *restoreContext) itemsAvailable(action framework.RestoreItemResolvedAc
timeout = restoreItemOut.AdditionalItemsReadyTimeout
}
err := wait.PollImmediate(time.Second, timeout, func() (bool, error) {
err := wait.PollUntilContextTimeout(go_context.Background(), time.Second, timeout, true, func(go_context.Context) (bool, error) {
var err error
available, err = action.AreAdditionalItemsReady(restoreItemOut.AdditionalItems, ctx.restore)
@@ -1025,7 +1025,7 @@ func (ctx *restoreContext) itemsAvailable(action framework.RestoreItemResolvedAc
return available, nil
})
if err == wait.ErrWaitTimeout {
if wait.Interrupted(err) {
ctx.log.Debug("timeout reached waiting for AdditionalItems to be ready")
}
@@ -2153,10 +2153,10 @@ type restoreableItem struct {
func (ctx *restoreContext) getOrderedResourceCollection(
backupResources map[string]*archive.ResourceItems,
restoreResourceCollection []restoreableResource,
processedResources sets.String,
processedResources sets.Set[string],
resourcePriorities Priorities,
includeAllResources bool,
) ([]restoreableResource, sets.String, results.Result, results.Result) {
) ([]restoreableResource, sets.Set[string], results.Result, results.Result) {
var warnings, errs results.Result
// Iterate through an ordered list of resources to restore, checking each
// one to see if it should be restored. Note that resources *may* be in this
pkg/restore/restore_test.go
+6 -6
View File
@@ -25,7 +25,7 @@ import (
"testing"
"time"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
"github.com/pkg/errors"
"github.com/sirupsen/logrus"
"github.com/stretchr/testify/assert"
@@ -2238,7 +2238,7 @@ func TestShouldRestore(t *testing.T) {
),
},
want: false,
wantErr: errors.New("timed out waiting for the condition"),
wantErr: errors.New("context deadline exceeded"),
},
{
name: "when PV is found, has associated PVC that's not deleting, has associated NS that is terminating, result is false + timeout error",
@@ -2253,7 +2253,7 @@ func TestShouldRestore(t *testing.T) {
builder.ForNamespace("ns-1").Phase(corev1api.NamespaceTerminating).Result(),
},
want: false,
wantErr: errors.New("timed out waiting for the condition"),
wantErr: errors.New("context deadline exceeded"),
},
{
name: "when PV is found, has associated PVC that's not deleting, has associated NS that has deletion timestamp, result is false + timeout error",
@@ -2268,7 +2268,7 @@ func TestShouldRestore(t *testing.T) {
builder.ForNamespace("ns-1").ObjectMeta(builder.WithDeletionTimestamp(time.Now())).Result(),
},
want: false,
wantErr: errors.New("timed out waiting for the condition"),
wantErr: errors.New("context deadline exceeded"),
},
{
name: "when PV is found, associated PVC is not found, result is false + timeout error",
@@ -2279,7 +2279,7 @@ func TestShouldRestore(t *testing.T) {
),
},
want: false,
wantErr: errors.New("timed out waiting for the condition"),
wantErr: errors.New("context deadline exceeded"),
},
{
name: "when PV is found, has associated PVC, associated namespace not found, result is false + timeout error",
@@ -2291,7 +2291,7 @@ func TestShouldRestore(t *testing.T) {
test.PVCs(builder.ForPersistentVolumeClaim("ns-1", "pvc-1").Result()),
},
want: false,
wantErr: errors.New("timed out waiting for the condition"),
wantErr: errors.New("context deadline exceeded"),
},
}
pkg/test/fake_controller_runtime_client.go
+1 -1
View File
@@ -19,7 +19,7 @@ package test
import (
"testing"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
"github.com/stretchr/testify/require"
appsv1api "k8s.io/api/apps/v1"
corev1api "k8s.io/api/core/v1"
pkg/test/mocks.go
+2 -2
View File
@@ -1,8 +1,8 @@
package test
import (
snapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
snapshotv1listers "github.com/kubernetes-csi/external-snapshotter/client/v4/listers/volumesnapshot/v1"
snapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
snapshotv1listers "github.com/kubernetes-csi/external-snapshotter/client/v7/listers/volumesnapshot/v1"
"k8s.io/apimachinery/pkg/labels"
)
pkg/test/mocks/VolumeSnapshotLister.go
+2 -2
View File
@@ -6,9 +6,9 @@ import (
mock "github.com/stretchr/testify/mock"
labels "k8s.io/apimachinery/pkg/labels"
v1 "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
v1 "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
volumesnapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v4/listers/volumesnapshot/v1"
volumesnapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v7/listers/volumesnapshot/v1"
)
// VolumeSnapshotLister is an autogenerated mock type for the VolumeSnapshotLister type
pkg/util/csi/volume_snapshot.go
+7 -8
View File
@@ -33,8 +33,8 @@ import (
"github.com/vmware-tanzu/velero/pkg/util/stringptr"
"github.com/vmware-tanzu/velero/pkg/util/stringslice"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
snapshotter "github.com/kubernetes-csi/external-snapshotter/client/v4/clientset/versioned/typed/volumesnapshot/v1"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
snapshotter "github.com/kubernetes-csi/external-snapshotter/client/v7/clientset/versioned/typed/volumesnapshot/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -52,7 +52,7 @@ func WaitVolumeSnapshotReady(ctx context.Context, snapshotClient snapshotter.Sna
var updated *snapshotv1api.VolumeSnapshot
errMessage := sets.NewString()
err := wait.PollImmediate(waitInternal, timeout, func() (bool, error) {
err := wait.PollUntilContextTimeout(ctx, waitInternal, timeout, true, func(ctx context.Context) (bool, error) {
tmpVS, err := snapshotClient.VolumeSnapshots(volumeSnapshotNS).Get(ctx, volumeSnapshot, metav1.GetOptions{})
if err != nil {
return false, errors.Wrapf(err, fmt.Sprintf("error to get volumesnapshot %s/%s", volumeSnapshotNS, volumeSnapshot))
@@ -74,7 +74,7 @@ func WaitVolumeSnapshotReady(ctx context.Context, snapshotClient snapshotter.Sna
return true, nil
})
if err == wait.ErrWaitTimeout {
if wait.Interrupted(err) {
err = errors.Errorf("volume snapshot is not ready until timeout, errors: %v", errMessage.List())
}
@@ -132,7 +132,7 @@ func EnsureDeleteVS(ctx context.Context, snapshotClient snapshotter.SnapshotV1In
return errors.Wrap(err, "error to delete volume snapshot")
}
err = wait.PollImmediate(waitInternal, timeout, func() (bool, error) {
err = wait.PollUntilContextTimeout(ctx, waitInternal, timeout, true, func(ctx context.Context) (bool, error) {
_, err := snapshotClient.VolumeSnapshots(vsNamespace).Get(ctx, vsName, metav1.GetOptions{})
if err != nil {
if apierrors.IsNotFound(err) {
@@ -153,7 +153,7 @@ func EnsureDeleteVS(ctx context.Context, snapshotClient snapshotter.SnapshotV1In
}
func RemoveVSCProtect(ctx context.Context, snapshotClient snapshotter.SnapshotV1Interface, vscName string, timeout time.Duration) error {
err := wait.PollImmediate(waitInternal, timeout, func() (bool, error) {
err := wait.PollUntilContextTimeout(ctx, waitInternal, timeout, true, func(ctx context.Context) (bool, error) {
vsc, err := snapshotClient.VolumeSnapshotContents().Get(ctx, vscName, metav1.GetOptions{})
if err != nil {
return false, errors.Wrapf(err, "error to get VolumeSnapshotContent %s", vscName)
@@ -183,8 +183,7 @@ func EnsureDeleteVSC(ctx context.Context, snapshotClient snapshotter.SnapshotV1I
if err != nil && !apierrors.IsNotFound(err) {
return errors.Wrap(err, "error to delete volume snapshot content")
}
err = wait.PollImmediate(waitInternal, timeout, func() (bool, error) {
err = wait.PollUntilContextTimeout(ctx, waitInternal, timeout, true, func(ctx context.Context) (bool, error) {
_, err := snapshotClient.VolumeSnapshotContents().Get(ctx, vscName, metav1.GetOptions{})
if err != nil {
if apierrors.IsNotFound(err) {
pkg/util/csi/volume_snapshot_test.go
+3 -3
View File
@@ -22,8 +22,8 @@ import (
"testing"
"time"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v4/apis/volumesnapshot/v1"
snapshotFake "github.com/kubernetes-csi/external-snapshotter/client/v4/clientset/versioned/fake"
snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v7/apis/volumesnapshot/v1"
snapshotFake "github.com/kubernetes-csi/external-snapshotter/client/v7/clientset/versioned/fake"
"github.com/stretchr/testify/assert"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -705,7 +705,7 @@ func TestRemoveVSCProtect(t *testing.T) {
},
},
timeout: time.Second,
err: "timed out waiting for the condition",
err: "context deadline exceeded",
},
{
name: "succeed",
pkg/util/kube/event_handler.go
+13 -11
View File
@@ -17,6 +17,8 @@ limitations under the License.
package kube
import (
"context"
"k8s.io/client-go/util/workqueue"
"sigs.k8s.io/controller-runtime/pkg/client"
"sigs.k8s.io/controller-runtime/pkg/event"
@@ -24,7 +26,7 @@ import (
"sigs.k8s.io/controller-runtime/pkg/reconcile"
)
type MapUpdateFunc func(client.Object) []reconcile.Request
type MapUpdateFunc func(context.Context, client.Object) []reconcile.Request
// EnqueueRequestsFromMapUpdateFunc has the same purpose with handler.EnqueueRequestsFromMapFunc.
// MapUpdateFunc is simpler on Update event because mapAndEnqueue is called once with the new object. EnqueueRequestsFromMapFunc is called twice with the old and new object.
@@ -41,29 +43,29 @@ type enqueueRequestsFromMapFunc struct {
}
// Create implements EventHandler.
func (e *enqueueRequestsFromMapFunc) Create(evt event.CreateEvent, q workqueue.RateLimitingInterface) {
e.mapAndEnqueue(q, evt.Object)
func (e *enqueueRequestsFromMapFunc) Create(ctx context.Context, evt event.CreateEvent, q workqueue.RateLimitingInterface) {
e.mapAndEnqueue(ctx, q, evt.Object)
}
// Update implements EventHandler.
func (e *enqueueRequestsFromMapFunc) Update(evt event.UpdateEvent, q workqueue.RateLimitingInterface) {
e.mapAndEnqueue(q, evt.ObjectNew)
func (e *enqueueRequestsFromMapFunc) Update(ctx context.Context, evt event.UpdateEvent, q workqueue.RateLimitingInterface) {
e.mapAndEnqueue(ctx, q, evt.ObjectNew)
}
// Delete implements EventHandler.
func (e *enqueueRequestsFromMapFunc) Delete(evt event.DeleteEvent, q workqueue.RateLimitingInterface) {
e.mapAndEnqueue(q, evt.Object)
func (e *enqueueRequestsFromMapFunc) Delete(ctx context.Context, evt event.DeleteEvent, q workqueue.RateLimitingInterface) {
e.mapAndEnqueue(ctx, q, evt.Object)
}
// Generic implements EventHandler.
func (e *enqueueRequestsFromMapFunc) Generic(evt event.GenericEvent, q workqueue.RateLimitingInterface) {
e.mapAndEnqueue(q, evt.Object)
func (e *enqueueRequestsFromMapFunc) Generic(ctx context.Context, evt event.GenericEvent, q workqueue.RateLimitingInterface) {
e.mapAndEnqueue(ctx, q, evt.Object)
}
func (e *enqueueRequestsFromMapFunc) mapAndEnqueue(q workqueue.RateLimitingInterface, object client.Object) {
func (e *enqueueRequestsFromMapFunc) mapAndEnqueue(ctx context.Context, q workqueue.RateLimitingInterface, object client.Object) {
reqs := map[reconcile.Request]struct{}{}
for _, req := range e.toRequests(object) {
for _, req := range e.toRequests(ctx, object) {
_, ok := reqs[req]
if !ok {
q.Add(req)
pkg/util/kube/pod.go
+1 -1
View File
@@ -92,7 +92,7 @@ func EnsureDeletePod(ctx context.Context, podGetter corev1client.CoreV1Interface
return errors.Wrapf(err, "error to delete pod %s", pod)
}
err = wait.PollImmediate(waitInternal, timeout, func() (bool, error) {
err = wait.PollUntilContextTimeout(ctx, waitInternal, timeout, true, func(ctx context.Context) (bool, error) {
_, err := podGetter.Pods(namespace).Get(ctx, pod, metav1.GetOptions{})
if err != nil {
if apierrors.IsNotFound(err) {
pkg/util/kube/pvc_pv.go
+5 -4
View File
@@ -78,7 +78,7 @@ func DeletePVAndPVCIfAny(ctx context.Context, client corev1client.CoreV1Interfac
func WaitPVCBound(ctx context.Context, pvcGetter corev1client.CoreV1Interface,
pvGetter corev1client.CoreV1Interface, pvc string, namespace string, timeout time.Duration) (*corev1api.PersistentVolume, error) {
var updated *corev1api.PersistentVolumeClaim
err := wait.PollImmediate(waitInternal, timeout, func() (bool, error) {
err := wait.PollUntilContextTimeout(ctx, waitInternal, timeout, true, func(ctx context.Context) (bool, error) {
tmpPVC, err := pvcGetter.PersistentVolumeClaims(namespace).Get(ctx, pvc, metav1.GetOptions{})
if err != nil {
return false, errors.Wrapf(err, fmt.Sprintf("error to get pvc %s/%s", namespace, pvc))
@@ -124,7 +124,7 @@ func EnsureDeletePVC(ctx context.Context, pvcGetter corev1client.CoreV1Interface
return errors.Wrapf(err, "error to delete pvc %s", pvc)
}
err = wait.PollImmediate(waitInternal, timeout, func() (bool, error) {
err = wait.PollUntilContextTimeout(ctx, waitInternal, timeout, true, func(ctx context.Context) (bool, error) {
_, err := pvcGetter.PersistentVolumeClaims(namespace).Get(ctx, pvc, metav1.GetOptions{})
if err != nil {
if apierrors.IsNotFound(err) {
@@ -262,7 +262,8 @@ func WaitPVCConsumed(ctx context.Context, pvcGetter corev1client.CoreV1Interface
selectedNode := ""
var updated *corev1api.PersistentVolumeClaim
var storageClass *storagev1api.StorageClass
err := wait.PollImmediate(waitInternal, timeout, func() (bool, error) {
err := wait.PollUntilContextTimeout(ctx, waitInternal, timeout, true, func(ctx context.Context) (bool, error) {
tmpPVC, err := pvcGetter.PersistentVolumeClaims(namespace).Get(ctx, pvc, metav1.GetOptions{})
if err != nil {
return false, errors.Wrapf(err, "error to get pvc %s/%s", namespace, pvc)
@@ -299,7 +300,7 @@ func WaitPVCConsumed(ctx context.Context, pvcGetter corev1client.CoreV1Interface
// WaitPVBound wait for binding of a PV specified by name and returns the bound PV object
func WaitPVBound(ctx context.Context, pvGetter corev1client.CoreV1Interface, pvName string, pvcName string, pvcNamespace string, timeout time.Duration) (*corev1api.PersistentVolume, error) {
var updated *corev1api.PersistentVolume
err := wait.PollImmediate(waitInternal, timeout, func() (bool, error) {
err := wait.PollUntilContextTimeout(ctx, waitInternal, timeout, true, func(ctx context.Context) (bool, error) {
tmpPV, err := pvGetter.PersistentVolumes().Get(ctx, pvName, metav1.GetOptions{})
if err != nil {
return false, errors.Wrapf(err, fmt.Sprintf("failed to get pv %s", pvName))
pkg/util/kube/pvc_pv_test.go
+4 -4
View File
@@ -88,7 +88,7 @@ func TestWaitPVCBound(t *testing.T) {
kubeClientObj: []runtime.Object{
pvcObject,
},
err: "error to wait for rediness of PVC: timed out waiting for the condition",
err: "error to wait for rediness of PVC: context deadline exceeded",
},
{
name: "get pv fail",
@@ -249,7 +249,7 @@ func TestWaitPVCConsumed(t *testing.T) {
pvcObjectWithSC,
scObjWaitBind,
},
err: "error to wait for PVC: timed out waiting for the condition",
err: "error to wait for PVC: context deadline exceeded",
},
{
name: "success on sc without wait binding mode",
@@ -955,7 +955,7 @@ func TestWaitPVBound(t *testing.T) {
},
},
},
err: "error to wait for bound of PV: timed out waiting for the condition",
err: "error to wait for bound of PV: context deadline exceeded",
},
{
name: "pvc status not bound",
@@ -967,7 +967,7 @@ func TestWaitPVBound(t *testing.T) {
},
},
},
err: "error to wait for bound of PV: timed out waiting for the condition",
err: "error to wait for bound of PV: context deadline exceeded",
},
{
name: "pvc claimRef pvc name mismatch",
pkg/util/kube/utils.go
+2 -2
View File
@@ -72,8 +72,8 @@ func EnsureNamespaceExistsAndIsReady(namespace *corev1api.Namespace, client core
// required for keeping track of number of restored items
var nsCreated bool
var ready bool
err := wait.PollImmediate(time.Second, timeout, func() (bool, error) {
clusterNS, err := client.Get(context.TODO(), namespace.Name, metav1.GetOptions{})
err := wait.PollUntilContextTimeout(context.Background(), time.Second, timeout, true, func(ctx context.Context) (bool, error) {
clusterNS, err := client.Get(ctx, namespace.Name, metav1.GetOptions{})
if apierrors.IsNotFound(err) {
// Namespace isn't in cluster, we're good to create.