Repository: argoproj-labs/argo-rollouts-manager
Pull request #39 (2024-02-14 12:51:46)

feat: Cluster-scoped Rollouts support
**What does this PR do / why we need it**:
This PR is to add support for cluster scoped Rollouts. Currently Rollouts controller can reconcile CR created in same namespace, this PR has changes which would allow Rollouts controller to reconcile CR created in other namespaces as well.
There are few limitations for creating Cluster scoped Rollouts controller, please check original issue created for this feature.

**Have you updated the necessary documentation?**
* [ ] Documentation update is required by this PR, and has been updated.

**Which issue(s) this PR fixes**:
Fixes #?
https://github.com/argoproj-labs/argo-rollouts-manager/issues/20

**How to test changes / Special notes to the reviewer**:
1. Create `RolloutManager` CR and change scope to namespace or cluster using `Spec.NamespaceScoped` field of `RolloutManager`.
2. Create `Rollouts` CR if `RolloutManager` is cluster scoped and it should be reconciled.
3. Try combinations of cluster and namespace scoped Rollouts controller to confirm it works as expected and doesn't allow unsupported scenarios.

jparsai: Addressed all review comments, PTAL.
jgwest: @jparsai did you miss pushing a commit, by chance? I started re-reviewing by checking the first [two](https://github.com/argoproj-labs/argo-rollouts-manager/pull/39#discussion_r1491687467) [comments](https://github.com/argoproj-labs/argo-rollouts-manager/pull/39#discussion_r1491691186) , and it seems like both are still present in the code.
jparsai: Yes @jgwest the first comment about duplicate import I had fixed but it seems when I copied some code from another file to rollout_test.go it was added again because of auto import in VSCode.

About 2nd comment I removed the error Type and using only one type "Reconciled", but Reason has to have multiple values.
Type=Reconciled | Status =  true | Reason=Success | Message = ""
Type=Reconciled | Status =  false | Reason=ErrorOccurred | Message = "Some error message"

I removed duplicate import, PTAL.
jgwest: @jparsai Perfect, it's likely I misread reason and type, thanks for double checking!
jparsai: /test
jparsai: @jgwest PR is ready for review now, PTAL.

Patch:
diff --git a/Makefile b/Makefile
index 1924da8..15edc2b 100644
--- a/Makefile
+++ b/Makefile
@@ -5,6 +5,8 @@
 # - use environment variables to overwrite this value (e.g export VERSION=0.0.2)
 VERSION ?= 0.0.1
 
+NAMESPACE_SCOPED_ARGO_ROLLOUTS ?= false
+
 # CHANNELS define the bundle channels used in the bundle.
 # Add a new line here if you would like to change its default config. (E.g CHANNELS = "candidate,fast,stable")
 # To re-generate a bundle for other specific channels without changing the standard setup, you can:
@@ -102,6 +104,8 @@ fmt: ## Run go fmt against code.
 vet: ## Run go vet against code.
 	go vet ./...
 
+##@ Test
+
 .PHONY: test
 test: manifests generate fmt vet envtest ## Run tests.
 	KUBEBUILDER_ASSETS="$(shell $(ENVTEST) use $(ENVTEST_K8S_VERSION) --bin-dir $(LOCALBIN) -p path)" go test -coverprofile cover.out `go list ./... | grep -v 'tests/e2e'`
@@ -109,13 +113,44 @@ test: manifests generate fmt vet envtest ## Run tests.
 
 .PHONY: start-e2e
 start-e2e: ## Start the operator, to run the e2e tests
-	hack/start-rollouts-manager-for-e2e-tests.sh
-
+	NAMESPACE_SCOPED_ARGO_ROLLOUTS=$(NAMESPACE_SCOPED_ARGO_ROLLOUTS) hack/start-rollouts-manager-for-e2e-tests.sh
 
 .PHONY: test-e2e
 test-e2e: ## Run operator e2e tests
+	NAMESPACE_SCOPED_ARGO_ROLLOUTS=$(NAMESPACE_SCOPED_ARGO_ROLLOUTS) hack/run-rollouts-manager-e2e-tests.sh
+
+
+
+.PHONY: start-e2e-namespace-scoped
+start-e2e-namespace-scoped: ## Start the operator, to run the e2e tests
+	NAMESPACE_SCOPED_ARGO_ROLLOUTS=true hack/start-rollouts-manager-for-e2e-tests.sh
+
+.PHONY: start-e2e-cluster-scoped
+start-e2e-cluster-scoped: ## Start the operator, to run the e2e tests
+	NAMESPACE_SCOPED_ARGO_ROLLOUTS=false hack/start-rollouts-manager-for-e2e-tests.sh
+
+
+.PHONY: start-e2e-namespace-scoped-bg
+start-e2e-namespace-scoped-bg: ## Start the operator, to run the e2e tests
+	RUN_IN_BACKGROUND=true NAMESPACE_SCOPED_ARGO_ROLLOUTS=true hack/start-rollouts-manager-for-e2e-tests.sh
+
+.PHONY: start-e2e-cluster-scoped-bg 
+start-e2e-cluster-scoped-bg: ## Start the operator, to run the e2e tests
+	RUN_IN_BACKGROUND=true hack/start-rollouts-manager-for-e2e-tests.sh
+
+
+.PHONY: test-e2e-namespace-scoped
+test-e2e-namespace-scoped: ## Run operator e2e tests
+	NAMESPACE_SCOPED_ARGO_ROLLOUTS=true hack/run-rollouts-manager-e2e-tests.sh
+
+.PHONY: test-e2e-cluster-scoped
+test-e2e-cluster-scoped: ## Run operator e2e tests
 	hack/run-rollouts-manager-e2e-tests.sh
 
+.PHONY: start-test-e2e-all
+start-test-e2e-all: start-e2e-namespace-scoped-bg test-e2e-namespace-scoped start-e2e-cluster-scoped-bg test-e2e-cluster-scoped
+
+
 ##@ Build
 
 .PHONY: build
@@ -124,7 +159,7 @@ build: manifests generate fmt vet ## Build manager binary.
 
 .PHONY: run
 run: manifests generate fmt vet ## Run a controller from your host.
-	go run ./main.go
+	NAMESPACE_SCOPED_ARGO_ROLLOUTS=$(NAMESPACE_SCOPED_ARGO_ROLLOUTS) go run ./main.go
 
 # If you wish built the manager image targeting other platforms you can use the --platform flag.
 # (i.e. docker build --platform linux/arm64 ). However, you must enable docker buildKit for it.
diff --git a/api/v1alpha1/argorollouts_types.go b/api/v1alpha1/argorollouts_types.go
index 626e46a..cfeba9c 100644
--- a/api/v1alpha1/argorollouts_types.go
+++ b/api/v1alpha1/argorollouts_types.go
@@ -40,6 +40,9 @@ type RolloutManagerSpec struct {
 
 	// Version defines Argo Rollouts controller tag (optional)
 	Version string `json:"version,omitempty"`
+
+	// NamespaceScoped lets you specify if RolloutManager has to watch a namespace or the whole cluster
+	NamespaceScoped bool `json:"namespaceScoped,omitempty"`
 }
 
 // ArgoRolloutsNodePlacementSpec is used to specify NodeSelector and Tolerations for Rollouts workloads
@@ -64,6 +67,9 @@ type RolloutManagerStatus struct {
 	// Available: All of the resources for the RolloutManager are ready.
 	// Unknown: The state of the RolloutManager phase could not be obtained.
 	Phase RolloutControllerPhase `json:"phase,omitempty"`
+
+	// Conditions is an array of the RolloutManager's status conditions
+	Conditions []metav1.Condition `json:"conditions,omitempty"`
 }
 
 type RolloutControllerPhase string
@@ -75,6 +81,17 @@ const (
 	PhaseFailure   RolloutControllerPhase = "Failure"
 )
 
+const (
+	RolloutManagerConditionType = "Reconciled"
+)
+
+const (
+	RolloutManagerReasonSuccess                             = "Success"
+	RolloutManagerReasonErrorOccurred                       = "ErrorOccurred"
+	RolloutManagerReasonMultipleClusterScopedRolloutManager = "MultipleClusterScopedRolloutManager"
+	RolloutManagerReasonInvalidScoped                       = "InvalidRolloutManagerScope"
+)
+
 //+kubebuilder:object:root=true
 //+kubebuilder:subresource:status
 
diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go
index 01e2cbb..6f5e116 100644
--- a/api/v1alpha1/zz_generated.deepcopy.go
+++ b/api/v1alpha1/zz_generated.deepcopy.go
@@ -23,6 +23,7 @@ package v1alpha1
 
 import (
 	"k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	runtime "k8s.io/apimachinery/pkg/runtime"
 )
 
@@ -32,7 +33,7 @@ func (in *RolloutManager) DeepCopyInto(out *RolloutManager) {
 	out.TypeMeta = in.TypeMeta
 	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
 	in.Spec.DeepCopyInto(&out.Spec)
-	out.Status = in.Status
+	in.Status.DeepCopyInto(&out.Status)
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new RolloutManager.
@@ -120,6 +121,13 @@ func (in *RolloutManagerSpec) DeepCopy() *RolloutManagerSpec {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *RolloutManagerStatus) DeepCopyInto(out *RolloutManagerStatus) {
 	*out = *in
+	if in.Conditions != nil {
+		in, out := &in.Conditions, &out.Conditions
+		*out = make([]metav1.Condition, len(*in))
+		for i := range *in {
+			(*in)[i].DeepCopyInto(&(*out)[i])
+		}
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new RolloutManagerStatus.
diff --git a/config/crd/bases/argoproj.io_rolloutmanagers.yaml b/config/crd/bases/argoproj.io_rolloutmanagers.yaml
index 7840d69..be7ee2c 100644
--- a/config/crd/bases/argoproj.io_rolloutmanagers.yaml
+++ b/config/crd/bases/argoproj.io_rolloutmanagers.yaml
@@ -154,6 +154,10 @@ spec:
               image:
                 description: Image defines Argo Rollouts controller image (optional)
                 type: string
+              namespaceScoped:
+                description: NamespaceScoped lets you specify if RolloutManager has
+                  to watch a namespace or the whole cluster
+                type: boolean
               nodePlacement:
                 description: NodePlacement defines NodeSelectors and Taints for Rollouts
                   workloads
@@ -214,6 +218,76 @@ spec:
           status:
             description: RolloutManagerStatus defines the observed state of RolloutManager
             properties:
+              conditions:
+                description: Conditions is an array of the RolloutManager's status
+                  conditions
+                items:
+                  description: "Condition contains details for one aspect of the current
+                    state of this API Resource. --- This struct is intended for direct
+                    use as an array at the field path .status.conditions.  For example,
+                    \n type FooStatus struct{ // Represents the observations of a
+                    foo's current state. // Known .status.conditions.type are: \"Available\",
+                    \"Progressing\", and \"Degraded\" // +patchMergeKey=type // +patchStrategy=merge
+                    // +listType=map // +listMapKey=type Conditions []metav1.Condition
+                    `json:\"conditions,omitempty\" patchStrategy:\"merge\" patchMergeKey:\"type\"
+                    protobuf:\"bytes,1,rep,name=conditions\"` \n // other fields }"
+                  properties:
+                    lastTransitionTime:
+                      description: lastTransitionTime is the last time the condition
+                        transitioned from one status to another. This should be when
+                        the underlying condition changed.  If that is not known, then
+                        using the time when the API field changed is acceptable.
+                      format: date-time
+                      type: string
+                    message:
+                      description: message is a human readable message indicating
+                        details about the transition. This may be an empty string.
+                      maxLength: 32768
+                      type: string
+                    observedGeneration:
+                      description: observedGeneration represents the .metadata.generation
+                        that the condition was set based upon. For instance, if .metadata.generation
+                        is currently 12, but the .status.conditions[x].observedGeneration
+                        is 9, the condition is out of date with respect to the current
+                        state of the instance.
+                      format: int64
+                      minimum: 0
+                      type: integer
+                    reason:
+                      description: reason contains a programmatic identifier indicating
+                        the reason for the condition's last transition. Producers
+                        of specific condition types may define expected values and
+                        meanings for this field, and whether the values are considered
+                        a guaranteed API. The value should be a CamelCase string.
+                        This field may not be empty.
+                      maxLength: 1024
+                      minLength: 1
+                      pattern: ^[A-Za-z]([A-Za-z0-9_,:]*[A-Za-z0-9_])?$
+                      type: string
+                    status:
+                      description: status of the condition, one of True, False, Unknown.
+                      enum:
+                      - "True"
+                      - "False"
+                      - Unknown
+                      type: string
+                    type:
+                      description: type of condition in CamelCase or in foo.example.com/CamelCase.
+                        --- Many .condition.type values are consistent across resources
+                        like Available, but because arbitrary conditions can be useful
+                        (see .node.status.conditions), the ability to deconflict is
+                        important. The regex it matches is (dns1123SubdomainFmt/)?(qualifiedNameFmt)
+                      maxLength: 316
+                      pattern: ^([a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*/)?(([A-Za-z0-9][-A-Za-z0-9_.]*)?[A-Za-z0-9])$
+                      type: string
+                  required:
+                  - lastTransitionTime
+                  - message
+                  - reason
+                  - status
+                  - type
+                  type: object
+                type: array
               phase:
                 description: 'Phase is a simple, high-level summary of where the RolloutManager
                   is in its lifecycle. There are three possible phase values: Pending:
diff --git a/controllers/argorollouts_controller.go b/controllers/argorollouts_controller.go
index 182c4f1..3c09610 100644
--- a/controllers/argorollouts_controller.go
+++ b/controllers/argorollouts_controller.go
@@ -24,6 +24,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -39,6 +40,10 @@ type RolloutManagerReconciler struct {
 	client.Client
 	Scheme                       *runtime.Scheme
 	OpenShiftRoutePluginLocation string
+
+	// NamespaceScopedArgoRolloutsController is used to configure scope of Argo Rollouts controller
+	// If value is true then deploy namespace-scoped Argo Rollouts controller else cluster-scoped
+	NamespaceScopedArgoRolloutsController bool
 }
 
 var log = logr.Log.WithName("rollouts-controller")
@@ -94,8 +99,19 @@ func (r *RolloutManagerReconciler) Reconcile(ctx context.Context, req ctrl.Reque
 		return reconcile.Result{}, err
 	}
 
-	if err := r.reconcileRolloutsManager(ctx, rollouts); err != nil {
-		// Error reconciling RolloutManager sub-resources - requeue the request.
+	condition, err := r.reconcileRolloutsManager(ctx, rollouts)
+
+	if err := updateStatusConditionOfRolloutManager(ctx, condition, &rolloutsmanagerv1alpha1.RolloutManager{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      req.Name,
+			Namespace: req.Namespace,
+		},
+	}, r.Client, log); err != nil {
+		log.Error(err, "unable to update status of RolloutManager")
+		return reconcile.Result{}, err
+	}
+
+	if err != nil {
 		return reconcile.Result{}, err
 	}
 
@@ -123,8 +139,14 @@ func (r *RolloutManagerReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	// Watch for changes to Role sub-resources owned by RolloutManager.
 	bld.Owns(&rbacv1.Role{})
 
+	// Watch for changes to ClusterRole sub-resources owned by RolloutManager.
+	bld.Owns(&rbacv1.ClusterRole{})
+
 	// Watch for changes to RoleBinding sub-resources owned by RolloutManager.
 	bld.Owns(&rbacv1.RoleBinding{})
 
+	// Watch for changes to ClusterRoleBinding sub-resources owned by RolloutManager.
+	bld.Owns(&rbacv1.ClusterRoleBinding{})
+
 	return bld.Complete(r)
 }
diff --git a/controllers/configmap.go b/controllers/configmap.go
index 70e602b..0ee5a34 100644
--- a/controllers/configmap.go
+++ b/controllers/configmap.go
@@ -53,7 +53,7 @@ func (r *RolloutManagerReconciler) reconcileConfigMap(ctx context.Context, cr *r
 			log.Info("configMap not found, creating default configmap with openshift route plugin information")
 			return r.Client.Create(ctx, desiredConfigMap)
 		}
-		return fmt.Errorf("failed to get the serviceAccount associated with %s : %s", desiredConfigMap.Name, err)
+		return fmt.Errorf("failed to get the serviceAccount associated with %s: %w", desiredConfigMap.Name, err)
 	}
 
 	var actualTrafficRouterPlugins []types.PluginItem
@@ -72,7 +72,7 @@ func (r *RolloutManagerReconciler) reconcileConfigMap(ctx context.Context, cr *r
 
 	pluginString, err = yaml.Marshal(updatedTrafficRouterPlugins)
 	if err != nil {
-		return fmt.Errorf("error marshalling trafficRouterPlugin to string %s", err)
+		return fmt.Errorf("error marshalling trafficRouterPlugin to string %w", err)
 	}
 
 	actualConfigMap.Data[TrafficRouterPluginConfigMapKey] = string(pluginString)
diff --git a/controllers/default.go b/controllers/default.go
index b55ee6e..1c19a12 100644
--- a/controllers/default.go
+++ b/controllers/default.go
@@ -8,9 +8,9 @@ const (
 	DefaultArgoRolloutsMetricsServiceName = "argo-rollouts-metrics"
 	// ArgoRolloutsDefaultImage is the default image for rollouts controller.
 	DefaultArgoRolloutsImage = "quay.io/argoproj/argo-rollouts"
-	// ArgoRolloutsDefaultVersion is the default version for the rollouts controller.
+	// ArgoRolloutsDefaultVersion is the default version for the Rollouts controller.
 	DefaultArgoRolloutsVersion = "v1.6.6" // v1.6.6
-	// DefaultArgoRolloutsResourceName is the default name for rollout controller resources such as
+	// DefaultArgoRolloutsResourceName is the default name for Rollouts controller resources such as
 	// deployment, service, role, rolebinding and serviceaccount.
 	DefaultArgoRolloutsResourceName = "argo-rollouts"
 	// DefaultRolloutsNotificationSecretName is the default name for rollout controller secret resource.
@@ -25,4 +25,8 @@ const (
 	DefaultRolloutsConfigMapName = "argo-rollouts-config"
 
 	DefaultOpenShiftRoutePluginURL = "https://github.com/argoproj-labs/rollouts-plugin-trafficrouter-openshift/releases/download/commit-2749e0ac96ba00ce6f4af19dc6d5358048227d77/rollouts-plugin-trafficrouter-openshift-linux-amd64"
+
+	// NamespaceScopedArgoRolloutsController is an environment variable that can be used to configure scope of Argo Rollouts controller
+	// Set true to allow only namespace-scoped Argo Rollouts controller deployment and false for cluster-scoped
+	NamespaceScopedArgoRolloutsController = "NAMESPACE_SCOPED_ARGO_ROLLOUTS"
 )
diff --git a/controllers/deployment.go b/controllers/deployment.go
index 3c6a639..fc80a71 100644
--- a/controllers/deployment.go
+++ b/controllers/deployment.go
@@ -101,7 +101,7 @@ func (r *RolloutManagerReconciler) reconcileRolloutsDeployment(ctx context.Conte
 
 	if err := fetchObject(ctx, r.Client, cr.Namespace, DefaultArgoRolloutsResourceName, actualDeployment); err != nil {
 		if !errors.IsNotFound(err) {
-			return fmt.Errorf("failed to get the Deployment %s: %s", DefaultArgoRolloutsResourceName, err)
+			return fmt.Errorf("failed to get the Deployment %s: %w", DefaultArgoRolloutsResourceName, err)
 		}
 
 		if err := controllerutil.SetControllerReference(cr, &desiredDeployment, r.Scheme); err != nil {
@@ -453,7 +453,9 @@ func getRolloutsContainerImage(cr rolloutsmanagerv1alpha1.RolloutManager) string
 func getRolloutsCommandArgs(cr rolloutsmanagerv1alpha1.RolloutManager) []string {
 	args := make([]string, 0)
 
-	args = append(args, "--namespaced")
+	if cr.Spec.NamespaceScoped {
+		args = append(args, "--namespaced")
+	}
 
 	extraArgs := cr.Spec.ExtraCommandArgs
 	err := isMergable(extraArgs, args)
diff --git a/controllers/reconcile.go b/controllers/reconcile.go
index ad430c9..8c670ca 100644
--- a/controllers/reconcile.go
+++ b/controllers/reconcile.go
@@ -4,67 +4,117 @@ import (
 	"context"
 
 	rolloutsmanagerv1alpha1 "github.com/argoproj-labs/argo-rollouts-manager/api/v1alpha1"
+	rbacv1 "k8s.io/api/rbac/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-func (r *RolloutManagerReconciler) reconcileRolloutsManager(ctx context.Context, cr *rolloutsmanagerv1alpha1.RolloutManager) error {
+func (r *RolloutManagerReconciler) reconcileRolloutsManager(ctx context.Context, cr *rolloutsmanagerv1alpha1.RolloutManager) (metav1.Condition, error) {
 
-	log.Info("reconciling rollouts serviceaccount")
+	log.Info("validating RolloutManager's scope")
+	if err := validateRolloutsScope(ctx, r.Client, cr, r.NamespaceScopedArgoRolloutsController); err != nil {
+		if invalidRolloutScope(err) {
+			return createCondition(err.Error(), rolloutsmanagerv1alpha1.RolloutManagerReasonInvalidScoped), nil
+		}
+		log.Error(err, "failed to validate RolloutManager's scope.")
+		return createCondition(err.Error()), err
+	}
+
+	log.Info("searching for existing RolloutManagers")
+	if err := checkForExistingRolloutManager(ctx, r.Client, cr); err != nil {
+		if multipleRolloutManagersExist(err) {
+			return createCondition(err.Error(), rolloutsmanagerv1alpha1.RolloutManagerReasonMultipleClusterScopedRolloutManager), nil
+		}
+		log.Error(err, "failed to validate multiple RolloutManagers.")
+		return createCondition(err.Error()), err
+	}
+
+	log.Info("reconciling Rollouts ServiceAccount")
 	sa, err := r.reconcileRolloutsServiceAccount(ctx, cr)
 	if err != nil {
-		return err
+		log.Error(err, "failed to reconcile Rollout's ServiceAccount.")
+		return createCondition(err.Error()), err
 	}
 
-	log.Info("reconciling rollouts roles")
-	role, err := r.reconcileRolloutsRole(ctx, cr)
-	if err != nil {
-		return err
+	var role *rbacv1.Role
+	var clusterRole *rbacv1.ClusterRole
+
+	if cr.Spec.NamespaceScoped {
+		log.Info("reconciling Rollouts Roles")
+		role, err = r.reconcileRolloutsRole(ctx, cr)
+		if err != nil {
+			log.Error(err, "failed to reconcile Rollout's Role.")
+			return createCondition(err.Error()), err
+		}
+	} else {
+		log.Info("reconciling Rollouts ClusterRoles")
+		clusterRole, err = r.reconcileRolloutsClusterRole(ctx)
+		if err != nil {
+			log.Error(err, "failed to reconcile Rollout's ClusterRoles.")
+			return createCondition(err.Error()), err
+		}
 	}
 
 	log.Info("reconciling aggregate-to-admin ClusterRole")
-	if err := r.reconcileRolloutsAggregateToAdminClusterRole(ctx, cr); err != nil {
-		return err
+	if err := r.reconcileRolloutsAggregateToAdminClusterRole(ctx); err != nil {
+		log.Error(err, "failed to reconcile Rollout's aggregate-to-admin ClusterRoles.")
+		return createCondition(err.Error()), err
 	}
 
 	log.Info("reconciling aggregate-to-edit ClusterRole")
-	if err := r.reconcileRolloutsAggregateToEditClusterRole(ctx, cr); err != nil {
-		return err
+	if err := r.reconcileRolloutsAggregateToEditClusterRole(ctx); err != nil {
+		log.Error(err, "failed to reconcile Rollout's aggregate-to-edit ClusterRoles.")
+		return createCondition(err.Error()), err
 	}
 
 	log.Info("reconciling aggregate-to-view ClusterRole")
-	if err := r.reconcileRolloutsAggregateToViewClusterRole(ctx, cr); err != nil {
-		return err
+	if err := r.reconcileRolloutsAggregateToViewClusterRole(ctx); err != nil {
+		log.Error(err, "failed to reconcile Rollout's aggregate-to-view ClusterRoles.")
+		return createCondition(err.Error()), err
 	}
 
-	log.Info("reconciling rollouts role bindings")
-	if err := r.reconcileRolloutsRoleBinding(ctx, cr, role, sa); err != nil {
-		return err
+	if cr.Spec.NamespaceScoped {
+		log.Info("reconciling Rollouts RoleBindings")
+		if err := r.reconcileRolloutsRoleBinding(ctx, cr, role, sa); err != nil {
+			log.Error(err, "failed to reconcile Rollout's RoleBindings.")
+			return createCondition(err.Error()), err
+		}
+	} else {
+		log.Info("reconciling Rollouts ClusterRoleBinding")
+		if err := r.reconcileRolloutsClusterRoleBinding(ctx, clusterRole, sa); err != nil {
+			log.Error(err, "failed to reconcile Rollout's ClusterRoleBinding.")
+			return createCondition(err.Error()), err
+		}
 	}
 
-	log.Info("reconciling rollouts secret")
+	log.Info("reconciling Rollouts Secret")
 	if err := r.reconcileRolloutsSecrets(ctx, cr); err != nil {
-		return err
+		log.Error(err, "failed to reconcile Rollout's Secret.")
+		return createCondition(err.Error()), err
 	}
 
-	// reconcile configMap for plugins
-	log.Info("reconciling configMap for plugins")
+	log.Info("reconciling ConfigMap for plugins")
 	if err := r.reconcileConfigMap(ctx, cr); err != nil {
-		return err
+		log.Error(err, "failed to reconcile Rollout's ConfigMap.")
+		return createCondition(err.Error()), err
 	}
 
-	log.Info("reconciling rollouts deployment")
+	log.Info("reconciling Rollouts Deployment")
 	if err := r.reconcileRolloutsDeployment(ctx, cr, *sa); err != nil {
-		return err
+		log.Error(err, "failed to reconcile Rollout's Deployment.")
+		return createCondition(err.Error()), err
 	}
 
-	log.Info("reconciling rollouts metrics service")
+	log.Info("reconciling Rollouts Metrics Service")
 	if err := r.reconcileRolloutsMetricsService(ctx, cr); err != nil {
-		return err
+		log.Error(err, "failed to reconcile Rollout's Metrics Service.")
+		return createCondition(err.Error()), err
 	}
 
 	log.Info("reconciling status of workloads")
 	if err := r.reconcileStatus(ctx, cr); err != nil {
-		return err
+		log.Error(err, "failed to reconcile status of workloads.")
+		return createCondition(err.Error()), err
 	}
 
-	return nil
+	return createCondition(""), nil
 }
diff --git a/controllers/resources.go b/controllers/resources.go
index 208704d..9c8259b 100644
--- a/controllers/resources.go
+++ b/controllers/resources.go
@@ -15,7 +15,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 )
 
-// Reconciles rollouts serviceaccount.
+// Reconciles Rollouts ServiceAccount.
 func (r *RolloutManagerReconciler) reconcileRolloutsServiceAccount(ctx context.Context, cr *rolloutsmanagerv1alpha1.RolloutManager) (*corev1.ServiceAccount, error) {
 
 	sa := &corev1.ServiceAccount{
@@ -28,14 +28,14 @@ func (r *RolloutManagerReconciler) reconcileRolloutsServiceAccount(ctx context.C
 
 	if err := fetchObject(ctx, r.Client, cr.Namespace, sa.Name, sa); err != nil {
 		if !apierrors.IsNotFound(err) {
-			return nil, fmt.Errorf("failed to get the serviceAccount associated with %s : %s", sa.Name, err)
+			return nil, fmt.Errorf("failed to get the ServiceAccount associated with %s: %w", sa.Name, err)
 		}
 
 		if err := controllerutil.SetControllerReference(cr, sa, r.Scheme); err != nil {
 			return nil, err
 		}
 
-		log.Info(fmt.Sprintf("Creating serviceaccount %s", sa.Name))
+		log.Info(fmt.Sprintf("Creating ServiceAccount %s", sa.Name))
 		err := r.Client.Create(ctx, sa)
 		if err != nil {
 			return nil, err
@@ -45,7 +45,7 @@ func (r *RolloutManagerReconciler) reconcileRolloutsServiceAccount(ctx context.C
 	return sa, nil
 }
 
-// Reconciles rollouts role.
+// Reconciles Rollouts Role.
 func (r *RolloutManagerReconciler) reconcileRolloutsRole(ctx context.Context, cr *rolloutsmanagerv1alpha1.RolloutManager) (*rbacv1.Role, error) {
 
 	expectedPolicyRules := GetPolicyRules()
@@ -60,20 +60,21 @@ func (r *RolloutManagerReconciler) reconcileRolloutsRole(ctx context.Context, cr
 
 	if err := fetchObject(ctx, r.Client, cr.Namespace, role.Name, role); err != nil {
 		if !apierrors.IsNotFound(err) {
-			return nil, fmt.Errorf("failed to reconcile the role for the service account associated with %s : %s", role.Name, err)
+			return nil, fmt.Errorf("failed to reconcile the Role for the ServiceAccount associated with %s: %w", role.Name, err)
 		}
 
 		if err = controllerutil.SetControllerReference(cr, role, r.Scheme); err != nil {
 			return nil, err
 		}
 
-		log.Info(fmt.Sprintf("Creating role %s", role.Name))
+		log.Info(fmt.Sprintf("Creating Role %s", role.Name))
 		role.Rules = expectedPolicyRules
 		return role, r.Client.Create(ctx, role)
 	}
 
-	// Reconcile if the role already exists and modified.
+	// Reconcile if the Role already exists and modified.
 	if !reflect.DeepEqual(role.Rules, expectedPolicyRules) {
+		log.Info(fmt.Sprintf("PolicyRules of Role %s do not match the expected state, hence updating it", role.Name))
 		role.Rules = expectedPolicyRules
 		return role, r.Client.Update(ctx, role)
 	}
@@ -81,8 +82,49 @@ func (r *RolloutManagerReconciler) reconcileRolloutsRole(ctx context.Context, cr
 	return role, nil
 }
 
-// Reconcile rollouts rolebinding.
+// Reconciles Rollouts ClusterRole.
+func (r *RolloutManagerReconciler) reconcileRolloutsClusterRole(ctx context.Context) (*rbacv1.ClusterRole, error) {
+
+	expectedPolicyRules := GetPolicyRules()
+
+	clusterRole := &rbacv1.ClusterRole{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: DefaultArgoRolloutsResourceName,
+		},
+	}
+	setRolloutsLabels(&clusterRole.ObjectMeta)
+
+	if err := fetchObject(ctx, r.Client, "", clusterRole.Name, clusterRole); err != nil {
+		if !apierrors.IsNotFound(err) {
+			return nil, fmt.Errorf("failed to Reconcile the ClusterRole for the ServiceAccount associated with %s: %w", clusterRole.Name, err)
+		}
+
+		log.Info(fmt.Sprintf("Creating ClusterRole %s", clusterRole.Name))
+		clusterRole.Rules = expectedPolicyRules
+		return clusterRole, r.Client.Create(ctx, clusterRole)
+	}
+
+	// Reconcile if the ClusterRole already exists and modified.
+	if !reflect.DeepEqual(clusterRole.Rules, expectedPolicyRules) {
+		log.Info(fmt.Sprintf("PolicyRules of ClusterRole %s do not match the expected state, hence updating it", clusterRole.Name))
+		clusterRole.Rules = expectedPolicyRules
+		return clusterRole, r.Client.Update(ctx, clusterRole)
+	}
+
+	return clusterRole, nil
+}
+
+// Reconcile Rollouts RoleBinding.
 func (r *RolloutManagerReconciler) reconcileRolloutsRoleBinding(ctx context.Context, cr *rolloutsmanagerv1alpha1.RolloutManager, role *rbacv1.Role, sa *corev1.ServiceAccount) error {
+
+	if role == nil {
+		return fmt.Errorf("received Role is nil while reconciling RoleBinding")
+	}
+
+	if sa == nil {
+		return fmt.Errorf("received ServiceAccount is nil while reconciling RoleBinding")
+	}
+
 	expectedRoleBinding := &rbacv1.RoleBinding{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      DefaultArgoRolloutsResourceName,
@@ -107,24 +149,24 @@ func (r *RolloutManagerReconciler) reconcileRolloutsRoleBinding(ctx context.Cont
 
 	actualRoleBinding := &rbacv1.RoleBinding{}
 
-	// Fetch the rolebinding if exists and store that in actualRoleBinding.
+	// Fetch the RoleBinding if exists and store that in actualRoleBinding.
 	if err := fetchObject(ctx, r.Client, cr.Namespace, expectedRoleBinding.Name, actualRoleBinding); err != nil {
 		if !apierrors.IsNotFound(err) {
-			return fmt.Errorf("failed to get the rolebinding associated with %s : %s", expectedRoleBinding.Name, err)
+			return fmt.Errorf("failed to get the RoleBinding associated with %s: %w", expectedRoleBinding.Name, err)
 		}
 
 		if err := controllerutil.SetControllerReference(cr, expectedRoleBinding, r.Scheme); err != nil {
 			return err
 		}
 
-		log.Info(fmt.Sprintf("Creating rolebinding %s", expectedRoleBinding.Name))
+		log.Info(fmt.Sprintf("Creating RoleBinding %s", expectedRoleBinding.Name))
 		return r.Client.Create(ctx, expectedRoleBinding)
 	}
 
-	// Reconcile if the role already exists and modified.
+	// Reconcile if the RoleBinding already exists and modified.
 	if !reflect.DeepEqual(expectedRoleBinding.Subjects, actualRoleBinding.Subjects) {
+		log.Info(fmt.Sprintf("Subjects of RoleBinding %s do not match the expected state, hence updating it", actualRoleBinding.Name))
 		actualRoleBinding.Subjects = expectedRoleBinding.Subjects
-
 		if err := r.Client.Update(ctx, actualRoleBinding); err != nil {
 			return err
 		}
@@ -133,13 +175,69 @@ func (r *RolloutManagerReconciler) reconcileRolloutsRoleBinding(ctx context.Cont
 	return nil
 }
 
+// Reconcile Rollouts ClusterRoleBinding.
+func (r *RolloutManagerReconciler) reconcileRolloutsClusterRoleBinding(ctx context.Context, clusterRole *rbacv1.ClusterRole, sa *corev1.ServiceAccount) error {
+
+	if clusterRole == nil {
+		return fmt.Errorf("received ClusterRole is nil while reconciling ClusterRoleBinding")
+	}
+
+	if sa == nil {
+		return fmt.Errorf("received ServiceAccount is nil while reconciling ClusterRoleBinding")
+	}
+
+	expectedClusterRoleBinding := &rbacv1.ClusterRoleBinding{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: DefaultArgoRolloutsResourceName,
+		},
+	}
+	setRolloutsLabels(&expectedClusterRoleBinding.ObjectMeta)
+
+	expectedClusterRoleBinding.RoleRef = rbacv1.RoleRef{
+		APIGroup: rbacv1.GroupName,
+		Kind:     "ClusterRole",
+		Name:     clusterRole.Name,
+	}
+
+	expectedClusterRoleBinding.Subjects = []rbacv1.Subject{
+		{
+			Kind:      rbacv1.ServiceAccountKind,
+			Name:      sa.Name,
+			Namespace: sa.Namespace,
+		},
+	}
+
+	actualClusterRoleBinding := &rbacv1.ClusterRoleBinding{}
+
+	// Fetch the ClusterRoleBinding if exists and store that in actualClusterRoleBinding.
+	if err := fetchObject(ctx, r.Client, "", expectedClusterRoleBinding.Name, actualClusterRoleBinding); err != nil {
+		if !apierrors.IsNotFound(err) {
+			return fmt.Errorf("failed to get the ClusterRoleBinding associated with %s: %w", expectedClusterRoleBinding.Name, err)
+		}
+
+		log.Info(fmt.Sprintf("Creating ClusterRoleBinding %s", expectedClusterRoleBinding.Name))
+		return r.Client.Create(ctx, expectedClusterRoleBinding)
+	}
+
+	// Reconcile if the ClusterRoleBinding already exists and modified.
+	if !reflect.DeepEqual(expectedClusterRoleBinding.Subjects, actualClusterRoleBinding.Subjects) {
+		log.Info(fmt.Sprintf("Subjects of ClusterRoleBinding %s do not match the expected state, hence updating it", expectedClusterRoleBinding.Name))
+		actualClusterRoleBinding.Subjects = expectedClusterRoleBinding.Subjects
+		if err := r.Client.Update(ctx, actualClusterRoleBinding); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
 // Reconciles aggregate-to-admin ClusterRole.
-func (r *RolloutManagerReconciler) reconcileRolloutsAggregateToAdminClusterRole(ctx context.Context, cr *rolloutsmanagerv1alpha1.RolloutManager) error {
+func (r *RolloutManagerReconciler) reconcileRolloutsAggregateToAdminClusterRole(ctx context.Context) error {
 
 	var aggregationType string = "aggregate-to-admin"
 	name := fmt.Sprintf("%s-%s", DefaultArgoRolloutsResourceName, aggregationType)
 
-	expectedPolicyRules := getAggregateToAdminPolicyRules()
+	expectedPolicyRules := GetAggregateToAdminPolicyRules()
 
 	clusterRole := &rbacv1.ClusterRole{
 		ObjectMeta: metav1.ObjectMeta{
@@ -148,9 +246,9 @@ func (r *RolloutManagerReconciler) reconcileRolloutsAggregateToAdminClusterRole(
 	}
 	setRolloutsAggregatedClusterRoleLabels(&clusterRole.ObjectMeta, name, aggregationType)
 
-	if err := fetchObject(ctx, r.Client, cr.Namespace, clusterRole.Name, clusterRole); err != nil {
+	if err := fetchObject(ctx, r.Client, "", clusterRole.Name, clusterRole); err != nil {
 		if !apierrors.IsNotFound(err) {
-			return fmt.Errorf("failed to reconcile the aggregated ClusterRole %s : %s", clusterRole.Name, err)
+			return fmt.Errorf("failed to reconcile the aggregated ClusterRole %s: %w", clusterRole.Name, err)
 		}
 
 		log.Info(fmt.Sprintf("Creating aggregated ClusterRole %s", clusterRole.Name))
@@ -158,8 +256,9 @@ func (r *RolloutManagerReconciler) reconcileRolloutsAggregateToAdminClusterRole(
 		return r.Client.Create(ctx, clusterRole)
 	}
 
-	// Reconcile if the aggregated role already exists and modified.
+	// Reconcile if the aggregated CusterRole already exists and modified.
 	if !reflect.DeepEqual(clusterRole.Rules, expectedPolicyRules) {
+		log.Info(fmt.Sprintf("PolicyRules of ClusterRole %s do not match the expected state, hence updating it", clusterRole.Name))
 		clusterRole.Rules = expectedPolicyRules
 		return r.Client.Update(ctx, clusterRole)
 	}
@@ -168,12 +267,12 @@ func (r *RolloutManagerReconciler) reconcileRolloutsAggregateToAdminClusterRole(
 }
 
 // Reconciles aggregate-to-edit ClusterRole.
-func (r *RolloutManagerReconciler) reconcileRolloutsAggregateToEditClusterRole(ctx context.Context, cr *rolloutsmanagerv1alpha1.RolloutManager) error {
+func (r *RolloutManagerReconciler) reconcileRolloutsAggregateToEditClusterRole(ctx context.Context) error {
 
 	var aggregationType string = "aggregate-to-edit"
 	name := fmt.Sprintf("%s-%s", DefaultArgoRolloutsResourceName, aggregationType)
 
-	expectedPolicyRules := getAggregateToEditPolicyRules()
+	expectedPolicyRules := GetAggregateToEditPolicyRules()
 
 	clusterRole := &rbacv1.ClusterRole{
 		ObjectMeta: metav1.ObjectMeta{
@@ -182,9 +281,9 @@ func (r *RolloutManagerReconciler) reconcileRolloutsAggregateToEditClusterRole(c
 	}
 	setRolloutsAggregatedClusterRoleLabels(&clusterRole.ObjectMeta, name, aggregationType)
 
-	if err := fetchObject(ctx, r.Client, cr.Namespace, clusterRole.Name, clusterRole); err != nil {
+	if err := fetchObject(ctx, r.Client, "", clusterRole.Name, clusterRole); err != nil {
 		if !apierrors.IsNotFound(err) {
-			return fmt.Errorf("failed to reconcile the aggregated ClusterRole %s : %s", clusterRole.Name, err)
+			return fmt.Errorf("failed to reconcile the aggregated ClusterRole %s: %w", clusterRole.Name, err)
 		}
 
 		log.Info(fmt.Sprintf("Creating aggregated ClusterRole %s", clusterRole.Name))
@@ -192,8 +291,9 @@ func (r *RolloutManagerReconciler) reconcileRolloutsAggregateToEditClusterRole(c
 		return r.Client.Create(ctx, clusterRole)
 	}
 
-	// Reconcile if the aggregated role already exists and modified.
+	// Reconcile if the aggregated ClusterRole already exists and modified.
 	if !reflect.DeepEqual(clusterRole.Rules, expectedPolicyRules) {
+		log.Info(fmt.Sprintf("PolicyRules of ClusterRole %s do not match the expected state, hence updating it", clusterRole.Name))
 		clusterRole.Rules = expectedPolicyRules
 		return r.Client.Update(ctx, clusterRole)
 	}
@@ -202,12 +302,12 @@ func (r *RolloutManagerReconciler) reconcileRolloutsAggregateToEditClusterRole(c
 }
 
 // Reconciles aggregate-to-view ClusterRole.
-func (r *RolloutManagerReconciler) reconcileRolloutsAggregateToViewClusterRole(ctx context.Context, cr *rolloutsmanagerv1alpha1.RolloutManager) error {
+func (r *RolloutManagerReconciler) reconcileRolloutsAggregateToViewClusterRole(ctx context.Context) error {
 
 	var aggregationType string = "aggregate-to-view"
 	name := fmt.Sprintf("%s-%s", DefaultArgoRolloutsResourceName, aggregationType)
 
-	expectedPolicyRules := getAggregateToViewPolicyRules()
+	expectedPolicyRules := GetAggregateToViewPolicyRules()
 
 	clusterRole := &rbacv1.ClusterRole{
 		ObjectMeta: metav1.ObjectMeta{
@@ -216,9 +316,9 @@ func (r *RolloutManagerReconciler) reconcileRolloutsAggregateToViewClusterRole(c
 	}
 	setRolloutsAggregatedClusterRoleLabels(&clusterRole.ObjectMeta, name, aggregationType)
 
-	if err := fetchObject(ctx, r.Client, cr.Namespace, clusterRole.Name, clusterRole); err != nil {
+	if err := fetchObject(ctx, r.Client, "", clusterRole.Name, clusterRole); err != nil {
 		if !apierrors.IsNotFound(err) {
-			return fmt.Errorf("failed to reconcile the aggregated ClusterRole %s : %s", clusterRole.Name, err)
+			return fmt.Errorf("failed to reconcile the aggregated ClusterRole %s: %w", clusterRole.Name, err)
 		}
 
 		log.Info(fmt.Sprintf("Creating aggregated ClusterRole %s", clusterRole.Name))
@@ -226,8 +326,9 @@ func (r *RolloutManagerReconciler) reconcileRolloutsAggregateToViewClusterRole(c
 		return r.Client.Create(ctx, clusterRole)
 	}
 
-	// Reconcile if the aggregated role already exists and modified.
+	// Reconcile if the aggregated ClusterRole already exists and modified.
 	if !reflect.DeepEqual(clusterRole.Rules, expectedPolicyRules) {
+		log.Info(fmt.Sprintf("PolicyRules of ClusterRole %s do not match the expected state, hence updating it", clusterRole.Name))
 		clusterRole.Rules = expectedPolicyRules
 		return r.Client.Update(ctx, clusterRole)
 	}
@@ -235,7 +336,7 @@ func (r *RolloutManagerReconciler) reconcileRolloutsAggregateToViewClusterRole(c
 	return nil
 }
 
-// Reconcile rollouts metrics service.
+// Reconcile Rollouts Metrics Service.
 func (r *RolloutManagerReconciler) reconcileRolloutsMetricsService(ctx context.Context, cr *rolloutsmanagerv1alpha1.RolloutManager) error {
 
 	expectedSvc := &corev1.Service{
@@ -245,7 +346,7 @@ func (r *RolloutManagerReconciler) reconcileRolloutsMetricsService(ctx context.C
 		},
 	}
 	setRolloutsLabels(&expectedSvc.ObjectMeta)
-	// overwrite the annotations for rollouts metrics service
+	// overwrite the annotations for Rollouts Metrics Service
 	expectedSvc.ObjectMeta.Labels["app.kubernetes.io/name"] = DefaultArgoRolloutsMetricsServiceName
 	expectedSvc.ObjectMeta.Labels["app.kubernetes.io/component"] = "server"
 
@@ -264,21 +365,22 @@ func (r *RolloutManagerReconciler) reconcileRolloutsMetricsService(ctx context.C
 
 	actualSvc := &corev1.Service{}
 
-	// Fetch the service if exists and store that in actualSvc.
+	// Fetch the Service if exists and store that in actualSvc.
 	if err := fetchObject(ctx, r.Client, cr.Namespace, expectedSvc.Name, actualSvc); err != nil {
 		if !apierrors.IsNotFound(err) {
-			return fmt.Errorf("failed to get the service %s : %s", expectedSvc.Name, err)
+			return fmt.Errorf("failed to get the Service %s: %w", expectedSvc.Name, err)
 		}
 
 		if err := controllerutil.SetControllerReference(cr, expectedSvc, r.Scheme); err != nil {
 			return err
 		}
 
-		log.Info(fmt.Sprintf("Creating service %s", expectedSvc.Name))
+		log.Info(fmt.Sprintf("Creating Service %s", expectedSvc.Name))
 		return r.Client.Create(ctx, expectedSvc)
 	}
 
 	if !reflect.DeepEqual(actualSvc.Spec.Ports, expectedSvc.Spec.Ports) {
+		log.Info(fmt.Sprintf("Ports of Service %s do not match the expected state, hence updating it", actualSvc.Name))
 		actualSvc.Spec.Ports = expectedSvc.Spec.Ports
 		return r.Client.Create(ctx, actualSvc)
 	}
@@ -286,7 +388,7 @@ func (r *RolloutManagerReconciler) reconcileRolloutsMetricsService(ctx context.C
 	return nil
 }
 
-// Reconciles secrets for rollouts controller
+// Reconciles Secrets for Rollouts controller
 func (r *RolloutManagerReconciler) reconcileRolloutsSecrets(ctx context.Context, cr *rolloutsmanagerv1alpha1.RolloutManager) error {
 	secret := &corev1.Secret{
 		ObjectMeta: metav1.ObjectMeta{
@@ -298,14 +400,14 @@ func (r *RolloutManagerReconciler) reconcileRolloutsSecrets(ctx context.Context,
 
 	if err := fetchObject(ctx, r.Client, cr.Namespace, secret.Name, secret); err != nil {
 		if !apierrors.IsNotFound(err) {
-			return fmt.Errorf("failed to get the secret %s : %s", secret.Name, err)
+			return fmt.Errorf("failed to get the Secret %s: %w", secret.Name, err)
 		}
 
 		if err := controllerutil.SetControllerReference(cr, secret, r.Scheme); err != nil {
 			return err
 		}
 
-		log.Info(fmt.Sprintf("Creating secret %s", secret.Name))
+		log.Info(fmt.Sprintf("Creating Secret %s", secret.Name))
 		return r.Client.Create(ctx, secret)
 	}
 
@@ -734,7 +836,7 @@ func GetPolicyRules() []rbacv1.PolicyRule {
 }
 
 // Returns PolicyRules for the Cluster Role argo-rollouts-aggregate-to-admin
-func getAggregateToAdminPolicyRules() []rbacv1.PolicyRule {
+func GetAggregateToAdminPolicyRules() []rbacv1.PolicyRule {
 	return []rbacv1.PolicyRule{
 		{
 			APIGroups: []string{
@@ -764,7 +866,7 @@ func getAggregateToAdminPolicyRules() []rbacv1.PolicyRule {
 }
 
 // Returns PolicyRules for the Cluster Role argo-rollouts-aggregate-to-edit
-func getAggregateToEditPolicyRules() []rbacv1.PolicyRule {
+func GetAggregateToEditPolicyRules() []rbacv1.PolicyRule {
 	return []rbacv1.PolicyRule{
 		{
 			APIGroups: []string{
@@ -794,7 +896,7 @@ func getAggregateToEditPolicyRules() []rbacv1.PolicyRule {
 }
 
 // Returns PolicyRules for the Cluster Role argo-rollouts-aggregate-to-view
-func getAggregateToViewPolicyRules() []rbacv1.PolicyRule {
+func GetAggregateToViewPolicyRules() []rbacv1.PolicyRule {
 	return []rbacv1.PolicyRule{
 		{
 			APIGroups: []string{
diff --git a/controllers/utils.go b/controllers/utils.go
index 899603f..3b37cd9 100644
--- a/controllers/utils.go
+++ b/controllers/utils.go
@@ -8,12 +8,21 @@ import (
 	"sort"
 	"strings"
 
+	rolloutsmanagerv1alpha1 "github.com/argoproj-labs/argo-rollouts-manager/api/v1alpha1"
+	"github.com/go-logr/logr"
 	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
+const (
+	UnsupportedRolloutManagerConfiguration   = "when there exists a cluster-scoped RolloutManager on the cluster, there may not exist another: only a single cluster-scoped RolloutManager is supported"
+	UnsupportedRolloutManagerClusterScoped   = "when Subscription has environment variable NAMESPACE_SCOPED_ARGO_ROLLOUTS set to True, there may not exist any cluster-scoped RolloutManagers: in this case, only namespace-scoped RolloutManager resources are supported"
+	UnsupportedRolloutManagerNamespaceScoped = "when Subscription has environment variable NAMESPACE_SCOPED_ARGO_ROLLOUTS set to False, there may not exist any namespace-scoped RolloutManagers: only a single cluster-scoped RolloutManager is supported"
+)
+
 func setRolloutsLabels(obj *metav1.ObjectMeta) {
 	obj.Labels = map[string]string{}
 	obj.Labels["app.kubernetes.io/name"] = DefaultArgoRolloutsResourceName
@@ -24,6 +33,9 @@ func setRolloutsLabels(obj *metav1.ObjectMeta) {
 // fetchObject will retrieve the object with the given namespace and name using the Kubernetes API.
 // The result will be stored in the given object.
 func fetchObject(ctx context.Context, client client.Client, namespace string, name string, obj client.Object) error {
+	if namespace == "" {
+		return client.Get(ctx, types.NamespacedName{Name: name}, obj)
+	}
 	return client.Get(ctx, types.NamespacedName{Namespace: namespace, Name: name}, obj)
 }
 
@@ -132,3 +144,198 @@ func isMergable(extraArgs []string, cmd []string) error {
 	}
 	return nil
 }
+
+// validateRolloutsScope will check scope of Rollouts controller configured in RolloutManager and scope allowed by Admin (Configured in Subscription.Spec.Config.Env)
+func validateRolloutsScope(ctx context.Context, k8sClient client.Client, cr *rolloutsmanagerv1alpha1.RolloutManager, namespaceScopedArgoRolloutsController bool) error {
+
+	// If namespace-scoped Rollouts controller is allowed according to Subscription.Spec.Config.Env value
+	if namespaceScopedArgoRolloutsController {
+
+		// if RolloutManager being reconciled will create cluster-scoped Rollouts controller, then don't allow it.
+		if !cr.Spec.NamespaceScoped {
+			cr.Status.Phase = rolloutsmanagerv1alpha1.PhaseFailure
+			cr.Status.RolloutController = rolloutsmanagerv1alpha1.PhaseFailure
+
+			if err := k8sClient.Status().Update(ctx, cr); err != nil {
+				return fmt.Errorf("error updating the RolloutManager CR status: %w", err)
+			}
+
+			return fmt.Errorf(UnsupportedRolloutManagerClusterScoped)
+		}
+
+		// allow only namespace-scoped Rollouts controller
+		return nil
+
+	} else { // If cluster-scoped Rollout controller is allowed according to Subscription.Spec.Config.Env value
+
+		// if RolloutManager being reconciled will create namespace-scoped Rollouts controller, then don't allow it.
+		if cr.Spec.NamespaceScoped {
+			cr.Status.Phase = rolloutsmanagerv1alpha1.PhaseFailure
+			cr.Status.RolloutController = rolloutsmanagerv1alpha1.PhaseFailure
+
+			if err := k8sClient.Status().Update(ctx, cr); err != nil {
+				return fmt.Errorf("error updating the RolloutManager CR status: %w", err)
+			}
+
+			return fmt.Errorf(UnsupportedRolloutManagerNamespaceScoped)
+		}
+
+		// allow only cluster-scoped RolloutManager
+		return nil
+	}
+}
+
+// checkForExistingRolloutManager will return error if more than one cluster-scoped RolloutManagers are created.
+// because only one cluster-scoped or all namespace-scoped RolloutManagers are supported.
+func checkForExistingRolloutManager(ctx context.Context, k8sClient client.Client, cr *rolloutsmanagerv1alpha1.RolloutManager) error {
+
+	// if it is namespace-scoped then return no error
+	// because multiple namespace-scoped RolloutManagers are allowed if validateRolloutsScope check is passed earlier.
+	if cr.Spec.NamespaceScoped {
+		return nil
+	}
+
+	// get the list of all RolloutManagers available across all namespaces
+	rolloutManagerList := rolloutsmanagerv1alpha1.RolloutManagerList{}
+	if err := k8sClient.List(ctx, &rolloutManagerList); err != nil {
+		return fmt.Errorf("failed to get the list of RolloutManager CRs from cluster: %w", err)
+	}
+
+	// if there are more than one RolloutManagers available, then check if any cluster-scoped RolloutManager exists,
+	// if yes then return error for this CR, because only one cluster-scoped RolloutManagers is supported
+	for _, rolloutManager := range rolloutManagerList.Items {
+
+		// if current RolloutManager is being iterated, then skip it, because we are looking for other cluster-scoped RolloutManagers.
+		if rolloutManager.Name == cr.Name && rolloutManager.Namespace == cr.Namespace {
+			continue
+		}
+
+		// if there is a another cluster-scoped RolloutManager available in cluster then skip reconciliation of this one and set status to failure.
+		if !rolloutManager.Spec.NamespaceScoped {
+			cr.Status.Phase = rolloutsmanagerv1alpha1.PhaseFailure
+			cr.Status.RolloutController = rolloutsmanagerv1alpha1.PhaseFailure
+
+			if err := k8sClient.Status().Update(ctx, cr); err != nil {
+				return fmt.Errorf("error updating the RolloutManager CR status: %w", err)
+			}
+
+			return fmt.Errorf(UnsupportedRolloutManagerConfiguration)
+		}
+	}
+	// either there are no existing RolloutManagers or all are namespace-scoped or only one cluster-scoped RolloutManagers exists,
+	// so continue reconciliation of this CR
+	return nil
+}
+
+func multipleRolloutManagersExist(err error) bool {
+	return err.Error() == UnsupportedRolloutManagerConfiguration
+}
+
+func invalidRolloutScope(err error) bool {
+	return err.Error() == UnsupportedRolloutManagerClusterScoped ||
+		err.Error() == UnsupportedRolloutManagerNamespaceScoped
+}
+
+// updateStatusConditionOfRolloutManager calls Set Condition of RolloutManager status
+func updateStatusConditionOfRolloutManager(ctx context.Context, newCondition metav1.Condition, rm *rolloutsmanagerv1alpha1.RolloutManager, k8sClient client.Client, log logr.Logger) error {
+
+	if err := k8sClient.Get(ctx, client.ObjectKeyFromObject(rm), rm); err != nil {
+		if apierrors.IsNotFound(err) {
+			log.Error(err, "unable to fetch RolloutManager")
+			return nil
+		} else {
+			return err
+		}
+	}
+
+	changed, newConditions := insertOrUpdateConditionsInSlice(newCondition, rm.Status.Conditions)
+
+	if changed {
+		rm.Status.Conditions = newConditions
+
+		if err := k8sClient.Status().Update(ctx, rm); err != nil {
+			log.Error(err, "unable to update RolloutManager status condition")
+			return err
+		}
+	}
+	return nil
+}
+
+// insertOrUpdateConditionsInSlice is a generic function for inserting/updating metav1.Condition into a slice of []metav1.Condition
+func insertOrUpdateConditionsInSlice(newCondition metav1.Condition, existingConditions []metav1.Condition) (bool, []metav1.Condition) {
+
+	// Check if condition with same type is already set, if Yes then check if content is same,
+	// If content is not same update LastTransitionTime
+
+	index := -1
+	for i, Condition := range existingConditions {
+		if Condition.Type == newCondition.Type {
+			index = i
+			break
+		}
+	}
+
+	now := metav1.Now()
+
+	changed := false
+
+	if index == -1 {
+		newCondition.LastTransitionTime = now
+		existingConditions = append(existingConditions, newCondition)
+		changed = true
+
+	} else if existingConditions[index].Message != newCondition.Message ||
+		existingConditions[index].Reason != newCondition.Reason ||
+		existingConditions[index].Status != newCondition.Status {
+
+		newCondition.LastTransitionTime = now
+		existingConditions[index] = newCondition
+		changed = true
+	}
+
+	return changed, existingConditions
+
+}
+
+// createCondition returns Condition based on input provided.
+// 1. Returns Success condition if no error message is provided, all fields are default.
+// 2. If more than 1 reasons are there then its an internal error.
+// 3. If 1 Reason is provided, it returns Failed condition having all default fields except Reason.
+// 4. If Message is provided, it returns Failed condition having all default fields except Message.
+func createCondition(message string, reason ...string) metav1.Condition {
+
+	if message == "" {
+		return metav1.Condition{
+			Type:    rolloutsmanagerv1alpha1.RolloutManagerConditionType,
+			Reason:  rolloutsmanagerv1alpha1.RolloutManagerReasonSuccess,
+			Message: "",
+			Status:  metav1.ConditionTrue,
+		}
+	}
+
+	if len(reason) > 0 {
+
+		if len(reason) > 1 { // Only 0 or 1 reasons are supported.
+			return metav1.Condition{
+				Type:    rolloutsmanagerv1alpha1.RolloutManagerConditionType,
+				Reason:  rolloutsmanagerv1alpha1.RolloutManagerReasonErrorOccurred,
+				Message: "An internal error occurred",
+				Status:  metav1.ConditionTrue,
+			}
+		}
+
+		return metav1.Condition{
+			Type:    rolloutsmanagerv1alpha1.RolloutManagerConditionType,
+			Reason:  reason[0],
+			Message: message,
+			Status:  metav1.ConditionFalse,
+		}
+	}
+
+	return metav1.Condition{
+		Type:    rolloutsmanagerv1alpha1.RolloutManagerConditionType,
+		Reason:  rolloutsmanagerv1alpha1.RolloutManagerReasonErrorOccurred,
+		Message: message,
+		Status:  metav1.ConditionFalse,
+	}
+}
diff --git a/go.mod b/go.mod
index 4560f68..d1239d6 100644
--- a/go.mod
+++ b/go.mod
@@ -4,6 +4,7 @@ go 1.21
 
 require (
 	github.com/argoproj/argo-rollouts v1.6.6
+	github.com/go-logr/logr v1.2.4
 	github.com/onsi/ginkgo/v2 v2.11.0
 	github.com/onsi/gomega v1.27.10
 	go.uber.org/zap v1.24.0
@@ -22,7 +23,6 @@ require (
 	github.com/evanphx/json-patch v5.6.0+incompatible // indirect
 	github.com/evanphx/json-patch/v5 v5.6.0 // indirect
 	github.com/fsnotify/fsnotify v1.6.0 // indirect
-	github.com/go-logr/logr v1.2.4 // indirect
 	github.com/go-logr/zapr v1.2.3 // indirect
 	github.com/go-openapi/jsonpointer v0.19.5 // indirect
 	github.com/go-openapi/jsonreference v0.20.0 // indirect
diff --git a/main.go b/main.go
index 36d8272..61d863c 100644
--- a/main.go
+++ b/main.go
@@ -19,6 +19,7 @@ package main
 import (
 	"flag"
 	"os"
+	"strings"
 
 	// Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.)
 	// to ensure that exec-entrypoint and run can make use of them.
@@ -96,10 +97,19 @@ func main() {
 		openShiftRoutePluginLocation = controllers.DefaultOpenShiftRoutePluginURL
 	}
 
+	isNamespaceScoped := strings.ToLower(os.Getenv(controllers.NamespaceScopedArgoRolloutsController)) == "true"
+
+	if isNamespaceScoped {
+		setupLog.Info("Running in namespaced-scoped mode")
+	} else {
+		setupLog.Info("Running in cluster-scoped mode")
+	}
+
 	if err = (&controllers.RolloutManagerReconciler{
-		Client:                       mgr.GetClient(),
-		Scheme:                       mgr.GetScheme(),
-		OpenShiftRoutePluginLocation: openShiftRoutePluginLocation,
+		Client:                                mgr.GetClient(),
+		Scheme:                                mgr.GetScheme(),
+		OpenShiftRoutePluginLocation:          openShiftRoutePluginLocation,
+		NamespaceScopedArgoRolloutsController: isNamespaceScoped,
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "RolloutManager")
 		os.Exit(1)


Test patch:
diff --git a/.github/workflows/e2e_tests.yml b/.github/workflows/e2e_tests.yml
index ca25572..3313bb3 100644
--- a/.github/workflows/e2e_tests.yml
+++ b/.github/workflows/e2e_tests.yml
@@ -43,10 +43,7 @@ jobs:
       - name: Download Go dependencies
         run: |
           go mod download
-      - name: Run the operator locally
-        run: |
-          make start-e2e &
       - name: Run tests
         run: |
           set -o pipefail
-          make test-e2e 2>&1 | tee /tmp/e2e-test.log
+          make start-test-e2e-all 2>&1 | tee /tmp/e2e-test.log
diff --git a/controllers/argorollouts_controller_test.go b/controllers/argorollouts_controller_test.go
new file mode 100644
index 0000000..63ebe69
--- /dev/null
+++ b/controllers/argorollouts_controller_test.go
@@ -0,0 +1,744 @@
+package rollouts
+
+import (
+	"context"
+	"fmt"
+
+	rolloutsmanagerv1alpha1 "github.com/argoproj-labs/argo-rollouts-manager/api/v1alpha1"
+	"github.com/argoproj-labs/argo-rollouts-manager/tests/e2e/fixture/k8s"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/intstr"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+)
+
+var _ = Describe("RolloutManagerReconciler tests", func() {
+	var (
+		ctx context.Context
+		rm  *rolloutsmanagerv1alpha1.RolloutManager
+	)
+
+	BeforeEach(func() {
+		ctx = context.Background()
+		rm = makeTestRolloutManager()
+	})
+
+	When("NAMESPACE_SCOPED_ARGO_ROLLOUTS environment variable is set to False.", func() {
+
+		It("Should allow cluster-scoped RolloutManager CR to be reconciled and create expected resource.", func() {
+
+			By("Create cluster-scoped RolloutManager.")
+			r := makeTestReconciler(rm)
+			Expect(createNamespace(r, rm.Namespace)).To(Succeed())
+
+			req := reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Name:      rm.Name,
+					Namespace: rm.Namespace,
+				},
+			}
+
+			res, err := r.Reconcile(ctx, req)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(res.Requeue).Should(BeFalse(), "reconcile should not requeue request")
+
+			By("Check if RolloutManager's Status.Conditions are set.")
+			Expect(r.Client.Get(ctx, types.NamespacedName{Name: rm.Name, Namespace: rm.Namespace}, rm)).To(Succeed())
+			Expect(rm.Status.Conditions[0].Type == rolloutsmanagerv1alpha1.RolloutManagerConditionType &&
+				rm.Status.Conditions[0].Reason == rolloutsmanagerv1alpha1.RolloutManagerReasonSuccess &&
+				rm.Status.Conditions[0].Message == "" &&
+				rm.Status.Conditions[0].Status == metav1.ConditionTrue).To(BeTrue())
+
+			By("Check expected resources are created.")
+			validateArgoRolloutManagerResources(rm, r.Client, false)
+
+			By("Check Role and RoleBinding are not created.")
+
+			role := &rbacv1.Role{}
+			err = r.Client.Get(ctx, types.NamespacedName{
+				Name:      DefaultArgoRolloutsResourceName,
+				Namespace: testNamespace,
+			}, role)
+			Expect(errors.IsNotFound(err)).To(BeTrue(), "Role should not be created")
+
+			roleBinding := &rbacv1.RoleBinding{}
+			err = r.Client.Get(ctx, types.NamespacedName{
+				Name:      DefaultArgoRolloutsResourceName,
+				Namespace: testNamespace,
+			}, roleBinding)
+			Expect(errors.IsNotFound(err)).To(BeTrue(), "RoleBinding should not be created")
+		})
+
+		It("Should not allow namespace-scoped RolloutManager CR to be reconciled.", func() {
+
+			By("Create namespace-scoped RolloutManager.")
+			rm.Spec.NamespaceScoped = true
+
+			r := makeTestReconciler(rm)
+			Expect(createNamespace(r, rm.Namespace)).To(Succeed())
+
+			req := reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Name:      rm.Name,
+					Namespace: rm.Namespace,
+				},
+			}
+
+			res, err := r.Reconcile(ctx, req)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(res.Requeue).Should(BeFalse(), "reconcile should not requeue request")
+
+			By("Check if RolloutManager's Status.Conditions are set.")
+			Expect(r.Client.Get(ctx, types.NamespacedName{Name: rm.Name, Namespace: rm.Namespace}, rm)).To(Succeed())
+			Expect(rm.Status.Conditions[0].Type == rolloutsmanagerv1alpha1.RolloutManagerConditionType &&
+				rm.Status.Conditions[0].Reason == rolloutsmanagerv1alpha1.RolloutManagerReasonInvalidScoped &&
+				rm.Status.Conditions[0].Message == UnsupportedRolloutManagerNamespaceScoped &&
+				rm.Status.Conditions[0].Status == metav1.ConditionFalse).To(BeTrue())
+		})
+
+		It("If a failed namespace-scoped RolloutManager is available in cluster, cluster-scoped RolloutManager should still work.", func() {
+
+			By("1st RM: Create namespace-scoped RolloutManager.")
+			rm.Spec.NamespaceScoped = true
+
+			r := makeTestReconciler(rm)
+			Expect(createNamespace(r, rm.Namespace)).To(Succeed())
+
+			req := reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Name:      rm.Name,
+					Namespace: rm.Namespace,
+				},
+			}
+
+			res, err := r.Reconcile(ctx, req)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(res.Requeue).Should(BeFalse(), "reconcile should not requeue request")
+
+			By("1st RM: Check if RolloutManager's Status.Conditions are set.")
+			Expect(r.Client.Get(ctx, types.NamespacedName{Name: rm.Name, Namespace: rm.Namespace}, rm)).To(Succeed())
+			Expect(rm.Status.Conditions[0].Type == rolloutsmanagerv1alpha1.RolloutManagerConditionType &&
+				rm.Status.Conditions[0].Reason == rolloutsmanagerv1alpha1.RolloutManagerReasonInvalidScoped &&
+				rm.Status.Conditions[0].Message == UnsupportedRolloutManagerNamespaceScoped &&
+				rm.Status.Conditions[0].Status == metav1.ConditionFalse).To(BeTrue())
+
+			By("2nd RM: Create cluster-scoped RolloutManager.")
+			rm2 := makeTestRolloutManager()
+
+			r2 := makeTestReconciler(rm2)
+			Expect(createNamespace(r2, rm2.Namespace)).To(Succeed())
+
+			req2 := reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Name:      rm2.Name,
+					Namespace: rm2.Namespace,
+				},
+			}
+
+			res2, err := r2.Reconcile(ctx, req2)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(res2.Requeue).Should(BeFalse(), "reconcile should not requeue request")
+
+			By("2nd RM: Check if RolloutManager's Status.Conditions are set.")
+			Expect(r2.Client.Get(ctx, types.NamespacedName{Name: rm2.Name, Namespace: rm2.Namespace}, rm2)).To(Succeed())
+			Expect(rm2.Status.Conditions[0].Type == rolloutsmanagerv1alpha1.RolloutManagerConditionType &&
+				rm2.Status.Conditions[0].Reason == rolloutsmanagerv1alpha1.RolloutManagerReasonSuccess &&
+				rm2.Status.Conditions[0].Message == "" &&
+				rm2.Status.Conditions[0].Status == metav1.ConditionTrue).To(BeTrue())
+
+			By("2nd RM: Check expected resources are created.")
+			validateArgoRolloutManagerResources(rm2, r2.Client, false)
+
+			By("2nd RM: Check Role and RoleBinding are not created.")
+			role := &rbacv1.Role{}
+			err = r2.Client.Get(ctx, types.NamespacedName{
+				Name:      DefaultArgoRolloutsResourceName,
+				Namespace: testNamespace,
+			}, role)
+			Expect(errors.IsNotFound(err)).To(BeTrue(), "Role should not be created")
+
+			roleBinding := &rbacv1.RoleBinding{}
+			err = r2.Client.Get(ctx, types.NamespacedName{
+				Name:      DefaultArgoRolloutsResourceName,
+				Namespace: testNamespace,
+			}, roleBinding)
+			Expect(errors.IsNotFound(err)).To(BeTrue(), "RoleBinding should not be created")
+		})
+
+		It("should not allow more than one cluster-scoped RolloutManagers.", func() {
+
+			By("1st RM: Create 1st cluster-scoped RolloutManager.")
+			r := makeTestReconciler(rm)
+			Expect(createNamespace(r, rm.Namespace)).To(Succeed())
+
+			req := reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Name:      rm.Name,
+					Namespace: rm.Namespace,
+				},
+			}
+
+			res, err := r.Reconcile(ctx, req)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(res.Requeue).Should(BeFalse(), "reconcile should not requeue request")
+
+			By("1st RM: Check if RolloutManager's Status.Conditions are set.")
+			Expect(r.Client.Get(ctx, types.NamespacedName{Name: rm.Name, Namespace: rm.Namespace}, rm)).To(Succeed())
+			Expect(rm.Status.Conditions[0].Type == rolloutsmanagerv1alpha1.RolloutManagerConditionType &&
+				rm.Status.Conditions[0].Reason == rolloutsmanagerv1alpha1.RolloutManagerReasonSuccess &&
+				rm.Status.Conditions[0].Message == "" &&
+				rm.Status.Conditions[0].Status == metav1.ConditionTrue).To(BeTrue())
+
+			By("2nd RM: Create 2nd cluster-scoped RolloutManager and verify it failed.")
+			rm2 := makeTestRolloutManager()
+			rm2.Name = "test-rm"
+			rm2.Namespace = "test-ns"
+
+			Expect(createNamespace(r, rm2.Namespace)).To(Succeed())
+			Expect(r.Client.Create(ctx, rm2)).ToNot(HaveOccurred())
+
+			req2 := reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Name:      rm2.Name,
+					Namespace: rm2.Namespace,
+				},
+			}
+
+			res2, err := r.Reconcile(ctx, req2)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(res2.Requeue).Should(BeFalse(), "reconcile should not requeue request")
+
+			By("2nd RM: Check if RolloutManager's Status.Conditions are set.")
+			Expect(r.Client.Get(ctx, types.NamespacedName{Name: rm2.Name, Namespace: rm2.Namespace}, rm2)).To(Succeed())
+			Expect(rm2.Status.Conditions[0].Type == rolloutsmanagerv1alpha1.RolloutManagerConditionType &&
+				rm2.Status.Conditions[0].Reason == rolloutsmanagerv1alpha1.RolloutManagerReasonMultipleClusterScopedRolloutManager &&
+				rm2.Status.Conditions[0].Message == UnsupportedRolloutManagerConfiguration &&
+				rm2.Status.Conditions[0].Status == metav1.ConditionFalse).To(BeTrue())
+
+			By("1st RM: Reconcile 1st RolloutManager's once again and check it is also failed.")
+			res, err = r.Reconcile(ctx, req)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(res.Requeue).Should(BeFalse(), "reconcile should not requeue request")
+
+			By("1st RM: Check if RolloutManager's Status.Conditions are set.")
+			Expect(r.Client.Get(ctx, types.NamespacedName{Name: rm.Name, Namespace: rm.Namespace}, rm)).To(Succeed())
+			Expect(rm.Status.Conditions[0].Type == rolloutsmanagerv1alpha1.RolloutManagerConditionType &&
+				rm.Status.Conditions[0].Reason == rolloutsmanagerv1alpha1.RolloutManagerReasonMultipleClusterScopedRolloutManager &&
+				rm.Status.Conditions[0].Message == UnsupportedRolloutManagerConfiguration &&
+				rm.Status.Conditions[0].Status == metav1.ConditionFalse).To(BeTrue())
+		})
+	})
+
+	When("NAMESPACE_SCOPED_ARGO_ROLLOUTS environment variable is set to True.", func() {
+
+		It("Should allow namespace-scoped RolloutManager CR to be reconciled and create expected resource.", func() {
+
+			By("Create namespace-scoped RolloutManager.")
+			rm.Spec.NamespaceScoped = true
+
+			r := makeTestReconciler(rm)
+			r.NamespaceScopedArgoRolloutsController = true
+
+			Expect(createNamespace(r, rm.Namespace)).To(Succeed())
+
+			req := reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Name:      rm.Name,
+					Namespace: rm.Namespace,
+				},
+			}
+
+			res, err := r.Reconcile(ctx, req)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(res.Requeue).Should(BeFalse(), "reconcile should not requeue request")
+
+			By("Check if RolloutManager's Status.Conditions are set.")
+			Expect(r.Client.Get(ctx, types.NamespacedName{Name: rm.Name, Namespace: rm.Namespace}, rm)).To(Succeed())
+			Expect(rm.Status.Conditions[0].Type == rolloutsmanagerv1alpha1.RolloutManagerConditionType &&
+				rm.Status.Conditions[0].Reason == rolloutsmanagerv1alpha1.RolloutManagerReasonSuccess &&
+				rm.Status.Conditions[0].Message == "" &&
+				rm.Status.Conditions[0].Status == metav1.ConditionTrue).To(BeTrue())
+
+			By("Check expected resources are created.")
+			validateArgoRolloutManagerResources(rm, r.Client, true)
+
+			By("Check ClusterRole and ClusterRoleBinding are not created.")
+			clusterRole := &rbacv1.ClusterRole{}
+			err = r.Client.Get(ctx, types.NamespacedName{
+				Name:      DefaultArgoRolloutsResourceName,
+				Namespace: testNamespace,
+			}, clusterRole)
+			Expect(errors.IsNotFound(err)).To(BeTrue(), "ClusterRole should not be created")
+
+			clusterRoleBinding := &rbacv1.ClusterRoleBinding{}
+			err = r.Client.Get(ctx, types.NamespacedName{
+				Name:      DefaultArgoRolloutsResourceName,
+				Namespace: testNamespace,
+			}, clusterRoleBinding)
+			Expect(errors.IsNotFound(err)).To(BeTrue(), "ClusterRoleBinding should not be created")
+		})
+
+		It("Should not allow cluster-scoped RolloutManager CR to be reconciled.", func() {
+
+			By("Create cluster-scoped RolloutManager.")
+			r := makeTestReconciler(rm)
+			r.NamespaceScopedArgoRolloutsController = true
+			Expect(createNamespace(r, rm.Namespace)).To(Succeed())
+
+			req := reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Name:      rm.Name,
+					Namespace: rm.Namespace,
+				},
+			}
+
+			res, err := r.Reconcile(ctx, req)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(res.Requeue).Should(BeFalse(), "reconcile should not requeue request")
+
+			By("Check if RolloutManager's Status.Conditions are set.")
+			Expect(r.Client.Get(ctx, types.NamespacedName{Name: rm.Name, Namespace: rm.Namespace}, rm)).To(Succeed())
+			Expect(rm.Status.Conditions[0].Type == rolloutsmanagerv1alpha1.RolloutManagerConditionType &&
+				rm.Status.Conditions[0].Reason == rolloutsmanagerv1alpha1.RolloutManagerReasonInvalidScoped &&
+				rm.Status.Conditions[0].Message == UnsupportedRolloutManagerClusterScoped &&
+				rm.Status.Conditions[0].Status == metav1.ConditionFalse).To(BeTrue())
+		})
+
+		It("If a failed cluster-scoped RolloutManager is available in cluster, namespace-scoped RolloutManager should still work.", func() {
+
+			By("1st RM: Create cluster-scoped RolloutManager.")
+			r := makeTestReconciler(rm)
+			r.NamespaceScopedArgoRolloutsController = true
+			Expect(createNamespace(r, rm.Namespace)).To(Succeed())
+
+			req := reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Name:      rm.Name,
+					Namespace: rm.Namespace,
+				},
+			}
+
+			res, err := r.Reconcile(ctx, req)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(res.Requeue).Should(BeFalse(), "reconcile should not requeue request")
+
+			By("1st RM: Check if RolloutManager's Status.Conditions are set.")
+			Expect(r.Client.Get(ctx, types.NamespacedName{Name: rm.Name, Namespace: rm.Namespace}, rm)).To(Succeed())
+			Expect(rm.Status.Conditions[0].Type == rolloutsmanagerv1alpha1.RolloutManagerConditionType &&
+				rm.Status.Conditions[0].Reason == rolloutsmanagerv1alpha1.RolloutManagerReasonInvalidScoped &&
+				rm.Status.Conditions[0].Message == UnsupportedRolloutManagerClusterScoped &&
+				rm.Status.Conditions[0].Status == metav1.ConditionFalse).To(BeTrue())
+
+			By("2nd RM: Create namespace-scoped RolloutManager.")
+			rm2 := makeTestRolloutManager()
+			rm2.Name = "test-rm"
+			rm2.Namespace = "test-ns"
+			rm2.Spec.NamespaceScoped = true
+
+			r2 := makeTestReconciler(rm2)
+			r2.NamespaceScopedArgoRolloutsController = true
+
+			Expect(createNamespace(r2, rm2.Namespace)).To(Succeed())
+
+			req2 := reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Name:      rm2.Name,
+					Namespace: rm2.Namespace,
+				},
+			}
+
+			res2, err := r2.Reconcile(ctx, req2)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(res2.Requeue).Should(BeFalse(), "reconcile should not requeue request")
+
+			By("2nd RM: Check if RolloutManager's Status.Conditions are set.")
+			Expect(r2.Client.Get(ctx, types.NamespacedName{Name: rm2.Name, Namespace: rm2.Namespace}, rm2)).To(Succeed())
+			Expect(rm2.Status.Conditions[0].Type == rolloutsmanagerv1alpha1.RolloutManagerConditionType &&
+				rm2.Status.Conditions[0].Reason == rolloutsmanagerv1alpha1.RolloutManagerReasonSuccess &&
+				rm2.Status.Conditions[0].Message == "" &&
+				rm2.Status.Conditions[0].Status == metav1.ConditionTrue).To(BeTrue())
+
+			By("2nd RM: Check expected resources are created.")
+			validateArgoRolloutManagerResources(rm2, r2.Client, true)
+
+			By("2nd RM: Check ClusterRole and ClusterRoleBinding are not created.")
+			clusterRole := &rbacv1.ClusterRole{}
+			err = r2.Client.Get(ctx, types.NamespacedName{
+				Name:      DefaultArgoRolloutsResourceName,
+				Namespace: testNamespace,
+			}, clusterRole)
+			Expect(errors.IsNotFound(err)).To(BeTrue(), "ClusterRole should not be created")
+
+			clusterRoleBinding := &rbacv1.ClusterRoleBinding{}
+			err = r2.Client.Get(ctx, types.NamespacedName{
+				Name:      DefaultArgoRolloutsResourceName,
+				Namespace: testNamespace,
+			}, clusterRoleBinding)
+			Expect(errors.IsNotFound(err)).To(BeTrue(), "ClusterRoleBinding should not be created")
+		})
+
+		It("Should allow multiple namespace-scoped RolloutManagers in different namespaces.", func() {
+
+			By("1st RM: Create namespace-scoped RolloutManager.")
+			rm.Spec.NamespaceScoped = true
+
+			r := makeTestReconciler(rm)
+			r.NamespaceScopedArgoRolloutsController = true
+
+			Expect(createNamespace(r, rm.Namespace)).To(Succeed())
+
+			req := reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Name:      rm.Name,
+					Namespace: rm.Namespace,
+				},
+			}
+
+			res, err := r.Reconcile(ctx, req)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(res.Requeue).Should(BeFalse(), "reconcile should not requeue request")
+
+			By("1st RM: Check if RolloutManager's Status.Conditions are set.")
+			Expect(r.Client.Get(ctx, types.NamespacedName{Name: rm.Name, Namespace: rm.Namespace}, rm)).To(Succeed())
+			Expect(rm.Status.Conditions[0].Type == rolloutsmanagerv1alpha1.RolloutManagerConditionType &&
+				rm.Status.Conditions[0].Reason == rolloutsmanagerv1alpha1.RolloutManagerReasonSuccess &&
+				rm.Status.Conditions[0].Message == "" &&
+				rm.Status.Conditions[0].Status == metav1.ConditionTrue).To(BeTrue())
+
+			By("1st RM: Check expected resources are created in 1st namespace.")
+			validateArgoRolloutManagerResources(rm, r.Client, true)
+
+			By("2nd RM: Create namespace-scoped RolloutManager.")
+			rm2 := makeTestRolloutManager()
+			rm2.Spec.NamespaceScoped = true
+			rm2.Name = "test-rm"
+			rm2.Namespace = "test-ns"
+
+			Expect(createNamespace(r, rm2.Namespace)).To(Succeed())
+
+			r2 := makeTestReconciler(rm2)
+			r2.NamespaceScopedArgoRolloutsController = true
+			Expect(createNamespace(r2, rm2.Namespace)).To(Succeed())
+
+			req2 := reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Name:      rm2.Name,
+					Namespace: rm2.Namespace,
+				},
+			}
+
+			res2, err := r2.Reconcile(ctx, req2)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(res2.Requeue).Should(BeFalse(), "reconcile should not requeue request")
+
+			By("2nd RM: Check if RolloutManager's Status.Conditions are set.")
+			Expect(r2.Client.Get(ctx, types.NamespacedName{Name: rm2.Name, Namespace: rm2.Namespace}, rm2)).To(Succeed())
+			Expect(rm2.Status.Conditions[0].Type == rolloutsmanagerv1alpha1.RolloutManagerConditionType &&
+				rm2.Status.Conditions[0].Reason == rolloutsmanagerv1alpha1.RolloutManagerReasonSuccess &&
+				rm2.Status.Conditions[0].Message == "" &&
+				rm2.Status.Conditions[0].Status == metav1.ConditionTrue).To(BeTrue())
+
+			By("2nd RM: Check expected resources are created in 2nd namespace.")
+			validateArgoRolloutManagerResources(rm2, r2.Client, true)
+		})
+	})
+})
+
+func validateArgoRolloutManagerResources(rolloutsManager *rolloutsmanagerv1alpha1.RolloutManager, k8sClient client.Client, namespaceScoped bool) {
+
+	By("Verify that ServiceAccount is created.")
+	validateServiceAccount(k8sClient, rolloutsManager)
+
+	if namespaceScoped {
+		By("Verify that Argo-Rollout Role is created.")
+		validateArgoRolloutsRole(k8sClient, rolloutsManager)
+	} else {
+		By("Verify that Argo-Rollout ClusterRoles is created.")
+		validateArgoRolloutsClusterRole(k8sClient)
+	}
+
+	By("Verify that Aggregate-to-Admin ClusterRole is created.")
+	validateAggregateToAdminClusterRole(k8sClient)
+
+	By("Verify that Aggregate-to-Edit ClusterRole is created.")
+	validateAggregateToEditClusterRole(k8sClient)
+
+	By("Verify that Aggregate-to-View ClusterRole is created.")
+	validateAggregateToViewClusterRole(k8sClient)
+
+	if namespaceScoped {
+		By("Verify that RoleBinding is created.")
+		validateRoleBinding(k8sClient, rolloutsManager)
+	} else {
+		By("Verify that ClusterRoleBinding is created.")
+		validateClusterRoleBinding(k8sClient, rolloutsManager)
+	}
+
+	By("Verify that Service is created.")
+	validateService(k8sClient, rolloutsManager)
+
+	By("Verify that Secret is created.")
+	validateSecret(k8sClient, rolloutsManager)
+
+	By("Verify that Argo-Rollouts Deployment is created and it is in Ready state.")
+	validateDeployment(k8sClient, rolloutsManager)
+}
+
+func validateServiceAccount(k8sClient client.Client, rolloutsManager *rolloutsmanagerv1alpha1.RolloutManager) {
+	sa := &corev1.ServiceAccount{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      DefaultArgoRolloutsResourceName,
+			Namespace: rolloutsManager.Namespace,
+		},
+	}
+	Eventually(sa, "10s", "1s").Should(k8s.ExistByName(k8sClient))
+
+	By("Verify that ServiceAccount has correct labels.")
+	ensureLabels(&sa.ObjectMeta)
+}
+
+func validateArgoRolloutsRole(k8sClient client.Client, rolloutsManager *rolloutsmanagerv1alpha1.RolloutManager) {
+	role := &rbacv1.Role{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      DefaultArgoRolloutsResourceName,
+			Namespace: rolloutsManager.Namespace,
+		},
+	}
+	Eventually(role, "10s", "1s").Should(k8s.ExistByName(k8sClient))
+
+	By("Verify that Role has correct labels.")
+	ensureLabels(&role.ObjectMeta)
+
+	By("Verify that Role has correct policy rules.")
+	Expect(role.Rules).To(ConsistOf(GetPolicyRules()))
+}
+
+func validateArgoRolloutsClusterRole(k8sClient client.Client) {
+	clusterRole := &rbacv1.ClusterRole{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: DefaultArgoRolloutsResourceName,
+		},
+	}
+	Eventually(clusterRole, "30s", "1s").Should(k8s.ExistByName(k8sClient))
+
+	By("Verify that ClusterRole has correct labels.")
+	ensureLabels(&clusterRole.ObjectMeta)
+
+	By("Verify that ClusterRole has correct policy rules.")
+	Expect(clusterRole.Rules).To(ConsistOf(GetPolicyRules()))
+}
+
+func validateAggregateToAdminClusterRole(k8sClient client.Client) {
+
+	aggregationType := "aggregate-to-admin"
+	clusterRoleName := fmt.Sprintf("%s-%s", DefaultArgoRolloutsResourceName, aggregationType)
+
+	clusterRole := &rbacv1.ClusterRole{ObjectMeta: metav1.ObjectMeta{
+		Name: clusterRoleName},
+	}
+	Eventually(clusterRole, "30s", "1s").Should(k8s.ExistByName(k8sClient))
+
+	By("Verify that ClusterRole has correct labels.")
+	ensureAggregateLabels(&clusterRole.ObjectMeta, aggregationType)
+
+	By("Verify that ClusterRole has correct policy rules.")
+	Expect(clusterRole.Rules).To(ConsistOf(GetAggregateToAdminPolicyRules()))
+}
+
+func validateAggregateToEditClusterRole(k8sClient client.Client) {
+
+	aggregationType := "aggregate-to-edit"
+	clusterRoleName := fmt.Sprintf("%s-%s", DefaultArgoRolloutsResourceName, aggregationType)
+
+	clusterRole := &rbacv1.ClusterRole{ObjectMeta: metav1.ObjectMeta{
+		Name: clusterRoleName,
+	},
+	}
+	Eventually(clusterRole, "30s", "1s").Should(k8s.ExistByName(k8sClient))
+
+	By("Verify that ClusterRole has correct labels.")
+	ensureAggregateLabels(&clusterRole.ObjectMeta, aggregationType)
+
+	By("Verify that ClusterRole has correct policy rules.")
+	Expect(clusterRole.Rules).To(ConsistOf(GetAggregateToEditPolicyRules()))
+}
+
+func validateAggregateToViewClusterRole(k8sClient client.Client) {
+
+	aggregationType := "aggregate-to-view"
+	clusterRoleName := fmt.Sprintf("%s-%s", DefaultArgoRolloutsResourceName, aggregationType)
+
+	clusterRole := &rbacv1.ClusterRole{ObjectMeta: metav1.ObjectMeta{
+		Name: clusterRoleName,
+	},
+	}
+	Eventually(clusterRole, "30s", "1s").Should(k8s.ExistByName(k8sClient))
+
+	By("Verify that ClusterRole has correct labels.")
+	ensureAggregateLabels(&clusterRole.ObjectMeta, aggregationType)
+
+	By("Verify that ClusterRole has correct policy rules.")
+	Expect(clusterRole.Rules).To(ConsistOf(GetAggregateToViewPolicyRules()))
+}
+
+func validateRoleBinding(k8sClient client.Client, rolloutsManager *rolloutsmanagerv1alpha1.RolloutManager) {
+	binding := &rbacv1.RoleBinding{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      DefaultArgoRolloutsResourceName,
+			Namespace: rolloutsManager.Namespace,
+		},
+	}
+	Eventually(binding, "30s", "1s").Should(k8s.ExistByName(k8sClient))
+
+	By("Verify that RoleBinding has correct labels.")
+	ensureLabels(&binding.ObjectMeta)
+
+	By("Verify that RoleBinding has correct RoleRef.")
+	Expect(binding.RoleRef).To(Equal(rbacv1.RoleRef{
+		APIGroup: rbacv1.GroupName,
+		Kind:     "Role",
+		Name:     DefaultArgoRolloutsResourceName,
+	}))
+
+	By("Verify that RoleBinding has correct Subjects.")
+	Expect(binding.Subjects).To(Equal(
+		[]rbacv1.Subject{
+			{
+				Kind:      rbacv1.ServiceAccountKind,
+				Name:      DefaultArgoRolloutsResourceName,
+				Namespace: rolloutsManager.Namespace,
+			},
+		},
+	))
+}
+
+func validateClusterRoleBinding(k8sClient client.Client, rolloutsManager *rolloutsmanagerv1alpha1.RolloutManager) {
+	clusterRoleBinding := &rbacv1.ClusterRoleBinding{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: DefaultArgoRolloutsResourceName,
+		},
+	}
+	Eventually(clusterRoleBinding, "30s", "1s").Should(k8s.ExistByName(k8sClient))
+
+	By("Verify that ClusterRoleBinding has correct labels.")
+	ensureLabels(&clusterRoleBinding.ObjectMeta)
+
+	By("Verify that ClusterRoleBinding has correct RoleRef.")
+	Expect(clusterRoleBinding.RoleRef).To(Equal(rbacv1.RoleRef{
+		APIGroup: rbacv1.GroupName,
+		Kind:     "ClusterRole",
+		Name:     DefaultArgoRolloutsResourceName,
+	}))
+
+	By("Verify that ClusterRoleBinding has correct Subjects.")
+	Expect(clusterRoleBinding.Subjects).To(Equal(
+		[]rbacv1.Subject{
+			{
+				Kind:      rbacv1.ServiceAccountKind,
+				Name:      DefaultArgoRolloutsResourceName,
+				Namespace: rolloutsManager.Namespace,
+			},
+		},
+	))
+}
+
+func validateService(k8sClient client.Client, rolloutsManager *rolloutsmanagerv1alpha1.RolloutManager) {
+	service := &corev1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      DefaultArgoRolloutsMetricsServiceName,
+			Namespace: rolloutsManager.Namespace,
+		},
+	}
+	Eventually(service, "10s", "1s").Should(k8s.ExistByName(k8sClient))
+
+	By("Verify that Service has correct labels.")
+	Expect(service.Labels["app.kubernetes.io/name"]).To(Equal(DefaultArgoRolloutsMetricsServiceName))
+	Expect(service.Labels["app.kubernetes.io/part-of"]).To(Equal(DefaultArgoRolloutsResourceName))
+	Expect(service.Labels["app.kubernetes.io/component"]).To(Equal("server"))
+
+	By("Verify that Service has correct Ports.")
+	Expect(service.Spec.Ports).To(Equal([]corev1.ServicePort{
+		{
+			Name:       "metrics",
+			Port:       8090,
+			Protocol:   corev1.ProtocolTCP,
+			TargetPort: intstr.FromInt(8090),
+		},
+	}))
+
+	By("Verify that Service has correct Selector.")
+	Expect(service.Spec.Selector).To(Equal(map[string]string{
+		DefaultRolloutsSelectorKey: DefaultArgoRolloutsResourceName,
+	}))
+}
+
+func validateSecret(k8sClient client.Client, rolloutsManager *rolloutsmanagerv1alpha1.RolloutManager) {
+	secret := &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      DefaultRolloutsNotificationSecretName,
+			Namespace: rolloutsManager.Namespace,
+		},
+	}
+	Eventually(secret, "30s", "1s").Should(k8s.ExistByName(k8sClient))
+
+	By("Verify that Secret has correct Type.")
+	Expect(secret.Type).To(Equal(corev1.SecretTypeOpaque))
+}
+
+func validateDeployment(k8sClient client.Client, rolloutsManager *rolloutsmanagerv1alpha1.RolloutManager) {
+	depl := appsv1.Deployment{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      DefaultArgoRolloutsResourceName,
+			Namespace: rolloutsManager.Namespace,
+		},
+	}
+	Eventually(&depl, "10s", "1s").Should(k8s.ExistByName(k8sClient))
+
+	By("Verify that Deployment has correct labels.")
+	ensureLabels(&depl.ObjectMeta)
+
+	By("Verify that Deployment has correct Selector.")
+	Expect(depl.Spec.Selector).To(Equal(&metav1.LabelSelector{
+		MatchLabels: map[string]string{
+			DefaultRolloutsSelectorKey: DefaultArgoRolloutsResourceName,
+		}}))
+
+	By("Verify that Deployment Template has correct Template.")
+	Expect(depl.Spec.Template.Labels).To(Equal(map[string]string{DefaultRolloutsSelectorKey: DefaultArgoRolloutsResourceName}))
+
+	By("Verify that Deployment Template has correct NodeSelector.")
+	Expect(depl.Spec.Template.Spec.NodeSelector).To(Equal(map[string]string{"kubernetes.io/os": "linux"}))
+
+	By("Verify that Deployment Template has correct SecurityContext.")
+	Expect(*depl.Spec.Template.Spec.SecurityContext.RunAsNonRoot).To(BeTrue())
+
+	By("Verify that Deployment Template has correct ServiceAccountName.")
+	Expect(depl.Spec.Template.Spec.ServiceAccountName).To(Equal(DefaultArgoRolloutsResourceName))
+
+	By("Verify that Deployment Template Container is not empty.")
+	Expect(depl.Spec.Template.Spec.Containers[0]).ToNot(Equal(corev1.Container{}))
+}
+
+func ensureLabels(object *metav1.ObjectMeta) {
+	GinkgoHelper()
+	Expect(len(object.Labels)).To(Equal(3))
+	Expect(object.Labels["app.kubernetes.io/name"]).To(Equal(DefaultArgoRolloutsResourceName))
+	Expect(object.Labels["app.kubernetes.io/part-of"]).To(Equal(DefaultArgoRolloutsResourceName))
+	Expect(object.Labels["app.kubernetes.io/component"]).To(Equal(DefaultArgoRolloutsResourceName))
+}
+
+func ensureAggregateLabels(object *metav1.ObjectMeta, aggregationType string) {
+	GinkgoHelper()
+	Expect(len(object.Labels)).To(Equal(4))
+	Expect(object.Labels["app.kubernetes.io/name"]).To(Equal(object.Name))
+	Expect(object.Labels["app.kubernetes.io/part-of"]).To(Equal(DefaultArgoRolloutsResourceName))
+	Expect(object.Labels["app.kubernetes.io/component"]).To(Equal("aggregate-cluster-role"))
+	Expect(object.Labels["rbac.authorization.k8s.io/"+aggregationType]).To(Equal("true"))
+}
diff --git a/controllers/configmap_test.go b/controllers/configmap_test.go
index 2792330..a900def 100644
--- a/controllers/configmap_test.go
+++ b/controllers/configmap_test.go
@@ -100,7 +100,7 @@ var _ = Describe("ConfigMap Test", func() {
 		By("calling reconcileConfigMap")
 		Expect(r.reconcileConfigMap(ctx, a)).To(Succeed())
 
-		By("verifing that when ConfigMap is reconciled, it contains both plugins")
+		By("verifying that when ConfigMap is reconciled, it contains both plugins")
 
 		Expect(fetchObject(ctx, r.Client, a.Namespace, expectedConfigMap.Name, fetchedConfigMap)).To(Succeed())
 		Expect(fetchedConfigMap.Data[TrafficRouterPluginConfigMapKey]).To(ContainSubstring("test/plugin"))
diff --git a/controllers/resources_test.go b/controllers/resources_test.go
index def5f30..32f83c0 100644
--- a/controllers/resources_test.go
+++ b/controllers/resources_test.go
@@ -16,92 +16,177 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 )
 
-var _ = Describe("ReconcileRolloutManager tests", func() {
-	var ctx context.Context
-	var req reconcile.Request
-	var a *v1alpha1.RolloutManager
-	var r *RolloutManagerReconciler
-
-	BeforeEach(func() {
-		ctx = context.Background()
-		a = makeTestRolloutManager()
-		r = makeTestReconciler(a)
-		Expect(createNamespace(r, a.Namespace)).To(Succeed())
-
-		req = reconcile.Request{
-			NamespacedName: types.NamespacedName{
-				Name:      a.Name,
-				Namespace: a.Namespace,
-			},
-		}
-	})
+var _ = Describe("Resource creation and cleanup tests", func() {
 
-	It("Test to verify RolloutsResources", func() {
-		res, err := r.Reconcile(ctx, req)
-		Expect(err).ToNot(HaveOccurred())
-		Expect(res.Requeue).Should(BeFalse(), "reconcile should not requeue request")
-
-		sa := &corev1.ServiceAccount{}
-		Expect(r.Client.Get(ctx, types.NamespacedName{
-			Name:      DefaultArgoRolloutsResourceName,
-			Namespace: testNamespace,
-		}, sa)).To(Succeed(), fmt.Sprintf("failed to find the rollouts serviceaccount: %#v\n", err))
-
-		role := &rbacv1.Role{}
-		Expect(r.Client.Get(ctx, types.NamespacedName{
-			Name:      DefaultArgoRolloutsResourceName,
-			Namespace: testNamespace,
-		}, role)).To(Succeed(), fmt.Sprintf("failed to find the rollouts role: %#v\n", err))
-
-		roleBinding := &rbacv1.RoleBinding{}
-		Expect(r.Client.Get(ctx, types.NamespacedName{
-			Name:      DefaultArgoRolloutsResourceName,
-			Namespace: testNamespace,
-		}, roleBinding)).To(Succeed(), "failed to find the rollouts rolebinding")
-
-		aggregateToAdminClusterRole := &rbacv1.ClusterRole{}
-		Expect(r.Client.Get(ctx, types.NamespacedName{
-			Name: "argo-rollouts-aggregate-to-admin",
-		}, aggregateToAdminClusterRole)).To(Succeed(), fmt.Sprintf("failed to find the aggregateToAdmin ClusterRole: %#v\n", err))
-
-		aggregateToEditClusterRole := &rbacv1.ClusterRole{}
-		Expect(r.Client.Get(ctx, types.NamespacedName{
-			Name: "argo-rollouts-aggregate-to-edit",
-		}, aggregateToEditClusterRole)).To(Succeed(), fmt.Sprintf("failed to find the aggregateToEdit ClusterRole: %#v\n", err))
-
-		aggregateToViewClusterRole := &rbacv1.ClusterRole{}
-		Expect(r.Client.Get(ctx, types.NamespacedName{
-			Name: "argo-rollouts-aggregate-to-view",
-		}, aggregateToViewClusterRole)).To(Succeed(), fmt.Sprintf("failed to find the aggregateToView ClusterRole: %#v\n", err))
-
-		service := &corev1.Service{}
-		Expect(r.Client.Get(ctx, types.NamespacedName{
-			Name:      DefaultArgoRolloutsMetricsServiceName,
-			Namespace: a.Namespace,
-		}, service)).To(Succeed(), fmt.Sprintf("failed to find the rollouts metrics service: %#v\n", err))
-
-		secret := &corev1.Secret{}
-		Expect(r.Client.Get(ctx, types.NamespacedName{
-			Name:      DefaultRolloutsNotificationSecretName,
-			Namespace: a.Namespace,
-		}, secret)).To(Succeed(), fmt.Sprintf("failed to find the rollouts secret: %#v\n", err))
-	})
+	Context("Resource creation test", func() {
+		var (
+			ctx context.Context
+			a   *v1alpha1.RolloutManager
+			r   *RolloutManagerReconciler
+		)
 
-	It("ReconcileAggregate to adminClusterRole test", func() {
-		Expect(r.reconcileRolloutsAggregateToAdminClusterRole(context.Background(), a)).To(Succeed())
-	})
+		BeforeEach(func() {
+			ctx = context.Background()
+			a = makeTestRolloutManager()
+			r = makeTestReconciler(a)
+			err := createNamespace(r, a.Namespace)
+			Expect(err).ToNot(HaveOccurred())
+		})
 
-	It("ReconcileAggregate to EditClusterRole test", func() {
-		Expect(r.reconcileRolloutsAggregateToEditClusterRole(context.Background(), a)).To(Succeed())
-	})
+		It("Test for reconcileRolloutsServiceAccount function", func() {
+			_, err := r.reconcileRolloutsServiceAccount(ctx, a)
+			Expect(err).ToNot(HaveOccurred())
+		})
 
-	It("ReconcileAggregate to ViewClusterRole", func() {
-		Expect(r.reconcileRolloutsAggregateToViewClusterRole(context.Background(), a)).To(Succeed())
-	})
+		It("Test for reconcileRolloutsRole function", func() {
+			role, err := r.reconcileRolloutsRole(ctx, a)
+			Expect(err).ToNot(HaveOccurred())
+
+			By("Modify Rules of Role.")
+			role.Rules[0].Verbs = append(role.Rules[0].Verbs, "test")
+			Expect(r.Client.Update(ctx, role)).To(Succeed())
+
+			By("Reconciler should revert modifications.")
+			role, err = r.reconcileRolloutsRole(ctx, a)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(role.Rules).To(Equal(GetPolicyRules()))
+		})
+
+		It("Test for reconcileRolloutsClusterRole function", func() {
+			clusterRole, err := r.reconcileRolloutsClusterRole(ctx)
+			Expect(err).ToNot(HaveOccurred())
+
+			By("Modify Rules of Role.")
+			clusterRole.Rules[0].Verbs = append(clusterRole.Rules[0].Verbs, "test")
+			Expect(r.Client.Update(ctx, clusterRole)).To(Succeed())
+
+			By("Reconciler should revert modifications.")
+			clusterRole, err = r.reconcileRolloutsClusterRole(ctx)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(clusterRole.Rules).To(Equal(GetPolicyRules()))
+		})
+
+		It("Test for reconcileRolloutsRoleBinding function", func() {
+			sa, err := r.reconcileRolloutsServiceAccount(ctx, a)
+			Expect(err).ToNot(HaveOccurred())
+			role, err := r.reconcileRolloutsRole(ctx, a)
+			Expect(err).ToNot(HaveOccurred())
+
+			Expect(r.reconcileRolloutsRoleBinding(ctx, a, role, sa)).To(Succeed())
+
+			By("Modify Subject of RoleBinding.")
+			rb := &rbacv1.RoleBinding{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      DefaultArgoRolloutsResourceName,
+					Namespace: a.Namespace,
+				},
+			}
+			Expect(fetchObject(ctx, r.Client, a.Namespace, rb.Name, rb)).To(Succeed())
+
+			subTemp := rb.Subjects
+			rb.Subjects = append(rb.Subjects, rbacv1.Subject{Kind: rbacv1.ServiceAccountKind, Name: "test", Namespace: "test"})
+			Expect(r.Client.Update(ctx, rb)).To(Succeed())
+
+			By("Reconciler should revert modifications.")
+			Expect(r.reconcileRolloutsRoleBinding(ctx, a, role, sa)).To(Succeed())
+			Expect(fetchObject(ctx, r.Client, a.Namespace, rb.Name, rb)).To(Succeed())
+			Expect(rb.Subjects).To(Equal(subTemp))
+		})
+
+		It("Test for reconcileRolloutsClusterRoleBinding function", func() {
+			sa, err := r.reconcileRolloutsServiceAccount(ctx, a)
+			Expect(err).ToNot(HaveOccurred())
+			clusterRole, err := r.reconcileRolloutsClusterRole(ctx)
+			Expect(err).ToNot(HaveOccurred())
 
-	Context("RolloutManager Cleaup tests", func() {
-		a = makeTestRolloutManager()
+			Expect(r.reconcileRolloutsClusterRoleBinding(ctx, clusterRole, sa)).To(Succeed())
+
+			By("Modify Subject of ClusterRoleBinding.")
+			crb := &rbacv1.ClusterRoleBinding{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: DefaultArgoRolloutsResourceName,
+				},
+			}
+			Expect(fetchObject(ctx, r.Client, "", crb.Name, crb)).To(Succeed())
+
+			subTemp := crb.Subjects
+			crb.Subjects = append(crb.Subjects, rbacv1.Subject{Kind: rbacv1.ServiceAccountKind, Name: "test", Namespace: "test"})
+			Expect(r.Client.Update(ctx, crb)).To(Succeed())
+
+			By("Reconciler should revert modifications.")
+			Expect(r.reconcileRolloutsClusterRoleBinding(ctx, clusterRole, sa)).To(Succeed())
+			Expect(fetchObject(ctx, r.Client, "", crb.Name, crb)).To(Succeed())
+			Expect(crb.Subjects).To(Equal(subTemp))
+		})
+
+		It("Test for reconcileRolloutsAggregateToAdminClusterRole function", func() {
+			Expect(r.reconcileRolloutsAggregateToAdminClusterRole(ctx)).To(Succeed())
+
+			By("Modify Rules of ClusterRole.")
+			clusterRole := &rbacv1.ClusterRole{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "argo-rollouts-aggregate-to-admin",
+				},
+			}
+			Expect(fetchObject(ctx, r.Client, "", clusterRole.Name, clusterRole)).To(Succeed())
+			clusterRole.Rules[0].Verbs = append(clusterRole.Rules[0].Verbs, "test")
+			Expect(r.Client.Update(ctx, clusterRole)).To(Succeed())
+
+			By("Reconciler should revert modifications.")
+			Expect(r.reconcileRolloutsAggregateToAdminClusterRole(ctx)).To(Succeed())
+			Expect(fetchObject(ctx, r.Client, "", clusterRole.Name, clusterRole)).To(Succeed())
+			Expect(clusterRole.Rules).To(Equal(GetAggregateToAdminPolicyRules()))
+		})
+
+		It("Test for reconcileRolloutsAggregateToEditClusterRole function", func() {
+			Expect(r.reconcileRolloutsAggregateToEditClusterRole(ctx)).To(Succeed())
+
+			By("Modify Rules of ClusterRole.")
+			clusterRole := &rbacv1.ClusterRole{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "argo-rollouts-aggregate-to-edit",
+				},
+			}
+			Expect(fetchObject(ctx, r.Client, "", clusterRole.Name, clusterRole)).To(Succeed())
+			clusterRole.Rules[0].Verbs = append(clusterRole.Rules[0].Verbs, "test")
+			Expect(r.Client.Update(ctx, clusterRole)).To(Succeed())
+
+			By("Reconciler should revert modifications.")
+			Expect(r.reconcileRolloutsAggregateToEditClusterRole(ctx)).To(Succeed())
+			Expect(fetchObject(ctx, r.Client, "", clusterRole.Name, clusterRole)).To(Succeed())
+			Expect(clusterRole.Rules).To(Equal(GetAggregateToEditPolicyRules()))
+		})
+
+		It("Test for reconcileRolloutsAggregateToViewClusterRole function", func() {
+			Expect(r.reconcileRolloutsAggregateToViewClusterRole(ctx)).To(Succeed())
+
+			By("Modify Rules of ClusterRole.")
+			clusterRole := &rbacv1.ClusterRole{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "argo-rollouts-aggregate-to-view",
+				},
+			}
+			Expect(fetchObject(ctx, r.Client, "", clusterRole.Name, clusterRole)).To(Succeed())
+			clusterRole.Rules[0].Verbs = append(clusterRole.Rules[0].Verbs, "test")
+			Expect(r.Client.Update(ctx, clusterRole)).To(Succeed())
+
+			By("Reconciler should revert modifications.")
+			Expect(r.reconcileRolloutsAggregateToViewClusterRole(ctx)).To(Succeed())
+			Expect(fetchObject(ctx, r.Client, "", clusterRole.Name, clusterRole)).To(Succeed())
+			Expect(clusterRole.Rules).To(Equal(GetAggregateToViewPolicyRules()))
+		})
+
+		It("Test for reconcileRolloutsMetricsService function", func() {
+			Expect(r.reconcileRolloutsMetricsService(ctx, a)).To(Succeed())
+		})
+
+		It("Test for reconcileRolloutsSecrets function", func() {
+			Expect(r.reconcileRolloutsSecrets(ctx, a)).To(Succeed())
+		})
+	})
 
+	Context("Resource Cleanup test", func() {
+		a := makeTestRolloutManager()
 		tt := []struct {
 			name     string
 			resource client.Object
diff --git a/controllers/utils_test.go b/controllers/utils_test.go
index 89c574b..01d923c 100644
--- a/controllers/utils_test.go
+++ b/controllers/utils_test.go
@@ -4,14 +4,255 @@ import (
 	"context"
 
 	rolloutsmanagerv1alpha1 "github.com/argoproj-labs/argo-rollouts-manager/api/v1alpha1"
+
+	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes/scheme"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	logger "sigs.k8s.io/controller-runtime/pkg/log"
 )
 
+var _ = Describe("checkForExistingRolloutManager tests", func() {
+
+	var (
+		ctx             context.Context
+		k8sClient       client.WithWatch
+		rolloutsManager rolloutsmanagerv1alpha1.RolloutManager
+	)
+
+	BeforeEach(func() {
+		s := scheme.Scheme
+		Expect(rolloutsmanagerv1alpha1.AddToScheme(s)).To(Succeed())
+
+		ctx = context.Background()
+		log = logger.FromContext(ctx)
+		k8sClient = fake.NewClientBuilder().WithScheme(s).Build()
+
+		rolloutsManager = rolloutsmanagerv1alpha1.RolloutManager{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "test-rm-1",
+				Namespace: "test-ns-1",
+			},
+			Spec: rolloutsmanagerv1alpha1.RolloutManagerSpec{
+				NamespaceScoped: false,
+			},
+		}
+	})
+
+	When("A single cluster-scoped RolloutsManager is created.", func() {
+
+		It("should not return any error, as it is a valid use case.", func() {
+
+			By("Create only one RolloutManager.")
+			Expect(k8sClient.Create(ctx, &rolloutsManager)).To(Succeed())
+
+			By("Verify there is no error returned.")
+			Expect(checkForExistingRolloutManager(ctx, k8sClient, &rolloutsManager)).To(Succeed())
+		})
+	})
+
+	When("Multiple namespace-scoped RolloutsManagers are created.", func() {
+
+		It("should not return error for any of them, because only one cluster-scoped or all namespace-scoped RolloutsManagers are allowed.", func() {
+
+			By("1st RM: Create namespace-scoped RolloutsManager.")
+			rolloutsManager.Spec.NamespaceScoped = true
+			Expect(k8sClient.Create(ctx, &rolloutsManager)).To(Succeed())
+
+			By("1st RM: Verify there is no error returned, as only one RolloutsManager is created yet.")
+			Expect(checkForExistingRolloutManager(ctx, k8sClient, &rolloutsManager)).To(Succeed())
+
+			By("2nd RM: Create namespace-scoped RolloutsManager.")
+			rolloutsManager2 := rolloutsmanagerv1alpha1.RolloutManager{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-rm-2",
+					Namespace: "test-ns-2",
+				},
+				Spec: rolloutsmanagerv1alpha1.RolloutManagerSpec{
+					NamespaceScoped: true,
+				},
+			}
+			Expect(k8sClient.Create(ctx, &rolloutsManager2)).To(Succeed())
+
+			By("2nd RM: Verify there is no error returned, as all namespace-scoped RolloutsManagers are created.")
+			Expect(checkForExistingRolloutManager(ctx, k8sClient, &rolloutsManager2)).To(Succeed())
+
+			By("1st RM: Recheck and it should still work, as all namespace-scoped RolloutsManagers are created.")
+			Expect(k8sClient.Get(ctx, types.NamespacedName{Name: rolloutsManager.Name, Namespace: rolloutsManager.Namespace}, &rolloutsManager)).To(Succeed())
+			Expect(checkForExistingRolloutManager(ctx, k8sClient, &rolloutsManager)).To(Succeed())
+		})
+	})
+
+	When("Multiple cluster-scoped RolloutsManagers are created.", func() {
+
+		It("should return error for all of them, because only one cluster-scoped RolloutsManagers is allowed.", func() {
+
+			By("1st RM: Create cluster-scoped RolloutsManager.")
+			Expect(k8sClient.Create(ctx, &rolloutsManager)).To(Succeed())
+
+			By("1st RM: Verify there is no error returned, as only one RolloutsManager is created yet.")
+			Expect(checkForExistingRolloutManager(ctx, k8sClient, &rolloutsManager)).To(Succeed())
+
+			By("2nd RM: Create another cluster-scoped RolloutsManager.")
+			rolloutsManager2 := rolloutsmanagerv1alpha1.RolloutManager{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-rm-2",
+					Namespace: "test-ns-2",
+				},
+				Spec: rolloutsmanagerv1alpha1.RolloutManagerSpec{
+					NamespaceScoped: false,
+				},
+			}
+			Expect(k8sClient.Create(ctx, &rolloutsManager2)).To(Succeed())
+
+			By("2nd RM: It should return error.")
+			err := checkForExistingRolloutManager(ctx, k8sClient, &rolloutsManager2)
+			Expect(err).To(HaveOccurred())
+			Expect(multipleRolloutManagersExist(err)).To(BeTrue())
+			Expect(rolloutsManager2.Status.Phase).To(Equal(rolloutsmanagerv1alpha1.PhaseFailure))
+			Expect(rolloutsManager2.Status.RolloutController).To(Equal(rolloutsmanagerv1alpha1.PhaseFailure))
+
+			By("1st RM: Recheck 1st RolloutsManager and it should also have error now. since multiple RolloutsManagers are created.")
+			err = checkForExistingRolloutManager(ctx, k8sClient, &rolloutsManager)
+			Expect(err).To(HaveOccurred())
+			Expect(multipleRolloutManagersExist(err)).To(BeTrue())
+			Expect(rolloutsManager.Status.Phase).To(Equal(rolloutsmanagerv1alpha1.PhaseFailure))
+			Expect(rolloutsManager.Status.RolloutController).To(Equal(rolloutsmanagerv1alpha1.PhaseFailure))
+		})
+
+		It("should return error when multiple cluster-scoped RolloutsManagers are created, and when one of them is deleted other one should start working.", func() {
+
+			By("1st RM: Create cluster-scoped RolloutsManager.")
+			Expect(k8sClient.Create(ctx, &rolloutsManager)).To(Succeed())
+
+			By("1st RM: Verify there is no error returned, as only one RolloutsManager is created yet.")
+			Expect(checkForExistingRolloutManager(ctx, k8sClient, &rolloutsManager)).To(Succeed())
+
+			By("2nd RM: Create another cluster-scoped RolloutsManager.")
+			rolloutsManager2 := rolloutsmanagerv1alpha1.RolloutManager{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "test-rm-2",
+					Namespace: "test-ns-2",
+				},
+				Spec: rolloutsmanagerv1alpha1.RolloutManagerSpec{
+					NamespaceScoped: false,
+				},
+			}
+			Expect(k8sClient.Create(ctx, &rolloutsManager2)).To(Succeed())
+
+			By("2nd RM: It should return error.")
+			err := checkForExistingRolloutManager(ctx, k8sClient, &rolloutsManager2)
+			Expect(err).To(HaveOccurred())
+			Expect(multipleRolloutManagersExist(err)).To(BeTrue())
+			Expect(rolloutsManager2.Status.Phase).To(Equal(rolloutsmanagerv1alpha1.PhaseFailure))
+			Expect(rolloutsManager2.Status.RolloutController).To(Equal(rolloutsmanagerv1alpha1.PhaseFailure))
+
+			By("1st RM: Recheck 1st RolloutsManager and it should also have error now. since multiple RolloutsManagers are created.")
+			Expect(k8sClient.Get(ctx, types.NamespacedName{Name: rolloutsManager2.Name, Namespace: rolloutsManager2.Namespace}, &rolloutsManager2)).To(Succeed())
+
+			err = checkForExistingRolloutManager(ctx, k8sClient, &rolloutsManager)
+			Expect(err).To(HaveOccurred())
+			Expect(multipleRolloutManagersExist(err)).To(BeTrue())
+			Expect(rolloutsManager.Status.Phase).To(Equal(rolloutsmanagerv1alpha1.PhaseFailure))
+			Expect(rolloutsManager.Status.RolloutController).To(Equal(rolloutsmanagerv1alpha1.PhaseFailure))
+
+			By("2nd RM: Delete one RolloutsManager.")
+			Expect(k8sClient.Delete(ctx, &rolloutsManager2)).To(Succeed())
+
+			By("1st RM: Verify it works now, as only one RolloutsManager is available.")
+			Expect(checkForExistingRolloutManager(ctx, k8sClient, &rolloutsManager)).To(Succeed())
+		})
+	})
+})
+
+var _ = Describe("validateRolloutsScope tests", func() {
+
+	var (
+		ctx             context.Context
+		k8sClient       client.WithWatch
+		rolloutsManager rolloutsmanagerv1alpha1.RolloutManager
+	)
+
+	BeforeEach(func() {
+		s := scheme.Scheme
+		Expect(rolloutsmanagerv1alpha1.AddToScheme(s)).To(Succeed())
+
+		ctx = context.Background()
+		log = logger.FromContext(ctx)
+		k8sClient = fake.NewClientBuilder().WithScheme(s).Build()
+
+		rolloutsManager = rolloutsmanagerv1alpha1.RolloutManager{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      "test-rm-1",
+				Namespace: "test-ns-1",
+			},
+			Spec: rolloutsmanagerv1alpha1.RolloutManagerSpec{
+				NamespaceScoped: false,
+			},
+		}
+	})
+
+	When("NAMESPACE_SCOPED_ARGO_ROLLOUTS environment variable is set to True.", func() {
+
+		namespaceScopedArgoRolloutsController := true
+
+		It("should return error, if cluster-scoped RolloutManager is created.", func() {
+
+			By("Create cluster-scoped RolloutManager.")
+			Expect(k8sClient.Create(ctx, &rolloutsManager)).To(Succeed())
+
+			By("Verify an error is returned.")
+			err := validateRolloutsScope(ctx, k8sClient, &rolloutsManager, namespaceScopedArgoRolloutsController)
+
+			Expect(err).To(HaveOccurred())
+			Expect(invalidRolloutScope(err)).To(BeTrue())
+		})
+
+		It("should not return any error, if namespace-scoped RolloutManager is created.", func() {
+
+			By("Create namespace-scoped RolloutManager.")
+			rolloutsManager.Spec.NamespaceScoped = true
+			Expect(k8sClient.Create(ctx, &rolloutsManager)).To(Succeed())
+
+			By("Verify there is no error returned.")
+			Expect(validateRolloutsScope(ctx, k8sClient, &rolloutsManager, namespaceScopedArgoRolloutsController)).To(Succeed())
+		})
+	})
+
+	When("NAMESPACE_SCOPED_ARGO_ROLLOUTS environment variable is set to False.", func() {
+
+		namespaceScopedArgoRolloutsController := false
+
+		It("should not return error, if cluster-scoped RolloutManager is created.", func() {
+
+			By("Create cluster-scoped RolloutManager.")
+			Expect(k8sClient.Create(ctx, &rolloutsManager)).To(Succeed())
+
+			By("Verify there is no error returned.")
+			Expect(validateRolloutsScope(ctx, k8sClient, &rolloutsManager, namespaceScopedArgoRolloutsController)).To(Succeed())
+		})
+
+		It("should return error, if namespace-scoped RolloutManager is created.", func() {
+
+			By("Create namespace-scoped RolloutManager.")
+			rolloutsManager.Spec.NamespaceScoped = true
+			Expect(k8sClient.Create(ctx, &rolloutsManager)).To(Succeed())
+
+			By("Verify an error is returned.")
+			err := validateRolloutsScope(ctx, k8sClient, &rolloutsManager, namespaceScopedArgoRolloutsController)
+
+			Expect(err).To(HaveOccurred())
+			Expect(invalidRolloutScope(err)).To(BeTrue())
+		})
+	})
+})
+
 const (
 	testNamespace          = "rollouts"
 	testRolloutManagerName = "rollouts"
diff --git a/hack/run-rollouts-manager-e2e-tests.sh b/hack/run-rollouts-manager-e2e-tests.sh
index 52d430b..728e431 100755
--- a/hack/run-rollouts-manager-e2e-tests.sh
+++ b/hack/run-rollouts-manager-e2e-tests.sh
@@ -10,23 +10,35 @@ cd "$SCRIPTPATH/.."
 set -o pipefail
 set -ex
 
-go test -v -p=1 -timeout=10m -race -count=1 -coverprofile=coverage.out ./tests/e2e
+if [ "$NAMESPACE_SCOPED_ARGO_ROLLOUTS" == "true" ]; then
+
+  go test -v -p=1 -timeout=30m -race -count=1 -coverprofile=coverage.out ./tests/e2e/.  ./tests/e2e/namespace-scoped
+
+else
+
+  go test -v -p=1 -timeout=30m -race -count=1 -coverprofile=coverage.out ./tests/e2e/cluster-scoped
+
+fi
+
 
 set +e
 
 # If the output from the E2E operator is available, then check it for errors
 if [ -f "/tmp/e2e-operator-run.log" ]; then
 
-    # Grep the log for unexpected errors
-    # - Ignore errors that are expected to occur
-
-    ERRORS_FOUND_TEXT=`cat /tmp/e2e-operator-run.log | grep "ERROR" | grep -v "unable to create new content in namespace argo-rollouts because it is being terminated" | grep -v "the object has been modified; please apply your changes to the latest version and try again"`
+  # Wait for the controller to flush to the file, before killing the controller
+  sleep 10
+  killall main
+  sleep 5
 
-    ERRORS_FOUND=`echo $ERRORS_FOUND_TEXT | grep "ERROR" | wc -l`
+  # Grep the log for unexpected errors
+  # - Ignore errors that are expected to occur
 
-    if [ "$ERRORS_FOUND" != "0" ]; then
-        echo "Unexpected errors found: $ERRORS_FOUND_TEXT"
-        exit 1
-    fi
+  UNEXPECTED_ERRORS_FOUND_TEXT=`cat /tmp/e2e-operator-run.log | grep "ERROR" | grep -v "because it is being terminated" | grep -v "the object has been modified; please apply your changes to the latest version and try again" | grep -v "unable to fetch" | grep -v "StorageError"`
+  UNEXPECTED_ERRORS_COUNT=`echo $UNEXPECTED_ERRORS_FOUND_TEXT | grep "ERROR" | wc -l`
 
+  if [ "$UNEXPECTED_ERRORS_COUNT" != "0" ]; then
+      echo "Unexpected errors found: $UNEXPECTED_ERRORS_FOUND_TEXT"
+      exit 1
+  fi
 fi
diff --git a/hack/start-rollouts-manager-for-e2e-tests.sh b/hack/start-rollouts-manager-for-e2e-tests.sh
index 94ab0ee..5dcdd0f 100755
--- a/hack/start-rollouts-manager-for-e2e-tests.sh
+++ b/hack/start-rollouts-manager-for-e2e-tests.sh
@@ -7,10 +7,19 @@ SCRIPTPATH="$(
 
 cd "$SCRIPTPATH/.."
 
+killall main
+
+sleep 5s
+
 rm -f /tmp/e2e-operator-run.log || true
 
 set -o pipefail
 set -ex
 
 make install generate fmt vet
-go run ./main.go 2>&1 | tee /tmp/e2e-operator-run.log
+
+if [ "$RUN_IN_BACKGROUND" == "true" ]; then
+  go run ./main.go 2>&1 | tee /tmp/e2e-operator-run.log &
+else
+  go run ./main.go 2>&1 | tee /tmp/e2e-operator-run.log
+fi
diff --git a/tests/e2e/cluster-scoped/cluster_scoped_rollouts_test.go b/tests/e2e/cluster-scoped/cluster_scoped_rollouts_test.go
new file mode 100644
index 0000000..07c2e6a
--- /dev/null
+++ b/tests/e2e/cluster-scoped/cluster_scoped_rollouts_test.go
@@ -0,0 +1,265 @@
+package e2e
+
+import (
+	"context"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	utils "github.com/argoproj-labs/argo-rollouts-manager/tests/e2e"
+	"github.com/argoproj-labs/argo-rollouts-manager/tests/e2e/fixture"
+	"github.com/argoproj-labs/argo-rollouts-manager/tests/e2e/fixture/k8s"
+	rmFixture "github.com/argoproj-labs/argo-rollouts-manager/tests/e2e/fixture/rolloutmanager"
+
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	rmv1alpha1 "github.com/argoproj-labs/argo-rollouts-manager/api/v1alpha1"
+
+	controllers "github.com/argoproj-labs/argo-rollouts-manager/controllers"
+	rv1alpha1 "github.com/argoproj/argo-rollouts/pkg/apis/rollouts/v1alpha1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+)
+
+var _ = Describe("Cluster-scoped RolloutManager tests", func() {
+
+	Context("Testing cluster-scoped RolloutManager behaviour", func() {
+
+		var (
+			err       error
+			ctx       context.Context
+			k8sClient client.Client
+			scheme    *runtime.Scheme
+		)
+
+		BeforeEach(func() {
+			Expect(fixture.EnsureCleanSlate()).To(Succeed())
+
+			k8sClient, scheme, err = fixture.GetE2ETestKubeClient()
+			Expect(err).ToNot(HaveOccurred())
+
+			err = rv1alpha1.AddToScheme(scheme)
+			Expect(err).ToNot(HaveOccurred())
+
+			ctx = context.Background()
+		})
+
+		/*
+			In this test a cluster-scoped RolloutManager is created in one namespace.
+			Then operator should create required resources (ServiceAccount, ClusterRoles, ClusterRoleBinding, Service, Secret, Deployment) in that namespace.
+			Now when a Rollout CR is created in a different namespace, operator should still be able to reconcile it.
+		*/
+		It("After creating cluster-scoped RolloutManager in one namespace, operator should create appropriate K8s resources and watch argo rollouts CR in another namespace.", func() {
+
+			nsName := "test-ro-ns"
+
+			By("Create cluster-scoped RolloutManager in default namespace.")
+			rolloutsManager, err := utils.CreateRolloutManager(ctx, k8sClient, "test-rollouts-manager-1", fixture.TestE2ENamespace, false)
+			Expect(err).ToNot(HaveOccurred())
+
+			By("Verify that RolloutManager is successfully created.")
+			Eventually(rolloutsManager, "1m", "1s").Should(rmFixture.HavePhase(rmv1alpha1.PhaseAvailable))
+
+			By("Verify that Status.Condition is having success condition.")
+			Eventually(rolloutsManager, "1m", "1s").Should(rmFixture.HaveSuccessCondition())
+
+			By("Verify that expected resources are created.")
+			utils.ValidateArgoRolloutManagerResources(ctx, rolloutsManager, k8sClient, false)
+
+			By("Verify argo Rollouts controller is able to reconcile CR of other namespace.")
+
+			By("Create a different namespace.")
+			Expect(utils.CreateNamespace(ctx, k8sClient, nsName)).To(Succeed())
+
+			By("Create and validate rollouts.")
+			utils.ValidateArgoRolloutsResources(ctx, k8sClient, nsName, 31000, 32000)
+		})
+
+		/*
+			In this test a cluster-scoped RolloutManager is created in a namespace.
+			Then operator should create required resources (ServiceAccount, ClusterRoles, ClusterRoleBinding, Service, Secret, Deployment) in one namespace.
+			Now when a Rollouts CR is created in multiple namespaces, operator should still be able to reconcile all of them.
+		*/
+		It("After creating cluster-scoped RolloutManager in a namespace, operator should create appropriate K8s resources and watch argo rollouts CR in multiple namespace.", func() {
+
+			nsName1, nsName2, nsName3 := "rom-ns-1", "ro-ns-1", "ro-ns-2"
+
+			By("Create a namespace for RolloutManager.")
+			Expect(utils.CreateNamespace(ctx, k8sClient, nsName1)).To(Succeed())
+
+			By("Create cluster-scoped RolloutManager.")
+			rolloutsManager, err := utils.CreateRolloutManager(ctx, k8sClient, "test-rollouts-manager-1", nsName1, false)
+			Expect(err).ToNot(HaveOccurred())
+
+			By("Verify that RolloutManager is successfully created.")
+			Eventually(rolloutsManager, "1m", "1s").Should(rmFixture.HavePhase(rmv1alpha1.PhaseAvailable))
+
+			By("Verify that Status.Condition is having success condition.")
+			Eventually(rolloutsManager, "1m", "1s").Should(rmFixture.HaveSuccessCondition())
+
+			By("Verify that expected resources are created.")
+			utils.ValidateArgoRolloutManagerResources(ctx, rolloutsManager, k8sClient, false)
+
+			By("Verify argo Rollouts controller is able to reconcile CR of multiple namespaces.")
+
+			By("1st Namespace: Create a different namespace for Rollout.")
+			Expect(utils.CreateNamespace(ctx, k8sClient, nsName2)).To(Succeed())
+
+			By("1st Namespace: Create and validate Rollout in 1st namespace.")
+			utils.ValidateArgoRolloutsResources(ctx, k8sClient, nsName2, 31000, 32000)
+
+			By("2nd Namespace: Create a another namespace for 2nd Rollout.")
+			Expect(utils.CreateNamespace(ctx, k8sClient, nsName3)).To(Succeed())
+
+			By("2nd Namespace: Create and validate Rollout in 2nd namespace.")
+			utils.ValidateArgoRolloutsResources(ctx, k8sClient, nsName3, 31001, 32002)
+		})
+
+		/*
+			In this test a cluster-scoped RolloutManager is created in a namespace.
+			Then operator should create required resources (ServiceAccount, ClusterRoles, ClusterRoleBinding, Service, Secret, Deployment) in namespace.
+			Now when a namespace-scoped RolloutManager is created in 2nd namespace, it should not be accepted by operator,
+			since NAMESPACE_SCOPED_ARGO_ROLLOUTS is set to false, it means user can only create cluster-scoped RolloutManagers,
+			but Rollouts controller of 1st namespace should still be able to reconcile Rollout CR of 2nd namespace and failed RolloutManager of 2nd namespace should not cause any issues.
+			When 1st cluster-scoped RolloutManager is reconciled again it should still work, and should be able to reconcile Rollout CR created in any namespace.
+		*/
+		It("Should allow cluster-scoped RolloutManager, but not namespace-scoped.", func() {
+
+			nsName1, nsName2 := "test-rom-ns-1", "test-ro-ns-2"
+
+			By("1st RM: Create cluster-scoped RolloutManager in a namespace.")
+			rolloutsManagerCl, err := utils.CreateRolloutManager(ctx, k8sClient, "test-rollouts-manager-1", fixture.TestE2ENamespace, false)
+			Expect(err).ToNot(HaveOccurred())
+
+			By("1st RM: Verify that RolloutManager is successfully created.")
+			Eventually(rolloutsManagerCl, "1m", "1s").Should(rmFixture.HavePhase(rmv1alpha1.PhaseAvailable))
+
+			By("1st RM: Verify that Status.Condition is having success condition.")
+			Eventually(rolloutsManagerCl, "1m", "1s").Should(rmFixture.HaveSuccessCondition())
+
+			By("1st RM: Create and validate Rollout in 1st namespace.")
+			utils.ValidateArgoRolloutsResources(ctx, k8sClient, fixture.TestE2ENamespace, 31000, 32000)
+
+			By("2nd RM: Create 2nd namespace.")
+			Expect(utils.CreateNamespace(ctx, k8sClient, nsName1)).To(Succeed())
+
+			By("2nd RM: Create namespace-scoped RolloutManager in 2nd namespace.")
+			rolloutsManagerNs, err := utils.CreateRolloutManager(ctx, k8sClient, "test-rollouts-manager-2", nsName1, true)
+			Expect(err).ToNot(HaveOccurred())
+
+			By("2nd RM: Verify that RolloutManager is not working.")
+			Eventually(rolloutsManagerNs, "1m", "1s").Should(rmFixture.HavePhase(rmv1alpha1.PhaseFailure))
+
+			By("2nd RM: Verify that Status.Condition is having error message.")
+			Eventually(rolloutsManagerNs, "1m", "1s").Should(rmFixture.HaveCondition(
+				metav1.Condition{
+					Type:    rmv1alpha1.RolloutManagerConditionType,
+					Status:  metav1.ConditionFalse,
+					Reason:  rmv1alpha1.RolloutManagerReasonInvalidScoped,
+					Message: controllers.UnsupportedRolloutManagerNamespaceScoped,
+				}))
+
+			By("2nd RM: Create and validate Rollout in 2nd namespace.")
+			utils.ValidateArgoRolloutsResources(ctx, k8sClient, nsName1, 31001, 32002)
+
+			By("1st RM: Update cluster-scoped RolloutManager, after reconciliation it should still work.")
+			err = k8s.UpdateWithoutConflict(ctx, &rolloutsManagerCl, k8sClient, func(obj client.Object) {
+				goObj, ok := obj.(*rmv1alpha1.RolloutManager)
+				Expect(ok).To(BeTrue())
+
+				goObj.Spec.Env = append(goObj.Spec.Env, corev1.EnvVar{Name: "test-name", Value: "test-value"})
+			})
+			Expect(err).ToNot(HaveOccurred())
+
+			By("1st RM: Verify that cluster-scoped RolloutManager is still working.")
+			Eventually(rolloutsManagerCl, "1m", "1s").Should(rmFixture.HavePhase(rmv1alpha1.PhaseAvailable))
+
+			By("1st RM: Verify that Status.Condition is not having error message.")
+			Eventually(rolloutsManagerCl, "3m", "1s").Should(rmFixture.HaveSuccessCondition())
+
+			By("3rd RM: Create 3rd namespace.")
+			Expect(utils.CreateNamespace(ctx, k8sClient, nsName2)).To(Succeed())
+
+			By("2nd RM: Create and validate Rollout in 3rd namespace.")
+			utils.ValidateArgoRolloutsResources(ctx, k8sClient, nsName2, 31002, 32003)
+		})
+
+		/*
+			In this test a cluster-scoped RolloutManager is created in a namespace.
+			Then operator should create required resources (ServiceAccount, ClusterRoles, ClusterRoleBinding, Service, Secret, Deployment) in namespace.
+			Now when another cluster-scoped RolloutManager is created in 2nd namespace, it should not be accepted by operator,
+			since there in an existing RolloutManager watching entire cluster,
+			but Rollouts controller of 1st namespace should still be able to reconcile Rollout CR of 2nd namespace and failed RolloutManager of 2nd namespace should not cause any issues.
+			When cluster-scoped RolloutManager is reconciled again it should also have error, because only one cluster-scoped or all namespace-scoped RolloutManagers are supported.
+			but Rollouts controller deployed in 1st namespace should still work and reconcile Rollout CR created in any namespace,
+			it means even both cluster-scoped RolloutManagers are having error in status, but Rollout controller created before error occurred should still work normally.
+		*/
+		It("After creating cluster-scoped RolloutManager in a namespace, another cluster-scoped RolloutManager should not be allowed.", func() {
+
+			nsName1, nsName2 := "test-rom-ns-1", "test-ro-ns-2"
+
+			By("1st RM: Create cluster-scoped RolloutManager in 1st namespace.")
+			rolloutsManagerCl, err := utils.CreateRolloutManager(ctx, k8sClient, "test-rollouts-manager-1", fixture.TestE2ENamespace, false)
+			Expect(err).ToNot(HaveOccurred())
+
+			By("1st RM: Verify that RolloutManager is successfully created.")
+			Eventually(rolloutsManagerCl, "1m", "1s").Should(rmFixture.HavePhase(rmv1alpha1.PhaseAvailable))
+
+			By("1st RM: Verify that Status.Condition is having success condition.")
+			Eventually(rolloutsManagerCl, "1m", "1s").Should(rmFixture.HaveSuccessCondition())
+
+			By("1st RM: Create and validate Rollout in 1st namespace.")
+			utils.ValidateArgoRolloutsResources(ctx, k8sClient, fixture.TestE2ENamespace, 31000, 32000)
+
+			By("2nd RM: Create 2nd namespace.")
+			Expect(utils.CreateNamespace(ctx, k8sClient, nsName1)).To(Succeed())
+
+			By("2nd RM: Create cluster-scoped RolloutManager in 2nd namespace.")
+			rolloutsManagerCl2, err := utils.CreateRolloutManager(ctx, k8sClient, "test-rollouts-manager-2", nsName1, false)
+			Expect(err).ToNot(HaveOccurred())
+
+			By("2nd RM: Verify that RolloutManager is not working.")
+			Eventually(rolloutsManagerCl2, "1m", "1s").Should(rmFixture.HavePhase(rmv1alpha1.PhaseFailure))
+
+			By("2nd RM: Verify that Status.Condition is having error message.")
+			Eventually(rolloutsManagerCl2, "1m", "1s").Should(rmFixture.HaveCondition(
+				metav1.Condition{
+					Type:    rmv1alpha1.RolloutManagerConditionType,
+					Status:  metav1.ConditionFalse,
+					Reason:  rmv1alpha1.RolloutManagerReasonMultipleClusterScopedRolloutManager,
+					Message: controllers.UnsupportedRolloutManagerConfiguration,
+				}))
+
+			By("2nd RM: Create and validate Rollout in 2nd namespace.")
+			utils.ValidateArgoRolloutsResources(ctx, k8sClient, nsName1, 31001, 32001)
+
+			By("1st RM: Update first RolloutManager, after reconciliation it should also stop working.")
+			err = k8s.UpdateWithoutConflict(ctx, &rolloutsManagerCl, k8sClient, func(obj client.Object) {
+				goObj, ok := obj.(*rmv1alpha1.RolloutManager)
+				Expect(ok).To(BeTrue())
+
+				goObj.Spec.Env = append(goObj.Spec.Env, corev1.EnvVar{Name: "test-name", Value: "test-value"})
+			})
+			Expect(err).ToNot(HaveOccurred())
+
+			By("1st RM: Verify that now first RolloutManager is also not working.")
+			Eventually(rolloutsManagerCl, "1m", "1s").Should(rmFixture.HavePhase(rmv1alpha1.PhaseFailure))
+
+			By("1st RM: Verify that Status.Condition is now having error message.")
+			Eventually(rolloutsManagerCl, "3m", "1s").Should(rmFixture.HaveCondition(
+				metav1.Condition{
+					Type:    rmv1alpha1.RolloutManagerConditionType,
+					Status:  metav1.ConditionFalse,
+					Reason:  rmv1alpha1.RolloutManagerReasonMultipleClusterScopedRolloutManager,
+					Message: controllers.UnsupportedRolloutManagerConfiguration,
+				}))
+
+			By("1st RM: Create 3rd namespace.")
+			Expect(utils.CreateNamespace(ctx, k8sClient, nsName2)).To(Succeed())
+
+			By("1st RM: Create and validate Rollout in 3rd namespace.")
+			utils.ValidateArgoRolloutsResources(ctx, k8sClient, nsName2, 31002, 32002)
+		})
+	})
+})
diff --git a/tests/e2e/cluster-scoped/cluster_scoped_suite_test.go b/tests/e2e/cluster-scoped/cluster_scoped_suite_test.go
new file mode 100644
index 0000000..4af4e3a
--- /dev/null
+++ b/tests/e2e/cluster-scoped/cluster_scoped_suite_test.go
@@ -0,0 +1,24 @@
+package e2e
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"go.uber.org/zap/zapcore"
+
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+)
+
+var _ = BeforeSuite(func() {
+	logf.SetLogger(zap.New(zap.WriteTo(GinkgoWriter), zap.UseDevMode(true), zap.Level(zapcore.DebugLevel)))
+})
+
+func TestClusterScoped(t *testing.T) {
+	suiteConfig, _ := GinkgoConfiguration()
+
+	RegisterFailHandler(Fail)
+
+	RunSpecs(t, "Cluster-scoped Suite", suiteConfig)
+}
diff --git a/tests/e2e/fixture/fixture.go b/tests/e2e/fixture/fixture.go
index 1ce2eaa..b2fd309 100644
--- a/tests/e2e/fixture/fixture.go
+++ b/tests/e2e/fixture/fixture.go
@@ -8,9 +8,10 @@ import (
 	"time"
 
 	. "github.com/onsi/ginkgo/v2"
-	// . "github.com/onsi/gomega"
 
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/selection"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
@@ -27,15 +28,19 @@ import (
 
 const (
 	TestE2ENamespace = "argo-rollouts"
+	LabelsKey        = "app"
+	LabelsValue      = "test-argo-app"
 )
 
+var NamespaceLabels = map[string]string{LabelsKey: LabelsValue}
+
 type Cleaner struct {
 	cxt       context.Context
 	k8sClient client.Client
 }
 
-func NewCleaner() (*Cleaner, error) {
-	k8sClient, err := GetE2ETestKubeClient()
+func newCleaner() (*Cleaner, error) {
+	k8sClient, _, err := GetE2ETestKubeClient()
 	if err != nil {
 		return nil, err
 	}
@@ -47,17 +52,24 @@ func NewCleaner() (*Cleaner, error) {
 }
 
 func EnsureCleanSlate() error {
-	cleaner, err := NewCleaner()
+	cleaner, err := newCleaner()
+	if err != nil {
+		return err
+	}
+
+	// ensure namespaces created during test are deleted
+	err = cleaner.ensureTestNamespaceDeleted()
 	if err != nil {
 		return err
 	}
 
-	err = cleaner.EnsureDestinationNamespaceExists(TestE2ENamespace)
+	// create default namespace used for Rollouts controller
+	err = cleaner.ensureRlloutNamespaceExists(TestE2ENamespace)
 	if err != nil {
 		return err
 	}
 
-	err = cleaner.DeleteRolloutsClusterRoles()
+	err = cleaner.deleteRolloutsClusterRoles()
 	if err != nil {
 		return err
 	}
@@ -65,8 +77,8 @@ func EnsureCleanSlate() error {
 	return nil
 }
 
-func (cleaner *Cleaner) EnsureDestinationNamespaceExists(namespaceParam string) error {
-	if err := cleaner.DeleteNamespace(namespaceParam); err != nil {
+func (cleaner *Cleaner) ensureRlloutNamespaceExists(namespaceParam string) error {
+	if err := cleaner.deleteNamespace(namespaceParam); err != nil {
 		return fmt.Errorf("unable to delete namespace '%s': %w", namespaceParam, err)
 	}
 
@@ -81,7 +93,7 @@ func (cleaner *Cleaner) EnsureDestinationNamespaceExists(namespaceParam string)
 	return nil
 }
 
-func (cleaner *Cleaner) DeleteRolloutsClusterRoles() error {
+func (cleaner *Cleaner) deleteRolloutsClusterRoles() error {
 	crList := rbacv1.ClusterRoleList{}
 	if err := cleaner.k8sClient.List(cleaner.cxt, &crList, &client.ListOptions{}); err != nil {
 		return err
@@ -100,8 +112,8 @@ func (cleaner *Cleaner) DeleteRolloutsClusterRoles() error {
 	return nil
 }
 
-// DeleteNamespace deletes a namespace, and waits for it to be reported as deleted.
-func (cleaner *Cleaner) DeleteNamespace(namespaceParam string) error {
+// deleteNamespace deletes a namespace, and waits for it to be reported as deleted.
+func (cleaner *Cleaner) deleteNamespace(namespaceParam string) error {
 
 	// Delete the namespace:
 	// - Issue a request to Delete the namespace
@@ -138,55 +150,55 @@ func (cleaner *Cleaner) DeleteNamespace(namespaceParam string) error {
 	return nil
 }
 
-func GetE2ETestKubeClient() (client.Client, error) {
-	config, err := GetSystemKubeConfig()
+func GetE2ETestKubeClient() (client.Client, *runtime.Scheme, error) {
+	config, err := getSystemKubeConfig()
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
-	k8sClient, err := GetKubeClient(config)
+	k8sClient, scheme, err := getKubeClient(config)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
-	return k8sClient, nil
+	return k8sClient, scheme, nil
 }
 
-// GetKubeClient returns a controller-runtime Client for accessing K8s API resources used by the controller.
-func GetKubeClient(config *rest.Config) (client.Client, error) {
+// getKubeClient returns a controller-runtime Client for accessing K8s API resources used by the controller.
+func getKubeClient(config *rest.Config) (client.Client, *runtime.Scheme, error) {
 
 	scheme := runtime.NewScheme()
 
 	if err := rolloutsmanagerv1alpha1.AddToScheme(scheme); err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
 	if err := corev1.AddToScheme(scheme); err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
 	if err := apps.AddToScheme(scheme); err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 	if err := rbacv1.AddToScheme(scheme); err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
 	if err := admissionv1.AddToScheme(scheme); err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
 	k8sClient, err := client.New(config, client.Options{Scheme: scheme})
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
-	return k8sClient, nil
+	return k8sClient, scheme, nil
 
 }
 
 // Retrieve the system-level Kubernetes config (e.g. ~/.kube/config or service account config from volume)
-func GetSystemKubeConfig() (*rest.Config, error) {
+func getSystemKubeConfig() (*rest.Config, error) {
 
 	overrides := clientcmd.ConfigOverrides{}
 
@@ -199,3 +211,39 @@ func GetSystemKubeConfig() (*rest.Config, error) {
 	}
 	return restConfig, nil
 }
+
+// Delete all namespaces having a specific label used to identify namespaces that are created by e2e tests.
+func (cleaner *Cleaner) ensureTestNamespaceDeleted() error {
+
+	// fetch all namespaces having given label
+	nsList, err := listE2ETestNamespaces(cleaner.cxt, cleaner.k8sClient)
+	if err != nil {
+		return fmt.Errorf("unable to delete test namespace: %w", err)
+	}
+
+	// delete selected namespaces
+	for _, namespace := range nsList.Items {
+		if err := cleaner.deleteNamespace(namespace.Name); err != nil {
+			return fmt.Errorf("unable to delete namespace '%s': %w", namespace.Name, err)
+		}
+	}
+	return nil
+}
+
+// Retrieve list of namespaces having a specific label used to identify namespaces that are created by e2e tests.
+func listE2ETestNamespaces(ctx context.Context, k8sClient client.Client) (corev1.NamespaceList, error) {
+	nsList := corev1.NamespaceList{}
+
+	// set e2e label
+	req, err := labels.NewRequirement(LabelsKey, selection.Equals, []string{LabelsValue})
+	if err != nil {
+		return nsList, fmt.Errorf("unable to set labels while fetching list of test namespace: %w", err)
+	}
+
+	// fetch all namespaces having given label
+	err = k8sClient.List(ctx, &nsList, &client.ListOptions{LabelSelector: labels.NewSelector().Add(*req)})
+	if err != nil {
+		return nsList, fmt.Errorf("unable to fetch list of test namespace: %w", err)
+	}
+	return nsList, nil
+}
diff --git a/tests/e2e/fixture/k8s/fixture.go b/tests/e2e/fixture/k8s/fixture.go
index f84f957..400b83c 100644
--- a/tests/e2e/fixture/k8s/fixture.go
+++ b/tests/e2e/fixture/k8s/fixture.go
@@ -6,6 +6,7 @@ import (
 
 	. "github.com/onsi/gomega"
 	matcher "github.com/onsi/gomega/types"
+	"k8s.io/client-go/util/retry"
 
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
@@ -25,3 +26,20 @@ func ExistByName(k8sClient client.Client) matcher.GomegaMatcher {
 		return err == nil
 	}, BeTrue())
 }
+
+// UpdateWithoutConflict will keep trying to update object until it succeeds, or times out.
+func UpdateWithoutConflict(ctx context.Context, obj client.Object, k8sClient client.Client, modify func(client.Object)) error {
+	err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		// Retrieve the latest version of the object
+		err := k8sClient.Get(context.Background(), client.ObjectKeyFromObject(obj), obj)
+		if err != nil {
+			return err
+		}
+		modify(obj)
+
+		// Attempt to update the object
+		return k8sClient.Update(ctx, obj)
+	})
+
+	return err
+}
diff --git a/tests/e2e/fixture/rolloutmanager/fixture.go b/tests/e2e/fixture/rolloutmanager/fixture.go
index 87c59a1..51ea866 100644
--- a/tests/e2e/fixture/rolloutmanager/fixture.go
+++ b/tests/e2e/fixture/rolloutmanager/fixture.go
@@ -6,6 +6,7 @@ import (
 
 	. "github.com/onsi/gomega"
 	matcher "github.com/onsi/gomega/types"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	rolloutsmanagerv1alpha1 "github.com/argoproj-labs/argo-rollouts-manager/api/v1alpha1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -14,11 +15,11 @@ import (
 )
 
 // This is intentionally NOT exported, for now. Create another function in this file/package that calls this function, and export that.
-func expectedCondition(f func(app rolloutsmanagerv1alpha1.RolloutManager) bool) matcher.GomegaMatcher {
+func fetchRolloutManager(f func(app rolloutsmanagerv1alpha1.RolloutManager) bool) matcher.GomegaMatcher {
 
 	return WithTransform(func(app rolloutsmanagerv1alpha1.RolloutManager) bool {
 
-		k8sClient, err := fixture.GetE2ETestKubeClient()
+		k8sClient, _, err := fixture.GetE2ETestKubeClient()
 		if err != nil {
 			fmt.Println(err)
 			return false
@@ -37,7 +38,57 @@ func expectedCondition(f func(app rolloutsmanagerv1alpha1.RolloutManager) bool)
 }
 
 func HavePhase(phase rolloutsmanagerv1alpha1.RolloutControllerPhase) matcher.GomegaMatcher {
-	return expectedCondition(func(app rolloutsmanagerv1alpha1.RolloutManager) bool {
+	return fetchRolloutManager(func(app rolloutsmanagerv1alpha1.RolloutManager) bool {
+		fmt.Println("HavePhase:", "expected: ", phase, "actual: ", app.Status.Phase)
 		return app.Status.Phase == phase
 	})
 }
+
+func HaveCondition(expected metav1.Condition) matcher.GomegaMatcher {
+	return fetchRolloutManager(func(app rolloutsmanagerv1alpha1.RolloutManager) bool {
+
+		if len(app.Status.Conditions) == 0 {
+			fmt.Println("HaveCondition: Conditions is nil")
+			return false
+		}
+
+		for _, condition := range app.Status.Conditions {
+			if condition.Type == expected.Type {
+				fmt.Println("HaveCondition:", "expected: ", expected, "actual: ", condition)
+				return condition.Type == expected.Type &&
+					condition.Status == expected.Status &&
+					condition.Reason == expected.Reason &&
+					condition.Message == expected.Message
+			}
+		}
+		return false
+	})
+}
+
+func HaveSuccessCondition() matcher.GomegaMatcher {
+	return fetchRolloutManager(func(app rolloutsmanagerv1alpha1.RolloutManager) bool {
+
+		if len(app.Status.Conditions) == 0 {
+			fmt.Println("HaveSuccessCondition: Conditions is nil")
+			return false
+		}
+
+		expected := metav1.Condition{
+			Type:    rolloutsmanagerv1alpha1.RolloutManagerConditionType,
+			Status:  metav1.ConditionTrue,
+			Reason:  rolloutsmanagerv1alpha1.RolloutManagerReasonSuccess,
+			Message: "",
+		}
+
+		for _, condition := range app.Status.Conditions {
+			if condition.Type == rolloutsmanagerv1alpha1.RolloutManagerConditionType {
+				fmt.Println("HaveSuccessCondition:", "expected: ", expected, "actual: ", condition)
+				return condition.Type == expected.Type &&
+					condition.Status == expected.Status &&
+					condition.Reason == expected.Reason &&
+					condition.Message == expected.Message
+			}
+		}
+		return false
+	})
+}
diff --git a/tests/e2e/namespace-scoped/namespace_scoped_rollouts_test.go b/tests/e2e/namespace-scoped/namespace_scoped_rollouts_test.go
new file mode 100644
index 0000000..297cf22
--- /dev/null
+++ b/tests/e2e/namespace-scoped/namespace_scoped_rollouts_test.go
@@ -0,0 +1,302 @@
+package e2e
+
+import (
+	"context"
+	"reflect"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	utils "github.com/argoproj-labs/argo-rollouts-manager/tests/e2e"
+	"github.com/argoproj-labs/argo-rollouts-manager/tests/e2e/fixture"
+	"github.com/argoproj-labs/argo-rollouts-manager/tests/e2e/fixture/k8s"
+	rmFixture "github.com/argoproj-labs/argo-rollouts-manager/tests/e2e/fixture/rolloutmanager"
+
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	rmv1alpha1 "github.com/argoproj-labs/argo-rollouts-manager/api/v1alpha1"
+
+	controllers "github.com/argoproj-labs/argo-rollouts-manager/controllers"
+	rv1alpha1 "github.com/argoproj/argo-rollouts/pkg/apis/rollouts/v1alpha1"
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+)
+
+var _ = Describe("Namespace-scoped RolloutManager tests", func() {
+
+	Context("Testing namespace-scoped RolloutManager behaviour", func() {
+
+		var (
+			err       error
+			ctx       context.Context
+			k8sClient client.Client
+			scheme    *runtime.Scheme
+		)
+
+		BeforeEach(func() {
+			Expect(fixture.EnsureCleanSlate()).To(Succeed())
+
+			k8sClient, scheme, err = fixture.GetE2ETestKubeClient()
+			Expect(err).ToNot(HaveOccurred())
+
+			err = rv1alpha1.AddToScheme(scheme)
+			Expect(err).ToNot(HaveOccurred())
+
+			ctx = context.Background()
+		})
+
+		/*
+			In this test a namespace-scoped RolloutManager is created in a namespace.
+			Then operator should create required resources (ServiceAccount, Roles, RoleBinding, Service, Secret, Deployment) in namespace.
+			Now when a Rollouts CR is created in same namespace, operator should be able to reconcile it.
+		*/
+		It("After creating namespace-scoped RolloutManager in a namespace, operator should create appropriate K8s resources and watch argo rollouts CR in same namespace.", func() {
+
+			nsName := "test-rom-ns"
+
+			By("Create a namespace for RolloutManager.")
+			Expect(utils.CreateNamespace(ctx, k8sClient, nsName)).To(Succeed())
+
+			By("Create namespace-scoped RolloutManager in same namespace.")
+			rolloutsManager, err := utils.CreateRolloutManager(ctx, k8sClient, "test-rollouts-manager-1", nsName, true)
+			Expect(err).ToNot(HaveOccurred())
+
+			By("Verify that RolloutManager is successfully created.")
+			Eventually(rolloutsManager, "2m", "1s").Should(rmFixture.HavePhase(rmv1alpha1.PhaseAvailable))
+
+			By("Verify that Status.Condition is having success condition.")
+			Eventually(rolloutsManager, "2m", "1s").Should(rmFixture.HaveSuccessCondition())
+
+			By("Verify that expected resources are created.")
+			utils.ValidateArgoRolloutManagerResources(ctx, rolloutsManager, k8sClient, true)
+
+			By("Verify argo Rollouts controller is able to reconcile CR.")
+
+			By("Create and validate rollouts.")
+			utils.ValidateArgoRolloutsResources(ctx, k8sClient, nsName, 31000, 32000)
+		})
+
+		/*
+			In this test namespace-scoped RolloutManagers are created in multiple namespaces.
+			Then operator should create required resources (ServiceAccount, Roles, RoleBinding, Service, Secret, Deployment) in each namespace.
+			Now when a Rollouts CR is created in each namespace, Rollouts controller running in respective should be able to reconcile them.
+			When RolloutManagers is deleted in any namespace it also deletes Rollouts controller from that namespace.
+			After that if any Rollout CR is created in that namespace, it should not be reconciled.
+		*/
+		It("After creating namespace-scoped RolloutManager in a namespace, another namespace-scoped RolloutManager in different namespace should also work.", func() {
+
+			nsName1 := "test-rom-ns"
+
+			By("1st RM: Create namespace-scoped RolloutManager in 1st namespace.")
+			rolloutsManagerNs1, err := utils.CreateRolloutManager(ctx, k8sClient, "test-rollouts-manager-1", fixture.TestE2ENamespace, true)
+			Expect(err).ToNot(HaveOccurred())
+
+			By("1st RM: Verify that RolloutManager is successfully created.")
+			Eventually(rolloutsManagerNs1, "2m", "1s").Should(rmFixture.HavePhase(rmv1alpha1.PhaseAvailable))
+
+			By("1st RM: Verify that Status.Condition is having success condition.")
+			Eventually(rolloutsManagerNs1, "2m", "1s").Should(rmFixture.HaveSuccessCondition())
+
+			By("1st RM: Verify argo Rollouts controller of 1st namespace is able to reconcile CR created in 1st namespace.")
+
+			By("1st RM: Create and validate rollouts.")
+			utils.ValidateArgoRolloutsResources(ctx, k8sClient, fixture.TestE2ENamespace, 31000, 32000)
+
+			By("2nd RM: Create 2nd namespace.")
+			Expect(utils.CreateNamespace(ctx, k8sClient, nsName1)).To(Succeed())
+
+			By("2nd RM: Create 2nd namespace-scoped RolloutManager in 2nd namespace.")
+			rolloutsManagerNs2, err := utils.CreateRolloutManager(ctx, k8sClient, "test-rollouts-manager-2", nsName1, true)
+			Expect(err).ToNot(HaveOccurred())
+
+			By("2nd RM: Verify that RolloutManager of 2nd namespace is successfully created.")
+			Eventually(rolloutsManagerNs2, "1m", "1s").Should(rmFixture.HavePhase(rmv1alpha1.PhaseAvailable))
+
+			By("2nd RM: Verify that Status.Condition is having success condition.")
+			Eventually(rolloutsManagerNs2, "2m", "1s").Should(rmFixture.HaveSuccessCondition())
+
+			By("2nd RM: Verify argo Rollouts controller of 2nd namespace is able to reconcile CR created in 2nd namespace.")
+
+			By("2nd RM: Create and validate rollouts in 2nd namespace.")
+			utils.ValidateArgoRolloutsResources(ctx, k8sClient, nsName1, 31001, 32001)
+
+			By("1st RM: Update 1st RolloutManager, after reconciliation it should still work.")
+			err = k8s.UpdateWithoutConflict(ctx, &rolloutsManagerNs1, k8sClient, func(obj client.Object) {
+				goObj, ok := obj.(*rmv1alpha1.RolloutManager)
+				Expect(ok).To(BeTrue())
+
+				goObj.Spec.Env = append(goObj.Spec.Env, corev1.EnvVar{Name: "test-name", Value: "test-value"})
+			})
+			Expect(err).ToNot(HaveOccurred())
+
+			By("1st RM: Verify that now 1st RolloutManager is still working.")
+			Eventually(rolloutsManagerNs1, "2m", "1s").Should(rmFixture.HavePhase(rmv1alpha1.PhaseAvailable))
+
+			By("2nd RM: Delete 2nd RolloutManager and ensure 2nd Rollouts controller is also deleted.")
+			Expect(k8sClient.Delete(ctx, &rolloutsManagerNs2)).To(Succeed())
+			Eventually(&appsv1.Deployment{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      controllers.DefaultArgoRolloutsResourceName,
+					Namespace: nsName1,
+				},
+			}, "3m", "2s").ShouldNot(k8s.ExistByName(k8sClient))
+
+			By("2nd RM: Delete 2nd Rollout CR and ensure it is not recreated.")
+			rollout2 := rv1alpha1.Rollout{
+				ObjectMeta: metav1.ObjectMeta{Name: utils.RolloutsName, Namespace: nsName1},
+			}
+			Expect(k8sClient.Delete(ctx, &rollout2)).To(Succeed())
+			Eventually(func() error {
+				return k8sClient.Get(ctx, client.ObjectKeyFromObject(&rollout2), &rollout2)
+			}, "1m", "1s").ShouldNot(BeNil())
+
+			By("2nd RM: Create 3rd Rollout in 2nd namespace and ensure it is not reconciled, since RolloutsManager is deleted from 2nd namespace.")
+			rollout3, err := utils.CreateArgoRollout(ctx, k8sClient, "simple-rollout-1", nsName1, utils.RolloutsActiveServiceName, utils.RolloutsPreviewServiceName)
+			Expect(err).ToNot(HaveOccurred())
+			Consistently(func() bool {
+				if err := k8sClient.Get(ctx, client.ObjectKeyFromObject(&rollout3), &rollout3); err != nil {
+					return false
+				}
+				return reflect.DeepEqual(rollout3.Status, rv1alpha1.RolloutStatus{})
+			}, "1m", "1s").Should(
+				BeTrue(),
+			)
+		})
+
+		/*
+			In this a namespace-scoped RolloutManager is created in a namespaces.
+			Then operator should create required resources (ServiceAccount, Roles, RoleBinding, Service, Secret, Deployment) in namespace.
+			Now when a Rollouts CR is created in another namespace (which doesn't have RolloutsManagers), operator should not reconcile it.
+		*/
+		It("After creating namespace-scoped RolloutManager in a namespace, operator should create appropriate K8s resources, but it should not watch argo rollouts CR in other namespace.", func() {
+
+			nsName1, nsName2 := "test-rom-ns", "test-ro-ns"
+
+			By("1st NS: Create a namespace for RolloutManager.")
+			Expect(utils.CreateNamespace(ctx, k8sClient, nsName1)).To(Succeed())
+
+			By("1st NS: Create namespace-scoped RolloutManager in 1st namespace.")
+			rolloutsManager, err := utils.CreateRolloutManager(ctx, k8sClient, "test-rollouts-manager-1", nsName1, true)
+			Expect(err).ToNot(HaveOccurred())
+
+			By("1st NS: Verify that RolloutManager is successfully created in 1st namespace.")
+			Eventually(rolloutsManager, "2m", "1s").Should(rmFixture.HavePhase(rmv1alpha1.PhaseAvailable))
+
+			By("1st NS: Verify that Status.Condition is having success condition.")
+			Eventually(rolloutsManager, "2m", "1s").Should(rmFixture.HaveSuccessCondition())
+
+			By("1st NS: Verify that expected resources are created in 1st namespace.")
+			utils.ValidateArgoRolloutManagerResources(ctx, rolloutsManager, k8sClient, true)
+
+			By("2nd NS: Verify argo Rollouts controller of 1st namespace is not able to reconcile CR from a different namespace.")
+
+			By("2nd NS: Create 2nd namespace for RolloutManager.")
+			Expect(utils.CreateNamespace(ctx, k8sClient, nsName2)).To(Succeed())
+
+			By("2nd NS: Create active and preview services required for Rollout CR in 2nd namespace.")
+			rolloutServiceActive, err := utils.CreateService(ctx, k8sClient, utils.RolloutsActiveServiceName, nsName2, 31000)
+			Expect(err).ToNot(HaveOccurred())
+			Eventually(&rolloutServiceActive, "10s", "1s").Should(k8s.ExistByName(k8sClient))
+
+			rolloutServicePreview, err := utils.CreateService(ctx, k8sClient, utils.RolloutsPreviewServiceName, nsName2, 32000)
+			Expect(err).ToNot(HaveOccurred())
+			Eventually(&rolloutServicePreview, "10s", "1s").Should(k8s.ExistByName(k8sClient))
+
+			By("2nd NS: Create Argo Rollout CR in a 2nd namespace and verify that it is not reconciled by Rollouts controller of 1st namespace.")
+			rollout, err := utils.CreateArgoRollout(ctx, k8sClient, utils.RolloutsName, nsName2, rolloutServiceActive.Name, rolloutServicePreview.Name)
+			Expect(err).ToNot(HaveOccurred())
+			Consistently(func() bool {
+				if err := k8sClient.Get(ctx, client.ObjectKeyFromObject(&rollout), &rollout); err != nil {
+					return false
+				}
+				return reflect.DeepEqual(rollout.Status, rv1alpha1.RolloutStatus{})
+			}, "1m", "1s").Should(
+				BeTrue(),
+			)
+		})
+
+		/*
+			In this test a namespace-scoped RolloutManager is created in a namespace.
+			Then operator should create required resources (ServiceAccount, Roles, RoleBinding, Service, Secret, Deployment) in namespace.
+			Now when a cluster-scoped RolloutManager is created, it should not be accepted by operator,
+			since NAMESPACE_SCOPED_ARGO_ROLLOUTS is set to true, it means user can only create namespace-scoped RolloutManagers.
+			but namespace-scoped Rollouts controller of 1st namespace should still be able to reconcile Rollout CR of 1st namespace and failed cluster-scoped RolloutManager of 2nd namespace should not cause any issues.
+			When namespace-scoped RolloutManager is reconciled again it should still work, and should be able to reconcile Rollout CR created in 1st namespace.
+		*/
+		It("Should allow namespace-scoped RolloutManager, but not cluster-scoped.", func() {
+
+			nsName := "test-rom-ns"
+
+			By("1st RM: Create namespace-scoped RolloutManager in 1st namespace.")
+			rolloutsManagerNs, err := utils.CreateRolloutManager(ctx, k8sClient, "test-rollouts-manager-1", fixture.TestE2ENamespace, true)
+			Expect(err).ToNot(HaveOccurred())
+
+			By("1st RM: Verify that RolloutManager is successfully created in 1st namespace.")
+			Eventually(rolloutsManagerNs, "1m", "1s").Should(rmFixture.HavePhase(rmv1alpha1.PhaseAvailable))
+
+			By("1st RM: Verify that Status.Condition is having success condition.")
+			Eventually(rolloutsManagerNs, "1m", "1s").Should(rmFixture.HaveSuccessCondition())
+
+			By("2nd RM: Create 2nd namespace.")
+			Expect(utils.CreateNamespace(ctx, k8sClient, nsName)).To(Succeed())
+
+			By("2nd RM: Create cluster-scoped RolloutManager in 2nd namespace.")
+			rolloutsManagerCl, err := utils.CreateRolloutManager(ctx, k8sClient, "test-rollouts-manager-2", nsName, false)
+			Expect(err).ToNot(HaveOccurred())
+
+			By("2nd RM: Verify that RolloutManager is not working in 2nd namespace.")
+			Eventually(rolloutsManagerCl, "1m", "1s").Should(rmFixture.HavePhase(rmv1alpha1.PhaseFailure))
+
+			By("2nd RM: Verify that Status.Condition is having error message.")
+			Eventually(rolloutsManagerCl, "1m", "1s").Should(rmFixture.HaveCondition(
+				metav1.Condition{
+					Type:    rmv1alpha1.RolloutManagerConditionType,
+					Status:  metav1.ConditionFalse,
+					Reason:  rmv1alpha1.RolloutManagerReasonInvalidScoped,
+					Message: controllers.UnsupportedRolloutManagerClusterScoped,
+				}))
+
+			By("1st RM: Update namespace-scoped RolloutManager of 1st namespace, after reconciliation it should still be working.")
+			err = k8s.UpdateWithoutConflict(ctx, &rolloutsManagerNs, k8sClient, func(obj client.Object) {
+				goObj, ok := obj.(*rmv1alpha1.RolloutManager)
+				Expect(ok).To(BeTrue())
+
+				goObj.Spec.Env = append(goObj.Spec.Env, corev1.EnvVar{Name: "test-name", Value: "test-value"})
+			})
+			Expect(err).ToNot(HaveOccurred())
+
+			Eventually(rolloutsManagerNs, "1m", "1s").Should(rmFixture.HavePhase(rmv1alpha1.PhaseAvailable))
+
+			By("1st RM: Verify that Status.Condition is not having any error.")
+			Eventually(rolloutsManagerNs, "1m", "1s").Should(rmFixture.HaveSuccessCondition())
+
+			By("1st RM: Create Rollout CR in 1st namespace and ensure it is reconciled.")
+			utils.ValidateArgoRolloutsResources(ctx, k8sClient, fixture.TestE2ENamespace, 31000, 32000)
+
+			By("2nd RM: Create Rollout in 2nd namespace and it should not be reconciled as 2nd RolloutManager failed.")
+
+			By("2nd RM: Create active and preview services in 2nd namespace.")
+			rolloutServiceActive, err := utils.CreateService(ctx, k8sClient, utils.RolloutsActiveServiceName, nsName, 31001)
+			Expect(err).ToNot(HaveOccurred())
+			Eventually(&rolloutServiceActive, "10s", "1s").Should(k8s.ExistByName(k8sClient))
+
+			rolloutServicePreview, err := utils.CreateService(ctx, k8sClient, utils.RolloutsPreviewServiceName, nsName, 32002)
+			Expect(err).ToNot(HaveOccurred())
+			Eventually(&rolloutServicePreview, "10s", "1s").Should(k8s.ExistByName(k8sClient))
+
+			By("2nd RM: Create Argo Rollout CR in 2nd namespace and verify that it is not reconciled.")
+			rollout, err := utils.CreateArgoRollout(ctx, k8sClient, "simple-rollout-1", nsName, rolloutServiceActive.Name, rolloutServicePreview.Name)
+			Expect(err).ToNot(HaveOccurred())
+			Consistently(func() bool {
+				if err := k8sClient.Get(ctx, client.ObjectKeyFromObject(&rollout), &rollout); err != nil {
+					return false
+				}
+				return reflect.DeepEqual(rollout.Status, rv1alpha1.RolloutStatus{})
+			}, "30s", "1s").Should(
+				BeTrue(),
+			)
+		})
+	})
+})
diff --git a/tests/e2e/namespace-scoped/namespace_scoped_suite_test.go b/tests/e2e/namespace-scoped/namespace_scoped_suite_test.go
new file mode 100644
index 0000000..1cae4e3
--- /dev/null
+++ b/tests/e2e/namespace-scoped/namespace_scoped_suite_test.go
@@ -0,0 +1,23 @@
+package e2e
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"go.uber.org/zap/zapcore"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+)
+
+var _ = BeforeSuite(func() {
+	logf.SetLogger(zap.New(zap.WriteTo(GinkgoWriter), zap.UseDevMode(true), zap.Level(zapcore.DebugLevel)))
+})
+
+func TestNamespaceScoped(t *testing.T) {
+	suiteConfig, _ := GinkgoConfiguration()
+
+	RegisterFailHandler(Fail)
+
+	RunSpecs(t, "Namespace-scoped Suite", suiteConfig)
+}
diff --git a/tests/e2e/rollouts_test.go b/tests/e2e/rollouts_test.go
index 8c706aa..6472aff 100644
--- a/tests/e2e/rollouts_test.go
+++ b/tests/e2e/rollouts_test.go
@@ -35,7 +35,7 @@ var _ = Describe("RolloutManager tests", func() {
 			Expect(fixture.EnsureCleanSlate()).To(Succeed())
 
 			var err error
-			k8sClient, err = fixture.GetE2ETestKubeClient()
+			k8sClient, _, err = fixture.GetE2ETestKubeClient()
 			Expect(err).ToNot(HaveOccurred())
 			ctx = context.Background()
 
@@ -44,7 +44,9 @@ var _ = Describe("RolloutManager tests", func() {
 					Name:      "basic-rollouts-manager",
 					Namespace: fixture.TestE2ENamespace,
 				},
-				Spec: rolloutsmanagerv1alpha1.RolloutManagerSpec{},
+				Spec: rolloutsmanagerv1alpha1.RolloutManagerSpec{
+					NamespaceScoped: true,
+				},
 			}
 		})
 
@@ -55,90 +57,8 @@ var _ = Describe("RolloutManager tests", func() {
 				By("setting the phase to \"Available\"")
 				Eventually(rolloutManager, "60s", "1s").Should(rolloutManagerFixture.HavePhase(rolloutsmanagerv1alpha1.PhaseAvailable))
 
-				By("creating a service account")
-				sa := corev1.ServiceAccount{
-					ObjectMeta: metav1.ObjectMeta{Name: controllers.DefaultArgoRolloutsResourceName, Namespace: rolloutManager.Namespace},
-				}
-				Eventually(&sa, "10s", "1s").Should(k8s.ExistByName(k8sClient))
-
-				By("ensuring the service account has the correct labels")
-				ensureLabels(&sa.ObjectMeta)
-
-				By("creating a role")
-				role := rbacv1.Role{
-					ObjectMeta: metav1.ObjectMeta{Name: controllers.DefaultArgoRolloutsResourceName, Namespace: rolloutManager.Namespace},
-				}
-				Eventually(&role, "10s", "1s").Should(k8s.ExistByName(k8sClient))
-
-				By("ensuring the role has the correct labels")
-				ensureLabels(&role.ObjectMeta)
-
-				By("ensuring the role has the correct policy rules")
-				Expect(role.Rules).To(ConsistOf(controllers.GetPolicyRules()))
-
-				By("creating a role binding")
-				binding := rbacv1.RoleBinding{
-					ObjectMeta: metav1.ObjectMeta{Name: controllers.DefaultArgoRolloutsResourceName, Namespace: rolloutManager.Namespace},
-				}
-				Eventually(&binding, "10s", "1s").Should(k8s.ExistByName(k8sClient))
-
-				By("ensuring the role binding has the correct labels")
-				ensureLabels(&binding.ObjectMeta)
-
-				By("creating three cluster roles")
-				clusterRoleSuffixes := []string{"aggregate-to-admin", "aggregate-to-edit", "aggregate-to-view"}
-				for _, suffix := range clusterRoleSuffixes {
-					clusterRoleName := "argo-rollouts-" + suffix
-					clusterRole := rbacv1.ClusterRole{
-						ObjectMeta: metav1.ObjectMeta{Name: clusterRoleName},
-					}
-					Eventually(&clusterRole, "30s", "1s").Should(k8s.ExistByName(k8sClient))
-
-					Expect(len(binding.Labels)).To(Equal(3))
-					Expect(clusterRole.Labels["app.kubernetes.io/name"]).To(Equal(clusterRoleName))
-					Expect(clusterRole.Labels["app.kubernetes.io/part-of"]).To(Equal(controllers.DefaultArgoRolloutsResourceName))
-					Expect(clusterRole.Labels["app.kubernetes.io/component"]).To(Equal("aggregate-cluster-role"))
-					Expect(clusterRole.Labels["rbac.authorization.k8s.io/"+suffix]).To(Equal("true"))
-				}
-
-				By("creating a deployment")
-				deployment := appsv1.Deployment{
-					ObjectMeta: metav1.ObjectMeta{Name: controllers.DefaultArgoRolloutsResourceName, Namespace: rolloutManager.Namespace},
-				}
-				Eventually(&deployment, "10s", "1s").Should(k8s.ExistByName(k8sClient))
-
-				By("ensuring the deployment has the correct labels")
-				ensureLabels(&deployment.ObjectMeta)
-
-				By("creating a service")
-				service := corev1.Service{
-					ObjectMeta: metav1.ObjectMeta{Name: controllers.DefaultArgoRolloutsMetricsServiceName, Namespace: rolloutManager.Namespace},
-				}
-				Eventually(&service, "10s", "1s").Should(k8s.ExistByName(k8sClient))
-
-				By("ensuring the service has the correct labels")
-				Expect(service.Labels["app.kubernetes.io/name"]).To(Equal(controllers.DefaultArgoRolloutsMetricsServiceName))
-				Expect(service.Labels["app.kubernetes.io/part-of"]).To(Equal(controllers.DefaultArgoRolloutsResourceName))
-				Expect(service.Labels["app.kubernetes.io/component"]).To(Equal("server"))
-
-				By("having the deployment become ready")
-				Eventually(func() bool {
-
-					depl := appsv1.Deployment{
-						ObjectMeta: metav1.ObjectMeta{Name: controllers.DefaultArgoRolloutsResourceName, Namespace: rolloutManager.Namespace},
-					}
-					if err := k8sClient.Get(ctx, client.ObjectKeyFromObject(&depl), &depl); err != nil {
-						return false
-					}
-
-					return depl.Status.ReadyReplicas == 1
-
-				}, "120s", "1s").Should(BeTrue())
-
-				By("creating a secret")
-				Eventually(&corev1.Secret{
-					ObjectMeta: metav1.ObjectMeta{Name: controllers.DefaultRolloutsNotificationSecretName, Namespace: rolloutManager.Namespace},
-				}, "30s", "1s").Should(k8s.ExistByName(k8sClient))
+				By("Verify that expected resources are created.")
+				ValidateArgoRolloutManagerResources(ctx, rolloutManager, k8sClient, true)
 			})
 		})
 
@@ -199,8 +119,11 @@ var _ = Describe("RolloutManager tests", func() {
 						"--loglevel",
 						"error",
 					},
+					NamespaceScoped: true,
 				}
 				Expect(k8sClient.Create(ctx, &rolloutManager)).To(Succeed())
+				Eventually(rolloutManager, "1m", "1s").Should(rolloutManagerFixture.HavePhase(rolloutsmanagerv1alpha1.PhaseAvailable))
+
 				deployment := appsv1.Deployment{
 					ObjectMeta: metav1.ObjectMeta{Name: controllers.DefaultArgoRolloutsResourceName, Namespace: rolloutManager.Namespace},
 				}
@@ -208,14 +131,21 @@ var _ = Describe("RolloutManager tests", func() {
 				Expect(deployment.Spec.Template.Spec.Containers[0].Args).To(Equal([]string{"--namespaced", "--loglevel", "error"}))
 
 				By("updating the deployment when the argument in the RolloutManager is updated")
-				Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(&rolloutManager), &rolloutManager)).To(Succeed())
-				rolloutManager.Spec = rolloutsmanagerv1alpha1.RolloutManagerSpec{
-					ExtraCommandArgs: []string{
-						"--logformat",
-						"text",
-					},
-				}
-				Expect(k8sClient.Update(ctx, &rolloutManager)).To(Succeed())
+
+				err := k8s.UpdateWithoutConflict(ctx, &rolloutManager, k8sClient, func(obj client.Object) {
+					goObj, ok := obj.(*rolloutsmanagerv1alpha1.RolloutManager)
+					Expect(ok).To(BeTrue())
+
+					goObj.Spec = rolloutsmanagerv1alpha1.RolloutManagerSpec{
+						ExtraCommandArgs: []string{
+							"--logformat",
+							"text",
+						},
+						NamespaceScoped: true,
+					}
+				})
+				Expect(err).ToNot(HaveOccurred())
+
 				Eventually(func() []string {
 					Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(&deployment), &deployment)).To(Succeed())
 					return deployment.Spec.Template.Spec.Containers[0].Args
@@ -226,13 +156,15 @@ var _ = Describe("RolloutManager tests", func() {
 		When("A RolloutManager specifies environment variables", func() {
 			It("should reflect those variables in the deployment", func() {
 				By("creating the deployment with the environment variables specified in the RolloutManager")
-				rolloutManager.Spec = rolloutsmanagerv1alpha1.RolloutManagerSpec{
-					Env: []corev1.EnvVar{
-						{Name: "EDITOR", Value: "emacs"},
-						{Name: "LANG", Value: "en_CA.UTF-8"},
-					},
+
+				rolloutManager.Spec.Env = []corev1.EnvVar{
+					{Name: "EDITOR", Value: "emacs"},
+					{Name: "LANG", Value: "en_CA.UTF-8"},
 				}
+
 				Expect(k8sClient.Create(ctx, &rolloutManager)).To(Succeed())
+				Eventually(rolloutManager, "1m", "1s").Should(rolloutManagerFixture.HavePhase(rolloutsmanagerv1alpha1.PhaseAvailable))
+
 				deployment := appsv1.Deployment{
 					ObjectMeta: metav1.ObjectMeta{Name: controllers.DefaultArgoRolloutsResourceName, Namespace: rolloutManager.Namespace},
 				}
@@ -246,14 +178,18 @@ var _ = Describe("RolloutManager tests", func() {
 				))
 
 				By("updating the deployment when the environment variables in the RolloutManager are updated")
-				Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(&rolloutManager), &rolloutManager)).To(Succeed())
-				rolloutManager.Spec = rolloutsmanagerv1alpha1.RolloutManagerSpec{
-					Env: []corev1.EnvVar{
+
+				err := k8s.UpdateWithoutConflict(ctx, &rolloutManager, k8sClient, func(obj client.Object) {
+					goObj, ok := obj.(*rolloutsmanagerv1alpha1.RolloutManager)
+					Expect(ok).To(BeTrue())
+
+					goObj.Spec.Env = []corev1.EnvVar{
 						{Name: "LANG", Value: "en_US.UTF-8"},
 						{Name: "TERM", Value: "xterm-256color"},
-					},
-				}
-				Expect(k8sClient.Update(ctx, &rolloutManager)).To(Succeed())
+					}
+				})
+				Expect(err).ToNot(HaveOccurred())
+
 				Eventually(func() []corev1.EnvVar {
 					Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(&deployment), &deployment)).To(Succeed())
 					return deployment.Spec.Template.Spec.Containers[0].Env
@@ -270,11 +206,12 @@ var _ = Describe("RolloutManager tests", func() {
 		When("A RolloutManager specifies an image", func() {
 			It("should reflect that image in the deployment", func() {
 				By("creating the deployment with the image specified in the RolloutManager")
-				rolloutManager.Spec = rolloutsmanagerv1alpha1.RolloutManagerSpec{
-					Image:   "quay.io/prometheus/busybox",
-					Version: "latest",
-				}
+
+				rolloutManager.Spec.Image = "quay.io/prometheus/busybox"
+				rolloutManager.Spec.Version = "latest"
+
 				Expect(k8sClient.Create(ctx, &rolloutManager)).To(Succeed())
+				Eventually(rolloutManager, "1m", "1s").Should(rolloutManagerFixture.HavePhase(rolloutsmanagerv1alpha1.PhasePending))
 
 				deployment := appsv1.Deployment{
 					ObjectMeta: metav1.ObjectMeta{Name: controllers.DefaultArgoRolloutsResourceName, Namespace: rolloutManager.Namespace},
@@ -285,12 +222,16 @@ var _ = Describe("RolloutManager tests", func() {
 				Expect(deployment.Spec.Template.Spec.Containers[0].Image).To(Equal(expectedVersion))
 
 				By("updating the deployment when the image in the RolloutManager is updated")
-				Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(&rolloutManager), &rolloutManager)).To(Succeed())
-				rolloutManager.Spec = rolloutsmanagerv1alpha1.RolloutManagerSpec{
-					Image:   controllers.DefaultArgoRolloutsImage,
-					Version: controllers.DefaultArgoRolloutsVersion,
-				}
-				Expect(k8sClient.Update(ctx, &rolloutManager)).To(Succeed())
+
+				err := k8s.UpdateWithoutConflict(ctx, &rolloutManager, k8sClient, func(obj client.Object) {
+					goObj, ok := obj.(*rolloutsmanagerv1alpha1.RolloutManager)
+					Expect(ok).To(BeTrue())
+					goObj.Spec.Image = controllers.DefaultArgoRolloutsImage
+					goObj.Spec.Version = controllers.DefaultArgoRolloutsVersion
+
+				})
+				Expect(err).ToNot(HaveOccurred())
+
 				expectedVersion = controllers.DefaultArgoRolloutsImage + ":" + controllers.DefaultArgoRolloutsVersion
 				Eventually(func() string {
 					Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(&deployment), &deployment)).To(Succeed())
@@ -300,11 +241,3 @@ var _ = Describe("RolloutManager tests", func() {
 		})
 	})
 })
-
-func ensureLabels(object *metav1.ObjectMeta) {
-	GinkgoHelper()
-	Expect(len(object.Labels)).To(Equal(3))
-	Expect(object.Labels["app.kubernetes.io/name"]).To(Equal(controllers.DefaultArgoRolloutsResourceName))
-	Expect(object.Labels["app.kubernetes.io/part-of"]).To(Equal(controllers.DefaultArgoRolloutsResourceName))
-	Expect(object.Labels["app.kubernetes.io/component"]).To(Equal(controllers.DefaultArgoRolloutsResourceName))
-}
diff --git a/tests/e2e/test_utils.go b/tests/e2e/test_utils.go
new file mode 100644
index 0000000..4f00466
--- /dev/null
+++ b/tests/e2e/test_utils.go
@@ -0,0 +1,449 @@
+package e2e
+
+import (
+	"context"
+	"fmt"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	"github.com/argoproj-labs/argo-rollouts-manager/tests/e2e/fixture"
+	"github.com/argoproj-labs/argo-rollouts-manager/tests/e2e/fixture/k8s"
+
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	rmv1alpha1 "github.com/argoproj-labs/argo-rollouts-manager/api/v1alpha1"
+
+	controllers "github.com/argoproj-labs/argo-rollouts-manager/controllers"
+	rv1alpha1 "github.com/argoproj/argo-rollouts/pkg/apis/rollouts/v1alpha1"
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
+)
+
+const (
+	RolloutsActiveServiceName  = "rollout-bluegreen-active"
+	RolloutsPreviewServiceName = "rollout-bluegreen-preview"
+	RolloutsName               = "simple-rollout"
+)
+
+// Create namespace for tests having a specific label for identification
+func CreateNamespace(ctx context.Context, k8sClient client.Client, name string) error {
+	return k8sClient.Create(ctx,
+		&corev1.Namespace{ObjectMeta: metav1.ObjectMeta{
+			Name:   name,
+			Labels: fixture.NamespaceLabels,
+		}})
+}
+
+// Create RolloutManager CR
+func CreateRolloutManager(ctx context.Context, k8sClient client.Client, name, namespace string, namespaceScoped bool) (rmv1alpha1.RolloutManager, error) {
+	rolloutsManager := rmv1alpha1.RolloutManager{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+		},
+		Spec: rmv1alpha1.RolloutManagerSpec{
+			NamespaceScoped: namespaceScoped,
+		},
+	}
+	return rolloutsManager, k8sClient.Create(ctx, &rolloutsManager)
+}
+
+// Create Service used by Rollout
+func CreateService(ctx context.Context, k8sClient client.Client, name, namespace string, nodePort int32) (corev1.Service, error) {
+	service := corev1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+		},
+		Spec: corev1.ServiceSpec{
+			Type:     corev1.ServiceTypeNodePort,
+			Selector: fixture.NamespaceLabels,
+			Ports: []corev1.ServicePort{
+				{
+					NodePort: nodePort,
+					Protocol: corev1.ProtocolTCP,
+					Port:     8080,
+				},
+			},
+		},
+	}
+	return service, k8sClient.Create(ctx, &service)
+}
+
+// Create Argo Rollout CR
+func CreateArgoRollout(ctx context.Context, k8sClient client.Client, name, namespace, activeService, previewService string) (rv1alpha1.Rollout, error) {
+	var num int32 = 2
+	autoPromotionEnabled := false
+
+	rollout := rv1alpha1.Rollout{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+		},
+		Spec: rv1alpha1.RolloutSpec{
+			Replicas: &num,
+			Strategy: rv1alpha1.RolloutStrategy{
+				BlueGreen: &rv1alpha1.BlueGreenStrategy{
+					ActiveService:        activeService,
+					PreviewService:       previewService,
+					AutoPromotionEnabled: &autoPromotionEnabled,
+				},
+			},
+			RevisionHistoryLimit: &num,
+			Selector: &metav1.LabelSelector{
+				MatchLabels: fixture.NamespaceLabels,
+			},
+			Template: corev1.PodTemplateSpec{
+				ObjectMeta: metav1.ObjectMeta{
+					Labels: fixture.NamespaceLabels,
+				},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name:  "webserver-simple",
+							Image: "docker.io/kostiscodefresh/gitops-canary-app:v1.0",
+							Ports: []corev1.ContainerPort{
+								{
+									Name:          "http",
+									ContainerPort: 8080,
+									Protocol:      corev1.ProtocolTCP,
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+	return rollout, k8sClient.Create(ctx, &rollout)
+}
+
+// Check resources created after creating of RolloutManager CR and verify that they are healthy.
+func ValidateArgoRolloutManagerResources(ctx context.Context, rolloutsManager rmv1alpha1.RolloutManager, k8sClient client.Client, namespaceScoped bool) {
+
+	By("Verify that ServiceAccount is created.")
+	validateServiceAccount(k8sClient, rolloutsManager)
+
+	if namespaceScoped {
+		By("Verify that argo-rollout Role is created.")
+		validateArgoRolloutsRole(k8sClient, rolloutsManager)
+	} else {
+		By("Verify that argo-rollout ClusterRoles is created.")
+		validateArgoRolloutsClusterRole(k8sClient)
+	}
+
+	By("Verify that aggregate-to-admin ClusterRole is created.")
+	validateAggregateToAdminClusterRole(k8sClient)
+
+	By("Verify that aggregate-to-edit ClusterRole is created.")
+	validateAggregateToEditClusterRole(k8sClient)
+
+	By("Verify that aggregate-to-view ClusterRole is created.")
+	validateAggregateToViewClusterRole(k8sClient)
+
+	if namespaceScoped {
+		By("Verify that RoleBinding is created.")
+		validateRoleBinding(k8sClient, rolloutsManager)
+	} else {
+		By("Verify that ClusterRoleBinding is created.")
+		validateClusterRoleBinding(k8sClient, rolloutsManager)
+	}
+
+	By("Verify that Service is created.")
+	validateService(k8sClient, rolloutsManager)
+
+	By("Verify that Secret is created.")
+	validateSecret(k8sClient, rolloutsManager)
+
+	By("Verify that argo rollouts deployment is created and it is in Ready state.")
+	validateDeployment(ctx, k8sClient, rolloutsManager)
+}
+
+// Create Argo Rollout CR and Services required by it and verify that they are healthy.
+func ValidateArgoRolloutsResources(ctx context.Context, k8sClient client.Client, nsName string, port1, port2 int32) {
+
+	By("Create active Services in given namespace")
+	rolloutServiceActive, err := CreateService(ctx, k8sClient, RolloutsActiveServiceName, nsName, port1)
+	Expect(err).ToNot(HaveOccurred())
+	Eventually(&rolloutServiceActive, "10s", "1s").Should(k8s.ExistByName(k8sClient))
+
+	By("Create preview Services in given namespace")
+	rolloutServicePreview, err := CreateService(ctx, k8sClient, RolloutsPreviewServiceName, nsName, port2)
+	Expect(err).ToNot(HaveOccurred())
+	Eventually(&rolloutServicePreview, "10s", "1s").Should(k8s.ExistByName(k8sClient))
+
+	By("Create Argo Rollout CR in given namespace and check it is reconciled successfully.")
+	rollout, err := CreateArgoRollout(ctx, k8sClient, RolloutsName, nsName, rolloutServiceActive.Name, rolloutServicePreview.Name)
+	Expect(err).ToNot(HaveOccurred())
+	Eventually(func() bool {
+		if err := k8sClient.Get(ctx, client.ObjectKeyFromObject(&rollout), &rollout); err != nil {
+			return false
+		}
+		return rollout.Status.Phase == rv1alpha1.RolloutPhaseHealthy
+	}, "3m", "1s").Should(BeTrue())
+}
+
+func validateServiceAccount(k8sClient client.Client, rolloutsManager rmv1alpha1.RolloutManager) {
+	sa := &corev1.ServiceAccount{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      controllers.DefaultArgoRolloutsResourceName,
+			Namespace: rolloutsManager.Namespace,
+		},
+	}
+	Eventually(sa, "10s", "1s").Should(k8s.ExistByName(k8sClient))
+
+	By("Verify that ServiceAccount has correct labels.")
+	validateLabels(&sa.ObjectMeta)
+}
+
+func validateArgoRolloutsRole(k8sClient client.Client, rolloutsManager rmv1alpha1.RolloutManager) {
+	role := &rbacv1.Role{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      controllers.DefaultArgoRolloutsResourceName,
+			Namespace: rolloutsManager.Namespace,
+		},
+	}
+	Eventually(role, "10s", "1s").Should(k8s.ExistByName(k8sClient))
+
+	By("Verify that Role has correct labels.")
+	validateLabels(&role.ObjectMeta)
+
+	By("Verify that Role has correct policy rules.")
+	Expect(role.Rules).To(ConsistOf(controllers.GetPolicyRules()))
+}
+
+func validateArgoRolloutsClusterRole(k8sClient client.Client) {
+	clusterRole := &rbacv1.ClusterRole{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: controllers.DefaultArgoRolloutsResourceName,
+		},
+	}
+	Eventually(clusterRole, "30s", "1s").Should(k8s.ExistByName(k8sClient))
+
+	By("Verify that ClusterRole has correct labels.")
+	validateLabels(&clusterRole.ObjectMeta)
+
+	By("Verify that ClusterRole has correct policy rules.")
+	Expect(clusterRole.Rules).To(ConsistOf(controllers.GetPolicyRules()))
+}
+
+func validateAggregateToAdminClusterRole(k8sClient client.Client) {
+
+	aggregationType := "aggregate-to-admin"
+	clusterRoleName := fmt.Sprintf("%s-%s", controllers.DefaultArgoRolloutsResourceName, aggregationType)
+
+	clusterRole := &rbacv1.ClusterRole{ObjectMeta: metav1.ObjectMeta{
+		Name: clusterRoleName,
+	},
+	}
+	Eventually(clusterRole, "30s", "1s").Should(k8s.ExistByName(k8sClient))
+
+	By("Verify that ClusterRole has correct labels.")
+	validateAggregateLabels(&clusterRole.ObjectMeta, aggregationType)
+
+	By("Verify that ClusterRole has correct policy rules.")
+	Expect(clusterRole.Rules).To(ConsistOf(controllers.GetAggregateToAdminPolicyRules()))
+}
+
+func validateAggregateToEditClusterRole(k8sClient client.Client) {
+
+	aggregationType := "aggregate-to-edit"
+	clusterRoleName := fmt.Sprintf("%s-%s", controllers.DefaultArgoRolloutsResourceName, aggregationType)
+
+	clusterRole := &rbacv1.ClusterRole{ObjectMeta: metav1.ObjectMeta{
+		Name: clusterRoleName,
+	},
+	}
+	Eventually(clusterRole, "30s", "1s").Should(k8s.ExistByName(k8sClient))
+
+	By("Verify that ClusterRole has correct labels.")
+	validateAggregateLabels(&clusterRole.ObjectMeta, aggregationType)
+
+	By("Verify that ClusterRole has correct policy rules.")
+	Expect(clusterRole.Rules).To(ConsistOf(controllers.GetAggregateToEditPolicyRules()))
+}
+
+func validateAggregateToViewClusterRole(k8sClient client.Client) {
+
+	aggregationType := "aggregate-to-view"
+	clusterRoleName := fmt.Sprintf("%s-%s", controllers.DefaultArgoRolloutsResourceName, aggregationType)
+
+	clusterRole := &rbacv1.ClusterRole{ObjectMeta: metav1.ObjectMeta{
+		Name: clusterRoleName,
+	},
+	}
+	Eventually(clusterRole, "30s", "1s").Should(k8s.ExistByName(k8sClient))
+
+	By("Verify that ClusterRole has correct labels.")
+	validateAggregateLabels(&clusterRole.ObjectMeta, aggregationType)
+
+	By("Verify that ClusterRole has correct policy rules.")
+	Expect(clusterRole.Rules).To(ConsistOf(controllers.GetAggregateToViewPolicyRules()))
+}
+
+func validateRoleBinding(k8sClient client.Client, rolloutsManager rmv1alpha1.RolloutManager) {
+	binding := &rbacv1.RoleBinding{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      controllers.DefaultArgoRolloutsResourceName,
+			Namespace: rolloutsManager.Namespace,
+		},
+	}
+	Eventually(binding, "30s", "1s").Should(k8s.ExistByName(k8sClient))
+
+	By("Verify that RoleBinding has correct labels.")
+	validateLabels(&binding.ObjectMeta)
+
+	By("Verify that RoleBinding has correct RoleRef.")
+	Expect(binding.RoleRef).To(Equal(rbacv1.RoleRef{
+		APIGroup: rbacv1.GroupName,
+		Kind:     "Role",
+		Name:     controllers.DefaultArgoRolloutsResourceName,
+	}))
+
+	By("Verify that RoleBinding has correct Subjects.")
+	Expect(binding.Subjects).To(Equal(
+		[]rbacv1.Subject{
+			{
+				Kind:      rbacv1.ServiceAccountKind,
+				Name:      controllers.DefaultArgoRolloutsResourceName,
+				Namespace: rolloutsManager.Namespace,
+			},
+		},
+	))
+}
+
+func validateClusterRoleBinding(k8sClient client.Client, rolloutsManager rmv1alpha1.RolloutManager) {
+	clusterRoleBinding := &rbacv1.ClusterRoleBinding{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: controllers.DefaultArgoRolloutsResourceName,
+		},
+	}
+	Eventually(clusterRoleBinding, "30s", "1s").Should(k8s.ExistByName(k8sClient))
+
+	By("Verify that ClusterRoleBinding has correct labels.")
+	validateLabels(&clusterRoleBinding.ObjectMeta)
+
+	By("Verify that ClusterRoleBinding has correct RoleRef.")
+	Expect(clusterRoleBinding.RoleRef).To(Equal(rbacv1.RoleRef{
+		APIGroup: rbacv1.GroupName,
+		Kind:     "ClusterRole",
+		Name:     controllers.DefaultArgoRolloutsResourceName,
+	}))
+
+	By("Verify that ClusterRoleBinding has correct Subjects.")
+	Expect(clusterRoleBinding.Subjects).To(Equal(
+		[]rbacv1.Subject{
+			{
+				Kind:      rbacv1.ServiceAccountKind,
+				Name:      controllers.DefaultArgoRolloutsResourceName,
+				Namespace: rolloutsManager.Namespace,
+			},
+		},
+	))
+}
+
+func validateService(k8sClient client.Client, rolloutsManager rmv1alpha1.RolloutManager) {
+	service := &corev1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      controllers.DefaultArgoRolloutsMetricsServiceName,
+			Namespace: rolloutsManager.Namespace,
+		},
+	}
+	Eventually(service, "10s", "1s").Should(k8s.ExistByName(k8sClient))
+
+	By("Verify that Service has correct labels.")
+	Expect(service.Labels["app.kubernetes.io/name"]).To(Equal(controllers.DefaultArgoRolloutsMetricsServiceName))
+	Expect(service.Labels["app.kubernetes.io/part-of"]).To(Equal(controllers.DefaultArgoRolloutsResourceName))
+	Expect(service.Labels["app.kubernetes.io/component"]).To(Equal("server"))
+
+	By("Verify that ClusterRoleBinding has correct Ports.")
+	Expect(service.Spec.Ports).To(Equal([]corev1.ServicePort{
+		{
+			Name:       "metrics",
+			Port:       8090,
+			Protocol:   corev1.ProtocolTCP,
+			TargetPort: intstr.FromInt(8090),
+		},
+	}))
+
+	By("Verify that ClusterRoleBinding has correct Ports.")
+	Expect(service.Spec.Selector).To(Equal(map[string]string{
+		controllers.DefaultRolloutsSelectorKey: controllers.DefaultArgoRolloutsResourceName,
+	}))
+}
+
+func validateSecret(k8sClient client.Client, rolloutsManager rmv1alpha1.RolloutManager) {
+	secret := &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      controllers.DefaultRolloutsNotificationSecretName,
+			Namespace: rolloutsManager.Namespace,
+		},
+	}
+	Eventually(secret, "30s", "1s").Should(k8s.ExistByName(k8sClient))
+
+	By("Verify that Secret has correct Type.")
+	Expect(secret.Type).To(Equal(corev1.SecretTypeOpaque))
+}
+
+func validateDeployment(ctx context.Context, k8sClient client.Client, rolloutsManager rmv1alpha1.RolloutManager) {
+	depl := appsv1.Deployment{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      controllers.DefaultArgoRolloutsResourceName,
+			Namespace: rolloutsManager.Namespace,
+		},
+	}
+	Eventually(&depl, "10s", "1s").Should(k8s.ExistByName(k8sClient))
+
+	By("Verify that Deployment replica is in Ready state.")
+	Eventually(func() bool {
+		if err := k8sClient.Get(ctx, client.ObjectKeyFromObject(&depl), &depl); err != nil {
+			return false
+		}
+		return depl.Status.ReadyReplicas == 1
+	}, "3m", "1s").Should(BeTrue())
+
+	By("Verify that Deployment has correct labels.")
+	validateLabels(&depl.ObjectMeta)
+
+	By("Verify that Deployment has correct Selector.")
+	Expect(depl.Spec.Selector).To(Equal(&metav1.LabelSelector{
+		MatchLabels: map[string]string{
+			controllers.DefaultRolloutsSelectorKey: controllers.DefaultArgoRolloutsResourceName,
+		}}))
+
+	By("Verify that Deployment Template has correct Template.")
+	Expect(depl.Spec.Template.Labels).To(Equal(map[string]string{controllers.DefaultRolloutsSelectorKey: controllers.DefaultArgoRolloutsResourceName}))
+
+	By("Verify that Deployment Template has correct NodeSelector.")
+	Expect(depl.Spec.Template.Spec.NodeSelector).To(Equal(map[string]string{"kubernetes.io/os": "linux"}))
+
+	By("Verify that Deployment Template has correct SecurityContext.")
+	Expect(*depl.Spec.Template.Spec.SecurityContext.RunAsNonRoot).To(BeTrue())
+
+	By("Verify that Deployment Template has correct ServiceAccountName.")
+	Expect(depl.Spec.Template.Spec.ServiceAccountName).To(Equal(controllers.DefaultArgoRolloutsResourceName))
+
+	By("Verify that Deployment Template Container is not empty.")
+	Expect(depl.Spec.Template.Spec.Containers[0]).ToNot(Equal(corev1.Container{}))
+}
+
+func validateLabels(object *metav1.ObjectMeta) {
+	GinkgoHelper()
+	Expect(len(object.Labels)).To(Equal(3))
+	Expect(object.Labels["app.kubernetes.io/name"]).To(Equal(controllers.DefaultArgoRolloutsResourceName))
+	Expect(object.Labels["app.kubernetes.io/part-of"]).To(Equal(controllers.DefaultArgoRolloutsResourceName))
+	Expect(object.Labels["app.kubernetes.io/component"]).To(Equal(controllers.DefaultArgoRolloutsResourceName))
+}
+
+func validateAggregateLabels(object *metav1.ObjectMeta, aggregationType string) {
+	GinkgoHelper()
+	Expect(len(object.Labels)).To(Equal(4))
+	Expect(object.Labels["app.kubernetes.io/name"]).To(Equal(object.Name))
+	Expect(object.Labels["app.kubernetes.io/part-of"]).To(Equal(controllers.DefaultArgoRolloutsResourceName))
+	Expect(object.Labels["app.kubernetes.io/component"]).To(Equal("aggregate-cluster-role"))
+	Expect(object.Labels["rbac.authorization.k8s.io/"+aggregationType]).To(Equal("true"))
+}