Merge pull request oracle#376 in OKE/oci-cloud-controller-manager from hotfix/service-controller to release-1.24

* commit '5a10f40efb215df6babbfd43ac45c0aa5ebd13d7':
  Deduplicating endpoints, simplifying virtual node check

Author: Akshay Kumar

pkg/cloudprovider/providers/oci/ccm.go

@@ -53,6 +53,10 @@ const (
 	// (OCI) cloud-provider.
 	providerName   = "oci"
 	providerPrefix = providerName + "://"
+
+	// endpointSliceUpdatesBatchPeriod is the duration after which a service
+	// owning an endpointslice is batched in the Service Controller
+	endpointSliceUpdatesBatchPeriod = 5*time.Second
 )
 
 // ProviderName uniquely identifies the Oracle Bare Metal Cloud Services (OCI)
@@ -279,7 +283,7 @@ func startOciServiceController(ctx context.Context, initContext cloudControllerM
 		completedConfig.SharedInformers.Core().V1().Nodes(),
 		completedConfig.SharedInformers.Discovery().V1().EndpointSlices(),
 		completedConfig.ComponentConfig.KubeCloudShared.ClusterName,
-		5*time.Second,
+		endpointSliceUpdatesBatchPeriod,
 		utilfeature.DefaultFeatureGate,
 	)
 	if err != nil {

pkg/cloudprovider/providers/oci/instances_test.go

@@ -305,6 +305,7 @@ var (
 					Conditions: v1discovery.EndpointConditions{
 						Ready:       &ready,
 					},
+					Addresses: []string{"0.0.0.9"},
 				},
 				{
 					TargetRef: &v1.ObjectReference{
@@ -314,6 +315,7 @@ var (
 					Conditions: v1discovery.EndpointConditions{
 						Ready:       &ready,
 					},
+					Addresses: []string{"0.0.0.10"},
 				},
 			},
 		},
@@ -332,6 +334,7 @@ var (
 					Conditions: v1discovery.EndpointConditions{
 						Ready:       &ready,
 					},
+					Addresses: []string{"0.0.0.19"},
 				},
 				{
 					TargetRef: &v1.ObjectReference{
@@ -341,6 +344,7 @@ var (
 					Conditions: v1discovery.EndpointConditions{
 						Ready:       &ready,
 					},
+					Addresses: []string{"0.0.0.20"},
 				},
 			},
 		},
@@ -360,6 +364,7 @@ var (
 					Conditions: v1discovery.EndpointConditions{
 						Ready:       &ready,
 					},
+					Addresses: []string{"0.0.0.9"},
 				},
 				{
 					TargetRef: &v1.ObjectReference{
@@ -369,6 +374,7 @@ var (
 					Conditions: v1discovery.EndpointConditions{
 						Ready:       &ready,
 					},
+					Addresses: []string{"0.0.0.19"},
 				},
 				{
 					TargetRef: &v1.ObjectReference{
@@ -378,6 +384,7 @@ var (
 					Conditions: v1discovery.EndpointConditions{
 						Ready:       &ready,
 					},
+					Addresses: []string{"0.0.0.10"},
 				},
 				{
 					TargetRef: &v1.ObjectReference{
@@ -387,6 +394,7 @@ var (
 					Conditions: v1discovery.EndpointConditions{
 						Ready:       &ready,
 					},
+					Addresses: []string{"0.0.0.20"},
 				},
 			},
 		},
@@ -405,6 +413,65 @@ var (
 					Conditions: v1discovery.EndpointConditions{
 						Ready:       &ready,
 					},
+					Addresses: []string{"0.0.0.100"},
+				},
+			},
+		},
+		"endpointSliceDuplicate1.1": {
+			ObjectMeta: metav1.ObjectMeta{
+				Labels: map[string]string{
+					v1discovery.LabelServiceName: "duplicateEndpointsService",
+				},
+			},
+			Endpoints: []v1discovery.Endpoint{
+				{
+					TargetRef: &v1.ObjectReference{
+						Kind: "Pod",
+						Name: "virtualPod1",
+					},
+					Conditions: v1discovery.EndpointConditions{
+						Ready:       &ready,
+					},
+					Addresses: []string{"0.0.0.10"},
+				},
+				{
+					TargetRef: &v1.ObjectReference{
+						Kind: "Pod",
+						Name: "virtualPod2",
+					},
+					Conditions: v1discovery.EndpointConditions{
+						Ready:       &ready,
+					},
+					Addresses: []string{"0.0.0.9"},
+				},
+			},
+		},
+		"endpointSliceDuplicate1.2": {
+			ObjectMeta: metav1.ObjectMeta{
+				Labels: map[string]string{
+					v1discovery.LabelServiceName: "duplicateEndpointsService",
+				},
+			},
+			Endpoints: []v1discovery.Endpoint{
+				{
+					TargetRef: &v1.ObjectReference{
+						Kind: "Pod",
+						Name: "virtualPod2",
+					},
+					Conditions: v1discovery.EndpointConditions{
+						Ready:       &ready,
+					},
+					Addresses: []string{"0.0.0.9"},
+				},
+				{
+					TargetRef: &v1.ObjectReference{
+						Kind: "Pod",
+						Name: "virtualPod1",
+					},
+					Conditions: v1discovery.EndpointConditions{
+						Ready:       &ready,
+					},
+					Addresses: []string{"0.0.0.10"},
 				},
 			},
 		},

pkg/cloudprovider/providers/oci/load_balancer.go

@@ -22,6 +22,7 @@ import (
 	"go.uber.org/zap"
 	v1 "k8s.io/api/core/v1"
 	discovery "k8s.io/api/discovery/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/util/sets"
@@ -374,31 +375,23 @@ func getNodeFilter(svc *v1.Service) (labels.Selector, error) {
 	return labels.Parse(labelSelector)
 }
 
-// filterNodes - Separates provisioned nodes based on the label selector,
-// that should be backends in the load balancer, and virtual nodes.
-func filterNodes(logger *zap.SugaredLogger, svc *v1.Service, nodes []*v1.Node) ([]*v1.Node, []*v1.Node, error) {
+// filterNodes based on the label selector, if present, and returns the set of nodes
+// that should be backends in the load balancer.
+func filterNodes(svc *v1.Service, nodes []*v1.Node) ([]*v1.Node, error) {
 
 	selector, err := getNodeFilter(svc)
 	if err != nil {
-		return nil, nil, err
+		return nil, err
 	}
 
-	var provisionedNodes, virtualNodes []*v1.Node
+	var filteredNodes []*v1.Node
 	for _, n := range nodes {
-		// Since virtual nodes should not be added as backends,
-		// labels are not checked
-		isVirtualNode, err := IsVirtualNode(n)
-		if err != nil {
-			logger.With(zap.Error(err)).Errorf("Failed to check if node is virtual: %s", n.Name)
-		}
-		if isVirtualNode {
-			virtualNodes = append(virtualNodes, n)
-		} else if selector.Matches(labels.Set(n.GetLabels())) {
-			provisionedNodes = append(provisionedNodes, n)
+		if selector.Matches(labels.Set(n.GetLabels())) {
+			filteredNodes = append(filteredNodes, n)
 		}
 	}
 
-	return provisionedNodes, virtualNodes, nil
+	return filteredNodes, nil
 }
 
 // EnsureLoadBalancer creates a new load balancer or updates the existing one.
@@ -409,22 +402,17 @@ func (cp *CloudProvider) EnsureLoadBalancer(ctx context.Context, clusterName str
 	loadBalancerType := getLoadBalancerType(service)
 	logger := cp.logger.With("loadBalancerName", lbName, "serviceName", service.Name, "loadBalancerType", loadBalancerType)
 
-	// Ideally virtual nodes will be excluded since they will have the LabelNodeExcludeBalancers set on them,
-	// However, the user could delete the label, so we need to filter out virtual nodes here
-	provisionedNodes, virtualNodes, err := filterNodes(logger, service, nodes)
+	nodes, err := filterNodes(service, nodes)
 	if err != nil {
 		logger.With(zap.Error(err)).Error("Failed to filter provisioned nodes with label selector and virtual nodes")
 		return nil, err
 	}
 
-	virtualNodeExists := len(virtualNodes) > 0
-	if !virtualNodeExists {
-		// Check if virtual nodes exist in the cluster
-		virtualNodeExists, err = cp.virtualNodeExists(logger)
-		if err != nil {
-			logger.With(zap.Error(err)).Error("Failed to check if cluster has virtual nodes")
-			return nil, errors.Wrap(err, "failed to check if cluster has virtual nodes")
-		}
+	// Check if virtual nodes exist in the cluster
+	virtualNodeExists, err := VirtualNodeExists(cp.NodeLister)
+	if err != nil {
+		logger.With(zap.Error(err)).Error("Failed to check if cluster has virtual nodes")
+		return nil, errors.Wrap(err, "failed to check if cluster has virtual nodes")
 	}
 
 	var virtualPods []*v1.Pod
@@ -437,7 +425,7 @@ func (cp *CloudProvider) EnsureLoadBalancer(ctx context.Context, clusterName str
 		}
 	}
 
-	logger.With("provisioned nodes", len(provisionedNodes), "virtual pods", len(virtualPods)).Info("Ensuring load balancer")
+	logger.With("provisioned nodes", len(nodes), "virtual pods", len(virtualPods)).Info("Ensuring load balancer")
 
 	dimensionsMap := make(map[string]string)
 
@@ -493,7 +481,7 @@ func (cp *CloudProvider) EnsureLoadBalancer(ctx context.Context, clusterName str
 		return nil, err
 	}
 
-	spec, err := NewLBSpec(logger, service, provisionedNodes, virtualPods, subnets, sslConfig, cp.securityListManagerFactory, cp.config.Tags)
+	spec, err := NewLBSpec(logger, service, nodes, virtualPods, subnets, sslConfig, cp.securityListManagerFactory, cp.config.Tags)
 	if err != nil {
 		logger.With(zap.Error(err)).Error("Failed to derive LBSpec")
 		errorType = util.GetError(err)
@@ -549,7 +537,8 @@ func (cp *CloudProvider) EnsureLoadBalancer(ctx context.Context, clusterName str
 		}
 	}
 
-	if len(provisionedNodes) == 0 && !virtualNodeExists {
+	// TODO: Revisit this condition when clusters with mixed node pools are introduced, possibly add len(virtualPods) == 0 check
+	if len(nodes) == 0 && !virtualNodeExists {
 		allNodesNotReady, err := cp.checkAllBackendNodesNotReady()
 		if err != nil {
 			logger.With(zap.Error(err)).Error("Failed to check if all backend nodes are not ready")
@@ -884,7 +873,7 @@ func (cp *CloudProvider) UpdateLoadBalancer(ctx context.Context, clusterName str
 }
 
 // getNodesAndPodsByIPs returns slices of Nodes and Pods corresponding to the given IP addresses.
-func (cp *CloudProvider) getNodesAndPodsByIPs(ctx context.Context, logger *zap.SugaredLogger, backendIPs []string, service *v1.Service) ([]*v1.Node, []*v1.Pod, error) {
+func (cp *CloudProvider) getNodesAndPodsByIPs(ctx context.Context, backendIPs []string, service *v1.Service) ([]*v1.Node, []*v1.Pod, error) {
 	ipToPodLookup := make(map[string]*v1.Pod)
 	ipToNodeLookup := make(map[string]*v1.Node)
 
@@ -895,11 +884,7 @@ func (cp *CloudProvider) getNodesAndPodsByIPs(ctx context.Context, logger *zap.S
 
 	var virtualNodeExists bool
 	for _, node := range nodeList {
-		isVirtualNode, err := IsVirtualNode(node)
-		if err != nil {
-			logger.With(zap.Error(err)).Errorf("Failed to check if node is virtual: %s", node.Name)
-		}
-		if isVirtualNode {
+		if IsVirtualNode(node) {
 			virtualNodeExists = true
 			continue
 		}
@@ -925,7 +910,7 @@ func (cp *CloudProvider) getNodesAndPodsByIPs(ctx context.Context, logger *zap.S
 		if node, nodeExists := ipToNodeLookup[ip]; nodeExists {
 			nodes = append(nodes, node)
 		} else if pod, podExists := ipToPodLookup[ip]; virtualNodeExists && podExists {
-			 pods = append(pods, pod)
+			pods = append(pods, pod)
 		} else {
 			return nil, nil, errors.Errorf("provisioned node or virtual pod was not found by IP %q", ip)
 		}
@@ -1021,7 +1006,7 @@ func (cp *CloudProvider) cleanupSecListForLoadBalancerDelete(lb *client.GenericL
 			ipSet.Insert(*backend.IpAddress)
 		}
 	}
-	nodes, _, err := cp.getNodesAndPodsByIPs(ctx, logger, ipSet.List(), service)
+	nodes, _, err := cp.getNodesAndPodsByIPs(ctx, ipSet.List(), service)
 	if err != nil {
 		logger.With(zap.Error(err)).Error("Failed to fetch nodes by internal ips")
 		return errors.Wrap(err, "fetching nodes by internal ips")
@@ -1175,48 +1160,45 @@ func (cp *CloudProvider) getVirtualPodsOfService(ctx context.Context, logger *za
 	if err != nil {
 		return nil, err
 	}
+
+	endpointSet := make(map[string]struct{})
 	var virtualPods []*v1.Pod
 	for _, es := range endpointSlices {
 		for _, e := range es.Endpoints {
 			if e.TargetRef.Kind == "Pod" {
+				if _, exists := endpointSet[e.Addresses[0]]; exists {
+					continue
+				}
+
 				pod, err := cp.kubeclient.CoreV1().Pods(es.Namespace).Get(ctx, e.TargetRef.Name, metav1.GetOptions{})
 				if err != nil {
+					if apierrors.IsNotFound(err) {
+						logger.With(zap.Error(err)).Errorf("Pod object does not exist: %s", e.TargetRef.Name)
+						continue
+					}
 					return nil, err
 				}
+
 				node, err := cp.NodeLister.Get(pod.Spec.NodeName)
 				if err != nil {
+					if apierrors.IsNotFound(err) {
+						logger.With(zap.Error(err)).Errorf("Node object does not exist: %s", pod.Spec.NodeName)
+						continue
+					}
 					return nil, err
 				}
-				isVirtualNode, err := IsVirtualNode(node)
-				if err != nil {
-					logger.With(zap.Error(err)).Errorf("Failed to check if node is virtual: %s", node.Name)
-				}
-				if isVirtualNode {
+
+				if IsVirtualNode(node) {
 					virtualPods = append(virtualPods, pod)
 				}
+
+				endpointSet[e.Addresses[0]] = struct{}{}
 			}
 		}
 	}
 	return virtualPods, nil
 }
 
-func (cp *CloudProvider) virtualNodeExists(logger *zap.SugaredLogger) (bool, error) {
-	nodeList, err := cp.NodeLister.List(labels.Everything())
-	if err != nil {
-		return false, err
-	}
-	for _, node := range nodeList {
-		isVirtualNode, err := IsVirtualNode(node)
-		if err != nil {
-			logger.With(zap.Error(err)).Errorf("Failed to check if node is virtual: %s", node.Name)
-		}
-		if isVirtualNode {
-			return true, nil
-		}
-	}
-	return false, nil
-}
-
 func (cp *CloudProvider) getEndpointSlicesForService(service *v1.Service) ([]*discovery.EndpointSlice, error) {
 	esLabelSelector := labels.Set(map[string]string{
 		discovery.LabelServiceName: service.Name,