use user context for metrics observation

share/availability/discovery/discovery.go

@@ -160,7 +160,7 @@ func (d *Discovery) Advertise(ctx context.Context) {
 	defer timer.Stop()
 	for {
 		ttl, err := d.disc.Advertise(ctx, rendezvousPoint)
-		d.metrics.observeAdvertise(err)
+		d.metrics.observeAdvertise(ctx, err)
 		if err != nil {
 			log.Debugf("Error advertising %s: %s", rendezvousPoint, err.Error())
 			if ctx.Err() != nil {
@@ -286,16 +286,16 @@ func (d *Discovery) discover(ctx context.Context) bool {
 		drainChannel(ticker.C)
 		select {
 		case <-findCtx.Done():
-			d.metrics.observeFindPeers(true, true)
+			d.metrics.observeFindPeers(ctx, true, true)
 			return true
 		case <-ticker.C:
-			d.metrics.observeDiscoveryStuck()
+			d.metrics.observeDiscoveryStuck(ctx)
 			log.Warn("wasn't able to find new peers for long time")
 			continue
 		case p, ok := <-peers:
 			if !ok {
 				isEnoughPeers := d.set.Size() >= d.set.Limit()
-				d.metrics.observeFindPeers(ctx.Err() != nil, isEnoughPeers)
+				d.metrics.observeFindPeers(ctx, ctx.Err() != nil, isEnoughPeers)
 				log.Debugw("discovery channel closed", "find_is_canceled", findCtx.Err() != nil)
 				return isEnoughPeers
 			}
@@ -333,19 +333,19 @@ func (d *Discovery) handleDiscoveredPeer(ctx context.Context, peer peer.AddrInfo
 	logger := log.With("peer", peer.ID)
 	switch {
 	case peer.ID == d.host.ID():
-		d.metrics.observeHandlePeer(handlePeerSkipSelf)
+		d.metrics.observeHandlePeer(ctx, handlePeerSkipSelf)
 		logger.Debug("skip handle: self discovery")
 		return false
 	case len(peer.Addrs) == 0:
-		d.metrics.observeHandlePeer(handlePeerEmptyAddrs)
+		d.metrics.observeHandlePeer(ctx, handlePeerEmptyAddrs)
 		logger.Debug("skip handle: empty address list")
 		return false
 	case d.set.Size() >= d.set.Limit():
-		d.metrics.observeHandlePeer(handlePeerEnoughPeers)
+		d.metrics.observeHandlePeer(ctx, handlePeerEnoughPeers)
 		logger.Debug("skip handle: enough peers found")
 		return false
 	case d.connector.HasBackoff(peer.ID):
-		d.metrics.observeHandlePeer(handlePeerBackoff)
+		d.metrics.observeHandlePeer(ctx, handlePeerBackoff)
 		logger.Debug("skip handle: backoff")
 		return false
 	}
@@ -356,7 +356,7 @@ func (d *Discovery) handleDiscoveredPeer(ctx context.Context, peer peer.AddrInfo
 	case network.NotConnected:
 		err := d.connector.Connect(ctx, peer)
 		if err != nil {
-			d.metrics.observeHandlePeer(handlePeerConnErr)
+			d.metrics.observeHandlePeer(ctx, handlePeerConnErr)
 			logger.Debugw("unable to connect", "err", err)
 			return false
 		}
@@ -365,12 +365,12 @@ func (d *Discovery) handleDiscoveredPeer(ctx context.Context, peer peer.AddrInfo
 	}
 
 	if !d.set.Add(peer.ID) {
-		d.metrics.observeHandlePeer(handlePeerInSet)
+		d.metrics.observeHandlePeer(ctx, handlePeerInSet)
 		logger.Debug("peer is already in discovery set")
 		return false
 	}
 	d.onUpdatedPeers(peer.ID, true)
-	d.metrics.observeHandlePeer(handlePeerConnected)
+	d.metrics.observeHandlePeer(ctx, handlePeerConnected)
 	logger.Debug("added peer to set")
 
 	// tag to protect peer from being killed by ConnManager

share/availability/discovery/metrics.go

@@ -3,8 +3,6 @@ package discovery
 import (
 	"context"
 	"fmt"
-	"time"
-
 	"github.com/libp2p/go-libp2p/core/peer"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric/global"
@@ -14,8 +12,6 @@ import (
 )
 
 const (
-	observeTimeout = 100 * time.Millisecond
-
 	discoveryEnougPeersKey  = "enough_peers"
 	discoveryFindCancledKey = "is_canceled"
 
@@ -125,35 +121,38 @@ func initMetrics(d *Discovery) (*metrics, error) {
 	return metrics, nil
 }
 
-func (m *metrics) observeFindPeers(canceled, isEnoughPeers bool) {
+func (m *metrics) observeFindPeers(ctx context.Context, canceled, isEnoughPeers bool) {
 	if m == nil {
 		return
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), observeTimeout)
-	defer cancel()
+	if ctx.Err() != nil {
+		ctx = context.Background()
+	}
 
 	m.discoveryResult.Add(ctx, 1,
 		attribute.Bool(discoveryFindCancledKey, canceled),
 		attribute.Bool(discoveryEnougPeersKey, isEnoughPeers))
 }
 
-func (m *metrics) observeHandlePeer(result handlePeerResult) {
+func (m *metrics) observeHandlePeer(ctx context.Context, result handlePeerResult) {
 	if m == nil {
 		return
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), observeTimeout)
-	defer cancel()
+	if ctx.Err() != nil {
+		ctx = context.Background()
+	}
 
 	m.handlePeerResult.Add(ctx, 1,
 		attribute.String(handlePeerResultKey, string(result)))
 }
 
-func (m *metrics) observeAdvertise(err error) {
+func (m *metrics) observeAdvertise(ctx context.Context, err error) {
 	if m == nil {
 		return
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), observeTimeout)
-	defer cancel()
+	if ctx.Err() != nil {
+		ctx = context.Background()
+	}
 
 	m.advertise.Add(ctx, 1,
 		attribute.Bool(advertiseFailedKey, err != nil))
@@ -163,8 +162,7 @@ func (m *metrics) observeOnPeersUpdate(_ peer.ID, isAdded bool) {
 	if m == nil {
 		return
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), observeTimeout)
-	defer cancel()
+	ctx := context.Background()
 
 	if isAdded {
 		m.peerAdded.Add(ctx, 1)
@@ -173,12 +171,13 @@ func (m *metrics) observeOnPeersUpdate(_ peer.ID, isAdded bool) {
 	m.peerRemoved.Add(ctx, 1)
 }
 
-func (m *metrics) observeDiscoveryStuck() {
+func (m *metrics) observeDiscoveryStuck(ctx context.Context) {
 	if m == nil {
 		return
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), observeTimeout)
-	defer cancel()
+	if ctx.Err() != nil {
+		ctx = context.Background()
+	}
 
 	m.discoveryStuck.Add(ctx, 1)
 }