diff --git a/pkg/descheduler/descheduler.go b/pkg/descheduler/descheduler.go index 8089e2754..1f0ae29c3 100644 --- a/pkg/descheduler/descheduler.go +++ b/pkg/descheduler/descheduler.go @@ -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 } diff --git a/pkg/descheduler/descheduler_test.go b/pkg/descheduler/descheduler_test.go index 37d845925..284b4edb6 100644 --- a/pkg/descheduler/descheduler_test.go +++ b/pkg/descheduler/descheduler_test.go @@ -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)