save allServices in prep for async iptables

pkg/proxy/iptables/proxier.go

@@ -193,18 +193,19 @@ type proxyEndpointMap map[proxy.ServicePortName][]*endpointsInfo
 // Proxier is an iptables based proxy for connections between a localhost:lport
 // and services that provide the actual backends.
 type Proxier struct {
-	mu                        sync.Mutex // protects the following fields
+	mu           sync.Mutex // protects the following fields
-	serviceMap                proxyServiceMap
+	serviceMap   proxyServiceMap
-	endpointsMap              proxyEndpointMap
+	endpointsMap proxyEndpointMap
-	portsMap                  map[localPort]closeable
+	portsMap     map[localPort]closeable
-	haveReceivedServiceUpdate bool // true once we've seen an OnServiceUpdate event
+	// allServices and allEndpoints should never be modified by proxier - the
-	// allEndpoints should never be modified by proxier - the pointers
+	// pointers are shared with higher layers of kube-proxy. They are guaranteed
-	// 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
-	// to not be modified in the meantime, but also require to be not
+	// by Proxier.
-	// modified by Proxier.
+	// nil until we have seen an On*Update event.
-	// nil until we have seen an OnEndpointsUpdate event.
+	allServices  []*api.Service
 	allEndpoints []*api.Endpoints
-	throttle     flowcontrol.RateLimiter
+
+	throttle flowcontrol.RateLimiter
 
 	// These are effectively const and do not need the mutex to be held.
 	syncPeriod     time.Duration
@@ -529,13 +530,12 @@ 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) {
-	start := time.Now()
-	defer func() {
-		glog.V(4).Infof("OnServiceUpdate took %v for %d services", time.Since(start), len(allServices))
-	}()
 	proxier.mu.Lock()
 	defer proxier.mu.Unlock()
-	proxier.haveReceivedServiceUpdate = true
+	if proxier.allServices == nil {
+		glog.V(2).Info("Received first Services update")
+	}
+	proxier.allServices = allServices
 
 	newServiceMap, hcAdd, hcDel, staleUDPServices := buildServiceMap(allServices, proxier.serviceMap)
 	for _, hc := range hcAdd {
@@ -769,7 +769,7 @@ func (proxier *Proxier) syncProxyRules() {
 		glog.V(4).Infof("syncProxyRules took %v", time.Since(start))
 	}()
 	// don't sync rules till we've received services and endpoints
-	if proxier.allEndpoints == nil || !proxier.haveReceivedServiceUpdate {
+	if proxier.allEndpoints == nil || proxier.allServices == nil {
 		glog.V(2).Info("Not syncing iptables until Services and Endpoints have been received from master")
 		return
 	}

pkg/proxy/iptables/proxier_test.go

@@ -365,16 +365,16 @@ func NewFakeProxier(ipt utiliptables.Interface) *Proxier {
 	// TODO: Call NewProxier after refactoring out the goroutine
 	// invocation into a Run() method.
 	return &Proxier{
-		exec:                      &exec.FakeExec{},
+		exec:          &exec.FakeExec{},
-		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,
+		allServices:   []*api.Service{},
-		hostname:                  testHostname,
+		hostname:      testHostname,
-		portsMap:                  make(map[localPort]closeable),
+		portsMap:      make(map[localPort]closeable),
-		portMapper:                &fakePortOpener{[]*localPort{}},
+		portMapper:    &fakePortOpener{[]*localPort{}},
-		healthChecker:             fakeHealthChecker{},
+		healthChecker: fakeHealthChecker{},
 	}
 }