Repository: liftbridge-io/liftbridge
Pull request #73 (2019-07-03 04:14:56)

Suppress logs during Raft recovery by default
Also add new config option `log.recovery` to allow enabling logs during
Raft recovery.


Patch:
diff --git a/documentation/configuration.md b/documentation/configuration.md
index ed0e065..65e4203 100644
--- a/documentation/configuration.md
+++ b/documentation/configuration.md
@@ -102,6 +102,7 @@ the setting in the configuration file and the CLI flag if it exists.
 | tls.key | tls-key | The private key file for server certificate. This must be set in combination with `tls.cert` to enable TLS. | string | |
 | tls.cert | tls-cert | The server certificate file. This must be set in combination with `tls.key` to enable TLS. | string | |
 | log.level | level | The logging level. | string | info | [debug, info, warn, error] |
+| log.recovery | | Log messages resulting from the replay of the Raft log on server recovery. | bool | false | |
 | data.dir | data-dir | The directory to store data in. | string | /tmp/liftbridge/namespace | |
 | batch.max.messages | | The maximum number of messages to batch when writing to disk. | int | 1024 |
 | batch.wait.time | | The time to wait to batch more messages when writing to disk. | duration | 0 | |
diff --git a/server/commitlog/commitlog.go b/server/commitlog/commitlog.go
index 3b6e487..abdc590 100644
--- a/server/commitlog/commitlog.go
+++ b/server/commitlog/commitlog.go
@@ -16,7 +16,6 @@ import (
 
 	atomic_file "github.com/natefinch/atomic"
 	"github.com/pkg/errors"
-	log "github.com/sirupsen/logrus"
 
 	"github.com/liftbridge-io/liftbridge/server/logger"
 	"github.com/liftbridge-io/liftbridge/server/proto"
@@ -71,7 +70,8 @@ func New(opts Options) (*CommitLog, error) {
 	}
 
 	if opts.Logger == nil {
-		opts.Logger = &log.Logger{Out: ioutil.Discard}
+		opts.Logger = logger.NewLogger(0)
+		opts.Logger.SetWriter(ioutil.Discard)
 	}
 
 	if opts.MaxSegmentBytes == 0 {
diff --git a/server/config.go b/server/config.go
index 0afee2c..b95a7c2 100644
--- a/server/config.go
+++ b/server/config.go
@@ -99,7 +99,8 @@ type Config struct {
 	Host                string
 	Port                int
 	LogLevel            uint32
-	NoLog               bool
+	LogRecovery         bool
+	LogSilent           bool
 	DataDir             string
 	BatchMaxMessages    int
 	BatchWaitTime       time.Duration
@@ -189,6 +190,8 @@ func NewConfig(configFile string) (*Config, error) {
 				return nil, err
 			}
 			config.LogLevel = level
+		case "log.recovery":
+			config.LogRecovery = v.(bool)
 		case "data.dir":
 			config.DataDir = v.(string)
 		case "batch.max.messages":
diff --git a/server/fsm.go b/server/fsm.go
index 37e5e4c..5f25035 100644
--- a/server/fsm.go
+++ b/server/fsm.go
@@ -4,6 +4,7 @@ import (
 	"encoding/binary"
 	"fmt"
 	"io"
+	"io/ioutil"
 	"strconv"
 
 	"github.com/dustin/go-humanize/english"
@@ -53,7 +54,7 @@ func (s *Server) recoverLatestCommittedFSMLog(applyIndex uint64) (*raft.Log, err
 // committed to Raft as part of the recovery process. As such, this should be
 // an idempotent call.
 func (s *Server) Apply(l *raft.Log) interface{} {
-	// If startedRecovery is false, the server was just started. We are going
+	// If recoveryStarted is false, the server was just started. We are going
 	// to recover the last committed Raft FSM log entry, if any, to determine
 	// the recovery high watermark. Once we apply all entries up to that point,
 	// we know we've completed the recovery process and subsequent entries are
@@ -61,17 +62,18 @@ func (s *Server) Apply(l *raft.Log) interface{} {
 	// streams will not be started until recovery is finished to avoid starting
 	// streams in an intermediate state. When recovery completes, we'll call
 	// finishedRecovery() to start the recovered streams.
-	if !s.startedRecovery {
+	if !s.recoveryStarted {
 		lastCommittedLog, err := s.recoverLatestCommittedFSMLog(l.Index)
 		// If this returns an error, something is very wrong.
 		if err != nil {
 			panic(err)
 		}
 		s.latestRecoveredLog = lastCommittedLog
-		s.startedRecovery = true
+		s.recoveryStarted = true
 		if s.latestRecoveredLog != nil {
 			s.logger.Debug("fsm: Replaying Raft log...")
 		}
+		s.startedRecovery()
 	}
 
 	// Check if this is a "recovered" Raft entry, meaning we are still applying
@@ -162,6 +164,39 @@ func (s *Server) apply(log *proto.RaftLog, index uint64, recovered bool) (interf
 	return nil, nil
 }
 
+// startedRecovery should be called when the FSM has started replaying any
+// unapplied log entries.
+func (s *Server) startedRecovery() {
+	if s.config.LogRecovery {
+		return
+	}
+	// If LogRecovery is disabled, we need to suppress logs while replaying the
+	// Raft log. Do this by discarding the log output.
+	s.loggerOut = s.logger.Writer()
+	s.logger.SetWriter(ioutil.Discard)
+}
+
+// finishedRecovery should be called when the FSM has finished replaying any
+// unapplied log entries. This will start any streams recovered during the
+// replay.
+func (s *Server) finishedRecovery() (int, error) {
+	// If LogRecovery is disabled, we need to restore the previous log output.
+	if !s.config.LogRecovery {
+		s.logger.SetWriter(s.loggerOut)
+	}
+	count := 0
+	for _, stream := range s.metadata.GetStreams() {
+		recovered, err := stream.StartRecovered()
+		if err != nil {
+			return 0, err
+		}
+		if recovered {
+			count++
+		}
+	}
+	return count, nil
+}
+
 // fsmSnapshot is returned by an FSM in response to a Snapshot. It must be safe
 // to invoke fsmSnapshot methods with concurrent calls to Apply.
 type fsmSnapshot struct {
diff --git a/server/logger/logger.go b/server/logger/logger.go
index 9a034b3..fd8972b 100644
--- a/server/logger/logger.go
+++ b/server/logger/logger.go
@@ -1,5 +1,12 @@
 package logger
 
+import (
+	"io"
+
+	log "github.com/sirupsen/logrus"
+)
+
+// Logger interface is used to allow tests to inject custom loggers.
 type Logger interface {
 	Fatalf(string, ...interface{})
 	Debugf(string, ...interface{})
@@ -10,4 +17,30 @@ type Logger interface {
 	Warn(...interface{})
 	Info(...interface{})
 	Fatal(...interface{})
+	Writer() io.Writer
+	SetWriter(io.Writer)
+}
+
+type logger struct {
+	*log.Logger
+}
+
+// NewLogger returns a new Logger instance backed by Logrus.
+func NewLogger(level uint32) Logger {
+	l := log.New()
+	l.SetLevel(log.Level(level))
+	logFormatter := &log.TextFormatter{
+		FullTimestamp:   true,
+		TimestampFormat: "2006-01-02 15:04:05",
+	}
+	l.Formatter = logFormatter
+	return &logger{l}
+}
+
+func (l *logger) Writer() io.Writer {
+	return l.Out
+}
+
+func (l *logger) SetWriter(writer io.Writer) {
+	l.Out = writer
 }
diff --git a/server/server.go b/server/server.go
index f7bd961..3029ff3 100644
--- a/server/server.go
+++ b/server/server.go
@@ -2,6 +2,7 @@ package server
 
 import (
 	"fmt"
+	"io"
 	"io/ioutil"
 	"math/rand"
 	"net"
@@ -16,7 +17,6 @@ import (
 	client "github.com/liftbridge-io/go-liftbridge/liftbridge-grpc"
 	"github.com/nats-io/go-nats"
 	"github.com/pkg/errors"
-	log "github.com/sirupsen/logrus"
 	"golang.org/x/net/context"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
@@ -42,12 +42,13 @@ type Server struct {
 	ncAcks             *nats.Conn
 	ncPublishes        *nats.Conn
 	logger             logger.Logger
+	loggerOut          io.Writer
 	api                *grpc.Server
 	metadata           *metadataAPI
 	shutdownCh         chan struct{}
 	raft               *raftNode
 	leaderSub          *nats.Subscription
-	startedRecovery    bool
+	recoveryStarted    bool
 	latestRecoveredLog *raft.Log
 	mu                 sync.RWMutex
 	shutdown           bool
@@ -70,15 +71,9 @@ func New(config *Config) *Server {
 	if config.DataDir == "" {
 		config.DataDir = filepath.Join("/tmp", "liftbridge", config.Clustering.Namespace)
 	}
-	logger := log.New()
-	logger.SetLevel(log.Level(config.LogLevel))
-	logFormatter := &log.TextFormatter{
-		FullTimestamp:   true,
-		TimestampFormat: "2006-01-02 15:04:05",
-	}
-	logger.Formatter = logFormatter
-	if config.NoLog {
-		logger.Out = ioutil.Discard
+	logger := logger.NewLogger(config.LogLevel)
+	if config.LogSilent {
+		logger.SetWriter(ioutil.Discard)
 	}
 	s := &Server{
 		config:     config,
@@ -352,23 +347,6 @@ func (s *Server) createNATSConn(name string) (*nats.Conn, error) {
 	return opts.Connect()
 }
 
-// finishedRecovery should be called when the FSM has finished replaying any
-// unapplied log entries. This will start any streams recovered during the
-// replay.
-func (s *Server) finishedRecovery() (int, error) {
-	count := 0
-	for _, stream := range s.metadata.GetStreams() {
-		recovered, err := stream.StartRecovered()
-		if err != nil {
-			return 0, err
-		}
-		if recovered {
-			count++
-		}
-	}
-	return count, nil
-}
-
 // startMetadataRaft creates and starts an embedded Raft node to participate in
 // the metadata cluster. This will bootstrap using the configured server
 // settings and start a goroutine for automatically responding to Raft


Test patch:
diff --git a/server/commitlog/delete_cleaner_test.go b/server/commitlog/delete_cleaner_test.go
index 2907a98..256e399 100644
--- a/server/commitlog/delete_cleaner_test.go
+++ b/server/commitlog/delete_cleaner_test.go
@@ -5,7 +5,6 @@ import (
 	"testing"
 	"time"
 
-	log "github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/require"
 
 	"github.com/liftbridge-io/liftbridge/server/logger"
@@ -13,7 +12,9 @@ import (
 )
 
 func noopLogger() logger.Logger {
-	return &log.Logger{Out: ioutil.Discard}
+	log := logger.NewLogger(0)
+	log.SetWriter(ioutil.Discard)
+	return log
 }
 
 func createSegment(t require.TestingT, dir string, baseOffset, maxBytes int64) *Segment {
diff --git a/server/common_test.go b/server/common_test.go
index b90d043..a5bf146 100644
--- a/server/common_test.go
+++ b/server/common_test.go
@@ -1,7 +1,9 @@
 package server
 
 import (
+	"bytes"
 	"fmt"
+	"io"
 	"runtime"
 	"strings"
 	"sync"
@@ -73,3 +75,9 @@ func (c *captureFatalLogger) Fatalf(format string, args ...interface{}) {
 	}
 	c.Unlock()
 }
+
+func (c *captureFatalLogger) Writer() io.Writer {
+	return bytes.NewBufferString(c.msg)
+}
+
+func (c *captureFatalLogger) SetWriter(writer io.Writer) {}
diff --git a/server/server_test.go b/server/server_test.go
index f3f85b1..155ef33 100644
--- a/server/server_test.go
+++ b/server/server_test.go
@@ -43,7 +43,7 @@ func getTestConfig(id string, bootstrap bool, port int) *Config {
 	config.Clustering.RaftLogging = true
 	config.LogLevel = uint32(log.DebugLevel)
 	config.NATS.Servers = []string{"nats://localhost:4222"}
-	config.NoLog = true
+	config.LogSilent = true
 	config.Port = port
 	return config
 }