RIS: Do not send withdraws when BMP session is lost (#372)

* RIS: Do not send withdraws when BMP session is lost

Co-authored-by: Oliver Geiselhardt-Herms <ogeiselhardt-herms@cloudflare.com>

Author: taktv6

cmd/ris-mirror/rismirror/router.go

@@ -38,6 +38,10 @@ func (r *Router) Address() net.IP {
 	return r.address
 }
 
+func (r *Router) Ready(vrf uint64, afi uint16) bool {
+	return true
+}
+
 // GetVRF gets a VRF by its ID
 func (r *Router) GetVRF(vrfID uint64) *vrf.VRF {
 	return r.vrfRegistry.GetVRFByRD(vrfID)

cmd/ris/risserver/server.go

@@ -57,6 +57,21 @@ func wrapGetRIBErr(err error, rtr string, vrfID uint64, version api.IP_Version)
 	return fmt.Errorf("unable to get RIB (%s/%s/v%d): %w", rtr, vrf.RouteDistinguisherHumanReadable(vrfID), version, err)
 }
 
+func wrapRIBNotReadyErr(err error, rtr string, vrfID uint64, version api.IP_Version) error {
+	return fmt.Errorf("RIB not ready yet (%s/%s/v%d): %w", rtr, vrf.RouteDistinguisherHumanReadable(vrfID), version, err)
+}
+
+func ipVersionFromProto(v netapi.IP_Version) uint16 {
+	switch v {
+	case netapi.IP_IPv4:
+		return 4
+	case netapi.IP_IPv6:
+		return 6
+	}
+
+	return 0
+}
+
 func (s Server) getRIB(rtr string, vrfID uint64, ipVersion netapi.IP_Version) (*locRIB.LocRIB, error) {
 	r := s.bmp.GetRouter(rtr)
 	if r == nil {
@@ -179,6 +194,10 @@ func (s *Server) ObserveRIB(req *pb.ObserveRIBRequest, stream pb.RoutingInformat
 		return status.New(codes.Unavailable, wrapGetRIBErr(err, req.Router, vrfID, ipVersion).Error()).Err()
 	}
 
+	if !s.bmp.GetRouter(req.Router).Ready(vrfID, ipVersionFromProto(ipVersion)) {
+		return status.New(codes.Unavailable, wrapRIBNotReadyErr(err, req.Router, vrfID, ipVersion).Error()).Err()
+	}
+
 	risObserveFIBClients.WithLabelValues(req.Router, fmt.Sprintf("%d", req.VrfId), fmt.Sprintf("%d", req.Afisafi)).Inc()
 	defer risObserveFIBClients.WithLabelValues(req.Router, fmt.Sprintf("%d", req.VrfId), fmt.Sprintf("%d", req.Afisafi)).Dec()
 

protocols/bgp/server/bmp_router.go

@@ -23,6 +23,7 @@ type RouterInterface interface {
 	Address() net.IP
 	GetVRF(vrfID uint64) *vrf.VRF
 	GetVRFs() []*vrf.VRF
+	Ready(vrf uint64, afi uint16) bool
 }
 
 // Router represents a BMP enabled route in BMP context
@@ -88,6 +89,46 @@ func newRouter(addr net.IP, port uint16, passive bool, arif adjRIBInFactoryI) *R
 	}
 }
 
+func (r *Router) Ready(vrf uint64, afi uint16) bool {
+	neighbors := r.neighborManager.list()
+	if len(neighbors) == 0 {
+		return false
+	}
+
+	if !neighborsIncludeVRF(neighbors, vrf) {
+		return false
+	}
+
+	for _, n := range neighbors {
+		if n.vrfID != vrf {
+			continue
+		}
+
+		var fsmAfi *fsmAddressFamily
+		if afi == 4 {
+			fsmAfi = n.fsm.ipv4Unicast
+		} else {
+			fsmAfi = n.fsm.ipv6Unicast
+		}
+
+		if !fsmAfi.endOfRIBMarkerReceived.Load() {
+			return false
+		}
+	}
+
+	return true
+}
+
+func neighborsIncludeVRF(neighbors []*neighbor, vrfID uint64) bool {
+	for _, n := range neighbors {
+		if n.vrfID == vrfID {
+			return true
+		}
+	}
+
+	return false
+}
+
 // GetVRF get's a VRF
 func (r *Router) GetVRF(rd uint64) *vrf.VRF {
 	return r.vrfRegistry.GetVRFByRD(rd)

protocols/bgp/server/fsm_address_family.go

@@ -1,6 +1,7 @@
 package server
 
 import (
+	"sync/atomic"
 	"time"
 
 	bnet "github.com/bio-routing/bio-rd/net"
@@ -35,7 +36,7 @@ type fsmAddressFamily struct {
 	multiProtocol bool
 
 	initialized            bool
-	endOfRIBMarkerReceived bool
+	endOfRIBMarkerReceived atomic.Bool
 }
 
 func newFSMAddressFamily(afi uint16, safi uint8, family *peerAddressFamily, fsm *FSM) *fsmAddressFamily {
@@ -181,7 +182,7 @@ func (f *fsmAddressFamily) processUpdate(u *packet.BGPUpdate, bmpPostPolicy bool
 	f.multiProtocolUpdates(u, bmpPostPolicy, timestamp)
 	if f.afi == packet.AFIIPv4 {
 		if u.IsEndOfRIBMarker() {
-			f.endOfRIBMarkerReceived = true
+			f.endOfRIBMarkerReceived.Store(true)
 		}
 
 		f.withdraws(u, bmpPostPolicy)
@@ -226,7 +227,7 @@ func (f *fsmAddressFamily) multiProtocolUpdates(u *packet.BGPUpdate, bmpPostPoli
 
 	if mpReachNLRI != nil && mpUnreachNLRI != nil {
 		if mpReachNLRI.NLRI == nil && mpUnreachNLRI.NLRI == nil {
-			f.endOfRIBMarkerReceived = true
+			f.endOfRIBMarkerReceived.Store(true)
 		}
 	}
 }

protocols/bgp/server/metrics_service.go

@@ -71,7 +71,7 @@ func metricsForFamily(family *fsmAddressFamily) *metrics.BGPAddressFamilyMetrics
 		AFI:                    family.afi,
 		SAFI:                   family.safi,
 		RoutesReceived:         uint64(family.adjRIBIn.RouteCount()),
-		EndOfRIBMarkerReceived: family.endOfRIBMarkerReceived,
+		EndOfRIBMarkerReceived: family.endOfRIBMarkerReceived.Load(),
 	}
 
 	if family.adjRIBOut != nil {

routingtable/client_manager.go

@@ -30,9 +30,10 @@ func (c *ClientOptions) GetMaxPaths(ecmpPaths uint) uint {
 
 // ClientManager manages clients of routing tables (observer pattern)
 type ClientManager struct {
-	clients map[RouteTableClient]ClientOptions
-	master  ClientManagerMaster
-	mu      sync.RWMutex
+	clients   map[RouteTableClient]ClientOptions
+	master    ClientManagerMaster
+	mu        sync.RWMutex
+	endOfLife bool // do not accept new clients when EOL
 }
 
 // NewClientManager creates and initializes a new client manager
@@ -62,6 +63,11 @@ func (c *ClientManager) GetOptions(client RouteTableClient) ClientOptions {
 // RegisterWithOptions registers a client with options for updates
 func (c *ClientManager) RegisterWithOptions(client RouteTableClient, opt ClientOptions) {
 	c.mu.Lock()
+
+	if c.endOfLife {
+		return
+	}
+
 	c.clients[client] = opt
 	c.mu.Unlock()
 
@@ -72,6 +78,11 @@ func (c *ClientManager) RegisterWithOptions(client RouteTableClient, opt ClientO
 func (c *ClientManager) Unregister(client RouteTableClient) bool {
 	c.mu.Lock()
 	defer c.mu.Unlock()
+
+	return c._unregister(client)
+}
+
+func (c *ClientManager) _unregister(client RouteTableClient) bool {
 	if _, ok := c.clients[client]; !ok {
 		return false
 	}
@@ -90,3 +101,14 @@ func (c *ClientManager) Clients() []RouteTableClient {
 
 	return ret
 }
+
+func (c *ClientManager) Dispose() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	c.endOfLife = true
+
+	for cli := range c.clients {
+		c._unregister(cli)
+	}
+}

routingtable/locRIB/loc_rib.go

@@ -352,5 +352,6 @@ func (a *LocRIB) RefreshRoute(*net.Prefix, []*route.Path) {
 func (a *LocRIB) Dispose() {
 	for _, c := range a.clientManager.Clients() {
 		c.Dispose()
+		a.clientManager.Unregister(c)
 	}
 }