From 411bf84633c6a88fa1d08b5238d74b55910cffd1 Mon Sep 17 00:00:00 2001 From: FUJITA Tomonori Date: Thu, 4 Aug 2016 17:05:57 +0900 Subject: support mrt table dump (TABLE_DUMPv2 format) Only global rib dumping is supported for now. Signed-off-by: FUJITA Tomonori --- server/mrt.go | 226 +++++++++++++++++++++++++++++++++++++++++++------------ server/server.go | 59 +++++++++------ 2 files changed, 212 insertions(+), 73 deletions(-) (limited to 'server') diff --git a/server/mrt.go b/server/mrt.go index 9241eabd..605d6fe0 100644 --- a/server/mrt.go +++ b/server/mrt.go @@ -17,19 +17,25 @@ package server import ( "bytes" + "fmt" "os" "time" log "github.com/Sirupsen/logrus" + "github.com/osrg/gobgp/config" + "github.com/osrg/gobgp/packet/bgp" "github.com/osrg/gobgp/packet/mrt" + "github.com/osrg/gobgp/table" ) type mrtWriter struct { - dead chan struct{} - s *BgpServer - filename string - file *os.File - interval uint64 + dead chan struct{} + s *BgpServer + filename string + file *os.File + rotationInterval uint64 + dumpInterval uint64 + dumpType config.MrtType } func (m *mrtWriter) Stop() { @@ -37,43 +43,111 @@ func (m *mrtWriter) Stop() { } func (m *mrtWriter) loop() error { - w := m.s.Watch(WatchUpdate(false)) - c := func() *time.Ticker { - if m.interval == 0 { + ops := []WatchOption{} + switch m.dumpType { + case config.MRT_TYPE_UPDATES: + ops = append(ops, WatchUpdate(false)) + case config.MRT_TYPE_TABLE: + } + w := m.s.Watch(ops...) + rotator := func() *time.Ticker { + if m.rotationInterval == 0 { + return &time.Ticker{} + } + return time.NewTicker(time.Second * time.Duration(m.rotationInterval)) + }() + dump := func() *time.Ticker { + if m.dumpInterval == 0 { return &time.Ticker{} } - return time.NewTicker(time.Second * time.Duration(m.interval)) + return time.NewTicker(time.Second * time.Duration(m.dumpInterval)) }() defer func() { if m.file != nil { m.file.Close() } - if m.interval != 0 { - c.Stop() + if m.rotationInterval != 0 { + rotator.Stop() + } + if m.dumpInterval == 0 { + dump.Stop() } w.Stop() }() for { - serialize := func(ev WatchEvent) ([]byte, error) { - m := ev.(*WatchEventUpdate) - subtype := mrt.MESSAGE_AS4 - mp := mrt.NewBGP4MPMessage(m.PeerAS, m.LocalAS, 0, m.PeerAddress.String(), m.LocalAddress.String(), m.FourBytesAs, nil) - mp.BGPMessagePayload = m.Payload - if m.FourBytesAs == false { - subtype = mrt.MESSAGE - } - bm, err := mrt.NewMRTMessage(uint32(m.Timestamp.Unix()), mrt.BGP4MP, subtype, mp) - if err != nil { - log.WithFields(log.Fields{ - "Topic": "mrt", - "Data": m, - "Error": err, - }).Warn("Failed to create MRT message in serialize()") - return nil, err + serialize := func(ev WatchEvent) []*mrt.MRTMessage { + msg := make([]*mrt.MRTMessage, 0, 1) + switch m := ev.(type) { + case *WatchEventUpdate: + subtype := mrt.MESSAGE_AS4 + mp := mrt.NewBGP4MPMessage(m.PeerAS, m.LocalAS, 0, m.PeerAddress.String(), m.LocalAddress.String(), m.FourBytesAs, nil) + mp.BGPMessagePayload = m.Payload + if m.FourBytesAs == false { + subtype = mrt.MESSAGE + } + if bm, err := mrt.NewMRTMessage(uint32(m.Timestamp.Unix()), mrt.BGP4MP, subtype, mp); err != nil { + log.WithFields(log.Fields{ + "Topic": "mrt", + "Data": m, + "Error": err, + }).Warn("Failed to create MRT message in serialize()") + } else { + msg = append(msg, bm) + } + case *WatchEventTable: + t := uint32(time.Now().Unix()) + peers := make([]*mrt.Peer, 0, len(m.Neighbor)) + for _, pconf := range m.Neighbor { + peers = append(peers, mrt.NewPeer(pconf.State.Description, pconf.Config.NeighborAddress, pconf.Config.PeerAs, true)) + } + if bm, err := mrt.NewMRTMessage(t, mrt.TABLE_DUMPv2, mrt.PEER_INDEX_TABLE, mrt.NewPeerIndexTable(m.RouterId, "", peers)); err != nil { + break + } else { + msg = append(msg, bm) + } + + idx := func(p *table.Path) uint16 { + for i, pconf := range m.Neighbor { + if p.GetSource().Address.String() == pconf.Config.NeighborAddress { + return uint16(i) + } + } + return uint16(len(m.Neighbor)) + } + + subtype := func(p *table.Path) mrt.MRTSubTypeTableDumpv2 { + switch p.GetRouteFamily() { + case bgp.RF_IPv4_UC: + return mrt.RIB_IPV4_UNICAST + case bgp.RF_IPv4_MC: + return mrt.RIB_IPV4_MULTICAST + case bgp.RF_IPv6_UC: + return mrt.RIB_IPV6_UNICAST + case bgp.RF_IPv6_MC: + return mrt.RIB_IPV6_MULTICAST + } + return mrt.RIB_GENERIC + } + + seq := uint32(0) + for _, pathList := range m.PathList { + entries := make([]*mrt.RibEntry, 0, len(pathList)) + for _, path := range pathList { + if path.IsLocal() { + continue + } + entries = append(entries, mrt.NewRibEntry(idx(path), uint32(path.GetTimestamp().Unix()), path.GetPathAttrs())) + } + if len(entries) > 0 { + bm, _ := mrt.NewMRTMessage(t, mrt.TABLE_DUMPv2, subtype(pathList[0]), mrt.NewRib(seq, pathList[0].GetNlri(), entries)) + msg = append(msg, bm) + seq++ + } + } } - return bm.Serialize() + return msg } drain := func(ev WatchEvent) { @@ -99,19 +173,20 @@ func (m *mrtWriter) loop() error { var b bytes.Buffer for _, e := range events { - buf, err := serialize(e) - if err != nil { - log.WithFields(log.Fields{ - "Topic": "mrt", - "Data": e, - "Error": err, - }).Warn("Failed to serialize event") - continue - } - b.Write(buf) - if b.Len() > 1*1000*1000 { - w(b.Bytes()) - b.Reset() + for _, m := range serialize(e) { + if buf, err := m.Serialize(); err != nil { + log.WithFields(log.Fields{ + "Topic": "mrt", + "Data": e, + "Error": err, + }).Warn("Failed to serialize event") + } else { + b.Write(buf) + if b.Len() > 1*1000*1000 { + w(b.Bytes()) + b.Reset() + } + } } } if b.Len() > 0 { @@ -124,9 +199,9 @@ func (m *mrtWriter) loop() error { return nil case e := <-w.Event(): drain(e) - case <-c.C: + case <-rotator.C: m.file.Close() - file, err := mrtFileOpen(m.filename, m.interval) + file, err := mrtFileOpen(m.filename, m.rotationInterval) if err == nil { m.file = file } else { @@ -135,6 +210,8 @@ func (m *mrtWriter) loop() error { "Error": err, }).Warn("can't rotate MRT file") } + case <-dump.C: + w.Generate(WATCH_EVENT_TYPE_TABLE) } } } @@ -181,17 +258,68 @@ func mrtFileOpen(filename string, interval uint64) (*os.File, error) { return file, err } -func newMrtWriter(s *BgpServer, dumpType int, filename string, interval uint64) (*mrtWriter, error) { - file, err := mrtFileOpen(filename, interval) +func newMrtWriter(s *BgpServer, dumpType config.MrtType, filename string, rInterval, dInterval uint64) (*mrtWriter, error) { + file, err := mrtFileOpen(filename, rInterval) if err != nil { return nil, err } m := mrtWriter{ - s: s, - filename: filename, - file: file, - interval: interval, + dumpType: dumpType, + s: s, + filename: filename, + file: file, + rotationInterval: rInterval, + dumpInterval: dInterval, } go m.loop() return &m, nil } + +type mrtManager struct { + bgpServer *BgpServer + writer map[string]*mrtWriter +} + +func (m *mrtManager) enable(c *config.MrtConfig) error { + if _, ok := m.writer[c.FileName]; ok { + return fmt.Errorf("%s already exists", c.FileName) + } + + rInterval := c.RotationInterval + if rInterval != 0 && rInterval < 30 { + log.Info("minimum mrt dump interval is 30 seconds") + rInterval = 30 + } + dInterval := c.DumpInterval + if c.DumpType == config.MRT_TYPE_TABLE { + if dInterval < 60 { + log.Info("minimum mrt dump interval is 30 seconds") + dInterval = 60 + } + } else if c.DumpType == config.MRT_TYPE_UPDATES { + dInterval = 0 + } + + w, err := newMrtWriter(m.bgpServer, c.DumpType, c.FileName, rInterval, dInterval) + if err == nil { + m.writer[c.FileName] = w + } + return err +} + +func (m *mrtManager) disable(c *config.MrtConfig) error { + if w, ok := m.writer[c.FileName]; !ok { + return fmt.Errorf("%s doesn't exists", c.FileName) + } else { + w.Stop() + delete(m.writer, c.FileName) + } + return nil +} + +func newMrtManager(s *BgpServer) *mrtManager { + return &mrtManager{ + bgpServer: s, + writer: make(map[string]*mrtWriter), + } +} diff --git a/server/server.go b/server/server.go index ae036a83..c0ac7c7c 100644 --- a/server/server.go +++ b/server/server.go @@ -104,7 +104,7 @@ type BgpServer struct { watcherMap map[WatchEventType][]*Watcher zclient *zebraClient bmpManager *bmpClientManager - mrt *mrtWriter + mrtManager *mrtManager } func NewBgpServer() *BgpServer { @@ -117,6 +117,7 @@ func NewBgpServer() *BgpServer { watcherMap: make(map[WatchEventType][]*Watcher), } s.bmpManager = newBmpClientManager(s) + s.mrtManager = newMrtManager(s) return s } @@ -2206,40 +2207,26 @@ func (s *BgpServer) ReplacePolicyAssignment(name string, dir table.PolicyDirecti return err } -func (s *BgpServer) EnableMrt(c *config.Mrt) (err error) { +func (s *BgpServer) EnableMrt(c *config.MrtConfig) (err error) { ch := make(chan struct{}) defer func() { <-ch }() s.mgmtCh <- func() { defer close(ch) - if s.mrt != nil { - err = fmt.Errorf("already enabled") - } else { - interval := c.Interval - - if interval != 0 && interval < 30 { - log.Info("minimum mrt dump interval is 30 seconds") - interval = 30 - } - s.mrt, err = newMrtWriter(s, c.DumpType.ToInt(), c.FileName, interval) - } + err = s.mrtManager.enable(c) } return err } -func (s *BgpServer) DisableMrt() (err error) { +func (s *BgpServer) DisableMrt(c *config.MrtConfig) (err error) { ch := make(chan struct{}) defer func() { <-ch }() s.mgmtCh <- func() { defer close(ch) - if s.mrt != nil { - s.mrt.Stop() - } else { - err = fmt.Errorf("not enabled") - } + err = s.mrtManager.disable(c) } return err } @@ -2372,6 +2359,7 @@ const ( WATCH_EVENT_TYPE_PRE_UPDATE WatchEventType = "preupdate" WATCH_EVENT_TYPE_POST_UPDATE WatchEventType = "postupdate" WATCH_EVENT_TYPE_PEER_STATE WatchEventType = "peerstate" + WATCH_EVENT_TYPE_TABLE WatchEventType = "table" ) type WatchEvent interface { @@ -2410,6 +2398,12 @@ type WatchEventAdjIn struct { PathList []*table.Path } +type WatchEventTable struct { + RouterId string + PathList map[string][]*table.Path + Neighbor []*config.Neighbor +} + type WatchEventBestPath struct { PathList []*table.Path MultiPathList [][]*table.Path @@ -2480,15 +2474,32 @@ func (w *Watcher) Generate(t WatchEventType) (err error) { switch t { case WATCH_EVENT_TYPE_PRE_UPDATE: + pathList := make([]*table.Path, 0) + for _, peer := range w.s.neighborMap { + pathList = append(pathList, peer.adjRibIn.PathList(peer.configuredRFlist(), false)...) + } + w.notify(&WatchEventAdjIn{PathList: clonePathList(pathList)}) + case WATCH_EVENT_TYPE_TABLE: + pathList := func() map[string][]*table.Path { + pathList := make(map[string][]*table.Path) + for _, t := range w.s.globalRib.Tables { + for _, dst := range t.GetSortedDestinations() { + if paths := dst.GetKnownPathList(table.GLOBAL_RIB_NAME); len(paths) > 0 { + pathList[dst.GetNlri().String()] = clonePathList(paths) + } + } + } + return pathList + }() + l := make([]*config.Neighbor, 0, len(w.s.neighborMap)) + for _, peer := range w.s.neighborMap { + l = append(l, peer.ToConfig()) + } + w.notify(&WatchEventTable{PathList: pathList, Neighbor: l}) default: err = fmt.Errorf("unsupported type ", t) return } - pathList := make([]*table.Path, 0) - for _, peer := range w.s.neighborMap { - pathList = append(pathList, peer.adjRibIn.PathList(peer.configuredRFlist(), false)...) - } - w.notify(&WatchEventAdjIn{PathList: clonePathList(pathList)}) } return err } -- cgit v1.2.3