Merge pull request #298 from roffe/healthcheck

Healthchecks

Author: roffe

Documentation/health.md

@@ -0,0 +1,24 @@
+# Health checking kube-router
+
+kube-router currently has basic health checking in form of heartbeats sent from each controller to the healthcontroller each time the main loop completes successfully.
+
+The health port is by default 20244 but can be changed with the startup option.
+The health path is `/healthz`
+
+    --health-port=<port number>
+
+If port is set to 0 (zero) no HTTP endpoint will be made availible but the health controller will still run and print out any missed heartbeats to STDERR of kube-router
+
+If a controller does not send a heartbeat within controllersynctime + 5 seconds the component will be flagged as unhealthy.
+
+If any of the running components is failing the whole kube-router state will be marked as failed in the /healthz endpoint
+
+E.g
+
+kube-router is started with
+
+    --run-router=true
+    --run-firewall=true
+    --run-service-proxy=true
+
+If the route controller, policy controller or service controller exits it's main loop and does not publish a heartbeat the /healthz endpoint will return a error 500 signaling that kube-router is not healthy.

app/controllers/health_controller.go

@@ -0,0 +1,180 @@
+package controllers
+
+import (
+	"net/http"
+	"strconv"
+	"sync"
+	"time"
+
+	"github.com/cloudnativelabs/kube-router/app/options"
+	"github.com/golang/glog"
+	"golang.org/x/net/context"
+)
+
+//ControllerHeartbeat is the structure to hold the heartbeats sent by controlers
+type ControllerHeartbeat struct {
+	Component     string
+	Lastheartbeat time.Time
+}
+
+//HealthController reports the health of the controller loops as a http endpoint
+type HealthController struct {
+	HealthPort  uint16
+	HTTPenabled bool
+	Status      HealthStats
+	Config      *options.KubeRouterConfig
+}
+
+//HealthStats is holds the latest heartbeats
+type HealthStats struct {
+	sync.Mutex
+	Healthy                        bool
+	MetricsControllerAlive         time.Time
+	NetworkPolicyControllerAlive   time.Time
+	NetworkRoutingControllerAlive  time.Time
+	NetworkServicesControllerAlive time.Time
+}
+
+//sendHeartBeat sends a heartbeat on the passed channel
+func sendHeartBeat(channel chan<- *ControllerHeartbeat, controller string) {
+	heartbeat := ControllerHeartbeat{
+		Component:     controller,
+		Lastheartbeat: time.Now(),
+	}
+	channel <- &heartbeat
+}
+
+//Handler writes HTTP responses to the health path
+func (hc *HealthController) Handler(w http.ResponseWriter, req *http.Request) {
+	if hc.Status.Healthy {
+		w.WriteHeader(http.StatusOK)
+		w.Write([]byte("OK\n"))
+	} else {
+		w.WriteHeader(http.StatusInternalServerError)
+		/*
+			statusText := fmt.Sprintf("Service controller last alive %s\n ago"+
+				"Routing controller last alive: %s\n ago"+
+				"Policy controller last alive: %s\n ago"+
+				"Metrics controller last alive: %s\n ago",
+				time.Since(hc.Status.NetworkServicesControllerAlive),
+				time.Since(hc.Status.NetworkRoutingControllerAlive),
+				time.Since(hc.Status.NetworkPolicyControllerAlive),
+				time.Since(hc.Status.MetricsControllerAlive))
+			w.Write([]byte(statusText))
+		*/
+		w.Write([]byte("Unhealthy"))
+	}
+}
+
+//HandleHeartbeat handles recevied heartbeats onthe health channel
+func (hc *HealthController) HandleHeartbeat(beat *ControllerHeartbeat) {
+	glog.V(3).Infof("Received heartbeat from %s", beat.Component)
+
+	hc.Status.Lock()
+	defer hc.Status.Unlock()
+
+	switch component := beat.Component; component {
+	case "NSC":
+		hc.Status.NetworkServicesControllerAlive = time.Now()
+	case "NRC":
+		hc.Status.NetworkRoutingControllerAlive = time.Now()
+	case "NPC":
+		hc.Status.NetworkPolicyControllerAlive = time.Now()
+	case "MC":
+		hc.Status.MetricsControllerAlive = time.Now()
+	}
+}
+
+//CheckHealth evaluates the time since last heartbeat to decide if the controller is running or not
+func (hc *HealthController) CheckHealth() bool {
+	health := true
+	if hc.Config.RunFirewall {
+		if time.Since(hc.Status.NetworkPolicyControllerAlive) > hc.Config.IPTablesSyncPeriod+5*time.Second {
+			glog.Error("Network Policy Controller heartbeat missed")
+			health = false
+		}
+	}
+
+	if hc.Config.RunRouter {
+		if time.Since(hc.Status.NetworkRoutingControllerAlive) > hc.Config.RoutesSyncPeriod+5*time.Second {
+			glog.Error("Network Routing Controller heartbeat missed")
+			health = false
+		}
+	}
+
+	if hc.Config.RunServiceProxy {
+		if time.Since(hc.Status.NetworkServicesControllerAlive) > hc.Config.IpvsSyncPeriod+5*time.Second {
+			glog.Error("NetworkService Controller heartbeat missed")
+			health = false
+		}
+	}
+
+	if hc.Config.MetricsEnabled {
+		if time.Since(hc.Status.MetricsControllerAlive) > 5*time.Second {
+			glog.Error("Metrics Controller heartbeat missed")
+			health = false
+		}
+	}
+
+	return health
+}
+
+//Run starts the HealthController
+func (hc *HealthController) Run(healthChan <-chan *ControllerHeartbeat, stopCh <-chan struct{}, wg *sync.WaitGroup) error {
+	Started := time.Now()
+	t := time.NewTicker(500 * time.Millisecond)
+	defer wg.Done()
+	glog.Info("Starting health controller")
+
+	srv := &http.Server{Addr: ":" + strconv.Itoa(int(hc.HealthPort)), Handler: http.DefaultServeMux}
+
+	http.HandleFunc("/healthz", hc.Handler)
+
+	if (hc.Config.HealthPort > 0) && (hc.Config.HealthPort <= 65535) {
+		hc.HTTPenabled = true
+		go func() {
+			if err := srv.ListenAndServe(); err != nil {
+				// cannot panic, because this probably is an intentional close
+				glog.Errorf("Health controller error: %s", err)
+			}
+		}()
+	} else if hc.Config.MetricsPort > 65535 {
+		glog.Errorf("Metrics port must be over 0 and under 65535, given port: %d", hc.Config.MetricsPort)
+	} else {
+		hc.HTTPenabled = false
+	}
+
+	for {
+		//Give the controllers a few seconds to start before checking health
+		if time.Since(Started) > 5*time.Second {
+			hc.Status.Healthy = hc.CheckHealth()
+		}
+		select {
+		case <-stopCh:
+			glog.Infof("Shutting down health controller")
+			if hc.HTTPenabled {
+				if err := srv.Shutdown(context.Background()); err != nil {
+					glog.Errorf("could not shutdown: %v", err)
+				}
+			}
+			return nil
+		case heartbeat := <-healthChan:
+			hc.HandleHeartbeat(heartbeat)
+		case <-t.C:
+			glog.V(4).Info("Health controller tick")
+		}
+	}
+
+}
+
+//NewHealthController creates a new healh controller and returns a reference to it
+func NewHealthController(config *options.KubeRouterConfig) (*HealthController, error) {
+	hc := HealthController{
+		Config:     config,
+		HealthPort: config.HealthPort,
+		Status: HealthStats{
+			Healthy: false,
+		},
+	}
+	return &hc, nil
+}

app/controllers/metrics_controller.go

@@ -5,6 +5,7 @@ import (
 	"net/http"
 	"strconv"
 	"sync"
+	"time"
 
 	"github.com/cloudnativelabs/kube-router/app/options"
 	"github.com/golang/glog"
@@ -118,7 +119,8 @@ type MetricsController struct {
 }
 
 // Run prometheus metrics controller
-func (mc *MetricsController) Run(stopCh <-chan struct{}, wg *sync.WaitGroup) error {
+func (mc *MetricsController) Run(healthChan chan<- *ControllerHeartbeat, stopCh <-chan struct{}, wg *sync.WaitGroup) error {
+	t := time.NewTicker(3 * time.Second)
 	defer wg.Done()
 	glog.Info("Starting metrics controller")
 
@@ -136,13 +138,19 @@ func (mc *MetricsController) Run(stopCh <-chan struct{}, wg *sync.WaitGroup) err
 			glog.Errorf("Metrics controller error: %s", err)
 		}
 	}()
-
-	<-stopCh
-	glog.Infof("Shutting down metrics controller")
-	if err := srv.Shutdown(context.Background()); err != nil {
-		glog.Errorf("could not shutdown: %v", err)
+	for {
+		sendHeartBeat(healthChan, "MC")
+		select {
+		case <-stopCh:
+			glog.Infof("Shutting down metrics controller")
+			if err := srv.Shutdown(context.Background()); err != nil {
+				glog.Errorf("could not shutdown: %v", err)
+			}
+			return nil
+		case <-t.C:
+			glog.V(4).Info("Metrics controller tick")
+		}
 	}
-	return nil
 }
 
 // NewMetricsController returns new MetricController object

app/controllers/network_policy_controller.go

@@ -100,7 +100,7 @@ type protocolAndPort struct {
 }
 
 // Run runs forver till we receive notification on stopCh
-func (npc *NetworkPolicyController) Run(stopCh <-chan struct{}, wg *sync.WaitGroup) {
+func (npc *NetworkPolicyController) Run(healthChan chan<- *ControllerHeartbeat, stopCh <-chan struct{}, wg *sync.WaitGroup) {
 	t := time.NewTicker(npc.syncPeriod)
 	defer t.Stop()
 	defer wg.Done()
@@ -121,6 +121,8 @@ func (npc *NetworkPolicyController) Run(stopCh <-chan struct{}, wg *sync.WaitGro
 			err := npc.Sync()
 			if err != nil {
 				glog.Errorf("Error during periodic sync: " + err.Error())
+			} else {
+				sendHeartBeat(healthChan, "NPC")
 			}
 		} else {
 			continue

app/controllers/network_routes_controller.go

@@ -82,7 +82,7 @@ const (
 )
 
 // Run runs forever until we are notified on stop channel
-func (nrc *NetworkRoutingController) Run(stopCh <-chan struct{}, wg *sync.WaitGroup) {
+func (nrc *NetworkRoutingController) Run(healthChan chan<- *ControllerHeartbeat, stopCh <-chan struct{}, wg *sync.WaitGroup) {
 	cidr, err := utils.GetPodCidrFromCniSpec("/etc/cni/net.d/10-kuberouter.conf")
 	if err != nil {
 		glog.Errorf("Failed to get pod CIDR from CNI conf file: %s", err.Error())
@@ -254,6 +254,8 @@ func (nrc *NetworkRoutingController) Run(stopCh <-chan struct{}, wg *sync.WaitGr
 			nrc.syncInternalPeers()
 		}
 
+		sendHeartBeat(healthChan, "NRC")
+
 		select {
 		case <-stopCh:
 			glog.Infof("Shutting down network routes controller")

app/controllers/network_services_controller.go

@@ -99,7 +99,7 @@ type endpointsInfo struct {
 type endpointsInfoMap map[string][]endpointsInfo
 
 // Run periodically sync ipvs configuration to reflect desired state of services and endpoints
-func (nsc *NetworkServicesController) Run(stopCh <-chan struct{}, wg *sync.WaitGroup) error {
+func (nsc *NetworkServicesController) Run(healthChan chan<- *ControllerHeartbeat, stopCh <-chan struct{}, wg *sync.WaitGroup) error {
 
 	t := time.NewTicker(nsc.syncPeriod)
 	defer t.Stop()
@@ -130,7 +130,12 @@ func (nsc *NetworkServicesController) Run(stopCh <-chan struct{}, wg *sync.WaitG
 
 		if watchers.PodWatcher.HasSynced() && watchers.NetworkPolicyWatcher.HasSynced() {
 			glog.V(1).Info("Performing periodic sync of ipvs services")
-			nsc.sync()
+			err := nsc.sync()
+			if err != nil {
+				glog.Errorf("Error during periodic ipvs sync: " + err.Error())
+			} else {
+				sendHeartBeat(healthChan, "NSC")
+			}
 		} else {
 			continue
 		}
@@ -144,20 +149,28 @@ func (nsc *NetworkServicesController) Run(stopCh <-chan struct{}, wg *sync.WaitG
 	}
 }
 
-func (nsc *NetworkServicesController) sync() {
+func (nsc *NetworkServicesController) sync() error {
+	var err error
 	nsc.mu.Lock()
 	defer nsc.mu.Unlock()
 
 	nsc.serviceMap = buildServicesInfo()
 	nsc.endpointsMap = buildEndpointsInfo()
-	err := nsc.syncHairpinIptablesRules()
+	err = nsc.syncHairpinIptablesRules()
 	if err != nil {
 		glog.Errorf("Error syncing hairpin iptable rules: %s", err.Error())
 	}
-	nsc.syncIpvsServices(nsc.serviceMap, nsc.endpointsMap)
+
+	err = nsc.syncIpvsServices(nsc.serviceMap, nsc.endpointsMap)
+	if err != nil {
+		glog.Errorf("Error syncing IPVS services: %s", err.Error())
+		return err
+	}
+
 	if nsc.MetricsEnabled {
 		nsc.publishMetrics(nsc.serviceMap)
 	}
+	return nil
 }
 
 func (nsc *NetworkServicesController) publishMetrics(serviceInfoMap serviceInfoMap) error {

app/options/options.go

@@ -41,6 +41,7 @@ type KubeRouterConfig struct {
 	MetricsPort         uint16
 	MetricsPath         string
 	VLevel              string
+	HealthPort          uint16
 	// FullMeshPassword    string
 }
 
@@ -113,10 +114,11 @@ func (s *KubeRouterConfig) AddFlags(fs *pflag.FlagSet) {
 		"Password for authenticating against the BGP peer defined with \"--peer-router-ips\".")
 	fs.BoolVar(&s.EnablePprof, "enable-pprof", false,
 		"Enables pprof for debugging performance and memory leak issues.")
-	fs.Uint16Var(&s.MetricsPort, "metrics-port", 0, "Prometheus metrics port, 0 = Disabled")
+	fs.Uint16Var(&s.MetricsPort, "metrics-port", 0, "Prometheus metrics port, (Default 0, Disabled)")
 	fs.StringVar(&s.MetricsPath, "metrics-path", "/metrics", "Prometheus metrics path")
 	// fs.StringVar(&s.FullMeshPassword, "nodes-full-mesh-password", s.FullMeshPassword,
 	// 	"Password that cluster-node BGP servers will use to authenticate one another when \"--nodes-full-mesh\" is set.")
 	fs.StringVarP(&s.VLevel, "v", "v", "0", "log level for V logs")
+	fs.Uint16Var(&s.HealthPort, "health-port", 20244, "Health check port, 0 = Disabled")
 
 }