feat: adding pipeline options

internal/scheduling/cinder/filter_weigher_pipeline_controller.go

@@ -121,7 +121,7 @@ func (c *FilterWeigherPipelineController) process(ctx context.Context, decision
 		return err
 	}
 
-	result, err := pipeline.Run(request)
+	result, err := pipeline.Run(request, lib.Options{})
 	if err != nil {
 		log.Error(err, "failed to run pipeline")
 		return err

internal/scheduling/lib/filter_monitor.go

@@ -43,6 +43,6 @@ func (fm *FilterMonitor[RequestType]) Validate(ctx context.Context, params v1alp
 }
 
 // Run the filter and observe its execution.
-func (fm *FilterMonitor[RequestType]) Run(traceLog *slog.Logger, request RequestType) (*FilterWeigherPipelineStepResult, error) {
-	return fm.monitor.RunWrapped(traceLog, request, fm.filter)
+func (fm *FilterMonitor[RequestType]) Run(traceLog *slog.Logger, request RequestType, opts Options) (*FilterWeigherPipelineStepResult, error) {
+	return fm.monitor.RunWrapped(traceLog, request, opts, fm.filter)
 }

internal/scheduling/lib/filter_monitor_test.go

@@ -100,7 +100,7 @@ func TestFilterMonitor_Run(t *testing.T) {
 		Weights: map[string]float64{"host1": 0.1, "host2": 0.2, "host3": 0.3},
 	}
 
-	result, err := fm.Run(slog.Default(), request)
+	result, err := fm.Run(slog.Default(), request, Options{})
 	if err != nil {
 		t.Errorf("expected no error, got %v", err)
 	}

internal/scheduling/lib/filter_test.go

@@ -31,7 +31,7 @@ func (m *mockFilter[RequestType]) Validate(ctx context.Context, params v1alpha1.
 	}
 	return m.ValidateFunc(ctx, params)
 }
-func (m *mockFilter[RequestType]) Run(traceLog *slog.Logger, request RequestType) (*FilterWeigherPipelineStepResult, error) {
+func (m *mockFilter[RequestType]) Run(traceLog *slog.Logger, request RequestType, opts Options) (*FilterWeigherPipelineStepResult, error) {
 	if m.RunFunc == nil {
 		return &FilterWeigherPipelineStepResult{}, nil
 	}

internal/scheduling/lib/filter_validation.go

@@ -35,8 +35,8 @@ func validateFilter[RequestType FilterWeigherPipelineRequest](filter Filter[Requ
 }
 
 // Run the filter and validate what happens.
-func (s *FilterValidator[RequestType]) Run(traceLog *slog.Logger, request RequestType) (*FilterWeigherPipelineStepResult, error) {
-	result, err := s.Filter.Run(traceLog, request)
+func (s *FilterValidator[RequestType]) Run(traceLog *slog.Logger, request RequestType, opts Options) (*FilterWeigherPipelineStepResult, error) {
+	result, err := s.Filter.Run(traceLog, request, opts)
 	if err != nil {
 		return nil, err
 	}

internal/scheduling/lib/filter_validation_test.go

@@ -156,7 +156,7 @@ func TestFilterValidator_Run(t *testing.T) {
 			}
 			traceLog := slog.Default()
 
-			result, err := validator.Run(traceLog, request)
+			result, err := validator.Run(traceLog, request, Options{})
 
 			if tt.expectError && err == nil {
 				t.Error("expected error but got nil")

internal/scheduling/lib/filter_weigher_pipeline.go

@@ -18,8 +18,8 @@ import (
 )
 
 type FilterWeigherPipeline[RequestType FilterWeigherPipelineRequest] interface {
-	// Run the scheduling pipeline with the given request.
-	Run(request RequestType) (v1alpha1.DecisionResult, error)
+	// Run the scheduling pipeline with the given request and call-time options.
+	Run(request RequestType, opts Options) (v1alpha1.DecisionResult, error)
 }
 
 // Pipeline of scheduler steps.
@@ -138,14 +138,15 @@ func InitNewFilterWeigherPipeline[RequestType FilterWeigherPipelineRequest](
 func (p *filterWeigherPipeline[RequestType]) runFilters(
 	log *slog.Logger,
 	request RequestType,
+	opts Options,
 ) (filteredRequest RequestType, stepResults []v1alpha1.StepResult) {
 
 	filteredRequest = request
 	for _, filterName := range p.filtersOrder {
 		filter := p.filters[filterName]
 		stepLog := log.With("filter", filterName)
 		stepLog.Info("scheduler: running filter")
-		result, err := filter.Run(stepLog, filteredRequest)
+		result, err := filter.Run(stepLog, filteredRequest, opts)
 		if errors.Is(err, ErrStepSkipped) {
 			stepLog.Info("scheduler: filter skipped")
 			continue
@@ -170,6 +171,7 @@ func (p *filterWeigherPipeline[RequestType]) runFilters(
 func (p *filterWeigherPipeline[RequestType]) runWeighers(
 	log *slog.Logger,
 	filteredRequest RequestType,
+	opts Options,
 ) map[string]map[string]float64 {
 
 	activationsByStep := map[string]map[string]float64{}
@@ -181,7 +183,7 @@ func (p *filterWeigherPipeline[RequestType]) runWeighers(
 		wg.Go(func() {
 			stepLog := log.With("weigher", weigherName)
 			stepLog.Info("scheduler: running weigher")
-			result, err := weigher.Run(stepLog, filteredRequest)
+			result, err := weigher.Run(stepLog, filteredRequest, opts)
 			if errors.Is(err, ErrStepSkipped) {
 				stepLog.Info("scheduler: weigher skipped")
 				return
@@ -262,7 +264,10 @@ func (s *filterWeigherPipeline[RequestType]) sortHostsByWeights(weights map[stri
 }
 
 // Evaluate the pipeline and return a list of hosts in order of preference.
-func (p *filterWeigherPipeline[RequestType]) Run(request RequestType) (v1alpha1.DecisionResult, error) {
+func (p *filterWeigherPipeline[RequestType]) Run(request RequestType, opts Options) (v1alpha1.DecisionResult, error) {
+	if err := opts.Validate(); err != nil {
+		return v1alpha1.DecisionResult{}, err
+	}
 	slogArgs := request.GetTraceLogArgs()
 	slogArgsAny := make([]any, 0, len(slogArgs))
 	for _, arg := range slogArgs {
@@ -279,7 +284,7 @@ func (p *filterWeigherPipeline[RequestType]) Run(request RequestType) (v1alpha1.
 
 	// Run filters first to reduce the number of hosts.
 	// Any weights assigned to filtered out hosts are ignored.
-	filteredRequest, filterStepResults := p.runFilters(traceLog, request)
+	filteredRequest, filterStepResults := p.runFilters(traceLog, request, opts)
 	traceLog.Info(
 		"scheduler: finished filters",
 		"remainingHosts", filteredRequest.GetHosts(),
@@ -290,13 +295,23 @@ func (p *filterWeigherPipeline[RequestType]) Run(request RequestType) (v1alpha1.
 	for _, host := range filteredRequest.GetHosts() {
 		remainingWeights[host] = inWeights[host]
 	}
-	stepWeights := p.runWeighers(traceLog, filteredRequest)
+	stepWeights := p.runWeighers(traceLog, filteredRequest, opts)
 	outWeights := p.applyWeights(traceLog, stepWeights, remainingWeights)
 	traceLog.Info("scheduler: output weights", "weights", outWeights)
 
 	hosts := p.sortHostsByWeights(outWeights)
 	traceLog.Info("scheduler: sorted hosts", "hosts", hosts)
 
+	if opts.MaxCandidates > 0 && len(hosts) > opts.MaxCandidates {
+		hosts = hosts[:opts.MaxCandidates]
+		// Drop trimmed hosts from outWeights so AggregatedOutWeights stays consistent.
+		for host := range outWeights {
+			if !slices.Contains(hosts, host) {
+				delete(outWeights, host)
+			}
+		}
+	}
+
 	// Collect some metrics about the pipeline execution.
 	go p.monitor.observePipelineResult(request, hosts)
 

internal/scheduling/lib/filter_weigher_pipeline_step.go

@@ -30,7 +30,9 @@ type FilterWeigherPipelineStep[RequestType FilterWeigherPipelineRequest] interfa
 	//
 	// A traceLog is provided that contains the global request id and should
 	// be used to log the step's execution.
-	Run(traceLog *slog.Logger, request RequestType) (*FilterWeigherPipelineStepResult, error)
+	//
+	// opts carries per-call behavioral options set by the pipeline caller.
+	Run(traceLog *slog.Logger, request RequestType, opts Options) (*FilterWeigherPipelineStepResult, error)
 }
 
 // Common base for all steps that provides some functionality

internal/scheduling/lib/filter_weigher_pipeline_step_monitor.go

@@ -65,6 +65,7 @@ func monitorStep[RequestType FilterWeigherPipelineRequest](stepName string, m Fi
 func (s *FilterWeigherPipelineStepMonitor[RequestType]) RunWrapped(
 	traceLog *slog.Logger,
 	request RequestType,
+	opts Options,
 	step FilterWeigherPipelineStep[RequestType],
 ) (*FilterWeigherPipelineStepResult, error) {
 
@@ -74,7 +75,7 @@ func (s *FilterWeigherPipelineStepMonitor[RequestType]) RunWrapped(
 	}
 
 	inWeights := request.GetWeights()
-	stepResult, err := step.Run(traceLog, request)
+	stepResult, err := step.Run(traceLog, request, opts)
 	if err != nil {
 		return nil, err
 	}

internal/scheduling/lib/filter_weigher_pipeline_step_monitor_test.go

@@ -38,7 +38,7 @@ func TestStepMonitorRun(t *testing.T) {
 		Hosts:   []string{"host1", "host2", "host3"},
 		Weights: map[string]float64{"host1": 0.2, "host2": 0.1, "host3": 0.0},
 	}
-	if _, err := monitor.RunWrapped(slog.Default(), request, step); err != nil {
+	if _, err := monitor.RunWrapped(slog.Default(), request, Options{}, step); err != nil {
 		t.Fatalf("Run() error = %v, want nil", err)
 	}
 	if len(removedHostsObserver.Observations) != 1 {

internal/scheduling/lib/filter_weigher_pipeline_test.go

@@ -72,7 +72,7 @@ func TestPipeline_Run(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			result, err := pipeline.Run(tt.request)
+			result, err := pipeline.Run(tt.request, Options{})
 			if err != nil {
 				t.Fatalf("expected no error, got %v", err)
 			}
@@ -221,7 +221,7 @@ func TestPipeline_RunFilters(t *testing.T) {
 		Weights: map[string]float64{"host1": 0.0, "host2": 0.0, "host3": 0.0},
 	}
 
-	req, _ := p.runFilters(slog.Default(), request)
+	req, _ := p.runFilters(slog.Default(), request, Options{})
 	if len(req.Hosts) != 2 {
 		t.Fatalf("expected 2 step results, got %d", len(req.Hosts))
 	}
@@ -372,3 +372,59 @@ func TestFilterWeigherPipelineMonitor_SubPipeline(t *testing.T) {
 		t.Error("original monitor should not be modified")
 	}
 }
+
+func TestPipeline_MaxCandidates(t *testing.T) {
+	// Pipeline that passes all 4 hosts with descending weights.
+	pipeline := &filterWeigherPipeline[mockFilterWeigherPipelineRequest]{
+		filters:       map[string]Filter[mockFilterWeigherPipelineRequest]{},
+		filtersOrder:  []string{},
+		weighersOrder: []string{},
+		weighers:      map[string]Weigher[mockFilterWeigherPipelineRequest]{},
+	}
+	request := mockFilterWeigherPipelineRequest{
+		Hosts:   []string{"host1", "host2", "host3", "host4"},
+		Weights: map[string]float64{"host1": 4.0, "host2": 3.0, "host3": 2.0, "host4": 1.0},
+	}
+
+	tests := []struct {
+		name          string
+		maxCandidates int
+		wantLen       int
+		wantFirst     string
+	}{
+		{"no limit", 0, 4, "host1"},
+		{"limit to 2", 2, 2, "host1"},
+		{"limit to 1", 1, 1, "host1"},
+		{"limit larger than hosts", 10, 4, "host1"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result, err := pipeline.Run(request, Options{MaxCandidates: tt.maxCandidates})
+			if err != nil {
+				t.Fatalf("expected no error, got %v", err)
+			}
+			if len(result.OrderedHosts) != tt.wantLen {
+				t.Errorf("expected %d hosts, got %d: %v", tt.wantLen, len(result.OrderedHosts), result.OrderedHosts)
+			}
+			if len(result.OrderedHosts) > 0 && result.OrderedHosts[0] != tt.wantFirst {
+				t.Errorf("expected first host %s, got %s", tt.wantFirst, result.OrderedHosts[0])
+			}
+			if tt.maxCandidates > 0 && len(result.OrderedHosts) <= tt.maxCandidates {
+				// AggregatedOutWeights must only contain returned hosts.
+				for host := range result.AggregatedOutWeights {
+					found := false
+					for _, h := range result.OrderedHosts {
+						if h == host {
+							found = true
+							break
+						}
+					}
+					if !found {
+						t.Errorf("AggregatedOutWeights contains trimmed host %s", host)
+					}
+				}
+			}
+		})
+	}
+}

internal/scheduling/lib/options.go

@@ -0,0 +1,47 @@
+// Copyright SAP SE
+// SPDX-License-Identifier: Apache-2.0
+
+package lib
+
+import (
+	"errors"
+
+	"github.com/cobaltcore-dev/cortex/api/v1alpha1"
+)
+
+// Options configure the behavior of a single pipeline run at call time.
+// These are distinct from per-step YAML options (FilterWeigherPipelineStepOpts),
+// which are static and set when the pipeline is initialized.
+//
+// Consumed by steps: ReadOnly, LockReservations, AssumeEmptyHosts, IgnoredReservationTypes.
+// Consumed by the controller after pipeline.Run(): RecordHistory, CreateInflight.
+type Options struct {
+	// ReadOnly means the pipeline could run without using the mutex, i.e. concurrent runs are ok.
+	ReadOnly bool
+	// LockReservations prevents reservation unlocking, e.g. in the capacity filter.
+	// Set when finding hosts for new reservations (failover, CR) to see true available capacity.
+	LockReservations bool
+	// AssumeEmptyHosts treats all hosts as having no running VMs.
+	AssumeEmptyHosts bool
+	// IgnoredReservationTypes lists reservation types the capacity filter skips entirely.
+	IgnoredReservationTypes []v1alpha1.ReservationType
+	// MaxCandidates limits the number of hosts returned after weighing. 0 means no limit.
+	MaxCandidates int
+
+	// RecordHistory records the placement decision in placement history.
+	// Replaces pipeline.Spec.CreateHistory once pipelines consolidate.
+	RecordHistory bool
+	// CreateInflight creates pessimistic blocking reservations for all returned candidates.
+	CreateInflight bool
+}
+
+// Validate checks for mutually exclusive or inconsistent option combinations.
+func (o Options) Validate() error {
+	if o.ReadOnly && o.RecordHistory {
+		return errors.New("ReadOnly and RecordHistory are mutually exclusive: read-only runs must not mutate state")
+	}
+	if o.ReadOnly && o.CreateInflight {
+		return errors.New("ReadOnly and CreateInflight are mutually exclusive: read-only runs must not mutate state")
+	}
+	return nil
+}

internal/scheduling/lib/weigher_monitor.go

@@ -43,6 +43,6 @@ func (wm *WeigherMonitor[RequestType]) Validate(ctx context.Context, params v1al
 }
 
 // Run the weigher and observe its execution.
-func (wm *WeigherMonitor[RequestType]) Run(traceLog *slog.Logger, request RequestType) (*FilterWeigherPipelineStepResult, error) {
-	return wm.monitor.RunWrapped(traceLog, request, wm.weigher)
+func (wm *WeigherMonitor[RequestType]) Run(traceLog *slog.Logger, request RequestType, opts Options) (*FilterWeigherPipelineStepResult, error) {
+	return wm.monitor.RunWrapped(traceLog, request, opts, wm.weigher)
 }

internal/scheduling/lib/weigher_monitor_test.go

@@ -100,7 +100,7 @@ func TestWeigherMonitor_Run(t *testing.T) {
 		Weights: map[string]float64{"host1": 0.1, "host2": 0.2, "host3": 0.3},
 	}
 
-	result, err := wm.Run(slog.Default(), request)
+	result, err := wm.Run(slog.Default(), request, Options{})
 	if err != nil {
 		t.Errorf("expected no error, got %v", err)
 	}

internal/scheduling/lib/weigher_test.go

@@ -34,7 +34,7 @@ func (m *mockWeigher[RequestType]) Validate(ctx context.Context, params v1alpha1
 	}
 	return m.ValidateFunc(ctx, params)
 }
-func (m *mockWeigher[RequestType]) Run(traceLog *slog.Logger, request RequestType) (*FilterWeigherPipelineStepResult, error) {
+func (m *mockWeigher[RequestType]) Run(traceLog *slog.Logger, request RequestType, opts Options) (*FilterWeigherPipelineStepResult, error) {
 	if m.RunFunc == nil {
 		return &FilterWeigherPipelineStepResult{}, nil
 	}

internal/scheduling/lib/weigher_validation.go

@@ -35,8 +35,8 @@ func validateWeigher[RequestType FilterWeigherPipelineRequest](weigher Weigher[R
 }
 
 // Run the weigher and validate what happens.
-func (s *WeigherValidator[RequestType]) Run(traceLog *slog.Logger, request RequestType) (*FilterWeigherPipelineStepResult, error) {
-	result, err := s.Weigher.Run(traceLog, request)
+func (s *WeigherValidator[RequestType]) Run(traceLog *slog.Logger, request RequestType, opts Options) (*FilterWeigherPipelineStepResult, error) {
+	result, err := s.Weigher.Run(traceLog, request, opts)
 	if err != nil {
 		return nil, err
 	}

internal/scheduling/lib/weigher_validation_test.go

@@ -96,7 +96,7 @@ func TestWeigherValidator_Run_ValidHosts(t *testing.T) {
 		Weigher: mockStep,
 	}
 
-	result, err := validator.Run(slog.Default(), request)
+	result, err := validator.Run(slog.Default(), request, Options{})
 	if err != nil {
 		t.Errorf("Run() error = %v, want nil", err)
 	}
@@ -130,7 +130,7 @@ func TestWeigherValidator_Run_HostNumberMismatch(t *testing.T) {
 		Weigher: mockStep,
 	}
 
-	result, err := validator.Run(slog.Default(), request)
+	result, err := validator.Run(slog.Default(), request, Options{})
 	if err == nil {
 		t.Errorf("Run() error = nil, want error")
 	}

internal/scheduling/machines/filter_weigher_pipeline_controller.go

@@ -144,7 +144,7 @@ func (c *FilterWeigherPipelineController) process(ctx context.Context, decision
 
 	// Execute the scheduling pipeline.
 	request := ironcore.MachinePipelineRequest{Pools: pools.Items}
-	result, err := pipeline.Run(request)
+	result, err := pipeline.Run(request, lib.Options{})
 	if err != nil {
 		log.V(1).Error(err, "failed to run scheduler pipeline")
 		return errors.New("failed to run scheduler pipeline")

internal/scheduling/machines/filter_weigher_pipeline_controller_test.go

@@ -516,7 +516,7 @@ func createMockPipeline() lib.FilterWeigherPipeline[ironcore.MachinePipelineRequ
 
 type mockMachinePipeline struct{}
 
-func (m *mockMachinePipeline) Run(request ironcore.MachinePipelineRequest) (v1alpha1.DecisionResult, error) {
+func (m *mockMachinePipeline) Run(request ironcore.MachinePipelineRequest, opts lib.Options) (v1alpha1.DecisionResult, error) {
 	if len(request.Pools) == 0 {
 		return v1alpha1.DecisionResult{}, nil
 	}

internal/scheduling/machines/plugins/filters/filter_noop.go

@@ -31,7 +31,7 @@ func (f *NoopFilter) Validate(ctx context.Context, params v1alpha1.Parameters) e
 // not in the map are considered as filtered out.
 // Provide a traceLog that contains the global request id and should
 // be used to log the step's execution.
-func (NoopFilter) Run(traceLog *slog.Logger, request ironcore.MachinePipelineRequest) (*lib.FilterWeigherPipelineStepResult, error) {
+func (NoopFilter) Run(traceLog *slog.Logger, request ironcore.MachinePipelineRequest, opts lib.Options) (*lib.FilterWeigherPipelineStepResult, error) {
 	activations := make(map[string]float64, len(request.Pools))
 	stats := make(map[string]lib.FilterWeigherPipelineStepStatistics)
 	// Usually you would do some filtering here, or adjust the weights.