Merge pull request #1133 from bprashanth/ubernetes_healthchecks

Ubernetes multizone and custom healthchecks

Author: bprashanth

ingress/controllers/gce/backends/backends.go

@@ -107,9 +107,9 @@ func (b *Backends) Get(port int64) (*compute.BackendService, error) {
 	return be, nil
 }
 
-func (b *Backends) create(ig *compute.InstanceGroup, namedPort *compute.NamedPort, name string) (*compute.BackendService, error) {
+func (b *Backends) create(igs []*compute.InstanceGroup, namedPort *compute.NamedPort, name string) (*compute.BackendService, error) {
 	// Create a new health check
-	if err := b.healthChecker.Add(namedPort.Port, ""); err != nil {
+	if err := b.healthChecker.Add(namedPort.Port); err != nil {
 		return nil, err
 	}
 	hc, err := b.healthChecker.Get(namedPort.Port)
@@ -120,11 +120,7 @@ func (b *Backends) create(ig *compute.InstanceGroup, namedPort *compute.NamedPor
 	backend := &compute.BackendService{
 		Name:     name,
 		Protocol: "HTTP",
-		Backends: []*compute.Backend{
-			{
-				Group: ig.SelfLink,
-			},
-		},
+		Backends: getBackendsForIGs(igs),
 		// Api expects one, means little to kubernetes.
 		HealthChecks: []string{hc.SelfLink},
 		Port:         namedPort.Port,
@@ -143,20 +139,24 @@ func (b *Backends) Add(port int64) error {
 	be := &compute.BackendService{}
 	defer func() { b.snapshotter.Add(portKey(port), be) }()
 
-	ig, namedPort, err := b.nodePool.AddInstanceGroup(b.namer.IGName(), port)
+	igs, namedPort, err := b.nodePool.AddInstanceGroup(b.namer.IGName(), port)
 	if err != nil {
 		return err
 	}
 	be, _ = b.Get(port)
 	if be == nil {
-		glog.Infof("Creating backend for instance group %v port %v named port %v",
-			ig.Name, port, namedPort)
-		be, err = b.create(ig, namedPort, b.namer.BeName(port))
+		glog.Infof("Creating backend for %d instance groups, port %v named port %v",
+			len(igs), port, namedPort)
+		be, err = b.create(igs, namedPort, b.namer.BeName(port))
 		if err != nil {
 			return err
 		}
 	}
-	if err := b.edgeHop(be, ig); err != nil {
+	// we won't find any igs till the node pool syncs nodes.
+	if len(igs) == 0 {
+		return nil
+	}
+	if err := b.edgeHop(be, igs); err != nil {
 		return err
 	}
 	return err
@@ -201,18 +201,31 @@ func (b *Backends) List() ([]interface{}, error) {
 	return interList, nil
 }
 
+func getBackendsForIGs(igs []*compute.InstanceGroup) []*compute.Backend {
+	backends := []*compute.Backend{}
+	for _, ig := range igs {
+		backends = append(backends, &compute.Backend{Group: ig.SelfLink})
+	}
+	return backends
+}
+
 // edgeHop checks the links of the given backend by executing an edge hop.
 // It fixes broken links.
-func (b *Backends) edgeHop(be *compute.BackendService, ig *compute.InstanceGroup) error {
-	if len(be.Backends) == 1 &&
-		utils.CompareLinks(be.Backends[0].Group, ig.SelfLink) {
-		return nil
+func (b *Backends) edgeHop(be *compute.BackendService, igs []*compute.InstanceGroup) error {
+	beIGs := sets.String{}
+	for _, beToIG := range be.Backends {
+		beIGs.Insert(beToIG.Group)
 	}
-	glog.Infof("Backend %v has a broken edge, adding link to %v",
-		be.Name, ig.Name)
-	be.Backends = []*compute.Backend{
-		{Group: ig.SelfLink},
+	igLinks := sets.String{}
+	for _, igToBE := range igs {
+		igLinks.Insert(igToBE.SelfLink)
+	}
+	if igLinks.Equal(beIGs) {
+		return nil
 	}
+	glog.Infof("Backend %v has a broken edge, expected igs %+v, current igs %+v",
+		be.Name, igLinks.List(), beIGs.List())
+	be.Backends = getBackendsForIGs(igs)
 	if err := b.cloud.UpdateBackendService(be); err != nil {
 		return err
 	}

ingress/controllers/gce/backends/backends_test.go

@@ -27,12 +27,16 @@ import (
 	"k8s.io/kubernetes/pkg/util/sets"
 )
 
+const defaultZone = "zone-a"
+
 func newBackendPool(f BackendServices, fakeIGs instances.InstanceGroups, syncWithCloud bool) BackendPool {
 	namer := &utils.Namer{}
+	nodePool := instances.NewNodePool(fakeIGs)
+	nodePool.Init(&instances.FakeZoneLister{[]string{defaultZone}})
+	healthChecks := healthchecks.NewHealthChecker(healthchecks.NewFakeHealthChecks(), "/", namer)
+	healthChecks.Init(&healthchecks.FakeHealthCheckGetter{nil})
 	return NewBackendPool(
-		f,
-		healthchecks.NewHealthChecker(healthchecks.NewFakeHealthChecks(), "/", namer),
-		instances.NewNodePool(fakeIGs, "default-zone"), namer, []int64{}, syncWithCloud)
+		f, healthChecks, nodePool, namer, []int64{}, syncWithCloud)
 }
 
 func TestBackendPoolAdd(t *testing.T) {
@@ -80,8 +84,14 @@ func TestBackendPoolAdd(t *testing.T) {
 			t.Fatalf("Unexpected create for existing backend service")
 		}
 	}
-	gotBackend, _ := f.GetBackendService(beName)
-	gotGroup, _ := fakeIGs.GetInstanceGroup(namer.IGName(), "default-zone")
+	gotBackend, err := f.GetBackendService(beName)
+	if err != nil {
+		t.Fatalf("Failed to find a backend with name %v: %v", beName, err)
+	}
+	gotGroup, err := fakeIGs.GetInstanceGroup(namer.IGName(), defaultZone)
+	if err != nil {
+		t.Fatalf("Failed to find instance group %v", namer.IGName())
+	}
 	if gotBackend.Backends[0].Group != gotGroup.SelfLink {
 		t.Fatalf(
 			"Broken instance group link: %v %v",

ingress/controllers/gce/backends/fakes.go

@@ -99,47 +99,3 @@ func (f *FakeBackendServices) GetHealth(name, instanceGroupLink string) (*comput
 	return &compute.BackendServiceGroupHealth{
 		HealthStatus: states}, nil
 }
-
-// NewFakeHealthChecks returns a health check fake.
-func NewFakeHealthChecks() *FakeHealthChecks {
-	return &FakeHealthChecks{hc: []*compute.HttpHealthCheck{}}
-}
-
-// FakeHealthChecks fakes out health checks.
-type FakeHealthChecks struct {
-	hc []*compute.HttpHealthCheck
-}
-
-// CreateHttpHealthCheck fakes health check creation.
-func (f *FakeHealthChecks) CreateHttpHealthCheck(hc *compute.HttpHealthCheck) error {
-	f.hc = append(f.hc, hc)
-	return nil
-}
-
-// GetHttpHealthCheck fakes getting a http health check.
-func (f *FakeHealthChecks) GetHttpHealthCheck(name string) (*compute.HttpHealthCheck, error) {
-	for _, h := range f.hc {
-		if h.Name == name {
-			return h, nil
-		}
-	}
-	return nil, fmt.Errorf("Health check %v not found.", name)
-}
-
-// DeleteHttpHealthCheck fakes deleting a http health check.
-func (f *FakeHealthChecks) DeleteHttpHealthCheck(name string) error {
-	healthChecks := []*compute.HttpHealthCheck{}
-	exists := false
-	for _, h := range f.hc {
-		if h.Name == name {
-			exists = true
-			continue
-		}
-		healthChecks = append(healthChecks, h)
-	}
-	if !exists {
-		return fmt.Errorf("Failed to find health check %v", name)
-	}
-	f.hc = healthChecks
-	return nil
-}

ingress/controllers/gce/backends/interfaces.go

@@ -42,17 +42,3 @@ type BackendServices interface {
 	ListBackendServices() (*compute.BackendServiceList, error)
 	GetHealth(name, instanceGroupLink string) (*compute.BackendServiceGroupHealth, error)
 }
-
-// SingleHealthCheck is an interface to manage a single GCE health check.
-type SingleHealthCheck interface {
-	CreateHttpHealthCheck(hc *compute.HttpHealthCheck) error
-	DeleteHttpHealthCheck(name string) error
-	GetHttpHealthCheck(name string) (*compute.HttpHealthCheck, error)
-}
-
-// HealthChecker is an interface to manage cloud HTTPHealthChecks.
-type HealthChecker interface {
-	Add(port int64, path string) error
-	Delete(port int64) error
-	Get(port int64) (*compute.HttpHealthCheck, error)
-}

ingress/controllers/gce/controller/cluster_manager.go

@@ -74,6 +74,23 @@ type ClusterManager struct {
 	backendPool            backends.BackendPool
 	l7Pool                 loadbalancers.LoadBalancerPool
 	firewallPool           firewalls.SingleFirewallPool
+
+	// TODO: Refactor so we simply init a health check pool.
+	// Currently health checks are tied to backends because each backend needs
+	// the link of the associated health, but both the backend pool and
+	// loadbalancer pool manage backends, because the lifetime of the default
+	// backend is tied to the last/first loadbalancer not the life of the
+	// nodeport service or Ingress.
+	healthCheckers []healthchecks.HealthChecker
+}
+
+// Init initializes the cluster manager.
+func (c *ClusterManager) Init(tr *GCETranslator) {
+	c.instancePool.Init(tr)
+	for _, h := range c.healthCheckers {
+		h.Init(tr)
+	}
+	// TODO: Initialize other members as needed.
 }
 
 // IsHealthy returns an error if the cluster manager is unhealthy.
@@ -217,7 +234,7 @@ func getGCEClient(config io.Reader) *gce.GCECloud {
 //   string passed to glbc via --gce-cluster-name.
 // - defaultBackendNodePort: is the node port of glbc's default backend. This is
 //	 the kubernetes Service that serves the 404 page if no urls match.
-// - defaultHealthCheckPath: is the default path used for L7 health checks, eg: "/healthz"
+// - defaultHealthCheckPath: is the default path used for L7 health checks, eg: "/healthz".
 func NewClusterManager(
 	configFilePath string,
 	name string,
@@ -244,21 +261,20 @@ func NewClusterManager(
 
 	// Names are fundamental to the cluster, the uid allocator makes sure names don't collide.
 	cluster := ClusterManager{ClusterNamer: &utils.Namer{name}}
-	zone, err := cloud.GetZone()
-	if err != nil {
-		return nil, err
-	}
 
 	// NodePool stores GCE vms that are in this Kubernetes cluster.
-	cluster.instancePool = instances.NewNodePool(cloud, zone.FailureDomain)
+	cluster.instancePool = instances.NewNodePool(cloud)
 
 	// BackendPool creates GCE BackendServices and associated health checks.
 	healthChecker := healthchecks.NewHealthChecker(cloud, defaultHealthCheckPath, cluster.ClusterNamer)
+	// Loadbalancer pool manages the default backend and its health check.
+	defaultBackendHealthChecker := healthchecks.NewHealthChecker(cloud, "/healthz", cluster.ClusterNamer)
+
+	cluster.healthCheckers = []healthchecks.HealthChecker{healthChecker, defaultBackendHealthChecker}
 
 	// TODO: This needs to change to a consolidated management of the default backend.
 	cluster.backendPool = backends.NewBackendPool(
 		cloud, healthChecker, cluster.instancePool, cluster.ClusterNamer, []int64{defaultBackendNodePort}, true)
-	defaultBackendHealthChecker := healthchecks.NewHealthChecker(cloud, "/healthz", cluster.ClusterNamer)
 	defaultBackendPool := backends.NewBackendPool(
 		cloud, defaultBackendHealthChecker, cluster.instancePool, cluster.ClusterNamer, []int64{}, false)
 	cluster.defaultBackendNodePort = defaultBackendNodePort

ingress/controllers/gce/controller/controller.go

@@ -44,18 +44,25 @@ var (
 	// DefaultClusterUID is the uid to use for clusters resources created by an
 	// L7 controller created without specifying the --cluster-uid flag.
 	DefaultClusterUID = ""
+
+	// Frequency to poll on local stores to sync.
+	storeSyncPollPeriod = 5 * time.Second
 )
 
 // LoadBalancerController watches the kubernetes api and adds/removes services
 // from the loadbalancer, via loadBalancerConfig.
 type LoadBalancerController struct {
-	client              *client.Client
-	ingController       *framework.Controller
-	nodeController      *framework.Controller
-	svcController       *framework.Controller
-	ingLister           StoreToIngressLister
-	nodeLister          cache.StoreToNodeLister
-	svcLister           cache.StoreToServiceLister
+	client         *client.Client
+	ingController  *framework.Controller
+	nodeController *framework.Controller
+	svcController  *framework.Controller
+	podController  *framework.Controller
+	ingLister      StoreToIngressLister
+	nodeLister     cache.StoreToNodeLister
+	svcLister      cache.StoreToServiceLister
+	// Health checks are the readiness probes of containers on pods.
+	podLister cache.StoreToPodLister
+	// TODO: Watch secrets
 	CloudClusterManager *ClusterManager
 	recorder            record.EventRecorder
 	nodeQueue           *taskQueue
@@ -69,6 +76,9 @@ type LoadBalancerController struct {
 	shutdown bool
 	// tlsLoader loads secrets from the Kubernetes apiserver for Ingresses.
 	tlsLoader tlsLoader
+	// hasSynced returns true if all associated sub-controllers have synced.
+	// Abstracted into a func for testing.
+	hasSynced func() bool
 }
 
 // NewLoadBalancerController creates a controller for gce loadbalancers.
@@ -90,6 +100,7 @@ func NewLoadBalancerController(kubeClient *client.Client, clusterManager *Cluste
 	}
 	lbc.nodeQueue = NewTaskQueue(lbc.syncNodes)
 	lbc.ingQueue = NewTaskQueue(lbc.sync)
+	lbc.hasSynced = lbc.storesSynced
 
 	// Ingress watch handlers
 	pathHandlers := framework.ResourceEventHandlerFuncs{
@@ -130,12 +141,19 @@ func NewLoadBalancerController(kubeClient *client.Client, clusterManager *Cluste
 			lbc.client, "services", namespace, fields.Everything()),
 		&api.Service{}, resyncPeriod, svcHandlers)
 
+	lbc.podLister.Indexer, lbc.podController = framework.NewIndexerInformer(
+		cache.NewListWatchFromClient(lbc.client, "pods", namespace, fields.Everything()),
+		&api.Pod{},
+		resyncPeriod,
+		framework.ResourceEventHandlerFuncs{},
+		cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
+	)
+
 	nodeHandlers := framework.ResourceEventHandlerFuncs{
 		AddFunc:    lbc.nodeQueue.enqueue,
 		DeleteFunc: lbc.nodeQueue.enqueue,
 		// Nodes are updated every 10s and we don't care, so no update handler.
 	}
-
 	// Node watch handlers
 	lbc.nodeLister.Store, lbc.nodeController = framework.NewInformer(
 		&cache.ListWatch{
@@ -194,6 +212,7 @@ func (lbc *LoadBalancerController) Run() {
 	go lbc.ingController.Run(lbc.stopCh)
 	go lbc.nodeController.Run(lbc.stopCh)
 	go lbc.svcController.Run(lbc.stopCh)
+	go lbc.podController.Run(lbc.stopCh)
 	go lbc.ingQueue.run(time.Second, lbc.stopCh)
 	go lbc.nodeQueue.run(time.Second, lbc.stopCh)
 	<-lbc.stopCh
@@ -224,8 +243,29 @@ func (lbc *LoadBalancerController) Stop(deleteAll bool) error {
 	return nil
 }
 
+// storesSynced returns true if all the sub-controllers have finished their
+// first sync with apiserver.
+func (lbc *LoadBalancerController) storesSynced() bool {
+	return (
+	// wait for pods to sync so we don't allocate a default health check when
+	// an endpoint has a readiness probe.
+	lbc.podController.HasSynced() &&
+		// wait for services so we don't thrash on backend creation.
+		lbc.svcController.HasSynced() &&
+		// wait for nodes so we don't disconnect a backend from an instance
+		// group just because we don't realize there are nodes in that zone.
+		lbc.nodeController.HasSynced() &&
+		// Wait for ingresses as a safety measure. We don't really need this.
+		lbc.ingController.HasSynced())
+}
+
 // sync manages Ingress create/updates/deletes.
 func (lbc *LoadBalancerController) sync(key string) {
+	if !lbc.hasSynced() {
+		time.Sleep(storeSyncPollPeriod)
+		lbc.ingQueue.requeue(key, fmt.Errorf("Waiting for stores to sync"))
+		return
+	}
 	glog.V(3).Infof("Syncing %v", key)
 
 	paths, err := lbc.ingLister.List()

ingress/controllers/gce/controller/controller_test.go

@@ -55,6 +55,7 @@ func newLoadBalancerController(t *testing.T, cm *fakeClusterManager, masterUrl s
 	if err != nil {
 		t.Fatalf("%v", err)
 	}
+	lb.hasSynced = func() bool { return true }
 	return lb
 }
 

ingress/controllers/gce/controller/fakes.go

@@ -30,7 +30,6 @@ import (
 
 const (
 	testDefaultBeNodePort = int64(3000)
-	defaultZone           = "default-zone"
 )
 
 var testBackendPort = intstr.IntOrString{Type: intstr.Int, IntVal: 80}
@@ -50,8 +49,13 @@ func NewFakeClusterManager(clusterName string) *fakeClusterManager {
 	fakeIGs := instances.NewFakeInstanceGroups(sets.NewString())
 	fakeHCs := healthchecks.NewFakeHealthChecks()
 	namer := &utils.Namer{clusterName}
-	nodePool := instances.NewNodePool(fakeIGs, defaultZone)
+
+	nodePool := instances.NewNodePool(fakeIGs)
+	nodePool.Init(&instances.FakeZoneLister{[]string{"zone-a"}})
+
 	healthChecker := healthchecks.NewHealthChecker(fakeHCs, "/", namer)
+	healthChecker.Init(&healthchecks.FakeHealthCheckGetter{nil})
+
 	backendPool := backends.NewBackendPool(
 		fakeBackends,
 		healthChecker, nodePool, namer, []int64{}, false)