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,25 @@ func main() {
 		log.Fatalf("cannot get default BGP config %s", err)
 	}
 
+	// Create secret watcher (no longer uses events)
+	secretWatcher, err := watchers.NewSecretWatcher(k8sclient)
+	if err != nil {
+		log.Fatalf("could not create secret watcher: %s", err)
+	}
+
+	// Create peer handler and set it in felix server (following the existing pattern)
+	peerHandler := routing.NewPeerHandler(secretWatcher, log.WithFields(logrus.Fields{"component": "peer-handler"}))
+	peerHandler.SetBGPConf(bgpConf)
+	felixServer.SetPeerHandler(peerHandler)
+
+	// Set peer handler in routing server
+	routingServer.SetPeerHandler(peerHandler)
+
+	// Create peer watcher (no longer uses events from routing server)
+	peerWatcher := watchers.NewPeerWatcher(clientv3, peerHandler, secretWatcher, log.WithFields(logrus.Fields{"component": "peer-watcher"}))
 	peerWatcher.SetBGPConf(bgpConf)
+
 	routingServer.SetBGPConf(bgpConf)
-	felixServer.SetBGPConf(bgpConf)
 
 	watchDog := watchdog.NewWatchDog(log.WithFields(logrus.Fields{"component": "watchDog"}), &t)
 	Go(felixServer.ServeFelix)

calico-vpp-agent/common/pubsub.go

@@ -45,10 +45,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/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"
@@ -34,6 +33,7 @@ import (
 	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/felix/connectivity"
 	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/felix/policies"
 	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/felix/services"
+	"github.com/projectcalico/vpp-dataplane/v3/calico-vpp-agent/routing"
 	"github.com/projectcalico/vpp-dataplane/v3/config"
 	"github.com/projectcalico/vpp-dataplane/v3/vpplink"
 	"github.com/projectcalico/vpp-dataplane/v3/vpplink/types"
@@ -59,6 +59,7 @@ type Server struct {
 	cniHandler          *cni.CNIHandler
 	connectivityHandler *connectivity.ConnectivityHandler
 	serviceHandler      *services.ServiceHandler
+	peerHandler         *routing.PeerHandler
 }
 
 // NewFelixServer creates a felix server
@@ -104,8 +105,19 @@ 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
+}
+
+// SetPeerHandler sets the peer handler for the felix server
+func (s *Server) SetPeerHandler(peerHandler *routing.PeerHandler) {
+	s.peerHandler = peerHandler
+}
+
+// GetPeerHandler returns the peer handler
+func (s *Server) GetPeerHandler() *routing.PeerHandler {
+	return s.peerHandler
 }
 
 func (s *Server) getMainInterface() *config.UplinkStatus {

calico-vpp-agent/routing/bgp_watcher.go

@@ -669,6 +669,12 @@ func (s *Server) WatchBGPPath(t *tomb.Tomb) error {
 				}
 				s.log.Infof("bgp(del) filter deleted: %s", filter.Name)
 				delete(s.bgpFilters, filter.Name)
+			case common.PeerNodeStateChanged:
+				old, _ := evt.Old.(*common.LocalNodeSpec)
+				new, _ := evt.New.(*common.LocalNodeSpec)
+				if s.peerHandler != nil {
+					s.peerHandler.OnPeerNodeStateChanged(old, new)
+				}
 			}
 		}
 	}