mrt: support reset/rotate commands

$ gobgp mrt update reset

create a new file or reopen the existing file and continue to write
update messages to it.

$ gobgp mrt update <filename>

rename the current logging file to <filename> and truncate the logging
file and continue to write update messages to it.

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

3 files changed, 100 insertions, 4 deletions

diff --git a/server/grpc_server.go b/server/grpc_server.go
index c475f232..5ec5947b 100644
--- a/server/grpc_server.go
+++ b/server/grpc_server.go
@@ -48,6 +48,7 @@ const (
 	REQ_MONITOR_NEIGHBOR_PEER_STATE
 	REQ_MRT_GLOBAL_RIB
 	REQ_MRT_LOCAL_RIB
+	REQ_MOD_MRT
 	REQ_RPKI
 	REQ_ROA
 	REQ_VRF
@@ -259,6 +260,10 @@ func (s *Server) GetMrt(arg *api.MrtArguments, stream api.GobgpApi_GetMrtServer)
 	})
 }
 
+func (s *Server) ModMrt(ctx context.Context, arg *api.ModMrtArguments) (*api.Error, error) {
+	return s.mod(REQ_MOD_MRT, arg)
+}
+
 func (s *Server) GetRPKI(arg *api.Arguments, stream api.GobgpApi_GetRPKIServer) error {
 	req := NewGrpcRequest(REQ_RPKI, "", bgp.RouteFamily(arg.Rf), nil)
 	s.bgpServerCh <- req
diff --git a/server/server.go b/server/server.go
index 8a5cabdf..46b9b44d 100644
--- a/server/server.go
+++ b/server/server.go
@@ -1691,6 +1691,8 @@ func (server *BgpServer) handleGrpc(grpcReq *GrpcRequest) []*SenderMsg {
 		server.broadcastReqs = append(server.broadcastReqs, grpcReq)
 	case REQ_MRT_GLOBAL_RIB, REQ_MRT_LOCAL_RIB:
 		server.handleMrt(grpcReq)
+	case REQ_MOD_MRT:
+		server.handleModMrt(grpcReq)
 	case REQ_ROA, REQ_RPKI:
 		server.roaClient.handleGRPC(grpcReq)
 	case REQ_VRF, REQ_VRFS, REQ_VRF_MOD:
@@ -2197,6 +2199,46 @@ func (server *BgpServer) handleGrpcModPolicyAssignment(grpcReq *GrpcRequest) err
 	return err
 }
 
+func (server *BgpServer) handleModMrt(grpcReq *GrpcRequest) {
+	done := func(e error) {
+		result := &GrpcResponse{
+			ResponseErr: e,
+		}
+		grpcReq.ResponseCh <- result
+		close(grpcReq.ResponseCh)
+	}
+	arg := grpcReq.Data.(*api.ModMrtArguments)
+	w, y := server.watchers[WATCHER_MRT]
+	if arg.Operation == api.Operation_ADD {
+		if y {
+			done(fmt.Errorf("already enabled"))
+			return
+		}
+	} else {
+		if !y {
+			done(fmt.Errorf("not enabled yet"))
+			return
+		}
+	}
+	switch arg.Operation {
+	case api.Operation_ADD:
+		w, err := newMrtWatcher(arg.Filename)
+		if err == nil {
+			server.watchers[WATCHER_MRT] = w
+		}
+		done(err)
+	case api.Operation_DEL:
+		delete(server.watchers, WATCHER_MRT)
+		w.stop()
+		done(nil)
+	case api.Operation_REPLACE:
+		go func() {
+			err := w.restart(arg.Filename)
+			done(err)
+		}()
+	}
+}
+
 func (server *BgpServer) handleMrt(grpcReq *GrpcRequest) {
 	now := uint32(time.Now().Unix())
 	view := ""
diff --git a/server/watcher.go b/server/watcher.go
index a4c3a192..bed70538 100644
--- a/server/watcher.go
+++ b/server/watcher.go
@@ -16,6 +16,7 @@
 package server
 
 import (
+	"fmt"
 	log "github.com/Sirupsen/logrus"
 	"github.com/osrg/gobgp/packet"
 	"gopkg.in/tomb.v2"
@@ -66,14 +67,21 @@ type watcherEventUpdateMsg struct {
 
 type watcher interface {
 	notify(watcherEventType) chan watcherEvent
+	restart(string) error
 	stop()
 }
 
+type mrtWatcherOp struct {
+	filename string //used for rotate
+	result   chan error
+}
+
 type mrtWatcher struct {
 	t        tomb.Tomb
 	filename string
 	file     *os.File
 	ch       chan watcherEvent
+	opCh     chan *mrtWatcherOp
 }
 
 func (w *mrtWatcher) notify(t watcherEventType) chan watcherEvent {
@@ -87,7 +95,21 @@ func (w *mrtWatcher) stop() {
 	w.t.Kill(nil)
 }
 
+func (w *mrtWatcher) restart(filename string) error {
+	adminOp := &mrtWatcherOp{
+		filename: filename,
+		result:   make(chan error),
+	}
+	select {
+	case w.opCh <- adminOp:
+	default:
+		return fmt.Errorf("already an admin operaiton in progress")
+	}
+	return <-adminOp.result
+}
+
 func (w *mrtWatcher) loop() error {
+	defer w.file.Close()
 	for {
 		write := func(ev watcherEvent) {
 			m := ev.(*watcherEventUpdateMsg)
@@ -117,28 +139,55 @@ func (w *mrtWatcher) loop() error {
 			}
 		}
 
-		select {
-		case <-w.t.Dying():
+		drain := func() {
 			for len(w.ch) > 0 {
 				m := <-w.ch
 				write(m)
 			}
+		}
+
+		select {
+		case <-w.t.Dying():
+			drain()
 			return nil
 		case m := <-w.ch:
 			write(m)
+		case adminOp := <-w.opCh:
+			var err error
+			if adminOp.filename != "" {
+				err = os.Rename(w.file.Name(), adminOp.filename)
+			}
+			if err == nil {
+				var file *os.File
+				file, err = mrtFileOpen(w.file.Name())
+				if err == nil {
+					w.file.Close()
+					w.file = file
+				}
+			}
+			adminOp.result <- err
 		}
 	}
 }
 
-func newMrtWatcher(filename string) (*mrtWatcher, error) {
+func mrtFileOpen(filename string) (*os.File, error) {
 	file, err := os.OpenFile(filename, os.O_CREATE|os.O_RDWR|os.O_APPEND, 0644)
 	if err != nil {
-		log.Fatal(err)
+		log.Warn(err)
+	}
+	return file, err
+}
+
+func newMrtWatcher(filename string) (*mrtWatcher, error) {
+	file, err := mrtFileOpen(filename)
+	if err != nil {
+		return nil, err
 	}
 	w := mrtWatcher{
 		filename: filename,
 		file:     file,
 		ch:       make(chan watcherEvent),
+		opCh:     make(chan *mrtWatcherOp, 1),
 	}
 	w.t.Go(w.loop)
 	return &w, nil