Merge pull request #234 from rramkumar1/mci-enqueue

Add ability for MCI controller to enqueue ingresses
kubernetes · Apr 19, 2018 · e5c5c8b · e5c5c8b
2 parents 34c5945 + 8828ea2
commit e5c5c8b
Show file tree

Hide file tree

Showing 4 changed files with 35 additions and 3 deletions.
diff --git a/cmd/glbc/main.go b/cmd/glbc/main.go
@@ -126,7 +126,7 @@ func main() {
 	}
 
 	if flags.F.MultiCluster {
-		mciController, _ := mci.NewController(ctx, flags.F.ResyncPeriod)
+		mciController, _ := mci.NewController(ctx, flags.F.ResyncPeriod, lbc)
 		go mciController.Run(stopCh)
 		glog.V(0).Infof("Multi-Cluster Ingress Controller started")
 	}

diff --git a/pkg/context/context.go b/pkg/context/context.go
@@ -54,6 +54,7 @@ type ControllerContext struct {
 type MultiClusterContext struct {
 	RegistryClient  crclient.Interface
 	ClusterInformer cache.SharedIndexInformer
+	MCIEnabled      bool
 }
 
 // NewControllerContext returns a new shared set of informers.
@@ -76,6 +77,7 @@ func NewControllerContext(kubeClient kubernetes.Interface, registryClient crclie
 	}
 	if context.MC.RegistryClient != nil {
 		context.MC.ClusterInformer = crinformerv1alpha1.NewClusterInformer(registryClient, resyncPeriod, newIndexer())
+		context.MC.MCIEnabled = true
 	}
 
 	return context

diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go
@@ -184,6 +184,18 @@ func NewLoadBalancerController(ctx *context.ControllerContext, clusterManager *C
 	return &lbc, nil
 }
 
+// Implements MCIEnqueue
+func (lbc *LoadBalancerController) EnqueueAllIngresses() error {
+	ings, err := lbc.ingLister.ListGCEIngresses()
+	if err != nil {
+		return err
+	}
+	for _, ing := range ings.Items {
+		lbc.ingQueue.Enqueue(&ing)
+	}
+	return nil
+}
+
 // enqueueIngressForService enqueues all the Ingress' for a Service.
 func (lbc *LoadBalancerController) enqueueIngressForService(obj interface{}) {
 	svc := obj.(*apiv1.Service)
@@ -242,6 +254,12 @@ func (lbc *LoadBalancerController) sync(key string) (retErr error) {
 	}
 	glog.V(3).Infof("Syncing %v", key)
 
+	if lbc.ctx.MC.MCIEnabled {
+		// This is a temporary short-circuit to just verify that
+		// the MCI controller properly queues ingresses.
+		return nil
+	}
+
 	gceIngresses, err := lbc.ingLister.ListGCEIngresses()
 	if err != nil {
 		return err

diff --git a/pkg/mci/controller.go b/pkg/mci/controller.go
@@ -34,11 +34,15 @@ type Controller struct {
 
 	clusterSynced cache.InformerSynced
 	clusterLister cache.Indexer
+}
 
-	// TODO(rramkumar): Add lister for service extension CRD.
+// MCIEnqueue is a interface to allow the MCI controller to enqueue ingresses
+// based on events it receives.
+type MCIEnqueue interface {
+	EnqueueAllIngresses() error
 }
 
-func NewController(ctx *context.ControllerContext, resyncPeriod time.Duration) (*Controller, error) {
+func NewController(ctx *context.ControllerContext, resyncPeriod time.Duration, enqueue MCIEnqueue) (*Controller, error) {
 	mciController := &Controller{
 		resyncPeriod:  resyncPeriod,
 		clusterSynced: ctx.MC.ClusterInformer.HasSynced,
@@ -49,10 +53,18 @@ func NewController(ctx *context.ControllerContext, resyncPeriod time.Duration) (
 		AddFunc: func(obj interface{}) {
 			c := obj.(*crv1alpha1.Cluster)
 			glog.V(3).Infof("Cluster %v added", c.Name)
+			err := enqueue.EnqueueAllIngresses()
+			if err != nil {
+				glog.V(3).Infof("Error enqueuing ingress on cluster add: %v", err)
+			}
 		},
 		DeleteFunc: func(obj interface{}) {
 			c := obj.(*crv1alpha1.Cluster)
 			glog.V(3).Infof("Cluster %v deleted", c.Name)
+			err := enqueue.EnqueueAllIngresses()
+			if err != nil {
+				glog.V(3).Infof("Error enqueuing ingress on cluster delete: %v", err)
+			}
 		},
 		UpdateFunc: func(obj, cur interface{}) {
 			c := obj.(*crv1alpha1.Cluster)