diff --git a/Taskfile.yaml b/Taskfile.yaml
index 18036c4..1176f31 100644
--- a/Taskfile.yaml
+++ b/Taskfile.yaml
@@ -9,7 +9,7 @@ includes:
       NESTED_MODULES: api
       API_DIRS: '{{.ROOT_DIR}}/api/provider/v1alpha1/... {{.ROOT_DIR}}/api/clusters/v1alpha1/...'
       MANIFEST_OUT: '{{.ROOT_DIR}}/api/crds/manifests'
-      CODE_DIRS: '{{.ROOT_DIR}}/cmd/... {{.ROOT_DIR}}/internal/... {{.ROOT_DIR}}/test/... {{.ROOT_DIR}}/api/provider/v1alpha1/... {{.ROOT_DIR}}/api/clusters/v1alpha1/...'
+      CODE_DIRS: '{{.ROOT_DIR}}/cmd/... {{.ROOT_DIR}}/internal/... {{.ROOT_DIR}}/api/provider/v1alpha1/... {{.ROOT_DIR}}/api/clusters/v1alpha1/...'
       COMPONENTS: 'openmcp-operator'
       REPO_URL: 'https://github.com/openmcp-project/openmcp-operator'
       GENERATE_DOCS_INDEX: "true"
diff --git a/api/clusters/v1alpha1/accessrequest_types.go b/api/clusters/v1alpha1/accessrequest_types.go
index 46fd616..b04208f 100644
--- a/api/clusters/v1alpha1/accessrequest_types.go
+++ b/api/clusters/v1alpha1/accessrequest_types.go
@@ -12,6 +12,12 @@ type AccessRequestSpec struct {
 	// +kubebuilder:validation:XValidation:rule="self == oldSelf",message="clusterRef is immutable"
 	ClusterRef NamespacedObjectReference `json:"clusterRef"`
 
+	// RequestRef is the reference to the ClusterRequest for whose Cluster access is requested.
+	// Exactly one of clusterRef or requestRef must be set.
+	// This value is immutable.
+	// +kubebuilder:validation:XValidation:rule="self == oldSelf",message="requestRef is immutable"
+	RequestRef NamespacedObjectReference `json:"requestRef"`
+
 	// Permissions are the requested permissions.
 	Permissions []PermissionsRequest `json:"permissions"`
 }
@@ -32,14 +38,20 @@ type AccessRequestStatus struct {
 	CommonStatus `json:",inline"`
 
 	// Phase is the current phase of the request.
+	// +kubebuilder:default=Pending
+	// +kubebuilder:validation:Enum=Pending;Granted;Denied
 	Phase RequestPhase `json:"phase"`
 
-	// TODO: expose actual access information
+	// SecretRef holds the reference to the secret that contains the actual credentials.
+	SecretRef *NamespacedObjectReference `json:"secretRef,omitempty"`
 }
 
 // +kubebuilder:object:root=true
 // +kubebuilder:subresource:status
-// +kubebuilder:metadata:labels="openmcp.cloud/cluster=onboarding"
+// +kubebuilder:resource:shortName=ar;areq
+// +kubebuilder:metadata:labels="openmcp.cloud/cluster=platform"
+// +kubebuilder:selectablefield:JSONPath=".status.phase"
+// +kubebuilder:printcolumn:JSONPath=".status.phase",name="Phase",type=string
 
 // AccessRequest is the Schema for the accessrequests API
 type AccessRequest struct {
diff --git a/api/clusters/v1alpha1/cluster_types.go b/api/clusters/v1alpha1/cluster_types.go
index d49aab9..3a01c8e 100644
--- a/api/clusters/v1alpha1/cluster_types.go
+++ b/api/clusters/v1alpha1/cluster_types.go
@@ -6,6 +6,7 @@ import (
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/sets"
 )
 
 // ClusterSpec defines the desired state of Cluster
@@ -81,15 +82,14 @@ const (
 
 // +kubebuilder:object:root=true
 // +kubebuilder:subresource:status
-// +kubebuilder:metadata:labels="openmcp.cloud/cluster=onboarding"
-// +kubebuilder:selectablefield:JSONPath=".spec.clusterProfileRef.name"
+// +kubebuilder:metadata:labels="openmcp.cloud/cluster=platform"
+// +kubebuilder:selectablefield:JSONPath=".spec.profile"
 // +kubebuilder:printcolumn:JSONPath=".spec.purposes",name="Purposes",type=string
 // +kubebuilder:printcolumn:JSONPath=`.status.phase`,name="Phase",type=string
 // +kubebuilder:printcolumn:JSONPath=`.metadata.annotations["clusters.openmcp.cloud/k8sversion"]`,name="Version",type=string
 // +kubebuilder:printcolumn:JSONPath=`.metadata.annotations["clusters.openmcp.cloud/profile"]`,name="Profile",type=string
-// +kubebuilder:printcolumn:JSONPath=`.metadata.labels["environment.clusters.openmcp.cloud"]`,name="Env",type=string,priority=10
 // +kubebuilder:printcolumn:JSONPath=`.metadata.labels["provider.clusters.openmcp.cloud"]`,name="Provider",type=string, priority=10
-// +kubebuilder:printcolumn:JSONPath=".spec.clusterProfileRef.name",name="ProfileRef",type=string,priority=10
+// +kubebuilder:printcolumn:JSONPath=".spec.profile",name="ProfileRef",type=string,priority=10
 // +kubebuilder:printcolumn:JSONPath=`.metadata.annotations["clusters.openmcp.cloud/providerinfo"]`,name="Info",type=string,priority=10
 // +kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp"
 
@@ -132,12 +132,19 @@ func (cs *ClusterStatus) SetProviderStatus(from any) error {
 
 // GetTenancyCount returns the number of ClusterRequests currently pointing to this cluster.
 // This is determined by counting the finalizers that have the corresponding prefix.
+// Note that only unique finalizers are counted, so if there are multiple identical request finalizers
+// (which should not happen), this method's return value might not match the actual number of finalizers with the prefix.
 func (c *Cluster) GetTenancyCount() int {
-	count := 0
+	return c.GetRequestUIDs().Len()
+}
+
+// GetRequestUIDs returns the UIDs of all ClusterRequests that have marked this cluster with a corresponding finalizer.
+func (c *Cluster) GetRequestUIDs() sets.Set[string] {
+	res := sets.New[string]()
 	for _, fin := range c.Finalizers {
 		if strings.HasPrefix(fin, RequestFinalizerOnClusterPrefix) {
-			count++
+			res.Insert(strings.TrimPrefix(fin, RequestFinalizerOnClusterPrefix))
 		}
 	}
-	return count
+	return res
 }
diff --git a/api/clusters/v1alpha1/clusterprofile_types.go b/api/clusters/v1alpha1/clusterprofile_types.go
index b39e669..1ea3025 100644
--- a/api/clusters/v1alpha1/clusterprofile_types.go
+++ b/api/clusters/v1alpha1/clusterprofile_types.go
@@ -6,9 +6,6 @@ import (
 
 // ClusterProfileSpec defines the desired state of Provider.
 type ClusterProfileSpec struct {
-	// Environment is the environment in which the ClusterProvider resides.
-	Environment string `json:"environment"`
-
 	// ProviderRef is a reference to the ClusterProvider
 	ProviderRef ObjectReference `json:"providerRef"`
 
@@ -30,11 +27,9 @@ type SupportedK8sVersion struct {
 
 // +kubebuilder:object:root=true
 // +kubebuilder:resource:scope=Cluster,shortName=cprof;profile
-// +kubebuilder:metadata:labels="openmcp.cloud/cluster=onboarding"
-// +kubebuilder:selectablefield:JSONPath=".spec.environment"
+// +kubebuilder:metadata:labels="openmcp.cloud/cluster=platform"
 // +kubebuilder:selectablefield:JSONPath=".spec.providerRef.name"
 // +kubebuilder:selectablefield:JSONPath=".spec.providerConfigRef.name"
-// +kubebuilder:printcolumn:JSONPath=".spec.environment",name="Env",type=string
 // +kubebuilder:printcolumn:JSONPath=".spec.providerRef.name",name="Provider",type=string
 // +kubebuilder:printcolumn:JSONPath=".spec.providerConfigRef.name",name="Config",type=string
 
diff --git a/api/clusters/v1alpha1/clusterrequest_types.go b/api/clusters/v1alpha1/clusterrequest_types.go
index 2034a2f..7407900 100644
--- a/api/clusters/v1alpha1/clusterrequest_types.go
+++ b/api/clusters/v1alpha1/clusterrequest_types.go
@@ -16,20 +16,35 @@ type ClusterRequestStatus struct {
 	CommonStatus `json:",inline"`
 
 	// Phase is the current phase of the request.
+	// +kubebuilder:default=Pending
+	// +kubebuilder:validation:Enum=Pending;Granted;Denied
 	Phase RequestPhase `json:"phase"`
 
-	// ClusterRef is the reference to the Cluster that was returned as a result of a granted request.
+	// Cluster is the reference to the Cluster that was returned as a result of a granted request.
 	// Note that this information needs to be recoverable in case this status is lost, e.g. by adding a back reference in form of a finalizer to the Cluster resource.
 	// +optional
 	// +kubebuilder:validation:XValidation:rule="self == oldSelf",message="clusterRef is immutable"
-	ClusterRef *NamespacedObjectReference `json:"clusterRef,omitempty"`
+	Cluster *NamespacedObjectReference `json:"clusterRef,omitempty"`
 }
 
 type RequestPhase string
 
+func (p RequestPhase) IsGranted() bool {
+	return p == REQUEST_GRANTED
+}
+
+func (p RequestPhase) IsDenied() bool {
+	return p == REQUEST_DENIED
+}
+
+func (p RequestPhase) IsPending() bool {
+	return p == "" || p == REQUEST_PENDING
+}
+
 // +kubebuilder:object:root=true
 // +kubebuilder:subresource:status
-// +kubebuilder:metadata:labels="openmcp.cloud/cluster=onboarding"
+// +kubebuilder:resource:shortName=cr;creq
+// +kubebuilder:metadata:labels="openmcp.cloud/cluster=platform"
 // +kubebuilder:selectablefield:JSONPath=".spec.purpose"
 // +kubebuilder:selectablefield:JSONPath=".status.phase"
 // +kubebuilder:printcolumn:JSONPath=".spec.purpose",name="Purpose",type=string
diff --git a/api/clusters/v1alpha1/constants.go b/api/clusters/v1alpha1/constants.go
index 464b58e..1ef2c2f 100644
--- a/api/clusters/v1alpha1/constants.go
+++ b/api/clusters/v1alpha1/constants.go
@@ -75,6 +75,10 @@ const (
 	EnvironmentAnnotation = "clusters.openmcp.cloud/environment"
 	// ProviderAnnotation can be used to display the provider of the cluster.
 	ProviderAnnotation = "clusters.openmcp.cloud/provider"
+
+	// DeleteWithoutRequestsLabel marks that the corresponding cluster can be deleted if the scheduler removes the last request pointing to it.
+	// Its value must be "true" for the label to take effect.
+	DeleteWithoutRequestsLabel = "clusters.openmcp.cloud/delete-without-requests"
 )
 
 const (
diff --git a/api/clusters/v1alpha1/zz_generated.deepcopy.go b/api/clusters/v1alpha1/zz_generated.deepcopy.go
index 6583174..9188737 100644
--- a/api/clusters/v1alpha1/zz_generated.deepcopy.go
+++ b/api/clusters/v1alpha1/zz_generated.deepcopy.go
@@ -72,6 +72,7 @@ func (in *AccessRequestList) DeepCopyObject() runtime.Object {
 func (in *AccessRequestSpec) DeepCopyInto(out *AccessRequestSpec) {
 	*out = *in
 	out.ClusterRef = in.ClusterRef
+	out.RequestRef = in.RequestRef
 	if in.Permissions != nil {
 		in, out := &in.Permissions, &out.Permissions
 		*out = make([]PermissionsRequest, len(*in))
@@ -95,6 +96,11 @@ func (in *AccessRequestSpec) DeepCopy() *AccessRequestSpec {
 func (in *AccessRequestStatus) DeepCopyInto(out *AccessRequestStatus) {
 	*out = *in
 	in.CommonStatus.DeepCopyInto(&out.CommonStatus)
+	if in.SecretRef != nil {
+		in, out := &in.SecretRef, &out.SecretRef
+		*out = new(NamespacedObjectReference)
+		**out = **in
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new AccessRequestStatus.
@@ -339,8 +345,8 @@ func (in *ClusterRequestSpec) DeepCopy() *ClusterRequestSpec {
 func (in *ClusterRequestStatus) DeepCopyInto(out *ClusterRequestStatus) {
 	*out = *in
 	in.CommonStatus.DeepCopyInto(&out.CommonStatus)
-	if in.ClusterRef != nil {
-		in, out := &in.ClusterRef, &out.ClusterRef
+	if in.Cluster != nil {
+		in, out := &in.Cluster, &out.Cluster
 		*out = new(NamespacedObjectReference)
 		**out = **in
 	}
diff --git a/api/crds/manifests/clusters.openmcp.cloud_accessrequests.yaml b/api/crds/manifests/clusters.openmcp.cloud_accessrequests.yaml
index 7d66bed..990fecc 100644
--- a/api/crds/manifests/clusters.openmcp.cloud_accessrequests.yaml
+++ b/api/crds/manifests/clusters.openmcp.cloud_accessrequests.yaml
@@ -5,7 +5,7 @@ metadata:
   annotations:
     controller-gen.kubebuilder.io/version: v0.17.3
   labels:
-    openmcp.cloud/cluster: onboarding
+    openmcp.cloud/cluster: platform
   name: accessrequests.clusters.openmcp.cloud
 spec:
   group: clusters.openmcp.cloud
@@ -13,10 +13,17 @@ spec:
     kind: AccessRequest
     listKind: AccessRequestList
     plural: accessrequests
+    shortNames:
+    - ar
+    - areq
     singular: accessrequest
   scope: Namespaced
   versions:
-  - name: v1alpha1
+  - additionalPrinterColumns:
+    - jsonPath: .status.phase
+      name: Phase
+      type: string
+    name: v1alpha1
     schema:
       openAPIV3Schema:
         description: AccessRequest is the Schema for the accessrequests API
@@ -125,9 +132,30 @@ spec:
                   - rules
                   type: object
                 type: array
+              requestRef:
+                description: |-
+                  RequestRef is the reference to the ClusterRequest for whose Cluster access is requested.
+                  Exactly one of clusterRef or requestRef must be set.
+                  This value is immutable.
+                properties:
+                  name:
+                    description: Name is the name of the referenced resource.
+                    minLength: 1
+                    type: string
+                  namespace:
+                    description: Namespace is the namespace of the referenced resource.
+                    type: string
+                required:
+                - name
+                - namespace
+                type: object
+                x-kubernetes-validations:
+                - message: requestRef is immutable
+                  rule: self == oldSelf
             required:
             - clusterRef
             - permissions
+            - requestRef
             type: object
           status:
             description: AccessRequestStatus defines the observed state of AccessRequest
@@ -181,18 +209,40 @@ spec:
                 format: int64
                 type: integer
               phase:
+                default: Pending
                 description: Phase is the current phase of the request.
+                enum:
+                - Pending
+                - Granted
+                - Denied
                 type: string
               reason:
                 description: Reason is expected to contain a CamelCased string that
                   provides further information in a machine-readable format.
                 type: string
+              secretRef:
+                description: SecretRef holds the reference to the secret that contains
+                  the actual credentials.
+                properties:
+                  name:
+                    description: Name is the name of the referenced resource.
+                    minLength: 1
+                    type: string
+                  namespace:
+                    description: Namespace is the namespace of the referenced resource.
+                    type: string
+                required:
+                - name
+                - namespace
+                type: object
             required:
             - lastReconcileTime
             - observedGeneration
             - phase
             type: object
         type: object
+    selectableFields:
+    - jsonPath: .status.phase
     served: true
     storage: true
     subresources:
diff --git a/api/crds/manifests/clusters.openmcp.cloud_clusterprofiles.yaml b/api/crds/manifests/clusters.openmcp.cloud_clusterprofiles.yaml
index efaca70..502e4e7 100644
--- a/api/crds/manifests/clusters.openmcp.cloud_clusterprofiles.yaml
+++ b/api/crds/manifests/clusters.openmcp.cloud_clusterprofiles.yaml
@@ -5,7 +5,7 @@ metadata:
   annotations:
     controller-gen.kubebuilder.io/version: v0.17.3
   labels:
-    openmcp.cloud/cluster: onboarding
+    openmcp.cloud/cluster: platform
   name: clusterprofiles.clusters.openmcp.cloud
 spec:
   group: clusters.openmcp.cloud
@@ -20,9 +20,6 @@ spec:
   scope: Cluster
   versions:
   - additionalPrinterColumns:
-    - jsonPath: .spec.environment
-      name: Env
-      type: string
     - jsonPath: .spec.providerRef.name
       name: Provider
       type: string
@@ -53,10 +50,6 @@ spec:
           spec:
             description: ClusterProfileSpec defines the desired state of Provider.
             properties:
-              environment:
-                description: Environment is the environment in which the ClusterProvider
-                  resides.
-                type: string
               providerConfigRef:
                 description: ProviderConfigRef is a reference to the provider-specific
                   configuration.
@@ -94,14 +87,12 @@ spec:
                   type: object
                 type: array
             required:
-            - environment
             - providerConfigRef
             - providerRef
             - supportedVersions
             type: object
         type: object
     selectableFields:
-    - jsonPath: .spec.environment
     - jsonPath: .spec.providerRef.name
     - jsonPath: .spec.providerConfigRef.name
     served: true
diff --git a/api/crds/manifests/clusters.openmcp.cloud_clusterrequests.yaml b/api/crds/manifests/clusters.openmcp.cloud_clusterrequests.yaml
index 5439540..4ccecd8 100644
--- a/api/crds/manifests/clusters.openmcp.cloud_clusterrequests.yaml
+++ b/api/crds/manifests/clusters.openmcp.cloud_clusterrequests.yaml
@@ -5,7 +5,7 @@ metadata:
   annotations:
     controller-gen.kubebuilder.io/version: v0.17.3
   labels:
-    openmcp.cloud/cluster: onboarding
+    openmcp.cloud/cluster: platform
   name: clusterrequests.clusters.openmcp.cloud
 spec:
   group: clusters.openmcp.cloud
@@ -13,6 +13,9 @@ spec:
     kind: ClusterRequest
     listKind: ClusterRequestList
     plural: clusterrequests
+    shortNames:
+    - cr
+    - creq
     singular: clusterrequest
   scope: Namespaced
   versions:
@@ -61,7 +64,7 @@ spec:
             properties:
               clusterRef:
                 description: |-
-                  ClusterRef is the reference to the Cluster that was returned as a result of a granted request.
+                  Cluster is the reference to the Cluster that was returned as a result of a granted request.
                   Note that this information needs to be recoverable in case this status is lost, e.g. by adding a back reference in form of a finalizer to the Cluster resource.
                 properties:
                   name:
@@ -127,7 +130,12 @@ spec:
                 format: int64
                 type: integer
               phase:
+                default: Pending
                 description: Phase is the current phase of the request.
+                enum:
+                - Pending
+                - Granted
+                - Denied
                 type: string
               reason:
                 description: Reason is expected to contain a CamelCased string that
diff --git a/api/crds/manifests/clusters.openmcp.cloud_clusters.yaml b/api/crds/manifests/clusters.openmcp.cloud_clusters.yaml
index f16c32d..c6dc363 100644
--- a/api/crds/manifests/clusters.openmcp.cloud_clusters.yaml
+++ b/api/crds/manifests/clusters.openmcp.cloud_clusters.yaml
@@ -5,7 +5,7 @@ metadata:
   annotations:
     controller-gen.kubebuilder.io/version: v0.17.3
   labels:
-    openmcp.cloud/cluster: onboarding
+    openmcp.cloud/cluster: platform
   name: clusters.clusters.openmcp.cloud
 spec:
   group: clusters.openmcp.cloud
@@ -29,15 +29,11 @@ spec:
     - jsonPath: .metadata.annotations["clusters.openmcp.cloud/profile"]
       name: Profile
       type: string
-    - jsonPath: .metadata.labels["environment.clusters.openmcp.cloud"]
-      name: Env
-      priority: 10
-      type: string
     - jsonPath: .metadata.labels["provider.clusters.openmcp.cloud"]
       name: Provider
       priority: 10
       type: string
-    - jsonPath: .spec.clusterProfileRef.name
+    - jsonPath: .spec.profile
       name: ProfileRef
       priority: 10
       type: string
@@ -194,7 +190,7 @@ spec:
             type: object
         type: object
     selectableFields:
-    - jsonPath: .spec.clusterProfileRef.name
+    - jsonPath: .spec.profile
     served: true
     storage: true
     subresources:
diff --git a/api/install/install.go b/api/install/install.go
index 00d0856..0126953 100644
--- a/api/install/install.go
+++ b/api/install/install.go
@@ -1,11 +1,13 @@
 package install
 
 import (
-	providerv1alpha1 "github.com/openmcp-project/openmcp-operator/api/provider/v1alpha1"
 	apiextv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
+
+	clustersv1alpha1 "github.com/openmcp-project/openmcp-operator/api/clusters/v1alpha1"
+	providerv1alpha1 "github.com/openmcp-project/openmcp-operator/api/provider/v1alpha1"
 )
 
 // InstallCRDAPIs installs the CRD APIs in the scheme.
@@ -20,6 +22,7 @@ func InstallCRDAPIs(scheme *runtime.Scheme) *runtime.Scheme {
 func InstallOperatorAPIs(scheme *runtime.Scheme) *runtime.Scheme {
 	utilruntime.Must(clientgoscheme.AddToScheme(scheme))
 	utilruntime.Must(providerv1alpha1.AddToScheme(scheme))
+	utilruntime.Must(clustersv1alpha1.AddToScheme(scheme))
 
 	return scheme
 }
diff --git a/cmd/openmcp-operator/app/run.go b/cmd/openmcp-operator/app/run.go
index a2c1a32..2a72c00 100644
--- a/cmd/openmcp-operator/app/run.go
+++ b/cmd/openmcp-operator/app/run.go
@@ -26,10 +26,11 @@ import (
 	"github.com/openmcp-project/openmcp-operator/api/install"
 	"github.com/openmcp-project/openmcp-operator/api/provider/v1alpha1"
 	"github.com/openmcp-project/openmcp-operator/internal/controllers/provider"
+	"github.com/openmcp-project/openmcp-operator/internal/controllers/scheduler"
 )
 
 var setupLog logging.Logger
-var allControllers = []string{}
+var allControllers = []string{strings.ToLower(scheduler.ControllerName)}
 
 func NewRunCommand(so *SharedOptions) *cobra.Command {
 	opts := &RunOptions{
@@ -73,7 +74,7 @@ func (o *RunOptions) AddFlags(cmd *cobra.Command) {
 	cmd.Flags().StringVar(&o.MetricsCertKey, "metrics-cert-key", "tls.key", "The name of the metrics server key file.")
 	cmd.Flags().BoolVar(&o.EnableHTTP2, "enable-http2", false, "If set, HTTP/2 will be enabled for the metrics and webhook servers")
 
-	cmd.Flags().StringSliceVar(&o.Controllers, "controllers", allControllers, "List of active controllers.")
+	cmd.Flags().StringSliceVar(&o.Controllers, "controllers", allControllers, "List of active controllers. Separate with comma or specify flag multiple times to activate multiple controllers.")
 }
 
 type RawRunOptions struct {
@@ -284,14 +285,20 @@ func (o *RunOptions) Run(ctx context.Context) error {
 		return fmt.Errorf("unable to add onboarding cluster to manager: %w", err)
 	}
 
-	// setup cluster controller
-	// if slices.Contains(o.Controllers, strings.ToLower(cluster.ControllerName)) {
-	// 	if _, _, _, err := controllers.SetupClusterControllersWithManager(mgr, o.Clusters.Platform, o.Clusters.Onboarding, swMgr); err != nil {
-	// 		return fmt.Errorf("unable to setup cluster controllers: %w", err)
-	// 	}
-	// }
+	// setup cluster scheduler
+	if slices.Contains(o.Controllers, strings.ToLower(scheduler.ControllerName)) {
+		sc, err := scheduler.NewClusterScheduler(&setupLog, o.Clusters.Platform, o.Config.Scheduler)
+		if err != nil {
+			return fmt.Errorf("unable to initialize cluster scheduler: %w", err)
+		}
+		if err := sc.SetupWithManager(mgr); err != nil {
+			return fmt.Errorf("unable to setup cluster scheduler with manager: %w", err)
+		}
+	}
 
 	// setup deployment controller
+	// TODO: Can we use a variable/constant/function instead of a hardcoded string for the controller name here?
+	// TODO: This value has to be added to the allControllers variable too, because I guess we want it to be enabled by default.
 	if slices.Contains(o.Controllers, strings.ToLower("deploymentcontroller")) {
 		utilruntime.Must(clientgoscheme.AddToScheme(mgr.GetScheme()))
 		utilruntime.Must(api.AddToScheme(mgr.GetScheme()))
diff --git a/docs/README.md b/docs/README.md
index 8dc010a..05b077e 100644
--- a/docs/README.md
+++ b/docs/README.md
@@ -1,6 +1,12 @@
 <!-- Do not edit this file, as it is auto-generated!-->
 # Documentation Index
 
+## Controller
+
+- [Cluster Scheduler](controller/scheduler.md)
+
 ## Resources
 
+- [Cluster Provider: Gardener [Resource Example]](resources/cluster-provider-gardener.md)
+- [Service Provider: Landscaper [Resource Example]](resources/service-provider-landscaper.md)
 
diff --git a/docs/controller/.docnames b/docs/controller/.docnames
new file mode 100644
index 0000000..8b49e1e
--- /dev/null
+++ b/docs/controller/.docnames
@@ -0,0 +1,3 @@
+{
+    "header": "Controller"
+}
\ No newline at end of file
diff --git a/docs/controller/scheduler.md b/docs/controller/scheduler.md
new file mode 100644
index 0000000..592b3e5
--- /dev/null
+++ b/docs/controller/scheduler.md
@@ -0,0 +1,122 @@
+# Cluster Scheduler
+
+The _Cluster Scheduler_ that is part of the openMCP Operator is responsible for answering `ClusterRequest` resources by either creating new `Cluster` resources or referencing existing ones.
+
+## Configuration
+
+To disable the scheduler, make sure that `scheduler` is not part of the `--controllers` flag when running the binary. It is included by default.
+
+If not disabled, the scheduler requires a config that looks like this:
+```yaml
+scheduler:
+  scope: Namespaced  # optional
+  strategy: Balanced # optional
+
+  selectors:                  # optional
+    clusters:                 # optional
+      matchLabels: <...>      # optional
+      matchExpressions: <...> # optional
+    requests:                 # optional
+      matchLabels: <...>      # optional
+      matchExpressions: <...> # optional
+
+  purposeMappings:
+    mcp:
+      selector:                 # optional
+        matchLabels: <...>      # optional
+        matchExpressions: <...> # optional
+      template:
+        metadata:
+          namespace: mcp-clusters
+        spec:
+          profile: gcp-workerless
+          tenancy: Exclusive
+    platform:
+      template:
+        metadata:
+          labels:
+            clusters.openmcp.cloud/delete-without-requests: "false"
+        spec:
+          profile: gcp-large
+          tenancy: Shared
+    onboarding:
+      template:
+        metadata:
+          labels:
+            clusters.openmcp.cloud/delete-without-requests: "false"
+        spec:
+          profile: gcp-workerless
+          tenancy: Shared
+    workload:
+      tenancyCount: 20
+      template:
+        metadata:
+          namespace: workload-clusters
+        spec:
+          profile: gcp-small
+          tenancy: Shared
+```
+
+The following fields can be specified inside the `scheduler` field:
+- `scope` _(optional, defaults to `Namespaced`)_
+  - Valid values: `Namespaced`, `Cluster`
+  - Determines whether the scheduler takes `Cluster` resources in all namespaces into accounts or only in a specific one.
+    - In `Namespaced` mode, only `Cluster` resources from a single namespace are taken into account when checking for existing clusters to schedule requests to. If the cluster template that corresponds to the purpose specified in the request has a `metadata.namespace` set, this namespace is used to check for `Cluster` resources and also to create new ones. If not, the namespace of the `ClusterRequest` resource is used instead.
+    - In `Cluster` mode, the scheduler takes all clusters into account when trying to find existing clusters that can be reused. New clusters are still created in the namespace specified in the cluster template, or in the request's namespace, if the former one is not set.
+- `strategy` _(optional, defaults to `Balanced`)_
+  - Valid values: `Balanced`, `Random`, `Simple`
+  - Determines how the scheduler chooses a cluster if multiple existing ones qualify for a request.
+    - With the `Balanced` strategy, the scheduler chooses the cluster with the fewest requests pointing to it. In case of a tie, the first one is chosen.
+    - With the `Random` strategy, a cluster is chosen randomly.
+    - With the `Simple` strategy, the first cluster in the list (should be in alphabetical order) is chosen.
+- `selectors.clusters` _(optional)_
+  - A label selector that restricts which `Cluster` resources are evaluated by the scheduler. Clusters that don't match the selector are treated as if they didn't exist.
+    - The selector syntax is the default k8s one, as it is used in `Deployment` resources, for example.
+    - Validation of the configuration will fail if any of the cluster templates from the `purposeMappings` field does not match the selector - this would otherwise result in the scheduler creating clusters that it would be unable to find again later on.
+    - Note that the scheduler might run into naming conflicts with existing `Cluster` resources that don't match the selectors. See below for further information on how new clusters are named.
+  - Selectors specified in `selectors.clusters` apply to all `Cluster` resources, while selectors in `purposeMappings[*].selector` are only applied for that specific purpose.
+- `selectors.requests` _(optional)_
+  - A label selector that restricts which `ClusterRequest` resources are reconciled by the scheduler. Requests that don't match the selector are not reconciled.
+    - The selector syntax is the default k8s one, as it is used in `Deployment` resources, for example.
+- `purposeMappings`
+  - This is a map where each entry maps a purpose to a cluster template and some additional information. When a `ClusterRequest` is reconciled, its `spec.purpose` is looked up in this map and the result determines how existing clusters are evaluated and how new ones are created.
+  - The structure for a mapping is explained below.
+
+Each value of a purpose mapping takes the following fields:
+- `template`
+  - A `Cluster` template, consisting of `metadata` (optional) and `spec`. This is used when new clusters are created, but it is also partly evaluated when checking for existing ones.
+    - `metadata.name` and `metadata.generateName` can be used to influence how newly created clusters are named. See below for further explanation on cluster naming.
+    - `metadata.namespace` is the namespace in which newly created clusters for this purpose are created. If empty, the request's namespace is used instead. If the scheduler runs in `Namespaced` mode, this is also the only namespace that is evaluated when checking for existing clusters (again falling back to the request's namespace, if not specified). In `Cluster` mode, existing clusters are checked across all namespaces.
+    - `metadata.labels` and `metadata.annotations` are simply passed to newly created clusters. If label selectors for clusters are specified, `metadata.labels` has to satisfy the selectors, otherwise the validation will fail.
+    - `spec.profile` and `spec.tenancy` have to be set. The latter value determines whether the cluster is for the creating request exlusively (`Exclusive`) or whether other requests are allowed to point to the same cluster (`Shared`).
+    - If `spec.purposes` does not contain the purpose this template is mapped to, it will be added during creation of the `Cluster`.
+- `tenancyCount` _(optional, defaults to `0`)_
+  - This value specifies how many requests may point to a cluster with this purpose.
+    - If `template.spec.tenancy` is `Exclusive`, this value has to be `0` and does not have any effect.
+    - If `template.spec.tenancy` is `Shared`, this value must be equal to or greater than `0`.
+      - If greater than `0`, this is the amount of requests that may point to the same cluster.
+        - A value of `1` behaves similar to an exclusive cluster, but the cluster is marked as shared and other requests may refer to it at a later point, if the value is increased or the scheduler logic is changed.
+      - If `0`, the cluster is shared with an unlimited amount of requests that may refer to it. This basically means that there will ever be only one cluster for this purpose and all requests with this purpose will refer to this one cluster.
+        - Note that 'one cluster' means 'within the boundaries specified by namespace and label selectors'. If the scheduler runs in `Namespaced` mode and the template does not specify a namespace, for example, one cluster per namespace will be created, not one in total.
+- `selector` _(optional)_
+  - A label selector that is used to filter `Cluster` resources when checking for existing clusters for this purpose. This is merged with any selectors from `selectors.clusters`.
+
+## Names and Namespaces of Clusters created by the Scheduler
+
+If the scheduler needs to create a new `Cluster` resource, because none of the existing ones fits the criteria for a `ClusterRequest`, it has to choose name and namespace for the `Cluster` resource. This is done according to the following logic:
+
+#### Namespace
+
+If the cluster template from the configuration for the requested purpose has `metadata.namespace` set, that is used as namespace. Otherwise, the cluster is created in the same namespace as the `ClusterRequest` that caused its creation.
+
+#### Name
+
+For clusters with `Exclusive` tenancy, or for `Shared` ones with a limited tenancy count, the scheduler uses `metadata.generateName` from the cluster template or defaults it to `<purpose>-`, if not set.
+
+For clusters with unlimited tenancy count, `metadata.generateName` takes precedence, if specified in the template, with `metadata.name` being evaluated second. If neither is specified, `<purpose>` is used as `metadata.name` (as there should be only one instance of this cluster in this namespace due to the unlimited tenancy count, there is no need to add a randomized suffix to the name).
+
+## Deletion of Clusters
+
+By default, the scheduler marks every `Cluster` that it has created itself with a `clusters.openmcp.cloud/delete-without-requests: "true"` label. When a `ClusterRequest` is deleted, the scheduler removes the request's finalizers from all clusters and if it was the last request finalizer on that cluster and the cluster has the aforementioned label, the scheduler will delete the cluster.
+
+To prevent the scheduler from deleting a cluster that was created by it after the last request finalizer has been removed from the `Cluster` resource, add the label with any value except `"true"` to the cluster's template in the scheduler configuration.
diff --git a/docs/resources/cluster-provider-gardener.yaml b/docs/resources/cluster-provider-gardener.md
similarity index 77%
rename from docs/resources/cluster-provider-gardener.yaml
rename to docs/resources/cluster-provider-gardener.md
index e89fd23..cb88ae8 100644
--- a/docs/resources/cluster-provider-gardener.yaml
+++ b/docs/resources/cluster-provider-gardener.md
@@ -1,3 +1,6 @@
+# Cluster Provider: Gardener [Resource Example]
+
+```yaml
 apiVersion: openmcp.cloud/v1alpha1
 kind: ClusterProvider
 metadata:
@@ -5,3 +8,4 @@ metadata:
 spec:
   image: "ghcr.io/openmcp-project/images/cluster-provider-gardener:v0.0.1"
   imagePullSecrets: []
+```
diff --git a/docs/resources/service-provider-landscaper.yaml b/docs/resources/service-provider-landscaper.md
similarity index 76%
rename from docs/resources/service-provider-landscaper.yaml
rename to docs/resources/service-provider-landscaper.md
index 4709c70..c438582 100644
--- a/docs/resources/service-provider-landscaper.yaml
+++ b/docs/resources/service-provider-landscaper.md
@@ -1,3 +1,6 @@
+# Service Provider: Landscaper [Resource Example]
+
+```yaml
 apiVersion: openmcp.cloud/v1alpha1
 kind: ServiceProvider
 metadata:
@@ -5,3 +8,4 @@ metadata:
 spec:
   image: "ghcr.io/openmcp-project/images/service-provider-landscaper:v0.0.4"
   imagePullSecrets: []
+```
diff --git a/internal/config/config.go b/internal/config/config.go
index ab01b42..61ffdeb 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -23,8 +23,12 @@ type Config struct {
 	// If a field's value has a 'Complete() error' method, it will be called to complete the config.
 	// These methods will be called in the order Default -> Validate -> Complete.
 	// The config printed during startup, therefore its fields should contain json markers.
+
+	// Scheduler is the configuration for the cluster scheduler.
+	Scheduler *SchedulerConfig `json:"scheduler,omitempty"`
 }
 
+// Dump is used for logging and debugging purposes.
 func (c *Config) Dump(out io.Writer) error {
 	data, err := yaml.Marshal(c)
 	if err != nil {
diff --git a/internal/config/config_scheduler.go b/internal/config/config_scheduler.go
new file mode 100644
index 0000000..0f6165a
--- /dev/null
+++ b/internal/config/config_scheduler.go
@@ -0,0 +1,223 @@
+package config
+
+import (
+	"fmt"
+	"maps"
+	"slices"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/util/validation/field"
+
+	clustersv1alpha1 "github.com/openmcp-project/openmcp-operator/api/clusters/v1alpha1"
+)
+
+type SchedulerConfig struct {
+	// Scope determines whether the scheduler considers all clusters or only the ones in the same namespace as the ClusterRequest.
+	// Defaults to "Namespaced".
+	Scope SchedulerScope `json:"scope"`
+
+	// Strategy determines how the scheduler chooses between multiple fitting clusters:
+	// - Random: chooses a random cluster
+	// - Simple: chooses the first cluster in the list
+	// - Balanced: chooses the cluster with the least number of requests (first one in case of a tie)
+	// Defaults to "Balanced".
+	Strategy Strategy `json:"strategy"`
+
+	// +optional
+	Selectors *SchedulerSelectors `json:"selectors,omitempty"`
+	// Note that CompletedSelectors.Clusters holds the global cluster selector.
+	// During Complete(), the local selector is merged with the global one (or set to the global one if nil).
+	// This means that always the local completed selector should be used, unless the task is not tied to a specific ClusterDefinition.
+	CompletedSelectors CompletedSchedulerSelectors `json:"-"`
+
+	PurposeMappings map[string]*ClusterDefinition `json:"purposeMappings"`
+}
+
+type SchedulerScope string
+
+const (
+	SCOPE_CLUSTER    SchedulerScope = "Cluster"
+	SCOPE_NAMESPACED SchedulerScope = "Namespaced"
+)
+
+type Strategy string
+
+const (
+	STRATEGY_BALANCED Strategy = "Balanced"
+	STRATEGY_RANDOM   Strategy = "Random"
+	STRATEGY_SIMPLE   Strategy = "Simple"
+)
+
+type ClusterDefinition struct {
+	// TenancyCount determines how many ClusterRequests may point to the same Cluster.
+	// Has no effect if the tenancy in the Cluster template is set to "Exclusive".
+	// Must be equal to or greater than 0 otherwise, with 0 meaning "unlimited".
+	TenancyCount int `json:"tenancyCount,omitempty"`
+
+	Template          ClusterTemplate       `json:"template"`
+	Selector          *metav1.LabelSelector `json:"selector,omitempty"`
+	CompletedSelector labels.Selector       `json:"-"`
+}
+
+type ClusterTemplate struct {
+	metav1.ObjectMeta `json:"metadata"`
+	Spec              clustersv1alpha1.ClusterSpec `json:"spec"`
+}
+
+type SchedulerSelectors struct {
+	Clusters *metav1.LabelSelector `json:"clusters,omitempty"`
+	Requests *metav1.LabelSelector `json:"requests,omitempty"`
+}
+type CompletedSchedulerSelectors struct {
+	Clusters labels.Selector
+	Requests labels.Selector
+}
+
+func (c *SchedulerConfig) Default(_ *field.Path) error {
+	if c.Scope == "" {
+		c.Scope = SCOPE_NAMESPACED
+	}
+	if c.Strategy == "" {
+		c.Strategy = STRATEGY_BALANCED
+	}
+	if c.PurposeMappings == nil {
+		c.PurposeMappings = map[string]*ClusterDefinition{}
+	}
+	return nil
+}
+
+func (c *SchedulerConfig) Validate(fldPath *field.Path) error {
+	errs := field.ErrorList{}
+
+	// validate scope and strategy
+	validScopes := []string{string(SCOPE_CLUSTER), string(SCOPE_NAMESPACED)}
+	if !slices.Contains(validScopes, string(c.Scope)) {
+		errs = append(errs, field.NotSupported(fldPath.Child("scope"), string(c.Scope), validScopes))
+	}
+	validStrategies := []string{string(STRATEGY_BALANCED), string(STRATEGY_RANDOM), string(STRATEGY_SIMPLE)}
+	if !slices.Contains(validStrategies, string(c.Strategy)) {
+		errs = append(errs, field.NotSupported(fldPath.Child("strategy"), string(c.Strategy), validStrategies))
+	}
+
+	// validate label selectors
+	var cls labels.Selector
+	if c.Selectors != nil {
+		if c.Selectors.Clusters != nil {
+			var err error
+			cls, err = metav1.LabelSelectorAsSelector(c.Selectors.Clusters)
+			if err != nil {
+				errs = append(errs, field.Invalid(fldPath.Child("selectors").Child("clusters"), c.Selectors.Clusters, err.Error()))
+			}
+		}
+		if c.Selectors.Requests != nil {
+			_, err := metav1.LabelSelectorAsSelector(c.Selectors.Requests)
+			if err != nil {
+				errs = append(errs, field.Invalid(fldPath.Child("selectors").Child("requests"), c.Selectors.Requests, err.Error()))
+			}
+		}
+	}
+
+	// validate purpose mappings
+	validTenancies := []string{string(clustersv1alpha1.TENANCY_EXCLUSIVE), string(clustersv1alpha1.TENANCY_SHARED)}
+	fldPath = fldPath.Child("purposeMappings")
+	for purpose, definition := range c.PurposeMappings {
+		pPath := fldPath.Key(purpose)
+		if purpose == "" {
+			errs = append(errs, field.Invalid(fldPath, purpose, "purpose must not be empty"))
+		}
+		if definition == nil {
+			errs = append(errs, field.Required(pPath, "definition must not be nil"))
+			continue
+		}
+		if definition.TenancyCount < 0 {
+			errs = append(errs, field.Invalid(pPath.Child("tenancyCount"), definition.TenancyCount, "tenancyCount must be greater than or equal to 0"))
+		}
+		if definition.Template.Spec.Profile == "" {
+			errs = append(errs, field.Required(pPath.Child("template").Child("spec").Child("profile"), definition.Template.Spec.Profile))
+		}
+		if definition.Template.Spec.Tenancy == "" {
+			errs = append(errs, field.Required(pPath.Child("template").Child("spec").Child("tenancy"), string(definition.Template.Spec.Tenancy)))
+			continue
+		} else if !slices.Contains(validTenancies, string(definition.Template.Spec.Tenancy)) {
+			errs = append(errs, field.NotSupported(pPath.Child("template").Child("spec").Child("tenancy"), string(definition.Template.Spec.Tenancy), validTenancies))
+			continue
+		}
+		if definition.Template.Spec.Tenancy == clustersv1alpha1.TENANCY_EXCLUSIVE && definition.TenancyCount != 0 {
+			errs = append(errs, field.Invalid(pPath.Child("tenancyCount"), definition.TenancyCount, fmt.Sprintf("tenancyCount must be 0 if the template specifies '%s' tenancy", string(clustersv1alpha1.TENANCY_EXCLUSIVE))))
+		}
+		if cls != nil && !cls.Matches(labels.Set(definition.Template.Labels)) {
+			errs = append(errs, field.Invalid(pPath.Child("template").Child("metadata").Child("labels"), definition.Template.Labels, "labels do not match specified global cluster selector"))
+		}
+		var lcls labels.Selector
+		if definition.Selector != nil {
+			var err error
+			lcls, err = metav1.LabelSelectorAsSelector(definition.Selector)
+			if err != nil {
+				errs = append(errs, field.Invalid(pPath.Child("selector"), definition.Selector, err.Error()))
+			}
+		}
+		if lcls != nil && !lcls.Matches(labels.Set(definition.Template.Labels)) {
+			errs = append(errs, field.Invalid(pPath.Child("template").Child("metadata").Child("labels"), definition.Template.Labels, "labels do not match specified local cluster selector"))
+		}
+	}
+	return errs.ToAggregate()
+}
+
+func (c *SchedulerConfig) Complete(fldPath *field.Path) error {
+	if c.Selectors != nil {
+		if c.Selectors.Clusters != nil {
+			var err error
+			c.CompletedSelectors.Clusters, err = metav1.LabelSelectorAsSelector(c.Selectors.Clusters)
+			if err != nil {
+				return field.Invalid(fldPath.Child("selectors").Child("clusters"), c.Selectors.Clusters, err.Error())
+			}
+		}
+		if c.Selectors.Requests != nil {
+			var err error
+			c.CompletedSelectors.Requests, err = metav1.LabelSelectorAsSelector(c.Selectors.Requests)
+			if err != nil {
+				return field.Invalid(fldPath.Child("selectors").Child("requests"), c.Selectors.Requests, err.Error())
+			}
+		}
+	}
+	if c.CompletedSelectors.Clusters == nil {
+		c.CompletedSelectors.Clusters = labels.Everything()
+	}
+	if c.CompletedSelectors.Requests == nil {
+		c.CompletedSelectors.Requests = labels.Everything()
+	}
+
+	for purpose, definition := range c.PurposeMappings {
+		pPath := fldPath.Child("purposeMappings").Key(purpose)
+		if definition.Selector != nil {
+			var combinedSelector *metav1.LabelSelector
+			if c.Selectors.Clusters == nil {
+				combinedSelector = definition.Selector
+			} else if definition.Selector == nil {
+				combinedSelector = c.Selectors.Clusters
+			} else {
+				combinedSelector = c.Selectors.Clusters.DeepCopy()
+				if combinedSelector.MatchLabels == nil {
+					combinedSelector.MatchLabels = definition.Selector.MatchLabels
+				} else if definition.Selector.MatchLabels != nil {
+					maps.Insert(combinedSelector.MatchLabels, maps.All(definition.Selector.MatchLabels))
+				}
+				if combinedSelector.MatchExpressions == nil {
+					combinedSelector.MatchExpressions = definition.Selector.MatchExpressions
+				} else if definition.Selector.MatchExpressions != nil {
+					combinedSelector.MatchExpressions = append(combinedSelector.MatchExpressions, definition.Selector.MatchExpressions...)
+				}
+			}
+			var err error
+			definition.CompletedSelector, err = metav1.LabelSelectorAsSelector(combinedSelector)
+			if err != nil {
+				return field.Invalid(pPath.Child("selector"), combinedSelector, fmt.Sprintf("the combination of the global and local selector is invalid: %s", err.Error()))
+			}
+		} else {
+			definition.CompletedSelector = c.CompletedSelectors.Clusters
+		}
+	}
+
+	return nil
+}
diff --git a/internal/controllers/scheduler/controller.go b/internal/controllers/scheduler/controller.go
new file mode 100644
index 0000000..6332aa2
--- /dev/null
+++ b/internal/controllers/scheduler/controller.go
@@ -0,0 +1,433 @@
+package scheduler
+
+import (
+	"context"
+	"fmt"
+	"math/rand/v2"
+	"slices"
+	"strings"
+
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/util/sets"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+
+	"github.com/openmcp-project/controller-utils/pkg/clusters"
+	"github.com/openmcp-project/controller-utils/pkg/collections/filters"
+	ctrlutils "github.com/openmcp-project/controller-utils/pkg/controller"
+	errutils "github.com/openmcp-project/controller-utils/pkg/errors"
+	"github.com/openmcp-project/controller-utils/pkg/logging"
+
+	clustersv1alpha1 "github.com/openmcp-project/openmcp-operator/api/clusters/v1alpha1"
+	cconst "github.com/openmcp-project/openmcp-operator/api/clusters/v1alpha1/constants"
+	"github.com/openmcp-project/openmcp-operator/internal/config"
+)
+
+const ControllerName = "Scheduler"
+
+func NewClusterScheduler(setupLog *logging.Logger, platformCluster *clusters.Cluster, config *config.SchedulerConfig) (*ClusterScheduler, error) {
+	if platformCluster == nil {
+		return nil, fmt.Errorf("onboarding cluster must not be nil")
+	}
+	if config == nil {
+		return nil, fmt.Errorf("scheduler config must not be nil")
+	}
+	if setupLog != nil {
+		setupLog.WithName(ControllerName).Info("Initializing cluster scheduler", "scope", string(config.Scope), "strategy", string(config.Strategy), "knownPurposes", strings.Join(sets.List(sets.KeySet(config.PurposeMappings)), ","))
+	}
+	return &ClusterScheduler{
+		PlatformCluster: platformCluster,
+		Config:          config,
+	}, nil
+}
+
+type ClusterScheduler struct {
+	PlatformCluster *clusters.Cluster
+	Config          *config.SchedulerConfig
+}
+
+var _ reconcile.Reconciler = &ClusterScheduler{}
+
+type ReconcileResult = ctrlutils.ReconcileResult[*clustersv1alpha1.ClusterRequest, clustersv1alpha1.ConditionStatus]
+
+func (r *ClusterScheduler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
+	log := logging.FromContextOrPanic(ctx).WithName(ControllerName)
+	ctx = logging.NewContext(ctx, log)
+	log.Info("Starting reconcile")
+	rr := r.reconcile(ctx, log, req)
+	// status update
+	return ctrlutils.NewStatusUpdaterBuilder[*clustersv1alpha1.ClusterRequest, clustersv1alpha1.RequestPhase, clustersv1alpha1.ConditionStatus]().
+		WithNestedStruct("CommonStatus").
+		WithFieldOverride(ctrlutils.STATUS_FIELD_PHASE, "Phase").
+		WithoutFields(ctrlutils.STATUS_FIELD_CONDITIONS).
+		WithPhaseUpdateFunc(func(obj *clustersv1alpha1.ClusterRequest, rr ReconcileResult) (clustersv1alpha1.RequestPhase, error) {
+			if rr.ReconcileError != nil || rr.Object == nil || rr.Object.Status.Cluster == nil {
+				return clustersv1alpha1.REQUEST_PENDING, nil
+			}
+			return clustersv1alpha1.REQUEST_GRANTED, nil
+		}).
+		Build().
+		UpdateStatus(ctx, r.PlatformCluster.Client(), rr)
+}
+
+func (r *ClusterScheduler) reconcile(ctx context.Context, log logging.Logger, req reconcile.Request) ReconcileResult {
+	// get ClusterRequest resource
+	cr := &clustersv1alpha1.ClusterRequest{}
+	if err := r.PlatformCluster.Client().Get(ctx, req.NamespacedName, cr); err != nil {
+		if apierrors.IsNotFound(err) {
+			log.Info("Resource not found")
+			return ReconcileResult{}
+		}
+		return ReconcileResult{ReconcileError: errutils.WithReason(fmt.Errorf("unable to get resource '%s' from cluster: %w", req.NamespacedName.String(), err), cconst.ReasonPlatformClusterInteractionProblem)}
+	}
+
+	// handle operation annotation
+	if cr.GetAnnotations() != nil {
+		op, ok := cr.GetAnnotations()[clustersv1alpha1.OperationAnnotation]
+		if ok {
+			switch op {
+			case clustersv1alpha1.OperationAnnotationValueIgnore:
+				log.Info("Ignoring resource due to ignore operation annotation")
+				return ReconcileResult{}
+			case clustersv1alpha1.OperationAnnotationValueReconcile:
+				log.Debug("Removing reconcile operation annotation from resource")
+				if err := ctrlutils.EnsureAnnotation(ctx, r.PlatformCluster.Client(), cr, clustersv1alpha1.OperationAnnotation, "", true, ctrlutils.DELETE); err != nil {
+					return ReconcileResult{ReconcileError: errutils.WithReason(fmt.Errorf("error removing operation annotation: %w", err), cconst.ReasonPlatformClusterInteractionProblem)}
+				}
+			}
+		}
+	}
+
+	inDeletion := !cr.DeletionTimestamp.IsZero()
+	var rr ReconcileResult
+	if !inDeletion {
+		rr = r.handleCreateOrUpdate(ctx, req, cr)
+	} else {
+		rr = r.handleDelete(ctx, req, cr)
+	}
+
+	return rr
+}
+
+func (r *ClusterScheduler) handleCreateOrUpdate(ctx context.Context, req reconcile.Request, cr *clustersv1alpha1.ClusterRequest) ReconcileResult {
+	log := logging.FromContextOrPanic(ctx)
+	rr := ReconcileResult{
+		Object:    cr,
+		OldObject: cr.DeepCopy(),
+	}
+
+	log.Info("Creating/updating resource")
+
+	// ensure finalizer
+	if controllerutil.AddFinalizer(cr, clustersv1alpha1.ClusterRequestFinalizer) {
+		log.Info("Adding finalizer")
+		if err := r.PlatformCluster.Client().Patch(ctx, cr, client.MergeFrom(rr.OldObject)); err != nil {
+			rr.ReconcileError = errutils.WithReason(fmt.Errorf("error patching finalizer on resource '%s': %w", req.NamespacedName.String(), err), cconst.ReasonPlatformClusterInteractionProblem)
+			return rr
+		}
+	}
+
+	// check if request is already granted
+	if cr.Status.Cluster != nil {
+		log.Info("Request already contains a cluster reference, nothing to do", "clusterName", cr.Status.Cluster.Name, "clusterNamespace", cr.Status.Cluster.Namespace)
+		return rr
+	}
+	log.Debug("Request status does not contain a cluster reference, checking for existing clusters with referencing finalizers")
+
+	// fetch cluster definition
+	purpose := cr.Spec.Purpose
+	cDef, ok := r.Config.PurposeMappings[purpose]
+	if !ok {
+		rr.ReconcileError = errutils.WithReason(fmt.Errorf("no cluster definition found for purpose '%s'", purpose), cconst.ReasonConfigurationProblem)
+		return rr
+	}
+
+	clusters, rerr := r.fetchRelevantClusters(ctx, cr, cDef)
+	if rerr != nil {
+		rr.ReconcileError = rerr
+		return rr
+	}
+
+	// check if status was lost, but there exists a cluster that was already assigned to this request
+	reqFin := cr.FinalizerForCluster()
+	var cluster *clustersv1alpha1.Cluster
+	for _, c := range clusters {
+		if slices.Contains(c.Finalizers, reqFin) {
+			cluster = c
+			break
+		}
+	}
+	if cluster != nil {
+		log.Info("Recovered cluster from referencing finalizer", "clusterName", cluster.Name, "clusterNamespace", cluster.Namespace)
+		rr.Object.Status.Cluster = &clustersv1alpha1.NamespacedObjectReference{}
+		rr.Object.Status.Cluster.Name = cluster.Name
+		rr.Object.Status.Cluster.Namespace = cluster.Namespace
+		return rr
+	}
+
+	if cDef.Template.Spec.Tenancy == clustersv1alpha1.TENANCY_SHARED {
+		log.Debug("Cluster template allows sharing, checking for fitting clusters", "purpose", purpose, "tenancyCount", cDef.TenancyCount)
+		// unless the cluster template for the requested purpose allows unlimited sharing, filter out all clusters that are already at their tenancy limit
+		if cDef.TenancyCount > 0 {
+			clusters = filters.FilterSlice(clusters, func(args ...any) bool {
+				c, ok := args[0].(*clustersv1alpha1.Cluster)
+				if !ok {
+					return false
+				}
+				return c.GetTenancyCount() < cDef.TenancyCount
+			})
+		}
+		if len(clusters) == 1 {
+			cluster = clusters[0]
+			log.Debug("One existing cluster qualifies for request", "clusterName", cluster.Name, "clusterNamespace", cluster.Namespace)
+		} else if len(clusters) > 0 {
+			log.Debug("Multiple existing clusters qualify for request, choosing one according to strategy", "strategy", string(r.Config.Strategy), "count", len(clusters))
+			switch r.Config.Strategy {
+			case config.STRATEGY_SIMPLE:
+				cluster = clusters[0]
+			case config.STRATEGY_RANDOM:
+				cluster = clusters[rand.IntN(len(clusters))]
+			case "":
+				// default to balanced, if empty
+				fallthrough
+			case config.STRATEGY_BALANCED:
+				// find cluster with least number of requests
+				cluster = clusters[0]
+				count := cluster.GetTenancyCount()
+				for _, c := range clusters[1:] {
+					tmp := c.GetTenancyCount()
+					if tmp < count {
+						count = tmp
+						cluster = c
+					}
+				}
+			default:
+				rr.ReconcileError = errutils.WithReason(fmt.Errorf("unknown strategy '%s'", r.Config.Strategy), cconst.ReasonConfigurationProblem)
+				return rr
+			}
+		}
+	}
+
+	if cluster != nil {
+		log.Info("Existing cluster qualifies for request, using it", "clusterName", cluster.Name, "clusterNamespace", cluster.Namespace)
+
+		// patch finalizer into Cluster
+		oldCluster := cluster.DeepCopy()
+		fin := cr.FinalizerForCluster()
+		if controllerutil.AddFinalizer(cluster, fin) {
+			log.Debug("Adding finalizer to cluster", "clusterName", cluster.Name, "clusterNamespace", cluster.Namespace, "finalizer", fin)
+			if err := r.PlatformCluster.Client().Patch(ctx, cluster, client.MergeFrom(oldCluster)); err != nil {
+				rr.ReconcileError = errutils.WithReason(fmt.Errorf("error patching finalizer '%s' on cluster '%s/%s': %w", fin, cluster.Namespace, cluster.Name, err), cconst.ReasonPlatformClusterInteractionProblem)
+				return rr
+			}
+		}
+	} else {
+		cluster = r.initializeNewCluster(ctx, cr, cDef)
+
+		// create Cluster resource
+		if err := r.PlatformCluster.Client().Create(ctx, cluster); err != nil {
+			if apierrors.IsAlreadyExists(err) {
+				rr.ReconcileError = errutils.WithReason(fmt.Errorf("Cluster '%s/%s' already exists, this is not supposed to happen", cluster.Namespace, cluster.Name), cconst.ReasonInternalError)
+				return rr
+			}
+			rr.ReconcileError = errutils.WithReason(fmt.Errorf("error creating cluster '%s/%s': %w", cluster.Namespace, cluster.Name, err), cconst.ReasonPlatformClusterInteractionProblem)
+			return rr
+		}
+	}
+
+	// add cluster reference to request
+	rr.Object.Status.Cluster = &clustersv1alpha1.NamespacedObjectReference{}
+	rr.Object.Status.Cluster.Name = cluster.Name
+	rr.Object.Status.Cluster.Namespace = cluster.Namespace
+
+	return rr
+}
+
+func (r *ClusterScheduler) handleDelete(ctx context.Context, req reconcile.Request, cr *clustersv1alpha1.ClusterRequest) ReconcileResult {
+	log := logging.FromContextOrPanic(ctx)
+	rr := ReconcileResult{
+		Object:    cr,
+		OldObject: cr.DeepCopy(),
+	}
+
+	log.Info("Deleting resource")
+
+	// fetch all clusters and filter for the ones that have a finalizer from this request
+	fin := cr.FinalizerForCluster()
+	clusterList := &clustersv1alpha1.ClusterList{}
+	if err := r.PlatformCluster.Client().List(ctx, clusterList, client.MatchingLabelsSelector{Selector: r.Config.CompletedSelectors.Clusters}); err != nil {
+		rr.ReconcileError = errutils.WithReason(fmt.Errorf("error listing Clusters: %w", err), cconst.ReasonPlatformClusterInteractionProblem)
+		return rr
+	}
+	clusters := make([]*clustersv1alpha1.Cluster, len(clusterList.Items))
+	for i := range clusterList.Items {
+		clusters[i] = &clusterList.Items[i]
+	}
+	clusters = filters.FilterSlice(clusters, func(args ...any) bool {
+		c, ok := args[0].(*clustersv1alpha1.Cluster)
+		if !ok {
+			return false
+		}
+		return slices.Contains(c.Finalizers, fin)
+	})
+
+	// remove finalizer from all clusters
+	errs := errutils.NewReasonableErrorList()
+	for _, c := range clusters {
+		log.Debug("Removing finalizer from cluster", "clusterName", c.Name, "clusterNamespace", c.Namespace, "finalizer", fin)
+		oldCluster := c.DeepCopy()
+		if controllerutil.RemoveFinalizer(c, fin) {
+			if err := r.PlatformCluster.Client().Patch(ctx, c, client.MergeFrom(oldCluster)); err != nil {
+				errs.Append(errutils.WithReason(fmt.Errorf("error patching finalizer '%s' on cluster '%s/%s': %w", fin, c.Namespace, c.Name, err), cconst.ReasonPlatformClusterInteractionProblem))
+			}
+		}
+		if c.GetTenancyCount() == 0 && ctrlutils.HasLabelWithValue(c, clustersv1alpha1.DeleteWithoutRequestsLabel, "true") {
+			log.Info("Deleting cluster without requests", "clusterName", c.Name, "clusterNamespace", c.Namespace)
+			if err := r.PlatformCluster.Client().Delete(ctx, c); err != nil {
+				if apierrors.IsNotFound(err) {
+					log.Info("Cluster already deleted", "clusterName", c.Name, "clusterNamespace", c.Namespace)
+				} else {
+					errs.Append(errutils.WithReason(fmt.Errorf("error deleting cluster '%s/%s': %w", c.Namespace, c.Name, err), cconst.ReasonPlatformClusterInteractionProblem))
+				}
+			}
+		}
+	}
+	rr.ReconcileError = errs.Aggregate()
+	if rr.ReconcileError != nil {
+		return rr
+	}
+
+	// remove finalizer
+	if controllerutil.RemoveFinalizer(cr, clustersv1alpha1.ClusterRequestFinalizer) {
+		log.Info("Removing finalizer")
+		if err := r.PlatformCluster.Client().Patch(ctx, cr, client.MergeFrom(rr.OldObject)); err != nil {
+			rr.ReconcileError = errutils.WithReason(fmt.Errorf("error removing finalizer from resource '%s': %w", req.NamespacedName.String(), err), cconst.ReasonPlatformClusterInteractionProblem)
+			return rr
+		}
+	}
+	rr.Object = nil // this prevents the controller from trying to update an already deleted resource
+
+	return rr
+}
+
+// SetupWithManager sets up the controller with the Manager.
+func (r *ClusterScheduler) SetupWithManager(mgr ctrl.Manager) error {
+	return ctrl.NewControllerManagedBy(mgr).
+		// watch ClusterRequest resources
+		For(&clustersv1alpha1.ClusterRequest{}).
+		WithEventFilter(predicate.And(
+			ctrlutils.LabelSelectorPredicate(r.Config.CompletedSelectors.Requests),
+			predicate.Or(
+				predicate.GenerationChangedPredicate{},
+				ctrlutils.DeletionTimestampChangedPredicate{},
+				ctrlutils.GotAnnotationPredicate(clustersv1alpha1.OperationAnnotation, clustersv1alpha1.OperationAnnotationValueReconcile),
+				ctrlutils.LostAnnotationPredicate(clustersv1alpha1.OperationAnnotation, clustersv1alpha1.OperationAnnotationValueIgnore),
+			),
+			predicate.Not(
+				ctrlutils.HasAnnotationPredicate(clustersv1alpha1.OperationAnnotation, clustersv1alpha1.OperationAnnotationValueIgnore),
+			),
+		)).
+		Complete(r)
+}
+
+// fetchRelevantClusters fetches all Cluster resources that could qualify for the given ClusterRequest.
+func (r *ClusterScheduler) fetchRelevantClusters(ctx context.Context, cr *clustersv1alpha1.ClusterRequest, cDef *config.ClusterDefinition) ([]*clustersv1alpha1.Cluster, errutils.ReasonableError) {
+	// fetch clusters
+	purpose := cr.Spec.Purpose
+	namespace := cr.Namespace
+	if r.Config.Scope == config.SCOPE_CLUSTER {
+		// in cluster scope, search all namespaces
+		namespace = ""
+	} else if cDef.Template.Namespace != "" {
+		// in namespaced scope, use template namespace if set, and request namespace otherwise
+		namespace = cDef.Template.Namespace
+	}
+	clusterList := &clustersv1alpha1.ClusterList{}
+	if err := r.PlatformCluster.Client().List(ctx, clusterList, client.InNamespace(namespace), client.MatchingLabelsSelector{Selector: cDef.CompletedSelector}); err != nil {
+		return nil, errutils.WithReason(fmt.Errorf("error listing Clusters: %w", err), cconst.ReasonPlatformClusterInteractionProblem)
+	}
+	clusters := make([]*clustersv1alpha1.Cluster, len(clusterList.Items))
+	for i := range clusterList.Items {
+		clusters[i] = &clusterList.Items[i]
+	}
+
+	// filter clusters by the desired purpose
+	clusters = filters.FilterSlice(clusters, func(args ...any) bool {
+		c, ok := args[0].(*clustersv1alpha1.Cluster)
+		if !ok {
+			return false
+		}
+		return slices.Contains(c.Spec.Purposes, purpose)
+	})
+
+	return clusters, nil
+}
+
+// initializeNewCluster creates a new Cluster resource based on the given ClusterRequest and ClusterDefinition.
+func (r *ClusterScheduler) initializeNewCluster(ctx context.Context, cr *clustersv1alpha1.ClusterRequest, cDef *config.ClusterDefinition) *clustersv1alpha1.Cluster {
+	log := logging.FromContextOrPanic(ctx)
+	purpose := cr.Spec.Purpose
+	cluster := &clustersv1alpha1.Cluster{}
+	// choose a name for the cluster
+	// priority as follows:
+	// - for singleton clusters (shared unlimited):
+	//  1. generateName of template
+	//  2. name of template
+	//  3. purpose
+	// - for exclusive clusters or shared limited:
+	//  1. generateName of template
+	//  2. purpose used as generateName
+	if cDef.Template.Spec.Tenancy == clustersv1alpha1.TENANCY_SHARED && cDef.TenancyCount == 0 {
+		// there will only be one instance of this cluster
+		if cDef.Template.GenerateName != "" {
+			cluster.SetGenerateName(cDef.Template.GenerateName)
+		} else if cDef.Template.Name != "" {
+			cluster.SetName(cDef.Template.Name)
+		} else {
+			cluster.SetName(purpose)
+		}
+	} else {
+		// there might be multiple instances of this cluster
+		if cDef.Template.GenerateName != "" {
+			cluster.SetGenerateName(cDef.Template.GenerateName)
+		} else {
+			cluster.SetGenerateName(purpose + "-")
+		}
+	}
+	// choose a namespace for the cluster
+	// priority as follows:
+	//  1. namespace of template
+	//  2. namespace of request
+	if cDef.Template.Namespace != "" {
+		cluster.SetNamespace(cDef.Template.Namespace)
+	} else {
+		cluster.SetNamespace(cr.Namespace)
+	}
+	log.Info("Creating new cluster", "clusterName", cluster.Name, "clusterNamespace", cluster.Namespace)
+
+	// set finalizer
+	cluster.SetFinalizers([]string{cr.FinalizerForCluster()})
+	// take over labels, annotations, and spec from the template
+	cluster.SetLabels(cDef.Template.Labels)
+	if err := ctrlutils.EnsureLabel(ctx, nil, cluster, clustersv1alpha1.DeleteWithoutRequestsLabel, "true", false); err != nil {
+		if !ctrlutils.IsMetadataEntryAlreadyExistsError(err) {
+			log.Error(err, "error setting label", "label", clustersv1alpha1.DeleteWithoutRequestsLabel, "value", "true")
+		}
+	}
+	cluster.SetAnnotations(cDef.Template.Annotations)
+	cluster.Spec = cDef.Template.Spec
+
+	// set purpose, if not set
+	if len(cluster.Spec.Purposes) == 0 {
+		cluster.Spec.Purposes = []string{purpose}
+	} else {
+		if !slices.Contains(cluster.Spec.Purposes, purpose) {
+			cluster.Spec.Purposes = append(cluster.Spec.Purposes, purpose)
+		}
+	}
+
+	return cluster
+}
diff --git a/internal/controllers/scheduler/controller_test.go b/internal/controllers/scheduler/controller_test.go
new file mode 100644
index 0000000..38a889a
--- /dev/null
+++ b/internal/controllers/scheduler/controller_test.go
@@ -0,0 +1,449 @@
+package scheduler_test
+
+import (
+	"fmt"
+	"path/filepath"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	. "github.com/onsi/gomega/gstruct"
+
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/util/uuid"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+
+	"github.com/openmcp-project/controller-utils/pkg/clusters"
+	ctrlutils "github.com/openmcp-project/controller-utils/pkg/controller"
+	testutils "github.com/openmcp-project/controller-utils/pkg/testing"
+
+	clustersv1alpha1 "github.com/openmcp-project/openmcp-operator/api/clusters/v1alpha1"
+	"github.com/openmcp-project/openmcp-operator/api/install"
+	"github.com/openmcp-project/openmcp-operator/internal/config"
+	"github.com/openmcp-project/openmcp-operator/internal/controllers/scheduler"
+)
+
+var scheme = install.InstallOperatorAPIs(runtime.NewScheme())
+
+// defaultTestSetup initializes a new environment for testing the scheduler controller.
+// Expected folder structure is a 'config.yaml' file next to a folder named 'cluster' containing the manifests.
+func defaultTestSetup(testDirPathSegments ...string) (*scheduler.ClusterScheduler, *testutils.Environment) {
+	cfg, err := config.LoadFromFiles(filepath.Join(append(testDirPathSegments, "config.yaml")...))
+	Expect(err).ToNot(HaveOccurred())
+	Expect(cfg.Default()).To(Succeed())
+	Expect(cfg.Validate()).To(Succeed())
+	Expect(cfg.Complete()).To(Succeed())
+	env := testutils.NewEnvironmentBuilder().WithFakeClient(scheme).WithInitObjectPath(append(testDirPathSegments, "cluster")...).WithReconcilerConstructor(func(c client.Client) reconcile.Reconciler {
+		r, err := scheduler.NewClusterScheduler(nil, clusters.NewTestClusterFromClient("onboarding", c), cfg.Scheduler)
+		Expect(err).ToNot(HaveOccurred())
+		return r
+	}).Build()
+	sc, ok := env.Reconciler().(*scheduler.ClusterScheduler)
+	Expect(ok).To(BeTrue(), "Reconciler is not of type ClusterScheduler")
+	return sc, env
+}
+
+var _ = Describe("Scheduler", func() {
+
+	Context("Scope: Namespaced", func() {
+
+		It("should create a new exclusive cluster if no cluster exists", func() {
+			clusterNamespace := "exclusive"
+			sc, env := defaultTestSetup("testdata", "test-01")
+			Expect(env.Client().DeleteAllOf(env.Ctx, &clustersv1alpha1.Cluster{}, client.InNamespace(clusterNamespace))).To(Succeed())
+			existingClusters := &clustersv1alpha1.ClusterList{}
+			Expect(env.Client().List(env.Ctx, existingClusters, client.InNamespace(clusterNamespace))).To(Succeed())
+			Expect(existingClusters.Items).To(BeEmpty())
+
+			req := &clustersv1alpha1.ClusterRequest{}
+			Expect(env.Client().Get(env.Ctx, ctrlutils.ObjectKey("exclusive", "foo"), req)).To(Succeed())
+			Expect(req.Status.Cluster).To(BeNil())
+
+			env.ShouldReconcile(testutils.RequestFromObject(req))
+			Expect(env.Client().Get(env.Ctx, client.ObjectKeyFromObject(req), req)).To(Succeed())
+			Expect(req.Status.Cluster).ToNot(BeNil())
+
+			Expect(env.Client().List(env.Ctx, existingClusters, client.InNamespace(clusterNamespace))).To(Succeed())
+			Expect(existingClusters.Items).To(HaveLen(1))
+			cluster := existingClusters.Items[0]
+			Expect(cluster.Name).To(Equal(req.Status.Cluster.Name))
+			Expect(cluster.Namespace).To(Equal(clusterNamespace))
+			Expect(cluster.Name).To(HavePrefix(fmt.Sprintf("%s-", req.Spec.Purpose)))
+			Expect(cluster.Namespace).To(Equal(sc.Config.PurposeMappings[req.Spec.Purpose].Template.ObjectMeta.Namespace))
+			Expect(cluster.Spec.Tenancy).To(BeEquivalentTo(sc.Config.PurposeMappings[req.Spec.Purpose].Template.Spec.Tenancy))
+			Expect(cluster.Finalizers).To(ContainElements(req.FinalizerForCluster()))
+		})
+
+		It("should create a new exclusive cluster if a cluster exists", func() {
+			clusterNamespace := "exclusive"
+			sc, env := defaultTestSetup("testdata", "test-01")
+			existingClusters := &clustersv1alpha1.ClusterList{}
+			Expect(env.Client().List(env.Ctx, existingClusters, client.InNamespace(clusterNamespace))).To(Succeed())
+			oldCount := len(existingClusters.Items)
+
+			req := &clustersv1alpha1.ClusterRequest{}
+			Expect(env.Client().Get(env.Ctx, ctrlutils.ObjectKey("exclusive", "foo"), req)).To(Succeed())
+			Expect(req.Status.Cluster).To(BeNil())
+
+			env.ShouldReconcile(testutils.RequestFromObject(req))
+			Expect(env.Client().Get(env.Ctx, client.ObjectKeyFromObject(req), req)).To(Succeed())
+			Expect(req.Status.Cluster).ToNot(BeNil())
+
+			Expect(env.Client().List(env.Ctx, existingClusters, client.InNamespace(clusterNamespace))).To(Succeed())
+			Expect(existingClusters.Items).To(HaveLen(oldCount + 1))
+			Expect(existingClusters.Items).To(ContainElements(MatchFields(IgnoreExtras, Fields{
+				"ObjectMeta": MatchFields(IgnoreExtras, Fields{
+					"Name":       Equal(req.Status.Cluster.Name),
+					"Namespace":  Equal(req.Status.Cluster.Namespace),
+					"Finalizers": ContainElements(req.FinalizerForCluster()),
+				}),
+				"Spec": MatchFields(IgnoreExtras, Fields{
+					"Tenancy": BeEquivalentTo(sc.Config.PurposeMappings[req.Spec.Purpose].Template.Spec.Tenancy),
+				}),
+			})))
+		})
+
+		It("should create a new shared cluster if no cluster exists", func() {
+			clusterNamespace := "shared-twice"
+			sc, env := defaultTestSetup("testdata", "test-01")
+			Expect(env.Client().DeleteAllOf(env.Ctx, &clustersv1alpha1.Cluster{}, client.InNamespace(clusterNamespace))).To(Succeed())
+			existingClusters := &clustersv1alpha1.ClusterList{}
+			Expect(env.Client().List(env.Ctx, existingClusters, client.InNamespace(clusterNamespace))).To(Succeed())
+			Expect(existingClusters.Items).To(BeEmpty())
+
+			req := &clustersv1alpha1.ClusterRequest{}
+			Expect(env.Client().Get(env.Ctx, ctrlutils.ObjectKey("shared", "foo"), req)).To(Succeed())
+			Expect(req.Status.Cluster).To(BeNil())
+
+			env.ShouldReconcile(testutils.RequestFromObject(req))
+			Expect(env.Client().Get(env.Ctx, client.ObjectKeyFromObject(req), req)).To(Succeed())
+			Expect(req.Status.Cluster).ToNot(BeNil())
+
+			Expect(env.Client().List(env.Ctx, existingClusters, client.InNamespace(clusterNamespace))).To(Succeed())
+			Expect(existingClusters.Items).To(HaveLen(1))
+			cluster := existingClusters.Items[0]
+			Expect(cluster.Name).To(Equal(req.Status.Cluster.Name))
+			Expect(cluster.Namespace).To(Equal(clusterNamespace))
+			Expect(cluster.Name).To(HavePrefix(fmt.Sprintf("%s-", req.Spec.Purpose)))
+			Expect(cluster.Namespace).To(Equal(sc.Config.PurposeMappings[req.Spec.Purpose].Template.ObjectMeta.Namespace))
+			Expect(cluster.Spec.Tenancy).To(BeEquivalentTo(sc.Config.PurposeMappings[req.Spec.Purpose].Template.Spec.Tenancy))
+			Expect(cluster.Finalizers).To(ContainElements(req.FinalizerForCluster()))
+		})
+
+		It("should share a shared cluster if it still has capacity and create a new one otherwise", func() {
+			clusterNamespace := "shared-twice"
+			sc, env := defaultTestSetup("testdata", "test-01")
+			existingClusters := &clustersv1alpha1.ClusterList{}
+			Expect(env.Client().List(env.Ctx, existingClusters, client.InNamespace(clusterNamespace))).To(Succeed())
+			oldCount := len(existingClusters.Items)
+
+			// first request
+			// should use existing cluster
+			req := &clustersv1alpha1.ClusterRequest{}
+			Expect(env.Client().Get(env.Ctx, ctrlutils.ObjectKey("shared", "foo"), req)).To(Succeed())
+			Expect(req.Status.Cluster).To(BeNil())
+
+			env.ShouldReconcile(testutils.RequestFromObject(req))
+			Expect(env.Client().Get(env.Ctx, client.ObjectKeyFromObject(req), req)).To(Succeed())
+			Expect(req.Status.Cluster).ToNot(BeNil())
+
+			Expect(env.Client().List(env.Ctx, existingClusters, client.InNamespace(clusterNamespace))).To(Succeed())
+			Expect(existingClusters.Items).To(HaveLen(oldCount))
+			Expect(existingClusters.Items).To(ContainElements(MatchFields(IgnoreExtras, Fields{
+				"ObjectMeta": MatchFields(IgnoreExtras, Fields{
+					"Name":       Equal(req.Status.Cluster.Name),
+					"Namespace":  Equal(req.Status.Cluster.Namespace),
+					"Finalizers": ContainElements(req.FinalizerForCluster()),
+				}),
+				"Spec": MatchFields(IgnoreExtras, Fields{
+					"Tenancy": BeEquivalentTo(sc.Config.PurposeMappings[req.Spec.Purpose].Template.Spec.Tenancy),
+				}),
+			})))
+
+			// second request
+			// should use existing cluster
+			req2 := &clustersv1alpha1.ClusterRequest{}
+			Expect(env.Client().Get(env.Ctx, ctrlutils.ObjectKey("shared2", "foo"), req2)).To(Succeed())
+			Expect(req2.Status.Cluster).To(BeNil())
+
+			env.ShouldReconcile(testutils.RequestFromObject(req2))
+			Expect(env.Client().Get(env.Ctx, client.ObjectKeyFromObject(req2), req2)).To(Succeed())
+			Expect(req2.Status.Cluster).ToNot(BeNil())
+
+			Expect(env.Client().List(env.Ctx, existingClusters, client.InNamespace(clusterNamespace))).To(Succeed())
+			Expect(existingClusters.Items).To(HaveLen(oldCount))
+			Expect(existingClusters.Items).To(ContainElements(MatchFields(IgnoreExtras, Fields{
+				"ObjectMeta": MatchFields(IgnoreExtras, Fields{
+					"Name":       Equal(req2.Status.Cluster.Name),
+					"Namespace":  Equal(req2.Status.Cluster.Namespace),
+					"Finalizers": ContainElements(req2.FinalizerForCluster()),
+				}),
+				"Spec": MatchFields(IgnoreExtras, Fields{
+					"Tenancy": BeEquivalentTo(sc.Config.PurposeMappings[req2.Spec.Purpose].Template.Spec.Tenancy),
+				}),
+			})))
+			Expect(req2.Status.Cluster.Name).To(Equal(req.Status.Cluster.Name))
+
+			// third request
+			// should create a new cluster
+			req3 := &clustersv1alpha1.ClusterRequest{}
+			Expect(env.Client().Get(env.Ctx, ctrlutils.ObjectKey("shared3", "foo"), req3)).To(Succeed())
+			Expect(req3.Status.Cluster).To(BeNil())
+
+			env.ShouldReconcile(testutils.RequestFromObject(req3))
+			Expect(env.Client().Get(env.Ctx, client.ObjectKeyFromObject(req3), req3)).To(Succeed())
+			Expect(req3.Status.Cluster).ToNot(BeNil())
+
+			Expect(env.Client().List(env.Ctx, existingClusters, client.InNamespace(clusterNamespace))).To(Succeed())
+			Expect(existingClusters.Items).To(HaveLen(oldCount + 1))
+			Expect(existingClusters.Items).To(ContainElements(MatchFields(IgnoreExtras, Fields{
+				"ObjectMeta": MatchFields(IgnoreExtras, Fields{
+					"Name":       Equal(req3.Status.Cluster.Name),
+					"Namespace":  Equal(req3.Status.Cluster.Namespace),
+					"Finalizers": ContainElements(req3.FinalizerForCluster()),
+				}),
+				"Spec": MatchFields(IgnoreExtras, Fields{
+					"Tenancy": BeEquivalentTo(sc.Config.PurposeMappings[req3.Spec.Purpose].Template.Spec.Tenancy),
+				}),
+			})))
+			Expect(req3.Status.Cluster.Name).ToNot(Equal(req.Status.Cluster.Name))
+			Expect(req3.Status.Cluster.Name).ToNot(Equal(req2.Status.Cluster.Name))
+		})
+
+		It("should only create a new cluster if none exists for unlimitedly shared clusters", func() {
+			clusterNamespace := "shared-unlimited"
+			sc, env := defaultTestSetup("testdata", "test-01")
+			reqCount := 20
+			requests := make([]*clustersv1alpha1.ClusterRequest, reqCount)
+			for i := range reqCount {
+				requests[i] = &clustersv1alpha1.ClusterRequest{}
+				requests[i].SetName(fmt.Sprintf("req-%d", i))
+				requests[i].SetNamespace("foo")
+				requests[i].SetUID(uuid.NewUUID())
+				requests[i].Spec.Purpose = "shared-unlimited"
+				Expect(env.Client().Create(env.Ctx, requests[i])).To(Succeed())
+				env.ShouldReconcile(testutils.RequestFromObject(requests[i]))
+			}
+			for _, req := range requests {
+				Expect(env.Client().Get(env.Ctx, client.ObjectKeyFromObject(req), req)).To(Succeed())
+				Expect(req.Status.Cluster).ToNot(BeNil())
+				Expect(req.Status.Cluster.Name).To(Equal(requests[0].Status.Cluster.Name))
+				Expect(req.Status.Cluster.Namespace).To(Equal(requests[0].Status.Cluster.Namespace))
+			}
+			existingClusters := &clustersv1alpha1.ClusterList{}
+			Expect(env.Client().List(env.Ctx, existingClusters, client.InNamespace(clusterNamespace))).To(Succeed())
+			Expect(existingClusters.Items).To(HaveLen(1))
+			cluster := existingClusters.Items[0]
+			Expect(cluster.Name).To(Equal(requests[0].Status.Cluster.Name))
+			Expect(cluster.Namespace).To(Equal(clusterNamespace))
+			Expect(cluster.Name).To(Equal(requests[0].Spec.Purpose))
+			Expect(cluster.Namespace).To(Equal(sc.Config.PurposeMappings[requests[0].Spec.Purpose].Template.ObjectMeta.Namespace))
+			Expect(cluster.Spec.Tenancy).To(BeEquivalentTo(clustersv1alpha1.TENANCY_SHARED))
+			Expect(cluster.Finalizers).To(ContainElements(requests[0].FinalizerForCluster()))
+		})
+
+		It("should take over annotations and labels from the cluster template", func() {
+			_, env := defaultTestSetup("testdata", "test-02")
+
+			req := &clustersv1alpha1.ClusterRequest{}
+			Expect(env.Client().Get(env.Ctx, ctrlutils.ObjectKey("exclusive", "foo"), req)).To(Succeed())
+			Expect(req.Status.Cluster).To(BeNil())
+
+			env.ShouldReconcile(testutils.RequestFromObject(req))
+			Expect(env.Client().Get(env.Ctx, client.ObjectKeyFromObject(req), req)).To(Succeed())
+			Expect(req.Status.Cluster).ToNot(BeNil())
+
+			cluster := &clustersv1alpha1.Cluster{}
+			Expect(env.Client().Get(env.Ctx, ctrlutils.ObjectKey(req.Status.Cluster.Name, req.Status.Cluster.Namespace), cluster)).To(Succeed())
+			Expect(cluster.Labels).To(HaveKeyWithValue("foo.bar.baz/foobar", "true"))
+			Expect(cluster.Annotations).To(HaveKeyWithValue("foo.bar.baz/foobar", "false"))
+		})
+
+		It("should use the request's namespace if none is specified in the template and ignore clusters that don't match the label selector", func() {
+			clusterNamespace := "foo"
+			_, env := defaultTestSetup("testdata", "test-02")
+
+			req := &clustersv1alpha1.ClusterRequest{}
+			Expect(env.Client().Get(env.Ctx, ctrlutils.ObjectKey("shared", "foo"), req)).To(Succeed())
+			Expect(req.Status.Cluster).To(BeNil())
+
+			fooClusters := &clustersv1alpha1.ClusterList{}
+			Expect(env.Client().List(env.Ctx, fooClusters, client.InNamespace(clusterNamespace))).To(Succeed())
+			Expect(fooClusters.Items).ToNot(BeEmpty())
+			oldCountFoo := len(fooClusters.Items)
+			for _, cluster := range fooClusters.Items {
+				Expect(cluster.Labels).ToNot(HaveKeyWithValue("foo.bar.baz/foobar", "true"))
+			}
+
+			// this should create a new cluster in 'foo'
+			// because the existing ones' labels don't match the selector
+			env.ShouldReconcile(testutils.RequestFromObject(req))
+			Expect(env.Client().Get(env.Ctx, client.ObjectKeyFromObject(req), req)).To(Succeed())
+			Expect(req.Status.Cluster).ToNot(BeNil())
+			Expect(env.Client().List(env.Ctx, fooClusters, client.InNamespace(clusterNamespace))).To(Succeed())
+			Expect(fooClusters.Items).To(HaveLen(oldCountFoo + 1))
+			oldCountFoo = len(fooClusters.Items)
+
+			// this should create a new cluster in 'bar'
+			req2 := &clustersv1alpha1.ClusterRequest{}
+			Expect(env.Client().Get(env.Ctx, ctrlutils.ObjectKey("shared", "bar"), req2)).To(Succeed())
+			Expect(req2.Status.Cluster).To(BeNil())
+			barClusters := &clustersv1alpha1.ClusterList{}
+			Expect(env.Client().List(env.Ctx, barClusters, client.InNamespace("bar"))).To(Succeed())
+			oldCountBar := len(barClusters.Items)
+			env.ShouldReconcile(testutils.RequestFromObject(req2))
+			Expect(env.Client().Get(env.Ctx, client.ObjectKeyFromObject(req2), req2)).To(Succeed())
+			Expect(req2.Status.Cluster).ToNot(BeNil())
+			Expect(env.Client().List(env.Ctx, barClusters, client.InNamespace("bar"))).To(Succeed())
+			Expect(barClusters.Items).To(HaveLen(oldCountBar + 1))
+
+			// this should re-use the existing cluster in 'foo'
+			req3 := &clustersv1alpha1.ClusterRequest{}
+			Expect(env.Client().Get(env.Ctx, ctrlutils.ObjectKey("shared2", "foo"), req3)).To(Succeed())
+			Expect(req3.Status.Cluster).To(BeNil())
+			env.ShouldReconcile(testutils.RequestFromObject(req3))
+			Expect(env.Client().Get(env.Ctx, client.ObjectKeyFromObject(req3), req3)).To(Succeed())
+			Expect(req3.Status.Cluster).ToNot(BeNil())
+			Expect(req3.Status.Cluster.Name).To(Equal(req.Status.Cluster.Name))
+			Expect(req3.Status.Cluster.Namespace).To(Equal(req.Status.Cluster.Namespace))
+			Expect(env.Client().List(env.Ctx, fooClusters, client.InNamespace(clusterNamespace))).To(Succeed())
+			Expect(fooClusters.Items).To(HaveLen(oldCountFoo))
+		})
+
+	})
+
+	Context("Scope: Cluster", func() {
+
+		It("should evaluate all namespaces in cluster scope", func() {
+			_, env := defaultTestSetup("testdata", "test-03")
+
+			req := &clustersv1alpha1.ClusterRequest{}
+			Expect(env.Client().Get(env.Ctx, ctrlutils.ObjectKey("shared", "foo"), req)).To(Succeed())
+			Expect(req.Status.Cluster).To(BeNil())
+
+			clusters := &clustersv1alpha1.ClusterList{}
+			Expect(env.Client().List(env.Ctx, clusters)).To(Succeed())
+			Expect(clusters.Items).ToNot(BeEmpty())
+			oldCount := len(clusters.Items)
+			for _, cluster := range clusters.Items {
+				Expect(cluster.Labels).ToNot(HaveKeyWithValue("foo.bar.baz/foobar", "true"))
+			}
+
+			// this should create a new cluster in 'foo'
+			// because the existing ones' labels don't match the selector
+			env.ShouldReconcile(testutils.RequestFromObject(req))
+			Expect(env.Client().Get(env.Ctx, client.ObjectKeyFromObject(req), req)).To(Succeed())
+			Expect(req.Status.Cluster).ToNot(BeNil())
+			Expect(env.Client().List(env.Ctx, clusters)).To(Succeed())
+			Expect(clusters.Items).To(HaveLen(oldCount + 1))
+			oldCount = len(clusters.Items)
+
+			// this should re-use the existing cluster
+			req2 := &clustersv1alpha1.ClusterRequest{}
+			Expect(env.Client().Get(env.Ctx, ctrlutils.ObjectKey("shared2", "bar"), req2)).To(Succeed())
+			Expect(req2.Status.Cluster).To(BeNil())
+			env.ShouldReconcile(testutils.RequestFromObject(req2))
+			Expect(env.Client().Get(env.Ctx, client.ObjectKeyFromObject(req2), req2)).To(Succeed())
+			Expect(req2.Status.Cluster).ToNot(BeNil())
+			Expect(req2.Status.Cluster.Name).To(Equal(req.Status.Cluster.Name))
+			Expect(req2.Status.Cluster.Namespace).To(Equal(req.Status.Cluster.Namespace))
+			Expect(env.Client().List(env.Ctx, clusters)).To(Succeed())
+			Expect(clusters.Items).To(HaveLen(oldCount))
+
+			// this should re-use the existing cluster
+			req3 := &clustersv1alpha1.ClusterRequest{}
+			Expect(env.Client().Get(env.Ctx, ctrlutils.ObjectKey("shared3", "baz"), req3)).To(Succeed())
+			Expect(req3.Status.Cluster).To(BeNil())
+			env.ShouldReconcile(testutils.RequestFromObject(req3))
+			Expect(env.Client().Get(env.Ctx, client.ObjectKeyFromObject(req3), req3)).To(Succeed())
+			Expect(req3.Status.Cluster).ToNot(BeNil())
+			Expect(req3.Status.Cluster.Name).To(Equal(req.Status.Cluster.Name))
+			Expect(req3.Status.Cluster.Namespace).To(Equal(req.Status.Cluster.Namespace))
+			Expect(env.Client().List(env.Ctx, clusters)).To(Succeed())
+			Expect(clusters.Items).To(HaveLen(oldCount))
+		})
+
+	})
+
+	It("should combine cluster label selectors correctly", func() {
+		_, env := defaultTestSetup("testdata", "test-04")
+
+		// should use the existing cluster
+		req := &clustersv1alpha1.ClusterRequest{}
+		Expect(env.Client().Get(env.Ctx, ctrlutils.ObjectKey("shared", "foo"), req)).To(Succeed())
+		Expect(req.Status.Cluster).To(BeNil())
+		env.ShouldReconcile(testutils.RequestFromObject(req))
+		Expect(env.Client().Get(env.Ctx, client.ObjectKeyFromObject(req), req)).To(Succeed())
+		Expect(req.Status.Cluster).ToNot(BeNil())
+		Expect(req.Status.Cluster.Name).To(Equal("shared"))
+		Expect(req.Status.Cluster.Namespace).To(Equal("foo"))
+
+		// should create a new cluster
+		req2 := &clustersv1alpha1.ClusterRequest{}
+		Expect(env.Client().Get(env.Ctx, ctrlutils.ObjectKey("shared2", "foo"), req2)).To(Succeed())
+		Expect(req2.Status.Cluster).To(BeNil())
+		env.ShouldReconcile(testutils.RequestFromObject(req2))
+		Expect(env.Client().Get(env.Ctx, client.ObjectKeyFromObject(req2), req2)).To(Succeed())
+		Expect(req2.Status.Cluster).ToNot(BeNil())
+		Expect(req2.Status.Cluster.Name).To(Equal("shared2"))
+		Expect(req2.Status.Cluster.Namespace).To(Equal("foo"))
+
+		// should use the existing cluster
+		req3 := &clustersv1alpha1.ClusterRequest{}
+		Expect(env.Client().Get(env.Ctx, ctrlutils.ObjectKey("shared3", "foo"), req3)).To(Succeed())
+		Expect(req3.Status.Cluster).To(BeNil())
+		env.ShouldReconcile(testutils.RequestFromObject(req3))
+		Expect(env.Client().Get(env.Ctx, client.ObjectKeyFromObject(req3), req3)).To(Succeed())
+		Expect(req3.Status.Cluster).ToNot(BeNil())
+		Expect(req3.Status.Cluster.Name).To(Equal("shared2"))
+		Expect(req3.Status.Cluster.Namespace).To(Equal("foo"))
+	})
+
+	It("should handle the delete-without-requests label correctly", func() {
+		_, env := defaultTestSetup("testdata", "test-05")
+
+		// should create a new cluster
+		req := &clustersv1alpha1.ClusterRequest{}
+		Expect(env.Client().Get(env.Ctx, ctrlutils.ObjectKey("delete", "foo"), req)).To(Succeed())
+		Expect(req.Status.Cluster).To(BeNil())
+
+		env.ShouldReconcile(testutils.RequestFromObject(req))
+		Expect(env.Client().Get(env.Ctx, client.ObjectKeyFromObject(req), req)).To(Succeed())
+		Expect(req.Status.Cluster).ToNot(BeNil())
+		cluster := &clustersv1alpha1.Cluster{}
+		Expect(env.Client().Get(env.Ctx, ctrlutils.ObjectKey(req.Status.Cluster.Name, req.Status.Cluster.Namespace), cluster)).To(Succeed())
+		Expect(cluster.Labels).To(HaveKeyWithValue(clustersv1alpha1.DeleteWithoutRequestsLabel, "true"))
+
+		// should delete the cluster
+		Expect(env.Client().Delete(env.Ctx, req)).To(Succeed())
+		env.ShouldReconcile(testutils.RequestFromObject(req))
+		Eventually(func() bool {
+			return apierrors.IsNotFound(env.Client().Get(env.Ctx, client.ObjectKeyFromObject(req), req))
+		}, 3).Should(BeTrue(), "Request should be deleted")
+		Eventually(func() bool {
+			return apierrors.IsNotFound(env.Client().Get(env.Ctx, client.ObjectKeyFromObject(cluster), cluster))
+		}, 3).Should(BeTrue(), "Cluster should be deleted")
+
+		// should create a new cluster
+		req2 := &clustersv1alpha1.ClusterRequest{}
+		Expect(env.Client().Get(env.Ctx, ctrlutils.ObjectKey("no-delete", "foo"), req2)).To(Succeed())
+		Expect(req2.Status.Cluster).To(BeNil())
+
+		env.ShouldReconcile(testutils.RequestFromObject(req2))
+		Expect(env.Client().Get(env.Ctx, client.ObjectKeyFromObject(req2), req2)).To(Succeed())
+		Expect(req2.Status.Cluster).ToNot(BeNil())
+		Expect(env.Client().Get(env.Ctx, ctrlutils.ObjectKey(req2.Status.Cluster.Name, req2.Status.Cluster.Namespace), cluster)).To(Succeed())
+		Expect(cluster.Labels).To(HaveKeyWithValue(clustersv1alpha1.DeleteWithoutRequestsLabel, "false"))
+
+		// should not delete the cluster
+		Expect(env.Client().Delete(env.Ctx, req2)).To(Succeed())
+		env.ShouldReconcile(testutils.RequestFromObject(req2))
+		Eventually(func() bool {
+			return apierrors.IsNotFound(env.Client().Get(env.Ctx, client.ObjectKeyFromObject(req2), req2))
+		}, 3).Should(BeTrue(), "Request should be deleted")
+		Expect(env.Client().Get(env.Ctx, client.ObjectKeyFromObject(cluster), cluster)).To(Succeed(), "Cluster should not be deleted")
+		Expect(cluster.DeletionTimestamp).To(BeZero(), "Cluster should not be marked for deletion")
+	})
+
+})
diff --git a/internal/controllers/scheduler/suite_test.go b/internal/controllers/scheduler/suite_test.go
new file mode 100644
index 0000000..2969d2c
--- /dev/null
+++ b/internal/controllers/scheduler/suite_test.go
@@ -0,0 +1,14 @@
+package scheduler_test
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+func TestComponentUtils(t *testing.T) {
+	RegisterFailHandler(Fail)
+
+	RunSpecs(t, "Scheduler Test Suite")
+}
diff --git a/internal/controllers/scheduler/testdata/test-01/cluster/cluster-0.yaml b/internal/controllers/scheduler/testdata/test-01/cluster/cluster-0.yaml
new file mode 100644
index 0000000..2374dd7
--- /dev/null
+++ b/internal/controllers/scheduler/testdata/test-01/cluster/cluster-0.yaml
@@ -0,0 +1,13 @@
+apiVersion: clusters.openmcp.cloud/v1alpha1
+kind: Cluster
+metadata:
+  name: exclusive-1
+  namespace: exclusive
+spec:
+  profile: test-profile
+  kubernetes:
+    version: "1.32"
+  purposes:
+  - test
+  - exclusive
+  tenancy: Exclusive
diff --git a/internal/controllers/scheduler/testdata/test-01/cluster/cluster-1.yaml b/internal/controllers/scheduler/testdata/test-01/cluster/cluster-1.yaml
new file mode 100644
index 0000000..1f5e48c
--- /dev/null
+++ b/internal/controllers/scheduler/testdata/test-01/cluster/cluster-1.yaml
@@ -0,0 +1,14 @@
+apiVersion: clusters.openmcp.cloud/v1alpha1
+kind: Cluster
+metadata:
+  name: shared-1
+  namespace: shared-twice
+spec:
+  profile: test-profile
+  kubernetes:
+    version: "1.32"
+  purposes:
+  - test
+  - shared-twice
+  tenancy: Shared
+  
\ No newline at end of file
diff --git a/internal/controllers/scheduler/testdata/test-01/cluster/req-0.yaml b/internal/controllers/scheduler/testdata/test-01/cluster/req-0.yaml
new file mode 100644
index 0000000..6a01132
--- /dev/null
+++ b/internal/controllers/scheduler/testdata/test-01/cluster/req-0.yaml
@@ -0,0 +1,8 @@
+apiVersion: clusters.openmcp.cloud/v1alpha1
+kind: ClusterRequest
+metadata:
+  name: exclusive
+  namespace: foo
+  uid: dc4b422b-8676-4bc0-951f-7a535d790c04
+spec:
+  purpose: exclusive
diff --git a/internal/controllers/scheduler/testdata/test-01/cluster/req-1.yaml b/internal/controllers/scheduler/testdata/test-01/cluster/req-1.yaml
new file mode 100644
index 0000000..9af2f88
--- /dev/null
+++ b/internal/controllers/scheduler/testdata/test-01/cluster/req-1.yaml
@@ -0,0 +1,8 @@
+apiVersion: clusters.openmcp.cloud/v1alpha1
+kind: ClusterRequest
+metadata:
+  name: shared
+  namespace: foo
+  uid: 4d6aa495-54c0-4df2-bc7b-05b103f02e69
+spec:
+  purpose: shared-twice
diff --git a/internal/controllers/scheduler/testdata/test-01/cluster/req-2.yaml b/internal/controllers/scheduler/testdata/test-01/cluster/req-2.yaml
new file mode 100644
index 0000000..1a3721a
--- /dev/null
+++ b/internal/controllers/scheduler/testdata/test-01/cluster/req-2.yaml
@@ -0,0 +1,8 @@
+apiVersion: clusters.openmcp.cloud/v1alpha1
+kind: ClusterRequest
+metadata:
+  name: shared2
+  namespace: foo
+  uid: bf7d8a52-ea1e-4303-a0f9-bf2ef0cc21ea
+spec:
+  purpose: shared-twice
diff --git a/internal/controllers/scheduler/testdata/test-01/cluster/req-3.yaml b/internal/controllers/scheduler/testdata/test-01/cluster/req-3.yaml
new file mode 100644
index 0000000..0aa7691
--- /dev/null
+++ b/internal/controllers/scheduler/testdata/test-01/cluster/req-3.yaml
@@ -0,0 +1,8 @@
+apiVersion: clusters.openmcp.cloud/v1alpha1
+kind: ClusterRequest
+metadata:
+  name: shared3
+  namespace: foo
+  uid: 7642e165-196c-478f-a225-765bd97a29d8
+spec:
+  purpose: shared-twice
diff --git a/internal/controllers/scheduler/testdata/test-01/config.yaml b/internal/controllers/scheduler/testdata/test-01/config.yaml
new file mode 100644
index 0000000..813ef76
--- /dev/null
+++ b/internal/controllers/scheduler/testdata/test-01/config.yaml
@@ -0,0 +1,24 @@
+scheduler:
+  purposeMappings:
+    exclusive:
+      template:
+        metadata:
+          namespace: exclusive
+        spec:
+          profile: test-profile
+          tenancy: Exclusive
+    shared-unlimited:
+      template:
+        metadata:
+          namespace: shared-unlimited
+        spec:
+          profile: test-profile
+          tenancy: Shared
+    shared-twice:
+      tenancyCount: 2
+      template:
+        metadata:
+          namespace: shared-twice
+        spec:
+          profile: test-profile
+          tenancy: Shared
diff --git a/internal/controllers/scheduler/testdata/test-02/cluster/cluster-0.yaml b/internal/controllers/scheduler/testdata/test-02/cluster/cluster-0.yaml
new file mode 100644
index 0000000..bcc7fc7
--- /dev/null
+++ b/internal/controllers/scheduler/testdata/test-02/cluster/cluster-0.yaml
@@ -0,0 +1,13 @@
+apiVersion: clusters.openmcp.cloud/v1alpha1
+kind: Cluster
+metadata:
+  name: out-of-scope
+  namespace: foo
+spec:
+  profile: test-profile
+  kubernetes:
+    version: "1.32"
+  purposes:
+  - test
+  - shared-unlimited
+  tenancy: Shared
diff --git a/internal/controllers/scheduler/testdata/test-02/cluster/req-0.yaml b/internal/controllers/scheduler/testdata/test-02/cluster/req-0.yaml
new file mode 100644
index 0000000..6a01132
--- /dev/null
+++ b/internal/controllers/scheduler/testdata/test-02/cluster/req-0.yaml
@@ -0,0 +1,8 @@
+apiVersion: clusters.openmcp.cloud/v1alpha1
+kind: ClusterRequest
+metadata:
+  name: exclusive
+  namespace: foo
+  uid: dc4b422b-8676-4bc0-951f-7a535d790c04
+spec:
+  purpose: exclusive
diff --git a/internal/controllers/scheduler/testdata/test-02/cluster/req-1.yaml b/internal/controllers/scheduler/testdata/test-02/cluster/req-1.yaml
new file mode 100644
index 0000000..897f6aa
--- /dev/null
+++ b/internal/controllers/scheduler/testdata/test-02/cluster/req-1.yaml
@@ -0,0 +1,8 @@
+apiVersion: clusters.openmcp.cloud/v1alpha1
+kind: ClusterRequest
+metadata:
+  name: shared
+  namespace: foo
+  uid: 4d6aa495-54c0-4df2-bc7b-05b103f02e69
+spec:
+  purpose: shared-unlimited
diff --git a/internal/controllers/scheduler/testdata/test-02/cluster/req-2.yaml b/internal/controllers/scheduler/testdata/test-02/cluster/req-2.yaml
new file mode 100644
index 0000000..b25b4e4
--- /dev/null
+++ b/internal/controllers/scheduler/testdata/test-02/cluster/req-2.yaml
@@ -0,0 +1,8 @@
+apiVersion: clusters.openmcp.cloud/v1alpha1
+kind: ClusterRequest
+metadata:
+  name: shared
+  namespace: bar
+  uid: bf7d8a52-ea1e-4303-a0f9-bf2ef0cc21ea
+spec:
+  purpose: shared-unlimited
diff --git a/internal/controllers/scheduler/testdata/test-02/cluster/req-3.yaml b/internal/controllers/scheduler/testdata/test-02/cluster/req-3.yaml
new file mode 100644
index 0000000..21268ff
--- /dev/null
+++ b/internal/controllers/scheduler/testdata/test-02/cluster/req-3.yaml
@@ -0,0 +1,8 @@
+apiVersion: clusters.openmcp.cloud/v1alpha1
+kind: ClusterRequest
+metadata:
+  name: shared2
+  namespace: foo
+  uid: 7642e165-196c-478f-a225-765bd97a29d8
+spec:
+  purpose: shared-unlimited
diff --git a/internal/controllers/scheduler/testdata/test-02/config.yaml b/internal/controllers/scheduler/testdata/test-02/config.yaml
new file mode 100644
index 0000000..94a2284
--- /dev/null
+++ b/internal/controllers/scheduler/testdata/test-02/config.yaml
@@ -0,0 +1,25 @@
+scheduler:
+  selectors:
+    clusters:
+      matchLabels:
+        foo.bar.baz/foobar: "true"
+  purposeMappings:
+    exclusive:
+      template:
+        metadata:
+          labels:
+            foo.bar.baz/foobar: "true"
+          annotations:
+            foo.bar.baz/foobar: "false"
+        spec:
+          profile: test-profile
+          tenancy: Exclusive
+    shared-unlimited:
+      template:
+        metadata:
+          name: singleton
+          labels:
+            foo.bar.baz/foobar: "true"
+        spec:
+          profile: test-profile
+          tenancy: Shared
diff --git a/internal/controllers/scheduler/testdata/test-03/cluster/cluster-0.yaml b/internal/controllers/scheduler/testdata/test-03/cluster/cluster-0.yaml
new file mode 100644
index 0000000..38bd334
--- /dev/null
+++ b/internal/controllers/scheduler/testdata/test-03/cluster/cluster-0.yaml
@@ -0,0 +1,13 @@
+apiVersion: clusters.openmcp.cloud/v1alpha1
+kind: Cluster
+metadata:
+  name: out-of-scope
+  namespace: asdf
+spec:
+  profile: test-profile
+  kubernetes:
+    version: "1.32"
+  purposes:
+  - test
+  - shared-unlimited
+  tenancy: Shared
diff --git a/internal/controllers/scheduler/testdata/test-03/cluster/req-0.yaml b/internal/controllers/scheduler/testdata/test-03/cluster/req-0.yaml
new file mode 100644
index 0000000..897f6aa
--- /dev/null
+++ b/internal/controllers/scheduler/testdata/test-03/cluster/req-0.yaml
@@ -0,0 +1,8 @@
+apiVersion: clusters.openmcp.cloud/v1alpha1
+kind: ClusterRequest
+metadata:
+  name: shared
+  namespace: foo
+  uid: 4d6aa495-54c0-4df2-bc7b-05b103f02e69
+spec:
+  purpose: shared-unlimited
diff --git a/internal/controllers/scheduler/testdata/test-03/cluster/req-1.yaml b/internal/controllers/scheduler/testdata/test-03/cluster/req-1.yaml
new file mode 100644
index 0000000..35940f0
--- /dev/null
+++ b/internal/controllers/scheduler/testdata/test-03/cluster/req-1.yaml
@@ -0,0 +1,8 @@
+apiVersion: clusters.openmcp.cloud/v1alpha1
+kind: ClusterRequest
+metadata:
+  name: shared2
+  namespace: bar
+  uid: bf7d8a52-ea1e-4303-a0f9-bf2ef0cc21ea
+spec:
+  purpose: shared-unlimited
diff --git a/internal/controllers/scheduler/testdata/test-03/cluster/req-2.yaml b/internal/controllers/scheduler/testdata/test-03/cluster/req-2.yaml
new file mode 100644
index 0000000..a2f0ee6
--- /dev/null
+++ b/internal/controllers/scheduler/testdata/test-03/cluster/req-2.yaml
@@ -0,0 +1,8 @@
+apiVersion: clusters.openmcp.cloud/v1alpha1
+kind: ClusterRequest
+metadata:
+  name: shared3
+  namespace: baz
+  uid: 7642e165-196c-478f-a225-765bd97a29d8
+spec:
+  purpose: shared-unlimited
diff --git a/internal/controllers/scheduler/testdata/test-03/config.yaml b/internal/controllers/scheduler/testdata/test-03/config.yaml
new file mode 100644
index 0000000..4f9e60f
--- /dev/null
+++ b/internal/controllers/scheduler/testdata/test-03/config.yaml
@@ -0,0 +1,16 @@
+scheduler:
+  scope: Cluster
+  selectors:
+    clusters:
+      matchLabels:
+        foo.bar.baz/foobar: "true"
+  purposeMappings:
+    shared-unlimited:
+      template:
+        metadata:
+          name: singleton
+          labels:
+            foo.bar.baz/foobar: "true"
+        spec:
+          profile: test-profile
+          tenancy: Shared
diff --git a/internal/controllers/scheduler/testdata/test-04/cluster/cluster-0.yaml b/internal/controllers/scheduler/testdata/test-04/cluster/cluster-0.yaml
new file mode 100644
index 0000000..f38cdca
--- /dev/null
+++ b/internal/controllers/scheduler/testdata/test-04/cluster/cluster-0.yaml
@@ -0,0 +1,16 @@
+apiVersion: clusters.openmcp.cloud/v1alpha1
+kind: Cluster
+metadata:
+  name: shared
+  namespace: foo
+  labels:
+    foo.bar.baz/foobar: "true"
+spec:
+  profile: test-profile
+  kubernetes:
+    version: "1.32"
+  purposes:
+  - test
+  - shared
+  - shared2
+  tenancy: Shared
diff --git a/internal/controllers/scheduler/testdata/test-04/cluster/req-0.yaml b/internal/controllers/scheduler/testdata/test-04/cluster/req-0.yaml
new file mode 100644
index 0000000..5d97d47
--- /dev/null
+++ b/internal/controllers/scheduler/testdata/test-04/cluster/req-0.yaml
@@ -0,0 +1,8 @@
+apiVersion: clusters.openmcp.cloud/v1alpha1
+kind: ClusterRequest
+metadata:
+  name: shared
+  namespace: foo
+  uid: 4d6aa495-54c0-4df2-bc7b-05b103f02e69
+spec:
+  purpose: shared
diff --git a/internal/controllers/scheduler/testdata/test-04/cluster/req-1.yaml b/internal/controllers/scheduler/testdata/test-04/cluster/req-1.yaml
new file mode 100644
index 0000000..f1a265a
--- /dev/null
+++ b/internal/controllers/scheduler/testdata/test-04/cluster/req-1.yaml
@@ -0,0 +1,8 @@
+apiVersion: clusters.openmcp.cloud/v1alpha1
+kind: ClusterRequest
+metadata:
+  name: shared2
+  namespace: foo
+  uid: bf7d8a52-ea1e-4303-a0f9-bf2ef0cc21ea
+spec:
+  purpose: shared2
diff --git a/internal/controllers/scheduler/testdata/test-04/cluster/req-2.yaml b/internal/controllers/scheduler/testdata/test-04/cluster/req-2.yaml
new file mode 100644
index 0000000..eb38a30
--- /dev/null
+++ b/internal/controllers/scheduler/testdata/test-04/cluster/req-2.yaml
@@ -0,0 +1,8 @@
+apiVersion: clusters.openmcp.cloud/v1alpha1
+kind: ClusterRequest
+metadata:
+  name: shared3
+  namespace: foo
+  uid: 7642e165-196c-478f-a225-765bd97a29d8
+spec:
+  purpose: shared2
diff --git a/internal/controllers/scheduler/testdata/test-04/config.yaml b/internal/controllers/scheduler/testdata/test-04/config.yaml
new file mode 100644
index 0000000..a8e4d64
--- /dev/null
+++ b/internal/controllers/scheduler/testdata/test-04/config.yaml
@@ -0,0 +1,27 @@
+scheduler:
+  scope: Cluster
+  selectors:
+    clusters:
+      matchLabels:
+        foo.bar.baz/foobar: "true"
+  purposeMappings:
+    shared:
+      template:
+        metadata:
+          labels:
+            foo.bar.baz/foobar: "true"
+        spec:
+          profile: test-profile
+          tenancy: Shared
+    shared2:
+      selector:
+        matchLabels:
+          foo.bar.baz/foobaz: "true"
+      template:
+        metadata:
+          labels:
+            foo.bar.baz/foobar: "true"
+            foo.bar.baz/foobaz: "true"
+        spec:
+          profile: test-profile
+          tenancy: Shared
diff --git a/internal/controllers/scheduler/testdata/test-05/cluster/req-0.yaml b/internal/controllers/scheduler/testdata/test-05/cluster/req-0.yaml
new file mode 100644
index 0000000..0f31b84
--- /dev/null
+++ b/internal/controllers/scheduler/testdata/test-05/cluster/req-0.yaml
@@ -0,0 +1,8 @@
+apiVersion: clusters.openmcp.cloud/v1alpha1
+kind: ClusterRequest
+metadata:
+  name: delete
+  namespace: foo
+  uid: 4d6aa495-54c0-4df2-bc7b-05b103f02e69
+spec:
+  purpose: delete
diff --git a/internal/controllers/scheduler/testdata/test-05/cluster/req-1.yaml b/internal/controllers/scheduler/testdata/test-05/cluster/req-1.yaml
new file mode 100644
index 0000000..3229736
--- /dev/null
+++ b/internal/controllers/scheduler/testdata/test-05/cluster/req-1.yaml
@@ -0,0 +1,8 @@
+apiVersion: clusters.openmcp.cloud/v1alpha1
+kind: ClusterRequest
+metadata:
+  name: no-delete
+  namespace: foo
+  uid: bf7d8a52-ea1e-4303-a0f9-bf2ef0cc21ea
+spec:
+  purpose: no-delete
diff --git a/internal/controllers/scheduler/testdata/test-05/config.yaml b/internal/controllers/scheduler/testdata/test-05/config.yaml
new file mode 100644
index 0000000..f2644db
--- /dev/null
+++ b/internal/controllers/scheduler/testdata/test-05/config.yaml
@@ -0,0 +1,16 @@
+scheduler:
+  scope: Cluster
+  purposeMappings:
+    delete:
+      template:
+        spec:
+          profile: test-profile
+          tenancy: Shared
+    no-delete:
+      template:
+        metadata:
+          labels:
+            clusters.openmcp.cloud/delete-without-requests: "false"
+        spec:
+          profile: test-profile
+          tenancy: Shared
diff --git a/test/e2e/e2e_suite_test.go b/test/e2e/e2e_suite_test.go
deleted file mode 100644
index 71ba077..0000000
--- a/test/e2e/e2e_suite_test.go
+++ /dev/null
@@ -1,89 +0,0 @@
-/*
-Copyright 2025.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package e2e
-
-import (
-	"fmt"
-	"os"
-	"os/exec"
-	"testing"
-
-	. "github.com/onsi/ginkgo/v2"
-	. "github.com/onsi/gomega"
-
-	"github.com/openmcp-project/openmcp-operator/test/utils"
-)
-
-var (
-	// Optional Environment Variables:
-	// - CERT_MANAGER_INSTALL_SKIP=true: Skips CertManager installation during test setup.
-	// These variables are useful if CertManager is already installed, avoiding
-	// re-installation and conflicts.
-	skipCertManagerInstall = os.Getenv("CERT_MANAGER_INSTALL_SKIP") == "true"
-	// isCertManagerAlreadyInstalled will be set true when CertManager CRDs be found on the cluster
-	isCertManagerAlreadyInstalled = false
-
-	// projectImage is the name of the image which will be build and loaded
-	// with the code source changes to be tested.
-	projectImage = "example.com/openmcp-operator:v0.0.1"
-)
-
-// TestE2E runs the end-to-end (e2e) test suite for the project. These tests execute in an isolated,
-// temporary environment to validate project changes with the the purposed to be used in CI jobs.
-// The default setup requires Kind, builds/loads the Manager Docker image locally, and installs
-// CertManager.
-func TestE2E(t *testing.T) {
-	//  RegisterFailHandler(Fail)
-	//  _, _ = fmt.Fprintf(GinkgoWriter, "Starting openmcp-operator integration test suite\n")
-	//  RunSpecs(t, "e2e suite")
-}
-
-var _ = BeforeSuite(func() {
-	By("building the manager(Operator) image")
-	cmd := exec.Command("make", "docker-build", fmt.Sprintf("IMG=%s", projectImage))
-	_, err := utils.Run(cmd)
-	ExpectWithOffset(1, err).NotTo(HaveOccurred(), "Failed to build the manager(Operator) image")
-
-	// TODO(user): If you want to change the e2e test vendor from Kind, ensure the image is
-	// built and available before running the tests. Also, remove the following block.
-	By("loading the manager(Operator) image on Kind")
-	err = utils.LoadImageToKindClusterWithName(projectImage)
-	ExpectWithOffset(1, err).NotTo(HaveOccurred(), "Failed to load the manager(Operator) image into Kind")
-
-	// The tests-e2e are intended to run on a temporary cluster that is created and destroyed for testing.
-	// To prevent errors when tests run in environments with CertManager already installed,
-	// we check for its presence before execution.
-	// Setup CertManager before the suite if not skipped and if not already installed
-	if !skipCertManagerInstall {
-		By("checking if cert manager is installed already")
-		isCertManagerAlreadyInstalled = utils.IsCertManagerCRDsInstalled()
-		if !isCertManagerAlreadyInstalled {
-			_, _ = fmt.Fprintf(GinkgoWriter, "Installing CertManager...\n")
-			Expect(utils.InstallCertManager()).To(Succeed(), "Failed to install CertManager")
-		} else {
-			_, _ = fmt.Fprintf(GinkgoWriter, "WARNING: CertManager is already installed. Skipping installation...\n")
-		}
-	}
-})
-
-var _ = AfterSuite(func() {
-	// Teardown CertManager after the suite if not skipped and if it was not already installed
-	if !skipCertManagerInstall && !isCertManagerAlreadyInstalled {
-		_, _ = fmt.Fprintf(GinkgoWriter, "Uninstalling CertManager...\n")
-		utils.UninstallCertManager()
-	}
-})
diff --git a/test/e2e/e2e_test.go b/test/e2e/e2e_test.go
deleted file mode 100644
index d3dd76d..0000000
--- a/test/e2e/e2e_test.go
+++ /dev/null
@@ -1,329 +0,0 @@
-/*
-Copyright 2025.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package e2e
-
-import (
-	"encoding/json"
-	"fmt"
-	"os"
-	"os/exec"
-	"path/filepath"
-	"time"
-
-	. "github.com/onsi/ginkgo/v2"
-	. "github.com/onsi/gomega"
-
-	"github.com/openmcp-project/openmcp-operator/test/utils"
-)
-
-// namespace where the project is deployed in
-const namespace = "openmcp-operator-system"
-
-// serviceAccountName created for the project
-const serviceAccountName = "openmcp-operator-controller-manager"
-
-// metricsServiceName is the name of the metrics service of the project
-const metricsServiceName = "openmcp-operator-controller-manager-metrics-service"
-
-// metricsRoleBindingName is the name of the RBAC that will be created to allow get the metrics data
-const metricsRoleBindingName = "openmcp-operator-metrics-binding"
-
-var _ = Describe("Manager", Ordered, func() {
-	var controllerPodName string
-
-	// Before running the tests, set up the environment by creating the namespace,
-	// enforce the restricted security policy to the namespace, installing CRDs,
-	// and deploying the controller.
-	BeforeAll(func() {
-		By("creating manager namespace")
-		cmd := exec.Command("kubectl", "create", "ns", namespace)
-		_, err := utils.Run(cmd)
-		Expect(err).NotTo(HaveOccurred(), "Failed to create namespace")
-
-		By("labeling the namespace to enforce the restricted security policy")
-		cmd = exec.Command("kubectl", "label", "--overwrite", "ns", namespace,
-			"pod-security.kubernetes.io/enforce=restricted")
-		_, err = utils.Run(cmd)
-		Expect(err).NotTo(HaveOccurred(), "Failed to label namespace with restricted policy")
-
-		By("installing CRDs")
-		cmd = exec.Command("make", "install")
-		_, err = utils.Run(cmd)
-		Expect(err).NotTo(HaveOccurred(), "Failed to install CRDs")
-
-		By("deploying the controller-manager")
-		cmd = exec.Command("make", "deploy", fmt.Sprintf("IMG=%s", projectImage))
-		_, err = utils.Run(cmd)
-		Expect(err).NotTo(HaveOccurred(), "Failed to deploy the controller-manager")
-	})
-
-	// After all tests have been executed, clean up by undeploying the controller, uninstalling CRDs,
-	// and deleting the namespace.
-	AfterAll(func() {
-		By("cleaning up the curl pod for metrics")
-		cmd := exec.Command("kubectl", "delete", "pod", "curl-metrics", "-n", namespace)
-		_, _ = utils.Run(cmd)
-
-		By("undeploying the controller-manager")
-		cmd = exec.Command("make", "undeploy")
-		_, _ = utils.Run(cmd)
-
-		By("uninstalling CRDs")
-		cmd = exec.Command("make", "uninstall")
-		_, _ = utils.Run(cmd)
-
-		By("removing manager namespace")
-		cmd = exec.Command("kubectl", "delete", "ns", namespace)
-		_, _ = utils.Run(cmd)
-	})
-
-	// After each test, check for failures and collect logs, events,
-	// and pod descriptions for debugging.
-	AfterEach(func() {
-		specReport := CurrentSpecReport()
-		if specReport.Failed() {
-			By("Fetching controller manager pod logs")
-			cmd := exec.Command("kubectl", "logs", controllerPodName, "-n", namespace)
-			controllerLogs, err := utils.Run(cmd)
-			if err == nil {
-				_, _ = fmt.Fprintf(GinkgoWriter, "Controller logs:\n %s", controllerLogs)
-			} else {
-				_, _ = fmt.Fprintf(GinkgoWriter, "Failed to get Controller logs: %s", err)
-			}
-
-			By("Fetching Kubernetes events")
-			cmd = exec.Command("kubectl", "get", "events", "-n", namespace, "--sort-by=.lastTimestamp")
-			eventsOutput, err := utils.Run(cmd)
-			if err == nil {
-				_, _ = fmt.Fprintf(GinkgoWriter, "Kubernetes events:\n%s", eventsOutput)
-			} else {
-				_, _ = fmt.Fprintf(GinkgoWriter, "Failed to get Kubernetes events: %s", err)
-			}
-
-			By("Fetching curl-metrics logs")
-			cmd = exec.Command("kubectl", "logs", "curl-metrics", "-n", namespace)
-			metricsOutput, err := utils.Run(cmd)
-			if err == nil {
-				_, _ = fmt.Fprintf(GinkgoWriter, "Metrics logs:\n %s", metricsOutput)
-			} else {
-				_, _ = fmt.Fprintf(GinkgoWriter, "Failed to get curl-metrics logs: %s", err)
-			}
-
-			By("Fetching controller manager pod description")
-			cmd = exec.Command("kubectl", "describe", "pod", controllerPodName, "-n", namespace)
-			podDescription, err := utils.Run(cmd)
-			if err == nil {
-				fmt.Println("Pod description:\n", podDescription)
-			} else {
-				fmt.Println("Failed to describe controller pod")
-			}
-		}
-	})
-
-	SetDefaultEventuallyTimeout(2 * time.Minute)
-	SetDefaultEventuallyPollingInterval(time.Second)
-
-	Context("Manager", func() {
-		It("should run successfully", func() {
-			By("validating that the controller-manager pod is running as expected")
-			verifyControllerUp := func(g Gomega) {
-				// Get the name of the controller-manager pod
-				cmd := exec.Command("kubectl", "get",
-					"pods", "-l", "control-plane=controller-manager",
-					"-o", "go-template={{ range .items }}"+
-						"{{ if not .metadata.deletionTimestamp }}"+
-						"{{ .metadata.name }}"+
-						"{{ \"\\n\" }}{{ end }}{{ end }}",
-					"-n", namespace,
-				)
-
-				podOutput, err := utils.Run(cmd)
-				g.Expect(err).NotTo(HaveOccurred(), "Failed to retrieve controller-manager pod information")
-				podNames := utils.GetNonEmptyLines(podOutput)
-				g.Expect(podNames).To(HaveLen(1), "expected 1 controller pod running")
-				controllerPodName = podNames[0]
-				g.Expect(controllerPodName).To(ContainSubstring("controller-manager"))
-
-				// Validate the pod's status
-				cmd = exec.Command("kubectl", "get",
-					"pods", controllerPodName, "-o", "jsonpath={.status.phase}",
-					"-n", namespace,
-				)
-				output, err := utils.Run(cmd)
-				g.Expect(err).NotTo(HaveOccurred())
-				g.Expect(output).To(Equal("Running"), "Incorrect controller-manager pod status")
-			}
-			Eventually(verifyControllerUp).Should(Succeed())
-		})
-
-		It("should ensure the metrics endpoint is serving metrics", func() {
-			By("creating a ClusterRoleBinding for the service account to allow access to metrics")
-			cmd := exec.Command("kubectl", "create", "clusterrolebinding", metricsRoleBindingName,
-				"--clusterrole=openmcp-operator-metrics-reader",
-				fmt.Sprintf("--serviceaccount=%s:%s", namespace, serviceAccountName),
-			)
-			_, err := utils.Run(cmd)
-			Expect(err).NotTo(HaveOccurred(), "Failed to create ClusterRoleBinding")
-
-			By("validating that the metrics service is available")
-			cmd = exec.Command("kubectl", "get", "service", metricsServiceName, "-n", namespace)
-			_, err = utils.Run(cmd)
-			Expect(err).NotTo(HaveOccurred(), "Metrics service should exist")
-
-			By("getting the service account token")
-			token, err := serviceAccountToken()
-			Expect(err).NotTo(HaveOccurred())
-			Expect(token).NotTo(BeEmpty())
-
-			By("waiting for the metrics endpoint to be ready")
-			verifyMetricsEndpointReady := func(g Gomega) {
-				cmd := exec.Command("kubectl", "get", "endpoints", metricsServiceName, "-n", namespace)
-				output, err := utils.Run(cmd)
-				g.Expect(err).NotTo(HaveOccurred())
-				g.Expect(output).To(ContainSubstring("8443"), "Metrics endpoint is not ready")
-			}
-			Eventually(verifyMetricsEndpointReady).Should(Succeed())
-
-			By("verifying that the controller manager is serving the metrics server")
-			verifyMetricsServerStarted := func(g Gomega) {
-				cmd := exec.Command("kubectl", "logs", controllerPodName, "-n", namespace)
-				output, err := utils.Run(cmd)
-				g.Expect(err).NotTo(HaveOccurred())
-				g.Expect(output).To(ContainSubstring("controller-runtime.metrics\tServing metrics server"),
-					"Metrics server not yet started")
-			}
-			Eventually(verifyMetricsServerStarted).Should(Succeed())
-
-			By("creating the curl-metrics pod to access the metrics endpoint")
-			cmd = exec.Command("kubectl", "run", "curl-metrics", "--restart=Never",
-				"--namespace", namespace,
-				"--image=curlimages/curl:latest",
-				"--overrides",
-				fmt.Sprintf(`{
-					"spec": {
-						"containers": [{
-							"name": "curl",
-							"image": "curlimages/curl:latest",
-							"command": ["/bin/sh", "-c"],
-							"args": ["curl -v -k -H 'Authorization: Bearer %s' https://%s.%s.svc.cluster.local:8443/metrics"],
-							"securityContext": {
-								"allowPrivilegeEscalation": false,
-								"capabilities": {
-									"drop": ["ALL"]
-								},
-								"runAsNonRoot": true,
-								"runAsUser": 1000,
-								"seccompProfile": {
-									"type": "RuntimeDefault"
-								}
-							}
-						}],
-						"serviceAccount": "%s"
-					}
-				}`, token, metricsServiceName, namespace, serviceAccountName))
-			_, err = utils.Run(cmd)
-			Expect(err).NotTo(HaveOccurred(), "Failed to create curl-metrics pod")
-
-			By("waiting for the curl-metrics pod to complete.")
-			verifyCurlUp := func(g Gomega) {
-				cmd := exec.Command("kubectl", "get", "pods", "curl-metrics",
-					"-o", "jsonpath={.status.phase}",
-					"-n", namespace)
-				output, err := utils.Run(cmd)
-				g.Expect(err).NotTo(HaveOccurred())
-				g.Expect(output).To(Equal("Succeeded"), "curl pod in wrong status")
-			}
-			Eventually(verifyCurlUp, 5*time.Minute).Should(Succeed())
-
-			By("getting the metrics by checking curl-metrics logs")
-			metricsOutput := getMetricsOutput()
-			Expect(metricsOutput).To(ContainSubstring(
-				"controller_runtime_reconcile_total",
-			))
-		})
-
-		// +kubebuilder:scaffold:e2e-webhooks-checks
-
-		// TODO: Customize the e2e test suite with scenarios specific to your project.
-		// Consider applying sample/CR(s) and check their status and/or verifying
-		// the reconciliation by using the metrics, i.e.:
-		// metricsOutput := getMetricsOutput()
-		// Expect(metricsOutput).To(ContainSubstring(
-		//    fmt.Sprintf(`controller_runtime_reconcile_total{controller="%s",result="success"} 1`,
-		//    strings.ToLower(<Kind>),
-		// ))
-	})
-})
-
-// serviceAccountToken returns a token for the specified service account in the given namespace.
-// It uses the Kubernetes TokenRequest API to generate a token by directly sending a request
-// and parsing the resulting token from the API response.
-func serviceAccountToken() (string, error) {
-	const tokenRequestRawString = `{
-		"apiVersion": "authentication.k8s.io/v1",
-		"kind": "TokenRequest"
-	}`
-
-	// Temporary file to store the token request
-	secretName := fmt.Sprintf("%s-token-request", serviceAccountName)
-	tokenRequestFile := filepath.Join("/tmp", secretName)
-	err := os.WriteFile(tokenRequestFile, []byte(tokenRequestRawString), os.FileMode(0o644))
-	if err != nil {
-		return "", err
-	}
-
-	var out string
-	verifyTokenCreation := func(g Gomega) {
-		// Execute kubectl command to create the token
-		cmd := exec.Command("kubectl", "create", "--raw", fmt.Sprintf(
-			"/api/v1/namespaces/%s/serviceaccounts/%s/token",
-			namespace,
-			serviceAccountName,
-		), "-f", tokenRequestFile)
-
-		output, err := cmd.CombinedOutput()
-		g.Expect(err).NotTo(HaveOccurred())
-
-		// Parse the JSON output to extract the token
-		var token tokenRequest
-		err = json.Unmarshal(output, &token)
-		g.Expect(err).NotTo(HaveOccurred())
-
-		out = token.Status.Token
-	}
-	Eventually(verifyTokenCreation).Should(Succeed())
-
-	return out, err
-}
-
-// getMetricsOutput retrieves and returns the logs from the curl pod used to access the metrics endpoint.
-func getMetricsOutput() string {
-	By("getting the curl-metrics logs")
-	cmd := exec.Command("kubectl", "logs", "curl-metrics", "-n", namespace)
-	metricsOutput, err := utils.Run(cmd)
-	Expect(err).NotTo(HaveOccurred(), "Failed to retrieve logs from curl pod")
-	Expect(metricsOutput).To(ContainSubstring("< HTTP/1.1 200 OK"))
-	return metricsOutput
-}
-
-// tokenRequest is a simplified representation of the Kubernetes TokenRequest API response,
-// containing only the token field that we need to extract.
-type tokenRequest struct {
-	Status struct {
-		Token string `json:"token"`
-	} `json:"status"`
-}
diff --git a/test/utils/utils.go b/test/utils/utils.go
deleted file mode 100644
index 04a5141..0000000
--- a/test/utils/utils.go
+++ /dev/null
@@ -1,251 +0,0 @@
-/*
-Copyright 2025.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package utils
-
-import (
-	"bufio"
-	"bytes"
-	"fmt"
-	"os"
-	"os/exec"
-	"strings"
-
-	. "github.com/onsi/ginkgo/v2" //nolint:golint,revive
-)
-
-const (
-	prometheusOperatorVersion = "v0.77.1"
-	prometheusOperatorURL     = "https://github.com/prometheus-operator/prometheus-operator/" +
-		"releases/download/%s/bundle.yaml"
-
-	certmanagerVersion = "v1.16.3"
-	certmanagerURLTmpl = "https://github.com/cert-manager/cert-manager/releases/download/%s/cert-manager.yaml"
-)
-
-func warnError(err error) {
-	_, _ = fmt.Fprintf(GinkgoWriter, "warning: %v\n", err)
-}
-
-// Run executes the provided command within this context
-func Run(cmd *exec.Cmd) (string, error) {
-	dir, _ := GetProjectDir()
-	cmd.Dir = dir
-
-	if err := os.Chdir(cmd.Dir); err != nil {
-		_, _ = fmt.Fprintf(GinkgoWriter, "chdir dir: %s\n", err)
-	}
-
-	cmd.Env = append(os.Environ(), "GO111MODULE=on")
-	command := strings.Join(cmd.Args, " ")
-	_, _ = fmt.Fprintf(GinkgoWriter, "running: %s\n", command)
-	output, err := cmd.CombinedOutput()
-	if err != nil {
-		return string(output), fmt.Errorf("%s failed with error: (%v) %s", command, err, string(output))
-	}
-
-	return string(output), nil
-}
-
-// InstallPrometheusOperator installs the prometheus Operator to be used to export the enabled metrics.
-func InstallPrometheusOperator() error {
-	url := fmt.Sprintf(prometheusOperatorURL, prometheusOperatorVersion)
-	cmd := exec.Command("kubectl", "create", "-f", url)
-	_, err := Run(cmd)
-	return err
-}
-
-// UninstallPrometheusOperator uninstalls the prometheus
-func UninstallPrometheusOperator() {
-	url := fmt.Sprintf(prometheusOperatorURL, prometheusOperatorVersion)
-	cmd := exec.Command("kubectl", "delete", "-f", url)
-	if _, err := Run(cmd); err != nil {
-		warnError(err)
-	}
-}
-
-// IsPrometheusCRDsInstalled checks if any Prometheus CRDs are installed
-// by verifying the existence of key CRDs related to Prometheus.
-func IsPrometheusCRDsInstalled() bool {
-	// List of common Prometheus CRDs
-	prometheusCRDs := []string{
-		"prometheuses.monitoring.coreos.com",
-		"prometheusrules.monitoring.coreos.com",
-		"prometheusagents.monitoring.coreos.com",
-	}
-
-	cmd := exec.Command("kubectl", "get", "crds", "-o", "custom-columns=NAME:.metadata.name")
-	output, err := Run(cmd)
-	if err != nil {
-		return false
-	}
-	crdList := GetNonEmptyLines(output)
-	for _, crd := range prometheusCRDs {
-		for _, line := range crdList {
-			if strings.Contains(line, crd) {
-				return true
-			}
-		}
-	}
-
-	return false
-}
-
-// UninstallCertManager uninstalls the cert manager
-func UninstallCertManager() {
-	url := fmt.Sprintf(certmanagerURLTmpl, certmanagerVersion)
-	cmd := exec.Command("kubectl", "delete", "-f", url)
-	if _, err := Run(cmd); err != nil {
-		warnError(err)
-	}
-}
-
-// InstallCertManager installs the cert manager bundle.
-func InstallCertManager() error {
-	url := fmt.Sprintf(certmanagerURLTmpl, certmanagerVersion)
-	cmd := exec.Command("kubectl", "apply", "-f", url)
-	if _, err := Run(cmd); err != nil {
-		return err
-	}
-	// Wait for cert-manager-webhook to be ready, which can take time if cert-manager
-	// was re-installed after uninstalling on a cluster.
-	cmd = exec.Command("kubectl", "wait", "deployment.apps/cert-manager-webhook",
-		"--for", "condition=Available",
-		"--namespace", "cert-manager",
-		"--timeout", "5m",
-	)
-
-	_, err := Run(cmd)
-	return err
-}
-
-// IsCertManagerCRDsInstalled checks if any Cert Manager CRDs are installed
-// by verifying the existence of key CRDs related to Cert Manager.
-func IsCertManagerCRDsInstalled() bool {
-	// List of common Cert Manager CRDs
-	certManagerCRDs := []string{
-		"certificates.cert-manager.io",
-		"issuers.cert-manager.io",
-		"clusterissuers.cert-manager.io",
-		"certificaterequests.cert-manager.io",
-		"orders.acme.cert-manager.io",
-		"challenges.acme.cert-manager.io",
-	}
-
-	// Execute the kubectl command to get all CRDs
-	cmd := exec.Command("kubectl", "get", "crds")
-	output, err := Run(cmd)
-	if err != nil {
-		return false
-	}
-
-	// Check if any of the Cert Manager CRDs are present
-	crdList := GetNonEmptyLines(output)
-	for _, crd := range certManagerCRDs {
-		for _, line := range crdList {
-			if strings.Contains(line, crd) {
-				return true
-			}
-		}
-	}
-
-	return false
-}
-
-// LoadImageToKindClusterWithName loads a local docker image to the kind cluster
-func LoadImageToKindClusterWithName(name string) error {
-	cluster := "kind"
-	if v, ok := os.LookupEnv("KIND_CLUSTER"); ok {
-		cluster = v
-	}
-	kindOptions := []string{"load", "docker-image", name, "--name", cluster}
-	cmd := exec.Command("kind", kindOptions...)
-	_, err := Run(cmd)
-	return err
-}
-
-// GetNonEmptyLines converts given command output string into individual objects
-// according to line breakers, and ignores the empty elements in it.
-func GetNonEmptyLines(output string) []string {
-	var res []string
-	elements := strings.Split(output, "\n")
-	for _, element := range elements {
-		if element != "" {
-			res = append(res, element)
-		}
-	}
-
-	return res
-}
-
-// GetProjectDir will return the directory where the project is
-func GetProjectDir() (string, error) {
-	wd, err := os.Getwd()
-	if err != nil {
-		return wd, err
-	}
-	wd = strings.Replace(wd, "/test/e2e", "", -1)
-	return wd, nil
-}
-
-// UncommentCode searches for target in the file and remove the comment prefix
-// of the target content. The target content may span multiple lines.
-func UncommentCode(filename, target, prefix string) error {
-	// false positive
-	// nolint:gosec
-	content, err := os.ReadFile(filename)
-	if err != nil {
-		return err
-	}
-	strContent := string(content)
-
-	idx := strings.Index(strContent, target)
-	if idx < 0 {
-		return fmt.Errorf("unable to find the code %s to be uncomment", target)
-	}
-
-	out := new(bytes.Buffer)
-	_, err = out.Write(content[:idx])
-	if err != nil {
-		return err
-	}
-
-	scanner := bufio.NewScanner(bytes.NewBufferString(target))
-	if !scanner.Scan() {
-		return nil
-	}
-	for {
-		_, err := out.WriteString(strings.TrimPrefix(scanner.Text(), prefix))
-		if err != nil {
-			return err
-		}
-		// Avoid writing a newline in case the previous line was the last in target.
-		if !scanner.Scan() {
-			break
-		}
-		if _, err := out.WriteString("\n"); err != nil {
-			return err
-		}
-	}
-
-	_, err = out.Write(content[idx+len(target):])
-	if err != nil {
-		return err
-	}
-	// false positive
-	// nolint:gosec
-	return os.WriteFile(filename, out.Bytes(), 0644)
-}