Improve node watch handling

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

Author: mblaschke

README.md

@@ -23,15 +23,18 @@ Usage:
   kube-pool-manager [OPTIONS]
 
 Application Options:
-      --debug     debug mode [$DEBUG]
-  -v, --verbose   verbose mode [$VERBOSE]
-      --log.json  Switch log output to json format [$LOG_JSON]
-      --dry-run   Dry run (do not apply to nodes) [$DRY_RUN]
-      --config=   Config path [$CONFIG]
-      --bind=     Server address (default: :8080) [$SERVER_BIND]
+      --debug                    debug mode [$DEBUG]
+  -v, --verbose                  verbose mode [$VERBOSE]
+      --log.json                 Switch log output to json format [$LOG_JSON]
+      --kube.node.labelselector= Node Label selector which nodes should be checked [$KUBE_NODE_LABELSELECTOR]
+      --kube.watch.timeout=      Timeout & full resync for node watch (time.Duration) (default: 24h)
+                                 [$KUBE_WATCH_TIMEOUT]
+      --dry-run                  Dry run (do not apply to nodes) [$DRY_RUN]
+      --config=                  Config path [$CONFIG]
+      --bind=                    Server address (default: :8080) [$SERVER_BIND]
 
 Help Options:
-  -h, --help      Show this help message
+  -h, --help                     Show this help message
 ```
 
 see [example.yaml](/example.yaml) for configuration file

config/config_test.go

@@ -94,7 +94,6 @@ func Test_NodeLabelMatcher(t *testing.T) {
 	}
 }
 
-
 func Test_NodeMultipleMatcher(t *testing.T) {
 	node := buildNode()
 

config/opts.go

@@ -3,6 +3,7 @@ package config
 import (
 	"encoding/json"
 	log "github.com/sirupsen/logrus"
+	"time"
 )
 
 type (
@@ -14,6 +15,11 @@ type (
 			LogJson bool `           long:"log.json"     env:"LOG_JSON" description:"Switch log output to json format"`
 		}
 
+		K8s struct {
+			NodeLabelSelector string        `long:"kube.node.labelselector"     env:"KUBE_NODE_LABELSELECTOR"     description:"Node Label selector which nodes should be checked"        default:""`
+			WatchTimeout      time.Duration `long:"kube.watch.timeout"          env:"KUBE_WATCH_TIMEOUT"          description:"Timeout & full resync for node watch (time.Duration)"     default:"24h"`
+		}
+
 		// general options
 		DryRun     bool   `long:"dry-run"  env:"DRY_RUN"       description:"Dry run (do not apply to nodes)"`
 		Config     string `long:"config"   env:"CONFIG"        description:"Config path"        required:"true"`

manager/manager.go

@@ -3,18 +3,19 @@ package manager
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"github.com/prometheus/client_golang/prometheus"
 	log "github.com/sirupsen/logrus"
 	"github.com/webdevops/kube-pool-manager/config"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/watch"
 	"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"
 )
 
 type (
@@ -27,7 +28,7 @@ type (
 
 		prometheus struct {
 			nodePoolStatus *prometheus.GaugeVec
-			nodeApplied *prometheus.GaugeVec
+			nodeApplied    *prometheus.GaugeVec
 		}
 	}
 )
@@ -83,31 +84,46 @@ func (r *KubePoolManager) initK8s() {
 }
 
 func (m *KubePoolManager) Start() {
-	m.startWatch()
+	go func() {
+		for {
+			log.Info("(re)starting node watch")
+			if err := m.startNodeWatch(); err != nil {
+				log.Errorf("node watcher stopped: %v", err)
+			}
+		}
+	}()
 }
 
-func (m *KubePoolManager) startWatch() {
-	watch, err := m.k8sClient.CoreV1().Nodes().Watch(m.ctx, metav1.ListOptions{Watch: true})
+func (m *KubePoolManager) startNodeWatch() error {
+	timeout := int64(m.Opts.K8s.WatchTimeout.Seconds())
+	watchOpts := metav1.ListOptions{
+		LabelSelector:  m.Opts.K8s.NodeLabelSelector,
+		TimeoutSeconds: &timeout,
+		Watch:          true,
+	}
+	nodeWatcher, err := m.k8sClient.CoreV1().Nodes().Watch(m.ctx, watchOpts)
 	if err != nil {
 		log.Panic(err)
 	}
+	defer nodeWatcher.Stop()
 
-	go func() {
-		for res := range watch.ResultChan() {
-			switch strings.ToLower(string(res.Type)) {
-			case "added":
-				if node, ok := res.Object.(*corev1.Node); ok {
-					m.applyNode(node)
-				}
+	for res := range nodeWatcher.ResultChan() {
+		switch res.Type {
+		case watch.Added:
+			if node, ok := res.Object.(*corev1.Node); ok {
+				m.applyNode(node)
 			}
+		case watch.Error:
+			log.Errorf("go watch error event %v", res.Object)
 		}
-	}()
+	}
+
+	return fmt.Errorf("terminated")
 }
 
 func (m *KubePoolManager) applyNode(node *corev1.Node) {
 	contextLogger := log.WithField("node", node.Name)
 
-
 	for _, poolConfig := range m.Config.Pools {
 		m.prometheus.nodePoolStatus.WithLabelValues(node.Name, poolConfig.Name).Set(0)
 		poolLogger := contextLogger.WithField("pool", poolConfig.Name)