Merge pull request #50476 from caesarxuchao/plumb-proxy

cmd/kube-apiserver/app/server.go

@@ -251,7 +251,7 @@ func CreateNodeDialer(s *options.ServerRunOptions) (tunneler.Tunneler, *http.Tra
 }
 
 // CreateKubeAPIServerConfig creates all the resources for running the API server, but runs none of them
-func CreateKubeAPIServerConfig(s *options.ServerRunOptions, nodeTunneler tunneler.Tunneler, proxyTransport http.RoundTripper) (*master.Config, informers.SharedInformerFactory, clientgoinformers.SharedInformerFactory, *kubeserver.InsecureServingInfo, aggregatorapiserver.ServiceResolver, error) {
+func CreateKubeAPIServerConfig(s *options.ServerRunOptions, nodeTunneler tunneler.Tunneler, proxyTransport *http.Transport) (*master.Config, informers.SharedInformerFactory, clientgoinformers.SharedInformerFactory, *kubeserver.InsecureServingInfo, aggregatorapiserver.ServiceResolver, error) {
 	// set defaults in the options before trying to create the generic config
 	if err := defaultOptions(s); err != nil {
 		return nil, nil, nil, nil, nil, err
@@ -269,8 +269,7 @@ func CreateKubeAPIServerConfig(s *options.ServerRunOptions, nodeTunneler tunnele
 			return nil, nil, nil, nil, nil, err
 		}
 	}
-
-	genericConfig, sharedInformers, versionedInformers, insecureServingOptions, serviceResolver, err := BuildGenericConfig(s)
+	genericConfig, sharedInformers, versionedInformers, insecureServingOptions, serviceResolver, err := BuildGenericConfig(s, proxyTransport)
 	if err != nil {
 		return nil, nil, nil, nil, nil, err
 	}
@@ -354,7 +353,7 @@ func CreateKubeAPIServerConfig(s *options.ServerRunOptions, nodeTunneler tunnele
 }
 
 // BuildGenericConfig takes the master server options and produces the genericapiserver.Config associated with it
-func BuildGenericConfig(s *options.ServerRunOptions) (*genericapiserver.Config, informers.SharedInformerFactory, clientgoinformers.SharedInformerFactory, *kubeserver.InsecureServingInfo, aggregatorapiserver.ServiceResolver, error) {
+func BuildGenericConfig(s *options.ServerRunOptions, proxyTransport *http.Transport) (*genericapiserver.Config, informers.SharedInformerFactory, clientgoinformers.SharedInformerFactory, *kubeserver.InsecureServingInfo, aggregatorapiserver.ServiceResolver, error) {
 	genericConfig := genericapiserver.NewConfig(api.Codecs)
 	if err := s.GenericServerRunOptions.ApplyTo(genericConfig); err != nil {
 		return nil, nil, nil, nil, nil, err
@@ -460,6 +459,7 @@ func BuildGenericConfig(s *options.ServerRunOptions) (*genericapiserver.Config,
 		sharedInformers,
 		genericConfig.Authorizer,
 		serviceResolver,
+		proxyTransport,
 	)
 	if err != nil {
 		return nil, nil, nil, nil, nil, fmt.Errorf("failed to create admission plugin initializer: %v", err)
@@ -476,7 +476,7 @@ func BuildGenericConfig(s *options.ServerRunOptions) (*genericapiserver.Config,
 }
 
 // BuildAdmissionPluginInitializer constructs the admission plugin initializer
-func BuildAdmissionPluginInitializer(s *options.ServerRunOptions, client internalclientset.Interface, externalClient clientset.Interface, sharedInformers informers.SharedInformerFactory, apiAuthorizer authorizer.Authorizer, serviceResolver aggregatorapiserver.ServiceResolver) (admission.PluginInitializer, error) {
+func BuildAdmissionPluginInitializer(s *options.ServerRunOptions, client internalclientset.Interface, externalClient clientset.Interface, sharedInformers informers.SharedInformerFactory, apiAuthorizer authorizer.Authorizer, serviceResolver aggregatorapiserver.ServiceResolver, proxyTransport *http.Transport) (admission.PluginInitializer, error) {
 	var cloudConfig []byte
 
 	if s.CloudProvider.CloudConfigFile != "" {
@@ -510,6 +510,7 @@ func BuildAdmissionPluginInitializer(s *options.ServerRunOptions, client interna
 	}
 
 	pluginInitializer = pluginInitializer.SetServiceResolver(serviceResolver)
+	pluginInitializer = pluginInitializer.SetProxyTransport(proxyTransport)
 
 	return pluginInitializer, nil
 }

pkg/kubeapiserver/admission/initializer.go

@@ -17,6 +17,7 @@ limitations under the License.
 package admission
 
 import (
+	"net/http"
 	"net/url"
 
 	"k8s.io/apimachinery/pkg/api/meta"
@@ -88,6 +89,12 @@ type ServiceResolver interface {
 	ResolveEndpoint(namespace, name string) (*url.URL, error)
 }
 
+// WantsProxyTransport defines a fuction that accepts a proxy transport for admission
+// plugins that need to make calls to pods.
+type WantsProxyTransport interface {
+	SetProxyTransport(proxyTransport *http.Transport)
+}
+
 type PluginInitializer struct {
 	internalClient  internalclientset.Interface
 	externalClient  clientset.Interface
@@ -101,6 +108,7 @@ type PluginInitializer struct {
 	// for proving we are apiserver in call-outs
 	clientCert     []byte
 	clientKey      []byte
+	proxyTransport *http.Transport
 }
 
 var _ admission.PluginInitializer = &PluginInitializer{}
@@ -142,6 +150,12 @@ func (i *PluginInitializer) SetClientCert(cert, key []byte) *PluginInitializer {
 	return i
 }
 
+// SetProxyTransport sets the proxyTransport which is needed by some plugins.
+func (i *PluginInitializer) SetProxyTransport(proxyTransport *http.Transport) *PluginInitializer {
+	i.proxyTransport = proxyTransport
+	return i
+}
+
 // Initialize checks the initialization interfaces implemented by each plugin
 // and provide the appropriate initialization data
 func (i *PluginInitializer) Initialize(plugin admission.Interface) {
@@ -186,4 +200,8 @@ func (i *PluginInitializer) Initialize(plugin admission.Interface) {
 		}
 		wants.SetClientCert(i.clientCert, i.clientKey)
 	}
+
+	if wants, ok := plugin.(WantsProxyTransport); ok {
+		wants.SetProxyTransport(i.proxyTransport)
+	}
 }

plugin/pkg/admission/webhook/admission.go

@@ -21,6 +21,8 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"net"
+	"net/http"
 	"sync"
 	"time"
 
@@ -106,6 +108,7 @@ type GenericAdmissionWebhook struct {
 	negotiatedSerializer runtime.NegotiatedSerializer
 	clientCert           []byte
 	clientKey            []byte
+	proxyTransport       *http.Transport
 }
 
 var (
@@ -114,6 +117,10 @@ var (
 	_ = admissioninit.WantsExternalKubeClientSet(&GenericAdmissionWebhook{})
 )
 
+func (a *GenericAdmissionWebhook) SetProxyTransport(pt *http.Transport) {
+	a.proxyTransport = pt
+}
+
 func (a *GenericAdmissionWebhook) SetServiceResolver(sr admissioninit.ServiceResolver) {
 	a.serviceResolver = sr
 }
@@ -242,11 +249,17 @@ func (a *GenericAdmissionWebhook) hookClient(h *v1alpha1.ExternalAdmissionHook)
 		return nil, err
 	}
 
+	var dial func(network, addr string) (net.Conn, error)
+	if a.proxyTransport != nil && a.proxyTransport.Dial != nil {
+		dial = a.proxyTransport.Dial
+	}
+
 	// TODO: cache these instead of constructing one each time
 	cfg := &rest.Config{
 		Host:    u.Host,
 		APIPath: u.Path,
 		TLSClientConfig: rest.TLSClientConfig{
+			ServerName: h.ClientConfig.Service.Name + "." + h.ClientConfig.Service.Namespace + ".svc",
 			CAData:     h.ClientConfig.CABundle,
 			CertData:   a.clientCert,
 			KeyData:    a.clientKey,
@@ -256,6 +269,7 @@ func (a *GenericAdmissionWebhook) hookClient(h *v1alpha1.ExternalAdmissionHook)
 		ContentConfig: rest.ContentConfig{
 			NegotiatedSerializer: a.negotiatedSerializer,
 		},
+		Dial: dial,
 	}
 	return rest.UnversionedRESTClientFor(cfg)
 }

plugin/pkg/admission/webhook/admission_test.go

@@ -53,6 +53,7 @@ func (f *fakeHookSource) Run(stopCh <-chan struct{}) {}
 
 type fakeServiceResolver struct {
 	base url.URL
+	path string
 }
 
 func (f fakeServiceResolver) ResolveEndpoint(namespace, name string) (*url.URL, error) {
@@ -60,7 +61,7 @@ func (f fakeServiceResolver) ResolveEndpoint(namespace, name string) (*url.URL,
 		return nil, fmt.Errorf("couldn't resolve service location")
 	}
 	u := f.base
-	u.Path = name
+	u.Path = f.path
 	return &u, nil
 }
 
@@ -89,7 +90,6 @@ func TestAdmit(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	wh.serviceResolver = fakeServiceResolver{*serverURL}
 	wh.clientCert = clientCert
 	wh.clientKey = clientKey
 
@@ -123,17 +123,17 @@ func TestAdmit(t *testing.T) {
 
 	type test struct {
 		hookSource    fakeHookSource
+		path          string
 		expectAllow   bool
 		errorContains string
 	}
-	ccfg := func(result string) registrationv1alpha1.AdmissionHookClientConfig {
-		return registrationv1alpha1.AdmissionHookClientConfig{
+	ccfg := registrationv1alpha1.AdmissionHookClientConfig{
 		Service: registrationv1alpha1.ServiceReference{
-				Name: result,
+			Name:      "webhook-test",
+			Namespace: "default",
 		},
 		CABundle: caCert,
 	}
-	}
 	matchEverythingRules := []registrationv1alpha1.RuleWithOperations{{
 		Operations: []registrationv1alpha1.OperationType{registrationv1alpha1.OperationAll},
 		Rule: registrationv1alpha1.Rule{
@@ -148,66 +148,72 @@ func TestAdmit(t *testing.T) {
 			hookSource: fakeHookSource{
 				hooks: []registrationv1alpha1.ExternalAdmissionHook{{
 					Name:         "nomatch",
-					ClientConfig: ccfg("disallow"),
+					ClientConfig: ccfg,
 					Rules: []registrationv1alpha1.RuleWithOperations{{
 						Operations: []registrationv1alpha1.OperationType{registrationv1alpha1.Create},
 					}},
 				}},
 			},
+			path:        "disallow",
 			expectAllow: true,
 		},
 		"match & allow": {
 			hookSource: fakeHookSource{
 				hooks: []registrationv1alpha1.ExternalAdmissionHook{{
 					Name:         "allow",
-					ClientConfig: ccfg("allow"),
+					ClientConfig: ccfg,
 					Rules:        matchEverythingRules,
 				}},
 			},
+			path:        "allow",
 			expectAllow: true,
 		},
 		"match & disallow": {
 			hookSource: fakeHookSource{
 				hooks: []registrationv1alpha1.ExternalAdmissionHook{{
 					Name:         "disallow",
-					ClientConfig: ccfg("disallow"),
+					ClientConfig: ccfg,
 					Rules:        matchEverythingRules,
 				}},
 			},
+			path:          "disallow",
 			errorContains: "without explanation",
 		},
 		"match & disallow ii": {
 			hookSource: fakeHookSource{
 				hooks: []registrationv1alpha1.ExternalAdmissionHook{{
 					Name:         "disallowReason",
-					ClientConfig: ccfg("disallowReason"),
+					ClientConfig: ccfg,
 					Rules:        matchEverythingRules,
 				}},
 			},
+			path:          "disallowReason",
 			errorContains: "you shall not pass",
 		},
 		"match & fail (but allow because fail open)": {
 			hookSource: fakeHookSource{
 				hooks: []registrationv1alpha1.ExternalAdmissionHook{{
 					Name:         "internalErr A",
-					ClientConfig: ccfg("internalErr"),
+					ClientConfig: ccfg,
 					Rules:        matchEverythingRules,
 				}, {
-					Name:         "invalidReq B",
-					ClientConfig: ccfg("invalidReq"),
+					Name:         "internalErr B",
+					ClientConfig: ccfg,
 					Rules:        matchEverythingRules,
 				}, {
-					Name:         "invalidResp C",
-					ClientConfig: ccfg("invalidResp"),
+					Name:         "internalErr C",
+					ClientConfig: ccfg,
 					Rules:        matchEverythingRules,
 				}},
 			},
+			path:        "internalErr",
 			expectAllow: true,
 		},
 	}
 
 	for name, tt := range table {
 		wh.hookSource = &tt.hookSource
+		wh.serviceResolver = fakeServiceResolver{base: *serverURL, path: tt.path}
 
 		err = wh.Admit(admission.NewAttributesRecord(&object, &oldObject, kind, namespace, name, resource, subResource, operation, &userInfo))
 		if tt.expectAllow != (err == nil) {

plugin/pkg/admission/webhook/gencerts.sh

@@ -61,7 +61,7 @@ openssl req -x509 -new -nodes -key badCAKey.pem -days 100000 -out badCACert.pem
 
 # Create a server certiticate
 openssl genrsa -out serverKey.pem 2048
-openssl req -new -key serverKey.pem -out server.csr -subj "/CN=${CN_BASE}_server" -config server.conf
+openssl req -new -key serverKey.pem -out server.csr -subj "/CN=webhook-test.default.svc" -config server.conf
 openssl x509 -req -in server.csr -CA caCert.pem -CAkey caKey.pem -CAcreateserial -out serverCert.pem -days 100000 -extensions v3_req -extfile server.conf
 
 # Create a client certiticate

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

@@ -114,6 +114,9 @@ type Config struct {
 	// The maximum length of time to wait before giving up on a server request. A value of zero means no timeout.
 	Timeout time.Duration
 
+	// Dial specifies the dial function for creating unencrypted TCP connections.
+	Dial func(network, addr string) (net.Conn, error)
+
 	// Version forces a specific version to be used (if registered)
 	// Do we need this?
 	// Version string

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

@@ -18,6 +18,7 @@ package rest
 
 import (
 	"io"
+	"net"
 	"net/http"
 	"path/filepath"
 	"reflect"
@@ -236,6 +237,8 @@ func TestAnonymousConfig(t *testing.T) {
 		func(r *clientcmdapi.AuthProviderConfig, f fuzz.Continue) {
 			r.Config = map[string]string{}
 		},
+		// Dial does not require fuzzer
+		func(r *func(network, addr string) (net.Conn, error), f fuzz.Continue) {},
 	)
 	for i := 0; i < 20; i++ {
 		original := &Config{}

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

@@ -96,5 +96,6 @@ func (c *Config) TransportConfig() (*transport.Config, error) {
 			Groups:   c.Impersonate.Groups,
 			Extra:    c.Impersonate.Extra,
 		},
+		Dial: c.Dial,
 	}, nil
 }

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

@@ -63,16 +63,20 @@ func (c *tlsTransportCache) get(config *Config) (http.RoundTripper, error) {
 		return http.DefaultTransport, nil
 	}
 
+	dial := config.Dial
+	if dial == nil {
+		dial = (&net.Dialer{
+			Timeout:   30 * time.Second,
+			KeepAlive: 30 * time.Second,
+		}).Dial
+	}
 	// Cache a single transport for these options
 	c.transports[key] = utilnet.SetTransportDefaults(&http.Transport{
 		Proxy:               http.ProxyFromEnvironment,
 		TLSHandshakeTimeout: 10 * time.Second,
 		TLSClientConfig:     tlsConfig,
 		MaxIdleConnsPerHost: idleConnsPerHost,
-		Dial: (&net.Dialer{
-			Timeout:   30 * time.Second,
-			KeepAlive: 30 * time.Second,
-		}).Dial,
+		Dial:                dial,
 	})
 	return c.transports[key], nil
 }

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

@@ -16,7 +16,10 @@ limitations under the License.
 
 package transport
 
-import "net/http"
+import (
+	"net"
+	"net/http"
+)
 
 // Config holds various options for establishing a transport.
 type Config struct {
@@ -52,6 +55,9 @@ type Config struct {
 	// config may layer other RoundTrippers on top of the returned
 	// RoundTripper.
 	WrapTransport func(rt http.RoundTripper) http.RoundTripper
+
+	// Dial specifies the dial function for creating unencrypted TCP connections.
+	Dial func(network, addr string) (net.Conn, error)
 }
 
 // ImpersonationConfig has all the available impersonation options