move gRPC-related code for REQ_ADD_NEIGHBOR and REQ_DELETE_NEIGHBOR to grpc_server.go

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

3 files changed, 102 insertions, 126 deletions

diff --git a/gobgpd/main.go b/gobgpd/main.go
index 31029635..67a0b2f4 100644
--- a/gobgpd/main.go
+++ b/gobgpd/main.go
@@ -248,17 +248,17 @@ func main() {
 				c = newConfig
 			}
 
-			for _, p := range added {
+			for i, p := range added {
 				log.Infof("Peer %v is added", p.Config.NeighborAddress)
-				bgpServer.PeerAdd(p)
+				bgpServer.AddNeighbor(&added[i])
 			}
-			for _, p := range deleted {
+			for i, p := range deleted {
 				log.Infof("Peer %v is deleted", p.Config.NeighborAddress)
-				bgpServer.PeerDelete(p)
+				bgpServer.DeleteNeighbor(&deleted[i])
 			}
-			for _, p := range updated {
+			for i, p := range updated {
 				log.Infof("Peer %v is updated", p.Config.NeighborAddress)
-				u, _ := bgpServer.PeerUpdate(p)
+				u, _ := bgpServer.UpdateNeighbor(&updated[i])
 				updatePolicy = updatePolicy || u
 			}
 
diff --git a/server/grpc_server.go b/server/grpc_server.go
index 12337403..15b45e5b 100644
--- a/server/grpc_server.go
+++ b/server/grpc_server.go
@@ -679,8 +679,8 @@ func (s *Server) DeleteVrf(ctx context.Context, arg *api.DeleteVrfRequest) (*api
 }
 
 func (s *Server) AddNeighbor(ctx context.Context, arg *api.AddNeighborRequest) (*api.AddNeighborResponse, error) {
-	c, err := func(a *api.Peer) (*config.Neighbor, error) {
-		pconf := &config.Neighbor{}
+	c, err := func(a *api.Peer) (config.Neighbor, error) {
+		pconf := config.Neighbor{}
 		if a.Conf != nil {
 			pconf.Config.NeighborAddress = a.Conf.NeighborAddress
 			pconf.Config.PeerAs = a.Conf.PeerAs
@@ -789,21 +789,13 @@ func (s *Server) AddNeighbor(ctx context.Context, arg *api.AddNeighborRequest) (
 	if err != nil {
 		return nil, err
 	}
-	d, err := s.get(REQ_ADD_NEIGHBOR, c)
-	if err != nil {
-		return nil, err
-	}
-	return d.(*api.AddNeighborResponse), err
+	return &api.AddNeighborResponse{}, s.bgpServer.AddNeighbor(&c)
 }
 
 func (s *Server) DeleteNeighbor(ctx context.Context, arg *api.DeleteNeighborRequest) (*api.DeleteNeighborResponse, error) {
-	d, err := s.get(REQ_DELETE_NEIGHBOR, &config.Neighbor{Config: config.NeighborConfig{
+	return &api.DeleteNeighborResponse{}, s.bgpServer.DeleteNeighbor(&config.Neighbor{Config: config.NeighborConfig{
 		NeighborAddress: arg.Peer.Conf.NeighborAddress,
 	}})
-	if err != nil {
-		return nil, err
-	}
-	return d.(*api.DeleteNeighborResponse), err
 }
 
 func NewPrefixFromApiStruct(a *api.Prefix) (*table.Prefix, error) {
diff --git a/server/server.go b/server/server.go
index 381f1f61..fd8a8d36 100644
--- a/server/server.go
+++ b/server/server.go
@@ -906,37 +906,6 @@ func (server *BgpServer) SetMrtConfig(c []config.Mrt) error {
 	return nil
 }
 
-func (server *BgpServer) PeerAdd(peer config.Neighbor) error {
-	ch := make(chan *GrpcResponse)
-	server.GrpcReqCh <- &GrpcRequest{
-		RequestType: REQ_ADD_NEIGHBOR,
-		Data:        &peer,
-		ResponseCh:  ch,
-	}
-	return (<-ch).Err()
-}
-
-func (server *BgpServer) PeerDelete(peer config.Neighbor) error {
-	ch := make(chan *GrpcResponse)
-	server.GrpcReqCh <- &GrpcRequest{
-		RequestType: REQ_DELETE_NEIGHBOR,
-		Data:        &peer,
-		ResponseCh:  ch,
-	}
-	return (<-ch).Err()
-}
-
-func (server *BgpServer) PeerUpdate(peer config.Neighbor) (bool, error) {
-	ch := make(chan *GrpcResponse)
-	server.GrpcReqCh <- &GrpcRequest{
-		RequestType: REQ_UPDATE_NEIGHBOR,
-		Data:        &peer,
-		ResponseCh:  ch,
-	}
-	res := <-ch
-	return res.Data.(bool), res.Err()
-}
-
 func (server *BgpServer) Shutdown() {
 	server.shutdown = true
 	for _, p := range server.neighborMap {
@@ -1441,7 +1410,7 @@ func (server *BgpServer) handleModConfig(grpcReq *GrpcRequest) error {
 		c = arg
 	case *api.StopServerRequest:
 		for k, _ := range server.neighborMap {
-			err := server.handleDeleteNeighbor(&config.Neighbor{Config: config.NeighborConfig{
+			err := server.deleteNeighbor(&config.Neighbor{Config: config.NeighborConfig{
 				NeighborAddress: k}}, bgp.BGP_ERROR_CEASE, bgp.BGP_ERROR_SUB_PEER_DECONFIGURED)
 			if err != nil {
 				return err
@@ -1922,27 +1891,6 @@ func (server *BgpServer) handleGrpc(grpcReq *GrpcRequest) {
 		}
 		grpcReq.ResponseCh <- result
 		close(grpcReq.ResponseCh)
-	case REQ_ADD_NEIGHBOR:
-		err := server.handleAddNeighbor(grpcReq.Data.(*config.Neighbor))
-		grpcReq.ResponseCh <- &GrpcResponse{
-			Data:        &api.AddNeighborResponse{},
-			ResponseErr: err,
-		}
-		close(grpcReq.ResponseCh)
-	case REQ_DELETE_NEIGHBOR:
-		err := server.handleDeleteNeighbor(grpcReq.Data.(*config.Neighbor), bgp.BGP_ERROR_CEASE, bgp.BGP_ERROR_SUB_PEER_DECONFIGURED)
-		grpcReq.ResponseCh <- &GrpcResponse{
-			Data:        &api.DeleteNeighborResponse{},
-			ResponseErr: err,
-		}
-		close(grpcReq.ResponseCh)
-	case REQ_UPDATE_NEIGHBOR:
-		policyUpdated, err := server.handleUpdateNeighbor(grpcReq.Data.(*config.Neighbor))
-		grpcReq.ResponseCh <- &GrpcResponse{
-			Data:        policyUpdated,
-			ResponseErr: err,
-		}
-		close(grpcReq.ResponseCh)
 	case REQ_MONITOR_RIB, REQ_MONITOR_NEIGHBOR_PEER_STATE:
 		if grpcReq.Name != "" {
 			if _, err = server.checkNeighborRequest(grpcReq); err != nil {
@@ -2030,7 +1978,7 @@ ERROR:
 	return
 }
 
-func (server *BgpServer) handleAddNeighbor(c *config.Neighbor) error {
+func (server *BgpServer) addNeighbor(c *config.Neighbor) error {
 	addr := c.Config.NeighborAddress
 	if _, y := server.neighborMap[addr]; y {
 		return fmt.Errorf("Can't overwrite the exising peer: %s", addr)
@@ -2048,9 +1996,7 @@ func (server *BgpServer) handleAddNeighbor(c *config.Neighbor) error {
 	}
 
 	peer := NewPeer(&server.bgpConfig.Global, c, server.globalRib, server.policy)
-	policyMutex.Lock()
 	server.setPolicyByConfig(peer.ID(), c.ApplyPolicy)
-	policyMutex.Unlock()
 	if peer.isRouteServerClient() {
 		pathList := make([]*table.Path, 0)
 		rfList := peer.configuredRFlist()
@@ -2071,7 +2017,22 @@ func (server *BgpServer) handleAddNeighbor(c *config.Neighbor) error {
 	return nil
 }
 
-func (server *BgpServer) handleDeleteNeighbor(c *config.Neighbor, code, subcode uint8) error {
+func (s *BgpServer) AddNeighbor(c *config.Neighbor) (err error) {
+	ch := make(chan struct{})
+	defer func() { <-ch }()
+
+	s.mgmtCh <- func() {
+		policyMutex.Lock()
+		defer func() {
+			policyMutex.Unlock()
+			close(ch)
+		}()
+		err = s.addNeighbor(c)
+	}
+	return err
+}
+
+func (server *BgpServer) deleteNeighbor(c *config.Neighbor, code, subcode uint8) error {
 	addr := c.Config.NeighborAddress
 	n, y := server.neighborMap[addr]
 	if !y {
@@ -2099,73 +2060,96 @@ func (server *BgpServer) handleDeleteNeighbor(c *config.Neighbor, code, subcode
 	return nil
 }
 
-func (server *BgpServer) handleUpdateNeighbor(c *config.Neighbor) (bool, error) {
-	addr := c.Config.NeighborAddress
-	peer := server.neighborMap[addr]
-	policyUpdated := false
+func (s *BgpServer) DeleteNeighbor(c *config.Neighbor) (err error) {
+	ch := make(chan struct{})
+	defer func() { <-ch }()
 
-	if !peer.fsm.pConf.ApplyPolicy.Equal(&c.ApplyPolicy) {
-		log.WithFields(log.Fields{
-			"Topic": "Peer",
-			"Key":   addr,
-		}).Info("Update ApplyPolicy")
+	s.mgmtCh <- func() {
 		policyMutex.Lock()
-		server.setPolicyByConfig(peer.ID(), c.ApplyPolicy)
-		peer.fsm.pConf.ApplyPolicy = c.ApplyPolicy
-		policyMutex.Unlock()
-		policyUpdated = true
+		defer func() {
+			policyMutex.Unlock()
+			close(ch)
+		}()
+		err = s.deleteNeighbor(c, bgp.BGP_ERROR_CEASE, bgp.BGP_ERROR_SUB_PEER_DECONFIGURED)
 	}
-	original := peer.fsm.pConf
+	return err
+}
 
-	if !original.Config.Equal(&c.Config) || !original.Transport.Config.Equal(&c.Transport.Config) || config.CheckAfiSafisChange(original.AfiSafis, c.AfiSafis) {
-		sub := uint8(bgp.BGP_ERROR_SUB_OTHER_CONFIGURATION_CHANGE)
-		if original.Config.AdminDown != c.Config.AdminDown {
-			sub = bgp.BGP_ERROR_SUB_ADMINISTRATIVE_SHUTDOWN
-			state := "Admin Down"
-			if c.Config.AdminDown == false {
-				state = "Admin Up"
-			}
+func (s *BgpServer) UpdateNeighbor(c *config.Neighbor) (policyUpdated bool, err error) {
+	ch := make(chan struct{})
+	defer func() { <-ch }()
+
+	s.mgmtCh <- func() {
+		policyMutex.Lock()
+		defer func() {
+			policyMutex.Unlock()
+			close(ch)
+		}()
+
+		addr := c.Config.NeighborAddress
+		peer := s.neighborMap[addr]
+
+		if !peer.fsm.pConf.ApplyPolicy.Equal(&c.ApplyPolicy) {
 			log.WithFields(log.Fields{
 				"Topic": "Peer",
-				"Key":   peer.ID(),
-				"State": state,
-			}).Info("update admin-state configuration")
-		} else if original.Config.PeerAs != c.Config.PeerAs {
-			sub = bgp.BGP_ERROR_SUB_PEER_DECONFIGURED
+				"Key":   addr,
+			}).Info("Update ApplyPolicy")
+			s.setPolicyByConfig(peer.ID(), c.ApplyPolicy)
+			peer.fsm.pConf.ApplyPolicy = c.ApplyPolicy
+			policyUpdated = true
+		}
+		original := peer.fsm.pConf
+
+		if !original.Config.Equal(&c.Config) || !original.Transport.Config.Equal(&c.Transport.Config) || config.CheckAfiSafisChange(original.AfiSafis, c.AfiSafis) {
+			sub := uint8(bgp.BGP_ERROR_SUB_OTHER_CONFIGURATION_CHANGE)
+			if original.Config.AdminDown != c.Config.AdminDown {
+				sub = bgp.BGP_ERROR_SUB_ADMINISTRATIVE_SHUTDOWN
+				state := "Admin Down"
+				if c.Config.AdminDown == false {
+					state = "Admin Up"
+				}
+				log.WithFields(log.Fields{
+					"Topic": "Peer",
+					"Key":   peer.ID(),
+					"State": state,
+				}).Info("update admin-state configuration")
+			} else if original.Config.PeerAs != c.Config.PeerAs {
+				sub = bgp.BGP_ERROR_SUB_PEER_DECONFIGURED
+			}
+			if err = s.deleteNeighbor(peer.fsm.pConf, bgp.BGP_ERROR_CEASE, sub); err != nil {
+				log.WithFields(log.Fields{
+					"Topic": "Peer",
+					"Key":   addr,
+				}).Error(err)
+				return
+			}
+			err = s.addNeighbor(c)
+			if err != nil {
+				log.WithFields(log.Fields{
+					"Topic": "Peer",
+					"Key":   addr,
+				}).Error(err)
+			}
+			return
 		}
-		if err := server.handleDeleteNeighbor(peer.fsm.pConf, bgp.BGP_ERROR_CEASE, sub); err != nil {
+
+		if !original.Timers.Config.Equal(&c.Timers.Config) {
 			log.WithFields(log.Fields{
 				"Topic": "Peer",
-				"Key":   addr,
-			}).Error(err)
-			return policyUpdated, err
+				"Key":   peer.ID(),
+			}).Info("update timer configuration")
+			peer.fsm.pConf.Timers.Config = c.Timers.Config
 		}
-		err := server.handleAddNeighbor(c)
+
+		err = peer.updatePrefixLimitConfig(c.AfiSafis)
 		if err != nil {
 			log.WithFields(log.Fields{
 				"Topic": "Peer",
 				"Key":   addr,
 			}).Error(err)
+			// rollback to original state
+			peer.fsm.pConf = original
 		}
-		return policyUpdated, err
-	}
-
-	if !original.Timers.Config.Equal(&c.Timers.Config) {
-		log.WithFields(log.Fields{
-			"Topic": "Peer",
-			"Key":   peer.ID(),
-		}).Info("update timer configuration")
-		peer.fsm.pConf.Timers.Config = c.Timers.Config
-	}
-
-	err := peer.updatePrefixLimitConfig(c.AfiSafis)
-	if err != nil {
-		log.WithFields(log.Fields{
-			"Topic": "Peer",
-			"Key":   addr,
-		}).Error(err)
-		// rollback to original state
-		peer.fsm.pConf = original
 	}
 	return policyUpdated, err
 }