add OperatorGroup reconciliation

Signed-off-by: akihikokuroda <[email protected]>

Author: akihikokuroda

pkg/controller/operators/catalog/installplan_sync.go

@@ -74,3 +74,50 @@ func (o *Operator) triggerInstallPlanRetry(obj interface{}) (syncError error) {
 	syncError = utilerrors.NewAggregate(errs)
 	return
 }
+
+// When the operatorgroup is created after the installplan creation, the installplan must be resync
+func (o *Operator) triggerInstallPlanUpdateForOperatorGroup(obj interface{}) (syncError error) {
+	metaObj, ok := obj.(metav1.Object)
+	if !ok {
+		syncError = errors.New("casting to metav1 object failed")
+		o.logger.Warn(syncError.Error())
+		return
+	}
+
+	ips, err := o.lister.OperatorsV1alpha1().InstallPlanLister().InstallPlans(metaObj.GetNamespace()).List(labels.Everything())
+	if err != nil {
+		syncError = err
+		return
+	}
+
+	update := func(ip *v1alpha1.InstallPlan) error {
+		out := ip.DeepCopy()
+		out.Status.Phase = v1alpha1.InstallPlanPhaseInstalling
+		now := o.now()
+		out.Status.SetCondition(v1alpha1.ConditionFailed(v1alpha1.InstallPlanInstalled, v1alpha1.InstallPlanReasonComponentFailed, "OperatorGroup updated", &now))
+		_, err := o.client.OperatorsV1alpha1().InstallPlans(ip.GetNamespace()).UpdateStatus(context.TODO(), out, metav1.UpdateOptions{})
+
+		return err
+	}
+
+	var errs []error
+	for _, ip := range ips {
+
+		logger := o.logger.WithFields(logrus.Fields{
+			"ip":        ip.GetName(),
+			"namespace": ip.GetNamespace(),
+			"phase":     ip.Status.Phase,
+		})
+
+		if updateErr := update(ip); updateErr != nil {
+			errs = append(errs, updateErr)
+			logger.WithError(updateErr).Warn("failed to kick off InstallPlan retry")
+			continue
+		}
+
+		logger.Info("InstallPlan condition message set to 'OperatorGroup updated' for retry")
+	}
+
+	syncError = utilerrors.NewAggregate(errs)
+	return
+}

pkg/controller/operators/catalog/operator.go

@@ -102,6 +102,7 @@ type Operator struct {
 	catsrcQueueSet           *queueinformer.ResourceQueueSet
 	subQueueSet              *queueinformer.ResourceQueueSet
 	ipQueueSet               *queueinformer.ResourceQueueSet
+	ogQueueSet               *queueinformer.ResourceQueueSet
 	nsResolveQueue           workqueue.RateLimitingInterface
 	namespace                string
 	recorder                 record.EventRecorder
@@ -179,6 +180,7 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 		catsrcQueueSet:           queueinformer.NewEmptyResourceQueueSet(),
 		subQueueSet:              queueinformer.NewEmptyResourceQueueSet(),
 		ipQueueSet:               queueinformer.NewEmptyResourceQueueSet(),
+		ogQueueSet:               queueinformer.NewEmptyResourceQueueSet(),
 		catalogSubscriberIndexer: map[string]cache.Indexer{},
 		serviceAccountQuerier:    scoped.NewUserDefinedServiceAccountQuerier(logger, crClient),
 		clientAttenuator:         scoped.NewClientAttenuator(logger, config, opClient),
@@ -272,6 +274,25 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 		return nil, err
 	}
 
+	// Wire OperatorGroup reconciliation
+	operatorGroupInformer := crInformerFactory.Operators().V1().OperatorGroups()
+	op.lister.OperatorsV1().RegisterOperatorGroupLister(metav1.NamespaceAll, operatorGroupInformer.Lister())
+	ogQueue := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "og")
+	op.ogQueueSet.Set(metav1.NamespaceAll, ogQueue)
+	operatorGroupQueueInformer, err := queueinformer.NewQueueInformer(
+		ctx,
+		queueinformer.WithLogger(op.logger),
+		queueinformer.WithQueue(ogQueue),
+		queueinformer.WithInformer(operatorGroupInformer.Informer()),
+		queueinformer.WithSyncer(queueinformer.LegacySyncHandler(op.syncOperatorGroup).ToSyncer()),
+	)
+	if err != nil {
+		return nil, err
+	}
+	if err := op.RegisterQueueInformer(operatorGroupQueueInformer); err != nil {
+		return nil, err
+	}
+
 	// Wire Subscriptions
 	subInformer := crInformerFactory.Operators().V1alpha1().Subscriptions()
 	op.lister.OperatorsV1alpha1().RegisterSubscriptionLister(metav1.NamespaceAll, subInformer.Lister())
@@ -526,10 +547,20 @@ func (o *Operator) syncObject(obj interface{}) (syncError error) {
 	}
 
 	o.requeueOwners(metaObj)
-
 	return o.triggerInstallPlanRetry(obj)
 }
 
+func (o *Operator) syncOperatorGroup(obj interface{}) (syncError error) {
+	// Assert as metav1.Object
+	_, ok := obj.(metav1.Object)
+	if !ok {
+		syncError = errors.New("casting to metav1 object failed")
+		o.logger.Warn(syncError.Error())
+		return
+	}
+	return o.triggerInstallPlanUpdateForOperatorGroup(obj)
+}
+
 func (o *Operator) handleDeletion(obj interface{}) {
 	metaObj, ok := obj.(metav1.Object)
 	if !ok {