Merge pull request kubernetes#127904 from towca/jtuznik/dra-autoscaling

DRA: allow Cluster Autoscaler to integrate with DRA scheduler plugin

Author: k8s-ci-robot

pkg/scheduler/framework/autoscaler_contract/framework_contract_test.go

@@ -36,6 +36,7 @@ import (
 type frameworkContract interface {
 	RunPreFilterPlugins(ctx context.Context, state *framework.CycleState, pod *v1.Pod) (*framework.PreFilterResult, *framework.Status, sets.Set[string])
 	RunFilterPlugins(context.Context, *framework.CycleState, *v1.Pod, *framework.NodeInfo) *framework.Status
+	RunReservePluginsReserve(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) *framework.Status
 }
 
 func TestFrameworkContract(t *testing.T) {

pkg/scheduler/framework/autoscaler_contract/lister_contract_test.go

@@ -21,12 +21,20 @@ limitations under the License.
 package contract
 
 import (
+	resourceapi "k8s.io/api/resource/v1alpha3"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/dynamic-resource-allocation/structured"
 	"k8s.io/kubernetes/pkg/scheduler/framework"
 )
 
 var _ framework.NodeInfoLister = &nodeInfoListerContract{}
 var _ framework.StorageInfoLister = &storageInfoListerContract{}
 var _ framework.SharedLister = &shareListerContract{}
+var _ framework.ResourceSliceLister = &resourceSliceListerContract{}
+var _ framework.DeviceClassLister = &deviceClassListerContract{}
+var _ framework.ResourceClaimTracker = &resourceClaimTrackerContract{}
+var _ framework.SharedDRAManager = &sharedDRAManagerContract{}
 
 type nodeInfoListerContract struct{}
 
@@ -61,3 +69,66 @@ func (c *shareListerContract) NodeInfos() framework.NodeInfoLister {
 func (c *shareListerContract) StorageInfos() framework.StorageInfoLister {
 	return nil
 }
+
+type resourceSliceListerContract struct{}
+
+func (c *resourceSliceListerContract) List() ([]*resourceapi.ResourceSlice, error) {
+	return nil, nil
+}
+
+type deviceClassListerContract struct{}
+
+func (c *deviceClassListerContract) List() ([]*resourceapi.DeviceClass, error) {
+	return nil, nil
+}
+
+func (c *deviceClassListerContract) Get(_ string) (*resourceapi.DeviceClass, error) {
+	return nil, nil
+}
+
+type resourceClaimTrackerContract struct{}
+
+func (r *resourceClaimTrackerContract) List() ([]*resourceapi.ResourceClaim, error) {
+	return nil, nil
+}
+
+func (r *resourceClaimTrackerContract) Get(_, _ string) (*resourceapi.ResourceClaim, error) {
+	return nil, nil
+}
+
+func (r *resourceClaimTrackerContract) ListAllAllocatedDevices() (sets.Set[structured.DeviceID], error) {
+	return nil, nil
+}
+
+func (r *resourceClaimTrackerContract) SignalClaimPendingAllocation(_ types.UID, _ *resourceapi.ResourceClaim) error {
+	return nil
+}
+
+func (r *resourceClaimTrackerContract) ClaimHasPendingAllocation(_ types.UID) bool {
+	return false
+}
+
+func (r *resourceClaimTrackerContract) RemoveClaimPendingAllocation(_ types.UID) (deleted bool) {
+	return false
+}
+
+func (r *resourceClaimTrackerContract) AssumeClaimAfterAPICall(_ *resourceapi.ResourceClaim) error {
+	return nil
+}
+
+func (r *resourceClaimTrackerContract) AssumedClaimRestore(_, _ string) {
+}
+
+type sharedDRAManagerContract struct{}
+
+func (s *sharedDRAManagerContract) ResourceClaims() framework.ResourceClaimTracker {
+	return nil
+}
+
+func (s *sharedDRAManagerContract) ResourceSlices() framework.ResourceSliceLister {
+	return nil
+}
+
+func (s *sharedDRAManagerContract) DeviceClasses() framework.DeviceClassLister {
+	return nil
+}

pkg/scheduler/framework/interface.go

@@ -39,7 +39,6 @@ import (
 	"k8s.io/klog/v2"
 	"k8s.io/kubernetes/pkg/scheduler/apis/config"
 	"k8s.io/kubernetes/pkg/scheduler/framework/parallelize"
-	"k8s.io/kubernetes/pkg/scheduler/util/assumecache"
 )
 
 // NodeScoreList declares a list of nodes and their scores.
@@ -820,10 +819,9 @@ type Handle interface {
 
 	SharedInformerFactory() informers.SharedInformerFactory
 
-	// ResourceClaimCache returns an assume cache of ResourceClaim objects
-	// which gets populated by the shared informer factory and the dynamic resources
-	// plugin.
-	ResourceClaimCache() *assumecache.AssumeCache
+	// SharedDRAManager can be used to obtain DRA objects, and track modifications to them in-memory - mainly by the DRA plugin.
+	// A non-default implementation can be plugged into the framework to simulate the state of DRA objects.
+	SharedDRAManager() SharedDRAManager
 
 	// RunFilterPluginsWithNominatedPods runs the set of configured filter plugins for nominated pod on the given node.
 	RunFilterPluginsWithNominatedPods(ctx context.Context, state *CycleState, pod *v1.Pod, info *NodeInfo) *Status

pkg/scheduler/framework/listers.go

@@ -16,6 +16,13 @@ limitations under the License.
 
 package framework
 
+import (
+	resourceapi "k8s.io/api/resource/v1alpha3"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/dynamic-resource-allocation/structured"
+)
+
 // NodeInfoLister interface represents anything that can list/get NodeInfo objects from node name.
 type NodeInfoLister interface {
 	// List returns the list of NodeInfos.
@@ -40,3 +47,65 @@ type SharedLister interface {
 	NodeInfos() NodeInfoLister
 	StorageInfos() StorageInfoLister
 }
+
+// ResourceSliceLister can be used to obtain ResourceSlices.
+type ResourceSliceLister interface {
+	// List returns a list of all ResourceSlices.
+	List() ([]*resourceapi.ResourceSlice, error)
+}
+
+// DeviceClassLister can be used to obtain DeviceClasses.
+type DeviceClassLister interface {
+	// List returns a list of all DeviceClasses.
+	List() ([]*resourceapi.DeviceClass, error)
+	// Get returns the DeviceClass with the given className.
+	Get(className string) (*resourceapi.DeviceClass, error)
+}
+
+// ResourceClaimTracker can be used to obtain ResourceClaims, and track changes to ResourceClaims in-memory.
+//
+// If the claims are meant to be allocated in the API during the binding phase (when used by scheduler), the tracker helps avoid
+// race conditions between scheduling and binding phases (as well as between the binding phase and the informer cache update).
+//
+// If the binding phase is not run (e.g. when used by Cluster Autoscaler which only runs the scheduling phase, and simulates binding in-memory),
+// the tracker allows the framework user to obtain the claim allocations produced by the DRA plugin, and persist them outside of the API (e.g. in-memory).
+type ResourceClaimTracker interface {
+	// List lists ResourceClaims. The result is guaranteed to immediately include any changes made via AssumeClaimAfterAPICall(),
+	// and SignalClaimPendingAllocation().
+	List() ([]*resourceapi.ResourceClaim, error)
+	// Get works like List(), but for a single claim.
+	Get(namespace, claimName string) (*resourceapi.ResourceClaim, error)
+	// ListAllAllocatedDevices lists all allocated Devices from allocated ResourceClaims. The result is guaranteed to immediately include
+	// any changes made via AssumeClaimAfterAPICall(), and SignalClaimPendingAllocation().
+	ListAllAllocatedDevices() (sets.Set[structured.DeviceID], error)
+
+	// SignalClaimPendingAllocation signals to the tracker that the given ResourceClaim will be allocated via an API call in the
+	// binding phase. This change is immediately reflected in the result of List() and the other accessors.
+	SignalClaimPendingAllocation(claimUID types.UID, allocatedClaim *resourceapi.ResourceClaim) error
+	// ClaimHasPendingAllocation answers whether a given claim has a pending allocation during the binding phase. It can be used to avoid
+	// race conditions in subsequent scheduling phases.
+	ClaimHasPendingAllocation(claimUID types.UID) bool
+	// RemoveClaimPendingAllocation removes the pending allocation for the given ResourceClaim from the tracker if any was signaled via
+	// SignalClaimPendingAllocation(). Returns whether there was a pending allocation to remove. List() and the other accessors immediately
+	// stop reflecting the pending allocation in the results.
+	RemoveClaimPendingAllocation(claimUID types.UID) (deleted bool)
+
+	// AssumeClaimAfterAPICall signals to the tracker that an API call modifying the given ResourceClaim was made in the binding phase, and the
+	// changes should be reflected in informers very soon. This change is immediately reflected in the result of List() and the other accessors.
+	// This mechanism can be used to avoid race conditions between the informer update and subsequent scheduling phases.
+	AssumeClaimAfterAPICall(claim *resourceapi.ResourceClaim) error
+	// AssumedClaimRestore signals to the tracker that something went wrong with the API call modifying the given ResourceClaim, and
+	// the changes won't be reflected in informers after all. List() and the other accessors immediately stop reflecting the assumed change,
+	// and go back to the informer version.
+	AssumedClaimRestore(namespace, claimName string)
+}
+
+// SharedDRAManager can be used to obtain DRA objects, and track modifications to them in-memory - mainly by the DRA plugin.
+// The plugin's default implementation obtains the objects from the API. A different implementation can be
+// plugged into the framework in order to simulate the state of DRA objects. For example, Cluster Autoscaler
+// can use this to provide the correct DRA object state to the DRA plugin when simulating scheduling changes in-memory.
+type SharedDRAManager interface {
+	ResourceClaims() ResourceClaimTracker
+	ResourceSlices() ResourceSliceLister
+	DeviceClasses() DeviceClassLister
+}