github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-31 07:20:13 +00:00
Code Issues Projects Releases Wiki Activity

Add port to ServiceResolvers

Browse Source
This commit is contained in:
Mehdy Bohlool 2019-03-01 16:32:50 -08:00
parent 404e2f7a30
commit 11f37d757f
19 changed files with 78 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
pkg/apis/admissionregistration/validation/validation_test.go
View File

@ -377,6 +377,7 @@ func TestValidateValidatingWebhookConfiguration(t *testing.T) {
Service: &admissionregistration.ServiceReference{
Namespace: "ns",
Name: "n",
Port: 443,
},
URL: strPtr("example.com/k8s/webhook"),
},
@ -478,6 +479,7 @@ func TestValidateValidatingWebhookConfiguration(t *testing.T) {
Namespace: "ns",
Name: "n",
Path: strPtr("foo/"),
Port: 443,
},
},
},
@ -494,6 +496,7 @@ func TestValidateValidatingWebhookConfiguration(t *testing.T) {
Namespace: "ns",
Name: "n",
Path: strPtr("/"),
Port: 443,
},
},
},
@ -510,6 +513,7 @@ func TestValidateValidatingWebhookConfiguration(t *testing.T) {
Namespace: "ns",
Name: "n",
Path: strPtr("/foo"),
Port: 443,
},
},
},
@ -526,6 +530,7 @@ func TestValidateValidatingWebhookConfiguration(t *testing.T) {
Namespace: "ns",
Name: "n",
Path: strPtr("//"),
Port: 443,
},
},
},
@ -542,6 +547,7 @@ func TestValidateValidatingWebhookConfiguration(t *testing.T) {
Namespace: "ns",
Name: "n",
Path: strPtr("/foo//bar/"),
Port: 443,
},
},
},
@ -557,6 +563,7 @@ func TestValidateValidatingWebhookConfiguration(t *testing.T) {
Namespace: "ns",
Name: "n",
Path: strPtr("/foo/bar//"),
Port: 443,
},
},
},
@ -573,6 +580,7 @@ func TestValidateValidatingWebhookConfiguration(t *testing.T) {
Namespace: "ns",
Name: "n",
Path: strPtr("/apis/foo.bar/v1alpha1/--bad"),
Port: 443,
},
},
},
@ -595,7 +603,7 @@ func TestValidateValidatingWebhookConfiguration(t *testing.T) {
},
},
}, true),
expectedError: `Invalid value: 0: port must be a valid number between 1 and 65535, inclusive`,
expectedError: `Invalid value: 0: port is not valid: must be between 1 and 65535, inclusive`,
},
{
name: "invalid port >65535",
@ -613,7 +621,7 @@ func TestValidateValidatingWebhookConfiguration(t *testing.T) {
},
},
}, true),
expectedError: `Invalid value: 65536: port must be a valid number between 1 and 65535, inclusive`,
expectedError: `Invalid value: 65536: port is not valid: must be between 1 and 65535, inclusive`,
},
{
name: "timeout seconds cannot be greater than 30",

12
pkg/apis/auditregistration/validation/validation_test.go
View File

@ -155,6 +155,7 @@ func TestValidateWebhookConfiguration(t *testing.T) {
Service: &auditregistration.ServiceReference{
Namespace: "ns",
Name: "n",
Port: 443,
},
URL: strPtr("example.com/k8s/webhook"),
},
@ -223,6 +224,7 @@ func TestValidateWebhookConfiguration(t *testing.T) {
Namespace: "ns",
Name: "n",
Path: strPtr("foo/"),
Port: 443,
},
},
},
@ -240,7 +242,7 @@ func TestValidateWebhookConfiguration(t *testing.T) {
},
},
},
expectedError: `Invalid value: 65536: port must be a valid number between 1 and 65535, inclusive`,
expectedError: `Invalid value: 65536: port is not valid: must be between 1 and 65535, inclusive`,
},
{
name: "invalid port 0",
@ -254,7 +256,7 @@ func TestValidateWebhookConfiguration(t *testing.T) {
},
},
},
expectedError: `Invalid value: 0: port must be a valid number between 1 and 65535, inclusive`,
expectedError: `Invalid value: 0: port is not valid: must be between 1 and 65535, inclusive`,
},
{
name: "path accepts slash",
@ -264,6 +266,7 @@ func TestValidateWebhookConfiguration(t *testing.T) {
Namespace: "ns",
Name: "n",
Path: strPtr("/"),
Port: 443,
},
},
},
@ -277,6 +280,7 @@ func TestValidateWebhookConfiguration(t *testing.T) {
Namespace: "ns",
Name: "n",
Path: strPtr("/foo"),
Port: 443,
},
},
},
@ -290,6 +294,7 @@ func TestValidateWebhookConfiguration(t *testing.T) {
Namespace: "ns",
Name: "n",
Path: strPtr("//"),
Port: 443,
},
},
},
@ -303,6 +308,7 @@ func TestValidateWebhookConfiguration(t *testing.T) {
Namespace: "ns",
Name: "n",
Path: strPtr("/foo//bar/"),
Port: 443,
},
},
},
@ -315,6 +321,7 @@ func TestValidateWebhookConfiguration(t *testing.T) {
Namespace: "ns",
Name: "n",
Path: strPtr("/foo/bar//"),
Port: 443,
},
},
},
@ -328,6 +335,7 @@ func TestValidateWebhookConfiguration(t *testing.T) {
Namespace: "ns",
Name: "n",
Path: strPtr("/apis/foo.bar/v1alpha1/--bad"),
Port: 443,
},
},
},

1
staging/src/k8s.io/apiextensions-apiserver/pkg/apiserver/conversion/webhook_converter.go
View File

@ -75,6 +75,7 @@ func webhookClientConfigForCRD(crd *internal.CustomResourceDefinition) *webhook.
ret.Service = &webhook.ClientConfigService{
Name: apiConfig.Service.Name,
Namespace: apiConfig.Service.Namespace,
Port: apiConfig.Service.Port,
}
if apiConfig.Service.Path != nil {
ret.Service.Path = *apiConfig.Service.Path

4
staging/src/k8s.io/apiextensions-apiserver/pkg/cmd/server/options/options.go
View File

@ -129,6 +129,6 @@ type serviceResolver struct {
services v1.ServiceLister
}
func (r *serviceResolver) ResolveEndpoint(namespace, name string) (*url.URL, error) {
return proxy.ResolveCluster(r.services, namespace, name)
func (r *serviceResolver) ResolveEndpoint(namespace, name string, port int32) (*url.URL, error) {
return proxy.ResolveCluster(r.services, namespace, name, port)
}

2
staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/initializer/initializer.go
View File

@ -32,7 +32,7 @@ type WantsServiceResolver interface {
// ServiceResolver knows how to convert a service reference into an actual
// location.
type ServiceResolver interface {
ResolveEndpoint(namespace, name string) (*url.URL, error)
ResolveEndpoint(namespace, name string, port int32) (*url.URL, error)
}
// WantsAuthenticationInfoResolverWrapper defines a function that wraps the standard AuthenticationInfoResolver

2
staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/initializer/initializer_test.go
View File

@ -30,7 +30,7 @@ func (doNothingAdmission) Handles(o admission.Operation) bool { return false }
type fakeServiceResolver struct{}
func (f *fakeServiceResolver) ResolveEndpoint(namespace, name string) (*url.URL, error) {
func (f *fakeServiceResolver) ResolveEndpoint(namespace, name string, port int32) (*url.URL, error) {
return nil, nil
}

2
staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/testing/service_resolver.go
View File

@ -33,7 +33,7 @@ func NewServiceResolver(base url.URL) webhook.ServiceResolver {
return &serviceResolver{base}
}
func (f serviceResolver) ResolveEndpoint(namespace, name string) (*url.URL, error) {
func (f serviceResolver) ResolveEndpoint(namespace, name string, port int32) (*url.URL, error) {
if namespace == "failResolve" {
return nil, fmt.Errorf("couldn't resolve service location")
}

5
staging/src/k8s.io/apiserver/pkg/admission/plugin/webhook/util/client_config.go
View File

@ -34,6 +34,11 @@ func HookClientConfigForWebhook(w *v1beta1.Webhook) webhook.ClientConfig {
Name: w.ClientConfig.Service.Name,
Namespace: w.ClientConfig.Service.Namespace,
}
if w.ClientConfig.Service.Port != nil {
ret.Service.Port = *w.ClientConfig.Service.Port
} else {
ret.Service.Port = 443
}
if w.ClientConfig.Service.Path != nil {
ret.Service.Path = *w.ClientConfig.Service.Path
}

6
staging/src/k8s.io/apiserver/pkg/audit/util/conversion.go
View File

@ -35,6 +35,12 @@ func HookClientConfigForSink(a *v1alpha1.AuditSink) webhook.ClientConfig {
Name: c.Service.Name,
Namespace: c.Service.Namespace,
}
if c.Service.Port != nil {
ret.Service.Port = *c.Service.Port
} else {
ret.Service.Port = 443
}
if c.Service.Path != nil {
ret.Service.Path = *c.Service.Path
}

20
staging/src/k8s.io/apiserver/pkg/util/proxy/proxy.go
View File

@ -26,28 +26,25 @@ import (
"k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/errors"
listersv1 "k8s.io/client-go/listers/core/v1"
"k8s.io/apimachinery/pkg/util/intstr"
)
// findServicePort finds the service port by name or numerically.
func findServicePort(svc *v1.Service, port intstr.IntOrString) (*v1.ServicePort, error) {
func findServicePort(svc *v1.Service, port int32) (*v1.ServicePort, error) {
for _, svcPort := range svc.Spec.Ports {
if (port.Type == intstr.Int && int32(svcPort.Port) == port.IntVal) || (port.Type == intstr.String && svcPort.Name == port.StrVal) {
if svcPort.Port == port {
return &svcPort, nil
}
}
return nil, errors.NewServiceUnavailable(fmt.Sprintf("no service port %q found for service %q", port.String(), svc.Name))
return nil, errors.NewServiceUnavailable(fmt.Sprintf("no service port %q found for service %q", port, svc.Name))
}
// ResourceLocation returns a URL to which one can send traffic for the specified service.
func ResolveEndpoint(services listersv1.ServiceLister, endpoints listersv1.EndpointsLister, namespace, id string) (*url.URL, error) {
func ResolveEndpoint(services listersv1.ServiceLister, endpoints listersv1.EndpointsLister, namespace, id string, port int32) (*url.URL, error) {
svc, err := services.Services(namespace).Get(id)
if err != nil {
return nil, err
}
port := intstr.FromInt(443)
svcPort, err := findServicePort(svc, port)
if err != nil {
return nil, err
@ -92,14 +89,12 @@ func ResolveEndpoint(services listersv1.ServiceLister, endpoints listersv1.Endpo
return nil, errors.NewServiceUnavailable(fmt.Sprintf("no endpoints available for service %q", id))
}
func ResolveCluster(services listersv1.ServiceLister, namespace, id string) (*url.URL, error) {
func ResolveCluster(services listersv1.ServiceLister, namespace, id string, port int32) (*url.URL, error) {
svc, err := services.Services(namespace).Get(id)
if err != nil {
return nil, err
}
port := intstr.FromInt(443)
switch {
case svc.Spec.Type == v1.ServiceTypeClusterIP && svc.Spec.ClusterIP == v1.ClusterIPNone:
return nil, fmt.Errorf(`cannot route to service with ClusterIP "None"`)
@ -114,12 +109,9 @@ func ResolveCluster(services listersv1.ServiceLister, namespace, id string) (*ur
Host: net.JoinHostPort(svc.Spec.ClusterIP, fmt.Sprintf("%d", svcPort.Port)),
}, nil
case svc.Spec.Type == v1.ServiceTypeExternalName:
if port.Type != intstr.Int {
return nil, fmt.Errorf("named ports not supported")
}
return &url.URL{
Scheme: "https",
Host: net.JoinHostPort(svc.Spec.ExternalName, port.String()),
Host: net.JoinHostPort(svc.Spec.ExternalName, fmt.Sprintf("%d", port)),
}, nil
default:
return nil, fmt.Errorf("unsupported service type %q", svc.Spec.Type)

4
staging/src/k8s.io/apiserver/pkg/util/proxy/proxy_test.go
View File

@ -234,10 +234,10 @@ func TestResolve(t *testing.T) {
}
}
clusterURL, err := ResolveCluster(serviceLister, "one", "alfa")
clusterURL, err := ResolveCluster(serviceLister, "one", "alfa", 443)
check("cluster", test.clusterMode, clusterURL, err)
endpointURL, err := ResolveEndpoint(serviceLister, endpointLister, "one", "alfa")
endpointURL, err := ResolveEndpoint(serviceLister, endpointLister, "one", "alfa", 443)
check("endpoint", test.endpointMode, endpointURL, err)
}
}

7
staging/src/k8s.io/apiserver/pkg/util/webhook/client.go
View File

@ -49,6 +49,7 @@ type ClientConfigService struct {
Name string
Namespace string
Path string
Port int32
}
// ClientManager builds REST clients to talk to webhooks. It caches the clients
@ -164,7 +165,11 @@ func (cm *ClientManager) HookClient(cc ClientConfig) (*rest.RESTClient, error) {
}
cfg.Dial = func(ctx context.Context, network, addr string) (net.Conn, error) {
if addr == host {
u, err := cm.serviceResolver.ResolveEndpoint(cc.Service.Namespace, cc.Service.Name)
port := cc.Service.Port
if port == 0 {
port = 443
}
u, err := cm.serviceResolver.ResolveEndpoint(cc.Service.Namespace, cc.Service.Name, port)
if err != nil {
return nil, err
}

13
staging/src/k8s.io/apiserver/pkg/util/webhook/serviceresolver.go
View File

@ -24,7 +24,7 @@ import (
// ServiceResolver knows how to convert a service reference into an actual location.
type ServiceResolver interface {
ResolveEndpoint(namespace, name string) (*url.URL, error)
ResolveEndpoint(namespace, name string, port int32) (*url.URL, error)
}
type defaultServiceResolver struct{}
@ -35,12 +35,13 @@ func NewDefaultServiceResolver() ServiceResolver {
}
// ResolveEndpoint constructs a service URL from a given namespace and name
// note that the name and namespace are required and by default all created addresses use HTTPS scheme.
// note that the name, namespace, and port are required and by default all
// created addresses use HTTPS scheme.
// for example:
// name=ross namespace=andromeda resolves to https://ross.andromeda.svc:443
func (sr defaultServiceResolver) ResolveEndpoint(namespace, name string) (*url.URL, error) {
if len(name) == 0 || len(namespace) == 0 {
return nil, errors.New("cannot resolve an empty service name or namespace")
func (sr defaultServiceResolver) ResolveEndpoint(namespace, name string, port int32) (*url.URL, error) {
if len(name) == 0 || len(namespace) == 0 || port == 0 {
return nil, errors.New("cannot resolve an empty service name or namespace or port")
}
return &url.URL{Scheme: "https", Host: fmt.Sprintf("%s.%s.svc:443", name, namespace)}, nil
return &url.URL{Scheme: "https", Host: fmt.Sprintf("%s.%s.svc:%d", name, namespace, port)}, nil
}

7
staging/src/k8s.io/apiserver/pkg/util/webhook/serviceresolver_test.go
View File

@ -25,22 +25,25 @@ func TestDefaultServiceResolver(t *testing.T) {
scenarios := []struct {
serviceName string
serviceNamespace string
port int32
expectedOutput string
expectError bool
}{
// scenario 1: a service name along with a namespace resolves
{serviceName: "ross", serviceNamespace: "andromeda", expectedOutput: "https://ross.andromeda.svc:443"},
{serviceName: "ross", serviceNamespace: "andromeda", port: 443, expectedOutput: "https://ross.andromeda.svc:443"},
// scenario 2: a service name without a namespace does not resolve
{serviceName: "ross", expectError: true},
// scenario 3: cannot resolve an empty service name
{serviceNamespace: "andromeda", expectError: true},
// scenario 1: a service name along with a namespace and different port resolves
{serviceName: "ross", serviceNamespace: "andromeda", port: 1002, expectedOutput: "https://ross.andromeda.svc:1002"},
}
// act
for index, scenario := range scenarios {
t.Run(fmt.Sprintf("scenario %d", index), func(t *testing.T) {
target := defaultServiceResolver{}
serviceURL, err := target.ResolveEndpoint(scenario.serviceNamespace, scenario.serviceName)
serviceURL, err := target.ResolveEndpoint(scenario.serviceNamespace, scenario.serviceName, scenario.port)
if err != nil && !scenario.expectError {
t.Errorf("unexpected error has occurred = %v", err)

2
staging/src/k8s.io/apiserver/pkg/util/webhook/validation.go
View File

@ -63,7 +63,7 @@ func ValidateWebhookService(fldPath *field.Path, namespace, name string, path *s
}
if errs := validation.IsValidPortNum(int(port)); errs != nil {
allErrors = append(allErrors, field.Invalid(fldPath.Child("port"), port, "port is not valid:"+strings.Join(errs, ",")))
allErrors = append(allErrors, field.Invalid(fldPath.Child("port"), port, "port is not valid: "+strings.Join(errs, ", ")))
}
if path == nil {

2
staging/src/k8s.io/kube-aggregator/pkg/apiserver/handler_proxy.go
View File

@ -128,7 +128,7 @@ func (r *proxyHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
// write a new location based on the existing request pointed at the target service
location := &url.URL{}
location.Scheme = "https"
rloc, err := r.serviceResolver.ResolveEndpoint(handlingInfo.serviceNamespace, handlingInfo.serviceName)
rloc, err := r.serviceResolver.ResolveEndpoint(handlingInfo.serviceNamespace, handlingInfo.serviceName, 443)
if err != nil {
klog.Errorf("error resolving %s/%s: %v", handlingInfo.serviceNamespace, handlingInfo.serviceName, err)
proxyError(w, req, "service unavailable", http.StatusServiceUnavailable)

2
staging/src/k8s.io/kube-aggregator/pkg/apiserver/handler_proxy_test.go
View File

@ -84,7 +84,7 @@ type mockedRouter struct {
err error
}
func (r *mockedRouter) ResolveEndpoint(namespace, name string) (*url.URL, error) {
func (r *mockedRouter) ResolveEndpoint(namespace, name string, port int32) (*url.URL, error) {
return &url.URL{Scheme: "https", Host: r.destinationHost}, r.err
}

19
staging/src/k8s.io/kube-aggregator/pkg/apiserver/resolvers.go
View File

@ -25,7 +25,7 @@ import (
// A ServiceResolver knows how to get a URL given a service.
type ServiceResolver interface {
ResolveEndpoint(namespace, name string) (*url.URL, error)
ResolveEndpoint(namespace, name string, port int32) (*url.URL, error)
}
// NewEndpointServiceResolver returns a ServiceResolver that chooses one of the
@ -42,8 +42,8 @@ type aggregatorEndpointRouting struct {
endpoints listersv1.EndpointsLister
}
func (r *aggregatorEndpointRouting) ResolveEndpoint(namespace, name string) (*url.URL, error) {
return proxy.ResolveEndpoint(r.services, r.endpoints, namespace, name)
func (r *aggregatorEndpointRouting) ResolveEndpoint(namespace, name string, port int32) (*url.URL, error) {
return proxy.ResolveEndpoint(r.services, r.endpoints, namespace, name, port)
}
// NewClusterIPServiceResolver returns a ServiceResolver that directly calls the
@ -58,11 +58,12 @@ type aggregatorClusterRouting struct {
services listersv1.ServiceLister
}
func (r *aggregatorClusterRouting) ResolveEndpoint(namespace, name string) (*url.URL, error) {
return proxy.ResolveCluster(r.services, namespace, name)
func (r *aggregatorClusterRouting) ResolveEndpoint(namespace, name string, port int32) (*url.URL, error) {
return proxy.ResolveCluster(r.services, namespace, name, port)
}
// NewLoopbackServiceResolver returns a ServiceResolver that routes the kubernetes/default service to loopback.
// NewLoopbackServiceResolver returns a ServiceResolver that routes
// the kubernetes/default service with port 443 to loopback.
func NewLoopbackServiceResolver(delegate ServiceResolver, host *url.URL) ServiceResolver {
return &loopbackResolver{
delegate: delegate,
@ -75,9 +76,9 @@ type loopbackResolver struct {
host *url.URL
}
func (r *loopbackResolver) ResolveEndpoint(namespace, name string) (*url.URL, error) {
if namespace == "default" && name == "kubernetes" {
func (r *loopbackResolver) ResolveEndpoint(namespace, name string, port int32) (*url.URL, error) {
if namespace == "default" && name == "kubernetes" && port == 443 {
return r.host, nil
}
return r.delegate.ResolveEndpoint(namespace, name)
return r.delegate.ResolveEndpoint(namespace, name, port)
}

4
staging/src/k8s.io/kube-aggregator/pkg/controllers/status/available_controller.go
View File

@ -48,7 +48,7 @@ import (
// ServiceResolver knows how to convert a service reference into an actual location.
type ServiceResolver interface {
ResolveEndpoint(namespace, name string) (*url.URL, error)
ResolveEndpoint(namespace, name string, port int32) (*url.URL, error)
}
// AvailableConditionController handles checking the availability of registered API services.
@ -235,7 +235,7 @@ func (c *AvailableConditionController) sync(key string) error {
}
// actually try to hit the discovery endpoint when it isn't local and when we're routing as a service.
if apiService.Spec.Service != nil && c.serviceResolver != nil {
discoveryURL, err := c.serviceResolver.ResolveEndpoint(apiService.Spec.Service.Namespace, apiService.Spec.Service.Name)
discoveryURL, err := c.serviceResolver.ResolveEndpoint(apiService.Spec.Service.Namespace, apiService.Spec.Service.Name, apiService.Spec.Service.Port)
if err != nil {
return err
}