Add EOR type to Table's filter on Watch method

New EOR filter provides opprtunity to watch eor messages from
different peers for any route family.

CLOUD-118121

api/gobgp.proto

@@ -125,7 +125,7 @@ message WatchEventRequest {
 
   message Table {
     message Filter {
-      enum Type { BEST = 0; ADJIN = 1; POST_POLICY = 2; }
+      enum Type { BEST = 0; ADJIN = 1; POST_POLICY = 2; EOR = 3; }
       Type type = 1;
       bool init = 2;
     }

internal/pkg/table/path.go

@@ -1246,6 +1246,12 @@ func (p *Path) GetHash() uint32 {
 	return p.attrsHash
 }
 
+func (p *Path) SetSource(peerInfo *PeerInfo) {
+	if p.info != nil {
+		p.info.source = peerInfo
+	}
+}
+
 func nlriToIPNet(nlri bgp.AddrPrefixInterface) *net.IPNet {
 	switch T := nlri.(type) {
 	case *bgp.IPAddrPrefix:

pkg/packet/bgp/bgp.go

@@ -1973,11 +1973,19 @@ func (l *LabeledVPNIPAddrPrefix) SAFI() uint8 {
 }
 
 func (l *LabeledVPNIPAddrPrefix) IPPrefixLen() uint8 {
-	return l.Length - 8*uint8(l.Labels.Len()+l.RD.Len())
+	rdLen := 0
+	if l.RD != nil {
+		rdLen = l.RD.Len()
+	}
+	return l.Length - 8*uint8(l.Labels.Len()+rdLen)
 }
 
 func (l *LabeledVPNIPAddrPrefix) Len(options ...*MarshallingOption) int {
-	return 1 + l.Labels.Len() + l.RD.Len() + int((l.IPPrefixLen()+7)/8)
+	rdLen := 0
+	if l.RD != nil {
+		rdLen = l.RD.Len()
+	}
+	return 1 + l.Labels.Len() + rdLen + int((l.IPPrefixLen()+7)/8)
 }
 
 func (l *LabeledVPNIPAddrPrefix) String() string {

pkg/server/grpc_server_test.go

@@ -1,9 +1,16 @@
 package server
 
 import (
+	"net"
 	"testing"
+	"time"
 
+	api "github.com/osrg/gobgp/v3/api"
+	"github.com/osrg/gobgp/v3/internal/pkg/table"
+	"github.com/osrg/gobgp/v3/pkg/apiutil"
+	"github.com/osrg/gobgp/v3/pkg/packet/bgp"
 	"github.com/stretchr/testify/assert"
+	"google.golang.org/protobuf/types/known/anypb"
 )
 
 func TestParseHost(t *testing.T) {
@@ -41,3 +48,115 @@ func TestParseHost(t *testing.T) {
 		})
 	}
 }
+
+func TestToPathApi(t *testing.T) {
+	type args struct {
+		path            *table.Path
+		v               *table.Validation
+		onlyBinary      bool
+		nlriBinary      bool
+		attributeBinary bool
+	}
+	tests := []struct {
+		name string
+		args args
+		want *api.Path
+	}{
+		{
+			name: "ipv4 path",
+			args: args{
+				path: table.NewPath(&table.PeerInfo{
+					ID:           net.IP{10, 10, 10, 10},
+					LocalID:      net.IP{10, 11, 11, 11},
+					Address:      net.IP{10, 12, 12, 12},
+					LocalAddress: net.IP{10, 13, 13, 13},
+				},
+					bgp.NewIPAddrPrefix(8, "10.0.0.0"),
+					false,
+					[]bgp.PathAttributeInterface{bgp.NewPathAttributeOrigin(0)},
+					time.Time{},
+					false),
+			},
+			want: &api.Path{
+				Nlri:   anyNlri(bgp.NewIPAddrPrefix(8, "10.0.0.0")),
+				Pattrs: anyAttrs([]bgp.PathAttributeInterface{bgp.NewPathAttributeOrigin(0)}),
+				Family: &api.Family{
+					Afi:  api.Family_AFI_IP,
+					Safi: api.Family_SAFI_UNICAST,
+				},
+				Validation: &api.Validation{},
+				NeighborIp: "10.12.12.12",
+				SourceId:   "10.10.10.10",
+			},
+		},
+		{
+			name: "eor ipv4 path",
+			args: args{
+				path: eor(bgp.RF_IPv4_UC),
+			},
+			want: &api.Path{
+				Nlri: anyEorNlri(bgp.AFI_IP, bgp.SAFI_UNICAST),
+				Family: &api.Family{
+					Afi:  api.Family_AFI_IP,
+					Safi: api.Family_SAFI_UNICAST,
+				},
+				Pattrs:     []*anypb.Any{},
+				Validation: &api.Validation{},
+				NeighborIp: "10.12.12.12",
+				SourceId:   "10.10.10.10",
+			},
+		},
+		{
+			name: "eor vpn path",
+			args: args{
+				path: eor(bgp.RF_IPv4_VPN),
+			},
+			want: &api.Path{
+				Nlri: anyEorNlri(bgp.AFI_IP, bgp.SAFI_MPLS_VPN),
+				Family: &api.Family{
+					Afi:  api.Family_AFI_IP,
+					Safi: api.Family_SAFI_MPLS_VPN,
+				},
+				Pattrs:     []*anypb.Any{},
+				Validation: &api.Validation{},
+				NeighborIp: "10.12.12.12",
+				SourceId:   "10.10.10.10",
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			apiPath := toPathApi(tt.args.path, tt.args.v, tt.args.onlyBinary, tt.args.nlriBinary, tt.args.attributeBinary)
+			assert.Equal(t, tt.want.Nlri, apiPath.Nlri, "not equal nlri")
+			assert.Equal(t, tt.want.Pattrs, apiPath.Pattrs, "not equal attrs")
+			assert.Equal(t, tt.want.Family, apiPath.Family, "not equal family")
+			assert.Equal(t, tt.want.NeighborIp, apiPath.NeighborIp, "not equal neighbor")
+		})
+	}
+}
+
+func eor(f bgp.RouteFamily) *table.Path {
+	p := table.NewEOR(f)
+	p.SetSource(&table.PeerInfo{
+		ID:           net.IP{10, 10, 10, 10},
+		LocalID:      net.IP{10, 11, 11, 11},
+		Address:      net.IP{10, 12, 12, 12},
+		LocalAddress: net.IP{10, 13, 13, 13},
+	})
+	return p
+}
+
+func anyEorNlri(afi uint16, safi uint8) *anypb.Any {
+	n, _ := bgp.NewPrefixFromRouteFamily(afi, safi)
+	return anyNlri(n)
+}
+
+func anyNlri(nlri bgp.AddrPrefixInterface) *anypb.Any {
+	anyNlri, _ := apiutil.MarshalNLRI(nlri)
+	return anyNlri
+}
+
+func anyAttrs(attrs []bgp.PathAttributeInterface) []*anypb.Any {
+	anyPattrs, _ := apiutil.MarshalPathAttributes(attrs)
+	return anyPattrs
+}

pkg/server/server.go

@@ -1667,6 +1667,21 @@ func (s *BgpServer) handleFSMMessage(peer *peer, e *fsmMsg) {
 					if f == bgp.RF_RTC_UC {
 						rtc = true
 					}
+					peer.fsm.lock.RLock()
+					peerInfo := &table.PeerInfo{
+						AS:           peer.fsm.peerInfo.AS,
+						ID:           peer.fsm.peerInfo.ID,
+						LocalAS:      peer.fsm.peerInfo.LocalAS,
+						LocalID:      peer.fsm.peerInfo.LocalID,
+						Address:      peer.fsm.peerInfo.Address,
+						LocalAddress: peer.fsm.peerInfo.LocalAddress,
+					}
+					peer.fsm.lock.RUnlock()
+					ev := &watchEventEor{
+						Family:   f,
+						PeerInfo: peerInfo,
+					}
+					s.notifyWatcher(watchEventTypeEor, ev)
 					for i, a := range peerAfiSafis {
 						if a.State.Family == f {
 							peer.fsm.lock.Lock()
@@ -4015,6 +4030,8 @@ func (s *BgpServer) WatchEvent(ctx context.Context, r *api.WatchEventRequest, fn
 				opts = append(opts, watchUpdate(filter.Init, ""))
 			case api.WatchEventRequest_Table_Filter_POST_POLICY:
 				opts = append(opts, watchPostUpdate(filter.Init, ""))
+			case api.WatchEventRequest_Table_Filter_EOR:
+				opts = append(opts, watchEor(filter.Init))
 			}
 		}
 	}
@@ -4067,6 +4084,19 @@ func (s *BgpServer) WatchEvent(ctx context.Context, r *api.WatchEventRequest, fn
 							},
 						},
 					})
+				case *watchEventEor:
+					eor := table.NewEOR(msg.Family)
+					eor.SetSource(msg.PeerInfo)
+					path := toPathApi(eor, nil, false, false, false)
+
+					fn(&api.WatchEventResponse{
+						Event: &api.WatchEventResponse_Table{
+							Table: &api.WatchEventResponse_TableEvent{
+								Paths: []*api.Path{path},
+							},
+						},
+					})
+
 				case *watchEventPeer:
 					fn(&api.WatchEventResponse{
 						Event: &api.WatchEventResponse_Peer{
@@ -4137,6 +4167,7 @@ const (
 	watchEventTypePeerState  watchEventType = "peerstate"
 	watchEventTypeTable      watchEventType = "table"
 	watchEventTypeRecvMsg    watchEventType = "receivedmessage"
+	watchEventTypeEor        watchEventType = "eor"
 )
 
 type watchEvent interface {
@@ -4214,6 +4245,11 @@ type watchEventMessage struct {
 	IsSent       bool
 }
 
+type watchEventEor struct {
+	Family   bgp.RouteFamily
+	PeerInfo *table.PeerInfo
+}
+
 type watchOptions struct {
 	bestpath       bool
 	preUpdate      bool
@@ -4225,6 +4261,8 @@ type watchOptions struct {
 	tableName      string
 	recvMessage    bool
 	peerAddress    string
+	initEor        bool
+	eor            bool
 }
 
 type watchOption func(*watchOptions)
@@ -4258,6 +4296,15 @@ func watchPostUpdate(current bool, peerAddress string) watchOption {
 	}
 }
 
+func watchEor(current bool) watchOption {
+	return func(o *watchOptions) {
+		o.eor = true
+		if current {
+			o.initEor = true
+		}
+	}
+}
+
 func watchPeer() watchOption {
 	return func(o *watchOptions) {
 		o.peerState = true
@@ -4412,6 +4459,9 @@ func (s *BgpServer) watch(opts ...watchOption) (w *watcher) {
 		if w.opts.postUpdate {
 			register(watchEventTypePostUpdate, w)
 		}
+		if w.opts.eor {
+			register(watchEventTypeEor, w)
+		}
 		if w.opts.peerState {
 			for _, p := range s.neighborMap {
 				w.notify(newWatchEventPeer(p, nil, p.fsm.state, PEER_EVENT_INIT))
@@ -4427,6 +4477,31 @@ func (s *BgpServer) watch(opts ...watchOption) (w *watcher) {
 				MultiPathList: s.globalRib.GetBestMultiPathList(table.GLOBAL_RIB_NAME, nil),
 			})
 		}
+		if w.opts.initEor && s.active() == nil {
+			for _, p := range s.neighborMap {
+				func() {
+					p.fsm.lock.RLock()
+					defer p.fsm.lock.RUnlock()
+					for _, a := range p.fsm.pConf.AfiSafis {
+						if s := a.MpGracefulRestart.State; s.Enabled && s.EndOfRibReceived {
+							family := a.State.Family
+							peerInfo := &table.PeerInfo{
+								AS:           p.fsm.peerInfo.AS,
+								ID:           p.fsm.peerInfo.ID,
+								LocalAS:      p.fsm.peerInfo.LocalAS,
+								LocalID:      p.fsm.peerInfo.LocalID,
+								Address:      p.fsm.peerInfo.Address,
+								LocalAddress: p.fsm.peerInfo.LocalAddress,
+							}
+							w.notify(&watchEventEor{
+								Family:   family,
+								PeerInfo: peerInfo,
+							})
+						}
+					}
+				}()
+			}
+		}
 		if w.opts.initUpdate {
 			for _, peer := range s.neighborMap {
 				peer.fsm.lock.RLock()