Repository: operator-framework/operator-lifecycle-manager
Pull request #2499 (2021-11-25 17:27:45)

Address Invalid Address in GRPC Catalogs
Problem: Within the catalogSource resource, the RegistryServiceStatus
stores service information that is used to generate an address that OLM
relies on in order to establish a connection with the associated pod.
If the RegistryStatusService is not nil and is missing the namespace,
name, and port information for its service, OLM is unable to recover
until the catalogService's associated pod has an invalid image or
spec.

Solution: When reconciling a CatalogSource, OLM will now ensure that
the RegistryServiceStatus of the catalogSource is valid and will update
the catalogSource's status to reflect the change.  Additionally, this
address is stored within the status of the catalogSource within the
status.GRPCConnectionState.Address field. If the address changes, OLM
will update this field to reflect the new address as well.

<!--

Before making a PR, please read our contributing guidelines https://github.com/operator-framework/operator-lifecycle-manager/blob/master/CONTRIBUTING.md

Note: Make sure your branch is rebased to the latest upstream master.

-->

**Description of the change:**


**Motivation for the change:**

**Reviewer Checklist**
- [ ] Implementation matches the proposed design, or proposal is updated to match implementation
- [ ] Sufficient unit test coverage
- [ ] Sufficient end-to-end test coverage
- [ ] Docs updated or added to `/doc`
- [ ] Commit messages sensible and descriptive


<!--

Note: If this PR is fixing an issue make sure to add a note saying:
Closes #<ISSUE_NUMBER>

-->


openshift-ci[bot]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/operator-framework/operator-lifecycle-manager/pull/2499#" title="Author self-approved">awgreene</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=operator-framework%2Foperator-lifecycle-manager).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/operator-framework/operator-lifecycle-manager/blob/master/OWNERS)~~ [awgreene]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
openshift-ci[bot]: @awgreene: This pull request references [Bugzilla bug 2026343](https://bugzilla.redhat.com/show_bug.cgi?id=2026343), which is invalid:
 - expected the bug to target the "4.10.0" release, but it targets "---" instead

Comment <code>/bugzilla refresh</code> to re-evaluate validity if changes to the Bugzilla bug are made, or edit the title of this pull request to link to a different bug.

<details>

In response to [this](https://github.com/operator-framework/operator-lifecycle-manager/pull/2499):

>Bug 2026343: Address Invalid Address in GRPC Catalogs


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>
openshift-ci[bot]: @awgreene: No Bugzilla bug is referenced in the title of this pull request.
To reference a bug, add 'Bug XXX:' to the title of this pull request and request another bug refresh with <code>/bugzilla refresh</code>.

<details>

In response to [this](https://github.com/operator-framework/operator-lifecycle-manager/pull/2499):

>Address Invalid Address in GRPC Catalogs


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>
timflannagan: (removed bugzilla reference from the PR title)
awgreene: /retest

Patch:
diff --git a/deploy/chart/templates/0000_50_olm_02-olmconfig.yaml b/deploy/chart/templates/0000_50_olm_02-olmconfig.yaml
new file mode 100644
index 000000000..ad881b7be
--- /dev/null
+++ b/deploy/chart/templates/0000_50_olm_02-olmconfig.yaml
@@ -0,0 +1,4 @@
+apiVersion: operators.coreos.com/v1
+kind: OLMConfig
+metadata:
+  name: cluster
diff --git a/pkg/controller/operators/catalog/operator.go b/pkg/controller/operators/catalog/operator.go
index d152b399a..a749a9c11 100644
--- a/pkg/controller/operators/catalog/operator.go
+++ b/pkg/controller/operators/catalog/operator.go
@@ -803,6 +803,7 @@ func (o *Operator) syncConnection(logger *logrus.Entry, in *v1alpha1.CatalogSour
 		// Set connection status and return.
 		out.Status.GRPCConnectionState.LastConnectTime = now
 		out.Status.GRPCConnectionState.LastObservedState = source.ConnectionState.String()
+		out.Status.GRPCConnectionState.Address = source.Address
 	}
 
 	return
diff --git a/pkg/controller/operators/olm/operator.go b/pkg/controller/operators/olm/operator.go
index 47586f997..23a6e021c 100644
--- a/pkg/controller/operators/olm/operator.go
+++ b/pkg/controller/operators/olm/operator.go
@@ -14,9 +14,11 @@ import (
 	rbacv1 "k8s.io/api/rbac/v1"
 	extinf "k8s.io/apiextensions-apiserver/pkg/client/informers/externalversions"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/selection"
 	utilclock "k8s.io/apimachinery/pkg/util/clock"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
@@ -68,6 +70,7 @@ type Operator struct {
 	copiedCSVLister       operatorsv1alpha1listers.ClusterServiceVersionLister
 	ogQueueSet            *queueinformer.ResourceQueueSet
 	csvQueueSet           *queueinformer.ResourceQueueSet
+	olmConfigQueue        workqueue.RateLimitingInterface
 	csvCopyQueueSet       *queueinformer.ResourceQueueSet
 	copiedCSVGCQueueSet   *queueinformer.ResourceQueueSet
 	objGCQueueSet         *queueinformer.ResourceQueueSet
@@ -124,6 +127,7 @@ func newOperatorWithConfig(ctx context.Context, config *operatorConfig) (*Operat
 		client:                config.externalClient,
 		ogQueueSet:            queueinformer.NewEmptyResourceQueueSet(),
 		csvQueueSet:           queueinformer.NewEmptyResourceQueueSet(),
+		olmConfigQueue:        workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "olmConfig"),
 		csvCopyQueueSet:       queueinformer.NewEmptyResourceQueueSet(),
 		copiedCSVGCQueueSet:   queueinformer.NewEmptyResourceQueueSet(),
 		objGCQueueSet:         queueinformer.NewEmptyResourceQueueSet(),
@@ -433,6 +437,26 @@ func newOperatorWithConfig(ctx context.Context, config *operatorConfig) (*Operat
 		return nil, err
 	}
 
+	// Register QueueInformer for olmConfig
+	olmConfigInformer := externalversions.NewSharedInformerFactoryWithOptions(
+		op.client,
+		config.resyncPeriod(),
+	).Operators().V1().OLMConfigs().Informer()
+	olmConfigQueueInformer, err := queueinformer.NewQueueInformer(
+		ctx,
+		queueinformer.WithInformer(olmConfigInformer),
+		queueinformer.WithLogger(op.logger),
+		queueinformer.WithQueue(op.olmConfigQueue),
+		queueinformer.WithIndexer(olmConfigInformer.GetIndexer()),
+		queueinformer.WithSyncer(queueinformer.LegacySyncHandler(op.syncOLMConfig).ToSyncer()),
+	)
+	if err != nil {
+		return nil, err
+	}
+	if err := op.RegisterQueueInformer(olmConfigQueueInformer); err != nil {
+		return nil, err
+	}
+
 	k8sInformerFactory := informers.NewSharedInformerFactory(op.opClient.KubernetesInterface(), config.resyncPeriod())
 	clusterRoleInformer := k8sInformerFactory.Rbac().V1().ClusterRoles()
 	op.lister.RbacV1().RegisterClusterRoleLister(clusterRoleInformer.Lister())
@@ -1211,6 +1235,127 @@ func (a *Operator) syncClusterServiceVersion(obj interface{}) (syncError error)
 	return
 }
 
+func (a *Operator) allNamespaceOperatorGroups() ([]*v1.OperatorGroup, error) {
+	operatorGroups, err := a.lister.OperatorsV1().OperatorGroupLister().List(labels.Everything())
+	if err != nil {
+		return nil, err
+	}
+
+	result := []*v1.OperatorGroup{}
+	for _, operatorGroup := range operatorGroups {
+		if NewNamespaceSet(operatorGroup.Status.Namespaces).IsAllNamespaces() {
+			result = append(result, operatorGroup)
+		}
+	}
+	return result, nil
+}
+
+func (a *Operator) syncOLMConfig(obj interface{}) (syncError error) {
+	a.logger.Info("Processing olmConfig")
+	olmConfig, ok := obj.(*v1.OLMConfig)
+	if !ok {
+		return fmt.Errorf("casting OLMConfig failed")
+	}
+
+	// Generate an array of allNamespace OperatorGroups
+	allNSOperatorGroups, err := a.allNamespaceOperatorGroups()
+	if err != nil {
+		return err
+	}
+
+	nonCopiedCSVRequirement, err := labels.NewRequirement(v1alpha1.CopiedLabelKey, selection.DoesNotExist, []string{})
+	if err != nil {
+		return err
+	}
+
+	csvIsRequeued := false
+	for _, og := range allNSOperatorGroups {
+		// Get all copied CSVs owned by this operatorGroup
+		copiedCSVRequirement, err := labels.NewRequirement(v1alpha1.CopiedLabelKey, selection.Equals, []string{og.GetNamespace()})
+		if err != nil {
+			return err
+		}
+
+		copiedCSVs, err := a.copiedCSVLister.List(labels.NewSelector().Add(*copiedCSVRequirement))
+		if err != nil {
+			return err
+		}
+
+		// Filter to unique copies
+		uniqueCopiedCSVs := map[string]struct{}{}
+		for _, copiedCSV := range copiedCSVs {
+			uniqueCopiedCSVs[copiedCSV.GetName()] = struct{}{}
+		}
+
+		csvs, err := a.lister.OperatorsV1alpha1().ClusterServiceVersionLister().ClusterServiceVersions(og.GetNamespace()).List(labels.NewSelector().Add(*nonCopiedCSVRequirement))
+		if err != nil {
+			return err
+		}
+
+		for _, csv := range csvs {
+			// If the correct number of copied CSVs were found, continue
+			if _, ok := uniqueCopiedCSVs[csv.GetName()]; ok == olmConfig.CopiedCSVsAreEnabled() {
+				continue
+			}
+
+			if err := a.csvQueueSet.Requeue(csv.GetNamespace(), csv.GetName()); err != nil {
+				a.logger.WithError(err).Warn("unable to requeue")
+			}
+			csvIsRequeued = true
+		}
+	}
+
+	// Update the olmConfig status if it has changed.
+	condition := getCopiedCSVsCondition(!olmConfig.CopiedCSVsAreEnabled(), csvIsRequeued)
+	if !isStatusConditionPresentAndAreTypeReasonMessageStatusEqual(olmConfig.Status.Conditions, condition) {
+		meta.SetStatusCondition(&olmConfig.Status.Conditions, condition)
+		if _, err := a.client.OperatorsV1().OLMConfigs().UpdateStatus(context.TODO(), olmConfig, metav1.UpdateOptions{}); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func isStatusConditionPresentAndAreTypeReasonMessageStatusEqual(conditions []metav1.Condition, condition metav1.Condition) bool {
+	foundCondition := meta.FindStatusCondition(conditions, condition.Type)
+	if foundCondition == nil {
+		return false
+	}
+	return foundCondition.Type == condition.Type &&
+		foundCondition.Reason == condition.Reason &&
+		foundCondition.Message == condition.Message &&
+		foundCondition.Status == condition.Status
+}
+
+func getCopiedCSVsCondition(isDisabled, csvIsRequeued bool) metav1.Condition {
+	condition := metav1.Condition{
+		Type:               v1.DisabledCopiedCSVsConditionType,
+		LastTransitionTime: metav1.Now(),
+		Status:             metav1.ConditionFalse,
+	}
+	if !isDisabled {
+		condition.Reason = "CopiedCSVsEnabled"
+		condition.Message = "Copied CSVs are enabled and present accross the cluster"
+		if csvIsRequeued {
+			condition.Message = "Copied CSVs are enabled and at least one copied CSVs is missing"
+		}
+		return condition
+	}
+
+	if csvIsRequeued {
+		condition.Reason = "CopiedCSVsFound"
+		condition.Message = "Copied CSVs are disabled and at least one copied CSV was found for an operator installed in AllNamespace mode"
+		return condition
+	}
+
+	condition.Status = metav1.ConditionTrue
+	condition.Reason = "NoCopiedCSVsFound"
+	condition.Message = "Copied CSVs are disabled and none were found for operators installed in AllNamespace mode"
+
+	return condition
+}
+
 func (a *Operator) syncCopyCSV(obj interface{}) (syncError error) {
 	clusterServiceVersion, ok := obj.(*v1alpha1.ClusterServiceVersion)
 	if !ok {
@@ -1218,6 +1363,15 @@ func (a *Operator) syncCopyCSV(obj interface{}) (syncError error) {
 		return fmt.Errorf("casting ClusterServiceVersion failed")
 	}
 
+	olmConfig, err := a.client.OperatorsV1().OLMConfigs().Get(context.TODO(), "cluster", metav1.GetOptions{})
+	if err != nil && !k8serrors.IsNotFound(err) {
+		return err
+	}
+
+	if err == nil {
+		go a.olmConfigQueue.AddAfter(olmConfig, time.Second*5)
+	}
+
 	logger := a.logger.WithFields(logrus.Fields{
 		"id":        queueinformer.NewLoopID(),
 		"csv":       clusterServiceVersion.GetName(),
@@ -1239,15 +1393,145 @@ func (a *Operator) syncCopyCSV(obj interface{}) (syncError error) {
 		"targetNamespaces": strings.Join(operatorGroup.Status.Namespaces, ","),
 	}).Debug("copying csv to targets")
 
+	copiedCSVsAreEnabled, err := a.copiedCSVsAreEnabled()
+	if err != nil {
+		return err
+	}
+
 	// Check if we need to do any copying / annotation for the operatorgroup
-	if err := a.ensureCSVsInNamespaces(clusterServiceVersion, operatorGroup, NewNamespaceSet(operatorGroup.Status.Namespaces)); err != nil {
-		logger.WithError(err).Info("couldn't copy CSV to target namespaces")
-		syncError = err
+	namespaceSet := NewNamespaceSet(operatorGroup.Status.Namespaces)
+	if copiedCSVsAreEnabled || !namespaceSet.IsAllNamespaces() {
+		if err := a.ensureCSVsInNamespaces(clusterServiceVersion, operatorGroup, namespaceSet); err != nil {
+			logger.WithError(err).Info("couldn't copy CSV to target namespaces")
+			syncError = err
+		}
+
+		// If the CSV was installed in AllNamespace mode, remove any "CSV Copying Disabled" events
+		// in which the related object's name, namespace, and uid match the given CSV's.
+		if namespaceSet.IsAllNamespaces() {
+			if err := a.deleteCSVCopyingDisabledEvent(clusterServiceVersion); err != nil {
+				return err
+			}
+		}
+		return
+	}
+
+	requirement, err := labels.NewRequirement(v1alpha1.CopiedLabelKey, selection.Equals, []string{clusterServiceVersion.Namespace})
+	if err != nil {
+		return err
+	}
+
+	copiedCSVs, err := a.copiedCSVLister.List(labels.NewSelector().Add(*requirement))
+	if err != nil {
+		return err
+	}
+
+	for _, copiedCSV := range copiedCSVs {
+		err := a.client.OperatorsV1alpha1().ClusterServiceVersions(copiedCSV.Namespace).Delete(context.TODO(), copiedCSV.Name, metav1.DeleteOptions{})
+		if err != nil && !k8serrors.IsNotFound(err) {
+			return err
+		}
+	}
+
+	if err := a.createCSVCopyingDisabledEvent(clusterServiceVersion); err != nil {
+		return err
 	}
 
 	return
 }
 
+// copiedCSVsAreEnabled determines if csv copying is enabled for OLM.
+//
+// This method will first attempt to get the "cluster" olmConfig resource,
+// if any error other than "IsNotFound" is encountered, false and the error
+// will be returned.
+//
+// If the "cluster" olmConfig resource is found, the value of
+// olmConfig.spec.features.disableCopiedCSVs will be returned along with a
+// nil error.
+//
+// If the "cluster" olmConfig resource is not found, true will be returned
+// without an error.
+func (a *Operator) copiedCSVsAreEnabled() (bool, error) {
+	olmConfig, err := a.client.OperatorsV1().OLMConfigs().Get(context.TODO(), "cluster", metav1.GetOptions{})
+	if err != nil {
+		// Default to true if olmConfig singleton cannot be found
+		if k8serrors.IsNotFound(err) {
+			return true, nil
+		}
+		// If there was an error that wasn't an IsNotFound, return the error
+		return false, err
+	}
+
+	// If there was no error, return value based on olmConfig singleton
+	return olmConfig.CopiedCSVsAreEnabled(), nil
+}
+
+func (a *Operator) getCopiedCSVDisabledEventsForCSV(csv *v1alpha1.ClusterServiceVersion) ([]corev1.Event, error) {
+	result := []corev1.Event{}
+	if csv == nil {
+		return result, nil
+	}
+
+	events, err := a.opClient.KubernetesInterface().CoreV1().Events(csv.GetNamespace()).List(context.TODO(), metav1.ListOptions{})
+	if err != nil {
+		return nil, err
+	}
+
+	for _, event := range events.Items {
+		if event.InvolvedObject.Namespace == csv.GetNamespace() &&
+			event.InvolvedObject.Name == csv.GetName() &&
+			event.InvolvedObject.UID == csv.GetUID() &&
+			event.Reason == v1.DisabledCopiedCSVsConditionType {
+			result = append(result, event)
+		}
+	}
+
+	return result, nil
+}
+
+func (a *Operator) deleteCSVCopyingDisabledEvent(csv *v1alpha1.ClusterServiceVersion) error {
+	events, err := a.getCopiedCSVDisabledEventsForCSV(csv)
+	if err != nil {
+		return err
+	}
+
+	// Remove existing events.
+	return a.deleteEvents(events)
+}
+
+func (a *Operator) deleteEvents(events []corev1.Event) error {
+	for _, event := range events {
+		err := a.opClient.KubernetesInterface().EventsV1().Events(event.GetNamespace()).Delete(context.TODO(), event.GetName(), metav1.DeleteOptions{})
+		if err != nil && !k8serrors.IsNotFound(err) {
+			return err
+		}
+	}
+	return nil
+}
+
+func (a *Operator) createCSVCopyingDisabledEvent(csv *v1alpha1.ClusterServiceVersion) error {
+	events, err := a.getCopiedCSVDisabledEventsForCSV(csv)
+	if err != nil {
+		return err
+	}
+
+	if len(events) == 1 {
+		return nil
+	}
+
+	// Remove existing events.
+	if len(events) > 1 {
+		if err := a.deleteEvents(events); err != nil {
+			return err
+		}
+	}
+
+	a.recorder.Eventf(csv, corev1.EventTypeWarning, v1.DisabledCopiedCSVsConditionType, "CSV copying disabled for %s/%s", csv.GetNamespace(), csv.GetName())
+
+	return nil
+}
+
 func (a *Operator) syncGcCsv(obj interface{}) (syncError error) {
 	clusterServiceVersion, ok := obj.(*v1alpha1.ClusterServiceVersion)
 	if !ok {
diff --git a/pkg/controller/registry/reconciler/grpc.go b/pkg/controller/registry/reconciler/grpc.go
index 8406740be..145d29bce 100644
--- a/pkg/controller/registry/reconciler/grpc.go
+++ b/pkg/controller/registry/reconciler/grpc.go
@@ -191,7 +191,7 @@ func (c *GrpcRegistryReconciler) EnsureRegistryServer(catalogSource *v1alpha1.Ca
 	source := grpcCatalogSourceDecorator{catalogSource}
 
 	// if service status is nil, we force create every object to ensure they're created the first time
-	overwrite := source.Status.RegistryServiceStatus == nil
+	overwrite := source.Status.RegistryServiceStatus == nil || !isRegistryServiceStatusValid(&source)
 
 	//TODO: if any of these error out, we should write a status back (possibly set RegistryServiceStatus to nil so they get recreated)
 	sa, err := c.ensureSA(source)
@@ -216,17 +216,33 @@ func (c *GrpcRegistryReconciler) EnsureRegistryServer(catalogSource *v1alpha1.Ca
 
 	if overwritePod {
 		now := c.now()
+		service := source.Service()
 		catalogSource.Status.RegistryServiceStatus = &v1alpha1.RegistryServiceStatus{
 			CreatedAt:        now,
 			Protocol:         "grpc",
-			ServiceName:      source.Service().GetName(),
+			ServiceName:      service.GetName(),
 			ServiceNamespace: source.GetNamespace(),
-			Port:             fmt.Sprintf("%d", source.Service().Spec.Ports[0].Port),
+			Port:             getPort(service),
 		}
 	}
 	return nil
 }
 
+func getPort(service *corev1.Service) string {
+	return fmt.Sprintf("%d", service.Spec.Ports[0].Port)
+}
+
+func isRegistryServiceStatusValid(source *grpcCatalogSourceDecorator) bool {
+	service := source.Service()
+	if source.Status.RegistryServiceStatus.ServiceName != service.GetName() ||
+		source.Status.RegistryServiceStatus.ServiceNamespace != service.GetNamespace() ||
+		source.Status.RegistryServiceStatus.Port != getPort(service) ||
+		source.Status.RegistryServiceStatus.Protocol != "grpc" {
+		return false
+	}
+	return true
+}
+
 func (c *GrpcRegistryReconciler) ensurePod(source grpcCatalogSourceDecorator, saName string, overwrite bool) error {
 	// currentLivePods refers to the currently live instances of the catalog source
 	currentLivePods := c.currentPods(source)


Test patch:
diff --git a/pkg/controller/operators/catalog/operator_test.go b/pkg/controller/operators/catalog/operator_test.go
index f6aceb8bc..8abd022d9 100644
--- a/pkg/controller/operators/catalog/operator_test.go
+++ b/pkg/controller/operators/catalog/operator_test.go
@@ -49,6 +49,7 @@ import (
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/clientset/versioned/fake"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/informers/externalversions"
 	olmerrors "github.com/operator-framework/operator-lifecycle-manager/pkg/controller/errors"
+	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/registry"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/registry/grpc"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/registry/reconciler"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/registry/resolver"
@@ -758,6 +759,12 @@ func TestExecutePlanDynamicResources(t *testing.T) {
 	}
 }
 
+func withStatus(catalogSource v1alpha1.CatalogSource, status v1alpha1.CatalogSourceStatus) *v1alpha1.CatalogSource {
+	copy := catalogSource.DeepCopy()
+	copy.Status = status
+	return copy
+}
+
 func TestSyncCatalogSources(t *testing.T) {
 	clockFake := utilclock.NewFakeClock(time.Date(2018, time.January, 26, 20, 40, 0, 0, time.UTC))
 	now := metav1.NewTime(clockFake.Now())
@@ -786,14 +793,15 @@ func TestSyncCatalogSources(t *testing.T) {
 		},
 	}
 	tests := []struct {
-		testName       string
-		namespace      string
-		catalogSource  *v1alpha1.CatalogSource
-		k8sObjs        []runtime.Object
-		configMap      *corev1.ConfigMap
-		expectedStatus *v1alpha1.CatalogSourceStatus
-		expectedObjs   []runtime.Object
-		expectedError  error
+		testName        string
+		namespace       string
+		catalogSource   *v1alpha1.CatalogSource
+		k8sObjs         []runtime.Object
+		configMap       *corev1.ConfigMap
+		expectedStatus  *v1alpha1.CatalogSourceStatus
+		expectedObjs    []runtime.Object
+		expectedError   error
+		existingSources []sourceAddress
 	}{
 		{
 			testName:  "CatalogSourceWithInvalidSourceType",
@@ -1013,6 +1021,47 @@ func TestSyncCatalogSources(t *testing.T) {
 			},
 			expectedError: nil,
 		},
+		{
+			testName:  "GRPCConnectionStateAddressIsUpdated",
+			namespace: "cool-namespace",
+			catalogSource: withStatus(*grpcCatalog, v1alpha1.CatalogSourceStatus{
+				RegistryServiceStatus: &v1alpha1.RegistryServiceStatus{
+					Protocol:         "grpc",
+					ServiceName:      "cool-catalog",
+					ServiceNamespace: "cool-namespace",
+					Port:             "50051",
+					CreatedAt:        now,
+				},
+				GRPCConnectionState: &v1alpha1.GRPCConnectionState{
+					Address: "..svc:", // Needs to be updated to cool-catalog.cool-namespace.svc:50051
+				},
+			}),
+			k8sObjs: []runtime.Object{
+				pod(*grpcCatalog),
+				service(grpcCatalog.GetName(), grpcCatalog.GetNamespace()),
+			},
+			existingSources: []sourceAddress{
+				{
+					sourceKey: registry.CatalogKey{Name: "cool-catalog", Namespace: "cool-namespace"},
+					address:   "cool-catalog.cool-namespace.svc:50051",
+				},
+			},
+			expectedStatus: &v1alpha1.CatalogSourceStatus{
+				RegistryServiceStatus: &v1alpha1.RegistryServiceStatus{
+					Protocol:         "grpc",
+					ServiceName:      "cool-catalog",
+					ServiceNamespace: "cool-namespace",
+					Port:             "50051",
+					CreatedAt:        now,
+				},
+				GRPCConnectionState: &v1alpha1.GRPCConnectionState{
+					Address:           "cool-catalog.cool-namespace.svc:50051",
+					LastObservedState: "",
+					LastConnectTime:   now,
+				},
+			},
+			expectedError: nil,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.testName, func(t *testing.T) {
@@ -1023,7 +1072,7 @@ func TestSyncCatalogSources(t *testing.T) {
 			ctx, cancel := context.WithCancel(context.TODO())
 			defer cancel()
 
-			op, err := NewFakeOperator(ctx, tt.namespace, []string{tt.namespace}, withClock(clockFake), withClientObjs(clientObjs...), withK8sObjs(tt.k8sObjs...))
+			op, err := NewFakeOperator(ctx, tt.namespace, []string{tt.namespace}, withClock(clockFake), withClientObjs(clientObjs...), withK8sObjs(tt.k8sObjs...), withSources(tt.existingSources...))
 			require.NoError(t, err)
 
 			// Run sync
@@ -1040,6 +1089,13 @@ func TestSyncCatalogSources(t *testing.T) {
 			require.NotEmpty(t, updated)
 
 			if tt.expectedStatus != nil {
+				if tt.expectedStatus.GRPCConnectionState != nil {
+					updated.Status.GRPCConnectionState.LastConnectTime = now
+					// Ignore LastObservedState difference if an expected LastObservedState is no provided
+					if tt.expectedStatus.GRPCConnectionState.LastObservedState == "" {
+						updated.Status.GRPCConnectionState.LastObservedState = ""
+					}
+				}
 				require.NotEmpty(t, updated.Status)
 				require.Equal(t, *tt.expectedStatus, updated.Status)
 
@@ -1384,6 +1440,7 @@ type fakeOperatorConfig struct {
 	resolver      resolver.StepResolver
 	recorder      record.EventRecorder
 	reconciler    reconciler.RegistryReconcilerFactory
+	sources       []sourceAddress
 }
 
 // fakeOperatorOption applies an option to the given fake operator configuration.
@@ -1395,6 +1452,12 @@ func withResolver(res resolver.StepResolver) fakeOperatorOption {
 	}
 }
 
+func withSources(sources ...sourceAddress) fakeOperatorOption {
+	return func(config *fakeOperatorConfig) {
+		config.sources = sources
+	}
+}
+
 func withReconciler(rec reconciler.RegistryReconcilerFactory) fakeOperatorOption {
 	return func(config *fakeOperatorConfig) {
 		config.reconciler = rec
@@ -1431,6 +1494,11 @@ func withFakeClientOptions(options ...clientfake.Option) fakeOperatorOption {
 	}
 }
 
+type sourceAddress struct {
+	address   string
+	sourceKey registry.CatalogKey
+}
+
 // NewFakeOperator creates a new operator using fake clients.
 func NewFakeOperator(ctx context.Context, namespace string, namespaces []string, fakeOptions ...fakeOperatorOption) (*Operator, error) {
 	// Apply options to default config
@@ -1548,6 +1616,9 @@ func NewFakeOperator(ctx context.Context, namespace string, namespaces []string,
 
 	op.RunInformers(ctx)
 	op.sources.Start(ctx)
+	for _, source := range config.sources {
+		op.sources.Add(source.sourceKey, source.address)
+	}
 
 	if ok := cache.WaitForCacheSync(ctx.Done(), op.HasSynced); !ok {
 		return nil, fmt.Errorf("failed to wait for caches to sync")
diff --git a/pkg/controller/registry/reconciler/grpc_test.go b/pkg/controller/registry/reconciler/grpc_test.go
index 882728621..6ac340648 100644
--- a/pkg/controller/registry/reconciler/grpc_test.go
+++ b/pkg/controller/registry/reconciler/grpc_test.go
@@ -49,6 +49,11 @@ func grpcCatalogSourceWithSecret(secretNames []string) *v1alpha1.CatalogSource {
 		},
 	}
 }
+func grpcCatalogSourceWithStatus(status v1alpha1.CatalogSourceStatus) *v1alpha1.CatalogSource {
+	catsrc := validGrpcCatalogSource("image", "")
+	catsrc.Status = status
+	return catsrc
+}
 
 func grpcCatalogSourceWithAnnotations(annotations map[string]string) *v1alpha1.CatalogSource {
 	catsrc := validGrpcCatalogSource("image", "")
@@ -284,6 +289,29 @@ func TestGrpcRegistryReconciler(t *testing.T) {
 				},
 			},
 		},
+		{
+			testName: "Grpc/ExistingRegistry/UpdateInvalidRegistryServiceStatus",
+			in: in{
+				cluster: cluster{
+					k8sObjs: objectsForCatalogSource(validGrpcCatalogSource("image", "")),
+				},
+				catsrc: grpcCatalogSourceWithStatus(v1alpha1.CatalogSourceStatus{
+					RegistryServiceStatus: &v1alpha1.RegistryServiceStatus{
+						CreatedAt: now(),
+						Protocol:  "grpc",
+					},
+				}),
+			},
+			out: out{
+				status: &v1alpha1.RegistryServiceStatus{
+					CreatedAt:        now(),
+					Protocol:         "grpc",
+					ServiceName:      "img-catalog",
+					ServiceNamespace: testNamespace,
+					Port:             "50051",
+				},
+			},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.testName, func(t *testing.T) {
diff --git a/test/e2e/csv_e2e_test.go b/test/e2e/csv_e2e_test.go
index 2f7bf6ca8..6d3e95ee8 100644
--- a/test/e2e/csv_e2e_test.go
+++ b/test/e2e/csv_e2e_test.go
@@ -18,9 +18,12 @@ import (
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apimachinery/pkg/api/equality"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	k8slabels "k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/selection"
+	apitypes "k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/diff"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/apimachinery/pkg/util/wait"
@@ -4171,6 +4174,253 @@ var _ = Describe("ClusterServiceVersion", func() {
 	})
 })
 
+var _ = Describe("Disabling copied CSVs", func() {
+	// Define namespace, operatorGroup, and csv upfront
+	ns := &corev1.Namespace{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: genName("csv-toggle-test-"),
+		},
+	}
+
+	operatorGroup := operatorsv1.OperatorGroup{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      genName("csv-toggle-test-"),
+			Namespace: ns.GetName(),
+		},
+	}
+
+	csv := operatorsv1alpha1.ClusterServiceVersion{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      genName("csv-toggle-test-"),
+			Namespace: ns.GetName(),
+		},
+		Spec: operatorsv1alpha1.ClusterServiceVersionSpec{
+			InstallStrategy: newNginxInstallStrategy(genName("csv-toggle-test-"), nil, nil),
+			InstallModes: []operatorsv1alpha1.InstallMode{
+				{
+					Type:      operatorsv1alpha1.InstallModeTypeAllNamespaces,
+					Supported: true,
+				},
+			},
+		},
+	}
+
+	When("an operator is installed in AllNamespace mode", func() {
+		BeforeEach(func() {
+			Eventually(func() error {
+				if err := ctx.Ctx().Client().Create(context.TODO(), ns); err != nil && !k8serrors.IsAlreadyExists(err) {
+					ctx.Ctx().Logf("Unable to create ns: %v", err)
+					return err
+				}
+
+				if err := ctx.Ctx().Client().Create(context.TODO(), &operatorGroup); err != nil && !k8serrors.IsAlreadyExists(err) {
+					ctx.Ctx().Logf("Unable to create og: %v", err)
+					return err
+				}
+
+				if err := ctx.Ctx().Client().Create(context.TODO(), &csv); err != nil && !k8serrors.IsAlreadyExists(err) {
+					ctx.Ctx().Logf("Unable to create csv: %v", err)
+					return err
+				}
+
+				return nil
+			}).Should(Succeed())
+		})
+
+		It("should have Copied CSVs in all other namespaces", func() {
+			Eventually(func() error {
+				requirement, err := k8slabels.NewRequirement(operatorsv1alpha1.CopiedLabelKey, selection.Equals, []string{csv.GetNamespace()})
+				if err != nil {
+					return err
+				}
+
+				var copiedCSVs operatorsv1alpha1.ClusterServiceVersionList
+				err = ctx.Ctx().Client().List(context.TODO(), &copiedCSVs, &client.ListOptions{
+					LabelSelector: k8slabels.NewSelector().Add(*requirement),
+				})
+				if err != nil {
+					return err
+				}
+
+				var namespaces corev1.NamespaceList
+				if err := ctx.Ctx().Client().List(context.TODO(), &namespaces, &client.ListOptions{}); err != nil {
+					return err
+				}
+
+				if len(namespaces.Items)-1 != len(copiedCSVs.Items) {
+					return fmt.Errorf("%d copied CSVs found, expected %d", len(copiedCSVs.Items), len(namespaces.Items)-1)
+				}
+
+				return nil
+			}).Should(Succeed())
+		})
+	})
+
+	When("Copied CSVs are disabled", func() {
+		BeforeEach(func() {
+			Eventually(func() error {
+				var olmConfig operatorsv1.OLMConfig
+				if err := ctx.Ctx().Client().Get(context.TODO(), apitypes.NamespacedName{Name: "cluster"}, &olmConfig); err != nil {
+					ctx.Ctx().Logf("Error getting olmConfig %v", err)
+					return err
+				}
+
+				// Exit early if copied CSVs are disabled.
+				if !olmConfig.CopiedCSVsAreEnabled() {
+					return nil
+				}
+
+				olmConfig.Spec = operatorsv1.OLMConfigSpec{
+					Features: &operatorsv1.Features{
+						DisableCopiedCSVs: getPointer(true),
+					},
+				}
+
+				if err := ctx.Ctx().Client().Update(context.TODO(), &olmConfig); err != nil {
+					ctx.Ctx().Logf("Error setting olmConfig %v", err)
+					return err
+				}
+
+				return nil
+			}).Should(Succeed())
+		})
+
+		It("should not have any copied CSVs", func() {
+			Eventually(func() error {
+				requirement, err := k8slabels.NewRequirement(operatorsv1alpha1.CopiedLabelKey, selection.Equals, []string{csv.GetNamespace()})
+				if err != nil {
+					return err
+				}
+
+				var copiedCSVs operatorsv1alpha1.ClusterServiceVersionList
+				err = ctx.Ctx().Client().List(context.TODO(), &copiedCSVs, &client.ListOptions{
+					LabelSelector: k8slabels.NewSelector().Add(*requirement),
+				})
+				if err != nil {
+					return err
+				}
+
+				if numCSVs := len(copiedCSVs.Items); numCSVs != 0 {
+					return fmt.Errorf("Found %d copied CSVs, should be 0", numCSVs)
+				}
+				return nil
+			}).Should(Succeed())
+		})
+
+		It("should be reflected in the olmConfig.Status.Condition array that the expected number of copied CSVs exist", func() {
+			Eventually(func() error {
+				var olmConfig operatorsv1.OLMConfig
+				if err := ctx.Ctx().Client().Get(context.TODO(), apitypes.NamespacedName{Name: "cluster"}, &olmConfig); err != nil {
+					return err
+				}
+
+				foundCondition := meta.FindStatusCondition(olmConfig.Status.Conditions, operatorsv1.DisabledCopiedCSVsConditionType)
+				if foundCondition == nil {
+					return fmt.Errorf("%s condition not found", operatorsv1.DisabledCopiedCSVsConditionType)
+				}
+
+				expectedCondition := metav1.Condition{
+					Reason:  "NoCopiedCSVsFound",
+					Message: "Copied CSVs are disabled and none were found for operators installed in AllNamespace mode",
+					Status:  metav1.ConditionTrue,
+				}
+
+				if foundCondition.Reason != expectedCondition.Reason ||
+					foundCondition.Message != expectedCondition.Message ||
+					foundCondition.Status != expectedCondition.Status {
+					return fmt.Errorf("condition does not have expected reason, message, and status. Expected %v, got %v", expectedCondition, foundCondition)
+				}
+
+				return nil
+			}).Should(Succeed())
+		})
+	})
+
+	When("Copied CSVs are toggled back on", func() {
+		BeforeEach(func() {
+			Eventually(func() error {
+				var olmConfig operatorsv1.OLMConfig
+				if err := ctx.Ctx().Client().Get(context.TODO(), apitypes.NamespacedName{Name: "cluster"}, &olmConfig); err != nil {
+					return err
+				}
+
+				// Exit early if copied CSVs are enabled.
+				if olmConfig.CopiedCSVsAreEnabled() {
+					return nil
+				}
+
+				olmConfig.Spec = operatorsv1.OLMConfigSpec{
+					Features: &operatorsv1.Features{
+						DisableCopiedCSVs: getPointer(false),
+					},
+				}
+
+				if err := ctx.Ctx().Client().Update(context.TODO(), &olmConfig); err != nil {
+					return err
+				}
+
+				return nil
+			}).Should(Succeed())
+		})
+
+		It("should have copied CSVs in all other Namespaces", func() {
+			Eventually(func() error {
+				// find copied csvs...
+				requirement, err := k8slabels.NewRequirement(operatorsv1alpha1.CopiedLabelKey, selection.Equals, []string{csv.GetNamespace()})
+				if err != nil {
+					return err
+				}
+
+				var copiedCSVs operatorsv1alpha1.ClusterServiceVersionList
+				err = ctx.Ctx().Client().List(context.TODO(), &copiedCSVs, &client.ListOptions{
+					LabelSelector: k8slabels.NewSelector().Add(*requirement),
+				})
+				if err != nil {
+					return err
+				}
+
+				var namespaces corev1.NamespaceList
+				if err := ctx.Ctx().Client().List(context.TODO(), &namespaces, &client.ListOptions{}); err != nil {
+					return err
+				}
+
+				if len(namespaces.Items)-1 != len(copiedCSVs.Items) {
+					return fmt.Errorf("%d copied CSVs found, expected %d", len(copiedCSVs.Items), len(namespaces.Items)-1)
+				}
+
+				return nil
+			}).Should(Succeed())
+		})
+
+		It("should be reflected in the olmConfig.Status.Condition array that the expected number of copied CSVs exist", func() {
+			Eventually(func() error {
+				var olmConfig operatorsv1.OLMConfig
+				if err := ctx.Ctx().Client().Get(context.TODO(), apitypes.NamespacedName{Name: "cluster"}, &olmConfig); err != nil {
+					return err
+				}
+				foundCondition := meta.FindStatusCondition(olmConfig.Status.Conditions, operatorsv1.DisabledCopiedCSVsConditionType)
+				if foundCondition == nil {
+					return fmt.Errorf("%s condition not found", operatorsv1.DisabledCopiedCSVsConditionType)
+				}
+
+				expectedCondition := metav1.Condition{
+					Reason:  "CopiedCSVsEnabled",
+					Message: "Copied CSVs are enabled and present accross the cluster",
+					Status:  metav1.ConditionFalse,
+				}
+
+				if foundCondition.Reason != expectedCondition.Reason ||
+					foundCondition.Message != expectedCondition.Message ||
+					foundCondition.Status != expectedCondition.Status {
+					return fmt.Errorf("condition does not have expected reason, message, and status. Expected %v, got %v", expectedCondition, foundCondition)
+				}
+
+				return nil
+			}).Should(Succeed())
+		})
+	})
+})
+
 var singleInstance = int32(1)
 
 type cleanupFunc func()
@@ -4219,6 +4469,10 @@ func buildCSVCleanupFunc(c operatorclient.ClientInterface, crc versioned.Interfa
 	}
 }
 
+func getPointer(b bool) *bool {
+	return &b
+}
+
 func createCSV(c operatorclient.ClientInterface, crc versioned.Interface, csv operatorsv1alpha1.ClusterServiceVersion, namespace string, cleanupCRDs, cleanupAPIServices bool) (cleanupFunc, error) {
 	csv.Kind = operatorsv1alpha1.ClusterServiceVersionKind
 	csv.APIVersion = operatorsv1alpha1.SchemeGroupVersion.String()