Merge pull request #127 from qiangzii/master

Optimized for scenarios with unhealthy nodes and no backends

Makefile

@@ -42,6 +42,7 @@ test: fmt vet
 	go test -v -cover  ./pkg/...
 fmt:
 	go fmt ./pkg/... ./cmd/...
+
 vet:
 	go vet ./pkg/... ./cmd/...
 

pkg/controllers/clusternode/cluster_node_controller.go

@@ -21,6 +21,7 @@ import (
 	genericcontrollermanager "k8s.io/controller-manager/app"
 	"k8s.io/klog"
 
+	"github.com/yunify/qingcloud-cloud-controller-manager/pkg/controllers/utils"
 	"github.com/yunify/qingcloud-cloud-controller-manager/pkg/qingcloud"
 )
 
@@ -175,16 +176,18 @@ func (cnc *ClusterNodeController) handleNodesUpdate(key string) error {
 
 	// 1. get node list
 	var nodes []*corev1.Node
-	nodeList, err := cnc.nodeLister.List(labels.NewSelector())
+	nodeList, err := cnc.nodeLister.List(labels.Everything())
 	if err != nil {
 		return fmt.Errorf("get node list error: %v", err)
 	}
 	for i, _ := range nodeList {
-		nodes = append(nodes, nodeList[i])
+		if utils.NodeConditionCheck(nodeList[i]) {
+			nodes = append(nodes, nodeList[i])
+		}
 	}
 
 	// 2. list all service
-	svcs, err := cnc.serviceLister.List(labels.NewSelector())
+	svcs, err := cnc.serviceLister.List(labels.Everything())
 	if err != nil {
 		return fmt.Errorf("list service  error: %v", err)
 	}

pkg/controllers/endpoint/endpoint_controller.go

@@ -20,6 +20,8 @@ import (
 	cloudcontrollerconfig "k8s.io/cloud-provider/app/config"
 	genericcontrollermanager "k8s.io/controller-manager/app"
 	"k8s.io/klog"
+
+	"github.com/yunify/qingcloud-cloud-controller-manager/pkg/controllers/utils"
 )
 
 const (
@@ -181,12 +183,14 @@ func (epc *EndpointController) handleEndpointsUpdate(key string) error {
 
 	// 2. get node list
 	var nodes []*corev1.Node
-	nodeList, err := epc.nodeLister.List(labels.NewSelector())
+	nodeList, err := epc.nodeLister.List(labels.Everything())
 	if err != nil {
 		return fmt.Errorf("get node list error: %v", err)
 	}
 	for i, _ := range nodeList {
-		nodes = append(nodes, nodeList[i])
+		if utils.NodeConditionCheck(nodeList[i]) {
+			nodes = append(nodes, nodeList[i])
+		}
 	}
 
 	// 3. update lb

pkg/controllers/utils/node.go

@@ -0,0 +1,26 @@
+package utils
+
+import (
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/klog"
+)
+
+// check if the node should be backend or not
+func NodeConditionCheck(node *corev1.Node) bool {
+	if _, hasExcludeBalancerLabel := node.Labels[corev1.LabelNodeExcludeBalancers]; hasExcludeBalancerLabel {
+		return false
+	}
+
+	// If we have no info, don't accept
+	if len(node.Status.Conditions) == 0 {
+		return false
+	}
+	for _, cond := range node.Status.Conditions {
+		// We consider the node for load balancing only when its NodeReady condition status is ConditionTrue
+		if cond.Type == corev1.NodeReady && cond.Status != corev1.ConditionTrue {
+			klog.V(4).Infof("Ignoring node %v with %v condition status %v", node.Name, cond.Type, cond.Status)
+			return false
+		}
+	}
+	return true
+}

pkg/qingcloud/qingcloud.go

@@ -200,10 +200,10 @@ func (qc *QingCloud) EnsureLoadBalancer(ctx context.Context, _ string, service *
 	}
 
 	// filter nodes by service externalTrafficPolicy
-	nodes, err = qc.filterNodes(ctx, service, nodes, conf)
-	if err != nil {
-		klog.Errorf("filterNodes for service %s with externalTrafficPolicy %s error: %v", service.Name, service.Spec.ExternalTrafficPolicy, err)
-		return nil, err
+	nodes, e := qc.filterNodes(ctx, service, nodes, conf)
+	if e != nil {
+		klog.Errorf("filterNodes for service %s/%s with externalTrafficPolicy %s error: %v", service.Namespace, service.Name, service.Spec.ExternalTrafficPolicy, e)
+		return nil, e
 	}
 
 	//1. ensure & update lb
@@ -262,7 +262,7 @@ func (qc *QingCloud) EnsureLoadBalancer(ctx context.Context, _ string, service *
 					toDelete, toAdd := diffBackend(listener, nodes)
 
 					if len(toDelete) > 0 {
-						klog.Infof("backend %s will be deleted for listener %s(%s) of lb %s",
+						klog.Infof("backends %s will be deleted for listener %s(%s) of lb %s",
 							spew.Sdump(toDelete), *listener.Spec.LoadBalancerListenerName, *listener.Spec.LoadBalancerListenerID, *lb.Status.LoadBalancerID)
 						err = qc.Client.DeleteBackends(toDelete)
 						if err != nil {
@@ -273,7 +273,7 @@ func (qc *QingCloud) EnsureLoadBalancer(ctx context.Context, _ string, service *
 
 					toAddBackends := generateLoadBalancerBackends(toAdd, listener, service.Spec.Ports)
 					if len(toAddBackends) > 0 {
-						klog.Infof("backend %s will be added for listener %s(%s) of lb %s",
+						klog.Infof("backends %s will be added for listener %s(%s) of lb %s",
 							spew.Sdump(toAddBackends), *listener.Spec.LoadBalancerListenerName, *listener.Spec.LoadBalancerListenerID, *lb.Status.LoadBalancerID)
 						_, err = qc.Client.CreateBackends(toAddBackends)
 						if err != nil {
@@ -414,7 +414,7 @@ func (qc *QingCloud) UpdateLoadBalancer(ctx context.Context, _ string, service *
 		toDelete, toAdd := diffBackend(listener, nodes)
 
 		if len(toDelete) > 0 {
-			klog.Infof("backend %s will be deleted for listener %s(%s) of lb %s",
+			klog.Infof("backends %s will be deleted for listener %s(%s) of lb %s",
 				spew.Sdump(toDelete), *listener.Spec.LoadBalancerListenerName, *listener.Spec.LoadBalancerListenerID, *lb.Status.LoadBalancerID)
 			err = qc.Client.DeleteBackends(toDelete)
 			if err != nil {
@@ -425,7 +425,7 @@ func (qc *QingCloud) UpdateLoadBalancer(ctx context.Context, _ string, service *
 
 		toAddBackends := generateLoadBalancerBackends(toAdd, listener, service.Spec.Ports)
 		if len(toAddBackends) > 0 {
-			klog.Infof("backend %s will be added for listener %s(%s) of lb %s",
+			klog.Infof("backends %s will be added for listener %s(%s) of lb %s",
 				spew.Sdump(toAddBackends), *listener.Spec.LoadBalancerListenerName, *listener.Spec.LoadBalancerListenerID, *lb.Status.LoadBalancerID)
 			_, err = qc.Client.CreateBackends(toAddBackends)
 			if err != nil {
@@ -567,6 +567,11 @@ func (qc *QingCloud) filterNodes(ctx context.Context, svc *v1.Service, nodes []*
 					newNodes = append(newNodes, nodes[i])
 				}
 			}
+			// if there are no available nodes , use all nodes
+			if len(newNodes) == 0 {
+				klog.Infof("there are no available nodes for service %s/%s, use all nodes!", svc.Namespace, svc.Name)
+				newNodes = nodes
+			}
 		} else {
 			// no need to filter
 			newNodes = nodes