replace lb function call with ovnClient (#2598)

pkg/controller/controller_test.go

@@ -5,17 +5,22 @@ import (
 
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/require"
+
+	"k8s.io/client-go/informers"
+	coreinformers "k8s.io/client-go/informers/core/v1"
+	"k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/util/workqueue"
 
 	mockovs "github.com/kubeovn/kube-ovn/mocks/pkg/ovs"
-	"github.com/kubeovn/kube-ovn/pkg/client/clientset/versioned/fake"
-	informerfactory "github.com/kubeovn/kube-ovn/pkg/client/informers/externalversions"
+	kubeovnfake "github.com/kubeovn/kube-ovn/pkg/client/clientset/versioned/fake"
+	kubeovninformerfactory "github.com/kubeovn/kube-ovn/pkg/client/informers/externalversions"
 	kubeovninformer "github.com/kubeovn/kube-ovn/pkg/client/informers/externalversions/kubeovn/v1"
 )
 
 type fakeControllerInformers struct {
-	vpcInformer    kubeovninformer.VpcInformer
-	sbunetInformer kubeovninformer.SubnetInformer
+	vpcInformer     kubeovninformer.VpcInformer
+	sbunetInformer  kubeovninformer.SubnetInformer
+	serviceInformer coreinformers.ServiceInformer
 }
 
 type fakeController struct {
@@ -27,21 +32,28 @@ type fakeController struct {
 func alwaysReady() bool { return true }
 
 func newFakeController(t *testing.T) *fakeController {
-	/* kube ovn fake client */
-	kubeovnClient := fake.NewSimpleClientset()
-	kubeovnInformerFactory := informerfactory.NewSharedInformerFactory(kubeovnClient, 0)
+	/* fake kube client */
+	kubeClient := fake.NewSimpleClientset()
+	kubeInformerFactory := informers.NewSharedInformerFactory(kubeClient, 0)
+	serviceInformer := kubeInformerFactory.Core().V1().Services()
+
+	/* fake kube ovn client */
+	kubeovnClient := kubeovnfake.NewSimpleClientset()
+	kubeovnInformerFactory := kubeovninformerfactory.NewSharedInformerFactory(kubeovnClient, 0)
 	vpcInformer := kubeovnInformerFactory.Kubeovn().V1().Vpcs()
 	sbunetInformer := kubeovnInformerFactory.Kubeovn().V1().Subnets()
 
 	fakeInformers := &fakeControllerInformers{
-		vpcInformer:    vpcInformer,
-		sbunetInformer: sbunetInformer,
+		vpcInformer:     vpcInformer,
+		sbunetInformer:  sbunetInformer,
+		serviceInformer: serviceInformer,
 	}
 
 	/* ovn fake client */
 	mockOvnClient := mockovs.NewMockOvnClient(gomock.NewController(t))
 
 	ctrl := &Controller{
+		servicesLister:        serviceInformer.Lister(),
 		vpcsLister:            vpcInformer.Lister(),
 		vpcSynced:             alwaysReady,
 		subnetsLister:         sbunetInformer.Lister(),

pkg/controller/endpoint.go

@@ -192,18 +192,18 @@ func (c *Controller) handleUpdateEndpoint(key string) error {
 
 			// for performance reason delete lb with no backends
 			if len(backends) != 0 {
-				err = c.ovnLegacyClient.CreateLoadBalancerRule(lb, vip, backends, string(port.Protocol))
-				if err != nil {
+				if err = c.ovnClient.LoadBalancerAddVips(lb, map[string]string{vip: backends}); err != nil {
 					klog.Errorf("failed to add vip %s with backends %s to LB %s: %v", vip, backends, lb, err)
 					return err
 				}
 			} else {
-				if err = c.ovnLegacyClient.DeleteLoadBalancerVip(vip, lb); err != nil {
+				if err := c.ovnClient.LoadBalancerDeleteVips(lb, vip); err != nil {
 					klog.Errorf("failed to delete vip %s from LB %s: %v", vip, lb, err)
 					return err
 				}
-				if err = c.ovnLegacyClient.DeleteLoadBalancerVip(vip, oldLb); err != nil {
-					klog.Errorf("failed to delete vip %s from LB %s: %v", vip, oldLb, err)
+
+				if err := c.ovnClient.LoadBalancerDeleteVips(oldLb, vip); err != nil {
+					klog.Errorf("failed to delete vip %s from LB %s: %v", vip, lb, err)
 					return err
 				}
 			}

pkg/controller/gc.go

@@ -17,6 +17,7 @@ import (
 	"github.com/kubeovn/kube-ovn/pkg/ovs"
 	"github.com/kubeovn/kube-ovn/pkg/ovsdb/ovnnb"
 	"github.com/kubeovn/kube-ovn/pkg/util"
+	"github.com/ovn-org/libovsdb/ovsdb"
 )
 
 var lastNoPodLSP map[string]bool
@@ -405,15 +406,8 @@ func (c *Controller) gcLoadBalancer() error {
 					continue
 				}
 
-				err = c.ovnLegacyClient.RemoveLbFromLogicalSwitch(
-					subnetName,
-					vpc.Status.TcpLoadBalancer,
-					vpc.Status.TcpSessionLoadBalancer,
-					vpc.Status.UdpLoadBalancer,
-					vpc.Status.UdpSessionLoadBalancer,
-					vpc.Status.SctpLoadBalancer,
-					vpc.Status.SctpSessionLoadBalancer)
-				if err != nil {
+				lbs := []string{vpc.Status.TcpLoadBalancer, vpc.Status.TcpSessionLoadBalancer, vpc.Status.UdpLoadBalancer, vpc.Status.UdpSessionLoadBalancer, vpc.Status.SctpLoadBalancer, vpc.Status.SctpSessionLoadBalancer}
+				if err := c.ovnClient.LogicalSwitchUpdateLoadBalancers(subnetName, ovsdb.MutateOperationDelete, lbs...); err != nil {
 					return err
 				}
 			}
@@ -434,14 +428,9 @@ func (c *Controller) gcLoadBalancer() error {
 			}
 		}
 
-		// delete
-		ovnLbs, err := c.ovnLegacyClient.ListLoadBalancer()
-		if err != nil {
-			klog.Errorf("failed to list load balancer, %v", err)
-			return err
-		}
-		if err = c.ovnLegacyClient.DeleteLoadBalancer(ovnLbs...); err != nil {
-			klog.Errorf("failed to delete load balancer, %v", err)
+		// lbs will remove from logical switch automatically when delete lbs
+		if err = c.ovnClient.DeleteLoadBalancers(nil); err != nil {
+			klog.Errorf("delete all load balancers: %v", err)
 			return err
 		}
 		return nil
@@ -502,19 +491,21 @@ func (c *Controller) gcLoadBalancer() error {
 		tcpSessLb, udpSessLb, sctpSessLb := vpc.Status.TcpSessionLoadBalancer, vpc.Status.UdpSessionLoadBalancer, vpc.Status.SctpSessionLoadBalancer
 		vpcLbs = append(vpcLbs, tcpLb, udpLb, sctpLb, tcpSessLb, udpSessLb, sctpSessLb)
 
-		removeVIP := func(lb string, svcVips map[string]struct{}) error {
-			if lb == "" {
+		removeVIP := func(lbName string, svcVips map[string]struct{}) error {
+			if lbName == "" {
 				return nil
 			}
-			vips, err := c.ovnLegacyClient.GetLoadBalancerVips(lb)
+
+			lb, err := c.ovnClient.GetLoadBalancer(lbName, false)
 			if err != nil {
-				klog.Errorf("failed to get vips of LB %s: %v", lb, err)
+				klog.Errorf("get LB %s: %v", lbName, err)
 				return err
 			}
-			for vip := range vips {
+
+			for vip := range lb.Vips {
 				if _, ok := svcVips[vip]; !ok {
-					if err = c.ovnLegacyClient.DeleteLoadBalancerVip(vip, lb); err != nil {
-						klog.Errorf("failed to delete vip %s from LB %s: %v", vip, lb, err)
+					if err = c.ovnClient.LoadBalancerDeleteVips(lbName, vip); err != nil {
+						klog.Errorf("failed to delete vip %s from LB %s: %v", vip, lbName, err)
 						return err
 					}
 				}
@@ -542,23 +533,14 @@ func (c *Controller) gcLoadBalancer() error {
 		}
 	}
 
-	ovnLbs, err := c.ovnLegacyClient.ListLoadBalancer()
-	if err != nil {
-		klog.Errorf("failed to list load balancer, %v", err)
+	// delete lbs
+	if err = c.ovnClient.DeleteLoadBalancers(func(lb *ovnnb.LoadBalancer) bool {
+		return !util.ContainsString(vpcLbs, lb.Name)
+	}); err != nil {
+		klog.Errorf("delete load balancers: %v", err)
 		return err
 	}
 
-	klog.Infof("vpcLbs: %v", vpcLbs)
-	klog.Infof("ovnLbs: %v", ovnLbs)
-	for _, lb := range ovnLbs {
-		if util.ContainsString(vpcLbs, lb) {
-			continue
-		}
-		klog.Infof("start to destroy load balancer %s", lb)
-		if err := c.ovnLegacyClient.DeleteLoadBalancer(lb); err != nil {
-			return err
-		}
-	}
 	return nil
 }
 

pkg/controller/init.go

@@ -202,26 +202,19 @@ func (c *Controller) initClusterRouter() error {
 
 func (c *Controller) initLB(name, protocol string, sessionAffinity bool) error {
 	protocol = strings.ToLower(protocol)
-	uuid, err := c.ovnLegacyClient.FindLoadbalancer(name)
-	if err != nil {
-		return fmt.Errorf("failed to find %s LB: %v", protocol, err)
+
+	var selectFields string
+	if sessionAffinity {
+		selectFields = string(ovnnb.LoadBalancerSelectionFieldsIPSrc)
 	}
-	if uuid == "" {
-		klog.Infof("init cluster %s load balancer %s", protocol, name)
-		var selectFields string
-		if sessionAffinity {
-			selectFields = string(ovnnb.LoadBalancerSelectionFieldsIPSrc)
-		}
-		if err = c.ovnLegacyClient.CreateLoadBalancer(name, protocol, selectFields); err != nil {
-			klog.Errorf("failed to create cluster %s load balancer: %v", protocol, err)
-			return err
-		}
-	} else {
-		klog.Infof("%s load balancer %s already exists: uuid = %s", protocol, name, uuid)
+
+	if err := c.ovnClient.CreateLoadBalancer(name, protocol, selectFields); err != nil {
+		klog.Errorf("create load balancer %s: %v", name, err)
+		return err
 	}
 
 	if sessionAffinity {
-		if err = c.ovnLegacyClient.SetLoadBalancerAffinityTimeout(name, util.DefaultServiceSessionStickinessTimeout); err != nil {
+		if err := c.ovnClient.SetLoadBalancerAffinityTimeout(name, util.DefaultServiceSessionStickinessTimeout); err != nil {
 			klog.Errorf("failed to set affinity timeout of %s load balancer %s: %v", protocol, name, err)
 			return err
 		}

pkg/controller/ovn_dnat.go

@@ -4,15 +4,17 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"net"
+
 	kubeovnv1 "github.com/kubeovn/kube-ovn/pkg/apis/kubeovn/v1"
 	"github.com/kubeovn/kube-ovn/pkg/util"
+	"github.com/ovn-org/libovsdb/ovsdb"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/klog/v2"
-	"net"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 )
 
@@ -589,13 +591,18 @@ func (c *Controller) AddDnatRule(vpcName, dnatName, externalIp, internalIp, exte
 	externalEndpoint := net.JoinHostPort(externalIp, externalPort)
 	internalEndpoint := net.JoinHostPort(internalIp, internalPort)
 
-	if err := c.ovnLegacyClient.CreateLoadBalancerRule(dnatName, externalEndpoint, internalEndpoint, protocol); err != nil {
-		klog.Errorf("failed to create loadBalancer rule, %v", err)
+	if err := c.ovnClient.CreateLoadBalancer(dnatName, protocol, ""); err != nil {
+		klog.Errorf("create loadBalancer %s: %v", dnatName, err)
 		return err
 	}
 
-	if err := c.ovnLegacyClient.AddLoadBalancerToLogicalRouter(dnatName, vpcName); err != nil {
-		klog.Errorf("failed to add lb %s to vpc %s, %v", dnatName, vpcName, err)
+	if err := c.ovnClient.LoadBalancerAddVips(dnatName, map[string]string{externalEndpoint: internalEndpoint}); err != nil {
+		klog.Errorf("add vip %s with backends %s to LB %s: %v", externalEndpoint, internalEndpoint, dnatName, err)
+		return err
+	}
+
+	if err := c.ovnClient.LogicalRouterUpdateLoadBalancers(vpcName, ovsdb.MutateOperationInsert, dnatName); err != nil {
+		klog.Errorf("add lb %s to vpc %s: %v", dnatName, vpcName, err)
 		return err
 	}
 	return nil
@@ -604,14 +611,15 @@ func (c *Controller) AddDnatRule(vpcName, dnatName, externalIp, internalIp, exte
 func (c *Controller) DelDnatRule(vpcName, dnatName, externalIp, externalPort string) error {
 	externalEndpoint := net.JoinHostPort(externalIp, externalPort)
 
-	if err := c.ovnLegacyClient.DeleteLoadBalancerVip(externalEndpoint, dnatName); err != nil {
-		klog.Errorf("failed to delete loadBalancer rule, %v", err)
+	if err := c.ovnClient.LoadBalancerDeleteVips(dnatName, externalEndpoint); err != nil {
+		klog.Errorf("delete loadBalancer vips %s: %v", externalEndpoint, err)
 		return err
 	}
 
-	if err := c.ovnLegacyClient.RemoveLoadBalancerFromLogicalRouter(dnatName, vpcName); err != nil {
-		klog.Errorf("failed to remove lb %s from vpc %s, dnatName, vpcName, %v", err)
+	if err := c.ovnClient.LogicalRouterUpdateLoadBalancers(vpcName, ovsdb.MutateOperationDelete, dnatName); err != nil {
+		klog.Errorf("failed to remove lb %s from vpc %s: %v", dnatName, vpcName, err)
 		return err
 	}
+
 	return nil
 }