refactor endpoint code to eliminate code duplication (#52726)

* wip

Signed-off-by: Rama Chavali <[email protected]>

* refactor endpoint code to eliminate code duplication

Signed-off-by: Rama Chavali <[email protected]>

* fix bad merge

Signed-off-by: Rama Chavali <[email protected]>

---------

Signed-off-by: Rama Chavali <[email protected]>

Author: ramaraochavali

pilot/pkg/model/push_context.go

@@ -297,6 +297,8 @@ type ConsolidatedDestRule struct {
 	from []types.NamespacedName
 }
 
+type EdsUpdateFn func(shard ShardKey, hostname string, namespace string, entry []*IstioEndpoint)
+
 // XDSUpdater is used for direct updates of the xDS model and incremental push.
 // Pilot uses multiple registries - for example each K8S cluster is a registry
 // instance. Each registry is responsible for tracking a set

pilot/pkg/serviceregistry/serviceentry/controller.go

@@ -324,12 +324,12 @@ func (s *Controller) workloadEntryHandler(old, curr config.Config, event model.E
 		if event == model.EventAdd {
 			s.XdsUpdater.ProxyUpdate(s.Cluster(), wle.Address)
 		}
-		s.edsUpdate(allInstances)
+		s.edsUpdate(allInstances, true)
 		return
 	}
 
 	// update eds cache only
-	s.edsCacheUpdate(allInstances)
+	s.edsUpdate(allInstances, false)
 
 	pushReq := &model.PushRequest{
 		Full:           true,
@@ -451,7 +451,7 @@ func (s *Controller) serviceEntryHandler(old, curr config.Config, event model.Ev
 	fullPush := len(configsUpdated) > 0
 	// if not full push needed, at least one service unchanged
 	if !fullPush {
-		s.edsUpdate(serviceInstances)
+		s.edsUpdate(serviceInstances, true)
 		return
 	}
 
@@ -468,7 +468,7 @@ func (s *Controller) serviceEntryHandler(old, curr config.Config, event model.Ev
 		keys.Insert(instancesKey{hostname: svc.Hostname, namespace: curr.Namespace})
 	}
 
-	s.queueEdsEvent(keys, s.doEdsCacheUpdate)
+	s.queueEdsEvent(keys, false)
 
 	pushReq := &model.PushRequest{
 		Full:           true,
@@ -604,7 +604,7 @@ func (s *Controller) WorkloadInstanceHandler(wi *model.WorkloadInstance, event m
 
 	s.mutex.Unlock()
 
-	s.edsUpdate(append(instances, instancesDeleted...))
+	s.edsUpdate(append(instances, instancesDeleted...), true)
 
 	// ServiceEntry with WorkloadEntry results in STRICT_DNS cluster with hardcoded endpoints
 	// need to update CDS to refresh endpoints
@@ -691,7 +691,7 @@ func (s *Controller) ResyncEDS() {
 	s.mutex.RLock()
 	allInstances := s.serviceInstances.getAll()
 	s.mutex.RUnlock()
-	s.edsUpdate(allInstances)
+	s.edsUpdate(allInstances, true)
 	// HACK to workaround Service syncing after WorkloadEntry: https://github.com/istio/istio/issues/45114
 	s.workloadInstances.ForEach(func(wi *model.WorkloadInstance) {
 		if wi.Kind == model.WorkloadEntryKind {
@@ -703,35 +703,27 @@ func (s *Controller) ResyncEDS() {
 // edsUpdate triggers an EDS push serially such that we can prevent all instances
 // got at t1 can accidentally override that got at t2 if multiple threads are
 // running this function. Queueing ensures latest updated wins.
-func (s *Controller) edsUpdate(instances []*model.ServiceInstance) {
+func (s *Controller) edsUpdate(instances []*model.ServiceInstance, pushEds bool) {
 	// Find all keys we need to lookup
 	keys := sets.NewWithLength[instancesKey](len(instances))
 	for _, i := range instances {
 		keys.Insert(makeInstanceKey(i))
 	}
-	s.queueEdsEvent(keys, s.doEdsUpdate)
-}
-
-// edsCacheUpdate updates eds cache serially such that we can prevent allinstances
-// got at t1 can accidentally override that got at t2 if multiple threads are
-// running this function. Queueing ensures latest updated wins.
-func (s *Controller) edsCacheUpdate(instances []*model.ServiceInstance) {
-	// Find all keys we need to lookup
-	keys := map[instancesKey]struct{}{}
-	for _, i := range instances {
-		keys[makeInstanceKey(i)] = struct{}{}
-	}
-	s.queueEdsEvent(keys, s.doEdsCacheUpdate)
+	s.queueEdsEvent(keys, pushEds)
 }
 
 // queueEdsEvent processes eds events sequentially for the passed keys and invokes the passed function.
-func (s *Controller) queueEdsEvent(keys sets.Set[instancesKey], edsFn func(keys sets.Set[instancesKey])) {
+func (s *Controller) queueEdsEvent(keys sets.Set[instancesKey], pushEds bool) {
 	// wait for the cache update finished
 	waitCh := make(chan struct{})
 	// trigger update eds endpoint shards
 	s.edsQueue.Push(func() error {
 		defer close(waitCh)
-		edsFn(keys)
+		xdsUpdateFn := s.XdsUpdater.EDSCacheUpdate
+		if pushEds {
+			xdsUpdateFn = s.XdsUpdater.EDSUpdate
+		}
+		s.doEdsUpdate(keys, xdsUpdateFn)
 		return nil
 	})
 	select {
@@ -744,34 +736,17 @@ func (s *Controller) queueEdsEvent(keys sets.Set[instancesKey], edsFn func(keys
 	}
 }
 
-// doEdsCacheUpdate invokes XdsUpdater's EDSCacheUpdate to update endpoint shards.
-func (s *Controller) doEdsCacheUpdate(keys sets.Set[instancesKey]) {
+func (s *Controller) doEdsUpdate(keys sets.Set[instancesKey], xdsUpdateFn model.EdsUpdateFn) {
 	endpoints := s.buildEndpoints(keys)
 	shard := model.ShardKeyFromRegistry(s)
 
 	for k := range keys {
 		if eps, ok := endpoints[k]; ok {
 			// Update the cache with the generated endpoints.
-			s.XdsUpdater.EDSCacheUpdate(shard, string(k.hostname), k.namespace, eps)
-		} else {
-			// Handle deletions by sending a nil endpoints update.
-			s.XdsUpdater.EDSCacheUpdate(shard, string(k.hostname), k.namespace, nil)
-		}
-	}
-}
-
-// doEdsUpdate invokes XdsUpdater's eds update to trigger eds push.
-func (s *Controller) doEdsUpdate(keys sets.Set[instancesKey]) {
-	endpoints := s.buildEndpoints(keys)
-	shard := model.ShardKeyFromRegistry(s)
-
-	for k := range keys {
-		if eps, ok := endpoints[k]; ok {
-			// Update with the generated endpoints.
-			s.XdsUpdater.EDSUpdate(shard, string(k.hostname), k.namespace, eps)
+			xdsUpdateFn(shard, string(k.hostname), k.namespace, eps)
 		} else {
 			// Handle deletions by sending a nil endpoints update.
-			s.XdsUpdater.EDSUpdate(shard, string(k.hostname), k.namespace, nil)
+			xdsUpdateFn(shard, string(k.hostname), k.namespace, nil)
 		}
 	}
 }