refactor: make BGP peer and secret management system event-driven

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

Author: Aritra Basu

calico-vpp-agent/cmd/calico_vpp_dataplane.go

@@ -159,10 +159,11 @@ func main() {
 		log.Fatalf("cannot get default BGP config %s", err)
 	}
 
-	peerWatcher.SetBGPConf(bgpConf)
 	routingServer.SetBGPConf(bgpConf)
 	felixServer.SetBGPConf(bgpConf)
 
+	routingServer.SetPeerHandler(felixServer.GetPeerHandler())
+
 	watchDog := watchdog.NewWatchDog(log.WithFields(logrus.Fields{"component": "watchDog"}), &t)
 	Go(felixServer.ServeFelix)
 	Go(felixWatcher.WatchFelix)

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"
@@ -65,6 +64,15 @@ const (
 
 	IpamPoolUpdate CalicoVppEventType = "IpamPoolUpdate"
 	IpamPoolRemove CalicoVppEventType = "IpamPoolRemove"
+
+	PeersChanged CalicoVppEventType = "PeersChanged"
+	PeerAdded    CalicoVppEventType = "PeerAdded"
+	PeerUpdated  CalicoVppEventType = "PeerUpdated"
+	PeerDeleted  CalicoVppEventType = "PeerDeleted"
+
+	SecretAdded   CalicoVppEventType = "SecretAdded"
+	SecretChanged CalicoVppEventType = "SecretChanged"
+	SecretDeleted CalicoVppEventType = "SecretDeleted"
 )
 
 var (

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,45 @@ type ServiceEndpointsUpdate struct {
 type ServiceEndpointsDelete struct {
 	Meta *metav1.ObjectMeta
 }
+
+// PeersChangedEvent is emitted when the list of BGP peers changes
+type PeersChangedEvent struct {
+	Peers []calicov3.BGPPeer
+}
+
+// PeerAddedEvent is emitted when a BGP peer is added
+type PeerAddedEvent struct {
+	Peer calicov3.BGPPeer
+}
+
+// PeerUpdatedEvent is emitted when a BGP peer is updated
+type PeerUpdatedEvent struct {
+	Old calicov3.BGPPeer
+	New calicov3.BGPPeer
+}
+
+// PeerDeletedEvent is emitted when a BGP peer is deleted
+type PeerDeletedEvent struct {
+	Peer calicov3.BGPPeer
+}
+
+// SecretData represents secret information
+type SecretData struct {
+	Name string
+	Data map[string][]byte
+}
+
+// SecretAddedEvent is emitted when a secret is added
+type SecretAddedEvent struct {
+	Secret *SecretData
+}
+
+// SecretChangedEvent is emitted when a secret changes
+type SecretChangedEvent struct {
+	SecretName string
+}
+
+// SecretDeletedEvent is emitted when a secret is deleted
+type SecretDeletedEvent struct {
+	SecretName string
+}

calico-vpp-agent/felix/felix_server.go

@@ -34,6 +34,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 +60,7 @@ type Server struct {
 	cniHandler          *cni.CNIHandler
 	connectivityHandler *connectivity.ConnectivityHandler
 	serviceHandler      *services.ServiceHandler
+	peerHandler         *routing.PeerHandler
 }
 
 // NewFelixServer creates a felix server
@@ -77,6 +79,7 @@ func NewFelixServer(vpp *vpplink.VppLink, clientv3 calicov3cli.Interface, log *l
 		cniHandler:          cni.NewCNIHandler(vpp, cache, log),
 		connectivityHandler: connectivity.NewConnectivityHandler(vpp, cache, clientv3, log),
 		serviceHandler:      services.NewServiceHandler(vpp, cache, log),
+		peerHandler:         routing.NewPeerHandler(cache, log),
 	}
 
 	reg := common.RegisterHandler(server.felixServerEventChan, "felix server events")
@@ -91,6 +94,13 @@ func NewFelixServer(vpp *vpplink.VppLink, clientv3 calicov3cli.Interface, log *l
 		common.ConnectivityDeleted,
 		common.SRv6PolicyAdded,
 		common.SRv6PolicyDeleted,
+		common.PeersChanged,
+		common.PeerAdded,
+		common.PeerUpdated,
+		common.PeerDeleted,
+		common.SecretAdded,
+		common.SecretChanged,
+		common.SecretDeleted,
 	)
 
 	return server
@@ -104,6 +114,10 @@ func (s *Server) GetCache() *cache.Cache {
 	return s.cache
 }
 
+func (s *Server) GetPeerHandler() *routing.PeerHandler {
+	return s.peerHandler
+}
+
 func (s *Server) SetBGPConf(bgpConf *calicov3.BGPConfigurationSpec) {
 	s.cache.BGPConf = bgpConf
 }
@@ -404,6 +418,60 @@ func (s *Server) handleFelixServerEvents(msg interface{}) (err error) {
 			if err != nil {
 				s.log.Errorf("Error while deleting SRv6 Policy %s", err)
 			}
+		case common.PeersChanged:
+			peersEvent, ok := evt.New.(*common.PeersChangedEvent)
+			if !ok {
+				return fmt.Errorf("evt.New is not a (*common.PeersChangedEvent) %v", evt.New)
+			}
+			err := s.peerHandler.ProcessPeers(peersEvent.Peers)
+			if err != nil {
+				s.log.Errorf("Error processing peers: %v", err)
+			}
+		case common.PeerAdded:
+			peerEvent, ok := evt.New.(*common.PeerAddedEvent)
+			if !ok {
+				return fmt.Errorf("evt.New is not a (*common.PeerAddedEvent) %v", evt.New)
+			}
+			err := s.peerHandler.OnPeerAdded(&peerEvent.Peer)
+			if err != nil {
+				s.log.Errorf("Error adding peer: %v", err)
+			}
+		case common.PeerUpdated:
+			peerEvent, ok := evt.New.(*common.PeerUpdatedEvent)
+			if !ok {
+				return fmt.Errorf("evt.New is not a (*common.PeerUpdatedEvent) %v", evt.New)
+			}
+			err := s.peerHandler.OnPeerUpdated(&peerEvent.Old, &peerEvent.New)
+			if err != nil {
+				s.log.Errorf("Error updating peer: %v", err)
+			}
+		case common.PeerDeleted:
+			peerEvent, ok := evt.New.(*common.PeerDeletedEvent)
+			if !ok {
+				return fmt.Errorf("evt.New is not a (*common.PeerDeletedEvent) %v", evt.New)
+			}
+			err := s.peerHandler.OnPeerDeleted(&peerEvent.Peer)
+			if err != nil {
+				s.log.Errorf("Error deleting peer: %v", err)
+			}
+		case common.SecretAdded:
+			secretEvent, ok := evt.New.(*common.SecretAddedEvent)
+			if !ok {
+				return fmt.Errorf("evt.New is not a (*common.SecretAddedEvent) %v", evt.New)
+			}
+			s.peerHandler.OnSecretAdded(secretEvent.Secret)
+		case common.SecretChanged:
+			secretEvent, ok := evt.New.(*common.SecretChangedEvent)
+			if !ok {
+				return fmt.Errorf("evt.New is not a (*common.SecretChangedEvent) %v", evt.New)
+			}
+			s.peerHandler.OnSecretChanged(secretEvent.SecretName)
+		case common.SecretDeleted:
+			secretEvent, ok := evt.New.(*common.SecretDeletedEvent)
+			if !ok {
+				return fmt.Errorf("evt.New is not a (*common.SecretDeletedEvent) %v", evt.New)
+			}
+			s.peerHandler.OnSecretDeleted(secretEvent.SecretName)
 		default:
 			s.log.Warnf("Unhandled CalicoVppEvent.Type: %s", evt.Type)
 		}

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)
+				}
 			}
 		}
 	}