Skip to content

Commit 3af3a51

Browse files
openshift-merge-robotopenshift-merge-robot
authored
Merge pull request #679 from ecordell/fix-multiple-installplans
fix(installplan): fix bug where too many installplans can be created
2 parents 85363e8 + d1b2c38 commit 3af3a51

File tree

12 files changed

+404
-114
lines changed

12 files changed

+404
-114
lines changed

‎go.sum

Lines changed: 1 addition & 48 deletions
Large diffs are not rendered by default.

‎pkg/controller/operators/catalog/operator.go

Lines changed: 139 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -56,16 +56,17 @@ var timeNow = func() metav1.Time { return metav1.NewTime(time.Now().UTC()) }
5656
// resolving dependencies in a catalog.
5757
type Operator struct {
5858
*queueinformer.Operator
59-
client versioned.Interface
60-
lister operatorlister.OperatorLister
61-
namespace string
62-
sources map[resolver.CatalogKey]resolver.SourceRef
63-
sourcesLock sync.RWMutex
64-
sourcesLastUpdate metav1.Time
65-
resolver resolver.Resolver
66-
subQueue workqueue.RateLimitingInterface
67-
catSrcQueueSet queueinformer.ResourceQueueSet
68-
reconciler reconciler.ReconcilerFactory
59+
client versioned.Interface
60+
lister operatorlister.OperatorLister
61+
namespace string
62+
sources map[resolver.CatalogKey]resolver.SourceRef
63+
sourcesLock sync.RWMutex
64+
sourcesLastUpdate metav1.Time
65+
resolver resolver.Resolver
66+
subQueue workqueue.RateLimitingInterface
67+
catSrcQueueSet queueinformer.ResourceQueueSet
68+
namespaceResolveQueue workqueue.RateLimitingInterface
69+
reconciler reconciler.ReconcilerFactory
6970
}
7071

7172
// NewOperator creates a new Catalog Operator.
@@ -95,6 +96,7 @@ func NewOperator(kubeconfigPath string, logger *logrus.Logger, wakeupInterval ti
9596
// resolver needs subscription and csv listers
9697
lister.OperatorsV1alpha1().RegisterSubscriptionLister(namespace, nsInformerFactory.Operators().V1alpha1().Subscriptions().Lister())
9798
lister.OperatorsV1alpha1().RegisterClusterServiceVersionLister(namespace, nsInformerFactory.Operators().V1alpha1().ClusterServiceVersions().Lister())
99+
lister.OperatorsV1alpha1().RegisterInstallPlanLister(namespace, nsInformerFactory.Operators().V1alpha1().InstallPlans().Lister())
98100
}
99101

100102
// Create a new queueinformer-based operator.
@@ -209,6 +211,24 @@ func NewOperator(kubeconfigPath string, logger *logrus.Logger, wakeupInterval ti
209211
OpClient: op.OpClient,
210212
Lister: op.lister,
211213
}
214+
215+
// Namespace sync for resolving subscriptions
216+
namespaceInformer := informers.NewSharedInformerFactory(op.OpClient.KubernetesInterface(), wakeupInterval).Core().V1().Namespaces()
217+
resolvingNamespaceQueue := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "resolver")
218+
namespaceQueueInformer := queueinformer.NewInformer(
219+
resolvingNamespaceQueue,
220+
namespaceInformer.Informer(),
221+
op.syncResolvingNamespace,
222+
nil,
223+
"resolver",
224+
metrics.NewMetricsNil(),
225+
logger,
226+
)
227+
228+
op.RegisterQueueInformer(namespaceQueueInformer)
229+
op.lister.CoreV1().RegisterNamespaceLister(namespaceInformer.Lister())
230+
op.namespaceResolveQueue = resolvingNamespaceQueue
231+
212232
return op, nil
213233
}
214234

@@ -319,6 +339,7 @@ func (o *Operator) syncCatalogSources(obj interface{}) (syncError error) {
319339

320340
logger := o.Log.WithFields(logrus.Fields{
321341
"source": catsrc.GetName(),
342+
"id": queueinformer.NewLoopID(),
322343
})
323344
logger.Debug("syncing catsrc")
324345
out := catsrc.DeepCopy()
@@ -456,9 +477,8 @@ func (o *Operator) syncCatalogSources(obj interface{}) (syncError error) {
456477
return err
457478
}
458479

459-
// Sync any dependent Subscriptions
460-
// TODO: this should go away, we should resync the namespace instead
461-
o.syncDependentSubscriptions(logger, out.GetName(), out.GetNamespace())
480+
// Trigger a resolve, will pick up any subscriptions that depend on the catalog
481+
o.resolveNamespace(out.GetNamespace())
462482

463483
return nil
464484
}
@@ -487,36 +507,59 @@ func (o *Operator) syncDependentSubscriptions(logger *logrus.Entry, catalogSourc
487507
}
488508
}
489509

490-
func (o *Operator) syncSubscriptions(obj interface{}) error {
491-
sub, ok := obj.(*v1alpha1.Subscription)
510+
func (o *Operator) syncResolvingNamespace(obj interface{}) error {
511+
ns, ok := obj.(*corev1.Namespace)
492512
if !ok {
493513
o.Log.Debugf("wrong type: %#v", obj)
494-
return fmt.Errorf("casting Subscription failed")
514+
return fmt.Errorf("casting Namespace failed")
495515
}
496-
namespace := sub.GetNamespace()
516+
namespace := ns.GetName()
497517

498518
logger := o.Log.WithFields(logrus.Fields{
499-
"sub": sub.GetName(),
500-
"namespace": sub.GetNamespace(),
501-
"source": sub.Spec.CatalogSource,
502-
"pkg": sub.Spec.Package,
503-
"channel": sub.Spec.Channel,
519+
"namespace": namespace,
520+
"id": queueinformer.NewLoopID(),
504521
})
505522

506-
// record the current state of the desired corresponding CSV in the status. no-op if we don't know the csv yet.
507-
sub, err := o.ensureSubscriptionCSVState(logger, sub)
523+
// get the set of sources that should be used for resolution and best-effort get their connections working
524+
logger.Debug("resolving sources")
525+
resolverSources := o.ensureResolverSources(logger, namespace)
526+
527+
logger.Debug("checking if subscriptions need update")
528+
529+
subs, err := o.lister.OperatorsV1alpha1().SubscriptionLister().Subscriptions(namespace).List(labels.Everything())
508530
if err != nil {
531+
logger.WithError(err).Debug("couldn't list subscriptions")
509532
return err
510533
}
511534

512-
// return early if the subscription is up to date
513-
if o.nothingToUpdate(logger, sub) {
535+
shouldUpdate := false
536+
for _, sub := range subs {
537+
logger := logger.WithFields(logrus.Fields{
538+
"sub": sub.GetName(),
539+
"source": sub.Spec.CatalogSource,
540+
"pkg": sub.Spec.Package,
541+
"channel": sub.Spec.Channel,
542+
})
543+
544+
// ensure the installplan reference is correct
545+
sub, err := o.ensureSubscriptionInstallPlanState(logger, sub)
546+
if err != nil {
547+
return err
548+
}
549+
550+
// record the current state of the desired corresponding CSV in the status. no-op if we don't know the csv yet.
551+
sub, err = o.ensureSubscriptionCSVState(logger, sub)
552+
if err != nil {
553+
return err
554+
}
555+
shouldUpdate = shouldUpdate || !o.nothingToUpdate(logger, sub)
556+
}
557+
if !shouldUpdate {
558+
logger.Debug("all subscriptions up to date")
514559
return nil
515560
}
516561

517-
// get the set of sources that should be used for resolution and best-effort get their connections working
518-
logger.Debugf("resolving sources for %s", namespace)
519-
resolverSources := o.ensureResolverSources(logger, namespace)
562+
logger.Debug("resolving subscriptions in namespace")
520563

521564
// resolve a set of steps to apply to a cluster, a set of subscriptions to create/update, and any errors
522565
steps, subs, err := o.resolver.ResolveSteps(namespace, resolver.NewNamespaceSourceQuerier(resolverSources))
@@ -533,20 +576,35 @@ func (o *Operator) syncSubscriptions(obj interface{}) error {
533576
break
534577
}
535578
}
536-
537579
installplanReference, err := o.createInstallPlan(namespace, subs, installPlanApproval, steps)
538580
if err != nil {
539581
logger.WithError(err).Debug("error creating installplan")
540582
return err
541583
}
542584

543-
if err := o.ensureSubscriptionInstallPlanState(namespace, subs, installplanReference); err != nil {
585+
if err := o.updateSubscriptionSetInstallPlanState(namespace, subs, installplanReference); err != nil {
544586
logger.WithError(err).Debug("error ensuring subscription installplan state")
545587
return err
546588
}
547589
return nil
548590
}
549591

592+
func (o *Operator) syncSubscriptions(obj interface{}) error {
593+
sub, ok := obj.(*v1alpha1.Subscription)
594+
if !ok {
595+
o.Log.Debugf("wrong type: %#v", obj)
596+
return fmt.Errorf("casting Subscription failed")
597+
}
598+
599+
o.resolveNamespace(sub.GetNamespace())
600+
601+
return nil
602+
}
603+
604+
func (o *Operator) resolveNamespace(namespace string) {
605+
o.namespaceResolveQueue.AddRateLimited(namespace)
606+
}
607+
550608
func (o *Operator) ensureResolverSources(logger *logrus.Entry, namespace string) map[resolver.CatalogKey]registryclient.Interface {
551609
// TODO: record connection status onto an object
552610
resolverSources := make(map[resolver.CatalogKey]registryclient.Interface, 0)
@@ -606,6 +664,44 @@ func (o *Operator) nothingToUpdate(logger *logrus.Entry, sub *v1alpha1.Subscript
606664
return false
607665
}
608666

667+
func (o *Operator) ensureSubscriptionInstallPlanState(logger *logrus.Entry, sub *v1alpha1.Subscription) (*v1alpha1.Subscription, error) {
668+
if sub.Status.Install != nil {
669+
return sub, nil
670+
}
671+
672+
logger.Debug("checking for existing installplan")
673+
674+
// check if there's an installplan that created this subscription (only if it doesn't have a reference yet)
675+
// this indicates it was newly resolved by another operator, and we should reference that installplan in the status
676+
ips, err := o.lister.OperatorsV1alpha1().InstallPlanLister().InstallPlans(sub.GetNamespace()).List(labels.Everything())
677+
if err != nil {
678+
logger.WithError(err).Debug("couldn't get installplans")
679+
// if we can't list, just continue processing
680+
return sub, nil
681+
}
682+
683+
out := sub.DeepCopy()
684+
685+
for _, ip := range ips {
686+
for _, step := range ip.Status.Plan {
687+
// TODO: is this enough? should we check equality of pkg/channel?
688+
if step != nil && step.Resource.Kind == v1alpha1.SubscriptionKind && step.Resource.Name == sub.GetName() {
689+
logger.WithField("installplan", ip.GetName()).Debug("found subscription in steps of existing installplan")
690+
out.Status.Install = o.referenceForInstallPlan(ip)
691+
out.Status.State = v1alpha1.SubscriptionStateUpgradePending
692+
if updated, err := o.client.OperatorsV1alpha1().Subscriptions(sub.GetNamespace()).UpdateStatus(out); err != nil {
693+
return nil, err
694+
} else {
695+
return updated, nil
696+
}
697+
}
698+
}
699+
}
700+
logger.Debug("did not find subscription in steps of existing installplan")
701+
702+
return sub, nil
703+
}
704+
609705
func (o *Operator) ensureSubscriptionCSVState(logger *logrus.Entry, sub *v1alpha1.Subscription) (*v1alpha1.Subscription, error) {
610706
if sub.Status.CurrentCSV == "" {
611707
return sub, nil
@@ -637,7 +733,7 @@ func (o *Operator) ensureSubscriptionCSVState(logger *logrus.Entry, sub *v1alpha
637733
return sub, nil
638734
}
639735

640-
func (o *Operator) ensureSubscriptionInstallPlanState(namespace string, subs []*v1alpha1.Subscription, installPlanRef *v1alpha1.InstallPlanReference) error {
736+
func (o *Operator) updateSubscriptionSetInstallPlanState(namespace string, subs []*v1alpha1.Subscription, installPlanRef *v1alpha1.InstallPlanReference) error {
641737
// TODO: parallel, sync waitgroup
642738
for _, sub := range subs {
643739
sub.Status.Install = installPlanRef
@@ -699,13 +795,17 @@ func (o *Operator) createInstallPlan(namespace string, subs []*v1alpha1.Subscrip
699795
if err != nil {
700796
return nil, err
701797
}
798+
return o.referenceForInstallPlan(res), nil
799+
800+
}
801+
802+
func (o *Operator) referenceForInstallPlan(ip *v1alpha1.InstallPlan) *v1alpha1.InstallPlanReference {
702803
return &v1alpha1.InstallPlanReference{
703-
UID: res.GetUID(),
704-
Name: res.GetName(),
804+
UID: ip.GetUID(),
805+
Name: ip.GetName(),
705806
APIVersion: v1alpha1.SchemeGroupVersion.String(),
706807
Kind: v1alpha1.InstallPlanKind,
707-
}, nil
708-
808+
}
709809
}
710810

711811
func (o *Operator) requeueSubscription(name, namespace string) {
@@ -723,6 +823,7 @@ func (o *Operator) syncInstallPlans(obj interface{}) (syncError error) {
723823
}
724824

725825
logger := o.Log.WithFields(logrus.Fields{
826+
"id": queueinformer.NewLoopID(),
726827
"ip": plan.GetName(),
727828
"namespace": plan.GetNamespace(),
728829
"phase": plan.Status.Phase,
@@ -749,7 +850,7 @@ func (o *Operator) syncInstallPlans(obj interface{}) (syncError error) {
749850
// notify subscription loop of installplan changes
750851
if ownerutil.IsOwnedByKind(outInstallPlan, v1alpha1.SubscriptionKind) {
751852
oref := ownerutil.GetOwnerByKind(outInstallPlan, v1alpha1.SubscriptionKind)
752-
logger.Info("requeuing installplan owning subscription")
853+
logger.WithField("owner", oref).Debug("requeueing installplan owner")
753854
o.requeueSubscription(oref.Name, outInstallPlan.GetNamespace())
754855
}
755856

@@ -900,7 +1001,7 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
9001001

9011002
// Attempt to create the Subscription
9021003
sub.SetNamespace(namespace)
903-
created, err := o.client.OperatorsV1alpha1().Subscriptions(sub.GetNamespace()).Create(&sub)
1004+
_, err = o.client.OperatorsV1alpha1().Subscriptions(sub.GetNamespace()).Create(&sub)
9041005
if k8serrors.IsAlreadyExists(err) {
9051006
// If it already existed, mark the step as Present.
9061007
plan.Status.Plan[i].Status = v1alpha1.StepStatusPresent
@@ -909,15 +1010,6 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
9091010
} else {
9101011
// If no error occurred, mark the step as Created.
9111012
plan.Status.Plan[i].Status = v1alpha1.StepStatusCreated
912-
created.Status.Install = &v1alpha1.InstallPlanReference{
913-
UID: plan.GetUID(),
914-
Name: plan.GetName(),
915-
APIVersion: v1alpha1.SchemeGroupVersion.String(),
916-
Kind: v1alpha1.InstallPlanKind,
917-
}
918-
if _, err := o.client.OperatorsV1alpha1().Subscriptions(sub.GetNamespace()).UpdateStatus(created); err != nil {
919-
o.Log.WithError(err).Warn("couldn't set installplan reference on created subscription")
920-
}
9211013
}
9221014
case secretKind:
9231015
// TODO: this will confuse bundle users that include secrets in their bundles - this only handles pull secrets

‎pkg/controller/operators/catalog/operator_test.go

Lines changed: 18 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ import (
2121
"k8s.io/client-go/informers"
2222
k8sfake "k8s.io/client-go/kubernetes/fake"
2323
"k8s.io/client-go/tools/cache"
24+
"k8s.io/client-go/util/workqueue"
2425
apiregistrationfake "k8s.io/kube-aggregator/pkg/client/clientset_generated/clientset/fake"
2526

2627
"github.com/operator-framework/operator-lifecycle-manager/pkg/api/apis/operators/v1alpha1"
@@ -139,23 +140,23 @@ func TestExecutePlan(t *testing.T) {
139140
Resource: v1alpha1.StepResource{
140141
CatalogSource: "catalog",
141142
CatalogSourceNamespace: namespace,
142-
Group: "",
143-
Version: "v1",
144-
Kind: "Service",
145-
Name: "service",
146-
Manifest: toManifest(service("service", namespace)),
143+
Group: "",
144+
Version: "v1",
145+
Kind: "Service",
146+
Name: "service",
147+
Manifest: toManifest(service("service", namespace)),
147148
},
148149
Status: v1alpha1.StepStatusUnknown,
149150
},
150151
&v1alpha1.Step{
151152
Resource: v1alpha1.StepResource{
152153
CatalogSource: "catalog",
153154
CatalogSourceNamespace: namespace,
154-
Group: "operators.coreos.com",
155-
Version: "v1alpha1",
156-
Kind: "ClusterServiceVersion",
157-
Name: "csv",
158-
Manifest: toManifest(csv("csv", namespace, nil, nil)),
155+
Group: "operators.coreos.com",
156+
Version: "v1alpha1",
157+
Kind: "ClusterServiceVersion",
158+
Name: "csv",
159+
Manifest: toManifest(csv("csv", namespace, nil, nil)),
159160
},
160161
Status: v1alpha1.StepStatusUnknown,
161162
},
@@ -507,12 +508,13 @@ func NewFakeOperator(clientObjs []runtime.Object, k8sObjs []runtime.Object, extO
507508
// Create the new operator
508509
queueOperator, err := queueinformer.NewOperatorFromClient(opClientFake, logrus.New())
509510
op := &Operator{
510-
Operator: queueOperator,
511-
client: clientFake,
512-
lister: lister,
513-
namespace: namespace,
514-
sources: make(map[resolver.CatalogKey]resolver.SourceRef),
515-
resolver: &fakes.FakeResolver{},
511+
Operator: queueOperator,
512+
client: clientFake,
513+
lister: lister,
514+
namespace: namespace,
515+
namespaceResolveQueue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "resolver"),
516+
sources: make(map[resolver.CatalogKey]resolver.SourceRef),
517+
resolver: &fakes.FakeResolver{},
516518
}
517519

518520
op.reconciler = &reconciler.RegistryReconcilerFactory{

0 commit comments

Comments
 (0)