adapt to upstream interface changes

- rename scheduler v1alpha1 package - rename master package - podutil -> corev1helpers - FrameworkHandle -> Handle
togashidm · Apr 6, 2021 · 751944f · 751944f
1 parent e8e5771
commit 751944f
Show file tree

Hide file tree

Showing 31 changed files with 85 additions and 94 deletions.
diff --git a/cmd/scheduler/main_test.go b/cmd/scheduler/main_test.go
@@ -404,12 +404,12 @@ profiles:
 		},
 		"FilterPlugin": {
 			{Name: "NodeUnschedulable"},
-			{Name: "NodeResourcesFit"},
 			{Name: "NodeName"},
-			{Name: "NodePorts"},
+			{Name: "TaintToleration"},
 			{Name: "NodeAffinity"},
+			{Name: "NodePorts"},
+			{Name: "NodeResourcesFit"},
 			{Name: "VolumeRestrictions"},
-			{Name: "TaintToleration"},
 			{Name: "EBSLimits"},
 			{Name: "GCEPDLimits"},
 			{Name: "NodeVolumeLimits"},
@@ -426,7 +426,6 @@ profiles:
 			{Name: "InterPodAffinity"},
 			{Name: "PodTopologySpread"},
 			{Name: "TaintToleration"},
-			{Name: "SelectorSpread"},
 		},
 		"ScorePlugin": {
 			{Name: "NodeResourcesBalancedAllocation", Weight: 1},
@@ -437,7 +436,6 @@ profiles:
 			{Name: "NodePreferAvoidPods", Weight: 10000},
 			{Name: "PodTopologySpread", Weight: 2},
 			{Name: "TaintToleration", Weight: 1},
-			{Name: "SelectorSpread", Weight: 1},
 		},
 		"BindPlugin":    {{Name: "DefaultBinder"}},
 		"ReservePlugin": {{Name: "VolumeBinding"}},

diff --git a/kep/2-lightweight-coscheduling/README.md b/kep/2-lightweight-coscheduling/README.md
@@ -69,7 +69,7 @@ labels:
 ```go
 // Coscheduling is a plugin that implements the mechanism of gang scheduling.
 type Coscheduling struct {
-    FrameworkHandle         framework.FrameworkHandle
+    FrameworkHandle         framework.Handle
     PodLister               corelisters.PodLister
     // Key is the name of PodGroup.
     PodGroupInfos           map[string]PodGroupInfo

diff --git a/pkg/capacityscheduling/capacity_scheduling.go b/pkg/capacityscheduling/capacity_scheduling.go
@@ -31,16 +31,17 @@ import (
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes"
+	corelisters "k8s.io/client-go/listers/core/v1"
 	policylisters "k8s.io/client-go/listers/policy/v1beta1"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/clientcmd"
+	corev1helpers "k8s.io/component-helpers/scheduling/corev1"
 	"k8s.io/klog/v2"
 	extenderv1 "k8s.io/kube-scheduler/extender/v1"
-	podutil "k8s.io/kubernetes/pkg/api/v1/pod"
 	kubefeatures "k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/scheduler/core"
+	"k8s.io/kubernetes/pkg/scheduler/framework"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/defaultpreemption"
-	framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
 	"k8s.io/kubernetes/pkg/scheduler/util"
 
 	"sigs.k8s.io/scheduler-plugins/pkg/apis/config"
@@ -54,7 +55,8 @@ import (
 // CapacityScheduling is a plugin that implements the mechanism of capacity scheduling.
 type CapacityScheduling struct {
 	sync.RWMutex
-	frameworkHandle    framework.FrameworkHandle
+	frameworkHandle    framework.Handle
+	podLister          corelisters.PodLister
 	pdbLister          policylisters.PodDisruptionBudgetLister
 	elasticQuotaLister externalv1alpha1.ElasticQuotaLister
 	elasticQuotaInfos  ElasticQuotaInfos
@@ -101,7 +103,7 @@ func (c *CapacityScheduling) Name() string {
 }
 
 // New initializes a new plugin and returns it.
-func New(obj runtime.Object, handle framework.FrameworkHandle) (framework.Plugin, error) {
+func New(obj runtime.Object, handle framework.Handle) (framework.Plugin, error) {
 	args, ok := obj.(*config.CapacitySchedulingArgs)
 	if !ok {
 		return nil, fmt.Errorf("want args to be of type CapacitySchedulingArgs, got %T", obj)
@@ -111,6 +113,7 @@ func New(obj runtime.Object, handle framework.FrameworkHandle) (framework.Plugin
 	c := &CapacityScheduling{
 		frameworkHandle:   handle,
 		elasticQuotaInfos: NewElasticQuotaInfos(),
+		podLister:         handle.SharedInformerFactory().Core().V1().Pods().Lister(),
 		pdbLister:         getPDBLister(handle.SharedInformerFactory()),
 	}
 
@@ -297,9 +300,11 @@ func (c *CapacityScheduling) preempt(ctx context.Context, state *framework.Cycle
 	ph := c.frameworkHandle.PreemptHandle()
 	nodeLister := c.frameworkHandle.SnapshotSharedLister().NodeInfos()
 
-	// 0) Fetch the latest version of <pod>.
-	// TODO(Huang-Wei): get pod from informer cache instead of API server.
-	pod, err := util.GetUpdatedPod(client, pod)
+	// Fetch the latest version of <pod>.
+	// It's safe to directly fetch pod here. Because the informer cache has already been
+	// initialized when creating the Scheduler obj, i.e., factory.go#MakeDefaultErrorFunc().
+	// However, tests may need to manually initialize the shared pod informer.
+	pod, err := c.podLister.Pods(pod.Namespace).Get(pod.Name)
 	if err != nil {
 		klog.Errorf("Error getting the updated preemptor pod object: %v", err)
 		return "", err
@@ -460,7 +465,7 @@ func selectVictimsOnNode(
 	}
 
 	elasticQuotaInfos := elasticQuotaSnapshotState.elasticQuotaInfos
-	podPriority := podutil.GetPodPriority(pod)
+	podPriority := corev1helpers.PodPriority(pod)
 	preemptorElasticQuotaInfo, preemptorWithElasticQuota := elasticQuotaInfos[pod.Namespace]
 
 	var moreThanMinWithPreemptor bool
@@ -486,7 +491,7 @@ func selectVictimsOnNode(
 				// quotas. So that we will select the pods which subject to the
 				// same quota(namespace) with the lower priority than the
 				// preemptor's priority as potential victims in a node.
-				if p.Pod.Namespace == pod.Namespace && podutil.GetPodPriority(p.Pod) < podPriority {
+				if p.Pod.Namespace == pod.Namespace && corev1helpers.PodPriority(p.Pod) < podPriority {
 					potentialVictims = append(potentialVictims, p.Pod)
 					if err := removePod(p.Pod); err != nil {
 						return nil, 0, false
@@ -514,7 +519,7 @@ func selectVictimsOnNode(
 			if pWithElasticQuota {
 				continue
 			}
-			if podutil.GetPodPriority(p.Pod) < podPriority {
+			if corev1helpers.PodPriority(p.Pod) < podPriority {
 				potentialVictims = append(potentialVictims, p.Pod)
 				if err := removePod(p.Pod); err != nil {
 					return nil, 0, false

diff --git a/pkg/capacityscheduling/capacity_scheduling_test.go b/pkg/capacityscheduling/capacity_scheduling_test.go
@@ -29,12 +29,12 @@ import (
 	clientsetfake "k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/tools/events"
 	extenderv1 "k8s.io/kube-scheduler/extender/v1"
+	"k8s.io/kubernetes/pkg/scheduler/framework"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/defaultbinder"
 	dp "k8s.io/kubernetes/pkg/scheduler/framework/plugins/defaultpreemption"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/noderesources"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/queuesort"
 	frameworkruntime "k8s.io/kubernetes/pkg/scheduler/framework/runtime"
-	framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
 	st "k8s.io/kubernetes/pkg/scheduler/testing"
 	imageutils "k8s.io/kubernetes/test/utils/image"
 

diff --git a/pkg/capacityscheduling/elasticquota.go b/pkg/capacityscheduling/elasticquota.go
@@ -19,7 +19,7 @@ package capacityscheduling
 import (
 	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
-	framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
+	"k8s.io/kubernetes/pkg/scheduler/framework"
 )
 
 type ElasticQuotaInfos map[string]*ElasticQuotaInfo

diff --git a/pkg/capacityscheduling/elasticquota_test.go b/pkg/capacityscheduling/elasticquota_test.go
@@ -21,7 +21,7 @@ import (
 	"testing"
 
 	"k8s.io/api/core/v1"
-	framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
+	"k8s.io/kubernetes/pkg/scheduler/framework"
 )
 
 func TestReserveResource(t *testing.T) {

diff --git a/pkg/coscheduling/core/core.go b/pkg/coscheduling/core/core.go
@@ -31,7 +31,7 @@ import (
 	informerv1 "k8s.io/client-go/informers/core/v1"
 	listerv1 "k8s.io/client-go/listers/core/v1"
 	"k8s.io/klog/v2"
-	framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
+	"k8s.io/kubernetes/pkg/scheduler/framework"
 
 	"sigs.k8s.io/scheduler-plugins/pkg/apis/scheduling/v1alpha1"
 	pgclientset "sigs.k8s.io/scheduler-plugins/pkg/generated/clientset/versioned"

diff --git a/pkg/coscheduling/core/core_test.go b/pkg/coscheduling/core/core_test.go
@@ -29,7 +29,7 @@ import (
 	"k8s.io/client-go/informers"
 	clientsetfake "k8s.io/client-go/kubernetes/fake"
 	clicache "k8s.io/client-go/tools/cache"
-	framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
+	"k8s.io/kubernetes/pkg/scheduler/framework"
 	st "k8s.io/kubernetes/pkg/scheduler/testing"
 
 	"sigs.k8s.io/scheduler-plugins/pkg/apis/scheduling/v1alpha1"

diff --git a/pkg/coscheduling/coscheduling.go b/pkg/coscheduling/coscheduling.go
@@ -29,9 +29,9 @@ import (
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/clientcmd"
+	corev1helpers "k8s.io/component-helpers/scheduling/corev1"
 	"k8s.io/klog/v2"
-	podutil "k8s.io/kubernetes/pkg/api/v1/pod"
-	framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
+	"k8s.io/kubernetes/pkg/scheduler/framework"
 
 	"sigs.k8s.io/scheduler-plugins/pkg/apis/config"
 	"sigs.k8s.io/scheduler-plugins/pkg/coscheduling/core"
@@ -42,7 +42,7 @@ import (
 
 // Coscheduling is a plugin that schedules pods in a group.
 type Coscheduling struct {
-	frameworkHandler framework.FrameworkHandle
+	frameworkHandler framework.Handle
 	pgMgr            core.Manager
 	scheduleTimeout  *time.Duration
 }
@@ -60,7 +60,7 @@ const (
 )
 
 // New initializes and returns a new Coscheduling plugin.
-func New(obj runtime.Object, handle framework.FrameworkHandle) (framework.Plugin, error) {
+func New(obj runtime.Object, handle framework.Handle) (framework.Plugin, error) {
 	args, ok := obj.(*config.CoschedulingArgs)
 	if !ok {
 		return nil, fmt.Errorf("want args to be of type CoschedulingArgs, got %T", obj)
@@ -114,8 +114,8 @@ func (cs *Coscheduling) Name() string {
 // 2. Compare the initialization timestamps of PodGroups or Pods.
 // 3. Compare the keys of PodGroups/Pods: <namespace>/<podname>.
 func (cs *Coscheduling) Less(podInfo1, podInfo2 *framework.QueuedPodInfo) bool {
-	prio1 := podutil.GetPodPriority(podInfo1.Pod)
-	prio2 := podutil.GetPodPriority(podInfo2.Pod)
+	prio1 := corev1helpers.PodPriority(podInfo1.Pod)
+	prio2 := corev1helpers.PodPriority(podInfo2.Pod)
 	if prio1 != prio2 {
 		return prio1 > prio2
 	}

diff --git a/pkg/coscheduling/coscheduling_test.go b/pkg/coscheduling/coscheduling_test.go
@@ -27,7 +27,7 @@ import (
 	"k8s.io/client-go/kubernetes"
 	clientsetfake "k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/tools/events"
-	framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
+	"k8s.io/kubernetes/pkg/scheduler/framework"
 	st "k8s.io/kubernetes/pkg/scheduler/testing"
 
 	_ "sigs.k8s.io/scheduler-plugins/pkg/apis/config/scheme"
@@ -403,7 +403,7 @@ func TestPostFilter(t *testing.T) {
 type fakeHandler struct {
 }
 
-var _ framework.FrameworkHandle = &fakeHandler{}
+var _ framework.Handle = &fakeHandler{}
 
 func (f fakeHandler) SnapshotSharedLister() framework.SharedLister {
 	return nil

diff --git a/pkg/crossnodepreemption/cross_node_preemption.go b/pkg/crossnodepreemption/cross_node_preemption.go
@@ -21,12 +21,12 @@ import (
 
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	corelisters "k8s.io/client-go/listers/core/v1"
+	corev1helpers "k8s.io/component-helpers/scheduling/corev1"
 	"k8s.io/klog/v2"
-	podutil "k8s.io/kubernetes/pkg/api/v1/pod"
 	"k8s.io/kubernetes/pkg/scheduler/core"
+	"k8s.io/kubernetes/pkg/scheduler/framework"
 	dp "k8s.io/kubernetes/pkg/scheduler/framework/plugins/defaultpreemption"
-	framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
-	"k8s.io/kubernetes/pkg/scheduler/util"
 )
 
 const (
@@ -36,7 +36,8 @@ const (
 
 // CrossNodePreemption is a PostFilter plugin implements the preemption logic.
 type CrossNodePreemption struct {
-	fh framework.FrameworkHandle
+	fh        framework.Handle
+	podLister corelisters.PodLister
 }
 
 var _ framework.PostFilterPlugin = &CrossNodePreemption{}
@@ -47,9 +48,10 @@ func (pl *CrossNodePreemption) Name() string {
 }
 
 // New initializes a new plugin and returns it.
-func New(_ runtime.Object, fh framework.FrameworkHandle) (framework.Plugin, error) {
+func New(_ runtime.Object, fh framework.Handle) (framework.Plugin, error) {
 	pl := CrossNodePreemption{
-		fh: fh,
+		fh:        fh,
+		podLister: fh.SharedInformerFactory().Core().V1().Pods().Lister(),
 	}
 	return &pl, nil
 }
@@ -71,9 +73,11 @@ func (pl *CrossNodePreemption) preempt(ctx context.Context, state *framework.Cyc
 	ph := pl.fh.PreemptHandle()
 	nodeLister := pl.fh.SnapshotSharedLister().NodeInfos()
 
-	// 0) Fetch the latest version of <pod>.
-	// TODO(Huang-Wei): get pod from informer cache instead of API server.
-	pod, err := util.GetUpdatedPod(cs, pod)
+	// Fetch the latest version of <pod>.
+	// It's safe to directly fetch pod here. Because the informer cache has already been
+	// initialized when creating the Scheduler obj, i.e., factory.go#MakeDefaultErrorFunc().
+	// However, tests may need to manually initialize the shared pod informer.
+	pod, err := pl.podLister.Pods(pod.Namespace).Get(pod.Name)
 	if err != nil {
 		klog.Errorf("Error getting the updated preemptor pod object: %v", err)
 		return "", err
@@ -133,12 +137,12 @@ func FindCandidates(ctx context.Context, state *framework.CycleState, pod *v1.Po
 func bruteForceDryRunPreemption(ctx context.Context, ph framework.PreemptHandle, state *framework.CycleState,
 	pod *v1.Pod, potentialNodes []*framework.NodeInfo, nodeLister framework.NodeInfoLister) []dp.Candidate {
 	// Loop over <potentialNodes> and collect the pods that has lower priority than <pod>.
-	priority := podutil.GetPodPriority(pod)
+	priority := corev1helpers.PodPriority(pod)
 	var pods []*v1.Pod
 	for _, node := range potentialNodes {
 		for i := range node.Pods {
 			p := node.Pods[i].Pod
-			if podutil.GetPodPriority(p) < priority {
+			if corev1helpers.PodPriority(p) < priority {
 				pods = append(pods, p)
 			}
 		}

diff --git a/pkg/crossnodepreemption/cross_node_preemption_test.go b/pkg/crossnodepreemption/cross_node_preemption_test.go
@@ -26,14 +26,14 @@ import (
 	"k8s.io/client-go/informers"
 	clientsetfake "k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/tools/events"
+	"k8s.io/kubernetes/pkg/scheduler/framework"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/defaultbinder"
 	dp "k8s.io/kubernetes/pkg/scheduler/framework/plugins/defaultpreemption"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/interpodaffinity"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/noderesources"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/podtopologyspread"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/queuesort"
 	frameworkruntime "k8s.io/kubernetes/pkg/scheduler/framework/runtime"
-	framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
 	st "k8s.io/kubernetes/pkg/scheduler/testing"
 
 	testutil "sigs.k8s.io/scheduler-plugins/test/util"

diff --git a/pkg/noderesources/allocatable.go b/pkg/noderesources/allocatable.go
@@ -25,15 +25,15 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/klog/v2"
 	schedulerconfig "k8s.io/kube-scheduler/config/v1"
-	framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
+	"k8s.io/kubernetes/pkg/scheduler/framework"
 
 	"sigs.k8s.io/scheduler-plugins/pkg/apis/config"
 )
 
 // Allocatable is a score plugin that favors nodes based on their allocatable
 // resources.
 type Allocatable struct {
-	handle framework.FrameworkHandle
+	handle framework.Handle
 	resourceAllocationScorer
 }
 
@@ -77,7 +77,7 @@ func (alloc *Allocatable) ScoreExtensions() framework.ScoreExtensions {
 }
 
 // NewAllocatable initializes a new plugin and returns it.
-func NewAllocatable(allocArgs runtime.Object, h framework.FrameworkHandle) (framework.Plugin, error) {
+func NewAllocatable(allocArgs runtime.Object, h framework.Handle) (framework.Plugin, error) {
 	// Start with default values.
 	mode := config.Least
 	resToWeightMap := defaultResourcesToWeightMap

diff --git a/pkg/noderesources/allocatable_test.go b/pkg/noderesources/allocatable_test.go
@@ -27,11 +27,11 @@ import (
 	"k8s.io/client-go/informers"
 	clientsetfake "k8s.io/client-go/kubernetes/fake"
 	schedulerconfig "k8s.io/kube-scheduler/config/v1"
+	"k8s.io/kubernetes/pkg/scheduler/framework"
+	fakeframework "k8s.io/kubernetes/pkg/scheduler/framework/fake"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/defaultbinder"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/queuesort"
 	frameworkruntime "k8s.io/kubernetes/pkg/scheduler/framework/runtime"
-	framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
-	fakeframework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1/fake"
 	st "k8s.io/kubernetes/pkg/scheduler/testing"
 
 	"sigs.k8s.io/scheduler-plugins/pkg/apis/config"

diff --git a/pkg/noderesources/resource_allocation.go b/pkg/noderesources/resource_allocation.go
@@ -23,9 +23,8 @@ import (
 	v1 "k8s.io/api/core/v1"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/klog/v2"
-	v1helper "k8s.io/kubernetes/pkg/apis/core/v1/helper"
 	"k8s.io/kubernetes/pkg/features"
-	framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
+	"k8s.io/kubernetes/pkg/scheduler/framework"
 	schedutil "k8s.io/kubernetes/pkg/scheduler/util"
 )
 
@@ -106,7 +105,7 @@ func calculateResourceAllocatableRequest(nodeInfo *framework.NodeInfo, pod *v1.P
 	case v1.ResourceEphemeralStorage:
 		return nodeInfo.Allocatable.EphemeralStorage, (nodeInfo.Requested.EphemeralStorage + podRequest)
 	default:
-		if v1helper.IsScalarResourceName(resource) {
+		if schedutil.IsScalarResourceName(resource) {
 			return nodeInfo.Allocatable.ScalarResources[resource], (nodeInfo.Requested.ScalarResources[resource] + podRequest)
 		}
 	}