ensure that requests are never scheduled in parallel

Author: Diaphteiros

internal/controllers/scheduler/controller.go

@@ -38,17 +38,14 @@ func NewClusterScheduler(setupLog *logging.Logger, platformCluster *clusters.Clu
 		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{
-		Preemptive: PreemptiveScheduler{
-			PlatformCluster: platformCluster,
-			Config:          config,
-		},
+		Preemptive:      NewPreemptiveScheduler(platformCluster, config),
 		PlatformCluster: platformCluster,
 		Config:          config,
 	}, nil
 }
 
 type ClusterScheduler struct {
-	Preemptive      PreemptiveScheduler
+	Preemptive      *PreemptiveScheduler
 	PlatformCluster *clusters.Cluster
 	Config          *config.SchedulerConfig
 }
@@ -60,6 +57,8 @@ type ReconcileResult = ctrlutils.ReconcileResult[*clustersv1alpha1.ClusterReques
 func (r *ClusterScheduler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
 	log := logging.FromContextOrPanic(ctx).WithName(ControllerName)
 	ctx = logging.NewContext(ctx, log)
+	r.Preemptive.Lock.Lock()
+	defer r.Preemptive.Lock.Unlock()
 	log.Info("Starting reconcile")
 	rr := r.reconcile(ctx, req)
 	// status update

internal/controllers/scheduler/controller_preemptive.go

@@ -3,6 +3,7 @@ package scheduler
 import (
 	"context"
 	"fmt"
+	"sync"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -24,17 +25,33 @@ import (
 const PreemptiveControllerName = "PreemptiveScheduler"
 
 type PreemptiveScheduler struct {
+	Lock            *sync.Mutex
 	PlatformCluster *clusters.Cluster
 	Config          *config.SchedulerConfig
 }
 
+// NewPreemptiveScheduler creates a new PreemptiveScheduler.
+// Note that this is already called by the NewScheduler constructor function, there should be no need to call this outside of tests.
+func NewPreemptiveScheduler(platformCluster *clusters.Cluster, cfg *config.SchedulerConfig) *PreemptiveScheduler {
+	if cfg == nil {
+		cfg = &config.SchedulerConfig{}
+	}
+	return &PreemptiveScheduler{
+		Lock:            &sync.Mutex{},
+		PlatformCluster: platformCluster,
+		Config:          cfg,
+	}
+}
+
 var _ reconcile.Reconciler = &PreemptiveScheduler{}
 
 type PreemptiveReconcileResult = ctrlutils.ReconcileResult[*clustersv1alpha1.PreemptiveClusterRequest, clustersv1alpha1.ConditionStatus]
 
 func (r *PreemptiveScheduler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
 	log := logging.FromContextOrPanic(ctx).WithName(PreemptiveControllerName)
 	ctx = logging.NewContext(ctx, log)
+	r.Lock.Lock()
+	defer r.Lock.Unlock()
 	log.Info("Starting reconcile")
 	rr := r.reconcile(ctx, req)
 	// status update

internal/controllers/scheduler/controller_test.go

@@ -60,10 +60,7 @@ func defaultTestSetup(testDirPathSegments ...string) (*config.SchedulerConfig, *
 			return r
 		}, platform).
 		WithReconcilerConstructor(pSchedulerRec, func(c ...client.Client) reconcile.Reconciler {
-			return &scheduler.PreemptiveScheduler{
-				PlatformCluster: clusters.NewTestClusterFromClient(platform, c[0]),
-				Config:          cfg.Scheduler,
-			}
+			return scheduler.NewPreemptiveScheduler(clusters.NewTestClusterFromClient(platform, c[0]), cfg.Scheduler)
 		}, platform).
 		WithUIDs(platform).
 		Build()