Switch node list to watch

Signed-off-by: Markus Blaschke <[email protected]>

Author: mblaschke

autopilot/lib.k8s.go

@@ -1,48 +1,66 @@
 package autopilot
 
 import (
-	"context"
 	"encoding/json"
 	"fmt"
 	log "github.com/sirupsen/logrus"
 	"github.com/webdevopos/azure-k8s-autopilot/k8s"
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/watch"
 	"time"
 )
 
-func (r *AzureK8sAutopilot) getK8sNodeList() (nodeList *k8s.NodeList, err error) {
-	ctx := context.Background()
+func (r *AzureK8sAutopilot) startNodeWatch() error {
+	// init list
+	r.nodeList.lock.Lock()
+	r.nodeList.list = map[string]k8s.Node{}
+	r.nodeList.lock.Unlock()
 
-	opts := metav1.ListOptions{}
-	opts.LabelSelector = r.Config.K8S.NodeLabelSelector
-	list, k8sError := r.k8sClient.CoreV1().Nodes().List(ctx, opts)
-	if k8sError != nil {
-		err = k8sError
-		return
+	timeout := int64(60 * 60 * 1)
+	nodeWatcher, err := r.k8sClient.CoreV1().Nodes().Watch(r.ctx, metav1.ListOptions{TimeoutSeconds: &timeout, Watch: true})
+	if err != nil {
+		log.Panic(err)
 	}
-
-	nodeList = &k8s.NodeList{NodeList: list}
-
-	// fetch all nodes
-	for {
-		if list.RemainingItemCount == nil || *list.RemainingItemCount == 0 {
-			break
+	defer nodeWatcher.Stop()
+
+	for res := range nodeWatcher.ResultChan() {
+		switch res.Type {
+		case watch.Added:
+			r.nodeList.lock.Lock()
+			if node, ok := res.Object.(*corev1.Node); ok {
+				r.nodeList.list[node.Name] = k8s.Node{Node: node}
+			}
+			r.nodeList.lock.Unlock()
+		case watch.Deleted:
+			r.nodeList.lock.Lock()
+			if node, ok := res.Object.(*corev1.Node); ok {
+				delete(r.nodeList.list, node.Name)
+			}
+			r.nodeList.lock.Unlock()
+		case watch.Modified:
+			r.nodeList.lock.Lock()
+			if node, ok := res.Object.(*corev1.Node); ok {
+				r.nodeList.list[node.Name] = k8s.Node{Node: node}
+			}
+			r.nodeList.lock.Unlock()
+		case watch.Error:
+			return fmt.Errorf("unable to understand watch event %v", res.Type)
 		}
+	}
 
-		opts.Continue = list.Continue
+	return fmt.Errorf("terminated")
+}
 
-		remainList, k8sError := r.k8sClient.CoreV1().Nodes().List(ctx, opts)
-		if k8sError != nil {
-			err = k8sError
-			return
-		}
+func (r *AzureK8sAutopilot) getK8sNodeList() (nodeList *k8s.NodeList, err error) {
+	nodeList = &k8s.NodeList{}
 
-		list.Continue = remainList.Continue
-		list.RemainingItemCount = remainList.RemainingItemCount
-		nodeList.Items = append(nodeList.Items, remainList.Items...)
+	r.nodeList.lock.Lock()
+	for _, node := range r.nodeList.list {
+		nodeList.List = append(nodeList.List, &node)
 	}
-
+	r.nodeList.lock.Unlock()
 	return
 }
 

autopilot/main.go

@@ -14,13 +14,14 @@ import (
 	"github.com/webdevopos/azure-k8s-autopilot/k8s"
 	"golang.org/x/net/context"
 	"k8s.io/api/policy/v1beta1"
-	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 	_ "k8s.io/client-go/plugin/pkg/client/auth"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
 	"os"
 	"strings"
+	"sync"
 	"time"
 )
 
@@ -59,6 +60,11 @@ type (
 		cache          *cache.Cache
 		nodeRepairLock *cache.Cache
 		nodeUpdateLock *cache.Cache
+
+		nodeList struct {
+			list map[string]k8s.Node
+			lock sync.Mutex
+		}
 	}
 )
 
@@ -183,6 +189,15 @@ func (r *AzureK8sAutopilot) Start() {
 		r.leaderElect()
 		log.Infof("starting autopilot")
 
+		go func() {
+			for {
+				log.Info("(re)starting node watch")
+				if err := r.startNodeWatch(); err != nil {
+					log.Errorf("node watcher stopped: %v", err)
+				}
+			}
+		}()
+
 		if r.Config.Repair.Crontab != "" {
 			r.startAutopilotRepair()
 		}
@@ -294,7 +309,7 @@ func (r *AzureK8sAutopilot) checkSelfEviction(node *k8s.Node) bool {
 		}
 
 		eviction := v1beta1.Eviction{
-			ObjectMeta: v1.ObjectMeta{
+			ObjectMeta: metav1.ObjectMeta{
 				Name:      *r.Config.Instance.Pod,
 				Namespace: *r.Config.Instance.Namespace,
 			},

autopilot/task.repair.go

@@ -19,7 +19,7 @@ func (r *AzureK8sAutopilot) repairRun(contextLogger *log.Entry) {
 
 	r.nodeRepairLock.DeleteExpired()
 
-	contextLogger.Debugf("found %v nodes in cluster (%v in locked state)", len(nodeList.Items), r.nodeRepairLock.ItemCount())
+	contextLogger.Debugf("found %v nodes in cluster (%v in locked state)", len(nodeList.GetNodes()), r.nodeRepairLock.ItemCount())
 
 nodeLoop:
 	for _, node := range nodeList.GetNodes() {

k8s/nodelist.go

@@ -2,21 +2,16 @@ package k8s
 
 import (
 	"fmt"
-	v1 "k8s.io/api/core/v1"
 )
 
 type (
 	NodeList struct {
-		*v1.NodeList
+		List []*Node
 	}
 )
 
 func (n *NodeList) GetNodes() (list []*Node) {
-	for _, value := range n.Items {
-		node := value
-		list = append(list, &Node{Node: &node})
-	}
-	return
+	return n.List
 }
 
 func (n *NodeList) GetAzureVmssList() (vmssList map[string]*NodeInfo, err error) {
@@ -47,9 +42,9 @@ func (n *NodeList) GetAzureVmssList() (vmssList map[string]*NodeInfo, err error)
 }
 
 func (n *NodeList) FindNodeByProviderId(providerId string) (ret *Node) {
-	for _, node := range n.Items {
+	for _, node := range n.List {
 		if node.Spec.ProviderID == providerId {
-			ret = &Node{&node}
+			ret = node
 			break
 		}
 	}