mirror of
https://github.com/kubernetes/kubernetes.git
synced 2026-07-16 13:14:10 -04:00
Merge pull request #138274 from wtravO/wtravo/placement-cycle-state
Add PlacementCycleState to WAS scheduler framework
This commit is contained in:
Kubernetes Prow Robot
GitHub
commit
ae84ac1a16
12 changed files with 384 additions and 43 deletions
|
|
@ -46,6 +46,10 @@ type CycleState struct {
|
|||
// or doesn't belong to any pod group.
|
||||
// This field can only be non-nil when GenericWorkload feature flag is enabled.
|
||||
podGroupCycleState fwk.PodGroupCycleState
|
||||
// placementCycleState contains the CycleState for the current Placement being evaluated.
|
||||
// If set to nil, it means this pod is not being scheduled within a placement context.
|
||||
// This field can only be non-nil when GenericWorkload feature flag is enabled.
|
||||
placementCycleState fwk.PlacementCycleState
|
||||
}
|
||||
|
||||
// NewCycleState initializes a new CycleState and returns its pointer.
|
||||
|
|
@ -113,6 +117,14 @@ func (c *CycleState) GetPodGroupSchedulingCycle() fwk.PodGroupCycleState {
|
|||
return c.podGroupCycleState
|
||||
}
|
||||
|
||||
func (c *CycleState) GetPlacementCycleState() fwk.PlacementCycleState {
|
||||
return c.placementCycleState
|
||||
}
|
||||
|
||||
func (c *CycleState) SetPlacementCycleState(placementCycleState fwk.PlacementCycleState) {
|
||||
c.placementCycleState = placementCycleState
|
||||
}
|
||||
|
||||
func (c *CycleState) SetSkipAllPostFilterPlugins(flag bool) {
|
||||
c.skipAllPostFilterPlugins = flag
|
||||
}
|
||||
|
|
@ -140,6 +152,7 @@ func (c *CycleState) Clone() fwk.CycleState {
|
|||
copy.skipPreBindPlugins = c.skipPreBindPlugins
|
||||
copy.parallelPreBindPlugins = c.parallelPreBindPlugins
|
||||
copy.podGroupCycleState = c.podGroupCycleState
|
||||
copy.placementCycleState = c.placementCycleState
|
||||
copy.skipAllPostFilterPlugins = c.skipAllPostFilterPlugins
|
||||
|
||||
return copy
|
||||
|
|
|
|||
|
|
@ -193,3 +193,94 @@ func TestCycleStateClone(t *testing.T) {
|
|||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestPlacementCycleState(t *testing.T) {
|
||||
t.Run("nil by default", func(t *testing.T) {
|
||||
state := NewCycleState()
|
||||
if state.GetPlacementCycleState() != nil {
|
||||
t.Errorf("expected nil PlacementCycleState on fresh CycleState")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("set and get", func(t *testing.T) {
|
||||
state := NewCycleState()
|
||||
podGroupState := NewCycleState()
|
||||
placementState := NewCycleState()
|
||||
placementState.SetPodGroupSchedulingCycle(podGroupState)
|
||||
placementState.Write("testkey", &fakeData{data: "placementdata"})
|
||||
|
||||
state.SetPlacementCycleState(placementState)
|
||||
|
||||
got := state.GetPlacementCycleState()
|
||||
if got == nil {
|
||||
t.Fatal("expected non-nil PlacementCycleState after Set")
|
||||
}
|
||||
|
||||
data, err := got.Read("testkey")
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error reading from PlacementCycleState: %v", err)
|
||||
}
|
||||
if data.(*fakeData).data != "placementdata" {
|
||||
t.Errorf("expected 'placementdata', got %q", data.(*fakeData).data)
|
||||
}
|
||||
if got.GetPodGroupSchedulingCycle() != podGroupState {
|
||||
t.Errorf("expected PlacementCycleState to expose its PodGroupCycleState")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("set to nil clears", func(t *testing.T) {
|
||||
state := NewCycleState()
|
||||
state.SetPlacementCycleState(NewCycleState())
|
||||
state.SetPlacementCycleState(nil)
|
||||
|
||||
if state.GetPlacementCycleState() != nil {
|
||||
t.Errorf("expected nil PlacementCycleState after setting to nil")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("clone preserves reference", func(t *testing.T) {
|
||||
state := NewCycleState()
|
||||
state.Write(key, &fakeData{data: "pod-data"})
|
||||
|
||||
placementState := NewCycleState()
|
||||
placementState.Write("pkey", &fakeData{data: "placement-data"})
|
||||
state.SetPlacementCycleState(placementState)
|
||||
|
||||
cloned := state.Clone().(*CycleState)
|
||||
|
||||
// The cloned state should reference the same PlacementCycleState.
|
||||
if cloned.GetPlacementCycleState() == nil {
|
||||
t.Fatal("cloned state should have non-nil PlacementCycleState")
|
||||
}
|
||||
|
||||
data, err := cloned.GetPlacementCycleState().Read("pkey")
|
||||
if err != nil {
|
||||
t.Fatalf("unexpected error reading from cloned PlacementCycleState: %v", err)
|
||||
}
|
||||
if data.(*fakeData).data != "placement-data" {
|
||||
t.Errorf("expected 'placement-data', got %q", data.(*fakeData).data)
|
||||
}
|
||||
|
||||
// Writes to the PlacementCycleState via the clone should be visible from the original,
|
||||
// since it's a shared reference (same as podGroupCycleState behavior).
|
||||
cloned.GetPlacementCycleState().Write("newkey", &fakeData{data: "new"})
|
||||
newData, err := state.GetPlacementCycleState().Read("newkey")
|
||||
if err != nil {
|
||||
t.Fatalf("write via clone's PlacementCycleState should be visible from original: %v", err)
|
||||
}
|
||||
if newData.(*fakeData).data != "new" {
|
||||
t.Errorf("expected 'new', got %q", newData.(*fakeData).data)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("clone with nil placement state", func(t *testing.T) {
|
||||
state := NewCycleState()
|
||||
state.Write(key, &fakeData{data: "data"})
|
||||
// Do not set PlacementCycleState — leave nil.
|
||||
|
||||
cloned := state.Clone().(*CycleState)
|
||||
if cloned.GetPlacementCycleState() != nil {
|
||||
t.Errorf("cloned state should have nil PlacementCycleState when original has nil")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
|
|
|||
|
|
@ -192,7 +192,7 @@ type PlacementFeasiblePlugin interface {
|
|||
// The scheduler will give up this placement and won't even evaluate remaining pods. The placement will remain eligible for preemption.
|
||||
// Return Success status if the pod group can be scheduled in the current partially evaluated placement.
|
||||
// After returning Success, the plugin should keep returning Success for the remaining pods.
|
||||
PlacementFeasible(ctx context.Context, placementCycleState fwk.PodGroupCycleState, podGroupInfo fwk.PodGroupInfo) *fwk.Status
|
||||
PlacementFeasible(ctx context.Context, placementCycleState fwk.PlacementCycleState, podGroupInfo fwk.PodGroupInfo) *fwk.Status
|
||||
}
|
||||
|
||||
// Framework manages the set of plugins in use by the scheduling framework.
|
||||
|
|
@ -283,7 +283,7 @@ type Framework interface {
|
|||
// If any plugin returns invalid status, the result will be Error and the remaining plugins won't be invoked.
|
||||
// Otherwise, if at least 1 plugin returns UnschedulableAndUnresolvable, the remaining plugins won't be invoked and the result will be UnschdulableAndUnresolvable. The placement will remain eligible for preemption.
|
||||
// Otherwise, if at least 1 plugin returns Unschedulable, the remaining plugins will be invoked and the result will be Unschedulable.
|
||||
RunPlacementFeasiblePlugins(ctx context.Context, placementCycleState fwk.PodGroupCycleState, podGroupInfo fwk.PodGroupInfo) *fwk.Status
|
||||
RunPlacementFeasiblePlugins(ctx context.Context, placementCycleState fwk.PlacementCycleState, podGroupInfo fwk.PodGroupInfo) *fwk.Status
|
||||
|
||||
// AddWaitingPod creates a waiting pod instance and adds it to the framework.
|
||||
// It takes the pluginsWaitTime map returned by the RunPermitPlugins.
|
||||
|
|
@ -308,7 +308,8 @@ type Framework interface {
|
|||
// It returns a list that stores scores from each plugin and total score for each Placement.
|
||||
// It also returns *Status, which is set to non-success if any of the plugins returns
|
||||
// a non-success status.
|
||||
RunPlacementScorePlugins(ctx context.Context, state fwk.PodGroupCycleState, podGroupInfo fwk.PodGroupInfo, placements []*fwk.PodGroupAssignments) (ns []fwk.PlacementPluginScores, status *fwk.Status)
|
||||
// Each PlacementCycleState is passed to ScorePlacement for the PodGroupAssignments at the same index.
|
||||
RunPlacementScorePlugins(ctx context.Context, state fwk.PodGroupCycleState, podGroupInfo fwk.PodGroupInfo, placements []*fwk.PodGroupAssignments, placementStates []fwk.PlacementCycleState) (ns []fwk.PlacementPluginScores, status *fwk.Status)
|
||||
|
||||
// HasFilterPlugins returns true if at least one Filter plugin is defined.
|
||||
HasFilterPlugins() bool
|
||||
|
|
|
|||
|
|
@ -225,7 +225,7 @@ func (s *placementFeasibleState) Clone() fwk.StateData {
|
|||
}
|
||||
}
|
||||
|
||||
func getPlacementFeasibleState(placementCycleState fwk.PodGroupCycleState) *placementFeasibleState {
|
||||
func getPlacementFeasibleState(placementCycleState fwk.PlacementCycleState) *placementFeasibleState {
|
||||
state, err := placementCycleState.Read(placementFeasibleStateKey)
|
||||
if err != nil {
|
||||
state = &placementFeasibleState{}
|
||||
|
|
@ -237,7 +237,7 @@ func getPlacementFeasibleState(placementCycleState fwk.PodGroupCycleState) *plac
|
|||
// PlacementFeasible is responsible for enforcing the gang's MinCount constraint in the pod group scheduling cycle.
|
||||
// The function will only return success once the gang's MinCount is satisfied or if the pod group is not using gang scheduling policy.
|
||||
// In case there are not enough remaining pods to satisfy the gang's MinCount, it returns UnschedulableAndUnresolvable which will terminate the pod group scheduling cycle early.
|
||||
func (pl *GangScheduling) PlacementFeasible(ctx context.Context, placementCycleState fwk.PodGroupCycleState, podGroupInfo fwk.PodGroupInfo) *fwk.Status {
|
||||
func (pl *GangScheduling) PlacementFeasible(ctx context.Context, placementCycleState fwk.PlacementCycleState, podGroupInfo fwk.PodGroupInfo) *fwk.Status {
|
||||
pg, err := pl.podGroupLister.PodGroups(podGroupInfo.GetNamespace()).Get(podGroupInfo.GetName())
|
||||
if err != nil {
|
||||
return fwk.AsStatus(fmt.Errorf("failed to get podGroup %s to compute gang feasibility: %w", klog.KObj(podGroupInfo), err))
|
||||
|
|
|
|||
|
|
@ -760,6 +760,6 @@ func (f *Fit) PlacementScoreExtensions() fwk.PlacementScoreExtensions {
|
|||
}
|
||||
|
||||
// ScorePlacement scores capacity ratio on all nodes in the placement including all assigned pod group pods' resource requests.
|
||||
func (f *Fit) ScorePlacement(ctx context.Context, state fwk.PodGroupCycleState, podGroup fwk.PodGroupInfo, placement *fwk.PodGroupAssignments) (int64, *fwk.Status) {
|
||||
func (f *Fit) ScorePlacement(ctx context.Context, state fwk.PlacementCycleState, podGroup fwk.PodGroupInfo, placement *fwk.PodGroupAssignments) (int64, *fwk.Status) {
|
||||
return f.placementScorer.scorePlacement(ctx, podGroup, placement)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -49,7 +49,7 @@ func (pl *PodGroupPodsCount) Name() string {
|
|||
// Both scheduled (assumed/assigned) pods and the proposed assignments are taken into consideration
|
||||
// when computing the score. This ensures that the relative difference between choices is reduced,
|
||||
// and small changes to the total count result in small changes to the score.
|
||||
func (pl *PodGroupPodsCount) ScorePlacement(ctx context.Context, state fwk.PodGroupCycleState, podGroup fwk.PodGroupInfo, placement *fwk.PodGroupAssignments) (int64, *fwk.Status) {
|
||||
func (pl *PodGroupPodsCount) ScorePlacement(ctx context.Context, state fwk.PlacementCycleState, podGroup fwk.PodGroupInfo, placement *fwk.PodGroupAssignments) (int64, *fwk.Status) {
|
||||
pgState, err := pl.handle.SnapshotSharedLister().PodGroupStates().Get(podGroup.GetNamespace(), podGroup.GetName())
|
||||
if err != nil {
|
||||
return 0, fwk.AsStatus(err)
|
||||
|
|
|
|||
|
|
@ -1482,13 +1482,21 @@ func (f *frameworkImpl) runScoreExtension(ctx context.Context, pl fwk.ScorePlugi
|
|||
// It also returns *Status, which is set to non-success if any of the plugins returns
|
||||
// a non-success status.
|
||||
//
|
||||
// placementStates holds the per-placement cycle state for each entry in podGroupAssignments
|
||||
// (1:1 by index): placementStates[i] is the state with which podGroupAssignments[i] was
|
||||
// processed, and is passed to ScorePlacement for that placement.
|
||||
//
|
||||
// This function mostly duplicates RunScorePlugins. Any changes to it should likely be reflected in both places.
|
||||
func (f *frameworkImpl) RunPlacementScorePlugins(ctx context.Context, state fwk.PodGroupCycleState, podGroupInfo fwk.PodGroupInfo, podGroupAssignments []*fwk.PodGroupAssignments) (ps []fwk.PlacementPluginScores, status *fwk.Status) {
|
||||
func (f *frameworkImpl) RunPlacementScorePlugins(ctx context.Context, state fwk.PodGroupCycleState, podGroupInfo fwk.PodGroupInfo, podGroupAssignments []*fwk.PodGroupAssignments, placementStates []fwk.PlacementCycleState) (ps []fwk.PlacementPluginScores, status *fwk.Status) {
|
||||
startTime := time.Now()
|
||||
defer func() {
|
||||
metrics.FrameworkExtensionPointDuration.WithLabelValues(metrics.PlacementScore, status.Code().String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
|
||||
}()
|
||||
|
||||
if len(podGroupAssignments) != len(placementStates) {
|
||||
return nil, fwk.AsStatus(fmt.Errorf("expected one PlacementCycleState per PodGroupAssignments, got %d placement states for %d assignments", len(placementStates), len(podGroupAssignments)))
|
||||
}
|
||||
|
||||
allPlacementPluginScores := make([]fwk.PlacementPluginScores, len(podGroupAssignments))
|
||||
numPlugins := len(f.placementScorePlugins)
|
||||
plugins := make([]fwk.PlacementScorePlugin, 0, numPlugins)
|
||||
|
|
@ -1520,7 +1528,7 @@ func (f *frameworkImpl) RunPlacementScorePlugins(ctx context.Context, state fwk.
|
|||
logger := klog.LoggerWithName(logger, pl.Name())
|
||||
ctx = klog.NewContext(ctx, logger)
|
||||
}
|
||||
s, status := f.runPlacementScorePlugin(ctx, pl, state, podGroupInfo, pga)
|
||||
s, status := f.runPlacementScorePlugin(ctx, pl, placementStates[index], podGroupInfo, pga)
|
||||
if !status.IsSuccess() {
|
||||
err := fmt.Errorf("plugin %q failed with: %w", pl.Name(), status.AsError())
|
||||
errCh.SendWithCancel(err, cancel)
|
||||
|
|
@ -1589,7 +1597,7 @@ func (f *frameworkImpl) RunPlacementScorePlugins(ctx context.Context, state fwk.
|
|||
return allPlacementPluginScores, nil
|
||||
}
|
||||
|
||||
func (f *frameworkImpl) runPlacementScorePlugin(ctx context.Context, pl fwk.PlacementScorePlugin, state fwk.PodGroupCycleState, podGroup fwk.PodGroupInfo, placement *fwk.PodGroupAssignments) (int64, *fwk.Status) {
|
||||
func (f *frameworkImpl) runPlacementScorePlugin(ctx context.Context, pl fwk.PlacementScorePlugin, state fwk.PlacementCycleState, podGroup fwk.PodGroupInfo, placement *fwk.PodGroupAssignments) (int64, *fwk.Status) {
|
||||
if !state.ShouldRecordPluginMetrics() {
|
||||
return pl.ScorePlacement(ctx, state, podGroup, placement)
|
||||
}
|
||||
|
|
@ -2014,7 +2022,7 @@ func (f *frameworkImpl) runPermitPlugin(ctx context.Context, pl fwk.PermitPlugin
|
|||
// If any plugin returns invalid status, the result will be Error and the remaining plugins won't be invoked.
|
||||
// Otherwise, if at least 1 plugin returns UnschedulableAndUnresolvable, the remaining plugins won't be invoked and the result will be UnschdulableAndUnresolvable.
|
||||
// Otherwise, if at least 1 plugin returns Unschedulable, the remaining plugins will be invoked and the result will be Unschedulable.
|
||||
func (f *frameworkImpl) RunPlacementFeasiblePlugins(ctx context.Context, placementCycleState fwk.PodGroupCycleState, podGroupInfo fwk.PodGroupInfo) (status *fwk.Status) {
|
||||
func (f *frameworkImpl) RunPlacementFeasiblePlugins(ctx context.Context, placementCycleState fwk.PlacementCycleState, podGroupInfo fwk.PodGroupInfo) (status *fwk.Status) {
|
||||
startTime := time.Now()
|
||||
defer func() {
|
||||
metrics.FrameworkExtensionPointDuration.WithLabelValues(metrics.PlacementFeasible, status.Code().String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
|
||||
|
|
@ -2041,7 +2049,7 @@ func (f *frameworkImpl) RunPlacementFeasiblePlugins(ctx context.Context, placeme
|
|||
return status
|
||||
}
|
||||
|
||||
func (f *frameworkImpl) runPlacementFeasiblePlugin(ctx context.Context, pl framework.PlacementFeasiblePlugin, state fwk.PodGroupCycleState, podGroup fwk.PodGroupInfo) *fwk.Status {
|
||||
func (f *frameworkImpl) runPlacementFeasiblePlugin(ctx context.Context, pl framework.PlacementFeasiblePlugin, state fwk.PlacementCycleState, podGroup fwk.PodGroupInfo) *fwk.Status {
|
||||
if !state.ShouldRecordPluginMetrics() {
|
||||
return pl.PlacementFeasible(ctx, state, podGroup)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -228,7 +228,7 @@ func (pl *TestPlugin) PreFilter(ctx context.Context, state fwk.CycleState, p *v1
|
|||
return pl.inj.PreFilterResult, fwk.NewStatus(fwk.Code(pl.inj.PreFilterStatus), injectReason)
|
||||
}
|
||||
|
||||
func (pl *TestPlugin) PlacementFeasible(ctx context.Context, state fwk.PodGroupCycleState, podGroup fwk.PodGroupInfo) *fwk.Status {
|
||||
func (pl *TestPlugin) PlacementFeasible(ctx context.Context, state fwk.PlacementCycleState, podGroup fwk.PodGroupInfo) *fwk.Status {
|
||||
return fwk.NewStatus(fwk.Code(pl.inj.PlacementFeasibleStatus), injectReason)
|
||||
}
|
||||
|
||||
|
|
@ -278,7 +278,7 @@ func (pl *TestPlugin) GeneratePlacements(ctx context.Context, state fwk.PodGroup
|
|||
return &fwk.GeneratePlacementsResult{Placements: pl.inj.GeneratePlacementsResult}, fwk.NewStatus(fwk.Code(pl.inj.GeneratePlacementsStatus), injectReason)
|
||||
}
|
||||
|
||||
func (pl *TestPlugin) ScorePlacement(ctx context.Context, state fwk.PodGroupCycleState, podGroup fwk.PodGroupInfo, placement *fwk.PodGroupAssignments) (int64, *fwk.Status) {
|
||||
func (pl *TestPlugin) ScorePlacement(ctx context.Context, state fwk.PlacementCycleState, podGroup fwk.PodGroupInfo, placement *fwk.PodGroupAssignments) (int64, *fwk.Status) {
|
||||
return 0, fwk.NewStatus(fwk.Code(pl.inj.PlacementScoreStatus), injectReason)
|
||||
}
|
||||
|
||||
|
|
@ -795,7 +795,7 @@ type mockPlacementFeasiblePlugin struct {
|
|||
|
||||
func (p *mockPlacementFeasiblePlugin) Name() string { return p.name }
|
||||
|
||||
func (p *mockPlacementFeasiblePlugin) PlacementFeasible(ctx context.Context, state fwk.PodGroupCycleState, podGroup fwk.PodGroupInfo) *fwk.Status {
|
||||
func (p *mockPlacementFeasiblePlugin) PlacementFeasible(ctx context.Context, state fwk.PlacementCycleState, podGroup fwk.PodGroupInfo) *fwk.Status {
|
||||
p.called = true
|
||||
return p.status
|
||||
}
|
||||
|
|
@ -3666,7 +3666,7 @@ func TestRecordingMetrics(t *testing.T) {
|
|||
{
|
||||
name: "PlacementScore - Success",
|
||||
action: func(ctx context.Context, f framework.Framework) {
|
||||
f.RunPlacementScorePlugins(ctx, state, nil, []*fwk.PodGroupAssignments{{Placement: &fwk.Placement{}}})
|
||||
f.RunPlacementScorePlugins(ctx, state, nil, []*fwk.PodGroupAssignments{{Placement: &fwk.Placement{}}}, []fwk.PlacementCycleState{state})
|
||||
},
|
||||
wantExtensionPoint: "PlacementScore",
|
||||
wantStatus: fwk.Success,
|
||||
|
|
@ -3750,7 +3750,7 @@ func TestRecordingMetrics(t *testing.T) {
|
|||
{
|
||||
name: "PlacementScore - Error",
|
||||
action: func(ctx context.Context, f framework.Framework) {
|
||||
f.RunPlacementScorePlugins(ctx, state, nil, []*fwk.PodGroupAssignments{{Placement: &fwk.Placement{}}})
|
||||
f.RunPlacementScorePlugins(ctx, state, nil, []*fwk.PodGroupAssignments{{Placement: &fwk.Placement{}}}, []fwk.PlacementCycleState{state})
|
||||
},
|
||||
inject: injectedResult{PlacementScoreStatus: int(fwk.Error)},
|
||||
wantExtensionPoint: "PlacementScore",
|
||||
|
|
@ -4440,7 +4440,7 @@ func (pl *testPlacementScorePlugin) Name() string {
|
|||
return pl.name
|
||||
}
|
||||
|
||||
func (pl *testPlacementScorePlugin) ScorePlacement(ctx context.Context, state fwk.PodGroupCycleState, podGroup fwk.PodGroupInfo, placement *fwk.PodGroupAssignments) (int64, *fwk.Status) {
|
||||
func (pl *testPlacementScorePlugin) ScorePlacement(ctx context.Context, state fwk.PlacementCycleState, podGroup fwk.PodGroupInfo, placement *fwk.PodGroupAssignments) (int64, *fwk.Status) {
|
||||
r := pl.results[placement.Placement]
|
||||
return r.score, r.status
|
||||
}
|
||||
|
|
@ -4682,11 +4682,13 @@ func TestRunPlacementScorePlugins(t *testing.T) {
|
|||
t.Fatalf("Unexpected error during calling NewFramework, got %v", err)
|
||||
}
|
||||
assumedPlacements := make([]*fwk.PodGroupAssignments, len(placements))
|
||||
placementStates := make([]fwk.PlacementCycleState, len(placements))
|
||||
for i := range placements {
|
||||
assumedPlacements[i] = &fwk.PodGroupAssignments{Placement: placements[i]}
|
||||
placementStates[i] = framework.NewCycleState()
|
||||
}
|
||||
|
||||
result, status := fw.RunPlacementScorePlugins(ctx, framework.NewCycleState(), nil, assumedPlacements)
|
||||
result, status := fw.RunPlacementScorePlugins(ctx, framework.NewCycleState(), nil, assumedPlacements, placementStates)
|
||||
if status.Code() != tt.wantStatusCode {
|
||||
t.Errorf("got status code %s, want %s", status.Code(), tt.wantStatusCode)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -129,7 +129,7 @@ type podSchedulingContext struct {
|
|||
}
|
||||
|
||||
// initPodSchedulingContext initializes the scheduling context of a single pod for pod group scheduling cycle.
|
||||
func initPodSchedulingContext(ctx context.Context, pod *v1.Pod, podGroupState *framework.CycleState, postFilterMode podGroupPostFilterMode) *podSchedulingContext {
|
||||
func initPodSchedulingContext(ctx context.Context, pod *v1.Pod, placementCycleState *framework.CycleState, postFilterMode podGroupPostFilterMode) *podSchedulingContext {
|
||||
logger := klog.FromContext(ctx)
|
||||
// TODO(knelasevero): Remove duplicated keys from log entry calls
|
||||
// When contextualized logging hits GA
|
||||
|
|
@ -147,8 +147,11 @@ func initPodSchedulingContext(ctx context.Context, pod *v1.Pod, podGroupState *f
|
|||
podsToActivate := framework.NewPodsToActivate()
|
||||
state.Write(framework.PodsToActivateKey, podsToActivate)
|
||||
|
||||
podGroupCycleState := placementCycleState.GetPodGroupSchedulingCycle()
|
||||
// Marks this cycle as a pod group scheduling cycle.
|
||||
state.SetPodGroupSchedulingCycle(podGroupState)
|
||||
state.SetPodGroupSchedulingCycle(podGroupCycleState)
|
||||
// Set the placement cycle state so per-pod plugins can access placement-scoped data.
|
||||
state.SetPlacementCycleState(placementCycleState)
|
||||
|
||||
// Skip post filters if requested.
|
||||
switch postFilterMode {
|
||||
|
|
@ -305,30 +308,29 @@ type podGroupAlgorithmResult struct {
|
|||
// waitingOnPreemption indicates whether this pod group requires or is waiting for preemption to complete.
|
||||
// This can only be set to true when the status is Unschedulable.
|
||||
waitingOnPreemption bool
|
||||
// placementCycleState is the state with which this placement was processed.
|
||||
placementCycleState fwk.PlacementCycleState
|
||||
}
|
||||
|
||||
// podGroupSchedulingDefaultAlgorithm runs the default algorithm for scheduling a pod group.
|
||||
// It tries to schedule each pod using standard filtering and scoring logic in a fixed order.
|
||||
// If a pod requires preemption to be schedulable, subsequent pods in the algorithm
|
||||
// treat that pod as already scheduled on that node with victims being already removed in memory.
|
||||
func (sched *Scheduler) podGroupSchedulingDefaultAlgorithm(ctx context.Context, schedFwk framework.Framework, podGroupCycleState *framework.CycleState, podGroupInfo *framework.QueuedPodGroupInfo, postFilterMode podGroupPostFilterMode) podGroupAlgorithmResult {
|
||||
func (sched *Scheduler) podGroupSchedulingDefaultAlgorithm(ctx context.Context, schedFwk framework.Framework, placementCycleState *framework.CycleState, podGroupInfo *framework.QueuedPodGroupInfo, postFilterMode podGroupPostFilterMode) podGroupAlgorithmResult {
|
||||
result := podGroupAlgorithmResult{
|
||||
podResults: make([]algorithmResult, 0, len(podGroupInfo.QueuedPodInfos)),
|
||||
status: fwk.NewStatus(fwk.Unschedulable).WithError(errPodGroupUnschedulable),
|
||||
waitingOnPreemption: false,
|
||||
placementCycleState: placementCycleState,
|
||||
}
|
||||
|
||||
placementCycleState := framework.NewCycleState()
|
||||
placementCycleState.SetRecordPluginMetrics(true)
|
||||
placementCycleState.SetPodGroupSchedulingCycle(podGroupCycleState)
|
||||
|
||||
logger := klog.FromContext(ctx)
|
||||
logger.V(5).Info("Running a pod group scheduling algorithm", "podGroup", klog.KObj(podGroupInfo), "unscheduledPodsCount", len(podGroupInfo.QueuedPodInfos))
|
||||
|
||||
requiresPreemption := false
|
||||
anyScheduled := false
|
||||
for _, podInfo := range podGroupInfo.QueuedPodInfos {
|
||||
podResult, revertFn := sched.podGroupPodSchedulingAlgorithm(ctx, schedFwk, podGroupCycleState, podGroupInfo, podInfo, postFilterMode)
|
||||
podResult, revertFn := sched.podGroupPodSchedulingAlgorithm(ctx, schedFwk, placementCycleState, podGroupInfo, podInfo, postFilterMode)
|
||||
result.podResults = append(result.podResults, podResult)
|
||||
if revertFn != nil {
|
||||
// We unreserve the pod at the end of the whole algorithm (via defer) because it should be ultimately returned to the queue,
|
||||
|
|
@ -382,9 +384,9 @@ func (sched *Scheduler) podGroupSchedulingDefaultAlgorithm(ctx context.Context,
|
|||
|
||||
// podGroupPodSchedulingAlgorithm runs a scheduling algorithm for individual pod from a pod group.
|
||||
// It returns the algorithm result together with the revert function.
|
||||
func (sched *Scheduler) podGroupPodSchedulingAlgorithm(ctx context.Context, schedFwk framework.Framework, podGroupCycleState *framework.CycleState, podGroupInfo *framework.QueuedPodGroupInfo, podInfo *framework.QueuedPodInfo, postFilterMode podGroupPostFilterMode) (algorithmResult, func()) {
|
||||
func (sched *Scheduler) podGroupPodSchedulingAlgorithm(ctx context.Context, schedFwk framework.Framework, placementCycleState *framework.CycleState, podGroupInfo *framework.QueuedPodGroupInfo, podInfo *framework.QueuedPodInfo, postFilterMode podGroupPostFilterMode) (algorithmResult, func()) {
|
||||
pod := podInfo.Pod
|
||||
podCtx := initPodSchedulingContext(ctx, pod, podGroupCycleState, postFilterMode)
|
||||
podCtx := initPodSchedulingContext(ctx, pod, placementCycleState, postFilterMode)
|
||||
logger := podCtx.logger
|
||||
ctx = klog.NewContext(ctx, logger)
|
||||
start := time.Now()
|
||||
|
|
@ -451,9 +453,11 @@ func completePodGroupAlgorithmResult(ctx context.Context, podGroupInfo *framewor
|
|||
copy(newResults, podGroupResult.podResults)
|
||||
for i := numInResult; i < numInQueue; i++ {
|
||||
pInfo := podGroupInfo.QueuedPodInfos[i]
|
||||
placementCycleState := framework.NewCycleState()
|
||||
placementCycleState.SetPodGroupSchedulingCycle(podGroupState)
|
||||
newResults[i] = algorithmResult{
|
||||
pod: pInfo.Pod,
|
||||
podCtx: initPodSchedulingContext(ctx, pInfo.Pod, podGroupState, postFilterMode),
|
||||
podCtx: initPodSchedulingContext(ctx, pInfo.Pod, placementCycleState, postFilterMode),
|
||||
status: podGroupResult.status.Clone(),
|
||||
}
|
||||
}
|
||||
|
|
@ -656,7 +660,9 @@ func (sched *Scheduler) podGroupSchedulingPlacementAlgorithm(ctx context.Context
|
|||
status: fwk.AsStatus(fmt.Errorf("failed to assume pod group placement: %w", err)),
|
||||
}
|
||||
}
|
||||
result := sched.podGroupSchedulingDefaultAlgorithm(ctx, schedFwk, podGroupCycleState, podGroupInfo, postFilterMode)
|
||||
placementCycleState := framework.NewCycleState()
|
||||
placementCycleState.SetPodGroupSchedulingCycle(podGroupCycleState)
|
||||
result := sched.podGroupSchedulingDefaultAlgorithm(ctx, schedFwk, placementCycleState, podGroupInfo, postFilterMode)
|
||||
sched.nodeInfoSnapshot.ForgetPlacement()
|
||||
if result.status.IsError() {
|
||||
return result
|
||||
|
|
@ -696,9 +702,9 @@ func (sched *Scheduler) podGroupSchedulingPlacementAlgorithm(ctx context.Context
|
|||
|
||||
// findBestPlacement uses PlacementScore plugins to determine the best placement based on the scheduling results.
|
||||
func (sched *Scheduler) findBestPlacement(ctx context.Context, schedFwk framework.Framework, podGroupCycleState fwk.PodGroupCycleState, podGroupInfo *framework.QueuedPodGroupInfo, successfulResults map[*fwk.Placement]*podGroupAlgorithmResult) (*fwk.Placement, *fwk.Status) {
|
||||
placementPodGroupAssignments := makePodGroupAssignments(successfulResults)
|
||||
placementPodGroupAssignments, placementStates := makePodGroupAssignments(successfulResults)
|
||||
|
||||
scores, status := schedFwk.RunPlacementScorePlugins(ctx, podGroupCycleState, podGroupInfo, placementPodGroupAssignments)
|
||||
scores, status := schedFwk.RunPlacementScorePlugins(ctx, podGroupCycleState, podGroupInfo, placementPodGroupAssignments, placementStates)
|
||||
if !status.IsSuccess() {
|
||||
return nil, status
|
||||
}
|
||||
|
|
@ -728,16 +734,18 @@ func (sched *Scheduler) findBestPlacement(ctx context.Context, schedFwk framewor
|
|||
return bestScore.Placement, nil
|
||||
}
|
||||
|
||||
func makePodGroupAssignments(successfulResults map[*fwk.Placement]*podGroupAlgorithmResult) []*fwk.PodGroupAssignments {
|
||||
func makePodGroupAssignments(successfulResults map[*fwk.Placement]*podGroupAlgorithmResult) ([]*fwk.PodGroupAssignments, []fwk.PlacementCycleState) {
|
||||
placementPodGroupAssignments := make([]*fwk.PodGroupAssignments, 0, len(successfulResults))
|
||||
placementStates := make([]fwk.PlacementCycleState, 0, len(successfulResults))
|
||||
for placement, result := range successfulResults {
|
||||
proposedAssignments := makeProposedAssignments(result)
|
||||
placementPodGroupAssignments = append(placementPodGroupAssignments, &fwk.PodGroupAssignments{
|
||||
Placement: placement,
|
||||
ProposedAssignments: proposedAssignments,
|
||||
})
|
||||
placementStates = append(placementStates, result.placementCycleState)
|
||||
}
|
||||
return placementPodGroupAssignments
|
||||
return placementPodGroupAssignments, placementStates
|
||||
}
|
||||
|
||||
// makeProposedAssignments builds a list of proposedAssignments from the result of a pod group scheduling attempt.
|
||||
|
|
@ -759,5 +767,10 @@ func (sched *Scheduler) podGroupSchedulingAlgorithm(ctx context.Context, schedFw
|
|||
if utilfeature.DefaultFeatureGate.Enabled(features.TopologyAwareWorkloadScheduling) {
|
||||
return sched.podGroupSchedulingPlacementAlgorithm(podGroupCycleCtx, schedFwk, podGroupCycleState, podGroupInfo, postFilterMode)
|
||||
}
|
||||
return sched.podGroupSchedulingDefaultAlgorithm(podGroupCycleCtx, schedFwk, podGroupCycleState, podGroupInfo, postFilterMode)
|
||||
// The non-TAS default algorithm does not evaluate placement candidates, but it
|
||||
// still runs in a single implicit placement context so placement-scoped
|
||||
// extension points can use the same state plumbing as TAS.
|
||||
placementCycleState := framework.NewCycleState()
|
||||
placementCycleState.SetPodGroupSchedulingCycle(podGroupCycleState)
|
||||
return sched.podGroupSchedulingDefaultAlgorithm(podGroupCycleCtx, schedFwk, placementCycleState, podGroupInfo, postFilterMode)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -139,7 +139,7 @@ func (mp *fakePlacementFeasiblePlugin) Name() string {
|
|||
// - The outer slice represents distinct placements (e.g., when evaluating multiple topology placements).
|
||||
// - The inner slice represents the pod-by-pod evaluation within a single placement.
|
||||
// It uses placementCycleState to track how many pods have been evaluated in the current placement.
|
||||
func (mp *fakePlacementFeasiblePlugin) PlacementFeasible(ctx context.Context, placementCycleState fwk.PodGroupCycleState, podGroupInfo fwk.PodGroupInfo) *fwk.Status {
|
||||
func (mp *fakePlacementFeasiblePlugin) PlacementFeasible(ctx context.Context, placementCycleState fwk.PlacementCycleState, podGroupInfo fwk.PodGroupInfo) *fwk.Status {
|
||||
// If no mock statuses are configured, always succeed.
|
||||
if len(mp.placementFeasibleStatuses) == 0 {
|
||||
return nil
|
||||
|
|
@ -1642,7 +1642,9 @@ func TestSubmitPodGroupAlgorithmResult(t *testing.T) {
|
|||
|
||||
for i := range tt.algorithmResult.podResults {
|
||||
pod := podGroupInfo.QueuedPodInfos[i].Pod
|
||||
podCtx := initPodSchedulingContext(ctx, pod, podGroupCycleState, runAllPostFilters)
|
||||
placementCycleState := framework.NewCycleState()
|
||||
placementCycleState.SetPodGroupSchedulingCycle(podGroupCycleState)
|
||||
podCtx := initPodSchedulingContext(ctx, pod, placementCycleState, runAllPostFilters)
|
||||
tt.algorithmResult.podResults[i].podCtx = podCtx
|
||||
}
|
||||
|
||||
|
|
@ -2091,7 +2093,7 @@ func (mp *fakePlacementPlugin) PlacementScoreExtensions() fwk.PlacementScoreExte
|
|||
return nil
|
||||
}
|
||||
|
||||
func (mp *fakePlacementPlugin) ScorePlacement(ctx context.Context, state fwk.PodGroupCycleState, podGroup fwk.PodGroupInfo, placement *fwk.PodGroupAssignments) (int64, *fwk.Status) {
|
||||
func (mp *fakePlacementPlugin) ScorePlacement(ctx context.Context, state fwk.PlacementCycleState, podGroup fwk.PodGroupInfo, placement *fwk.PodGroupAssignments) (int64, *fwk.Status) {
|
||||
return mp.scorePlacementsResult[placement.Name], mp.scorePlacementsStatus[placement.Name]
|
||||
}
|
||||
|
||||
|
|
@ -2428,6 +2430,7 @@ func TestPodGroupSchedulingPlacementAlgorithm(t *testing.T) {
|
|||
algorithmResult{},
|
||||
ScheduleResult{},
|
||||
fwk.Status{}),
|
||||
cmpopts.IgnoreFields(podGroupAlgorithmResult{}, "placementCycleState"),
|
||||
cmpopts.IgnoreFields(algorithmResult{}, "podCtx", "schedulingDuration"),
|
||||
statusCmpOpt,
|
||||
}
|
||||
|
|
@ -2593,6 +2596,182 @@ func TestPodGroupSchedulingPlacementAlgorithm_Scoring(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
// placementStateTracker is a fake plugin that writes to PlacementCycleState during Filter
|
||||
// and reads from it during ScorePlacement, to verify the lifecycle of PlacementCycleState.
|
||||
type placementStateTracker struct {
|
||||
name string
|
||||
mu sync.Mutex
|
||||
// scoreReadValues records what value was read from PlacementCycleState
|
||||
// during each ScorePlacement call, keyed by placement name.
|
||||
scoreReadValues map[string]string
|
||||
// generatePlacementsResult defines the placements to generate.
|
||||
generatePlacementsResult map[string][]string
|
||||
}
|
||||
|
||||
type placementStateData struct {
|
||||
value string
|
||||
}
|
||||
|
||||
func (d *placementStateData) Clone() fwk.StateData { return d }
|
||||
|
||||
var placementStateKey fwk.StateKey = "placementStateTracker"
|
||||
|
||||
var _ fwk.FilterPlugin = &placementStateTracker{}
|
||||
var _ fwk.PlacementGeneratePlugin = &placementStateTracker{}
|
||||
var _ fwk.PlacementScorePlugin = &placementStateTracker{}
|
||||
|
||||
func (p *placementStateTracker) Name() string { return p.name }
|
||||
|
||||
func (p *placementStateTracker) Filter(ctx context.Context, state fwk.CycleState, pod *v1.Pod, nodeInfo fwk.NodeInfo) *fwk.Status {
|
||||
placementState := state.GetPlacementCycleState()
|
||||
if placementState == nil {
|
||||
return fwk.NewStatus(fwk.Error, "PlacementCycleState is nil during Filter")
|
||||
}
|
||||
|
||||
// Write the node name as a marker so ScorePlacement can verify
|
||||
// which placement's state it received.
|
||||
placementState.Write(placementStateKey, &placementStateData{value: nodeInfo.Node().Name})
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *placementStateTracker) ScorePlacement(ctx context.Context, state fwk.PlacementCycleState, podGroup fwk.PodGroupInfo, placement *fwk.PodGroupAssignments) (int64, *fwk.Status) {
|
||||
if state == nil {
|
||||
return 0, fwk.NewStatus(fwk.Error, "PlacementCycleState is nil during ScorePlacement")
|
||||
}
|
||||
|
||||
data, err := state.Read(placementStateKey)
|
||||
if err != nil {
|
||||
return 0, fwk.NewStatus(fwk.Error, fmt.Sprintf("failed to read PlacementCycleState for %s: %v", placement.Name, err))
|
||||
}
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
p.scoreReadValues[placement.Name] = data.(*placementStateData).value
|
||||
return 1, nil
|
||||
}
|
||||
|
||||
func (p *placementStateTracker) PlacementScoreExtensions() fwk.PlacementScoreExtensions {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *placementStateTracker) GeneratePlacements(ctx context.Context, state fwk.PodGroupCycleState, podGroup fwk.PodGroupInfo, parentPlacement *fwk.Placement) (*fwk.GeneratePlacementsResult, *fwk.Status) {
|
||||
parentNodes := map[string]fwk.NodeInfo{}
|
||||
for _, node := range parentPlacement.Nodes {
|
||||
parentNodes[node.Node().Name] = node
|
||||
}
|
||||
|
||||
resultPlacements := make([]*fwk.Placement, 0, len(p.generatePlacementsResult))
|
||||
for placementName, nodeNames := range p.generatePlacementsResult {
|
||||
placement := &fwk.Placement{Name: placementName}
|
||||
for _, nodeName := range nodeNames {
|
||||
placement.Nodes = append(placement.Nodes, parentNodes[nodeName])
|
||||
}
|
||||
resultPlacements = append(resultPlacements, placement)
|
||||
}
|
||||
return &fwk.GeneratePlacementsResult{Placements: resultPlacements}, nil
|
||||
}
|
||||
|
||||
func TestPlacementCycleStateLifecycle(t *testing.T) {
|
||||
featuregatetesting.SetFeatureGatesDuringTest(t, utilfeature.DefaultFeatureGate, featuregatetesting.FeatureOverrides{
|
||||
features.TopologyAwareWorkloadScheduling: true,
|
||||
features.GenericWorkload: true,
|
||||
features.GangScheduling: true,
|
||||
})
|
||||
|
||||
// A single scenario exercises both isolation and continuity:
|
||||
// - Filter writes a node-name marker into PlacementCycleState during each placement's simulation.
|
||||
// - ScorePlacement reads from the placement state after all simulations.
|
||||
// Assertions verify:
|
||||
// 1. Each placement's scorer reads only the value its own simulation wrote (isolation).
|
||||
// 2. Data written during each placement's simulation remains readable during its scoring (continuity from simulation to scoring).
|
||||
|
||||
nodes := []*v1.Node{
|
||||
st.MakeNode().Name("node1").Obj(),
|
||||
st.MakeNode().Name("node2").Obj(),
|
||||
}
|
||||
podGroupPod := st.MakePod().Name("foo").UID("foo").PodGroupName("pg").Obj()
|
||||
|
||||
logger, ctx := ktesting.NewTestContext(t)
|
||||
|
||||
informerFactory := informers.NewSharedInformerFactory(clientsetfake.NewClientset(), 0)
|
||||
queue := internalqueue.NewSchedulingQueue(nil, informerFactory)
|
||||
|
||||
tracker := &placementStateTracker{
|
||||
name: "StateTracker",
|
||||
scoreReadValues: make(map[string]string),
|
||||
generatePlacementsResult: map[string][]string{
|
||||
"placementA": {nodes[0].Name},
|
||||
"placementB": {nodes[1].Name},
|
||||
},
|
||||
}
|
||||
|
||||
registry := []tf.RegisterPluginFunc{
|
||||
tf.RegisterPlacementGeneratePlugin(tracker.Name(), func(_ context.Context, _ runtime.Object, _ fwk.Handle) (fwk.Plugin, error) {
|
||||
return tracker, nil
|
||||
}),
|
||||
tf.RegisterPlacementScorePlugin(tracker.Name(), func(_ context.Context, _ runtime.Object, _ fwk.Handle) (fwk.Plugin, error) {
|
||||
return tracker, nil
|
||||
}, 1),
|
||||
tf.RegisterFilterPlugin(tracker.Name(), func(_ context.Context, _ runtime.Object, _ fwk.Handle) (fwk.Plugin, error) {
|
||||
return tracker, nil
|
||||
}),
|
||||
}
|
||||
|
||||
snapshot := internalcache.NewEmptySnapshot()
|
||||
schedFwk, err := tf.NewFramework(ctx,
|
||||
append(registry,
|
||||
tf.RegisterQueueSortPlugin(queuesort.Name, queuesort.New),
|
||||
tf.RegisterBindPlugin(defaultbinder.Name, defaultbinder.New),
|
||||
),
|
||||
"test-scheduler",
|
||||
frameworkruntime.WithInformerFactory(informerFactory),
|
||||
frameworkruntime.WithSnapshotSharedLister(snapshot),
|
||||
frameworkruntime.WithPodNominator(queue),
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to create framework: %v", err)
|
||||
}
|
||||
|
||||
cache := internalcache.New(ctx, nil, true)
|
||||
for _, node := range nodes {
|
||||
cache.AddNode(logger, node)
|
||||
}
|
||||
|
||||
sched := &Scheduler{
|
||||
Cache: cache,
|
||||
nodeInfoSnapshot: snapshot,
|
||||
SchedulingQueue: queue,
|
||||
Profiles: profile.Map{"test-scheduler": schedFwk},
|
||||
}
|
||||
sched.SchedulePod = sched.schedulePod
|
||||
|
||||
if err := sched.Cache.UpdateSnapshot(logger, sched.nodeInfoSnapshot); err != nil {
|
||||
t.Fatalf("Failed to update snapshot: %v", err)
|
||||
}
|
||||
|
||||
pgInfo := &framework.QueuedPodGroupInfo{
|
||||
QueuedPodInfos: []*framework.QueuedPodInfo{
|
||||
{PodInfo: &framework.PodInfo{Pod: podGroupPod}},
|
||||
},
|
||||
PodGroupInfo: &framework.PodGroupInfo{
|
||||
UnscheduledPods: []*v1.Pod{podGroupPod},
|
||||
},
|
||||
}
|
||||
|
||||
result := sched.podGroupSchedulingPlacementAlgorithm(ctx, schedFwk, framework.NewCycleState(), pgInfo, runAllPostFilters)
|
||||
if !result.status.IsSuccess() {
|
||||
t.Fatalf("Expected success, got: %v", result.status)
|
||||
}
|
||||
|
||||
// Each placement's scorer must read only what its own simulation wrote
|
||||
// (placementA simulated on node1, placementB on node2). This proves both:
|
||||
// - Continuity: data written during a placement's simulation is readable during its scoring.
|
||||
// - Isolation: a placement's scorer does not see another placement's writes.
|
||||
expectedScoreReadValues := map[string]string{"placementA": "node1", "placementB": "node2"}
|
||||
if diff := cmp.Diff(expectedScoreReadValues, tracker.scoreReadValues); diff != "" {
|
||||
t.Errorf("Unexpected scoreReadValues (-want,+got)\n%s", diff)
|
||||
}
|
||||
}
|
||||
|
||||
type fakeDefaultPreemption struct {
|
||||
*fakePodGroupPlugin
|
||||
}
|
||||
|
|
|
|||
|
|
@ -96,14 +96,47 @@ type CycleState interface {
|
|||
// IsPodGroupSchedulingCycle returns true if this cycle is a pod group scheduling cycle.
|
||||
// If set to false, it means that the pod referencing this CycleState either passed the pod group cycle
|
||||
// or doesn't belong to any pod group.
|
||||
// This field can only be set to true when GenericWorkload feature flag is enabled.
|
||||
// This field can only be true when GenericWorkload feature flag is enabled.
|
||||
IsPodGroupSchedulingCycle() bool
|
||||
// GetPodGroupSchedulingCycle gets the cycle state of the PodGroup for a Pod.
|
||||
// This should be only used when GenericWorkload feature flag is enabled.
|
||||
GetPodGroupSchedulingCycle() PodGroupCycleState
|
||||
// SetPodGroupSchedulingCycle sets the cycle state of the PodGroup for a Pod.
|
||||
// GetPlacementCycleState gets the cycle state of the current Placement for a Pod.
|
||||
// Returns nil if this pod is not being scheduled within a placement context.
|
||||
// This should be only used when GenericWorkload feature flag is enabled.
|
||||
SetPodGroupSchedulingCycle(PodGroupCycleState)
|
||||
GetPlacementCycleState() PlacementCycleState
|
||||
// SetPlacementCycleState sets the cycle state of the current Placement for a Pod.
|
||||
// This should be only used when GenericWorkload feature flag is enabled.
|
||||
SetPlacementCycleState(PlacementCycleState)
|
||||
}
|
||||
|
||||
// PlacementCycleState provides a mechanism for plugins to store and retrieve arbitrary data
|
||||
// scoped to a single placement candidate within a pod group scheduling cycle.
|
||||
// Data stored in PlacementCycleState is shared across all pods scheduled within the same
|
||||
// placement iteration and is passed to ScorePlacement for that placement.
|
||||
// PlacementCycleState does not provide any data protection, as all plugins are assumed to be
|
||||
// trusted.
|
||||
type PlacementCycleState interface {
|
||||
// ShouldRecordPluginMetrics returns whether metrics.PluginExecutionDuration metrics
|
||||
// should be recorded.
|
||||
// This function is mostly for the scheduling framework runtime, plugins usually don't have to use it.
|
||||
ShouldRecordPluginMetrics() bool
|
||||
// Read retrieves data with the given "key" from PlacementCycleState. If the key is not
|
||||
// present, ErrNotFound is returned.
|
||||
//
|
||||
// See PlacementCycleState for notes on concurrency.
|
||||
Read(key StateKey) (StateData, error)
|
||||
// Write stores the given "val" in PlacementCycleState with the given "key".
|
||||
//
|
||||
// See PlacementCycleState for notes on concurrency.
|
||||
Write(key StateKey, val StateData)
|
||||
// Delete deletes data with the given key from PlacementCycleState.
|
||||
//
|
||||
// See PlacementCycleState for notes on concurrency.
|
||||
Delete(key StateKey)
|
||||
// GetPodGroupSchedulingCycle gets the cycle state of the PodGroup for this Placement.
|
||||
// This should be only used when GenericWorkload feature flag is enabled.
|
||||
GetPodGroupSchedulingCycle() PodGroupCycleState
|
||||
}
|
||||
|
||||
// PodGroupCycleState provides a mechanism for plugins that operate on pod groups to store and retrieve arbitrary data.
|
||||
|
|
|
|||
|
|
@ -795,9 +795,10 @@ type PlacementScorePlugin interface {
|
|||
// ScorePlacement calculates a score for a given Placement.
|
||||
// This function is called only for Placements that have been deemed feasible for the sufficient number of pods in the PodGroup scheduling cycle.
|
||||
// The PodGroupAssignments indicates the node assigned to each pod within this Placement.
|
||||
// The state is scoped to the Placement being scored and can be used to access the owning PodGroup cycle state.
|
||||
// The returned score is a int64 with higher scores generally indicating more preferable Placements.
|
||||
// Plugins can implement various scoring strategies, such as bin packing to minimize resource fragmentation.
|
||||
ScorePlacement(ctx context.Context, state PodGroupCycleState, podGroup PodGroupInfo, placement *PodGroupAssignments) (int64, *Status)
|
||||
ScorePlacement(ctx context.Context, state PlacementCycleState, podGroup PodGroupInfo, placement *PodGroupAssignments) (int64, *Status)
|
||||
|
||||
// PlacementScoreExtensions returns a PlacementScoreExtensions interface if it implements one, or nil if does not.
|
||||
PlacementScoreExtensions() PlacementScoreExtensions
|
||||
|
|
|
|||
Loading…
Reference in a new issue