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

Client-go: add support for a pool of connections

staging/src/k8s.io/client-go/features/known_features.go

@@ -55,6 +55,14 @@ const (
 	// "application/json" or "application/apply-patch+yaml", respectively.
 	ClientsAllowCBOR Feature = "ClientsAllowCBOR"
 
+	// owner: @mborsz
+	// beta: v1.38
+	//
+	// If enabled, setting rest.Config.MinConnectionPoolSize or transport.Config.MinConnectionPoolSize
+	// greater than 1 make clients spread requests across a pool of independent transports.
+	// Disabling it restores a single shared transport per config.
+	ClientsAllowConnectionPool Feature = "ClientsAllowConnectionPool"
+
 	// owner: @enj
 	// beta: v1.36
 	//
@@ -126,6 +134,9 @@ var defaultVersionedKubernetesFeatureGates = map[Feature]VersionedSpecs{
 	ClientsAllowCBOR: {
 		{Version: version.MustParse("1.32"), Default: false, PreRelease: Alpha},
 	},
+	ClientsAllowConnectionPool: {
+		{Version: version.MustParse("1.38"), Default: true, PreRelease: Beta},
+	},
 	ClientsAllowTLSCacheGC: {
 		{Version: version.MustParse("1.36"), Default: true, PreRelease: Beta},
 	},

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

@@ -102,6 +102,15 @@ type Config struct {
 	// server.
 	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 Transport is set or if the
+	// ClientsAllowConnectionPool feature gate is disabled.
+	MinConnectionPoolSize int
+
 	// Transport may be used for custom HTTP behavior. This attribute may not
 	// be specified with the TLS client certificate options. Use WrapTransport
 	// to provide additional per-server middleware behavior.
@@ -659,6 +668,7 @@ func AnonymousClientConfig(config *Config) *Config {
 		WarningHandlerWithContext: config.WarningHandlerWithContext,
 		UserAgent:                 config.UserAgent,
 		DisableCompression:        config.DisableCompression,
+		MinConnectionPoolSize:     config.MinConnectionPoolSize,
 		QPS:                       config.QPS,
 		Burst:                     config.Burst,
 		Timeout:                   config.Timeout,
@@ -699,6 +709,7 @@ func CopyConfig(config *Config) *Config {
 		},
 		UserAgent:                 config.UserAgent,
 		DisableCompression:        config.DisableCompression,
+		MinConnectionPoolSize:     config.MinConnectionPoolSize,
 		Transport:                 config.Transport,
 		WrapTransport:             config.WrapTransport,
 		QPS:                       config.QPS,

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

@@ -652,7 +652,7 @@ func TestConfigSprint(t *testing.T) {
 		Proxy:                     fakeProxyFunc,
 	}
 	want := fmt.Sprintf(
-		`&rest.Config{Host:"localhost:8080", APIPath:"v1", ContentConfig:rest.ContentConfig{AcceptContentTypes:"application/json", ContentType:"application/json", GroupVersion:(*schema.GroupVersion)(nil), NegotiatedSerializer:runtime.NegotiatedSerializer(nil), DropManagedFields:false}, Username:"gopher", Password:"--- REDACTED ---", BearerToken:"--- REDACTED ---", BearerTokenFile:"", Impersonate:rest.ImpersonationConfig{UserName:"gopher2", UID:"uid123", Groups:[]string(nil), Extra:map[string][]string(nil)}, AuthProvider:api.AuthProviderConfig{Name: "gopher", Config: map[string]string{--- REDACTED ---}}, AuthConfigPersister:rest.AuthProviderConfigPersister(--- REDACTED ---), ExecProvider:api.ExecConfig{Command: "sudo", Args: []string{"--- REDACTED ---"}, Env: []ExecEnvVar{--- REDACTED ---}, APIVersion: "", ProvideClusterInfo: true, Config: runtime.Object(--- REDACTED ---), StdinUnavailable: false}, TLSClientConfig:rest.sanitizedTLSClientConfig{Insecure:false, ServerName:"", CertFile:"a.crt", KeyFile:"a.key", CAFile:"", CertData:[]uint8{0x2d, 0x2d, 0x2d, 0x20, 0x54, 0x52, 0x55, 0x4e, 0x43, 0x41, 0x54, 0x45, 0x44, 0x20, 0x2d, 0x2d, 0x2d}, KeyData:[]uint8{0x2d, 0x2d, 0x2d, 0x20, 0x52, 0x45, 0x44, 0x41, 0x43, 0x54, 0x45, 0x44, 0x20, 0x2d, 0x2d, 0x2d}, CAData:[]uint8(nil), NextProtos:[]string{"h2", "http/1.1"}}, UserAgent:"gobot", DisableCompression:false, Transport:(*rest.fakeRoundTripper)(%p), WrapTransport:(transport.WrapperFunc)(%p), QPS:1, Burst:2, RateLimiter:(*rest.fakeLimiter)(%p), WarningHandler:rest.fakeWarningHandler{}, WarningHandlerWithContext:rest.fakeWarningHandlerWithContext{}, Timeout:3000000000, Dial:(func(context.Context, string, string) (net.Conn, error))(%p), Proxy:(func(*http.Request) (*url.URL, error))(%p)}`,
+		`&rest.Config{Host:"localhost:8080", APIPath:"v1", ContentConfig:rest.ContentConfig{AcceptContentTypes:"application/json", ContentType:"application/json", GroupVersion:(*schema.GroupVersion)(nil), NegotiatedSerializer:runtime.NegotiatedSerializer(nil), DropManagedFields:false}, Username:"gopher", Password:"--- REDACTED ---", BearerToken:"--- REDACTED ---", BearerTokenFile:"", Impersonate:rest.ImpersonationConfig{UserName:"gopher2", UID:"uid123", Groups:[]string(nil), Extra:map[string][]string(nil)}, AuthProvider:api.AuthProviderConfig{Name: "gopher", Config: map[string]string{--- REDACTED ---}}, AuthConfigPersister:rest.AuthProviderConfigPersister(--- REDACTED ---), ExecProvider:api.ExecConfig{Command: "sudo", Args: []string{"--- REDACTED ---"}, Env: []ExecEnvVar{--- REDACTED ---}, APIVersion: "", ProvideClusterInfo: true, Config: runtime.Object(--- REDACTED ---), StdinUnavailable: false}, TLSClientConfig:rest.sanitizedTLSClientConfig{Insecure:false, ServerName:"", CertFile:"a.crt", KeyFile:"a.key", CAFile:"", CertData:[]uint8{0x2d, 0x2d, 0x2d, 0x20, 0x54, 0x52, 0x55, 0x4e, 0x43, 0x41, 0x54, 0x45, 0x44, 0x20, 0x2d, 0x2d, 0x2d}, KeyData:[]uint8{0x2d, 0x2d, 0x2d, 0x20, 0x52, 0x45, 0x44, 0x41, 0x43, 0x54, 0x45, 0x44, 0x20, 0x2d, 0x2d, 0x2d}, CAData:[]uint8(nil), NextProtos:[]string{"h2", "http/1.1"}}, UserAgent:"gobot", DisableCompression:false, MinConnectionPoolSize:0, Transport:(*rest.fakeRoundTripper)(%p), WrapTransport:(transport.WrapperFunc)(%p), QPS:1, Burst:2, RateLimiter:(*rest.fakeLimiter)(%p), WarningHandler:rest.fakeWarningHandler{}, WarningHandlerWithContext:rest.fakeWarningHandlerWithContext{}, Timeout:3000000000, Dial:(func(context.Context, string, string) (net.Conn, error))(%p), Proxy:(func(*http.Request) (*url.URL, error))(%p)}`,
 		c.Transport, fakeWrapperFunc, c.RateLimiter, fakeDialFunc, fakeProxyFunc,
 	)
 

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

@@ -292,6 +292,7 @@ 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

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

@@ -85,10 +85,11 @@ func HTTPWrappersForConfig(config *Config, rt http.RoundTripper) (http.RoundTrip
 func (c *Config) TransportConfig() (*transport.Config, error) {
 	metrics.EnsureRegistered()
 	conf := &transport.Config{
-		UserAgent:          c.UserAgent,
-		Transport:          c.Transport,
-		WrapTransport:      c.WrapTransport,
-		DisableCompression: c.DisableCompression,
+		UserAgent:             c.UserAgent,
+		Transport:             c.Transport,
+		WrapTransport:         c.WrapTransport,
+		DisableCompression:    c.DisableCompression,
+		MinConnectionPoolSize: c.MinConnectionPoolSize,
 		TLS: transport.TLSConfig{
 			Insecure:   c.Insecure,
 			ServerName: c.ServerName,

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

@@ -124,6 +124,15 @@ uml0obOEy+ON91k+SWTJ3ggmF/U=
 				},
 			},
 		},
+		{
+			name: "with connection pool",
+			config: &Config{
+				Host:                  "localhost:8080",
+				Username:              "gopher",
+				Password:              "g0ph3r",
+				MinConnectionPoolSize: 3,
+			},
+		},
 	}
 	var wg sync.WaitGroup
 
@@ -146,3 +155,14 @@ uml0obOEy+ON91k+SWTJ3ggmF/U=
 	}
 	wg.Wait()
 }
+
+func TestTransportConfigMinConnectionPoolSize(t *testing.T) {
+	config := &Config{Host: "localhost:8080", MinConnectionPoolSize: 3}
+	transportConfig, err := config.TransportConfig()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if transportConfig.MinConnectionPoolSize != 3 {
+		t.Errorf("expected MinConnectionPoolSize 3, got %d", transportConfig.MinConnectionPoolSize)
+	}
+}

staging/src/k8s.io/client-go/transport/ca_rotation.go

@@ -19,6 +19,8 @@ package transport
 import (
 	"bytes"
 	"context"
+	"crypto/tls"
+	"crypto/x509"
 	"net/http"
 	"os"
 	"sync"
@@ -42,7 +44,7 @@ type atomicTransportHolder struct {
 	caRefreshDuration time.Duration
 	// mu covers transport and transportLastChecked
 	mu                   sync.RWMutex
-	transport            *http.Transport
+	transport            rotatableCARoundTripper
 	transportLastChecked time.Time
 }
 
@@ -57,7 +59,7 @@ func (h *atomicTransportHolder) WrappedRoundTripper() http.RoundTripper {
 	return h.transport
 }
 
-func (h *atomicTransportHolder) getTransport(ctx context.Context) *http.Transport {
+func (h *atomicTransportHolder) getTransport(ctx context.Context) rotatableCARoundTripper {
 	if rt := h.getTransportIfFresh(); rt != nil {
 		return rt
 	}
@@ -69,7 +71,7 @@ func (h *atomicTransportHolder) getTransport(ctx context.Context) *http.Transpor
 	return h.transport
 }
 
-func (h *atomicTransportHolder) getTransportIfFresh() *http.Transport {
+func (h *atomicTransportHolder) getTransportIfFresh() rotatableCARoundTripper {
 	h.mu.RLock()
 	defer h.mu.RUnlock()
 
@@ -124,7 +126,7 @@ func (h *atomicTransportHolder) tryRefreshTransportLocked(ctx context.Context) {
 		return
 	}
 	newTransport := h.transport.Clone()
-	newTransport.TLSClientConfig.RootCAs = newCAs
+	newTransport.SetRootCAs(newCAs)
 	oldTransport := h.transport
 	h.transport = newTransport
 	// Update our tracking of current CA data
@@ -137,11 +139,48 @@ func (h *atomicTransportHolder) tryRefreshTransportLocked(ctx context.Context) {
 	metrics.TransportCAReloads.Increment("success", "updated")
 }
 
+type rotatableCARoundTripper interface {
+	http.RoundTripper
+	Clone() rotatableCARoundTripper
+	GetRootCAs() *x509.CertPool
+	SetRootCAs(*x509.CertPool)
+	CloseIdleConnections()
+}
+
+// rotatableCAHTTPTransport adapts an *http.Transport to rotatableCARoundTripper.
+type rotatableCAHTTPTransport struct {
+	*http.Transport
+}
+
+var _ utilnet.RoundTripperWrapper = &rotatableCAHTTPTransport{}
+
+// WrappedRoundTripper returns the underlying transport, so that helpers like
+// utilnet.TLSClientConfig and utilnet.DialerFor can inspect it.
+func (h *rotatableCAHTTPTransport) WrappedRoundTripper() http.RoundTripper {
+	return h.Transport
+}
+
+func (h *rotatableCAHTTPTransport) Clone() rotatableCARoundTripper {
+	return &rotatableCAHTTPTransport{h.Transport.Clone()}
+}
+func (h *rotatableCAHTTPTransport) GetRootCAs() *x509.CertPool {
+	if h.Transport.TLSClientConfig == nil {
+		return nil
+	}
+	return h.Transport.TLSClientConfig.RootCAs
+}
+func (h *rotatableCAHTTPTransport) SetRootCAs(pool *x509.CertPool) {
+	if h.Transport.TLSClientConfig == nil {
+		h.Transport.TLSClientConfig = &tls.Config{}
+	}
+	h.Transport.TLSClientConfig.RootCAs = pool
+}
+
 // newAtomicTransportHolder creates a new holder for CA file reloading scenarios.
 // The caFile must be specified.
 // caData may be empty but should correspond to the contents of caFile.
 // transport must have a TLS config and its root CAs should match caData.
-func newAtomicTransportHolder(caFile string, caData []byte, transport *http.Transport) *atomicTransportHolder {
+func newAtomicTransportHolder(caFile string, caData []byte, transport rotatableCARoundTripper) *atomicTransportHolder {
 	c := clock.RealClock{}
 	return &atomicTransportHolder{
 		caFile:               caFile,

staging/src/k8s.io/client-go/transport/ca_rotation_test.go

@@ -28,6 +28,7 @@ import (
 	"testing"
 	"time"
 
+	utilnet "k8s.io/apimachinery/pkg/util/net"
 	"k8s.io/apimachinery/pkg/util/wait"
 	clientgofeaturegate "k8s.io/client-go/features"
 	clientfeaturestesting "k8s.io/client-go/features/testing"
@@ -174,7 +175,8 @@ func TestCheckCAFileAndRotate(t *testing.T) {
 			}
 
 			clock := testingclock.NewFakeClock(time.Now())
-			holder := newAtomicTransportHolder(caFile, tt.setupCA, transport)
+			initialTransport := &rotatableCAHTTPTransport{transport}
+			holder := newAtomicTransportHolder(caFile, tt.setupCA, initialTransport)
 			holder.clock = clock
 			holder.transportLastChecked = clock.Now()
 
@@ -190,13 +192,13 @@ func TestCheckCAFileAndRotate(t *testing.T) {
 
 			// Check CA file rotation
 			newTransport := holder.getTransport(t.Context())
-			newRoots := newTransport.TLSClientConfig.RootCAs
+			newRoots := newTransport.GetRootCAs()
 
 			if newRoots == nil || !expectedRoots.Equal(newRoots) {
 				t.Error("new roots did not match expected roots")
 			}
 
-			transportRotated := newTransport != transport
+			transportRotated := newTransport != initialTransport
 			if tt.expectRotation != transportRotated {
 				t.Error("transport rotation did not match")
 			}
@@ -205,6 +207,30 @@ func TestCheckCAFileAndRotate(t *testing.T) {
 	}
 }
 
+// TestCAReloadTransportUnwrapping verifies that helpers that inspect the
+// underlying *http.Transport still work through a CA reloading holder.
+func TestCAReloadTransportUnwrapping(t *testing.T) {
+	caData := []byte(testCACert1)
+	transport := createTestTransport(t, caData)
+	transport.DialContext = (&net.Dialer{}).DialContext
+	holder := newAtomicTransportHolder(writeCAFile(t, caData), caData, &rotatableCAHTTPTransport{transport})
+
+	tlsConfig, err := utilnet.TLSClientConfig(holder)
+	if err != nil {
+		t.Fatalf("TLSClientConfig: %v", err)
+	}
+	if tlsConfig != transport.TLSClientConfig {
+		t.Error("expected the TLS config of the underlying transport")
+	}
+	dial, err := utilnet.DialerFor(holder)
+	if err != nil {
+		t.Fatalf("DialerFor: %v", err)
+	}
+	if dial == nil {
+		t.Error("expected the dialer of the underlying transport")
+	}
+}
+
 func generateServerCertAndCA(t testing.TB) (servingCertPEM, servingKeyPEM, caCertPEM []byte) {
 	t.Helper()
 	certPEM, keyPEM, err := cert.GenerateSelfSignedCertKey("127.0.0.1", nil, nil)
@@ -467,7 +493,7 @@ func TestCARotationMetricsEmitted(t *testing.T) {
 	transport := createTestTransport(t, caData)
 
 	clock := testingclock.NewFakeClock(time.Now())
-	holder := newAtomicTransportHolder(caFile, caData, transport)
+	holder := newAtomicTransportHolder(caFile, caData, &rotatableCAHTTPTransport{transport})
 	holder.clock = clock
 	holder.transportLastChecked = clock.Now()
 

staging/src/k8s.io/client-go/transport/cache.go

@@ -18,6 +18,7 @@ package transport
 
 import (
 	"context"
+	"crypto/tls"
 	"fmt"
 	"net"
 	"net/http"
@@ -64,6 +65,8 @@ type tlsCacheKey struct {
 	serverName         string
 	nextProtos         string
 	disableCompression bool
+	// minConnectionPoolSize is 0 unless a connection pool is used, see connectionPoolSize.
+	minConnectionPoolSize int
 	// these functions are wrapped to allow them to be used as map keys
 	getCert *GetCertHolder
 	dial    *DialHolder
@@ -74,8 +77,8 @@ func (t tlsCacheKey) String() string {
 	if len(t.keyData) > 0 {
 		keyText = "<redacted>"
 	}
-	return fmt.Sprintf("insecure:%v, caData:%#v, caFile:%s, certData:%#v, keyData:%s, serverName:%s, disableCompression:%t, getCert:%p, dial:%p",
-		t.insecure, t.caData, t.caFile, t.certData, keyText, t.serverName, t.disableCompression, t.getCert, t.dial)
+	return fmt.Sprintf("insecure:%v, caData:%#v, caFile:%s, certData:%#v, keyData:%s, serverName:%s, disableCompression:%t, minConnectionPoolSize:%d, getCert:%p, dial:%p",
+		t.insecure, t.caData, t.caFile, t.certData, keyText, t.serverName, t.disableCompression, t.minConnectionPoolSize, t.getCert, t.dial)
 }
 
 func (c *tlsTransportCache) get(config *Config) (http.RoundTripper, error) {
@@ -110,7 +113,7 @@ func (c *tlsTransportCache) get(config *Config) (http.RoundTripper, error) {
 		return nil, err
 	}
 	// The options didn't require a custom TLS config
-	if tlsConfig == nil && config.DialHolder == nil && config.Proxy == nil {
+	if tlsConfig == nil && config.DialHolder == nil && config.Proxy == nil && connectionPoolSize(config) == 0 {
 		return http.DefaultTransport, nil
 	}
 
@@ -139,23 +142,25 @@ func (c *tlsTransportCache) get(config *Config) (http.RoundTripper, error) {
 		go dynamicCertDialer.run(ctx.Done())
 	}
 
-	proxy := http.ProxyFromEnvironment
-	if config.Proxy != nil {
-		proxy = config.Proxy
+	httpTransport := newTransport(config, tlsConfig, dial)
+	var transport http.RoundTripper = httpTransport
+	var rotatable rotatableCARoundTripper = &rotatableCAHTTPTransport{httpTransport}
+
+	if n := connectionPoolSize(config); n > 1 {
+		// Every member is a separate http.Transport with its own connections.
+		// They share dial, and so the cert rotation above, but each gets its
+		// own copy of tlsConfig, because net/http modifies it on first use.
+		members := []rotatableCARoundTripper{rotatable}
+		for len(members) < n {
+			members = append(members, &rotatableCAHTTPTransport{newTransport(config, tlsConfig.Clone(), dial)})
+		}
+		pool := newConnectionPool(members)
+		transport, rotatable = pool, pool
 	}
 
-	httpTransport := utilnet.SetTransportDefaults(&http.Transport{
-		Proxy:               proxy,
-		TLSHandshakeTimeout: 10 * time.Second,
-		TLSClientConfig:     tlsConfig,
-		MaxIdleConnsPerHost: idleConnsPerHost,
-		DialContext:         dial,
-		DisableCompression:  config.DisableCompression,
-	})
-	var transport http.RoundTripper = httpTransport
-
+	// For a pool, a single holder rotates the CAs of all members at once.
 	if config.TLS.ReloadCAFiles && tlsConfig != nil && tlsConfig.RootCAs != nil && len(config.TLS.CAFile) > 0 {
-		transport = newAtomicTransportHolder(config.TLS.CAFile, config.TLS.CAData, httpTransport)
+		transport = newAtomicTransportHolder(config.TLS.CAFile, config.TLS.CAData, rotatable)
 	}
 
 	if !canCache && cancel == nil {
@@ -204,6 +209,24 @@ func (c *tlsTransportCache) get(config *Config) (http.RoundTripper, error) {
 	return transportWithGC, nil
 }
 
+// newTransport returns a new, uncached http.Transport for config that uses
+// tlsConfig and dial.
+func newTransport(config *Config, tlsConfig *tls.Config, dial func(ctx context.Context, network, address string) (net.Conn, error)) *http.Transport {
+	proxy := http.ProxyFromEnvironment
+	if config.Proxy != nil {
+		proxy = config.Proxy
+	}
+
+	return utilnet.SetTransportDefaults(&http.Transport{
+		Proxy:               proxy,
+		TLSHandshakeTimeout: 10 * time.Second,
+		TLSClientConfig:     tlsConfig,
+		MaxIdleConnsPerHost: idleConnsPerHost,
+		DialContext:         dial,
+		DisableCompression:  config.DisableCompression,
+	})
+}
+
 func (c *tlsTransportCache) getLocked(key tlsCacheKey) (http.RoundTripper, bool) {
 	if !clientgofeaturegate.FeatureGates().Enabled(clientgofeaturegate.ClientsAllowTLSCacheGC) {
 		v, ok := c.strongTransports[key]
@@ -249,6 +272,16 @@ func (v *trackedTransport) WrappedRoundTripper() http.RoundTripper {
 	return v.rt
 }
 
+// connectionPoolSize returns the number of transports to pool for c, or 0 if c
+// does not ask for a connection pool or the ClientsAllowConnectionPool feature
+// is disabled. All configs without a pool get 0, so they share cache entries.
+func connectionPoolSize(c *Config) int {
+	if c.MinConnectionPoolSize <= 1 || !clientgofeaturegate.FeatureGates().Enabled(clientgofeaturegate.ClientsAllowConnectionPool) {
+		return 0
+	}
+	return c.MinConnectionPoolSize
+}
+
 // tlsConfigKey returns a unique key for tls.Config objects returned from TLSConfigFor
 func tlsConfigKey(c *Config) (tlsCacheKey, bool, error) {
 	// Make sure ca/key/cert content is loaded
@@ -270,6 +303,8 @@ func tlsConfigKey(c *Config) (tlsCacheKey, bool, error) {
 		dial:               c.DialHolder,
 	}
 
+	k.minConnectionPoolSize = connectionPoolSize(c)
+
 	if c.TLS.ReloadTLSFiles {
 		k.certFile = c.TLS.CertFile
 		k.keyFile = c.TLS.KeyFile

staging/src/k8s.io/client-go/transport/cache_test.go

@@ -20,8 +20,11 @@ import (
 	"context"
 	"crypto/tls"
 	"crypto/x509"
+	"fmt"
+	"io"
 	"net"
 	"net/http"
+	"net/http/httptest"
 	"net/url"
 	"os"
 	"reflect"
@@ -32,10 +35,12 @@ import (
 	"time"
 	"weak"
 
+	utilnet "k8s.io/apimachinery/pkg/util/net"
 	"k8s.io/apimachinery/pkg/util/wait"
 	clientgofeaturegate "k8s.io/client-go/features"
 	clientfeaturestesting "k8s.io/client-go/features/testing"
 	"k8s.io/client-go/tools/metrics"
+	testingclock "k8s.io/utils/clock/testing"
 )
 
 func TestTLSConfigKey(t *testing.T) {
@@ -43,12 +48,13 @@ func TestTLSConfigKey(t *testing.T) {
 	clientfeaturestesting.SetFeatureDuringTest(t, clientgofeaturegate.ClientsAllowCARotation, true)
 	// Make sure config fields that don't affect the tls config don't affect the cache key
 	identicalConfigurations := map[string]*Config{
-		"empty":          {},
-		"basic":          {Username: "bob", Password: "password"},
-		"bearer":         {BearerToken: "token"},
-		"user agent":     {UserAgent: "useragent"},
-		"transport":      {Transport: http.DefaultTransport},
-		"wrap transport": {WrapTransport: func(http.RoundTripper) http.RoundTripper { return nil }},
+		"empty":                {},
+		"basic":                {Username: "bob", Password: "password"},
+		"bearer":               {BearerToken: "token"},
+		"user agent":           {UserAgent: "useragent"},
+		"transport":            {Transport: http.DefaultTransport},
+		"wrap transport":       {WrapTransport: func(http.RoundTripper) http.RoundTripper { return nil }},
+		"connection pool of 1": {MinConnectionPoolSize: 1},
 	}
 	for nameA, valueA := range identicalConfigurations {
 		for nameB, valueB := range identicalConfigurations {
@@ -88,6 +94,8 @@ func TestTLSConfigKey(t *testing.T) {
 	uniqueConfigurations := map[string]*Config{
 		"proxy":                         {Proxy: func(request *http.Request) (*url.URL, error) { return nil, nil }},
 		"no tls":                        {},
+		"connection pool of 2":          {MinConnectionPoolSize: 2},
+		"connection pool of 3":          {MinConnectionPoolSize: 3},
 		"dialer":                        {DialHolder: &DialHolder{Dial: dialer.DialContext}},
 		"dialer2":                       {DialHolder: &DialHolder{Dial: func(ctx context.Context, network, address string) (net.Conn, error) { return nil, nil }}},
 		"insecure":                      {TLS: TLSConfig{Insecure: true}},
@@ -414,11 +422,11 @@ func TestEmptyCAFileRotationLifecycle(t *testing.T) {
 	// Verify the initial state: RootCAs should be non-nil but empty
 	initialTransport := holder.getTransport(context.Background())
 
-	if initialTransport.TLSClientConfig == nil || initialTransport.TLSClientConfig.RootCAs == nil {
+	if initialTransport.GetRootCAs() == nil {
 		t.Fatal("Expected RootCAs to be non-nil for an empty CA file (should be an empty CertPool)")
 	}
 	emptyPool := x509.NewCertPool()
-	if !initialTransport.TLSClientConfig.RootCAs.Equal(emptyPool) {
+	if !initialTransport.GetRootCAs().Equal(emptyPool) {
 		t.Fatal("Expected initially empty RootCAs")
 	}
 
@@ -435,7 +443,7 @@ func TestEmptyCAFileRotationLifecycle(t *testing.T) {
 	refreshedTransport := holder.getTransport(context.Background())
 
 	// Verify the refresh succeeded and the cert pool is now populated
-	if refreshedTransport.TLSClientConfig.RootCAs.Equal(emptyPool) {
+	if refreshedTransport.GetRootCAs().Equal(emptyPool) {
 		t.Fatal("Expected RootCAs to be populated after writing valid cert data and refreshing")
 	}
 }
@@ -997,3 +1005,283 @@ func installFakeMetrics(t *testing.T) (*recordingCreateCalls, *recordingCacheGCC
 	})
 	return createCalls, gcCalls, rotationGCCalls, cacheEntries
 }
+
+// TestConnectionPoolIsCached verifies that a connection pool is cached like any
+// other transport, so that constructing clients repeatedly reuses the pool
+// instead of creating new transports and connections.
+func TestConnectionPoolIsCached(t *testing.T) {
+	for name, config := range map[string]*Config{
+		"tls":    {MinConnectionPoolSize: 3, TLS: TLSConfig{ServerName: "pool-test"}},
+		"no tls": {MinConnectionPoolSize: 3},
+	} {
+		t.Run(name, func(t *testing.T) {
+			createCalls, _, _, _ := installFakeMetrics(t)
+			cache := newTLSCache()
+
+			rt1, err := cache.get(config)
+			if err != nil {
+				t.Fatal(err)
+			}
+			rt2, err := cache.get(config)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if rt1 != rt2 {
+				t.Error("expected the same transport (cache hit)")
+			}
+			if calls := createCalls.reset(); !reflect.DeepEqual(calls, []string{"miss", "hit"}) {
+				t.Errorf("expected [miss hit], got %v", calls)
+			}
+			requireCacheLen(t, cache, 1)
+
+			pool := unwrapConnectionPool(t, rt1)
+			if len(pool.members) != 3 {
+				t.Fatalf("expected 3 members, got %d", len(pool.members))
+			}
+			seen := map[http.RoundTripper]bool{}
+			for i, m := range pool.members {
+				if _, ok := m.rt.(*rotatableCAHTTPTransport); !ok {
+					t.Errorf("member %d: expected *rotatableCAHTTPTransport, got %T", i, m.rt)
+				}
+				if seen[m.rt] {
+					t.Errorf("member %d: expected a separate transport", i)
+				}
+				seen[m.rt] = true
+			}
+		})
+	}
+}
+
+// TestConnectionPoolFeatureGateDisabled verifies that with the
+// ClientsAllowConnectionPool feature disabled, MinConnectionPoolSize is ignored:
+// the config gets the same transport as the same config without a pool.
+func TestConnectionPoolFeatureGateDisabled(t *testing.T) {
+	clientfeaturestesting.SetFeatureDuringTest(t, clientgofeaturegate.ClientsAllowConnectionPool, false)
+
+	for name, tc := range map[string]struct{ pooled, plain *Config }{
+		"tls": {
+			pooled: &Config{MinConnectionPoolSize: 3, TLS: TLSConfig{ServerName: "pool-test"}},
+			plain:  &Config{TLS: TLSConfig{ServerName: "pool-test"}},
+		},
+		"no tls": {
+			pooled: &Config{MinConnectionPoolSize: 3},
+			plain:  &Config{},
+		},
+	} {
+		t.Run(name, func(t *testing.T) {
+			cache := newTLSCache()
+			pooled, err := cache.get(tc.pooled)
+			if err != nil {
+				t.Fatal(err)
+			}
+			plain, err := cache.get(tc.plain)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if pooled != plain {
+				t.Errorf("expected the same transport as without a connection pool, got %T and %T", pooled, plain)
+			}
+		})
+	}
+}
+
+// TestConnectionPoolCARotation verifies that with CA reloading, a single
+// holder rotates the CAs of all pool members at once, and that a failed reload
+// leaves all members on the old CAs.
+func TestConnectionPoolCARotation(t *testing.T) {
+	clientfeaturestesting.SetFeatureDuringTest(t, clientgofeaturegate.ClientsAllowCARotation, true)
+
+	caFile := writeCAFile(t, []byte(testCACert1))
+	rt, err := newTLSCache().get(&Config{MinConnectionPoolSize: 3, TLS: TLSConfig{CAFile: caFile}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	holder, ok := rt.(*trackedTransport).rt.(*atomicTransportHolder)
+	if !ok {
+		t.Fatalf("expected a single atomicTransportHolder around the pool, got %T", rt.(*trackedTransport).rt)
+	}
+	clock := testingclock.NewFakeClock(time.Now())
+	holder.clock = clock
+	holder.transportLastChecked = clock.Now()
+
+	requireMembersCAs := func(t *testing.T, pool *connectionPool, want []byte) {
+		t.Helper()
+		wantPool, err := rootCertPool(want)
+		if err != nil {
+			t.Fatal(err)
+		}
+		for i, m := range pool.members {
+			if !m.rt.GetRootCAs().Equal(wantPool) {
+				t.Errorf("member %d: unexpected root CAs", i)
+			}
+		}
+	}
+	initial := holder.getTransport(t.Context()).(*connectionPool)
+	requireMembersCAs(t, initial, []byte(testCACert1))
+
+	// A failed reload leaves all members on the old CAs.
+	if err := os.WriteFile(caFile, nil, 0644); err != nil {
+		t.Fatal(err)
+	}
+	clock.Step(holder.caRefreshDuration + time.Second)
+	if got := holder.getTransport(t.Context()); got != initial {
+		t.Fatal("expected the pool not to be replaced after a failed reload")
+	}
+	requireMembersCAs(t, initial, []byte(testCACert1))
+
+	// A successful reload replaces the CAs of all members at once.
+	_, _, newCA := generateServerCertAndCA(t)
+	if err := os.WriteFile(caFile, newCA, 0644); err != nil {
+		t.Fatal(err)
+	}
+	clock.Step(holder.caRefreshDuration + time.Second)
+	rotated := holder.getTransport(t.Context()).(*connectionPool)
+	if rotated == initial {
+		t.Fatal("expected a new pool after the CA file changed")
+	}
+	if len(rotated.members) != 3 {
+		t.Fatalf("expected 3 members, got %d", len(rotated.members))
+	}
+	requireMembersCAs(t, rotated, newCA)
+	requireMembersCAs(t, initial, []byte(testCACert1))
+}
+
+// TestConnectionPoolCertRotationLeak verifies that a connection pool starts a
+// single cert rotation goroutine, shared by all members, and that it is
+// stopped when the pool is garbage collected.
+func TestConnectionPoolCertRotationLeak(t *testing.T) {
+	clientfeaturestesting.SetFeatureDuringTest(t, clientgofeaturegate.ClientsAllowTLSCacheGC, true)
+	_, gcCalls, rotationGCCalls, _ := installFakeMetrics(t)
+
+	certFile := writeCAFile(t, []byte(certData))
+	keyFile := writeCAFile(t, []byte(keyData))
+
+	cache := newTLSCache()
+
+	baseline := runtime.NumGoroutine()
+
+	rt, err := cache.get(&Config{
+		MinConnectionPoolSize: 3,
+		TLS: TLSConfig{
+			CertFile:   certFile,
+			KeyFile:    keyFile,
+			ServerName: "pool-cert-rotation-test",
+		},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	requireCacheLen(t, cache, 1)
+
+	runtime.KeepAlive(rt)
+
+	gcCalls.reset()
+	pollCacheSizeWithGC(t, cache, 0)
+
+	err = wait.PollUntilContextTimeout(t.Context(), 10*time.Millisecond, 10*time.Second, true, func(_ context.Context) (bool, error) {
+		runtime.GC()
+		return runtime.NumGoroutine() <= baseline, nil
+	})
+	if err != nil {
+		t.Errorf("goroutine leak: cert rotation goroutine was not stopped for connection pool (baseline=%d current=%d)", baseline, runtime.NumGoroutine())
+	}
+
+	if n := rotationGCCalls.count.Load(); n != 1 {
+		t.Errorf("expected TransportCertRotationGCCalls=1, got %d", n)
+	}
+	if calls := gcCalls.reset(); len(calls) != 1 || calls[0] != "deleted" {
+		t.Errorf("expected [deleted], got %v", calls)
+	}
+}
+
+// TestConnectionPoolConnections verifies that a connection pool of size N
+// opens N connections to the server, and that constructing the client again
+// for the same config reuses them.
+func TestConnectionPoolConnections(t *testing.T) {
+	t.Setenv("DISABLE_HTTP2", "")
+
+	var mu sync.Mutex
+	var conns map[string]bool
+	server := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.ProtoMajor != 2 {
+			t.Errorf("expected HTTP/2, got %s", r.Proto)
+		}
+		mu.Lock()
+		defer mu.Unlock()
+		conns[r.RemoteAddr] = true
+	}))
+	server.EnableHTTP2 = true
+	server.StartTLS()
+	defer server.Close()
+
+	for _, tc := range []struct {
+		poolSize  int
+		wantConns int
+	}{
+		{poolSize: 0, wantConns: 1},
+		{poolSize: 1, wantConns: 1},
+		{poolSize: 4, wantConns: 4},
+	} {
+		t.Run(fmt.Sprintf("pool size %d", tc.poolSize), func(t *testing.T) {
+			mu.Lock()
+			conns = map[string]bool{}
+			mu.Unlock()
+
+			cache := newTLSCache()
+			config := &Config{MinConnectionPoolSize: tc.poolSize, TLS: TLSConfig{Insecure: true}}
+
+			// Construct the client a few times, like callers that rebuild their
+			// clients, and send concurrent requests with each of them.
+			var rts []http.RoundTripper
+			for range 3 {
+				rt, err := cache.get(config)
+				if err != nil {
+					t.Fatal(err)
+				}
+				rts = append(rts, rt)
+
+				client := &http.Client{Transport: rt}
+				var wg sync.WaitGroup
+				for range 32 {
+					wg.Go(func() {
+						resp, err := client.Get(server.URL)
+						if err != nil {
+							t.Error(err)
+							return
+						}
+						_, _ = io.Copy(io.Discard, resp.Body)
+						_ = resp.Body.Close()
+					})
+				}
+				wg.Wait()
+			}
+			for _, rt := range rts {
+				if rt != rts[0] {
+					t.Error("expected the same transport for the same config")
+				}
+			}
+			utilnet.CloseIdleConnectionsFor(rts[0])
+
+			mu.Lock()
+			defer mu.Unlock()
+			if len(conns) != tc.wantConns {
+				t.Errorf("expected %d connections, got %d: %v", tc.wantConns, len(conns), conns)
+			}
+		})
+	}
+}
+
+func unwrapConnectionPool(t *testing.T, rt http.RoundTripper) *connectionPool {
+	t.Helper()
+	for {
+		switch v := rt.(type) {
+		case *connectionPool:
+			return v
+		case utilnet.RoundTripperWrapper:
+			rt = v.WrappedRoundTripper()
+		default:
+			t.Fatalf("expected a connection pool, got %T", rt)
+			return nil
+		}
+	}
+}

staging/src/k8s.io/client-go/transport/config.go

@@ -52,6 +52,15 @@ type Config struct {
 	// server.
 	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 Transport is set or if the
+	// ClientsAllowConnectionPool feature gate is disabled.
+	MinConnectionPoolSize int
+
 	// Transport may be used for custom HTTP behavior. This attribute may
 	// not be specified with the TLS client certificate options. Use
 	// WrapTransport for most client level operations.

staging/src/k8s.io/client-go/transport/connection_pool.go

@@ -0,0 +1,169 @@
+/*
+Copyright 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 transport
+
+import (
+	"crypto/x509"
+	"io"
+	"math/rand/v2"
+	"net/http"
+	"sync"
+	"sync/atomic"
+
+	utilnet "k8s.io/apimachinery/pkg/util/net"
+)
+
+// connectionPool spreads requests across a fixed set of independent
+// RoundTrippers ("members"), each of which keeps its own connections.
+//
+// For every request it picks two members at random and uses the one with
+// fewer requests in flight ("power of two choices"). A request is in flight
+// until its response body is closed or fully read, so long-running requests
+// such as watches are taken into account.
+type connectionPool struct {
+	members []*poolMember
+	// intn returns a random number in [0, n).
+	intn func(n int) int
+}
+
+type poolMember struct {
+	rt       rotatableCARoundTripper
+	inFlight atomic.Int64
+}
+
+var _ rotatableCARoundTripper = &connectionPool{}
+var _ utilnet.RoundTripperWrapper = &connectionPool{}
+
+// newConnectionPool returns a pool of the given RoundTrippers. rts must not be empty.
+func newConnectionPool(rts []rotatableCARoundTripper) *connectionPool {
+	members := make([]*poolMember, len(rts))
+	for i, rt := range rts {
+		members[i] = &poolMember{rt: rt}
+	}
+	return &connectionPool{members: members, intn: rand.IntN}
+}
+
+// pick returns the less loaded of two distinct, randomly chosen members.
+func (p *connectionPool) pick() *poolMember {
+	n := len(p.members)
+	if n == 1 {
+		return p.members[0]
+	}
+	i := p.intn(n)
+	// Pick j uniformly from the other n-1 members: draw from [0, n-1) and
+	// skip over i.
+	j := p.intn(n - 1)
+	if j >= i {
+		j++
+	}
+	a, b := p.members[i], p.members[j]
+	if b.inFlight.Load() < a.inFlight.Load() {
+		return b
+	}
+	return a
+}
+
+func (p *connectionPool) RoundTrip(req *http.Request) (*http.Response, error) {
+	m := p.pick()
+	m.inFlight.Add(1)
+	resp, err := m.rt.RoundTrip(req)
+	if err != nil || resp == nil || resp.Body == nil {
+		m.inFlight.Add(-1)
+		return resp, err
+	}
+	resp.Body = newInFlightBody(resp.Body, func() { m.inFlight.Add(-1) })
+	return resp, nil
+}
+
+func (p *connectionPool) CloseIdleConnections() {
+	for _, m := range p.members {
+		m.rt.CloseIdleConnections()
+	}
+}
+
+// Clone returns a new pool with clones of all members. The new pool starts
+// with no requests in flight.
+func (p *connectionPool) Clone() rotatableCARoundTripper {
+	rts := make([]rotatableCARoundTripper, len(p.members))
+	for i, m := range p.members {
+		rts[i] = m.rt.Clone()
+	}
+	clone := newConnectionPool(rts)
+	clone.intn = p.intn
+	return clone
+}
+
+// GetRootCAs returns the root CAs of the members, which all use the same ones.
+func (p *connectionPool) GetRootCAs() *x509.CertPool {
+	return p.members[0].rt.GetRootCAs()
+}
+
+// SetRootCAs sets the root CAs of all members.
+func (p *connectionPool) SetRootCAs(pool *x509.CertPool) {
+	for _, m := range p.members {
+		m.rt.SetRootCAs(pool)
+	}
+}
+
+func (p *connectionPool) CancelRequest(req *http.Request) {
+	for _, m := range p.members {
+		tryCancelRequest(m.rt, req)
+	}
+}
+
+// WrappedRoundTripper returns the first member. All members are configured
+// the same way, including their root CAs, which are only ever set for all
+// members at once, so it can be used to inspect e.g. the TLS config.
+func (p *connectionPool) WrappedRoundTripper() http.RoundTripper {
+	return p.members[0].rt
+}
+
+// inFlightBody calls done once the body is closed or fully read.
+type inFlightBody struct {
+	io.ReadCloser
+	once sync.Once
+	done func()
+}
+
+func (b *inFlightBody) Read(p []byte) (int, error) {
+	n, err := b.ReadCloser.Read(p)
+	if err == io.EOF {
+		b.once.Do(b.done)
+	}
+	return n, err
+}
+
+func (b *inFlightBody) Close() error {
+	err := b.ReadCloser.Close()
+	b.once.Do(b.done)
+	return err
+}
+
+// inFlightReadWriteCloser keeps the body of "101 Switching Protocols"
+// responses writable.
+type inFlightReadWriteCloser struct {
+	*inFlightBody
+	io.Writer
+}
+
+func newInFlightBody(body io.ReadCloser, done func()) io.ReadCloser {
+	b := &inFlightBody{ReadCloser: body, done: done}
+	if w, ok := body.(io.Writer); ok {
+		return &inFlightReadWriteCloser{inFlightBody: b, Writer: w}
+	}
+	return b
+}

staging/src/k8s.io/client-go/transport/connection_pool_test.go

@@ -0,0 +1,224 @@
+/*
+Copyright 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 transport
+
+import (
+	"crypto/x509"
+	"errors"
+	"io"
+	"net/http"
+	"strings"
+	"testing"
+
+	utilnet "k8s.io/apimachinery/pkg/util/net"
+)
+
+type fakePoolMemberRT struct {
+	calls       int
+	idleClosed  int
+	err         error
+	upgradeBody bool
+	rootCAs     *x509.CertPool
+}
+
+func (f *fakePoolMemberRT) Clone() rotatableCARoundTripper {
+	return &fakePoolMemberRT{err: f.err, upgradeBody: f.upgradeBody, rootCAs: f.rootCAs}
+}
+
+func (f *fakePoolMemberRT) GetRootCAs() *x509.CertPool     { return f.rootCAs }
+func (f *fakePoolMemberRT) SetRootCAs(pool *x509.CertPool) { f.rootCAs = pool }
+
+func (f *fakePoolMemberRT) RoundTrip(req *http.Request) (*http.Response, error) {
+	f.calls++
+	if f.err != nil {
+		return nil, f.err
+	}
+	if f.upgradeBody {
+		return &http.Response{StatusCode: http.StatusSwitchingProtocols, Body: &fakeReadWriteCloser{}}, nil
+	}
+	return &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader("ok"))}, nil
+}
+
+func (f *fakePoolMemberRT) CloseIdleConnections() {
+	f.idleClosed++
+}
+
+type fakeReadWriteCloser struct {
+	strings.Builder
+}
+
+func (f *fakeReadWriteCloser) Read(p []byte) (int, error) { return 0, io.EOF }
+func (f *fakeReadWriteCloser) Close() error               { return nil }
+
+func newFakePool(n int) (*connectionPool, []*fakePoolMemberRT) {
+	fakes := make([]*fakePoolMemberRT, n)
+	rts := make([]rotatableCARoundTripper, n)
+	for i := range n {
+		fakes[i] = &fakePoolMemberRT{}
+		rts[i] = fakes[i]
+	}
+	return newConnectionPool(rts), fakes
+}
+
+func newPoolTestRequest(t *testing.T) *http.Request {
+	t.Helper()
+	req, err := http.NewRequest(http.MethodGet, "https://example.com", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return req
+}
+
+func requireInFlight(t *testing.T, pool *connectionPool, want ...int64) {
+	t.Helper()
+	for i, m := range pool.members {
+		if got := m.inFlight.Load(); got != want[i] {
+			t.Fatalf("member %d: expected %d requests in flight, got %d", i, want[i], got)
+		}
+	}
+}
+
+func TestConnectionPoolPicksLessLoadedMember(t *testing.T) {
+	pool, fakes := newFakePool(3)
+	pool.members[0].inFlight.Store(2)
+	pool.members[1].inFlight.Store(1)
+	// intn always returning 0 makes the pool compare members 0 and 1.
+	pool.intn = func(int) int { return 0 }
+
+	resp, err := pool.RoundTrip(newPoolTestRequest(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = resp.Body.Close() }()
+	if fakes[1].calls != 1 {
+		t.Fatalf("expected the request to go to member 1, got calls %d/%d/%d", fakes[0].calls, fakes[1].calls, fakes[2].calls)
+	}
+	requireInFlight(t, pool, 2, 2, 0)
+}
+
+func TestConnectionPoolInFlightAccounting(t *testing.T) {
+	pool, fakes := newFakePool(2)
+	pool.intn = func(int) int { return 0 } // always pick member 0 (ties go to the first candidate)
+
+	// A request is in flight until its body is closed. Closing twice decrements once.
+	resp, err := pool.RoundTrip(newPoolTestRequest(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	requireInFlight(t, pool, 1, 0)
+	_ = resp.Body.Close()
+	_ = resp.Body.Close()
+	requireInFlight(t, pool, 0, 0)
+
+	// Reading the body to EOF also ends the request.
+	resp, err = pool.RoundTrip(newPoolTestRequest(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := io.ReadAll(resp.Body); err != nil {
+		t.Fatal(err)
+	}
+	requireInFlight(t, pool, 0, 0)
+	_ = resp.Body.Close()
+	requireInFlight(t, pool, 0, 0)
+
+	// A failed request is not in flight.
+	fakes[0].err = errors.New("connection refused")
+	if _, err := pool.RoundTrip(newPoolTestRequest(t)); err == nil {
+		t.Fatal("expected an error")
+	}
+	requireInFlight(t, pool, 0, 0)
+}
+
+func TestConnectionPoolKeepsUpgradedBodyWritable(t *testing.T) {
+	pool, fakes := newFakePool(2)
+	fakes[0].upgradeBody = true
+	fakes[1].upgradeBody = true
+
+	resp, err := pool.RoundTrip(newPoolTestRequest(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	rwc, ok := resp.Body.(io.ReadWriteCloser)
+	if !ok {
+		t.Fatalf("expected body to implement io.ReadWriteCloser, got %T", resp.Body)
+	}
+	if _, err := rwc.Write([]byte("ping")); err != nil {
+		t.Fatal(err)
+	}
+	_ = rwc.Close()
+	requireInFlight(t, pool, 0, 0)
+}
+
+func TestConnectionPoolSpreadsLongRunningRequests(t *testing.T) {
+	const members, requests = 4, 400
+	pool, _ := newFakePool(members)
+
+	// Keep all responses open, like watches.
+	for range requests {
+		if _, err := pool.RoundTrip(newPoolTestRequest(t)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	for i, m := range pool.members {
+		// Power of two choices keeps the members within a few requests of each other.
+		if got := m.inFlight.Load(); got < requests/members-10 || got > requests/members+10 {
+			t.Errorf("member %d: expected about %d requests in flight, got %d", i, requests/members, got)
+		}
+	}
+}
+
+func TestConnectionPoolCloseIdleConnections(t *testing.T) {
+	pool, fakes := newFakePool(3)
+	utilnet.CloseIdleConnectionsFor(pool)
+	for i, f := range fakes {
+		if f.idleClosed != 1 {
+			t.Errorf("member %d: expected CloseIdleConnections to be called once, got %d", i, f.idleClosed)
+		}
+	}
+}
+
+func TestConnectionPoolCloneAndSetRootCAs(t *testing.T) {
+	pool, fakes := newFakePool(3)
+	oldCAs, newCAs := x509.NewCertPool(), x509.NewCertPool()
+	pool.SetRootCAs(oldCAs)
+	pool.members[0].inFlight.Store(5)
+
+	clone := pool.Clone().(*connectionPool)
+	clone.SetRootCAs(newCAs)
+
+	if len(clone.members) != len(pool.members) {
+		t.Fatalf("expected %d members, got %d", len(pool.members), len(clone.members))
+	}
+	for i, m := range clone.members {
+		if m.rt == pool.members[i].rt {
+			t.Errorf("member %d: expected a clone, got the same member", i)
+		}
+		if m.rt.GetRootCAs() != newCAs {
+			t.Errorf("member %d of the clone: expected the new root CAs", i)
+		}
+		if m.inFlight.Load() != 0 {
+			t.Errorf("member %d of the clone: expected no requests in flight, got %d", i, m.inFlight.Load())
+		}
+		if fakes[i].rootCAs != oldCAs {
+			t.Errorf("member %d of the original pool: expected the old root CAs", i)
+		}
+	}
+	if clone.GetRootCAs() != newCAs || pool.GetRootCAs() != oldCAs {
+		t.Error("expected GetRootCAs to return the root CAs of the members")
+	}
+}

test/compatibility_lifecycle/reference/feature_list.md

@@ -31,6 +31,7 @@
 | ClearingNominatedNodeNameAfterBinding | :ballot_box_with_check: 1.35+ |  | 1.34 | 1.35– |  |  |  | [code](https://cs.k8s.io/?q=%5CbClearingNominatedNodeNameAfterBinding%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbClearingNominatedNodeNameAfterBinding%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) |
 | ClientsAllowCARotation | :ballot_box_with_check: 1.36+ |  |  | 1.36– |  |  |  | [code](https://cs.k8s.io/?q=%5CbClientsAllowCARotation%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbClientsAllowCARotation%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) |
 | ClientsAllowCBOR |  |  | 1.32– |  |  |  |  | [code](https://cs.k8s.io/?q=%5CbClientsAllowCBOR%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbClientsAllowCBOR%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) |
+| ClientsAllowConnectionPool | :ballot_box_with_check: 1.38+ |  |  | 1.38– |  |  |  | [code](https://cs.k8s.io/?q=%5CbClientsAllowConnectionPool%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbClientsAllowConnectionPool%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) |
 | ClientsAllowTLSCacheGC | :ballot_box_with_check: 1.36+ |  |  | 1.36– |  |  |  | [code](https://cs.k8s.io/?q=%5CbClientsAllowTLSCacheGC%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbClientsAllowTLSCacheGC%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) |
 | ClientsPreferCBOR |  |  | 1.32– |  |  |  |  | [code](https://cs.k8s.io/?q=%5CbClientsPreferCBOR%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbClientsPreferCBOR%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) |
 | CloudControllerManagerWatchBasedRoutesReconciliation |  |  | 1.35– |  |  |  |  | [code](https://cs.k8s.io/?q=%5CbCloudControllerManagerWatchBasedRoutesReconciliation%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/kubernetes) [KEPs](https://cs.k8s.io/?q=%5CbCloudControllerManagerWatchBasedRoutesReconciliation%5Cb&i=nope&files=&excludeFiles=CHANGELOG&repos=kubernetes/enhancements) |