feat(endpoint_slice): refactor and support arbitrary providers

Signed-off-by: SkalaNetworks <contact@skala.network>

Author: SkalaNetworks

pkg/controller/endpoint_slice.go

@@ -85,7 +85,6 @@ func (c *Controller) handleUpdateEndpointSlice(key string) error {
 	svc := cachedService.DeepCopy()
 
 	var (
-		pods                     []*v1.Pod
 		lbVips                   []string
 		vip, vpcName, subnetName string
 		ok                       bool
@@ -123,11 +122,7 @@ func (c *Controller) handleUpdateEndpointSlice(key string) error {
 		}
 	}
 
-	if pods, err = c.podsLister.Pods(namespace).List(labels.Set(svc.Spec.Selector).AsSelector()); err != nil {
-		klog.Errorf("failed to get pods for service %s in namespace %s: %v", name, namespace, err)
-		return err
-	}
-	vpcName, subnetName = c.getVpcSubnetName(pods, endpointSlices, svc)
+	vpcName, subnetName = c.getVpcSubnetName(endpointSlices, svc)
 
 	var (
 		vpc    *kubeovnv1.Vpc
@@ -241,40 +236,55 @@ func (c *Controller) handleUpdateEndpointSlice(key string) error {
 	return nil
 }
 
-func (c *Controller) getVpcSubnetName(pods []*v1.Pod, endpointSlices []*discoveryv1.EndpointSlice, service *v1.Service) (string, string) {
-	var (
-		vpcName    string
-		subnetName string
-	)
-
-	for _, pod := range pods {
-		if len(pod.Annotations) == 0 {
-			continue
+func (c *Controller) getVpcSubnetName(endpointSlices []*discoveryv1.EndpointSlice, service *v1.Service) (vpcName, subnetName string) {
+	// Let the user self-determine what VPC and subnet to use
+	if service.Annotations != nil {
+		if vpc := service.Annotations[util.LogicalRouterAnnotation]; vpc != "" {
+			vpcName = vpc
 		}
-		if subnetName == "" {
-			subnetName = pod.Annotations[util.LogicalSwitchAnnotation]
+		if subnet := service.Annotations[util.LogicalSwitchAnnotation]; subnet != "" {
+			subnetName = subnet
 		}
 
-	LOOP:
-		for _, endpointSlice := range endpointSlices {
-			for _, endpoint := range endpointSlice.Endpoints {
-				for _, addr := range endpoint.Addresses {
-					for _, podIP := range pod.Status.PodIPs {
-						if addr == podIP.IP {
-							if vpcName == "" {
-								vpcName = pod.Annotations[util.LogicalRouterAnnotation]
-							}
-							if vpcName != "" {
-								break LOOP
-							}
-						}
-					}
-				}
-			}
+		if vpcName != "" && subnetName != "" {
+			return vpcName, subnetName
 		}
+	}
 
-		if vpcName != "" && subnetName != "" {
-			break
+	for _, slice := range endpointSlices {
+		for _, endpoint := range slice.Endpoints {
+			if endpoint.TargetRef == nil {
+				continue
+			}
+
+			namespace := endpoint.TargetRef.Namespace
+			name := endpoint.TargetRef.Name
+
+			pod, err := c.podsLister.Pods(namespace).Get(name)
+			if err != nil {
+				err := fmt.Errorf("couldn't retrieve pod %s/%s", namespace, name)
+				klog.Error(err)
+				continue
+			}
+
+			vpc, subnet, err := c.getEndpointVpcAndSubnet(pod, endpoint.Addresses)
+			if err != nil {
+				err := fmt.Errorf("couldn't retrieve get subnet/vpc for pod %s/%s", namespace, name)
+				klog.Error(err)
+				continue
+			}
+
+			if vpcName == "" {
+				vpcName = vpc
+			}
+
+			if subnetName == "" {
+				subnetName = subnet
+			}
+
+			if vpcName != "" && subnetName != "" {
+				return vpcName, subnetName
+			}
 		}
 	}
 
@@ -375,7 +385,14 @@ func (c *Controller) getIPPortMappingBackend(endpointSlices []*discoveryv1.Endpo
 
 		for _, endpoint := range endpointSlice.Endpoints {
 			if isGenIPPortMapping && endpoint.TargetRef.Name != "" {
-				lspName, err := c.getEndpointTargetLSP(endpoint.TargetRef.Name, endpoint.TargetRef.Namespace, endpoint.Addresses)
+				pod, err := c.podsLister.Pods(endpoint.TargetRef.Namespace).Get(endpoint.TargetRef.Name)
+				if err != nil {
+					err := fmt.Errorf("failed to get pod %s/%s: %w", endpoint.TargetRef.Namespace, endpoint.TargetRef.Name, err)
+					klog.Error(err)
+					continue
+				}
+
+				lspName, err := c.getEndpointTargetLSPName(pod, endpoint.Addresses)
 				if err != nil {
 					err := fmt.Errorf("couldn't get LSP for the endpoint's target: %w", err)
 					klog.Error(err)
@@ -408,6 +425,23 @@ func endpointReady(endpoint discoveryv1.Endpoint) bool {
 	return endpoint.Conditions.Ready == nil || *endpoint.Conditions.Ready
 }
 
+// getPodProviders returns all the providers available on a pod
+func (c *Controller) getPodProviders(pod *v1.Pod) ([]string, error) {
+	// Get all the networks to which the pod is attached
+	podNetworks, err := c.getPodKubeovnNets(pod)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get pod networks: %w", err)
+	}
+
+	// Retrieve all the providers
+	var providers []string
+	for _, podNetwork := range podNetworks {
+		providers = append(providers, podNetwork.ProviderName)
+	}
+
+	return providers, nil
+}
+
 // getMatchingProviderForAddress returns the provider linked to a subnet in which a particular address is present
 func getMatchingProviderForAddress(pod *v1.Pod, providers []string, address string) string {
 	if pod.Annotations == nil {
@@ -430,10 +464,29 @@ func getMatchingProviderForAddress(pod *v1.Pod, providers []string, address stri
 	return ""
 }
 
-// getEndpointTargetLSPName returns the name of the LSP for a pod targeted by an endpoint.
+// getEndpointProvider returns the provider linked to the addresses of an endpoint
+func (c *Controller) getEndpointProvider(pod *v1.Pod, addresses []string) (string, error) {
+	// Retrieve all the providers of the pod
+	providers, err := c.getPodProviders(pod)
+	if err != nil {
+		return "", err
+	}
+
+	// Get the first matching provider for any of the address in the endpoint
+	var provider string
+	for _, address := range addresses {
+		if provider = getMatchingProviderForAddress(pod, providers, address); provider != "" {
+			return provider, nil
+		}
+	}
+
+	return "", nil
+}
+
+// getEndpointTargetLSPNameFromProvider returns the name of the LSP for a pod targeted by an endpoint.
 // A custom provider can be specified if the LSP is within a subnet that doesn't use
 // the default "ovn" provider.
-func getEndpointTargetLSPName(pod *v1.Pod, provider string) string {
+func getEndpointTargetLSPNameFromProvider(pod *v1.Pod, provider string) string {
 	// If no provider is specified, use the default one
 	if provider == "" {
 		provider = util.OvnProvider
@@ -451,32 +504,59 @@ func getEndpointTargetLSPName(pod *v1.Pod, provider string) string {
 }
 
 // getEndpointTargetLSP returns the name of the LSP on which addresses are attached for a specific pod
-func (c *Controller) getEndpointTargetLSP(pod, namespace string, addresses []string) (string, error) {
-	// Retrieve the pod object from its namespace and name
-	podObj, err := c.podsLister.Pods(namespace).Get(pod)
+func (c *Controller) getEndpointTargetLSPName(pod *v1.Pod, addresses []string) (string, error) {
+	// Retrieve the provider for those addresses
+	provider, err := c.getEndpointProvider(pod, addresses)
 	if err != nil {
-		return "", fmt.Errorf("failed to get pod %s/%s: %w", namespace, pod, err)
+		return "", err
 	}
 
-	// Get all the networks to which the pod is attached
-	podNetworks, err := c.getPodKubeovnNets(podObj)
+	return getEndpointTargetLSPNameFromProvider(pod, provider), nil
+}
+
+// getSubnetByProvider returns the subnet linked to a provider on a pod
+func (c *Controller) getSubnetByProvider(pod *v1.Pod, provider string) (string, error) {
+	subnetName, exists := pod.Annotations[fmt.Sprintf(util.LogicalSwitchAnnotationTemplate, provider)]
+	if !exists {
+		return "", fmt.Errorf("couldn't find subnet linked to provider %s", provider)
+	}
+
+	return subnetName, nil
+}
+
+// getVpcByProvider returns the VPC linked to a provider on a pod
+func (c *Controller) getVpcByProvider(pod *v1.Pod, provider string) (string, error) {
+	vpcName, exists := pod.Annotations[fmt.Sprintf(util.LogicalRouterAnnotationTemplate, provider)]
+	if !exists {
+		return "", fmt.Errorf("couldn't find vpc linked to provider %s", provider)
+	}
+
+	return vpcName, nil
+}
+
+// getEndpointVpcAndSubnet returns the VPC/subnet for a pod and a set of addresses attached to it
+func (c *Controller) getEndpointVpcAndSubnet(pod *v1.Pod, addresses []string) (string, string, error) {
+	// Retrieve the provider for those addresses
+	provider, err := c.getEndpointProvider(pod, addresses)
 	if err != nil {
-		return "", fmt.Errorf("failed to get pod networks: %w", err)
+		return "", "", err
 	}
 
-	// Retrieve all the providers
-	var providers []string
-	for _, podNetwork := range podNetworks {
-		providers = append(providers, podNetwork.ProviderName)
+	if provider == "" {
+		return "", "", nil
 	}
 
-	// Get the first matching provider for any of the address in the endpoint
-	var provider string
-	for _, address := range addresses {
-		if provider = getMatchingProviderForAddress(podObj, providers, address); provider != "" {
-			break
-		}
+	// Retrieve the subnet
+	subnet, err := c.getSubnetByProvider(pod, provider)
+	if err != nil {
+		return "", "", err
+	}
+
+	// Retrieve the VPC
+	vpc, err := c.getVpcByProvider(pod, provider)
+	if err != nil {
+		return "", "", err
 	}
 
-	return getEndpointTargetLSPName(podObj, provider), nil
+	return vpc, subnet, nil
 }

pkg/controller/endpoint_slice_test.go

@@ -222,7 +222,7 @@ func TestGetEndpointTargetLSPName(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			result := getEndpointTargetLSPName(tt.pod, tt.provider)
+			result := getEndpointTargetLSPNameFromProvider(tt.pod, tt.provider)
 			if result != tt.expected {
 				t.Errorf("getEndpointTargetLSPName() = %q, want %q", result, tt.expected)
 			}

pkg/controller/switch_lb_rule.go

@@ -358,9 +358,32 @@ func generateHeadlessService(slr *kubeovnv1.SwitchLBRule, oldSvc *corev1.Service
 			},
 		}
 	}
+
+	// If the user supplies a VPC/subnet for the SLR, propagate it to the service
+	setUserDefinedNetwork(newSvc, slr)
+
 	return newSvc
 }
 
+// setUserDefinedNetwork propagates user-defined VPC/subnet from the SLR to the Service
+func setUserDefinedNetwork(service *corev1.Service, slr *kubeovnv1.SwitchLBRule) {
+	if service == nil || slr == nil || slr.Annotations == nil {
+		return
+	}
+
+	if service.Annotations == nil {
+		service.Annotations = make(map[string]string)
+	}
+
+	if vpc := slr.Annotations[util.LogicalRouterAnnotation]; vpc != "" {
+		service.Annotations[util.LogicalRouterAnnotation] = vpc
+	}
+
+	if subnet := slr.Annotations[util.LogicalSwitchAnnotation]; subnet != "" {
+		service.Annotations[util.LogicalSwitchAnnotation] = subnet
+	}
+}
+
 func generateEndpoints(slr *kubeovnv1.SwitchLBRule, oldEps *corev1.Endpoints) *corev1.Endpoints {
 	var (
 		name    string

pkg/controller/switch_lb_rule_test.go

@@ -4,6 +4,11 @@ import (
 	"reflect"
 	"testing"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	kubeovnv1 "github.com/kubeovn/kube-ovn/pkg/apis/kubeovn/v1"
+	"github.com/kubeovn/kube-ovn/pkg/util"
+
 	corev1 "k8s.io/api/core/v1"
 )
 
@@ -72,3 +77,85 @@ func Test_getIPFamilies(t *testing.T) {
 		})
 	}
 }
+
+func Test_setUserDefinedNetwork(t *testing.T) {
+	tests := []struct {
+		name    string
+		service *corev1.Service
+		slr     *kubeovnv1.SwitchLBRule
+		result  *corev1.Service
+	}{
+		{
+			name:    "Propagate VPC",
+			service: &corev1.Service{},
+			slr: &kubeovnv1.SwitchLBRule{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						util.LogicalRouterAnnotation: "test",
+					},
+				},
+			},
+			result: &corev1.Service{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						util.LogicalRouterAnnotation: "test",
+					},
+				},
+			},
+		},
+		{
+			name:    "Propagate Subnet",
+			service: &corev1.Service{},
+			slr: &kubeovnv1.SwitchLBRule{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						util.LogicalSwitchAnnotation: "test",
+					},
+				},
+			},
+			result: &corev1.Service{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						util.LogicalSwitchAnnotation: "test",
+					},
+				},
+			},
+		},
+		{
+			name:    "Propagate VPC/Subnet",
+			service: &corev1.Service{},
+			slr: &kubeovnv1.SwitchLBRule{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						util.LogicalRouterAnnotation: "test1",
+						util.LogicalSwitchAnnotation: "test2",
+					},
+				},
+			},
+			result: &corev1.Service{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						util.LogicalRouterAnnotation: "test1",
+						util.LogicalSwitchAnnotation: "test2",
+					},
+				},
+			},
+		},
+		{
+			name:    "Propagate nothing",
+			service: &corev1.Service{},
+			slr:     &kubeovnv1.SwitchLBRule{},
+			result:  &corev1.Service{},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			setUserDefinedNetwork(tt.service, tt.slr)
+
+			if !reflect.DeepEqual(*tt.service, *tt.result) {
+				t.Errorf("Expected service %v, but got %v", *tt.service, *tt.result)
+			}
+		})
+	}
+}