make scheduling logic independent from API types

Author: Diaphteiros

api/clusters/v1alpha1/constants/reasons.go

@@ -13,4 +13,6 @@ const (
 	ReasonInternalError = "InternalError"
 	// ReasonPreemptiveRequest indicates that the ClusterRequest is preemptive and AccessRequests referencing it are denied.
 	ReasonPreemptiveRequest = "PreemptiveRequest"
+	// ReasonSchedulingFailed indicates that there was a problem with scheduling a request.
+	ReasonSchedulingFailed = "SchedulingFailed"
 )

internal/config/config_scheduler.go

@@ -42,9 +42,10 @@ const (
 type Strategy string
 
 const (
-	STRATEGY_BALANCED Strategy = "Balanced"
-	STRATEGY_RANDOM   Strategy = "Random"
-	STRATEGY_SIMPLE   Strategy = "Simple"
+	STRATEGY_BALANCED_IGNORE_EMPTY Strategy = "BalancedIgnoreEmpty"
+	STRATEGY_BALANCED              Strategy = "Balanced"
+	STRATEGY_RANDOM                Strategy = "Random"
+	STRATEGY_SIMPLE                Strategy = "Simple"
 )
 
 type ClusterDefinition struct {

internal/controllers/scheduler/controller.go

@@ -3,12 +3,12 @@ package scheduler
 import (
 	"context"
 	"fmt"
-	"math/rand/v2"
 	"slices"
 	"strings"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/utils/ptr"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
@@ -20,11 +20,13 @@ import (
 	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"
+	"github.com/openmcp-project/controller-utils/pkg/pairs"
 
 	clustersv1alpha1 "github.com/openmcp-project/openmcp-operator/api/clusters/v1alpha1"
 	cconst "github.com/openmcp-project/openmcp-operator/api/clusters/v1alpha1/constants"
 	apiconst "github.com/openmcp-project/openmcp-operator/api/constants"
 	"github.com/openmcp-project/openmcp-operator/internal/config"
+	"github.com/openmcp-project/openmcp-operator/internal/controllers/scheduler/strategy"
 )
 
 const ControllerName = "Scheduler"
@@ -177,62 +179,30 @@ func (r *ClusterScheduler) handleCreateOrUpdate(ctx context.Context, req reconci
 		return rr
 	}
 
-	// if no cluster was found, check if there is an existing cluster that qualifies for the request
-	// skip this check for preemptive requests with purposes with exclusive tenancy, as they will always result in a new cluster
-	if !(cDef.IsExclusive() && cr.Spec.Preemptive) { //nolint:staticcheck // QF1001
-		cluster, rerr = r.pickFittingCluster(ctx, cr, clusters, cDef)
-		if rerr != nil {
-			rr.ReconcileError = rerr
-			return rr
-		}
+	sr, err := Schedule(ctx, clusters, cDef, strategy.FromConfig(r.Config.Strategy), NewSchedulingRequest(string(cr.UID), cr.Spec.Preemptive, false, cr.Namespace, cr.Spec.Purpose))
+	if err != nil {
+		rr.ReconcileError = errutils.WithReason(fmt.Errorf("error scheduling request '%s': %w", req.String(), err), cconst.ReasonSchedulingFailed)
+		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()
-		changed := controllerutil.AddFinalizer(cluster, fin) // should always be true at this point
-		replacedPreemptiveUID := ""
-		if !cr.Spec.Preemptive && !cDef.IsSharedUnlimitedly() {
-			// if the request is not preemptive, it can potentially replace a preemptive request that was already assigned to this cluster
-			// this check can be skipped for clusters which are shared unlimitedly, as the preemptive request would point to the same cluster again
-			suffix, removed := RemoveFinalizerWithPrefix(cluster, clustersv1alpha1.PreemptiveRequestFinalizerOnClusterPrefix)
-			replacedPreemptiveUID = suffix
-			if removed {
-				log.Info("Replacing preemptive request", "uid", replacedPreemptiveUID)
-				if rerr := r.ReschedulePreemptiveRequest(ctx, replacedPreemptiveUID); rerr != nil {
-					if rerr.Reason() == cconst.ReasonInternalError {
-						// if the problem is that the preemptive request was not found, only log an error
-						log.Error(rerr, "Error rescheduling preemptive request")
-					} else {
-						rr.ReconcileError = rerr
-						return rr
-					}
-				}
-			}
-		}
-		if changed {
-			log.Debug("Adding finalizer to cluster", "clusterName", cluster.Name, "clusterNamespace", cluster.Namespace, "finalizer", fin, "replacedPreemptiveUID", replacedPreemptiveUID)
-			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
+	// apply required changes to the cluster
+	updated, err := sr.Apply(ctx, r.PlatformCluster.Client())
+	if err != nil {
+		rr.ReconcileError = errutils.WithReason(fmt.Errorf("error applying scheduling result for request '%s': %w", req.String(), err), cconst.ReasonPlatformClusterInteractionProblem)
+		return rr
+	}
+	// identify the cluster that was chosen for this request
+	for _, c := range updated {
+		if slices.Contains(c.Finalizers, cr.FinalizerForCluster()) {
+			cluster = c
+			log.Debug("Request has been scheduled", "clusterName", cluster.Name, "clusterNamespace", cluster.Namespace, "preemptive", cr.Spec.Preemptive)
+			break
 		}
 	}
+	if cluster == nil {
+		rr.ReconcileError = errutils.WithReason(fmt.Errorf("unable to determine cluster for request '%s', this should not happen", req.String()), cconst.ReasonSchedulingFailed)
+		return rr
+	}
 
 	// add cluster reference to request
 	rr.Object.Status.Cluster = &clustersv1alpha1.NamespacedObjectReference{}
@@ -369,72 +339,6 @@ func (r *ClusterScheduler) fetchRelevantClusters(ctx context.Context, cr *cluste
 	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.IsSharedUnlimitedly() {
-		// 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, "preemptive", cr.Spec.Preemptive)
-
-	// 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
-}
-
 // TODO: move to controller-utils
 // RemoveFinalizerWithPrefix removes the first finalizer with the given prefix from the object.
 // The bool return value indicates whether a finalizer was removed.
@@ -460,10 +364,32 @@ func RemoveFinalizerWithPrefix(obj client.Object, prefix string) (string, bool)
 	return suffix, length != index
 }
 
+// TODO: move to controller-utils
+// GetAny returns an arbitrary key-value pair from the map as a pointer to a pairs.Pair.
+// If the map is empty, it returns nil.
+func GetAny[K comparable, V any](m map[K]V) *pairs.Pair[K, V] {
+	for k, v := range m {
+		return ptr.To(pairs.New(k, v))
+	}
+	return nil
+}
+
+// TODO: move to controller-utils
+// MapValues returns a slice of all values in the map.
+// The order is unspecified.
+// The values are not deep-copied, so changes to the values could affect the original map.
+func MapValues[K comparable, V any](m map[K]V) []V {
+	values := make([]V, 0, len(m))
+	for _, v := range m {
+		values = append(values, v)
+	}
+	return values
+}
+
 // ReschedulePreemptiveRequest fetches the ClusterRequest with the corresponding UID from the cluster and adds a reconcile annotation to it.
-func (r *ClusterScheduler) ReschedulePreemptiveRequest(ctx context.Context, uid string) errutils.ReasonableError {
+func ReschedulePreemptiveRequest(ctx context.Context, platformClient client.Client, uid string) errutils.ReasonableError {
 	crl := &clustersv1alpha1.ClusterRequestList{}
-	if err := r.PlatformCluster.Client().List(ctx, crl, client.MatchingLabelsSelector{Selector: r.Config.Selectors.Requests.Completed()}, client.MatchingFields{"spec.preemptive": "true"}); err != nil {
+	if err := platformClient.List(ctx, crl, client.MatchingFields{"spec.preemptive": "true"}); err != nil {
 		return errutils.WithReason(fmt.Errorf("error listing ClusterRequests: %w", err), cconst.ReasonPlatformClusterInteractionProblem)
 	}
 	var cr *clustersv1alpha1.ClusterRequest
@@ -476,86 +402,8 @@ func (r *ClusterScheduler) ReschedulePreemptiveRequest(ctx context.Context, uid
 	if cr == nil {
 		return errutils.WithReason(fmt.Errorf("unable to find preemptive ClusterRequest with UID '%s'", uid), cconst.ReasonInternalError)
 	}
-	if err := ctrlutils.EnsureAnnotation(ctx, r.PlatformCluster.Client(), cr, apiconst.OperationAnnotation, apiconst.OperationAnnotationValueReconcile, true); err != nil {
+	if err := ctrlutils.EnsureAnnotation(ctx, platformClient, cr, apiconst.OperationAnnotation, apiconst.OperationAnnotationValueReconcile, true); err != nil {
 		return errutils.WithReason(fmt.Errorf("error adding reconcile annotation to preemptive ClusterRequest '%s': %w", cr.Name, err), cconst.ReasonPlatformClusterInteractionProblem)
 	}
 	return nil
 }
-
-// pickFittingCluster gets a list of existing clusters that match the request's purpose and tries to pick one for the request.
-// If the returned cluster is nil, no fitting cluster was found. This can e.g. happen if all clusters are already at their tenancy limit.
-func (r *ClusterScheduler) pickFittingCluster(ctx context.Context, cr *clustersv1alpha1.ClusterRequest, clusters []*clustersv1alpha1.Cluster, cDef *config.ClusterDefinition) (*clustersv1alpha1.Cluster, errutils.ReasonableError) {
-	log := logging.FromContextOrPanic(ctx)
-	log.Debug("Checking for fitting clusters", "purpose", cr.Spec.Purpose, "tenancyCount", cDef.TenancyCount, "preemptive", cr.Spec.Preemptive)
-	// remove all clusters with a non-zero deletion timestamp from the list of candidates
-	clusters = filters.FilterSlice(clusters, func(args ...any) bool {
-		c, ok := args[0].(*clustersv1alpha1.Cluster)
-		if !ok {
-			return false
-		}
-		return c.DeletionTimestamp.IsZero()
-	})
-	// avoid picking clusters which are not ready if ready ones are available
-	readyClusters := filters.FilterSlice(clusters, func(args ...any) bool {
-		c, ok := args[0].(*clustersv1alpha1.Cluster)
-		if !ok {
-			return false
-		}
-		return c.Status.Phase == clustersv1alpha1.CLUSTER_PHASE_READY
-	})
-	if len(readyClusters) > 0 {
-		log.Debug("Excluding non-ready clusters from scheduling", "readyCount", len(readyClusters), "totalCount", len(clusters))
-		clusters = readyClusters
-	}
-	// unless the cluster template for the requested purpose allows unlimited sharing, filter out all clusters that are already at their tenancy limit
-	// for preemptive requests, other preemptive requests count towards the tenancy limit, for regular requests, preemptive requests are ignored
-	if cDef.IsExclusive() || cDef.TenancyCount > 0 {
-		clusters = filters.FilterSlice(clusters, func(args ...any) bool {
-			c, ok := args[0].(*clustersv1alpha1.Cluster)
-			if !ok {
-				return false
-			}
-			preTenCount := c.GetPreemptiveTenancyCount()
-			if cDef.IsExclusive() && preTenCount == 0 {
-				// for exclusive tenancy, take only clusters into account that have been preemptively requested
-				return false
-			}
-			tenCount := c.GetTenancyCount()
-			if cr.Spec.Preemptive {
-				// for preemptive requests, also take preemptive "workloads" into account
-				tenCount += preTenCount
-			}
-			return tenCount < cDef.TenancyCount
-		})
-	}
-	var cluster *clustersv1alpha1.Cluster
-	if len(clusters) == 1 {
-		cluster = clusters[0]
-		log.Debug("One existing cluster qualifies for request", "clusterName", cluster.Name, "clusterNamespace", cluster.Namespace, "preemptive", cr.Spec.Preemptive)
-	} 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), "preemptive", cr.Spec.Preemptive)
-		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:
-			return nil, errutils.WithReason(fmt.Errorf("unknown strategy '%s'", r.Config.Strategy), cconst.ReasonConfigurationProblem)
-		}
-	}
-	return cluster, nil
-}