Descheduling profile with PoC fake plugin (#1093)

* Descheduling profile * Fake plugin + profile unit testing * Rename Profile config type into DeschedulerProfile To avoid resamblance with profileImpl * First run deschedule, then balance extension points
2026-01-26 13:29:11 +01:00 · 2023-03-23 17:14:33 +01:00
parent 0872b214ff
commit bcc6c8eb2a
18 changed files with 1088 additions and 301 deletions
--- a/pkg/framework/profile/profile.go
+++ b/pkg/framework/profile/profile.go
@@ -0,0 +1,310 @@
+/*
+Copyright 2023 The Kubernetes Authors.
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package profile
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"sigs.k8s.io/descheduler/metrics"
+	"sigs.k8s.io/descheduler/pkg/api"
+	"sigs.k8s.io/descheduler/pkg/descheduler/evictions"
+	podutil "sigs.k8s.io/descheduler/pkg/descheduler/pod"
+	"sigs.k8s.io/descheduler/pkg/framework"
+	"sigs.k8s.io/descheduler/pkg/framework/pluginregistry"
+	"sigs.k8s.io/descheduler/pkg/framework/plugins/defaultevictor"
+
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/errors"
+	"k8s.io/client-go/informers"
+	clientset "k8s.io/client-go/kubernetes"
+
+	"k8s.io/klog/v2"
+)
+
+// evictorImpl implements the Evictor interface so plugins
+// can evict a pod without importing a specific pod evictor
+type evictorImpl struct {
+	podEvictor    *evictions.PodEvictor
+	evictorFilter framework.EvictorPlugin
+}
+
+var _ framework.Evictor = &evictorImpl{}
+
+// Filter checks if a pod can be evicted
+func (ei *evictorImpl) Filter(pod *v1.Pod) bool {
+	return ei.evictorFilter.Filter(pod)
+}
+
+// PreEvictionFilter checks if pod can be evicted right before eviction
+func (ei *evictorImpl) PreEvictionFilter(pod *v1.Pod) bool {
+	return ei.evictorFilter.PreEvictionFilter(pod)
+}
+
+// Evict evicts a pod (no pre-check performed)
+func (ei *evictorImpl) Evict(ctx context.Context, pod *v1.Pod, opts evictions.EvictOptions) bool {
+	return ei.podEvictor.EvictPod(ctx, pod, opts)
+}
+
+func (ei *evictorImpl) NodeLimitExceeded(node *v1.Node) bool {
+	return ei.podEvictor.NodeLimitExceeded(node)
+}
+
+// handleImpl implements the framework handle which gets passed to plugins
+type handleImpl struct {
+	clientSet                 clientset.Interface
+	getPodsAssignedToNodeFunc podutil.GetPodsAssignedToNodeFunc
+	sharedInformerFactory     informers.SharedInformerFactory
+	evictor                   *evictorImpl
+}
+
+var _ framework.Handle = &handleImpl{}
+
+// ClientSet retrieves kube client set
+func (hi *handleImpl) ClientSet() clientset.Interface {
+	return hi.clientSet
+}
+
+// GetPodsAssignedToNodeFunc retrieves GetPodsAssignedToNodeFunc implementation
+func (hi *handleImpl) GetPodsAssignedToNodeFunc() podutil.GetPodsAssignedToNodeFunc {
+	return hi.getPodsAssignedToNodeFunc
+}
+
+// SharedInformerFactory retrieves shared informer factory
+func (hi *handleImpl) SharedInformerFactory() informers.SharedInformerFactory {
+	return hi.sharedInformerFactory
+}
+
+// Evictor retrieves evictor so plugins can filter and evict pods
+func (hi *handleImpl) Evictor() framework.Evictor {
+	return hi.evictor
+}
+
+type profileImpl struct {
+	profileName string
+	podEvictor  *evictions.PodEvictor
+
+	deschedulePlugins []framework.DeschedulePlugin
+	balancePlugins    []framework.BalancePlugin
+}
+
+// Option for the handleImpl.
+type Option func(*handleImplOpts)
+
+type handleImplOpts struct {
+	clientSet                 clientset.Interface
+	sharedInformerFactory     informers.SharedInformerFactory
+	getPodsAssignedToNodeFunc podutil.GetPodsAssignedToNodeFunc
+	podEvictor                *evictions.PodEvictor
+}
+
+// WithClientSet sets clientSet for the scheduling frameworkImpl.
+func WithClientSet(clientSet clientset.Interface) Option {
+	return func(o *handleImplOpts) {
+		o.clientSet = clientSet
+	}
+}
+
+func WithSharedInformerFactory(sharedInformerFactory informers.SharedInformerFactory) Option {
+	return func(o *handleImplOpts) {
+		o.sharedInformerFactory = sharedInformerFactory
+	}
+}
+
+func WithPodEvictor(podEvictor *evictions.PodEvictor) Option {
+	return func(o *handleImplOpts) {
+		o.podEvictor = podEvictor
+	}
+}
+
+func WithGetPodsAssignedToNodeFnc(getPodsAssignedToNodeFunc podutil.GetPodsAssignedToNodeFunc) Option {
+	return func(o *handleImplOpts) {
+		o.getPodsAssignedToNodeFunc = getPodsAssignedToNodeFunc
+	}
+}
+
+func getPluginConfig(pluginName string, pluginConfigs []api.PluginConfig) (*api.PluginConfig, int) {
+	for idx, pluginConfig := range pluginConfigs {
+		if pluginConfig.Name == pluginName {
+			return &pluginConfig, idx
+		}
+	}
+	return nil, 0
+}
+
+func buildPlugin(config api.DeschedulerProfile, pluginName string, handle *handleImpl, reg pluginregistry.Registry) (framework.Plugin, error) {
+	pc, _ := getPluginConfig(pluginName, config.PluginConfigs)
+	if pc == nil {
+		klog.ErrorS(fmt.Errorf("unable to get plugin config"), "skipping plugin", "plugin", pluginName, "profile", config.Name)
+		return nil, fmt.Errorf("unable to find %q plugin config", pluginName)
+	}
+
+	registryPlugin, ok := reg[pluginName]
+	if !ok {
+		klog.ErrorS(fmt.Errorf("unable to find plugin in the pluginsMap"), "skipping plugin", "plugin", pluginName)
+		return nil, fmt.Errorf("unable to find %q plugin in the pluginsMap", pluginName)
+	}
+	pg, err := registryPlugin.PluginBuilder(pc.Args, handle)
+	if err != nil {
+		klog.ErrorS(err, "unable to initialize a plugin", "pluginName", pluginName)
+		return nil, fmt.Errorf("unable to initialize %q plugin: %v", pluginName, err)
+	}
+	return pg, nil
+}
+
+func NewProfile(config api.DeschedulerProfile, reg pluginregistry.Registry, opts ...Option) (*profileImpl, error) {
+	hOpts := &handleImplOpts{}
+	for _, optFnc := range opts {
+		optFnc(hOpts)
+	}
+
+	if hOpts.clientSet == nil {
+		return nil, fmt.Errorf("clientSet missing")
+	}
+
+	if hOpts.sharedInformerFactory == nil {
+		return nil, fmt.Errorf("sharedInformerFactory missing")
+	}
+
+	if hOpts.podEvictor == nil {
+		return nil, fmt.Errorf("podEvictor missing")
+	}
+
+	evictorPlugin, err := buildPlugin(config, defaultevictor.PluginName, &handleImpl{
+		clientSet:                 hOpts.clientSet,
+		getPodsAssignedToNodeFunc: hOpts.getPodsAssignedToNodeFunc,
+		sharedInformerFactory:     hOpts.sharedInformerFactory,
+	}, reg)
+	if err != nil {
+		return nil, fmt.Errorf("unable to build %v plugin: %v", defaultevictor.PluginName, err)
+	}
+	if evictorPlugin == nil {
+		return nil, fmt.Errorf("empty plugin build for %v plugin: %v", defaultevictor.PluginName, err)
+	}
+
+	handle := &handleImpl{
+		clientSet:                 hOpts.clientSet,
+		getPodsAssignedToNodeFunc: hOpts.getPodsAssignedToNodeFunc,
+		sharedInformerFactory:     hOpts.sharedInformerFactory,
+		evictor: &evictorImpl{
+			podEvictor:    hOpts.podEvictor,
+			evictorFilter: evictorPlugin.(framework.EvictorPlugin),
+		},
+	}
+
+	deschedulePlugins := []framework.DeschedulePlugin{}
+	balancePlugins := []framework.BalancePlugin{}
+
+	descheduleEnabled := make(map[string]struct{})
+	balanceEnabled := make(map[string]struct{})
+	for _, name := range config.Plugins.Deschedule.Enabled {
+		descheduleEnabled[name] = struct{}{}
+	}
+	for _, name := range config.Plugins.Balance.Enabled {
+		balanceEnabled[name] = struct{}{}
+	}
+
+	// Assuming only a list of enabled extension points.
+	// Later, when a default list of plugins and their extension points is established,
+	// compute the list of enabled extension points as (DefaultEnabled + Enabled - Disabled)
+	for _, plugin := range append(config.Plugins.Deschedule.Enabled, config.Plugins.Balance.Enabled...) {
+		pg, err := buildPlugin(config, plugin, handle, reg)
+		if err != nil {
+			return nil, fmt.Errorf("unable to build %v plugin: %v", plugin, err)
+		}
+		if pg != nil {
+			// pg can be of any of each type, or both
+
+			if _, exists := descheduleEnabled[plugin]; exists {
+				_, ok := pg.(framework.DeschedulePlugin)
+				if ok {
+					deschedulePlugins = append(deschedulePlugins, pg.(framework.DeschedulePlugin))
+				}
+			}
+
+			if _, exists := balanceEnabled[plugin]; exists {
+				_, ok := pg.(framework.BalancePlugin)
+				if ok {
+					balancePlugins = append(balancePlugins, pg.(framework.BalancePlugin))
+				}
+			}
+		}
+	}
+
+	return &profileImpl{
+		profileName:       config.Name,
+		podEvictor:        hOpts.podEvictor,
+		deschedulePlugins: deschedulePlugins,
+		balancePlugins:    balancePlugins,
+	}, nil
+}
+
+func (d profileImpl) RunDeschedulePlugins(ctx context.Context, nodes []*v1.Node) *framework.Status {
+	errs := []error{}
+	for _, pl := range d.deschedulePlugins {
+		evicted := d.podEvictor.TotalEvicted()
+		// TODO: strategyName should be accessible from within the strategy using a framework
+		// handle or function which the Evictor has access to. For migration/in-progress framework
+		// work, we are currently passing this via context. To be removed
+		// (See discussion thread https://github.com/kubernetes-sigs/descheduler/pull/885#discussion_r919962292)
+		strategyStart := time.Now()
+		childCtx := context.WithValue(ctx, "strategyName", pl.Name())
+		status := pl.Deschedule(childCtx, nodes)
+		metrics.DeschedulerStrategyDuration.With(map[string]string{"strategy": pl.Name(), "profile": d.profileName}).Observe(time.Since(strategyStart).Seconds())
+
+		if status != nil && status.Err != nil {
+			errs = append(errs, fmt.Errorf("plugin %q finished with error: %v", pl.Name(), status.Err))
+		}
+		klog.V(1).InfoS("Total number of pods evicted", "extension point", "Deschedule", "evictedPods", d.podEvictor.TotalEvicted()-evicted)
+	}
+
+	aggrErr := errors.NewAggregate(errs)
+	if aggrErr == nil {
+		return &framework.Status{}
+	}
+
+	return &framework.Status{
+		Err: fmt.Errorf("%v", aggrErr.Error()),
+	}
+}
+
+func (d profileImpl) RunBalancePlugins(ctx context.Context, nodes []*v1.Node) *framework.Status {
+	errs := []error{}
+	for _, pl := range d.balancePlugins {
+		evicted := d.podEvictor.TotalEvicted()
+		// TODO: strategyName should be accessible from within the strategy using a framework
+		// handle or function which the Evictor has access to. For migration/in-progress framework
+		// work, we are currently passing this via context. To be removed
+		// (See discussion thread https://github.com/kubernetes-sigs/descheduler/pull/885#discussion_r919962292)
+		strategyStart := time.Now()
+		childCtx := context.WithValue(ctx, "strategyName", pl.Name())
+		status := pl.Balance(childCtx, nodes)
+		metrics.DeschedulerStrategyDuration.With(map[string]string{"strategy": pl.Name(), "profile": d.profileName}).Observe(time.Since(strategyStart).Seconds())
+
+		if status != nil && status.Err != nil {
+			errs = append(errs, fmt.Errorf("plugin %q finished with error: %v", pl.Name(), status.Err))
+		}
+		klog.V(1).InfoS("Total number of pods evicted", "extension point", "Balance", "evictedPods", d.podEvictor.TotalEvicted()-evicted)
+	}
+
+	aggrErr := errors.NewAggregate(errs)
+	if aggrErr == nil {
+		return &framework.Status{}
+	}
+
+	return &framework.Status{
+		Err: fmt.Errorf("%v", aggrErr.Error()),
+	}
+}
--- a/pkg/framework/profile/profile_test.go
+++ b/pkg/framework/profile/profile_test.go
@@ -0,0 +1,283 @@
+package profile
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	v1 "k8s.io/api/core/v1"
+	policy "k8s.io/api/policy/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/informers"
+	fakeclientset "k8s.io/client-go/kubernetes/fake"
+	core "k8s.io/client-go/testing"
+
+	"sigs.k8s.io/descheduler/pkg/api"
+	"sigs.k8s.io/descheduler/pkg/descheduler/evictions"
+	podutil "sigs.k8s.io/descheduler/pkg/descheduler/pod"
+	"sigs.k8s.io/descheduler/pkg/framework"
+	fakeplugin "sigs.k8s.io/descheduler/pkg/framework/fake/plugin"
+	"sigs.k8s.io/descheduler/pkg/framework/pluginregistry"
+	"sigs.k8s.io/descheduler/pkg/framework/plugins/defaultevictor"
+	"sigs.k8s.io/descheduler/pkg/utils"
+	testutils "sigs.k8s.io/descheduler/test"
+)
+
+func TestProfileTopExtensionPoints(t *testing.T) {
+	tests := []struct {
+		name             string
+		config           api.DeschedulerProfile
+		extensionPoint   framework.ExtensionPoint
+		expectedEviction bool
+	}{
+		{
+			name: "profile with deschedule extension point enabled single eviction",
+			config: api.DeschedulerProfile{
+				Name: "strategy-test-profile-with-deschedule",
+				PluginConfigs: []api.PluginConfig{
+					{
+						Name: defaultevictor.PluginName,
+						Args: &defaultevictor.DefaultEvictorArgs{
+							PriorityThreshold: &api.PriorityThreshold{
+								Value: nil,
+							},
+						},
+					},
+					{
+						Name: "FakePlugin",
+						Args: &fakeplugin.FakePluginArgs{},
+					},
+				},
+				Plugins: api.Plugins{
+					Deschedule: api.PluginSet{
+						Enabled: []string{"FakePlugin"},
+					},
+					Evict: api.PluginSet{
+						Enabled: []string{defaultevictor.PluginName},
+					},
+				},
+			},
+			extensionPoint:   framework.DescheduleExtensionPoint,
+			expectedEviction: true,
+		},
+		{
+			name: "profile with balance extension point enabled single eviction",
+			config: api.DeschedulerProfile{
+				Name: "strategy-test-profile-with-balance",
+				PluginConfigs: []api.PluginConfig{
+					{
+						Name: defaultevictor.PluginName,
+						Args: &defaultevictor.DefaultEvictorArgs{
+							PriorityThreshold: &api.PriorityThreshold{
+								Value: nil,
+							},
+						},
+					},
+					{
+						Name: "FakePlugin",
+						Args: &fakeplugin.FakePluginArgs{},
+					},
+				},
+				Plugins: api.Plugins{
+					Balance: api.PluginSet{
+						Enabled: []string{"FakePlugin"},
+					},
+					Evict: api.PluginSet{
+						Enabled: []string{defaultevictor.PluginName},
+					},
+				},
+			},
+			extensionPoint:   framework.BalanceExtensionPoint,
+			expectedEviction: true,
+		},
+		{
+			name: "profile with deschedule extension point balance enabled no eviction",
+			config: api.DeschedulerProfile{
+				Name: "strategy-test-profile-with-deschedule",
+				PluginConfigs: []api.PluginConfig{
+					{
+						Name: defaultevictor.PluginName,
+						Args: &defaultevictor.DefaultEvictorArgs{
+							PriorityThreshold: &api.PriorityThreshold{
+								Value: nil,
+							},
+						},
+					},
+					{
+						Name: "FakePlugin",
+						Args: &fakeplugin.FakePluginArgs{},
+					},
+				},
+				Plugins: api.Plugins{
+					Balance: api.PluginSet{
+						Enabled: []string{"FakePlugin"},
+					},
+					Evict: api.PluginSet{
+						Enabled: []string{defaultevictor.PluginName},
+					},
+				},
+			},
+			extensionPoint:   framework.DescheduleExtensionPoint,
+			expectedEviction: false,
+		},
+		{
+			name: "profile with balance extension point deschedule enabled no eviction",
+			config: api.DeschedulerProfile{
+				Name: "strategy-test-profile-with-deschedule",
+				PluginConfigs: []api.PluginConfig{
+					{
+						Name: defaultevictor.PluginName,
+						Args: &defaultevictor.DefaultEvictorArgs{
+							PriorityThreshold: &api.PriorityThreshold{
+								Value: nil,
+							},
+						},
+					},
+					{
+						Name: "FakePlugin",
+						Args: &fakeplugin.FakePluginArgs{},
+					},
+				},
+				Plugins: api.Plugins{
+					Deschedule: api.PluginSet{
+						Enabled: []string{"FakePlugin"},
+					},
+					Evict: api.PluginSet{
+						Enabled: []string{defaultevictor.PluginName},
+					},
+				},
+			},
+			extensionPoint:   framework.BalanceExtensionPoint,
+			expectedEviction: false,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			ctx, cancel := context.WithCancel(context.TODO())
+			defer cancel()
+
+			n1 := testutils.BuildTestNode("n1", 2000, 3000, 10, nil)
+			n2 := testutils.BuildTestNode("n2", 2000, 3000, 10, nil)
+			nodes := []*v1.Node{n1, n2}
+
+			p1 := testutils.BuildTestPod(fmt.Sprintf("pod_1_%s", n1.Name), 200, 0, n1.Name, nil)
+			p1.ObjectMeta.OwnerReferences = []metav1.OwnerReference{{}}
+
+			fakePlugin := fakeplugin.FakePlugin{}
+			if test.extensionPoint == framework.DescheduleExtensionPoint {
+				fakePlugin.AddReactor(string(framework.DescheduleExtensionPoint), func(action fakeplugin.Action) (handled bool, err error) {
+					if dAction, ok := action.(fakeplugin.DescheduleAction); ok {
+						if dAction.Handle().Evictor().Evict(ctx, p1, evictions.EvictOptions{}) {
+							return true, nil
+						}
+						return true, fmt.Errorf("pod not evicted")
+					}
+					return false, nil
+				})
+			}
+			if test.extensionPoint == framework.BalanceExtensionPoint {
+				fakePlugin.AddReactor(string(framework.BalanceExtensionPoint), func(action fakeplugin.Action) (handled bool, err error) {
+					if dAction, ok := action.(fakeplugin.BalanceAction); ok {
+						if dAction.Handle().Evictor().Evict(ctx, p1, evictions.EvictOptions{}) {
+							return true, nil
+						}
+						return true, fmt.Errorf("pod not evicted")
+					}
+					return false, nil
+				})
+			}
+
+			pluginregistry.PluginRegistry = pluginregistry.NewRegistry()
+			pluginregistry.Register(
+				"FakePlugin",
+				fakeplugin.NewPluginFncFromFake(&fakePlugin),
+				&fakeplugin.FakePluginArgs{},
+				fakeplugin.ValidateFakePluginArgs,
+				fakeplugin.SetDefaults_FakePluginArgs,
+				pluginregistry.PluginRegistry,
+			)
+
+			pluginregistry.Register(
+				defaultevictor.PluginName,
+				defaultevictor.New,
+				&defaultevictor.DefaultEvictorArgs{},
+				defaultevictor.ValidateDefaultEvictorArgs,
+				defaultevictor.SetDefaults_DefaultEvictorArgs,
+				pluginregistry.PluginRegistry,
+			)
+
+			client := fakeclientset.NewSimpleClientset(n1, n2, p1)
+			var evictedPods []string
+			client.PrependReactor("create", "pods", podEvictionReactionFuc(&evictedPods))
+
+			sharedInformerFactory := informers.NewSharedInformerFactory(client, 0)
+			podInformer := sharedInformerFactory.Core().V1().Pods().Informer()
+			getPodsAssignedToNode, err := podutil.BuildGetPodsAssignedToNodeFunc(podInformer)
+			if err != nil {
+				t.Fatalf("build get pods assigned to node function error: %v", err)
+			}
+
+			sharedInformerFactory.Start(ctx.Done())
+			sharedInformerFactory.WaitForCacheSync(ctx.Done())
+
+			eventClient := fakeclientset.NewSimpleClientset(n1, n2, p1)
+			eventBroadcaster, eventRecorder := utils.GetRecorderAndBroadcaster(ctx, eventClient)
+			defer eventBroadcaster.Shutdown()
+
+			podEvictor := evictions.NewPodEvictor(client, "policy/v1", false, nil, nil, nodes, true, eventRecorder)
+
+			prfl, err := NewProfile(
+				test.config,
+				pluginregistry.PluginRegistry,
+				WithClientSet(client),
+				WithSharedInformerFactory(sharedInformerFactory),
+				WithPodEvictor(podEvictor),
+				WithGetPodsAssignedToNodeFnc(getPodsAssignedToNode),
+			)
+			if err != nil {
+				t.Fatalf("unable to create %q profile: %v", test.config.Name, err)
+			}
+
+			var status *framework.Status
+			switch test.extensionPoint {
+			case framework.DescheduleExtensionPoint:
+				status = prfl.RunDeschedulePlugins(ctx, nodes)
+			case framework.BalanceExtensionPoint:
+				status = prfl.RunBalancePlugins(ctx, nodes)
+			default:
+				t.Fatalf("unknown %q extension point", test.extensionPoint)
+			}
+
+			if status == nil {
+				t.Fatalf("Unexpected nil status")
+			}
+			if status.Err != nil {
+				t.Fatalf("Expected nil error in status, got %q instead", status.Err)
+			}
+
+			if test.expectedEviction && len(evictedPods) < 1 {
+				t.Fatalf("Expected eviction, got none")
+			}
+			if !test.expectedEviction && len(evictedPods) > 0 {
+				t.Fatalf("Unexpected eviction, expected none")
+			}
+		})
+	}
+}
+
+func podEvictionReactionFuc(evictedPods *[]string) func(action core.Action) (bool, runtime.Object, error) {
+	return func(action core.Action) (bool, runtime.Object, error) {
+		if action.GetSubresource() == "eviction" {
+			createAct, matched := action.(core.CreateActionImpl)
+			if !matched {
+				return false, nil, fmt.Errorf("unable to convert action to core.CreateActionImpl")
+			}
+			if eviction, matched := createAct.Object.(*policy.Eviction); matched {
+				*evictedPods = append(*evictedPods, eviction.GetName())
+			}
+		}
+		return false, nil, nil // fallback to the default reactor
+	}
+}