Move joiner methods into context

pkg/context/context.go

@@ -20,13 +20,16 @@ import (
 	"github.com/golang/glog"
 
 	apiv1 "k8s.io/api/core/v1"
+	extensions "k8s.io/api/extensions/v1beta1"
 	informerv1 "k8s.io/client-go/informers/core/v1"
 	informerv1beta1 "k8s.io/client-go/informers/extensions/v1beta1"
 	"k8s.io/client-go/kubernetes"
 	scheme "k8s.io/client-go/kubernetes/scheme"
 	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/record"
+	backendconfigv1beta1 "k8s.io/ingress-gce/pkg/apis/backendconfig/v1beta1"
+	"k8s.io/ingress-gce/pkg/backendconfig"
 	backendconfigclient "k8s.io/ingress-gce/pkg/backendconfig/client/clientset/versioned"
 	informerbackendconfig "k8s.io/ingress-gce/pkg/backendconfig/client/informers/externalversions/backendconfig/v1beta1"
 	"k8s.io/ingress-gce/pkg/utils"
@@ -180,3 +183,34 @@ func (ctx *ControllerContext) Start(stopCh chan struct{}) {
 		go ctx.BackendConfigInformer.Run(stopCh)
 	}
 }
+
+// IngressesForService gets all the Ingresses that reference a Service.
+func (ctx *ControllerContext) IngressesForService(svc *apiv1.Service) (ingList []*extensions.Ingress) {
+	ingLister := utils.StoreToIngressLister{Store: ctx.IngressInformer.GetStore()}
+	ings, err := ingLister.GetServiceIngress(svc, ctx.DefaultBackendSvcPortID)
+	if err != nil {
+		glog.V(4).Infof("ignoring service %v: %v", svc.Name, err)
+		return
+	}
+	for _, ing := range ings {
+		if !utils.IsGCEIngress(&ing) {
+			continue
+		}
+		ingList = append(ingList, &ing)
+	}
+	return
+}
+
+// IngressesForBackendConfig gets all Ingresses that reference (indirectly) a BackendConfig.
+// TODO(rramkumar): This can be optimized to remove nested loops.
+func (ctx *ControllerContext) IngressesForBackendConfig(beConfig *backendconfigv1beta1.BackendConfig) (ingList []*extensions.Ingress) {
+	// Get all the Services associated with this BackendConfig.
+	svcLister := ctx.ServiceInformer.GetStore()
+	linkedSvcs := backendconfig.GetServicesForBackendConfig(svcLister, beConfig)
+	// Return all the Ingresses associated with each Service.
+	for _, svc := range linkedSvcs {
+		ingsForSvc := ctx.IngressesForService(svc)
+		ingList = append(ingList, ingsForSvc...)
+	}
+	return
+}

pkg/controller/controller.go

@@ -49,7 +49,6 @@ type LoadBalancerController struct {
 	client kubernetes.Interface
 	ctx    *context.ControllerContext
 
-	joiner     *utils.Joiner
 	ingLister  utils.StoreToIngressLister
 	nodeLister cache.Indexer
 	nodes      *NodeController
@@ -96,7 +95,6 @@ func NewLoadBalancerController(
 		hasSynced:           ctx.HasSynced,
 	}
 	lbc.ingQueue = utils.NewPeriodicTaskQueue("ingresses", lbc.sync)
-	lbc.joiner = utils.NewJoiner(lbc.ingLister, ctx.ServiceInformer.GetIndexer(), ctx.DefaultBackendSvcPortID)
 
 	// Ingress event handlers.
 	ctx.IngressInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
@@ -140,13 +138,13 @@ func NewLoadBalancerController(
 	ctx.ServiceInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
 			svc := obj.(*apiv1.Service)
-			ings := lbc.joiner.IngressesForService(svc)
+			ings := lbc.ctx.IngressesForService(svc)
 			lbc.ingQueue.Enqueue(convert(ings)...)
 		},
 		UpdateFunc: func(old, cur interface{}) {
 			if !reflect.DeepEqual(old, cur) {
 				svc := cur.(*apiv1.Service)
-				ings := lbc.joiner.IngressesForService(svc)
+				ings := lbc.ctx.IngressesForService(svc)
 				lbc.ingQueue.Enqueue(convert(ings)...)
 			}
 		},
@@ -158,20 +156,20 @@ func NewLoadBalancerController(
 		ctx.BackendConfigInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
 			AddFunc: func(obj interface{}) {
 				beConfig := obj.(*backendconfigv1beta1.BackendConfig)
-				ings := lbc.joiner.IngressesForBackendConfig(beConfig)
+				ings := lbc.ctx.IngressesForBackendConfig(beConfig)
 				lbc.ingQueue.Enqueue(convert(ings)...)
 
 			},
 			UpdateFunc: func(old, cur interface{}) {
 				if !reflect.DeepEqual(old, cur) {
 					beConfig := cur.(*backendconfigv1beta1.BackendConfig)
-					ings := lbc.joiner.IngressesForBackendConfig(beConfig)
+					ings := lbc.ctx.IngressesForBackendConfig(beConfig)
 					lbc.ingQueue.Enqueue(convert(ings)...)
 				}
 			},
 			DeleteFunc: func(obj interface{}) {
 				beConfig := obj.(*backendconfigv1beta1.BackendConfig)
-				ings := lbc.joiner.IngressesForBackendConfig(beConfig)
+				ings := lbc.ctx.IngressesForBackendConfig(beConfig)
 				lbc.ingQueue.Enqueue(convert(ings)...)
 			},
 		})

pkg/firewalls/controller.go

@@ -45,7 +45,6 @@ type FirewallController struct {
 	ctx          *context.ControllerContext
 	firewallPool SingleFirewallPool
 	queue        utils.TaskQueue
-	joiner       *utils.Joiner
 	ingLister    utils.StoreToIngressLister
 	translator   *translator.Translator
 	nodeLister   cache.Indexer
@@ -70,7 +69,6 @@ func NewFirewallController(
 	}
 
 	fwc.queue = utils.NewPeriodicTaskQueue("firewall", fwc.sync)
-	fwc.joiner = utils.NewJoiner(fwc.ingLister, ctx.ServiceInformer.GetIndexer(), ctx.DefaultBackendSvcPortID)
 
 	// Ingress event handlers.
 	ctx.IngressInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
@@ -101,15 +99,15 @@ func NewFirewallController(
 	ctx.ServiceInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
 			svc := obj.(*apiv1.Service)
-			ings := fwc.joiner.IngressesForService(svc)
+			ings := fwc.ctx.IngressesForService(svc)
 			if len(ings) > 0 {
 				fwc.queue.Enqueue(queueKey)
 			}
 		},
 		UpdateFunc: func(old, cur interface{}) {
 			if !reflect.DeepEqual(old, cur) {
 				svc := cur.(*apiv1.Service)
-				ings := fwc.joiner.IngressesForService(svc)
+				ings := fwc.ctx.IngressesForService(svc)
 				if len(ings) > 0 {
 					fwc.queue.Enqueue(queueKey)
 				}