gossip: add digest/delta metrics (#154)

Adds gossip metrics for incoming/outgoing digest and delta entries.
andydunstall · Aug 15, 2024 · 689648d · 689648d
1 parent fa28e5d
commit 689648d
Show file tree

Hide file tree

Showing 4 changed files with 77 additions and 4 deletions.
diff --git a/pkg/gossip/gossip.go b/pkg/gossip/gossip.go
@@ -381,6 +381,8 @@ func (g *Gossip) gossip(node NodeMetadata) error {
 			break
 		}
 		bufLen = buf.Len()
+
+		g.metrics.DigestEntriesOutbound.Inc()
 	}
 
 	udpAddr, err := net.ResolveUDPAddr("udp", node.Addr)
@@ -438,13 +440,17 @@ func (g *Gossip) join(addr string) (string, error) {
 		return "", fmt.Errorf("encode: %w", err)
 	}
 
+	delta := g.state.LocalDelta()
 	if err := encoder.Encode(g.state.LocalDelta()); err != nil {
 		return "", fmt.Errorf("encode: %w", err)
 	}
+	g.metrics.DeltaEntriesOutbound.Add(float64(delta.EntriesTotal()))
 
-	if err := encoder.Encode(g.state.Digest()); err != nil {
+	digest := g.state.Digest()
+	if err := encoder.Encode(digest); err != nil {
 		return "", fmt.Errorf("encode: %w", err)
 	}
+	g.metrics.DigestEntriesOutbound.Add(float64(len(digest)))
 
 	if err := w.Flush(); err != nil {
 		return "", fmt.Errorf("flush: %w", err)
@@ -457,10 +463,10 @@ func (g *Gossip) join(addr string) (string, error) {
 		return "", fmt.Errorf("decode: %w", err)
 	}
 
-	var delta delta
 	if err := decoder.Decode(&delta); err != nil {
 		return "", fmt.Errorf("decode: %w", err)
 	}
+	g.metrics.DeltaEntriesInbound.Add(float64(delta.EntriesTotal()))
 
 	g.state.ApplyDelta(delta)
 
@@ -509,9 +515,11 @@ func (g *Gossip) leave(addr string) error {
 		return fmt.Errorf("encode: %w", err)
 	}
 
-	if err := encoder.Encode(g.state.LocalDelta()); err != nil {
+	delta := g.state.LocalDelta()
+	if err := encoder.Encode(delta); err != nil {
 		return fmt.Errorf("encode: %w", err)
 	}
+	g.metrics.DeltaEntriesOutbound.Add(float64(delta.EntriesTotal()))
 
 	if err := w.Flush(); err != nil {
 		return fmt.Errorf("flush: %w", err)

diff --git a/pkg/gossip/listener.go b/pkg/gossip/listener.go
@@ -131,10 +131,12 @@ func (l *streamListener) join(r io.Reader, w *bufio.Writer) error {
 	if err := decoder.Decode(&delta); err != nil {
 		return fmt.Errorf("decode: %w", err)
 	}
+	l.metrics.DeltaEntriesInbound.Add(float64(delta.EntriesTotal()))
 	var digest digest
 	if err := decoder.Decode(&digest); err != nil {
 		return fmt.Errorf("decode: %w", err)
 	}
+	l.metrics.DigestEntriesInbound.Add(float64(len(digest)))
 
 	// Apply unknown state from the delta.
 	l.state.ApplyDelta(delta)
@@ -156,6 +158,7 @@ func (l *streamListener) join(r io.Reader, w *bufio.Writer) error {
 	if err := encoder.Encode(delta); err != nil {
 		return fmt.Errorf("encode: %w", err)
 	}
+	l.metrics.DeltaEntriesOutbound.Add(float64(delta.EntriesTotal()))
 
 	if err := w.Flush(); err != nil {
 		return fmt.Errorf("flush: %w", err)
@@ -174,6 +177,7 @@ func (l *streamListener) leave(r io.Reader, w *bufio.Writer) error {
 	if err := decoder.Decode(&delta); err != nil {
 		return fmt.Errorf("decode: %w", err)
 	}
+	l.metrics.DeltaEntriesInbound.Add(float64(delta.EntriesTotal()))
 
 	// Apply unknown state from the delta.
 	l.state.ApplyDelta(delta)
@@ -285,6 +289,7 @@ func (l *packetListener) digest(b []byte) error {
 	if err != nil {
 		return fmt.Errorf("decode: %w", err)
 	}
+	l.metrics.DigestEntriesInbound.Add(float64(len(digest)))
 
 	// Discover any unknown nodes from the digest.
 	l.state.ApplyDigest(digest)
@@ -293,16 +298,19 @@ func (l *packetListener) digest(b []byte) error {
 	if err := l.sendDelta(delta, header.Addr); err != nil {
 		return fmt.Errorf("send delta: %w", err)
 	}
+	l.metrics.DeltaEntriesOutbound.Add(float64(delta.EntriesTotal()))
 
 	// If the digest was a request, send our own digest response.
 	if header.Request {
+		digest := l.state.Digest()
 		if err := l.sendDigest(
-			l.state.Digest(),
+			digest,
 			header.Addr,
 			false,
 		); err != nil {
 			return fmt.Errorf("send digest: %w", err)
 		}
+		l.metrics.DigestEntriesOutbound.Add(float64(len(digest)))
 	}
 
 	return nil
@@ -313,6 +321,7 @@ func (l *packetListener) delta(b []byte) error {
 	if err != nil {
 		return fmt.Errorf("decode: %w", err)
 	}
+	l.metrics.DeltaEntriesInbound.Add(float64(delta.EntriesTotal()))
 
 	l.failureDetector.Report(header.NodeID)
 

diff --git a/pkg/gossip/metrics.go b/pkg/gossip/metrics.go
@@ -27,6 +27,18 @@ type Metrics struct {
 	// connection.
 	PacketBytesOutbound prometheus.Counter
 
+	// DigestEntriesInbound is the total number of incoming digest entries.
+	DigestEntriesInbound prometheus.Counter
+
+	// DeltaEntriesInbound is the total number of incoming delta entries.
+	DeltaEntriesInbound prometheus.Counter
+
+	// DigestEntriesOutbound is the total number of outgoing digest entries.
+	DigestEntriesOutbound prometheus.Counter
+
+	// DeltaEntriesOutbound is the total number of outgoing delta entries.
+	DeltaEntriesOutbound prometheus.Counter
+
 	// Entries is the number of entries labelled by node_id, deleted and
 	// internal.
 	Entries *prometheus.GaugeVec
@@ -82,6 +94,38 @@ func newMetrics() *Metrics {
 				Help:      "Total number of written bytes via a packet connection",
 			},
 		),
+		DigestEntriesInbound: prometheus.NewCounter(
+			prometheus.CounterOpts{
+				Namespace: "piko",
+				Subsystem: "gossip",
+				Name:      "digest_entries_inbound_total",
+				Help:      "Total number of inbound digest entries",
+			},
+		),
+		DeltaEntriesInbound: prometheus.NewCounter(
+			prometheus.CounterOpts{
+				Namespace: "piko",
+				Subsystem: "gossip",
+				Name:      "delta_entries_inbound_total",
+				Help:      "Total number of inbound digest entries",
+			},
+		),
+		DigestEntriesOutbound: prometheus.NewCounter(
+			prometheus.CounterOpts{
+				Namespace: "piko",
+				Subsystem: "gossip",
+				Name:      "digest_entries_outbound_total",
+				Help:      "Total number of outbound digest entries",
+			},
+		),
+		DeltaEntriesOutbound: prometheus.NewCounter(
+			prometheus.CounterOpts{
+				Namespace: "piko",
+				Subsystem: "gossip",
+				Name:      "delta_entries_outbound_total",
+				Help:      "Total number of outbound delta entries",
+			},
+		),
 		Entries: prometheus.NewGaugeVec(
 			prometheus.GaugeOpts{
 				Namespace: "piko",
@@ -102,6 +146,10 @@ func (m *Metrics) Register(reg *prometheus.Registry) {
 		m.ConnectionsOutbound,
 		m.StreamBytesOutbound,
 		m.PacketBytesOutbound,
+		m.DigestEntriesInbound,
+		m.DeltaEntriesInbound,
+		m.DigestEntriesOutbound,
+		m.DeltaEntriesOutbound,
 		m.Entries,
 	)
 }
diff --git a/pkg/gossip/state.go b/pkg/gossip/state.go
@@ -80,6 +80,14 @@ type deltaEntry struct {
 
 type delta []deltaEntry
 
+func (d delta) EntriesTotal() int {
+	total := 0
+	for _, e := range d {
+		total += len(e.Entries)
+	}
+	return total
+}
+
 type nodeState struct {
 	NodeMetadata