Improve health check for cache synchronization (#498)

* improve Control flow logic
* drop log,comment,sleep
* update user-guide.md
* set cache-sync-timeout default to 1m

Author: jimmy-zh

docs/user-guide.md

@@ -31,6 +31,7 @@ Usage of kube-router:
       --advertise-loadbalancer-ip        Add LoadbBalancer IP of service status as set by the LB provider to the RIB so that it gets advertised to the BGP peers.
       --advertise-pod-cidr               Add Node's POD cidr to the RIB so that it gets advertised to the BGP peers. (default true)
       --bgp-graceful-restart             Enables the BGP Graceful Restart capability so that routes are preserved on unexpected restarts
+      --cache-sync-timeout duration      The timeout for cache synchronization (e.g. '5s', '1m'). Must be greater than 0. (default 1m0s)
       --cleanup-config                   Cleanup iptables rules, ipvs, ipset configuration and exit.
       --cluster-asn uint                 ASN number under which cluster nodes will run iBGP.
       --cluster-cidr string              CIDR range of pods in the cluster. It is used to identify traffic originating from and destinated to pods.

pkg/cmd/kube-router.go

@@ -21,6 +21,7 @@ import (
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
+	"time"
 )
 
 // These get set at build time via -ldflags magic
@@ -76,24 +77,40 @@ func CleanupConfigAndExit() {
 func (kr *KubeRouter) Run() error {
 	var err error
 	var wg sync.WaitGroup
-
 	healthChan := make(chan *healthcheck.ControllerHeartbeat, 10)
 	defer close(healthChan)
-
 	stopCh := make(chan struct{})
 
+	if !(kr.Config.RunFirewall || kr.Config.RunServiceProxy || kr.Config.RunRouter) {
+		glog.Info("Router, Firewall or Service proxy functionality must be specified. Exiting!")
+		os.Exit(0)
+	}
+
 	hc, err := healthcheck.NewHealthController(kr.Config)
 	if err != nil {
 		return errors.New("Failed to create health controller: " + err.Error())
 	}
 	wg.Add(1)
-	go hc.Run(healthChan, stopCh, &wg)
+	go hc.RunServer(stopCh, &wg)
 
-	if !(kr.Config.RunFirewall || kr.Config.RunServiceProxy || kr.Config.RunRouter) {
-		glog.Info("Router, Firewall or Service proxy functionality must be specified. Exiting!")
-		os.Exit(0)
+	informerFactory := informers.NewSharedInformerFactory(kr.Client, 0)
+	svcInformer := informerFactory.Core().V1().Services().Informer()
+	epInformer := informerFactory.Core().V1().Endpoints().Informer()
+	podInformer := informerFactory.Core().V1().Pods().Informer()
+	nodeInformer := informerFactory.Core().V1().Nodes().Informer()
+	nsInformer := informerFactory.Core().V1().Namespaces().Informer()
+	npInformer := informerFactory.Networking().V1().NetworkPolicies().Informer()
+	informerFactory.Start(stopCh)
+
+	err = kr.CacheSyncOrTimeout(informerFactory, stopCh)
+	if err != nil {
+		return errors.New("Failed to synchronize cache: " + err.Error())
 	}
 
+	hc.SetAlive()
+	wg.Add(1)
+	go hc.RunCheck(healthChan, stopCh, &wg)
+
 	if (kr.Config.MetricsPort > 0) && (kr.Config.MetricsPort <= 65535) {
 		kr.Config.MetricsEnabled = true
 		mc, err := metrics.NewMetricsController(kr.Client, kr.Config)
@@ -110,18 +127,6 @@ func (kr *KubeRouter) Run() error {
 		kr.Config.MetricsEnabled = false
 	}
 
-	informerFactory := informers.NewSharedInformerFactory(kr.Client, 0)
-
-	svcInformer := informerFactory.Core().V1().Services().Informer()
-	epInformer := informerFactory.Core().V1().Endpoints().Informer()
-	podInformer := informerFactory.Core().V1().Pods().Informer()
-	nodeInformer := informerFactory.Core().V1().Nodes().Informer()
-	nsInformer := informerFactory.Core().V1().Namespaces().Informer()
-	npInformer := informerFactory.Networking().V1().NetworkPolicies().Informer()
-
-	informerFactory.Start(stopCh)
-	informerFactory.WaitForCacheSync(stopCh)
-
 	if kr.Config.RunFirewall {
 		npc, err := netpol.NewNetworkPolicyController(kr.Client,
 			kr.Config, podInformer, npInformer, nsInformer)
@@ -177,6 +182,22 @@ func (kr *KubeRouter) Run() error {
 	return nil
 }
 
+// CacheSync performs cache synchronization under timeout limit
+func (kr *KubeRouter) CacheSyncOrTimeout(informerFactory informers.SharedInformerFactory, stopCh <-chan struct{}) error {
+	syncOverCh := make(chan struct{})
+	go func() {
+		informerFactory.WaitForCacheSync(stopCh)
+		close(syncOverCh)
+	}()
+
+	select {
+	case <-time.After(kr.Config.CacheSyncTimeout):
+		return errors.New(kr.Config.CacheSyncTimeout.String() + " timeout")
+	case <-syncOverCh:
+		return nil
+	}
+}
+
 func PrintVersion(logOutput bool) {
 	output := fmt.Sprintf("Running %v version %s, built on %s, %s\n", os.Args[0], version, buildDate, runtime.Version())
 

pkg/healthcheck/health_controller.go

@@ -107,7 +107,6 @@ func (hc *HealthController) CheckHealth() bool {
 	graceTime := time.Duration(1500 * time.Millisecond)
 
 	if hc.Config.RunFirewall {
-
 		if time.Since(hc.Status.NetworkPolicyControllerAlive) > hc.Config.IPTablesSyncPeriod+hc.Status.NetworkPolicyControllerAliveTTL+graceTime {
 			glog.Error("Network Policy Controller heartbeat missed")
 			health = false
@@ -138,16 +137,11 @@ func (hc *HealthController) CheckHealth() bool {
 	return health
 }
 
-//Run starts the HealthController
-func (hc *HealthController) Run(healthChan <-chan *ControllerHeartbeat, stopCh <-chan struct{}, wg *sync.WaitGroup) error {
-	t := time.NewTicker(5000 * time.Millisecond)
+//RunServer starts the HealthController's server
+func (hc *HealthController) RunServer(stopCh <-chan struct{}, wg *sync.WaitGroup) error {
 	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() {
@@ -162,15 +156,26 @@ func (hc *HealthController) Run(healthChan <-chan *ControllerHeartbeat, stopCh <
 		hc.HTTPEnabled = false
 	}
 
+	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
+	}
+}
+
+//RunCheck starts the HealthController's check
+func (hc *HealthController) RunCheck(healthChan <-chan *ControllerHeartbeat, stopCh <-chan struct{}, wg *sync.WaitGroup) error {
+	t := time.NewTicker(5000 * time.Millisecond)
+	defer wg.Done()
 	for {
 		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)
-				}
-			}
+			glog.Infof("Shutting down HealthController RunCheck")
 			return nil
 		case heartbeat := <-healthChan:
 			hc.HandleHeartbeat(heartbeat)
@@ -179,7 +184,16 @@ func (hc *HealthController) Run(healthChan <-chan *ControllerHeartbeat, stopCh <
 		}
 		hc.Status.Healthy = hc.CheckHealth()
 	}
+}
+
+func (hc *HealthController) SetAlive() {
+
+	now := time.Now()
 
+	hc.Status.MetricsControllerAlive = now
+	hc.Status.NetworkPolicyControllerAlive = now
+	hc.Status.NetworkRoutingControllerAlive = now
+	hc.Status.NetworkServicesControllerAlive = now
 }
 
 //NewHealthController creates a new health controller and returns a reference to it
@@ -188,11 +202,7 @@ func NewHealthController(config *options.KubeRouterConfig) (*HealthController, e
 		Config:     config,
 		HealthPort: config.HealthPort,
 		Status: HealthStats{
-			Healthy:                        true,
-			MetricsControllerAlive:         time.Now(),
-			NetworkPolicyControllerAlive:   time.Now(),
-			NetworkRoutingControllerAlive:  time.Now(),
-			NetworkServicesControllerAlive: time.Now(),
+			Healthy: true,
 		},
 	}
 	return &hc, nil

pkg/options/options.go

@@ -13,6 +13,7 @@ type KubeRouterConfig struct {
 	AdvertiseNodePodCidr    bool
 	AdvertiseLoadBalancerIp bool
 	BGPGracefulRestart      bool
+	CacheSyncTimeout        time.Duration
 	CleanupConfig           bool
 	ClusterAsn              uint
 	ClusterCIDR             string
@@ -50,6 +51,7 @@ type KubeRouterConfig struct {
 
 func NewKubeRouterConfig() *KubeRouterConfig {
 	return &KubeRouterConfig{
+		CacheSyncTimeout:   1 * time.Minute,
 		IpvsSyncPeriod:     5 * time.Minute,
 		IPTablesSyncPeriod: 5 * time.Minute,
 		RoutesSyncPeriod:   5 * time.Minute,
@@ -62,6 +64,8 @@ func (s *KubeRouterConfig) AddFlags(fs *pflag.FlagSet) {
 		"Print usage information.")
 	fs.BoolVarP(&s.Version, "version", "V", false,
 		"Print version information.")
+	fs.DurationVar(&s.CacheSyncTimeout, "cache-sync-timeout", s.CacheSyncTimeout,
+		"The timeout for cache synchronization (e.g. '5s', '1m'). Must be greater than 0.")
 	fs.BoolVar(&s.RunServiceProxy, "run-service-proxy", true,
 		"Enables Service Proxy -- sets up IPVS for Kubernetes Services.")
 	fs.BoolVar(&s.RunFirewall, "run-firewall", true,