Skip to content

K
k3s
Commit 71d1e899 authored by Kubernetes Submit Queue's avatar Kubernetes Submit Queue Committed by GitHub
Browse files
Options

Merge pull request #42989 from wojtek-t/avoid_copies_in_kube_proxy

Automatic merge from submit-queue (batch tested with PRs 41000, 42989) Avoid unnecessary copies of endpoints object in kube-proxy
parents a5ddb828 2a6083c3
Show whitespace changes
Inline Side-by-side
Showing with 173 additions and 157 deletions
pkg/kubemark/hollow_proxy.go
View file @ 71d1e899
...@@ -41,7 +41,7 @@ type HollowProxy struct { ...@@ -41,7 +41,7 @@ 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{}
......
pkg/proxy/config/api_test.go
View file @ 71d1e899
...@@ -250,15 +250,15 @@ func (s *svcHandler) OnServiceUpdate(services []api.Service) { ...@@ -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) { ...@@ -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.
......
pkg/proxy/config/config.go
View file @ 71d1e899
...@@ -63,7 +63,14 @@ type EndpointsConfigHandler interface { ...@@ -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 { ...@@ -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 { ...@@ -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
......
pkg/proxy/config/config_test.go
View file @ 71d1e899
...@@ -83,7 +83,7 @@ func (h *ServiceHandlerMock) ValidateServices(t *testing.T, expectedServices []a ...@@ -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 { ...@@ -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. ...@@ -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 ...@@ -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 ...@@ -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 ...@@ -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 ...@@ -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)
} }
......
pkg/proxy/iptables/proxier.go
View file @ 71d1e899
...@@ -198,7 +198,12 @@ type Proxier struct { ...@@ -198,7 +198,12 @@ type Proxier struct {
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
// 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 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.
...@@ -559,7 +564,7 @@ func (proxier *Proxier) OnServiceUpdate(allServices []api.Service) { ...@@ -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) { ...@@ -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 ...@@ -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 ...@@ -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
......
pkg/proxy/iptables/proxier_test.go
View file @ 71d1e899
...@@ -369,7 +369,7 @@ func NewFakeProxier(ipt utiliptables.Interface) *Proxier { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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 ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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 { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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{{
......
pkg/proxy/userspace/proxier_test.go
View file @ 71d1e899
...@@ -202,7 +202,7 @@ func waitForNumProxyClients(t *testing.T, s *ServiceInfo, want int, timeout time ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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 { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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")
......
pkg/proxy/userspace/roundrobin.go
View file @ 71d1e899
...@@ -246,14 +246,15 @@ func (lb *LoadBalancerRR) updateAffinityMap(svcPort proxy.ServicePortName, newEn ...@@ -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.
......
pkg/proxy/userspace/roundrobin_test.go
View file @ 71d1e899
...@@ -67,7 +67,7 @@ func TestFilterWorks(t *testing.T) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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}}},
......
pkg/proxy/winuserspace/proxier_test.go
View file @ 71d1e899
...@@ -216,7 +216,7 @@ func getPortNum(t *testing.T, addr string) int { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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 { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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")
......
pkg/proxy/winuserspace/roundrobin.go
View file @ 71d1e899
...@@ -236,14 +236,15 @@ func (lb *LoadBalancerRR) updateAffinityMap(svcPort proxy.ServicePortName, newEn ...@@ -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.
......
pkg/proxy/winuserspace/roundrobin_test.go
View file @ 71d1e899
...@@ -67,7 +67,7 @@ func TestFilterWorks(t *testing.T) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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}}},
......
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