Merge pull request #143729 from yongruilin/managedfields-opt-out-scheduler

KEP-5958: Drop managedFields in kube-scheduler informers

cmd/kube-scheduler/app/options/options.go

@@ -33,6 +33,7 @@ import (
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/dynamic/dynamicinformer"
+	clientfeatures "k8s.io/client-go/features"
 	clientset "k8s.io/client-go/kubernetes"
 	restclient "k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
@@ -316,7 +317,7 @@ func (o *Options) Config(ctx context.Context) (*schedulerappconfig.Config, error
 	}
 
 	// Prepare kube clients.
-	client, eventClient, err := createClients(c.KubeConfig)
+	client, informerClient, eventClient, err := createClients(c.KubeConfig)
 	if err != nil {
 		return nil, err
 	}
@@ -339,8 +340,11 @@ func (o *Options) Config(ctx context.Context) (*schedulerappconfig.Config, error
 	}
 
 	c.Client = client
-	c.InformerFactory = scheduler.NewInformerFactory(client, 0, o.InformerName)
-	dynClient := dynamic.NewForConfigOrDie(c.KubeConfig)
+	c.InformerFactory = scheduler.NewInformerFactory(informerClient, 0, o.InformerName)
+	// dynClient only serves the dynamic informers.
+	dynConfig := restclient.CopyConfig(c.KubeConfig)
+	dynConfig.DropManagedFields = true
+	dynClient := dynamic.NewForConfigOrDie(dynConfig)
 	c.DynInformerFactory = dynamicinformer.NewFilteredDynamicSharedInformerFactory(dynClient, 0, corev1.NamespaceAll, nil)
 	c.LeaderElection = leaderElectionConfig
 	c.ComponentGlobalsRegistry = o.ComponentGlobalsRegistry
@@ -399,17 +403,29 @@ func createKubeConfig(config componentbaseconfig.ClientConnectionConfiguration,
 	return kubeConfig, nil
 }
 
-// createClients creates a kube client and an event client from the given kubeConfig
-func createClients(kubeConfig *restclient.Config) (clientset.Interface, clientset.Interface, error) {
+// createClients creates a kube client, an informer client and an event client from the given kubeConfig
+func createClients(kubeConfig *restclient.Config) (clientset.Interface, clientset.Interface, clientset.Interface, error) {
 	client, err := clientset.NewForConfig(restclient.AddUserAgent(kubeConfig, "scheduler"))
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, nil, err
+	}
+
+	// Only the informers drop managedFields: plugins get client through handle.ClientSet().
+	// Without the gate, the informers keep sharing client, including its rate limiter.
+	informerClient := client
+	if clientfeatures.FeatureGates().Enabled(clientfeatures.ManagedFieldsOptOutClient) {
+		informerConfig := restclient.CopyConfig(kubeConfig)
+		informerConfig.DropManagedFields = true
+		informerClient, err = clientset.NewForConfig(informerConfig)
+		if err != nil {
+			return nil, nil, nil, err
+		}
 	}
 
 	eventClient, err := clientset.NewForConfig(kubeConfig)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, nil, err
 	}
 
-	return client, eventClient, nil
+	return client, informerClient, eventClient, nil
 }

cmd/kube-scheduler/app/options/options_test.go

@@ -23,6 +23,7 @@ import (
 	"net/http/httptest"
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
 	"time"
 
@@ -32,6 +33,10 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	apiserveroptions "k8s.io/apiserver/pkg/server/options"
+	clientfeatures "k8s.io/client-go/features"
+	clientfeaturestesting "k8s.io/client-go/features/testing"
+	clientset "k8s.io/client-go/kubernetes"
+	restclient "k8s.io/client-go/rest"
 	basecompatibility "k8s.io/component-base/compatibility"
 	componentbaseconfig "k8s.io/component-base/config"
 	"k8s.io/component-base/logs"
@@ -1009,3 +1014,39 @@ profiles:
 		})
 	}
 }
+
+func TestCreateClients(t *testing.T) {
+	var accept string
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
+		accept = req.Header.Get("Accept")
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = w.Write([]byte(`{"kind":"PodList","apiVersion":"v1","items":[]}`))
+	}))
+	defer server.Close()
+
+	for _, enabled := range []bool{false, true} {
+		t.Run(fmt.Sprintf("ManagedFieldsOptOutClient=%t", enabled), func(t *testing.T) {
+			clientfeaturestesting.SetFeatureDuringTest(t, clientfeatures.ManagedFieldsOptOutClient, enabled)
+			client, informerClient, _, err := createClients(&restclient.Config{Host: server.URL})
+			if err != nil {
+				t.Fatalf("unexpected error creating clients: %v", err)
+			}
+			// Without the gate, the informers keep sharing client, including its rate limiter.
+			if !enabled && informerClient != client {
+				t.Error("expected the informers to share the scheduler's client")
+			}
+			list := func(c clientset.Interface) string {
+				if _, err := c.CoreV1().Pods("").List(context.Background(), metav1.ListOptions{}); err != nil {
+					t.Fatalf("unexpected error listing pods: %v", err)
+				}
+				return accept
+			}
+			if got := list(client); strings.Contains(got, "drop=") {
+				t.Errorf("client Accept = %q, want no drop parameter", got)
+			}
+			if got := list(informerClient); strings.Contains(got, "drop=metadata.managedFields") != enabled {
+				t.Errorf("informer client Accept = %q, want drop parameter: %t", got, enabled)
+			}
+		})
+	}
+}

pkg/scheduler/scheduler.go

@@ -29,6 +29,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/client-go/dynamic/dynamicinformer"
+	clientfeatures "k8s.io/client-go/features"
 	"k8s.io/client-go/informers"
 	coreinformers "k8s.io/client-go/informers/core/v1"
 	"k8s.io/client-go/informers/internalinterfaces"
@@ -628,17 +629,20 @@ func newPodInformer(cs clientset.Interface, resyncPeriod time.Duration, informer
 		InformerName:     informerName,
 	})
 
-	// Dropping `.metadata.managedFields` to improve memory usage.
-	// The Extract workflow (i.e. `ExtractPod`) should be unused.
-	trim := func(obj interface{}) (interface{}, error) {
-		if accessor, err := meta.Accessor(obj); err == nil {
-			if accessor.GetManagedFields() != nil {
-				accessor.SetManagedFields(nil)
+	// With ManagedFieldsOptOutClient, kube-scheduler's informer client drops managedFields instead.
+	if !clientfeatures.FeatureGates().Enabled(clientfeatures.ManagedFieldsOptOutClient) {
+		// Dropping `.metadata.managedFields` to improve memory usage.
+		// The Extract workflow (i.e. `ExtractPod`) should be unused.
+		trim := func(obj interface{}) (interface{}, error) {
+			if accessor, err := meta.Accessor(obj); err == nil {
+				if accessor.GetManagedFields() != nil {
+					accessor.SetManagedFields(nil)
+				}
 			}
+			return obj, nil
 		}
-		return obj, nil
+		_ = informer.SetTransform(trim)
 	}
-	informer.SetTransform(trim)
 	return informer
 }
 

pkg/scheduler/scheduler_test.go

@@ -38,6 +38,8 @@ import (
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/util/version"
 	"k8s.io/apiserver/pkg/util/feature"
+	clientfeatures "k8s.io/client-go/features"
+	clientfeaturestesting "k8s.io/client-go/features/testing"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/kubernetes/fake"
@@ -1522,6 +1524,8 @@ func (f fakePermitPlugin) Permit(ctx context.Context, state fwk.CycleState, p *v
 var _ fwk.PermitPlugin = &fakePermitPlugin{}
 
 func TestNewInformerFactoryTrim(t *testing.T) {
+	// The transform only runs with ManagedFieldsOptOutClient disabled.
+	clientfeaturestesting.SetFeatureDuringTest(t, clientfeatures.ManagedFieldsOptOutClient, false)
 	cs := fake.NewClientset()
 
 	pd := &v1.Pod{ObjectMeta: metav1.ObjectMeta{