server/api: add api to monitor changes of best paths

Signed-off-by: ISHIDA Wataru <ishida.wataru@lab.ntt.co.jp>

2 files changed, 60 insertions, 0 deletions

diff --git a/server/grpc_server.go b/server/grpc_server.go
index b14845b2..aa36a264 100644
--- a/server/grpc_server.go
+++ b/server/grpc_server.go
@@ -48,6 +48,7 @@ const (
 	REQ_GLOBAL_RIB
 	REQ_GLOBAL_ADD
 	REQ_GLOBAL_DELETE
+	REQ_GLOBAL_MONITOR_BEST_CHANGED
 	REQ_POLICY_PREFIX
 	REQ_POLICY_PREFIXES
 	REQ_POLICY_PREFIX_ADD
@@ -205,6 +206,37 @@ func (s *Server) GetRib(arg *api.Arguments, stream api.Grpc_GetRibServer) error
 	return nil
 }
 
+func (s *Server) MonitorBestChanged(arg *api.Arguments, stream api.Grpc_MonitorBestChangedServer) error {
+	var reqType int
+	switch arg.Resource {
+	case api.Resource_GLOBAL:
+		reqType = REQ_GLOBAL_MONITOR_BEST_CHANGED
+	default:
+		return fmt.Errorf("unsupported resource type: %v", arg.Resource)
+	}
+
+	rf, err := convertAf2Rf(arg.Af)
+	if err != nil {
+		return err
+	}
+
+	req := NewGrpcRequest(reqType, "", rf, nil)
+	s.bgpServerCh <- req
+
+	for res := range req.ResponseCh {
+		if err = res.Err(); err != nil {
+			log.Debug(err.Error())
+			goto END
+		}
+		if err = stream.Send(res.Data.(*api.Path)); err != nil {
+			goto END
+		}
+	}
+END:
+	req.EndCh <- struct{}{}
+	return err
+}
+
 func (s *Server) neighbor(reqType int, arg *api.Arguments) (*api.Error, error) {
 	rf, err := convertAf2Rf(arg.Af)
 	if err != nil {
@@ -512,6 +544,7 @@ type GrpcRequest struct {
 	RemoteAddr  string
 	RouteFamily bgp.RouteFamily
 	ResponseCh  chan *GrpcResponse
+	EndCh       chan struct{}
 	Err         error
 	Data        interface{}
 }
@@ -522,6 +555,7 @@ func NewGrpcRequest(reqType int, remoteAddr string, rf bgp.RouteFamily, d interf
 		RouteFamily: rf,
 		RemoteAddr:  remoteAddr,
 		ResponseCh:  make(chan *GrpcResponse),
+		EndCh:       make(chan struct{}),
 		Data:        d,
 	}
 	return r
diff --git a/server/server.go b/server/server.go
index 939c2f56..cc19ecc7 100644
--- a/server/server.go
+++ b/server/server.go
@@ -52,6 +52,7 @@ type BgpServer struct {
 	policyUpdateCh chan config.RoutingPolicy
 	policyMap      map[string]*policy.Policy
 	routingPolicy  config.RoutingPolicy
+	broadcastReqs  []*GrpcRequest
 
 	neighborMap map[string]*Peer
 	localRibMap map[string]*LocalRib
@@ -354,6 +355,9 @@ func (server *BgpServer) dropPeerAllRoutes(peer *Peer) []*SenderMsg {
 			if len(pathList) == 0 {
 				continue
 			}
+
+			server.broadcastBests(pathList)
+
 			msgList := table.CreateUpdateMsgFromPaths(pathList)
 			for _, targetPeer := range server.neighborMap {
 				if targetPeer.isRouteServerClient() || targetPeer.fsm.state != bgp.BGP_FSM_ESTABLISHED {
@@ -404,6 +408,24 @@ func applyPolicies(peer *Peer, loc *LocalRib, isExport bool, pathList []table.Pa
 	return ret
 }
 
+func (server *BgpServer) broadcastBests(bests []table.Path) {
+	for _, path := range bests {
+		result := &GrpcResponse{
+			Data: path.ToApiStruct(),
+		}
+		remainReqs := make([]*GrpcRequest, 0, len(server.broadcastReqs))
+		for _, req := range server.broadcastReqs {
+			select {
+			case <-req.EndCh:
+				continue
+			case req.ResponseCh <- result:
+			}
+			remainReqs = append(remainReqs, req)
+		}
+		server.broadcastReqs = remainReqs
+	}
+}
+
 func (server *BgpServer) propagateUpdate(neighborAddress string, RouteServerClient bool, pathList []table.Path) []*SenderMsg {
 	msgs := make([]*SenderMsg, 0)
 
@@ -432,6 +454,8 @@ func (server *BgpServer) propagateUpdate(neighborAddress string, RouteServerClie
 			return msgs
 		}
 
+		server.broadcastBests(sendPathList)
+
 		for _, targetPeer := range server.neighborMap {
 			if targetPeer.isRouteServerClient() || targetPeer.fsm.state != bgp.BGP_FSM_ESTABLISHED {
 				continue
@@ -790,6 +814,8 @@ func (server *BgpServer) handleGrpc(grpcReq *GrpcRequest) []*SenderMsg {
 			close(grpcReq.ResponseCh)
 		}
 
+	case REQ_GLOBAL_MONITOR_BEST_CHANGED:
+		server.broadcastReqs = append(server.broadcastReqs, grpcReq)
 	case REQ_NEIGHBORS:
 		for _, peer := range server.neighborMap {
 			result := &GrpcResponse{