summaryrefslogtreecommitdiffhomepage
path: root/server
diff options
context:
space:
mode:
Diffstat (limited to 'server')
-rw-r--r--server/dumper.go71
-rw-r--r--server/server.go27
-rw-r--r--server/watcher.go145
3 files changed, 165 insertions, 78 deletions
diff --git a/server/dumper.go b/server/dumper.go
deleted file mode 100644
index 46281ca6..00000000
--- a/server/dumper.go
+++ /dev/null
@@ -1,71 +0,0 @@
-// Copyright (C) 2015 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 (
- log "github.com/Sirupsen/logrus"
- "github.com/osrg/gobgp/packet"
- "os"
- "time"
-)
-
-type dumper struct {
- ch chan *broadcastBGPMsg
-}
-
-func (d *dumper) sendCh() chan *broadcastBGPMsg {
- return d.ch
-}
-
-func newDumper(filename string) (*dumper, error) {
- f, err := os.Create(filename)
- if err != nil {
- return nil, err
- }
-
- ch := make(chan *broadcastBGPMsg, 16)
-
- go func() {
- for {
- m := <-ch
- subtype := bgp.MESSAGE_AS4
- mp := bgp.NewBGP4MPMessage(m.peerAS, m.localAS, 0, m.peerAddress.String(), m.localAddress.String(), m.fourBytesAs, m.message)
- if m.fourBytesAs == false {
- subtype = bgp.MESSAGE
- }
- bm, err := bgp.NewMRTMessage(uint32(time.Now().Unix()), bgp.BGP4MP, subtype, mp)
- if err != nil {
- log.WithFields(log.Fields{
- "Topic": "mrt",
- "Data": m,
- }).Warn(err)
- continue
- }
- buf, err := bm.Serialize()
- if err != nil {
- log.WithFields(log.Fields{
- "Topic": "mrt",
- "Data": m,
- }).Warn(err)
- } else {
- f.Write(buf)
- }
- }
- }()
- return &dumper{
- ch: ch,
- }, nil
-}
diff --git a/server/server.go b/server/server.go
index 1e54da39..8a5cabdf 100644
--- a/server/server.go
+++ b/server/server.go
@@ -87,7 +87,6 @@ type BgpServer struct {
fsmincomingCh chan *FsmMsg
rpkiConfigCh chan config.RpkiServers
bmpConfigCh chan config.BmpServers
- dumper *dumper
GrpcReqCh chan *GrpcRequest
listenPort int
@@ -103,6 +102,7 @@ type BgpServer struct {
bmpClient *bmpClient
bmpConnCh chan *bmpConn
shutdown bool
+ watchers map[watcherType]watcher
}
func NewBgpServer(port int) *BgpServer {
@@ -118,6 +118,7 @@ func NewBgpServer(port int) *BgpServer {
b.policyUpdateCh = make(chan config.RoutingPolicy)
b.neighborMap = make(map[string]*Peer)
b.listenPort = port
+ b.watchers = make(map[watcherType]watcher)
return &b
}
@@ -163,11 +164,11 @@ func (server *BgpServer) Serve() {
server.roaClient, _ = newROAClient(g.GlobalConfig.As, config.RpkiServers{})
if g.Mrt.FileName != "" {
- d, err := newDumper(g.Mrt.FileName)
+ w, err := newMrtWatcher(g.Mrt.FileName)
if err != nil {
log.Warn(err)
} else {
- server.dumper = d
+ server.watchers[WATCHER_MRT] = w
}
}
@@ -799,20 +800,32 @@ func (server *BgpServer) handleFSMMessage(peer *Peer, e *FsmMsg, incoming chan *
}
}
if m.Header.Type == bgp.BGP_MSG_UPDATE {
- if server.dumper != nil {
+ listener := make(map[watcher]chan watcherEvent)
+ for _, watcher := range server.watchers {
+ if ch := watcher.notify(WATCHER_EVENT_UPDATE_MSG); ch != nil {
+ listener[watcher] = ch
+ }
+ }
+ if len(listener) > 0 {
_, y := peer.capMap[bgp.BGP_CAP_FOUR_OCTET_AS_NUMBER]
l, _ := peer.fsm.LocalHostPort()
- bm := &broadcastBGPMsg{
+ ev := &watcherEventUpdateMsg{
message: m,
peerAS: peer.fsm.peerInfo.AS,
localAS: peer.fsm.peerInfo.LocalAS,
peerAddress: peer.fsm.peerInfo.Address,
localAddress: net.ParseIP(l),
fourBytesAs: y,
- ch: server.dumper.sendCh(),
}
- server.broadcastMsgs = append(server.broadcastMsgs, bm)
+ for _, ch := range listener {
+ bm := &broadcastWatcherMsg{
+ ch: ch,
+ event: ev,
+ }
+ server.broadcastMsgs = append(server.broadcastMsgs, bm)
+ }
}
+
if ch := server.bmpClient.send(); ch != nil {
bm := &broadcastBMPMsg{
ch: ch,
diff --git a/server/watcher.go b/server/watcher.go
new file mode 100644
index 00000000..a4c3a192
--- /dev/null
+++ b/server/watcher.go
@@ -0,0 +1,145 @@
+// Copyright (C) 2015 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 (
+ log "github.com/Sirupsen/logrus"
+ "github.com/osrg/gobgp/packet"
+ "gopkg.in/tomb.v2"
+ "net"
+ "os"
+ "time"
+)
+
+type broadcastWatcherMsg struct {
+ ch chan watcherEvent
+ event watcherEvent
+}
+
+func (m *broadcastWatcherMsg) send() {
+ m.ch <- m.event
+}
+
+type watcherType uint8
+
+const (
+ _ watcherType = iota
+ WATCHER_MRT // UPDATE MSG
+ WATCHER_BMP
+ WATCHER_ZEBRA
+ WATCHER_GRPC_BESTPATH
+)
+
+type watcherEventType uint8
+
+const (
+ _ watcherEventType = iota
+ WATCHER_EVENT_UPDATE_MSG
+ WATCHER_EVENT_STATE_CHANGE
+ WATCHER_EVENT_BESTPATH_CHANGE
+)
+
+type watcherEvent interface {
+}
+
+type watcherEventUpdateMsg struct {
+ message *bgp.BGPMessage
+ peerAS uint32
+ localAS uint32
+ peerAddress net.IP
+ localAddress net.IP
+ fourBytesAs bool
+}
+
+type watcher interface {
+ notify(watcherEventType) chan watcherEvent
+ stop()
+}
+
+type mrtWatcher struct {
+ t tomb.Tomb
+ filename string
+ file *os.File
+ ch chan watcherEvent
+}
+
+func (w *mrtWatcher) notify(t watcherEventType) chan watcherEvent {
+ if t == WATCHER_EVENT_UPDATE_MSG {
+ return w.ch
+ }
+ return nil
+}
+
+func (w *mrtWatcher) stop() {
+ w.t.Kill(nil)
+}
+
+func (w *mrtWatcher) loop() error {
+ for {
+ write := func(ev watcherEvent) {
+ m := ev.(*watcherEventUpdateMsg)
+ subtype := bgp.MESSAGE_AS4
+ mp := bgp.NewBGP4MPMessage(m.peerAS, m.localAS, 0, m.peerAddress.String(), m.localAddress.String(), m.fourBytesAs, m.message)
+ if m.fourBytesAs == false {
+ subtype = bgp.MESSAGE
+ }
+ bm, err := bgp.NewMRTMessage(uint32(time.Now().Unix()), bgp.BGP4MP, subtype, mp)
+ if err != nil {
+ log.WithFields(log.Fields{
+ "Topic": "mrt",
+ "Data": m,
+ }).Warn(err)
+ return
+ }
+ buf, err := bm.Serialize()
+ if err == nil {
+ _, err = w.file.Write(buf)
+ }
+
+ if err != nil {
+ log.WithFields(log.Fields{
+ "Topic": "mrt",
+ "Data": m,
+ }).Warn(err)
+ }
+ }
+
+ select {
+ case <-w.t.Dying():
+ for len(w.ch) > 0 {
+ m := <-w.ch
+ write(m)
+ }
+ return nil
+ case m := <-w.ch:
+ write(m)
+ }
+ }
+}
+
+func newMrtWatcher(filename string) (*mrtWatcher, error) {
+ file, err := os.OpenFile(filename, os.O_CREATE|os.O_RDWR|os.O_APPEND, 0644)
+ if err != nil {
+ log.Fatal(err)
+ }
+ w := mrtWatcher{
+ filename: filename,
+ file: file,
+ ch: make(chan watcherEvent),
+ }
+ w.t.Go(w.loop)
+ return &w, nil
+}
generated by cgit v1.2.3 (git 2.39.1) at 2024-11-16 06:34:31 +0000