github
/
k3s
mirror of https://github.com/k3s-io/k3s
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Avoid copying endpoints object in kube-proxy

pull/6/head
Wojciech Tyczynski 2017-03-09 16:42:45 +01:00
parent 595998c43c
commit 2a6083c3e7
12 changed files with 176 additions and 160 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
pkg/kubemark/hollow_proxy.go
View File

@ -40,8 +40,8 @@ type HollowProxy struct {
type FakeProxyHandler struct{} type FakeProxyHandler struct{}
func (*FakeProxyHandler) OnServiceUpdate(services []api.Service) {} func (*FakeProxyHandler) OnServiceUpdate(services []api.Service) {}
func (*FakeProxyHandler) OnEndpointsUpdate(endpoints []api.Endpoints) {} func (*FakeProxyHandler) OnEndpointsUpdate(endpoints []*api.Endpoints) {}
type FakeProxier struct{} type FakeProxier struct{}

8
pkg/proxy/config/api_test.go
View File

@ -250,15 +250,15 @@ func (s *svcHandler) OnServiceUpdate(services []api.Service) {
type epsHandler struct { type epsHandler struct {
t *testing.T t *testing.T
expected []api.Endpoints expected []*api.Endpoints
done func() done func()
} }
func newEpsHandler(t *testing.T, eps []api.Endpoints, done func()) *epsHandler { func newEpsHandler(t *testing.T, eps []*api.Endpoints, done func()) *epsHandler {
return &epsHandler{t: t, expected: eps, done: done} return &epsHandler{t: t, expected: eps, done: done}
} }
func (e *epsHandler) OnEndpointsUpdate(endpoints []api.Endpoints) { func (e *epsHandler) OnEndpointsUpdate(endpoints []*api.Endpoints) {
defer e.done() defer e.done()
sort.Sort(sortedEndpoints(endpoints)) sort.Sort(sortedEndpoints(endpoints))
if !reflect.DeepEqual(e.expected, endpoints) { if !reflect.DeepEqual(e.expected, endpoints) {
@ -290,7 +290,7 @@ func TestInitialSync(t *testing.T) {
epsConfig := NewEndpointsConfig() epsConfig := NewEndpointsConfig()
svcHandler := newSvcHandler(t, []api.Service{*svc2, *svc1}, wg.Done) svcHandler := newSvcHandler(t, []api.Service{*svc2, *svc1}, wg.Done)
svcConfig.RegisterHandler(svcHandler) svcConfig.RegisterHandler(svcHandler)
epsHandler := newEpsHandler(t, []api.Endpoints{*eps2, *eps1}, wg.Done) epsHandler := newEpsHandler(t, []*api.Endpoints{eps2, eps1}, wg.Done)
epsConfig.RegisterHandler(epsHandler) epsConfig.RegisterHandler(epsHandler)
// Setup fake api client. // Setup fake api client.

15
pkg/proxy/config/config.go
View File

@ -63,7 +63,14 @@ type EndpointsConfigHandler interface {
// OnEndpointsUpdate gets called when endpoints configuration is changed for a given // OnEndpointsUpdate gets called when endpoints configuration is changed for a given
// service on any of the configuration sources. An example is when a new // service on any of the configuration sources. An example is when a new
// service comes up, or when containers come up or down for an existing service. // service comes up, or when containers come up or down for an existing service.
OnEndpointsUpdate(endpoints []api.Endpoints) //
// NOTE: For efficiency, endpoints are being passed by reference, thus,
// OnEndpointsUpdate should NOT modify pointers of a given slice.
// Those endpoints objects are shared with other layers of the system and
// are guaranteed to be immutable with the assumption that are also
// not mutated by those handlers. Make a deep copy if you need to modify
// them in your code.
OnEndpointsUpdate(endpoints []*api.Endpoints)
} }
// EndpointsConfig tracks a set of endpoints configurations. // EndpointsConfig tracks a set of endpoints configurations.
@ -93,7 +100,7 @@ func NewEndpointsConfig() *EndpointsConfig {
func (c *EndpointsConfig) RegisterHandler(handler EndpointsConfigHandler) { func (c *EndpointsConfig) RegisterHandler(handler EndpointsConfigHandler) {
c.bcaster.Add(config.ListenerFunc(func(instance interface{}) { c.bcaster.Add(config.ListenerFunc(func(instance interface{}) {
glog.V(3).Infof("Calling handler.OnEndpointsUpdate()") glog.V(3).Infof("Calling handler.OnEndpointsUpdate()")
handler.OnEndpointsUpdate(instance.([]api.Endpoints)) handler.OnEndpointsUpdate(instance.([]*api.Endpoints))
})) }))
} }
@ -158,10 +165,10 @@ func (s *endpointsStore) Merge(source string, change interface{}) error {
func (s *endpointsStore) MergedState() interface{} { func (s *endpointsStore) MergedState() interface{} {
s.endpointLock.RLock() s.endpointLock.RLock()
defer s.endpointLock.RUnlock() defer s.endpointLock.RUnlock()
endpoints := make([]api.Endpoints, 0) endpoints := make([]*api.Endpoints, 0)
for _, sourceEndpoints := range s.endpoints { for _, sourceEndpoints := range s.endpoints {
for _, value := range sourceEndpoints { for _, value := range sourceEndpoints {
endpoints = append(endpoints, *value) endpoints = append(endpoints, value)
} }
} }
return endpoints return endpoints

22
pkg/proxy/config/config_test.go
View File

@ -83,7 +83,7 @@ func (h *ServiceHandlerMock) ValidateServices(t *testing.T, expectedServices []a
} }
} }
type sortedEndpoints []api.Endpoints type sortedEndpoints []*api.Endpoints
func (s sortedEndpoints) Len() int { func (s sortedEndpoints) Len() int {
return len(s) return len(s)
@ -96,24 +96,24 @@ func (s sortedEndpoints) Less(i, j int) bool {
} }
type EndpointsHandlerMock struct { type EndpointsHandlerMock struct {
updated chan []api.Endpoints updated chan []*api.Endpoints
waits int waits int
} }
func NewEndpointsHandlerMock() *EndpointsHandlerMock { func NewEndpointsHandlerMock() *EndpointsHandlerMock {
return &EndpointsHandlerMock{updated: make(chan []api.Endpoints, 5)} return &EndpointsHandlerMock{updated: make(chan []*api.Endpoints, 5)}
} }
func (h *EndpointsHandlerMock) OnEndpointsUpdate(endpoints []api.Endpoints) { func (h *EndpointsHandlerMock) OnEndpointsUpdate(endpoints []*api.Endpoints) {
sort.Sort(sortedEndpoints(endpoints)) sort.Sort(sortedEndpoints(endpoints))
h.updated <- endpoints h.updated <- endpoints
} }
func (h *EndpointsHandlerMock) ValidateEndpoints(t *testing.T, expectedEndpoints []api.Endpoints) { func (h *EndpointsHandlerMock) ValidateEndpoints(t *testing.T, expectedEndpoints []*api.Endpoints) {
// We might get 1 or more updates for N endpoint updates, because we // We might get 1 or more updates for N endpoint updates, because we
// over write older snapshots of endpoints from the producer go-routine // over write older snapshots of endpoints from the producer go-routine
// if the consumer falls behind. Unittests will hard timeout in 5m. // if the consumer falls behind. Unittests will hard timeout in 5m.
var endpoints []api.Endpoints var endpoints []*api.Endpoints
for { for {
select { select {
case endpoints = <-h.updated: case endpoints = <-h.updated:
@ -254,7 +254,7 @@ func TestNewMultipleSourcesEndpointsMultipleHandlersAddedAndNotified(t *testing.
channelOne <- endpointsUpdate1 channelOne <- endpointsUpdate1
channelTwo <- endpointsUpdate2 channelTwo <- endpointsUpdate2
endpoints := []api.Endpoints{*endpointsUpdate2.Endpoints, *endpointsUpdate1.Endpoints} endpoints := []*api.Endpoints{endpointsUpdate2.Endpoints, endpointsUpdate1.Endpoints}
handler.ValidateEndpoints(t, endpoints) handler.ValidateEndpoints(t, endpoints)
handler2.ValidateEndpoints(t, endpoints) handler2.ValidateEndpoints(t, endpoints)
} }
@ -285,7 +285,7 @@ func TestNewMultipleSourcesEndpointsMultipleHandlersAddRemoveSetAndNotified(t *t
channelOne <- endpointsUpdate1 channelOne <- endpointsUpdate1
channelTwo <- endpointsUpdate2 channelTwo <- endpointsUpdate2
endpoints := []api.Endpoints{*endpointsUpdate2.Endpoints, *endpointsUpdate1.Endpoints} endpoints := []*api.Endpoints{endpointsUpdate2.Endpoints, endpointsUpdate1.Endpoints}
handler.ValidateEndpoints(t, endpoints) handler.ValidateEndpoints(t, endpoints)
handler2.ValidateEndpoints(t, endpoints) handler2.ValidateEndpoints(t, endpoints)
@ -298,7 +298,7 @@ func TestNewMultipleSourcesEndpointsMultipleHandlersAddRemoveSetAndNotified(t *t
}}, }},
}) })
channelTwo <- endpointsUpdate3 channelTwo <- endpointsUpdate3
endpoints = []api.Endpoints{*endpointsUpdate2.Endpoints, *endpointsUpdate1.Endpoints, *endpointsUpdate3.Endpoints} endpoints = []*api.Endpoints{endpointsUpdate2.Endpoints, endpointsUpdate1.Endpoints, endpointsUpdate3.Endpoints}
handler.ValidateEndpoints(t, endpoints) handler.ValidateEndpoints(t, endpoints)
handler2.ValidateEndpoints(t, endpoints) handler2.ValidateEndpoints(t, endpoints)
@ -311,7 +311,7 @@ func TestNewMultipleSourcesEndpointsMultipleHandlersAddRemoveSetAndNotified(t *t
}}, }},
}) })
channelOne <- endpointsUpdate1 channelOne <- endpointsUpdate1
endpoints = []api.Endpoints{*endpointsUpdate2.Endpoints, *endpointsUpdate1.Endpoints, *endpointsUpdate3.Endpoints} endpoints = []*api.Endpoints{endpointsUpdate2.Endpoints, endpointsUpdate1.Endpoints, endpointsUpdate3.Endpoints}
handler.ValidateEndpoints(t, endpoints) handler.ValidateEndpoints(t, endpoints)
handler2.ValidateEndpoints(t, endpoints) handler2.ValidateEndpoints(t, endpoints)
@ -319,7 +319,7 @@ func TestNewMultipleSourcesEndpointsMultipleHandlersAddRemoveSetAndNotified(t *t
endpointsUpdate2 = CreateEndpointsUpdate(REMOVE, &api.Endpoints{ObjectMeta: metav1.ObjectMeta{Namespace: "testnamespace", Name: "bar"}}) endpointsUpdate2 = CreateEndpointsUpdate(REMOVE, &api.Endpoints{ObjectMeta: metav1.ObjectMeta{Namespace: "testnamespace", Name: "bar"}})
channelTwo <- endpointsUpdate2 channelTwo <- endpointsUpdate2
endpoints = []api.Endpoints{*endpointsUpdate1.Endpoints, *endpointsUpdate3.Endpoints} endpoints = []*api.Endpoints{endpointsUpdate1.Endpoints, endpointsUpdate3.Endpoints}
handler.ValidateEndpoints(t, endpoints) handler.ValidateEndpoints(t, endpoints)
handler2.ValidateEndpoints(t, endpoints) handler2.ValidateEndpoints(t, endpoints)
} }

19
pkg/proxy/iptables/proxier.go
View File

@ -197,9 +197,14 @@ type Proxier struct {
serviceMap proxyServiceMap serviceMap proxyServiceMap
endpointsMap proxyEndpointMap endpointsMap proxyEndpointMap
portsMap map[localPort]closeable portsMap map[localPort]closeable
haveReceivedServiceUpdate bool // true once we've seen an OnServiceUpdate event haveReceivedServiceUpdate bool // true once we've seen an OnServiceUpdate event
allEndpoints []api.Endpoints // nil until we have seen an OnEndpointsUpdate event // allEndpoints should never be modified by proxier - the pointers
throttle flowcontrol.RateLimiter // are shared with higher layers of kube-proxy. They are guaranteed
// to not be modified in the meantime, but also require to be not
// modified by Proxier.
// nil until we have seen an OnEndpointsUpdate event.
allEndpoints []*api.Endpoints
throttle flowcontrol.RateLimiter
// These are effectively const and do not need the mutex to be held. // These are effectively const and do not need the mutex to be held.
syncPeriod time.Duration syncPeriod time.Duration
@ -559,7 +564,7 @@ func (proxier *Proxier) OnServiceUpdate(allServices []api.Service) {
} }
// OnEndpointsUpdate takes in a slice of updated endpoints. // OnEndpointsUpdate takes in a slice of updated endpoints.
func (proxier *Proxier) OnEndpointsUpdate(allEndpoints []api.Endpoints) { func (proxier *Proxier) OnEndpointsUpdate(allEndpoints []*api.Endpoints) {
proxier.mu.Lock() proxier.mu.Lock()
defer proxier.mu.Unlock() defer proxier.mu.Unlock()
if proxier.allEndpoints == nil { if proxier.allEndpoints == nil {
@ -580,7 +585,7 @@ func (proxier *Proxier) OnEndpointsUpdate(allEndpoints []api.Endpoints) {
} }
// Convert a slice of api.Endpoints objects into a map of service-port -> endpoints. // Convert a slice of api.Endpoints objects into a map of service-port -> endpoints.
func updateEndpoints(allEndpoints []api.Endpoints, curMap proxyEndpointMap, hostname string, func updateEndpoints(allEndpoints []*api.Endpoints, curMap proxyEndpointMap, hostname string,
healthChecker healthChecker) (newMap proxyEndpointMap, staleSet map[endpointServicePair]bool) { healthChecker healthChecker) (newMap proxyEndpointMap, staleSet map[endpointServicePair]bool) {
// return values // return values
@ -589,7 +594,7 @@ func updateEndpoints(allEndpoints []api.Endpoints, curMap proxyEndpointMap, host
// Update endpoints for services. // Update endpoints for services.
for i := range allEndpoints { for i := range allEndpoints {
accumulateEndpointsMap(&allEndpoints[i], hostname, curMap, &newMap) accumulateEndpointsMap(allEndpoints[i], hostname, curMap, &newMap)
} }
// Check stale connections against endpoints missing from the update. // Check stale connections against endpoints missing from the update.
// TODO: we should really only mark a connection stale if the proto was UDP // TODO: we should really only mark a connection stale if the proto was UDP
@ -630,6 +635,8 @@ func updateEndpoints(allEndpoints []api.Endpoints, curMap proxyEndpointMap, host
// scope - it only knows one Endpoints, but sees the whole current map. That // scope - it only knows one Endpoints, but sees the whole current map. That
// cleanup has to be done above. // cleanup has to be done above.
// //
// NOTE: endpoints object should NOT be modified.
//
// TODO: this could be simplified: // TODO: this could be simplified:
// - hostPortInfo and endpointsInfo overlap too much // - hostPortInfo and endpointsInfo overlap too much
// - the test for this is overlapped by the test for updateEndpoints // - the test for this is overlapped by the test for updateEndpoints

54
pkg/proxy/iptables/proxier_test.go
View File

@ -369,7 +369,7 @@ func NewFakeProxier(ipt utiliptables.Interface) *Proxier {
serviceMap: make(map[proxy.ServicePortName]*serviceInfo), serviceMap: make(map[proxy.ServicePortName]*serviceInfo),
iptables: ipt, iptables: ipt,
clusterCIDR: "10.0.0.0/24", clusterCIDR: "10.0.0.0/24",
allEndpoints: []api.Endpoints{}, allEndpoints: []*api.Endpoints{},
haveReceivedServiceUpdate: true, haveReceivedServiceUpdate: true,
hostname: testHostname, hostname: testHostname,
portsMap: make(map[localPort]closeable), portsMap: make(map[localPort]closeable),
@ -570,7 +570,7 @@ func TestClusterIPEndpointsJump(t *testing.T) {
ip := "10.180.0.1" ip := "10.180.0.1"
port := 80 port := 80
ep := fmt.Sprintf("%s:%d", ip, port) ep := fmt.Sprintf("%s:%d", ip, port)
allEndpoints := []api.Endpoints{ allEndpoints := []*api.Endpoints{
makeTestEndpoints("ns1", svcName, func(ept *api.Endpoints) { makeTestEndpoints("ns1", svcName, func(ept *api.Endpoints) {
ept.Subsets = []api.EndpointSubset{{ ept.Subsets = []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{ Addresses: []api.EndpointAddress{{
@ -624,7 +624,7 @@ func TestLoadBalancer(t *testing.T) {
ip := "10.180.0.1" ip := "10.180.0.1"
port := 80 port := 80
fp.allEndpoints = []api.Endpoints{ fp.allEndpoints = []*api.Endpoints{
makeTestEndpoints("ns1", svcName, func(ept *api.Endpoints) { makeTestEndpoints("ns1", svcName, func(ept *api.Endpoints) {
ept.Subsets = []api.EndpointSubset{{ ept.Subsets = []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{ Addresses: []api.EndpointAddress{{
@ -669,7 +669,7 @@ func TestNodePort(t *testing.T) {
ip := "10.180.0.1" ip := "10.180.0.1"
port := 80 port := 80
fp.allEndpoints = []api.Endpoints{ fp.allEndpoints = []*api.Endpoints{
makeTestEndpoints("ns1", svcName, func(ept *api.Endpoints) { makeTestEndpoints("ns1", svcName, func(ept *api.Endpoints) {
ept.Subsets = []api.EndpointSubset{{ ept.Subsets = []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{ Addresses: []api.EndpointAddress{{
@ -732,7 +732,7 @@ func TestOnlyLocalLoadBalancing(t *testing.T) {
port := 80 port := 80
nonLocalEp := fmt.Sprintf("%s:%d", ip1, port) nonLocalEp := fmt.Sprintf("%s:%d", ip1, port)
localEp := fmt.Sprintf("%s:%d", ip2, port) localEp := fmt.Sprintf("%s:%d", ip2, port)
allEndpoints := []api.Endpoints{ allEndpoints := []*api.Endpoints{
makeTestEndpoints("ns1", svcName, func(ept *api.Endpoints) { makeTestEndpoints("ns1", svcName, func(ept *api.Endpoints) {
ept.Subsets = []api.EndpointSubset{{ ept.Subsets = []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{ Addresses: []api.EndpointAddress{{
@ -810,7 +810,7 @@ func onlyLocalNodePorts(t *testing.T, fp *Proxier, ipt *iptablestest.FakeIPTable
port := 80 port := 80
nonLocalEp := fmt.Sprintf("%s:%d", ip1, port) nonLocalEp := fmt.Sprintf("%s:%d", ip1, port)
localEp := fmt.Sprintf("%s:%d", ip2, port) localEp := fmt.Sprintf("%s:%d", ip2, port)
allEndpoints := []api.Endpoints{ allEndpoints := []*api.Endpoints{
makeTestEndpoints("ns1", svcName, func(ept *api.Endpoints) { makeTestEndpoints("ns1", svcName, func(ept *api.Endpoints) {
ept.Subsets = []api.EndpointSubset{{ ept.Subsets = []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{ Addresses: []api.EndpointAddress{{
@ -1147,7 +1147,7 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
// This is a coarse test, but it offers some modicum of confidence as the code is evolved. // This is a coarse test, but it offers some modicum of confidence as the code is evolved.
func Test_accumulateEndpointsMap(t *testing.T) { func Test_accumulateEndpointsMap(t *testing.T) {
testCases := []struct { testCases := []struct {
newEndpoints api.Endpoints newEndpoints *api.Endpoints
oldEndpoints map[proxy.ServicePortName][]*endpointsInfo oldEndpoints map[proxy.ServicePortName][]*endpointsInfo
expectedNew map[proxy.ServicePortName][]*endpointsInfo expectedNew map[proxy.ServicePortName][]*endpointsInfo
}{{ }{{
@ -1356,7 +1356,7 @@ func Test_accumulateEndpointsMap(t *testing.T) {
for tci, tc := range testCases { for tci, tc := range testCases {
// outputs // outputs
newEndpoints := make(proxyEndpointMap) newEndpoints := make(proxyEndpointMap)
accumulateEndpointsMap(&tc.newEndpoints, "host", tc.oldEndpoints, &newEndpoints) accumulateEndpointsMap(tc.newEndpoints, "host", tc.oldEndpoints, &newEndpoints)
if len(newEndpoints) != len(tc.expectedNew) { if len(newEndpoints) != len(tc.expectedNew) {
t.Errorf("[%d] expected %d new, got %d: %v", tci, len(tc.expectedNew), len(newEndpoints), spew.Sdump(newEndpoints)) t.Errorf("[%d] expected %d new, got %d: %v", tci, len(tc.expectedNew), len(newEndpoints), spew.Sdump(newEndpoints))
@ -1375,14 +1375,14 @@ func Test_accumulateEndpointsMap(t *testing.T) {
} }
} }
func makeTestEndpoints(namespace, name string, eptFunc func(*api.Endpoints)) api.Endpoints { func makeTestEndpoints(namespace, name string, eptFunc func(*api.Endpoints)) *api.Endpoints {
ept := api.Endpoints{ ept := &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{ ObjectMeta: metav1.ObjectMeta{
Name: name, Name: name,
Namespace: namespace, Namespace: namespace,
}, },
} }
eptFunc(&ept) eptFunc(ept)
return ept return ept
} }
@ -1398,19 +1398,19 @@ func makeServicePortName(ns, name, port string) proxy.ServicePortName {
func Test_updateEndpoints(t *testing.T) { func Test_updateEndpoints(t *testing.T) {
testCases := []struct { testCases := []struct {
newEndpoints []api.Endpoints newEndpoints []*api.Endpoints
oldEndpoints map[proxy.ServicePortName][]*endpointsInfo oldEndpoints map[proxy.ServicePortName][]*endpointsInfo
expectedResult map[proxy.ServicePortName][]*endpointsInfo expectedResult map[proxy.ServicePortName][]*endpointsInfo
expectedStale []endpointServicePair expectedStale []endpointServicePair
}{{ }{{
// Case[0]: nothing // Case[0]: nothing
newEndpoints: []api.Endpoints{}, newEndpoints: []*api.Endpoints{},
oldEndpoints: map[proxy.ServicePortName][]*endpointsInfo{}, oldEndpoints: map[proxy.ServicePortName][]*endpointsInfo{},
expectedResult: map[proxy.ServicePortName][]*endpointsInfo{}, expectedResult: map[proxy.ServicePortName][]*endpointsInfo{},
expectedStale: []endpointServicePair{}, expectedStale: []endpointServicePair{},
}, { }, {
// Case[1]: no change, unnamed port // Case[1]: no change, unnamed port
newEndpoints: []api.Endpoints{ newEndpoints: []*api.Endpoints{
makeTestEndpoints("ns1", "ep1", func(ept *api.Endpoints) { makeTestEndpoints("ns1", "ep1", func(ept *api.Endpoints) {
ept.Subsets = []api.EndpointSubset{{ ept.Subsets = []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{ Addresses: []api.EndpointAddress{{
@ -1435,7 +1435,7 @@ func Test_updateEndpoints(t *testing.T) {
expectedStale: []endpointServicePair{}, expectedStale: []endpointServicePair{},
}, { }, {
// Case[2]: no change, named port // Case[2]: no change, named port
newEndpoints: []api.Endpoints{ newEndpoints: []*api.Endpoints{
makeTestEndpoints("ns1", "ep1", func(ept *api.Endpoints) { makeTestEndpoints("ns1", "ep1", func(ept *api.Endpoints) {
ept.Subsets = []api.EndpointSubset{{ ept.Subsets = []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{ Addresses: []api.EndpointAddress{{
@ -1461,7 +1461,7 @@ func Test_updateEndpoints(t *testing.T) {
expectedStale: []endpointServicePair{}, expectedStale: []endpointServicePair{},
}, { }, {
// Case[3]: no change, multiple subsets // Case[3]: no change, multiple subsets
newEndpoints: []api.Endpoints{ newEndpoints: []*api.Endpoints{
makeTestEndpoints("ns1", "ep1", func(ept *api.Endpoints) { makeTestEndpoints("ns1", "ep1", func(ept *api.Endpoints) {
ept.Subsets = []api.EndpointSubset{{ ept.Subsets = []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{ Addresses: []api.EndpointAddress{{
@ -1501,7 +1501,7 @@ func Test_updateEndpoints(t *testing.T) {
expectedStale: []endpointServicePair{}, expectedStale: []endpointServicePair{},
}, { }, {
// Case[4]: no change, multiple subsets, multiple ports // Case[4]: no change, multiple subsets, multiple ports
newEndpoints: []api.Endpoints{ newEndpoints: []*api.Endpoints{
makeTestEndpoints("ns1", "ep1", func(ept *api.Endpoints) { makeTestEndpoints("ns1", "ep1", func(ept *api.Endpoints) {
ept.Subsets = []api.EndpointSubset{{ ept.Subsets = []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{ Addresses: []api.EndpointAddress{{
@ -1550,7 +1550,7 @@ func Test_updateEndpoints(t *testing.T) {
expectedStale: []endpointServicePair{}, expectedStale: []endpointServicePair{},
}, { }, {
// Case[5]: no change, multiple endpoints, subsets, IPs, and ports // Case[5]: no change, multiple endpoints, subsets, IPs, and ports
newEndpoints: []api.Endpoints{ newEndpoints: []*api.Endpoints{
makeTestEndpoints("ns1", "ep1", func(ept *api.Endpoints) { makeTestEndpoints("ns1", "ep1", func(ept *api.Endpoints) {
ept.Subsets = []api.EndpointSubset{{ ept.Subsets = []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{ Addresses: []api.EndpointAddress{{
@ -1652,7 +1652,7 @@ func Test_updateEndpoints(t *testing.T) {
expectedStale: []endpointServicePair{}, expectedStale: []endpointServicePair{},
}, { }, {
// Case[6]: add an Endpoints // Case[6]: add an Endpoints
newEndpoints: []api.Endpoints{ newEndpoints: []*api.Endpoints{
makeTestEndpoints("ns1", "ep1", func(ept *api.Endpoints) { makeTestEndpoints("ns1", "ep1", func(ept *api.Endpoints) {
ept.Subsets = []api.EndpointSubset{{ ept.Subsets = []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{ Addresses: []api.EndpointAddress{{
@ -1673,7 +1673,7 @@ func Test_updateEndpoints(t *testing.T) {
expectedStale: []endpointServicePair{}, expectedStale: []endpointServicePair{},
}, { }, {
// Case[7]: remove an Endpoints // Case[7]: remove an Endpoints
newEndpoints: []api.Endpoints{ /* empty */ }, newEndpoints: []*api.Endpoints{ /* empty */ },
oldEndpoints: map[proxy.ServicePortName][]*endpointsInfo{ oldEndpoints: map[proxy.ServicePortName][]*endpointsInfo{
makeServicePortName("ns1", "ep1", ""): { makeServicePortName("ns1", "ep1", ""): {
{"1.1.1.1:11", false}, {"1.1.1.1:11", false},
@ -1686,7 +1686,7 @@ func Test_updateEndpoints(t *testing.T) {
}}, }},
}, { }, {
// Case[8]: add an IP and port // Case[8]: add an IP and port
newEndpoints: []api.Endpoints{ newEndpoints: []*api.Endpoints{
makeTestEndpoints("ns1", "ep1", func(ept *api.Endpoints) { makeTestEndpoints("ns1", "ep1", func(ept *api.Endpoints) {
ept.Subsets = []api.EndpointSubset{{ ept.Subsets = []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{ Addresses: []api.EndpointAddress{{
@ -1722,7 +1722,7 @@ func Test_updateEndpoints(t *testing.T) {
expectedStale: []endpointServicePair{}, expectedStale: []endpointServicePair{},
}, { }, {
// Case[9]: remove an IP and port // Case[9]: remove an IP and port
newEndpoints: []api.Endpoints{ newEndpoints: []*api.Endpoints{
makeTestEndpoints("ns1", "ep1", func(ept *api.Endpoints) { makeTestEndpoints("ns1", "ep1", func(ept *api.Endpoints) {
ept.Subsets = []api.EndpointSubset{{ ept.Subsets = []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{ Addresses: []api.EndpointAddress{{
@ -1762,7 +1762,7 @@ func Test_updateEndpoints(t *testing.T) {
}}, }},
}, { }, {
// Case[10]: add a subset // Case[10]: add a subset
newEndpoints: []api.Endpoints{ newEndpoints: []*api.Endpoints{
makeTestEndpoints("ns1", "ep1", func(ept *api.Endpoints) { makeTestEndpoints("ns1", "ep1", func(ept *api.Endpoints) {
ept.Subsets = []api.EndpointSubset{{ ept.Subsets = []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{ Addresses: []api.EndpointAddress{{
@ -1799,7 +1799,7 @@ func Test_updateEndpoints(t *testing.T) {
expectedStale: []endpointServicePair{}, expectedStale: []endpointServicePair{},
}, { }, {
// Case[11]: remove a subset // Case[11]: remove a subset
newEndpoints: []api.Endpoints{ newEndpoints: []*api.Endpoints{
makeTestEndpoints("ns1", "ep1", func(ept *api.Endpoints) { makeTestEndpoints("ns1", "ep1", func(ept *api.Endpoints) {
ept.Subsets = []api.EndpointSubset{{ ept.Subsets = []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{ Addresses: []api.EndpointAddress{{
@ -1831,7 +1831,7 @@ func Test_updateEndpoints(t *testing.T) {
}}, }},
}, { }, {
// Case[12]: rename a port // Case[12]: rename a port
newEndpoints: []api.Endpoints{ newEndpoints: []*api.Endpoints{
makeTestEndpoints("ns1", "ep1", func(ept *api.Endpoints) { makeTestEndpoints("ns1", "ep1", func(ept *api.Endpoints) {
ept.Subsets = []api.EndpointSubset{{ ept.Subsets = []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{ Addresses: []api.EndpointAddress{{
@ -1860,7 +1860,7 @@ func Test_updateEndpoints(t *testing.T) {
}}, }},
}, { }, {
// Case[13]: renumber a port // Case[13]: renumber a port
newEndpoints: []api.Endpoints{ newEndpoints: []*api.Endpoints{
makeTestEndpoints("ns1", "ep1", func(ept *api.Endpoints) { makeTestEndpoints("ns1", "ep1", func(ept *api.Endpoints) {
ept.Subsets = []api.EndpointSubset{{ ept.Subsets = []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{ Addresses: []api.EndpointAddress{{
@ -1889,7 +1889,7 @@ func Test_updateEndpoints(t *testing.T) {
}}, }},
}, { }, {
// Case[14]: complex add and remove // Case[14]: complex add and remove
newEndpoints: []api.Endpoints{ newEndpoints: []*api.Endpoints{
makeTestEndpoints("ns1", "ep1", func(ept *api.Endpoints) { makeTestEndpoints("ns1", "ep1", func(ept *api.Endpoints) {
ept.Subsets = []api.EndpointSubset{{ ept.Subsets = []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{ Addresses: []api.EndpointAddress{{

40
pkg/proxy/userspace/proxier_test.go
View File

@ -202,7 +202,7 @@ func waitForNumProxyClients(t *testing.T, s *ServiceInfo, want int, timeout time
func TestTCPProxy(t *testing.T) { func TestTCPProxy(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"}
lb.OnEndpointsUpdate([]api.Endpoints{ lb.OnEndpointsUpdate([]*api.Endpoints{
{ {
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
@ -232,7 +232,7 @@ func TestTCPProxy(t *testing.T) {
func TestUDPProxy(t *testing.T) { func TestUDPProxy(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"}
lb.OnEndpointsUpdate([]api.Endpoints{ lb.OnEndpointsUpdate([]*api.Endpoints{
{ {
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
@ -262,7 +262,7 @@ func TestUDPProxy(t *testing.T) {
func TestUDPProxyTimeout(t *testing.T) { func TestUDPProxyTimeout(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"}
lb.OnEndpointsUpdate([]api.Endpoints{ lb.OnEndpointsUpdate([]*api.Endpoints{
{ {
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
@ -297,7 +297,7 @@ func TestMultiPortProxy(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
serviceP := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo-p"}, Port: "p"} serviceP := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo-p"}, Port: "p"}
serviceQ := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo-q"}, Port: "q"} serviceQ := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo-q"}, Port: "q"}
lb.OnEndpointsUpdate([]api.Endpoints{{ lb.OnEndpointsUpdate([]*api.Endpoints{{
ObjectMeta: metav1.ObjectMeta{Name: serviceP.Name, Namespace: serviceP.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: serviceP.Name, Namespace: serviceP.Namespace},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}}, Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}},
@ -397,7 +397,7 @@ func stopProxyByName(proxier *Proxier, service proxy.ServicePortName) error {
func TestTCPProxyStop(t *testing.T) { func TestTCPProxyStop(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"}
lb.OnEndpointsUpdate([]api.Endpoints{ lb.OnEndpointsUpdate([]*api.Endpoints{
{ {
ObjectMeta: metav1.ObjectMeta{Namespace: service.Namespace, Name: service.Name}, ObjectMeta: metav1.ObjectMeta{Namespace: service.Namespace, Name: service.Name},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
@ -444,7 +444,7 @@ func TestTCPProxyStop(t *testing.T) {
func TestUDPProxyStop(t *testing.T) { func TestUDPProxyStop(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"}
lb.OnEndpointsUpdate([]api.Endpoints{ lb.OnEndpointsUpdate([]*api.Endpoints{
{ {
ObjectMeta: metav1.ObjectMeta{Namespace: service.Namespace, Name: service.Name}, ObjectMeta: metav1.ObjectMeta{Namespace: service.Namespace, Name: service.Name},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
@ -485,7 +485,7 @@ func TestUDPProxyStop(t *testing.T) {
func TestTCPProxyUpdateDelete(t *testing.T) { func TestTCPProxyUpdateDelete(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"}
lb.OnEndpointsUpdate([]api.Endpoints{ lb.OnEndpointsUpdate([]*api.Endpoints{
{ {
ObjectMeta: metav1.ObjectMeta{Namespace: service.Namespace, Name: service.Name}, ObjectMeta: metav1.ObjectMeta{Namespace: service.Namespace, Name: service.Name},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
@ -525,7 +525,7 @@ func TestTCPProxyUpdateDelete(t *testing.T) {
func TestUDPProxyUpdateDelete(t *testing.T) { func TestUDPProxyUpdateDelete(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"}
lb.OnEndpointsUpdate([]api.Endpoints{ lb.OnEndpointsUpdate([]*api.Endpoints{
{ {
ObjectMeta: metav1.ObjectMeta{Namespace: service.Namespace, Name: service.Name}, ObjectMeta: metav1.ObjectMeta{Namespace: service.Namespace, Name: service.Name},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
@ -565,14 +565,14 @@ func TestUDPProxyUpdateDelete(t *testing.T) {
func TestTCPProxyUpdateDeleteUpdate(t *testing.T) { func TestTCPProxyUpdateDeleteUpdate(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"}
endpoint := api.Endpoints{ endpoint := &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}}, Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}},
Ports: []api.EndpointPort{{Name: "p", Port: tcpServerPort}}, Ports: []api.EndpointPort{{Name: "p", Port: tcpServerPort}},
}}, }},
} }
lb.OnEndpointsUpdate([]api.Endpoints{endpoint}) lb.OnEndpointsUpdate([]*api.Endpoints{endpoint})
fexec := makeFakeExec() fexec := makeFakeExec()
@ -601,7 +601,7 @@ func TestTCPProxyUpdateDeleteUpdate(t *testing.T) {
waitForNumProxyLoops(t, p, 0) waitForNumProxyLoops(t, p, 0)
// need to add endpoint here because it got clean up during service delete // need to add endpoint here because it got clean up during service delete
lb.OnEndpointsUpdate([]api.Endpoints{endpoint}) lb.OnEndpointsUpdate([]*api.Endpoints{endpoint})
p.OnServiceUpdate([]api.Service{{ p.OnServiceUpdate([]api.Service{{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Spec: api.ServiceSpec{ClusterIP: "1.2.3.4", Ports: []api.ServicePort{{ Spec: api.ServiceSpec{ClusterIP: "1.2.3.4", Ports: []api.ServicePort{{
@ -621,14 +621,14 @@ func TestTCPProxyUpdateDeleteUpdate(t *testing.T) {
func TestUDPProxyUpdateDeleteUpdate(t *testing.T) { func TestUDPProxyUpdateDeleteUpdate(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"}
endpoint := api.Endpoints{ endpoint := &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}}, Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}},
Ports: []api.EndpointPort{{Name: "p", Port: udpServerPort}}, Ports: []api.EndpointPort{{Name: "p", Port: udpServerPort}},
}}, }},
} }
lb.OnEndpointsUpdate([]api.Endpoints{endpoint}) lb.OnEndpointsUpdate([]*api.Endpoints{endpoint})
fexec := makeFakeExec() fexec := makeFakeExec()
@ -657,7 +657,7 @@ func TestUDPProxyUpdateDeleteUpdate(t *testing.T) {
waitForNumProxyLoops(t, p, 0) waitForNumProxyLoops(t, p, 0)
// need to add endpoint here because it got clean up during service delete // need to add endpoint here because it got clean up during service delete
lb.OnEndpointsUpdate([]api.Endpoints{endpoint}) lb.OnEndpointsUpdate([]*api.Endpoints{endpoint})
p.OnServiceUpdate([]api.Service{{ p.OnServiceUpdate([]api.Service{{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Spec: api.ServiceSpec{ClusterIP: "1.2.3.4", Ports: []api.ServicePort{{ Spec: api.ServiceSpec{ClusterIP: "1.2.3.4", Ports: []api.ServicePort{{
@ -677,7 +677,7 @@ func TestUDPProxyUpdateDeleteUpdate(t *testing.T) {
func TestTCPProxyUpdatePort(t *testing.T) { func TestTCPProxyUpdatePort(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"}
lb.OnEndpointsUpdate([]api.Endpoints{ lb.OnEndpointsUpdate([]*api.Endpoints{
{ {
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
@ -728,7 +728,7 @@ func TestTCPProxyUpdatePort(t *testing.T) {
func TestUDPProxyUpdatePort(t *testing.T) { func TestUDPProxyUpdatePort(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"}
lb.OnEndpointsUpdate([]api.Endpoints{ lb.OnEndpointsUpdate([]*api.Endpoints{
{ {
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
@ -776,7 +776,7 @@ func TestUDPProxyUpdatePort(t *testing.T) {
func TestProxyUpdatePublicIPs(t *testing.T) { func TestProxyUpdatePublicIPs(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"}
lb.OnEndpointsUpdate([]api.Endpoints{ lb.OnEndpointsUpdate([]*api.Endpoints{
{ {
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
@ -831,14 +831,14 @@ func TestProxyUpdatePublicIPs(t *testing.T) {
func TestProxyUpdatePortal(t *testing.T) { func TestProxyUpdatePortal(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"}
endpoint := api.Endpoints{ endpoint := &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}}, Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}},
Ports: []api.EndpointPort{{Name: "p", Port: tcpServerPort}}, Ports: []api.EndpointPort{{Name: "p", Port: tcpServerPort}},
}}, }},
} }
lb.OnEndpointsUpdate([]api.Endpoints{endpoint}) lb.OnEndpointsUpdate([]*api.Endpoints{endpoint})
fexec := makeFakeExec() fexec := makeFakeExec()
@ -890,7 +890,7 @@ func TestProxyUpdatePortal(t *testing.T) {
Protocol: "TCP", Protocol: "TCP",
}}}, }}},
}}) }})
lb.OnEndpointsUpdate([]api.Endpoints{endpoint}) lb.OnEndpointsUpdate([]*api.Endpoints{endpoint})
svcInfo, exists = p.getServiceInfo(service) svcInfo, exists = p.getServiceInfo(service)
if !exists { if !exists {
t.Fatalf("service with ClusterIP set not found in the proxy") t.Fatalf("service with ClusterIP set not found in the proxy")

5
pkg/proxy/userspace/roundrobin.go
View File

@ -246,14 +246,15 @@ func (lb *LoadBalancerRR) updateAffinityMap(svcPort proxy.ServicePortName, newEn
// OnEndpointsUpdate manages the registered service endpoints. // OnEndpointsUpdate manages the registered service endpoints.
// Registered endpoints are updated if found in the update set or // Registered endpoints are updated if found in the update set or
// unregistered if missing from the update set. // unregistered if missing from the update set.
func (lb *LoadBalancerRR) OnEndpointsUpdate(allEndpoints []api.Endpoints) { func (lb *LoadBalancerRR) OnEndpointsUpdate(allEndpoints []*api.Endpoints) {
registeredEndpoints := make(map[proxy.ServicePortName]bool) registeredEndpoints := make(map[proxy.ServicePortName]bool)
lb.lock.Lock() lb.lock.Lock()
defer lb.lock.Unlock() defer lb.lock.Unlock()
// Update endpoints for services. // Update endpoints for services.
for i := range allEndpoints { for i := range allEndpoints {
svcEndpoints := &allEndpoints[i] // svcEndpoints object should NOT be modified.
svcEndpoints := allEndpoints[i]
// We need to build a map of portname -> all ip:ports for that // We need to build a map of portname -> all ip:ports for that
// portname. Explode Endpoints.Subsets[*] into this structure. // portname. Explode Endpoints.Subsets[*] into this structure.

62
pkg/proxy/userspace/roundrobin_test.go
View File

@ -67,7 +67,7 @@ func TestFilterWorks(t *testing.T) {
func TestLoadBalanceFailsWithNoEndpoints(t *testing.T) { func TestLoadBalanceFailsWithNoEndpoints(t *testing.T) {
loadBalancer := NewLoadBalancerRR() loadBalancer := NewLoadBalancerRR()
var endpoints []api.Endpoints var endpoints []*api.Endpoints
loadBalancer.OnEndpointsUpdate(endpoints) loadBalancer.OnEndpointsUpdate(endpoints)
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "foo"}, Port: "does-not-exist"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "foo"}, Port: "does-not-exist"}
endpoint, err := loadBalancer.NextEndpoint(service, nil, false) endpoint, err := loadBalancer.NextEndpoint(service, nil, false)
@ -106,8 +106,8 @@ func TestLoadBalanceWorksWithSingleEndpoint(t *testing.T) {
if err == nil || len(endpoint) != 0 { if err == nil || len(endpoint) != 0 {
t.Errorf("Didn't fail with non-existent service") t.Errorf("Didn't fail with non-existent service")
} }
endpoints := make([]api.Endpoints, 1) endpoints := make([]*api.Endpoints, 1)
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{IP: "endpoint1"}}, Addresses: []api.EndpointAddress{{IP: "endpoint1"}},
@ -144,8 +144,8 @@ func TestLoadBalanceWorksWithMultipleEndpoints(t *testing.T) {
if err == nil || len(endpoint) != 0 { if err == nil || len(endpoint) != 0 {
t.Errorf("Didn't fail with non-existent service") t.Errorf("Didn't fail with non-existent service")
} }
endpoints := make([]api.Endpoints, 1) endpoints := make([]*api.Endpoints, 1)
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{IP: "endpoint"}}, Addresses: []api.EndpointAddress{{IP: "endpoint"}},
@ -172,8 +172,8 @@ func TestLoadBalanceWorksWithMultipleEndpointsMultiplePorts(t *testing.T) {
if err == nil || len(endpoint) != 0 { if err == nil || len(endpoint) != 0 {
t.Errorf("Didn't fail with non-existent service") t.Errorf("Didn't fail with non-existent service")
} }
endpoints := make([]api.Endpoints, 1) endpoints := make([]*api.Endpoints, 1)
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: serviceP.Name, Namespace: serviceP.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: serviceP.Name, Namespace: serviceP.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{ {
@ -215,8 +215,8 @@ func TestLoadBalanceWorksWithMultipleEndpointsAndUpdates(t *testing.T) {
if err == nil || len(endpoint) != 0 { if err == nil || len(endpoint) != 0 {
t.Errorf("Didn't fail with non-existent service") t.Errorf("Didn't fail with non-existent service")
} }
endpoints := make([]api.Endpoints, 1) endpoints := make([]*api.Endpoints, 1)
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: serviceP.Name, Namespace: serviceP.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: serviceP.Name, Namespace: serviceP.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{ {
@ -255,7 +255,7 @@ func TestLoadBalanceWorksWithMultipleEndpointsAndUpdates(t *testing.T) {
// Then update the configuration with one fewer endpoints, make sure // Then update the configuration with one fewer endpoints, make sure
// we start in the beginning again // we start in the beginning again
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: serviceP.Name, Namespace: serviceP.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: serviceP.Name, Namespace: serviceP.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{ {
@ -289,7 +289,7 @@ func TestLoadBalanceWorksWithMultipleEndpointsAndUpdates(t *testing.T) {
expectEndpoint(t, loadBalancer, serviceQ, shuffledEndpoints[1], nil) expectEndpoint(t, loadBalancer, serviceQ, shuffledEndpoints[1], nil)
// Clear endpoints // Clear endpoints
endpoints[0] = api.Endpoints{ObjectMeta: metav1.ObjectMeta{Name: serviceP.Name, Namespace: serviceP.Namespace}, Subsets: nil} endpoints[0] = &api.Endpoints{ObjectMeta: metav1.ObjectMeta{Name: serviceP.Name, Namespace: serviceP.Namespace}, Subsets: nil}
loadBalancer.OnEndpointsUpdate(endpoints) loadBalancer.OnEndpointsUpdate(endpoints)
endpoint, err = loadBalancer.NextEndpoint(serviceP, nil, false) endpoint, err = loadBalancer.NextEndpoint(serviceP, nil, false)
@ -306,8 +306,8 @@ func TestLoadBalanceWorksWithServiceRemoval(t *testing.T) {
if err == nil || len(endpoint) != 0 { if err == nil || len(endpoint) != 0 {
t.Errorf("Didn't fail with non-existent service") t.Errorf("Didn't fail with non-existent service")
} }
endpoints := make([]api.Endpoints, 2) endpoints := make([]*api.Endpoints, 2)
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: fooServiceP.Name, Namespace: fooServiceP.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: fooServiceP.Name, Namespace: fooServiceP.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{ {
@ -316,7 +316,7 @@ func TestLoadBalanceWorksWithServiceRemoval(t *testing.T) {
}, },
}, },
} }
endpoints[1] = api.Endpoints{ endpoints[1] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: barServiceP.Name, Namespace: barServiceP.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: barServiceP.Name, Namespace: barServiceP.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{ {
@ -364,8 +364,8 @@ func TestStickyLoadBalanceWorksWithNewServiceCalledFirst(t *testing.T) {
// Call NewService() before OnEndpointsUpdate() // Call NewService() before OnEndpointsUpdate()
loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0)
endpoints := make([]api.Endpoints, 1) endpoints := make([]*api.Endpoints, 1)
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{Addresses: []api.EndpointAddress{{IP: "endpoint1"}}, Ports: []api.EndpointPort{{Port: 1}}}, {Addresses: []api.EndpointAddress{{IP: "endpoint1"}}, Ports: []api.EndpointPort{{Port: 1}}},
@ -420,8 +420,8 @@ func TestStickyLoadBalanceWorksWithNewServiceCalledSecond(t *testing.T) {
} }
// Call OnEndpointsUpdate() before NewService() // Call OnEndpointsUpdate() before NewService()
endpoints := make([]api.Endpoints, 1) endpoints := make([]*api.Endpoints, 1)
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{Addresses: []api.EndpointAddress{{IP: "endpoint1"}}, Ports: []api.EndpointPort{{Port: 1}}}, {Addresses: []api.EndpointAddress{{IP: "endpoint1"}}, Ports: []api.EndpointPort{{Port: 1}}},
@ -482,8 +482,8 @@ func TestStickyLoadBalanaceWorksWithMultipleEndpointsRemoveOne(t *testing.T) {
} }
loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0)
endpoints := make([]api.Endpoints, 1) endpoints := make([]*api.Endpoints, 1)
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{ {
@ -503,7 +503,7 @@ func TestStickyLoadBalanaceWorksWithMultipleEndpointsRemoveOne(t *testing.T) {
expectEndpoint(t, loadBalancer, service, shuffledEndpoints[2], client3) expectEndpoint(t, loadBalancer, service, shuffledEndpoints[2], client3)
client3Endpoint := shuffledEndpoints[2] client3Endpoint := shuffledEndpoints[2]
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{ {
@ -525,7 +525,7 @@ func TestStickyLoadBalanaceWorksWithMultipleEndpointsRemoveOne(t *testing.T) {
expectEndpoint(t, loadBalancer, service, client2Endpoint, client2) expectEndpoint(t, loadBalancer, service, client2Endpoint, client2)
expectEndpoint(t, loadBalancer, service, client3Endpoint, client3) expectEndpoint(t, loadBalancer, service, client3Endpoint, client3)
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{ {
@ -556,8 +556,8 @@ func TestStickyLoadBalanceWorksWithMultipleEndpointsAndUpdates(t *testing.T) {
} }
loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0)
endpoints := make([]api.Endpoints, 1) endpoints := make([]*api.Endpoints, 1)
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{ {
@ -577,7 +577,7 @@ func TestStickyLoadBalanceWorksWithMultipleEndpointsAndUpdates(t *testing.T) {
expectEndpoint(t, loadBalancer, service, shuffledEndpoints[1], client2) expectEndpoint(t, loadBalancer, service, shuffledEndpoints[1], client2)
// Then update the configuration with one fewer endpoints, make sure // Then update the configuration with one fewer endpoints, make sure
// we start in the beginning again // we start in the beginning again
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{ {
@ -596,7 +596,7 @@ func TestStickyLoadBalanceWorksWithMultipleEndpointsAndUpdates(t *testing.T) {
expectEndpoint(t, loadBalancer, service, shuffledEndpoints[1], client2) expectEndpoint(t, loadBalancer, service, shuffledEndpoints[1], client2)
// Clear endpoints // Clear endpoints
endpoints[0] = api.Endpoints{ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, Subsets: nil} endpoints[0] = &api.Endpoints{ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, Subsets: nil}
loadBalancer.OnEndpointsUpdate(endpoints) loadBalancer.OnEndpointsUpdate(endpoints)
endpoint, err = loadBalancer.NextEndpoint(service, nil, false) endpoint, err = loadBalancer.NextEndpoint(service, nil, false)
@ -616,8 +616,8 @@ func TestStickyLoadBalanceWorksWithServiceRemoval(t *testing.T) {
t.Errorf("Didn't fail with non-existent service") t.Errorf("Didn't fail with non-existent service")
} }
loadBalancer.NewService(fooService, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(fooService, api.ServiceAffinityClientIP, 0)
endpoints := make([]api.Endpoints, 2) endpoints := make([]*api.Endpoints, 2)
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: fooService.Name, Namespace: fooService.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: fooService.Name, Namespace: fooService.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{ {
@ -628,7 +628,7 @@ func TestStickyLoadBalanceWorksWithServiceRemoval(t *testing.T) {
} }
barService := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "bar"}, Port: ""} barService := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "bar"}, Port: ""}
loadBalancer.NewService(barService, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(barService, api.ServiceAffinityClientIP, 0)
endpoints[1] = api.Endpoints{ endpoints[1] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: barService.Name, Namespace: barService.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: barService.Name, Namespace: barService.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{ {
@ -685,8 +685,8 @@ func TestStickyLoadBalanceWorksWithEndpointFails(t *testing.T) {
// Call NewService() before OnEndpointsUpdate() // Call NewService() before OnEndpointsUpdate()
loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0)
endpoints := make([]api.Endpoints, 1) endpoints := make([]*api.Endpoints, 1)
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{Addresses: []api.EndpointAddress{{IP: "endpoint1"}}, Ports: []api.EndpointPort{{Port: 1}}}, {Addresses: []api.EndpointAddress{{IP: "endpoint1"}}, Ports: []api.EndpointPort{{Port: 1}}},

40
pkg/proxy/winuserspace/proxier_test.go
View File

@ -216,7 +216,7 @@ func getPortNum(t *testing.T, addr string) int {
func TestTCPProxy(t *testing.T) { func TestTCPProxy(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"}
lb.OnEndpointsUpdate([]api.Endpoints{ lb.OnEndpointsUpdate([]*api.Endpoints{
{ {
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
@ -245,7 +245,7 @@ func TestTCPProxy(t *testing.T) {
func TestUDPProxy(t *testing.T) { func TestUDPProxy(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"}
lb.OnEndpointsUpdate([]api.Endpoints{ lb.OnEndpointsUpdate([]*api.Endpoints{
{ {
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
@ -274,7 +274,7 @@ func TestUDPProxy(t *testing.T) {
func TestUDPProxyTimeout(t *testing.T) { func TestUDPProxyTimeout(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"}
lb.OnEndpointsUpdate([]api.Endpoints{ lb.OnEndpointsUpdate([]*api.Endpoints{
{ {
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
@ -308,7 +308,7 @@ func TestMultiPortProxy(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
serviceP := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo-p"}, Port: "p"} serviceP := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo-p"}, Port: "p"}
serviceQ := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo-q"}, Port: "q"} serviceQ := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo-q"}, Port: "q"}
lb.OnEndpointsUpdate([]api.Endpoints{{ lb.OnEndpointsUpdate([]*api.Endpoints{{
ObjectMeta: metav1.ObjectMeta{Name: serviceP.Name, Namespace: serviceP.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: serviceP.Name, Namespace: serviceP.Namespace},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}}, Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}},
@ -410,7 +410,7 @@ func stopProxyByName(proxier *Proxier, service ServicePortPortalName) error {
func TestTCPProxyStop(t *testing.T) { func TestTCPProxyStop(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"}
lb.OnEndpointsUpdate([]api.Endpoints{ lb.OnEndpointsUpdate([]*api.Endpoints{
{ {
ObjectMeta: metav1.ObjectMeta{Namespace: service.Namespace, Name: service.Name}, ObjectMeta: metav1.ObjectMeta{Namespace: service.Namespace, Name: service.Name},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
@ -456,7 +456,7 @@ func TestTCPProxyStop(t *testing.T) {
func TestUDPProxyStop(t *testing.T) { func TestUDPProxyStop(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"}
lb.OnEndpointsUpdate([]api.Endpoints{ lb.OnEndpointsUpdate([]*api.Endpoints{
{ {
ObjectMeta: metav1.ObjectMeta{Namespace: service.Namespace, Name: service.Name}, ObjectMeta: metav1.ObjectMeta{Namespace: service.Namespace, Name: service.Name},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
@ -496,7 +496,7 @@ func TestUDPProxyStop(t *testing.T) {
func TestTCPProxyUpdateDelete(t *testing.T) { func TestTCPProxyUpdateDelete(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"}
lb.OnEndpointsUpdate([]api.Endpoints{ lb.OnEndpointsUpdate([]*api.Endpoints{
{ {
ObjectMeta: metav1.ObjectMeta{Namespace: service.Namespace, Name: service.Name}, ObjectMeta: metav1.ObjectMeta{Namespace: service.Namespace, Name: service.Name},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
@ -536,7 +536,7 @@ func TestTCPProxyUpdateDelete(t *testing.T) {
func TestUDPProxyUpdateDelete(t *testing.T) { func TestUDPProxyUpdateDelete(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"}
lb.OnEndpointsUpdate([]api.Endpoints{ lb.OnEndpointsUpdate([]*api.Endpoints{
{ {
ObjectMeta: metav1.ObjectMeta{Namespace: service.Namespace, Name: service.Name}, ObjectMeta: metav1.ObjectMeta{Namespace: service.Namespace, Name: service.Name},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
@ -575,14 +575,14 @@ func TestUDPProxyUpdateDelete(t *testing.T) {
func TestTCPProxyUpdateDeleteUpdate(t *testing.T) { func TestTCPProxyUpdateDeleteUpdate(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"}
endpoint := api.Endpoints{ endpoint := &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}}, Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}},
Ports: []api.EndpointPort{{Name: "p", Port: tcpServerPort}}, Ports: []api.EndpointPort{{Name: "p", Port: tcpServerPort}},
}}, }},
} }
lb.OnEndpointsUpdate([]api.Endpoints{endpoint}) lb.OnEndpointsUpdate([]*api.Endpoints{endpoint})
listenIP := "0.0.0.0" listenIP := "0.0.0.0"
p, err := createProxier(lb, net.ParseIP(listenIP), netshtest.NewFake(), net.ParseIP("127.0.0.1"), time.Minute, udpIdleTimeoutForTest) p, err := createProxier(lb, net.ParseIP(listenIP), netshtest.NewFake(), net.ParseIP("127.0.0.1"), time.Minute, udpIdleTimeoutForTest)
@ -610,7 +610,7 @@ func TestTCPProxyUpdateDeleteUpdate(t *testing.T) {
waitForNumProxyLoops(t, p, 0) waitForNumProxyLoops(t, p, 0)
// need to add endpoint here because it got clean up during service delete // need to add endpoint here because it got clean up during service delete
lb.OnEndpointsUpdate([]api.Endpoints{endpoint}) lb.OnEndpointsUpdate([]*api.Endpoints{endpoint})
p.OnServiceUpdate([]api.Service{{ p.OnServiceUpdate([]api.Service{{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Spec: api.ServiceSpec{ClusterIP: listenIP, Ports: []api.ServicePort{{ Spec: api.ServiceSpec{ClusterIP: listenIP, Ports: []api.ServicePort{{
@ -630,14 +630,14 @@ func TestTCPProxyUpdateDeleteUpdate(t *testing.T) {
func TestUDPProxyUpdateDeleteUpdate(t *testing.T) { func TestUDPProxyUpdateDeleteUpdate(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"}
endpoint := api.Endpoints{ endpoint := &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}}, Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}},
Ports: []api.EndpointPort{{Name: "p", Port: udpServerPort}}, Ports: []api.EndpointPort{{Name: "p", Port: udpServerPort}},
}}, }},
} }
lb.OnEndpointsUpdate([]api.Endpoints{endpoint}) lb.OnEndpointsUpdate([]*api.Endpoints{endpoint})
listenIP := "0.0.0.0" listenIP := "0.0.0.0"
p, err := createProxier(lb, net.ParseIP(listenIP), netshtest.NewFake(), net.ParseIP("127.0.0.1"), time.Minute, udpIdleTimeoutForTest) p, err := createProxier(lb, net.ParseIP(listenIP), netshtest.NewFake(), net.ParseIP("127.0.0.1"), time.Minute, udpIdleTimeoutForTest)
@ -665,7 +665,7 @@ func TestUDPProxyUpdateDeleteUpdate(t *testing.T) {
waitForNumProxyLoops(t, p, 0) waitForNumProxyLoops(t, p, 0)
// need to add endpoint here because it got clean up during service delete // need to add endpoint here because it got clean up during service delete
lb.OnEndpointsUpdate([]api.Endpoints{endpoint}) lb.OnEndpointsUpdate([]*api.Endpoints{endpoint})
p.OnServiceUpdate([]api.Service{{ p.OnServiceUpdate([]api.Service{{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Spec: api.ServiceSpec{ClusterIP: listenIP, Ports: []api.ServicePort{{ Spec: api.ServiceSpec{ClusterIP: listenIP, Ports: []api.ServicePort{{
@ -685,7 +685,7 @@ func TestUDPProxyUpdateDeleteUpdate(t *testing.T) {
func TestTCPProxyUpdatePort(t *testing.T) { func TestTCPProxyUpdatePort(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"}
lb.OnEndpointsUpdate([]api.Endpoints{ lb.OnEndpointsUpdate([]*api.Endpoints{
{ {
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
@ -735,7 +735,7 @@ func TestTCPProxyUpdatePort(t *testing.T) {
func TestUDPProxyUpdatePort(t *testing.T) { func TestUDPProxyUpdatePort(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"}
lb.OnEndpointsUpdate([]api.Endpoints{ lb.OnEndpointsUpdate([]*api.Endpoints{
{ {
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
@ -782,7 +782,7 @@ func TestUDPProxyUpdatePort(t *testing.T) {
func TestProxyUpdatePublicIPs(t *testing.T) { func TestProxyUpdatePublicIPs(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"}
lb.OnEndpointsUpdate([]api.Endpoints{ lb.OnEndpointsUpdate([]*api.Endpoints{
{ {
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
@ -836,14 +836,14 @@ func TestProxyUpdatePublicIPs(t *testing.T) {
func TestProxyUpdatePortal(t *testing.T) { func TestProxyUpdatePortal(t *testing.T) {
lb := NewLoadBalancerRR() lb := NewLoadBalancerRR()
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "echo"}, Port: "p"}
endpoint := api.Endpoints{ endpoint := &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}}, Addresses: []api.EndpointAddress{{IP: "127.0.0.1"}},
Ports: []api.EndpointPort{{Name: "p", Port: tcpServerPort}}, Ports: []api.EndpointPort{{Name: "p", Port: tcpServerPort}},
}}, }},
} }
lb.OnEndpointsUpdate([]api.Endpoints{endpoint}) lb.OnEndpointsUpdate([]*api.Endpoints{endpoint})
listenIP := "0.0.0.0" listenIP := "0.0.0.0"
p, err := createProxier(lb, net.ParseIP(listenIP), netshtest.NewFake(), net.ParseIP("127.0.0.1"), time.Minute, udpIdleTimeoutForTest) p, err := createProxier(lb, net.ParseIP(listenIP), netshtest.NewFake(), net.ParseIP("127.0.0.1"), time.Minute, udpIdleTimeoutForTest)
@ -894,7 +894,7 @@ func TestProxyUpdatePortal(t *testing.T) {
Protocol: "TCP", Protocol: "TCP",
}}}, }}},
}}) }})
lb.OnEndpointsUpdate([]api.Endpoints{endpoint}) lb.OnEndpointsUpdate([]*api.Endpoints{endpoint})
svcInfo, exists = p.getServiceInfo(servicePortPortalName) svcInfo, exists = p.getServiceInfo(servicePortPortalName)
if !exists { if !exists {
t.Fatalf("service with ClusterIP set not found in the proxy") t.Fatalf("service with ClusterIP set not found in the proxy")

5
pkg/proxy/winuserspace/roundrobin.go
View File

@ -236,14 +236,15 @@ func (lb *LoadBalancerRR) updateAffinityMap(svcPort proxy.ServicePortName, newEn
// OnEndpointsUpdate manages the registered service endpoints. // OnEndpointsUpdate manages the registered service endpoints.
// Registered endpoints are updated if found in the update set or // Registered endpoints are updated if found in the update set or
// unregistered if missing from the update set. // unregistered if missing from the update set.
func (lb *LoadBalancerRR) OnEndpointsUpdate(allEndpoints []api.Endpoints) { func (lb *LoadBalancerRR) OnEndpointsUpdate(allEndpoints []*api.Endpoints) {
registeredEndpoints := make(map[proxy.ServicePortName]bool) registeredEndpoints := make(map[proxy.ServicePortName]bool)
lb.lock.Lock() lb.lock.Lock()
defer lb.lock.Unlock() defer lb.lock.Unlock()
// Update endpoints for services. // Update endpoints for services.
for i := range allEndpoints { for i := range allEndpoints {
svcEndpoints := &allEndpoints[i] // svcEndpoints should NOT be modified.
svcEndpoints := allEndpoints[i]
// We need to build a map of portname -> all ip:ports for that // We need to build a map of portname -> all ip:ports for that
// portname. Explode Endpoints.Subsets[*] into this structure. // portname. Explode Endpoints.Subsets[*] into this structure.

62
pkg/proxy/winuserspace/roundrobin_test.go
View File

@ -67,7 +67,7 @@ func TestFilterWorks(t *testing.T) {
func TestLoadBalanceFailsWithNoEndpoints(t *testing.T) { func TestLoadBalanceFailsWithNoEndpoints(t *testing.T) {
loadBalancer := NewLoadBalancerRR() loadBalancer := NewLoadBalancerRR()
var endpoints []api.Endpoints var endpoints []*api.Endpoints
loadBalancer.OnEndpointsUpdate(endpoints) loadBalancer.OnEndpointsUpdate(endpoints)
service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "foo"}, Port: "does-not-exist"} service := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "foo"}, Port: "does-not-exist"}
endpoint, err := loadBalancer.NextEndpoint(service, nil, false) endpoint, err := loadBalancer.NextEndpoint(service, nil, false)
@ -106,8 +106,8 @@ func TestLoadBalanceWorksWithSingleEndpoint(t *testing.T) {
if err == nil || len(endpoint) != 0 { if err == nil || len(endpoint) != 0 {
t.Errorf("Didn't fail with non-existent service") t.Errorf("Didn't fail with non-existent service")
} }
endpoints := make([]api.Endpoints, 1) endpoints := make([]*api.Endpoints, 1)
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{IP: "endpoint1"}}, Addresses: []api.EndpointAddress{{IP: "endpoint1"}},
@ -144,8 +144,8 @@ func TestLoadBalanceWorksWithMultipleEndpoints(t *testing.T) {
if err == nil || len(endpoint) != 0 { if err == nil || len(endpoint) != 0 {
t.Errorf("Didn't fail with non-existent service") t.Errorf("Didn't fail with non-existent service")
} }
endpoints := make([]api.Endpoints, 1) endpoints := make([]*api.Endpoints, 1)
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{{ Subsets: []api.EndpointSubset{{
Addresses: []api.EndpointAddress{{IP: "endpoint"}}, Addresses: []api.EndpointAddress{{IP: "endpoint"}},
@ -172,8 +172,8 @@ func TestLoadBalanceWorksWithMultipleEndpointsMultiplePorts(t *testing.T) {
if err == nil || len(endpoint) != 0 { if err == nil || len(endpoint) != 0 {
t.Errorf("Didn't fail with non-existent service") t.Errorf("Didn't fail with non-existent service")
} }
endpoints := make([]api.Endpoints, 1) endpoints := make([]*api.Endpoints, 1)
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: serviceP.Name, Namespace: serviceP.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: serviceP.Name, Namespace: serviceP.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{ {
@ -215,8 +215,8 @@ func TestLoadBalanceWorksWithMultipleEndpointsAndUpdates(t *testing.T) {
if err == nil || len(endpoint) != 0 { if err == nil || len(endpoint) != 0 {
t.Errorf("Didn't fail with non-existent service") t.Errorf("Didn't fail with non-existent service")
} }
endpoints := make([]api.Endpoints, 1) endpoints := make([]*api.Endpoints, 1)
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: serviceP.Name, Namespace: serviceP.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: serviceP.Name, Namespace: serviceP.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{ {
@ -255,7 +255,7 @@ func TestLoadBalanceWorksWithMultipleEndpointsAndUpdates(t *testing.T) {
// Then update the configuration with one fewer endpoints, make sure // Then update the configuration with one fewer endpoints, make sure
// we start in the beginning again // we start in the beginning again
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: serviceP.Name, Namespace: serviceP.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: serviceP.Name, Namespace: serviceP.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{ {
@ -289,7 +289,7 @@ func TestLoadBalanceWorksWithMultipleEndpointsAndUpdates(t *testing.T) {
expectEndpoint(t, loadBalancer, serviceQ, shuffledEndpoints[1], nil) expectEndpoint(t, loadBalancer, serviceQ, shuffledEndpoints[1], nil)
// Clear endpoints // Clear endpoints
endpoints[0] = api.Endpoints{ObjectMeta: metav1.ObjectMeta{Name: serviceP.Name, Namespace: serviceP.Namespace}, Subsets: nil} endpoints[0] = &api.Endpoints{ObjectMeta: metav1.ObjectMeta{Name: serviceP.Name, Namespace: serviceP.Namespace}, Subsets: nil}
loadBalancer.OnEndpointsUpdate(endpoints) loadBalancer.OnEndpointsUpdate(endpoints)
endpoint, err = loadBalancer.NextEndpoint(serviceP, nil, false) endpoint, err = loadBalancer.NextEndpoint(serviceP, nil, false)
@ -306,8 +306,8 @@ func TestLoadBalanceWorksWithServiceRemoval(t *testing.T) {
if err == nil || len(endpoint) != 0 { if err == nil || len(endpoint) != 0 {
t.Errorf("Didn't fail with non-existent service") t.Errorf("Didn't fail with non-existent service")
} }
endpoints := make([]api.Endpoints, 2) endpoints := make([]*api.Endpoints, 2)
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: fooServiceP.Name, Namespace: fooServiceP.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: fooServiceP.Name, Namespace: fooServiceP.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{ {
@ -316,7 +316,7 @@ func TestLoadBalanceWorksWithServiceRemoval(t *testing.T) {
}, },
}, },
} }
endpoints[1] = api.Endpoints{ endpoints[1] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: barServiceP.Name, Namespace: barServiceP.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: barServiceP.Name, Namespace: barServiceP.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{ {
@ -364,8 +364,8 @@ func TestStickyLoadBalanceWorksWithNewServiceCalledFirst(t *testing.T) {
// Call NewService() before OnEndpointsUpdate() // Call NewService() before OnEndpointsUpdate()
loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0)
endpoints := make([]api.Endpoints, 1) endpoints := make([]*api.Endpoints, 1)
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{Addresses: []api.EndpointAddress{{IP: "endpoint1"}}, Ports: []api.EndpointPort{{Port: 1}}}, {Addresses: []api.EndpointAddress{{IP: "endpoint1"}}, Ports: []api.EndpointPort{{Port: 1}}},
@ -420,8 +420,8 @@ func TestStickyLoadBalanceWorksWithNewServiceCalledSecond(t *testing.T) {
} }
// Call OnEndpointsUpdate() before NewService() // Call OnEndpointsUpdate() before NewService()
endpoints := make([]api.Endpoints, 1) endpoints := make([]*api.Endpoints, 1)
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{Addresses: []api.EndpointAddress{{IP: "endpoint1"}}, Ports: []api.EndpointPort{{Port: 1}}}, {Addresses: []api.EndpointAddress{{IP: "endpoint1"}}, Ports: []api.EndpointPort{{Port: 1}}},
@ -482,8 +482,8 @@ func TestStickyLoadBalanaceWorksWithMultipleEndpointsRemoveOne(t *testing.T) {
} }
loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0)
endpoints := make([]api.Endpoints, 1) endpoints := make([]*api.Endpoints, 1)
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{ {
@ -503,7 +503,7 @@ func TestStickyLoadBalanaceWorksWithMultipleEndpointsRemoveOne(t *testing.T) {
expectEndpoint(t, loadBalancer, service, shuffledEndpoints[2], client3) expectEndpoint(t, loadBalancer, service, shuffledEndpoints[2], client3)
client3Endpoint := shuffledEndpoints[2] client3Endpoint := shuffledEndpoints[2]
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{ {
@ -525,7 +525,7 @@ func TestStickyLoadBalanaceWorksWithMultipleEndpointsRemoveOne(t *testing.T) {
expectEndpoint(t, loadBalancer, service, client2Endpoint, client2) expectEndpoint(t, loadBalancer, service, client2Endpoint, client2)
expectEndpoint(t, loadBalancer, service, client3Endpoint, client3) expectEndpoint(t, loadBalancer, service, client3Endpoint, client3)
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{ {
@ -556,8 +556,8 @@ func TestStickyLoadBalanceWorksWithMultipleEndpointsAndUpdates(t *testing.T) {
} }
loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0)
endpoints := make([]api.Endpoints, 1) endpoints := make([]*api.Endpoints, 1)
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{ {
@ -577,7 +577,7 @@ func TestStickyLoadBalanceWorksWithMultipleEndpointsAndUpdates(t *testing.T) {
expectEndpoint(t, loadBalancer, service, shuffledEndpoints[1], client2) expectEndpoint(t, loadBalancer, service, shuffledEndpoints[1], client2)
// Then update the configuration with one fewer endpoints, make sure // Then update the configuration with one fewer endpoints, make sure
// we start in the beginning again // we start in the beginning again
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{ {
@ -596,7 +596,7 @@ func TestStickyLoadBalanceWorksWithMultipleEndpointsAndUpdates(t *testing.T) {
expectEndpoint(t, loadBalancer, service, shuffledEndpoints[1], client2) expectEndpoint(t, loadBalancer, service, shuffledEndpoints[1], client2)
// Clear endpoints // Clear endpoints
endpoints[0] = api.Endpoints{ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, Subsets: nil} endpoints[0] = &api.Endpoints{ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, Subsets: nil}
loadBalancer.OnEndpointsUpdate(endpoints) loadBalancer.OnEndpointsUpdate(endpoints)
endpoint, err = loadBalancer.NextEndpoint(service, nil, false) endpoint, err = loadBalancer.NextEndpoint(service, nil, false)
@ -616,8 +616,8 @@ func TestStickyLoadBalanceWorksWithServiceRemoval(t *testing.T) {
t.Errorf("Didn't fail with non-existent service") t.Errorf("Didn't fail with non-existent service")
} }
loadBalancer.NewService(fooService, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(fooService, api.ServiceAffinityClientIP, 0)
endpoints := make([]api.Endpoints, 2) endpoints := make([]*api.Endpoints, 2)
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: fooService.Name, Namespace: fooService.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: fooService.Name, Namespace: fooService.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{ {
@ -628,7 +628,7 @@ func TestStickyLoadBalanceWorksWithServiceRemoval(t *testing.T) {
} }
barService := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "bar"}, Port: ""} barService := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: "testnamespace", Name: "bar"}, Port: ""}
loadBalancer.NewService(barService, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(barService, api.ServiceAffinityClientIP, 0)
endpoints[1] = api.Endpoints{ endpoints[1] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: barService.Name, Namespace: barService.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: barService.Name, Namespace: barService.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{ {
@ -685,8 +685,8 @@ func TestStickyLoadBalanceWorksWithEndpointFails(t *testing.T) {
// Call NewService() before OnEndpointsUpdate() // Call NewService() before OnEndpointsUpdate()
loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0)
endpoints := make([]api.Endpoints, 1) endpoints := make([]*api.Endpoints, 1)
endpoints[0] = api.Endpoints{ endpoints[0] = &api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Subsets: []api.EndpointSubset{ Subsets: []api.EndpointSubset{
{Addresses: []api.EndpointAddress{{IP: "endpoint1"}}, Ports: []api.EndpointPort{{Port: 1}}}, {Addresses: []api.EndpointAddress{{IP: "endpoint1"}}, Ports: []api.EndpointPort{{Port: 1}}},