Repository: submariner-io/lighthouse
Pull request #587 (2021-07-13 18:22:07)

Optimize GlobalIngressIP retrieval
Use a resource watcher to maintain a cache instead of reading on demand.


submariner-bot: 🤖 Created branch: z_pr587/tpantelis/optimize_ingress_ip 
🚀 Full E2E won't run until the "ready-to-test" label is applied. I will add it automatically once the PR has 2 approvals, or you can add it manually.

Patch:
diff --git a/pkg/agent/controller/agent.go b/pkg/agent/controller/agent.go
index f51d32a..0abf4be 100644
--- a/pkg/agent/controller/agent.go
+++ b/pkg/agent/controller/agent.go
@@ -36,7 +36,6 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/apimachinery/pkg/runtime/schema"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/util/retry"
@@ -154,11 +153,6 @@ func New(spec *AgentSpecification, syncerConf broker.SyncerConfig, kubeClientSet
 		return nil, err
 	}
 
-	if agentController.globalnetEnabled {
-		gvr, _ := schema.ParseResourceArg("globalingressips.v1.submariner.io")
-		agentController.ingressIPClient = syncerConf.LocalClient.Resource(*gvr)
-	}
-
 	return agentController, nil
 }
 
@@ -531,15 +525,10 @@ func (a *Controller) getGlobalIP(service *corev1.Service) (ip, reason, msg strin
 }
 
 func (a *Controller) getIngressIP(name, namespace string) (*IngressIP, bool, error) {
-	obj, err := a.ingressIPClient.Namespace(namespace).Get(context.TODO(), name, metav1.GetOptions{})
-
-	if apierrors.IsNotFound(err) {
+	obj, found := a.serviceImportController.globalIngressIPCache.getForService(namespace, name)
+	if !found {
 		return nil, false, nil
 	}
 
-	if err != nil {
-		return nil, false, err
-	}
-
 	return parseIngressIP(obj), true, nil
 }
diff --git a/pkg/agent/controller/endpoint.go b/pkg/agent/controller/endpoint.go
index 0cbb03b..6f3bb24 100644
--- a/pkg/agent/controller/endpoint.go
+++ b/pkg/agent/controller/endpoint.go
@@ -42,7 +42,7 @@ import (
 
 func startEndpointController(localClient dynamic.Interface, restMapper meta.RESTMapper, scheme *runtime.Scheme,
 	serviceImport *mcsv1a1.ServiceImport, serviceImportNameSpace, serviceName, clusterID string,
-	globalnetEnabled bool) (*EndpointController, error) {
+	globalIngressIPCache *globalIngressIPCache) (*EndpointController, error) {
 	klog.V(log.DEBUG).Infof("Starting Endpoints controller for service %s/%s", serviceImportNameSpace, serviceName)
 
 	globalIngressIPGVR, _ := schema.ParseResourceArg("globalingressips.v1.submariner.io")
@@ -55,7 +55,7 @@ func startEndpointController(localClient dynamic.Interface, restMapper meta.REST
 		serviceName:                  serviceName,
 		stopCh:                       make(chan struct{}),
 		isHeadless:                   serviceImport.Spec.Type == mcsv1a1.Headless,
-		globalnetEnabled:             globalnetEnabled,
+		globalIngressIPCache:         globalIngressIPCache,
 		localClient:                  localClient,
 		ingressIPClient:              localClient.Resource(*globalIngressIPGVR),
 	}
@@ -189,7 +189,8 @@ func (e *EndpointController) getEndpointsFromAddresses(addresses []corev1.Endpoi
 	endpoints := []discovery.Endpoint{}
 	isIPv6AddressType := addressType == discovery.AddressTypeIPv6
 
-	for _, address := range addresses {
+	for i := range addresses {
+		address := &addresses[i]
 		if utilnet.IsIPv6String(address.IP) == isIPv6AddressType {
 			endpoint, retry := e.endpointFromAddress(address, ready)
 			if retry {
@@ -203,7 +204,7 @@ func (e *EndpointController) getEndpointsFromAddresses(addresses []corev1.Endpoi
 	return endpoints, false
 }
 
-func (e *EndpointController) endpointFromAddress(address corev1.EndpointAddress, ready bool) (*discovery.Endpoint, bool) {
+func (e *EndpointController) endpointFromAddress(address *corev1.EndpointAddress, ready bool) (*discovery.Endpoint, bool) {
 	topology := map[string]string{}
 	if address.NodeName != nil {
 		topology["kubernetes.io/hostname"] = *address.NodeName
@@ -252,20 +253,17 @@ func allAddressesIPv6(addresses []corev1.EndpointAddress) bool {
 	return true
 }
 
-func (e *EndpointController) getIP(address corev1.EndpointAddress) string {
-	if e.isHeadless && e.globalnetEnabled {
-		var ip string
-
-		name := "pod-" + address.TargetRef.Name
+func (e *EndpointController) getIP(address *corev1.EndpointAddress) string {
+	if e.isHeadless && e.globalIngressIPCache != nil {
+		obj, found := e.globalIngressIPCache.getForPod(e.serviceImportSourceNameSpace, address.TargetRef.Name)
 
-		// TODO: Replace with a syncer.GetResource() to reduce API calls
-		obj, err := e.ingressIPClient.Namespace(e.serviceImportSourceNameSpace).Get(context.TODO(), name, metav1.GetOptions{})
-		if err == nil {
+		var ip string
+		if found {
 			ip, _, _ = unstructured.NestedString(obj.Object, "status", "allocatedIP")
 		}
 
 		if ip == "" {
-			klog.Infof("GlobalIP for %s not allocated yet", name)
+			klog.Infof("GlobalIP for EndpointAddress %q is not allocated yet", address.TargetRef.Name)
 		}
 
 		return ip
diff --git a/pkg/agent/controller/global_ingressip_cache.go b/pkg/agent/controller/global_ingressip_cache.go
new file mode 100644
index 0000000..e238ee4
--- /dev/null
+++ b/pkg/agent/controller/global_ingressip_cache.go
@@ -0,0 +1,109 @@
+/*
+SPDX-License-Identifier: Apache-2.0
+
+Copyright Contributors to the Submariner project.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+package controller
+
+import (
+	"sync"
+
+	"github.com/submariner-io/admiral/pkg/watcher"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+)
+
+func newGlobalIngressIPCache(config watcher.Config) (*globalIngressIPCache, error) {
+	c := &globalIngressIPCache{}
+
+	config.ResourceConfigs = []watcher.ResourceConfig{
+		{
+			Name:         "GlobalIngressIP watcher",
+			ResourceType: GetGlobalIngressIPObj(),
+			Handler: watcher.EventHandlerFuncs{
+				OnCreateFunc: func(obj runtime.Object, numRequeues int) bool {
+					c.onCreateOrUpdate(obj.(*unstructured.Unstructured))
+					return false
+				},
+				OnUpdateFunc: func(obj runtime.Object, numRequeues int) bool {
+					c.onCreateOrUpdate(obj.(*unstructured.Unstructured))
+					return false
+				},
+				OnDeleteFunc: func(obj runtime.Object, numRequeues int) bool {
+					c.onDelete(obj.(*unstructured.Unstructured))
+					return false
+				},
+			},
+			SourceNamespace: metav1.NamespaceAll,
+		},
+	}
+
+	var err error
+
+	c.watcher, err = watcher.New(&config)
+
+	return c, err
+}
+
+func (c *globalIngressIPCache) start(stopCh <-chan struct{}) error {
+	return c.watcher.Start(stopCh)
+}
+
+func (c *globalIngressIPCache) onCreateOrUpdate(obj *unstructured.Unstructured) {
+	c.applyToCache(obj, func(to *sync.Map, key string, obj *unstructured.Unstructured) {
+		to.Store(key, obj)
+	})
+}
+
+func (c *globalIngressIPCache) onDelete(obj *unstructured.Unstructured) {
+	c.applyToCache(obj, func(to *sync.Map, key string, obj *unstructured.Unstructured) {
+		to.Delete(key)
+	})
+}
+
+func (c *globalIngressIPCache) applyToCache(obj *unstructured.Unstructured,
+	apply func(to *sync.Map, key string, obj *unstructured.Unstructured)) {
+	target, _, _ := unstructured.NestedString(obj.Object, "spec", "target")
+	switch target {
+	case ClusterIPService:
+		name, _, _ := unstructured.NestedString(obj.Object, "spec", "serviceRef", "name")
+		apply(&c.byService, c.key(obj.GetNamespace(), name), obj)
+	case HeadlessServicePod:
+		name, _, _ := unstructured.NestedString(obj.Object, "spec", "podRef", "name")
+		apply(&c.byPod, c.key(obj.GetNamespace(), name), obj)
+	}
+}
+
+func (c *globalIngressIPCache) getForService(namespace, name string) (*unstructured.Unstructured, bool) {
+	return c.get(&c.byService, namespace, name)
+}
+
+func (c *globalIngressIPCache) getForPod(namespace, name string) (*unstructured.Unstructured, bool) {
+	return c.get(&c.byPod, namespace, name)
+}
+
+func (c *globalIngressIPCache) get(from *sync.Map, namespace, name string) (*unstructured.Unstructured, bool) {
+	v, found := from.Load(c.key(namespace, name))
+	if !found {
+		return nil, false
+	}
+
+	return v.(*unstructured.Unstructured), true
+}
+
+func (c *globalIngressIPCache) key(ns, n string) string {
+	return ns + "/" + n
+}
diff --git a/pkg/agent/controller/globalingressip.go b/pkg/agent/controller/globalingressip.go
index cddb714..374c562 100644
--- a/pkg/agent/controller/globalingressip.go
+++ b/pkg/agent/controller/globalingressip.go
@@ -24,7 +24,7 @@ import (
 
 const (
 	ClusterIPService           = "ClusterIPService"
-	headlessServicePod         = "HeadlessServicePod"
+	HeadlessServicePod         = "HeadlessServicePod"
 	defaultReasonIPUnavailable = "ServiceGlobalIPUnavailable"
 	defaultMsgIPUnavailable    = "Service doesn't have a global IP yet"
 )
@@ -32,8 +32,6 @@ const (
 type IngressIP struct {
 	namespace         string
 	target            string
-	svcName           string
-	podName           string
 	allocatedIP       string
 	unallocatedReason string
 	unallocatedMsg    string
@@ -53,20 +51,6 @@ func parseIngressIP(obj *unstructured.Unstructured) *IngressIP {
 		return nil
 	}
 
-	gip.svcName, found, err = unstructured.NestedString(obj.Object, "spec", "serviceRef", "name")
-	if !found || err != nil {
-		klog.Errorf("ServiceRef not found in spec %#v", obj.Object)
-		return nil
-	}
-
-	if gip.target == headlessServicePod {
-		gip.podName, found, err = unstructured.NestedString(obj.Object, "spec", "podRef", "name")
-		if !found || err != nil {
-			klog.Errorf("Expected PodRef in spec %#v", obj.Object)
-			return nil
-		}
-	}
-
 	gip.allocatedIP, _, _ = unstructured.NestedString(obj.Object, "status", "allocatedIP")
 	if gip.allocatedIP == "" {
 		conditions, _, _ := unstructured.NestedSlice(obj.Object, "status", "conditions")
diff --git a/pkg/agent/controller/serviceimport.go b/pkg/agent/controller/serviceimport.go
index ba3ad28..3d22ae6 100644
--- a/pkg/agent/controller/serviceimport.go
+++ b/pkg/agent/controller/serviceimport.go
@@ -21,6 +21,7 @@ import (
 	"github.com/submariner-io/admiral/pkg/federate"
 	"github.com/submariner-io/admiral/pkg/log"
 	"github.com/submariner-io/admiral/pkg/syncer"
+	"github.com/submariner-io/admiral/pkg/watcher"
 	lhconstants "github.com/submariner-io/lighthouse/pkg/constants"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
@@ -34,12 +35,11 @@ import (
 func newServiceImportController(spec *AgentSpecification, serviceSyncer syncer.Interface, restMapper meta.RESTMapper,
 	localClient dynamic.Interface, scheme *runtime.Scheme) (*ServiceImportController, error) {
 	controller := &ServiceImportController{
-		serviceSyncer:    serviceSyncer,
-		localClient:      localClient,
-		restMapper:       restMapper,
-		clusterID:        spec.ClusterID,
-		scheme:           scheme,
-		globalnetEnabled: spec.GlobalnetEnabled,
+		serviceSyncer: serviceSyncer,
+		localClient:   localClient,
+		restMapper:    restMapper,
+		clusterID:     spec.ClusterID,
+		scheme:        scheme,
 	}
 
 	var err error
@@ -59,10 +59,24 @@ func newServiceImportController(spec *AgentSpecification, serviceSyncer syncer.I
 		return nil, err
 	}
 
-	return controller, nil
+	if spec.GlobalnetEnabled {
+		controller.globalIngressIPCache, err = newGlobalIngressIPCache(watcher.Config{
+			RestMapper: restMapper,
+			Client:     localClient,
+			Scheme:     scheme,
+		})
+	}
+
+	return controller, err
 }
 
 func (c *ServiceImportController) start(stopCh <-chan struct{}) error {
+	if c.globalIngressIPCache != nil {
+		if err := c.globalIngressIPCache.start(stopCh); err != nil {
+			return err
+		}
+	}
+
 	go func() {
 		<-stopCh
 
@@ -113,7 +127,7 @@ func (c *ServiceImportController) serviceImportCreatedOrUpdated(serviceImport *m
 	}
 
 	endpointController, err := startEndpointController(c.localClient, c.restMapper, c.scheme,
-		serviceImport, serviceNameSpace, serviceName, c.clusterID, c.globalnetEnabled)
+		serviceImport, serviceNameSpace, serviceName, c.clusterID, c.globalIngressIPCache)
 	if err != nil {
 		klog.Errorf(err.Error())
 		return true
diff --git a/pkg/agent/controller/types.go b/pkg/agent/controller/types.go
index 7030600..d47e09f 100644
--- a/pkg/agent/controller/types.go
+++ b/pkg/agent/controller/types.go
@@ -20,6 +20,7 @@ package controller
 import (
 	"sync"
 
+	"github.com/submariner-io/admiral/pkg/watcher"
 	"k8s.io/client-go/kubernetes"
 
 	"github.com/submariner-io/admiral/pkg/syncer"
@@ -41,7 +42,6 @@ type Controller struct {
 	endpointSliceSyncer     *broker.Syncer
 	serviceSyncer           syncer.Interface
 	serviceImportController *ServiceImportController
-	ingressIPClient         dynamic.NamespaceableResourceInterface
 }
 
 type AgentSpecification struct {
@@ -54,14 +54,14 @@ type AgentSpecification struct {
 // and creates an EndpointController in response. The EndpointController will use the app label as filter
 // to listen only for the endpoints event related to ServiceImport created
 type ServiceImportController struct {
-	serviceSyncer       syncer.Interface
-	localClient         dynamic.Interface
-	restMapper          meta.RESTMapper
-	serviceImportSyncer syncer.Interface
-	endpointControllers sync.Map
-	clusterID           string
-	scheme              *runtime.Scheme
-	globalnetEnabled    bool
+	serviceSyncer        syncer.Interface
+	localClient          dynamic.Interface
+	restMapper           meta.RESTMapper
+	serviceImportSyncer  syncer.Interface
+	endpointControllers  sync.Map
+	clusterID            string
+	scheme               *runtime.Scheme
+	globalIngressIPCache *globalIngressIPCache
 }
 
 // Each EndpointController listens for the endpoints that backs a service and have a ServiceImport
@@ -77,5 +77,12 @@ type EndpointController struct {
 	localClient                  dynamic.Interface
 	ingressIPClient              dynamic.NamespaceableResourceInterface
 	isHeadless                   bool
-	globalnetEnabled             bool
+	globalIngressIPCache         *globalIngressIPCache
+}
+
+type globalIngressIPCache struct {
+	sync.Mutex
+	byService sync.Map
+	byPod     sync.Map
+	watcher   watcher.Interface
 }


Test patch:
diff --git a/pkg/agent/controller/controller_suite_test.go b/pkg/agent/controller/controller_suite_test.go
index 14b1623..f3caf58 100644
--- a/pkg/agent/controller/controller_suite_test.go
+++ b/pkg/agent/controller/controller_suite_test.go
@@ -222,6 +222,14 @@ func (t *testDriver) newGlobalIngressIP(name, ip string) *unstructured.Unstructu
 	return ingressIP
 }
 
+func (t *testDriver) newHeadlessGlobalIngressIP(name, ip string) *unstructured.Unstructured {
+	ingressIP := t.newGlobalIngressIP("pod"+"-"+name, ip)
+	Expect(unstructured.SetNestedField(ingressIP.Object, controller.HeadlessServicePod, "spec", "target")).To(Succeed())
+	Expect(unstructured.SetNestedField(ingressIP.Object, name, "spec", "podRef", "name")).To(Succeed())
+
+	return ingressIP
+}
+
 func (t *testDriver) justBeforeEach() {
 	t.cluster1.start(t, *t.syncerConfig)
 	t.cluster2.start(t, *t.syncerConfig)
@@ -518,9 +526,9 @@ func (t *testDriver) createGlobalIngressIP(ingressIP *unstructured.Unstructured)
 
 func (t *testDriver) createEndpointIngressIPs() {
 	t.endpointGlobalIPs = []string{globalIP1, globalIP2, globalIP3}
-	t.createGlobalIngressIP(t.newGlobalIngressIP("pod-one", globalIP1))
-	t.createGlobalIngressIP(t.newGlobalIngressIP("pod-two", globalIP2))
-	t.createGlobalIngressIP(t.newGlobalIngressIP("pod-not-ready", globalIP3))
+	t.createGlobalIngressIP(t.newHeadlessGlobalIngressIP("one", globalIP1))
+	t.createGlobalIngressIP(t.newHeadlessGlobalIngressIP("two", globalIP2))
+	t.createGlobalIngressIP(t.newHeadlessGlobalIngressIP("not-ready", globalIP3))
 }
 
 func (t *testDriver) dynamicServiceClient() dynamic.ResourceInterface {