(feat) use scoped client for plan execution

For plan execution, use scoped client that is bound to the service
account specified in the operator group.

- When operator group is synced, check if a service account has been
  specified. If so, make sure it exists and update status accordingly.
- Refactor 'ExecutePlan' - move management of all step resource(s)
  except for CRD into a new layer 'StepEnsurer'
- Add a new strategy that provides appropriately scoped client(s) to
  be used for an operator that is being installed.

Author: tkashem

pkg/api/apis/operators/v1/operatorgroup_types.go

@@ -76,3 +76,21 @@ func (o *OperatorGroup) BuildTargetNamespaces() string {
 	sort.Strings(o.Status.Namespaces)
 	return strings.Join(o.Status.Namespaces, ",")
 }
+
+// IsServiceAccountSpecified returns true if the spec has a service account name specified.
+func (o *OperatorGroup) IsServiceAccountSpecified() bool {
+	if o.Spec.ServiceAccountName == "" {
+		return false
+	}
+
+	return true
+}
+
+// HasServiceAccountSynced returns true if the service account specified has been synced.
+func (o *OperatorGroup) HasServiceAccountSynced() bool {
+	if o.IsServiceAccountSpecified() && o.Status.ServiceAccountRef != nil {
+		return true
+	}
+
+	return false
+}

pkg/controller/operators/catalog/operator.go

@@ -23,11 +23,11 @@ import (
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/client-go/util/workqueue"
 
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/apis/operators/reference"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/apis/operators/v1alpha1"
-	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/client"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/clientset/versioned"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/informers/externalversions"
 	olmerrors "github.com/operator-framework/operator-lifecycle-manager/pkg/controller/errors"
@@ -39,6 +39,7 @@ import (
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/operatorlister"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/ownerutil"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/queueinformer"
+	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/scoped"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/metrics"
 )
 
@@ -75,6 +76,8 @@ type Operator struct {
 	resolver               resolver.Resolver
 	reconciler             reconciler.RegistryReconcilerFactory
 	csvProvidedAPIsIndexer map[string]cache.Indexer
+	clientAttenuator       *scoped.ClientAttenuator
+	serviceAccountQuerier  *scoped.UserDefinedServiceAccountQuerier
 }
 
 // NewOperator creates a new Catalog Operator.
@@ -84,8 +87,13 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 		watchedNamespaces = []string{metav1.NamespaceAll}
 	}
 
+	config, err := clientcmd.BuildConfigFromFlags("", kubeconfigPath)
+	if err != nil {
+		return nil, err
+	}
+
 	// Create a new client for OLM types (CRs)
-	crClient, err := client.NewClient(kubeconfigPath)
+	crClient, err := versioned.NewForConfig(config)
 	if err != nil {
 		return nil, err
 	}
@@ -114,6 +122,8 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 		catsrcQueueSet:         queueinformer.NewEmptyResourceQueueSet(),
 		subQueueSet:            queueinformer.NewEmptyResourceQueueSet(),
 		csvProvidedAPIsIndexer: map[string]cache.Indexer{},
+		serviceAccountQuerier:  scoped.NewUserDefinedServiceAccountQuerier(logger, crClient),
+		clientAttenuator:       scoped.NewClientAttenuator(logger, config, opClient, crClient),
 	}
 	op.reconciler = reconciler.NewRegistryReconcilerFactory(lister, opClient, configmapRegistryImage, op.now)
 
@@ -1037,6 +1047,18 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 		return err
 	}
 
+	// Does the namespace have an operator group that specifies a user defined
+	// service account? If so, then we should use a scoped client for plan
+	// execution.
+	getter := o.serviceAccountQuerier.NamespaceQuerier(namespace)
+	kubeclient, crclient, err := o.clientAttenuator.AttenuateClient(getter)
+	if err != nil {
+		o.logger.Errorf("failed to get a client for plan execution- %v", err)
+		return err
+	}
+
+	ensurer := newStepEnsurer(kubeclient, crclient)
+
 	for i, step := range plan.Status.Plan {
 		switch step.Status {
 		case v1alpha1.StepStatusPresent, v1alpha1.StepStatusCreated:
@@ -1112,16 +1134,14 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 
 				// Attempt to create the CSV.
 				csv.SetNamespace(namespace)
-				_, err = o.client.OperatorsV1alpha1().ClusterServiceVersions(csv.GetNamespace()).Create(&csv)
-				if k8serrors.IsAlreadyExists(err) {
-					// If it already existed, mark the step as Present.
-					plan.Status.Plan[i].Status = v1alpha1.StepStatusPresent
-				} else if err != nil {
-					return errorwrap.Wrapf(err, "error creating csv %s", csv.GetName())
-				} else {
-					// If no error occurred, mark the step as Created.
-					plan.Status.Plan[i].Status = v1alpha1.StepStatusCreated
+
+				status, err := ensurer.EnsureClusterServiceVersion(&csv)
+				if err != nil {
+					return err
 				}
+
+				plan.Status.Plan[i].Status = status
+
 			case v1alpha1.SubscriptionKind:
 				// Marshal the manifest into a subscription instance.
 				var sub v1alpha1.Subscription
@@ -1139,47 +1159,22 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 
 				// Attempt to create the Subscription
 				sub.SetNamespace(namespace)
-				_, err = o.client.OperatorsV1alpha1().Subscriptions(sub.GetNamespace()).Create(&sub)
-				if k8serrors.IsAlreadyExists(err) {
-					// If it already existed, mark the step as Present.
-					plan.Status.Plan[i].Status = v1alpha1.StepStatusPresent
-				} else if err != nil {
-					return errorwrap.Wrapf(err, "error creating subscription %s", sub.GetName())
-				} else {
-					// If no error occurred, mark the step as Created.
-					plan.Status.Plan[i].Status = v1alpha1.StepStatusCreated
-				}
-			case secretKind:
-				// TODO: this will confuse bundle users that include secrets in their bundles - this only handles pull secrets
-				// Get the pre-existing secret.
-				secret, err := o.opClient.KubernetesInterface().CoreV1().Secrets(o.namespace).Get(step.Resource.Name, metav1.GetOptions{})
-				if k8serrors.IsNotFound(err) {
-					return fmt.Errorf("secret %s does not exist", step.Resource.Name)
-				} else if err != nil {
-					return errorwrap.Wrapf(err, "error getting pull secret from olm namespace %s", secret.GetName())
+
+				status, err := ensurer.EnsureSubscription(&sub)
+				if err != nil {
+					return err
 				}
 
-				// Set the namespace to the InstallPlan's namespace and attempt to
-				// create a new secret.
-				secret.SetNamespace(namespace)
-				_, err = o.opClient.KubernetesInterface().CoreV1().Secrets(plan.Namespace).Create(&corev1.Secret{
-					ObjectMeta: metav1.ObjectMeta{
-						Name:      secret.Name,
-						Namespace: plan.Namespace,
-					},
-					Data: secret.Data,
-					Type: secret.Type,
-				})
-				if k8serrors.IsAlreadyExists(err) {
-					// If it already existed, mark the step as Present.
-					plan.Status.Plan[i].Status = v1alpha1.StepStatusPresent
-				} else if err != nil {
+				plan.Status.Plan[i].Status = status
+
+			case secretKind:
+				status, err := ensurer.EnsureSecret(o.namespace, plan.GetNamespace(), step.Resource.Name)
+				if err != nil {
 					return err
-				} else {
-					// If no error occured, mark the step as Created.
-					plan.Status.Plan[i].Status = v1alpha1.StepStatusCreated
 				}
 
+				plan.Status.Plan[i].Status = status
+
 			case clusterRoleKind:
 				// Marshal the manifest into a ClusterRole instance.
 				var cr rbacv1.ClusterRole
@@ -1188,23 +1183,13 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 					return errorwrap.Wrapf(err, "error parsing step manifest: %s", step.Resource.Name)
 				}
 
-				// Attempt to create the ClusterRole.
-				_, err = o.opClient.KubernetesInterface().RbacV1().ClusterRoles().Create(&cr)
-				if k8serrors.IsAlreadyExists(err) {
-					// if we're updating, point owner to the newest csv
-					cr.Labels[ownerutil.OwnerKey] = step.Resolving
-					_, err = o.opClient.UpdateClusterRole(&cr)
-					if err != nil {
-						return errorwrap.Wrapf(err, "error updating clusterrole %s", cr.GetName())
-					}
-					// If it already existed, mark the step as Present.
-					plan.Status.Plan[i].Status = v1alpha1.StepStatusPresent
-				} else if err != nil {
-					return errorwrap.Wrapf(err, "error creating clusterrole %s", cr.GetName())
-				} else {
-					// If no error occurred, mark the step as Created.
-					plan.Status.Plan[i].Status = v1alpha1.StepStatusCreated
+				status, err := ensurer.EnsureClusterRole(&cr, step)
+				if err != nil {
+					return err
 				}
+
+				plan.Status.Plan[i].Status = status
+
 			case clusterRoleBindingKind:
 				// Marshal the manifest into a RoleBinding instance.
 				var rb rbacv1.ClusterRoleBinding
@@ -1213,25 +1198,13 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 					return errorwrap.Wrapf(err, "error parsing step manifest: %s", step.Resource.Name)
 				}
 
-				// Attempt to create the ClusterRoleBinding.
-				_, err = o.opClient.KubernetesInterface().RbacV1().ClusterRoleBindings().Create(&rb)
-				if k8serrors.IsAlreadyExists(err) {
-					// if we're updating, point owner to the newest csv
-					rb.Labels[ownerutil.OwnerKey] = step.Resolving
-					_, err = o.opClient.UpdateClusterRoleBinding(&rb)
-					if err != nil {
-						return errorwrap.Wrapf(err, "error updating clusterrolebinding %s", rb.GetName())
-					}
-
-					// If it already existed, mark the step as Present.
-					plan.Status.Plan[i].Status = v1alpha1.StepStatusPresent
-				} else if err != nil {
-					return errorwrap.Wrapf(err, "error creating clusterrolebinding %s", rb.GetName())
-				} else {
-					// If no error occurred, mark the step as Created.
-					plan.Status.Plan[i].Status = v1alpha1.StepStatusCreated
+				status, err := ensurer.EnsureClusterRoleBinding(&rb, step)
+				if err != nil {
+					return err
 				}
 
+				plan.Status.Plan[i].Status = status
+
 			case roleKind:
 				// Marshal the manifest into a Role instance.
 				var r rbacv1.Role
@@ -1248,24 +1221,13 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 				r.SetOwnerReferences(updated)
 				r.SetNamespace(namespace)
 
-				// Attempt to create the Role.
-				_, err = o.opClient.KubernetesInterface().RbacV1().Roles(plan.Namespace).Create(&r)
-				if k8serrors.IsAlreadyExists(err) {
-					// If it already existed, mark the step as Present.
-					r.SetNamespace(plan.Namespace)
-					_, err = o.opClient.UpdateRole(&r)
-					if err != nil {
-						return errorwrap.Wrapf(err, "error updating role %s", r.GetName())
-					}
-
-					plan.Status.Plan[i].Status = v1alpha1.StepStatusPresent
-				} else if err != nil {
-					return errorwrap.Wrapf(err, "error creating role %s", r.GetName())
-				} else {
-					// If no error occurred, mark the step as Created.
-					plan.Status.Plan[i].Status = v1alpha1.StepStatusCreated
+				status, err := ensurer.EnsureRole(plan.Namespace, &r)
+				if err != nil {
+					return err
 				}
 
+				plan.Status.Plan[i].Status = status
+
 			case roleBindingKind:
 				// Marshal the manifest into a RoleBinding instance.
 				var rb rbacv1.RoleBinding
@@ -1282,24 +1244,13 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 				rb.SetOwnerReferences(updated)
 				rb.SetNamespace(namespace)
 
-				// Attempt to create the RoleBinding.
-				_, err = o.opClient.KubernetesInterface().RbacV1().RoleBindings(plan.Namespace).Create(&rb)
-				if k8serrors.IsAlreadyExists(err) {
-					rb.SetNamespace(plan.Namespace)
-					_, err = o.opClient.UpdateRoleBinding(&rb)
-					if err != nil {
-						return errorwrap.Wrapf(err, "error updating rolebinding %s", rb.GetName())
-					}
-
-					// If it already existed, mark the step as Present.
-					plan.Status.Plan[i].Status = v1alpha1.StepStatusPresent
-				} else if err != nil {
-					return errorwrap.Wrapf(err, "error creating rolebinding %s", rb.GetName())
-				} else {
-					// If no error occurred, mark the step as Created.
-					plan.Status.Plan[i].Status = v1alpha1.StepStatusCreated
+				status, err := ensurer.EnsureRoleBinding(plan.Namespace, &rb)
+				if err != nil {
+					return err
 				}
 
+				plan.Status.Plan[i].Status = status
+
 			case serviceAccountKind:
 				// Marshal the manifest into a ServiceAccount instance.
 				var sa corev1.ServiceAccount
@@ -1316,25 +1267,13 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 				sa.SetOwnerReferences(updated)
 				sa.SetNamespace(namespace)
 
-				// Attempt to create the ServiceAccount.
-				_, err = o.opClient.KubernetesInterface().CoreV1().ServiceAccounts(plan.Namespace).Create(&sa)
-				if k8serrors.IsAlreadyExists(err) {
-					// If it already exists we need to patch the existing SA with the new OwnerReferences
-					sa.SetNamespace(plan.Namespace)
-					_, err = o.opClient.UpdateServiceAccount(&sa)
-					if err != nil {
-						return errorwrap.Wrapf(err, "error updating service account: %s", sa.GetName())
-					}
-
-					// Mark as present
-					plan.Status.Plan[i].Status = v1alpha1.StepStatusPresent
-				} else if err != nil {
-					return errorwrap.Wrapf(err, "error creating service account: %s", sa.GetName())
-				} else {
-					// If no error occurred, mark the step as Created.
-					plan.Status.Plan[i].Status = v1alpha1.StepStatusCreated
+				status, err := ensurer.EnsureServiceAccount(namespace, &sa)
+				if err != nil {
+					return err
 				}
 
+				plan.Status.Plan[i].Status = status
+
 			case serviceKind:
 				// Marshal the manifest into a Service instance
 				var s corev1.Service
@@ -1351,25 +1290,13 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 				s.SetOwnerReferences(updated)
 				s.SetNamespace(namespace)
 
-				// Attempt to create the Service
-				_, err = o.opClient.KubernetesInterface().CoreV1().Services(plan.Namespace).Create(&s)
-				if k8serrors.IsAlreadyExists(err) {
-					// If it already exists we need to patch the existing SA with the new OwnerReferences
-					s.SetNamespace(plan.Namespace)
-					_, err = o.opClient.UpdateService(&s)
-					if err != nil {
-						return errorwrap.Wrapf(err, "error updating service: %s", s.GetName())
-					}
-
-					// Mark as present
-					plan.Status.Plan[i].Status = v1alpha1.StepStatusPresent
-				} else if err != nil {
-					return errorwrap.Wrapf(err, "error creating service: %s", s.GetName())
-				} else {
-					// If no error occurred, mark the step as Created
-					plan.Status.Plan[i].Status = v1alpha1.StepStatusCreated
+				status, err := ensurer.EnsureService(namespace, &s)
+				if err != nil {
+					return err
 				}
 
+				plan.Status.Plan[i].Status = status
+
 			default:
 				return v1alpha1.ErrInvalidInstallPlan
 			}

pkg/controller/operators/catalog/operator_test.go

@@ -15,6 +15,7 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
+	"k8s.io/client-go/rest"
 	"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
 	apiextensionsfake "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/fake"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -39,6 +40,7 @@ import (
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/operatorlister"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/ownerutil"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/queueinformer"
+	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/scoped"
 )
 
 type mockTransitioner struct {
@@ -663,6 +665,8 @@ func NewFakeOperator(ctx context.Context, namespace string, watchedNamespaces []
 
 	}
 
+	logger := logrus.New()
+
 	// Create the new operator
 	queueOperator, err := queueinformer.NewOperator(opClientFake.KubernetesInterface().Discovery())
 	for _, informer := range sharedInformers {
@@ -685,6 +689,8 @@ func NewFakeOperator(ctx context.Context, namespace string, watchedNamespaces []
 			), "resolver"),
 		sources:  make(map[resolver.CatalogKey]resolver.SourceRef),
 		resolver: &fakes.FakeResolver{},
+		clientAttenuator: scoped.NewClientAttenuator(logger, &rest.Config{}, opClientFake, clientFake),
+		serviceAccountQuerier:  scoped.NewUserDefinedServiceAccountQuerier(logger, clientFake),
 	}
 	op.reconciler = reconciler.NewRegistryReconcilerFactory(lister, op.opClient, "test:pod", op.now)