ConsensusLatency: Add consensus latency metrics

block.go

@@ -4,6 +4,7 @@ import (
 	"crypto/sha256"
 	"encoding/binary"
 	"fmt"
+	"time"
 )
 
 // Block contains a propsed "command", metadata for the protocol, and a link to the "parent" block.
@@ -15,6 +16,7 @@ type Block struct {
 	cmd      Command
 	cert     QuorumCert
 	view     View
+	ts       time.Time
 }
 
 // NewBlock creates a new Block
@@ -25,6 +27,7 @@ func NewBlock(parent Hash, cert QuorumCert, cmd Command, view View, proposer ID)
 		cmd:      cmd,
 		view:     view,
 		proposer: proposer,
+		ts:       time.Now(),
 	}
 	// cache the hash immediately because it is too racy to do it in Hash()
 	b.hash = sha256.Sum256(b.ToBytes())
@@ -72,6 +75,11 @@ func (b *Block) View() View {
 	return b.view
 }
 
+// TimeStamp returns the timestamp of the block
+func (b *Block) TimeStamp() time.Time {
+	return b.ts
+}
+
 // ToBytes returns the raw byte form of the Block, to be used for hashing, etc.
 func (b *Block) ToBytes() []byte {
 	buf := b.parent[:]

consensus/consensus.go

@@ -3,6 +3,7 @@ package consensus
 import (
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/relab/hotstuff"
 	"github.com/relab/hotstuff/eventloop"
@@ -282,6 +283,7 @@ func (cs *consensusBase) commitInner(block *hotstuff.Block) error {
 	} else {
 		return fmt.Errorf("failed to locate block: %s", block.Parent())
 	}
+	cs.eventLoop.AddEvent(hotstuff.ConsensusLatencyEvent{Latency: time.Since(block.TimeStamp())})
 	cs.logger.Debug("EXEC: ", block)
 	cs.executor.Exec(block)
 	cs.bExec = block

metrics/consensuslatency.go

@@ -0,0 +1,70 @@
+package metrics
+
+import (
+	"time"
+
+	"github.com/relab/hotstuff"
+	"github.com/relab/hotstuff/eventloop"
+	"github.com/relab/hotstuff/logging"
+	"github.com/relab/hotstuff/metrics/types"
+	"github.com/relab/hotstuff/modules"
+)
+
+func init() {
+	RegisterReplicaMetric("consensus-latency", func() any {
+		return &ConsensusLatency{}
+	})
+}
+
+// ConsensusLatency processes consensus latency measurements and writes them to the metrics logger.
+type ConsensusLatency struct {
+	metricsLogger Logger
+	id            hotstuff.ID
+	wf            Welford
+}
+
+// InitModule gives the module access to the other modules.
+func (lr *ConsensusLatency) InitModule(mods *modules.Core) {
+	var (
+		eventLoop *eventloop.EventLoop
+		logger    logging.Logger
+		opts      *modules.Options
+	)
+
+	mods.Get(
+		&lr.metricsLogger,
+		opts,
+		&eventLoop,
+		&logger,
+	)
+
+	lr.id = opts.ID()
+	eventLoop.RegisterHandler(hotstuff.ConsensusLatencyEvent{}, func(event any) {
+		latencyEvent := event.(hotstuff.ConsensusLatencyEvent)
+		lr.addLatency(latencyEvent.Latency)
+	})
+
+	eventLoop.RegisterHandler(types.TickEvent{}, func(event any) {
+		lr.tick(event.(types.TickEvent))
+	}, eventloop.Prioritize())
+
+	logger.Info("Consensus Latency metric enabled")
+}
+
+// AddLatency adds a latency data point to the current measurement.
+func (lr *ConsensusLatency) addLatency(latency time.Duration) {
+	millis := float64(latency) / float64(time.Millisecond)
+	lr.wf.Update(millis)
+}
+
+func (lr *ConsensusLatency) tick(_ types.TickEvent) {
+	mean, variance, count := lr.wf.Get()
+	event := &types.LatencyMeasurement{
+		Event:    types.NewReplicaEvent(uint32(lr.id), time.Now()),
+		Latency:  mean,
+		Variance: variance,
+		Count:    count,
+	}
+	lr.metricsLogger.Log(event)
+	lr.wf.Reset()
+}

types.go

@@ -9,6 +9,7 @@ import (
 	"io"
 	"strconv"
 	"strings"
+	"time"
 )
 
 // IDSet implements a set of replica IDs. It is used to show which replicas participated in some event.
@@ -383,3 +384,7 @@ func writeParticipants(wr io.Writer, participants IDSet) (err error) {
 	})
 	return err
 }
+
+type ConsensusLatencyEvent struct {
+	Latency time.Duration
+}