refactor: decouple routingServer into BGPWatcher and routingHandler

This commit replaces the routingServer with separate BGPWatcher and
routingHandler components coordinated by a central BGP manager.
This eliminates the pubsub event-driven model in favor of direct
component communication via dedicated handler interfaces.

Key changes:
- Split routingServer responsibilities between BGPWatcher (monitoring)
  and routingHandler (VPP programming)
- Add BGPManager as central orchestrator coordinating both components
- BGPWatcher now focuses only on monitoring GoBGP without Felix cache access
- routingHandler retains Felix cache access for VPP programming decisions
- Replace multiple pubsub events with direct interface calls with the goal
  of eventually moving away from the pubsub infra

Signed-off-by: Aritra Basu <[email protected]>

Author: Aritra Basu

calico-vpp-agent/cmd/calico_vpp_dataplane.go

@@ -36,8 +36,8 @@ import (
 
 	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/common"
 	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/felix"
+	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/felix/routing"
 	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/prometheus"
-	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/routing"
 	"github.com/projectcalico/vpp-dataplane/v3/config"
 
 	watchdog "github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/watch_dog"
@@ -138,7 +138,6 @@ func main() {
 	prefixWatcher := watchers.NewPrefixWatcher(client, log.WithFields(logrus.Fields{"subcomponent": "prefix-watcher"}))
 	bgpFilterWatcher := watchers.NewBGPFilterWatcher(clientv3, k8sclient, log.WithFields(logrus.Fields{"subcomponent": "BGPFilter-watcher"}))
 	netWatcher := watchers.NewNetWatcher(vpp, log.WithFields(logrus.Fields{"component": "net-watcher"}))
-	routingServer := routing.NewRoutingServer(vpp, bgpServer, log.WithFields(logrus.Fields{"component": "routing"}))
 	prometheusServer := prometheus.NewPrometheusServer(vpp, log.WithFields(logrus.Fields{"component": "prometheus"}))
 	localSIDWatcher := watchers.NewLocalSIDWatcher(vpp, clientv3, log.WithFields(logrus.Fields{"subcomponent": "localsid-watcher"}))
 	felixServer := felix.NewFelixServer(vpp, clientv3, log.WithFields(logrus.Fields{"component": "policy"}))
@@ -158,13 +157,23 @@ func main() {
 		log.Fatalf("cannot get default BGP config %s", err)
 	}
 
-	peerManager, err := felix.NewPeerManager(clientv3, k8sclient, felixServer, log.WithFields(logrus.Fields{"component": "peer-manager"}))
-	if err != nil {
-		log.Fatalf("could not create peer manager: %s", err)
-	}
+	// Create peer manager
+	// This also creates (i) secret watcher (ii) peer watcher (iii) peer handler with the peer manager as the orchestrator
+	peerManager := routing.NewPeerManager(clientv3, k8sclient, felixServer, log.WithFields(logrus.Fields{"component": "peer-manager"}))
+
+	// Create BGP manager
+	// This also creates (i) BGP watcher (ii) BGP handler (iii) routing handler with the BGP manager as the orchestrator
+	bgpManager := routing.NewBGPManager(bgpServer, vpp, felixServer.GetFelixServerEventChan(), log.WithFields(logrus.Fields{"component": "bgp-manager"}))
+
+	// Wire up BGP handlers after all components are created
+	prefixWatcher.SetBGPDefinedSetHandler(bgpManager)
+	prefixWatcher.SetBGPPathHandler(bgpManager)
+	bgpFilterWatcher.SetBGPFilterHandler(bgpManager)
+	localSIDWatcher.SetBGPPathHandler(bgpManager)
+	peerManager.SetBGPPeerHandler(bgpManager)
 
-	routingServer.SetBGPConf(bgpConf)
 	peerManager.SetBGPConf(bgpConf)
+	bgpManager.SetBGPConf(bgpConf)
 
 	watchDog := watchdog.NewWatchDog(log.WithFields(logrus.Fields{"component": "watchDog"}), &t)
 	Go(felixServer.ServeFelix)
@@ -181,7 +190,7 @@ func main() {
 			panic("ourBGPSpec is not *common.LocalNodeSpec")
 		}
 		prefixWatcher.SetOurBGPSpec(bgpSpec)
-		routingServer.SetOurBGPSpec(bgpSpec)
+		bgpManager.SetOurBGPSpec(bgpSpec)
 		localSIDWatcher.SetOurBGPSpec(bgpSpec)
 		netWatcher.SetOurBGPSpec(bgpSpec)
 	}
@@ -197,7 +206,8 @@ func main() {
 	Go(prefixWatcher.WatchPrefix)
 	Go(peerManager.Start)
 	Go(bgpFilterWatcher.WatchBGPFilters)
-	Go(routingServer.ServeRouting)
+	Go(bgpManager.StartBGPWatcher)
+	Go(bgpManager.StartRoutingHandler)
 	Go(serviceServer.ServeService)
 	Go(cniServer.ServeCNI)
 	Go(prometheusServer.ServePrometheus)

calico-vpp-agent/common/bgp.go

@@ -0,0 +1,88 @@
+// 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 common
+
+import (
+	bgpapi "github.com/osrg/gobgp/v3/api"
+	calicov3 "github.com/projectcalico/api/pkg/apis/projectcalico/v3"
+)
+
+// LocalBGPPeer represents a BGP peer with its configuration and policies
+type LocalBGPPeer struct {
+	Peer           *bgpapi.Peer
+	BGPFilterNames []string
+	BGPPolicies    map[string]*ImpExpPol
+	NeighborSet    *bgpapi.DefinedSet
+}
+
+// BGPPrefixesPolicyAndAssignment contains BGP policy and prefix information
+type BGPPrefixesPolicyAndAssignment struct {
+	PolicyAssignment *bgpapi.PolicyAssignment
+	Policy           *bgpapi.Policy
+	Prefixes         []*bgpapi.DefinedSet
+}
+
+// ImpExpPol contains import and export policies
+type ImpExpPol struct {
+	Imp *BGPPrefixesPolicyAndAssignment
+	Exp *BGPPrefixesPolicyAndAssignment
+}
+
+// BGPConnectivityHandler interface for handling connectivity events
+type BGPConnectivityHandler interface {
+	OnConnectivityAdded(connectivity *NodeConnectivity) error
+	OnConnectivityDeleted(connectivity *NodeConnectivity) error
+	OnSRv6PolicyAdded(connectivity *NodeConnectivity) error
+	OnSRv6PolicyDeleted(connectivity *NodeConnectivity) error
+}
+
+// BGPPathHandler interface for handling BGP path operations
+type BGPPathHandler interface {
+	HandleBGPPathAdded(path *bgpapi.Path) error
+	HandleBGPPathDeleted(path *bgpapi.Path) error
+}
+
+// BGPPeerHandler interface for handling BGP peer operations
+type BGPPeerHandler interface {
+	HandleBGPPeerAdded(localPeer *LocalBGPPeer) error
+	HandleBGPPeerUpdated(localPeer *LocalBGPPeer, oldPeer *LocalBGPPeer) error
+	HandleBGPPeerDeleted(peerIP string) error
+}
+
+// BGPFilterHandler interface for handling BGP filter operations
+type BGPFilterHandler interface {
+	HandleBGPFilterAddedOrUpdated(filter calicov3.BGPFilter) error
+	HandleBGPFilterDeleted(filter calicov3.BGPFilter) error
+}
+
+// BGPDefinedSetHandler interface for handling BGP defined set operations
+type BGPDefinedSetHandler interface {
+	HandleBGPDefinedSetAdded(definedSet *bgpapi.DefinedSet) error
+	HandleBGPDefinedSetDeleted(definedSet *bgpapi.DefinedSet) error
+}
+
+// BGPHandler interface for handling different BGP operations
+type BGPHandler interface {
+	BGPPathHandler
+	BGPPeerHandler
+	BGPFilterHandler
+	BGPDefinedSetHandler
+
+	SetBGPConnectivityHandler(handler BGPConnectivityHandler)
+	InjectRoute(path *bgpapi.Path) error
+	InjectSRv6Policy(path *bgpapi.Path) error
+	InitialPolicySetting(isv6 bool) error
+}

calico-vpp-agent/common/common.go

@@ -21,6 +21,8 @@ import (
 	"math"
 	"net"
 	"os"
+	"reflect"
+	"sort"
 	"strconv"
 	"strings"
 	"time"
@@ -534,3 +536,17 @@ func CompareIPList(newIPList, oldIPList []net.IP) (added []net.IP, deleted []net
 	changed = len(added)+len(deleted) > 0
 	return
 }
+
+// CompareStringSlices compares two string slices for equality (order-independent)
+func CompareStringSlices(slice1, slice2 []string) bool {
+	if len(slice1) != len(slice2) {
+		return false
+	}
+
+	// Sort the slices in ascending order
+	sort.Strings(slice1)
+	sort.Strings(slice2)
+
+	// Compare the sorted slices
+	return reflect.DeepEqual(slice1, slice2)
+}

calico-vpp-agent/common/pubsub.go

@@ -16,8 +16,6 @@
 package common
 
 import (
-	"fmt"
-
 	log "github.com/sirupsen/logrus"
 )
 
@@ -29,12 +27,6 @@ const (
 	IpamConfChanged CalicoVppEventType = "IpamConfChanged"
 	BGPConfChanged  CalicoVppEventType = "BGPConfChanged"
 
-	ConnectivityAdded   CalicoVppEventType = "ConnectivityAdded"
-	ConnectivityDeleted CalicoVppEventType = "ConnectivityDeleted"
-
-	SRv6PolicyAdded   CalicoVppEventType = "SRv6PolicyAdded"
-	SRv6PolicyDeleted CalicoVppEventType = "SRv6PolicyDeleted"
-
 	PodAdded   CalicoVppEventType = "PodAdded"
 	PodDeleted CalicoVppEventType = "PodDeleted"
 
@@ -44,19 +36,6 @@ const (
 	TunnelAdded   CalicoVppEventType = "TunnelAdded"
 	TunnelDeleted CalicoVppEventType = "TunnelDeleted"
 
-	BGPPeerAdded   CalicoVppEventType = "BGPPeerAdded"
-	BGPPeerDeleted CalicoVppEventType = "BGPPeerDeleted"
-	BGPPeerUpdated CalicoVppEventType = "BGPPeerUpdated"
-
-	BGPFilterAddedOrUpdated CalicoVppEventType = "BGPFilterAddedOrUpdated"
-	BGPFilterDeleted        CalicoVppEventType = "BGPFilterDeleted"
-
-	BGPDefinedSetAdded   CalicoVppEventType = "BGPDefinedSetAdded"
-	BGPDefinedSetDeleted CalicoVppEventType = "BGPDefinedSetDeleted"
-
-	BGPPathAdded   CalicoVppEventType = "BGPPathAdded"
-	BGPPathDeleted CalicoVppEventType = "BGPPathDeleted"
-
 	NetAddedOrUpdated CalicoVppEventType = "NetAddedOrUpdated"
 	NetDeleted        CalicoVppEventType = "NetDeleted"
 	NetsSynced        CalicoVppEventType = "NetsSynced"
@@ -106,18 +85,8 @@ func RegisterHandler(channel chan any, name string) *PubSubHandlerRegistration {
 	return reg
 }
 
-func redactPassword(event CalicoVppEvent) string {
-	switch event.Type {
-	case BGPPeerAdded:
-		return string(event.Type)
-	default:
-		return fmt.Sprintf("%+v", event)
-	}
-
-}
-
 func SendEvent(event CalicoVppEvent) {
-	ThePubSub.log.Debugf("Broadcasting event %s", redactPassword(event))
+	ThePubSub.log.Debugf("Broadcasting event %+v", event)
 	for _, reg := range ThePubSub.pubSubHandlerRegistrations {
 		if reg.expectedEvents[event.Type] {
 			reg.channel <- event

calico-vpp-agent/felix/felix_server.go

@@ -87,10 +87,6 @@ func NewFelixServer(vpp *vpplink.VppLink, clientv3 calicov3cli.Interface, log *l
 		common.TunnelDeleted,
 		common.NetAddedOrUpdated,
 		common.NetDeleted,
-		common.ConnectivityAdded,
-		common.ConnectivityDeleted,
-		common.SRv6PolicyAdded,
-		common.SRv6PolicyDeleted,
 	)
 
 	return server
@@ -369,42 +365,6 @@ func (s *Server) handleFelixServerEvents(msg interface{}) (err error) {
 				return fmt.Errorf("evt.Old not a uint32 %v", evt.Old)
 			}
 			s.policiesHandler.OnTunnelDelete(swIfIndex)
-		case common.ConnectivityAdded:
-			new, ok := evt.New.(*common.NodeConnectivity)
-			if !ok {
-				s.log.Errorf("evt.New is not a *common.NodeConnectivity %v", evt.New)
-			}
-			err := s.connectivityHandler.UpdateIPConnectivity(new, false /* isWithdraw */)
-			if err != nil {
-				s.log.Errorf("Error while adding connectivity %s", err)
-			}
-		case common.ConnectivityDeleted:
-			old, ok := evt.Old.(*common.NodeConnectivity)
-			if !ok {
-				s.log.Errorf("evt.Old is not a *common.NodeConnectivity %v", evt.Old)
-			}
-			err := s.connectivityHandler.UpdateIPConnectivity(old, true /* isWithdraw */)
-			if err != nil {
-				s.log.Errorf("Error while deleting connectivity %s", err)
-			}
-		case common.SRv6PolicyAdded:
-			new, ok := evt.New.(*common.NodeConnectivity)
-			if !ok {
-				s.log.Errorf("evt.New is not a *common.NodeConnectivity %v", evt.New)
-			}
-			err := s.connectivityHandler.UpdateSRv6Policy(new, false /* isWithdraw */)
-			if err != nil {
-				s.log.Errorf("Error while adding SRv6 Policy %s", err)
-			}
-		case common.SRv6PolicyDeleted:
-			old, ok := evt.Old.(*common.NodeConnectivity)
-			if !ok {
-				s.log.Errorf("evt.Old is not a *common.NodeConnectivity %v", evt.Old)
-			}
-			err := s.connectivityHandler.UpdateSRv6Policy(old, true /* isWithdraw */)
-			if err != nil {
-				s.log.Errorf("Error while deleting SRv6 Policy %s", err)
-			}
 		default:
 			s.log.Warnf("Unhandled CalicoVppEvent.Type: %s", evt.Type)
 		}