Use HandlerState in the OVN Handler

The mutex was kept to provide atomicity when updating the gateway and
host network dataplane since these are also invoked via the route config
syncer which runs on its own thread. The synchronization may actually not
be necessary but it's better to be safe and the overhead in negligible.

Signed-off-by: Tom Pantelis <tompantelis@gmail.com>

pkg/routeagent_driver/handlers/ovn/gateway_dataplane.go

@@ -53,6 +53,9 @@ func (ovn *Handler) cleanupGatewayDataplane() error {
 }
 
 func (ovn *Handler) updateGatewayDataplane() error {
+	ovn.mutex.Lock()
+	defer ovn.mutex.Unlock()
+
 	currentRuleRemotes, err := ovn.getExistingIPv4RuleSubnets()
 	if err != nil {
 		return errors.Wrapf(err, "error reading ip rule list for IPv4")

pkg/routeagent_driver/handlers/ovn/handler.go

@@ -58,8 +58,6 @@ type Handler struct {
 	HandlerConfig
 	mutex                     sync.Mutex
 	cableRoutingInterface     *net.Interface
-	remoteEndpoints           map[string]*submV1.Endpoint
-	isGateway                 bool
 	netLink                   netlink.Interface
 	ipt                       iptables.Interface
 	gatewayRouteController    *GatewayRouteController
@@ -77,11 +75,10 @@ func NewHandler(config *HandlerConfig) *Handler {
 	}
 
 	h := &Handler{
-		HandlerConfig:   *config,
-		remoteEndpoints: map[string]*submV1.Endpoint{},
-		netLink:         netlink.New(),
-		ipt:             ipt,
-		stopCh:          make(chan struct{}),
+		HandlerConfig: *config,
+		netLink:       netlink.New(),
+		ipt:           ipt,
+		stopCh:        make(chan struct{}),
 	}
 
 	if h.NewOVSDBClient == nil {
@@ -153,9 +150,6 @@ func (ovn *Handler) LocalEndpointCreated(endpoint *submV1.Endpoint) error {
 		}
 	}
 
-	ovn.mutex.Lock()
-	defer ovn.mutex.Unlock()
-
 	ovn.cableRoutingInterface = routingInterface
 
 	return nil
@@ -169,17 +163,12 @@ func (ovn *Handler) RemoteEndpointCreated(endpoint *submV1.Endpoint) error {
 		return nil
 	}
 
-	ovn.mutex.Lock()
-	defer ovn.mutex.Unlock()
-
-	ovn.remoteEndpoints[endpoint.Name] = endpoint
-
 	err := ovn.updateHostNetworkDataplane()
 	if err != nil {
 		return errors.Wrapf(err, "updateHostNetworkDataplane returned error")
 	}
 
-	if ovn.isGateway {
+	if ovn.State().IsOnGateway() {
 		for _, subnet := range endpoint.Spec.Subnets {
 			if err = ovn.addNoMasqueradeIPTables(subnet); err != nil {
 				return errors.Wrapf(err, "error adding no-masquerade rules for subnet %q", subnet)
@@ -199,35 +188,25 @@ func (ovn *Handler) RemoteEndpointUpdated(endpoint *submV1.Endpoint) error {
 		return nil
 	}
 
-	ovn.mutex.Lock()
-	defer ovn.mutex.Unlock()
-
-	ovn.remoteEndpoints[endpoint.Name] = endpoint
-
 	err := ovn.updateHostNetworkDataplane()
 	if err != nil {
 		return errors.Wrapf(err, "updateHostNetworkDataplane returned error")
 	}
 
-	if ovn.isGateway {
+	if ovn.State().IsOnGateway() {
 		return ovn.updateGatewayDataplane()
 	}
 
 	return nil
 }
 
 func (ovn *Handler) RemoteEndpointRemoved(endpoint *submV1.Endpoint) error {
-	ovn.mutex.Lock()
-	defer ovn.mutex.Unlock()
-
-	delete(ovn.remoteEndpoints, endpoint.Name)
-
 	err := ovn.updateHostNetworkDataplane()
 	if err != nil {
 		return errors.Wrapf(err, "updateHostNetworkDataplane returned error")
 	}
 
-	if ovn.isGateway {
+	if ovn.State().IsOnGateway() {
 		for _, subnet := range endpoint.Spec.Subnets {
 			if err = ovn.removeNoMasqueradeIPTables(subnet); err != nil {
 				return errors.Wrapf(err, "error removing no-masquerade rules for subnet %q", subnet)
@@ -241,12 +220,9 @@ func (ovn *Handler) RemoteEndpointRemoved(endpoint *submV1.Endpoint) error {
 }
 
 func (ovn *Handler) TransitionToNonGateway() error {
-	ovn.mutex.Lock()
-	defer ovn.mutex.Unlock()
-
-	ovn.isGateway = false
-	for _, endpoint := range ovn.remoteEndpoints {
-		for _, subnet := range endpoint.Spec.Subnets {
+	endpoints := ovn.State().GetRemoteEndpoints()
+	for i := range endpoints {
+		for _, subnet := range endpoints[i].Spec.Subnets {
 			if err := ovn.removeNoMasqueradeIPTables(subnet); err != nil {
 				return errors.Wrapf(err, "error removing no-masquerade rules for subnet %q", subnet)
 			}
@@ -257,13 +233,9 @@ func (ovn *Handler) TransitionToNonGateway() error {
 }
 
 func (ovn *Handler) TransitionToGateway() error {
-	ovn.mutex.Lock()
-	defer ovn.mutex.Unlock()
-
-	ovn.isGateway = true
-
-	for _, endpoint := range ovn.remoteEndpoints {
-		for _, subnet := range endpoint.Spec.Subnets {
+	endpoints := ovn.State().GetRemoteEndpoints()
+	for i := range endpoints {
+		for _, subnet := range endpoints[i].Spec.Subnets {
 			if err := ovn.addNoMasqueradeIPTables(subnet); err != nil {
 				return errors.Wrapf(err, "error adding no-masquerade rules for subnet %q", subnet)
 			}

pkg/routeagent_driver/handlers/ovn/host_networking.go

@@ -36,6 +36,9 @@ const (
 )
 
 func (ovn *Handler) updateHostNetworkDataplane() error {
+	ovn.mutex.Lock()
+	defer ovn.mutex.Unlock()
+
 	currentRuleRemotes, err := ovn.getExistingIPv4HostNetworkRoutes()
 	if err != nil {
 		return errors.Wrapf(err, "error reading ip rule list for IPv4")

pkg/routeagent_driver/handlers/ovn/route_config_syncer.go

@@ -131,9 +131,7 @@ func (ovn *Handler) handleInterfaceAddressChange() error {
 		logger.Infof("Waiting for interface %q to be ready: %v", OVNK8sMgmntIntfName, err)
 		return true
 	}, func() error {
-		ovn.mutex.Lock()
-		defer ovn.mutex.Unlock()
-		if ovn.isGateway {
+		if ovn.State().IsOnGateway() {
 			err = ovn.updateGatewayDataplane()
 			if err != nil {
 				return errors.Wrap(err, "error syncing gateway routes")

pkg/routeagent_driver/handlers/ovn/subnets.go

@@ -23,8 +23,9 @@ import "k8s.io/utils/set"
 func (ovn *Handler) getRemoteSubnets() set.Set[string] {
 	endpointSubnets := set.New[string]()
 
-	for _, endpoint := range ovn.remoteEndpoints {
-		for _, subnet := range endpoint.Spec.Subnets {
+	endpoints := ovn.State().GetRemoteEndpoints()
+	for i := range endpoints {
+		for _, subnet := range endpoints[i].Spec.Subnets {
 			endpointSubnets.Insert(subnet)
 		}
 	}