Support ProxyTerminatingEndpoints in AntreaProxy

Signed-off-by: Hongliang Liu <[email protected]>

pkg/agent/proxy/proxier.go

@@ -468,13 +468,7 @@ func (p *proxier) installServices() {
 				} else {
 					// If the type of the Service is ClusterIP, install a group according to internalTrafficPolicy.
 					groupID := p.groupCounter.AllocateIfNotExist(svcPortName, internalPolicyLocal)
-					var endpoints []k8sproxy.Endpoint
-					if internalPolicyLocal {
-						endpoints = localEndpoints
-					} else {
-						endpoints = clusterEndpoints
-					}
-					if err = p.ofClient.InstallServiceGroup(groupID, affinityTimeout != 0, endpoints); err != nil {
+					if err = p.ofClient.InstallServiceGroup(groupID, affinityTimeout != 0, allReachableEndpoints); err != nil {
 						klog.ErrorS(err, "Error when installing Group of Endpoints for Service", "Service", svcPortName)
 						continue
 					}
@@ -486,15 +480,11 @@ func (p *proxier) installServices() {
 				// uninstall the group which has all Endpoints; if both internalTrafficPolicy and externalTrafficPolicy are
 				// Cluster, install the group which has all Endpoints and unconditionally uninstall the group which has
 				// only local Endpoints. Note that, if a group doesn't exist on OVS, then the return value will be nil.
-				bothPolicyLocal := internalPolicyLocal && externalPolicyLocal
-				var endpoints []k8sproxy.Endpoint
-				if bothPolicyLocal {
-					endpoints = localEndpoints
-				} else {
-					endpoints = clusterEndpoints
-				}
+				// Note that, since internalTrafficPolicy and externalTrafficPolicy are the same, bothPolicyLocal just equals
+				// internalPolicyLocal.
+				bothPolicyLocal := internalPolicyLocal
 				groupID := p.groupCounter.AllocateIfNotExist(svcPortName, bothPolicyLocal)
-				if err = p.ofClient.InstallServiceGroup(groupID, affinityTimeout != 0, endpoints); err != nil {
+				if err = p.ofClient.InstallServiceGroup(groupID, affinityTimeout != 0, allReachableEndpoints); err != nil {
 					klog.ErrorS(err, "Error when installing Group of local Endpoints for Service", "Service", svcPortName)
 					continue
 				}
@@ -839,6 +829,9 @@ func (p *proxier) Run(stopCh <-chan struct{}) {
 		go p.serviceConfig.Run(stopCh)
 		if p.endpointSliceEnabled {
 			go p.endpointSliceConfig.Run(stopCh)
+			if p.topologyAwareHintsEnabled {
+				go p.nodeConfig.Run(stopCh)
+			}
 		} else {
 			go p.endpointsConfig.Run(stopCh)
 		}