Merge pull request #43702 from wojtek-t/edge_based_proxy

cmd/kube-proxy/app/server.go

@@ -220,7 +220,10 @@ func NewProxyServerDefault(config *options.ProxyServerConfig) (*ProxyServer, err
 
 	var proxier proxy.ProxyProvider
 	var servicesHandler proxyconfig.ServiceConfigHandler
+	// TODO: Migrate all handlers to EndpointsHandler type and
+	// get rid of this one.
 	var endpointsHandler proxyconfig.EndpointsConfigHandler
+	var endpointsEventHandler proxyconfig.EndpointsHandler
 
 	proxyMode := getProxyMode(string(config.Mode), client.Core().Nodes(), hostname, iptInterface, iptables.LinuxKernelCompatTester{})
 	if proxyMode == proxyModeIPTables {
@@ -247,7 +250,7 @@ func NewProxyServerDefault(config *options.ProxyServerConfig) (*ProxyServer, err
 		}
 		proxier = proxierIPTables
 		servicesHandler = proxierIPTables
-		endpointsHandler = proxierIPTables
+		endpointsEventHandler = 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)
@@ -257,7 +260,7 @@ func NewProxyServerDefault(config *options.ProxyServerConfig) (*ProxyServer, err
 			// 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
+			// set EndpointsHandler to our loadBalancer
 			endpointsHandler = loadBalancer
 			proxierUserspace, err := winuserspace.NewProxier(
 				loadBalancer,
@@ -278,7 +281,7 @@ func NewProxyServerDefault(config *options.ProxyServerConfig) (*ProxyServer, err
 			// our config.EndpointsConfigHandler.
 			loadBalancer := userspace.NewLoadBalancerRR()
 			// set EndpointsConfigHandler to our loadBalancer
-			endpointsHandler = loadBalancer
+			endpointsEventHandler = loadBalancer
 			proxierUserspace, err := userspace.NewProxier(
 				loadBalancer,
 				net.ParseIP(config.BindAddress),
@@ -318,7 +321,12 @@ func NewProxyServerDefault(config *options.ProxyServerConfig) (*ProxyServer, err
 	go serviceConfig.Run(wait.NeverStop)
 
 	endpointsConfig := proxyconfig.NewEndpointsConfig(informerFactory.Core().InternalVersion().Endpoints(), config.ConfigSyncPeriod)
-	endpointsConfig.RegisterHandler(endpointsHandler)
+	if endpointsHandler != nil {
+		endpointsConfig.RegisterHandler(endpointsHandler)
+	}
+	if endpointsEventHandler != nil {
+		endpointsConfig.RegisterEventHandler(endpointsEventHandler)
+	}
 	go endpointsConfig.Run(wait.NeverStop)
 
 	// This has to start after the calls to NewServiceConfig and NewEndpointsConfig because those

pkg/proxy/config/config.go

@@ -60,12 +60,31 @@ type EndpointsConfigHandler interface {
 	OnEndpointsUpdate(endpoints []*api.Endpoints)
 }
 
+// EndpointsHandler is an abstract interface o objects which receive
+// notifications about endpoints object changes.
+type EndpointsHandler interface {
+	// OnEndpointsAdd is called whenever creation of new endpoints object
+	// is observed.
+	OnEndpointsAdd(endpoints *api.Endpoints)
+	// OnEndpointsUpdate is called whenever modification of an existing
+	// endpoints object is observed.
+	OnEndpointsUpdate(oldEndpoints, endpoints *api.Endpoints)
+	// OnEndpointsDelete is called whever deletion of an existing endpoints
+	// object is observed.
+	OnEndpointsDelete(endpoints *api.Endpoints)
+	// OnEndpointsSynced is called once all the initial event handlers were
+	// called and the state is fully propagated to local cache.
+	OnEndpointsSynced()
+}
+
 // EndpointsConfig tracks a set of endpoints configurations.
 // It accepts "set", "add" and "remove" operations of endpoints via channels, and invokes registered handlers on change.
 type EndpointsConfig struct {
-	lister       listers.EndpointsLister
-	listerSynced cache.InformerSynced
-	handlers     []EndpointsConfigHandler
+	lister        listers.EndpointsLister
+	listerSynced  cache.InformerSynced
+	eventHandlers []EndpointsHandler
+	// TODO: Remove handlers by switching them to eventHandlers.
+	handlers []EndpointsConfigHandler
 	// updates channel is used to trigger registered handlers.
 	updates chan struct{}
 	stop    chan struct{}
@@ -101,6 +120,11 @@ func (c *EndpointsConfig) RegisterHandler(handler EndpointsConfigHandler) {
 	c.handlers = append(c.handlers, handler)
 }
 
+// RegisterEventHandler registers a handler which is called on every endpoints change.
+func (c *EndpointsConfig) RegisterEventHandler(handler EndpointsHandler) {
+	c.eventHandlers = append(c.eventHandlers, handler)
+}
+
 // Run starts the goroutine responsible for calling registered handlers.
 func (c *EndpointsConfig) Run(stopCh <-chan struct{}) {
 	if !cache.WaitForCacheSync(stopCh, c.listerSynced) {
@@ -111,6 +135,10 @@ func (c *EndpointsConfig) Run(stopCh <-chan struct{}) {
 	// We have synced informers. Now we can start delivering updates
 	// to the registered handler.
 	go func() {
+		for i := range c.eventHandlers {
+			glog.V(3).Infof("Calling handler.OnEndpointsSynced()")
+			c.eventHandlers[i].OnEndpointsSynced()
+		}
 		for {
 			select {
 			case <-c.updates:
@@ -140,15 +168,54 @@ func (c *EndpointsConfig) Run(stopCh <-chan struct{}) {
 	}()
 }
 
-func (c *EndpointsConfig) handleAddEndpoints(_ interface{}) {
+func (c *EndpointsConfig) handleAddEndpoints(obj interface{}) {
+	endpoints, ok := obj.(*api.Endpoints)
+	if !ok {
+		utilruntime.HandleError(fmt.Errorf("unexpected object type: %v", obj))
+		return
+	}
+	for i := range c.eventHandlers {
+		glog.V(4).Infof("Calling handler.OnEndpointsAdd")
+		c.eventHandlers[i].OnEndpointsAdd(endpoints)
+	}
 	c.dispatchUpdate()
 }
 
-func (c *EndpointsConfig) handleUpdateEndpoints(_, _ interface{}) {
+func (c *EndpointsConfig) handleUpdateEndpoints(oldObj, newObj interface{}) {
+	oldEndpoints, ok := oldObj.(*api.Endpoints)
+	if !ok {
+		utilruntime.HandleError(fmt.Errorf("unexpected object type: %v", oldObj))
+		return
+	}
+	endpoints, ok := newObj.(*api.Endpoints)
+	if !ok {
+		utilruntime.HandleError(fmt.Errorf("unexpected object type: %v", newObj))
+		return
+	}
+	for i := range c.eventHandlers {
+		glog.V(4).Infof("Calling handler.OnEndpointsUpdate")
+		c.eventHandlers[i].OnEndpointsUpdate(oldEndpoints, endpoints)
+	}
 	c.dispatchUpdate()
 }
 
-func (c *EndpointsConfig) handleDeleteEndpoints(_ interface{}) {
+func (c *EndpointsConfig) handleDeleteEndpoints(obj interface{}) {
+	endpoints, ok := obj.(*api.Endpoints)
+	if !ok {
+		tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
+		if !ok {
+			utilruntime.HandleError(fmt.Errorf("unexpected object type: %v", obj))
+			return
+		}
+		if endpoints, ok = tombstone.Obj.(*api.Endpoints); !ok {
+			utilruntime.HandleError(fmt.Errorf("unexpected object type: %v", obj))
+			return
+		}
+	}
+	for i := range c.eventHandlers {
+		glog.V(4).Infof("Calling handler.OnEndpointsUpdate")
+		c.eventHandlers[i].OnEndpointsDelete(endpoints)
+	}
 	c.dispatchUpdate()
 }
 

pkg/proxy/iptables/proxier.go

@@ -196,8 +196,8 @@ func newServiceInfo(serviceName proxy.ServicePortName, port *api.ServicePort, se
 	return info
 }
 
+type endpointsMap map[types.NamespacedName]*api.Endpoints
 type proxyServiceMap map[proxy.ServicePortName]*serviceInfo
-
 type proxyEndpointMap map[proxy.ServicePortName][]*endpointsInfo
 
 // Proxier is an iptables based proxy for connections between a localhost:lport
@@ -211,9 +211,14 @@ type Proxier struct {
 	// 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 On*Update event.
-	allServices  []*api.Service
-	allEndpoints []*api.Endpoints
+	allEndpoints endpointsMap
+	// allServices is nil until we have seen an OnServiceUpdate event.
+	allServices []*api.Service
+
+	// endpointsSynced is set to true when endpoints are synced after startup.
+	// This is used to avoid updating iptables with some partial data after
+	// kube-proxy restart.
+	endpointsSynced bool
 
 	throttle flowcontrol.RateLimiter
 
@@ -328,6 +333,7 @@ func NewProxier(ipt utiliptables.Interface,
 		serviceMap:     make(proxyServiceMap),
 		endpointsMap:   make(proxyEndpointMap),
 		portsMap:       make(map[localPort]closeable),
+		allEndpoints:   make(endpointsMap),
 		syncPeriod:     syncPeriod,
 		minSyncPeriod:  minSyncPeriod,
 		throttle:       throttle,
@@ -532,19 +538,42 @@ func (proxier *Proxier) OnServiceUpdate(allServices []*api.Service) {
 	proxier.syncProxyRules(syncReasonServices)
 }
 
-// OnEndpointsUpdate takes in a slice of updated endpoints.
-func (proxier *Proxier) OnEndpointsUpdate(allEndpoints []*api.Endpoints) {
+func (proxier *Proxier) OnEndpointsAdd(endpoints *api.Endpoints) {
+	namespacedName := types.NamespacedName{Namespace: endpoints.Namespace, Name: endpoints.Name}
+
 	proxier.mu.Lock()
 	defer proxier.mu.Unlock()
-	if proxier.allEndpoints == nil {
-		glog.V(2).Info("Received first Endpoints update")
-	}
-	proxier.allEndpoints = allEndpoints
+	proxier.allEndpoints[namespacedName] = endpoints
+	proxier.syncProxyRules(syncReasonEndpoints)
+}
+
+func (proxier *Proxier) OnEndpointsUpdate(_, endpoints *api.Endpoints) {
+	namespacedName := types.NamespacedName{Namespace: endpoints.Namespace, Name: endpoints.Name}
+
+	proxier.mu.Lock()
+	defer proxier.mu.Unlock()
+	proxier.allEndpoints[namespacedName] = endpoints
+	proxier.syncProxyRules(syncReasonEndpoints)
+}
+
+func (proxier *Proxier) OnEndpointsDelete(endpoints *api.Endpoints) {
+	namespacedName := types.NamespacedName{Namespace: endpoints.Namespace, Name: endpoints.Name}
+
+	proxier.mu.Lock()
+	defer proxier.mu.Unlock()
+	delete(proxier.allEndpoints, namespacedName)
+	proxier.syncProxyRules(syncReasonEndpoints)
+}
+
+func (proxier *Proxier) OnEndpointsSynced() {
+	proxier.mu.Lock()
+	defer proxier.mu.Unlock()
+	proxier.endpointsSynced = true
 	proxier.syncProxyRules(syncReasonEndpoints)
 }
 
 // Convert a slice of api.Endpoints objects into a map of service-port -> endpoints.
-func buildNewEndpointsMap(allEndpoints []*api.Endpoints, curMap proxyEndpointMap, hostname string) (newMap proxyEndpointMap, hcEndpoints map[types.NamespacedName]int, staleSet map[endpointServicePair]bool) {
+func buildNewEndpointsMap(allEndpoints endpointsMap, curMap proxyEndpointMap, hostname string) (newMap proxyEndpointMap, hcEndpoints map[types.NamespacedName]int, staleSet map[endpointServicePair]bool) {
 
 	// return values
 	newMap = make(proxyEndpointMap)
@@ -552,8 +581,8 @@ func buildNewEndpointsMap(allEndpoints []*api.Endpoints, curMap proxyEndpointMap
 	staleSet = make(map[endpointServicePair]bool)
 
 	// Update endpoints for services.
-	for i := range allEndpoints {
-		accumulateEndpointsMap(allEndpoints[i], hostname, &newMap)
+	for _, endpoints := range allEndpoints {
+		accumulateEndpointsMap(endpoints, hostname, &newMap)
 	}
 	// Check stale connections against endpoints missing from the update.
 	// TODO: we should really only mark a connection stale if the proto was UDP
@@ -608,7 +637,6 @@ func buildNewEndpointsMap(allEndpoints []*api.Endpoints, curMap proxyEndpointMap
 // NOTE: endpoints object should NOT be modified.
 //
 // TODO: this could be simplified:
-// - hostPortInfo and endpointsInfo overlap too much
 // - the test for this is overlapped by the test for buildNewEndpointsMap
 // - naming is poor and responsibilities are muddled
 func accumulateEndpointsMap(endpoints *api.Endpoints, hostname string, newEndpoints *proxyEndpointMap) {
@@ -733,7 +761,7 @@ func (proxier *Proxier) syncProxyRules(reason syncReason) {
 		glog.V(4).Infof("syncProxyRules(%s) took %v", reason, time.Since(start))
 	}()
 	// don't sync rules till we've received services and endpoints
-	if proxier.allEndpoints == nil || proxier.allServices == nil {
+	if !proxier.endpointsSynced || proxier.allServices == nil {
 		glog.V(2).Info("Not syncing iptables until Services and Endpoints have been received from master")
 		return
 	}

pkg/proxy/userspace/roundrobin.go

@@ -243,65 +243,92 @@ func (lb *LoadBalancerRR) updateAffinityMap(svcPort proxy.ServicePortName, newEn
 	}
 }
 
-// OnEndpointsUpdate manages the registered service endpoints.
-// Registered endpoints are updated if found in the update set or
-// unregistered if missing from the update set.
-func (lb *LoadBalancerRR) OnEndpointsUpdate(allEndpoints []*api.Endpoints) {
+// buildPortsToEndpointsMap builds a map of portname -> all ip:ports for that
+// portname. Expode Endpoints.Subsets[*] into this structure.
+func buildPortsToEndpointsMap(endpoints *api.Endpoints) map[string][]hostPortPair {
+	portsToEndpoints := map[string][]hostPortPair{}
+	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()
+	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]
+
+		if !exists || state == nil || len(newEndpoints) > 0 {
+			glog.V(1).Infof("LoadBalancerRR: Setting endpoints for %s to %+v", svcPort, newEndpoints)
+			lb.updateAffinityMap(svcPort, newEndpoints)
+			// OnEndpointsAdd 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
+		}
+	}
+}
+
+func (lb *LoadBalancerRR) OnEndpointsUpdate(oldEndpoints, endpoints *api.Endpoints) {
+	portsToEndpoints := buildPortsToEndpointsMap(endpoints)
+	oldPortsToEndpoints := buildPortsToEndpointsMap(oldEndpoints)
 	registeredEndpoints := make(map[proxy.ServicePortName]bool)
+
 	lb.lock.Lock()
 	defer lb.lock.Unlock()
 
-	// Update endpoints for services.
-	for i := range allEndpoints {
-		// svcEndpoints object should NOT be modified.
-		svcEndpoints := allEndpoints[i]
-
-		// We need to build a map of portname -> all ip:ports for that
-		// portname.  Explode Endpoints.Subsets[*] into this structure.
-		portsToEndpoints := map[string][]hostPortPair{}
-		for i := range svcEndpoints.Subsets {
-			ss := &svcEndpoints.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.
-				}
-			}
+	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
 		}
 
-		for portname := range portsToEndpoints {
-			svcPort := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: svcEndpoints.Namespace, Name: svcEndpoints.Name}, Port: portname}
-			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
+		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 {
-			glog.V(2).Infof("LoadBalancerRR: Removing endpoints for %s", k)
+
+	// Now remove all endpoints missing from the update.
+	for portname := range oldPortsToEndpoints {
+		svcPort := proxy.ServicePortName{NamespacedName: types.NamespacedName{Namespace: oldEndpoints.Namespace, Name: oldEndpoints.Name}, Port: portname}
+		if _, exists := registeredEndpoints[svcPort]; !exists {
+			glog.V(2).Infof("LoadBalancerRR: Removing endpoints for %s", svcPort)
 			// Reset but don't delete.
-			state := lb.services[k]
+			state := lb.services[svcPort]
 			state.endpoints = []string{}
 			state.index = 0
 			state.affinity.affinityMap = map[string]*affinityState{}
@@ -309,6 +336,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.
 func slicesEquiv(lhs, rhs []string) bool {
 	if len(lhs) != len(rhs) {