diff --git a/pkg/controller/backup_deletion_controller.go b/pkg/controller/backup_deletion_controller.go index 87662c506a..cde32aceda 100644 --- a/pkg/controller/backup_deletion_controller.go +++ b/pkg/controller/backup_deletion_controller.go @@ -102,7 +102,7 @@ func NewBackupDeletionReconciler( func (r *backupDeletionReconciler) SetupWithManager(mgr ctrl.Manager) error { // Make sure the expired requests can be deleted eventually - s := kube.NewPeriodicalEnqueueSource(r.logger, mgr.GetClient(), &velerov1api.DeleteBackupRequestList{}, time.Hour, kube.PeriodicalEnqueueSourceOption{}) + s := kube.NewPeriodicalEnqueueSource("backupDeletion", r.logger, mgr.GetClient(), &velerov1api.DeleteBackupRequestList{}, time.Hour, kube.PeriodicalEnqueueSourceOption{}) return ctrl.NewControllerManagedBy(mgr). For(&velerov1api.DeleteBackupRequest{}). WatchesRawSource(s, nil). diff --git a/pkg/controller/backup_operations_controller.go b/pkg/controller/backup_operations_controller.go index 73836a0d33..581d1941bc 100644 --- a/pkg/controller/backup_operations_controller.go +++ b/pkg/controller/backup_operations_controller.go @@ -84,7 +84,7 @@ func NewBackupOperationsReconciler( } func (c *backupOperationsReconciler) SetupWithManager(mgr ctrl.Manager) error { - s := kube.NewPeriodicalEnqueueSource(c.logger, mgr.GetClient(), &velerov1api.BackupList{}, c.frequency, kube.PeriodicalEnqueueSourceOption{}) + s := kube.NewPeriodicalEnqueueSource("backupOperations", c.logger, mgr.GetClient(), &velerov1api.BackupList{}, c.frequency, kube.PeriodicalEnqueueSourceOption{}) gp := kube.NewGenericEventPredicate(func(object client.Object) bool { backup := object.(*velerov1api.Backup) return (backup.Status.Phase == velerov1api.BackupPhaseWaitingForPluginOperations || diff --git a/pkg/controller/backup_repository_controller.go b/pkg/controller/backup_repository_controller.go index 0bc457a172..7e2e1eca1b 100644 --- a/pkg/controller/backup_repository_controller.go +++ b/pkg/controller/backup_repository_controller.go @@ -75,7 +75,7 @@ func NewBackupRepoReconciler(namespace string, logger logrus.FieldLogger, client } func (r *BackupRepoReconciler) SetupWithManager(mgr ctrl.Manager) error { - s := kube.NewPeriodicalEnqueueSource(r.logger, mgr.GetClient(), &velerov1api.BackupRepositoryList{}, repoSyncPeriod, kube.PeriodicalEnqueueSourceOption{}) + s := kube.NewPeriodicalEnqueueSource("backupRepo", r.logger, mgr.GetClient(), &velerov1api.BackupRepositoryList{}, repoSyncPeriod, kube.PeriodicalEnqueueSourceOption{}) return ctrl.NewControllerManagedBy(mgr). For(&velerov1api.BackupRepository{}, builder.WithPredicates(kube.SpecChangePredicate{})). diff --git a/pkg/controller/backup_storage_location_controller.go b/pkg/controller/backup_storage_location_controller.go index b10126eacb..66eb747b70 100644 --- a/pkg/controller/backup_storage_location_controller.go +++ b/pkg/controller/backup_storage_location_controller.go @@ -191,6 +191,7 @@ func (r *backupStorageLocationReconciler) logReconciledPhase(defaultFound bool, func (r *backupStorageLocationReconciler) SetupWithManager(mgr ctrl.Manager) error { g := kube.NewPeriodicalEnqueueSource( + "backupStorageLocation", r.log, mgr.GetClient(), &velerov1api.BackupStorageLocationList{}, diff --git a/pkg/controller/backup_sync_controller.go b/pkg/controller/backup_sync_controller.go index 5a9b445e6d..ac05a15433 100644 --- a/pkg/controller/backup_sync_controller.go +++ b/pkg/controller/backup_sync_controller.go @@ -348,6 +348,7 @@ func (b *backupSyncReconciler) filterBackupOwnerReferences(ctx context.Context, // SetupWithManager is used to setup controller and its watching sources. func (b *backupSyncReconciler) SetupWithManager(mgr ctrl.Manager) error { backupSyncSource := kube.NewPeriodicalEnqueueSource( + "backupSync", b.logger, mgr.GetClient(), &velerov1api.BackupStorageLocationList{}, diff --git a/pkg/controller/data_download_controller.go b/pkg/controller/data_download_controller.go index 7d6f6e0c6d..5921697d56 100644 --- a/pkg/controller/data_download_controller.go +++ b/pkg/controller/data_download_controller.go @@ -497,7 +497,7 @@ func (r *DataDownloadReconciler) OnDataDownloadProgress(ctx context.Context, nam // re-enqueue the previous related request once the related pod is in running status to keep going on the rest logic. and below logic will avoid handling the unwanted // pod status and also avoid block others CR handling func (r *DataDownloadReconciler) SetupWithManager(mgr ctrl.Manager) error { - s := kube.NewPeriodicalEnqueueSource(r.logger, r.client, &velerov2alpha1api.DataDownloadList{}, preparingMonitorFrequency, kube.PeriodicalEnqueueSourceOption{}) + s := kube.NewPeriodicalEnqueueSource("dataDownload", r.logger, r.client, &velerov2alpha1api.DataDownloadList{}, preparingMonitorFrequency, kube.PeriodicalEnqueueSourceOption{}) gp := kube.NewGenericEventPredicate(func(object client.Object) bool { dd := object.(*velerov2alpha1api.DataDownload) return (dd.Status.Phase == velerov2alpha1api.DataDownloadPhaseAccepted) diff --git a/pkg/controller/data_upload_controller.go b/pkg/controller/data_upload_controller.go index 5b3169621f..39e9c73da0 100644 --- a/pkg/controller/data_upload_controller.go +++ b/pkg/controller/data_upload_controller.go @@ -534,7 +534,7 @@ func (r *DataUploadReconciler) OnDataUploadProgress(ctx context.Context, namespa // re-enqueue the previous related request once the related pod is in running status to keep going on the rest logic. and below logic will avoid handling the unwanted // pod status and also avoid block others CR handling func (r *DataUploadReconciler) SetupWithManager(mgr ctrl.Manager) error { - s := kube.NewPeriodicalEnqueueSource(r.logger, r.client, &velerov2alpha1api.DataUploadList{}, preparingMonitorFrequency, kube.PeriodicalEnqueueSourceOption{}) + s := kube.NewPeriodicalEnqueueSource("dataUpload", r.logger, r.client, &velerov2alpha1api.DataUploadList{}, preparingMonitorFrequency, kube.PeriodicalEnqueueSourceOption{}) gp := kube.NewGenericEventPredicate(func(object client.Object) bool { du := object.(*velerov2alpha1api.DataUpload) return (du.Status.Phase == velerov2alpha1api.DataUploadPhaseAccepted) diff --git a/pkg/controller/download_request_controller.go b/pkg/controller/download_request_controller.go index 22e9d87075..3a2a721327 100644 --- a/pkg/controller/download_request_controller.go +++ b/pkg/controller/download_request_controller.go @@ -218,7 +218,7 @@ func (r *downloadRequestReconciler) Reconcile(ctx context.Context, req ctrl.Requ } func (r *downloadRequestReconciler) SetupWithManager(mgr ctrl.Manager) error { - downloadRequestSource := kube.NewPeriodicalEnqueueSource(r.log, mgr.GetClient(), + downloadRequestSource := kube.NewPeriodicalEnqueueSource("downloadRequest", r.log, mgr.GetClient(), &velerov1api.DownloadRequestList{}, defaultDownloadRequestSyncPeriod, kube.PeriodicalEnqueueSourceOption{}) downloadRequestPredicates := kube.NewGenericEventPredicate(func(object kbclient.Object) bool { downloadRequest := object.(*velerov1api.DownloadRequest) diff --git a/pkg/controller/gc_controller.go b/pkg/controller/gc_controller.go index 7f464523c1..a5d5f2bdf4 100644 --- a/pkg/controller/gc_controller.go +++ b/pkg/controller/gc_controller.go @@ -75,7 +75,7 @@ func NewGCReconciler( // Other Events will be filtered to decrease the number of reconcile call. Especially UpdateEvent must be filtered since we removed // the backup status as the sub-resource of backup in v1.9, every change on it will be treated as UpdateEvent and trigger reconcile call. func (c *gcReconciler) SetupWithManager(mgr ctrl.Manager) error { - s := kube.NewPeriodicalEnqueueSource(c.logger, mgr.GetClient(), &velerov1api.BackupList{}, c.frequency, kube.PeriodicalEnqueueSourceOption{}) + s := kube.NewPeriodicalEnqueueSource("gc", c.logger, mgr.GetClient(), &velerov1api.BackupList{}, c.frequency, kube.PeriodicalEnqueueSourceOption{}) return ctrl.NewControllerManagedBy(mgr). For(&velerov1api.Backup{}, builder.WithPredicates(predicate.Funcs{ UpdateFunc: func(ue event.UpdateEvent) bool { diff --git a/pkg/controller/restore_operations_controller.go b/pkg/controller/restore_operations_controller.go index 24d4698a56..c70de6be05 100644 --- a/pkg/controller/restore_operations_controller.go +++ b/pkg/controller/restore_operations_controller.go @@ -82,7 +82,7 @@ func NewRestoreOperationsReconciler( } func (r *restoreOperationsReconciler) SetupWithManager(mgr ctrl.Manager) error { - s := kube.NewPeriodicalEnqueueSource(r.logger, mgr.GetClient(), &velerov1api.RestoreList{}, r.frequency, kube.PeriodicalEnqueueSourceOption{}) + s := kube.NewPeriodicalEnqueueSource("restoreOperations", r.logger, mgr.GetClient(), &velerov1api.RestoreList{}, r.frequency, kube.PeriodicalEnqueueSourceOption{}) gp := kube.NewGenericEventPredicate(func(object client.Object) bool { restore := object.(*velerov1api.Restore) return (restore.Status.Phase == velerov1api.RestorePhaseWaitingForPluginOperations || diff --git a/pkg/controller/schedule_controller.go b/pkg/controller/schedule_controller.go index da15cdabf8..d1de7e5f09 100644 --- a/pkg/controller/schedule_controller.go +++ b/pkg/controller/schedule_controller.go @@ -69,7 +69,7 @@ func NewScheduleReconciler( } func (c *scheduleReconciler) SetupWithManager(mgr ctrl.Manager) error { - s := kube.NewPeriodicalEnqueueSource(c.logger, mgr.GetClient(), &velerov1.ScheduleList{}, scheduleSyncPeriod, kube.PeriodicalEnqueueSourceOption{}) + s := kube.NewPeriodicalEnqueueSource("schedule", c.logger, mgr.GetClient(), &velerov1.ScheduleList{}, scheduleSyncPeriod, kube.PeriodicalEnqueueSourceOption{}) return ctrl.NewControllerManagedBy(mgr). // global predicate, works for both For and Watch WithEventFilter(kube.NewAllEventPredicate(func(obj client.Object) bool { diff --git a/pkg/util/kube/periodical_enqueue_source.go b/pkg/util/kube/periodical_enqueue_source.go index b4c4f27b8e..e1da70dd4d 100644 --- a/pkg/util/kube/periodical_enqueue_source.go +++ b/pkg/util/kube/periodical_enqueue_source.go @@ -36,13 +36,16 @@ import ( ) func NewPeriodicalEnqueueSource( + controllerName string, logger logrus.FieldLogger, client client.Client, objList client.ObjectList, period time.Duration, option PeriodicalEnqueueSourceOption) *PeriodicalEnqueueSource { return &PeriodicalEnqueueSource{ - logger: logger.WithField("resource", reflect.TypeOf(objList).String()), + logger: logger. + WithField("resource", reflect.TypeOf(objList).String()). + WithField("controller", controllerName), Client: client, objList: objList, period: period, diff --git a/pkg/util/kube/periodical_enqueue_source_test.go b/pkg/util/kube/periodical_enqueue_source_test.go index a8b18c2291..b9028cc2cb 100644 --- a/pkg/util/kube/periodical_enqueue_source_test.go +++ b/pkg/util/kube/periodical_enqueue_source_test.go @@ -42,7 +42,7 @@ func TestStart(t *testing.T) { ctx, cancelFunc := context.WithCancel(context.TODO()) client := (&fake.ClientBuilder{}).Build() queue := workqueue.NewRateLimitingQueue(workqueue.DefaultItemBasedRateLimiter()) - source := NewPeriodicalEnqueueSource(logrus.WithContext(ctx), client, &velerov1.ScheduleList{}, 1*time.Second, PeriodicalEnqueueSourceOption{}) + source := NewPeriodicalEnqueueSource("PES_TEST", logrus.WithContext(ctx), client, &velerov1.ScheduleList{}, 1*time.Second, PeriodicalEnqueueSourceOption{}) require.NoError(t, source.Start(ctx, nil, queue)) @@ -75,6 +75,7 @@ func TestPredicate(t *testing.T) { client := (&fake.ClientBuilder{}).Build() queue := workqueue.NewRateLimitingQueue(workqueue.DefaultItemBasedRateLimiter()) source := NewPeriodicalEnqueueSource( + "PES_TEST", logrus.WithContext(ctx), client, &velerov1.BackupStorageLocationList{}, @@ -115,6 +116,7 @@ func TestOrder(t *testing.T) { client := (&fake.ClientBuilder{}).Build() queue := workqueue.NewRateLimitingQueue(workqueue.DefaultItemBasedRateLimiter()) source := NewPeriodicalEnqueueSource( + "PES_TEST", logrus.WithContext(ctx), client, &velerov1.BackupStorageLocationList{},