kubernetes#2019

sawsa307 · Mar 22, 2023 · 9a4c233 · 9a4c233
1 parent af91d0f
commit 9a4c233
Show file tree

Hide file tree

Showing 5 changed files with 432 additions and 57 deletions.
diff --git a/pkg/neg/syncers/endpoints_calculator.go b/pkg/neg/syncers/endpoints_calculator.go
@@ -199,7 +199,7 @@ func (l *L7EndpointsCalculator) Mode() types.EndpointsCalculatorMode {
 
 // CalculateEndpoints determines the endpoints in the NEGs based on the current service endpoints and the current NEGs.
 func (l *L7EndpointsCalculator) CalculateEndpoints(eds []types.EndpointsData, _ map[string]types.NetworkEndpointSet) (map[string]types.NetworkEndpointSet, types.EndpointPodMap, int, error) {
-	return toZoneNetworkEndpointMap(eds, l.zoneGetter, l.servicePortName, l.networkEndpointType, l.lpConfig)
+	return toZoneNetworkEndpointMap(eds, l.podLister, l.zoneGetter, l.servicePortName, l.networkEndpointType, l.lpConfig)
 }
 
 func nodeMapToString(nodeMap map[string][]*v1.Node) string {

diff --git a/pkg/neg/syncers/transaction.go b/pkg/neg/syncers/transaction.go
@@ -193,6 +193,7 @@ func (s *transactionSyncer) syncInternal() error {
 	start := time.Now()
 	err := s.syncInternalImpl()
 	if err != nil {
+		s.logger.V(3).Info("Setting error state", "err", err, "errorState", s.getErrorStateReason(err))
 		s.setErrorState(s.getErrorStateReason(err))
 	}
 	s.updateStatus(err)
@@ -261,8 +262,7 @@ func (s *transactionSyncer) syncInternalImpl() error {
 	endpointsData := negtypes.EndpointsDataFromEndpointSlices(endpointSlices)
 	targetMap, endpointPodMap, dupCount, err = s.endpointsCalculator.CalculateEndpoints(endpointsData, currentMap)
 	if err != nil {
-		s.setErrorState(s.getErrorStateReason(err))
-		return fmt.Errorf("endpoints calculation error in mode %q, err: %w", s.endpointsCalculator.Mode(), err)
+		return err
 	}
 	err = s.endpointsCalculator.ValidateEndpoints(endpointsData, endpointPodMap, dupCount)
 	if err != nil {
@@ -493,8 +493,9 @@ func (s *transactionSyncer) operationInternal(operation transactionOp, zone stri
 		endpointBatchErr := s.ValidateEndpointBatch(err, operation)
 		if endpointBatchErr != nil {
 			s.syncLock.Lock()
-			defer s.syncLock.Unlock()
+			s.logger.V(3).Info("Setting error state", "errorState", s.getErrorStateReason(endpointBatchErr))
 			s.setErrorState(s.getErrorStateReason(endpointBatchErr))
+			s.syncLock.Unlock()
 		}
 	}
 

diff --git a/pkg/neg/syncers/utils.go b/pkg/neg/syncers/utils.go
@@ -218,7 +218,7 @@ func ensureNetworkEndpointGroup(svcNamespace, svcName, negName, zone, negService
 }
 
 // toZoneNetworkEndpointMap translates addresses in endpoints object into zone and endpoints map, and also return the count for duplicated endpoints
-func toZoneNetworkEndpointMap(eds []negtypes.EndpointsData, zoneGetter negtypes.ZoneGetter, servicePortName string, networkEndpointType negtypes.NetworkEndpointType, lpConfig negtypes.PodLabelPropagationConfig) (map[string]negtypes.NetworkEndpointSet, negtypes.EndpointPodMap, int, error) {
+func toZoneNetworkEndpointMap(eds []negtypes.EndpointsData, podLister cache.Indexer, zoneGetter negtypes.ZoneGetter, servicePortName string, networkEndpointType negtypes.NetworkEndpointType, lpConfig negtypes.PodLabelPropagationConfig) (map[string]negtypes.NetworkEndpointSet, negtypes.EndpointPodMap, int, error) {
 	zoneNetworkEndpointMap := map[string]negtypes.NetworkEndpointSet{}
 	networkEndpointPodMap := negtypes.EndpointPodMap{}
 	dupCount := 0
@@ -251,19 +251,31 @@ func toZoneNetworkEndpointMap(eds []negtypes.EndpointsData, zoneGetter negtypes.
 			}
 			if endpointAddress.NodeName == nil || len(*endpointAddress.NodeName) == 0 {
 				klog.V(2).Infof("Detected unexpected error when checking missing nodeName. Endpoint %q in Endpoints %s/%s does not have an associated node. Skipping", endpointAddress.Addresses, ed.Meta.Namespace, ed.Meta.Name)
-				return nil, nil, dupCount, negtypes.ErrEPMissingNodeName
+				return nil, nil, dupCount, negtypes.ErrEPInvalidNodeName
 			}
 			if endpointAddress.TargetRef == nil {
 				klog.V(2).Infof("Endpoint %q in Endpoints %s/%s does not have an associated pod. Skipping", endpointAddress.Addresses, ed.Meta.Namespace, ed.Meta.Name)
-				continue
+				return nil, nil, dupCount, negtypes.ErrEPInvalidPod
 			}
 			zone, err := zoneGetter.GetZoneForNode(*endpointAddress.NodeName)
 			if err != nil {
-				return nil, nil, dupCount, negtypes.ErrNodeNotFound
+				return nil, nil, dupCount, negtypes.ErrEPInvalidNodeName
 			}
 			if zone == "" {
-				klog.V(2).Info("Detected unexpected error when checking missing zone")
-				return nil, nil, dupCount, negtypes.ErrEPMissingZone
+				klog.V(2).Info("Detected unexpected error when checking zone")
+				return nil, nil, dupCount, negtypes.ErrEPInvalidZone
+			}
+
+			key := fmt.Sprintf("%s/%s", endpointAddress.TargetRef.Namespace, endpointAddress.TargetRef.Name)
+			obj, exists, err := podLister.GetByKey(key)
+			if err != nil || !exists {
+				klog.V(2).Infof("Endpoint %q in Endpoints %s/%s does not correspond to an existing pod. Skipping", endpointAddress.Addresses, ed.Meta.Namespace, ed.Meta.Name)
+				return nil, nil, dupCount, negtypes.ErrEPInvalidPod
+			}
+			_, ok := obj.(*apiv1.Pod)
+			if !ok {
+				klog.V(2).Infof("Endpoint %q in Endpoints %s/%s does not correspond to an existing pod. Skipping", endpointAddress.Addresses, ed.Meta.Namespace, ed.Meta.Name)
+				return nil, nil, dupCount, negtypes.ErrEPInvalidPod
 			}
 			if zoneNetworkEndpointMap[zone] == nil {
 				zoneNetworkEndpointMap[zone] = negtypes.NewNetworkEndpointSet()