github/descheduler
1
0
Fork 0
mirror of https://github.com/kubernetes-sigs/descheduler.git synced 2026-08-23 22:46:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(pkg/descheduler): move the main descheduler run entry under a dedicated function

Browse Source
This commit is contained in:
Jan Chaloupka
2026-02-03 23:31:29 +01:00
parent f26ab740fe
commit d1671034df
2 changed files with 85 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/descheduler/descheduler.go
+28 -19
View File
@@ -547,6 +547,8 @@ const (
secretReconciliation
)
type runFncType func(context.Context) error
func bootstrapDescheduler(
ctx context.Context,
rs *options.DeschedulerServer,
@@ -555,16 +557,16 @@ func bootstrapDescheduler(
metricProviderTokenReconciliation tokenReconciliation,
sharedInformerFactory, namespacedSharedInformerFactory informers.SharedInformerFactory,
eventRecorder events.EventRecorder,
) (*descheduler, error) {
) (*descheduler, runFncType, error) {
// Always create descheduler with real client/factory first to register all informers
descheduler, err := newDescheduler(ctx, rs, deschedulerPolicy, evictionPolicyGroupVersion, eventRecorder, rs.Client, sharedInformerFactory, nil)
if err != nil {
return nil, fmt.Errorf("failed to create new descheduler: %v", err)
return nil, nil, fmt.Errorf("failed to create new descheduler: %v", err)
}
// Setup Prometheus provider (only for real client case, not for dry run)
if err := setupPrometheusProvider(descheduler, namespacedSharedInformerFactory); err != nil {
return nil, fmt.Errorf("failed to setup Prometheus provider: %v", err)
return nil, nil, fmt.Errorf("failed to setup Prometheus provider: %v", err)
}
// If in dry run mode, replace the descheduler with one using fake client/factory
@@ -572,7 +574,7 @@ func bootstrapDescheduler(
// Create sandbox with resources to mirror from real client
kubeClientSandbox, err := newDefaultKubeClientSandbox(rs.Client, sharedInformerFactory)
if err != nil {
return nil, fmt.Errorf("failed to create kube client sandbox: %v", err)
return nil, nil, fmt.Errorf("failed to create kube client sandbox: %v", err)
}
klog.V(3).Infof("Building a cached client from the cluster for the dry run")
@@ -582,7 +584,7 @@ func bootstrapDescheduler(
// Replace descheduler with one using fake client/factory
descheduler, err = newDescheduler(ctx, rs, deschedulerPolicy, evictionPolicyGroupVersion, eventRecorder, kubeClientSandbox.fakeClient(), kubeClientSandbox.fakeSharedInformerFactory(), kubeClientSandbox)
if err != nil {
return nil, fmt.Errorf("failed to create dry run descheduler: %v", err)
return nil, nil, fmt.Errorf("failed to create dry run descheduler: %v", err)
}
}
@@ -630,10 +632,26 @@ func bootstrapDescheduler(
}
if err := deschedulerInitFnc(ctx); err != nil {
return nil, err
return nil, nil, err
}
return descheduler, nil
runFnc := func(ctx context.Context) error {
if metricProviderTokenReconciliation == inClusterReconciliation {
// Read the sa token and assume it has the sufficient permissions to authenticate
if err := descheduler.promClientCtrl.reconcileInClusterSAToken(); err != nil {
return fmt.Errorf("unable to reconcile an in cluster SA token: %v", err)
}
}
err = descheduler.runDeschedulerLoop(ctx)
if err != nil {
return fmt.Errorf("failed to run descheduler loop: %v", err)
}
return nil
}
return descheduler, runFnc, nil
}
func prometheusProviderToTokenReconciliation(prometheusProvider *api.MetricsProvider) tokenReconciliation {
@@ -676,28 +694,19 @@ func RunDeschedulerStrategies(ctx context.Context, rs *options.DeschedulerServer
namespacedSharedInformerFactory = informers.NewSharedInformerFactoryWithOptions(rs.Client, 0, informers.WithTransform(trimManagedFields), informers.WithNamespace(prometheusProvider.Prometheus.AuthToken.SecretReference.Namespace))
}
descheduler, err := bootstrapDescheduler(ctx, rs, deschedulerPolicy, evictionPolicyGroupVersion, metricProviderTokenReconciliation, sharedInformerFactory, namespacedSharedInformerFactory, eventRecorder)
_, runLoop, err := bootstrapDescheduler(ctx, rs, deschedulerPolicy, evictionPolicyGroupVersion, metricProviderTokenReconciliation, sharedInformerFactory, namespacedSharedInformerFactory, eventRecorder)
if err != nil {
span.AddEvent("Failed to bootstrap a descheduler", trace.WithAttributes(attribute.String("err", err.Error())))
return err
}
wait.NonSlidingUntil(func() {
if metricProviderTokenReconciliation == inClusterReconciliation {
// Read the sa token and assume it has the sufficient permissions to authenticate
if err := descheduler.promClientCtrl.reconcileInClusterSAToken(); err != nil {
klog.ErrorS(err, "unable to reconcile an in cluster SA token")
return
}
}
// A next context is created here intentionally to avoid nesting the spans via context.
sCtx, sSpan := tracing.Tracer().Start(ctx, "NonSlidingUntil")
defer sSpan.End()
err = descheduler.runDeschedulerLoop(sCtx)
if err != nil {
sSpan.AddEvent("Failed to run descheduler loop", trace.WithAttributes(attribute.String("err", err.Error())))
if err := runLoop(sCtx); err != nil {
sSpan.AddEvent("Descheduling loop failed", trace.WithAttributes(attribute.String("err", err.Error())))
klog.Error(err)
return
}
pkg/descheduler/descheduler_test.go
+57 -23
View File
@@ -202,7 +202,7 @@ func lowNodeUtilizationPolicy(thresholds, targetThresholds api.ResourceThreshold
}
}
func initDescheduler(t *testing.T, ctx context.Context, featureGates featuregate.FeatureGate, internalDeschedulerPolicy *api.DeschedulerPolicy, metricsClient metricsclient.Interface, dryRun bool, objects ...runtime.Object) (*options.DeschedulerServer, *descheduler, *fakeclientset.Clientset) {
func initDescheduler(t *testing.T, ctx context.Context, featureGates featuregate.FeatureGate, internalDeschedulerPolicy *api.DeschedulerPolicy, metricsClient metricsclient.Interface, dryRun bool, objects ...runtime.Object) (*options.DeschedulerServer, *descheduler, runFncType, *fakeclientset.Clientset) {
client := fakeclientset.NewSimpleClientset(objects...)
eventClient := fakeclientset.NewSimpleClientset(objects...)
@@ -219,8 +219,16 @@ func initDescheduler(t *testing.T, ctx context.Context, featureGates featuregate
sharedInformerFactory := informers.NewSharedInformerFactoryWithOptions(rs.Client, 0, informers.WithTransform(trimManagedFields))
eventBroadcaster, eventRecorder := utils.GetRecorderAndBroadcaster(ctx, client)
var namespacedSharedInformerFactory informers.SharedInformerFactory
prometheusProvider := metricsProviderListToMap(internalDeschedulerPolicy.MetricsProviders)[api.PrometheusMetrics]
metricProviderTokenReconciliation := prometheusProviderToTokenReconciliation(prometheusProvider)
if metricProviderTokenReconciliation == secretReconciliation {
namespacedSharedInformerFactory = informers.NewSharedInformerFactoryWithOptions(rs.Client, 0, informers.WithTransform(trimManagedFields), informers.WithNamespace(prometheusProvider.Prometheus.AuthToken.SecretReference.Namespace))
}
// Always create descheduler with real client/factory first to register all informers
descheduler, err := bootstrapDescheduler(ctx, rs, internalDeschedulerPolicy, "v1", noReconciliation, sharedInformerFactory, nil, eventRecorder)
descheduler, runFnc, err := bootstrapDescheduler(ctx, rs, internalDeschedulerPolicy, "v1", metricProviderTokenReconciliation, sharedInformerFactory, namespacedSharedInformerFactory, eventRecorder)
if err != nil {
eventBroadcaster.Shutdown()
t.Fatalf("Failed to bootstrap a descheduler: %v", err)
@@ -250,7 +258,7 @@ func initDescheduler(t *testing.T, ctx context.Context, featureGates featuregate
}
}
return rs, descheduler, client
return rs, descheduler, runFnc, client
}
func TestTaintsUpdated(t *testing.T) {
@@ -571,7 +579,7 @@ func TestPodEvictorReset(t *testing.T) {
internalDeschedulerPolicy := removePodsViolatingNodeTaintsPolicy()
ctxCancel, cancel := context.WithCancel(ctx)
_, descheduler, client := initDescheduler(t, ctxCancel, initFeatureGates(), internalDeschedulerPolicy, nil, tc.dryRun, node1, node2, p1, p2)
_, descheduler, _, client := initDescheduler(t, ctxCancel, initFeatureGates(), internalDeschedulerPolicy, nil, tc.dryRun, node1, node2, p1, p2)
defer cancel()
var evictedPods []string
@@ -623,8 +631,8 @@ func checkTotals(t *testing.T, ctx context.Context, descheduler *descheduler, to
t.Logf("Total evictions: %v, total eviction requests: %v, total evictions and eviction requests: %v", totalEvicted, totalEvictionRequests, totalEvicted+totalEvictionRequests)
}
func runDeschedulingCycleAndCheckTotals(t *testing.T, ctx context.Context, nodes []*v1.Node, descheduler *descheduler, totalEvictionRequests, totalEvicted uint) {
err := descheduler.runDeschedulerLoop(ctx)
func runDeschedulingCycleAndCheckTotals(t *testing.T, ctx context.Context, nodes []*v1.Node, descheduler *descheduler, runFnc runFncType, totalEvictionRequests, totalEvicted uint) {
err := runFnc(ctx)
if err != nil {
t.Fatalf("Unable to run a descheduling loop: %v", err)
}
@@ -664,19 +672,19 @@ func TestEvictionRequestsCache(t *testing.T) {
featureGates.Add(map[featuregate.Feature]featuregate.FeatureSpec{
features.EvictionsInBackground: {Default: true, PreRelease: featuregate.Alpha},
})
_, descheduler, client := initDescheduler(t, ctxCancel, featureGates, internalDeschedulerPolicy, nil, false, node1, node2, p1, p2, p3, p4)
_, descheduler, runFnc, client := initDescheduler(t, ctxCancel, featureGates, internalDeschedulerPolicy, nil, false, node1, node2, p1, p2, p3, p4)
defer cancel()
var evictedPods []string
client.PrependReactor("create", "pods", podEvictionReactionTestingFnc(&evictedPods, func(name string) bool { return name == "p1" || name == "p2" }, nil))
klog.Infof("2 evictions in background expected, 2 normal evictions")
runDeschedulingCycleAndCheckTotals(t, ctx, nodes, descheduler, 2, 2)
runDeschedulingCycleAndCheckTotals(t, ctx, nodes, descheduler, runFnc, 2, 2)
klog.Infof("Repeat the same as previously to confirm no more evictions in background are requested")
// No evicted pod is actually deleted on purpose so the test can run the descheduling cycle repeatedly
// without recreating the pods.
runDeschedulingCycleAndCheckTotals(t, ctx, nodes, descheduler, 2, 2)
runDeschedulingCycleAndCheckTotals(t, ctx, nodes, descheduler, runFnc, 2, 2)
klog.Infof("Scenario: Eviction in background got initiated")
p2.Annotations[evictions.EvictionInProgressAnnotationKey] = ""
@@ -686,7 +694,7 @@ func TestEvictionRequestsCache(t *testing.T) {
time.Sleep(100 * time.Millisecond)
klog.Infof("Repeat the same as previously to confirm no more evictions in background are requested")
runDeschedulingCycleAndCheckTotals(t, ctx, nodes, descheduler, 2, 2)
runDeschedulingCycleAndCheckTotals(t, ctx, nodes, descheduler, runFnc, 2, 2)
klog.Infof("Scenario: Another eviction in background got initiated")
p1.Annotations[evictions.EvictionInProgressAnnotationKey] = ""
@@ -696,7 +704,7 @@ func TestEvictionRequestsCache(t *testing.T) {
time.Sleep(100 * time.Millisecond)
klog.Infof("Repeat the same as previously to confirm no more evictions in background are requested")
runDeschedulingCycleAndCheckTotals(t, ctx, nodes, descheduler, 2, 2)
runDeschedulingCycleAndCheckTotals(t, ctx, nodes, descheduler, runFnc, 2, 2)
klog.Infof("Scenario: Eviction in background completed")
if err := client.CoreV1().Pods(p1.Namespace).Delete(context.TODO(), p1.Name, metav1.DeleteOptions{}); err != nil {
@@ -705,7 +713,7 @@ func TestEvictionRequestsCache(t *testing.T) {
time.Sleep(100 * time.Millisecond)
klog.Infof("Check the number of evictions in background decreased")
runDeschedulingCycleAndCheckTotals(t, ctx, nodes, descheduler, 1, 2)
runDeschedulingCycleAndCheckTotals(t, ctx, nodes, descheduler, runFnc, 1, 2)
klog.Infof("Scenario: A new pod without eviction in background added")
if _, err := client.CoreV1().Pods(p5.Namespace).Create(context.TODO(), p5, metav1.CreateOptions{}); err != nil {
@@ -714,7 +722,7 @@ func TestEvictionRequestsCache(t *testing.T) {
time.Sleep(100 * time.Millisecond)
klog.Infof("Check the number of evictions increased after running a descheduling cycle")
runDeschedulingCycleAndCheckTotals(t, ctx, nodes, descheduler, 1, 3)
runDeschedulingCycleAndCheckTotals(t, ctx, nodes, descheduler, runFnc, 1, 3)
klog.Infof("Scenario: Eviction in background canceled => eviction in progress annotation removed")
delete(p2.Annotations, evictions.EvictionInProgressAnnotationKey)
@@ -727,7 +735,7 @@ func TestEvictionRequestsCache(t *testing.T) {
checkTotals(t, ctx, descheduler, 0, 3)
klog.Infof("Scenario: Re-run the descheduling cycle to re-request eviction in background")
runDeschedulingCycleAndCheckTotals(t, ctx, nodes, descheduler, 1, 3)
runDeschedulingCycleAndCheckTotals(t, ctx, nodes, descheduler, runFnc, 1, 3)
klog.Infof("Scenario: Eviction in background completed with a pod in completed state")
p2.Status.Phase = v1.PodSucceeded
@@ -737,7 +745,7 @@ func TestEvictionRequestsCache(t *testing.T) {
time.Sleep(100 * time.Millisecond)
klog.Infof("Check the number of evictions in background decreased")
runDeschedulingCycleAndCheckTotals(t, ctx, nodes, descheduler, 0, 3)
runDeschedulingCycleAndCheckTotals(t, ctx, nodes, descheduler, runFnc, 0, 3)
}
func TestDeschedulingLimits(t *testing.T) {
@@ -800,7 +808,7 @@ func TestDeschedulingLimits(t *testing.T) {
featureGates.Add(map[featuregate.Feature]featuregate.FeatureSpec{
features.EvictionsInBackground: {Default: true, PreRelease: featuregate.Alpha},
})
_, descheduler, client := initDescheduler(t, ctxCancel, featureGates, tc.policy, nil, false, node1, node2)
_, descheduler, runFnc, client := initDescheduler(t, ctxCancel, featureGates, tc.policy, nil, false, node1, node2)
defer cancel()
var evictedPods []string
@@ -832,7 +840,7 @@ func TestDeschedulingLimits(t *testing.T) {
time.Sleep(100 * time.Millisecond)
klog.Infof("2 evictions in background expected, 2 normal evictions")
err := descheduler.runDeschedulerLoop(ctx)
err := runFnc(ctx)
if err != nil {
t.Fatalf("Unable to run a descheduling loop: %v", err)
}
@@ -998,7 +1006,7 @@ func TestNodeLabelSelectorBasedEviction(t *testing.T) {
}
ctxCancel, cancel := context.WithCancel(ctx)
_, deschedulerInstance, client := initDescheduler(t, ctxCancel, initFeatureGates(), policy, nil, tc.dryRun, objects...)
_, deschedulerInstance, _, client := initDescheduler(t, ctxCancel, initFeatureGates(), policy, nil, tc.dryRun, objects...)
defer cancel()
// Verify all pods are created initially
@@ -1112,7 +1120,7 @@ func TestLoadAwareDescheduling(t *testing.T) {
policy.MetricsProviders = []api.MetricsProvider{{Source: api.KubernetesMetrics}}
ctxCancel, cancel := context.WithCancel(ctx)
_, descheduler, _ := initDescheduler(
_, descheduler, runFnc, _ := initDescheduler(
t,
ctxCancel,
initFeatureGates(),
@@ -1126,7 +1134,7 @@ func TestLoadAwareDescheduling(t *testing.T) {
// after newDescheduler in RunDeschedulerStrategies.
descheduler.metricsCollector.Collect(ctx)
err := descheduler.runDeschedulerLoop(ctx)
err := runFnc(ctx)
if err != nil {
t.Fatalf("Unable to run a descheduling loop: %v", err)
}
@@ -1465,6 +1473,14 @@ func TestPluginPrometheusClientAccess(t *testing.T) {
)
deschedulerPolicy := &api.DeschedulerPolicy{
MetricsProviders: []api.MetricsProvider{
{
Source: api.PrometheusMetrics,
Prometheus: &api.Prometheus{
URL: prometheusURL,
},
},
},
Profiles: []api.DeschedulerProfile{
{
Name: "test-profile",
@@ -1486,28 +1502,33 @@ func TestPluginPrometheusClientAccess(t *testing.T) {
node1 := test.BuildTestNode("node1", 1000, 2000, 9, nil)
node2 := test.BuildTestNode("node2", 1000, 2000, 9, nil)
_, descheduler, _ := initDescheduler(t, ctx, initFeatureGates(), deschedulerPolicy, nil, tc.dryRun, node1, node2)
_, descheduler, runFnc, _ := initDescheduler(t, ctx, initFeatureGates(), deschedulerPolicy, nil, tc.dryRun, node1, node2)
// Test cycles with different Prometheus client values
cycles := []struct {
name string
client promapi.Client
token string
}{
{
name: "initial client",
client: &mockPrometheusClient{name: "new-init-client"},
token: "init-token",
},
{
name: "nil client",
client: nil,
token: "",
},
{
name: "new client",
client: &mockPrometheusClient{name: "new-client"},
token: "new-token",
},
{
name: "another client",
client: &mockPrometheusClient{name: "another-client"},
token: "another-token",
},
}
@@ -1516,14 +1537,27 @@ func TestPluginPrometheusClientAccess(t *testing.T) {
// Set the descheduler's Prometheus client
t.Logf("Setting descheduler.promClientCtrl.promClient from %v to %v", descheduler.promClientCtrl.promClient, cycle.client)
descheduler.promClientCtrl.promClient = cycle.client
descheduler.promClientCtrl.inClusterConfig = func() (*rest.Config, error) {
return &rest.Config{BearerToken: cycle.token}, nil
}
descheduler.promClientCtrl.createPrometheusClient = func(url, token string) (promapi.Client, *http.Transport, error) {
if token != cycle.token {
t.Errorf("Expected token to be %q, got %q", cycle.token, token)
}
if url != prometheusURL {
t.Errorf("Expected url to be %q, got %q", prometheusURL, url)
}
return cycle.client, &http.Transport{}, nil
}
newInvoked = false
reactorInvoked = false
prometheusClientFromPluginNewHandle = nil
prometheusClientFromReactor = nil
descheduler.runProfiles(ctx)
if err := runFnc(ctx); err != nil {
t.Fatalf("Unexpected error during running a descheduling cycle: %v", err)
}
t.Logf("After cycle %d: prometheusClientFromReactor=%v, descheduler.promClientCtrl.promClient=%v", i+1, prometheusClientFromReactor, descheduler.promClientCtrl.promClient)