Receive notification of CSV(s) being watched

OLM operator watches all CSV resource(s) across all namespace(s). We
need a mechanism to receive notification of CSV reconciliation events
for add/update/delete.

Add an interface that a caller can implement to tap into these events.

Author: tkashem

pkg/controller/operators/olm/operator.go

@@ -69,6 +69,7 @@ type Operator struct {
 	apiLabeler       labeler.Labeler
 	csvSetGenerator  csvutility.SetGenerator
 	csvReplaceFinder csvutility.ReplaceFinder
+	csvNotification  csvutility.WatchNotification
 }
 
 func NewOperator(ctx context.Context, options ...OperatorOption) (*Operator, error) {
@@ -401,6 +402,14 @@ func (a *Operator) GetReplaceFinder() csvutility.ReplaceFinder {
 	return a.csvReplaceFinder
 }
 
+func (a *Operator) RegisterCSVWatchNotification(csvNotification csvutility.WatchNotification) {
+	if csvNotification == nil {
+		return
+	}
+
+	a.csvNotification = csvNotification
+}
+
 func (a *Operator) syncObject(obj interface{}) (syncError error) {
 	// Assert as metav1.Object
 	metaObj, ok := obj.(metav1.Object)
@@ -481,6 +490,10 @@ func (a *Operator) handleClusterServiceVersionDeletion(obj interface{}) {
 		}
 	}
 
+	if a.csvNotification != nil {
+		a.csvNotification.OnDelete(clusterServiceVersion)
+	}
+
 	logger := a.logger.WithFields(logrus.Fields{
 		"id":        queueinformer.NewLoopID(),
 		"csv":       clusterServiceVersion.GetName(),
@@ -640,6 +653,10 @@ func (a *Operator) syncClusterServiceVersion(obj interface{}) (syncError error)
 	})
 	logger.Debug("syncing CSV")
 
+	if a.csvNotification != nil {
+		a.csvNotification.OnAddOrUpdate(clusterServiceVersion)
+	}
+
 	if clusterServiceVersion.IsCopied() {
 		logger.Debug("skipping copied csv transition, schedule for gc check")
 		a.csvGCQueueSet.Requeue(clusterServiceVersion.GetNamespace(), clusterServiceVersion.GetName())
@@ -1201,7 +1218,7 @@ func (a *Operator) transitionCSVState(in v1alpha1.ClusterServiceVersion) (out *v
 				}
 			}
 		} else {
-			syncError = fmt.Errorf("csv marked as replacement, but no replacmenet csv found in cluster")
+			syncError = fmt.Errorf("CSV marked as replacement, but no replacement CSV found in cluster.")
 		}
 	case v1alpha1.CSVPhaseDeleting:
 		syncError = a.client.OperatorsV1alpha1().ClusterServiceVersions(out.GetNamespace()).Delete(out.GetName(), metav1.NewDeleteOptions(0))

pkg/lib/csv/csvset.go

@@ -22,6 +22,7 @@ func NewSetGenerator(logger *logrus.Logger, lister operatorlister.OperatorLister
 // CSV name; if metav1.NamespaceAll gets the set across all namespaces
 type SetGenerator interface {
 	WithNamespace(namespace string, phase v1alpha1.ClusterServiceVersionPhase) map[string]*v1alpha1.ClusterServiceVersion
+	WithNamespaceAndLabels(namespace string, phase v1alpha1.ClusterServiceVersionPhase, selector labels.Selector) map[string]*v1alpha1.ClusterServiceVersion
 }
 
 type csvSet struct {
@@ -35,6 +36,12 @@ func (s *csvSet) WithNamespace(namespace string, phase v1alpha1.ClusterServiceVe
 	return s.with(namespace, phase, labels.Everything())
 }
 
+// WithNamespaceAndLabels returns all ClusterServiceVersion resource(s) that
+// matches the specified phase and label selector from a given namespace.
+func (s *csvSet) WithNamespaceAndLabels(namespace string, phase v1alpha1.ClusterServiceVersionPhase, selector labels.Selector) map[string]*v1alpha1.ClusterServiceVersion {
+	return s.with(namespace, phase, selector)
+}
+
 func (s *csvSet) with(namespace string, phase v1alpha1.ClusterServiceVersionPhase, selector labels.Selector) map[string]*v1alpha1.ClusterServiceVersion {
 	csvsInNamespace, err := s.lister.OperatorsV1alpha1().ClusterServiceVersionLister().ClusterServiceVersions(namespace).List(selector)
 

pkg/lib/csv/notification.go

@@ -0,0 +1,17 @@
+package csv
+
+import (
+	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/apis/operators/v1alpha1"
+)
+
+// WatchNotification is an sink interface that can be used to get notification
+// of CSV reconciliation request(s) received by the operator.
+type WatchNotification interface {
+	// OnAddOrUpdate is invoked when a add or update reconciliation request has
+	// been received by the operator.
+	OnAddOrUpdate(in *v1alpha1.ClusterServiceVersion)
+
+	// OnDelete is invoked when a delete reconciliation request has
+	// been received by the operator.
+	OnDelete(in *v1alpha1.ClusterServiceVersion)
+}

pkg/lib/csv/replace_finder.go

@@ -20,6 +20,7 @@ func NewReplaceFinder(logger *logrus.Logger, client versioned.Interface) Replace
 type ReplaceFinder interface {
 	IsBeingReplaced(in *v1alpha1.ClusterServiceVersion, csvsInNamespace map[string]*v1alpha1.ClusterServiceVersion) (replacedBy *v1alpha1.ClusterServiceVersion)
 	IsReplacing(in *v1alpha1.ClusterServiceVersion) *v1alpha1.ClusterServiceVersion
+	GetFinalCSVInReplacing(in *v1alpha1.ClusterServiceVersion, csvsInNamespace map[string]*v1alpha1.ClusterServiceVersion) (replacedBy *v1alpha1.ClusterServiceVersion)
 }
 
 type replace struct {
@@ -68,3 +69,22 @@ func (r *replace) IsReplacing(in *v1alpha1.ClusterServiceVersion) *v1alpha1.Clus
 
 	return previous
 }
+
+// GetFinalCSVInReplacing returns the most recent ClustererviceVersion that is
+// in the replace chain.
+//
+// If the corresponding ClusterServiceVersion is not found nil is returned.
+func (r *replace) GetFinalCSVInReplacing(in *v1alpha1.ClusterServiceVersion, csvsInNamespace map[string]*v1alpha1.ClusterServiceVersion) (replacedBy *v1alpha1.ClusterServiceVersion) {
+	current := in
+	for {
+		next := r.IsBeingReplaced(current, csvsInNamespace)
+		if next == nil {
+			break
+		}
+
+		replacedBy = next
+		current = next
+	}
+
+	return
+}