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

kube-proxy: OnServiceUpdate takes pointers 

This signature is more consistent with OnEndpointsUpdate and removes a
copy loop.  This is part on ongoing cleanup to rate-limit iptables
calls.
pull/6/head
Tim Hockin 2017-03-31 21:48:39 -07:00
parent a8e552832d
commit adf30aa2e1
13 changed files with 97 additions and 98 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -218,6 +218,7 @@ func NewProxyServerDefault(config *options.ProxyServerConfig) (*ProxyServer, err
recorder := eventBroadcaster.NewRecorder(api.Scheme, clientv1.EventSource{Component: "kube-proxy", Host: hostname})
var proxier proxy.ProxyProvider
var servicesHandler proxyconfig.ServiceConfigHandler
var endpointsHandler proxyconfig.EndpointsConfigHandler
proxyMode := getProxyMode(string(config.Mode), client.Core().Nodes(), hostname, iptInterface, iptables.LinuxKernelCompatTester{})
@ -244,22 +245,20 @@ func NewProxyServerDefault(config *options.ProxyServerConfig) (*ProxyServer, err
glog.Fatalf("Unable to create proxier: %v", err)
}
proxier = proxierIPTables
servicesHandler = proxierIPTables
endpointsHandler = proxierIPTables
// No turning back. Remove artifacts that might still exist from the userspace Proxier.
glog.V(0).Info("Tearing down userspace rules.")
userspace.CleanupLeftovers(iptInterface)
} else {
glog.V(0).Info("Using userspace Proxier.")
var proxierUserspace proxy.ProxyProvider
if runtime.GOOS == "windows" {
// This is a proxy.LoadBalancer which NewProxier needs but has methods we don't need for
// our config.EndpointsConfigHandler.
loadBalancer := winuserspace.NewLoadBalancerRR()
// set EndpointsConfigHandler to our loadBalancer
endpointsHandler = loadBalancer
proxierUserspace, err = winuserspace.NewProxier(
proxierUserspace, err := winuserspace.NewProxier(
loadBalancer,
net.ParseIP(config.BindAddress),
netshInterface,
@ -268,13 +267,18 @@ func NewProxyServerDefault(config *options.ProxyServerConfig) (*ProxyServer, err
config.IPTablesSyncPeriod.Duration,
config.UDPIdleTimeout.Duration,
)
if err != nil {
glog.Fatalf("Unable to create proxier: %v", err)
}
servicesHandler = proxierUserspace
proxier = proxierUserspace
} else {
// This is a proxy.LoadBalancer which NewProxier needs but has methods we don't need for
// our config.EndpointsConfigHandler.
loadBalancer := userspace.NewLoadBalancerRR()
// set EndpointsConfigHandler to our loadBalancer
endpointsHandler = loadBalancer
proxierUserspace, err = userspace.NewProxier(
proxierUserspace, err := userspace.NewProxier(
loadBalancer,
net.ParseIP(config.BindAddress),
iptInterface,
@ -284,11 +288,12 @@ func NewProxyServerDefault(config *options.ProxyServerConfig) (*ProxyServer, err
config.IPTablesMinSyncPeriod.Duration,
config.UDPIdleTimeout.Duration,
)
}
if err != nil {
glog.Fatalf("Unable to create proxier: %v", err)
}
servicesHandler = proxierUserspace
proxier = proxierUserspace
}
// Remove artifacts from the pure-iptables Proxier, if not on Windows.
if runtime.GOOS != "windows" {
glog.V(0).Info("Tearing down pure-iptables proxy rules.")
@ -306,7 +311,7 @@ func NewProxyServerDefault(config *options.ProxyServerConfig) (*ProxyServer, err
// only notify on changes, and the initial update (on process start) may be lost if no handlers
// are registered yet.
serviceConfig := proxyconfig.NewServiceConfig(client.Core().RESTClient(), config.ConfigSyncPeriod)
serviceConfig.RegisterHandler(proxier)
serviceConfig.RegisterHandler(servicesHandler)
go serviceConfig.Run(wait.NeverStop)
endpointsConfig := proxyconfig.NewEndpointsConfig(client.Core().RESTClient(), config.ConfigSyncPeriod)

4
pkg/kubemark/hollow_proxy.go
View File

@ -39,12 +39,12 @@ type HollowProxy struct {
type FakeProxyHandler struct{}
func (*FakeProxyHandler) OnServiceUpdate(services []api.Service) {}
func (*FakeProxyHandler) OnServiceUpdate(services []*api.Service) {}
func (*FakeProxyHandler) OnEndpointsUpdate(endpoints []*api.Endpoints) {}
type FakeProxier struct{}
func (*FakeProxier) OnServiceUpdate(services []api.Service) {}
func (*FakeProxier) OnServiceUpdate(services []*api.Service) {}
func (*FakeProxier) Sync() {}
func (*FakeProxier) SyncLoop() {
select {}

5
pkg/proxy/BUILD
View File

@ -14,10 +14,7 @@ go_library(
"types.go",
],
tags = ["automanaged"],
deps = [
"//pkg/api:go_default_library",
"//vendor:k8s.io/apimachinery/pkg/types",
],
deps = ["//vendor:k8s.io/apimachinery/pkg/types"],
)
filegroup(

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

@ -74,27 +74,27 @@ func TestNewServicesSourceApi_UpdatesAndMultipleServices(t *testing.T) {
go serviceConfig.Run(stopCh)
// Add the first service
handler.expected = []api.Service{*service1v1}
handler.expected = []*api.Service{service1v1}
fakeWatch.Add(service1v1)
<-ch
// Add another service
handler.expected = []api.Service{*service1v1, *service2}
handler.expected = []*api.Service{service1v1, service2}
fakeWatch.Add(service2)
<-ch
// Modify service1
handler.expected = []api.Service{*service1v2, *service2}
handler.expected = []*api.Service{service1v2, service2}
fakeWatch.Modify(service1v2)
<-ch
// Delete service1
handler.expected = []api.Service{*service2}
handler.expected = []*api.Service{service2}
fakeWatch.Delete(service1v2)
<-ch
// Delete service2
handler.expected = []api.Service{}
handler.expected = []*api.Service{}
fakeWatch.Delete(service2)
<-ch
}
@ -174,15 +174,15 @@ func TestNewEndpointsSourceApi_UpdatesAndMultipleEndpoints(t *testing.T) {
type svcHandler struct {
t *testing.T
expected []api.Service
expected []*api.Service
done func()
}
func newSvcHandler(t *testing.T, svcs []api.Service, done func()) *svcHandler {
func newSvcHandler(t *testing.T, svcs []*api.Service, done func()) *svcHandler {
return &svcHandler{t: t, expected: svcs, done: done}
}
func (s *svcHandler) OnServiceUpdate(services []api.Service) {
func (s *svcHandler) OnServiceUpdate(services []*api.Service) {
defer s.done()
sort.Sort(sortedServices(services))
if !reflect.DeepEqual(s.expected, services) {
@ -242,7 +242,7 @@ func TestInitialSync(t *testing.T) {
svcConfig := newServiceConfig(svcLW, time.Minute)
epsConfig := newEndpointsConfig(epsLW, time.Minute)
svcHandler := newSvcHandler(t, []api.Service{*svc2, *svc1}, wg.Done)
svcHandler := newSvcHandler(t, []*api.Service{svc2, svc1}, wg.Done)
svcConfig.RegisterHandler(svcHandler)
epsHandler := newEpsHandler(t, []*api.Endpoints{eps2, eps1}, wg.Done)
epsConfig.RegisterHandler(epsHandler)

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

@ -33,9 +33,17 @@ import (
// ServiceConfigHandler is an abstract interface of objects which receive update notifications for the set of services.
type ServiceConfigHandler interface {
// OnServiceUpdate gets called when a configuration has been changed by one of the sources.
// This is the union of all the configuration sources.
OnServiceUpdate(services []api.Service)
// OnServiceUpdate gets called when a service is created, removed or changed
// on any of the configuration sources. An example is when a new service
// comes up.
//
// NOTE: For efficiency, services are being passed by reference, thus,
// OnServiceUpdate should NOT modify pointers of a given slice.
// Those service 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.
OnServiceUpdate(services []*api.Service)
}
// EndpointsConfigHandler is an abstract interface of objects which receive update notifications for the set of endpoints.
@ -208,24 +216,23 @@ func (c *ServiceConfig) Run(stopCh <-chan struct{}) {
return
}
// We hanve synced informers. Now we can start delivering updates
// We have synced informers. Now we can start delivering updates
// to the registered handler.
go func() {
for range c.updates {
services, err := c.lister.List(labels.Everything())
if err != nil {
glog.Errorf("Error while listing services from cache: %v", err)
// This will cause a retry (if there isnt' any other trigger in-flight).
// This will cause a retry (if there isn't any other trigger in-flight).
c.dispatchUpdate()
continue
}
svcs := make([]api.Service, 0, len(services))
for i := range services {
svcs = append(svcs, *services[i])
if services == nil {
services = []*api.Service{}
}
for i := range c.handlers {
glog.V(3).Infof("Calling handler.OnServiceUpdate()")
c.handlers[i].OnServiceUpdate(svcs)
c.handlers[i].OnServiceUpdate(services)
}
}
}()

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

@ -28,7 +28,7 @@ import (
"k8s.io/kubernetes/pkg/api"
)
type sortedServices []api.Service
type sortedServices []*api.Service
func (s sortedServices) Len() int {
return len(s)
@ -41,24 +41,24 @@ func (s sortedServices) Less(i, j int) bool {
}
type ServiceHandlerMock struct {
updated chan []api.Service
updated chan []*api.Service
waits int
}
func NewServiceHandlerMock() *ServiceHandlerMock {
return &ServiceHandlerMock{updated: make(chan []api.Service, 5)}
return &ServiceHandlerMock{updated: make(chan []*api.Service, 5)}
}
func (h *ServiceHandlerMock) OnServiceUpdate(services []api.Service) {
func (h *ServiceHandlerMock) OnServiceUpdate(services []*api.Service) {
sort.Sort(sortedServices(services))
h.updated <- services
}
func (h *ServiceHandlerMock) ValidateServices(t *testing.T, expectedServices []api.Service) {
func (h *ServiceHandlerMock) ValidateServices(t *testing.T, expectedServices []*api.Service) {
// We might get 1 or more updates for N service updates, because we
// over write older snapshots of services from the producer go-routine
// if the consumer falls behind.
var services []api.Service
var services []*api.Service
for {
select {
case services = <-h.updated:
@ -139,7 +139,7 @@ func TestNewServiceAddedAndNotified(t *testing.T) {
Spec: api.ServiceSpec{Ports: []api.ServicePort{{Protocol: "TCP", Port: 10}}},
}
fakeWatch.Add(service)
handler.ValidateServices(t, []api.Service{*service})
handler.ValidateServices(t, []*api.Service{service})
}
func TestServiceAddedRemovedSetAndNotified(t *testing.T) {
@ -161,18 +161,18 @@ func TestServiceAddedRemovedSetAndNotified(t *testing.T) {
Spec: api.ServiceSpec{Ports: []api.ServicePort{{Protocol: "TCP", Port: 10}}},
}
fakeWatch.Add(service1)
handler.ValidateServices(t, []api.Service{*service1})
handler.ValidateServices(t, []*api.Service{service1})
service2 := &api.Service{
ObjectMeta: metav1.ObjectMeta{Namespace: "testnamespace", Name: "bar"},
Spec: api.ServiceSpec{Ports: []api.ServicePort{{Protocol: "TCP", Port: 20}}},
}
fakeWatch.Add(service2)
services := []api.Service{*service2, *service1}
services := []*api.Service{service2, service1}
handler.ValidateServices(t, services)
fakeWatch.Delete(service1)
services = []api.Service{*service2}
services = []*api.Service{service2}
handler.ValidateServices(t, services)
}
@ -203,7 +203,7 @@ func TestNewServicesMultipleHandlersAddedAndNotified(t *testing.T) {
fakeWatch.Add(service1)
fakeWatch.Add(service2)
services := []api.Service{*service2, *service1}
services := []*api.Service{service2, service1}
handler.ValidateServices(t, services)
handler2.ValidateServices(t, services)
}

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

@ -457,13 +457,12 @@ type healthCheckPort struct {
// Accepts a list of Services and the existing service map. Returns the new
// service map, a list of healthcheck ports to add to or remove from the health
// checking listener service, and a set of stale UDP services.
func buildServiceMap(allServices []api.Service, oldServiceMap proxyServiceMap) (proxyServiceMap, []healthCheckPort, []healthCheckPort, sets.String) {
func buildServiceMap(allServices []*api.Service, oldServiceMap proxyServiceMap) (proxyServiceMap, []healthCheckPort, []healthCheckPort, sets.String) {
newServiceMap := make(proxyServiceMap)
healthCheckAdd := make([]healthCheckPort, 0)
healthCheckDel := make([]healthCheckPort, 0)
for i := range allServices {
service := &allServices[i]
for _, service := range allServices {
svcName := types.NamespacedName{
Namespace: service.Namespace,
Name: service.Name,
@ -529,7 +528,7 @@ func buildServiceMap(allServices []api.Service, oldServiceMap proxyServiceMap) (
// OnServiceUpdate tracks the active set of service proxies.
// They will be synchronized using syncProxyRules()
func (proxier *Proxier) OnServiceUpdate(allServices []api.Service) {
func (proxier *Proxier) OnServiceUpdate(allServices []*api.Service) {
start := time.Now()
defer func() {
glog.V(4).Infof("OnServiceUpdate took %v for %d services", time.Since(start), len(allServices))

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

@ -858,8 +858,8 @@ func onlyLocalNodePorts(t *testing.T, fp *Proxier, ipt *iptablestest.FakeIPTable
}
}
func makeTestService(namespace, name string, svcFunc func(*api.Service)) api.Service {
svc := api.Service{
func makeTestService(namespace, name string, svcFunc func(*api.Service)) *api.Service {
svc := &api.Service{
ObjectMeta: metav1.ObjectMeta{
Name: name,
Namespace: namespace,
@ -867,7 +867,7 @@ func makeTestService(namespace, name string, svcFunc func(*api.Service)) api.Ser
Spec: api.ServiceSpec{},
Status: api.ServiceStatus{},
}
svcFunc(&svc)
svcFunc(svc)
return svc
}
@ -883,7 +883,7 @@ func addTestPort(array []api.ServicePort, name string, protocol api.Protocol, po
}
func TestBuildServiceMapAddRemove(t *testing.T) {
services := []api.Service{
services := []*api.Service{
makeTestService("somewhere-else", "cluster-ip", func(svc *api.Service) {
svc.Spec.Type = api.ServiceTypeClusterIP
svc.Spec.ClusterIP = "172.16.55.4"
@ -959,7 +959,7 @@ func TestBuildServiceMapAddRemove(t *testing.T) {
}
// Remove some stuff
services = []api.Service{services[0]}
services = []*api.Service{services[0]}
services[0].Spec.Ports = []api.ServicePort{services[0].Spec.Ports[1]}
serviceMap, hcAdd, hcDel, staleUDPServices = buildServiceMap(services, serviceMap)
if len(serviceMap) != 1 {
@ -999,7 +999,7 @@ func TestBuildServiceMapAddRemove(t *testing.T) {
}
func TestBuildServiceMapServiceHeadless(t *testing.T) {
services := []api.Service{
services := []*api.Service{
makeTestService("somewhere-else", "headless", func(svc *api.Service) {
svc.Spec.Type = api.ServiceTypeClusterIP
svc.Spec.ClusterIP = api.ClusterIPNone
@ -1027,7 +1027,7 @@ func TestBuildServiceMapServiceHeadless(t *testing.T) {
}
func TestBuildServiceMapServiceTypeExternalName(t *testing.T) {
services := []api.Service{
services := []*api.Service{
makeTestService("somewhere-else", "external-name", func(svc *api.Service) {
svc.Spec.Type = api.ServiceTypeExternalName
svc.Spec.ClusterIP = "172.16.55.4" // Should be ignored
@ -1053,7 +1053,7 @@ func TestBuildServiceMapServiceTypeExternalName(t *testing.T) {
}
func TestBuildServiceMapServiceUpdate(t *testing.T) {
first := []api.Service{
first := []*api.Service{
makeTestService("somewhere", "some-service", func(svc *api.Service) {
svc.Spec.Type = api.ServiceTypeClusterIP
svc.Spec.ClusterIP = "172.16.55.4"
@ -1062,7 +1062,7 @@ func TestBuildServiceMapServiceUpdate(t *testing.T) {
}),
}
second := []api.Service{
second := []*api.Service{
makeTestService("somewhere", "some-service", func(svc *api.Service) {
svc.ObjectMeta.Annotations = map[string]string{
service.BetaAnnotationExternalTraffic: service.AnnotationValueExternalTrafficLocal,

5
pkg/proxy/types.go
View File

@ -20,15 +20,10 @@ import (
"fmt"
"k8s.io/apimachinery/pkg/types"
"k8s.io/kubernetes/pkg/api"
)
// ProxyProvider is the interface provided by proxier implementations.
type ProxyProvider interface {
// OnServiceUpdate manages the active set of service proxies.
// Active service proxies are reinitialized if found in the update set or
// removed if missing from the update set.
OnServiceUpdate(services []api.Service)
// Sync immediately synchronizes the ProxyProvider's current state to iptables.
Sync()
// SyncLoop runs periodic work.

6
pkg/proxy/userspace/proxier.go
View File

@ -400,12 +400,10 @@ func (proxier *Proxier) addServiceOnPort(service proxy.ServicePortName, serviceR
// OnServiceUpdate manages the active set of service proxies.
// Active service proxies are reinitialized if found in the update set or
// shutdown if missing from the update set.
func (proxier *Proxier) OnServiceUpdate(services []api.Service) {
func (proxier *Proxier) OnServiceUpdate(services []*api.Service) {
glog.V(4).Infof("Received update notice: %+v", services)
activeServices := make(map[proxy.ServicePortName]bool) // use a map as a set
for i := range services {
service := &services[i]
for _, service := range services {
// if ClusterIP is "None" or empty, skip proxying
if !api.IsServiceIPSet(service) {
glog.V(3).Infof("Skipping service %s due to clusterIP = %q", types.NamespacedName{Namespace: service.Namespace, Name: service.Name}, service.Spec.ClusterIP)

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

@ -350,7 +350,7 @@ func TestMultiPortOnServiceUpdate(t *testing.T) {
}
waitForNumProxyLoops(t, p, 0)
p.OnServiceUpdate([]api.Service{{
p.OnServiceUpdate([]*api.Service{{
ObjectMeta: metav1.ObjectMeta{Name: serviceP.Name, Namespace: serviceP.Namespace},
Spec: api.ServiceSpec{ClusterIP: "1.2.3.4", Ports: []api.ServicePort{{
Name: "p",
@ -515,7 +515,7 @@ func TestTCPProxyUpdateDelete(t *testing.T) {
conn.Close()
waitForNumProxyLoops(t, p, 1)
p.OnServiceUpdate([]api.Service{})
p.OnServiceUpdate([]*api.Service{})
if err := waitForClosedPortTCP(p, svcInfo.proxyPort); err != nil {
t.Fatalf(err.Error())
}
@ -555,7 +555,7 @@ func TestUDPProxyUpdateDelete(t *testing.T) {
conn.Close()
waitForNumProxyLoops(t, p, 1)
p.OnServiceUpdate([]api.Service{})
p.OnServiceUpdate([]*api.Service{})
if err := waitForClosedPortUDP(p, svcInfo.proxyPort); err != nil {
t.Fatalf(err.Error())
}
@ -594,7 +594,7 @@ func TestTCPProxyUpdateDeleteUpdate(t *testing.T) {
conn.Close()
waitForNumProxyLoops(t, p, 1)
p.OnServiceUpdate([]api.Service{})
p.OnServiceUpdate([]*api.Service{})
if err := waitForClosedPortTCP(p, svcInfo.proxyPort); err != nil {
t.Fatalf(err.Error())
}
@ -602,7 +602,7 @@ func TestTCPProxyUpdateDeleteUpdate(t *testing.T) {
// need to add endpoint here because it got clean up during service delete
lb.OnEndpointsUpdate([]*api.Endpoints{endpoint})
p.OnServiceUpdate([]api.Service{{
p.OnServiceUpdate([]*api.Service{{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Spec: api.ServiceSpec{ClusterIP: "1.2.3.4", Ports: []api.ServicePort{{
Name: "p",
@ -650,7 +650,7 @@ func TestUDPProxyUpdateDeleteUpdate(t *testing.T) {
conn.Close()
waitForNumProxyLoops(t, p, 1)
p.OnServiceUpdate([]api.Service{})
p.OnServiceUpdate([]*api.Service{})
if err := waitForClosedPortUDP(p, svcInfo.proxyPort); err != nil {
t.Fatalf(err.Error())
}
@ -658,7 +658,7 @@ func TestUDPProxyUpdateDeleteUpdate(t *testing.T) {
// need to add endpoint here because it got clean up during service delete
lb.OnEndpointsUpdate([]*api.Endpoints{endpoint})
p.OnServiceUpdate([]api.Service{{
p.OnServiceUpdate([]*api.Service{{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Spec: api.ServiceSpec{ClusterIP: "1.2.3.4", Ports: []api.ServicePort{{
Name: "p",
@ -703,7 +703,7 @@ func TestTCPProxyUpdatePort(t *testing.T) {
testEchoTCP(t, "127.0.0.1", svcInfo.proxyPort)
waitForNumProxyLoops(t, p, 1)
p.OnServiceUpdate([]api.Service{{
p.OnServiceUpdate([]*api.Service{{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Spec: api.ServiceSpec{ClusterIP: "1.2.3.4", Ports: []api.ServicePort{{
Name: "p",
@ -753,7 +753,7 @@ func TestUDPProxyUpdatePort(t *testing.T) {
}
waitForNumProxyLoops(t, p, 1)
p.OnServiceUpdate([]api.Service{{
p.OnServiceUpdate([]*api.Service{{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Spec: api.ServiceSpec{ClusterIP: "1.2.3.4", Ports: []api.ServicePort{{
Name: "p",
@ -802,7 +802,7 @@ func TestProxyUpdatePublicIPs(t *testing.T) {
testEchoTCP(t, "127.0.0.1", svcInfo.proxyPort)
waitForNumProxyLoops(t, p, 1)
p.OnServiceUpdate([]api.Service{{
p.OnServiceUpdate([]*api.Service{{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Spec: api.ServiceSpec{
Ports: []api.ServicePort{{
@ -856,7 +856,7 @@ func TestProxyUpdatePortal(t *testing.T) {
testEchoTCP(t, "127.0.0.1", svcInfo.proxyPort)
waitForNumProxyLoops(t, p, 1)
p.OnServiceUpdate([]api.Service{{
p.OnServiceUpdate([]*api.Service{{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Spec: api.ServiceSpec{ClusterIP: "", Ports: []api.ServicePort{{
Name: "p",
@ -869,7 +869,7 @@ func TestProxyUpdatePortal(t *testing.T) {
t.Fatalf("service with empty ClusterIP should not be included in the proxy")
}
p.OnServiceUpdate([]api.Service{{
p.OnServiceUpdate([]*api.Service{{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Spec: api.ServiceSpec{ClusterIP: "None", Ports: []api.ServicePort{{
Name: "p",
@ -882,7 +882,7 @@ func TestProxyUpdatePortal(t *testing.T) {
t.Fatalf("service with 'None' as ClusterIP should not be included in the proxy")
}
p.OnServiceUpdate([]api.Service{{
p.OnServiceUpdate([]*api.Service{{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Spec: api.ServiceSpec{ClusterIP: "1.2.3.4", Ports: []api.ServicePort{{
Name: "p",

6
pkg/proxy/winuserspace/proxier.go
View File

@ -317,12 +317,10 @@ func getListenIPPortMap(service *api.Service, listenPort int, nodePort int) map[
// OnServiceUpdate manages the active set of service proxies.
// Active service proxies are reinitialized if found in the update set or
// shutdown if missing from the update set.
func (proxier *Proxier) OnServiceUpdate(services []api.Service) {
func (proxier *Proxier) OnServiceUpdate(services []*api.Service) {
glog.V(4).Infof("Received update notice: %+v", services)
activeServicePortPortals := make(map[ServicePortPortalName]bool) // use a map as a set
for i := range services {
service := &services[i]
for _, service := range services {
// if ClusterIP is "None" or empty, skip proxying
if !api.IsServiceIPSet(service) {
glog.V(3).Infof("Skipping service %s due to clusterIP = %q", types.NamespacedName{Namespace: service.Namespace, Name: service.Name}, service.Spec.ClusterIP)

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

@ -359,7 +359,7 @@ func TestMultiPortOnServiceUpdate(t *testing.T) {
}
waitForNumProxyLoops(t, p, 0)
p.OnServiceUpdate([]api.Service{{
p.OnServiceUpdate([]*api.Service{{
ObjectMeta: metav1.ObjectMeta{Name: serviceP.Name, Namespace: serviceP.Namespace},
Spec: api.ServiceSpec{ClusterIP: "0.0.0.0", Ports: []api.ServicePort{{
Name: "p",
@ -526,7 +526,7 @@ func TestTCPProxyUpdateDelete(t *testing.T) {
conn.Close()
waitForNumProxyLoops(t, p, 1)
p.OnServiceUpdate([]api.Service{})
p.OnServiceUpdate([]*api.Service{})
if err := waitForClosedPortTCP(p, getPortNum(t, svcInfo.socket.Addr().String())); err != nil {
t.Fatalf(err.Error())
}
@ -565,7 +565,7 @@ func TestUDPProxyUpdateDelete(t *testing.T) {
conn.Close()
waitForNumProxyLoops(t, p, 1)
p.OnServiceUpdate([]api.Service{})
p.OnServiceUpdate([]*api.Service{})
if err := waitForClosedPortUDP(p, getPortNum(t, svcInfo.socket.Addr().String())); err != nil {
t.Fatalf(err.Error())
}
@ -603,7 +603,7 @@ func TestTCPProxyUpdateDeleteUpdate(t *testing.T) {
conn.Close()
waitForNumProxyLoops(t, p, 1)
p.OnServiceUpdate([]api.Service{})
p.OnServiceUpdate([]*api.Service{})
if err := waitForClosedPortTCP(p, getPortNum(t, svcInfo.socket.Addr().String())); err != nil {
t.Fatalf(err.Error())
}
@ -611,7 +611,7 @@ func TestTCPProxyUpdateDeleteUpdate(t *testing.T) {
// need to add endpoint here because it got clean up during service delete
lb.OnEndpointsUpdate([]*api.Endpoints{endpoint})
p.OnServiceUpdate([]api.Service{{
p.OnServiceUpdate([]*api.Service{{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Spec: api.ServiceSpec{ClusterIP: listenIP, Ports: []api.ServicePort{{
Name: "p",
@ -658,7 +658,7 @@ func TestUDPProxyUpdateDeleteUpdate(t *testing.T) {
conn.Close()
waitForNumProxyLoops(t, p, 1)
p.OnServiceUpdate([]api.Service{})
p.OnServiceUpdate([]*api.Service{})
if err := waitForClosedPortUDP(p, getPortNum(t, svcInfo.socket.Addr().String())); err != nil {
t.Fatalf(err.Error())
}
@ -666,7 +666,7 @@ func TestUDPProxyUpdateDeleteUpdate(t *testing.T) {
// need to add endpoint here because it got clean up during service delete
lb.OnEndpointsUpdate([]*api.Endpoints{endpoint})
p.OnServiceUpdate([]api.Service{{
p.OnServiceUpdate([]*api.Service{{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Spec: api.ServiceSpec{ClusterIP: listenIP, Ports: []api.ServicePort{{
Name: "p",
@ -710,7 +710,7 @@ func TestTCPProxyUpdatePort(t *testing.T) {
testEchoTCP(t, "127.0.0.1", getPortNum(t, svcInfo.socket.Addr().String()))
waitForNumProxyLoops(t, p, 1)
p.OnServiceUpdate([]api.Service{{
p.OnServiceUpdate([]*api.Service{{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Spec: api.ServiceSpec{ClusterIP: listenIP, Ports: []api.ServicePort{{
Name: "p",
@ -759,7 +759,7 @@ func TestUDPProxyUpdatePort(t *testing.T) {
}
waitForNumProxyLoops(t, p, 1)
p.OnServiceUpdate([]api.Service{{
p.OnServiceUpdate([]*api.Service{{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Spec: api.ServiceSpec{ClusterIP: listenIP, Ports: []api.ServicePort{{
Name: "p",
@ -807,7 +807,7 @@ func TestProxyUpdatePublicIPs(t *testing.T) {
testEchoTCP(t, "127.0.0.1", getPortNum(t, svcInfo.socket.Addr().String()))
waitForNumProxyLoops(t, p, 1)
p.OnServiceUpdate([]api.Service{{
p.OnServiceUpdate([]*api.Service{{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Spec: api.ServiceSpec{
Ports: []api.ServicePort{{
@ -860,7 +860,7 @@ func TestProxyUpdatePortal(t *testing.T) {
testEchoTCP(t, "127.0.0.1", getPortNum(t, svcInfo.socket.Addr().String()))
waitForNumProxyLoops(t, p, 1)
p.OnServiceUpdate([]api.Service{{
p.OnServiceUpdate([]*api.Service{{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Spec: api.ServiceSpec{ClusterIP: "", Ports: []api.ServicePort{{
Name: "p",
@ -873,7 +873,7 @@ func TestProxyUpdatePortal(t *testing.T) {
t.Fatalf("service with empty ClusterIP should not be included in the proxy")
}
p.OnServiceUpdate([]api.Service{{
p.OnServiceUpdate([]*api.Service{{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Spec: api.ServiceSpec{ClusterIP: "None", Ports: []api.ServicePort{{
Name: "p",
@ -886,7 +886,7 @@ func TestProxyUpdatePortal(t *testing.T) {
t.Fatalf("service with 'None' as ClusterIP should not be included in the proxy")
}
p.OnServiceUpdate([]api.Service{{
p.OnServiceUpdate([]*api.Service{{
ObjectMeta: metav1.ObjectMeta{Name: service.Name, Namespace: service.Namespace},
Spec: api.ServiceSpec{ClusterIP: listenIP, Ports: []api.ServicePort{{
Name: "p",