diff --git a/client/client.go b/client/client.go index 7af896a9..3a3b3a99 100644 --- a/client/client.go +++ b/client/client.go @@ -17,6 +17,7 @@ import ( "github.com/relab/hotstuff/core/eventloop" "github.com/relab/hotstuff/core/logging" "github.com/relab/hotstuff/internal/proto/clientpb" + "go.uber.org/zap" "golang.org/x/time/rate" "google.golang.org/grpc" "google.golang.org/grpc/credentials" @@ -62,7 +63,7 @@ type Config struct { // Client is a hotstuff client. type Client struct { eventLoop *eventloop.EventLoop - logger logging.Logger + logger logging.Logger2 id ID mut sync.Mutex @@ -83,7 +84,7 @@ type Client struct { // New returns a new Client. func New( eventLoop *eventloop.EventLoop, - logger logging.Logger, + logger logging.Logger2, id ID, conf Config, ) (client *Client) { @@ -155,14 +156,16 @@ func (c *Client) Run(ctx context.Context) { err := c.sendCommands(ctx) if err != nil && !errors.Is(err, io.EOF) { - c.logger.Panicf("Failed to send commands: %v", err) + c.logger.Error("Failed to send commands", zap.Error(err)) + panic(err) } c.close() commandStats := <-commandStatsChan - c.logger.Infof( - "Done sending commands (executed: %d, failed: %d, timeouts: %d)", - commandStats.executed, commandStats.failed, commandStats.timeout, + c.logger.Info("Done sending commands", + zap.Int("executed", commandStats.executed), + zap.Int("failed", commandStats.failed), + zap.Int("timeouts", commandStats.timeout), ) <-eventLoopDone close(c.done) @@ -185,7 +188,7 @@ func (c *Client) close() { c.mgr.Close() err := c.reader.Close() if err != nil { - c.logger.Warn("Failed to close reader: ", err) + c.logger.Warn("Failed to close reader", zap.Error(err)) } } @@ -249,7 +252,7 @@ loop: } if time.Now().After(nextLogTime) { - c.logger.Infof("%d commands sent so far", num) + c.logger.Info("commands sent so far", zap.Uint64("count", num)) nextLogTime = time.Now().Add(time.Second) } @@ -280,7 +283,7 @@ func (c *Client) handleCommands(ctx context.Context) (executed, failed, timeout c.logger.Debug("Command timed out.") timeout++ } else if !errors.Is(err, context.Canceled) { - c.logger.Debugf("Did not get enough replies for command: %v\n", err) + c.logger.Debug("Did not get enough replies for command", zap.Error(err)) failed++ } } else { diff --git a/cmd/logger/main.go b/cmd/logger/main.go new file mode 100644 index 00000000..8d9ad4b4 --- /dev/null +++ b/cmd/logger/main.go @@ -0,0 +1,77 @@ +package main + +import ( + "errors" + "time" + + "github.com/relab/hotstuff/core/logging" + "go.uber.org/zap" +) + +func main() { + // Original unstructured logging + something := logging.New("test") + something.Info("Testing...") + + // Native zap logger (for comparison) + logger, _ := zap.NewProduction() + defer logger.Sync() + url := "http://example.com" + logger.Info("failed to fetch URL", + // Structured context as strongly typed Field values. + zap.String("url", url), + zap.Int("attempt", 3), + zap.Duration("backoff", time.Second), + ) + + // Logger2 interface + structuredLogger := logging.New2("structured-test") + + structuredLogger.Info("Application started", + zap.String("version", "1.0.0"), + zap.Int("port", 8080), + zap.Bool("debug", true), + ) + + structuredLogger.Debug("Processing request", + zap.String("method", "GET"), + zap.String("path", "/api/health"), + zap.Duration("latency", time.Millisecond*150), + ) + + structuredLogger.Warn("High memory usage detected", + zap.Float64("memory_usage_gb", 7.8), + zap.Int64("available_gb", 2), + ) + + // Error logging with structured context + err := errors.New("database connection failed") + structuredLogger.Error("Failed to connect to database", + zap.Error(err), + zap.String("host", "localhost"), + zap.Int("port", 5432), + zap.String("database", "hotstuff_db"), + ) + + // Using With() to create a logger with pre-set fields + userLogger := structuredLogger.With( + zap.String("user_id", "12345"), + zap.String("session_id", "abcdef"), + ) + + userLogger.Info("User action performed", + zap.String("action", "login"), + zap.Duration("response_time", time.Millisecond*45), + ) + + // Using Named() to create a sub-logger + dbLogger := structuredLogger.Named("database") + dbLogger.Info("Query executed", + zap.String("query", "SELECT * FROM users"), + zap.Duration("execution_time", time.Millisecond*23), + zap.Int("rows_affected", 42), + ) +} + + +/*./hotstuff run --config scripts/local_config.cue --ssh-config scripts/ssh_config.local --log-level debug &> x.log*/ diff --git a/core/eventloop/context_test.go b/core/eventloop/context_test.go index 8bdeea78..b5f5d1f5 100644 --- a/core/eventloop/context_test.go +++ b/core/eventloop/context_test.go @@ -11,7 +11,7 @@ import ( // TestTimeoutContext tests that a timeout context is canceled after receiving a timeout event. func TestTimeoutContext(t *testing.T) { - logger := logging.New("test") + logger := logging.New2("test") eventloop := eventloop.New(logger, 10) ctx, cancel := eventloop.TimeoutContext() defer cancel() @@ -25,7 +25,7 @@ func TestTimeoutContext(t *testing.T) { // TestTimeoutContextView tests that a timeout context is canceled after receiving a view change event. func TestTimeoutContextView(t *testing.T) { - logger := logging.New("test") + logger := logging.New2("test") eventloop := eventloop.New(logger, 10) ctx, cancel := eventloop.TimeoutContext() defer cancel() @@ -39,7 +39,7 @@ func TestTimeoutContextView(t *testing.T) { // TestViewContext tests that a view context is canceled after receiving a view change event. func TestViewContext(t *testing.T) { - logger := logging.New("test") + logger := logging.New2("test") eventloop := eventloop.New(logger, 10) ctx, cancel := eventloop.ViewContext(nil) defer cancel() @@ -53,7 +53,7 @@ func TestViewContext(t *testing.T) { // TestViewContextEarlierView tests that a view context is not canceled when receiving a view change event for an earlier view. func TestViewContextEarlierView(t *testing.T) { - logger := logging.New("test") + logger := logging.New2("test") eventloop := eventloop.New(logger, 10) v := hotstuff.View(1) ctx, cancel := eventloop.ViewContext(&v) diff --git a/core/eventloop/eventloop.go b/core/eventloop/eventloop.go index a04f849e..9aa97141 100644 --- a/core/eventloop/eventloop.go +++ b/core/eventloop/eventloop.go @@ -14,6 +14,7 @@ import ( "time" "github.com/relab/hotstuff/core/logging" + "go.uber.org/zap" ) type handlerOpts struct { @@ -52,7 +53,7 @@ type handler struct { // EventLoop accepts events of any type and executes registered event handlers. type EventLoop struct { - logger logging.Logger + logger logging.Logger2 eventQ queue @@ -70,7 +71,7 @@ type EventLoop struct { // New returns a new event loop with the requested buffer size. func New( - logger logging.Logger, + logger logging.Logger2, bufferSize uint, ) *EventLoop { el := &EventLoop{ @@ -82,6 +83,9 @@ func New( handlers: make(map[reflect.Type][]handler), tickers: make(map[int]*ticker), } + logger.Debug("EventLoop created", + zap.Uint("buffer_size", bufferSize), + ) return el } @@ -135,7 +139,10 @@ func (el *EventLoop) AddEvent(event any) { el.processEvent(event, true) droppedEvent := el.eventQ.push(event) if droppedEvent != nil { - el.logger.Warnf("event queue is full, dropped event: %v", droppedEvent) + el.logger.Warn("Event queue is full, dropped event", + zap.String("dropped_event_type", reflect.TypeOf(droppedEvent).String()), + zap.Any("dropped_event", droppedEvent), + ) } } } @@ -301,7 +308,11 @@ func (el *EventLoop) AddTicker(interval time.Duration, callback func(tick time.T // so we need to start the ticker from the run loop. droppedEvent := el.eventQ.push(startTickerEvent{id}) if droppedEvent != nil { - el.logger.Warnf("event queue is full, dropped event: %v", droppedEvent) + el.logger.Warn("Event queue is full, dropped ticker start event", + zap.Int("ticker_id", id), + zap.Duration("interval", interval), + zap.String("dropped_event_type", reflect.TypeOf(droppedEvent).String()), + ) } return id diff --git a/core/eventloop/eventloop_test.go b/core/eventloop/eventloop_test.go index d2d9c158..5bd349be 100644 --- a/core/eventloop/eventloop_test.go +++ b/core/eventloop/eventloop_test.go @@ -13,7 +13,7 @@ import ( type testEvent int func TestHandler(t *testing.T) { - logger := logging.New("test") + logger := logging.New2("test") el := eventloop.New(logger, 10) c := make(chan any) eventloop.Register(el, func(event testEvent) { @@ -53,7 +53,7 @@ func TestPrioritize(t *testing.T) { handler bool } - logger := logging.New("test") + logger := logging.New2("test") el := eventloop.New(logger, 10) c := make(chan eventData) eventloop.Register(el, func(event testEvent) { @@ -103,7 +103,7 @@ func TestTicker(t *testing.T) { return } - logger := logging.New("test") + logger := logging.New2("test") el := eventloop.New(logger, 10) count := 0 eventloop.Register(el, func(event testEvent) { @@ -135,7 +135,7 @@ func TestTicker(t *testing.T) { } func TestDelayedEvent(t *testing.T) { - logger := logging.New("test") + logger := logging.New2("test") el := eventloop.New(logger, 10) c := make(chan testEvent) @@ -166,7 +166,7 @@ func TestDelayedEvent(t *testing.T) { } func BenchmarkEventLoopWithPrioritize(b *testing.B) { - logger := logging.New("test") + logger := logging.New2("test") el := eventloop.New(logger, 100) for range 100 { @@ -184,7 +184,7 @@ func BenchmarkEventLoopWithPrioritize(b *testing.B) { } func BenchmarkEventLoopWithUnsafeRunInAddEventHandlers(b *testing.B) { - logger := logging.New("test") + logger := logging.New2("test") el := eventloop.New(logger, 100) for range 100 { @@ -210,7 +210,7 @@ func BenchmarkEventLoopWithUnsafeRunInAddEventHandlers(b *testing.B) { } func BenchmarkDelay(b *testing.B) { - logger := logging.New("test") + logger := logging.New2("test") el := eventloop.New(logger, 100) for b.Loop() { diff --git a/core/logging/logging2.go b/core/logging/logging2.go new file mode 100644 index 00000000..57ed60d7 --- /dev/null +++ b/core/logging/logging2.go @@ -0,0 +1,207 @@ +// Package logging provides structured logging capabilities using zap.Field for type-safe logging. +package logging + +import ( + "io" + "os" + "runtime" + "strings" + "sync" + + "go.uber.org/zap" + "go.uber.org/zap/zapcore" + "golang.org/x/term" +) + +var ( + lokiCfg *LokiConfig + lokiMut sync.RWMutex + lokiCores []zapcore.Core // track active Loki cores for Sync on shutdown +) + +// SetLokiConfig sets the global Loki push configuration. +// When set, all loggers created via New2/New2WithDest will also push logs to Loki. +// Pass nil to disable Loki logging. +func SetLokiConfig(cfg *LokiConfig) { + lokiMut.Lock() + defer lokiMut.Unlock() + lokiCfg = cfg +} + +// SyncLoki flushes all buffered Loki log entries. Call this before application exit. +func SyncLoki() { + lokiMut.RLock() + defer lokiMut.RUnlock() + for _, c := range lokiCores { + _ = c.Sync() + } +} + +// newLokiCoreIfConfigured creates and registers a Loki core if LokiConfig is set. +func newLokiCoreIfConfigured(level zapcore.LevelEnabler) zapcore.Core { + lokiMut.RLock() + defer lokiMut.RUnlock() + if lokiCfg == nil { + return nil + } + core := NewLokiCore(*lokiCfg, level) + lokiCores = append(lokiCores, core) + return core +} + +// Logger2 provides structured logging with type-safe fields using zap.Field. +// This interface supports structured field logging for common log levels. +type Logger2 interface { + Debug(msg string, fields ...zap.Field) + Info(msg string, fields ...zap.Field) + Warn(msg string, fields ...zap.Field) + Error(msg string, fields ...zap.Field) + + // Additional convenience methods + With(fields ...zap.Field) Logger2 + Named(name string) Logger2 +} + +// wrapper2 implements the Logger2 interface with level management +type wrapper2 struct { + zapLogger *zap.Logger + level zap.AtomicLevel + mut sync.Mutex +} + +// updateLevel dynamically updates the log level based on package-specific settings +func (wr *wrapper2) updateLevel() { + var ( + file string + ok bool + ) + + mut.RLock() + defer mut.RUnlock() + + if len(packageLevels) < 1 { + // no need to do anything + return + } + + _, file, _, ok = runtime.Caller(2) + + if ok { + for k, v := range packageLevels { + if strings.Contains(file, k) { + wr.level.SetLevel(v) + return + } + } + } + + wr.level.SetLevel(logLevel) +} + +func (wr *wrapper2) Debug(msg string, fields ...zap.Field) { + wr.mut.Lock() + defer wr.mut.Unlock() + wr.updateLevel() + wr.zapLogger.Debug(msg, fields...) +} + +func (wr *wrapper2) Info(msg string, fields ...zap.Field) { + wr.mut.Lock() + defer wr.mut.Unlock() + wr.updateLevel() + wr.zapLogger.Info(msg, fields...) +} + +func (wr *wrapper2) Warn(msg string, fields ...zap.Field) { + wr.mut.Lock() + defer wr.mut.Unlock() + wr.updateLevel() + wr.zapLogger.Warn(msg, fields...) +} + +func (wr *wrapper2) Error(msg string, fields ...zap.Field) { + wr.mut.Lock() + defer wr.mut.Unlock() + wr.updateLevel() + wr.zapLogger.Error(msg, fields...) +} + +func (wr *wrapper2) With(fields ...zap.Field) Logger2 { + wr.mut.Lock() + defer wr.mut.Unlock() + return &wrapper2{ + zapLogger: wr.zapLogger.With(fields...), + level: wr.level, + } +} + +func (wr *wrapper2) Named(name string) Logger2 { + wr.mut.Lock() + defer wr.mut.Unlock() + return &wrapper2{ + zapLogger: wr.zapLogger.Named(name), + level: wr.level, + } +} + +// New2 returns a new structured logger for stderr with the given name. +// If a Loki configuration has been set via SetLokiConfig, the logger will +// also push structured log entries to Grafana Loki. +func New2(name string) Logger2 { + var config zap.Config + if strings.ToLower(os.Getenv("HOTSTUFF_LOG_TYPE")) == "json" { + config = zap.NewProductionConfig() + } else { + config = zap.NewDevelopmentConfig() + if term.IsTerminal(int(os.Stderr.Fd())) { + config.EncoderConfig.EncodeLevel = zapcore.CapitalColorLevelEncoder + } + } + mut.RLock() + config.Level.SetLevel(logLevel) + mut.RUnlock() + + opts := []zap.Option{zap.AddCallerSkip(1)} + + // If Loki is configured, wrap the core with a tee to push to both console and Loki. + if lokiCore := newLokiCoreIfConfigured(config.Level); lokiCore != nil { + l, err := config.Build() + if err != nil { + panic(err) + } + teeCore := zapcore.NewTee(l.Core(), lokiCore) + l = zap.New(teeCore, opts...) + return &wrapper2{ + zapLogger: l.Named(name), + level: config.Level, + } + } + + l, err := config.Build(opts...) + if err != nil { + panic(err) + } + return &wrapper2{ + zapLogger: l.Named(name), + level: config.Level, + } +} + +// New2WithDest returns a new structured logger for the given destination with the given name. +// If a Loki configuration has been set via SetLokiConfig, the logger will +// also push structured log entries to Grafana Loki. +func New2WithDest(dest io.Writer, name string) Logger2 { + atom := zap.NewAtomicLevelAt(logLevel) + core := zapcore.NewCore(zapcore.NewConsoleEncoder(zap.NewDevelopmentEncoderConfig()), zapcore.AddSync(dest), atom) + + // If Loki is configured, tee the core. + if lokiCore := newLokiCoreIfConfigured(atom); lokiCore != nil { + core = zapcore.NewTee(core, lokiCore) + } + + l := zap.New(core, zap.AddCallerSkip(1)) + return &wrapper2{ + zapLogger: l.Named(name), + level: atom, + } +} diff --git a/core/logging/loki.go b/core/logging/loki.go new file mode 100644 index 00000000..736d3cbd --- /dev/null +++ b/core/logging/loki.go @@ -0,0 +1,245 @@ +// Package logging provides a Loki integration for structured logging. +// This file implements a zapcore.Core that pushes log entries to Grafana Loki +// via its HTTP push API (/loki/api/v1/push). +package logging + +import ( + "bytes" + "encoding/json" + "fmt" + "net/http" + "strconv" + "sync" + "time" + + "go.uber.org/zap/zapcore" +) + +// LokiConfig holds configuration for the Loki log aggregation client. +type LokiConfig struct { + // URL is the Loki push API endpoint, e.g. "http://localhost:3100/loki/api/v1/push". + URL string + // Labels are static labels attached to every log stream pushed to Loki. + // Common labels: {"app": "hotstuff", "node": "replica-1"}. + Labels map[string]string + // BatchSize is the maximum number of log entries to buffer before flushing. + // Default: 100. + BatchSize int + // BatchWait is the maximum time to wait before flushing a partial batch. + // Default: 1 second. + BatchWait time.Duration + // TenantID is an optional multi-tenant header for Loki (X-Scope-OrgID). + TenantID string +} + +// lokiPushRequest matches the Loki push API JSON format. +type lokiPushRequest struct { + Streams []lokiStream `json:"streams"` +} + +type lokiStream struct { + Stream map[string]string `json:"stream"` + Values [][]string `json:"values"` +} + +// lokiCore implements zapcore.Core and pushes log entries to Loki. +type lokiCore struct { + cfg LokiConfig + encoder zapcore.Encoder + level zapcore.LevelEnabler + fields []zapcore.Field + + mu sync.Mutex + batch []lokiEntry + client *http.Client + quit chan struct{} + done chan struct{} + flushed chan struct{} +} + +type lokiEntry struct { + timestamp time.Time + line string +} + +// NewLokiCore creates a new zapcore.Core that sends logs to Grafana Loki. +// The core buffers log entries and pushes them in batches. +func NewLokiCore(cfg LokiConfig, level zapcore.LevelEnabler) zapcore.Core { + if cfg.BatchSize <= 0 { + cfg.BatchSize = 100 + } + if cfg.BatchWait <= 0 { + cfg.BatchWait = 1 * time.Second + } + if cfg.Labels == nil { + cfg.Labels = map[string]string{"app": "hotstuff"} + } + // Ensure "app" label exists. + if _, ok := cfg.Labels["app"]; !ok { + cfg.Labels["app"] = "hotstuff" + } + + encoderCfg := zapcore.EncoderConfig{ + TimeKey: "ts", + LevelKey: "level", + NameKey: "logger", + CallerKey: "caller", + FunctionKey: zapcore.OmitKey, + MessageKey: "msg", + StacktraceKey: "stacktrace", + LineEnding: "", + EncodeLevel: zapcore.LowercaseLevelEncoder, + EncodeTime: zapcore.ISO8601TimeEncoder, + EncodeDuration: zapcore.StringDurationEncoder, + EncodeCaller: zapcore.ShortCallerEncoder, + } + + lc := &lokiCore{ + cfg: cfg, + encoder: zapcore.NewJSONEncoder(encoderCfg), + level: level, + batch: make([]lokiEntry, 0, cfg.BatchSize), + client: &http.Client{Timeout: 5 * time.Second}, + quit: make(chan struct{}), + done: make(chan struct{}), + flushed: make(chan struct{}), + } + + go lc.runFlusher() + return lc +} + +// Enabled implements zapcore.Core. +func (lc *lokiCore) Enabled(lvl zapcore.Level) bool { + return lc.level.Enabled(lvl) +} + +// With implements zapcore.Core. +func (lc *lokiCore) With(fields []zapcore.Field) zapcore.Core { + clone := &lokiCore{ + cfg: lc.cfg, + encoder: lc.encoder.Clone(), + level: lc.level, + fields: append(lc.fields[:len(lc.fields):len(lc.fields)], fields...), + batch: lc.batch, + client: lc.client, + quit: lc.quit, + done: lc.done, + flushed: lc.flushed, + } + for _, f := range fields { + f.AddTo(clone.encoder) + } + return clone +} + +// Check implements zapcore.Core. +func (lc *lokiCore) Check(entry zapcore.Entry, ce *zapcore.CheckedEntry) *zapcore.CheckedEntry { + if lc.Enabled(entry.Level) { + return ce.AddCore(entry, lc) + } + return ce +} + +// Write implements zapcore.Core. +func (lc *lokiCore) Write(entry zapcore.Entry, fields []zapcore.Field) error { + buf, err := lc.encoder.EncodeEntry(entry, fields) + if err != nil { + return err + } + line := buf.String() + buf.Free() + + lc.mu.Lock() + lc.batch = append(lc.batch, lokiEntry{ + timestamp: entry.Time, + line: line, + }) + shouldFlush := len(lc.batch) >= lc.cfg.BatchSize + lc.mu.Unlock() + + if shouldFlush { + lc.flush() + } + return nil +} + +// Sync implements zapcore.Core. Flushes any remaining buffered logs. +func (lc *lokiCore) Sync() error { + close(lc.quit) + <-lc.done + lc.flush() + return nil +} + +// runFlusher periodically flushes batched log entries to Loki. +func (lc *lokiCore) runFlusher() { + defer close(lc.done) + ticker := time.NewTicker(lc.cfg.BatchWait) + defer ticker.Stop() + + for { + select { + case <-ticker.C: + lc.flush() + case <-lc.quit: + return + } + } +} + +// flush sends the current batch to Loki and clears it. +func (lc *lokiCore) flush() { + lc.mu.Lock() + if len(lc.batch) == 0 { + lc.mu.Unlock() + return + } + entries := lc.batch + lc.batch = make([]lokiEntry, 0, lc.cfg.BatchSize) + lc.mu.Unlock() + + values := make([][]string, len(entries)) + for i, e := range entries { + values[i] = []string{ + strconv.FormatInt(e.timestamp.UnixNano(), 10), + e.line, + } + } + + payload := lokiPushRequest{ + Streams: []lokiStream{ + { + Stream: lc.cfg.Labels, + Values: values, + }, + }, + } + + body, err := json.Marshal(payload) + if err != nil { + fmt.Printf("loki: failed to marshal push request: %v\n", err) + return + } + + req, err := http.NewRequest(http.MethodPost, lc.cfg.URL, bytes.NewReader(body)) + if err != nil { + fmt.Printf("loki: failed to create request: %v\n", err) + return + } + req.Header.Set("Content-Type", "application/json") + if lc.cfg.TenantID != "" { + req.Header.Set("X-Scope-OrgID", lc.cfg.TenantID) + } + + resp, err := lc.client.Do(req) + if err != nil { + fmt.Printf("loki: failed to push logs: %v\n", err) + return + } + defer resp.Body.Close() + + if resp.StatusCode/100 != 2 { + fmt.Printf("loki: unexpected response status: %s\n", resp.Status) + } +} diff --git a/core/logging/loki_test.go b/core/logging/loki_test.go new file mode 100644 index 00000000..07083375 --- /dev/null +++ b/core/logging/loki_test.go @@ -0,0 +1,212 @@ +package logging + +import ( + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "go.uber.org/zap" + "go.uber.org/zap/zapcore" +) + +func TestLokiCorePushesLogs(t *testing.T) { + var ( + mu sync.Mutex + received []lokiPushRequest + ) + + // Start a mock Loki server. + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + t.Errorf("expected POST, got %s", r.Method) + } + if ct := r.Header.Get("Content-Type"); ct != "application/json" { + t.Errorf("expected application/json content-type, got %s", ct) + } + + body, err := io.ReadAll(r.Body) + if err != nil { + t.Errorf("failed to read body: %v", err) + w.WriteHeader(http.StatusInternalServerError) + return + } + + var req lokiPushRequest + if err := json.Unmarshal(body, &req); err != nil { + t.Errorf("failed to unmarshal request: %v", err) + w.WriteHeader(http.StatusBadRequest) + return + } + + mu.Lock() + received = append(received, req) + mu.Unlock() + + w.WriteHeader(http.StatusNoContent) + })) + defer srv.Close() + + cfg := LokiConfig{ + URL: srv.URL, + Labels: map[string]string{"app": "hotstuff-test", "env": "test"}, + BatchSize: 2, + BatchWait: 50 * time.Millisecond, + } + + core := NewLokiCore(cfg, zapcore.DebugLevel) + logger := zap.New(core) + + // Write enough logs to trigger a batch flush (BatchSize=2). + logger.Info("first message", zap.String("key", "val1")) + logger.Info("second message", zap.String("key", "val2")) + + // Give the flusher time to send. + time.Sleep(200 * time.Millisecond) + + mu.Lock() + defer mu.Unlock() + + if len(received) == 0 { + t.Fatal("expected at least one push request to Loki, got none") + } + + // Verify the stream labels and values. + found := false + for _, req := range received { + for _, stream := range req.Streams { + if stream.Stream["app"] == "hotstuff-test" && stream.Stream["env"] == "test" { + found = true + if len(stream.Values) < 1 { + t.Error("expected at least 1 log value in stream") + } + } + } + } + if !found { + t.Error("did not find stream with expected labels") + } + + // Sync (stop flusher) should not panic. + _ = core.Sync() +} + +func TestLokiCoreTenantHeader(t *testing.T) { + var gotTenant string + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotTenant = r.Header.Get("X-Scope-OrgID") + w.WriteHeader(http.StatusNoContent) + })) + defer srv.Close() + + cfg := LokiConfig{ + URL: srv.URL, + Labels: map[string]string{"app": "hotstuff"}, + TenantID: "my-tenant", + BatchSize: 1, + BatchWait: 50 * time.Millisecond, + } + + core := NewLokiCore(cfg, zapcore.DebugLevel) + logger := zap.New(core) + logger.Info("tenant test") + + time.Sleep(200 * time.Millisecond) + + if gotTenant != "my-tenant" { + t.Errorf("expected tenant header 'my-tenant', got %q", gotTenant) + } + + _ = core.Sync() +} + +func TestLokiBatchWaitFlush(t *testing.T) { + var ( + mu sync.Mutex + received int + ) + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + var req lokiPushRequest + _ = json.Unmarshal(body, &req) + mu.Lock() + for _, s := range req.Streams { + received += len(s.Values) + } + mu.Unlock() + w.WriteHeader(http.StatusNoContent) + })) + defer srv.Close() + + cfg := LokiConfig{ + URL: srv.URL, + Labels: map[string]string{"app": "hotstuff"}, + BatchSize: 1000, // High batch size so it won't trigger size-based flush. + BatchWait: 100 * time.Millisecond, + } + + core := NewLokiCore(cfg, zapcore.DebugLevel) + logger := zap.New(core) + + // Send a single log that won't fill the batch. + logger.Info("timer flush test") + + // Wait for the batch timer to fire. + time.Sleep(300 * time.Millisecond) + + mu.Lock() + defer mu.Unlock() + if received < 1 { + t.Errorf("expected at least 1 log flushed by timer, got %d", received) + } + + _ = core.Sync() +} + +func TestSetLokiConfigIntegration(t *testing.T) { + var ( + mu sync.Mutex + received int + ) + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + var req lokiPushRequest + _ = json.Unmarshal(body, &req) + mu.Lock() + for _, s := range req.Streams { + received += len(s.Values) + } + mu.Unlock() + w.WriteHeader(http.StatusNoContent) + })) + defer srv.Close() + + // Configure Loki globally. + SetLokiConfig(&LokiConfig{ + URL: srv.URL, + Labels: map[string]string{"app": "hotstuff", "test": "integration"}, + BatchSize: 1, + BatchWait: 50 * time.Millisecond, + }) + defer SetLokiConfig(nil) + + // Create a Logger2 — it should automatically wire up Loki. + logger := New2WithDest(io.Discard, "test-loki") + logger.Info("hello loki", zap.String("foo", "bar")) + + time.Sleep(300 * time.Millisecond) + + mu.Lock() + defer mu.Unlock() + if received < 1 { + t.Errorf("expected Logger2 to push to Loki, got %d entries", received) + } + + SyncLoki() +} diff --git a/internal/cli/root.go b/internal/cli/root.go index 2d790a4d..4dbcd181 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -5,6 +5,7 @@ import ( "fmt" "os" "strings" + "time" "github.com/relab/hotstuff/core/logging" "github.com/relab/hotstuff/protocol/comm" @@ -100,6 +101,18 @@ func init() { cobra.CheckErr(viper.BindPFlag("log-level", rootCmd.PersistentFlags().Lookup("log-level"))) rootCmd.PersistentFlags().StringSlice("log-pkgs", []string{}, "set the log level on a per-package basis.") cobra.CheckErr(viper.BindPFlag("log-pkgs", rootCmd.PersistentFlags().Lookup("log-pkgs"))) + + // Loki log aggregation flags + rootCmd.PersistentFlags().String("loki-url", "", "Loki push API URL (e.g. http://localhost:3100/loki/api/v1/push). Enables Loki log shipping when set.") + cobra.CheckErr(viper.BindPFlag("loki-url", rootCmd.PersistentFlags().Lookup("loki-url"))) + rootCmd.PersistentFlags().StringSlice("loki-labels", []string{}, "static labels for Loki streams as key=value pairs (e.g. node=replica-1,env=dev)") + cobra.CheckErr(viper.BindPFlag("loki-labels", rootCmd.PersistentFlags().Lookup("loki-labels"))) + rootCmd.PersistentFlags().String("loki-tenant-id", "", "Loki tenant ID for multi-tenant setups (X-Scope-OrgID header)") + cobra.CheckErr(viper.BindPFlag("loki-tenant-id", rootCmd.PersistentFlags().Lookup("loki-tenant-id"))) + rootCmd.PersistentFlags().Int("loki-batch-size", 100, "max log entries to buffer before pushing to Loki") + cobra.CheckErr(viper.BindPFlag("loki-batch-size", rootCmd.PersistentFlags().Lookup("loki-batch-size"))) + rootCmd.PersistentFlags().Duration("loki-batch-wait", 1*time.Second, "max time to wait before flushing a partial batch to Loki") + cobra.CheckErr(viper.BindPFlag("loki-batch-wait", rootCmd.PersistentFlags().Lookup("loki-batch-wait"))) } // initConfig reads in config file and ENV variables if set. @@ -130,6 +143,25 @@ func initConfig() { logging.SetLogLevel(viper.GetString("log-level")) + // Configure Loki if URL is provided (via --loki-url flag or HOTSTUFF_LOKI_URL env var) + lokiURL := viper.GetString("loki-url") + if lokiURL != "" { + labels := map[string]string{"app": "hotstuff"} + for _, lbl := range viper.GetStringSlice("loki-labels") { + parts := strings.SplitN(lbl, "=", 2) + if len(parts) == 2 { + labels[parts[0]] = parts[1] + } + } + logging.SetLokiConfig(&logging.LokiConfig{ + URL: lokiURL, + Labels: labels, + TenantID: viper.GetString("loki-tenant-id"), + BatchSize: viper.GetInt("loki-batch-size"), + BatchWait: viper.GetDuration("loki-batch-wait"), + }) + } + packageLevels := viper.GetStringSlice("log-pkgs") for _, packageLevel := range packageLevels { diff --git a/internal/cli/run.go b/internal/cli/run.go index 4c9bb3eb..0dff9edf 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -25,6 +25,8 @@ import ( ) func runController() { + defer logging.SyncLoki() + cfg, err := config.NewViper() checkf("viper config error: %v", err) @@ -101,7 +103,7 @@ func runSingleExperiment(cfg *config.ExperimentConfig) { experiment, err := orchestration.NewExperiment( cfg, remoteWorkers, - logging.New("ctrl"), + logging.New2("ctrl"), ) checkf("config error: %v", err) @@ -158,7 +160,7 @@ func localWorker(globalOutput string, enableMetrics []string, interval time.Dura wr := bufio.NewWriter(f) defer func() { checkf("failed to flush writer: %v", wr.Flush()) }() - logger, err = metrics.NewJSONLogger(wr, logging.New("json")) + logger, err = metrics.NewJSONLogger(wr, logging.New2("json")) checkf("failed to create JSON logger: %v", err) defer func() { checkf("failed to close logger: %v", logger.Close()) }() } else { diff --git a/internal/cli/twins.go b/internal/cli/twins.go index 83cf1ba3..6df9ab93 100644 --- a/internal/cli/twins.go +++ b/internal/cli/twins.go @@ -86,7 +86,7 @@ func twinsRun() { err error ) if twinsSrc == "" { - source = newGen(logging.New("")) + source = newGen(logging.New2("")) } else { f, err := os.Open(twinsSrc) checkf("failed to open source file: %v", err) @@ -131,7 +131,7 @@ func twinsRun() { } func twinsGenerate() { - t, err := newInstance(newGen(logging.New(""))) + t, err := newInstance(newGen(logging.New2(""))) checkf("failed to create twins instance: %v", err) defer func() { checkf("failed to close twins instance: %v", t.closeOutput()) }() @@ -158,7 +158,7 @@ type twinsInstance struct { closeOutput func() error } -func newGen(logger logging.Logger) *twins.Generator { +func newGen(logger logging.Logger2) *twins.Generator { gen := twins.NewGenerator(logger, twins.Settings{ NumNodes: numReplicas, NumTwins: numTwins, diff --git a/internal/cli/worker.go b/internal/cli/worker.go index 1ba158ee..3d7d24c0 100644 --- a/internal/cli/worker.go +++ b/internal/cli/worker.go @@ -60,6 +60,8 @@ func init() { } func runWorker() { + defer logging.SyncLoki() + stopProfilers, err := profiling.StartProfilers(cpuProfile, memProfile, trace, fgprofProfile) checkf("failed to start profilers: %v", err) defer func() { @@ -72,7 +74,7 @@ func runWorker() { f, err := os.OpenFile(dataPath, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0644) checkf("failed to create data path: %v", err) writer := bufio.NewWriter(f) - metricsLogger, err = metrics.NewJSONLogger(writer, logging.New("json")) + metricsLogger, err = metrics.NewJSONLogger(writer, logging.New2("json")) defer func() { err = metricsLogger.Close() checkf("failed to close metrics logger: %v", err) diff --git a/internal/orchestration/controller.go b/internal/orchestration/controller.go index 13bb33cd..2d224a07 100644 --- a/internal/orchestration/controller.go +++ b/internal/orchestration/controller.go @@ -19,13 +19,14 @@ import ( "github.com/relab/hotstuff/internal/latency" "github.com/relab/hotstuff/internal/proto/orchestrationpb" "github.com/relab/hotstuff/security/crypto/keygen" + "go.uber.org/zap" "google.golang.org/protobuf/proto" ) // Experiment coordinates replicas and clients, controls experiment flow and // handles measurement output. type Experiment struct { - logger logging.Logger + logger logging.Logger2 workers map[string]RemoteWorker output string // path to output folder @@ -40,7 +41,7 @@ type Experiment struct { func NewExperiment( cfg *config.ExperimentConfig, workers map[string]RemoteWorker, - logger logging.Logger, + logger logging.Logger2, ) (*Experiment, error) { totalHostCount := len(cfg.ReplicaHosts) + len(cfg.ClientHosts) workerCount := len(workers) @@ -113,7 +114,7 @@ func (e *Experiment) Run() (err error) { } wait := 5 * replicaOpts.GetInitialTimeout().AsDuration() - e.logger.Infof("Waiting %s for replicas to finish.", wait) + e.logger.Info("Waiting for replicas to finish", zap.Duration("wait", wait)) // give the replicas some time to commit the last batch time.Sleep(wait) @@ -143,7 +144,7 @@ func (e *Experiment) createReplicas(replicaMap config.ReplicaMap) (cfg *orchestr return nil, err } req.Replicas[opt.ID] = opt - e.logger.Infof("replica %d assigned to host %s", opt.ID, host) + e.logger.Info("replica assigned to host", zap.Uint32("replica", uint32(opt.ID)), zap.String("host", host)) } worker := e.workers[host] @@ -154,8 +155,13 @@ func (e *Experiment) createReplicas(replicaMap config.ReplicaMap) (cfg *orchestr for id, replicaCfg := range wcfg.GetReplicas() { replicaCfg.Address = host - e.logger.Debugf("Replica %d: Address: %s, PublicKey: %t, ReplicaPort: %d, ClientPort: %d", - id, replicaCfg.Address, len(replicaCfg.PublicKey) > 0, replicaCfg.ReplicaPort, replicaCfg.ClientPort) + e.logger.Debug("Replica info", + zap.Uint32("id", uint32(id)), + zap.String("address", replicaCfg.Address), + zap.Bool("publicKey", len(replicaCfg.PublicKey) > 0), + zap.Uint32("replicaPort", uint32(replicaCfg.ReplicaPort)), + zap.Uint32("clientPort", uint32(replicaCfg.ClientPort)), + ) cfg.Replicas[id] = replicaCfg } } @@ -261,7 +267,7 @@ func (e *Experiment) startClients(cfg *orchestrationpb.ReplicaConfiguration, src clientOpts := proto.CloneOf(srcClientOpt) clientOpts.ID = uint32(id) req.Clients[uint32(id)] = clientOpts - e.logger.Infof("client %d assigned to host %s", id, host) + e.logger.Info("client assigned to host", zap.Uint32("client", uint32(id)), zap.String("host", host)) } _, err := worker.StartClient(req) if err != nil { diff --git a/internal/orchestration/orchestration_test.go b/internal/orchestration/orchestration_test.go index 3c7ba167..85e6375d 100644 --- a/internal/orchestration/orchestration_test.go +++ b/internal/orchestration/orchestration_test.go @@ -80,7 +80,7 @@ func run(t *testing.T, cfg *config.ExperimentConfig) { experiment, err := orchestration.NewExperiment( cfg, map[string]orchestration.RemoteWorker{"localhost": workerProxy}, - logging.New("ctrl"), + logging.New2("ctrl"), ) if err != nil { t.Fatal(err) @@ -223,7 +223,7 @@ func TestDeployment(t *testing.T) { experiment, err := orchestration.NewExperiment( cfg, workers, - logging.New("ctrl"), + logging.New2("ctrl"), ) if err != nil { t.Fatal(err) diff --git a/internal/orchestration/worker.go b/internal/orchestration/worker.go index 36f8c89d..ae6c8e86 100644 --- a/internal/orchestration/worker.go +++ b/internal/orchestration/worker.go @@ -28,6 +28,7 @@ import ( "github.com/relab/hotstuff/security/crypto/keygen" "github.com/relab/hotstuff/server" "github.com/relab/hotstuff/wiring" + "go.uber.org/zap" "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials" "google.golang.org/grpc/credentials/insecure" @@ -195,7 +196,7 @@ func (w *Worker) createReplica(opts *orchestrationpb.ReplicaOpts) (*replica.Repl depsCore.RuntimeCfg(), creds, ) - depsCore.Logger().Debugf("Initializing module (crypto): %s", opts.GetCrypto()) + depsCore.Logger().Debug("Initializing module (crypto)", zap.String("module", opts.GetCrypto())) base, err := crypto.New( depsCore.RuntimeCfg(), opts.GetCrypto(), @@ -243,7 +244,7 @@ func initConsensusModules( synchronizer.ViewDuration, error, ) { - depsCore.Logger().Debugf("Initializing module (consensus rules): %s", opts.GetConsensus()) + depsCore.Logger().Debug("Initializing module (consensus rules)", zap.String("module", opts.GetConsensus())) consensusRules, err := rules.New( depsCore.Logger(), depsCore.RuntimeCfg(), @@ -254,7 +255,7 @@ func initConsensusModules( return nil, nil, nil, nil, nil, err } if byzStrategy := opts.GetByzantineStrategy(); byzStrategy != "" { - depsCore.Logger().Debugf("Initializing module (byzantine strategy): %s", byzStrategy) + depsCore.Logger().Debug("Initializing module (byzantine strategy)", zap.String("module", byzStrategy)) byz, err := byzantine.Wrap( depsCore.RuntimeCfg(), depsSecure.Blockchain(), @@ -273,7 +274,7 @@ func initConsensusModules( if err != nil { return nil, nil, nil, nil, nil, err } - depsCore.Logger().Debugf("Initializing module (leader rotation): %s", opts.GetLeaderRotation()) + depsCore.Logger().Debug("Initializing module (leader rotation)", zap.String("module", opts.GetLeaderRotation())) leaderRotation, err := leaderrotation.New( depsCore.Logger(), depsCore.RuntimeCfg(), @@ -285,7 +286,7 @@ func initConsensusModules( if err != nil { return nil, nil, nil, nil, nil, err } - depsCore.Logger().Debugf("Initializing module (communication): %s", opts.GetCommunication()) + depsCore.Logger().Debug("Initializing module (communication)", zap.String("module", opts.GetCommunication())) comm, err := comm.New( depsCore.Logger(), depsCore.EventLoop(), @@ -391,7 +392,7 @@ func (w *Worker) startClients(req *orchestrationpb.StartClientRequest) (*orchest RateStepInterval: opts.GetRateStepInterval().AsDuration(), Timeout: opts.GetTimeout().AsDuration(), } - logger := logging.New("cli" + opts.ClientIDString()) + logger := logging.New2("cli" + opts.ClientIDString()) eventLoop := eventloop.New(logger, 1000) if w.measurementInterval > 0 { diff --git a/metrics/datalogger.go b/metrics/datalogger.go index 99a35a8f..5d8033d2 100644 --- a/metrics/datalogger.go +++ b/metrics/datalogger.go @@ -6,6 +6,7 @@ import ( "sync" "github.com/relab/hotstuff/core/logging" + "go.uber.org/zap" "google.golang.org/protobuf/encoding/protojson" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/anypb" @@ -18,7 +19,7 @@ type Logger interface { } type jsonLogger struct { - logger logging.Logger + logger logging.Logger2 mut sync.Mutex wr io.Writer @@ -26,7 +27,7 @@ type jsonLogger struct { } // NewJSONLogger returns a new metrics logger that logs to the specified writer. -func NewJSONLogger(wr io.Writer, logger logging.Logger) (Logger, error) { +func NewJSONLogger(wr io.Writer, logger logging.Logger2) (Logger, error) { _, err := io.WriteString(wr, "[\n") if err != nil { return nil, fmt.Errorf("failed to write start of JSON array: %v", err) @@ -43,13 +44,13 @@ func (dl *jsonLogger) Log(msg proto.Message) { if anyMsg, ok = msg.(*anypb.Any); !ok { anyMsg, err = anypb.New(msg) if err != nil { - dl.logger.Errorf("failed to create Any message: %v", err) + dl.logger.Error("failed to create Any message", zap.Error(err)) return } } err = dl.write(anyMsg) if err != nil { - dl.logger.Errorf("failed to write message to log: %v", err) + dl.logger.Error("failed to write message to log", zap.Error(err)) } } diff --git a/metrics/registry.go b/metrics/registry.go index a17aa5a1..725305ef 100644 --- a/metrics/registry.go +++ b/metrics/registry.go @@ -8,6 +8,7 @@ import ( "github.com/relab/hotstuff/client" "github.com/relab/hotstuff/core/eventloop" "github.com/relab/hotstuff/core/logging" + "go.uber.org/zap" ) // Enable enables logging of the specified client and replica metrics. @@ -17,7 +18,7 @@ import ( // Valid metric names are defined as constants in their respective metric files. func Enable[T client.ID | hotstuff.ID]( eventLoop *eventloop.EventLoop, - logger logging.Logger, + logger logging.Logger2, metricsLogger Logger, id T, measurementInterval time.Duration, @@ -53,7 +54,7 @@ func Enable[T client.ID | hotstuff.ID]( return fmt.Errorf("invalid metric: %s", name) } } - logger.Infof("Metrics enabled: %v", enabledMetrics) + logger.Info("Metrics enabled", zap.Strings("metrics", enabledMetrics)) addTicker(eventLoop, measurementInterval) return nil } diff --git a/network/sender.go b/network/sender.go index 270d08d5..e21fa42f 100644 --- a/network/sender.go +++ b/network/sender.go @@ -13,6 +13,7 @@ import ( "github.com/relab/hotstuff/core/eventloop" "github.com/relab/hotstuff/core/logging" "github.com/relab/hotstuff/internal/proto/hotstuffpb" + "go.uber.org/zap" "google.golang.org/grpc" "google.golang.org/grpc/credentials" "google.golang.org/grpc/credentials/insecure" @@ -22,7 +23,7 @@ import ( type GorumsSender struct { eventLoop *eventloop.EventLoop - logger logging.Logger + logger logging.Logger2 config *core.RuntimeConfig mgrOpts []gorums.ManagerOption @@ -36,7 +37,7 @@ type GorumsSender struct { func NewGorumsSender( el *eventloop.EventLoop, - logger logging.Logger, + logger logging.Logger2, config *core.RuntimeConfig, creds credentials.TransportCredentials, @@ -134,24 +135,24 @@ func (s *GorumsSender) replicaConnected(c hotstuff.ReplicaConnectedEvent) { id, err := s.config.PeerIDFromContext(c.Ctx) if err != nil { - s.logger.Warnf("Failed to get id for %v: %v", info.Addr, err) + s.logger.Warn("Failed to get id", zap.Stringer("addr", info.Addr), zap.Error(err)) return } _, ok := s.config.ReplicaInfo(id) if !ok { - s.logger.Warnf("Replica with id %d was not found", id) + s.logger.Warn("Replica not found", zap.Uint32("id", uint32(id))) return } replica := s.replicas[id] replica.md = readMetadata(md) if err := s.config.SetReplicaMetadata(replica.id, replica.md); err != nil { - s.logger.Errorf("failed to set replica metadata: %v", err) + s.logger.Error("failed to set replica metadata", zap.Error(err)) return } - s.logger.Debugf("Replica %d connected from address %v", id, info.Addr) + s.logger.Debug("Replica connected", zap.Uint32("id", uint32(id)), zap.Stringer("addr", info.Addr)) } // Timeout sends the timeout message to all replicas. @@ -175,7 +176,7 @@ func (s *GorumsSender) RequestBlock(ctx context.Context, hash hotstuff.Hash) (*h if err != nil { // filter out context errors if !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) { - s.logger.Infof("Failed to fetch block: %v", err) + s.logger.Info("Failed to fetch block", zap.Error(err)) } return nil, false } diff --git a/protocol/comm/factory.go b/protocol/comm/factory.go index 4021f18d..21a2321b 100644 --- a/protocol/comm/factory.go +++ b/protocol/comm/factory.go @@ -16,7 +16,7 @@ import ( ) func New( - logger logging.Logger, + logger logging.Logger2, eventLoop *eventloop.EventLoop, config *core.RuntimeConfig, blockchain *blockchain.Blockchain, diff --git a/protocol/comm/kauri.go b/protocol/comm/kauri.go index fffda5a4..ab6b002f 100644 --- a/protocol/comm/kauri.go +++ b/protocol/comm/kauri.go @@ -14,13 +14,14 @@ import ( "github.com/relab/hotstuff/protocol/comm/kauri" "github.com/relab/hotstuff/security/blockchain" "github.com/relab/hotstuff/security/cert" + "go.uber.org/zap" ) const NameKauri = "kauri" // Kauri implements tree-based dissemination and aggregation. type Kauri struct { - logger logging.Logger + logger logging.Logger2 eventLoop *eventloop.EventLoop config *core.RuntimeConfig blockchain *blockchain.Blockchain @@ -38,7 +39,7 @@ type Kauri struct { // NewKauri creates a new Kauri instance for communicating proposals and votes. func NewKauri( - logger logging.Logger, + logger logging.Logger2, el *eventloop.EventLoop, config *core.RuntimeConfig, blockchain *blockchain.Blockchain, @@ -86,7 +87,7 @@ func (k *Kauri) begin(p *hotstuff.ProposeMsg, pc hotstuff.PartialCert) error { // TODO(meling): This is not correct use of DelayUntil, see issue #267 eventloop.DelayUntil[network.ConnectedEvent](k.eventLoop, func() { if err := k.begin(p, pc); err != nil { - k.logger.Error(err) + k.logger.Error("Failed to begin", zap.Error(err)) } }) return nil @@ -112,7 +113,7 @@ func (k *Kauri) sendProposalToChildren(p *hotstuff.ProposeMsg) error { if err != nil { return fmt.Errorf("unable to send the proposal to children: %w", err) } - k.logger.Debug("Sending proposal to children ", children) + k.logger.Debug("Sending proposal to children", zap.Int("count", len(children))) childSender.Propose(p) go k.waitToAggregate() } else { @@ -134,11 +135,11 @@ func (k *Kauri) onContributionRecv(event kauri.ContributionRecvEvent) { return } contribution := event.Contribution - k.logger.Debugf("Processing the contribution from %d", contribution.ID) + k.logger.Debug("Processing the contribution", zap.Uint32("from", contribution.ID)) currentSignature := hotstuffpb.QuorumSignatureFromProto(contribution.Signature) err := k.mergeContribution(currentSignature) if err != nil { - k.logger.Errorf("Failed to merge contribution from %d: %v", contribution.ID, err) + k.logger.Error("Failed to merge contribution", zap.Uint32("from", contribution.ID), zap.Error(err)) return } k.senders = append(k.senders, hotstuff.ID(contribution.ID)) diff --git a/protocol/consensus/committer.go b/protocol/consensus/committer.go index d6c3a26a..8fd3f8a6 100644 --- a/protocol/consensus/committer.go +++ b/protocol/consensus/committer.go @@ -10,12 +10,13 @@ import ( "github.com/relab/hotstuff/internal/proto/clientpb" "github.com/relab/hotstuff/protocol" "github.com/relab/hotstuff/security/blockchain" + "go.uber.org/zap" ) // Committer commits the correct block for a view. type Committer struct { eventLoop *eventloop.EventLoop - logger logging.Logger + logger logging.Logger2 blockchain *blockchain.Blockchain viewStates *protocol.ViewStates ruler CommitRuler @@ -23,7 +24,7 @@ type Committer struct { func NewCommitter( eventLoop *eventloop.EventLoop, - logger logging.Logger, + logger logging.Logger2, blockchain *blockchain.Blockchain, viewStates *protocol.ViewStates, ruler CommitRuler, @@ -42,7 +43,7 @@ func NewCommitter( // This eligible block is then used as the starting point for recursively // committing its uncommitted ancestor blocks. func (cm *Committer) TryCommit(block *hotstuff.Block) error { - cm.logger.Debugf("TryCommit: %v", block) + cm.logger.Debug("TryCommit", zap.Stringer("block", block)) cm.blockchain.Store(block) // check commit rule and get the next block to commit. If it was nil, do nothing. if blockToCommit := cm.ruler.CommitRule(block); blockToCommit != nil { @@ -86,7 +87,7 @@ func (cm *Committer) commitInner(block, committedBlock *hotstuff.Block) error { } else { return fmt.Errorf("failed to locate block: %s", block.Parent().SmallString()) } - cm.logger.Debug("EXEC: ", block) + cm.logger.Debug("EXEC", zap.Stringer("block", block)) batch := block.Commands() // CommitEvent holds the entire block and is used in twins since it needs the hash. cm.eventLoop.AddEvent(hotstuff.CommitEvent{Block: block}) diff --git a/protocol/leaderrotation/carousel.go b/protocol/leaderrotation/carousel.go index 96c05a49..8477848a 100644 --- a/protocol/leaderrotation/carousel.go +++ b/protocol/leaderrotation/carousel.go @@ -10,6 +10,7 @@ import ( "github.com/relab/hotstuff/core/logging" "github.com/relab/hotstuff/protocol" "github.com/relab/hotstuff/security/blockchain" + "go.uber.org/zap" ) const NameCarousel = "carousel" @@ -18,7 +19,7 @@ type Carousel struct { blockchain *blockchain.Blockchain viewStates *protocol.ViewStates config *core.RuntimeConfig - logger logging.Logger + logger logging.Logger2 chainLength int } @@ -30,7 +31,7 @@ func NewCarousel( blockchain *blockchain.Blockchain, viewStates *protocol.ViewStates, config *core.RuntimeConfig, - logger logging.Logger, + logger logging.Logger2, ) *Carousel { return &Carousel{ blockchain: blockchain, @@ -50,7 +51,7 @@ func (c *Carousel) GetLeader(round hotstuff.View) hotstuff.ID { } if commitHead.View() != round-hotstuff.View(c.chainLength) { - c.logger.Debugf("fallback to round-robin (view=%d, commitHead=%d)", round, commitHead.View()) + c.logger.Debug("fallback to round-robin", zap.Uint64("view", uint64(round)), zap.Uint64("commitHead", uint64(commitHead.View()))) return ChooseRoundRobin(round, c.config.ReplicaCount()) } @@ -81,7 +82,7 @@ func (c *Carousel) GetLeader(round hotstuff.View) hotstuff.ID { rnd := rand.New(rand.NewSource(seed)) leader := candidates[rnd.Int()%len(candidates)] - c.logger.Debugf("chose id %d", leader) + c.logger.Debug("chose id", zap.Uint32("leader", uint32(leader))) return leader } diff --git a/protocol/leaderrotation/factory.go b/protocol/leaderrotation/factory.go index 8db44209..ec5c1679 100644 --- a/protocol/leaderrotation/factory.go +++ b/protocol/leaderrotation/factory.go @@ -11,7 +11,7 @@ import ( ) func New( - logger logging.Logger, + logger logging.Logger2, config *core.RuntimeConfig, blockchain *blockchain.Blockchain, viewStates *protocol.ViewStates, diff --git a/protocol/leaderrotation/reputation.go b/protocol/leaderrotation/reputation.go index 0444f1ea..841d3164 100644 --- a/protocol/leaderrotation/reputation.go +++ b/protocol/leaderrotation/reputation.go @@ -10,6 +10,7 @@ import ( "github.com/relab/hotstuff/core" "github.com/relab/hotstuff/core/logging" "github.com/relab/hotstuff/protocol" + "go.uber.org/zap" ) const NameReputation = "reputation" @@ -19,7 +20,7 @@ type reputationsMap map[hotstuff.ID]float64 type RepBased struct { viewStates *protocol.ViewStates config *core.RuntimeConfig - logger logging.Logger + logger logging.Logger2 chainLength int prevCommitHead *hotstuff.Block @@ -70,11 +71,11 @@ func (r *RepBased) GetLeader(view hotstuff.View) hotstuff.ID { r.prevCommitHead = block } - r.logger.Debug(weights) + r.logger.Debug("weighted choices calculated", zap.Int("count", len(weights))) chooser, err := wr.NewChooser(weights...) if err != nil { - r.logger.Error("weightedrand error: ", err) + r.logger.Error("weightedrand error", zap.Error(err)) return 0 } @@ -82,7 +83,7 @@ func (r *RepBased) GetLeader(view hotstuff.View) hotstuff.ID { rnd := rand.New(rand.NewSource(seed)) leader := chooser.PickSource(rnd).(hotstuff.ID) - r.logger.Debugf("picked leader %d for view %d using seed %d", leader, view, seed) + r.logger.Debug("picked leader", zap.Uint32("leader", uint32(leader)), zap.Uint64("view", uint64(view)), zap.Int64("seed", seed)) return leader } @@ -93,7 +94,7 @@ func NewRepBased( viewStates *protocol.ViewStates, config *core.RuntimeConfig, - logger logging.Logger, + logger logging.Logger2, ) *RepBased { return &RepBased{ viewStates: viewStates, diff --git a/protocol/rules/chainedhotstuff.go b/protocol/rules/chainedhotstuff.go index 97302769..5ea9e242 100644 --- a/protocol/rules/chainedhotstuff.go +++ b/protocol/rules/chainedhotstuff.go @@ -7,13 +7,14 @@ import ( "github.com/relab/hotstuff/internal/proto/clientpb" "github.com/relab/hotstuff/protocol/consensus" "github.com/relab/hotstuff/security/blockchain" + "go.uber.org/zap" ) const NameChainedHotStuff = "chainedhotstuff" // ChainedHotStuff implements the pipelined three-phase HotStuff protocol. type ChainedHotStuff struct { - logger logging.Logger + logger logging.Logger2 config *core.RuntimeConfig blockchain *blockchain.Blockchain @@ -24,7 +25,7 @@ type ChainedHotStuff struct { // NewChainedHotStuff returns a new instance of the chained HotStuff consensus ruleset. func NewChainedHotStuff( - logger logging.Logger, + logger logging.Logger2, config *core.RuntimeConfig, blockchain *blockchain.Blockchain, ) *ChainedHotStuff { @@ -53,7 +54,7 @@ func (hs *ChainedHotStuff) CommitRule(block *hotstuff.Block) *hotstuff.Block { // Note that we do not call UpdateHighQC here. // This is done through AdvanceView, which the Consensus implementation will call. - hs.logger.Debug("CommitRule - PRE_COMMIT: ", block1) + hs.logger.Debug("CommitRule - PRE_COMMIT", zap.Stringer("block", block1)) block2, ok := hs.qcRef(block1.QuorumCert()) if !ok { @@ -61,7 +62,7 @@ func (hs *ChainedHotStuff) CommitRule(block *hotstuff.Block) *hotstuff.Block { } if block2.View() > hs.bLock.View() { - hs.logger.Debug("CommitRule - COMMIT: ", block2) + hs.logger.Debug("CommitRule - COMMIT", zap.Stringer("block", block2)) hs.bLock = block2 } @@ -70,13 +71,13 @@ func (hs *ChainedHotStuff) CommitRule(block *hotstuff.Block) *hotstuff.Block { return nil } - // since our implementation does not create dummy blocks every view, + // since our implementation does not create dummy blocks every view, // we explicitly check that the parents are in the previous view - if block1.Parent() == block2.Hash() && + if block1.Parent() == block2.Hash() && block1.View() == block2.View()+1 && block2.Parent() == block3.Hash() && block2.View() == block3.View()+1 { - hs.logger.Debug("CommitRule - DECIDE: ", block3) + hs.logger.Debug("CommitRule - DECIDE", zap.Stringer("block", block3)) return block3 } diff --git a/protocol/rules/factory.go b/protocol/rules/factory.go index ef814ef7..60fbd152 100644 --- a/protocol/rules/factory.go +++ b/protocol/rules/factory.go @@ -11,7 +11,7 @@ import ( ) func New( - logger logging.Logger, + logger logging.Logger2, config *core.RuntimeConfig, blockchain *blockchain.Blockchain, name string, diff --git a/protocol/rules/fasthotstuff.go b/protocol/rules/fasthotstuff.go index c765986a..1156e624 100644 --- a/protocol/rules/fasthotstuff.go +++ b/protocol/rules/fasthotstuff.go @@ -7,6 +7,7 @@ import ( "github.com/relab/hotstuff/internal/proto/clientpb" "github.com/relab/hotstuff/protocol/consensus" "github.com/relab/hotstuff/security/blockchain" + "go.uber.org/zap" ) const NameFastHotStuff = "fasthotstuff" @@ -14,14 +15,14 @@ const NameFastHotStuff = "fasthotstuff" // FastHotStuff is an implementation of the Fast-HotStuff protocol. // See the paper for details: https://arxiv.org/abs/2010.11454 type FastHotStuff struct { - logger logging.Logger + logger logging.Logger2 config *core.RuntimeConfig blockchain *blockchain.Blockchain } // NewFastHotStuff returns a new instance of the FastHotStuff consensus ruleset. func NewFastHotStuff( - logger logging.Logger, + logger logging.Logger2, config *core.RuntimeConfig, blockchain *blockchain.Blockchain, ) *FastHotStuff { @@ -48,14 +49,14 @@ func (fhs *FastHotStuff) CommitRule(block *hotstuff.Block) *hotstuff.Block { if !ok { return nil } - fhs.logger.Debug("PRECOMMIT: ", parent) + fhs.logger.Debug("PRECOMMIT", zap.Stringer("block", parent)) grandparent, ok := fhs.qcRef(parent.QuorumCert()) if !ok { return nil } if block.Parent() == parent.Hash() && block.View() == parent.View()+1 && parent.Parent() == grandparent.Hash() && parent.View() == grandparent.View()+1 { - fhs.logger.Debug("COMMIT: ", grandparent) + fhs.logger.Debug("COMMIT", zap.Stringer("block", grandparent)) return grandparent } return nil diff --git a/protocol/rules/simplehotstuff.go b/protocol/rules/simplehotstuff.go index 2f75c0b9..a618dd4a 100644 --- a/protocol/rules/simplehotstuff.go +++ b/protocol/rules/simplehotstuff.go @@ -7,6 +7,7 @@ import ( "github.com/relab/hotstuff/internal/proto/clientpb" "github.com/relab/hotstuff/protocol/consensus" "github.com/relab/hotstuff/security/blockchain" + "go.uber.org/zap" ) const NameSimpleHotStuff = "simplehotstuff" @@ -16,7 +17,7 @@ const NameSimpleHotStuff = "simplehotstuff" // Based on the simplified algorithm described in the paper // "Formal Verification of HotStuff" by Leander Jehl. type SimpleHotStuff struct { - logger logging.Logger + logger logging.Logger2 config *core.RuntimeConfig blockchain *blockchain.Blockchain @@ -25,7 +26,7 @@ type SimpleHotStuff struct { // NewSimpleHotStuff creates a new instance of the simple HotStuff consensus ruleset. func NewSimpleHotStuff( - logger logging.Logger, + logger logging.Logger2, config *core.RuntimeConfig, blockchain *blockchain.Blockchain, ) *SimpleHotStuff { @@ -50,7 +51,7 @@ func (hs *SimpleHotStuff) VoteRule(view hotstuff.View, proposal hotstuff.Propose parent, ok := hs.blockchain.Get(block.QuorumCert().BlockHash()) if !ok { - hs.logger.Info("VoteRule: missing parent block: ", block.QuorumCert().BlockHash()) + hs.logger.Info("VoteRule: missing parent block", zap.String("hash", block.QuorumCert().BlockHash().String())) return false } @@ -74,7 +75,7 @@ func (hs *SimpleHotStuff) CommitRule(block *hotstuff.Block) *hotstuff.Block { gp, ok := hs.blockchain.Get(p.QuorumCert().BlockHash()) if ok && gp.View() > hs.locked.View() { hs.locked = gp - hs.logger.Debug("CommitRule: updated locked block: ", gp) + hs.logger.Debug("CommitRule: updated locked block", zap.Stringer("block", gp)) } else if !ok { return nil } diff --git a/protocol/synchronizer/synchronizer.go b/protocol/synchronizer/synchronizer.go index 728ca15e..560ff29e 100644 --- a/protocol/synchronizer/synchronizer.go +++ b/protocol/synchronizer/synchronizer.go @@ -14,12 +14,13 @@ import ( "github.com/relab/hotstuff/security/cert" "github.com/relab/hotstuff" + "go.uber.org/zap" ) // Synchronizer implements the DiemBFT pacemaker and is the main loop of the consensus protocol. type Synchronizer struct { eventLoop *eventloop.EventLoop - logger logging.Logger + logger logging.Logger2 config *core.RuntimeConfig auth *cert.Authority @@ -48,7 +49,7 @@ type Synchronizer struct { func New( // core dependencies el *eventloop.EventLoop, - logger logging.Logger, + logger logging.Logger2, config *core.RuntimeConfig, // security dependencies @@ -94,7 +95,7 @@ func New( s.OnRemoteTimeout(timeoutMsg) }) eventloop.Register(el, func(proposal hotstuff.ProposeMsg) { - s.logger.Debugf("Received proposal: %v", proposal.Block) + s.logger.Debug("Received proposal", zap.Stringer("block", proposal.Block)) // advance the view regardless of vote status s.advanceView(hotstuff.NewSyncInfoWith(proposal.Block.QuorumCert())) @@ -105,23 +106,23 @@ func New( // alpha limits acceptable view drift, 10 is a temporary heuristic. const alpha hotstuff.View = 10 if proposalView > localView+alpha { - s.logger.Warnf("Dropping proposal: proposal view too high (%v) >> replica's local view (%v)", proposalView, localView) + s.logger.Warn("Dropping proposal: proposal view too high", zap.Uint64("proposalView", uint64(proposalView)), zap.Uint64("localView", uint64(localView))) return } if proposalView > localView { - s.logger.Debugf("Delaying proposal until after next view change (proposal view %v > local view %v)", proposalView, localView) + s.logger.Debug("Delaying proposal until after next view change", zap.Uint64("proposalView", uint64(proposalView)), zap.Uint64("localView", uint64(localView))) eventloop.DelayUntil[hotstuff.ViewChangeEvent](s.eventLoop, proposal) return } // verify the incoming proposal before attempting to vote and try to commit. if err := s.voter.Verify(&proposal); err != nil { - s.logger.Infof("failed to verify incoming vote: %v", err) + s.logger.Info("failed to verify incoming vote", zap.Error(err)) return } err := s.voter.OnValidPropose(&proposal) if err != nil { - s.logger.Info(err) + s.logger.Info("Vote error after valid proposal", zap.Error(err)) } }) return s @@ -167,12 +168,12 @@ func (s *Synchronizer) Start(ctx context.Context) { proposal, err := s.proposer.CreateProposal(syncInfo) if err != nil { // debug log here since it may frequently fail due to lack of commands. - s.logger.Debugf("Failed to create proposal: %v", err) + s.logger.Debug("Failed to create initial proposal (no commands)", zap.Error(err)) return } s.logger.Debug("Propose") if err := s.proposer.Propose(&proposal); err != nil { - s.logger.Info(err) + s.logger.Info("Propose error", zap.Error(err)) return } } @@ -188,17 +189,17 @@ func (s *Synchronizer) OnLocalTimeout() { return } s.duration.ViewTimeout() // increase the duration of the next view - s.logger.Debugf("OnLocalTimeout: %v", currentView) + s.logger.Debug("OnLocalTimeout", zap.Uint64("view", uint64(currentView))) timeoutMsg, err := s.timeoutRules.LocalTimeoutRule(currentView, s.state.SyncInfo()) if err != nil { - s.logger.Warnf("Failed to create timeout message: %v", err) + s.logger.Warn("Failed to create timeout message", zap.Error(err)) return } s.lastTimeout = timeoutMsg if s.voter.StopVoting(currentView) { - s.logger.Debugf("Stopped voting for timed out view %d", currentView) + s.logger.Debug("Stopped voting for timed out view", zap.Uint64("view", uint64(currentView))) } s.sender.Timeout(*timeoutMsg) @@ -211,15 +212,15 @@ func (s *Synchronizer) OnRemoteTimeout(timeout hotstuff.TimeoutMsg) { defer s.timeouts.deleteOldViews(currView) if err := s.auth.Verify(timeout.ViewSignature, timeout.View.ToBytes()); err != nil { - s.logger.Infof("View timeout signature could not be verified: %v", err) + s.logger.Info("View timeout signature could not be verified", zap.Error(err)) return } - s.logger.Debug("OnRemoteTimeout (advancing view): ", timeout) + s.logger.Debug("OnRemoteTimeout (advancing view)", zap.Uint64("view", uint64(timeout.View))) s.advanceView(timeout.SyncInfo) timeoutList, quorum := s.timeouts.add(timeout) if !quorum { - s.logger.Debugf("OnRemoteTimeout: not enough timeouts for view %d, waiting for more", timeout.View) + s.logger.Debug("OnRemoteTimeout: not enough timeouts", zap.Uint64("view", uint64(timeout.View))) return } @@ -227,20 +228,20 @@ func (s *Synchronizer) OnRemoteTimeout(timeout hotstuff.TimeoutMsg) { if err != nil { // this can only happen if the timeout rule fails to create a quorum certificate // or aggregate certificate, e.g., due insufficient number of timeouts. - s.logger.Debugf("Failed to create sync info: %v", err) + s.logger.Debug("Failed to create sync info", zap.Error(err)) return } si.SetQC(s.state.HighQC()) // ensure sync info also has the high QC - s.logger.Debugf("OnRemoteTimeout (second advance)") + s.logger.Debug("OnRemoteTimeout (second advance)") s.advanceView(si) } // OnNewView handles an incoming consensus.NewViewMsg func (s *Synchronizer) OnNewView(newView hotstuff.NewViewMsg) { - s.logger.Debugf("OnNewView (from network: %t)", newView.FromNetwork) + s.logger.Debug("OnNewView", zap.Bool("fromNetwork", newView.FromNetwork)) if newView.FromNetwork { - s.logger.Debugf("new view msg from: %d", newView.ID) + s.logger.Debug("new view msg from", zap.Uint32("from", uint32(newView.ID))) } s.advanceView(newView.SyncInfo) } @@ -248,21 +249,21 @@ func (s *Synchronizer) OnNewView(newView hotstuff.NewViewMsg) { // advanceView attempts to advance to the next view using the given QC. // qc must be either a regular quorum certificate, or a timeout certificate. func (s *Synchronizer) advanceView(syncInfo hotstuff.SyncInfo) { - s.logger.Debugf("advanceView: %v", syncInfo) + s.logger.Debug("advanceView", zap.Any("syncInfo", syncInfo)) qc, view, timeout, err := s.timeoutRules.VerifySyncInfo(syncInfo) if err != nil { - s.logger.Infof("advanceView: Failed to verify sync info: %v", err) + s.logger.Info("advanceView: Failed to verify sync info", zap.Error(err)) return } if qc != nil { updated, err := s.state.UpdateHighQC(*qc) if err != nil { - s.logger.Warnf("advanceView: Failed to update HighQC: %v", err) + s.logger.Warn("advanceView: Failed to update HighQC", zap.Error(err)) } else if updated { s.logger.Debug("advanceView: Successfully updated HighQC") } else { - s.logger.Debugf("advanceView: HighQC not updated, current view: %d, new view: %d", s.state.View(), qc.View()) + s.logger.Debug("advanceView: HighQC not updated", zap.Uint64("currentView", uint64(s.state.View())), zap.Uint64("newView", uint64(qc.View()))) } // ensure that the true highQC is the one stored in the syncInfo syncInfo.SetQC(s.state.HighQC()) @@ -287,7 +288,7 @@ func (s *Synchronizer) advanceView(syncInfo hotstuff.SyncInfo) { s.startTimeoutTimer() - s.logger.Debugf("advanceView: Advanced to view %d", newView) + s.logger.Debug("advanceView: Advanced to view", zap.Uint64("view", uint64(newView))) s.eventLoop.AddEvent(hotstuff.ViewChangeEvent{View: newView, Timeout: timeout}) leader := s.leaderRotation.GetLeader(newView) @@ -295,15 +296,15 @@ func (s *Synchronizer) advanceView(syncInfo hotstuff.SyncInfo) { proposal, err := s.proposer.CreateProposal(syncInfo) if err != nil { // debug log here since it may frequently fail due to lack of commands. - s.logger.Debugf("Failed to create proposal: %v", err) + s.logger.Debug("Failed to create leader proposal", zap.Error(err)) return } if err := s.proposer.Propose(&proposal); err != nil { - s.logger.Info(err) + s.logger.Info("Leader propose error", zap.Error(err)) } return } if err := s.sender.NewView(leader, syncInfo); err != nil { - s.logger.Warnf("advanceView: error on sending new view: %v", err) + s.logger.Warn("advanceView: error on sending new view", zap.Error(err)) } } diff --git a/protocol/votingmachine/votingmachine.go b/protocol/votingmachine/votingmachine.go index 74960888..04ef2d50 100644 --- a/protocol/votingmachine/votingmachine.go +++ b/protocol/votingmachine/votingmachine.go @@ -11,11 +11,12 @@ import ( "github.com/relab/hotstuff/protocol" "github.com/relab/hotstuff/security/blockchain" "github.com/relab/hotstuff/security/cert" + "go.uber.org/zap" ) // VotingMachine collects and verifies votes. type VotingMachine struct { - logger logging.Logger + logger logging.Logger2 eventLoop *eventloop.EventLoop config *core.RuntimeConfig blockchain *blockchain.Blockchain @@ -27,7 +28,7 @@ type VotingMachine struct { } func New( - logger logging.Logger, + logger logging.Logger2, el *eventloop.EventLoop, config *core.RuntimeConfig, blockchain *blockchain.Blockchain, @@ -52,7 +53,7 @@ func New( // CollectVote handles an incoming vote. func (vm *VotingMachine) CollectVote(vote hotstuff.VoteMsg) { cert := vote.PartialCert - vm.logger.Debugf("CollectVote(from %d): %s", vote.ID, cert.BlockHash().SmallString()) + vm.logger.Debug("CollectVote", zap.Uint32("from", uint32(vote.ID)), zap.String("hash", cert.BlockHash().SmallString())) var ( block *hotstuff.Block ok bool @@ -63,7 +64,7 @@ func (vm *VotingMachine) CollectVote(vote hotstuff.VoteMsg) { if !ok { // if that does not work, we will try to handle this event later. // hopefully, the block has arrived by then. - vm.logger.Debugf("Local cache miss for block: %s", cert.BlockHash().SmallString()) + vm.logger.Debug("Local cache miss for block", zap.String("hash", cert.BlockHash().SmallString())) vote.Deferred = true eventloop.DelayUntil[hotstuff.ProposeMsg](vm.eventLoop, vote) return @@ -72,7 +73,7 @@ func (vm *VotingMachine) CollectVote(vote hotstuff.VoteMsg) { // if the block has not arrived at this point we will try to fetch it. block, ok = vm.blockchain.Get(cert.BlockHash()) if !ok { - vm.logger.Debugf("Could not find block for vote: %s.", cert.BlockHash().SmallString()) + vm.logger.Debug("Could not find block for vote", zap.String("hash", cert.BlockHash().SmallString())) return } } @@ -89,7 +90,7 @@ func (vm *VotingMachine) CollectVote(vote hotstuff.VoteMsg) { func (vm *VotingMachine) verifyCert(cert hotstuff.PartialCert, block *hotstuff.Block) { if err := vm.auth.VerifyPartialCert(cert); err != nil { - vm.logger.Infof("vote could not be verified: %v", err) + vm.logger.Info("vote could not be verified", zap.Error(err)) return } vm.mut.Lock() @@ -111,7 +112,7 @@ func (vm *VotingMachine) verifyCert(cert hotstuff.PartialCert, block *hotstuff.B // Check for duplicate votes from the same signer for _, v := range votes { if v.Signer() == cert.Signer() { - vm.logger.Debugf("Ignoring duplicate vote from signer %d", cert.Signer()) + vm.logger.Debug("Ignoring duplicate vote from signer", zap.Uint32("signer", uint32(cert.Signer()))) return } } @@ -122,10 +123,10 @@ func (vm *VotingMachine) verifyCert(cert hotstuff.PartialCert, block *hotstuff.B } qc, err := vm.auth.CreateQuorumCert(block, votes) if err != nil { - vm.logger.Infof("CollectVote: could not create QC for block: %v", err) + vm.logger.Info("CollectVote: could not create QC for block", zap.Error(err)) return } delete(vm.verifiedVotes, cert.BlockHash()) - vm.logger.Debugf("CollectVote: dispatching event for new view (current : %d)", vm.state.View()) + vm.logger.Debug("CollectVote: dispatching event for new view", zap.Uint64("current", uint64(vm.state.View()))) vm.eventLoop.AddEvent(hotstuff.NewViewMsg{ID: vm.config.ID(), SyncInfo: hotstuff.NewSyncInfoWith(qc)}) } diff --git a/scripts/loki/docker-compose.yml b/scripts/loki/docker-compose.yml new file mode 100644 index 00000000..2d9b0485 --- /dev/null +++ b/scripts/loki/docker-compose.yml @@ -0,0 +1,53 @@ +# Grafana Loki stack for HotStuff log aggregation. +# +# Usage: +# cd scripts/loki +# docker compose up -d +# +# Access: +# Grafana: http://localhost:3000 (admin / admin) +# Loki API: http://localhost:3100 +# +# Then run HotStuff with Loki enabled: +# ./hotstuff run --config scripts/local_config.cue \ +# --ssh-config scripts/ssh_config.local \ +# --log-level debug \ +# --loki-url http://localhost:3100/loki/api/v1/push \ +# --loki-labels node=controller,env=local + +services: + loki: + image: grafana/loki:3.4.2 + ports: + - "3100:3100" + volumes: + - ./loki-config.yaml:/etc/loki/local-config.yaml:ro + - loki-data:/loki + command: -config.file=/etc/loki/local-config.yaml + restart: unless-stopped + healthcheck: + test: [ "CMD-SHELL", "wget --no-verbose --tries=1 --spider http://localhost:3100/ready || exit 1" ] + interval: 10s + timeout: 5s + retries: 5 + + grafana: + image: grafana/grafana:11.5.2 + ports: + - "3000:3000" + environment: + - GF_SECURITY_ADMIN_USER=admin + - GF_SECURITY_ADMIN_PASSWORD=admin + - GF_AUTH_ANONYMOUS_ENABLED=true + - GF_AUTH_ANONYMOUS_ORG_ROLE=Admin + volumes: + - ./provisioning:/etc/grafana/provisioning:ro + - grafana-data:/var/lib/grafana + depends_on: + loki: + condition: service_healthy + restart: unless-stopped + +volumes: + loki-data: + grafana-data: diff --git a/scripts/loki/loki-config.yaml b/scripts/loki/loki-config.yaml new file mode 100644 index 00000000..b19ba198 --- /dev/null +++ b/scripts/loki/loki-config.yaml @@ -0,0 +1,47 @@ +# Loki configuration for local development with HotStuff. +# Docs: https://grafana.com/docs/loki/latest/configure/ + +auth_enabled: false + +server: + http_listen_port: 3100 + grpc_listen_port: 9096 + log_level: warn + +common: + instance_addr: 127.0.0.1 + path_prefix: /loki + storage: + filesystem: + chunks_directory: /loki/chunks + rules_directory: /loki/rules + replication_factor: 1 + ring: + kvstore: + store: inmemory + +query_range: + results_cache: + cache: + embedded_cache: + enabled: true + max_size_mb: 100 + +schema_config: + configs: + - from: "2020-10-24" + store: tsdb + object_store: filesystem + schema: v13 + index: + prefix: index_ + period: 24h + +limits_config: + reject_old_samples: true + reject_old_samples_max_age: 168h + ingestion_rate_mb: 16 + ingestion_burst_size_mb: 32 + +analytics: + reporting_enabled: false diff --git a/scripts/loki/provisioning/dashboards/dashboards.yaml b/scripts/loki/provisioning/dashboards/dashboards.yaml new file mode 100644 index 00000000..893bda44 --- /dev/null +++ b/scripts/loki/provisioning/dashboards/dashboards.yaml @@ -0,0 +1,13 @@ +# Grafana dashboard provisioning: auto-load HotStuff dashboards. +apiVersion: 1 + +providers: + - name: HotStuff + orgId: 1 + folder: HotStuff + type: file + disableDeletion: false + editable: true + options: + path: /etc/grafana/provisioning/dashboards/json + foldersFromFilesStructure: false diff --git a/scripts/loki/provisioning/dashboards/json/hotstuff-logs.json b/scripts/loki/provisioning/dashboards/json/hotstuff-logs.json new file mode 100644 index 00000000..d9ac4fff --- /dev/null +++ b/scripts/loki/provisioning/dashboards/json/hotstuff-logs.json @@ -0,0 +1,239 @@ +{ + "annotations": { + "list": [] + }, + "editable": true, + "fiscalYearStartMonth": 0, + "graphTooltip": 1, + "id": null, + "links": [], + "panels": [ + { + "title": "Log Volume by Level", + "type": "timeseries", + "gridPos": { + "h": 8, + "w": 24, + "x": 0, + "y": 0 + }, + "datasource": { + "type": "loki", + "uid": "" + }, + "targets": [ + { + "expr": "sum by (level) (count_over_time({app=\"hotstuff\"} | json | __error__=\"\" [1m]))", + "legendFormat": "{{ level }}", + "refId": "A" + } + ], + "fieldConfig": { + "defaults": { + "custom": { + "drawStyle": "bars", + "stacking": { + "mode": "normal" + }, + "fillOpacity": 80 + } + }, + "overrides": [ + { + "matcher": { + "id": "byName", + "options": "error" + }, + "properties": [ + { + "id": "color", + "value": { + "fixedColor": "red", + "mode": "fixed" + } + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "warn" + }, + "properties": [ + { + "id": "color", + "value": { + "fixedColor": "orange", + "mode": "fixed" + } + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "info" + }, + "properties": [ + { + "id": "color", + "value": { + "fixedColor": "green", + "mode": "fixed" + } + } + ] + }, + { + "matcher": { + "id": "byName", + "options": "debug" + }, + "properties": [ + { + "id": "color", + "value": { + "fixedColor": "blue", + "mode": "fixed" + } + } + ] + } + ] + } + }, + { + "title": "All Logs", + "type": "logs", + "gridPos": { + "h": 16, + "w": 24, + "x": 0, + "y": 8 + }, + "datasource": { + "type": "loki", + "uid": "" + }, + "targets": [ + { + "expr": "{app=\"hotstuff\"} | json", + "refId": "A" + } + ], + "options": { + "showTime": true, + "showLabels": true, + "showCommonLabels": false, + "wrapLogMessage": true, + "prettifyLogMessage": true, + "enableLogDetails": true, + "sortOrder": "Descending", + "dedupStrategy": "none" + } + }, + { + "title": "Error Logs", + "type": "logs", + "gridPos": { + "h": 12, + "w": 24, + "x": 0, + "y": 24 + }, + "datasource": { + "type": "loki", + "uid": "" + }, + "targets": [ + { + "expr": "{app=\"hotstuff\"} | json | level=\"error\"", + "refId": "A" + } + ], + "options": { + "showTime": true, + "showLabels": true, + "showCommonLabels": false, + "wrapLogMessage": true, + "prettifyLogMessage": true, + "enableLogDetails": true, + "sortOrder": "Descending", + "dedupStrategy": "none" + } + }, + { + "title": "Logs by Logger Name", + "type": "logs", + "gridPos": { + "h": 12, + "w": 24, + "x": 0, + "y": 36 + }, + "datasource": { + "type": "loki", + "uid": "" + }, + "targets": [ + { + "expr": "{app=\"hotstuff\"} | json | logger=~\"$logger\"", + "refId": "A" + } + ], + "options": { + "showTime": true, + "showLabels": true, + "showCommonLabels": false, + "wrapLogMessage": true, + "prettifyLogMessage": true, + "enableLogDetails": true, + "sortOrder": "Descending", + "dedupStrategy": "none" + } + } + ], + "refresh": "5s", + "schemaVersion": 39, + "tags": [ + "hotstuff", + "loki" + ], + "templating": { + "list": [ + { + "current": { + "selected": false, + "text": ".*", + "value": ".*" + }, + "description": "Filter by logger name (e.g. ctrl, json, database)", + "label": "Logger", + "name": "logger", + "query": ".*", + "type": "textbox" + }, + { + "current": { + "selected": false, + "text": "", + "value": "" + }, + "description": "Filter by node label", + "label": "Node", + "name": "node", + "query": "", + "type": "textbox" + } + ] + }, + "time": { + "from": "now-15m", + "to": "now" + }, + "timepicker": {}, + "timezone": "browser", + "title": "HotStuff Logs", + "uid": "hotstuff-logs", + "version": 1 +} diff --git a/scripts/loki/provisioning/datasources/loki.yaml b/scripts/loki/provisioning/datasources/loki.yaml new file mode 100644 index 00000000..7b52bdb4 --- /dev/null +++ b/scripts/loki/provisioning/datasources/loki.yaml @@ -0,0 +1,12 @@ +# Grafana datasource provisioning: auto-configure Loki as a datasource. +apiVersion: 1 + +datasources: + - name: Loki + type: loki + access: proxy + url: http://loki:3100 + isDefault: true + editable: true + jsonData: + maxLines: 1000 diff --git a/security/blockchain/blockchain.go b/security/blockchain/blockchain.go index 62abbc80..fae221e0 100644 --- a/security/blockchain/blockchain.go +++ b/security/blockchain/blockchain.go @@ -10,6 +10,7 @@ import ( "github.com/relab/hotstuff/core" "github.com/relab/hotstuff/core/eventloop" "github.com/relab/hotstuff/core/logging" + "go.uber.org/zap" ) // Blockchain stores a limited amount of blocks in a map. @@ -17,7 +18,7 @@ import ( type Blockchain struct { sender core.Sender eventLoop *eventloop.EventLoop - logger logging.Logger + logger logging.Logger2 mut sync.Mutex pruneHeight hotstuff.View @@ -31,7 +32,7 @@ type Blockchain struct { // Blocks are dropped in least recently used order. func New( eventLoop *eventloop.EventLoop, - logger logging.Logger, + logger logging.Logger2, sender core.Sender, ) *Blockchain { bc := &Blockchain{ @@ -54,7 +55,7 @@ func (chain *Blockchain) Store(block *hotstuff.Block) { // do not store existing blocks, otherwise something is terribly wrong. if _, ok := chain.blocks[block.Hash()]; ok { - chain.logger.Warnf("block already exists: %s", block.String()) + chain.logger.Warn("block already exists", zap.String("block", block.String())) return } @@ -113,7 +114,7 @@ func (chain *Blockchain) Get(hash hotstuff.Hash) (block *hotstuff.Block, ok bool chain.pendingFetch[hash] = cancel chain.mut.Unlock() - chain.logger.Debugf("Attempting to fetch block: %s", hash.SmallString()) + chain.logger.Debug("Attempting to fetch block", zap.String("hash", hash.SmallString())) block, ok = chain.sender.RequestBlock(ctx, hash) chain.mut.Lock() @@ -124,7 +125,7 @@ func (chain *Blockchain) Get(hash hotstuff.Hash) (block *hotstuff.Block, ok bool goto done } - chain.logger.Debugf("Successfully fetched block: %s", hash.SmallString()) + chain.logger.Debug("Successfully fetched block", zap.String("hash", hash.SmallString())) chain.blocks[hash] = block chain.blockAtHeight[block.View()] = block @@ -173,7 +174,7 @@ func (chain *Blockchain) PruneToHeight(committedHeight, height hotstuff.View) (f if !committedViews[h] { block, ok := chain.blockAtHeight[h] if ok { - chain.logger.Debugf("PruneToHeight: found forked block: %v", block) + chain.logger.Debug("PruneToHeight: found forked block", zap.Stringer("block", block)) forkedBlocks = append(forkedBlocks, block) } } diff --git a/server/clientio.go b/server/clientio.go index 8f88871e..7373d293 100644 --- a/server/clientio.go +++ b/server/clientio.go @@ -10,6 +10,7 @@ import ( "github.com/relab/hotstuff/core/eventloop" "github.com/relab/hotstuff/core/logging" "github.com/relab/hotstuff/internal/proto/clientpb" + "go.uber.org/zap" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" "google.golang.org/protobuf/types/known/emptypb" @@ -17,7 +18,7 @@ import ( // ClientIO serves a client. type ClientIO struct { - logger logging.Logger + logger logging.Logger2 cmdCache *clientpb.CommandCache mut sync.Mutex @@ -32,7 +33,7 @@ type ClientIO struct { // NewClientIO returns a new client IO server. func NewClientIO( el *eventloop.EventLoop, - logger logging.Logger, + logger logging.Logger2, cmdCache *clientpb.CommandCache, srvOpts ...gorums.ServerOption, ) (srv *ClientIO) { @@ -59,7 +60,7 @@ func (srv *ClientIO) StartOnListener(lis net.Listener) { go func() { err := srv.srv.Serve(lis) if err != nil { - srv.logger.Error(err) + srv.logger.Error("Server error", zap.Error(err)) } }() } @@ -107,7 +108,7 @@ func (srv *ClientIO) Exec(batch *clientpb.Batch) { srv.completeCommand(id, nil) srv.mut.Unlock() } - srv.logger.Debugf("Hash: %.8x", srv.hash.Sum(nil)) + srv.logger.Debug("Hash", zap.Binary("hash", srv.hash.Sum(nil))) } func (srv *ClientIO) Abort(batch *clientpb.Batch) { diff --git a/server/server.go b/server/server.go index dc97a550..3aa5cb50 100644 --- a/server/server.go +++ b/server/server.go @@ -16,6 +16,7 @@ import ( "github.com/relab/hotstuff" "github.com/relab/hotstuff/internal/latency" "github.com/relab/hotstuff/internal/proto/hotstuffpb" + "go.uber.org/zap" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" ) @@ -25,7 +26,7 @@ import ( type Server struct { blockchain *blockchain.Blockchain eventLoop *eventloop.EventLoop - logger logging.Logger + logger logging.Logger2 config *core.RuntimeConfig id hotstuff.ID @@ -36,7 +37,7 @@ type Server struct { // NewServer creates a new Server. func NewServer( eventLoop *eventloop.EventLoop, - logger logging.Logger, + logger logging.Logger2, config *core.RuntimeConfig, blockchain *blockchain.Blockchain, srvOpts ...ServerOption, @@ -73,7 +74,7 @@ func (srv *Server) addNetworkDelay(sender hotstuff.ID) { return } delay := srv.lm.Latency(srv.id, sender) - srv.logger.Debugf("Delay between %s and %s: %v\n", srv.lm.Location(srv.id), srv.lm.Location(sender), delay) + srv.logger.Debug("Delay between locations", zap.String("local", srv.lm.Location(srv.id)), zap.String("remote", srv.lm.Location(sender)), zap.Duration("delay", delay)) srv.lm.Delay(srv.id, sender) } @@ -97,7 +98,7 @@ func (srv *Server) StartOnListener(listener net.Listener) { go func() { err := srv.gorumsSrv.Serve(listener) if err != nil { - srv.logger.Errorf("An error occurred while serving: %v", err) + srv.logger.Error("An error occurred while serving", zap.Error(err)) } }() } @@ -116,7 +117,7 @@ type serviceImpl struct { func (impl *serviceImpl) Propose(ctx gorums.ServerCtx, proposal *hotstuffpb.Proposal) { id, err := impl.srv.config.PeerIDFromContext(ctx) if err != nil { - impl.srv.logger.Warnf("Could not get replica ID: %v", err) + impl.srv.logger.Warn("Could not get replica ID", zap.Error(err)) return } if impl.srv.config.HasKauriTree() { @@ -133,7 +134,7 @@ func (impl *serviceImpl) Propose(ctx gorums.ServerCtx, proposal *hotstuffpb.Prop func (impl *serviceImpl) Vote(ctx gorums.ServerCtx, cert *hotstuffpb.PartialCert) { id, err := impl.srv.config.PeerIDFromContext(ctx) if err != nil { - impl.srv.logger.Warnf("Could not get replica ID: %v", err) + impl.srv.logger.Warn("Could not get replica ID", zap.Error(err)) return } impl.srv.addNetworkDelay(id) @@ -147,7 +148,7 @@ func (impl *serviceImpl) Vote(ctx gorums.ServerCtx, cert *hotstuffpb.PartialCert func (impl *serviceImpl) NewView(ctx gorums.ServerCtx, msg *hotstuffpb.SyncInfo) { id, err := impl.srv.config.PeerIDFromContext(ctx) if err != nil { - impl.srv.logger.Warnf("Could not get replica ID: %v", err) + impl.srv.logger.Warn("Could not get replica ID", zap.Error(err)) return } impl.srv.addNetworkDelay(id) @@ -167,7 +168,7 @@ func (impl *serviceImpl) RequestBlock(_ gorums.ServerCtx, pb *hotstuffpb.BlockHa return nil, status.Errorf(codes.NotFound, "requested block was not found") } - impl.srv.logger.Debugf("On RequestBlock: %s", hash.SmallString()) + impl.srv.logger.Debug("On RequestBlock", zap.String("hash", hash.SmallString())) return hotstuffpb.BlockToProto(block), nil } @@ -176,7 +177,7 @@ func (impl *serviceImpl) RequestBlock(_ gorums.ServerCtx, pb *hotstuffpb.BlockHa func (impl *serviceImpl) Timeout(ctx gorums.ServerCtx, msg *hotstuffpb.TimeoutMsg) { id, err := impl.srv.config.PeerIDFromContext(ctx) if err != nil { - impl.srv.logger.Warnf("Could not get replica ID: %v", err) + impl.srv.logger.Warn("Could not get replica ID", zap.Error(err)) } timeoutMsg := hotstuffpb.TimeoutMsgFromProto(msg) timeoutMsg.ID = id diff --git a/twins/generator.go b/twins/generator.go index ab60ad36..82f8fa06 100644 --- a/twins/generator.go +++ b/twins/generator.go @@ -8,12 +8,13 @@ import ( "github.com/relab/hotstuff" "github.com/relab/hotstuff/core/logging" + "go.uber.org/zap" ) // Generator generates twins scenarios. type Generator struct { mut sync.Mutex - logger logging.Logger + logger logging.Logger2 remaining int64 allNodes []NodeID indices []int @@ -44,7 +45,7 @@ func assignNodeIDs(numNodes, numTwins uint8) (nodes, twins []NodeID) { } // NewGenerator creates a new generator. -func NewGenerator(logger logging.Logger, settings Settings) *Generator { +func NewGenerator(logger logging.Logger2, settings Settings) *Generator { g := &Generator{ logger: logger, allNodes: make([]NodeID, 0, settings.NumNodes+settings.NumTwins), @@ -73,10 +74,7 @@ func NewGenerator(logger logging.Logger, settings Settings) *Generator { g.remaining = int64(math.Pow(float64(len(g.leadersPartitions)), float64(g.settings.Views))) - g.logger.Infof( - "%d scenarios can be generated with current settings.", - g.remaining, - ) + g.logger.Info("scenarios can be generated with current settings", zap.Int64("count", g.remaining)) return g } diff --git a/twins/generator_test.go b/twins/generator_test.go index 6194632b..a1d4ea49 100644 --- a/twins/generator_test.go +++ b/twins/generator_test.go @@ -16,7 +16,7 @@ func TestPartitionsGenerator(t *testing.T) { } func TestGenerator(t *testing.T) { - g := NewGenerator(logging.New(""), Settings{ + g := NewGenerator(logging.New2(""), Settings{ NumNodes: 4, NumTwins: 1, Partitions: 3, @@ -66,11 +66,11 @@ func TestAssignNodeIDs(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { gotNodes, gotTwins := assignNodeIDs(tt.numNodes, tt.numTwins) - + if !reflect.DeepEqual(gotNodes, tt.wantNodes) { t.Errorf("assignNodeIDs() gotNodes = %v, want %v", gotNodes, tt.wantNodes) } - + if !reflect.DeepEqual(gotTwins, tt.wantTwins) { t.Errorf("assignNodeIDs() gotTwins = %v, want %v", gotTwins, tt.wantTwins) } diff --git a/twins/node.go b/twins/node.go index 5403af84..a6f34c56 100644 --- a/twins/node.go +++ b/twins/node.go @@ -24,7 +24,7 @@ import ( type node struct { config *core.RuntimeConfig - logger logging.Logger + logger logging.Logger2 sender *emulatedSender blockchain *blockchain.Blockchain commandCache *clientpb.CommandCache @@ -50,7 +50,7 @@ func newNode(n *Network, nodeID NodeID, consensusName string, pk *ecdsa.PrivateK config: core.NewRuntimeConfig(nodeID.ReplicaID, pk, allOpts...), commandCache: clientpb.NewCommandCache(1), } - node.logger = logging.NewWithDest(&n.log, fmt.Sprintf("r%dn%d", nodeID.ReplicaID, nodeID.TwinID)) + node.logger = logging.New2WithDest(&n.log, fmt.Sprintf("r%dn%d", nodeID.ReplicaID, nodeID.TwinID)) node.eventLoop = eventloop.New(node.logger, 100) node.sender = newSender(n, node) base, err := crypto.New( diff --git a/twins/timeoutmgr.go b/twins/timeoutmgr.go index 00d1b2f7..4e7c0e37 100644 --- a/twins/timeoutmgr.go +++ b/twins/timeoutmgr.go @@ -4,6 +4,7 @@ import ( "github.com/relab/hotstuff" "github.com/relab/hotstuff/core/eventloop" "github.com/relab/hotstuff/protocol" + "go.uber.org/zap" ) type timeoutManager struct { @@ -24,7 +25,7 @@ func (tm *timeoutManager) advance() { tm.countdown = tm.timeout if tm.node.effectiveView <= view { tm.node.effectiveView = view + 1 - tm.network.logger.Infof("node %v effective view is %d due to timeout", tm.node.id, tm.node.effectiveView) + tm.network.logger.Info("node effective view updated due to timeout", zap.Stringer("node", tm.node.id), zap.Uint64("view", uint64(tm.node.effectiveView))) } } } diff --git a/twins/twins_test.go b/twins/twins_test.go index dedcab5b..e76efe8c 100644 --- a/twins/twins_test.go +++ b/twins/twins_test.go @@ -15,7 +15,7 @@ func TestTwins(t *testing.T) { numTwins = 1 ) - g := twins.NewGenerator(logging.New(""), twins.Settings{ + g := twins.NewGenerator(logging.New2(""), twins.Settings{ NumNodes: numNodes, NumTwins: numTwins, Partitions: 2, diff --git a/twins/twinsrules.go b/twins/twinsrules.go index df8d86aa..2b4309b0 100644 --- a/twins/twinsrules.go +++ b/twins/twinsrules.go @@ -9,7 +9,7 @@ import ( ) func newTwinsConsensusRules( - logger logging.Logger, + logger logging.Logger2, config *core.RuntimeConfig, blockchain *blockchain.Blockchain, name string, diff --git a/twins/vulnfhs.go b/twins/vulnfhs.go index faee3614..57a54ad4 100644 --- a/twins/vulnfhs.go +++ b/twins/vulnfhs.go @@ -6,19 +6,20 @@ import ( "github.com/relab/hotstuff/protocol/consensus" "github.com/relab/hotstuff/protocol/rules" "github.com/relab/hotstuff/security/blockchain" + "go.uber.org/zap" ) const nameVulnerableFHS = "vulnerableFHS" // A wrapper around the FHS rules that swaps the commit rule for a vulnerable version type vulnerableFHS struct { - logger logging.Logger + logger logging.Logger2 blockchain *blockchain.Blockchain rules.FastHotStuff } func NewVulnFHS( - logger logging.Logger, + logger logging.Logger2, blockchain *blockchain.Blockchain, inner *rules.FastHotStuff, ) *vulnerableFHS { @@ -42,7 +43,7 @@ func (fhs *vulnerableFHS) CommitRule(block *hotstuff.Block) *hotstuff.Block { if !ok { return nil } - fhs.logger.Debug("PRECOMMIT: ", parent) + fhs.logger.Debug("PRECOMMIT", zap.Stringer("block", parent)) grandparent, ok := fhs.qcRef(parent.QuorumCert()) if !ok { return nil @@ -50,7 +51,7 @@ func (fhs *vulnerableFHS) CommitRule(block *hotstuff.Block) *hotstuff.Block { // NOTE: this does check for a direct link between the block and the grandparent. // This is what causes the safety violation. if block.Parent() == parent.Hash() && parent.Parent() == grandparent.Hash() { - fhs.logger.Debug("COMMIT(vulnerable): ", grandparent) + fhs.logger.Debug("COMMIT(vulnerable)", zap.Stringer("block", grandparent)) return grandparent } return nil diff --git a/wiring/client.go b/wiring/client.go index 1376aabe..e7797c8e 100644 --- a/wiring/client.go +++ b/wiring/client.go @@ -16,7 +16,7 @@ type Client struct { // NewClient returns a set of dependencies for serving clients through func NewClient( eventLoop *eventloop.EventLoop, - logger logging.Logger, + logger logging.Logger2, commandBatchSize uint32, clientSrvOpts ...gorums.ServerOption, ) *Client { diff --git a/wiring/consensus.go b/wiring/consensus.go index f95d98f6..fdf95215 100644 --- a/wiring/consensus.go +++ b/wiring/consensus.go @@ -21,7 +21,7 @@ type Consensus struct { func NewConsensus( eventLoop *eventloop.EventLoop, - logger logging.Logger, + logger logging.Logger2, config *core.RuntimeConfig, blockchain *blockchain.Blockchain, auth *cert.Authority, diff --git a/wiring/core.go b/wiring/core.go index 92ddb401..e86b462e 100644 --- a/wiring/core.go +++ b/wiring/core.go @@ -7,11 +7,12 @@ import ( "github.com/relab/hotstuff/core" "github.com/relab/hotstuff/core/eventloop" "github.com/relab/hotstuff/core/logging" + "go.uber.org/zap" ) type Core struct { eventLoop *eventloop.EventLoop - logger logging.Logger + logger logging.Logger2 config *core.RuntimeConfig } @@ -21,7 +22,11 @@ func NewCore( privKey hotstuff.PrivateKey, opts ...core.RuntimeOption, ) *Core { - logger := logging.New(fmt.Sprintf("%s%d", logTag, id)) + logger := logging.New2(fmt.Sprintf("%s%d", logTag, id)) + logger.Info("Initializing core", + zap.Uint32("replica_id", uint32(id)), + zap.String("log_tag", logTag), + ) return &Core{ config: core.NewRuntimeConfig(id, privKey, opts...), eventLoop: eventloop.New(logger, 100), @@ -35,7 +40,7 @@ func (c *Core) EventLoop() *eventloop.EventLoop { } // Logger returns the logger instance. -func (c *Core) Logger() logging.Logger { +func (c *Core) Logger() logging.Logger2 { return c.logger } diff --git a/wiring/security.go b/wiring/security.go index 84b90154..2bde7020 100644 --- a/wiring/security.go +++ b/wiring/security.go @@ -17,7 +17,7 @@ type Security struct { // NewSecurity returns a set of dependencies necessary for application security and integrity. func NewSecurity( eventLoop *eventloop.EventLoop, - logger logging.Logger, + logger logging.Logger2, config *core.RuntimeConfig, sender core.Sender, base crypto.Base,