fix(reacher): add metrics

Author: gacevicljubisa

pkg/p2p/libp2p/internal/reacher/metrics.go

@@ -0,0 +1,56 @@
+// Copyright 2021 The Swarm Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package reacher
+
+import (
+	m "github.com/ethersphere/bee/v2/pkg/metrics"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// metrics groups reacher related prometheus counters.
+type metrics struct {
+	Peers            prometheus.Gauge
+	PingAttemptCount prometheus.Counter
+	PingErrorCount   prometheus.Counter
+	PingDuration     prometheus.Histogram
+}
+
+// newMetrics is a convenient constructor for creating new metrics.
+func newMetrics() metrics {
+	const subsystem = "reacher"
+
+	return metrics{
+		Peers: prometheus.NewGauge(prometheus.GaugeOpts{
+			Namespace: m.Namespace,
+			Subsystem: subsystem,
+			Name:      "peers",
+			Help:      "Number of peers currently in the reacher queue.",
+		}),
+		PingAttemptCount: prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: m.Namespace,
+			Subsystem: subsystem,
+			Name:      "ping_attempt_count",
+			Help:      "Number of ping attempts.",
+		}),
+		PingErrorCount: prometheus.NewCounter(prometheus.CounterOpts{
+			Namespace: m.Namespace,
+			Subsystem: subsystem,
+			Name:      "ping_error_count",
+			Help:      "Number of failed ping attempts.",
+		}),
+		PingDuration: prometheus.NewHistogram(prometheus.HistogramOpts{
+			Namespace: m.Namespace,
+			Subsystem: subsystem,
+			Name:      "ping_duration_seconds",
+			Help:      "Ping latency distribution in seconds.",
+			Buckets:   []float64{.1, .25, .5, 1, 2, 5, 10, 15},
+		}),
+	}
+}
+
+// Metrics returns set of prometheus collectors.
+func (r *reacher) Metrics() []prometheus.Collector {
+	return m.PrometheusCollectorsFromFields(r.metrics)
+}

pkg/p2p/libp2p/internal/reacher/reacher.go

@@ -21,7 +21,7 @@ import (
 
 const (
 	pingTimeout               = time.Second * 15
-	workers                   = 4
+	workers                   = 8
 	retryAfterDuration        = time.Minute * 5
 	maxFailBackoffExponent    = 4   // caps failure backoff at retryAfterDuration * 2^4 = 80 min
 	maxSuccessBackoffExponent = 2   // caps success backoff at retryAfterDuration * 2^2 = 20 min
@@ -51,6 +51,7 @@ type reacher struct {
 
 	wg sync.WaitGroup
 
+	metrics metrics
 	options *Options
 	logger  log.Logger
 }
@@ -70,6 +71,7 @@ func New(streamer p2p.Pinger, notifier p2p.ReachableNotifier, o *Options, log lo
 		peerHeap:  make(peerHeap, 0),
 		peerIndex: make(map[string]*peer),
 		notifier:  notifier,
+		metrics:   newMetrics(),
 		logger:    log.WithName("reacher").Register(),
 	}
 
@@ -144,14 +146,19 @@ func (r *reacher) ping(c chan peer, ctx context.Context) {
 	defer r.wg.Done()
 	for p := range c {
 		func() {
+			r.metrics.PingAttemptCount.Inc()
 			ctxt, cancel := context.WithTimeout(ctx, r.options.PingTimeout)
 			defer cancel()
+			start := time.Now()
 			rtt, err := r.pinger.Ping(ctxt, p.addr)
 			if err != nil {
+				r.metrics.PingDuration.Observe(time.Since(start).Seconds())
+				r.metrics.PingErrorCount.Inc()
 				r.logger.Debug("ping failed", "peer", p.overlay.String(), "addr", p.addr.String(), "error", err)
 				r.notifier.Reachable(p.overlay, p2p.ReachabilityStatusPrivate)
 				r.notifyResult(p.overlay, false, p.generation)
 			} else {
+				r.metrics.PingDuration.Observe(rtt.Seconds())
 				r.logger.Debug("ping succeeded", "peer", p.overlay.String(), "addr", p.addr.String(), "rtt", rtt)
 				r.notifier.Reachable(p.overlay, p2p.ReachabilityStatusPublic)
 				r.notifyResult(p.overlay, true, p.generation)
@@ -210,6 +217,7 @@ func (r *reacher) Connected(overlay swarm.Address, addr ma.Multiaddr) {
 		p := &peer{overlay: overlay, addr: addr}
 		r.peerIndex[key] = p
 		heap.Push(&r.peerHeap, p)
+		r.metrics.Peers.Inc()
 	}
 
 	select {
@@ -267,6 +275,7 @@ func (r *reacher) Disconnected(overlay swarm.Address) {
 	if p, ok := r.peerIndex[key]; ok {
 		heap.Remove(&r.peerHeap, p.index)
 		delete(r.peerIndex, key)
+		r.metrics.Peers.Dec()
 	}
 }
 

pkg/p2p/libp2p/metrics.go

@@ -121,7 +121,11 @@ func newMetrics() metrics {
 }
 
 func (s *Service) Metrics() []prometheus.Collector {
-	return append(m.PrometheusCollectorsFromFields(s.metrics), s.handshakeService.Metrics()...)
+	collectors := append(m.PrometheusCollectorsFromFields(s.metrics), s.handshakeService.Metrics()...)
+	if mc, ok := s.reacher.(interface{ Metrics() []prometheus.Collector }); ok {
+		collectors = append(collectors, mc.Metrics()...)
+	}
+	return collectors
 }
 
 // StatusMetrics exposes metrics that are exposed on the status protocol.