Merge pull request #33733 from wojtek-t/fix_http2

Automatic merge from submit-queue

Use multiple clients in load test to avoid "no cached connections" er…

Fix #33711

pkg/client/restclient/plugin_test.go

@@ -61,7 +61,7 @@ func TestAuthPluginWrapTransport(t *testing.T) {
 		if len(tc.plugin) != 0 {
 			c.AuthProvider = &clientcmdapi.AuthProviderConfig{Name: tc.plugin}
 		}
-		tConfig, err := c.transportConfig()
+		tConfig, err := c.TransportConfig()
 		if err != nil {
 			// Unknown/bad plugins are expected to fail here.
 			if !tc.expectErr {

pkg/client/restclient/transport.go

@@ -26,7 +26,7 @@ import (
 // TLSConfigFor returns a tls.Config that will provide the transport level security defined
 // by the provided Config. Will return nil if no transport level security is requested.
 func TLSConfigFor(config *Config) (*tls.Config, error) {
-	cfg, err := config.transportConfig()
+	cfg, err := config.TransportConfig()
 	if err != nil {
 		return nil, err
 	}
@@ -37,7 +37,7 @@ func TLSConfigFor(config *Config) (*tls.Config, error) {
 // or transport level security defined by the provided Config. Will return the
 // default http.DefaultTransport if no special case behavior is needed.
 func TransportFor(config *Config) (http.RoundTripper, error) {
-	cfg, err := config.transportConfig()
+	cfg, err := config.TransportConfig()
 	if err != nil {
 		return nil, err
 	}
@@ -49,15 +49,15 @@ func TransportFor(config *Config) (http.RoundTripper, error) {
 // the underlying connection (like WebSocket or HTTP2 clients). Pure HTTP clients should use
 // the higher level TransportFor or RESTClientFor methods.
 func HTTPWrappersForConfig(config *Config, rt http.RoundTripper) (http.RoundTripper, error) {
-	cfg, err := config.transportConfig()
+	cfg, err := config.TransportConfig()
 	if err != nil {
 		return nil, err
 	}
 	return transport.HTTPWrappersForConfig(cfg, rt)
 }
 
-// transportConfig converts a client config to an appropriate transport config.
+// TransportConfig converts a client config to an appropriate transport config.
-func (c *Config) transportConfig() (*transport.Config, error) {
+func (c *Config) TransportConfig() (*transport.Config, error) {
 	wt := c.WrapTransport
 	if c.AuthProvider != nil {
 		provider, err := GetAuthProvider(c.Host, c.AuthProvider, c.AuthConfigPersister)

test/e2e/load.go

@@ -20,6 +20,8 @@ import (
 	"fmt"
 	"math"
 	"math/rand"
+	"net"
+	"net/http"
 	"os"
 	"strconv"
 	"sync"
@@ -28,9 +30,12 @@ import (
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
 	unversionedcore "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/typed/core/unversioned"
+	"k8s.io/kubernetes/pkg/client/restclient"
+	"k8s.io/kubernetes/pkg/client/transport"
 	client "k8s.io/kubernetes/pkg/client/unversioned"
 	"k8s.io/kubernetes/pkg/labels"
 	"k8s.io/kubernetes/pkg/util/intstr"
+	utilnet "k8s.io/kubernetes/pkg/util/net"
 	"k8s.io/kubernetes/test/e2e/framework"
 
 	. "github.com/onsi/ginkgo"
@@ -140,7 +145,7 @@ var _ = framework.KubeDescribe("Load capacity", func() {
 			namespaces = createNamespaces(f, nodeCount, itArg.podsPerNode)
 
 			totalPods := itArg.podsPerNode * nodeCount
-			configs = generateRCConfigs(totalPods, itArg.image, itArg.command, c, namespaces)
+			configs = generateRCConfigs(totalPods, itArg.image, itArg.command, namespaces)
 			var services []*api.Service
 			// Read the environment variable to see if we want to create services
 			createServices := os.Getenv("CREATE_SERVICES")
@@ -218,6 +223,52 @@ func createNamespaces(f *framework.Framework, nodeCount, podsPerNode int) []*api
 	return namespaces
 }
 
+func createClients(numberOfClients int) ([]*client.Client, error) {
+	clients := make([]*client.Client, numberOfClients)
+	for i := 0; i < numberOfClients; i++ {
+		config, err := framework.LoadConfig()
+		Expect(err).NotTo(HaveOccurred())
+		config.QPS = 100
+		config.Burst = 200
+		if framework.TestContext.KubeAPIContentType != "" {
+			config.ContentType = framework.TestContext.KubeAPIContentType
+		}
+
+		// For the purpose of this test, we want to force that clients
+		// do not share underlying transport (which is a default behavior
+		// in Kubernetes). Thus, we are explicitly creating transport for
+		// each client here.
+		transportConfig, err := config.TransportConfig()
+		if err != nil {
+			return nil, err
+		}
+		tlsConfig, err := transport.TLSConfigFor(transportConfig)
+		if err != nil {
+			return nil, err
+		}
+		config.Transport = utilnet.SetTransportDefaults(&http.Transport{
+			Proxy:               http.ProxyFromEnvironment,
+			TLSHandshakeTimeout: 10 * time.Second,
+			TLSClientConfig:     tlsConfig,
+			MaxIdleConnsPerHost: 100,
+			Dial: (&net.Dialer{
+				Timeout:   30 * time.Second,
+				KeepAlive: 30 * time.Second,
+			}).Dial,
+		})
+		// Overwrite TLS-related fields from config to avoid collision with
+		// Transport field.
+		config.TLSClientConfig = restclient.TLSClientConfig{}
+
+		c, err := client.New(config)
+		if err != nil {
+			return nil, err
+		}
+		clients[i] = c
+	}
+	return clients, nil
+}
+
 func computeRCCounts(total int) (int, int, int) {
 	// Small RCs owns ~0.5 of total number of pods, medium and big RCs ~0.25 each.
 	// For example for 3000 pods (100 nodes, 30 pods per node) there are:
@@ -232,22 +283,33 @@ func computeRCCounts(total int) (int, int, int) {
 	return smallRCCount, mediumRCCount, bigRCCount
 }
 
-func generateRCConfigs(totalPods int, image string, command []string, c *client.Client, nss []*api.Namespace) []*framework.RCConfig {
+func generateRCConfigs(totalPods int, image string, command []string, nss []*api.Namespace) []*framework.RCConfig {
 	configs := make([]*framework.RCConfig, 0)
 
 	smallRCCount, mediumRCCount, bigRCCount := computeRCCounts(totalPods)
-	configs = append(configs, generateRCConfigsForGroup(c, nss, smallRCGroupName, smallRCSize, smallRCCount, image, command)...)
+	configs = append(configs, generateRCConfigsForGroup(nss, smallRCGroupName, smallRCSize, smallRCCount, image, command)...)
-	configs = append(configs, generateRCConfigsForGroup(c, nss, mediumRCGroupName, mediumRCSize, mediumRCCount, image, command)...)
+	configs = append(configs, generateRCConfigsForGroup(nss, mediumRCGroupName, mediumRCSize, mediumRCCount, image, command)...)
-	configs = append(configs, generateRCConfigsForGroup(c, nss, bigRCGroupName, bigRCSize, bigRCCount, image, command)...)
+	configs = append(configs, generateRCConfigsForGroup(nss, bigRCGroupName, bigRCSize, bigRCCount, image, command)...)
+
+	// Create a number of clients to better simulate real usecase
+	// where not everyone is using exactly the same client.
+	rcsPerClient := 20
+	clients, err := createClients((len(configs) + rcsPerClient - 1) / rcsPerClient)
+	framework.ExpectNoError(err)
+
+	for i := 0; i < len(configs); i++ {
+		configs[i].Client = clients[i%len(clients)]
+	}
 
 	return configs
 }
 
-func generateRCConfigsForGroup(c *client.Client, nss []*api.Namespace, groupName string, size, count int, image string, command []string) []*framework.RCConfig {
+func generateRCConfigsForGroup(
+	nss []*api.Namespace, groupName string, size, count int, image string, command []string) []*framework.RCConfig {
 	configs := make([]*framework.RCConfig, 0, count)
 	for i := 1; i <= count; i++ {
 		config := &framework.RCConfig{
-			Client:     c,
+			Client:     nil, // this will be overwritten later
 			Name:       groupName + "-" + strconv.Itoa(i),
 			Namespace:  nss[i%len(nss)].Name,
 			Timeout:    10 * time.Minute,