[refactor] split routeWatcher into watcher and handler components

calico-vpp-agent/cmd/calico_vpp_dataplane.go

@@ -152,6 +152,8 @@ func main() {
 	cniServer := watchers.NewCNIServer(felixServer.GetFelixServerEventChan(), log.WithFields(logrus.Fields{"component": "cni"}))
 	serviceServer := watchers.NewServiceServer(felixServer.GetFelixServerEventChan(), k8sclient, log.WithFields(logrus.Fields{"component": "services"}))
 
+	felixServer.GetRouteHandler().SetRouteWatcher(routeWatcher)
+
 	err = watchers.InstallFelixPlugin()
 	if err != nil {
 		log.Fatalf("could not install felix plugin: %s", err)

calico-vpp-agent/common/pubsub.go

@@ -27,7 +27,6 @@ const (
 	ChanSize = 500
 
 	PeerNodeStateChanged CalicoVppEventType = "PeerNodeStateChanged"
-	IpamConfChanged      CalicoVppEventType = "IpamConfChanged"
 	BGPConfChanged       CalicoVppEventType = "BGPConfChanged"
 
 	ConnectivityAdded   CalicoVppEventType = "ConnectivityAdded"
@@ -61,6 +60,9 @@ const (
 
 	NetAddedOrUpdated CalicoVppEventType = "NetAddedOrUpdated"
 	NetDeleted        CalicoVppEventType = "NetDeleted"
+
+	IpamPoolUpdate CalicoVppEventType = "IpamPoolUpdate"
+	IpamPoolRemove CalicoVppEventType = "IpamPoolRemove"
 )
 
 var (

calico-vpp-agent/felix/felix_server.go

@@ -50,6 +50,7 @@ type Server struct {
 	cniHandler          *cni.CNIHandler
 	connectivityHandler *connectivity.ConnectivityHandler
 	serviceHandler      *services.ServiceHandler
+	routeHandler        *RouteHandler
 
 	prometheusServer *prometheus.PrometheusServer
 }
@@ -68,6 +69,7 @@ func NewFelixServer(vpp *vpplink.VppLink, clientv3 calicov3cli.Interface, log *l
 		cniHandler:          cni.NewCNIHandler(vpp, cache, log),
 		connectivityHandler: connectivity.NewConnectivityHandler(vpp, cache, clientv3, log.WithFields(logrus.Fields{"component": "connectivity"})),
 		serviceHandler:      services.NewServiceHandler(vpp, cache, log),
+		routeHandler:        NewRouteHandler(log),
 
 		prometheusServer: prometheus.NewPrometheusServer(vpp, log.WithFields(logrus.Fields{"component": "prometheus"})),
 	}
@@ -89,6 +91,10 @@ func NewFelixServer(vpp *vpplink.VppLink, clientv3 calicov3cli.Interface, log *l
 	return server
 }
 
+func (s *Server) GetRouteHandler() *RouteHandler {
+	return s.routeHandler
+}
+
 func (s *Server) GetFelixServerEventChan() chan any {
 	return s.felixServerEventChan
 }
@@ -322,6 +328,12 @@ func (s *Server) handleFelixServerEvents(msg interface{}) (err error) {
 			s.cache.NetworkDefinitions[new.Name] = new
 			s.cache.Networks[new.Vni] = new
 			s.cniHandler.OnNetAddedOrUpdated(old, new)
+			if s.routeHandler != nil {
+				err := s.routeHandler.OnNetAddedOrUpdated(new)
+				if err != nil {
+					s.log.Errorf("Failed to handle network update in RouteHandler: %v", err)
+				}
+			}
 		case common.NetDeleted:
 			netDef, ok := evt.Old.(*common.NetworkDefinition)
 			if !ok {
@@ -330,6 +342,12 @@ func (s *Server) handleFelixServerEvents(msg interface{}) (err error) {
 			delete(s.cache.NetworkDefinitions, netDef.Name)
 			delete(s.cache.Networks, netDef.Vni)
 			s.cniHandler.OnNetDeleted(netDef)
+			if s.routeHandler != nil {
+				err := s.routeHandler.OnNetDeleted(netDef)
+				if err != nil {
+					s.log.Errorf("Failed to handle network deletion in RouteHandler: %v", err)
+				}
+			}
 		case common.PodAdded:
 			podSpec, ok := evt.New.(*model.LocalPodSpec)
 			if !ok {

calico-vpp-agent/felix/felix_server_test.go

@@ -80,6 +80,17 @@ var _ = Describe("Felix functionality", func() {
 		// add interface to mock the tap0 because felix server needs it
 		CreateLoopbackAndTaggingItAsMain(vpp, log)
 		common.ThePubSub = common.NewPubSub(log.WithFields(logrus.Fields{"component": "pubsub"}))
+		common.VppManagerInfo = &config.VppManagerInfo{
+			UplinkStatuses: map[string]config.UplinkStatus{
+				"uplink": {
+					SwIfIndex:      1,
+					TapSwIfIndex:   1,
+					IsMain:         true,
+					FakeNextHopIP4: net.ParseIP("169.254.1.1"),
+					FakeNextHopIP6: net.ParseIP("fd00::1"),
+				},
+			},
+		}
 		felixServer = NewFelixServer(vpp, nil, log.WithFields(logrus.Fields{"component": "policy"}))
 		policiesHandler = felixServer.policiesHandler
 		policiesHandler.OnFelixSocketStateChanged(&common.FelixSocketStateChanged{NewState: common.StateInSync})

calico-vpp-agent/felix/ipam.go

@@ -21,8 +21,6 @@ import (
 	"github.com/pkg/errors"
 
 	"github.com/projectcalico/calico/felix/proto"
-
-	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/common"
 )
 
 func (s *Server) handleIpamPoolUpdate(msg *proto.IPAMPoolUpdate) (err error) {
@@ -49,11 +47,12 @@ func (s *Server) handleIpamPoolUpdate(msg *proto.IPAMPoolUpdate) (err error) {
 			}
 			s.connectivityHandler.OnIpamConfChanged(oldIpamPool, newIpamPool)
 			s.cniHandler.OnIpamConfChanged(oldIpamPool, newIpamPool)
-			common.SendEvent(common.CalicoVppEvent{
-				Type: common.IpamConfChanged,
-				Old:  ipamPoolCopy(oldIpamPool),
-				New:  ipamPoolCopy(newIpamPool),
-			})
+			if s.routeHandler != nil {
+				err := s.routeHandler.OnIpamConfChanged(oldIpamPool, newIpamPool)
+				if err != nil {
+					s.log.Errorf("Failed to handle IPAM update in RouteHandler: %v", err)
+				}
+			}
 		}
 	} else {
 		s.log.Infof("Adding pool: %s, nat:%t", msg.GetId(), newIpamPool.GetMasquerade())
@@ -65,10 +64,12 @@ func (s *Server) handleIpamPoolUpdate(msg *proto.IPAMPoolUpdate) (err error) {
 		}
 		s.connectivityHandler.OnIpamConfChanged(nil /*old*/, newIpamPool)
 		s.cniHandler.OnIpamConfChanged(nil /*old*/, newIpamPool)
-		common.SendEvent(common.CalicoVppEvent{
-			Type: common.IpamConfChanged,
-			New:  ipamPoolCopy(newIpamPool),
-		})
+		if s.routeHandler != nil {
+			err := s.routeHandler.OnIpamConfChanged(nil, newIpamPool)
+			if err != nil {
+				s.log.Errorf("Failed to handle IPAM addition in RouteHandler: %v", err)
+			}
+		}
 	}
 	return nil
 }
@@ -88,32 +89,21 @@ func (s *Server) handleIpamPoolRemove(msg *proto.IPAMPoolRemove) (err error) {
 		if err != nil {
 			return errors.Wrap(err, "error handling ipam deletion")
 		}
-		common.SendEvent(common.CalicoVppEvent{
-			Type: common.IpamConfChanged,
-			Old:  ipamPoolCopy(oldIpamPool),
-			New:  nil,
-		})
 		s.connectivityHandler.OnIpamConfChanged(oldIpamPool, nil /* new */)
 		s.cniHandler.OnIpamConfChanged(oldIpamPool, nil /* new */)
+		if s.routeHandler != nil {
+			err := s.routeHandler.OnIpamConfChanged(oldIpamPool, nil)
+			if err != nil {
+				s.log.Errorf("Failed to handle IPAM deletion in RouteHandler: %v", err)
+			}
+		}
 	} else {
 		s.log.Warnf("Deleting unknown ippool")
 		return nil
 	}
 	return nil
 }
 
-func ipamPoolCopy(ipamPool *proto.IPAMPool) *proto.IPAMPool {
-	if ipamPool != nil {
-		return &proto.IPAMPool{
-			Cidr:       ipamPool.Cidr,
-			Masquerade: ipamPool.Masquerade,
-			IpipMode:   ipamPool.IpipMode,
-			VxlanMode:  ipamPool.VxlanMode,
-		}
-	}
-	return nil
-}
-
 // Compare only the fields that make a difference for this agent i.e. the fields that have an impact on routing
 func ipamPoolEquals(a *proto.IPAMPool, b *proto.IPAMPool) bool {
 	if (a == nil || b == nil) && a != b {

calico-vpp-agent/felix/route_handler.go

@@ -0,0 +1,149 @@
+// Copyright (C) 2025 Cisco Systems Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package felix
+
+import (
+	"net"
+	"syscall"
+
+	"github.com/pkg/errors"
+	"github.com/projectcalico/calico/felix/proto"
+	"github.com/sirupsen/logrus"
+	"github.com/vishvananda/netlink"
+
+	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/common"
+	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/watchers"
+)
+
+// RouteHandler handles network and IPAM events by updating VPP routing configuration
+type RouteHandler struct {
+	log          *logrus.Entry
+	routeWatcher *watchers.RouteWatcher
+}
+
+// NewRouteHandler creates a new RouteHandler instance
+func NewRouteHandler(log *logrus.Entry) *RouteHandler {
+	return &RouteHandler{
+		log:          log,
+		routeWatcher: nil,
+	}
+}
+
+// SetRouteWatcher sets the route watcher for performing route operations
+func (h *RouteHandler) SetRouteWatcher(routeWatcher *watchers.RouteWatcher) {
+	h.routeWatcher = routeWatcher
+}
+
+// OnNetDeleted handles network deletion events
+func (h *RouteHandler) OnNetDeleted(netDef *common.NetworkDefinition) error {
+	key := netDef.Range
+	routes, err := h.getNetworkRoute(key, netDef.PhysicalNetworkName)
+	if err != nil {
+		h.log.Errorf("Error getting route from network deletion: %v", err)
+		return err
+	}
+	for _, route := range routes {
+		err = h.routeWatcher.DelRoute(route)
+		if err != nil {
+			h.log.Errorf("Cannot delete pool route %s through vpp tap: %v", key, err)
+			return err
+		}
+	}
+	return nil
+}
+
+// OnNetAddedOrUpdated handles network addition/update events
+func (h *RouteHandler) OnNetAddedOrUpdated(netDef *common.NetworkDefinition) error {
+	key := netDef.Range
+	routes, err := h.getNetworkRoute(key, netDef.PhysicalNetworkName)
+	if err != nil {
+		h.log.Errorf("Error getting route from network addition/update: %v", err)
+		return err
+	}
+	for _, route := range routes {
+		err = h.routeWatcher.AddRoute(route)
+		if err != nil {
+			h.log.Errorf("Cannot add pool route %s through vpp tap: %v", key, err)
+			return err
+		}
+	}
+	return nil
+}
+
+// OnIpamConfChanged handles IPAM configuration changes
+func (h *RouteHandler) OnIpamConfChanged(oldPool, newPool *proto.IPAMPool) error {
+	h.log.Debugf("Received IPAM config update in route handler old:%+v new:%+v", oldPool, newPool)
+	if newPool == nil && oldPool != nil {
+		routes, err := h.getNetworkRoute(oldPool.Cidr, "")
+		if err != nil {
+			h.log.Errorf("Error getting route from ipam update: %v", err)
+			return err
+		}
+		for _, route := range routes {
+			err = h.routeWatcher.DelRoute(route)
+			if err != nil {
+				h.log.Errorf("Cannot delete pool route %s through vpp tap: %v", oldPool.Cidr, err)
+				return err
+			}
+		}
+	} else if newPool != nil {
+		routes, err := h.getNetworkRoute(newPool.Cidr, "")
+		if err != nil {
+			h.log.Errorf("Error getting route from ipam update: %v", err)
+			return err
+		}
+		for _, route := range routes {
+			err = h.routeWatcher.AddRoute(route)
+			if err != nil {
+				h.log.Errorf("Cannot add pool route %s through vpp tap: %v", newPool.Cidr, err)
+				return err
+			}
+		}
+	}
+	return nil
+}
+
+func (h *RouteHandler) getNetworkRoute(network string, physicalNet string) (route []*netlink.Route, err error) {
+	_, cidr, err := net.ParseCIDR(network)
+	if err != nil {
+		return nil, errors.Wrapf(err, "error parsing %s", network)
+	}
+	var routes []*netlink.Route
+	var order int
+	for _, uplinkStatus := range common.VppManagerInfo.UplinkStatuses {
+		if uplinkStatus.PhysicalNetworkName == physicalNet {
+			gw := uplinkStatus.FakeNextHopIP4
+			if cidr.IP.To4() == nil {
+				gw = uplinkStatus.FakeNextHopIP6
+			}
+			var priority int
+			if uplinkStatus.IsMain {
+				priority = 0
+			} else {
+				order += 1
+				priority = order
+			}
+			routes = append(routes, &netlink.Route{
+				Dst:      cidr,
+				Gw:       gw,
+				Protocol: syscall.RTPROT_STATIC,
+				MTU:      watchers.GetUplinkMtu(),
+				Priority: priority,
+			})
+		}
+	}
+	return routes, nil
+}

calico-vpp-agent/watchers/net_watcher.go

@@ -106,7 +106,7 @@ func (w *NetWatcher) resyncAndCreateWatchers() error {
 				return errors.Wrapf(err, "Listing NetworkAttachmentDefinitions failed")
 			}
 			for _, nad := range nadList.Items {
-				err = w.onNadAdded(&nad)
+				err = w.OnNadAdded(&nad)
 				if err != nil {
 					return errors.Wrapf(err, "OnNadAdded failed for %v", nad)
 				}
@@ -197,7 +197,7 @@ func (w *NetWatcher) WatchNetworks(t *tomb.Tomb) error {
 						w.log.Errorf("update.Object is not *NetworkAttachmentDefinition, %v", update.Object)
 						continue
 					}
-					err := w.onNadAdded(nad)
+					err := w.OnNadAdded(nad)
 					if err != nil {
 						w.log.Error(err)
 					}
@@ -207,7 +207,7 @@ func (w *NetWatcher) WatchNetworks(t *tomb.Tomb) error {
 						w.log.Errorf("update.Object is not *NetworkAttachmentDefinition, %v", update.Object)
 						continue
 					}
-					err := w.onNadDeleted(nad)
+					err := w.OnNadDeleted(nad)
 					if err != nil {
 						w.log.Error(err)
 					}
@@ -227,7 +227,7 @@ func (w *NetWatcher) Stop() {
 	close(w.stop)
 }
 
-func (w *NetWatcher) onNadDeleted(nad *netv1.NetworkAttachmentDefinition) error {
+func (w *NetWatcher) OnNadDeleted(nad *netv1.NetworkAttachmentDefinition) error {
 	delete(w.nads, nad.Namespace+"/"+nad.Name)
 	for key, net := range w.networkDefinitions {
 		if net.NetAttachDefs == nad.Namespace+"/"+nad.Name {
@@ -241,7 +241,7 @@ func (w *NetWatcher) onNadDeleted(nad *netv1.NetworkAttachmentDefinition) error
 	return nil
 }
 
-func (w *NetWatcher) onNadAdded(nad *netv1.NetworkAttachmentDefinition) error {
+func (w *NetWatcher) OnNadAdded(nad *netv1.NetworkAttachmentDefinition) error {
 	var nadConfig nadv1.NetConfList
 	err := json.Unmarshal([]byte(nad.Spec.Config), &nadConfig)
 	if err != nil {