Merge pull request #143734 from mborsz/client-go-connection-pool-kubeconfig

client-go: add min-connection-pool-size to kubeconfig

api/api-rules/violation_exceptions.list

@@ -128,11 +128,13 @@ API rule violation: names_match,k8s.io/apimachinery/pkg/util/intstr,IntOrString,
 API rule violation: names_match,k8s.io/client-go/pkg/apis/clientauthentication/v1,Cluster,CertificateAuthorityData
 API rule violation: names_match,k8s.io/client-go/pkg/apis/clientauthentication/v1,Cluster,DisableCompression
 API rule violation: names_match,k8s.io/client-go/pkg/apis/clientauthentication/v1,Cluster,InsecureSkipTLSVerify
+API rule violation: names_match,k8s.io/client-go/pkg/apis/clientauthentication/v1,Cluster,MinConnectionPoolSize
 API rule violation: names_match,k8s.io/client-go/pkg/apis/clientauthentication/v1,Cluster,ProxyURL
 API rule violation: names_match,k8s.io/client-go/pkg/apis/clientauthentication/v1,Cluster,TLSServerName
 API rule violation: names_match,k8s.io/client-go/pkg/apis/clientauthentication/v1beta1,Cluster,CertificateAuthorityData
 API rule violation: names_match,k8s.io/client-go/pkg/apis/clientauthentication/v1beta1,Cluster,DisableCompression
 API rule violation: names_match,k8s.io/client-go/pkg/apis/clientauthentication/v1beta1,Cluster,InsecureSkipTLSVerify
+API rule violation: names_match,k8s.io/client-go/pkg/apis/clientauthentication/v1beta1,Cluster,MinConnectionPoolSize
 API rule violation: names_match,k8s.io/client-go/pkg/apis/clientauthentication/v1beta1,Cluster,ProxyURL
 API rule violation: names_match,k8s.io/client-go/pkg/apis/clientauthentication/v1beta1,Cluster,TLSServerName
 API rule violation: names_match,k8s.io/cloud-provider/config/v1alpha1,CloudControllerManagerConfiguration,Generic

pkg/generated/openapi/zz_generated.openapi.go

@@ -70331,6 +70331,13 @@ func schema_pkg_apis_clientauthentication_v1_Cluster(ref common.ReferenceCallbac
 							Format:      "",
 						},
 					},
+					"min-connection-pool-size": {
+						SchemaProps: spec.SchemaProps{
+							Description: "MinConnectionPoolSize, if greater than 1, spreads requests to the server across at least this many independent transports, each with its own connections, preferring the one with fewer requests in flight. More connections may be opened, for example when a connection reaches the server's limit of concurrent HTTP/2 streams. Values <= 1 keep the default of a single shared transport. Ignored if the ClientsAllowConnectionPool feature gate is disabled.",
+							Type:        []string{"integer"},
+							Format:      "int32",
+						},
+					},
 					"config": {
 						SchemaProps: spec.SchemaProps{
 							Description: "Config holds additional config data that is specific to the exec plugin with regards to the cluster being authenticated to.\n\nThis data is sourced from the clientcmd Cluster object's extensions[client.authentication.k8s.io/exec] field:\n\nclusters: - name: my-cluster\n  cluster:\n    ...\n    extensions:\n    - name: client.authentication.k8s.io/exec  # reserved extension name for per cluster exec config\n      extension:\n        audience: 06e3fbd18de8  # arbitrary config\n\nIn some environments, the user config may be exactly the same across many clusters (i.e. call this exec plugin) minus some details that are specific to each cluster such as the audience.  This field allows the per cluster config to be directly specified with the cluster info.  Using this field to store secret data is not recommended as one of the prime benefits of exec plugins is that no secrets need to be stored directly in the kubeconfig.",
@@ -70515,6 +70522,13 @@ func schema_pkg_apis_clientauthentication_v1beta1_Cluster(ref common.ReferenceCa
 							Format:      "",
 						},
 					},
+					"min-connection-pool-size": {
+						SchemaProps: spec.SchemaProps{
+							Description: "MinConnectionPoolSize, if greater than 1, spreads requests to the server across at least this many independent transports, each with its own connections, preferring the one with fewer requests in flight. More connections may be opened, for example when a connection reaches the server's limit of concurrent HTTP/2 streams. Values <= 1 keep the default of a single shared transport. Ignored if the ClientsAllowConnectionPool feature gate is disabled.",
+							Type:        []string{"integer"},
+							Format:      "int32",
+						},
+					},
 					"config": {
 						SchemaProps: spec.SchemaProps{
 							Description: "Config holds additional config data that is specific to the exec plugin with regards to the cluster being authenticated to.\n\nThis data is sourced from the clientcmd Cluster object's extensions[client.authentication.k8s.io/exec] field:\n\nclusters: - name: my-cluster\n  cluster:\n    ...\n    extensions:\n    - name: client.authentication.k8s.io/exec  # reserved extension name for per cluster exec config\n      extension:\n        audience: 06e3fbd18de8  # arbitrary config\n\nIn some environments, the user config may be exactly the same across many clusters (i.e. call this exec plugin) minus some details that are specific to each cluster such as the audience.  This field allows the per cluster config to be directly specified with the cluster info.  Using this field to store secret data is not recommended as one of the prime benefits of exec plugins is that no secrets need to be stored directly in the kubeconfig.",

staging/src/k8s.io/client-go/pkg/apis/clientauthentication/types.go

@@ -103,6 +103,12 @@ type Cluster struct {
 	// compression (server-side) and decompression (client-side): https://github.com/kubernetes/kubernetes/issues/112296.
 	// +optional
 	DisableCompression bool
+	// MinConnectionPoolSize, if greater than 1, spreads requests to the server across at least this many independent
+	// transports, each with its own connections, preferring the one with fewer requests in flight. More connections may be
+	// opened, for example when a connection reaches the server's limit of concurrent HTTP/2 streams. Values <= 1 keep the
+	// default of a single shared transport. Ignored if the ClientsAllowConnectionPool feature gate is disabled.
+	// +optional
+	MinConnectionPoolSize int32
 	// Config holds additional config data that is specific to the exec
 	// plugin with regards to the cluster being authenticated to.
 	//

staging/src/k8s.io/client-go/pkg/apis/clientauthentication/v1/types.go

@@ -101,6 +101,12 @@ type Cluster struct {
 	// compression (server-side) and decompression (client-side): https://github.com/kubernetes/kubernetes/issues/112296.
 	// +optional
 	DisableCompression bool `json:"disable-compression,omitempty"`
+	// MinConnectionPoolSize, if greater than 1, spreads requests to the server across at least this many independent
+	// transports, each with its own connections, preferring the one with fewer requests in flight. More connections may be
+	// opened, for example when a connection reaches the server's limit of concurrent HTTP/2 streams. Values <= 1 keep the
+	// default of a single shared transport. Ignored if the ClientsAllowConnectionPool feature gate is disabled.
+	// +optional
+	MinConnectionPoolSize int32 `json:"min-connection-pool-size,omitempty"`
 	// Config holds additional config data that is specific to the exec
 	// plugin with regards to the cluster being authenticated to.
 	//

staging/src/k8s.io/client-go/pkg/apis/clientauthentication/v1/zz_generated.conversion.go

@@ -86,6 +86,7 @@ func autoConvert_v1_Cluster_To_clientauthentication_Cluster(in *Cluster, out *cl
 	out.CertificateAuthorityData = *(*[]byte)(unsafe.Pointer(&in.CertificateAuthorityData))
 	out.ProxyURL = in.ProxyURL
 	out.DisableCompression = in.DisableCompression
+	out.MinConnectionPoolSize = in.MinConnectionPoolSize
 	if err := runtime.Convert_runtime_RawExtension_To_runtime_Object(&in.Config, &out.Config, s); err != nil {
 		return err
 	}
@@ -104,6 +105,7 @@ func autoConvert_clientauthentication_Cluster_To_v1_Cluster(in *clientauthentica
 	out.CertificateAuthorityData = *(*[]byte)(unsafe.Pointer(&in.CertificateAuthorityData))
 	out.ProxyURL = in.ProxyURL
 	out.DisableCompression = in.DisableCompression
+	out.MinConnectionPoolSize = in.MinConnectionPoolSize
 	if err := runtime.Convert_runtime_Object_To_runtime_RawExtension(&in.Config, &out.Config, s); err != nil {
 		return err
 	}

staging/src/k8s.io/client-go/pkg/apis/clientauthentication/v1beta1/types.go

@@ -101,6 +101,12 @@ type Cluster struct {
 	// compression (server-side) and decompression (client-side): https://github.com/kubernetes/kubernetes/issues/112296.
 	// +optional
 	DisableCompression bool `json:"disable-compression,omitempty"`
+	// MinConnectionPoolSize, if greater than 1, spreads requests to the server across at least this many independent
+	// transports, each with its own connections, preferring the one with fewer requests in flight. More connections may be
+	// opened, for example when a connection reaches the server's limit of concurrent HTTP/2 streams. Values <= 1 keep the
+	// default of a single shared transport. Ignored if the ClientsAllowConnectionPool feature gate is disabled.
+	// +optional
+	MinConnectionPoolSize int32 `json:"min-connection-pool-size,omitempty"`
 	// Config holds additional config data that is specific to the exec
 	// plugin with regards to the cluster being authenticated to.
 	//

staging/src/k8s.io/client-go/pkg/apis/clientauthentication/v1beta1/zz_generated.conversion.go

@@ -86,6 +86,7 @@ func autoConvert_v1beta1_Cluster_To_clientauthentication_Cluster(in *Cluster, ou
 	out.CertificateAuthorityData = *(*[]byte)(unsafe.Pointer(&in.CertificateAuthorityData))
 	out.ProxyURL = in.ProxyURL
 	out.DisableCompression = in.DisableCompression
+	out.MinConnectionPoolSize = in.MinConnectionPoolSize
 	if err := runtime.Convert_runtime_RawExtension_To_runtime_Object(&in.Config, &out.Config, s); err != nil {
 		return err
 	}
@@ -104,6 +105,7 @@ func autoConvert_clientauthentication_Cluster_To_v1beta1_Cluster(in *clientauthe
 	out.CertificateAuthorityData = *(*[]byte)(unsafe.Pointer(&in.CertificateAuthorityData))
 	out.ProxyURL = in.ProxyURL
 	out.DisableCompression = in.DisableCompression
+	out.MinConnectionPoolSize = in.MinConnectionPoolSize
 	if err := runtime.Convert_runtime_Object_To_runtime_RawExtension(&in.Config, &out.Config, s); err != nil {
 		return err
 	}

staging/src/k8s.io/client-go/rest/exec.go

@@ -56,6 +56,7 @@ func ConfigToExecCluster(config *Config) (*clientauthenticationapi.Cluster, erro
 		CertificateAuthorityData: caData,
 		ProxyURL:                 proxyURL,
 		DisableCompression:       config.DisableCompression,
+		MinConnectionPoolSize:    int32(config.MinConnectionPoolSize),
 		Config:                   config.ExecProvider.Config,
 	}, nil
 }
@@ -80,7 +81,8 @@ func ExecClusterToConfig(cluster *clientauthenticationapi.Cluster) (*Config, err
 			ServerName: cluster.TLSServerName,
 			CAData:     cluster.CertificateAuthorityData,
 		},
-		Proxy:              proxy,
-		DisableCompression: cluster.DisableCompression,
+		Proxy:                 proxy,
+		DisableCompression:    cluster.DisableCompression,
+		MinConnectionPoolSize: int(cluster.MinConnectionPoolSize),
 	}, nil
 }

staging/src/k8s.io/client-go/rest/exec_test.go

@@ -70,7 +70,8 @@ func TestConfigToExecCluster(t *testing.T) {
 					Insecure:   true,
 					CAData:     []byte("some-ca-data"),
 				},
-				Proxy: proxy,
+				Proxy:                 proxy,
+				MinConnectionPoolSize: 4,
 			},
 			wantOut: clientauthenticationapi.Cluster{
 				Server:                   "some-host",
@@ -78,6 +79,7 @@ func TestConfigToExecCluster(t *testing.T) {
 				InsecureSkipTLSVerify:    true,
 				CertificateAuthorityData: []byte("some-ca-data"),
 				ProxyURL:                 proxyURL,
+				MinConnectionPoolSize:    4,
 				Config: &runtime.Unknown{
 					Raw: []byte("stuff"),
 				},
@@ -292,7 +294,6 @@ func TestConfigToExecClusterRoundtrip(t *testing.T) {
 		expected.TLSClientConfig.NextProtos = nil
 		expected.UserAgent = ""
 		expected.DisableCompression = false
-		expected.MinConnectionPoolSize = 0
 		expected.Transport = nil
 		expected.WrapTransport = nil
 		expected.QPS = 0.0
@@ -302,6 +303,8 @@ func TestConfigToExecClusterRoundtrip(t *testing.T) {
 		expected.WarningHandlerWithContext = nil
 		expected.Timeout = 0
 		expected.Dial = nil
+		// MinConnectionPoolSize is an int32 in the exec cluster info.
+		expected.MinConnectionPoolSize = int(int32(expected.MinConnectionPoolSize))
 
 		// Manually set URLs so we don't get an error when parsing these during the roundtrip.
 		if expected.Host != "" {

staging/src/k8s.io/client-go/tools/clientcmd/api/types.go

@@ -102,6 +102,13 @@ type Cluster struct {
 	// compression (server-side) and decompression (client-side): https://github.com/kubernetes/kubernetes/issues/112296.
 	// +optional
 	DisableCompression bool `json:"disable-compression,omitempty"`
+	// MinConnectionPoolSize, if greater than 1, spreads requests to the server across at least this many independent
+	// transports, each with its own connections, preferring the one with fewer requests in flight. More connections may be
+	// opened, for example when a connection reaches the server's limit of concurrent HTTP/2 streams. 0 and 1 keep the
+	// default of a single shared transport. Values below 0 or above 100 are invalid. Ignored if the
+	// ClientsAllowConnectionPool feature gate is disabled.
+	// +optional
+	MinConnectionPoolSize int32 `json:"min-connection-pool-size,omitempty"`
 	// Extensions holds additional information. This is useful for extenders so that reads and writes don't clobber unknown fields
 	// +optional
 	Extensions map[string]runtime.Object `json:"extensions,omitempty"`

staging/src/k8s.io/client-go/tools/clientcmd/api/v1/types.go

@@ -94,6 +94,13 @@ type Cluster struct {
 	// compression (server-side) and decompression (client-side): https://github.com/kubernetes/kubernetes/issues/112296.
 	// +optional
 	DisableCompression bool `json:"disable-compression,omitempty"`
+	// MinConnectionPoolSize, if greater than 1, spreads requests to the server across at least this many independent
+	// transports, each with its own connections, preferring the one with fewer requests in flight. More connections may be
+	// opened, for example when a connection reaches the server's limit of concurrent HTTP/2 streams. 0 and 1 keep the
+	// default of a single shared transport. Values below 0 or above 100 are invalid. Ignored if the
+	// ClientsAllowConnectionPool feature gate is disabled.
+	// +optional
+	MinConnectionPoolSize int32 `json:"min-connection-pool-size,omitempty"`
 	// Extensions holds additional information. This is useful for extenders so that reads and writes don't clobber unknown fields
 	// +optional
 	Extensions []NamedExtension `json:"extensions,omitempty"`

staging/src/k8s.io/client-go/tools/clientcmd/api/v1/zz_generated.conversion.go

@@ -256,6 +256,7 @@ func autoConvert_v1_Cluster_To_api_Cluster(in *Cluster, out *api.Cluster, s conv
 	out.CertificateAuthorityData = *(*[]byte)(unsafe.Pointer(&in.CertificateAuthorityData))
 	out.ProxyURL = in.ProxyURL
 	out.DisableCompression = in.DisableCompression
+	out.MinConnectionPoolSize = in.MinConnectionPoolSize
 	if err := Convert_Slice_v1_NamedExtension_To_Map_string_To_runtime_Object(&in.Extensions, &out.Extensions, s); err != nil {
 		return err
 	}
@@ -276,6 +277,7 @@ func autoConvert_api_Cluster_To_v1_Cluster(in *api.Cluster, out *Cluster, s conv
 	out.CertificateAuthorityData = *(*[]byte)(unsafe.Pointer(&in.CertificateAuthorityData))
 	out.ProxyURL = in.ProxyURL
 	out.DisableCompression = in.DisableCompression
+	out.MinConnectionPoolSize = in.MinConnectionPoolSize
 	if err := Convert_Map_string_To_runtime_Object_To_Slice_v1_NamedExtension(&in.Extensions, &out.Extensions, s); err != nil {
 		return err
 	}

staging/src/k8s.io/client-go/tools/clientcmd/client_config.go

@@ -204,6 +204,7 @@ func (config *DirectClientConfig) ClientConfig() (*restclient.Config, error) {
 	}
 
 	clientConfig.DisableCompression = configClusterInfo.DisableCompression
+	clientConfig.MinConnectionPoolSize = int(configClusterInfo.MinConnectionPoolSize)
 
 	if config.overrides != nil && len(config.overrides.Timeout) > 0 {
 		timeout, err := ParseTimeout(config.overrides.Timeout)

staging/src/k8s.io/client-go/tools/clientcmd/client_config_test.go

@@ -300,6 +300,59 @@ func TestDisableCompression(t *testing.T) {
 	matchBoolArg(true, actualCfg.DisableCompression, t)
 }
 
+func TestMinConnectionPoolSize(t *testing.T) {
+	kubeconfig := []byte(`apiVersion: v1
+kind: Config
+clusters:
+- name: clean
+  cluster:
+    server: https://localhost:8443
+    min-connection-pool-size: 4
+contexts:
+- name: clean
+  context:
+    cluster: clean
+current-context: clean
+`)
+	config, err := Load(kubeconfig)
+	if err != nil {
+		t.Fatal(err)
+	}
+	matchIntArg(4, int(config.Clusters["clean"].MinConnectionPoolSize), t)
+
+	// Writing and loading the kubeconfig again keeps the value.
+	written, err := Write(*config)
+	if err != nil {
+		t.Fatal(err)
+	}
+	config, err = Load(written)
+	if err != nil {
+		t.Fatal(err)
+	}
+	matchIntArg(4, int(config.Clusters["clean"].MinConnectionPoolSize), t)
+
+	restConfig, err := NewNonInteractiveClientConfig(*config, "clean", &ConfigOverrides{}, nil).ClientConfig()
+	if err != nil {
+		t.Fatal(err)
+	}
+	matchIntArg(4, restConfig.MinConnectionPoolSize, t)
+
+	for _, size := range []int32{0, 1, 100} {
+		cluster := *config.Clusters["clean"]
+		cluster.MinConnectionPoolSize = size
+		if errs := validateClusterInfo("clean", cluster); len(errs) != 0 {
+			t.Errorf("unexpected validation errors for min-connection-pool-size %d: %v", size, errs)
+		}
+	}
+	for _, size := range []int32{-1, 101} {
+		cluster := *config.Clusters["clean"]
+		cluster.MinConnectionPoolSize = size
+		if errs := validateClusterInfo("clean", cluster); len(errs) != 1 {
+			t.Errorf("expected one validation error for min-connection-pool-size %d, got %v", size, errs)
+		}
+	}
+}
+
 func TestInsecureOverridesCA(t *testing.T) {
 	config := createCAValidTestConfig()
 	clientBuilder := NewNonInteractiveClientConfig(*config, "clean", &ConfigOverrides{

staging/src/k8s.io/client-go/tools/clientcmd/validation.go

@@ -223,6 +223,10 @@ func ConfirmUsable(config clientcmdapi.Config, passedContextName string) error {
 	return newErrConfigurationInvalid(validationErrors)
 }
 
+// maxMinConnectionPoolSize is the largest min-connection-pool-size accepted in
+// a kubeconfig. Every unit is a separate transport created with the client.
+const maxMinConnectionPoolSize = 100
+
 // validateClusterInfo looks for conflicts and errors in the cluster info
 func validateClusterInfo(clusterName string, clusterInfo clientcmdapi.Cluster) []error {
 	validationErrors := make([]error, 0)
@@ -244,6 +248,9 @@ func validateClusterInfo(clusterName string, clusterInfo clientcmdapi.Cluster) [
 			validationErrors = append(validationErrors, fmt.Errorf("invalid 'proxy-url' %q for cluster %q: %w", proxyURL, clusterName, err))
 		}
 	}
+	if clusterInfo.MinConnectionPoolSize < 0 || clusterInfo.MinConnectionPoolSize > maxMinConnectionPoolSize {
+		validationErrors = append(validationErrors, fmt.Errorf("invalid 'min-connection-pool-size' %d for cluster %q: must be between 0 and %d", clusterInfo.MinConnectionPoolSize, clusterName, maxMinConnectionPoolSize))
+	}
 	// Make sure CA data and CA file aren't both specified
 	if len(clusterInfo.CertificateAuthority) != 0 && len(clusterInfo.CertificateAuthorityData) != 0 {
 		validationErrors = append(validationErrors, fmt.Errorf("certificate-authority-data and certificate-authority are both specified for %v. certificate-authority-data will override.", clusterName))

staging/src/k8s.io/kubectl/pkg/cmd/config/set.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"io"
 	"reflect"
+	"strconv"
 	"strings"
 
 	"github.com/spf13/cobra"
@@ -224,6 +225,17 @@ func modifyConfig(curr reflect.Value, steps *navigationSteps, propertyValue stri
 		actualCurrValue.SetBool(boolValue)
 		return nil
 
+	case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
+		if steps.moreStepsRemaining() {
+			return fmt.Errorf("can't have more steps after an integer. %v", steps)
+		}
+		intValue, err := strconv.ParseInt(propertyValue, 10, actualCurrValue.Type().Bits())
+		if err != nil {
+			return fmt.Errorf("error parsing integer value %q: %w", propertyValue, err)
+		}
+		actualCurrValue.SetInt(intValue)
+		return nil
+
 	case reflect.Struct:
 		for fieldIndex := 0; fieldIndex < actualCurrValue.NumField(); fieldIndex++ {
 			currFieldValue := actualCurrValue.Field(fieldIndex)

staging/src/k8s.io/kubectl/pkg/cmd/config/set_test.go

@@ -55,6 +55,48 @@ func TestSetConfigCurrentContext(t *testing.T) {
 	test.run(t)
 }
 
+func TestSetConfigInteger(t *testing.T) {
+	conf := clientcmdapi.NewConfig()
+	conf.Clusters["my-cluster"] = &clientcmdapi.Cluster{Server: "https://localhost:8443"}
+	fakeKubeFile, err := os.CreateTemp(os.TempDir(), "")
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	defer utiltesting.CloseAndRemove(t, fakeKubeFile)
+	if err := clientcmd.WriteToFile(*conf, fakeKubeFile.Name()); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	pathOptions := clientcmd.NewDefaultPathOptions()
+	pathOptions.GlobalFile = fakeKubeFile.Name()
+	pathOptions.EnvVar = ""
+
+	for _, tc := range []struct {
+		property  string
+		value     string
+		expectErr bool
+		expected  int32
+	}{
+		{property: "clusters.my-cluster.min-connection-pool-size", value: "4", expected: 4},
+		{property: "clusters.my-cluster.min-connection-pool-size", value: "four", expectErr: true, expected: 4},
+		{property: "clusters.my-cluster.min-connection-pool-size", value: "4294967296", expectErr: true, expected: 4},
+		{property: "clusters.my-cluster.min-connection-pool-size.foo", value: "4", expectErr: true, expected: 4},
+		{property: "clusters.my-cluster.min-connection-pool-size", value: "0", expected: 0},
+	} {
+		opts := setOptions{configAccess: pathOptions, propertyName: tc.property, propertyValue: tc.value}
+		err := opts.run()
+		if tc.expectErr != (err != nil) {
+			t.Errorf("%s=%s: expected error %v, got %v", tc.property, tc.value, tc.expectErr, err)
+		}
+		config, err := clientcmd.LoadFromFile(fakeKubeFile.Name())
+		if err != nil {
+			t.Fatalf("unexpected error loading kubeconfig file: %v", err)
+		}
+		if got := config.Clusters["my-cluster"].MinConnectionPoolSize; got != tc.expected {
+			t.Errorf("%s=%s: expected min-connection-pool-size %d, got %d", tc.property, tc.value, tc.expected, got)
+		}
+	}
+}
+
 func (test setConfigTest) run(t *testing.T) {
 	fakeKubeFile, err := os.CreateTemp(os.TempDir(), "")
 	if err != nil {