add collector feature

dump the update messages and the state change of peers into influxdb:

[collector.config]
  url = "http://localhost:8086"
  db-name = "gobgp"

Signed-off-by: FUJITA Tomonori <fujita.tomonori@lab.ntt.co.jp>

1 files changed, 37 insertions, 0 deletions

diff --git a/server/server.go b/server/server.go
index 97f9228c..25153aaa 100644
--- a/server/server.go
+++ b/server/server.go
@@ -146,6 +146,7 @@ type BgpServer struct {
 	fsmStateCh    chan *FsmMsg
 	acceptCh      chan *net.TCPConn
 	zapiMsgCh     chan *zebra.Message
+	collector     *Collector
 
 	GrpcReqCh     chan *GrpcRequest
 	policy        *table.RoutingPolicy
@@ -997,6 +998,22 @@ func (server *BgpServer) SetGlobalType(g config.Global) error {
 	return nil
 }
 
+func (server *BgpServer) SetCollector(c config.Collector) error {
+	if len(c.Config.Url) == 0 {
+		return nil
+	}
+	ch := make(chan *GrpcResponse)
+	server.GrpcReqCh <- &GrpcRequest{
+		RequestType: REQ_INITIALIZE_COLLECTOR,
+		Data:        &c.Config,
+		ResponseCh:  ch,
+	}
+	if err := (<-ch).Err(); err != nil {
+		return err
+	}
+	return nil
+}
+
 func (server *BgpServer) SetZebraConfig(z config.Zebra) error {
 	if !z.Config.Enabled {
 		return nil
@@ -2338,6 +2355,26 @@ func (server *BgpServer) handleGrpc(grpcReq *GrpcRequest) []*SenderMsg {
 			ResponseErr: err,
 		}
 		close(grpcReq.ResponseCh)
+	case REQ_INITIALIZE_COLLECTOR:
+		c := grpcReq.Data.(*config.CollectorConfig)
+		collector, err := NewCollector(server.GrpcReqCh, c.Url, c.DbName, c.TableDumpInterval)
+		if err == nil {
+			server.collector = collector
+			server.watchers[WATCHER_COLLECTOR] = collector
+		}
+		grpcReq.ResponseCh <- &GrpcResponse{
+			ResponseErr: err,
+		}
+		close(grpcReq.ResponseCh)
+	case REQ_WATCHER_ADJ_RIB_IN:
+		pathList := make([]*table.Path, 0)
+		for _, peer := range server.neighborMap {
+			pathList = append(pathList, peer.adjRibIn.PathList(peer.configuredRFlist(), false)...)
+		}
+
+		grpcReq.ResponseCh <- &GrpcResponse{}
+		close(grpcReq.ResponseCh)
+		server.notify2watchers(WATCHER_EVENT_ADJ_IN, &watcherEventAdjInMsg{pathList: pathList})
 	default:
 		err = fmt.Errorf("Unknown request type: %v", grpcReq.RequestType)
 		goto ERROR