Remove peers_watcher dependency on pubsub

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

Author: Aritra Basu

calico-vpp-agent/cmd/calico_vpp_dataplane.go

@@ -136,7 +136,6 @@ func main() {
 	linkWatcher := watchers.NewLinkWatcher(common.VppManagerInfo.UplinkStatuses, log.WithFields(logrus.Fields{"subcomponent": "host-link-watcher"}))
 	bgpConfigurationWatcher := watchers.NewBGPConfigurationWatcher(clientv3, log.WithFields(logrus.Fields{"subcomponent": "bgp-conf-watch"}))
 	prefixWatcher := watchers.NewPrefixWatcher(client, log.WithFields(logrus.Fields{"subcomponent": "prefix-watcher"}))
-	peerWatcher := watchers.NewPeerWatcher(clientv3, k8sclient, log.WithFields(logrus.Fields{"subcomponent": "peer-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"}))
@@ -159,9 +158,13 @@ func main() {
 		log.Fatalf("cannot get default BGP config %s", err)
 	}
 
-	peerWatcher.SetBGPConf(bgpConf)
+	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)
+	}
+
 	routingServer.SetBGPConf(bgpConf)
-	felixServer.SetBGPConf(bgpConf)
+	peerManager.SetBGPConf(bgpConf)
 
 	watchDog := watchdog.NewWatchDog(log.WithFields(logrus.Fields{"component": "watchDog"}), &t)
 	Go(felixServer.ServeFelix)
@@ -192,7 +195,7 @@ func main() {
 	Go(linkWatcher.WatchLinks)
 	Go(bgpConfigurationWatcher.WatchBGPConfiguration)
 	Go(prefixWatcher.WatchPrefix)
-	Go(peerWatcher.WatchBGPPeers)
+	Go(peerManager.Start)
 	Go(bgpFilterWatcher.WatchBGPFilters)
 	Go(routingServer.ServeRouting)
 	Go(serviceServer.ServeService)

calico-vpp-agent/common/pubsub.go

@@ -26,9 +26,8 @@ type CalicoVppEventType string
 const (
 	ChanSize = 500
 
-	PeerNodeStateChanged CalicoVppEventType = "PeerNodeStateChanged"
-	IpamConfChanged      CalicoVppEventType = "IpamConfChanged"
-	BGPConfChanged       CalicoVppEventType = "BGPConfChanged"
+	IpamConfChanged CalicoVppEventType = "IpamConfChanged"
+	BGPConfChanged  CalicoVppEventType = "BGPConfChanged"
 
 	ConnectivityAdded   CalicoVppEventType = "ConnectivityAdded"
 	ConnectivityDeleted CalicoVppEventType = "ConnectivityDeleted"
@@ -45,10 +44,9 @@ const (
 	TunnelAdded   CalicoVppEventType = "TunnelAdded"
 	TunnelDeleted CalicoVppEventType = "TunnelDeleted"
 
-	BGPPeerAdded     CalicoVppEventType = "BGPPeerAdded"
-	BGPPeerDeleted   CalicoVppEventType = "BGPPeerDeleted"
-	BGPPeerUpdated   CalicoVppEventType = "BGPPeerUpdated"
-	BGPSecretChanged CalicoVppEventType = "BGPSecretChanged"
+	BGPPeerAdded   CalicoVppEventType = "BGPPeerAdded"
+	BGPPeerDeleted CalicoVppEventType = "BGPPeerDeleted"
+	BGPPeerUpdated CalicoVppEventType = "BGPPeerUpdated"
 
 	BGPFilterAddedOrUpdated CalicoVppEventType = "BGPFilterAddedOrUpdated"
 	BGPFilterDeleted        CalicoVppEventType = "BGPFilterDeleted"

calico-vpp-agent/common/types.go

@@ -16,6 +16,7 @@
 package common
 
 import (
+	calicov3 "github.com/projectcalico/api/pkg/apis/projectcalico/v3"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
@@ -70,3 +71,11 @@ type ServiceEndpointsUpdate struct {
 type ServiceEndpointsDelete struct {
 	Meta *metav1.ObjectMeta
 }
+
+// BGPPeerState represents the state of a BGP peer
+type BGPPeerState struct {
+	AS            uint32
+	SweepFlag     bool
+	BGPPeerSpec   *calicov3.BGPPeerSpec
+	SecretChanged bool
+}

calico-vpp-agent/felix/connectivity/connectivity_handler.go

@@ -32,6 +32,11 @@ import (
 	"github.com/projectcalico/vpp-dataplane/v3/vpplink"
 )
 
+// PeerNodeStateChangeHandler defines the interface for handling peer node state changes
+type PeerNodeStateChangeHandler interface {
+	OnPeerNodeStateChanged(old, new *common.LocalNodeSpec)
+}
+
 type ConnectivityHandler struct {
 	log   *logrus.Entry
 	vpp   *vpplink.VppLink
@@ -40,6 +45,9 @@ type ConnectivityHandler struct {
 	providers         map[string]ConnectivityProvider
 	connectivityMap   map[string]common.NodeConnectivity
 	nodeByWGPublicKey map[string]string
+
+	// Peer node state change handlers
+	peerNodeStateChangeHandlers []PeerNodeStateChangeHandler
 }
 
 func NewConnectivityHandler(vpp *vpplink.VppLink, cache *cache.Cache, clientv3 calicov3cli.Interface, log *logrus.Entry) *ConnectivityHandler {
@@ -56,10 +64,16 @@ func NewConnectivityHandler(vpp *vpplink.VppLink, cache *cache.Cache, clientv3 c
 			WIREGUARD: NewWireguardProvider(vpp, clientv3, cache, log),
 			SRv6:      NewSRv6Provider(vpp, clientv3, cache, log),
 		},
-		nodeByWGPublicKey: make(map[string]string),
+		nodeByWGPublicKey:           make(map[string]string),
+		peerNodeStateChangeHandlers: make([]PeerNodeStateChangeHandler, 0),
 	}
 }
 
+// RegisterPeerNodeStateChangeHandler registers a handler for peer node state changes
+func (s *ConnectivityHandler) RegisterPeerNodeStateChangeHandler(handler PeerNodeStateChangeHandler) {
+	s.peerNodeStateChangeHandlers = append(s.peerNodeStateChangeHandlers, handler)
+}
+
 type change uint8
 
 const (
@@ -112,11 +126,11 @@ func (s *ConnectivityHandler) OnPeerNodeStateChanged(old, new *common.LocalNodeS
 			s.cache.NodeByAddr[new.IPv6Address.IP.String()] = *new
 		}
 	}
-	common.SendEvent(common.CalicoVppEvent{
-		Type: common.PeerNodeStateChanged,
-		Old:  old,
-		New:  new,
-	})
+
+	// Notify all registered peer node state change handlers
+	for _, handler := range s.peerNodeStateChangeHandlers {
+		handler.OnPeerNodeStateChanged(old, new)
+	}
 }
 
 func (s *ConnectivityHandler) UpdateSRv6Policy(cn *common.NodeConnectivity, IsWithdraw bool) (err error) {

calico-vpp-agent/felix/felix_server.go

@@ -24,7 +24,6 @@ import (
 	"github.com/sirupsen/logrus"
 	"gopkg.in/tomb.v2"
 
-	calicov3 "github.com/projectcalico/api/pkg/apis/projectcalico/v3"
 	"github.com/projectcalico/calico/felix/proto"
 
 	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/common"
@@ -64,6 +63,7 @@ type Server struct {
 // NewFelixServer creates a felix server
 func NewFelixServer(vpp *vpplink.VppLink, clientv3 calicov3cli.Interface, log *logrus.Entry) *Server {
 	cache := cache.NewCache(log)
+
 	server := &Server{
 		log: log,
 		vpp: vpp,
@@ -104,8 +104,9 @@ func (s *Server) GetCache() *cache.Cache {
 	return s.cache
 }
 
-func (s *Server) SetBGPConf(bgpConf *calicov3.BGPConfigurationSpec) {
-	s.cache.BGPConf = bgpConf
+// GetConnectivityHandler returns the connectivity handler
+func (s *Server) GetConnectivityHandler() *connectivity.ConnectivityHandler {
+	return s.connectivityHandler
 }
 
 func (s *Server) getMainInterface() *config.UplinkStatus {