Skip to content

K
k3s
Commit 1f6baa65 authored by BenTheElder's avatar BenTheElder
Browse files
Options

Move userspace code to sub-package in proxy.

Moves the userspace code in proxy to a sub-package and adds the ProxyProvider interface. This is in preparation for landing an implementation of https://github.com/GoogleCloudPlatform/kubernetes/issues/3760, which will mostly be in another sub package for iptables.
parent 991775e7
Expand all Hide whitespace changes
Inline Side-by-side
Showing with 129 additions and 71 deletions
cmd/kube-proxy/app/server.go
View file @ 1f6baa65
......@@ -29,8 +29,8 @@ import (
"k8s.io/kubernetes/pkg/client/clientcmd"
clientcmdapi "k8s.io/kubernetes/pkg/client/clientcmd/api"
"k8s.io/kubernetes/pkg/kubelet/qos"
"k8s.io/kubernetes/pkg/proxy"
"k8s.io/kubernetes/pkg/proxy/config"
"k8s.io/kubernetes/pkg/proxy/userspace"
"k8s.io/kubernetes/pkg/util"
"k8s.io/kubernetes/pkg/util/exec"
"k8s.io/kubernetes/pkg/util/iptables"
......@@ -97,8 +97,8 @@ func (s *ProxyServer) Run(_ []string) error {
if net.IP(s.BindAddress).To4() == nil {
protocol = iptables.ProtocolIpv6
}
loadBalancer := proxy.NewLoadBalancerRR()
proxier, err := proxy.NewProxier(loadBalancer, net.IP(s.BindAddress), iptables.New(exec.New(), protocol), s.PortRange)
loadBalancer := userspace.NewLoadBalancerRR()
proxier, err := userspace.NewProxier(loadBalancer, net.IP(s.BindAddress), iptables.New(exec.New(), protocol), s.PortRange)
if err != nil {
glog.Fatalf("Unable to create proxer: %v", err)
}
......
pkg/proxy/proxy_provider.go 0 → 100644
View file @ 1f6baa65
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
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 proxy
import (
"k8s.io/kubernetes/pkg/api"
)
// ProxyProvider is the interface provided by proxier implementations.
type ProxyProvider interface {
// OnUpdate manages the active set of service proxies.
// Active service proxies are reinitialized if found in the update set or
// removed if missing from the update set.
OnUpdate(services []api.Service)
// SyncLoop runs periodic work.
// This is expected to run as a goroutine or as the main loop of the app.
// It does not return.
SyncLoop()
}
pkg/proxy/loadbalancer.go → pkg/proxy/service_port_name.go
View file @ 1f6baa65
/*
Copyright 2014 The Kubernetes Authors All rights reserved.
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
......@@ -18,21 +18,10 @@ package proxy
import (
"fmt"
"net"
"k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/types"
)
// LoadBalancer is an interface for distributing incoming requests to service endpoints.
type LoadBalancer interface {
// NextEndpoint returns the endpoint to handle a request for the given
// service-port and source address.
NextEndpoint(service ServicePortName, srcAddr net.Addr) (string, error)
NewService(service ServicePortName, sessionAffinityType api.ServiceAffinity, stickyMaxAgeMinutes int) error
CleanupStaleStickySessions(service ServicePortName)
}
// ServicePortName carries a namespace + name + portname. This is the unique
// identfier for a load-balanced service.
type ServicePortName struct {
......
pkg/proxy/userspace/loadbalancer.go 0 → 100644
View file @ 1f6baa65
/*
Copyright 2014 The Kubernetes Authors All rights reserved.
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 userspace
import (
"net"
"k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/proxy"
)
// LoadBalancer is an interface for distributing incoming requests to service endpoints.
type LoadBalancer interface {
// NextEndpoint returns the endpoint to handle a request for the given
// service-port and source address.
NextEndpoint(service proxy.ServicePortName, srcAddr net.Addr) (string, error)
NewService(service proxy.ServicePortName, sessionAffinityType api.ServiceAffinity, stickyMaxAgeMinutes int) error
CleanupStaleStickySessions(service proxy.ServicePortName)
}
pkg/proxy/port_allocator.go → pkg/proxy/userspace/port_allocator.go
View file @ 1f6baa65
......@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
limitations under the License.
*/
package proxy
package userspace
import (
"errors"
......
pkg/proxy/port_allocator_test.go → pkg/proxy/userspace/port_allocator_test.go
View file @ 1f6baa65
......@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
limitations under the License.
*/
package proxy
package userspace
import (
"reflect"
......
pkg/proxy/proxier_test.go → pkg/proxy/userspace/proxier_test.go
View file @ 1f6baa65
......@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
limitations under the License.
*/
package proxy
package userspace
import (
"fmt"
......@@ -29,6 +29,7 @@ import (
"time"
"k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/proxy"
"k8s.io/kubernetes/pkg/types"
"k8s.io/kubernetes/pkg/util"
"k8s.io/kubernetes/pkg/util/iptables"
......@@ -211,7 +212,7 @@ func waitForNumProxyLoops(t *testing.T, p *Proxier, want int32) {
func TestTCPProxy(t *testing.T) {
lb := NewLoadBalancerRR()
service := ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "p"}
service := proxy.ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "p"}
lb.OnUpdate([]api.Endpoints{
{
ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
......@@ -238,7 +239,7 @@ func TestTCPProxy(t *testing.T) {
func TestUDPProxy(t *testing.T) {
lb := NewLoadBalancerRR()
service := ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "p"}
service := proxy.ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "p"}
lb.OnUpdate([]api.Endpoints{
{
ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
......@@ -265,8 +266,8 @@ func TestUDPProxy(t *testing.T) {
func TestMultiPortProxy(t *testing.T) {
lb := NewLoadBalancerRR()
serviceP := ServicePortName{types.NamespacedName{"testnamespace", "echo-p"}, "p"}
serviceQ := ServicePortName{types.NamespacedName{"testnamespace", "echo-q"}, "q"}
serviceP := proxy.ServicePortName{types.NamespacedName{"testnamespace", "echo-p"}, "p"}
serviceQ := proxy.ServicePortName{types.NamespacedName{"testnamespace", "echo-q"}, "q"}
lb.OnUpdate([]api.Endpoints{{
ObjectMeta: api.ObjectMeta{Name: serviceP.Name, Namespace: serviceP.Namespace},
Subsets: []api.EndpointSubset{{
......@@ -304,9 +305,9 @@ func TestMultiPortProxy(t *testing.T) {
func TestMultiPortOnUpdate(t *testing.T) {
lb := NewLoadBalancerRR()
serviceP := ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "p"}
serviceQ := ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "q"}
serviceX := ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "x"}
serviceP := proxy.ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "p"}
serviceQ := proxy.ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "q"}
serviceX := proxy.ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "x"}
p, err := createProxier(lb, net.ParseIP("0.0.0.0"), &fakeIptables{}, net.ParseIP("127.0.0.1"), nil)
if err != nil {
......@@ -350,7 +351,7 @@ func TestMultiPortOnUpdate(t *testing.T) {
}
// Helper: Stops the proxy for the named service.
func stopProxyByName(proxier *Proxier, service ServicePortName) error {
func stopProxyByName(proxier *Proxier, service proxy.ServicePortName) error {
info, found := proxier.getServiceInfo(service)
if !found {
return fmt.Errorf("unknown service: %s", service)
......@@ -360,7 +361,7 @@ func stopProxyByName(proxier *Proxier, service ServicePortName) error {
func TestTCPProxyStop(t *testing.T) {
lb := NewLoadBalancerRR()
service := ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "p"}
service := proxy.ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "p"}
lb.OnUpdate([]api.Endpoints{
{
ObjectMeta: api.ObjectMeta{Namespace: service.Namespace, Name: service.Name},
......@@ -398,7 +399,7 @@ func TestTCPProxyStop(t *testing.T) {
func TestUDPProxyStop(t *testing.T) {
lb := NewLoadBalancerRR()
service := ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "p"}
service := proxy.ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "p"}
lb.OnUpdate([]api.Endpoints{
{
ObjectMeta: api.ObjectMeta{Namespace: service.Namespace, Name: service.Name},
......@@ -436,7 +437,7 @@ func TestUDPProxyStop(t *testing.T) {
func TestTCPProxyUpdateDelete(t *testing.T) {
lb := NewLoadBalancerRR()
service := ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "p"}
service := proxy.ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "p"}
lb.OnUpdate([]api.Endpoints{
{
ObjectMeta: api.ObjectMeta{Namespace: service.Namespace, Name: service.Name},
......@@ -473,7 +474,7 @@ func TestTCPProxyUpdateDelete(t *testing.T) {
func TestUDPProxyUpdateDelete(t *testing.T) {
lb := NewLoadBalancerRR()
service := ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "p"}
service := proxy.ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "p"}
lb.OnUpdate([]api.Endpoints{
{
ObjectMeta: api.ObjectMeta{Namespace: service.Namespace, Name: service.Name},
......@@ -510,7 +511,7 @@ func TestUDPProxyUpdateDelete(t *testing.T) {
func TestTCPProxyUpdateDeleteUpdate(t *testing.T) {
lb := NewLoadBalancerRR()
service := ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "p"}
service := proxy.ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "p"}
lb.OnUpdate([]api.Endpoints{
{
ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
......@@ -562,7 +563,7 @@ func TestTCPProxyUpdateDeleteUpdate(t *testing.T) {
func TestUDPProxyUpdateDeleteUpdate(t *testing.T) {
lb := NewLoadBalancerRR()
service := ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "p"}
service := proxy.ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "p"}
lb.OnUpdate([]api.Endpoints{
{
ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
......@@ -614,7 +615,7 @@ func TestUDPProxyUpdateDeleteUpdate(t *testing.T) {
func TestTCPProxyUpdatePort(t *testing.T) {
lb := NewLoadBalancerRR()
service := ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "p"}
service := proxy.ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "p"}
lb.OnUpdate([]api.Endpoints{
{
ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
......@@ -662,7 +663,7 @@ func TestTCPProxyUpdatePort(t *testing.T) {
func TestUDPProxyUpdatePort(t *testing.T) {
lb := NewLoadBalancerRR()
service := ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "p"}
service := proxy.ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "p"}
lb.OnUpdate([]api.Endpoints{
{
ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
......@@ -707,7 +708,7 @@ func TestUDPProxyUpdatePort(t *testing.T) {
func TestProxyUpdatePublicIPs(t *testing.T) {
lb := NewLoadBalancerRR()
service := ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "p"}
service := proxy.ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "p"}
lb.OnUpdate([]api.Endpoints{
{
ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
......@@ -759,7 +760,7 @@ func TestProxyUpdatePublicIPs(t *testing.T) {
func TestProxyUpdatePortal(t *testing.T) {
lb := NewLoadBalancerRR()
service := ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "p"}
service := proxy.ServicePortName{types.NamespacedName{"testnamespace", "echo"}, "p"}
lb.OnUpdate([]api.Endpoints{
{
ObjectMeta: api.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
......
pkg/proxy/proxysocket.go → pkg/proxy/userspace/proxysocket.go
View file @ 1f6baa65
......@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
limitations under the License.
*/
package proxy
package userspace
import (
"fmt"
......@@ -27,6 +27,7 @@ import (
"github.com/golang/glog"
"k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/proxy"
"k8s.io/kubernetes/pkg/util"
)
......@@ -39,7 +40,7 @@ type proxySocket interface {
// while sessions are active.
Close() error
// ProxyLoop proxies incoming connections for the specified service to the service endpoints.
ProxyLoop(service ServicePortName, info *serviceInfo, proxier *Proxier)
ProxyLoop(service proxy.ServicePortName, info *serviceInfo, proxier *Proxier)
// ListenPort returns the host port that the proxySocket is listening on
ListenPort() int
}
......@@ -81,7 +82,7 @@ func (tcp *tcpProxySocket) ListenPort() int {
return tcp.port
}
func tryConnect(service ServicePortName, srcAddr net.Addr, protocol string, proxier *Proxier) (out net.Conn, err error) {
func tryConnect(service proxy.ServicePortName, srcAddr net.Addr, protocol string, proxier *Proxier) (out net.Conn, err error) {
for _, retryTimeout := range endpointDialTimeout {
endpoint, err := proxier.loadBalancer.NextEndpoint(service, srcAddr)
if err != nil {
......@@ -104,7 +105,7 @@ func tryConnect(service ServicePortName, srcAddr net.Addr, protocol string, prox
return nil, fmt.Errorf("failed to connect to an endpoint.")
}
func (tcp *tcpProxySocket) ProxyLoop(service ServicePortName, myInfo *serviceInfo, proxier *Proxier) {
func (tcp *tcpProxySocket) ProxyLoop(service proxy.ServicePortName, myInfo *serviceInfo, proxier *Proxier) {
for {
if info, exists := proxier.getServiceInfo(service); !exists || info != myInfo {
// The service port was closed or replaced.
......@@ -190,7 +191,7 @@ func newClientCache() *clientCache {
return &clientCache{clients: map[string]net.Conn{}}
}
func (udp *udpProxySocket) ProxyLoop(service ServicePortName, myInfo *serviceInfo, proxier *Proxier) {
func (udp *udpProxySocket) ProxyLoop(service proxy.ServicePortName, myInfo *serviceInfo, proxier *Proxier) {
activeClients := newClientCache()
var buffer [4096]byte // 4KiB should be enough for most whole-packets
for {
......@@ -235,7 +236,7 @@ func (udp *udpProxySocket) ProxyLoop(service ServicePortName, myInfo *serviceInf
}
}
func (udp *udpProxySocket) getBackendConn(activeClients *clientCache, cliAddr net.Addr, proxier *Proxier, service ServicePortName, timeout time.Duration) (net.Conn, error) {
func (udp *udpProxySocket) getBackendConn(activeClients *clientCache, cliAddr net.Addr, proxier *Proxier, service proxy.ServicePortName, timeout time.Duration) (net.Conn, error) {
activeClients.mu.Lock()
defer activeClients.mu.Unlock()
......
pkg/proxy/roundrobin.go → pkg/proxy/userspace/roundrobin.go
View file @ 1f6baa65
......@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
limitations under the License.
*/
package proxy
package userspace
import (
"errors"
......@@ -27,6 +27,7 @@ import (
"github.com/golang/glog"
"k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/proxy"
"k8s.io/kubernetes/pkg/types"
"k8s.io/kubernetes/pkg/util/slice"
)
......@@ -53,7 +54,7 @@ type affinityPolicy struct {
// LoadBalancerRR is a round-robin load balancer.
type LoadBalancerRR struct {
lock sync.RWMutex
services map[ServicePortName]*balancerState
services map[proxy.ServicePortName]*balancerState
}
// Ensure this implements LoadBalancer.
......@@ -76,11 +77,11 @@ func newAffinityPolicy(affinityType api.ServiceAffinity, ttlMinutes int) *affini
// NewLoadBalancerRR returns a new LoadBalancerRR.
func NewLoadBalancerRR() *LoadBalancerRR {
return &LoadBalancerRR{
services: map[ServicePortName]*balancerState{},
services: map[proxy.ServicePortName]*balancerState{},
}
}
func (lb *LoadBalancerRR) NewService(svcPort ServicePortName, affinityType api.ServiceAffinity, ttlMinutes int) error {
func (lb *LoadBalancerRR) NewService(svcPort proxy.ServicePortName, affinityType api.ServiceAffinity, ttlMinutes int) error {
lb.lock.Lock()
defer lb.lock.Unlock()
lb.newServiceInternal(svcPort, affinityType, ttlMinutes)
......@@ -88,7 +89,7 @@ func (lb *LoadBalancerRR) NewService(svcPort ServicePortName, affinityType api.S
}
// This assumes that lb.lock is already held.
func (lb *LoadBalancerRR) newServiceInternal(svcPort ServicePortName, affinityType api.ServiceAffinity, ttlMinutes int) *balancerState {
func (lb *LoadBalancerRR) newServiceInternal(svcPort proxy.ServicePortName, affinityType api.ServiceAffinity, ttlMinutes int) *balancerState {
if ttlMinutes == 0 {
ttlMinutes = 180 //default to 3 hours if not specified. Should 0 be unlimeted instead????
}
......@@ -113,7 +114,7 @@ func isSessionAffinity(affinity *affinityPolicy) bool {
// NextEndpoint returns a service endpoint.
// The service endpoint is chosen using the round-robin algorithm.
func (lb *LoadBalancerRR) NextEndpoint(svcPort ServicePortName, srcAddr net.Addr) (string, error) {
func (lb *LoadBalancerRR) NextEndpoint(svcPort proxy.ServicePortName, srcAddr net.Addr) (string, error) {
// Coarse locking is simple. We can get more fine-grained if/when we
// can prove it matters.
lb.lock.Lock()
......@@ -190,7 +191,7 @@ func flattenValidEndpoints(endpoints []hostPortPair) []string {
}
// Remove any session affinity records associated to a particular endpoint (for example when a pod goes down).
func removeSessionAffinityByEndpoint(state *balancerState, svcPort ServicePortName, endpoint string) {
func removeSessionAffinityByEndpoint(state *balancerState, svcPort proxy.ServicePortName, endpoint string) {
for _, affinity := range state.affinity.affinityMap {
if affinity.endpoint == endpoint {
glog.V(4).Infof("Removing client: %s from affinityMap for service %q", affinity.endpoint, svcPort)
......@@ -202,7 +203,7 @@ func removeSessionAffinityByEndpoint(state *balancerState, svcPort ServicePortNa
// Loop through the valid endpoints and then the endpoints associated with the Load Balancer.
// Then remove any session affinity records that are not in both lists.
// This assumes the lb.lock is held.
func (lb *LoadBalancerRR) updateAffinityMap(svcPort ServicePortName, newEndpoints []string) {
func (lb *LoadBalancerRR) updateAffinityMap(svcPort proxy.ServicePortName, newEndpoints []string) {
allEndpoints := map[string]int{}
for _, newEndpoint := range newEndpoints {
allEndpoints[newEndpoint] = 1
......@@ -226,7 +227,7 @@ func (lb *LoadBalancerRR) updateAffinityMap(svcPort ServicePortName, newEndpoint
// Registered endpoints are updated if found in the update set or
// unregistered if missing from the update set.
func (lb *LoadBalancerRR) OnUpdate(allEndpoints []api.Endpoints) {
registeredEndpoints := make(map[ServicePortName]bool)
registeredEndpoints := make(map[proxy.ServicePortName]bool)
lb.lock.Lock()
defer lb.lock.Unlock()
......@@ -250,7 +251,7 @@ func (lb *LoadBalancerRR) OnUpdate(allEndpoints []api.Endpoints) {
}
for portname := range portsToEndpoints {
svcPort := ServicePortName{types.NamespacedName{svcEndpoints.Namespace, svcEndpoints.Name}, portname}
svcPort := proxy.ServicePortName{types.NamespacedName{svcEndpoints.Namespace, svcEndpoints.Name}, portname}
state, exists := lb.services[svcPort]
curEndpoints := []string{}
if state != nil {
......@@ -294,7 +295,7 @@ func slicesEquiv(lhs, rhs []string) bool {
return false
}
func (lb *LoadBalancerRR) CleanupStaleStickySessions(svcPort ServicePortName) {
func (lb *LoadBalancerRR) CleanupStaleStickySessions(svcPort proxy.ServicePortName) {
lb.lock.Lock()
defer lb.lock.Unlock()
......
pkg/proxy/roundrobin_test.go → pkg/proxy/userspace/roundrobin_test.go
View file @ 1f6baa65
......@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
limitations under the License.
*/
package proxy
package userspace
import (
"net"
......@@ -67,7 +67,7 @@ func TestLoadBalanceFailsWithNoEndpoints(t *testing.T) {
loadBalancer := NewLoadBalancerRR()
var endpoints []api.Endpoints
loadBalancer.OnUpdate(endpoints)
service := ServicePortName{types.NamespacedName{"testnamespace", "foo"}, "does-not-exist"}
service := proxy.servicePortName{types.NamespacedName{"testnamespace", "foo"}, "does-not-exist"}
endpoint, err := loadBalancer.NextEndpoint(service, nil)
if err == nil {
t.Errorf("Didn't fail with non-existent service")
......@@ -77,7 +77,7 @@ func TestLoadBalanceFailsWithNoEndpoints(t *testing.T) {
}
}
func expectEndpoint(t *testing.T, loadBalancer *LoadBalancerRR, service ServicePortName, expected string, netaddr net.Addr) {
func expectEndpoint(t *testing.T, loadBalancer *LoadBalancerRR, service proxy.ServicePortName, expected string, netaddr net.Addr) {
endpoint, err := loadBalancer.NextEndpoint(service, netaddr)
if err != nil {
t.Errorf("Didn't find a service for %s, expected %s, failed with: %v", service, expected, err)
......@@ -89,7 +89,7 @@ func expectEndpoint(t *testing.T, loadBalancer *LoadBalancerRR, service ServiceP
func TestLoadBalanceWorksWithSingleEndpoint(t *testing.T) {
loadBalancer := NewLoadBalancerRR()
service := ServicePortName{types.NamespacedName{"testnamespace", "foo"}, "p"}
service := proxy.ServicePortName{types.NamespacedName{"testnamespace", "foo"}, "p"}
endpoint, err := loadBalancer.NextEndpoint(service, nil)
if err == nil || len(endpoint) != 0 {
t.Errorf("Didn't fail with non-existent service")
......@@ -127,7 +127,7 @@ func stringsInSlice(haystack []string, needles ...string) bool {
func TestLoadBalanceWorksWithMultipleEndpoints(t *testing.T) {
loadBalancer := NewLoadBalancerRR()
service := ServicePortName{types.NamespacedName{"testnamespace", "foo"}, "p"}
service := proxy.ServicePortName{types.NamespacedName{"testnamespace", "foo"}, "p"}
endpoint, err := loadBalancer.NextEndpoint(service, nil)
if err == nil || len(endpoint) != 0 {
t.Errorf("Didn't fail with non-existent service")
......@@ -154,8 +154,8 @@ func TestLoadBalanceWorksWithMultipleEndpoints(t *testing.T) {
func TestLoadBalanceWorksWithMultipleEndpointsMultiplePorts(t *testing.T) {
loadBalancer := NewLoadBalancerRR()
serviceP := ServicePortName{types.NamespacedName{"testnamespace", "foo"}, "p"}
serviceQ := ServicePortName{types.NamespacedName{"testnamespace", "foo"}, "q"}
serviceP := proxy.ServicePortName{types.NamespacedName{"testnamespace", "foo"}, "p"}
serviceQ := proxy.ServicePortName{types.NamespacedName{"testnamespace", "foo"}, "q"}
endpoint, err := loadBalancer.NextEndpoint(serviceP, nil)
if err == nil || len(endpoint) != 0 {
t.Errorf("Didn't fail with non-existent service")
......@@ -197,8 +197,8 @@ func TestLoadBalanceWorksWithMultipleEndpointsMultiplePorts(t *testing.T) {
func TestLoadBalanceWorksWithMultipleEndpointsAndUpdates(t *testing.T) {
loadBalancer := NewLoadBalancerRR()
serviceP := ServicePortName{types.NamespacedName{"testnamespace", "foo"}, "p"}
serviceQ := ServicePortName{types.NamespacedName{"testnamespace", "foo"}, "q"}
serviceP := proxy.ServicePortName{types.NamespacedName{"testnamespace", "foo"}, "p"}
serviceQ := proxy.ServicePortName{types.NamespacedName{"testnamespace", "foo"}, "q"}
endpoint, err := loadBalancer.NextEndpoint(serviceP, nil)
if err == nil || len(endpoint) != 0 {
t.Errorf("Didn't fail with non-existent service")
......@@ -288,8 +288,8 @@ func TestLoadBalanceWorksWithMultipleEndpointsAndUpdates(t *testing.T) {
func TestLoadBalanceWorksWithServiceRemoval(t *testing.T) {
loadBalancer := NewLoadBalancerRR()
fooServiceP := ServicePortName{types.NamespacedName{"testnamespace", "foo"}, "p"}
barServiceP := ServicePortName{types.NamespacedName{"testnamespace", "bar"}, "p"}
fooServiceP := proxy.ServicePortName{types.NamespacedName{"testnamespace", "foo"}, "p"}
barServiceP := proxy.ServicePortName{types.NamespacedName{"testnamespace", "bar"}, "p"}
endpoint, err := loadBalancer.NextEndpoint(fooServiceP, nil)
if err == nil || len(endpoint) != 0 {
t.Errorf("Didn't fail with non-existent service")
......@@ -344,7 +344,7 @@ func TestLoadBalanceWorksWithServiceRemoval(t *testing.T) {
func TestStickyLoadBalanceWorksWithNewServiceCalledFirst(t *testing.T) {
loadBalancer := NewLoadBalancerRR()
service := ServicePortName{types.NamespacedName{"testnamespace", "foo"}, ""}
service := proxy.ServicePortName{types.NamespacedName{"testnamespace", "foo"}, ""}
endpoint, err := loadBalancer.NextEndpoint(service, nil)
if err == nil || len(endpoint) != 0 {
t.Errorf("Didn't fail with non-existent service")
......@@ -401,7 +401,7 @@ func TestStickyLoadBalanceWorksWithNewServiceCalledFirst(t *testing.T) {
func TestStickyLoadBalanceWorksWithNewServiceCalledSecond(t *testing.T) {
loadBalancer := NewLoadBalancerRR()
service := ServicePortName{types.NamespacedName{"testnamespace", "foo"}, ""}
service := proxy.ServicePortName{types.NamespacedName{"testnamespace", "foo"}, ""}
endpoint, err := loadBalancer.NextEndpoint(service, nil)
if err == nil || len(endpoint) != 0 {
t.Errorf("Didn't fail with non-existent service")
......@@ -463,7 +463,7 @@ func TestStickyLoadBalanaceWorksWithMultipleEndpointsRemoveOne(t *testing.T) {
client5 := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 5), Port: 0}
client6 := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 6), Port: 0}
loadBalancer := NewLoadBalancerRR()
service := ServicePortName{types.NamespacedName{"testnamespace", "foo"}, ""}
service := proxy.ServicePortName{types.NamespacedName{"testnamespace", "foo"}, ""}
endpoint, err := loadBalancer.NextEndpoint(service, nil)
if err == nil || len(endpoint) != 0 {
t.Errorf("Didn't fail with non-existent service")
......@@ -537,7 +537,7 @@ func TestStickyLoadBalanceWorksWithMultipleEndpointsAndUpdates(t *testing.T) {
client2 := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 2), Port: 0}
client3 := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 3), Port: 0}
loadBalancer := NewLoadBalancerRR()
service := ServicePortName{types.NamespacedName{"testnamespace", "foo"}, ""}
service := proxy.ServicePortName{types.NamespacedName{"testnamespace", "foo"}, ""}
endpoint, err := loadBalancer.NextEndpoint(service, nil)
if err == nil || len(endpoint) != 0 {
t.Errorf("Didn't fail with non-existent service")
......@@ -598,7 +598,7 @@ func TestStickyLoadBalanceWorksWithServiceRemoval(t *testing.T) {
client2 := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 2), Port: 0}
client3 := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 3), Port: 0}
loadBalancer := NewLoadBalancerRR()
fooService := ServicePortName{types.NamespacedName{"testnamespace", "foo"}, ""}
fooService := proxy.ServicePortName{types.NamespacedName{"testnamespace", "foo"}, ""}
endpoint, err := loadBalancer.NextEndpoint(fooService, nil)
if err == nil || len(endpoint) != 0 {
t.Errorf("Didn't fail with non-existent service")
......@@ -614,7 +614,7 @@ func TestStickyLoadBalanceWorksWithServiceRemoval(t *testing.T) {
},
},
}
barService := ServicePortName{types.NamespacedName{"testnamespace", "bar"}, ""}
barService := proxy.ServicePortName{types.NamespacedName{"testnamespace", "bar"}, ""}
loadBalancer.NewService(barService, api.ServiceAffinityClientIP, 0)
endpoints[1] = api.Endpoints{
ObjectMeta: api.ObjectMeta{Name: barService.Name, Namespace: barService.Namespace},
......
pkg/proxy/udp_server.go → pkg/proxy/userspace/udp_server.go
View file @ 1f6baa65
......@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
limitations under the License.
*/
package proxy
package userspace
import (
"fmt"
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment