fix race condition when using spec.providerID

node added event is too early for newer k8s clusters so wait for node ready signal to apply settings

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

Author: mblaschke

README.md

@@ -15,6 +15,8 @@ Sets following settings on nodes if matched:
 - node annotations
 - node [configSource](https://kubernetes.io/docs/tasks/administer-cluster/reconfigure-kubelet/) 
 
+Node settings are applied on startup and for new nodes (delayed until they are ready) and (optional) on watch timeout.
+
 Configuration
 -------------
 
@@ -30,8 +32,8 @@ Application Options:
       --instance.namespace=      Name of namespace where autopilot is running [$INSTANCE_NAMESPACE]
       --instance.pod=            Name of pod where autopilot is running [$INSTANCE_POD]
       --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]
+      --kube.watch.timeout=      Timeout & full resync for node watch (time.Duration) (default: 24h) [$KUBE_WATCH_TIMEOUT]
+      --kube.watch.reapply       Reapply node settings on watch timeout [$KUBE_WATCH_REAPPLY]
       --lease.enable             Enable lease (leader election; enabled by default in docker images) [$LEASE_ENABLE]
       --lease.name=              Name of lease lock (default: kube-pool-manager-leader) [$LEASE_NAME]
       --dry-run                  Dry run (do not apply to nodes) [$DRY_RUN]

config/opts.go

@@ -23,8 +23,9 @@ type (
 		}
 
 		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"`
+			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"`
+			ReapplyOnWatchTimeout bool          `long:"kube.watch.reapply"          env:"KUBE_WATCH_REAPPLY"          description:"Reapply node settings on watch timeout"`
 		}
 
 		// lease

example.yaml

@@ -17,14 +17,14 @@ pools:
       # sets the kubernetes node role
       role: [windows]
 
-  - pool: agents
+  - pool: azure
     selector:
       - path: "{.spec.providerID}"
         # simple match
-        match: "azure:///subscriptions/d86bcf13-ddf7-45ea-82f1-6f656767a318/resourceGroups/mc_k8s_mblaschke_westeurope/providers/Microsoft.Compute/virtualMachineScaleSets/aks-agents-35471996-vmss/virtualMachines/30"
+        regexp: "azure://.*"
     node:
       # sets the kubernetes node role
-      roles: [agent,foobar]
+      roles: [azure]
 
       # dynamic kubelet configuration
       # see https://kubernetes.io/docs/tasks/administer-cluster/reconfigure-kubelet/

manager/manager.go

@@ -27,6 +27,8 @@ type (
 		ctx       context.Context
 		k8sClient *kubernetes.Clientset
 
+		nodePatchStatus map[string]bool
+
 		prometheus struct {
 			nodePoolStatus *prometheus.GaugeVec
 			nodeApplied    *prometheus.GaugeVec
@@ -36,6 +38,7 @@ type (
 
 func (m *KubePoolManager) Init() {
 	m.ctx = context.Background()
+	m.nodePatchStatus = map[string]bool{}
 	m.initK8s()
 	m.initPrometheus()
 }
@@ -87,10 +90,19 @@ func (r *KubePoolManager) initK8s() {
 func (m *KubePoolManager) Start() {
 	go func() {
 		m.leaderElect()
+
+		log.Info("initial node pool apply")
+		m.startupApply()
+
 		for {
 			log.Info("(re)starting node watch")
 			if err := m.startNodeWatch(); err != nil {
-				log.Errorf("node watcher stopped: %v", err)
+				log.Warnf("node watcher stopped: %v", err)
+			}
+
+			if m.Opts.K8s.ReapplyOnWatchTimeout {
+				log.Info("reapply node pool settings")
+				m.startupApply()
 			}
 		}
 	}()
@@ -116,6 +128,20 @@ func (m *KubePoolManager) leaderElect() {
 	}
 }
 
+func (m *KubePoolManager) startupApply() {
+	listOpts := metav1.ListOptions{}
+	nodeList, err := m.k8sClient.CoreV1().Nodes().List(m.ctx, listOpts)
+	if err != nil {
+		log.Panic(err)
+	}
+
+	m.nodePatchStatus = map[string]bool{}
+	for _, node := range nodeList.Items {
+		m.nodePatchStatus[node.Name] = true
+		m.applyNode(&node)
+	}
+}
+
 func (m *KubePoolManager) startNodeWatch() error {
 	timeout := int64(m.Opts.K8s.WatchTimeout.Seconds())
 	watchOpts := metav1.ListOptions{
@@ -131,9 +157,20 @@ func (m *KubePoolManager) startNodeWatch() error {
 
 	for res := range nodeWatcher.ResultChan() {
 		switch res.Type {
-		case watch.Added:
+		case watch.Modified:
 			if node, ok := res.Object.(*corev1.Node); ok {
-				m.applyNode(node)
+				if _, exists := m.nodePatchStatus[node.Name]; !exists {
+					m.nodePatchStatus[node.Name] = false
+				}
+
+				if !m.nodePatchStatus[node.Name] && m.checkNodeCondition(node) {
+					m.applyNode(node)
+					m.nodePatchStatus[node.Name] = true
+				}
+			}
+		case watch.Deleted:
+			if node, ok := res.Object.(*corev1.Node); ok {
+				delete(m.nodePatchStatus, node.Name)
 			}
 		case watch.Error:
 			log.Errorf("go watch error event %v", res.Object)
@@ -143,6 +180,16 @@ func (m *KubePoolManager) startNodeWatch() error {
 	return fmt.Errorf("terminated")
 }
 
+func (m *KubePoolManager) checkNodeCondition(node *corev1.Node) bool {
+	for _, condition := range node.Status.Conditions {
+		if stringCompare(string(condition.Type), "ready") && stringCompare(condition.Reason, "kubeletready") && stringCompare(string(condition.Status), "true") {
+			return true
+		}
+	}
+
+	return false
+}
+
 func (m *KubePoolManager) applyNode(node *corev1.Node) {
 	contextLogger := log.WithField("node", node.Name)
 

manager/misc.go

@@ -0,0 +1,7 @@
+package manager
+
+import "strings"
+
+func stringCompare(a, b string) bool {
+	return strings.EqualFold(a, b)
+}