Skip to content
Permalink
Browse files

Move PodPriorityResolution e2e to integration

  • Loading branch information
damemi committed Jul 31, 2019
1 parent 001f2cd commit c6bacc73a3f86f15dcbc821a3c026b7972437b36
@@ -46,31 +46,31 @@ const (
// Register registers a plugin
func Register(plugins *admission.Plugins) {
plugins.Register(PluginName, func(config io.Reader) (admission.Interface, error) {
return newPlugin(), nil
return NewPlugin(), nil
})
}

// priorityPlugin is an implementation of admission.Interface.
type priorityPlugin struct {
// Plugin is an implementation of admission.Interface.
type Plugin struct {
*admission.Handler
client kubernetes.Interface
lister schedulingv1listers.PriorityClassLister
}

var _ admission.MutationInterface = &priorityPlugin{}
var _ admission.ValidationInterface = &priorityPlugin{}
var _ = genericadmissioninitializers.WantsExternalKubeInformerFactory(&priorityPlugin{})
var _ = genericadmissioninitializers.WantsExternalKubeClientSet(&priorityPlugin{})
var _ admission.MutationInterface = &Plugin{}
var _ admission.ValidationInterface = &Plugin{}
var _ = genericadmissioninitializers.WantsExternalKubeInformerFactory(&Plugin{})
var _ = genericadmissioninitializers.WantsExternalKubeClientSet(&Plugin{})

// NewPlugin creates a new priority admission plugin.
func newPlugin() *priorityPlugin {
return &priorityPlugin{
func NewPlugin() *Plugin {
return &Plugin{
Handler: admission.NewHandler(admission.Create, admission.Update, admission.Delete),
}
}

// ValidateInitialization implements the InitializationValidator interface.
func (p *priorityPlugin) ValidateInitialization() error {
func (p *Plugin) ValidateInitialization() error {
if p.client == nil {
return fmt.Errorf("%s requires a client", PluginName)
}
@@ -80,13 +80,13 @@ func (p *priorityPlugin) ValidateInitialization() error {
return nil
}

// SetInternalKubeClientSet implements the WantsInternalKubeClientSet interface.
func (p *priorityPlugin) SetExternalKubeClientSet(client kubernetes.Interface) {
// SetExternalKubeClientSet implements the WantsInternalKubeClientSet interface.
func (p *Plugin) SetExternalKubeClientSet(client kubernetes.Interface) {
p.client = client
}

// SetInternalKubeInformerFactory implements the WantsInternalKubeInformerFactory interface.
func (p *priorityPlugin) SetExternalKubeInformerFactory(f informers.SharedInformerFactory) {
// SetExternalKubeInformerFactory implements the WantsInternalKubeInformerFactory interface.
func (p *Plugin) SetExternalKubeInformerFactory(f informers.SharedInformerFactory) {
priorityInformer := f.Scheduling().V1().PriorityClasses()
p.lister = priorityInformer.Lister()
p.SetReadyFunc(priorityInformer.Informer().HasSynced)
@@ -99,7 +99,7 @@ var (

// Admit checks Pods and admits or rejects them. It also resolves the priority of pods based on their PriorityClass.
// Note that pod validation mechanism prevents update of a pod priority.
func (p *priorityPlugin) Admit(ctx context.Context, a admission.Attributes, o admission.ObjectInterfaces) error {
func (p *Plugin) Admit(ctx context.Context, a admission.Attributes, o admission.ObjectInterfaces) error {
operation := a.GetOperation()
// Ignore all calls to subresources
if len(a.GetSubresource()) != 0 {
@@ -119,7 +119,7 @@ func (p *priorityPlugin) Admit(ctx context.Context, a admission.Attributes, o ad
}

// Validate checks PriorityClasses and admits or rejects them.
func (p *priorityPlugin) Validate(ctx context.Context, a admission.Attributes, o admission.ObjectInterfaces) error {
func (p *Plugin) Validate(ctx context.Context, a admission.Attributes, o admission.ObjectInterfaces) error {
operation := a.GetOperation()
// Ignore all calls to subresources
if len(a.GetSubresource()) != 0 {
@@ -153,7 +153,7 @@ func priorityClassPermittedInNamespace(priorityClassName string, namespace strin
}

// admitPod makes sure a new pod does not set spec.Priority field. It also makes sure that the PriorityClassName exists if it is provided and resolves the pod priority from the PriorityClassName.
func (p *priorityPlugin) admitPod(a admission.Attributes) error {
func (p *Plugin) admitPod(a admission.Attributes) error {
operation := a.GetOperation()
pod, ok := a.GetObject().(*api.Pod)
if !ok {
@@ -226,7 +226,7 @@ func (p *priorityPlugin) admitPod(a admission.Attributes) error {
}

// validatePriorityClass ensures that the value field is not larger than the highest user definable priority. If the GlobalDefault is set, it ensures that there is no other PriorityClass whose GlobalDefault is set.
func (p *priorityPlugin) validatePriorityClass(a admission.Attributes) error {
func (p *Plugin) validatePriorityClass(a admission.Attributes) error {
operation := a.GetOperation()
pc, ok := a.GetObject().(*scheduling.PriorityClass)
if !ok {
@@ -248,7 +248,7 @@ func (p *priorityPlugin) validatePriorityClass(a admission.Attributes) error {
return nil
}

func (p *priorityPlugin) getDefaultPriorityClass() (*schedulingv1.PriorityClass, error) {
func (p *Plugin) getDefaultPriorityClass() (*schedulingv1.PriorityClass, error) {
list, err := p.lister.List(labels.Everything())
if err != nil {
return nil, err
@@ -266,7 +266,7 @@ func (p *priorityPlugin) getDefaultPriorityClass() (*schedulingv1.PriorityClass,
return defaultPC, nil
}

func (p *priorityPlugin) getDefaultPriority() (string, int32, *apiv1.PreemptionPolicy, error) {
func (p *Plugin) getDefaultPriority() (string, int32, *apiv1.PreemptionPolicy, error) {
dpc, err := p.getDefaultPriorityClass()
if err != nil {
return "", 0, nil, err
@@ -37,7 +37,7 @@ import (
"k8s.io/kubernetes/pkg/features"
)

func addPriorityClasses(ctrl *priorityPlugin, priorityClasses []*scheduling.PriorityClass) error {
func addPriorityClasses(ctrl *Plugin, priorityClasses []*scheduling.PriorityClass) error {
informerFactory := informers.NewSharedInformerFactory(nil, controller.NoResyncPeriodFunc())
ctrl.SetExternalKubeInformerFactory(informerFactory)
// First add the existing classes to the cache.
@@ -173,7 +173,7 @@ func TestPriorityClassAdmission(t *testing.T) {
for _, test := range tests {
klog.V(4).Infof("starting test %q", test.name)

ctrl := newPlugin()
ctrl := NewPlugin()
// Add existing priority classes.
if err := addPriorityClasses(ctrl, test.existingClasses); err != nil {
t.Errorf("Test %q: unable to add object to informer: %v", test.name, err)
@@ -274,7 +274,7 @@ func TestDefaultPriority(t *testing.T) {

for _, test := range tests {
klog.V(4).Infof("starting test %q", test.name)
ctrl := newPlugin()
ctrl := NewPlugin()
if err := addPriorityClasses(ctrl, test.classesBefore); err != nil {
t.Errorf("Test %q: unable to add object to informer: %v", test.name, err)
}
@@ -682,7 +682,7 @@ func TestPodAdmission(t *testing.T) {
for _, test := range tests {
klog.V(4).Infof("starting test %q", test.name)

ctrl := newPlugin()
ctrl := NewPlugin()
// Add existing priority classes.
if err := addPriorityClasses(ctrl, test.existingClasses); err != nil {
t.Errorf("Test %q: unable to add object to informer: %v", test.name, err)
@@ -221,43 +221,6 @@ var _ = SIGDescribe("SchedulerPreemption [Serial]", func() {
})
})

var _ = SIGDescribe("PodPriorityResolution [Serial]", func() {
var cs clientset.Interface
var ns string
f := framework.NewDefaultFramework("sched-pod-priority")

ginkgo.BeforeEach(func() {
cs = f.ClientSet
ns = f.Namespace.Name

err := framework.CheckTestingNSDeletedExcept(cs, ns)
framework.ExpectNoError(err)
})

// This test verifies that system critical priorities are created automatically and resolved properly.
ginkgo.It("validates critical system priorities are created and resolved", func() {
// Create pods that use system critical priorities and
ginkgo.By("Create pods that use critical system priorities.")
systemPriorityClasses := []string{
scheduling.SystemNodeCritical, scheduling.SystemClusterCritical,
}
for i, spc := range systemPriorityClasses {
pod := createPausePod(f, pausePodConfig{
Name: fmt.Sprintf("pod%d-%v", i, spc),
Namespace: metav1.NamespaceSystem,
PriorityClassName: spc,
})
defer func() {
// Clean-up the pod.
err := f.ClientSet.CoreV1().Pods(pod.Namespace).Delete(pod.Name, metav1.NewDeleteOptions(0))
framework.ExpectNoError(err)
}()
gomega.Expect(pod.Spec.Priority).NotTo(gomega.BeNil())
framework.Logf("Created pod: %v", pod.Name)
}
})
})

// construct a fakecpu so as to set it to status of Node object
// otherwise if we update CPU/Memory/etc, those values will be corrected back by kubelet
var fakecpu v1.ResourceName = "example.com/fakecpu"
@@ -24,6 +24,7 @@ go_test(
deps = [
"//pkg/api/legacyscheme:go_default_library",
"//pkg/api/v1/pod:go_default_library",
"//pkg/apis/scheduling:go_default_library",
"//pkg/controller/nodelifecycle:go_default_library",
"//pkg/features:go_default_library",
"//pkg/scheduler:go_default_library",
@@ -37,6 +38,7 @@ go_test(
"//pkg/scheduler/testing:go_default_library",
"//plugin/pkg/admission/podtolerationrestriction:go_default_library",
"//plugin/pkg/admission/podtolerationrestriction/apis/podtolerationrestriction:go_default_library",
"//plugin/pkg/admission/priority:go_default_library",
"//staging/src/k8s.io/api/core/v1:go_default_library",
"//staging/src/k8s.io/api/policy/v1beta1:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/api/errors:go_default_library",
@@ -27,12 +27,18 @@ import (
policy "k8s.io/api/policy/v1beta1"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime/schema"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/intstr"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/client-go/informers"
"k8s.io/client-go/kubernetes"
clientset "k8s.io/client-go/kubernetes"
restclient "k8s.io/client-go/rest"
podutil "k8s.io/kubernetes/pkg/api/v1/pod"
"k8s.io/kubernetes/pkg/apis/scheduling"
_ "k8s.io/kubernetes/pkg/scheduler/algorithmprovider"
"k8s.io/kubernetes/plugin/pkg/admission/priority"
testutils "k8s.io/kubernetes/test/utils"

"k8s.io/klog"
@@ -366,6 +372,78 @@ func TestDisablePreemption(t *testing.T) {
}
}

func TestPodPriorityResolution(t *testing.T) {
admission := priority.NewPlugin()
context := initTestScheduler(t, initTestMaster(t, "preemption", admission), true, nil)
defer cleanupTest(t, context)
cs := context.clientSet

// Build clientset and informers for controllers.
externalClientset := kubernetes.NewForConfigOrDie(&restclient.Config{
QPS: -1,
Host: context.httpServer.URL,
ContentConfig: restclient.ContentConfig{GroupVersion: &schema.GroupVersion{Group: "", Version: "v1"}}})
externalInformers := informers.NewSharedInformerFactory(externalClientset, time.Second)
admission.SetExternalKubeClientSet(externalClientset)
admission.SetExternalKubeInformerFactory(externalInformers)
externalInformers.Start(context.stopCh)
externalInformers.WaitForCacheSync(context.stopCh)

tests := []struct {
PriorityClass string
Pod *v1.Pod
ExpectedPriority int32
}{
{
PriorityClass: scheduling.SystemNodeCritical,
ExpectedPriority: scheduling.SystemCriticalPriority + 1000,
Pod: initPausePod(cs, &pausePodConfig{
Name: fmt.Sprintf("pod1-%v", scheduling.SystemNodeCritical),
Namespace: metav1.NamespaceSystem,
PriorityClassName: scheduling.SystemNodeCritical,
}),
},
{
PriorityClass: scheduling.SystemClusterCritical,
ExpectedPriority: scheduling.SystemCriticalPriority,
Pod: initPausePod(cs, &pausePodConfig{
Name: fmt.Sprintf("pod2-%v", scheduling.SystemClusterCritical),
Namespace: metav1.NamespaceSystem,
PriorityClassName: scheduling.SystemClusterCritical,
}),
},
}

// Create a node with some resources and a label.
nodeRes := &v1.ResourceList{
v1.ResourcePods: *resource.NewQuantity(32, resource.DecimalSI),
v1.ResourceCPU: *resource.NewMilliQuantity(500, resource.DecimalSI),
v1.ResourceMemory: *resource.NewQuantity(500, resource.DecimalSI),
}
_, err := createNode(context.clientSet, "node1", nodeRes)
if err != nil {
t.Fatalf("Error creating nodes: %v", err)
}

pods := make([]*v1.Pod, len(tests))
for i, test := range tests {
pod, err := runPausePod(cs, test.Pod)
if err != nil {
t.Fatalf("Test [PodPriority/%v]: Error running pause pod: %v", test.PriorityClass, err)
}
pods[i] = pod
if pods[i].Spec.Priority != nil {
if *pods[i].Spec.Priority != test.ExpectedPriority {
t.Errorf("Expected pod %v to have priority %v but was %v", pods[i].Name, test.ExpectedPriority, pods[i].Spec.Priority)
}
} else {
t.Errorf("Expected pod %v to have priority %v but was nil", pods[i].Name, test.PriorityClass)
}
}
cleanupPods(cs, t, pods)
cleanupNodes(cs, t)
}

func mkPriorityPodWithGrace(tc *testContext, name string, priority int32, grace int64) *v1.Pod {
defaultPodRes := &v1.ResourceRequirements{Requests: v1.ResourceList{
v1.ResourceCPU: *resource.NewMilliQuantity(100, resource.DecimalSI),
@@ -462,6 +462,7 @@ type pausePodConfig struct {
NodeName string
SchedulerName string
Priority *int32
PriorityClassName string
}

// initPausePod initializes a pod API object from the given config. It is used
@@ -483,10 +484,11 @@ func initPausePod(cs clientset.Interface, conf *pausePodConfig) *v1.Pod {
Image: imageutils.GetPauseImageName(),
},
},
Tolerations: conf.Tolerations,
NodeName: conf.NodeName,
SchedulerName: conf.SchedulerName,
Priority: conf.Priority,
Tolerations: conf.Tolerations,
NodeName: conf.NodeName,
SchedulerName: conf.SchedulerName,
Priority: conf.Priority,
PriorityClassName: conf.PriorityClassName,
},
}
if conf.Resources != nil {

0 comments on commit c6bacc7

Please sign in to comment.
You can’t perform that action at this time.