feat(sub): add installplan reconciler to syncer

- Add InstallPlan status reconciler to Subscription Syncer
- Add InstallPlan Informer option to Subscription Syncer config
- Plumb now function through transitionInstallPlanState(...)

Author: njhale

pkg/controller/operators/catalog/operator.go

@@ -169,6 +169,7 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 			subscription.WithOperatorLister(op.lister),
 			subscription.WithSubscriptionInformer(subInformer.Informer()),
 			subscription.WithCatalogInformer(catsrcInformer.Informer()),
+			subscription.WithInstallPlanInformer(ipInformer.Informer()),
 			subscription.WithSubscriptionQueue(subQueue),
 			subscription.WithAppendedReconcilers(subscription.ReconcilerFromLegacySyncHandler(op.syncSubscriptions, nil)),
 			subscription.WithRegistryReconcilerFactory(op.reconciler),
@@ -919,7 +920,7 @@ func (o *Operator) syncInstallPlans(obj interface{}) (syncError error) {
 		return
 	}
 
-	outInstallPlan, syncError := transitionInstallPlanState(logger.Logger, o, *plan)
+	outInstallPlan, syncError := transitionInstallPlanState(logger.Logger, o, *plan, o.now())
 
 	if syncError != nil {
 		logger = logger.WithField("syncError", syncError)
@@ -964,7 +965,7 @@ type installPlanTransitioner interface {
 
 var _ installPlanTransitioner = &Operator{}
 
-func transitionInstallPlanState(log *logrus.Logger, transitioner installPlanTransitioner, in v1alpha1.InstallPlan) (*v1alpha1.InstallPlan, error) {
+func transitionInstallPlanState(log *logrus.Logger, transitioner installPlanTransitioner, in v1alpha1.InstallPlan, now metav1.Time) (*v1alpha1.InstallPlan, error) {
 	out := in.DeepCopy()
 
 	switch in.Status.Phase {
@@ -981,11 +982,11 @@ func transitionInstallPlanState(log *logrus.Logger, transitioner installPlanTran
 		log.Debug("attempting to install")
 		if err := transitioner.ExecutePlan(out); err != nil {
 			out.Status.SetCondition(v1alpha1.ConditionFailed(v1alpha1.InstallPlanInstalled,
-				v1alpha1.InstallPlanReasonComponentFailed, err))
+				v1alpha1.InstallPlanReasonComponentFailed, err.Error(), &now))
 			out.Status.Phase = v1alpha1.InstallPlanPhaseFailed
 			return out, err
 		}
-		out.Status.SetCondition(v1alpha1.ConditionMet(v1alpha1.InstallPlanInstalled))
+		out.Status.SetCondition(v1alpha1.ConditionMet(v1alpha1.InstallPlanInstalled, &now))
 		out.Status.Phase = v1alpha1.InstallPlanPhaseComplete
 		return out, nil
 	default:

pkg/controller/operators/catalog/operator_test.go

@@ -58,6 +58,8 @@ func (m *mockTransitioner) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 func TestTransitionInstallPlan(t *testing.T) {
 	errMsg := "transition test error"
 	err := errors.New(errMsg)
+	clockFake := utilclock.NewFakeClock(time.Date(2018, time.January, 26, 20, 40, 0, 0, time.UTC))
+	now := metav1.NewTime(clockFake.Now())
 
 	installed := &v1alpha1.InstallPlanCondition{
 		Type:   v1alpha1.InstallPlanInstalled,
@@ -103,7 +105,7 @@ func TestTransitionInstallPlan(t *testing.T) {
 		transitioner := &mockTransitioner{tt.transError}
 
 		// Attempt to transition phases.
-		out, _ := transitionInstallPlanState(logrus.New(), transitioner, *plan)
+		out, _ := transitionInstallPlanState(logrus.New(), transitioner, *plan, now)
 
 		// Assert that the final phase is as expected.
 		require.Equal(t, tt.expected, out.Status.Phase)

pkg/controller/operators/catalog/subscription/config.go

@@ -21,6 +21,7 @@ type syncerConfig struct {
 	lister                    operatorlister.OperatorLister
 	subscriptionInformer      cache.SharedIndexInformer
 	catalogInformer           cache.SharedIndexInformer
+	installPlanInformer       cache.SharedIndexInformer
 	subscriptionQueue         workqueue.RateLimitingInterface
 	reconcilers               kubestate.ReconcilerChain
 	registryReconcilerFactory reconciler.RegistryReconcilerFactory
@@ -72,13 +73,20 @@ func WithSubscriptionInformer(subscriptionInformer cache.SharedIndexInformer) Sy
 	}
 }
 
-// WithCatalogInformer sets the informer a syncer will wire dependent subscription notifications to.
+// WithCatalogInformer sets a CatalogSource informer to act as an event source for dependent Subscriptions.
 func WithCatalogInformer(catalogInformer cache.SharedIndexInformer) SyncerOption {
 	return func(config *syncerConfig) {
 		config.catalogInformer = catalogInformer
 	}
 }
 
+// WithInstallPlanInformer sets an InstallPlan informer to act as an event source for dependent Subscriptions.
+func WithInstallPlanInformer(installPlanInformer cache.SharedIndexInformer) SyncerOption {
+	return func(config *syncerConfig) {
+		config.installPlanInformer = installPlanInformer
+	}
+}
+
 // WithOperatorLister sets a syncer's operator lister.
 func WithOperatorLister(lister operatorlister.OperatorLister) SyncerOption {
 	return func(config *syncerConfig) {
@@ -138,6 +146,8 @@ func (s *syncerConfig) validate() (err error) {
 		err = newInvalidConfigError("nil subscription informer")
 	case s.catalogInformer == nil:
 		err = newInvalidConfigError("nil catalog informer")
+	case s.installPlanInformer == nil:
+		err = newInvalidConfigError("nil installplan informer")
 	case s.subscriptionQueue == nil:
 		err = newInvalidConfigError("nil subscription queue")
 	case len(s.reconcilers) == 0:

pkg/controller/operators/catalog/subscription/syncer.go

@@ -12,7 +12,9 @@ import (
 
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/apis/operators/install"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/apis/operators/v1alpha1"
+	listers "github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/listers/operators/v1alpha1"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/kubestate"
+	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/ownerutil"
 )
 
 var scheme = runtime.NewScheme()
@@ -28,6 +30,7 @@ type subscriptionSyncer struct {
 	clock                  utilclock.Clock
 	reconcilers            kubestate.ReconcilerChain
 	subscriptionCache      cache.Indexer
+	installPlanLister      listers.InstallPlanLister
 	globalCatalogNamespace string
 	notify                 kubestate.NotifyFunc
 }
@@ -96,9 +99,9 @@ func (s *subscriptionSyncer) catalogSubscriptionKeys(namespace string) ([]string
 	return keys, err
 }
 
-// catalogNotification notifies dependent subscriptions of the change with the given object.
-// The given object is assumed to be a Subscription, Subscription tombstone, or a cache.ExplicitKey.
-func (s *subscriptionSyncer) catalogNotification(ctx context.Context, obj interface{}) {
+// notifyOnCatalog notifies dependent subscriptions of the change with the given object.
+// The given object is assumed to be a CatalogSource, CatalogSource tombstone, or a cache.ExplicitKey.
+func (s *subscriptionSyncer) notifyOnCatalog(ctx context.Context, obj interface{}) {
 	k, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
 	if err != nil {
 		s.logger.WithField("resource", obj).Warn("could not unpack key")
@@ -127,6 +130,66 @@ func (s *subscriptionSyncer) catalogNotification(ctx context.Context, obj interf
 	logger.Trace("dependent subscriptions notified")
 }
 
+// notifyOnInstallPlan notifies dependent subscriptions of the change with the given object.
+// The given object is assumed to be an InstallPlan, InstallPlan tombstone, or a cache.ExplicitKey.
+func (s *subscriptionSyncer) notifyOnInstallPlan(ctx context.Context, obj interface{}) {
+	plan, ok := obj.(*v1alpha1.InstallPlan)
+	if !ok {
+		s.logger.WithField("obj", fmt.Sprintf("%v", obj)).Trace("could not cast as installplan directly while notifying subscription syncer")
+		if tombstone, ok := obj.(cache.DeletedFinalStateUnknown); ok {
+			if plan, ok = tombstone.Obj.(*v1alpha1.InstallPlan); !ok {
+				s.logger.WithField("tombstone", tombstone).Warn("could not cast as installplan")
+				return
+			}
+		} else {
+			k, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
+			if err != nil {
+				s.logger.WithField("resource", obj).Warn("could not unpack key")
+				return
+			}
+			logger := s.logger.WithField("key", k)
+
+			ns, name, err := cache.SplitMetaNamespaceKey(k)
+			if err != nil {
+				logger.Warn("could not split meta key")
+				return
+			}
+
+			if plan, err = s.installPlanLister.InstallPlans(ns).Get(name); err != nil {
+				logger.WithError(err).Warn("could not get installplan")
+				return
+			}
+		}
+	}
+
+	logger := s.logger.WithFields(logrus.Fields{
+		"namespace":   plan.GetNamespace(),
+		"installplan": plan.GetName(),
+	})
+
+	// Notify dependent owner Subscriptions
+	owners := ownerutil.GetOwnersByKind(plan, v1alpha1.SubscriptionKind)
+	for _, owner := range owners {
+		subKey := fmt.Sprintf("%s/%s", plan.GetNamespace(), owner.Name)
+		logger.Tracef("notifying subscription %s", subKey)
+		s.Notify(kubestate.NewResourceEvent(kubestate.ResourceUpdated, cache.ExplicitKey(subKey)))
+	}
+}
+
+func eventHandlers(ctx context.Context, notify func(ctx context.Context, obj interface{})) cache.ResourceEventHandlerFuncs {
+	return cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			notify(ctx, obj)
+		},
+		UpdateFunc: func(oldObj, newObj interface{}) {
+			notify(ctx, newObj)
+		},
+		DeleteFunc: func(obj interface{}) {
+			notify(ctx, obj)
+		},
+	}
+}
+
 // NewSyncer returns a syncer that syncs Subscription resources.
 func NewSyncer(ctx context.Context, options ...SyncerOption) (kubestate.Syncer, error) {
 	config := defaultSyncerConfig()
@@ -145,6 +208,7 @@ func newSyncerWithConfig(ctx context.Context, config *syncerConfig) (kubestate.S
 		clock:             config.clock,
 		reconcilers:       config.reconcilers,
 		subscriptionCache: config.subscriptionInformer.GetIndexer(),
+		installPlanLister: config.lister.OperatorsV1alpha1().InstallPlanLister(),
 		notify: func(event kubestate.ResourceEvent) {
 			// Notify Subscriptions by enqueuing to the Subscription queue.
 			config.subscriptionQueue.Add(event)
@@ -154,6 +218,11 @@ func newSyncerWithConfig(ctx context.Context, config *syncerConfig) (kubestate.S
 	// Build a reconciler chain from the default and configured reconcilers
 	// Default reconcilers should always come first in the chain
 	defaultReconcilers := kubestate.ReconcilerChain{
+		&installPlanReconciler{
+			now:               s.now,
+			client:            config.client,
+			installPlanLister: config.lister.OperatorsV1alpha1().InstallPlanLister(),
+		},
 		&catalogHealthReconciler{
 			now:                       s.now,
 			client:                    config.client,
@@ -165,17 +234,8 @@ func newSyncerWithConfig(ctx context.Context, config *syncerConfig) (kubestate.S
 	s.reconcilers = append(defaultReconcilers, s.reconcilers...)
 
 	// Add dependency notifications
-	config.catalogInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
-		AddFunc: func(obj interface{}) {
-			s.catalogNotification(ctx, obj)
-		},
-		UpdateFunc: func(oldObj, newObj interface{}) {
-			s.catalogNotification(ctx, newObj)
-		},
-		DeleteFunc: func(obj interface{}) {
-			s.catalogNotification(ctx, obj)
-		},
-	})
+	config.installPlanInformer.AddEventHandler(eventHandlers(ctx, s.notifyOnInstallPlan))
+	config.catalogInformer.AddEventHandler(eventHandlers(ctx, s.notifyOnCatalog))
 
 	return s, nil
 }