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

Edge-based winuserspace proxy

pull/6/head
Wojciech Tyczynski 2017-04-11 09:49:47 +02:00
parent bd3146a2a4
commit c3e9467b63
4 changed files with 209 additions and 192 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cmd/kube-proxy/app/server.go
View File

@ -261,7 +261,7 @@ func NewProxyServerDefault(config *options.ProxyServerConfig) (*ProxyServer, err
// our config.EndpointsConfigHandler. // our config.EndpointsConfigHandler.
loadBalancer := winuserspace.NewLoadBalancerRR() loadBalancer := winuserspace.NewLoadBalancerRR()
// set EndpointsHandler to our loadBalancer // set EndpointsHandler to our loadBalancer
endpointsHandler = loadBalancer endpointsEventHandler = loadBalancer
proxierUserspace, err := winuserspace.NewProxier( proxierUserspace, err := winuserspace.NewProxier(
loadBalancer, loadBalancer,
net.ParseIP(config.BindAddress), net.ParseIP(config.BindAddress),

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

@ -216,14 +216,12 @@ 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.OnEndpointsAdd(&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}}, }},
}},
},
}) })
listenIP := "0.0.0.0" listenIP := "0.0.0.0"
@ -245,14 +243,12 @@ 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.OnEndpointsAdd(&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}}, }},
}},
},
}) })
listenIP := "0.0.0.0" listenIP := "0.0.0.0"
@ -274,14 +270,12 @@ 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.OnEndpointsAdd(&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}}, }},
}},
},
}) })
listenIP := "0.0.0.0" listenIP := "0.0.0.0"
@ -308,19 +302,20 @@ 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.OnEndpointsAdd(&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"}},
Ports: []api.EndpointPort{{Name: "p", Protocol: "TCP", Port: tcpServerPort}}, Ports: []api.EndpointPort{{Name: "p", Protocol: "TCP", Port: tcpServerPort}},
}}, }},
}, { })
lb.OnEndpointsAdd(&api.Endpoints{
ObjectMeta: metav1.ObjectMeta{Name: serviceQ.Name, Namespace: serviceQ.Namespace}, ObjectMeta: metav1.ObjectMeta{Name: serviceQ.Name, Namespace: serviceQ.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: "q", Protocol: "UDP", Port: udpServerPort}}, Ports: []api.EndpointPort{{Name: "q", Protocol: "UDP", Port: udpServerPort}},
}}, }},
}}) })
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)
@ -410,14 +405,12 @@ 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.OnEndpointsAdd(&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{{ 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}}, }},
}},
},
}) })
listenIP := "0.0.0.0" listenIP := "0.0.0.0"
@ -456,14 +449,12 @@ 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.OnEndpointsAdd(&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{{ 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}}, }},
}},
},
}) })
listenIP := "0.0.0.0" listenIP := "0.0.0.0"
@ -496,14 +487,12 @@ 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.OnEndpointsAdd(&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{{ 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}}, }},
}},
},
}) })
listenIP := "0.0.0.0" listenIP := "0.0.0.0"
@ -536,14 +525,12 @@ 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.OnEndpointsAdd(&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{{ 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}}, }},
}},
},
}) })
listenIP := "0.0.0.0" listenIP := "0.0.0.0"
@ -582,7 +569,7 @@ func TestTCPProxyUpdateDeleteUpdate(t *testing.T) {
Ports: []api.EndpointPort{{Name: "p", Port: tcpServerPort}}, Ports: []api.EndpointPort{{Name: "p", Port: tcpServerPort}},
}}, }},
} }
lb.OnEndpointsUpdate([]*api.Endpoints{endpoint}) lb.OnEndpointsAdd(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 +597,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.OnEndpointsAdd(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{{
@ -637,7 +624,7 @@ func TestUDPProxyUpdateDeleteUpdate(t *testing.T) {
Ports: []api.EndpointPort{{Name: "p", Port: udpServerPort}}, Ports: []api.EndpointPort{{Name: "p", Port: udpServerPort}},
}}, }},
} }
lb.OnEndpointsUpdate([]*api.Endpoints{endpoint}) lb.OnEndpointsAdd(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 +652,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.OnEndpointsAdd(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,14 +672,12 @@ 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.OnEndpointsAdd(&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}}, }},
}},
},
}) })
listenIP := "0.0.0.0" listenIP := "0.0.0.0"
@ -735,14 +720,12 @@ 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.OnEndpointsAdd(&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}}, }},
}},
},
}) })
listenIP := "0.0.0.0" listenIP := "0.0.0.0"
@ -782,14 +765,12 @@ 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.OnEndpointsAdd(&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}}, }},
}},
},
}) })
listenIP := "0.0.0.0" listenIP := "0.0.0.0"
@ -843,7 +824,7 @@ func TestProxyUpdatePortal(t *testing.T) {
Ports: []api.EndpointPort{{Name: "p", Port: tcpServerPort}}, Ports: []api.EndpointPort{{Name: "p", Port: tcpServerPort}},
}}, }},
} }
lb.OnEndpointsUpdate([]*api.Endpoints{endpoint}) lb.OnEndpointsAdd(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 +875,7 @@ func TestProxyUpdatePortal(t *testing.T) {
Protocol: "TCP", Protocol: "TCP",
}}}, }}},
}}) }})
lb.OnEndpointsUpdate([]*api.Endpoints{endpoint}) lb.OnEndpointsAdd(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")

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

@ -233,65 +233,91 @@ func (lb *LoadBalancerRR) updateAffinityMap(svcPort proxy.ServicePortName, newEn
} }
} }
// OnEndpointsUpdate manages the registered service endpoints. // buildPortsToEndpointsMap builds a map of portname -> all ip:ports for that
// Registered endpoints are updated if found in the update set or // portname. Explode Endpoints.Subsets[*] into this structure.
// unregistered if missing from the update set. func buildPortsToEndpointsMap(endpoints *api.Endpoints) map[string][]hostPortPair {
func (lb *LoadBalancerRR) OnEndpointsUpdate(allEndpoints []*api.Endpoints) { portsToEndpoints := map[string][]hostPortPair{}
registeredEndpoints := make(map[proxy.ServicePortName]bool) for i := range endpoints.Subsets {
ss := &endpoints.Subsets[i]
for i := range ss.Ports {
port := &ss.Ports[i]
for i := range ss.Addresses {
addr := &ss.Addresses[i]
portsToEndpoints[port.Name] = append(portsToEndpoints[port.Name], hostPortPair{addr.IP, int(port.Port)})
// Ignore the protocol field - we'll get that from the Service objects.
}
}
}
return portsToEndpoints
}
func (lb *LoadBalancerRR) OnEndpointsAdd(endpoints *api.Endpoints) {
portsToEndpoints := buildPortsToEndpointsMap(endpoints)
lb.lock.Lock() lb.lock.Lock()
defer lb.lock.Unlock() defer lb.lock.Unlock()
// Update endpoints for services. for portname := range portsToEndpoints {
for i := range allEndpoints { svcPort := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: endpoints.Namespace, Name: endpoints.Name}, Port: portname}
// svcEndpoints should NOT be modified. newEndpoints := flattenValidEndpoints(portsToEndpoints[portname])
svcEndpoints := allEndpoints[i] state, exists := lb.services[svcPort]
// We need to build a map of portname -> all ip:ports for that if !exists || state == nil || len(newEndpoints) > 0 {
// portname. Explode Endpoints.Subsets[*] into this structure. glog.V(1).Infof("LoadBalancerRR: Setting endpoints for %s to %+v", svcPort, newEndpoints)
portsToEndpoints := map[string][]hostPortPair{} lb.updateAffinityMap(svcPort, newEndpoints)
for i := range svcEndpoints.Subsets { // OnEndpointsAdd can be called without NewService being called externally.
ss := &svcEndpoints.Subsets[i] // To be safe we will call it here. A new service will only be created
for i := range ss.Ports { // if one does not already exist. The affinity will be updated
port := &ss.Ports[i] // later, once NewService is called.
for i := range ss.Addresses { state = lb.newServiceInternal(svcPort, api.ServiceAffinity(""), 0)
addr := &ss.Addresses[i] state.endpoints = slice.ShuffleStrings(newEndpoints)
portsToEndpoints[port.Name] = append(portsToEndpoints[port.Name], hostPortPair{addr.IP, int(port.Port)})
// Ignore the protocol field - we'll get that from the Service objects.
}
}
}
for portname := range portsToEndpoints { // Reset the round-robin index.
svcPort := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: svcEndpoints.Namespace, Name: svcEndpoints.Name}, Port: portname} state.index = 0
state, exists := lb.services[svcPort]
curEndpoints := []string{}
if state != nil {
curEndpoints = state.endpoints
}
newEndpoints := flattenValidEndpoints(portsToEndpoints[portname])
if !exists || state == nil || len(curEndpoints) != len(newEndpoints) || !slicesEquiv(slice.CopyStrings(curEndpoints), newEndpoints) {
glog.V(1).Infof("LoadBalancerRR: Setting endpoints for %s to %+v", svcPort, newEndpoints)
lb.updateAffinityMap(svcPort, newEndpoints)
// OnEndpointsUpdate can be called without NewService being called externally.
// To be safe we will call it here. A new service will only be created
// if one does not already exist. The affinity will be updated
// later, once NewService is called.
state = lb.newServiceInternal(svcPort, api.ServiceAffinity(""), 0)
state.endpoints = slice.ShuffleStrings(newEndpoints)
// Reset the round-robin index.
state.index = 0
}
registeredEndpoints[svcPort] = true
} }
} }
// Remove endpoints missing from the update. }
for k := range lb.services {
if _, exists := registeredEndpoints[k]; !exists { func (lb *LoadBalancerRR) OnEndpointsUpdate(oldEndpoints, endpoints *api.Endpoints) {
glog.V(2).Infof("LoadBalancerRR: Removing endpoints for %s", k) portsToEndpoints := buildPortsToEndpointsMap(endpoints)
oldPortsToEndpoints := buildPortsToEndpointsMap(oldEndpoints)
registeredEndpoints := make(map[proxy.ServicePortName]bool)
lb.lock.Lock()
defer lb.lock.Unlock()
for portname := range portsToEndpoints {
svcPort := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: endpoints.Namespace, Name: endpoints.Name}, Port: portname}
newEndpoints := flattenValidEndpoints(portsToEndpoints[portname])
state, exists := lb.services[svcPort]
curEndpoints := []string{}
if state != nil {
curEndpoints = state.endpoints
}
if !exists || state == nil || len(curEndpoints) != len(newEndpoints) || !slicesEquiv(slice.CopyStrings(curEndpoints), newEndpoints) {
glog.V(1).Infof("LoadBalancerRR: Setting endpoints for %s to %+v", svcPort, newEndpoints)
lb.updateAffinityMap(svcPort, newEndpoints)
// OnEndpointsUpdate can be called without NewService being called externally.
// To be safe we will call it here. A new service will only be created
// if one does not already exist. The affinity will be updated
// later, once NewService is called.
state = lb.newServiceInternal(svcPort, api.ServiceAffinity(""), 0)
state.endpoints = slice.ShuffleStrings(newEndpoints)
// Reset the round-robin index.
state.index = 0
}
registeredEndpoints[svcPort] = true
}
for portname := range oldPortsToEndpoints {
svcPort := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: endpoints.Namespace, Name: endpoints.Name}, Port: portname}
if _, exists := registeredEndpoints[svcPort]; !exists {
glog.V(2).Infof("LoadBalancerRR: Removing endpoints for %s", svcPort)
// Reset but don't delete. // Reset but don't delete.
state := lb.services[k] state := lb.services[svcPort]
state.endpoints = []string{} state.endpoints = []string{}
state.index = 0 state.index = 0
state.affinity.affinityMap = map[string]*affinityState{} state.affinity.affinityMap = map[string]*affinityState{}
@ -299,6 +325,27 @@ func (lb *LoadBalancerRR) OnEndpointsUpdate(allEndpoints []*api.Endpoints) {
} }
} }
func (lb *LoadBalancerRR) OnEndpointsDelete(endpoints *api.Endpoints) {
portsToEndpoints := buildPortsToEndpointsMap(endpoints)
lb.lock.Lock()
defer lb.lock.Unlock()
for portname := range portsToEndpoints {
svcPort := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: endpoints.Namespace, Name: endpoints.Name}, Port: portname}
glog.V(2).Infof("LoadBalancerRR: Removing endpoints for %s", svcPort)
// If the service is still around, reset but don't delete.
if state, ok := lb.services[svcPort]; ok {
state.endpoints = []string{}
state.index = 0
state.affinity.affinityMap = map[string]*affinityState{}
}
}
}
func (lb *LoadBalancerRR) OnEndpointsSynced() {
}
// Tests whether two slices are equivalent. This sorts both slices in-place. // Tests whether two slices are equivalent. This sorts both slices in-place.
func slicesEquiv(lhs, rhs []string) bool { func slicesEquiv(lhs, rhs []string) bool {
if len(lhs) != len(rhs) { if len(lhs) != len(rhs) {

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

@ -67,8 +67,6 @@ func TestFilterWorks(t *testing.T) {
func TestLoadBalanceFailsWithNoEndpoints(t *testing.T) { func TestLoadBalanceFailsWithNoEndpoints(t *testing.T) {
loadBalancer := NewLoadBalancerRR() loadBalancer := NewLoadBalancerRR()
var endpoints []*api.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)
if err == nil { if err == nil {
@ -106,15 +104,14 @@ 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 := &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"}},
Ports: []api.EndpointPort{{Name: "p", Port: 40}}, Ports: []api.EndpointPort{{Name: "p", Port: 40}},
}}, }},
} }
loadBalancer.OnEndpointsUpdate(endpoints) loadBalancer.OnEndpointsAdd(endpoints)
expectEndpoint(t, loadBalancer, service, "endpoint1:40", nil) expectEndpoint(t, loadBalancer, service, "endpoint1:40", nil)
expectEndpoint(t, loadBalancer, service, "endpoint1:40", nil) expectEndpoint(t, loadBalancer, service, "endpoint1:40", nil)
expectEndpoint(t, loadBalancer, service, "endpoint1:40", nil) expectEndpoint(t, loadBalancer, service, "endpoint1:40", nil)
@ -144,15 +141,14 @@ 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 := &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"}},
Ports: []api.EndpointPort{{Name: "p", Port: 1}, {Name: "p", Port: 2}, {Name: "p", Port: 3}}, Ports: []api.EndpointPort{{Name: "p", Port: 1}, {Name: "p", Port: 2}, {Name: "p", Port: 3}},
}}, }},
} }
loadBalancer.OnEndpointsUpdate(endpoints) loadBalancer.OnEndpointsAdd(endpoints)
shuffledEndpoints := loadBalancer.services[service].endpoints shuffledEndpoints := loadBalancer.services[service].endpoints
if !stringsInSlice(shuffledEndpoints, "endpoint:1", "endpoint:2", "endpoint:3") { if !stringsInSlice(shuffledEndpoints, "endpoint:1", "endpoint:2", "endpoint:3") {
@ -172,8 +168,7 @@ 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 := &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{
{ {
@ -186,7 +181,7 @@ func TestLoadBalanceWorksWithMultipleEndpointsMultiplePorts(t *testing.T) {
}, },
}, },
} }
loadBalancer.OnEndpointsUpdate(endpoints) loadBalancer.OnEndpointsAdd(endpoints)
shuffledEndpoints := loadBalancer.services[serviceP].endpoints shuffledEndpoints := loadBalancer.services[serviceP].endpoints
if !stringsInSlice(shuffledEndpoints, "endpoint1:1", "endpoint2:1", "endpoint3:3") { if !stringsInSlice(shuffledEndpoints, "endpoint1:1", "endpoint2:1", "endpoint3:3") {
@ -215,8 +210,7 @@ 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) endpointsv1 := &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{
{ {
@ -233,7 +227,7 @@ func TestLoadBalanceWorksWithMultipleEndpointsAndUpdates(t *testing.T) {
}, },
}, },
} }
loadBalancer.OnEndpointsUpdate(endpoints) loadBalancer.OnEndpointsAdd(endpointsv1)
shuffledEndpoints := loadBalancer.services[serviceP].endpoints shuffledEndpoints := loadBalancer.services[serviceP].endpoints
if !stringsInSlice(shuffledEndpoints, "endpoint1:1", "endpoint2:2", "endpoint3:3") { if !stringsInSlice(shuffledEndpoints, "endpoint1:1", "endpoint2:2", "endpoint3:3") {
@ -255,7 +249,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{ endpointsv2 := &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{
{ {
@ -268,7 +262,7 @@ func TestLoadBalanceWorksWithMultipleEndpointsAndUpdates(t *testing.T) {
}, },
}, },
} }
loadBalancer.OnEndpointsUpdate(endpoints) loadBalancer.OnEndpointsUpdate(endpointsv1, endpointsv2)
shuffledEndpoints = loadBalancer.services[serviceP].endpoints shuffledEndpoints = loadBalancer.services[serviceP].endpoints
if !stringsInSlice(shuffledEndpoints, "endpoint4:4", "endpoint5:5") { if !stringsInSlice(shuffledEndpoints, "endpoint4:4", "endpoint5:5") {
@ -289,8 +283,8 @@ 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} endpointsv3 := &api.Endpoints{ObjectMeta: metav1.ObjectMeta{Name: serviceP.Name, Namespace: serviceP.Namespace}, Subsets: nil}
loadBalancer.OnEndpointsUpdate(endpoints) loadBalancer.OnEndpointsUpdate(endpointsv2, endpointsv3)
endpoint, err = loadBalancer.NextEndpoint(serviceP, nil, false) endpoint, err = loadBalancer.NextEndpoint(serviceP, nil, false)
if err == nil || len(endpoint) != 0 { if err == nil || len(endpoint) != 0 {
@ -306,8 +300,7 @@ 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) endpoints1 := &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 +309,7 @@ func TestLoadBalanceWorksWithServiceRemoval(t *testing.T) {
}, },
}, },
} }
endpoints[1] = &api.Endpoints{ endpoints2 := &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{
{ {
@ -325,7 +318,8 @@ func TestLoadBalanceWorksWithServiceRemoval(t *testing.T) {
}, },
}, },
} }
loadBalancer.OnEndpointsUpdate(endpoints) loadBalancer.OnEndpointsAdd(endpoints1)
loadBalancer.OnEndpointsAdd(endpoints2)
shuffledFooEndpoints := loadBalancer.services[fooServiceP].endpoints shuffledFooEndpoints := loadBalancer.services[fooServiceP].endpoints
expectEndpoint(t, loadBalancer, fooServiceP, shuffledFooEndpoints[0], nil) expectEndpoint(t, loadBalancer, fooServiceP, shuffledFooEndpoints[0], nil)
expectEndpoint(t, loadBalancer, fooServiceP, shuffledFooEndpoints[1], nil) expectEndpoint(t, loadBalancer, fooServiceP, shuffledFooEndpoints[1], nil)
@ -341,7 +335,7 @@ func TestLoadBalanceWorksWithServiceRemoval(t *testing.T) {
expectEndpoint(t, loadBalancer, barServiceP, shuffledBarEndpoints[1], nil) expectEndpoint(t, loadBalancer, barServiceP, shuffledBarEndpoints[1], nil)
// Then update the configuration by removing foo // Then update the configuration by removing foo
loadBalancer.OnEndpointsUpdate(endpoints[1:]) loadBalancer.OnEndpointsDelete(endpoints1)
endpoint, err = loadBalancer.NextEndpoint(fooServiceP, nil, false) endpoint, err = loadBalancer.NextEndpoint(fooServiceP, nil, false)
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")
@ -364,8 +358,7 @@ 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 := &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}}},
@ -373,7 +366,7 @@ func TestStickyLoadBalanceWorksWithNewServiceCalledFirst(t *testing.T) {
{Addresses: []api.EndpointAddress{{IP: "endpoint3"}}, Ports: []api.EndpointPort{{Port: 3}}}, {Addresses: []api.EndpointAddress{{IP: "endpoint3"}}, Ports: []api.EndpointPort{{Port: 3}}},
}, },
} }
loadBalancer.OnEndpointsUpdate(endpoints) loadBalancer.OnEndpointsAdd(endpoints)
client1 := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0} client1 := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0}
client2 := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 2), Port: 0} client2 := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 2), Port: 0}
@ -420,15 +413,14 @@ func TestStickyLoadBalanceWorksWithNewServiceCalledSecond(t *testing.T) {
} }
// Call OnEndpointsUpdate() before NewService() // Call OnEndpointsUpdate() before NewService()
endpoints := make([]*api.Endpoints, 1) endpoints := &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}}},
{Addresses: []api.EndpointAddress{{IP: "endpoint2"}}, Ports: []api.EndpointPort{{Port: 2}}}, {Addresses: []api.EndpointAddress{{IP: "endpoint2"}}, Ports: []api.EndpointPort{{Port: 2}}},
}, },
} }
loadBalancer.OnEndpointsUpdate(endpoints) loadBalancer.OnEndpointsAdd(endpoints)
loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0)
client1 := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0} client1 := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0}
@ -482,8 +474,7 @@ func TestStickyLoadBalanaceWorksWithMultipleEndpointsRemoveOne(t *testing.T) {
} }
loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0)
endpoints := make([]*api.Endpoints, 1) endpointsv1 := &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{
{ {
@ -492,7 +483,7 @@ func TestStickyLoadBalanaceWorksWithMultipleEndpointsRemoveOne(t *testing.T) {
}, },
}, },
} }
loadBalancer.OnEndpointsUpdate(endpoints) loadBalancer.OnEndpointsAdd(endpointsv1)
shuffledEndpoints := loadBalancer.services[service].endpoints shuffledEndpoints := loadBalancer.services[service].endpoints
expectEndpoint(t, loadBalancer, service, shuffledEndpoints[0], client1) expectEndpoint(t, loadBalancer, service, shuffledEndpoints[0], client1)
client1Endpoint := shuffledEndpoints[0] client1Endpoint := shuffledEndpoints[0]
@ -503,7 +494,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{ endpointsv2 := &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{
{ {
@ -512,7 +503,7 @@ func TestStickyLoadBalanaceWorksWithMultipleEndpointsRemoveOne(t *testing.T) {
}, },
}, },
} }
loadBalancer.OnEndpointsUpdate(endpoints) loadBalancer.OnEndpointsUpdate(endpointsv1, endpointsv2)
shuffledEndpoints = loadBalancer.services[service].endpoints shuffledEndpoints = loadBalancer.services[service].endpoints
if client1Endpoint == "endpoint:3" { if client1Endpoint == "endpoint:3" {
client1Endpoint = shuffledEndpoints[0] client1Endpoint = shuffledEndpoints[0]
@ -525,7 +516,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{ endpointsv3 := &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{
{ {
@ -534,7 +525,7 @@ func TestStickyLoadBalanaceWorksWithMultipleEndpointsRemoveOne(t *testing.T) {
}, },
}, },
} }
loadBalancer.OnEndpointsUpdate(endpoints) loadBalancer.OnEndpointsUpdate(endpointsv2, endpointsv3)
shuffledEndpoints = loadBalancer.services[service].endpoints shuffledEndpoints = loadBalancer.services[service].endpoints
expectEndpoint(t, loadBalancer, service, client1Endpoint, client1) expectEndpoint(t, loadBalancer, service, client1Endpoint, client1)
expectEndpoint(t, loadBalancer, service, client2Endpoint, client2) expectEndpoint(t, loadBalancer, service, client2Endpoint, client2)
@ -556,8 +547,7 @@ func TestStickyLoadBalanceWorksWithMultipleEndpointsAndUpdates(t *testing.T) {
} }
loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0) loadBalancer.NewService(service, api.ServiceAffinityClientIP, 0)
endpoints := make([]*api.Endpoints, 1) endpointsv1 := &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{
{ {
@ -566,7 +556,7 @@ func TestStickyLoadBalanceWorksWithMultipleEndpointsAndUpdates(t *testing.T) {
}, },
}, },
} }
loadBalancer.OnEndpointsUpdate(endpoints) loadBalancer.OnEndpointsAdd(endpointsv1)
shuffledEndpoints := loadBalancer.services[service].endpoints shuffledEndpoints := loadBalancer.services[service].endpoints
expectEndpoint(t, loadBalancer, service, shuffledEndpoints[0], client1) expectEndpoint(t, loadBalancer, service, shuffledEndpoints[0], client1)
expectEndpoint(t, loadBalancer, service, shuffledEndpoints[0], client1) expectEndpoint(t, loadBalancer, service, shuffledEndpoints[0], client1)
@ -577,7 +567,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{ endpointsv2 := &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{
{ {
@ -586,7 +576,7 @@ func TestStickyLoadBalanceWorksWithMultipleEndpointsAndUpdates(t *testing.T) {
}, },
}, },
} }
loadBalancer.OnEndpointsUpdate(endpoints) loadBalancer.OnEndpointsUpdate(endpointsv1, endpointsv2)
shuffledEndpoints = loadBalancer.services[service].endpoints shuffledEndpoints = loadBalancer.services[service].endpoints
expectEndpoint(t, loadBalancer, service, shuffledEndpoints[0], client1) expectEndpoint(t, loadBalancer, service, shuffledEndpoints[0], client1)
expectEndpoint(t, loadBalancer, service, shuffledEndpoints[1], client2) expectEndpoint(t, loadBalancer, service, shuffledEndpoints[1], client2)
@ -596,8 +586,8 @@ 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} endpointsv3 := &api.Endpoints{ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace}, Subsets: nil}
loadBalancer.OnEndpointsUpdate(endpoints) loadBalancer.OnEndpointsUpdate(endpointsv2, endpointsv3)
endpoint, err = loadBalancer.NextEndpoint(service, nil, false) endpoint, err = loadBalancer.NextEndpoint(service, nil, false)
if err == nil || len(endpoint) != 0 { if err == nil || len(endpoint) != 0 {
@ -616,8 +606,7 @@ 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) endpoints1 := &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 +617,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{ endpoints2 := &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{
{ {
@ -637,7 +626,8 @@ func TestStickyLoadBalanceWorksWithServiceRemoval(t *testing.T) {
}, },
}, },
} }
loadBalancer.OnEndpointsUpdate(endpoints) loadBalancer.OnEndpointsAdd(endpoints1)
loadBalancer.OnEndpointsAdd(endpoints2)
shuffledFooEndpoints := loadBalancer.services[fooService].endpoints shuffledFooEndpoints := loadBalancer.services[fooService].endpoints
expectEndpoint(t, loadBalancer, fooService, shuffledFooEndpoints[0], client1) expectEndpoint(t, loadBalancer, fooService, shuffledFooEndpoints[0], client1)
@ -659,7 +649,7 @@ func TestStickyLoadBalanceWorksWithServiceRemoval(t *testing.T) {
expectEndpoint(t, loadBalancer, barService, shuffledBarEndpoints[1], client2) expectEndpoint(t, loadBalancer, barService, shuffledBarEndpoints[1], client2)
// Then update the configuration by removing foo // Then update the configuration by removing foo
loadBalancer.OnEndpointsUpdate(endpoints[1:]) loadBalancer.OnEndpointsDelete(endpoints1)
endpoint, err = loadBalancer.NextEndpoint(fooService, nil, false) endpoint, err = loadBalancer.NextEndpoint(fooService, nil, false)
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")
@ -685,8 +675,7 @@ 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 := &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}}},
@ -694,7 +683,7 @@ func TestStickyLoadBalanceWorksWithEndpointFails(t *testing.T) {
{Addresses: []api.EndpointAddress{{IP: "endpoint3"}}, Ports: []api.EndpointPort{{Port: 3}}}, {Addresses: []api.EndpointAddress{{IP: "endpoint3"}}, Ports: []api.EndpointPort{{Port: 3}}},
}, },
} }
loadBalancer.OnEndpointsUpdate(endpoints) loadBalancer.OnEndpointsAdd(endpoints)
client1 := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0} client1 := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0}
client2 := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 2), Port: 0} client2 := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 2), Port: 0}