occm: implement a support for atomic routes update (#2134)

* occm: implement an optional atomic routes update

* Fix IPv6 detection logic in routes

Author: kayrus

docs/openstack-cloud-controller-manager/using-openstack-cloud-controller-manager.md

@@ -174,6 +174,11 @@ The options in `Global` section are used for openstack-cloud-controller-manager
   For example, this option can be useful when having multiple or dual-stack interfaces attached to a node and needing a user-controlled, deterministic way of sorting the addresses.
   Default: ""
 
+### Router
+
+* `router-id`
+  Specifies the Neutron router ID to manage Kubernetes cluster routes, e.g. for load balancers or compute instances that are not part of the Kubernetes cluster.
+
 ###  Load Balancer
 
 Although the openstack-cloud-controller-manager was initially implemented with Neutron-LBaaS support, Octavia is recommended now because Neutron-LBaaS has been deprecated since Queens OpenStack release cycle and no longer accepted new feature enhancements. As a result, lots of advanced features in openstack-cloud-controller-manager rely on Octavia, even the CI is running based on Octavia enabled OpenStack environment. Functionalities are not guaranteed if using Neutron-LBaaS.

go.mod

@@ -7,7 +7,7 @@ replace github.com/onsi/ginkgo/v2 => github.com/onsi/ginkgo/v2 v2.4.0
 require (
 	github.com/container-storage-interface/spec v1.7.0
 	github.com/go-chi/chi/v5 v5.0.8
-	github.com/gophercloud/gophercloud v1.2.0
+	github.com/gophercloud/gophercloud v1.2.1-0.20230227135528-e7de1a394a6e
 	github.com/gophercloud/utils v0.0.0-20230301065655-769528992f29
 	github.com/hashicorp/go-version v1.6.0
 	github.com/kubernetes-csi/csi-lib-utils v0.13.0

go.sum

@@ -218,8 +218,8 @@ github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+
 github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=
 github.com/googleapis/google-cloud-go-testing v0.0.0-20200911160855-bcd43fbb19e8/go.mod h1:dvDLG8qkwmyD9a/MJJN3XJcT3xFxOKAvTZGvuZmac9g=
 github.com/gophercloud/gophercloud v1.1.1/go.mod h1:aAVqcocTSXh2vYFZ1JTvx4EQmfgzxRcNupUfxZbBNDM=
-github.com/gophercloud/gophercloud v1.2.0 h1:1oXyj4g54KBg/kFtCdMM6jtxSzeIyg8wv4z1HoGPp1E=
-github.com/gophercloud/gophercloud v1.2.0/go.mod h1:aAVqcocTSXh2vYFZ1JTvx4EQmfgzxRcNupUfxZbBNDM=
+github.com/gophercloud/gophercloud v1.2.1-0.20230227135528-e7de1a394a6e h1:tzpcnvGylThnTal3tFmmG8zwcmeP1Wicbt698WfUhss=
+github.com/gophercloud/gophercloud v1.2.1-0.20230227135528-e7de1a394a6e/go.mod h1:aAVqcocTSXh2vYFZ1JTvx4EQmfgzxRcNupUfxZbBNDM=
 github.com/gophercloud/utils v0.0.0-20230301065655-769528992f29 h1:A2dT0wlliiBwnkWVPtSLXH1m36486l7sOwPRrjEm3hE=
 github.com/gophercloud/utils v0.0.0-20230301065655-769528992f29/go.mod h1:z4Dey7xsTUXgcB1C8elMvGRKTjV1ez0eoYQlMrduG1g=
 github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc=

pkg/openstack/openstack.go

@@ -131,7 +131,7 @@ type NetworkingOpts struct {
 
 // RouterOpts is used for Neutron routes
 type RouterOpts struct {
-	RouterID string `gcfg:"router-id"` // required
+	RouterID string `gcfg:"router-id"`
 }
 
 type ServerAttributesExt struct {
@@ -458,18 +458,23 @@ func (os *OpenStack) Routes() (cloudprovider.Routes, bool) {
 		return nil, false
 	}
 
-	if !netExts["extraroute"] {
+	if !netExts["extraroute"] && !netExts["extraroute-atomic"] {
 		klog.V(3).Info("Neutron extraroute extension not found, required for Routes support")
 		return nil, false
 	}
 
-	r, err := NewRoutes(os, network)
+	r, err := NewRoutes(os, network, netExts["extraroute-atomic"])
 	if err != nil {
 		klog.Warningf("Error initialising Routes support: %v", err)
 		return nil, false
 	}
 
-	klog.V(1).Info("Claiming to support Routes")
+	if netExts["extraroute-atomic"] {
+		klog.V(1).Info("Claiming to support Routes with atomic updates")
+	} else {
+		klog.V(1).Info("Claiming to support Routes")
+	}
+
 	return r, true
 }
 

pkg/openstack/routes.go

@@ -22,6 +22,7 @@ import (
 	"sync"
 
 	"github.com/gophercloud/gophercloud"
+	"github.com/gophercloud/gophercloud/openstack/networking/v2/extensions/layer3/extraroutes"
 	"github.com/gophercloud/gophercloud/openstack/networking/v2/extensions/layer3/routers"
 	"github.com/gophercloud/gophercloud/openstack/networking/v2/ports"
 
@@ -40,21 +41,25 @@ type Routes struct {
 	os      *OpenStack
 	// router's private network IDs
 	networkIDs []string
-	// OpenStack can modify only one route at once
+	// whether Neutron supports "extraroute-atomic" extension
+	atomicRoutes bool
+	// Neutron with no "extraroute-atomic" extension can modify only one route at
+	// once
 	sync.Mutex
 }
 
 var _ cloudprovider.Routes = &Routes{}
 
 // NewRoutes creates a new instance of Routes
-func NewRoutes(os *OpenStack, network *gophercloud.ServiceClient) (cloudprovider.Routes, error) {
+func NewRoutes(os *OpenStack, network *gophercloud.ServiceClient, atomicRoutes bool) (cloudprovider.Routes, error) {
 	if os.routeOpts.RouterID == "" {
 		return nil, errors.ErrNoRouterID
 	}
 
 	return &Routes{
-		network: network,
-		os:      os,
+		network:      network,
+		os:           os,
+		atomicRoutes: atomicRoutes,
 	}, nil
 }
 
@@ -90,9 +95,7 @@ func (r *Routes) ListRoutes(ctx context.Context, clusterName string) ([]*cloudpr
 	}
 
 	// detect router's private network ID for further VM ports filtering
-	r.Lock()
 	r.networkIDs, err = getRouterNetworkIDs(r.network, r.os.routeOpts.RouterID)
-	r.Unlock()
 	if err != nil {
 		return nil, err
 	}
@@ -146,10 +149,16 @@ func getAddrByNodeName(name types.NodeName, needIPv6 bool, nodes []*v1.Node) str
 					if ip == nil {
 						continue
 					}
-					if needIPv6 && ip.To4() == nil {
+					isIPv6 := ip.To4() == nil
+					if needIPv6 {
+						if isIPv6 {
+							return v.Address
+						}
+						continue
+					}
+					if !isIPv6 {
 						return v.Address
 					}
-					return v.Address
 				}
 			}
 		}
@@ -184,6 +193,52 @@ func updateRoutes(network *gophercloud.ServiceClient, router *routers.Router, ne
 	return unwinder, nil
 }
 
+func addRoute(network *gophercloud.ServiceClient, routerID string, newRoute []routers.Route) (func(), error) {
+	mc := metrics.NewMetricContext("router", "update")
+	_, err := extraroutes.Add(network, routerID, extraroutes.Opts{
+		Routes: &newRoute,
+	}).Extract()
+	if mc.ObserveRequest(err) != nil {
+		return nil, err
+	}
+
+	unwinder := func() {
+		klog.V(4).Infof("Reverting routes change to router %v", routerID)
+		mc := metrics.NewMetricContext("router", "update")
+		_, err := extraroutes.Remove(network, routerID, extraroutes.Opts{
+			Routes: &newRoute,
+		}).Extract()
+		if mc.ObserveRequest(err) != nil {
+			klog.Warningf("Unable to reset routes during error unwind: %v", err)
+		}
+	}
+
+	return unwinder, nil
+}
+
+func removeRoute(network *gophercloud.ServiceClient, routerID string, oldRoute []routers.Route) (func(), error) {
+	mc := metrics.NewMetricContext("router", "update")
+	_, err := extraroutes.Remove(network, routerID, extraroutes.Opts{
+		Routes: &oldRoute,
+	}).Extract()
+	if mc.ObserveRequest(err) != nil {
+		return nil, err
+	}
+
+	unwinder := func() {
+		klog.V(4).Infof("Reverting routes change to router %v", routerID)
+		mc := metrics.NewMetricContext("router", "update")
+		_, err := extraroutes.Add(network, routerID, extraroutes.Opts{
+			Routes: &oldRoute,
+		}).Extract()
+		if mc.ObserveRequest(err) != nil {
+			klog.Warningf("Unable to reset routes during error unwind: %v", err)
+		}
+	}
+
+	return unwinder, nil
+}
+
 func updateAllowedAddressPairs(network *gophercloud.ServiceClient, port *ports.Port, newPairs []ports.AddressPair) (func(), error) {
 	origPairs := port.AllowedAddressPairs // shallow copy
 
@@ -211,9 +266,6 @@ func updateAllowedAddressPairs(network *gophercloud.ServiceClient, port *ports.P
 
 // CreateRoute creates the described managed route
 func (r *Routes) CreateRoute(ctx context.Context, clusterName string, nameHint string, route *cloudprovider.Route) error {
-	r.Lock()
-	defer r.Unlock()
-
 	ip, _, _ := net.ParseCIDR(route.DestinationCIDR)
 	isCIDRv6 := ip.To4() == nil
 
@@ -232,31 +284,50 @@ func (r *Routes) CreateRoute(ctx context.Context, clusterName string, nameHint s
 
 	klog.V(4).Infof("Using nexthop %v for node %v", addr, route.TargetNode)
 
-	mc := metrics.NewMetricContext("router", "get")
-	router, err := routers.Get(r.network, r.os.routeOpts.RouterID).Extract()
-	if mc.ObserveRequest(err) != nil {
-		return err
-	}
+	if !r.atomicRoutes {
+		// classical logic
+		r.Lock()
+		defer r.Unlock()
 
-	routes := router.Routes
+		mc := metrics.NewMetricContext("router", "get")
+		router, err := routers.Get(r.network, r.os.routeOpts.RouterID).Extract()
+		if mc.ObserveRequest(err) != nil {
+			return err
+		}
 
-	for _, item := range routes {
-		if item.DestinationCIDR == route.DestinationCIDR && item.NextHop == addr {
-			klog.V(4).Infof("Skipping existing route: %v", route)
-			return nil
+		routes := router.Routes
+
+		for _, item := range routes {
+			if item.DestinationCIDR == route.DestinationCIDR && item.NextHop == addr {
+				klog.V(4).Infof("Skipping existing route: %v", route)
+				return nil
+			}
 		}
-	}
 
-	routes = append(routes, routers.Route{
-		DestinationCIDR: route.DestinationCIDR,
-		NextHop:         addr,
-	})
+		routes = append(routes, routers.Route{
+			DestinationCIDR: route.DestinationCIDR,
+			NextHop:         addr,
+		})
 
-	unwind, err := updateRoutes(r.network, router, routes)
-	if err != nil {
-		return err
+		unwind, err := updateRoutes(r.network, router, routes)
+		if err != nil {
+			return err
+		}
+
+		defer onFailure.call(unwind)
+	} else {
+		// atomic route update
+		route := []routers.Route{{
+			DestinationCIDR: route.DestinationCIDR,
+			NextHop:         addr,
+		}}
+		unwind, err := addRoute(r.network, r.os.routeOpts.RouterID, route)
+		if err != nil {
+			return err
+		}
+
+		defer onFailure.call(unwind)
 	}
-	defer onFailure.call(unwind)
 
 	// get the port of addr on target node.
 	port, err := getPortByIP(r.network, addr, r.networkIDs)
@@ -291,9 +362,6 @@ func (r *Routes) CreateRoute(ctx context.Context, clusterName string, nameHint s
 
 // DeleteRoute deletes the specified managed route
 func (r *Routes) DeleteRoute(ctx context.Context, clusterName string, route *cloudprovider.Route) error {
-	r.Lock()
-	defer r.Unlock()
-
 	klog.V(4).Infof("DeleteRoute(%v, %v)", clusterName, route)
 
 	onFailure := newCaller()
@@ -314,36 +382,57 @@ func (r *Routes) DeleteRoute(ctx context.Context, clusterName string, route *clo
 		}
 	}
 
-	mc := metrics.NewMetricContext("router", "get")
-	router, err := routers.Get(r.network, r.os.routeOpts.RouterID).Extract()
-	if mc.ObserveRequest(err) != nil {
-		return err
-	}
+	if !r.atomicRoutes {
+		// classical logic
+		r.Lock()
+		defer r.Unlock()
 
-	routes := router.Routes
-	index := -1
-	for i, item := range routes {
-		if item.DestinationCIDR == route.DestinationCIDR && (item.NextHop == addr || route.Blackhole && item.NextHop == string(route.TargetNode)) {
-			index = i
-			break
+		mc := metrics.NewMetricContext("router", "get")
+		router, err := routers.Get(r.network, r.os.routeOpts.RouterID).Extract()
+		if mc.ObserveRequest(err) != nil {
+			return err
 		}
-	}
 
-	if index == -1 {
-		klog.V(4).Infof("Skipping non-existent route: %v", route)
-		return nil
-	}
+		routes := router.Routes
+		index := -1
+		for i, item := range routes {
+			if item.DestinationCIDR == route.DestinationCIDR && (item.NextHop == addr || route.Blackhole && item.NextHop == string(route.TargetNode)) {
+				index = i
+				break
+			}
+		}
 
-	// Delete element `index`
-	routes[index] = routes[len(routes)-1]
-	routes = routes[:len(routes)-1]
+		if index == -1 {
+			klog.V(4).Infof("Skipping non-existent route: %v", route)
+			return nil
+		}
 
-	unwind, err := updateRoutes(r.network, router, routes)
-	// If this was a blackhole route we are done, there are no ports to update
-	if err != nil || route.Blackhole {
-		return err
+		// Delete element `index`
+		routes[index] = routes[len(routes)-1]
+		routes = routes[:len(routes)-1]
+
+		unwind, err := updateRoutes(r.network, router, routes)
+		// If this was a blackhole route we are done, there are no ports to update
+		if err != nil || route.Blackhole {
+			return err
+		}
+
+		defer onFailure.call(unwind)
+	} else {
+		// atomic route update
+		blackhole := route.Blackhole
+		route := []routers.Route{{
+			DestinationCIDR: route.DestinationCIDR,
+			NextHop:         addr,
+		}}
+		unwind, err := removeRoute(r.network, r.os.routeOpts.RouterID, route)
+		// If this was a blackhole route we are done, there are no ports to update
+		if err != nil || blackhole {
+			return err
+		}
+
+		defer onFailure.call(unwind)
 	}
-	defer onFailure.call(unwind)
 
 	// get the port of addr on target node.
 	port, err := getPortByIP(r.network, addr, r.networkIDs)
@@ -352,7 +441,7 @@ func (r *Routes) DeleteRoute(ctx context.Context, clusterName string, route *clo
 	}
 
 	addrPairs := port.AllowedAddressPairs
-	index = -1
+	index := -1
 	for i, item := range addrPairs {
 		if item.IPAddress == route.DestinationCIDR {
 			index = i