bmp: support post-policy routing monitoring

"RouteMonitoringPolicy" option added:

0: pre-policy (by default)
1: post-policy
2: both

=
[BmpServers]
  [[BmpServers.BmpServerList]]
    [BmpServers.BmpServerList.BmpServerConfig]
      Address = "127.0.0.1"
      Port = 11019
      RouteMonitoringPolicy = 2

Signed-off-by: FUJITA Tomonori <fujita.tomonori@lab.ntt.co.jp>

5 files changed, 90 insertions, 19 deletions

diff --git a/config/bgp_configs.go b/config/bgp_configs.go
index 8cf443e8..5611124d 100644
--- a/config/bgp_configs.go
+++ b/config/bgp_configs.go
@@ -112,6 +112,15 @@ const (
 	BGP_SET_COMMUNITY_OPTION_TYPE_REPLACE
 )
 
+// typedef for typedef gobgp:bmp-route-monitoring-policy-type
+type BmpRouteMonitoringPolicyType int
+
+const (
+	BMP_ROUTE_MONITORING_POLICY_TYPE_PRE_POLICY  BmpRouteMonitoringPolicyType = 0
+	BMP_ROUTE_MONITORING_POLICY_TYPE_POST_POLICY                              = 1
+	BMP_ROUTE_MONITORING_POLICY_TYPE_BOTH                                     = 2
+)
+
 // typedef for typedef gobgp:rpki-validation-result-type
 type RpkiValidationResultType int
 
@@ -142,6 +151,8 @@ type BmpServerConfig struct {
 	Address net.IP
 	// original -> gobgp:port
 	Port uint32
+	// original -> gobgp:route-monitoring-policy
+	RouteMonitoringPolicy BmpRouteMonitoringPolicyType
 }
 
 //struct for container gobgp:bmp-server
diff --git a/server/bmp.go b/server/bmp.go
index 5384c3f6..9496236d 100644
--- a/server/bmp.go
+++ b/server/bmp.go
@@ -29,7 +29,6 @@ type broadcastBMPMsg struct {
 	ch      chan *broadcastBMPMsg
 	msgList []*bgp.BMPMessage
 	conn    *net.TCPConn
-	addr    string
 }
 
 func (m *broadcastBMPMsg) send() {
@@ -38,7 +37,7 @@ func (m *broadcastBMPMsg) send() {
 
 type bmpConn struct {
 	conn *net.TCPConn
-	addr string
+	host string
 }
 
 type bmpClient struct {
@@ -55,16 +54,16 @@ func newBMPClient(conf config.BmpServers, connCh chan *bmpConn) (*bmpClient, err
 	b.ch = make(chan *broadcastBMPMsg)
 	b.connCh = connCh
 
-	tryConnect := func(addr string) {
+	tryConnect := func(host string) {
 		for {
-			conn, err := net.Dial("tcp", addr)
+			conn, err := net.Dial("tcp", host)
 			if err != nil {
 				time.Sleep(30 * time.Second)
 			} else {
-				log.Info("bmp server is connected, ", addr)
+				log.Info("bmp server is connected, ", host)
 				connCh <- &bmpConn{
 					conn: conn.(*net.TCPConn),
-					addr: addr,
+					host: host,
 				}
 				break
 			}
@@ -86,21 +85,47 @@ func newBMPClient(conf config.BmpServers, connCh chan *bmpConn) (*bmpClient, err
 					buf, _ := i.Serialize()
 					_, err := m.conn.Write(buf)
 					if err == nil {
-						connMap[m.addr] = m.conn
+						connMap[m.conn.RemoteAddr().String()] = m.conn
 					}
 				}
 
-				for addr, conn := range connMap {
+				for host, conn := range connMap {
 					if m.conn != nil && m.conn != conn {
 						continue
 					}
 
 					for _, msg := range m.msgList {
+						if msg.Header.Type == bgp.BMP_MSG_ROUTE_MONITORING {
+							c := func() *config.BmpServerConfig {
+								for _, c := range conf.BmpServerList {
+									b := &c.BmpServerConfig
+									if host == net.JoinHostPort(b.Address.String(), strconv.Itoa(int(b.Port))) {
+										return b
+									}
+								}
+								return nil
+							}()
+							if c == nil {
+								log.Fatal(host)
+							}
+							ph := msg.PeerHeader
+							switch c.RouteMonitoringPolicy {
+							case config.BMP_ROUTE_MONITORING_POLICY_TYPE_PRE_POLICY:
+								if ph.IsPostPolicy != false {
+									continue
+								}
+							case config.BMP_ROUTE_MONITORING_POLICY_TYPE_POST_POLICY:
+								if ph.IsPostPolicy != true {
+									continue
+								}
+							}
+
+						}
 						b, _ := msg.Serialize()
 						_, err := conn.Write(b)
 						if err != nil {
-							delete(connMap, addr)
-							go tryConnect(addr)
+							delete(connMap, host)
+							go tryConnect(host)
 							break
 						}
 					}
diff --git a/server/server.go b/server/server.go
index c8745c42..7f3e5756 100644
--- a/server/server.go
+++ b/server/server.go
@@ -341,7 +341,6 @@ func (server *BgpServer) Serve() {
 			m := &broadcastBMPMsg{
 				ch:      server.bmpClient.send(),
 				conn:    c.conn,
-				addr:    c.addr,
 				msgList: bmpMsgList,
 			}
 			server.broadcastMsgs = append(server.broadcastMsgs, m)
@@ -685,9 +684,10 @@ func (server *BgpServer) RSimportPaths(peer *Peer, pathList []*table.Path) []*ta
 	return moded
 }
 
-func (server *BgpServer) propagateUpdate(peer *Peer, pathList []*table.Path) []*SenderMsg {
+func (server *BgpServer) propagateUpdate(peer *Peer, pathList []*table.Path) ([]*SenderMsg, []*table.Path) {
 	msgs := make([]*SenderMsg, 0)
 	rib := server.globalRib
+	var alteredPathList []*table.Path
 	if peer != nil && peer.isRouteServerClient() {
 		for _, path := range pathList {
 			path.Filter(peer.ID(), table.POLICY_DIRECTION_IMPORT)
@@ -720,6 +720,7 @@ func (server *BgpServer) propagateUpdate(peer *Peer, pathList []*table.Path) []*
 		for idx, path := range pathList {
 			pathList[idx] = server.policy.ApplyPolicy(table.GLOBAL_RIB_NAME, table.POLICY_DIRECTION_IMPORT, path)
 		}
+		alteredPathList = pathList
 		dsts := rib.ProcessPaths(pathList)
 		sendPathList := make([]*table.Path, 0, len(dsts))
 		for _, dst := range dsts {
@@ -729,7 +730,7 @@ func (server *BgpServer) propagateUpdate(peer *Peer, pathList []*table.Path) []*
 			}
 		}
 		if len(sendPathList) == 0 {
-			return msgs
+			return msgs, alteredPathList
 		}
 
 		server.broadcastBests(sendPathList)
@@ -753,7 +754,7 @@ func (server *BgpServer) propagateUpdate(peer *Peer, pathList []*table.Path) []*
 			msgs = append(msgs, newSenderMsg(targetPeer, msgList))
 		}
 	}
-	return msgs
+	return msgs, alteredPathList
 }
 
 func (server *BgpServer) handleFSMMessage(peer *Peer, e *FsmMsg, incoming chan *FsmMsg) []*SenderMsg {
@@ -875,7 +876,19 @@ func (server *BgpServer) handleFSMMessage(peer *Peer, e *FsmMsg, incoming chan *
 			}
 			if len(pathList) > 0 {
 				server.roaClient.validate(pathList)
-				msgs = append(msgs, server.propagateUpdate(peer, pathList)...)
+				m, altered := server.propagateUpdate(peer, pathList)
+				msgs = append(msgs, m...)
+
+				if ch := server.bmpClient.send(); ch != nil {
+					for _, u := range table.CreateUpdateMsgFromPaths(altered) {
+						payload, _ := u.Serialize()
+						bm := &broadcastBMPMsg{
+							ch:      ch,
+							msgList: []*bgp.BMPMessage{bmpPeerRoute(bgp.BMP_PEER_TYPE_GLOBAL, true, 0, peer.fsm.peerInfo, e.timestamp.Unix(), payload)},
+						}
+						server.broadcastMsgs = append(server.broadcastMsgs, bm)
+					}
+				}
 			}
 		default:
 			log.WithFields(log.Fields{
@@ -1516,7 +1529,7 @@ func (server *BgpServer) handleGrpc(grpcReq *GrpcRequest) []*SenderMsg {
 	case REQ_MOD_PATH:
 		pathList := server.handleModPathRequest(grpcReq)
 		if len(pathList) > 0 {
-			msgs = server.propagateUpdate(nil, pathList)
+			msgs, _ = server.propagateUpdate(nil, pathList)
 			grpcReq.ResponseCh <- &GrpcResponse{}
 			close(grpcReq.ResponseCh)
 		}
@@ -1651,7 +1664,8 @@ func (server *BgpServer) handleGrpc(grpcReq *GrpcRequest) []*SenderMsg {
 					pathList = append(pathList, path.Clone(peer.conf.NeighborConfig.NeighborAddress, false))
 				}
 			}
-			msgs = append(msgs, server.propagateUpdate(peer, pathList)...)
+			m, _ := server.propagateUpdate(peer, pathList)
+			msgs = append(msgs, m...)
 		}
 
 		if grpcReq.RequestType == REQ_NEIGHBOR_SOFT_RESET_IN {
@@ -1811,7 +1825,7 @@ func (server *BgpServer) handleGrpc(grpcReq *GrpcRequest) []*SenderMsg {
 	case REQ_VRF, REQ_VRFS, REQ_VRF_MOD:
 		pathList := server.handleVrfRequest(grpcReq)
 		if len(pathList) > 0 {
-			msgs = server.propagateUpdate(nil, pathList)
+			msgs, _ = server.propagateUpdate(nil, pathList)
 		}
 	default:
 		err = fmt.Errorf("Unknown request type: %v", grpcReq.RequestType)
diff --git a/server/zclient.go b/server/zclient.go
index 419942de..fabd4205 100644
--- a/server/zclient.go
+++ b/server/zclient.go
@@ -157,7 +157,7 @@ func handleZapiMsg(msg *zebra.Message, server *BgpServer) []*SenderMsg {
 
 		if b.Prefix != nil && len(b.Nexthops) > 0 && b.Type != zebra.ROUTE_KERNEL {
 			p := createPathFromIPRouteMessage(msg, pi)
-			msgs := server.propagateUpdate(nil, []*table.Path{p})
+			msgs, _ := server.propagateUpdate(nil, []*table.Path{p})
 			return msgs
 		}
 	}
diff --git a/tools/pyang_plugins/gobgp.yang b/tools/pyang_plugins/gobgp.yang
index 7168e669..1f0f7474 100644
--- a/tools/pyang_plugins/gobgp.yang
+++ b/tools/pyang_plugins/gobgp.yang
@@ -48,6 +48,23 @@ module bgp-gobgp {
     }
   }
 
+  typedef bmp-route-monitoring-policy-type {
+    type enumeration {
+      enum PRE-POLICY {
+        value 0;
+        description "send pre-policy routes";
+      }
+      enum POST-POLICY {
+        value 1;
+        description "send post-policy routes";
+      }
+      enum BOTH {
+        value 2;
+        description "send both pre and post-policy routes";
+      }
+    }
+  }
+
   grouping gobgp-message-counter {
     description
       "Counters for all BGPMessage types";
@@ -410,6 +427,10 @@ module bgp-gobgp {
       description
         "Reference to the port of the BMP server";
     }
+    leaf route-monitoring-policy {
+      type bmp-route-monitoring-policy-type;
+      default PRE-POLICY;
+    }
   }
 
   grouping gobgp-bmp-server-state {