server: use grpc req handling code for global init via config file

preparation for Operation_DEL support of REQ_MOD_GLOBAL_CONFIG

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

2 files changed, 154 insertions, 144 deletions

diff --git a/gobgpd/main.go b/gobgpd/main.go
index f60f250b..1f64c76b 100644
--- a/gobgpd/main.go
+++ b/gobgpd/main.go
@@ -201,7 +201,9 @@ func main() {
 			var added, deleted, updated []config.Neighbor
 
 			if bgpConfig == nil {
-				bgpServer.SetGlobalType(newConfig.Bgp.Global)
+				if err := bgpServer.SetGlobalType(newConfig.Bgp.Global); err != nil {
+					log.Fatalf("failed to set global config: %s", err)
+				}
 				bgpConfig = &newConfig.Bgp
 				bgpServer.SetRpkiConfig(newConfig.Bgp.RpkiServers)
 				added = newConfig.Bgp.Neighbors
diff --git a/server/server.go b/server/server.go
index a03fd92e..c300e823 100644
--- a/server/server.go
+++ b/server/server.go
@@ -89,13 +89,14 @@ func (ws Watchers) watching(typ watcherEventType) bool {
 
 type BgpServer struct {
 	bgpConfig     config.Bgp
-	globalTypeCh  chan config.Global
 	addedPeerCh   chan config.Neighbor
 	deletedPeerCh chan config.Neighbor
 	updatedPeerCh chan config.Neighbor
 	fsmincomingCh *channels.InfiniteChannel
 	fsmStateCh    chan *FsmMsg
 	rpkiConfigCh  chan []config.RpkiServer
+	acceptCh      chan *net.TCPConn
+	zapiMsgCh     chan *zebra.Message
 
 	GrpcReqCh      chan *GrpcRequest
 	policyUpdateCh chan config.RoutingPolicy
@@ -113,7 +114,6 @@ type BgpServer struct {
 
 func NewBgpServer() *BgpServer {
 	b := BgpServer{}
-	b.globalTypeCh = make(chan config.Global, 1)
 	b.addedPeerCh = make(chan config.Neighbor)
 	b.deletedPeerCh = make(chan config.Neighbor)
 	b.updatedPeerCh = make(chan config.Neighbor)
@@ -184,64 +184,11 @@ func (server *BgpServer) Listeners(addr string) []*net.TCPListener {
 }
 
 func (server *BgpServer) Serve() {
-	var g config.Global
-	for {
-		select {
-		case g = <-server.globalTypeCh:
-			server.bgpConfig.Global = g
-			server.globalTypeCh = nil
-		default:
-		}
-
-		if server.globalTypeCh == nil {
-			break
-		}
-
-		select {
-		case grpcReq := <-server.GrpcReqCh:
-			server.handleGrpc(grpcReq)
-		case g = <-server.globalTypeCh:
-			server.bgpConfig.Global = g
-			server.globalTypeCh = nil
-		}
-	}
-
-	server.roaManager, _ = newROAManager(g.Config.As, nil)
-
-	if g.Mrt.FileName != "" {
-		w, err := newMrtWatcher(g.Mrt.FileName)
-		if err != nil {
-			log.Warn(err)
-		} else {
-			server.watchers[WATCHER_MRT] = w
-		}
-	}
-
-	if len(g.BmpServers) > 0 {
-		w, err := newBmpWatcher(server.GrpcReqCh)
-		if err != nil {
-			log.Warn(err)
-		} else {
-			for _, server := range g.BmpServers {
-				if err := w.addServer(server.Config); err != nil {
-					log.Warn(err)
-				}
-			}
-			server.watchers[WATCHER_BMP] = w
-		}
-	}
+	server.roaManager, _ = newROAManager(0, nil)
 
 	w, _ := newGrpcIncomingWatcher()
 	server.watchers[WATCHER_GRPC_INCOMING] = w
 
-	if g.Zebra.Enabled {
-		cli, err := NewZclient(g.Zebra.Url, g.Zebra.RedistributeRouteTypeList)
-		if err != nil {
-			log.Error(err)
-		}
-		server.zclient = cli
-	}
-
 	senderCh := make(chan *SenderMsg, 1<<16)
 	go func(ch chan *SenderMsg) {
 		for {
@@ -275,34 +222,11 @@ func (server *BgpServer) Serve() {
 		}
 	}(broadcastCh)
 
-	rfs, _ := config.AfiSafis(g.AfiSafis).ToRfList()
-	server.globalRib = table.NewTableManager(rfs, g.MplsLabelRange.MinLabel, g.MplsLabelRange.MaxLabel)
 	server.listeners = make([]*net.TCPListener, 0, 2)
-	acceptCh := make(chan *net.TCPConn, 4096)
-	if g.ListenConfig.Port > 0 {
-		list := []string{"0.0.0.0", "::"}
-		if len(g.ListenConfig.LocalAddressList) > 0 {
-			list = g.ListenConfig.LocalAddressList
-		}
-		for _, addr := range list {
-			l, err := listenAndAccept(addr, uint32(g.ListenConfig.Port), acceptCh)
-			if err != nil {
-				log.Fatal(err)
-				os.Exit(1)
-			}
-			server.listeners = append(server.listeners, l)
-		}
-	}
-
 	server.fsmincomingCh = channels.NewInfiniteChannel()
 	server.fsmStateCh = make(chan *FsmMsg, 4096)
 	var senderMsgs []*SenderMsg
 
-	var zapiMsgCh chan *zebra.Message
-	if server.zclient != nil {
-		zapiMsgCh = server.zclient.Receive()
-	}
-
 	handleFsmMsg := func(e *FsmMsg) {
 		peer, found := server.neighborMap[e.MsgSrc]
 		if !found {
@@ -373,7 +297,7 @@ func (server *BgpServer) Serve() {
 			if len(m) > 0 {
 				senderMsgs = append(senderMsgs, m...)
 			}
-		case conn := <-acceptCh:
+		case conn := <-server.acceptCh:
 			passConn(conn)
 		default:
 		}
@@ -393,12 +317,12 @@ func (server *BgpServer) Serve() {
 			server.roaManager, _ = newROAManager(server.bgpConfig.Global.Config.As, c)
 		case rmsg := <-server.roaManager.recieveROA():
 			server.roaManager.handleROAEvent(rmsg)
-		case zmsg := <-zapiMsgCh:
+		case zmsg := <-server.zapiMsgCh:
 			m := handleZapiMsg(zmsg, server)
 			if len(m) > 0 {
 				senderMsgs = append(senderMsgs, m...)
 			}
-		case conn := <-acceptCh:
+		case conn := <-server.acceptCh:
 			passConn(conn)
 		case config := <-server.addedPeerCh:
 			addr := config.Config.NeighborAddress
@@ -407,12 +331,12 @@ func (server *BgpServer) Serve() {
 				log.Warn("Can't overwrite the exising peer ", addr)
 				continue
 			}
-			if g.ListenConfig.Port > 0 {
+			if server.bgpConfig.Global.ListenConfig.Port > 0 {
 				for _, l := range server.Listeners(addr) {
 					SetTcpMD5SigSockopts(l, addr, config.Config.AuthPassword)
 				}
 			}
-			peer := NewPeer(g, config, server.globalRib, server.policy)
+			peer := NewPeer(server.bgpConfig.Global, config, server.globalRib, server.policy)
 			server.setPolicyByConfig(peer.ID(), config.ApplyPolicy)
 			if peer.isRouteServerClient() {
 				pathList := make([]*table.Path, 0)
@@ -1177,10 +1101,53 @@ func (server *BgpServer) handleFSMMessage(peer *Peer, e *FsmMsg) []*SenderMsg {
 	return msgs
 }
 
-func (server *BgpServer) SetGlobalType(g config.Global) {
-	if server.globalTypeCh != nil {
-		server.globalTypeCh <- g
+func (server *BgpServer) SetGlobalType(g config.Global) error {
+	{
+		ch := make(chan *GrpcResponse)
+		server.GrpcReqCh <- &GrpcRequest{
+			RequestType: REQ_MOD_GLOBAL_CONFIG,
+			Data:        &g,
+			ResponseCh:  ch,
+		}
+		if err := (<-ch).Err(); err != nil {
+			return err
+		}
+	}
+	if g.Mrt.FileName != "" {
+		ch := make(chan *GrpcResponse)
+		server.GrpcReqCh <- &GrpcRequest{
+			RequestType: REQ_MOD_MRT,
+			Data: &api.ModMrtArguments{
+				Operation: api.Operation_ADD,
+				Filename:  g.Mrt.FileName,
+			},
+			ResponseCh: ch,
+		}
+		if err := (<-ch).Err(); err != nil {
+			return err
+		}
+	}
+	for _, s := range g.BmpServers {
+		ch := make(chan *GrpcResponse)
+		server.GrpcReqCh <- &GrpcRequest{
+			RequestType: REQ_MOD_BMP,
+			Data:        &s.Config,
+			ResponseCh:  ch,
+		}
+		if err := (<-ch).Err(); err != nil {
+			return err
+		}
+	}
+
+	if g.Zebra.Enabled {
+		cli, err := NewZclient(g.Zebra.Url, g.Zebra.RedistributeRouteTypeList)
+		if err != nil {
+			return err
+		}
+		server.zclient = cli
+		server.zapiMsgCh = server.zclient.Receive()
 	}
+	return nil
 }
 
 func (server *BgpServer) SetRpkiConfig(c []config.RpkiServer) {
@@ -1663,56 +1630,87 @@ END:
 }
 
 func (server *BgpServer) handleModConfig(grpcReq *GrpcRequest) error {
-	arg := grpcReq.Data.(*api.ModGlobalConfigArguments)
-	if arg.Operation != api.Operation_ADD {
-		return fmt.Errorf("invalid operation %s", arg.Operation)
-	}
-	if server.globalTypeCh == nil {
-		return fmt.Errorf("gobgp is already started")
-	}
-	g := arg.Global
-	id := net.ParseIP(g.RouterId)
-	if id == nil {
-		return fmt.Errorf("invalid router-id format: %s", g.RouterId)
-	}
-	families := make([]config.AfiSafi, 0, len(g.Families))
-	for _, f := range g.Families {
-		families = append(families, config.AfiSafi{
-			Config: config.AfiSafiConfig{
-				AfiSafiName: config.AfiSafiType(bgp.RouteFamily(f).String()),
-				Enabled:     true,
+	var c *config.Global
+	switch arg := grpcReq.Data.(type) {
+	case *api.ModGlobalConfigArguments:
+		if arg.Operation != api.Operation_ADD {
+			return fmt.Errorf("invalid operation %s", arg.Operation)
+		}
+		g := arg.Global
+		id := net.ParseIP(g.RouterId)
+		if id == nil {
+			return fmt.Errorf("invalid router-id format: %s", g.RouterId)
+		}
+		families := make([]config.AfiSafi, 0, len(g.Families))
+		for _, f := range g.Families {
+			name := config.AfiSafiType(bgp.RouteFamily(f).String())
+			families = append(families, config.AfiSafi{
+				AfiSafiName: name,
+				Config: config.AfiSafiConfig{
+					AfiSafiName: name,
+					Enabled:     true,
+				},
+				State: config.AfiSafiState{
+					AfiSafiName: name,
+				},
+			})
+		}
+		b := &config.Bgp{
+			Global: config.Global{
+				Config: config.GlobalConfig{
+					As:       g.As,
+					RouterId: g.RouterId,
+				},
+				ListenConfig: config.ListenConfig{
+					Port:             g.ListenPort,
+					LocalAddressList: g.ListenAddresses,
+				},
+				MplsLabelRange: config.MplsLabelRange{
+					MinLabel: g.MplsLabelMin,
+					MaxLabel: g.MplsLabelMax,
+				},
+				AfiSafis: families,
+				Collector: config.Collector{
+					Enabled: g.Collector,
+				},
 			},
-		})
+		}
+		if err := config.SetDefaultConfigValues(nil, b); err != nil {
+			return err
+		}
+		c = &b.Global
+	case *config.Global:
+		c = arg
 	}
-	c := config.Bgp{
-		Global: config.Global{
-			Config: config.GlobalConfig{
-				As:       g.As,
-				RouterId: g.RouterId,
-			},
-			ListenConfig: config.ListenConfig{
-				Port:             g.ListenPort,
-				LocalAddressList: g.ListenAddresses,
-			},
-			MplsLabelRange: config.MplsLabelRange{
-				MinLabel: g.MplsLabelMin,
-				MaxLabel: g.MplsLabelMax,
-			},
-			AfiSafis: families,
-			Collector: config.Collector{
-				Enabled: g.Collector,
-			},
-		},
+
+	if server.bgpConfig.Global.Config.As != 0 {
+		return fmt.Errorf("gobgp is already started")
 	}
-	err := config.SetDefaultConfigValues(nil, &c)
-	if err != nil {
-		return err
+
+	if c.ListenConfig.Port > 0 {
+		acceptCh := make(chan *net.TCPConn, 4096)
+		list := []string{"0.0.0.0", "::"}
+		if len(c.ListenConfig.LocalAddressList) > 0 {
+			list = c.ListenConfig.LocalAddressList
+		}
+		for _, addr := range list {
+			l, err := listenAndAccept(addr, uint32(c.ListenConfig.Port), acceptCh)
+			if err != nil {
+				return err
+			}
+			server.listeners = append(server.listeners, l)
+		}
+		server.acceptCh = acceptCh
 	}
+
+	rfs, _ := config.AfiSafis(c.AfiSafis).ToRfList()
+	server.globalRib = table.NewTableManager(rfs, c.MplsLabelRange.MinLabel, c.MplsLabelRange.MaxLabel)
+
 	p := config.RoutingPolicy{}
 	if err := server.SetRoutingPolicy(p); err != nil {
-		log.Fatal(err)
+		return err
 	}
-	server.globalTypeCh <- c.Global
+	server.bgpConfig.Global = *c
 	return nil
 }
 
@@ -1766,7 +1764,7 @@ func (server *BgpServer) handleGrpc(grpcReq *GrpcRequest) []*SenderMsg {
 		return results
 	}
 
-	if server.globalTypeCh != nil && grpcReq.RequestType != REQ_MOD_GLOBAL_CONFIG {
+	if server.bgpConfig.Global.Config.As == 0 && grpcReq.RequestType != REQ_MOD_GLOBAL_CONFIG {
 		grpcReq.ResponseCh <- &GrpcResponse{
 			ResponseErr: fmt.Errorf("bgpd main loop is not started yet"),
 		}
@@ -2824,31 +2822,41 @@ func (server *BgpServer) handleModMrt(grpcReq *GrpcRequest) {
 }
 
 func (server *BgpServer) handleModBmp(grpcReq *GrpcRequest) {
-	arg := grpcReq.Data.(*api.ModBmpArguments)
+	var op api.Operation
+	var c *config.BmpServerConfig
+	switch arg := grpcReq.Data.(type) {
+	case *api.ModBmpArguments:
+		c = &config.BmpServerConfig{
+			Address: arg.Address,
+			Port:    arg.Port,
+			RouteMonitoringPolicy: config.BmpRouteMonitoringPolicyType(arg.Type),
+		}
+		op = arg.Operation
+	case *config.BmpServerConfig:
+		c = arg
+		op = api.Operation_ADD
+	}
+
 	w, y := server.watchers[WATCHER_BMP]
 	if !y {
-		if arg.Operation == api.Operation_ADD {
+		if op == api.Operation_ADD {
 			w, _ = newBmpWatcher(server.GrpcReqCh)
 			server.watchers[WATCHER_BMP] = w
-		} else if arg.Operation == api.Operation_DEL {
+		} else if op == api.Operation_DEL {
 			grpcDone(grpcReq, fmt.Errorf("not enabled yet"))
 			return
 		}
 	}
-	c := config.BmpServerConfig{
-		Address: arg.Address,
-		Port:    arg.Port,
-		RouteMonitoringPolicy: config.BmpRouteMonitoringPolicyType(arg.Type),
-	}
-	switch arg.Operation {
+
+	switch op {
 	case api.Operation_ADD:
-		err := w.(*bmpWatcher).addServer(c)
+		err := w.(*bmpWatcher).addServer(*c)
 		grpcDone(grpcReq, err)
 	case api.Operation_DEL:
-		err := w.(*bmpWatcher).deleteServer(c)
+		err := w.(*bmpWatcher).deleteServer(*c)
 		grpcDone(grpcReq, err)
 	default:
-		grpcDone(grpcReq, fmt.Errorf("unsupported operation: %s", arg.Operation))
+		grpcDone(grpcReq, fmt.Errorf("unsupported operation: %s", op))
 	}
 }