fix(subscriptions): fix race between subscription sync and cache

go.mod

@@ -17,7 +17,7 @@ require (
 	github.com/google/gofuzz v0.0.0-20170612174753-24818f796faf // indirect
 	github.com/googleapis/gnostic v0.2.0 // indirect
 	github.com/grpc-ecosystem/go-grpc-middleware v1.0.0 // indirect
-	github.com/grpc-ecosystem/grpc-gateway v1.6.3 // indirect
+	github.com/grpc-ecosystem/grpc-gateway v1.6.3
 	github.com/imdario/mergo v0.3.6 // indirect
 	github.com/inconshreveable/mousetrap v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.5 // indirect

go.sum

@@ -21,7 +21,7 @@ github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDk
 github.com/coreos/bbolt v1.3.0 h1:HIgH5xUWXT914HCI671AxuTTqjj64UOFr7pHn48LUTI=
 github.com/coreos/bbolt v1.3.0/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkEiiKk=
 github.com/coreos/etcd v3.3.9+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
-github.com/coreos/etcd v3.3.10+incompatible h1:jFneRYjIvLMLhDLCzuTuU4rSJUjRplcJQ7pD7MnhC04=
+github.com/coreos/etcd v3.3.10+incompatible h1:KjVWqrZ5U0wa3CxY2AxlH6/UcB+PK2td1DcsYhA+HRs=
 github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
 github.com/coreos/go-semver v0.2.0 h1:3Jm3tLmsgAYcjC+4Up7hJrFBPr+n7rAqYeSw/SZazuY=
 github.com/coreos/go-semver v0.2.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
@@ -107,7 +107,7 @@ github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoA
 github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=
 github.com/gregjones/httpcache v0.0.0-20181110185634-c63ab54fda8f h1:ShTPMJQes6tubcjzGMODIVG5hlrCeImaBnZzKF2N8SM=
 github.com/gregjones/httpcache v0.0.0-20181110185634-c63ab54fda8f/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=
-github.com/grpc-ecosystem/go-grpc-middleware v1.0.0 h1:Iju5GlWwrvL6UBg4zJJt3btmonfrMlCDdsejg4CZE7c=
+github.com/grpc-ecosystem/go-grpc-middleware v1.0.0 h1:BWIsLfhgKhV5g/oF34aRjniBHLTZe5DNekSjbAjIS6c=
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=
 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 h1:Ovs26xHkKqVztRpIrF/92BcuyuQ/YW4NSIpoGtfXNho=
 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0/go.mod h1:8NvIoxWQoOIhqOTXgfV/d3M/q6VIi02HzZEHgUlZvzk=
@@ -294,5 +294,5 @@ k8s.io/kube-aggregator v0.0.0-20181204002017-122bac39d429/go.mod h1:8sbzT4QQKDEm
 k8s.io/kube-openapi v0.0.0-20181031203759-72693cb1fadd h1:ggv/Vfza0i5xuhUZyYyxcc25AmQvHY8Zi1C2m8WgBvA=
 k8s.io/kube-openapi v0.0.0-20181031203759-72693cb1fadd/go.mod h1:BXM9ceUBTj2QnfH2MK1odQs778ajze1RxcmP6S8RVVc=
 k8s.io/kubernetes v1.11.7-beta.0.0.20181219023948-b875d52ea96d/go.mod h1:ocZa8+6APFNC2tX1DZASIbocyYT5jHzqFVsY5aoB7Jk=
-k8s.io/kubernetes v1.11.8-beta.0.0.20190124204751-3a10094374f2 h1:CzIOMOEjH+sQw35LY1Gl0jwthkyOojzaq2HIeYZYOrM=
+k8s.io/kubernetes v1.11.8-beta.0.0.20190124204751-3a10094374f2 h1:Q4hIsjqTbRprTaPk+gVDUuVipXpGJtTz7Lg2FS3xpmw=
 k8s.io/kubernetes v1.11.8-beta.0.0.20190124204751-3a10094374f2/go.mod h1:ocZa8+6APFNC2tX1DZASIbocyYT5jHzqFVsY5aoB7Jk=

pkg/controller/operators/catalog/operator.go

@@ -47,6 +47,8 @@ const (
 	serviceKind            = "Service"
 	roleKind               = "Role"
 	roleBindingKind        = "RoleBinding"
+
+	generatedByKey  = "olm/generated-by"
 )
 
 // for test stubbing and for ensuring standardization of timezones to UTC
@@ -675,33 +677,29 @@ func (o *Operator) ensureSubscriptionInstallPlanState(logger *logrus.Entry, sub
 
 	// check if there's an installplan that created this subscription (only if it doesn't have a reference yet)
 	// this indicates it was newly resolved by another operator, and we should reference that installplan in the status
-	ips, err := o.lister.OperatorsV1alpha1().InstallPlanLister().InstallPlans(sub.GetNamespace()).List(labels.Everything())
-	if err != nil {
-		logger.WithError(err).Debug("couldn't get installplans")
-		// if we can't list, just continue processing
+	ipName, ok := sub.GetAnnotations()[generatedByKey]
+	if !ok {
+		// err := fmt.Errorf("no installplan reference or %s annotation found", generatedByKey)
+		// logger.WithField("err", err.Error()).Error("an error occurred while associating a subscription with an installplan")
 		return sub, nil
 	}
 
-	out := sub.DeepCopy()
+	ip, err := o.lister.OperatorsV1alpha1().InstallPlanLister().InstallPlans(sub.GetNamespace()).Get(ipName)
+	if err != nil {
+		logger.WithField("installplan", ipName).Warn("unable to get installplan from cache")
+		return nil, err
+	}
+	logger.WithField("installplan", ipName).Debug("found installplan that generated subscription")
 
-	for _, ip := range ips {
-		for _, step := range ip.Status.Plan {
-			// TODO: is this enough? should we check equality of pkg/channel?
-			if step != nil && step.Resource.Kind == v1alpha1.SubscriptionKind && step.Resource.Name == sub.GetName() {
-				logger.WithField("installplan", ip.GetName()).Debug("found subscription in steps of existing installplan")
-				out.Status.Install = o.referenceForInstallPlan(ip)
-				out.Status.State = v1alpha1.SubscriptionStateUpgradePending
-				if updated, err := o.client.OperatorsV1alpha1().Subscriptions(sub.GetNamespace()).UpdateStatus(out); err != nil {
-					return nil, err
-				} else {
-					return updated, nil
-				}
-			}
-		}
+	out := sub.DeepCopy()
+	out.Status.Install = o.referenceForInstallPlan(ip)
+	out.Status.State = v1alpha1.SubscriptionStateUpgradePending
+	updated, err := o.client.OperatorsV1alpha1().Subscriptions(sub.GetNamespace()).UpdateStatus(out)
+	if err != nil {
+		return nil, err
 	}
-	logger.Debug("did not find subscription in steps of existing installplan")
 
-	return sub, nil
+	return updated, nil
 }
 
 func (o *Operator) ensureSubscriptionCSVState(logger *logrus.Entry, sub *v1alpha1.Subscription) (*v1alpha1.Subscription, error) {
@@ -1001,6 +999,13 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 					return errorwrap.Wrapf(err, "error parsing step manifest: %s", step.Resource.Name)
 				}
 
+				// Add the InstallPlan's name as an annotation
+				if annotations := sub.GetAnnotations(); annotations != nil {
+					annotations[generatedByKey] = plan.GetName()
+				} else {
+					sub.SetAnnotations(map[string]string{generatedByKey: plan.GetName()})
+				}
+
 				// Attempt to create the Subscription
 				sub.SetNamespace(namespace)
 				_, err = o.client.OperatorsV1alpha1().Subscriptions(sub.GetNamespace()).Create(&sub)

pkg/controller/registry/reconciler/configmap.go

@@ -106,15 +106,15 @@ func (s *configMapCatalogSourceDecorator) Pod(image string) *v1.Pod {
 								Command: []string{"grpc_health_probe", "-addr=localhost:50051"},
 							},
 						},
-						InitialDelaySeconds: 5,
+						InitialDelaySeconds: 1,
 					},
 					LivenessProbe: &v1.Probe{
 						Handler: v1.Handler{
 							Exec: &v1.ExecAction{
 								Command: []string{"grpc_health_probe", "-addr=localhost:50051"},
 							},
 						},
-						InitialDelaySeconds: 10,
+						InitialDelaySeconds: 2,
 					},
 				},
 			},

test/e2e/installplan_e2e_test.go

@@ -567,6 +567,7 @@ func TestCreateInstallPlanWithPreExistingCRDOwners(t *testing.T) {
 		// existing cleanup should remove this
 		createSubscriptionForCatalog(t, crc, testNamespace, subscriptionName, mainCatalogSourceName, mainPackageName, betaChannel, v1alpha1.ApprovalAutomatic)
 
+		// time.Sleep(5 * time.Minute)
 		subscription, err = fetchSubscription(t, crc, testNamespace, subscriptionName, subscriptionHasInstallPlanChecker)
 		require.NoError(t, err)
 		require.NotNil(t, subscription)

test/e2e/util_test.go

@@ -237,7 +237,7 @@ func createFieldNotEqualSelector(field string, names ...string) string {
 func cleanupOLM(t *testing.T, namespace string) {
 	var immediate int64 = 0
 	crc := newCRClient(t)
-	//c := newKubeClient(t)
+	c := newKubeClient(t)
 
 	// Cleanup non persistent OLM CRs
 	t.Log("cleaning up any remaining non persistent resources...")
@@ -250,7 +250,7 @@ func cleanupOLM(t *testing.T, namespace string) {
 
 	// error: the server does not allow this method on the requested resource
 	// Cleanup non persistent configmaps
-	//require.NoError(t, c.KubernetesInterface().CoreV1().ConfigMaps(namespace).DeleteCollection(deleteOptions, metav1.ListOptions{FieldSelector: nonPersistentConfigMapsFieldSelector}))
+	require.NoError(t, c.KubernetesInterface().CoreV1().Pods(namespace).DeleteCollection(deleteOptions, metav1.ListOptions{}))
 }
 
 func buildCatalogSourceCleanupFunc(t *testing.T, crc versioned.Interface, namespace string, catalogSource *v1alpha1.CatalogSource) cleanupFunc {