move filter before to prevent unneeded processing (istio#46718)

* move filter before

* move filter before

* Fix test

* revert log

pilot/pkg/bootstrap/server.go

@@ -847,18 +847,12 @@ func (s *Server) initRegistryEventHandlers() {
  log.Info("initializing registry event handlers")
  // Flush cached discovery responses whenever services configuration change.
  serviceHandler := func(prev, curr *model.Service, event model.Event) {
- needsPush := true
- if event == model.EventUpdate {
- needsPush = serviceUpdateNeedsPush(prev, curr)
- }
- if needsPush {
- pushReq := &model.PushRequest{
- Full: true,
- ConfigsUpdated: sets.New(model.ConfigKey{Kind: kind.ServiceEntry, Name: string(curr.Hostname), Namespace: curr.Attributes.Namespace}),
- Reason: model.NewReasonStats(model.ServiceUpdate),
- }
- s.XDSServer.ConfigUpdate(pushReq)
+ pushReq := &model.PushRequest{
+ Full: true,
+ ConfigsUpdated: sets.New(model.ConfigKey{Kind: kind.ServiceEntry, Name: string(curr.Hostname), Namespace: curr.Attributes.Namespace}),
+ Reason: model.NewReasonStats(model.ServiceUpdate),
  }
+ s.XDSServer.ConfigUpdate(pushReq)
  }
  s.ServiceController().AppendServiceHandler(serviceHandler)
 
@@ -1335,13 +1329,3 @@ func (s *Server) initReadinessProbes() {
  s.addReadinessProbe(name, probe)
  }
 }
-
-func serviceUpdateNeedsPush(prev, curr *model.Service) bool {
- if !features.EnableOptimizedServicePush {
- return true
- }
- if prev == nil {
- return true
- }
- return !prev.Equals(curr)
-}

pilot/pkg/serviceregistry/kube/controller/controller.go

@@ -370,7 +370,7 @@ func (c *Controller) Cleanup() error {
  return nil
 }
 
-func (c *Controller) onServiceEvent(_, curr *v1.Service, event model.Event) error {
+func (c *Controller) onServiceEvent(pre, curr *v1.Service, event model.Event) error {
  log.Debugf("Handle event %s for service %s in namespace %s", event, curr.Name, curr.Namespace)
 
  // Create the standard (cluster.local) service.
@@ -409,6 +409,12 @@ func (c *Controller) deleteService(svc *model.Service) {
 }
 
 func (c *Controller) addOrUpdateService(curr *v1.Service, currConv *model.Service, event model.Event, updateEDSCache bool) {
+ // instance conversion is only required when service is added/updated.
+ c.Lock()
+ prevConv := c.servicesMap[currConv.Hostname]
+ c.servicesMap[currConv.Hostname] = currConv
+ c.Unlock()
+
  needsFullPush := false
  // First, process nodePort gateway service, whose externalIPs specified
  // and loadbalancer gateway service
@@ -429,12 +435,6 @@ func (c *Controller) addOrUpdateService(curr *v1.Service, currConv *model.Servic
  if curr != nil && curr.Spec.Type == v1.ServiceTypeExternalName {
  updateEDSCache = true
  }
- var prevConv *model.Service
- // instance conversion is only required when service is added/updated.
- c.Lock()
- prevConv = c.servicesMap[currConv.Hostname]
- c.servicesMap[currConv.Hostname] = currConv
- c.Unlock()
 
  // This full push needed to update ALL ends endpoints, even though we do a full push on service add/update
  // as that full push is only triggered for the specific service.
@@ -455,6 +455,11 @@ func (c *Controller) addOrUpdateService(curr *v1.Service, currConv *model.Servic
  }
  }
 
+ // filter out same service event
+ if event == model.EventUpdate && !serviceUpdateNeedsPush(prevConv, currConv) {
+ return
+ }
+
  c.opts.XDSUpdater.SvcUpdate(shard, string(currConv.Hostname), ns, event)
  c.handlers.NotifyServiceHandlers(prevConv, currConv, event)
 }
@@ -1131,3 +1136,13 @@ func (c *Controller) servicesForNamespacedName(name types.NamespacedName) []*mod
  }
  return nil
 }
+
+func serviceUpdateNeedsPush(prev, curr *model.Service) bool {
+ if !features.EnableOptimizedServicePush {
+ return true
+ }
+ if prev == nil {
+ return true
+ }
+ return !prev.Equals(curr)
+}

pilot/pkg/serviceregistry/kube/controller/serviceimportcache.go

@@ -183,7 +183,7 @@ func (ic *serviceImportCacheImpl) onServiceImportEvent(_, obj controllers.Object
 
  // The service already existed. Treat it as an update.
  event = model.EventUpdate
-
+ mcsService = mcsService.DeepCopy()
  if ic.updateIPs(mcsService, ips) {
  needsFullPush = true
  }
@@ -193,6 +193,7 @@ func (ic *serviceImportCacheImpl) onServiceImportEvent(_, obj controllers.Object
  // a change to the discoverability policy.
  ic.addOrUpdateService(nil, mcsService, event, true)
 
+ // TODO: do we really need a full push, we should do it in `addOrUpdateService`.
  if needsFullPush {
  ic.doFullPush(mcsHost, si.GetNamespace())
  }