Add maximum retries in controller (#98165)

* Add maximum retries in controller

* Retry only for apierrors ServiceIsUnavailable

* Return error OnUpdate and OnCreate

* Trigger sync job for both again

* Remove initialized status

pkg/apis/provisioning/v0alpha1/types.go

@@ -122,10 +122,6 @@ type EditingOptions struct {
 // This is expected never to be created by a kubectl call or similar, and is expected to rarely (if ever) be edited manually.
 // As such, it is also a little less well structured than the spec, such as conditional-but-ever-present fields.
 type RepositoryStatus struct {
-	// FIXME: this is temporary field until we use different methods in the controller for create and update
-	// Initialized is true when the repository has been initialized
-	Initialized bool `json:"initialized"`
-
 	// The generation of the spec last time reconciliation ran
 	ObservedGeneration int64 `json:"observedGeneration"`
 

pkg/apis/provisioning/v0alpha1/zz_generated.openapi.go

@@ -877,14 +877,6 @@ func schema_pkg_apis_provisioning_v0alpha1_RepositoryStatus(ref common.Reference
 				Description: "The status of a Repository. This is expected never to be created by a kubectl call or similar, and is expected to rarely (if ever) be edited manually. As such, it is also a little less well structured than the spec, such as conditional-but-ever-present fields.",
 				Type:        []string{"object"},
 				Properties: map[string]spec.Schema{
-					"initialized": {
-						SchemaProps: spec.SchemaProps{
-							Description: "FIXME: this is temporary field until we use different methods in the controller for create and update Initialized is true when the repository has been initialized",
-							Default:     false,
-							Type:        []string{"boolean"},
-							Format:      "",
-						},
-					},
 					"observedGeneration": {
 						SchemaProps: spec.SchemaProps{
 							Description: "The generation of the spec last time reconciliation ran",
@@ -914,7 +906,7 @@ func schema_pkg_apis_provisioning_v0alpha1_RepositoryStatus(ref common.Reference
 						},
 					},
 				},
-				Required: []string{"initialized", "observedGeneration", "health", "sync", "webhook"},
+				Required: []string{"observedGeneration", "health", "sync", "webhook"},
 			},
 		},
 		Dependencies: []string{

pkg/generated/applyconfiguration/provisioning/v0alpha1/repositorystatus.go

@@ -7,7 +7,6 @@ package v0alpha1
 // RepositoryStatusApplyConfiguration represents a declarative configuration of the RepositoryStatus type for use
 // with apply.
 type RepositoryStatusApplyConfiguration struct {
-	Initialized        *bool                            `json:"initialized,omitempty"`
 	ObservedGeneration *int64                           `json:"observedGeneration,omitempty"`
 	Health             *HealthStatusApplyConfiguration  `json:"health,omitempty"`
 	Sync               *SyncStatusApplyConfiguration    `json:"sync,omitempty"`
@@ -20,14 +19,6 @@ func RepositoryStatus() *RepositoryStatusApplyConfiguration {
 	return &RepositoryStatusApplyConfiguration{}
 }
 
-// WithInitialized sets the Initialized field in the declarative configuration to the given value
-// and returns the receiver, so that objects can be built by chaining "With" function invocations.
-// If called multiple times, the Initialized field is set to the value of the last call.
-func (b *RepositoryStatusApplyConfiguration) WithInitialized(value bool) *RepositoryStatusApplyConfiguration {
-	b.Initialized = &value
-	return b
-}
-
 // WithObservedGeneration sets the ObservedGeneration field in the declarative configuration to the given value
 // and returns the receiver, so that objects can be built by chaining "With" function invocations.
 // If called multiple times, the ObservedGeneration field is set to the value of the last call.

pkg/registry/apis/provisioning/repository_controller.go

@@ -31,10 +31,13 @@ const (
 	operationDelete
 )
 
+const maxAttempts = 3
+
 type queueItem struct {
-	key string
-	op  operation
-	obj interface{}
+	key      string
+	op       operation
+	obj      interface{}
+	attempts int
 }
 
 // RepositoryController controls how and when CRD is established.
@@ -161,14 +164,15 @@ func (rc *RepositoryController) deleteRepository(obj interface{}) {
 
 // processNextWorkItem deals with one key off the queue.
 // It returns false when it's time to quit.
-func (rc *RepositoryController) processNextWorkItem(_ context.Context) bool {
+func (rc *RepositoryController) processNextWorkItem(ctx context.Context) bool {
 	item, quit := rc.queue.Get()
 	if quit {
 		return false
 	}
 	defer rc.queue.Done(item)
 
-	rc.logger.Info("RepositoryController processing key", "key", item)
+	logger := rc.logger.With("key", item.key)
+	logger.InfoContext(ctx, "RepositoryController processing key")
 
 	err := rc.processFn(item)
 	if err == nil {
@@ -176,6 +180,22 @@ func (rc *RepositoryController) processNextWorkItem(_ context.Context) bool {
 		return true
 	}
 
+	item.attempts++
+	logger = logger.With("error", err, "attempts", item.attempts)
+	if item.attempts >= maxAttempts {
+		logger.ErrorContext(ctx, "RepositoryController failed too many times")
+		rc.queue.Forget(item)
+		return true
+	}
+
+	if !apierrors.IsServiceUnavailable(err) {
+		logger.InfoContext(ctx, "RepositoryController will not retry")
+		rc.queue.Forget(item)
+		return true
+	} else {
+		logger.InfoContext(ctx, "RepositoryController will retry as service is unavailable")
+	}
+
 	utilruntime.HandleError(fmt.Errorf("%v failed with: %v", item, err))
 	rc.queue.AddRateLimited(item)
 
@@ -207,10 +227,10 @@ func (rc *RepositoryController) process(item *queueItem) error {
 	}
 
 	cachedRepo, err := rc.repoLister.Repositories(namespace).Get(name)
-	if apierrors.IsNotFound(err) {
+	switch {
+	case apierrors.IsNotFound(err):
 		return errors.New("repository not found in cache")
-	}
-	if err != nil {
+	case err != nil:
 		return err
 	}
 
@@ -235,24 +255,22 @@ func (rc *RepositoryController) process(item *queueItem) error {
 	logger.InfoContext(ctx, "repository spec changed", "previous", cachedRepo.Status.ObservedGeneration, "current", cachedRepo.Generation)
 
 	var status *provisioning.RepositoryStatus
-	if cachedRepo.Status.Initialized {
+	if cachedRepo.Status.ObservedGeneration > 0 {
 		logger.InfoContext(ctx, "handle repository update")
 		status, err = repo.OnUpdate(ctx, logger)
 		if err != nil {
-			rc.logger.Error("OnUpdate", "error", err)
+			return fmt.Errorf("handle repository update: %w", err)
 		}
 	} else {
 		logger.InfoContext(ctx, "handle repository init")
 		status, err = repo.OnCreate(ctx, logger)
 		if err != nil {
-			rc.logger.Error("OnCreate", "error", err)
+			return fmt.Errorf("handle repository create: %w", err)
 		}
 	}
 
 	if status == nil {
 		status = cachedRepo.Status.DeepCopy()
-	} else {
-		status.Initialized = true
 	}
 	status.ObservedGeneration = cachedRepo.Generation