Only send server updates to listeners when the opaque protocol changes (

#11907)

Whenever the destination controller's informer receives an update of a Server resource, it checks every portPublisher in the endpointsWatcher to see if the Server selects any pods in that servicePort and updates those pods' opaque protocol field.  Regardless of if any pods were matched or if the opaque protocol changed, an update is sent to each listener.  This results in an update to every endpointTranslator each time a Server is updated.  During a resync, we get an update for every Server in the cluster which results in N updates to each endpointTranslator where N is the number of Servers in the cluster.

If N is greater than 100, it becomes possible that these N updates could overflow the endpointTranslator update queue if the queue is not being drained fast enough.

We change this to only send the update for a Server if at least one of the servicePort addresses was selected by that server AND it's opaque protocol field changed.

Signed-off-by: Alex Leong <[email protected]>

controller/api/destination/watcher/endpoints_watcher.go

@@ -1182,6 +1182,7 @@ func (pp *portPublisher) unsubscribe(listener EndpointUpdateListener) {
 }
 
 func (pp *portPublisher) updateServer(server *v1beta1.Server, selector labels.Selector, isAdd bool) {
+	updated := false
 	for id, address := range pp.addresses.Addresses {
 		if address.Pod != nil && selector.Matches(labels.Set(address.Pod.Labels)) {
 			var portMatch bool
@@ -1207,12 +1208,18 @@ func (pp *portPublisher) updateServer(server *v1beta1.Server, selector labels.Se
 				} else {
 					address.OpaqueProtocol = false
 				}
-				pp.addresses.Addresses[id] = address
+				if pp.addresses.Addresses[id].OpaqueProtocol != address.OpaqueProtocol {
+					pp.addresses.Addresses[id] = address
+					updated = true
+				}
 			}
 		}
 	}
-	for _, listener := range pp.listeners {
-		listener.Add(pp.addresses)
+	if updated {
+		for _, listener := range pp.listeners {
+			listener.Add(pp.addresses)
+		}
+		pp.metrics.incUpdates()
 	}
 }