Change eviction logic in NodeController and make it Zone-aware

pkg/controller/node/controller_utils.go

@@ -35,18 +35,34 @@ import (
 	"github.com/golang/glog"
 )
 
+const (
+	// Number of Nodes that needs to be in the cluster for it to be treated as "large"
+	LargeClusterThreshold = 20
+)
+
 // This function is expected to get a slice of NodeReadyConditions for all Nodes in a given zone.
+// The zone is considered:
+// - fullyDisrupted if there're no Ready Nodes,
+// - partiallyDisrupted if more than 1/3 of Nodes (at least 3) are not Ready,
+// - normal otherwise
 func ComputeZoneState(nodeReadyConditions []*api.NodeCondition) zoneState {
-	seenReady := false
+	readyNodes := 0
+	notReadyNodes := 0
 	for i := range nodeReadyConditions {
 		if nodeReadyConditions[i] != nil && nodeReadyConditions[i].Status == api.ConditionTrue {
-			seenReady = true
+			readyNodes++
+		} else {
+			notReadyNodes++
 		}
 	}
-	if seenReady {
+	switch {
+	case readyNodes == 0 && notReadyNodes > 0:
+		return stateFullDisruption
+	case notReadyNodes > 2 && 2*notReadyNodes > readyNodes:
+		return statePartialDisruption
+	default:
 		return stateNormal
 	}
-	return stateFullSegmentation
 }
 
 // cleanupOrphanedPods deletes pods that are bound to nodes that don't
@@ -320,3 +336,15 @@ func terminatePods(kubeClient clientset.Interface, recorder record.EventRecorder
 	}
 	return complete, nextAttempt, nil
 }
+
+func HealthyQPSFunc(nodeNum int, defaultQPS float32) float32 {
+	return defaultQPS
+}
+
+// If the cluster is large make evictions slower, if they're small stop evictions altogether.
+func ReducedQPSFunc(nodeNum int, defaultQPS float32) float32 {
+	if nodeNum > LargeClusterThreshold {
+		return defaultQPS / 10
+	}
+	return 0
+}

pkg/controller/node/nodecontroller.go

@@ -67,9 +67,10 @@ const (
 type zoneState string
 
 const (
-	stateNormal              = zoneState("Normal")
-	stateFullSegmentation    = zoneState("FullSegmentation")
-	statePartialSegmentation = zoneState("PartialSegmentation")
+	stateInitial           = zoneState("Initial")
+	stateNormal            = zoneState("Normal")
+	stateFullDisruption    = zoneState("FullDisruption")
+	statePartialDisruption = zoneState("PartialDisruption")
 )
 
 type nodeStatusData struct {
@@ -136,9 +137,11 @@ type NodeController struct {
 	// allocate/recycle CIDRs for node if allocateNodeCIDRs == true
 	cidrAllocator CIDRAllocator
 
-	forcefullyDeletePod       func(*api.Pod) error
-	nodeExistsInCloudProvider func(string) (bool, error)
-	computeZoneStateFunc      func(nodeConditions []*api.NodeCondition) zoneState
+	forcefullyDeletePod        func(*api.Pod) error
+	nodeExistsInCloudProvider  func(string) (bool, error)
+	computeZoneStateFunc       func(nodeConditions []*api.NodeCondition) zoneState
+	enterPartialDisruptionFunc func(nodeNum int, defaultQPS float32) float32
+	enterFullDisruptionFunc    func(nodeNum int, defaultQPS float32) float32
 
 	zoneStates map[string]zoneState
 
@@ -192,28 +195,30 @@ func NewNodeController(
 	}
 
 	nc := &NodeController{
-		cloud:                     cloud,
-		knownNodeSet:              make(map[string]*api.Node),
-		kubeClient:                kubeClient,
-		recorder:                  recorder,
-		podEvictionTimeout:        podEvictionTimeout,
-		maximumGracePeriod:        5 * time.Minute,
-		zonePodEvictor:            make(map[string]*RateLimitedTimedQueue),
-		zoneTerminationEvictor:    make(map[string]*RateLimitedTimedQueue),
-		nodeStatusMap:             make(map[string]nodeStatusData),
-		nodeMonitorGracePeriod:    nodeMonitorGracePeriod,
-		nodeMonitorPeriod:         nodeMonitorPeriod,
-		nodeStartupGracePeriod:    nodeStartupGracePeriod,
-		lookupIP:                  net.LookupIP,
-		now:                       unversioned.Now,
-		clusterCIDR:               clusterCIDR,
-		serviceCIDR:               serviceCIDR,
-		allocateNodeCIDRs:         allocateNodeCIDRs,
-		forcefullyDeletePod:       func(p *api.Pod) error { return forcefullyDeletePod(kubeClient, p) },
-		nodeExistsInCloudProvider: func(nodeName string) (bool, error) { return nodeExistsInCloudProvider(cloud, nodeName) },
-		computeZoneStateFunc:      ComputeZoneState,
-		evictionLimiterQPS:        evictionLimiterQPS,
-		zoneStates:                make(map[string]zoneState),
+		cloud:                      cloud,
+		knownNodeSet:               make(map[string]*api.Node),
+		kubeClient:                 kubeClient,
+		recorder:                   recorder,
+		podEvictionTimeout:         podEvictionTimeout,
+		maximumGracePeriod:         5 * time.Minute,
+		zonePodEvictor:             make(map[string]*RateLimitedTimedQueue),
+		zoneTerminationEvictor:     make(map[string]*RateLimitedTimedQueue),
+		nodeStatusMap:              make(map[string]nodeStatusData),
+		nodeMonitorGracePeriod:     nodeMonitorGracePeriod,
+		nodeMonitorPeriod:          nodeMonitorPeriod,
+		nodeStartupGracePeriod:     nodeStartupGracePeriod,
+		lookupIP:                   net.LookupIP,
+		now:                        unversioned.Now,
+		clusterCIDR:                clusterCIDR,
+		serviceCIDR:                serviceCIDR,
+		allocateNodeCIDRs:          allocateNodeCIDRs,
+		forcefullyDeletePod:        func(p *api.Pod) error { return forcefullyDeletePod(kubeClient, p) },
+		nodeExistsInCloudProvider:  func(nodeName string) (bool, error) { return nodeExistsInCloudProvider(cloud, nodeName) },
+		enterPartialDisruptionFunc: ReducedQPSFunc,
+		enterFullDisruptionFunc:    HealthyQPSFunc,
+		computeZoneStateFunc:       ComputeZoneState,
+		evictionLimiterQPS:         evictionLimiterQPS,
+		zoneStates:                 make(map[string]zoneState),
 	}
 
 	podInformer.AddEventHandler(framework.ResourceEventHandlerFuncs{
@@ -491,7 +496,7 @@ func (nc *NodeController) monitorNodeStatus() error {
 				"Skipping - no pods will be evicted.", node.Name)
 			continue
 		}
-		// We do not treat a master node as a part of the cluster for network segmentation checking.
+		// We do not treat a master node as a part of the cluster for network disruption checking.
 		if !system.IsMasterNode(node) {
 			zoneToNodeConditions[utilnode.GetZoneKey(node)] = append(zoneToNodeConditions[utilnode.GetZoneKey(node)], currentReadyCondition)
 		}
@@ -550,37 +555,108 @@ func (nc *NodeController) monitorNodeStatus() error {
 			}
 		}
 	}
+	nc.handleDisruption(zoneToNodeConditions, nodes)
 
+	return nil
+}
+
+func (nc *NodeController) handleDisruption(zoneToNodeConditions map[string][]*api.NodeCondition, nodes *api.NodeList) {
+	newZoneStates := map[string]zoneState{}
+	allAreFullyDisrupted := true
 	for k, v := range zoneToNodeConditions {
 		newState := nc.computeZoneStateFunc(v)
-		if newState == nc.zoneStates[k] {
+		if newState != stateFullDisruption {
+			allAreFullyDisrupted = false
+		}
+		newZoneStates[k] = newState
+		if _, had := nc.zoneStates[k]; !had {
+			nc.zoneStates[k] = stateInitial
+		}
+	}
+
+	allWasFullyDisrupted := true
+	for k, v := range nc.zoneStates {
+		if _, have := zoneToNodeConditions[k]; !have {
+			delete(nc.zoneStates, k)
 			continue
 		}
-		if newState == stateFullSegmentation {
-			glog.V(2).Infof("NodeController is entering network segmentation mode in zone %v.", k)
-		} else if newState == stateNormal {
-			glog.V(2).Infof("NodeController exited network segmentation mode in zone %v.", k)
+		if v != stateFullDisruption {
+			allWasFullyDisrupted = false
+			break
 		}
-		for i := range nodes.Items {
-			if utilnode.GetZoneKey(&nodes.Items[i]) == k {
-				if newState == stateFullSegmentation {
-					// When zone is fully segmented we stop the eviction all together.
-					nc.cancelPodEviction(&nodes.Items[i])
-				}
-				if newState == stateNormal && nc.zoneStates[k] == stateFullSegmentation {
-					// When exiting segmentation mode update probe timestamps on all Nodes.
-					now := nc.now()
-					v := nc.nodeStatusMap[nodes.Items[i].Name]
-					v.probeTimestamp = now
-					v.readyTransitionTimestamp = now
-					nc.nodeStatusMap[nodes.Items[i].Name] = v
-				}
+	}
+
+	// At least one node was responding in previous pass or in the current pass. Semantics is as follows:
+	// - if the new state is "partialDisruption" we call a user defined function that returns a new limiter to use,
+	// - if the new state is "normal" we resume normal operation (go back to default limiter settings),
+	// - if new state is "fullDisruption" we restore normal eviction rate,
+	//   - unless all zones in the cluster are in "fullDisruption" - in that case we stop all evictions.
+	if !allAreFullyDisrupted || !allWasFullyDisrupted {
+		// We're switching to full disruption mode
+		if allAreFullyDisrupted {
+			glog.V(0).Info("NodeController detected that all Nodes are not-Ready. Entering master disruption mode.")
+			for i := range nodes.Items {
+				nc.cancelPodEviction(&nodes.Items[i])
+			}
+			// We stop all evictions.
+			for k := range nc.zonePodEvictor {
+				nc.zonePodEvictor[k].SwapLimiter(0)
+				nc.zoneTerminationEvictor[k].SwapLimiter(0)
+			}
+			for k := range nc.zoneStates {
+				nc.zoneStates[k] = stateFullDisruption
 			}
+			// All rate limiters are updated, so we can return early here.
+			return
+		}
+		// We're exiting full disruption mode
+		if allWasFullyDisrupted {
+			glog.V(0).Info("NodeController detected that some Nodes are Ready. Exiting master disruption mode.")
+			// When exiting disruption mode update probe timestamps on all Nodes.
+			now := nc.now()
+			for i := range nodes.Items {
+				v := nc.nodeStatusMap[nodes.Items[i].Name]
+				v.probeTimestamp = now
+				v.readyTransitionTimestamp = now
+				nc.nodeStatusMap[nodes.Items[i].Name] = v
+			}
+			// We reset all rate limiters to settings appropriate for the given state.
+			for k := range nc.zonePodEvictor {
+				nc.setLimiterInZone(k, len(zoneToNodeConditions[k]), newZoneStates[k])
+				nc.zoneStates[k] = newZoneStates[k]
+			}
+			return
+		}
+		// We know that there's at least one not-fully disrupted so,
+		// we can use default behavior for rate limiters
+		for k, v := range nc.zoneStates {
+			newState := newZoneStates[k]
+			if v == newState {
+				continue
+			}
+			glog.V(0).Infof("NodeController detected that zone %v is now in state %v.", k, newState)
+			nc.setLimiterInZone(k, len(zoneToNodeConditions[k]), newState)
+			nc.zoneStates[k] = newState
 		}
-		nc.zoneStates[k] = newState
 	}
+}
 
-	return nil
+func (nc *NodeController) setLimiterInZone(zone string, zoneSize int, state zoneState) {
+	switch state {
+	case stateNormal:
+		nc.zonePodEvictor[zone].SwapLimiter(nc.evictionLimiterQPS)
+		nc.zoneTerminationEvictor[zone].SwapLimiter(nc.evictionLimiterQPS)
+	case statePartialDisruption:
+		nc.zonePodEvictor[zone].SwapLimiter(
+			nc.enterPartialDisruptionFunc(zoneSize, nc.evictionLimiterQPS))
+		nc.zoneTerminationEvictor[zone].SwapLimiter(
+			nc.enterPartialDisruptionFunc(zoneSize, nc.evictionLimiterQPS))
+	case stateFullDisruption:
+		nc.zonePodEvictor[zone].SwapLimiter(
+			nc.enterFullDisruptionFunc(zoneSize, nc.evictionLimiterQPS))
+		nc.zoneTerminationEvictor[zone].SwapLimiter(
+			nc.enterFullDisruptionFunc(zoneSize, nc.evictionLimiterQPS))
+	}
 }
 
 // For a given node checks its conditions and tries to update it. Returns grace period to which given node
@@ -791,16 +867,5 @@ func (nc *NodeController) cancelPodEviction(node *api.Node) bool {
 func (nc *NodeController) evictPods(node *api.Node) bool {
 	nc.evictorLock.Lock()
 	defer nc.evictorLock.Unlock()
-	foundHealty := false
-	for _, state := range nc.zoneStates {
-		if state != stateFullSegmentation {
-			foundHealty = true
-			break
-		}
-	}
-	if !foundHealty {
-		return false
-	}
-	zone := utilnode.GetZoneKey(node)
-	return nc.zonePodEvictor[zone].Add(node.Name)
+	return nc.zonePodEvictor[utilnode.GetZoneKey(node)].Add(node.Name)
 }

pkg/controller/node/rate_limited_queue.go

@@ -21,9 +21,10 @@ import (
 	"sync"
 	"time"
 
-	"github.com/golang/glog"
 	"k8s.io/kubernetes/pkg/util/flowcontrol"
 	"k8s.io/kubernetes/pkg/util/sets"
+
+	"github.com/golang/glog"
 )
 
 // TimedValue is a value that should be processed at a designated time.
@@ -179,7 +180,7 @@ func (q *RateLimitedTimedQueue) Try(fn ActionFunc) {
 	for ok {
 		// rate limit the queue checking
 		if !q.limiter.TryAccept() {
-			glog.V(10).Info("Try rate limited...")
+			glog.V(10).Infof("Try rate limited for value: %v", val)
 			// Try again later
 			break
 		}

pkg/controller/node/test_utils.go

@@ -247,3 +247,7 @@ func getZones(nodeHandler *FakeNodeHandler) []string {
 	}
 	return zones.List()
 }
+
+func createZoneID(region, zone string) string {
+	return region + ":\x00:" + zone
+}