mirror of
https://github.com/osrg/gobgp.git
synced 2024-05-11 05:55:10 +00:00
server: kill peerMsg
Peers send and receive messages via channels, which could lead to a deadlock. With this patch, multiple goroutines are used for network I/Os per peer but one goroutine handle all ribs (including the global rib). So there is no messages via channels between peers. Signed-off-by: FUJITA Tomonori <[email protected]>
This commit is contained in:
FUJITA Tomonori
+85
-21
@@ -23,6 +23,7 @@ import (
|
||||
"gopkg.in/tomb.v2"
|
||||
"io"
|
||||
"net"
|
||||
"strconv"
|
||||
"time"
|
||||
)
|
||||
|
||||
@@ -36,6 +37,7 @@ const (
|
||||
|
||||
type fsmMsg struct {
|
||||
MsgType fsmMsgType
|
||||
MsgSrc string
|
||||
MsgData interface{}
|
||||
}
|
||||
|
||||
@@ -63,6 +65,7 @@ func (s AdminState) String() string {
|
||||
}
|
||||
|
||||
type FSM struct {
|
||||
t tomb.Tomb
|
||||
globalConfig *config.Global
|
||||
peerConfig *config.Neighbor
|
||||
keepaliveTicker *time.Ticker
|
||||
@@ -74,6 +77,8 @@ type FSM struct {
|
||||
negotiatedHoldTime float64
|
||||
adminState AdminState
|
||||
adminStateCh chan AdminState
|
||||
getActiveCh chan struct{}
|
||||
h *FSMHandler
|
||||
}
|
||||
|
||||
func (fsm *FSM) bgpMessageStateUpdate(MessageType uint8, isIn bool) {
|
||||
@@ -124,16 +129,19 @@ func (fsm *FSM) bgpMessageStateUpdate(MessageType uint8, isIn bool) {
|
||||
}
|
||||
}
|
||||
|
||||
func NewFSM(gConfig *config.Global, pConfig *config.Neighbor, connCh chan net.Conn) *FSM {
|
||||
return &FSM{
|
||||
func NewFSM(gConfig *config.Global, pConfig *config.Neighbor) *FSM {
|
||||
fsm := &FSM{
|
||||
globalConfig: gConfig,
|
||||
peerConfig: pConfig,
|
||||
state: bgp.BGP_FSM_IDLE,
|
||||
connCh: connCh,
|
||||
connCh: make(chan net.Conn),
|
||||
opensentHoldTime: float64(HOLDTIME_OPENSENT),
|
||||
adminState: ADMIN_STATE_UP,
|
||||
adminStateCh: make(chan AdminState, 1),
|
||||
getActiveCh: make(chan struct{}),
|
||||
}
|
||||
fsm.t.Go(fsm.connectLoop)
|
||||
return fsm
|
||||
}
|
||||
|
||||
func (fsm *FSM) StateChange(nextState bgp.FSMState) {
|
||||
@@ -144,6 +152,18 @@ func (fsm *FSM) StateChange(nextState bgp.FSMState) {
|
||||
"new": nextState.String(),
|
||||
}).Debug("state changed")
|
||||
fsm.state = nextState
|
||||
switch nextState {
|
||||
case bgp.BGP_FSM_ESTABLISHED:
|
||||
fsm.peerConfig.BgpNeighborCommonState.Uptime = time.Now().Unix()
|
||||
fsm.peerConfig.BgpNeighborCommonState.EstablishedCount++
|
||||
case bgp.BGP_FSM_ACTIVE:
|
||||
if !fsm.peerConfig.TransportOptions.PassiveMode {
|
||||
fsm.getActiveCh <- struct{}{}
|
||||
}
|
||||
fallthrough
|
||||
default:
|
||||
fsm.peerConfig.BgpNeighborCommonState.Downtime = time.Now().Unix()
|
||||
}
|
||||
}
|
||||
|
||||
func (fsm *FSM) LocalAddr() net.IP {
|
||||
@@ -179,6 +199,55 @@ func (fsm *FSM) sendNotification(conn net.Conn, code, subType uint8, data []byte
|
||||
fsm.sendNotificatonFromErrorMsg(conn, e.(*bgp.MessageError))
|
||||
}
|
||||
|
||||
func (fsm *FSM) connectLoop() error {
|
||||
var tick int
|
||||
if tick = int(fsm.peerConfig.Timers.ConnectRetry); tick < MIN_CONNECT_RETRY {
|
||||
tick = MIN_CONNECT_RETRY
|
||||
}
|
||||
|
||||
ticker := time.NewTicker(time.Duration(tick) * time.Second)
|
||||
ticker.Stop()
|
||||
|
||||
connect := func() {
|
||||
if bgp.FSMState(fsm.peerConfig.BgpNeighborCommonState.State) == bgp.BGP_FSM_ACTIVE {
|
||||
var host string
|
||||
addr := fsm.peerConfig.NeighborAddress
|
||||
|
||||
if addr.To4() != nil {
|
||||
host = addr.String() + ":" + strconv.Itoa(bgp.BGP_PORT)
|
||||
} else {
|
||||
host = "[" + addr.String() + "]:" + strconv.Itoa(bgp.BGP_PORT)
|
||||
}
|
||||
|
||||
conn, err := net.DialTimeout("tcp", host, time.Duration(MIN_CONNECT_RETRY-1)*time.Second)
|
||||
if err == nil {
|
||||
fsm.connCh <- conn
|
||||
} else {
|
||||
log.WithFields(log.Fields{
|
||||
"Topic": "Peer",
|
||||
"Key": fsm.peerConfig.NeighborAddress,
|
||||
}).Debugf("failed to connect: %s", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-fsm.t.Dying():
|
||||
log.WithFields(log.Fields{
|
||||
"Topic": "Peer",
|
||||
"Key": fsm.peerConfig.NeighborAddress,
|
||||
}).Debug("stop connect loop")
|
||||
ticker.Stop()
|
||||
return nil
|
||||
case <-ticker.C:
|
||||
connect()
|
||||
case <-fsm.getActiveCh:
|
||||
ticker = time.NewTicker(time.Duration(tick) * time.Second)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
type FSMHandler struct {
|
||||
t tomb.Tomb
|
||||
fsm *FSM
|
||||
@@ -203,15 +272,6 @@ func NewFSMHandler(fsm *FSM, incoming chan *fsmMsg, outgoing chan *bgp.BGPMessag
|
||||
return f
|
||||
}
|
||||
|
||||
func (h *FSMHandler) Wait() error {
|
||||
return h.t.Wait()
|
||||
}
|
||||
|
||||
func (h *FSMHandler) Stop() error {
|
||||
h.t.Kill(nil)
|
||||
return h.t.Wait()
|
||||
}
|
||||
|
||||
func (h *FSMHandler) idle() bgp.FSMState {
|
||||
fsm := h.fsm
|
||||
|
||||
@@ -354,6 +414,7 @@ func (h *FSMHandler) recvMessageWithError() error {
|
||||
}).Warn("malformed BGP Header")
|
||||
h.msgCh <- &fsmMsg{
|
||||
MsgType: FSM_MSG_BGP_MESSAGE,
|
||||
MsgSrc: h.fsm.peerConfig.NeighborAddress.String(),
|
||||
MsgData: err,
|
||||
}
|
||||
return err
|
||||
@@ -381,11 +442,13 @@ func (h *FSMHandler) recvMessageWithError() error {
|
||||
}).Warn("malformed BGP message")
|
||||
fmsg = &fsmMsg{
|
||||
MsgType: FSM_MSG_BGP_MESSAGE,
|
||||
MsgSrc: h.fsm.peerConfig.NeighborAddress.String(),
|
||||
MsgData: err,
|
||||
}
|
||||
} else {
|
||||
fmsg = &fsmMsg{
|
||||
MsgType: FSM_MSG_BGP_MESSAGE,
|
||||
MsgSrc: h.fsm.peerConfig.NeighborAddress.String(),
|
||||
MsgData: m,
|
||||
}
|
||||
if h.fsm.state == bgp.BGP_FSM_ESTABLISHED {
|
||||
@@ -456,6 +519,7 @@ func (h *FSMHandler) opensent() bgp.FSMState {
|
||||
|
||||
e := &fsmMsg{
|
||||
MsgType: FSM_MSG_BGP_MESSAGE,
|
||||
MsgSrc: fsm.peerConfig.NeighborAddress.String(),
|
||||
MsgData: m,
|
||||
}
|
||||
h.incoming <- e
|
||||
@@ -656,12 +720,12 @@ func (h *FSMHandler) sendMessageloop() error {
|
||||
// connection is closed and tried to kill us,
|
||||
// we need to die immediately. Otherwise fms
|
||||
// doesn't go to idle.
|
||||
for len(h.outgoing) > 0 {
|
||||
m := <-h.outgoing
|
||||
err := send(m)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
//
|
||||
// we always try to send. in case b), the
|
||||
// connection was already closed so it
|
||||
// correctly works in both cases.
|
||||
if h.fsm.state == bgp.BGP_FSM_ESTABLISHED {
|
||||
send(bgp.NewBGPNotificationMessage(bgp.BGP_ERROR_CEASE, bgp.BGP_ERROR_SUB_PEER_DECONFIGURED, nil))
|
||||
}
|
||||
return nil
|
||||
case m := <-h.outgoing:
|
||||
@@ -759,8 +823,8 @@ func (h *FSMHandler) loop() error {
|
||||
switch fsm.state {
|
||||
case bgp.BGP_FSM_IDLE:
|
||||
nextState = h.idle()
|
||||
// case bgp.BGP_FSM_CONNECT:
|
||||
// return h.connect()
|
||||
// case bgp.BGP_FSM_CONNECT:
|
||||
// nextState = h.connect()
|
||||
case bgp.BGP_FSM_ACTIVE:
|
||||
nextState = h.active()
|
||||
case bgp.BGP_FSM_OPENSENT:
|
||||
@@ -790,6 +854,7 @@ func (h *FSMHandler) loop() error {
|
||||
if nextState >= bgp.BGP_FSM_IDLE {
|
||||
e := &fsmMsg{
|
||||
MsgType: FSM_MSG_STATE_CHANGE,
|
||||
MsgSrc: fsm.peerConfig.NeighborAddress.String(),
|
||||
MsgData: nextState,
|
||||
}
|
||||
h.incoming <- e
|
||||
@@ -800,7 +865,6 @@ func (h *FSMHandler) loop() error {
|
||||
func (h *FSMHandler) changeAdminState(s AdminState) error {
|
||||
fsm := h.fsm
|
||||
if fsm.adminState != s {
|
||||
|
||||
log.WithFields(log.Fields{
|
||||
"Topic": "Peer",
|
||||
"Key": fsm.peerConfig.NeighborAddress,
|
||||
|
||||
+4
-10
@@ -289,19 +289,14 @@ func makePeerAndHandler() (*Peer, *FSMHandler) {
|
||||
|
||||
p := &Peer{
|
||||
globalConfig: globalConfig,
|
||||
peerConfig: neighborConfig,
|
||||
connCh: make(chan net.Conn),
|
||||
serverMsgCh: make(chan *serverMsg),
|
||||
peerMsgCh: make(chan *peerMsg),
|
||||
getActiveCh: make(chan struct{}),
|
||||
config: neighborConfig,
|
||||
capMap: make(map[bgp.BGPCapabilityCode]bgp.ParameterCapabilityInterface),
|
||||
}
|
||||
|
||||
p.siblings = make(map[string]*serverMsgDataPeer)
|
||||
p.fsm = NewFSM(&globalConfig, &neighborConfig, p.connCh)
|
||||
p.fsm = NewFSM(&globalConfig, &neighborConfig)
|
||||
|
||||
incoming := make(chan *fsmMsg, FSM_CHANNEL_LENGTH)
|
||||
p.outgoing = make(chan *bgp.BGPMessage, FSM_CHANNEL_LENGTH)
|
||||
incoming := make(chan *fsmMsg, 4096)
|
||||
p.outgoing = make(chan *bgp.BGPMessage, 4096)
|
||||
|
||||
h := &FSMHandler{
|
||||
fsm: p.fsm,
|
||||
@@ -309,7 +304,6 @@ func makePeerAndHandler() (*Peer, *FSMHandler) {
|
||||
incoming: incoming,
|
||||
outgoing: p.outgoing,
|
||||
}
|
||||
p.t.Go(p.connectLoop)
|
||||
|
||||
return p, h
|
||||
|
||||
|
||||
+175
-907
File diff suppressed because it is too large
Load Diff
@@ -1,522 +0,0 @@
|
||||
// Copyright (C) 2014 Nippon Telegraph and Telephone Corporation.
|
||||
//
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
||||
// implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
package server
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
log "github.com/Sirupsen/logrus"
|
||||
"github.com/osrg/gobgp/api"
|
||||
"github.com/osrg/gobgp/config"
|
||||
"github.com/osrg/gobgp/packet"
|
||||
"github.com/osrg/gobgp/table"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"net"
|
||||
"reflect"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func peerRC3() *table.PeerInfo {
|
||||
peer := &table.PeerInfo{
|
||||
AS: 66003,
|
||||
ID: net.ParseIP("10.0.255.3").To4(),
|
||||
LocalID: net.ParseIP("10.0.255.1").To4(),
|
||||
}
|
||||
return peer
|
||||
}
|
||||
|
||||
func createAsPathAttribute(ases []uint32) *bgp.PathAttributeAsPath {
|
||||
aspathParam := []bgp.AsPathParamInterface{bgp.NewAs4PathParam(2, ases)}
|
||||
aspath := bgp.NewPathAttributeAsPath(aspathParam)
|
||||
return aspath
|
||||
}
|
||||
|
||||
func createMpReach(nexthop string, prefix []bgp.AddrPrefixInterface) *bgp.PathAttributeMpReachNLRI {
|
||||
mp_reach := bgp.NewPathAttributeMpReachNLRI(nexthop, prefix)
|
||||
return mp_reach
|
||||
}
|
||||
|
||||
func update_fromRC3() *bgp.BGPMessage {
|
||||
pathAttributes := []bgp.PathAttributeInterface{
|
||||
bgp.NewPathAttributeOrigin(1),
|
||||
createAsPathAttribute([]uint32{66003, 4000, 70000}),
|
||||
createMpReach("2001:db8::3",
|
||||
[]bgp.AddrPrefixInterface{bgp.NewIPv6AddrPrefix(64, "38:38:38:38::")}),
|
||||
}
|
||||
return bgp.NewBGPUpdateMessage([]bgp.WithdrawnRoute{}, pathAttributes, []bgp.NLRInfo{})
|
||||
}
|
||||
|
||||
func TestProcessBGPUpdate_fourbyteAS(t *testing.T) {
|
||||
rib1 := table.NewTableManager("peer_test", []bgp.RouteFamily{bgp.RF_IPv4_UC, bgp.RF_IPv6_UC})
|
||||
|
||||
m := update_fromRC3()
|
||||
peerInfo := peerRC3()
|
||||
pathList := table.ProcessMessage(m, peerInfo)
|
||||
|
||||
pList, _ := rib1.ProcessPaths(pathList)
|
||||
assert.Equal(t, len(pList), 1)
|
||||
assert.Equal(t, pList[0].IsWithdraw(), false)
|
||||
fmt.Println(pList)
|
||||
sendMsg := table.CreateUpdateMsgFromPaths(pList)
|
||||
assert.Equal(t, len(sendMsg), 1)
|
||||
table.UpdatePathAttrs2ByteAs(sendMsg[0].Body.(*bgp.BGPUpdate))
|
||||
update := sendMsg[0].Body.(*bgp.BGPUpdate)
|
||||
assert.Equal(t, len(update.PathAttributes), 4)
|
||||
assert.Equal(t, reflect.TypeOf(update.PathAttributes[3]).String(), "*bgp.PathAttributeAs4Path")
|
||||
attr := update.PathAttributes[3].(*bgp.PathAttributeAs4Path)
|
||||
assert.Equal(t, len(attr.Value), 1)
|
||||
assert.Equal(t, attr.Value[0].AS, []uint32{66003, 70000})
|
||||
attrAS := update.PathAttributes[1].(*bgp.PathAttributeAsPath)
|
||||
assert.Equal(t, len(attrAS.Value), 1)
|
||||
assert.Equal(t, attrAS.Value[0].(*bgp.AsPathParam).AS, []uint16{bgp.AS_TRANS, 4000, bgp.AS_TRANS})
|
||||
|
||||
rib2 := table.NewTableManager("peer_test", []bgp.RouteFamily{bgp.RF_IPv4_UC, bgp.RF_IPv6_UC})
|
||||
pList2, _ := rib2.ProcessPaths(pathList)
|
||||
assert.Equal(t, len(pList2), 1)
|
||||
assert.Equal(t, pList[0].IsWithdraw(), false)
|
||||
sendMsg2 := table.CreateUpdateMsgFromPaths(pList2)
|
||||
assert.Equal(t, len(sendMsg2), 1)
|
||||
update2 := sendMsg2[0].Body.(*bgp.BGPUpdate)
|
||||
assert.Equal(t, len(update2.PathAttributes), 3)
|
||||
attrAS2 := update2.PathAttributes[1].(*bgp.PathAttributeAsPath)
|
||||
assert.Equal(t, len(attrAS2.Value), 1)
|
||||
assert.Equal(t, attrAS2.Value[0].(*bgp.As4PathParam).AS, []uint32{66003, 4000, 70000})
|
||||
}
|
||||
|
||||
func TestPeerAdminShutdownWhileEstablished(t *testing.T) {
|
||||
log.SetLevel(log.DebugLevel)
|
||||
assert := assert.New(t)
|
||||
m := NewMockConnection()
|
||||
globalConfig := config.Global{}
|
||||
peerConfig := config.Neighbor{}
|
||||
peerConfig.PeerAs = 100000
|
||||
peerConfig.Timers.KeepaliveInterval = 5
|
||||
peer := makePeer(globalConfig, peerConfig)
|
||||
peer.fsm.opensentHoldTime = 10
|
||||
|
||||
peer.t.Go(peer.loop)
|
||||
pushPackets := func() {
|
||||
o, _ := open().Serialize()
|
||||
m.setData(o)
|
||||
k, _ := keepalive().Serialize()
|
||||
m.setData(k)
|
||||
}
|
||||
go pushPackets()
|
||||
|
||||
waitUntil(assert, bgp.BGP_FSM_ACTIVE, peer, 1000)
|
||||
peer.connCh <- m
|
||||
waitUntil(assert, bgp.BGP_FSM_ESTABLISHED, peer, 1000)
|
||||
|
||||
grpcReq := NewGrpcRequest(REQ_NEIGHBOR_DISABLE, "0.0.0.0", bgp.RF_IPv4_UC, nil)
|
||||
msg := &serverMsg{
|
||||
msgType: SRV_MSG_API,
|
||||
msgData: grpcReq,
|
||||
}
|
||||
|
||||
peer.serverMsgCh <- msg
|
||||
result := <-grpcReq.ResponseCh
|
||||
err := result.Data.(api.Error)
|
||||
assert.Equal(err.Code, api.Error_SUCCESS)
|
||||
|
||||
waitUntil(assert, bgp.BGP_FSM_IDLE, peer, 1000)
|
||||
|
||||
assert.Equal(bgp.BGP_FSM_IDLE, peer.fsm.state)
|
||||
assert.Equal(ADMIN_STATE_DOWN, peer.fsm.adminState)
|
||||
lastMsg := m.sendBuf[len(m.sendBuf)-1]
|
||||
sent, _ := bgp.ParseBGPMessage(lastMsg)
|
||||
assert.Equal(uint8(bgp.BGP_MSG_NOTIFICATION), sent.Header.Type)
|
||||
assert.Equal(uint8(bgp.BGP_ERROR_CEASE), sent.Body.(*bgp.BGPNotification).ErrorCode)
|
||||
assert.Equal(uint8(bgp.BGP_ERROR_SUB_ADMINISTRATIVE_SHUTDOWN), sent.Body.(*bgp.BGPNotification).ErrorSubcode)
|
||||
assert.True(m.isClosed)
|
||||
|
||||
// check counter
|
||||
counter := peer.fsm.peerConfig.BgpNeighborCommonState
|
||||
assertCounter(assert, counter)
|
||||
}
|
||||
|
||||
func TestPeerAdminShutdownWhileIdle(t *testing.T) {
|
||||
log.SetLevel(log.DebugLevel)
|
||||
assert := assert.New(t)
|
||||
|
||||
globalConfig := config.Global{}
|
||||
peerConfig := config.Neighbor{}
|
||||
peerConfig.PeerAs = 100000
|
||||
peerConfig.Timers.KeepaliveInterval = 5
|
||||
peer := makePeer(globalConfig, peerConfig)
|
||||
peer.fsm.opensentHoldTime = 10
|
||||
peer.fsm.idleHoldTime = 5
|
||||
peer.t.Go(peer.loop)
|
||||
|
||||
waitUntil(assert, bgp.BGP_FSM_IDLE, peer, 1000)
|
||||
|
||||
grpcReq := NewGrpcRequest(REQ_NEIGHBOR_DISABLE, "0.0.0.0", bgp.RF_IPv4_UC, nil)
|
||||
msg := &serverMsg{
|
||||
msgType: SRV_MSG_API,
|
||||
msgData: grpcReq,
|
||||
}
|
||||
|
||||
peer.serverMsgCh <- msg
|
||||
result := <-grpcReq.ResponseCh
|
||||
err := result.Data.(api.Error)
|
||||
assert.Equal(err.Code, api.Error_SUCCESS)
|
||||
|
||||
waitUntil(assert, bgp.BGP_FSM_IDLE, peer, 100)
|
||||
assert.Equal(bgp.BGP_FSM_IDLE, peer.fsm.state)
|
||||
assert.Equal(ADMIN_STATE_DOWN, peer.fsm.adminState)
|
||||
|
||||
// check counter
|
||||
counter := peer.fsm.peerConfig.BgpNeighborCommonState
|
||||
assertCounter(assert, counter)
|
||||
}
|
||||
|
||||
func TestPeerAdminShutdownWhileActive(t *testing.T) {
|
||||
log.SetLevel(log.DebugLevel)
|
||||
assert := assert.New(t)
|
||||
|
||||
globalConfig := config.Global{}
|
||||
peerConfig := config.Neighbor{}
|
||||
peerConfig.PeerAs = 100000
|
||||
peerConfig.Timers.KeepaliveInterval = 5
|
||||
peer := makePeer(globalConfig, peerConfig)
|
||||
peer.fsm.opensentHoldTime = 10
|
||||
peer.t.Go(peer.loop)
|
||||
|
||||
waitUntil(assert, bgp.BGP_FSM_ACTIVE, peer, 1000)
|
||||
|
||||
grpcReq := NewGrpcRequest(REQ_NEIGHBOR_DISABLE, "0.0.0.0", bgp.RF_IPv4_UC, nil)
|
||||
msg := &serverMsg{
|
||||
msgType: SRV_MSG_API,
|
||||
msgData: grpcReq,
|
||||
}
|
||||
|
||||
peer.serverMsgCh <- msg
|
||||
result := <-grpcReq.ResponseCh
|
||||
err := result.Data.(api.Error)
|
||||
assert.Equal(err.Code, api.Error_SUCCESS)
|
||||
|
||||
waitUntil(assert, bgp.BGP_FSM_IDLE, peer, 100)
|
||||
assert.Equal(bgp.BGP_FSM_IDLE, peer.fsm.state)
|
||||
assert.Equal(ADMIN_STATE_DOWN, peer.fsm.adminState)
|
||||
|
||||
// check counter
|
||||
counter := peer.fsm.peerConfig.BgpNeighborCommonState
|
||||
assertCounter(assert, counter)
|
||||
}
|
||||
|
||||
func TestPeerAdminShutdownWhileOpensent(t *testing.T) {
|
||||
log.SetLevel(log.DebugLevel)
|
||||
assert := assert.New(t)
|
||||
m := NewMockConnection()
|
||||
globalConfig := config.Global{}
|
||||
peerConfig := config.Neighbor{}
|
||||
peerConfig.PeerAs = 100000
|
||||
peerConfig.Timers.KeepaliveInterval = 5
|
||||
peer := makePeer(globalConfig, peerConfig)
|
||||
peer.fsm.opensentHoldTime = 1
|
||||
peer.t.Go(peer.loop)
|
||||
|
||||
waitUntil(assert, bgp.BGP_FSM_ACTIVE, peer, 1000)
|
||||
peer.connCh <- m
|
||||
waitUntil(assert, bgp.BGP_FSM_OPENSENT, peer, 1000)
|
||||
|
||||
grpcReq := NewGrpcRequest(REQ_NEIGHBOR_DISABLE, "0.0.0.0", bgp.RF_IPv4_UC, nil)
|
||||
msg := &serverMsg{
|
||||
msgType: SRV_MSG_API,
|
||||
msgData: grpcReq,
|
||||
}
|
||||
|
||||
peer.serverMsgCh <- msg
|
||||
result := <-grpcReq.ResponseCh
|
||||
err := result.Data.(api.Error)
|
||||
assert.Equal(err.Code, api.Error_SUCCESS)
|
||||
|
||||
waitUntil(assert, bgp.BGP_FSM_IDLE, peer, 100)
|
||||
assert.Equal(bgp.BGP_FSM_IDLE, peer.fsm.state)
|
||||
assert.Equal(ADMIN_STATE_DOWN, peer.fsm.adminState)
|
||||
lastMsg := m.sendBuf[len(m.sendBuf)-1]
|
||||
sent, _ := bgp.ParseBGPMessage(lastMsg)
|
||||
assert.NotEqual(bgp.BGP_MSG_NOTIFICATION, sent.Header.Type)
|
||||
assert.True(m.isClosed)
|
||||
|
||||
// check counter
|
||||
counter := peer.fsm.peerConfig.BgpNeighborCommonState
|
||||
assertCounter(assert, counter)
|
||||
}
|
||||
|
||||
func TestPeerAdminShutdownWhileOpenconfirm(t *testing.T) {
|
||||
log.SetLevel(log.DebugLevel)
|
||||
assert := assert.New(t)
|
||||
m := NewMockConnection()
|
||||
globalConfig := config.Global{}
|
||||
peerConfig := config.Neighbor{}
|
||||
peerConfig.PeerAs = 100000
|
||||
peerConfig.Timers.KeepaliveInterval = 5
|
||||
peer := makePeer(globalConfig, peerConfig)
|
||||
peer.fsm.opensentHoldTime = 10
|
||||
peer.t.Go(peer.loop)
|
||||
pushPackets := func() {
|
||||
o, _ := open().Serialize()
|
||||
m.setData(o)
|
||||
}
|
||||
go pushPackets()
|
||||
waitUntil(assert, bgp.BGP_FSM_ACTIVE, peer, 1000)
|
||||
peer.connCh <- m
|
||||
waitUntil(assert, bgp.BGP_FSM_OPENCONFIRM, peer, 1000)
|
||||
|
||||
grpcReq := NewGrpcRequest(REQ_NEIGHBOR_DISABLE, "0.0.0.0", bgp.RF_IPv4_UC, nil)
|
||||
msg := &serverMsg{
|
||||
msgType: SRV_MSG_API,
|
||||
msgData: grpcReq,
|
||||
}
|
||||
|
||||
peer.serverMsgCh <- msg
|
||||
result := <-grpcReq.ResponseCh
|
||||
err := result.Data.(api.Error)
|
||||
assert.Equal(err.Code, api.Error_SUCCESS)
|
||||
|
||||
waitUntil(assert, bgp.BGP_FSM_IDLE, peer, 1000)
|
||||
assert.Equal(bgp.BGP_FSM_IDLE, peer.fsm.state)
|
||||
assert.Equal(ADMIN_STATE_DOWN, peer.fsm.adminState)
|
||||
lastMsg := m.sendBuf[len(m.sendBuf)-1]
|
||||
sent, _ := bgp.ParseBGPMessage(lastMsg)
|
||||
assert.NotEqual(bgp.BGP_MSG_NOTIFICATION, sent.Header.Type)
|
||||
assert.True(m.isClosed)
|
||||
|
||||
// check counter
|
||||
counter := peer.fsm.peerConfig.BgpNeighborCommonState
|
||||
assertCounter(assert, counter)
|
||||
|
||||
}
|
||||
|
||||
func TestPeerAdminEnable(t *testing.T) {
|
||||
log.SetLevel(log.DebugLevel)
|
||||
assert := assert.New(t)
|
||||
m := NewMockConnection()
|
||||
globalConfig := config.Global{}
|
||||
peerConfig := config.Neighbor{}
|
||||
peerConfig.PeerAs = 100000
|
||||
peerConfig.Timers.KeepaliveInterval = 5
|
||||
peer := makePeer(globalConfig, peerConfig)
|
||||
|
||||
peer.fsm.opensentHoldTime = 5
|
||||
peer.t.Go(peer.loop)
|
||||
pushPackets := func() {
|
||||
o, _ := open().Serialize()
|
||||
m.setData(o)
|
||||
k, _ := keepalive().Serialize()
|
||||
m.setData(k)
|
||||
}
|
||||
go pushPackets()
|
||||
|
||||
waitUntil(assert, bgp.BGP_FSM_ACTIVE, peer, 1000)
|
||||
peer.connCh <- m
|
||||
waitUntil(assert, bgp.BGP_FSM_ESTABLISHED, peer, 1000)
|
||||
|
||||
// shutdown peer at first
|
||||
grpcReq := NewGrpcRequest(REQ_NEIGHBOR_DISABLE, "0.0.0.0", bgp.RF_IPv4_UC, nil)
|
||||
msg := &serverMsg{
|
||||
msgType: SRV_MSG_API,
|
||||
msgData: grpcReq,
|
||||
}
|
||||
peer.serverMsgCh <- msg
|
||||
result := <-grpcReq.ResponseCh
|
||||
err := result.Data.(api.Error)
|
||||
assert.Equal(err.Code, api.Error_SUCCESS)
|
||||
|
||||
waitUntil(assert, bgp.BGP_FSM_IDLE, peer, 100)
|
||||
assert.Equal(bgp.BGP_FSM_IDLE, peer.fsm.state)
|
||||
assert.Equal(ADMIN_STATE_DOWN, peer.fsm.adminState)
|
||||
|
||||
// enable peer
|
||||
grpcReq = NewGrpcRequest(REQ_NEIGHBOR_ENABLE, "0.0.0.0", bgp.RF_IPv4_UC, nil)
|
||||
msg = &serverMsg{
|
||||
msgType: SRV_MSG_API,
|
||||
msgData: grpcReq,
|
||||
}
|
||||
peer.serverMsgCh <- msg
|
||||
result = <-grpcReq.ResponseCh
|
||||
err = result.Data.(api.Error)
|
||||
assert.Equal(err.Code, api.Error_SUCCESS)
|
||||
|
||||
waitUntil(assert, bgp.BGP_FSM_ACTIVE, peer, (HOLDTIME_IDLE+1)*1000)
|
||||
assert.Equal(bgp.BGP_FSM_ACTIVE, peer.fsm.state)
|
||||
|
||||
m2 := NewMockConnection()
|
||||
pushPackets = func() {
|
||||
o, _ := open().Serialize()
|
||||
m2.setData(o)
|
||||
k, _ := keepalive().Serialize()
|
||||
m2.setData(k)
|
||||
}
|
||||
go pushPackets()
|
||||
|
||||
peer.connCh <- m2
|
||||
|
||||
waitUntil(assert, bgp.BGP_FSM_ESTABLISHED, peer, 1000)
|
||||
assert.Equal(bgp.BGP_FSM_ESTABLISHED, peer.fsm.state)
|
||||
}
|
||||
|
||||
func TestPeerAdminShutdownReject(t *testing.T) {
|
||||
log.SetLevel(log.DebugLevel)
|
||||
assert := assert.New(t)
|
||||
m := NewMockConnection()
|
||||
m.wait = 500
|
||||
|
||||
globalConfig := config.Global{}
|
||||
peerConfig := config.Neighbor{}
|
||||
peerConfig.PeerAs = 100000
|
||||
peerConfig.Timers.KeepaliveInterval = 5
|
||||
peer := makePeer(globalConfig, peerConfig)
|
||||
peer.fsm.opensentHoldTime = 1
|
||||
peer.t.Go(peer.loop)
|
||||
|
||||
waitUntil(assert, bgp.BGP_FSM_ACTIVE, peer, 1000)
|
||||
peer.connCh <- m
|
||||
waitUntil(assert, bgp.BGP_FSM_OPENSENT, peer, 1000)
|
||||
|
||||
grpcReq := NewGrpcRequest(REQ_NEIGHBOR_DISABLE, "0.0.0.0", bgp.RF_IPv4_UC, nil)
|
||||
msg := &serverMsg{
|
||||
msgType: SRV_MSG_API,
|
||||
msgData: grpcReq,
|
||||
}
|
||||
|
||||
peer.fsm.adminStateCh <- ADMIN_STATE_DOWN
|
||||
|
||||
peer.serverMsgCh <- msg
|
||||
result := <-grpcReq.ResponseCh
|
||||
err := result.Data.(api.Error)
|
||||
assert.Equal(err.Code, api.Error_FAIL)
|
||||
|
||||
grpcReq = NewGrpcRequest(REQ_NEIGHBOR_ENABLE, "0.0.0.0", bgp.RF_IPv4_UC, nil)
|
||||
msg = &serverMsg{
|
||||
msgType: SRV_MSG_API,
|
||||
msgData: grpcReq,
|
||||
}
|
||||
|
||||
peer.serverMsgCh <- msg
|
||||
result = <-grpcReq.ResponseCh
|
||||
err = result.Data.(api.Error)
|
||||
assert.Equal(err.Code, api.Error_FAIL)
|
||||
|
||||
waitUntil(assert, bgp.BGP_FSM_IDLE, peer, 1000)
|
||||
assert.Equal(bgp.BGP_FSM_IDLE, peer.fsm.state)
|
||||
assert.Equal(ADMIN_STATE_DOWN, peer.fsm.adminState)
|
||||
|
||||
}
|
||||
|
||||
func TestPeerSelectSmallerHoldtime(t *testing.T) {
|
||||
log.SetLevel(log.DebugLevel)
|
||||
assert := assert.New(t)
|
||||
m := NewMockConnection()
|
||||
|
||||
globalConfig := config.Global{}
|
||||
peerConfig := config.Neighbor{}
|
||||
peerConfig.PeerAs = 65001
|
||||
peerConfig.Timers.KeepaliveInterval = 5
|
||||
peer := makePeer(globalConfig, peerConfig)
|
||||
peer.fsm.opensentHoldTime = 1
|
||||
peerConfig.Timers.HoldTime = 5
|
||||
peer.t.Go(peer.loop)
|
||||
|
||||
pushPackets := func() {
|
||||
opn := bgp.NewBGPOpenMessage(65001, 0, "10.0.0.1", []bgp.OptionParameterInterface{})
|
||||
o, _ := opn.Serialize()
|
||||
m.setData(o)
|
||||
}
|
||||
go pushPackets()
|
||||
|
||||
waitUntil(assert, bgp.BGP_FSM_ACTIVE, peer, 1000)
|
||||
peer.connCh <- m
|
||||
waitUntil(assert, bgp.BGP_FSM_OPENCONFIRM, peer, 1000)
|
||||
|
||||
assert.Equal(float64(0), peer.fsm.negotiatedHoldTime)
|
||||
}
|
||||
|
||||
func assertCounter(assert *assert.Assertions, counter config.BgpNeighborCommonState) {
|
||||
assert.Equal(uint32(0), counter.OpenIn)
|
||||
assert.Equal(uint32(0), counter.OpenOut)
|
||||
assert.Equal(uint32(0), counter.UpdateIn)
|
||||
assert.Equal(uint32(0), counter.UpdateOut)
|
||||
assert.Equal(uint32(0), counter.KeepaliveIn)
|
||||
assert.Equal(uint32(0), counter.KeepaliveOut)
|
||||
assert.Equal(uint32(0), counter.NotifyIn)
|
||||
assert.Equal(uint32(0), counter.NotifyOut)
|
||||
assert.Equal(uint32(0), counter.EstablishedCount)
|
||||
assert.Equal(uint32(0), counter.TotalIn)
|
||||
assert.Equal(uint32(0), counter.TotalOut)
|
||||
assert.Equal(uint32(0), counter.RefreshIn)
|
||||
assert.Equal(uint32(0), counter.RefreshOut)
|
||||
assert.Equal(uint32(0), counter.DynamicCapIn)
|
||||
assert.Equal(uint32(0), counter.DynamicCapOut)
|
||||
assert.Equal(uint32(0), counter.EstablishedCount)
|
||||
assert.Equal(uint32(0), counter.Flops)
|
||||
}
|
||||
|
||||
func waitUntil(assert *assert.Assertions, state bgp.FSMState, peer *Peer, timeout int64) {
|
||||
isTimeout := false
|
||||
expire := func() {
|
||||
isTimeout = true
|
||||
}
|
||||
time.AfterFunc((time.Duration)(timeout)*time.Millisecond, expire)
|
||||
|
||||
for {
|
||||
time.Sleep(1 * time.Millisecond)
|
||||
|
||||
if peer.fsm.state == state || isTimeout {
|
||||
assert.Equal(state, peer.fsm.state, "timeout")
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func makePeer(globalConfig config.Global, peerConfig config.Neighbor) *Peer {
|
||||
|
||||
sch := make(chan *serverMsg, 8)
|
||||
pch := make(chan *peerMsg, 4096)
|
||||
|
||||
p := &Peer{
|
||||
globalConfig: globalConfig,
|
||||
peerConfig: peerConfig,
|
||||
connCh: make(chan net.Conn),
|
||||
serverMsgCh: sch,
|
||||
peerMsgCh: pch,
|
||||
getActiveCh: make(chan struct{}),
|
||||
rfMap: make(map[bgp.RouteFamily]bool),
|
||||
capMap: make(map[bgp.BGPCapabilityCode]bgp.ParameterCapabilityInterface),
|
||||
}
|
||||
p.siblings = make(map[string]*serverMsgDataPeer)
|
||||
|
||||
p.fsm = NewFSM(&globalConfig, &peerConfig, p.connCh)
|
||||
peerConfig.BgpNeighborCommonState.State = uint32(bgp.BGP_FSM_IDLE)
|
||||
peerConfig.BgpNeighborCommonState.Downtime = time.Now().Unix()
|
||||
if peerConfig.NeighborAddress.To4() != nil {
|
||||
p.rfMap[bgp.RF_IPv4_UC] = true
|
||||
} else {
|
||||
p.rfMap[bgp.RF_IPv6_UC] = true
|
||||
}
|
||||
|
||||
p.peerInfo = &table.PeerInfo{
|
||||
AS: peerConfig.PeerAs,
|
||||
LocalID: globalConfig.RouterId,
|
||||
Address: peerConfig.NeighborAddress,
|
||||
}
|
||||
rfList := []bgp.RouteFamily{bgp.RF_IPv4_UC, bgp.RF_IPv6_UC}
|
||||
p.adjRib = table.NewAdjRib(rfList)
|
||||
p.rib = table.NewTableManager(p.peerConfig.NeighborAddress.String(), rfList)
|
||||
p.t.Go(p.connectLoop)
|
||||
|
||||
return p
|
||||
}
|
||||
+857
-152
File diff suppressed because it is too large
Load Diff
@@ -56,7 +56,7 @@ type Destination interface {
|
||||
setNlri(nlri bgp.AddrPrefixInterface)
|
||||
getBestPathReason() string
|
||||
setBestPathReason(string)
|
||||
getBestPath() Path
|
||||
GetBestPath() Path
|
||||
setBestPath(path Path)
|
||||
getKnownPathList() []Path
|
||||
setKnownPathList([]Path)
|
||||
@@ -102,7 +102,7 @@ func (dd *DestinationDefault) ToApiStruct() *api.Destination {
|
||||
|
||||
idx := func() int {
|
||||
for i, p := range dd.knownPathList {
|
||||
if p == dd.getBestPath() {
|
||||
if p == dd.GetBestPath() {
|
||||
return i
|
||||
}
|
||||
}
|
||||
@@ -152,7 +152,7 @@ func (dd *DestinationDefault) setBestPathReason(reason string) {
|
||||
dd.bestPathReason = reason
|
||||
}
|
||||
|
||||
func (dd *DestinationDefault) getBestPath() Path {
|
||||
func (dd *DestinationDefault) GetBestPath() Path {
|
||||
return dd.bestPath
|
||||
}
|
||||
|
||||
@@ -906,7 +906,7 @@ func (ipv6d *IPv6Destination) MarshalJSON() ([]byte, error) {
|
||||
prefix := ipv6d.getNlri().(*bgp.IPv6AddrPrefix).Prefix
|
||||
idx := func() int {
|
||||
for i, p := range ipv6d.DestinationDefault.knownPathList {
|
||||
if p == ipv6d.DestinationDefault.getBestPath() {
|
||||
if p == ipv6d.DestinationDefault.GetBestPath() {
|
||||
return i
|
||||
}
|
||||
}
|
||||
@@ -962,7 +962,7 @@ func (ipv4vpnd *IPv4VPNDestination) MarshalJSON() ([]byte, error) {
|
||||
prefix := ipv4vpnd.getNlri().(*bgp.LabelledVPNIPAddrPrefix).Prefix
|
||||
idx := func() int {
|
||||
for i, p := range ipv4vpnd.DestinationDefault.knownPathList {
|
||||
if p == ipv4vpnd.DestinationDefault.getBestPath() {
|
||||
if p == ipv4vpnd.DestinationDefault.GetBestPath() {
|
||||
return i
|
||||
}
|
||||
}
|
||||
@@ -1000,7 +1000,7 @@ func (evpnd *EVPNDestination) MarshalJSON() ([]byte, error) {
|
||||
nlri := evpnd.getNlri().(*bgp.EVPNNLRI)
|
||||
idx := func() int {
|
||||
for i, p := range evpnd.DestinationDefault.knownPathList {
|
||||
if p == evpnd.DestinationDefault.getBestPath() {
|
||||
if p == evpnd.DestinationDefault.GetBestPath() {
|
||||
return i
|
||||
}
|
||||
}
|
||||
|
||||
@@ -82,7 +82,7 @@ func TestDestinationSetBestPath(t *testing.T) {
|
||||
pathD := DestCreatePath(peerD)
|
||||
ipv4d := NewIPv4Destination(pathD[0].GetNlri())
|
||||
ipv4d.setBestPath(pathD[0])
|
||||
r_pathD := ipv4d.getBestPath()
|
||||
r_pathD := ipv4d.GetBestPath()
|
||||
assert.Equal(t, r_pathD, pathD[0])
|
||||
}
|
||||
func TestDestinationGetBestPath(t *testing.T) {
|
||||
@@ -90,7 +90,7 @@ func TestDestinationGetBestPath(t *testing.T) {
|
||||
pathD := DestCreatePath(peerD)
|
||||
ipv4d := NewIPv4Destination(pathD[0].GetNlri())
|
||||
ipv4d.setBestPath(pathD[0])
|
||||
r_pathD := ipv4d.getBestPath()
|
||||
r_pathD := ipv4d.GetBestPath()
|
||||
assert.Equal(t, r_pathD, pathD[0])
|
||||
}
|
||||
func TestDestinationCalculate(t *testing.T) {
|
||||
|
||||
+2
-2
@@ -152,7 +152,7 @@ func TestASPathLen(t *testing.T) {
|
||||
bgp.NewAsPathParam(bgp.BGP_ASPATH_ATTR_TYPE_SEQ, []uint16{65001, 65002, 65003, 65004, 65004, 65004, 65004, 65004, 65005}),
|
||||
bgp.NewAsPathParam(bgp.BGP_ASPATH_ATTR_TYPE_SET, []uint16{65001, 65002, 65003, 65004, 65005}),
|
||||
bgp.NewAsPathParam(bgp.BGP_ASPATH_ATTR_TYPE_CONFED_SEQ, []uint16{65100, 65101, 65102}),
|
||||
bgp.NewAsPathParam(bgp.BGP_ASPATH_ATTR_TYPE_CONFED_SET, []uint16{65100, 65101}),}
|
||||
bgp.NewAsPathParam(bgp.BGP_ASPATH_ATTR_TYPE_CONFED_SET, []uint16{65100, 65101})}
|
||||
aspath := bgp.NewPathAttributeAsPath(aspathParam)
|
||||
nexthop := bgp.NewPathAttributeNextHop("192.168.50.1")
|
||||
med := bgp.NewPathAttributeMultiExitDisc(0)
|
||||
@@ -170,7 +170,7 @@ func TestASPathLen(t *testing.T) {
|
||||
update := bgpmsg.Body.(*bgp.BGPUpdate)
|
||||
UpdatePathAttrs4ByteAs(update)
|
||||
peer := PathCreatePeer()
|
||||
p, _:= CreatePath(peer[0], &update.NLRI[0], update.PathAttributes, false, time.Now())
|
||||
p, _ := CreatePath(peer[0], &update.NLRI[0], update.PathAttributes, false, time.Now())
|
||||
assert.Equal(10, p.GetAsPathLen())
|
||||
}
|
||||
|
||||
|
||||
+10
-5
@@ -139,6 +139,10 @@ func NewTableManager(owner string, rfList []bgp.RouteFamily) *TableManager {
|
||||
return t
|
||||
}
|
||||
|
||||
func (manager *TableManager) OwnerName() string {
|
||||
return manager.owner
|
||||
}
|
||||
|
||||
func (manager *TableManager) calculate(destinationList []Destination) ([]Path, error) {
|
||||
newPaths := make([]Path, 0)
|
||||
|
||||
@@ -159,7 +163,7 @@ func (manager *TableManager) calculate(destinationList []Destination) ([]Path, e
|
||||
}
|
||||
|
||||
destination.setBestPathReason(reason)
|
||||
currentBestPath := destination.getBestPath()
|
||||
currentBestPath := destination.GetBestPath()
|
||||
|
||||
if newBestPath != nil && currentBestPath == newBestPath {
|
||||
// best path is not changed
|
||||
@@ -192,7 +196,7 @@ func (manager *TableManager) calculate(destinationList []Destination) ([]Path, e
|
||||
"next_hop": currentBestPath.GetNexthop().String(),
|
||||
}).Debug("best path is lost")
|
||||
|
||||
p := destination.getBestPath()
|
||||
p := destination.GetBestPath()
|
||||
newPaths = append(newPaths, p.Clone(true))
|
||||
}
|
||||
destination.setBestPath(nil)
|
||||
@@ -218,7 +222,7 @@ func (manager *TableManager) calculate(destinationList []Destination) ([]Path, e
|
||||
destination.setBestPath(newBestPath)
|
||||
}
|
||||
|
||||
if len(destination.getKnownPathList()) == 0 && destination.getBestPath() == nil {
|
||||
if len(destination.getKnownPathList()) == 0 && destination.GetBestPath() == nil {
|
||||
rf := destination.getRouteFamily()
|
||||
t := manager.Tables[rf]
|
||||
deleteDest(t, destination)
|
||||
@@ -259,7 +263,7 @@ func (manager *TableManager) GetPathList(rf bgp.RouteFamily) []Path {
|
||||
}
|
||||
var paths []Path
|
||||
for _, dest := range manager.Tables[rf].GetDestinations() {
|
||||
paths = append(paths, dest.getBestPath())
|
||||
paths = append(paths, dest.GetBestPath())
|
||||
}
|
||||
return paths
|
||||
}
|
||||
@@ -360,10 +364,11 @@ func (adj *AdjRib) GetOutCount(rf bgp.RouteFamily) int {
|
||||
return len(adj.adjRibOut[rf])
|
||||
}
|
||||
|
||||
func (adj *AdjRib) DropAllIn(rf bgp.RouteFamily) {
|
||||
func (adj *AdjRib) DropAll(rf bgp.RouteFamily) {
|
||||
if _, ok := adj.adjRibIn[rf]; ok {
|
||||
// replace old one
|
||||
adj.adjRibIn[rf] = make(map[string]*ReceivedRoute)
|
||||
adj.adjRibOut[rf] = make(map[string]*ReceivedRoute)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user