just subscription sync changes

Signed-off-by: Per Goncalves da Silva <[email protected]>

Author: Per Goncalves da Silva

pkg/controller/operators/catalog/operator.go

@@ -2,7 +2,6 @@ package catalog
 
 import (
 	"context"
-	"crypto/sha256"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -48,6 +47,7 @@ import (
 	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/client-go/tools/pager"
 	"k8s.io/client-go/tools/record"
+	"k8s.io/client-go/util/retry"
 	"k8s.io/client-go/util/workqueue"
 	utilclock "k8s.io/utils/clock"
 
@@ -1342,9 +1342,6 @@ func (o *Operator) syncResolvingNamespace(obj interface{}) error {
 		return err
 	}
 
-	// Remove resolutionfailed condition from subscriptions
-	o.removeSubsCond(subs, v1alpha1.SubscriptionResolutionFailed)
-
 	// Attempt to unpack bundles before installing
 	// Note: This should probably use the attenuated client to prevent users from resolving resources they otherwise don't have access to.
 	if len(bundleLookups) > 0 {
@@ -1472,6 +1469,9 @@ func (o *Operator) syncResolvingNamespace(obj interface{}) error {
 	// Remove BundleUnpackFailed condition from subscriptions
 	o.removeSubsCond(subs, v1alpha1.SubscriptionBundleUnpackFailed)
 
+	// Remove resolutionfailed condition from subscriptions
+	o.removeSubsCond(subs, v1alpha1.SubscriptionResolutionFailed)
+
 	newSub := true
 	for _, updatedSub := range updatedSubs {
 		updatedSub.Status.RemoveConditions(v1alpha1.SubscriptionResolutionFailed)
@@ -1499,6 +1499,18 @@ func (o *Operator) syncResolvingNamespace(obj interface{}) error {
 	return nil
 }
 
+func (o *Operator) syncSubscriptions(obj interface{}) error {
+	sub, ok := obj.(*v1alpha1.Subscription)
+	if !ok {
+		o.logger.Infof("wrong type: %#v", obj)
+		return fmt.Errorf("casting Subscription failed")
+	}
+
+	o.nsResolveQueue.Add(sub.GetNamespace())
+
+	return nil
+}
+
 // syncOperatorGroups requeues the namespace resolution queue on changes to an operatorgroup
 // This is because the operatorgroup is now an input to resolution via the global catalog exclusion annotation
 func (o *Operator) syncOperatorGroups(obj interface{}) error {
@@ -1660,20 +1672,7 @@ func (o *Operator) createInstallPlan(namespace string, gen int, subs []*v1alpha1
 		return nil, nil
 	}
 
-	sha, err := func() (string, error) {
-		jsonBytes, err := json.Marshal(steps)
-		if err != nil {
-			return "", err
-		}
-		hash := sha256.Sum256(jsonBytes)
-		return fmt.Sprintf("%x", hash), nil
-	}()
-
-	if err != nil {
-		return nil, err
-	}
-
-	var csvNames []string
+	csvNames := []string{}
 	catalogSourceMap := map[string]struct{}{}
 	for _, s := range steps {
 		if s.Resource.Kind == "ClusterServiceVersion" {
@@ -1682,7 +1681,7 @@ func (o *Operator) createInstallPlan(namespace string, gen int, subs []*v1alpha1
 		catalogSourceMap[s.Resource.CatalogSource] = struct{}{}
 	}
 
-	var catalogSources []string
+	catalogSources := []string{}
 	for s := range catalogSourceMap {
 		catalogSources = append(catalogSources, s)
 	}
@@ -1693,8 +1692,8 @@ func (o *Operator) createInstallPlan(namespace string, gen int, subs []*v1alpha1
 	}
 	ip := &v1alpha1.InstallPlan{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      fmt.Sprintf("install-%s", sha),
-			Namespace: namespace,
+			GenerateName: "install-",
+			Namespace:    namespace,
 		},
 		Spec: v1alpha1.InstallPlanSpec{
 			ClusterServiceVersionNames: csvNames,
@@ -1708,8 +1707,7 @@ func (o *Operator) createInstallPlan(namespace string, gen int, subs []*v1alpha1
 	}
 
 	res, err := o.client.OperatorsV1alpha1().InstallPlans(namespace).Create(context.TODO(), ip, metav1.CreateOptions{})
-	if err != nil && !apierrors.IsAlreadyExists(err) {
-
+	if err != nil {
 		return nil, err
 	}
 
@@ -1719,10 +1717,11 @@ func (o *Operator) createInstallPlan(namespace string, gen int, subs []*v1alpha1
 		CatalogSources: catalogSources,
 		BundleLookups:  bundleLookups,
 	}
-
-	if _, err = o.client.OperatorsV1alpha1().InstallPlans(namespace).UpdateStatus(context.TODO(), res, metav1.UpdateOptions{}); err != nil {
+	res, err = o.client.OperatorsV1alpha1().InstallPlans(namespace).UpdateStatus(context.TODO(), res, metav1.UpdateOptions{})
+	if err != nil {
 		return nil, err
 	}
+
 	return reference.GetReference(res)
 }
 
@@ -1766,6 +1765,7 @@ func (o *Operator) updateSubscriptionStatuses(subs []*v1alpha1.Subscription) ([]
 		errs       []error
 		mu         sync.Mutex
 		wg         sync.WaitGroup
+		getOpts    = metav1.GetOptions{}
 		updateOpts = metav1.UpdateOptions{}
 	)
 
@@ -1774,29 +1774,22 @@ func (o *Operator) updateSubscriptionStatuses(subs []*v1alpha1.Subscription) ([]
 		go func(sub *v1alpha1.Subscription) {
 			defer wg.Done()
 
-			_, err := o.client.OperatorsV1alpha1().Subscriptions(sub.Namespace).UpdateStatus(context.TODO(), sub, updateOpts)
-			if err != nil {
+			update := func() error {
+				// Update the status of the latest revision
+				latest, err := o.client.OperatorsV1alpha1().Subscriptions(sub.GetNamespace()).Get(context.TODO(), sub.GetName(), getOpts)
+				if err != nil {
+					return err
+				}
+				latest.Status = sub.Status
+				*sub = *latest
+				_, err = o.client.OperatorsV1alpha1().Subscriptions(sub.Namespace).UpdateStatus(context.TODO(), latest, updateOpts)
+				return err
+			}
+			if err := retry.RetryOnConflict(retry.DefaultRetry, update); err != nil {
 				mu.Lock()
 				defer mu.Unlock()
 				errs = append(errs, err)
 			}
-
-			//update := func() error {
-			//	// Update the status of the latest revision
-			//	latest, err := o.client.OperatorsV1alpha1().Subscriptions(sub.GetNamespace()).Get(context.TODO(), sub.GetName(), getOpts)
-			//	if err != nil {
-			//		return err
-			//	}
-			//	latest.Status = sub.Status
-			//	*sub = *latest
-			//	_, err = o.client.OperatorsV1alpha1().Subscriptions(sub.Namespace).UpdateStatus(context.TODO(), latest, updateOpts)
-			//	return err
-			//}
-			//if err := retry.RetryOnConflict(retry.DefaultRetry, update); err != nil {
-			//	mu.Lock()
-			//	defer mu.Unlock()
-			//	errs = append(errs, err)
-			//}
 		}(sub)
 	}
 	wg.Wait()
@@ -2857,11 +2850,31 @@ func (o *Operator) getUpdatedOwnerReferences(refs []metav1.OwnerReference, names
 }
 
 func (o *Operator) listSubscriptions(namespace string) (subs []*v1alpha1.Subscription, err error) {
-	return o.lister.OperatorsV1alpha1().SubscriptionLister().Subscriptions(namespace).List(labels.Everything())
+	list, err := o.client.OperatorsV1alpha1().Subscriptions(namespace).List(context.TODO(), metav1.ListOptions{})
+	if err != nil {
+		return
+	}
+
+	subs = make([]*v1alpha1.Subscription, 0)
+	for i := range list.Items {
+		subs = append(subs, &list.Items[i])
+	}
+
+	return
 }
 
 func (o *Operator) listInstallPlans(namespace string) (ips []*v1alpha1.InstallPlan, err error) {
-	return o.lister.OperatorsV1alpha1().InstallPlanLister().InstallPlans(namespace).List(labels.Everything())
+	list, err := o.client.OperatorsV1alpha1().InstallPlans(namespace).List(context.TODO(), metav1.ListOptions{})
+	if err != nil {
+		return
+	}
+
+	ips = make([]*v1alpha1.InstallPlan, 0)
+	for i := range list.Items {
+		ips = append(ips, &list.Items[i])
+	}
+
+	return
 }
 
 // competingCRDOwnersExist returns true if there exists a CSV that owns at least one of the given CSVs owned CRDs (that's not the given CSV)

pkg/controller/registry/resolver/step_resolver.go

@@ -2,6 +2,7 @@
 package resolver
 
 import (
+	"context"
 	"fmt"
 
 	"github.com/operator-framework/api/pkg/operators/v1alpha1"
@@ -15,6 +16,7 @@ import (
 	"github.com/sirupsen/logrus"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 )
 
@@ -89,7 +91,7 @@ func NewOperatorStepResolver(lister operatorlister.OperatorLister, client versio
 }
 
 func (r *OperatorStepResolver) ResolveSteps(namespace string) ([]*v1alpha1.Step, []v1alpha1.BundleLookup, []*v1alpha1.Subscription, error) {
-	subs, err := r.subLister.Subscriptions(namespace).List(labels.Everything())
+	subs, err := r.listSubscriptions(namespace)
 	if err != nil {
 		return nil, nil, nil, err
 	}
@@ -117,9 +119,9 @@ func (r *OperatorStepResolver) ResolveSteps(namespace string) ([]*v1alpha1.Step,
 
 	// if there's no error, we were able to satisfy all constraints in the subscription set, so we calculate what
 	// changes to persist to the cluster and write them out as `steps`
-	var steps []*v1alpha1.Step
-	var updatedSubs []*v1alpha1.Subscription
-	var bundleLookups []v1alpha1.BundleLookup
+	steps := []*v1alpha1.Step{}
+	updatedSubs := []*v1alpha1.Subscription{}
+	bundleLookups := []v1alpha1.BundleLookup{}
 	for _, op := range operators {
 		// Find any existing subscriptions that resolve to this operator.
 		existingSubscriptions := make(map[*v1alpha1.Subscription]bool)
@@ -217,9 +219,8 @@ func (r *OperatorStepResolver) ResolveSteps(namespace string) ([]*v1alpha1.Step,
 				continue
 			}
 			// update existing subscription status
-			updatedSub := sub.DeepCopy()
-			updatedSub.Status.CurrentCSV = op.Name
-			updatedSubs = append(updatedSubs, updatedSub)
+			sub.Status.CurrentCSV = op.Name
+			updatedSubs = append(updatedSubs, sub)
 		}
 	}
 
@@ -242,19 +243,19 @@ func (r *OperatorStepResolver) hasExistingCurrentCSV(sub *v1alpha1.Subscription)
 	return false, err // Can't answer this question right now.
 }
 
-//func (r *OperatorStepResolver) listSubscriptions(namespace string) ([]*v1alpha1.Subscription, error) {
-//	list, err := r.client.OperatorsV1alpha1().Subscriptions(namespace).List(context.TODO(), metav1.ListOptions{})
-//	if err != nil {
-//		return nil, err
-//	}
-//
-//	var subs []*v1alpha1.Subscription
-//	for i := range list.Items {
-//		subs = append(subs, &list.Items[i])
-//	}
-//
-//	return subs, nil
-//}
+func (r *OperatorStepResolver) listSubscriptions(namespace string) ([]*v1alpha1.Subscription, error) {
+	list, err := r.client.OperatorsV1alpha1().Subscriptions(namespace).List(context.TODO(), metav1.ListOptions{})
+	if err != nil {
+		return nil, err
+	}
+
+	var subs []*v1alpha1.Subscription
+	for i := range list.Items {
+		subs = append(subs, &list.Items[i])
+	}
+
+	return subs, nil
+}
 
 type mergedSourceProvider struct {
 	sps    []cache.SourceProvider

pkg/lib/queueinformer/config.go

@@ -233,6 +233,6 @@ func (c *operatorConfig) validate() (err error) {
 func defaultOperatorConfig() *operatorConfig {
 	return &operatorConfig{
 		logger:     logrus.New(),
-		numWorkers: 1,
+		numWorkers: 2,
 	}
 }