diff --git a/manifests/0000_50_olm_02-olmconfig.yaml b/manifests/0000_50_olm_02-olmconfig.yaml
new file mode 100644
index 0000000000..f485ac7917
--- /dev/null
+++ b/manifests/0000_50_olm_02-olmconfig.yaml
@@ -0,0 +1,9 @@
+apiVersion: operators.coreos.com/v1
+kind: OLMConfig
+metadata:
+  name: cluster
+  annotations:
+    release.openshift.io/create-only: "true"
+    include.release.openshift.io/ibm-cloud-managed: "true"
+    include.release.openshift.io/self-managed-high-availability: "true"
+    include.release.openshift.io/single-node-developer: "true"
diff --git a/scripts/cluster-olmconfig.patch.yaml b/scripts/cluster-olmconfig.patch.yaml
new file mode 100644
index 0000000000..9912a17a50
--- /dev/null
+++ b/scripts/cluster-olmconfig.patch.yaml
@@ -0,0 +1,3 @@
+metadata:
+  annotations:
+    release.openshift.io/create-only: "true"
diff --git a/scripts/generate_crds_manifests.sh b/scripts/generate_crds_manifests.sh
index b15bc2b1b7..69636be167 100755
--- a/scripts/generate_crds_manifests.sh
+++ b/scripts/generate_crds_manifests.sh
@@ -64,6 +64,7 @@ ${YQ} merge --inplace -d'0' manifests/0000_50_olm_00-namespace.yaml scripts/moni
 ${YQ} write --inplace -s scripts/olm-deployment.patch.yaml manifests/0000_50_olm_07-olm-operator.deployment.yaml
 ${YQ} write --inplace -s scripts/catalog-deployment.patch.yaml manifests/0000_50_olm_08-catalog-operator.deployment.yaml
 ${YQ} write --inplace -s scripts/packageserver-deployment.patch.yaml manifests/0000_50_olm_15-packageserver.clusterserviceversion.yaml
+${YQ} merge --inplace manifests/0000_50_olm_02-olmconfig.yaml scripts/cluster-olmconfig.patch.yaml
 mv manifests/0000_50_olm_15-packageserver.clusterserviceversion.yaml pkg/manifests/csv.yaml
 cp scripts/packageserver-pdb.yaml manifests/0000_50_olm_00-packageserver.pdb.yaml
 
diff --git a/staging/operator-lifecycle-manager/deploy/chart/templates/0000_50_olm_02-olmconfig.yaml b/staging/operator-lifecycle-manager/deploy/chart/templates/0000_50_olm_02-olmconfig.yaml
new file mode 100644
index 0000000000..ad881b7be7
--- /dev/null
+++ b/staging/operator-lifecycle-manager/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/staging/operator-lifecycle-manager/pkg/controller/operators/olm/operator.go b/staging/operator-lifecycle-manager/pkg/controller/operators/olm/operator.go
index 8e0f175fb9..2ab92f2a65 100644
--- a/staging/operator-lifecycle-manager/pkg/controller/operators/olm/operator.go
+++ b/staging/operator-lifecycle-manager/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())
@@ -1194,6 +1218,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 {
@@ -1201,6 +1346,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(),
@@ -1222,15 +1376,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/staging/operator-lifecycle-manager/test/e2e/csv_e2e_test.go b/staging/operator-lifecycle-manager/test/e2e/csv_e2e_test.go
index 2f7bf6ca87..6d3e95ee8c 100644
--- a/staging/operator-lifecycle-manager/test/e2e/csv_e2e_test.go
+++ b/staging/operator-lifecycle-manager/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()
diff --git a/vendor/github.com/operator-framework/operator-lifecycle-manager/pkg/controller/operators/olm/operator.go b/vendor/github.com/operator-framework/operator-lifecycle-manager/pkg/controller/operators/olm/operator.go
index 8e0f175fb9..2ab92f2a65 100644
--- a/vendor/github.com/operator-framework/operator-lifecycle-manager/pkg/controller/operators/olm/operator.go
+++ b/vendor/github.com/operator-framework/operator-lifecycle-manager/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())
@@ -1194,6 +1218,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 {
@@ -1201,6 +1346,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(),
@@ -1222,15 +1376,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 {