Handle updates to the node OVN transit switch IP

...used for NonGatewayRoutes. This addresses a TODO.

Added a TransitSwitchIP interface and implementation that encapsulates
the transit switch IP extracted from the local node's annotation value.
The instance is shared between the NonGatewayRouteController and the
NonGatewayRouteHandler. The NonGatewayRouteHandler watches for node
updates and updates the TransitSwitchIP value and the
NonGatewayRoutes NextHops field. The NonGatewayRouteController is a
read-only consumer of TransitSwitchIP.

Signed-off-by: Tom Pantelis <[email protected]>

Author: tpantelis

pkg/routeagent_driver/handlers/ovn/handler.go

@@ -43,14 +43,15 @@ import (
 type NewOVSDBClientFn func(_ model.ClientDBModel, _ ...libovsdbclient.Option) (libovsdbclient.Client, error)
 
 type HandlerConfig struct {
-	Namespace      string
-	ClusterCIDR    []string
-	ServiceCIDR    []string
-	SubmClient     clientset.Interface
-	K8sClient      kubernetes.Interface
-	DynClient      dynamic.Interface
-	WatcherConfig  *watcher.Config
-	NewOVSDBClient NewOVSDBClientFn
+	Namespace       string
+	ClusterCIDR     []string
+	ServiceCIDR     []string
+	SubmClient      clientset.Interface
+	K8sClient       kubernetes.Interface
+	DynClient       dynamic.Interface
+	WatcherConfig   *watcher.Config
+	NewOVSDBClient  NewOVSDBClientFn
+	TransitSwitchIP TransitSwitchIPGetter
 }
 
 type Handler struct {
@@ -124,7 +125,7 @@ func (ovn *Handler) Init() error {
 		return err
 	}
 
-	nonGatewayRouteController, err := NewNonGatewayRouteController(*ovn.WatcherConfig, connectionHandler, ovn.K8sClient, ovn.Namespace)
+	nonGatewayRouteController, err := NewNonGatewayRouteController(*ovn.WatcherConfig, connectionHandler, ovn.Namespace, ovn.TransitSwitchIP)
 	if err != nil {
 		return err
 	}

pkg/routeagent_driver/handlers/ovn/handler_test.go

@@ -51,7 +51,10 @@ const (
 var _ = Describe("Handler", func() {
 	t := newTestDriver()
 
-	var ovsdbClient *fakeovn.OVSDBClient
+	var (
+		ovsdbClient     *fakeovn.OVSDBClient
+		transitSwitchIP ovn.TransitSwitchIP
+	)
 
 	BeforeEach(func() {
 		ovsdbClient = fakeovn.NewOVSDBClient()
@@ -79,6 +82,8 @@ var _ = Describe("Handler", func() {
 
 		restMapper := test.GetRESTMapperFor(&submarinerv1.GatewayRoute{}, &submarinerv1.NonGatewayRoute{})
 
+		transitSwitchIP = ovn.NewTransitSwitchIP()
+
 		t.Start(ovn.NewHandler(&ovn.HandlerConfig{
 			Namespace:   testing.Namespace,
 			ClusterCIDR: []string{clusterCIDR},
@@ -93,9 +98,11 @@ var _ = Describe("Handler", func() {
 			NewOVSDBClient: func(_ model.ClientDBModel, _ ...libovsdbclient.Option) (libovsdbclient.Client, error) {
 				return ovsdbClient, nil
 			},
+			TransitSwitchIP: transitSwitchIP,
 		}))
 
 		Expect(ovsdbClient.Connected()).To(BeTrue())
+		Expect(transitSwitchIP.Init(t.k8sClient)).To(Succeed())
 	})
 
 	When("a remote Endpoint is created, updated, and deleted", func() {
@@ -254,37 +261,89 @@ var _ = Describe("Handler", func() {
 		})
 	})
 
-	When("a NonGatewayRoute is created and deleted", func() {
+	When("NonGatewayRoutes are created, updated and deleted", func() {
+		verifyLogicalRouterPolicies := func(ngr *submarinerv1.NonGatewayRoute, nextHop string) {
+			for _, cidr := range ngr.RoutePolicySpec.RemoteCIDRs {
+				ovsdbClient.AwaitModel(&nbdb.LogicalRouterPolicy{
+					Match:   cidr,
+					Nexthop: ptr.To(nextHop),
+				})
+			}
+		}
+
+		verifyNoLogicalRouterPolicies := func(ngr *submarinerv1.NonGatewayRoute, nextHop string) {
+			for _, cidr := range ngr.RoutePolicySpec.RemoteCIDRs {
+				ovsdbClient.AwaitNoModel(&nbdb.LogicalRouterPolicy{
+					Match:   cidr,
+					Nexthop: ptr.To(nextHop),
+				})
+			}
+		}
+
 		It("should correctly reconcile OVN router policies", func() {
 			client := t.dynClient.Resource(submarinerv1.SchemeGroupVersion.WithResource("nongatewayroutes")).Namespace(testing.Namespace)
 
-			nonGWRoute := &submarinerv1.NonGatewayRoute{
+			By("Creating first NonGatewayRoute")
+
+			nextHop := "172.1.1.1"
+
+			nonGWRoute1 := &submarinerv1.NonGatewayRoute{
 				ObjectMeta: metav1.ObjectMeta{
-					Name: "test-nongateway-route",
+					Name: "test-nongateway-route1",
 				},
 				RoutePolicySpec: submarinerv1.RoutePolicySpec{
-					NextHops:    []string{"111.1.1.1"},
-					RemoteCIDRs: []string{"192.0.1.0/24", "192.0.2.0/24"},
+					NextHops:    []string{nextHop},
+					RemoteCIDRs: []string{"111.0.1.0/24", "111.0.2.0/24"},
 				},
 			}
 
-			test.CreateResource(client, nonGWRoute)
+			test.CreateResource(client, nonGWRoute1)
 
-			for _, cidr := range nonGWRoute.RoutePolicySpec.RemoteCIDRs {
-				ovsdbClient.AwaitModel(&nbdb.LogicalRouterPolicy{
-					Match:   cidr,
-					Nexthop: ptr.To(nonGWRoute.RoutePolicySpec.NextHops[0]),
-				})
-			}
+			verifyLogicalRouterPolicies(nonGWRoute1, nextHop)
 
-			Expect(client.Delete(context.Background(), nonGWRoute.Name, metav1.DeleteOptions{})).To(Succeed())
+			By("Creating second NonGatewayRoute")
 
-			for _, cidr := range nonGWRoute.RoutePolicySpec.RemoteCIDRs {
-				ovsdbClient.AwaitNoModel(&nbdb.LogicalRouterPolicy{
-					Match:   cidr,
-					Nexthop: ptr.To(nonGWRoute.RoutePolicySpec.NextHops[0]),
-				})
+			nonGWRoute2 := &submarinerv1.NonGatewayRoute{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "test-nongateway-route2",
+				},
+				RoutePolicySpec: submarinerv1.RoutePolicySpec{
+					NextHops:    []string{nextHop},
+					RemoteCIDRs: []string{"222.0.1.0/24", "222.0.2.0/24"},
+				},
 			}
+
+			test.CreateResource(client, nonGWRoute2)
+
+			verifyLogicalRouterPolicies(nonGWRoute1, nextHop)
+			verifyLogicalRouterPolicies(nonGWRoute2, nextHop)
+
+			By("Updating NextHop for first NonGatewayRoute")
+
+			prevNextHop := nextHop
+			nextHop = "172.1.1.2"
+			nonGWRoute1.RoutePolicySpec.NextHops[0] = nextHop
+
+			test.UpdateResource(client, nonGWRoute1)
+
+			verifyLogicalRouterPolicies(nonGWRoute1, nextHop)
+			verifyNoLogicalRouterPolicies(nonGWRoute1, prevNextHop)
+			verifyNoLogicalRouterPolicies(nonGWRoute2, prevNextHop)
+
+			By("Updating NextHop for second NonGatewayRoute")
+
+			nonGWRoute2.RoutePolicySpec.NextHops[0] = nextHop
+
+			test.UpdateResource(client, nonGWRoute2)
+
+			verifyLogicalRouterPolicies(nonGWRoute1, nextHop)
+			verifyLogicalRouterPolicies(nonGWRoute2, nextHop)
+
+			By("Deleting first NonGatewayRoute")
+
+			Expect(client.Delete(context.Background(), nonGWRoute1.Name, metav1.DeleteOptions{})).To(Succeed())
+
+			verifyNoLogicalRouterPolicies(nonGWRoute1, nextHop)
 		})
 	})
 

pkg/routeagent_driver/handlers/ovn/non_gateway_route_controller.go

@@ -22,23 +22,21 @@ import (
 	"github.com/pkg/errors"
 	"github.com/submariner-io/admiral/pkg/watcher"
 	submarinerv1 "github.com/submariner-io/submariner/pkg/apis/submariner.io/v1"
-	nodeutil "github.com/submariner-io/submariner/pkg/node"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/sets"
-	clientset "k8s.io/client-go/kubernetes"
 )
 
 type NonGatewayRouteController struct {
 	nonGatewayRouteWatcher watcher.Interface
 	connectionHandler      *ConnectionHandler
 	remoteSubnets          sets.Set[string]
 	stopCh                 chan struct{}
-	transitSwitchIP        string
+	transitSwitchIP        TransitSwitchIPGetter
 }
 
 //nolint:gocritic // Ignore hugeParam
 func NewNonGatewayRouteController(config watcher.Config, connectionHandler *ConnectionHandler,
-	k8sClientSet clientset.Interface, namespace string,
+	namespace string, transitSwitchIP TransitSwitchIPGetter,
 ) (*NonGatewayRouteController, error) {
 	// We'll panic if config is nil, this is intentional
 	var err error
@@ -47,6 +45,7 @@ func NewNonGatewayRouteController(config watcher.Config, connectionHandler *Conn
 		connectionHandler: connectionHandler,
 		remoteSubnets:     sets.New[string](),
 		stopCh:            make(chan struct{}),
+		transitSwitchIP:   transitSwitchIP,
 	}
 
 	config.ResourceConfigs = []watcher.ResourceConfig{
@@ -62,25 +61,6 @@ func NewNonGatewayRouteController(config watcher.Config, connectionHandler *Conn
 		},
 	}
 
-	node, err := nodeutil.GetLocalNode(k8sClientSet)
-	if err != nil {
-		return nil, errors.Wrap(err, "error getting the local node info")
-	}
-
-	annotations := node.GetAnnotations()
-
-	transitSwitchIP := annotations["k8s.ovn.org/node-transit-switch-port-ifaddr"]
-	if transitSwitchIP == "" {
-		// This is a non-IC setup , so this controller will not be started.
-		logger.Infof("No transit switch IP configured on node %q", node.Name)
-		return controller, nil
-	}
-
-	controller.transitSwitchIP, err = jsonToIP(transitSwitchIP)
-	if err != nil {
-		return nil, errors.Wrapf(err, "error parsing transit switch IP")
-	}
-
 	controller.nonGatewayRouteWatcher, err = watcher.New(&config)
 	if err != nil {
 		return nil, errors.Wrap(err, "error creating resource watcher")
@@ -129,7 +109,7 @@ func (g *NonGatewayRouteController) reconcileRemoteSubnets(submNonGWRoute *subma
 	}
 
 	// If this node belongs to same zone as gateway node, ignore the event.
-	if submNonGWRoute.RoutePolicySpec.NextHops[0] != g.transitSwitchIP {
+	if submNonGWRoute.RoutePolicySpec.NextHops[0] != g.transitSwitchIP.Get() {
 		for _, subnet := range submNonGWRoute.RoutePolicySpec.RemoteCIDRs {
 			if addSubnet {
 				g.remoteSubnets.Insert(subnet)
@@ -145,7 +125,5 @@ func (g *NonGatewayRouteController) reconcileRemoteSubnets(submNonGWRoute *subma
 }
 
 func (g *NonGatewayRouteController) stop() {
-	if g.transitSwitchIP != "" {
-		close(g.stopCh)
-	}
+	close(g.stopCh)
 }

pkg/routeagent_driver/handlers/ovn/non_gateway_route_handler.go

@@ -28,47 +28,32 @@ import (
 	submarinerClientset "github.com/submariner-io/submariner/pkg/client/clientset/versioned"
 	"github.com/submariner-io/submariner/pkg/cni"
 	"github.com/submariner-io/submariner/pkg/event"
-	nodeutil "github.com/submariner-io/submariner/pkg/node"
-	"github.com/submariner-io/submariner/pkg/routeagent_driver/constants"
+	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	clientset "k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes"
 )
 
 type NonGatewayRouteHandler struct {
 	event.HandlerBase
+	event.NodeHandlerBase
 	smClient        submarinerClientset.Interface
-	k8sClient       clientset.Interface
-	transitSwitchIP string
+	k8sClient       kubernetes.Interface
+	transitSwitchIP TransitSwitchIP
 }
 
-func NewNonGatewayRouteHandler(smClient submarinerClientset.Interface, k8sClient clientset.Interface) *NonGatewayRouteHandler {
+func NewNonGatewayRouteHandler(smClient submarinerClientset.Interface, k8sClient kubernetes.Interface, transitSwitchIP TransitSwitchIP,
+) *NonGatewayRouteHandler {
 	return &NonGatewayRouteHandler{
-		smClient:  smClient,
-		k8sClient: k8sClient,
+		smClient:        smClient,
+		k8sClient:       k8sClient,
+		transitSwitchIP: transitSwitchIP,
 	}
 }
 
 func (h *NonGatewayRouteHandler) Init() error {
 	logger.Info("Starting NonGatewayRouteHandler")
-
-	node, err := nodeutil.GetLocalNode(h.k8sClient)
-	if err != nil {
-		return errors.Wrap(err, "error getting the g/w node")
-	}
-
-	annotations := node.GetAnnotations()
-
-	// TODO transitSwitchIP changes support needs to be added.
-	transitSwitchIP, ok := annotations[constants.OvnTransitSwitchIPAnnotation]
-	if !ok {
-		logger.Infof("No transit switch IP configured")
-		return nil
-	}
-
-	h.transitSwitchIP, err = jsonToIP(transitSwitchIP)
-
-	return errors.Wrapf(err, "error parsing the transit switch IP")
+	return errors.Wrap(h.transitSwitchIP.Init(h.k8sClient), "error initializing TransitSwitchIP")
 }
 
 func (h *NonGatewayRouteHandler) GetName() string {
@@ -80,7 +65,7 @@ func (h *NonGatewayRouteHandler) GetNetworkPlugins() []string {
 }
 
 func (h *NonGatewayRouteHandler) RemoteEndpointCreated(endpoint *submarinerv1.Endpoint) error {
-	if !h.State().IsOnGateway() || h.transitSwitchIP == "" {
+	if !h.State().IsOnGateway() || h.transitSwitchIP.Get() == "" {
 		return nil
 	}
 
@@ -98,7 +83,7 @@ func (h *NonGatewayRouteHandler) RemoteEndpointCreated(endpoint *submarinerv1.En
 }
 
 func (h *NonGatewayRouteHandler) RemoteEndpointRemoved(endpoint *submarinerv1.Endpoint) error {
-	if !h.State().IsOnGateway() || h.transitSwitchIP == "" {
+	if !h.State().IsOnGateway() || h.transitSwitchIP.Get() == "" {
 		return nil
 	}
 
@@ -113,7 +98,7 @@ func (h *NonGatewayRouteHandler) RemoteEndpointRemoved(endpoint *submarinerv1.En
 }
 
 func (h *NonGatewayRouteHandler) TransitionToGateway() error {
-	if h.transitSwitchIP == "" {
+	if h.transitSwitchIP.Get() == "" {
 		return nil
 	}
 
@@ -141,6 +126,38 @@ func (h *NonGatewayRouteHandler) TransitionToGateway() error {
 	return nil
 }
 
+func (h *NonGatewayRouteHandler) NodeUpdated(node *corev1.Node) error {
+	updated, err := h.transitSwitchIP.UpdateFrom(node)
+	if err != nil {
+		logger.Errorf(err, "Error updating transit switch IP from node: %s", resource.ToJSON(node))
+		return nil
+	}
+
+	if !updated {
+		return nil
+	}
+
+	logger.Infof("Transit switch IP updated to %s", h.transitSwitchIP.Get())
+
+	if !h.State().IsOnGateway() {
+		return nil
+	}
+
+	endpoints := h.State().GetRemoteEndpoints()
+	for i := range endpoints {
+		err = util.Update(context.TODO(), NonGatewayResourceInterface(h.smClient, endpoints[i].Namespace),
+			h.newNonGatewayRoute(&endpoints[i]), func(existing *submarinerv1.NonGatewayRoute) (*submarinerv1.NonGatewayRoute, error) {
+				existing.RoutePolicySpec.NextHops = []string{h.transitSwitchIP.Get()}
+				return existing, nil
+			})
+		if err != nil {
+			return errors.Wrapf(err, "error updating NonGatewayRoute")
+		}
+	}
+
+	return nil
+}
+
 func (h *NonGatewayRouteHandler) newNonGatewayRoute(endpoint *submarinerv1.Endpoint) *submarinerv1.NonGatewayRoute {
 	return &submarinerv1.NonGatewayRoute{
 		ObjectMeta: metav1.ObjectMeta{
@@ -149,7 +166,7 @@ func (h *NonGatewayRouteHandler) newNonGatewayRoute(endpoint *submarinerv1.Endpo
 		},
 		RoutePolicySpec: submarinerv1.RoutePolicySpec{
 			RemoteCIDRs: endpoint.Spec.Subnets,
-			NextHops:    []string{h.transitSwitchIP},
+			NextHops:    []string{h.transitSwitchIP.Get()},
 		},
 	}
 }