replace ovn legacy client with libovsdb (#3018)

cmd/controller/controller.go

@@ -22,7 +22,6 @@ import (
 
 	kubeovnv1 "github.com/kubeovn/kube-ovn/pkg/apis/kubeovn/v1"
 	"github.com/kubeovn/kube-ovn/pkg/controller"
-	"github.com/kubeovn/kube-ovn/pkg/ovs"
 	"github.com/kubeovn/kube-ovn/pkg/util"
 	"github.com/kubeovn/kube-ovn/versions"
 )
@@ -54,7 +53,6 @@ func CmdMain() {
 		util.LogFatalAndExit(err, "failed to check permission")
 	}
 
-	go loopOvnNbctlDaemon(config)
 	go func() {
 		mux := http.NewServeMux()
 		if config.EnableMetrics {
@@ -135,28 +133,6 @@ func CmdMain() {
 	})
 }
 
-func loopOvnNbctlDaemon(config *controller.Configuration) {
-	for {
-		daemonSocket := os.Getenv("OVN_NB_DAEMON")
-		time.Sleep(5 * time.Second)
-
-		if _, err := os.Stat(daemonSocket); os.IsNotExist(err) || daemonSocket == "" {
-			if err := ovs.StartOvnNbctlDaemon(config.OvnNbAddr); err != nil {
-				klog.Errorf("failed to start ovn-nbctl daemon %v", err)
-			}
-		}
-
-		// ovn-nbctl daemon may hang and cannot process further request.
-		// In case of that, we need to start a new daemon.
-		if err := ovs.CheckAlive(); err != nil {
-			klog.Warningf("ovn-nbctl daemon doesn't return, start a new daemon")
-			if err := ovs.StartOvnNbctlDaemon(config.OvnNbAddr); err != nil {
-				klog.Errorf("failed to start ovn-nbctl daemon %v", err)
-			}
-		}
-	}
-}
-
 func checkPermission(config *controller.Configuration) error {
 	resources := []string{"vpcs", "subnets", "ips", "vlans", "vpc-nat-gateways"}
 	for _, res := range resources {

cmd/controller_health_check/controller_health_check.go

@@ -8,23 +8,10 @@ import (
 	"time"
 
 	kubeovnv1 "github.com/kubeovn/kube-ovn/pkg/apis/kubeovn/v1"
-	"github.com/kubeovn/kube-ovn/pkg/ovs"
 	"github.com/kubeovn/kube-ovn/pkg/util"
 )
 
 func CmdMain() {
-	content, err := os.ReadFile("/var/run/ovn/ovn-nbctl.pid")
-	if err != nil {
-		util.LogFatalAndExit(err, "failed to get ovn-nbctl daemon pid")
-	}
-	daemonPid := strings.TrimSuffix(string(content), "\n")
-	if err := os.Setenv("OVN_NB_DAEMON", fmt.Sprintf("/var/run/ovn/ovn-nbctl.%s.ctl", daemonPid)); err != nil {
-		util.LogFatalAndExit(err, "failed to set env OVN_NB_DAEMON")
-	}
-	if err := ovs.CheckAlive(); err != nil {
-		os.Exit(1)
-	}
-
 	addr := "127.0.0.1:10660"
 	if os.Getenv("ENABLE_BIND_LOCAL_IP") == "true" {
 		podIpsEnv := os.Getenv("POD_IPS")

dist/images/Dockerfile.base

@@ -101,12 +101,6 @@ RUN curl -sSf -L --retry 3 -o /usr/local/bin/bfdd-control https://github.com/bob
     curl -sSf -L --retry 3 -o /usr/local/bin/bfdd-beacon https://github.com/bobz965/bfd-binary-for-kube-ovn-cni/releases/download/${BFDD_VERSION}/bfdd-beacon && \
     chmod +x /usr/local/bin/bfdd-control /usr/local/bin/bfdd-beacon
 
-ENV DUMB_INIT_VERSION="1.2.5"
-RUN dumb_init_arch="x86_64"; \
-    if [ "$ARCH" = "arm64" ]; then dumb_init_arch="aarch64"; fi; \
-    curl -sSf -L --retry 5 -o /usr/bin/dumb-init https://github.com/Yelp/dumb-init/releases/download/v${DUMB_INIT_VERSION}/dumb-init_${DUMB_INIT_VERSION}_${dumb_init_arch} && \
-    chmod +x /usr/bin/dumb-init
-
 RUN --mount=type=bind,target=/packages,from=ovs-builder,source=/packages  \
     dpkg -i /packages/openvswitch-*.deb /packages/python3-openvswitch*.deb && \
     dpkg -i --ignore-depends=openvswitch-switch,openvswitch-common /packages/ovn-*.deb && \
@@ -119,5 +113,3 @@ RUN --mount=type=bind,target=/packages,from=ovs-builder,source=/packages \
         rm -rf /var/lib/apt/lists/* && \
         dpkg -i --ignore-depends=openvswitch-switch,openvswitch-common /packages/*.ddeb; \
     fi
-
-ENTRYPOINT ["/usr/bin/dumb-init", "--"]

dist/images/start-controller.sh

@@ -32,29 +32,6 @@ function gen_conn_str {
 nb_addr="$(gen_conn_str 6641)"
 sb_addr="$(gen_conn_str 6642)"
 
-for ((i=0; i<3; i++)); do
-  if [[ "$ENABLE_SSL" == "false" ]]; then
-    OVN_NB_DAEMON=$(ovn-nbctl --db="$nb_addr" --pidfile --detach --overwrite-pidfile)
-  else
-    OVN_NB_DAEMON=$(ovn-nbctl -p /var/run/tls/key -c /var/run/tls/cert -C /var/run/tls/cacert --db="$nb_addr" --pidfile --detach --overwrite-pidfile)
-  fi
-  if echo -n "${OVN_NB_DAEMON}" | grep -qE '^/var/run/ovn/ovn-nbctl\.[0-9]+\.ctl$'; then
-    export OVN_NB_DAEMON
-    break
-  fi
-  if [ $(echo ${OVN_NB_DAEMON} | wc -c) -gt 64 ]; then
-    OVN_NB_DAEMON="$(echo ${OVN_NB_DAEMON} | cut -c1-64)..."
-  fi
-  echo "invalid ovn-nbctl daemon socket: \"${OVN_NB_DAEMON}\""
-  unset OVN_NB_DAEMON
-  pkill -f ovn-nbctl
-done
-
-if [ -z "${OVN_NB_DAEMON}" ]; then
-  echo "failed to start ovn-nbctl daemon"
-  exit 1
-fi
-
 exec ./kube-ovn-controller --ovn-nb-addr="$nb_addr" \
                            --ovn-sb-addr="$sb_addr" \
                            $@

dist/images/start-webhook.sh

@@ -1,4 +1,2 @@
 #!/usr/bin/env bash
-set -euo pipefail
-export OVN_NB_DAEMON=$(ovn-nbctl --db=tcp:${OVN_NB_SERVICE_HOST}:${OVN_NB_SERVICE_PORT} --pidfile --detach)
-exec ./kube-ovn-webhook --ovn-nb-host=${OVN_NB_SERVICE_HOST} --ovn-nb-port=${OVN_NB_SERVICE_PORT} $@
+exec ./kube-ovn-webhook

pkg/controller/controller.go

@@ -310,7 +310,7 @@ func Run(ctx context.Context, config *Configuration) {
 		vpcs:              &sync.Map{},
 		podSubnetMap:      &sync.Map{},
 		deletingPodObjMap: &sync.Map{},
-		ovnLegacyClient:   ovs.NewLegacyClient(config.OvnNbAddr, config.OvnTimeout, config.OvnSbAddr, config.ClusterRouter, config.ClusterTcpLoadBalancer, config.ClusterUdpLoadBalancer, config.ClusterTcpSessionLoadBalancer, config.ClusterUdpSessionLoadBalancer, config.NodeSwitch, config.NodeSwitchCIDR),
+		ovnLegacyClient:   ovs.NewLegacyClient(config.OvnTimeout, config.OvnSbAddr, config.ClusterRouter, config.ClusterTcpLoadBalancer, config.ClusterUdpLoadBalancer, config.ClusterTcpSessionLoadBalancer, config.ClusterUdpSessionLoadBalancer, config.NodeSwitch, config.NodeSwitchCIDR),
 		ipam:              ovnipam.NewIPAM(),
 		namedPort:         NewNamedPort(),
 

pkg/controller/external-gw.go

@@ -65,7 +65,6 @@ func (c *Controller) resyncExternalGateway() {
 		}
 		exGwEnabled = "true"
 		lastExGwCM = cm.Data
-		c.ovnLegacyClient.ExternalGatewayType = cm.Data["type"]
 		c.ExternalGatewayType = cm.Data["type"]
 		if err := c.updateDefaultVpcExternal(true); err != nil {
 			klog.Error("failed to update default vpc, %v", err)

pkg/controller/gc.go

@@ -119,11 +119,11 @@ func (c *Controller) gcLogicalSwitch() error {
 		klog.Errorf("failed to list subnet, %v", err)
 		return err
 	}
-	subnetNames := make([]string, 0, len(subnets))
+	subnetNames := strset.NewWithSize(len(subnets))
 	subnetMap := make(map[string]*kubeovnv1.Subnet, len(subnets))
 	for _, s := range subnets {
 		subnetMap[s.Name] = s
-		subnetNames = append(subnetNames, s.Name)
+		subnetNames.Add(s.Name)
 	}
 
 	lss, err := c.ovnClient.ListLogicalSwitch(c.config.EnableExternalVpc, nil)
@@ -152,21 +152,20 @@ func (c *Controller) gcLogicalSwitch() error {
 	}
 
 	klog.Infof("start to gc dhcp options")
-	dhcpOptions, err := c.ovnLegacyClient.ListDHCPOptions(c.config.EnableExternalVpc, "", "")
+	dhcpOptions, err := c.ovnClient.ListDHCPOptions(c.config.EnableExternalVpc, nil)
 	if err != nil {
 		klog.Errorf("failed to list dhcp options, %v", err)
 		return err
 	}
 	var uuidToDeleteList = []string{}
 	for _, item := range dhcpOptions {
-		ls := item.ExternalIds["ls"]
-		if !util.IsStringIn(ls, subnetNames) {
+		if len(item.ExternalIDs) == 0 || !subnetNames.Has(item.ExternalIDs["ls"]) {
 			uuidToDeleteList = append(uuidToDeleteList, item.UUID)
 		}
 	}
 	klog.Infof("gc dhcp options %v", uuidToDeleteList)
 	if len(uuidToDeleteList) > 0 {
-		if err = c.ovnLegacyClient.DeleteDHCPOptionsByUUIDs(uuidToDeleteList); err != nil {
+		if err = c.ovnClient.DeleteDHCPOptionsByUUIDs(uuidToDeleteList...); err != nil {
 			klog.Errorf("failed to delete dhcp options by uuids, %v", err)
 			return err
 		}
@@ -641,11 +640,14 @@ func (c *Controller) gcStaticRoute() error {
 			continue
 		}
 		if route.IPPrefix != "0.0.0.0/0" && route.IPPrefix != "::/0" && c.ipam.ContainAddress(route.IPPrefix) {
-			exist, err := c.ovnLegacyClient.NatRuleExists(route.IPPrefix)
-			if exist || err != nil {
+			exist, err := c.ovnClient.NatExists(c.config.ClusterRouter, "", "", route.IPPrefix)
+			if err != nil {
 				klog.Errorf("failed to get NatRule by LogicalIP %s, %v", route.IPPrefix, err)
 				continue
 			}
+			if exist {
+				continue
+			}
 			klog.Infof("gc static route %s %v %s %s", route.RouteTable, route.Policy, route.IPPrefix, route.Nexthop)
 			if err = c.ovnClient.DeleteLogicalRouterStaticRoute(c.config.ClusterRouter, &route.RouteTable, route.Policy, route.IPPrefix, route.Nexthop); err != nil {
 				klog.Errorf("failed to delete stale route %s %v %s %s: %v", route.RouteTable, route.Policy, route.IPPrefix, route.Nexthop, err)

pkg/controller/ovn_fip.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"strconv"
 
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -14,6 +15,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 
 	kubeovnv1 "github.com/kubeovn/kube-ovn/pkg/apis/kubeovn/v1"
+	"github.com/kubeovn/kube-ovn/pkg/ovsdb/ovnnb"
 	"github.com/kubeovn/kube-ovn/pkg/util"
 )
 
@@ -251,8 +253,9 @@ func (c *Controller) handleAddOvnFip(key string) error {
 		return err
 	}
 	// ovn add fip
-	if err = c.ovnLegacyClient.AddFipRule(vpcName, cachedEip.Status.V4Ip,
-		internalV4Ip, mac, cachedFip.Spec.IpName); err != nil {
+	if err = c.ovnClient.AddNat(vpcName, ovnnb.NATTypeDNATAndSNAT, cachedEip.Status.V4Ip,
+		internalV4Ip, mac, cachedFip.Spec.IpName,
+		map[string]string{"staleless": strconv.FormatBool(c.ExternalGatewayType == kubeovnv1.GWDistributedType)}); err != nil {
 		klog.Errorf("failed to create v4 fip, %v", err)
 		return err
 	}
@@ -327,8 +330,9 @@ func (c *Controller) handleUpdateOvnFip(key string) error {
 		return err
 	}
 	vpcName := subnet.Spec.Vpc
-	if err = c.ovnLegacyClient.AddFipRule(vpcName, cachedEip.Status.V4Ip,
-		internalV4Ip, mac, cachedFip.Spec.IpName); err != nil {
+	if err = c.ovnClient.AddNat(vpcName, ovnnb.NATTypeDNATAndSNAT, cachedEip.Status.V4Ip,
+		internalV4Ip, mac, cachedFip.Spec.IpName,
+		map[string]string{"staleless": strconv.FormatBool(c.ExternalGatewayType == kubeovnv1.GWDistributedType)}); err != nil {
 		klog.Errorf("failed to create v4 fip, %v", err)
 		return err
 	}
@@ -347,13 +351,14 @@ func (c *Controller) handleUpdateOvnFip(key string) error {
 	// fip change eip
 	if c.ovnFipChangeEip(fip, cachedEip) {
 		klog.V(3).Infof("fip change ip, old ip '%s', new ip %s", fip.Status.V4Ip, cachedEip.Status.V4Ip)
-		if err = c.ovnLegacyClient.DeleteFipRule(vpcName, fip.Status.V4Ip, internalV4Ip); err != nil {
+		if err = c.ovnClient.DeleteNat(vpcName, ovnnb.NATTypeDNATAndSNAT, fip.Status.V4Ip, internalV4Ip); err != nil {
 			klog.Errorf("failed to create fip, %v", err)
 			return err
 		}
 		// ovn add fip
-		if err = c.ovnLegacyClient.AddFipRule(vpcName, cachedEip.Status.V4Ip,
-			internalV4Ip, mac, cachedFip.Spec.IpName); err != nil {
+		if err = c.ovnClient.AddNat(vpcName, ovnnb.NATTypeDNATAndSNAT, cachedEip.Status.V4Ip,
+			internalV4Ip, mac, cachedFip.Spec.IpName,
+			map[string]string{"staleless": strconv.FormatBool(c.ExternalGatewayType == kubeovnv1.GWDistributedType)}); err != nil {
 			klog.Errorf("failed to create fip, %v", err)
 			return err
 		}
@@ -395,7 +400,7 @@ func (c *Controller) handleDelOvnFip(key string) error {
 	}
 	// ovn delete fip
 	if cachedFip.Status.Vpc != "" && cachedFip.Status.V4Eip != "" && cachedFip.Status.V4Ip != "" {
-		if err = c.ovnLegacyClient.DeleteFipRule(cachedFip.Status.Vpc, cachedFip.Status.V4Eip, cachedFip.Status.V4Ip); err != nil {
+		if err = c.ovnClient.DeleteNat(cachedFip.Status.Vpc, ovnnb.NATTypeDNATAndSNAT, cachedFip.Status.V4Eip, cachedFip.Status.V4Ip); err != nil {
 			klog.Errorf("failed to delete fip, %v", err)
 			return err
 		}