Merge pull request #19 from utilitywarehouse/replace-old-hc

Replace old healthcheck

deploy/example/namespace/daemonset.yaml

@@ -41,12 +41,15 @@ spec:
             capabilities:
               add:
                 - 'NET_ADMIN'
-          readinessProbe:
+          livenessProbe:
             httpGet:
               path: /healthz
               port: readiness-port
-            periodSeconds: 10
-            failureThreshold: 1
+            periodSeconds: 30
+            failureThreshold: 6
+            initialDelaySeconds: 10
+            successThreshold: 1
+            timeoutSeconds: 1
       volumes:
         - name: var-lib-semaphore-wireguard
           hostPath:

kube/node_watcher.go

@@ -29,8 +29,6 @@ type NodeWatcher struct {
 	store        cache.Store
 	controller   cache.Controller
 	eventHandler NodeEventHandler
-	ListHealthy  bool
-	WatchHealthy bool
 }
 
 // NewNodeWatcher returns a new node wathcer.
@@ -53,9 +51,6 @@ func (nw *NodeWatcher) Init() {
 			if err != nil {
 				log.Logger.Error("nw: list error", "err", err)
 				metrics.IncNodeWatcherFailures(nw.clusterName, "list")
-				nw.ListHealthy = false
-			} else {
-				nw.ListHealthy = true
 			}
 			return l, err
 		},
@@ -64,9 +59,6 @@ func (nw *NodeWatcher) Init() {
 			if err != nil {
 				log.Logger.Error("nw: watch error", "err", err)
 				metrics.IncNodeWatcherFailures(nw.clusterName, "watch")
-				nw.WatchHealthy = false
-			} else {
-				nw.WatchHealthy = true
 			}
 			return w, err
 		},
@@ -117,8 +109,3 @@ func (nw *NodeWatcher) List() ([]*v1.Node, error) {
 	}
 	return nodes, nil
 }
-
-// Healthy is true when both list and watch handlers are running without errors.
-func (nw *NodeWatcher) Healthy() bool {
-	return nw.ListHealthy && nw.WatchHealthy
-}

main.go

@@ -162,8 +162,12 @@ func listenAndServe(runners []*Runner) {
 	mux := http.NewServeMux()
 	mux.Handle("/metrics", promhttp.Handler())
 	mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) {
+		// if we reach listenAndServe func the wg devices should have
+		// been intialised and running. One could use the ruuners'
+		// initialised flag for a liveness probe to kick the deployment
+		// after some time
 		for _, r := range runners {
-			if !r.Healthy() {
+			if !r.initialised {
 				w.WriteHeader(http.StatusServiceUnavailable)
 				return
 			}

metrics/metrics.go

@@ -225,6 +225,7 @@ func (c *collector) Collect(ch chan<- prometheus.Metric) {
 	}
 }
 
+// SyncPeerAttempt increases the counter for attempts to sync wg peers
 func SyncPeerAttempt(device string, err error) {
 	s := "1"
 	if err != nil {
@@ -236,18 +237,21 @@ func SyncPeerAttempt(device string, err error) {
 	}).Inc()
 }
 
+// IncSyncQueueFullFailures increases sync queue failures counter
 func IncSyncQueueFullFailures(device string) {
 	syncQueueFullFailures.With(prometheus.Labels{
 		"device": device,
 	}).Inc()
 }
 
+// IncSyncRequeue increases requeue counter
 func IncSyncRequeue(device string) {
 	syncRequeue.With(prometheus.Labels{
 		"device": device,
 	}).Inc()
 }
 
+// IncNodeWatcherFailures increases node watcher failures counter
 func IncNodeWatcherFailures(c, v string) {
 	nodeWatcherFailures.With(prometheus.Labels{
 		"cluster": c,

runner.go

@@ -53,6 +53,7 @@ type Runner struct {
 	nodeWatcher *kube.NodeWatcher
 	peers       map[string]Peer
 	canSync     bool // Flag to allow updating wireguard peers only after initial node watcher sync
+	initialised bool // Flag to turn on after the successful initialisation of the runner to report healthy
 	annotations RunnerAnnotations
 	sync        chan struct{}
 	stop        chan struct{}
@@ -65,6 +66,7 @@ func newRunner(client, watchClient kubernetes.Interface, nodeName, wgDeviceName,
 		podSubnet:   podSubnet,
 		peers:       make(map[string]Peer),
 		canSync:     false,
+		initialised: false,
 		annotations: constructRunnerAnnotations(localClusterName, remoteClusterName),
 		sync:        make(chan struct{}),
 		stop:        make(chan struct{}),
@@ -100,17 +102,20 @@ func (r *Runner) Run() error {
 	if err := r.device.EnsureLinkUp(); err != nil {
 		return err
 	}
+	// Static route to the whole subnet cidr
+	if err := r.device.AddRouteToNet(r.podSubnet); err != nil {
+		return err
+	}
+	// At this point the runner should be considered successfully initialised
+	r.initialised = true
+
 	go r.nodeWatcher.Run()
 	// wait for node watcher to sync. TODO: atm dummy and could run forever
 	// if node cache fails to sync
 	stopCh := make(chan struct{})
 	if ok := cache.WaitForNamedCacheSync("nodeWatcher", stopCh, r.nodeWatcher.HasSynced); !ok {
 		return fmt.Errorf("failed to wait for nodes cache to sync")
 	}
-	// Static route to the whole subnet cidr
-	if err := r.device.AddRouteToNet(r.podSubnet); err != nil {
-		return err
-	}
 	r.canSync = true
 	r.enqueuePeersSync()
 	return nil
@@ -308,8 +313,3 @@ func (r *Runner) onPeerNodeDelete(node *v1.Node) {
 	}
 	r.enqueuePeersSync()
 }
-
-// Healthy is true if the node watcher is reporting healthy.
-func (r *Runner) Healthy() bool {
-	return r.nodeWatcher.Healthy()
-}