fix AddPeer timing issue, bgpserver interface changed.

Author: micho7

cmd/db-controller/main.go

@@ -60,12 +60,7 @@ func main() {
 	}
 	defer lockf.Close()
 
-	// for controlling the traffics that they're to the DB server port.
-	// the function returns nil if the expected chain is already exist.
-	nftConnect := nftables.NewDefaultConnector(logger)
-	if err := nftConnect.CreateChain(chainNameForDBAclFlag); err != nil {
-		panic(err)
-	}
+	logger.Info("Hello, Starting db-controller.")
 
 	// get my global ip address
 	myHostAddress, err := getNetIFAddress(globalInterfaceNameFlag)
@@ -74,42 +69,40 @@ func main() {
 	}
 	logger.Debug("host address", "address", myHostAddress)
 
-	// create context
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
+	dbReplicaPassword, err := readDBReplicaPassword(dbReplicaPasswordFilePathFlag)
+	if err != nil {
+		panic(err)
+	}
+
+	// for controlling the traffics that they're to the DB server port.
+	// the function returns nil if the expected chain is already exist.
+	nftConnect := nftables.NewDefaultConnector(logger)
+	if err := nftConnect.CreateChain(chainNameForDBAclFlag); err != nil {
+		panic(err)
+	}
+
+	var bgpPeers []bgpserver.Peer
+	for v := range strings.SplitSeq(bgpPeerAddressesFlag, ",") {
+		bgpPeers = append(bgpPeers, bgpserver.Peer{
+			Neighbor:             v,
+			RemoteAS:             uint32(bgpAsNumberFlag),
+			RemotePort:           uint32(bgpServingPortFlag),
+			KeepaliveIntervalSec: uint64(bgpKeepaliveIntervalSecFlag),
+		})
+	}
 
-	// start bgpserver
-	logger.Debug("starting bgpserver", "hostaddress", myHostAddress)
 	bgpServerConnect := bgpserver.NewDefaultConnector(
 		logger,
 		bgpserver.WithAsn(uint32(bgpAsNumberFlag)),
 		bgpserver.WithRouterId(myHostAddress),
 		bgpserver.WithListenPort(int32(bgpServingPortFlag)),
 		bgpserver.WithGrpcPort(gobgpGrpcPortFlag),
+		bgpserver.WithPeers(bgpPeers),
 	)
-	if err := bgpServerConnect.Start(ctx); err != nil {
-		panic(err)
-	}
-
-	// add peers to bgpserver
-	peers := strings.Split(bgpPeerAddressesFlag, ",")
-	if len(peers) != 2 {
-		panic("insufficient bgp peer addresses")
-	}
-	for _, v := range peers {
-		logger.Debug("adding peer", "neighbor", v)
-		bgpServerConnect.AddPeer(
-			v,
-			uint32(bgpAsNumberFlag),
-			uint32(bgpServingPortFlag),
-			uint64(bgpKeepaliveIntervalSecFlag),
-		)
-	}
 
-	dbReplicaPassword, err := readDBReplicaPassword(dbReplicaPasswordFilePathFlag)
-	if err != nil {
-		panic(err)
-	}
+	// create context
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
 
 	c := controller.NewController(
 		logger,
@@ -129,7 +122,10 @@ func main() {
 	wg.Add(1)
 	go func(ctx context.Context, wg *sync.WaitGroup, c *controller.Controller) {
 		defer wg.Done()
-		c.Start(ctx, time.Second*time.Duration(mainPollingSpanSecondFlag))
+		err := c.Start(ctx, time.Second*time.Duration(mainPollingSpanSecondFlag))
+		if err != nil {
+			panic(err)
+		}
 	}(ctx, wg, c)
 
 	if enablePrometheusExporterFlag {

pkg/bgpserver/connector.go

@@ -30,21 +30,28 @@ type Route struct {
 	Community uint32
 }
 
+type Peer struct {
+	Neighbor             string
+	RemoteAS             uint32
+	RemotePort           uint32
+	KeepaliveIntervalSec uint64
+}
+
 type Connector interface {
-	Start(ctx context.Context) error
-	AddPeer(neighbor string, remoteAS uint32, remotePort uint32, keepaliveIntervalSec uint64) error
+	Start() error
 	AddPath(prefix string, prefixLen uint32, nexthop string, community uint32) error
 	ListPath() ([]Route, error)
+	Stop()
 }
 
 type bgpServerConnector struct {
 	logger     *slog.Logger
 	server     *gobgpserver.BgpServer
-	ctx        context.Context
 	asn        uint32
 	routerId   string
 	listenPort int32
 	grpcPort   int
+	peers      []Peer
 }
 
 func NewDefaultConnector(logger *slog.Logger, configs ...func(*bgpServerConnector)) Connector {
@@ -85,48 +92,62 @@ func WithGrpcPort(grpcPort int) func(*bgpServerConnector) {
 	}
 }
 
-func (bs *bgpServerConnector) Start(ctx context.Context) error {
-	bs.ctx = ctx
+func WithPeers(peers []Peer) func(*bgpServerConnector) {
+	return func(c *bgpServerConnector) {
+		c.peers = peers
+	}
+}
+
+func (bs *bgpServerConnector) Start() error {
 	go bs.server.Serve()
-	return bs.server.StartBgp(ctx, &gobgpapi.StartBgpRequest{
+
+	err := bs.server.StartBgp(context.Background(), &gobgpapi.StartBgpRequest{
 		Global: &gobgpapi.Global{
 			Asn:             bs.asn,
 			RouterId:        bs.routerId,
 			ListenAddresses: []string{"0.0.0.0"},
 			ListenPort:      bs.listenPort,
 		},
 	})
-}
+	if err != nil {
+		return err
+	}
 
-func (bs *bgpServerConnector) AddPeer(neighbor string, remoteAS uint32, remotePort uint32, keepaliveIntervalSec uint64) error {
-	p := &gobgpapi.Peer{
-		Conf: &gobgpapi.PeerConf{
-			NeighborAddress: neighbor,
-			PeerAsn:         remoteAS,
-		},
-		Transport: &gobgpapi.Transport{
-			RemoteAddress: neighbor,
-			RemotePort:    remotePort,
-		},
-		Timers: &gobgpapi.Timers{
-			Config: &gobgpapi.TimersConfig{
-				KeepaliveInterval: keepaliveIntervalSec,
-				HoldTime:          keepaliveIntervalSec * 3,
+	for _, peer := range bs.peers {
+		p := &gobgpapi.Peer{
+			Conf: &gobgpapi.PeerConf{
+				NeighborAddress: peer.Neighbor,
+				PeerAsn:         peer.RemoteAS,
+			},
+			Transport: &gobgpapi.Transport{
+				RemoteAddress: peer.Neighbor,
+				RemotePort:    peer.RemoteAS,
+			},
+			Timers: &gobgpapi.Timers{
+				Config: &gobgpapi.TimersConfig{
+					KeepaliveInterval: peer.KeepaliveIntervalSec,
+					HoldTime:          peer.KeepaliveIntervalSec * 3,
+				},
 			},
-		},
 
-		// route reflector client is always on
-		RouteReflector: &gobgpapi.RouteReflector{
-			RouteReflectorClient: true,
-		},
+			// route reflector client is always on
+			RouteReflector: &gobgpapi.RouteReflector{
+				RouteReflectorClient: true,
+			},
+		}
+
+		err := bs.server.AddPeer(
+			context.Background(),
+			&gobgpapi.AddPeerRequest{
+				Peer: p,
+			},
+		)
+		if err != nil {
+			return err
+		}
 	}
 
-	return bs.server.AddPeer(
-		bs.ctx,
-		&gobgpapi.AddPeerRequest{
-			Peer: p,
-		},
-	)
+	return nil
 }
 
 func (bs *bgpServerConnector) AddPath(prefix string, prefixLen uint32, nexthop string, community uint32) error {
@@ -152,7 +173,7 @@ func (bs *bgpServerConnector) AddPath(prefix string, prefixLen uint32, nexthop s
 		attrs = []*apb.Any{attrOrigin, attrNextHop, attrCommunities}
 	}
 
-	_, err = bs.server.AddPath(bs.ctx, &gobgpapi.AddPathRequest{
+	_, err = bs.server.AddPath(context.Background(), &gobgpapi.AddPathRequest{
 		TableType: gobgpapi.TableType_GLOBAL,
 		Path: &gobgpapi.Path{
 			Family: &gobgpapi.Family{
@@ -170,15 +191,14 @@ func (bs *bgpServerConnector) AddPath(prefix string, prefixLen uint32, nexthop s
 func (bs *bgpServerConnector) ListPath() ([]Route, error) {
 	var routes []Route
 
-	err := bs.server.ListPath(bs.ctx, &gobgpapi.ListPathRequest{
+	err := bs.server.ListPath(context.Background(), &gobgpapi.ListPathRequest{
 		TableType: gobgpapi.TableType_GLOBAL,
 		Family: &gobgpapi.Family{
 			Afi:  gobgpapi.Family_AFI_IP,
 			Safi: gobgpapi.Family_SAFI_UNICAST,
 		},
 		EnableFiltered: true,
 	}, func(d *gobgpapi.Destination) {
-		bs.logger.Debug("ListPath: found prefix", "prefix", d.Prefix)
 		for _, path := range d.Paths {
 			for _, attr := range path.GetPattrs() {
 				m, err := attr.UnmarshalNew()
@@ -192,7 +212,6 @@ func (bs *bgpServerConnector) ListPath() ([]Route, error) {
 				}
 
 				for _, comm := range ca.Communities {
-					bs.logger.Debug("ListPath: found community", "community", EncodeCommunity(comm))
 					routes = append(routes, Route{
 						Prefix:    d.Prefix,
 						Community: comm,
@@ -205,6 +224,10 @@ func (bs *bgpServerConnector) ListPath() ([]Route, error) {
 	return routes, err
 }
 
+func (bs *bgpServerConnector) Stop() {
+	bs.server.Stop()
+}
+
 // EncodeCommunity converts plain community value to human readable notation(for example 65001:10)
 func EncodeCommunity(comm uint32) string {
 	upper := comm >> 16

pkg/bgpserver/fake_connector.go

@@ -15,7 +15,6 @@
 package bgpserver
 
 import (
-	"context"
 	"fmt"
 )
 
@@ -29,11 +28,7 @@ func NewFakeBgpServerConnector() Connector {
 	}
 }
 
-func (bs *FakeBgpServerConnector) Start(_ context.Context) error {
-	return nil
-}
-
-func (bs *FakeBgpServerConnector) AddPeer(_ string, _ uint32, _ uint32, _ uint64) error {
+func (bs *FakeBgpServerConnector) Start() error {
 	return nil
 }
 
@@ -47,3 +42,6 @@ func (bs *FakeBgpServerConnector) AddPath(prefix string, prefixLen uint32, _ str
 func (bs *FakeBgpServerConnector) ListPath() ([]Route, error) {
 	return nil, nil
 }
+
+func (bs *FakeBgpServerConnector) Stop() {
+}

pkg/controller/controller.go

@@ -184,18 +184,21 @@ func NewController(
 func (c *Controller) Start(
 	ctx context.Context,
 	ctrlerLoopInterval time.Duration,
-) {
-	c.logger.Info("Hello, Starting db-controller.")
+) error {
+	c.logger.Debug("controller: start bgpserver")
+	if err := c.bgpServerConnector.Start(); err != nil {
+		return err
+	}
+	defer c.bgpServerConnector.Stop()
 
 	ticker := time.NewTicker(ctrlerLoopInterval)
 	defer ticker.Stop()
 
-controllerLoop:
 	for {
 		select {
 		case <-ctx.Done():
-			c.onExit()
-			break controllerLoop
+			c.forceTransitionToFault()
+			return nil
 		case <-ticker.C:
 			// random sleep to avoid global synchronization
 			time.Sleep(time.Second * time.Duration(rand.Intn(2)+1))
@@ -250,15 +253,6 @@ func (c *Controller) decideNextState() State {
 	}
 }
 
-func (c *Controller) onExit() error {
-	c.setState(StateFault)
-	if err := c.triggerRunOnStateChangesToFault(); err != nil {
-		c.logger.Info("failed to TriggerRunOnStateChanges while going to fault. Ignore errors.", "error", err)
-	}
-
-	return nil
-}
-
 func (c *Controller) onStateHandler(nextState State) error {
 	if cannotTransitionTo(c.GetState(), nextState) {
 		panic("unreachable controller state was picked")
@@ -291,12 +285,9 @@ func (c *Controller) preDecideNextStateHandler() error {
 	if err != nil {
 		return err
 	}
-	c.logger.Debug("preDecideNextStateHandler", "routes", fmt.Sprintf("%#v", routes))
 
 	currentNeighbors := newNeighborSet()
 	for _, route := range routes {
-		c.logger.Debug(fmt.Sprintf("found route %s %s", route.Prefix, bgpserver.EncodeCommunity(route.Community)))
-
 		// parse community
 		state, ok := bgpCommunityToState[route.Community]
 		if !ok {
@@ -315,12 +306,10 @@ func (c *Controller) preDecideNextStateHandler() error {
 
 		// skip self originated route
 		if addr == c.hostAddress {
-			c.logger.Debug("it is my self host address, skipping.")
 			continue
 		}
 
 		if !slices.Contains(currentNeighbors[state], neighbor(addr)) {
-			c.logger.Debug(fmt.Sprintf("appending address %s to state %s", addr, state))
 			currentNeighbors[state] = append(currentNeighbors[state], neighbor(addr))
 		}
 	}